.NET: [Bug]: Checkpointing in Handoff Workflow

[Abalast8899]

Description

While experimenting with checkpointing in Handoff Workflow, I tested several scenarios using the RestoreCheckpointAsync and ResumeStreamingAsync functions.

Ideally, when I load a checkpoint, it should resume or restore at this step and continue from there.

This works in a custom workflow, but I encountered several problems when trying to use checkpoints in Handoff.
For instance, when I resume the workflow, the WorkflowErrorEvent fires with the following message: "Expected exactly one update for key 'SharedState"

In situations where the AI requires tool approval and I use RestoreCheckpointAsync, the following error is thrown by my agentingStreamingFunctionMiddleware:
System.InvalidOperationException: 'ToolApprovalRequestContent found with FunctionCall.CallId(s) 'call_LaKjwuOiutjf280LhZmPTqHv' that have no matching ToolApprovalResponseContent.'

When I use ResumeStreamingAsync, it does not throw an exception, though the ExecutorFailed and WorkflowError events are fired.
[ExecutorFailed] weather_agent_933c703323e343f0a2480c3ec3707dc8: Cannot have multiple simultaneous conversations in Handoff Orchestration.
[WorkflowError] Error invoking handler for Microsoft.Agents.AI.Workflows.Specialized.HandoffState

I also tried using the HITL custom workflow as an agent, which created another problem. When I use checkpoints, my ChatClient Middleware throws an error: _'System.ClientModel.ClientResultException
HResult=0x80131500
Message=HTTP 400 (invalid_request_error: )
Parameter: messages.[3].role

An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_yjczbxhnnB9MzfKyWRAfCa4B'_

The custom AI combines the ChatProtocol Executor and the Guessing Game Example.
I also noticed that the chat messages contained many "User Null" elements and that the Tool_call ContentAI was missing.

I created a workaround by removing all the "User Null" elements, creating the missing "ToolMessages," and injecting them into the "ToolMessages." However, I don't believe the system is designed to function this way.

Code Sample

public static Workflow CreateHandoffWorkflow(AIAgent mainAgent, List<AIAgent> agents)
        {
            #pragma warning disable MAAIW001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed.
            HandoffWorkflowBuilder workflowBuilder = AgentWorkflowBuilder.CreateHandoffBuilderWith(mainAgent)
                .WithHandoffs(mainAgent, agents) // Triage can route to either specialist
                .WithHandoffs(agents, mainAgent)
                .WithToolCallFilteringBehavior(HandoffToolCallFilteringBehavior.None)
                .EnableReturnToPrevious()
                .EmitAgentResponseUpdateEvents();
                #pragma warning restore MAAIW001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed.

            Workflow workflow = workflowBuilder.Build();
            WorkflowVisualizerTool.PrintAll(workflow);

            return workflow;
        }

    public static async Task<(List<ChatMessage>, List<CheckpointInfo>?, CheckpointManager?)> RestoreWorkflowAsync(
    Workflow workflow,
    List<ChatMessage> messages,
    string workflowName = "MasterAgent")
    {
        List<CheckpointInfo>? checkpoints = new();
        CheckpointManager? checkpointManager = CheckpointManager.Default;


        StreamingRun run = await InProcessExecution.RunStreamingAsync(
            workflow,
            messages,
            checkpointManager: checkpointManager);
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

        Console.WriteLine();
        Console.ForegroundColor = ConsoleColor.Red;
        Console.WriteLine($"{workflowName} > ");

        await HandleWorkflowAsync(run, messages, checkpoints);

        Console.WriteLine();
        Console.WriteLine("Restore Checkpoint");
        Console.WriteLine($"Checkpoints: {checkpoints.Count}");

        int index = checkpoints.Count / 2;

        Console.WriteLine($"Middle Checkpoint {index}: {checkpoints[index].SessionId} (ID: {checkpoints[index].CheckpointId})");
        Console.WriteLine("Will start from the middle");
        Console.WriteLine();
        Console.WriteLine();

        Console.WriteLine();
        Console.ForegroundColor = ConsoleColor.Red;
        Console.WriteLine($"{workflowName} > ");

        await run.RestoreCheckpointAsync(checkpoints[index], CancellationToken.None);
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await HandleWorkflowAsync(run, messages, checkpoints);

        await run.DisposeAsync();

        return (messages, checkpoints, checkpointManager);
    }

    public static async Task<(List<ChatMessage>, List<CheckpointInfo>?, CheckpointManager?)> RehydratingWorkflowAsync(
    Workflow workflow,
    List<ChatMessage> messages = null,
    string workflowName = "MasterAgent")
    {

        CheckpointManager? checkpointManager = CheckpointManager.CreateInMemory();
        List<CheckpointInfo>? checkpoints = new();

        StreamingRun run = await InProcessExecution.RunStreamingAsync(
            workflow,
            messages,
            checkpointManager: checkpointManager);
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

        Console.WriteLine();
        Console.ForegroundColor = ConsoleColor.Red;
        Console.WriteLine($"{workflowName} > ");

        await HandleWorkflowAsync(run, messages, checkpoints);

        Console.WriteLine();
        Console.WriteLine("Restore Checkpoint");
        Console.WriteLine($"Checkpoints: {checkpoints.Count}");

        int index = checkpoints.Count / 2;

        Console.WriteLine($"Middle Checkpoint {index}: {checkpoints[index].SessionId} (ID: {checkpoints[index].CheckpointId})");
        Console.WriteLine("Will start from the middle");
        Console.WriteLine();
        Console.WriteLine();

        Console.WriteLine();
        Console.ForegroundColor = ConsoleColor.Red;
        Console.WriteLine($"{workflowName} > ");

        await run.DisposeAsync();

        run = await InProcessExecution.ResumeStreamingAsync(
            workflow,
            checkpoints[index],
            checkpointManager: checkpointManager,
            CancellationToken.None);
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await HandleWorkflowAsync(run, messages, checkpoints);

        await run.DisposeAsync();

        return (messages, checkpoints, checkpointManager);
    }

