WorkflowSession.cs

// Copyright (c) Microsoft. All rights reserved.

using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows.Checkpointing;
using Microsoft.Agents.AI.Workflows.InProc;
using Microsoft.Extensions.AI;
using Microsoft.Shared.Diagnostics;

namespace Microsoft.Agents.AI.Workflows;

internal sealed class WorkflowSession : AgentSession
{
    private readonly Workflow _workflow;

    /// <summary>
    /// The execution environment for this session. Concrete type is required because
    /// <see cref="CreateOrResumeRunAsync"/> uses the internal
    /// <see cref="InProcessExecutionEnvironment.ResumeStreamingInternalAsync"/> API.
    /// </summary>
    private readonly InProcessExecutionEnvironment _inProcEnvironment;

    private readonly bool _includeExceptionDetails;
    private readonly bool _includeWorkflowOutputsInResponse;

    private InMemoryCheckpointManager? _inMemoryCheckpointManager;

    /// <summary>
    /// Tracks pending external requests by their workflow-facing request ID.
    /// This mapping enables converting incoming response content back to <see cref="ExternalResponse"/>
    /// when resuming a workflow from a checkpoint.
    /// </summary>
    /// <remarks>
    /// <para>
    /// Entries are added when a <see cref="RequestInfoEvent"/> is received during workflow execution,
    /// and removed when a matching response is delivered via <see cref="SendMessagesWithResponseConversionAsync"/>.
    /// </para>
    /// <para>
    /// The number of entries is bounded by the number of outstanding external requests in a single workflow run.
    /// When a session is abandoned, all pending requests are released with the session object.
    /// Request-level timeouts, if needed, should be implemented in the workflow definition itself
    /// (e.g., using a timer racing against an external event).
    /// </para>
    /// </remarks>
    private readonly Dictionary<string, ExternalRequest> _pendingRequests = [];

    internal static bool VerifyCheckpointingConfiguration(IWorkflowExecutionEnvironment executionEnvironment, [NotNullWhen(true)] out InProcessExecutionEnvironment? inProcEnv)
    {
        inProcEnv = null;
        if (executionEnvironment.IsCheckpointingEnabled)
        {
            return false;
        }

        if ((inProcEnv = executionEnvironment as InProcessExecutionEnvironment) == null)
        {
            throw new InvalidOperationException("Cannot use a non-checkpointed execution environment. Implicit checkpointing is supported only for InProcess.");
        }

        return true;
    }

    public WorkflowSession(Workflow workflow, string sessionId, IWorkflowExecutionEnvironment executionEnvironment, bool includeExceptionDetails = false, bool includeWorkflowOutputsInResponse = false)
    {
        this._workflow = Throw.IfNull(workflow);
        this._includeExceptionDetails = includeExceptionDetails;
        this._includeWorkflowOutputsInResponse = includeWorkflowOutputsInResponse;

        IWorkflowExecutionEnvironment env = Throw.IfNull(executionEnvironment);
        if (VerifyCheckpointingConfiguration(env, out InProcessExecutionEnvironment? inProcEnv))
        {
            // We have an InProcessExecutionEnvironment which is not configured for checkpointing. Ensure it has an externalizable checkpoint manager,
            // since we are responsible for maintaining the state.
            env = inProcEnv.WithCheckpointing(this.EnsureExternalizedInMemoryCheckpointing());
        }

        this._inProcEnvironment = env as InProcessExecutionEnvironment
            ?? throw new InvalidOperationException(
                $"WorkflowSession requires an {nameof(InProcessExecutionEnvironment)}, " +
                $"but received {env.GetType().Name}.");

        this.SessionId = Throw.IfNullOrEmpty(sessionId);
        this.ChatHistoryProvider = new WorkflowChatHistoryProvider();
    }

    private CheckpointManager EnsureExternalizedInMemoryCheckpointing()
    {
        return new(this._inMemoryCheckpointManager ??= new());
    }

    public WorkflowSession(Workflow workflow, JsonElement serializedSession, IWorkflowExecutionEnvironment executionEnvironment, bool includeExceptionDetails = false, bool includeWorkflowOutputsInResponse = false, JsonSerializerOptions? jsonSerializerOptions = null)
    {
        this._workflow = Throw.IfNull(workflow);
        this._includeExceptionDetails = includeExceptionDetails;
        this._includeWorkflowOutputsInResponse = includeWorkflowOutputsInResponse;

        IWorkflowExecutionEnvironment env = Throw.IfNull(executionEnvironment);

        JsonMarshaller marshaller = new(jsonSerializerOptions);
        SessionState sessionState = marshaller.Marshal<SessionState>(serializedSession);

        this._inMemoryCheckpointManager = sessionState.CheckpointManager;
        if (this._inMemoryCheckpointManager != null &&
            VerifyCheckpointingConfiguration(env, out InProcessExecutionEnvironment? inProcEnv))
        {
            env = inProcEnv.WithCheckpointing(this.EnsureExternalizedInMemoryCheckpointing());
        }
        else if (this._inMemoryCheckpointManager != null)
        {
            throw new ArgumentException("The session was saved with an externalized checkpoint manager, but the incoming execution environment does not support it.", nameof(executionEnvironment));
        }

        this._inProcEnvironment = env as InProcessExecutionEnvironment
            ?? throw new InvalidOperationException(
                $"WorkflowSession requires an {nameof(InProcessExecutionEnvironment)}, " +
                $"but received {env.GetType().Name}.");

        this.SessionId = sessionState.SessionId;
        this.ChatHistoryProvider = new WorkflowChatHistoryProvider();

        this.LastCheckpoint = sessionState.LastCheckpoint;
        this.StateBag = sessionState.StateBag;
        this._pendingRequests = sessionState.PendingRequests ?? [];
    }

    public CheckpointInfo? LastCheckpoint { get; set; }

    internal JsonElement Serialize(JsonSerializerOptions? jsonSerializerOptions = null)
    {
        JsonMarshaller marshaller = new(jsonSerializerOptions);
        SessionState info = new(
            this.SessionId,
            this.LastCheckpoint,
            this._inMemoryCheckpointManager,
            this.StateBag,
            this._pendingRequests);

        return marshaller.Marshal(info);
    }

    public AgentResponseUpdate CreateUpdate(string responseId, object raw, params AIContent[] parts)
    {
        Throw.IfNullOrEmpty(parts);

        return new(ChatRole.Assistant, parts)
        {
            CreatedAt = DateTimeOffset.UtcNow,
            MessageId = Guid.NewGuid().ToString("N"),
            Role = ChatRole.Assistant,
            ResponseId = responseId,
            RawRepresentation = raw
        };
    }

    public AgentResponseUpdate CreateUpdate(string responseId, object raw, ChatMessage message)
    {
        Throw.IfNull(message);

        return new(message.Role, message.Contents)
        {
            CreatedAt = message.CreatedAt ?? DateTimeOffset.UtcNow,
            MessageId = message.MessageId ?? Guid.NewGuid().ToString("N"),
            ResponseId = responseId,
            RawRepresentation = raw
        };
    }

    private async ValueTask<ResumeRunResult> CreateOrResumeRunAsync(List<ChatMessage> messages, CancellationToken cancellationToken = default)
    {
        // The workflow is validated to be a ChatProtocol workflow by the WorkflowHostAgent before creating the session,
        // and does not need to be checked again here.
        if (this.LastCheckpoint is not null)
        {
            // Use the internal resume path that suppresses pending request republishing.
            // WorkflowSession handles pending requests itself by converting matching responses
            // via SendMessagesWithResponseConversionAsync, so event-stream republishing would
            // cause unwanted duplicate events visible to the consumer.
            StreamingRun run =
                await this._inProcEnvironment
                            .ResumeStreamingInternalAsync(this._workflow,
                                               this.LastCheckpoint,
                                               republishPendingEvents: false,
                                               cancellationToken)
                            .ConfigureAwait(false);

            // Process messages: convert response content to ExternalResponse, send regular messages as-is
            ResumeDispatchInfo dispatchInfo = await this.SendMessagesWithResponseConversionAsync(run, messages).ConfigureAwait(false);
            return new ResumeRunResult(run, dispatchInfo);
        }

        StreamingRun newRun = await this._inProcEnvironment
                            .RunStreamingAsync(this._workflow,
                                         messages,
                                         this.SessionId,
                                         cancellationToken)
                            .ConfigureAwait(false);
        return new ResumeRunResult(newRun);
    }