HandleWorkflowAsync is the simple "await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))" function.

Error Messages / Stack Traces

System.InvalidOperationException
  HResult=0x80131509
  Message=ToolApprovalRequestContent found with FunctionCall.CallId(s) 'call_nwH4mt3P0aJCQrNQ89iVB5bb' that have no matching ToolApprovalResponseContent.
  Source=Microsoft.Extensions.AI
  StackTrace:
   at Microsoft.Shared.Diagnostics.Throw.InvalidOperationException(String message)
   at Microsoft.Extensions.AI.FunctionInvokingChatClient.ExtractAndRemoveApprovalRequestsAndResponses(List`1 messages)
   at Microsoft.Extensions.AI.FunctionInvokingChatClient.ProcessFunctionApprovalResponses(List`1 originalMessages, Boolean hasConversationId, String toolMessageId, String functionCallContentFallbackMessageId)
   at Microsoft.Extensions.AI.FunctionInvokingChatClient.<GetStreamingResponseAsync>d__42.MoveNext()
   at Microsoft.Extensions.AI.FunctionInvokingChatClient.<GetStreamingResponseAsync>d__42.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.MoveNext()
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.MoveNext()
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.MoveNext()
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.MoveNext()
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.MoveNext()
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.MoveNext()
   at Microsoft.Extensions.AI.ConfigureOptionsChatClient.<GetStreamingResponseAsync>d__3.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Agents.AI.ChatClientAgent.<RunCoreStreamingAsync>d__29.MoveNext()
   at Microsoft.Agents.AI.ChatClientAgent.<RunCoreStreamingAsync>d__29.MoveNext()
   at Microsoft.Agents.AI.ChatClientAgent.<RunCoreStreamingAsync>d__29.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.MoveNext()
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.MoveNext()
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.MoveNext()
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.MoveNext()
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at AgentTestSystem.Services.Middleware.MiddlewareClass.<AgentStreamingMiddleware>d__1.MoveNext() in D:\Entw\DevPhilip\COPA.AI\COPA.AI.Validation\COPA.AI_AgentTools\AgentTestSystem\Services\Middleware\Middleware.cs:line 47
   at AgentTestSystem.Services.Middleware.MiddlewareClass.<AgentStreamingMiddleware>d__1.MoveNext() in D:\Entw\DevPhilip\COPA.AI\COPA.AI.Validation\COPA.AI_AgentTools\AgentTestSystem\Services\Middleware\Middleware.cs:line 47
   at AgentTestSystem.Services.Middleware.MiddlewareClass.<AgentStreamingMiddleware>d__1.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Agents.AI.AIAgent.<RunStreamingAsync>d__33.MoveNext()



System.ClientModel.ClientResultException
  HResult=0x80131500
  Message=HTTP 400 (invalid_request_error: )
Parameter: messages.[3].role