    /// <summary>
    /// Sends messages to the run, converting FunctionResultContent and UserInputResponseContent
    /// to ExternalResponse when there's a matching pending request.
    /// </summary>
    /// <returns>
    /// Structured information about how resume content was dispatched.
    /// </returns>
    private async ValueTask<ResumeDispatchInfo> SendMessagesWithResponseConversionAsync(StreamingRun run, List<ChatMessage> messages)
    {
        List<ChatMessage> regularMessages = [];
        // Responses are deferred until after regular messages are queued so response handlers
        // can merge buffered regular content in the same continuation turn.
        List<(ExternalResponse Response, string RequestId)> externalResponses = [];
        bool hasMatchedResponseForStartExecutor = false;

        // Tracks content IDs already matched to pending requests within this invocation,
        // preventing duplicate responses for the same ID from being sent to the workflow engine.
        HashSet<string>? matchedContentIds = null;

        foreach (ChatMessage message in messages)
        {
            List<AIContent> regularContents = [];

            foreach (AIContent content in message.Contents)
            {
                string? contentId = GetResponseContentId(content);

                // Skip duplicate response content for an already-matched content ID
                if (contentId != null && matchedContentIds?.Contains(contentId) == true)
                {
                    continue;
                }

                if (contentId != null
                    && this.TryGetPendingRequest(contentId) is ExternalRequest pendingRequest)
                {
                    // For intercepted/complex topologies the port may not be registered in the EdgeMap.
                    // Treat unknown port as non-start-executor (conservative): TurnToken will still be sent.
                    if (run.TryGetResponsePortExecutorId(pendingRequest.PortInfo.PortId, out string? responseExecutorId))
                    {
                        hasMatchedResponseForStartExecutor |= string.Equals(responseExecutorId, this._workflow.StartExecutorId, StringComparison.Ordinal);
                    }

                    object normalizedResponseContent = NormalizeResponseContentForDelivery(content, pendingRequest);
                    externalResponses.Add((pendingRequest.CreateResponse(normalizedResponseContent), pendingRequest.RequestId));
                    (matchedContentIds ??= new(StringComparer.Ordinal)).Add(contentId);
                }
                else
                {
                    regularContents.Add(content);
                }
            }

            if (regularContents.Count > 0)
            {
                ChatMessage cloned = message.Clone();
                cloned.Contents = regularContents;
                regularMessages.Add(cloned);
            }
        }

        // Send regular messages first so response handlers can merge them with responses.
        bool hasRegularMessages = regularMessages.Count > 0;
        if (hasRegularMessages)
        {
            await run.TrySendMessageAsync(regularMessages).ConfigureAwait(false);
        }

        // Send external responses after regular messages.
        bool hasMatchedExternalResponses = false;
        foreach ((ExternalResponse response, string requestId) in externalResponses)
        {
            await run.SendResponseAsync(response).ConfigureAwait(false);
            hasMatchedExternalResponses = true;
            this.RemovePendingRequest(requestId);
        }

        return new ResumeDispatchInfo(
            hasRegularMessages,
            hasMatchedExternalResponses,
            hasMatchedResponseForStartExecutor);
    }

    /// <summary>
    /// Resolves the concrete request payload type from <see cref="RequestPortInfo.RequestType"/>
    /// and returns it as an <see cref="IExternalRequestEnvelope"/> if the type implements that
    /// abstraction. Resolving via the concrete <see cref="TypeId"/> (rather than asking the
    /// PortableValue to deserialize directly to <see cref="IExternalRequestEnvelope"/>) is
    /// required because checkpointed payloads round-trip as JSON which cannot be deserialized
    /// to an interface; the concrete type populates the deserialization cache so subsequent
    /// interface assignment succeeds.
    /// </summary>
    [UnconditionalSuppressMessage("Trimming", "IL2057:Unrecognized value passed to the parameter of method", Justification = "Higher-layer envelope types are explicitly preserved by the package that defines them.")]
    private static bool TryGetRequestEnvelope(ExternalRequest request, [NotNullWhen(true)] out IExternalRequestEnvelope? envelope)
    {
        envelope = null;

        TypeId requestType = request.PortInfo.RequestType;
        Type? concreteType = Type.GetType($"{requestType.TypeName}, {requestType.AssemblyName}", throwOnError: false);
        if (concreteType is null || !typeof(IExternalRequestEnvelope).IsAssignableFrom(concreteType))
        {
            return false;
        }

        if (!request.TryGetDataAs(concreteType, out object? data) || data is not IExternalRequestEnvelope env)
        {
            return false;
        }

        envelope = env;
        return true;
    }

    /// <summary>
    /// Creates the workflow-facing request content surfaced in response updates.
    /// </summary>
    private static AIContent CreateRequestContentForDelivery(ExternalRequest request)
    {
        // If the request payload is a higher-layer envelope (e.g., a declarative
        // ExternalInputRequest), surface its inner FCC/TARC to the host on the wire.
        if (TryGetRequestEnvelope(request, out IExternalRequestEnvelope? envelope))
        {
            AIContent? inner = envelope.GetInnerRequestContent();
            if (inner is ToolApprovalRequestContent toolApprovalRequest)
            {
                return CloneToolApprovalRequestContent(toolApprovalRequest, request.RequestId);
            }
            if (inner is FunctionCallContent functionCall)
            {
                return CloneFunctionCallContent(functionCall, request.RequestId);
            }
        }

        return request switch
        {
            ExternalRequest externalRequest when externalRequest.TryGetDataAs(out FunctionCallContent? functionCallContent)
                => CloneFunctionCallContent(functionCallContent, externalRequest.RequestId),
            ExternalRequest externalRequest when externalRequest.TryGetDataAs(out ToolApprovalRequestContent? toolApprovalRequestContent)
                => CloneToolApprovalRequestContent(toolApprovalRequestContent, externalRequest.RequestId),
            ExternalRequest externalRequest
                => externalRequest.ToFunctionCall(),
        };
    }

    /// <summary>
    /// Rewrites workflow-facing response content back to the original agent-owned content ID.
    /// </summary>
    private static object NormalizeResponseContentForDelivery(AIContent content, ExternalRequest request)
    {
        // If the request payload is a higher-layer envelope, recover the original
        // CallId/RequestId from the inner content and ask the envelope to wrap the
        // response back into its paired response type for delivery to the request port.
        if (TryGetRequestEnvelope(request, out IExternalRequestEnvelope? envelope))
        {
            AIContent? inner = envelope.GetInnerRequestContent();
            AIContent payload = (content, inner) switch
            {
                (FunctionResultContent functionResult, FunctionCallContent functionCall)
                    => CloneFunctionResultContent(functionResult, functionCall.CallId),
                (FunctionResultContent functionResult, ToolApprovalRequestContent toolApprovalRequest)
                    => CloneFunctionResultContent(functionResult, toolApprovalRequest.ToolCall.CallId),
                (ToolApprovalResponseContent toolApprovalResponse, ToolApprovalRequestContent toolApprovalRequest)
                    => CloneToolApprovalResponseContent(toolApprovalResponse, toolApprovalRequest.RequestId),
                _ => content,
            };

            ChatMessage message = new(ChatRole.Tool, [payload]);
            return envelope.CreateResponse([message]);
        }