An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_yjczbxhnnB9MzfKyWRAfCa4B
  Source=OpenAI
  StackTrace:
   at OpenAI.ClientPipelineExtensions.<ProcessMessageAsync>d__0.MoveNext()
   at System.Runtime.CompilerServices.ConfiguredValueTaskAwaitable`1.ConfiguredValueTaskAwaiter.GetResult()
   at OpenAI.Chat.ChatClient.<CompleteChatAsync>d__41.MoveNext()
   at OpenAI.Chat.ChatClient.<CompleteChatAsync>d__15.MoveNext()
   at Microsoft.Extensions.AI.OpenAIChatClient.<GetResponseAsync>d__6.MoveNext()
   at AgentTestSystem.Services.Middleware.MiddlewareClass.<CustomChatClientMiddleware>d__3.MoveNext() in D:\Entw\DevPhilip\COPA.AI\COPA.AI.Validation\COPA.AI_AgentTools\AgentTestSystem\Services\Middleware\Middleware.cs:line 94
   at Microsoft.Extensions.AI.AnonymousDelegatingChatClient.<<GetStreamingResponseAsync>g__GetStreamingResponseAsyncViaGetResponseAsync|6_0>d.MoveNext()
   at Microsoft.Extensions.AI.AnonymousDelegatingChatClient.<<GetStreamingResponseAsync>g__GetStreamingResponseAsyncViaGetResponseAsync|6_0>d.System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult(Int16 token)
   at Microsoft.Extensions.AI.FunctionInvokingChatClient.<GetStreamingResponseAsync>d__42.MoveNext()

Package Versions

Microsoft.Agents.AI.Workflows" Version="1.3.0" , Microsoft.Agents.AI" Version="1.3.0"

.NET Version

.NET 10.0

Additional Context

No response

[lokitoth]

Hi @Abalast8899, thank you for filing this issue.

Still looking into the tool issue (that looks like #4544, and as you noted, seem to be a race condition)

For the resume issue: It looks like you are sending a second TurnToken after you rehydrate. Because the workflow already has a live TurnToken it is passing around, the second one will cause a parallel conversation to be occurring on top of the original one, causing multiple writes to shared state, which triggers the error.

Your scenario should work if you comment out the line:

// await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

[Abalast8899]

Hi @lokitoth, thank you for your suggestion.

I tried commenting out run.TrySendMessageAsync().
While this did not throw an error, it did not act as expected.

This happened when I tried to rehydrate a workflow that ran with a HITL.

I assume the problem is that I chose the middle point of my checkpoint list arbitrarily, because if I test it with the first element, the checkpoint works. The same applies to the agent response. If I choose a checkpoint from the middle, the agent does not answer me anymore (and I assume that I chose the checkpoint before the agent had answered), whereas testing with a checkpoint from the beginning worked fine.

I have a question. When I tried to resume a workflow from a checkpoint with a function that I had approved in the first run, the system did not ask for approval again when I resumed it, but called the function directly. I'm wondering whether that is an intended behaviour?

[lokitoth]

When I tried to resume a workflow from a checkpoint with a function that I had approved in the first run, the system did not ask for approval again when I resumed it, but called the function directly. I'm wondering whether that is an intended behaviour?

Resume is a bit tricky in that the checkpoint is always at the end of the previous step. If you resume from a checkpoint, the following step will run. If you restore to a checkpoint after the response message has been processed (and the request removed from pending), then you will not need to resubmit the response, otherwise it will treat it as a non-serviced request. Tool approvals are modeled as request/responses (at the Workflow layer) so they inherit this behaviour.

This happened when I tried to rehydrate a workflow that ran with a HITL.

This is likely the same issue as #5350

[Abalast8899]

@lokitoth I got another question about Checkpointing.

I tried to use checkpointing with a subwork inside of a Workflow, but that did not work as expected. Is it a known issue?

[lokitoth]

@Abalast8899 - re: Subworkflow not working right under checkpointing - I do not know of an issue; please file it if you can.

[pravin271]

I am experiencing a similar issue with a handoff workflow agent.

My setup consists of a handoff workflow agent that invokes a sub-agent, where the sub-agent itself is another workflow (BatchCreator).

The sub-workflow raises a RequestPort for the required input, which is converted into FunctionCallContent. My web client receives this and responds with the corresponding FunctionResultContent. This successfully resumes the workflow execution the first time on the server.

However, the BatchCreator sub-workflow subsequently raises another RequestPort for additional input. The client again sends the correct FunctionResultContent, but this time the workflow does not resume.

I am seeing the following exceptions:

1. Expected exactly one update for key 'SharedState' (this appears to occur during the first resume).
2. No pending request with ID '38:BatchCreator_BatchCreator_FunctionCall:787ef28bae2a404dbf1f42174da2c543' found in the workflow context. - This occurs during the second resume. However, while debugging, I can see the same ID inside WorkflowSession.PendingRequest.

A few additional observations:

1. For each RequestPort, two FunctionCallContent instances are being produced.
2. I am passing the same conversation ID from the client and creating or restoring the WorkflowSession using that conversation ID.

I am not sure how to work around this issue.