        switch (content)
        {
            // If we got a FRC, and were expecting a FRC (because the request started out as a FCC, rather than getting converted to
            // on at the WorkflowSession boundary), clone it and send it in.
            case FunctionResultContent functionResultContent when request.TryGetDataAs(out FunctionCallContent? functionCallContent):
                return CloneFunctionResultContent(functionResultContent, functionCallContent.CallId);
            case FunctionResultContent functionResultContent when !request.PortInfo.ResponseType.IsMatchPolymorphic(typeof(FunctionResultContent)):
            {
                object? result = functionResultContent.Result;
                if (result != null)
                {
                    if (request.PortInfo.ResponseType.IsMatchPolymorphic(result.GetType()) || result is PortableValue)
                    {
                        return result;
                    }

                    throw new InvalidOperationException($"Unexpected result type in FunctionResultContent {result.GetType()}; expecting {request.PortInfo.ResponseType}");
                }

                throw new NotSupportedException($"Null result is not supported when using RequestPort with non-AIContent-typed requests. {functionResultContent}");
            }
            case ToolApprovalResponseContent toolApprovalResponseContent when request.TryGetDataAs(out ToolApprovalRequestContent? toolApprovalRequestContent):
                return CloneToolApprovalResponseContent(toolApprovalResponseContent, toolApprovalRequestContent.RequestId);
            default:
                return content;
        }
    }

    /// <summary>
    /// Gets the workflow-facing request ID from response content types.
    /// </summary>
    private static string? GetResponseContentId(AIContent content) => content switch
    {
        FunctionResultContent functionResultContent => functionResultContent.CallId,
        ToolApprovalResponseContent toolApprovalResponseContent => toolApprovalResponseContent.RequestId,
        _ => null
    };

    /// <summary>
    /// Tries to get a pending request by workflow-facing request ID.
    /// </summary>
    private ExternalRequest? TryGetPendingRequest(string requestId) =>
        this._pendingRequests.TryGetValue(requestId, out ExternalRequest? request) ? request : null;

    /// <summary>
    /// Adds a pending request indexed by workflow-facing request ID.
    /// </summary>
    private void AddPendingRequest(string requestId, ExternalRequest request) => this._pendingRequests[requestId] = request;

    /// <summary>
    /// Removes a pending request by workflow-facing request ID.
    /// </summary>
    private void RemovePendingRequest(string requestId) =>
        this._pendingRequests.Remove(requestId);

    internal async
    IAsyncEnumerable<AgentResponseUpdate> InvokeStageAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        this.LastResponseId = Guid.NewGuid().ToString("N");
        List<ChatMessage> messages = this.ChatHistoryProvider.GetFromBookmark(this).ToList();

        ResumeRunResult resumeResult =
            await this.CreateOrResumeRunAsync(messages, cancellationToken).ConfigureAwait(false);

#pragma warning disable CA2007 // Analyzer misfiring.
        await using StreamingRun run = resumeResult.Run;
#pragma warning restore CA2007

        ResumeDispatchInfo dispatchInfo = resumeResult.DispatchInfo;

        // Send a TurnToken to the start executor unless the only activity is an external
        // response directed at the start executor itself (which self-emits a TurnToken via
        // ContinueTurnAsync). Non-start executors (e.g., RequestInfoExecutor) do not emit
        // TurnTokens after processing responses, so the session must always provide one.
        bool shouldSendTurnToken =
            !dispatchInfo.HasMatchedExternalResponses
            || !dispatchInfo.HasMatchedResponseForStartExecutor;
        if (shouldSendTurnToken)
        {
            await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
        }
        await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken)
                                               .ConfigureAwait(false)
                                               .WithCancellation(cancellationToken))
        {
            switch (evt)
            {
                case AgentResponseUpdateEvent agentUpdate:
                    yield return agentUpdate.Update;
                    break;

                case RequestInfoEvent requestInfo:
                    AIContent requestContent = CreateRequestContentForDelivery(requestInfo.Request);

                    // Track the pending request so we can convert incoming responses back to ExternalResponse.
                    // External callers respond using the workflow-facing request ID, which is always RequestId.
                    this.AddPendingRequest(requestInfo.Request.RequestId, requestInfo.Request);

                    AgentResponseUpdate update = this.CreateUpdate(this.LastResponseId, evt, requestContent);
                    yield return update;
                    break;

                case AgentAIContextProviderMsgEvent requestMessages:
                    // Add the message in the AIContentProvider to the ChatHistoryProvider of the Workflow.
                    this.ChatHistoryProvider.AddMessages(this, requestMessages.Messages);
                    break;

                case WorkflowErrorEvent workflowError:
                    Exception? exception = workflowError.Exception;
                    if (exception is TargetInvocationException tie && tie.InnerException != null)
                    {
                        exception = tie.InnerException;
                    }

                    if (exception != null)
                    {
                        string message = this._includeExceptionDetails
                                       ? exception.Message
                                       : "An error occurred while executing the workflow.";

                        ErrorContent errorContent = new(message);
                        yield return this.CreateUpdate(this.LastResponseId, evt, errorContent);
                    }

                    break;

                case ExecutorFailedEvent executorFailed:
                    // Mirror WorkflowErrorEvent: never expose internal workflow graph
                    // identifiers (executor ID) to the client. Surface the exception
                    // message only when the host opts in via _includeExceptionDetails.
                    Exception? executorException = executorFailed.Data;
                    while (executorException is { InnerException: not null }
                        && (executorException is TargetInvocationException
                            || executorException.GetType().Name == "DeclarativeActionException"))
                    {
                        executorException = executorException.InnerException;
                    }

                    string executorMessage = this._includeExceptionDetails && executorException != null
                        ? executorException.Message
                        : "An error occurred while executing the workflow.";

                    yield return this.CreateUpdate(this.LastResponseId, evt, new ErrorContent(executorMessage));
                    break;

                case SuperStepCompletedEvent stepCompleted:
                    this.LastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint;
                    goto default;

                case AgentResponseEvent agentResponse:
                    // Under Futures.EnableAgentResponseOutputTaggingAndFiltering=true, mirror
                    // AgentResponseUpdateEvent's behavior: always forward, regardless of the
                    // _includeWorkflowOutputsInResponse host flag / "intermediate" tag. Under
                    // the legacy default, keep today's behavior — gated by the include flag.
                    if (!Futures.EnableAgentResponseOutputTaggingAndFiltering && !this._includeWorkflowOutputsInResponse)
                    {
                        goto default;
                    }

                    // Either EnableAgentResponseOutputTaggingAndFiltering -- so yield the Response
                    // regardless of whether it is tagged "intermediate" or whether the
                    // _includeWorkflowOutputInResponse flag is set. Reason being: The user specifies
                    // exclusion of an event by enabling filtering and then _not_ marking an Executor
                    // as an output executor.
                    foreach (ChatMessage message in agentResponse.Response.Messages)
                    {
                        yield return this.CreateUpdate(this.LastResponseId, evt, message);
                    }
                    break;

                case WorkflowOutputEvent output:
                    IEnumerable<ChatMessage>? updateMessages = output.Data switch
                    {
                        IEnumerable<ChatMessage> chatMessages => chatMessages,
                        ChatMessage chatMessage => [chatMessage],
                        _ => null
                    };

                    // Same assymetry as with AgentResponseEvent, but there is no EnableFiltering flag
                    // to consider. If this made it here (and since it is not an AgentResponse[Update]),
                    // it means it is already been selected as an Output() from the user. Intermediate
                    // is irrelevant here.
                    if (updateMessages == null || !this._includeWorkflowOutputsInResponse)
                    {
                        goto default;
                    }

                    foreach (ChatMessage message in updateMessages)
                    {
                        yield return this.CreateUpdate(this.LastResponseId, evt, message);
                    }
                    break;

                default:
                    // Emit all other workflow events for observability (DevUI, logging, etc.)
                    yield return new AgentResponseUpdate(ChatRole.Assistant, [])
                    {
                        CreatedAt = DateTimeOffset.UtcNow,
                        MessageId = Guid.NewGuid().ToString("N"),
                        Role = ChatRole.Assistant,
                        ResponseId = this.LastResponseId,
                        RawRepresentation = evt
                    };
                    break;
            }
        }
    }

    public string? LastResponseId { get; set; }

    public string SessionId { get; }

    /// <inheritdoc/>
    public WorkflowChatHistoryProvider ChatHistoryProvider { get; }

    /// <summary>
    /// Captures the outcome of creating or resuming a workflow run,
    /// indicating what types of messages were sent during resume.
    /// </summary>
    private readonly struct ResumeRunResult
    {
        /// <summary>The streaming run that was created or resumed.</summary>
        public StreamingRun Run { get; }

        /// <summary>How resume-time content was dispatched into the workflow runtime.</summary>
        public ResumeDispatchInfo DispatchInfo { get; }

        public ResumeRunResult(StreamingRun run, ResumeDispatchInfo dispatchInfo = default)
        {
            this.Run = Throw.IfNull(run);
            this.DispatchInfo = dispatchInfo;
        }
    }

    /// <summary>
    /// Captures how resumed input was split across regular-message and external-response delivery paths.
    /// </summary>
    private readonly struct ResumeDispatchInfo
    {
        public ResumeDispatchInfo(bool hasRegularMessages, bool hasMatchedExternalResponses, bool hasMatchedResponseForStartExecutor)
        {
            this.HasRegularMessages = hasRegularMessages;
            this.HasMatchedExternalResponses = hasMatchedExternalResponses;
            this.HasMatchedResponseForStartExecutor = hasMatchedResponseForStartExecutor;
        }

        public bool HasRegularMessages { get; }

        public bool HasMatchedExternalResponses { get; }

        public bool HasMatchedResponseForStartExecutor { get; }
    }

    /// <summary>
    /// Clones a <see cref="FunctionCallContent"/> with a workflow-facing call ID.
    /// </summary>
    private static FunctionCallContent CloneFunctionCallContent(FunctionCallContent content, string callId)
    {
        FunctionCallContent clone = new(callId, content.Name, content.Arguments)
        {
            Exception = content.Exception,
            InformationalOnly = content.InformationalOnly,
        };

        return CopyContentMetadata(content, clone);
    }

    /// <summary>
    /// Clones a <see cref="FunctionResultContent"/> with an agent-owned call ID.
    /// </summary>
    private static FunctionResultContent CloneFunctionResultContent(FunctionResultContent content, string callId)
    {
        FunctionResultContent clone = new(callId, content.Result)
        {
            Exception = content.Exception,
        };

        return CopyContentMetadata(content, clone);
    }

    /// <summary>
    /// Clones a <see cref="ToolApprovalRequestContent"/> with a workflow-facing request ID.
    /// </summary>
    private static ToolApprovalRequestContent CloneToolApprovalRequestContent(ToolApprovalRequestContent content, string id)
    {
        ToolApprovalRequestContent clone = new(id, content.ToolCall);
        return CopyContentMetadata(content, clone);
    }

    /// <summary>
    /// Clones a <see cref="ToolApprovalResponseContent"/> with an agent-owned request ID.
    /// </summary>
    private static ToolApprovalResponseContent CloneToolApprovalResponseContent(ToolApprovalResponseContent content, string id)
    {
        ToolApprovalResponseContent clone = new(id, content.Approved, content.ToolCall)
        {
            Reason = content.Reason,
        };

        return CopyContentMetadata(content, clone);
    }

    /// <summary>
    /// Copies shared <see cref="AIContent"/> metadata to a cloned content instance.
    /// </summary>
    private static TContent CopyContentMetadata<TContent>(AIContent source, TContent target)
        where TContent : AIContent
    {
        target.AdditionalProperties = source.AdditionalProperties;
        target.Annotations = source.Annotations;
        target.RawRepresentation = source.RawRepresentation;
        return target;
    }

    internal sealed class SessionState(
        string sessionId,
        CheckpointInfo? lastCheckpoint,
        InMemoryCheckpointManager? checkpointManager = null,
        AgentSessionStateBag? stateBag = null,
        Dictionary<string, ExternalRequest>? pendingRequests = null)
    {
        public string SessionId { get; } = sessionId;
        public CheckpointInfo? LastCheckpoint { get; } = lastCheckpoint;
        public InMemoryCheckpointManager? CheckpointManager { get; } = checkpointManager;
        public AgentSessionStateBag StateBag { get; } = stateBag ?? new();
        public Dictionary<string, ExternalRequest>? PendingRequests { get; } = pendingRequests;
    }
}