Merge branch 'main' into copilot/add-preconfigured-compaction-strategy

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs

@@ -167,6 +167,20 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
             return;
         }
 
+        if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowMcpToolFunctionEntryPoint)
+        {
+            if (mcpToolInvocationContext is null)
+            {
+                throw new InvalidOperationException($"MCP tool invocation context binding is missing for the invocation {context.InvocationId}.");
+            }
+
+            context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowMcpToolAsync(
+                mcpToolInvocationContext,
+                durableTaskClient,
+                context);
+            return;
+        }
+
         throw new InvalidOperationException($"Unsupported function entry point '{context.FunctionDefinition.EntryPoint}' for invocation {context.InvocationId}.");
     }
 

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs

@@ -29,6 +29,7 @@ internal static class BuiltInFunctions
     internal static readonly string InvokeWorkflowActivityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeWorkflowActivityAsync)}";
     internal static readonly string GetWorkflowStatusHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(GetWorkflowStatusAsync)}";
     internal static readonly string RespondToWorkflowHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RespondToWorkflowAsync)}";
+    internal static readonly string RunWorkflowMcpToolFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowMcpToolAsync)}";
 
 #pragma warning disable IL3000 // Avoid accessing Assembly file path when publishing as a single file - Azure Functions does not use single-file publishing
     internal static readonly string ScriptFile = Path.GetFileName(typeof(BuiltInFunctions).Assembly.Location);
@@ -378,6 +379,55 @@ internal static class BuiltInFunctions
         return agentResponse.Text;
     }
 
+    /// <summary>
+    /// Runs a workflow via MCP tool trigger.
+    /// Extracts the <c>input</c> argument, schedules a new orchestration, waits for completion, and returns the output.
+    /// </summary>
+    public static async Task<string?> RunWorkflowMcpToolAsync(
+        [McpToolTrigger("BuiltInWorkflowMcpTool")] ToolInvocationContext context,
+        [DurableClient] DurableTaskClient client,
+        FunctionContext functionContext)
+    {
+        if (context.Arguments is null)
+        {
+            throw new ArgumentException("MCP Tool invocation is missing required arguments.");
+        }
+
+        if (!context.Arguments.TryGetValue("input", out object? inputObj) || inputObj is not string input)
+        {
+            throw new ArgumentException("MCP Tool invocation is missing required 'input' argument of type string.");
+        }
+
+        string workflowName = context.Name;
+        string orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName);
+
+        DurableWorkflowInput<string> orchestrationInput = new() { Input = input };
+        string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, orchestrationInput);
+
+        OrchestrationMetadata? metadata = await client.WaitForInstanceCompletionAsync(
+            instanceId,
+            getInputsAndOutputs: true,
+            cancellation: functionContext.CancellationToken);
+
+        if (metadata is null)
+        {
+            throw new InvalidOperationException($"Workflow orchestration '{instanceId}' returned no metadata.");
+        }
+
+        if (metadata.RuntimeStatus is OrchestrationRuntimeStatus.Failed)
+        {
+            string errorMessage = metadata.FailureDetails?.ErrorMessage ?? "Unknown error";
+            throw new InvalidOperationException($"Workflow orchestration '{instanceId}' failed: {errorMessage}");
+        }
+
+        if (metadata.RuntimeStatus is not OrchestrationRuntimeStatus.Completed)
+        {
+            throw new InvalidOperationException($"Workflow orchestration '{instanceId}' ended with unexpected status '{metadata.RuntimeStatus}'.");
+        }
+
+        return metadata.ReadOutputAs<DurableWorkflowResult>()?.Result;
+    }
+
     /// <summary>
     /// Creates an error response with the specified status code and error message.
     /// </summary>

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [Unreleased]
 
+- Added MCP tool trigger support for durable workflows ([#4768](https://github.com/microsoft/agent-framework/pull/4768))
 - Added Azure Functions hosting support for durable workflows ([#4436](https://github.com/microsoft/agent-framework/pull/4436))
 
 ## v1.0.0-preview.251219.1

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DefaultFunctionsAgentOptionsProvider.cs

@@ -6,7 +6,8 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
 
 /// <summary>
 /// Provides access to agent-specific options for functions agents by name.
-/// Returns default options (HTTP trigger enabled, MCP tool disabled) when no explicit options were configured.
+/// Returns <see langword="false"/> when no explicit options have been configured for an agent,
+/// which distinguishes standalone agents from those auto-registered by workflows.
 /// </summary>
 internal sealed class DefaultFunctionsAgentOptionsProvider(IReadOnlyDictionary<string, FunctionsAgentOptions> functionsAgentOptions)
     : IFunctionsAgentOptionsProvider
@@ -14,32 +15,19 @@ internal sealed class DefaultFunctionsAgentOptionsProvider(IReadOnlyDictionary<s
     private readonly IReadOnlyDictionary<string, FunctionsAgentOptions> _functionsAgentOptions =
         functionsAgentOptions ?? throw new ArgumentNullException(nameof(functionsAgentOptions));
 
-    // Default options. HTTP trigger enabled, MCP tool disabled.
-    private static readonly FunctionsAgentOptions s_defaultOptions = new()
-    {
-        HttpTrigger = { IsEnabled = true },
-        McpToolTrigger = { IsEnabled = false }
-    };
-
     /// <summary>
     /// Attempts to retrieve the options associated with the specified agent name.
-    /// If not found, a default options instance (with HTTP trigger enabled) is returned.
+    /// Returns <see langword="false"/> when no options have been explicitly configured for the agent.
     /// </summary>
     /// <param name="agentName">The name of the agent whose options are to be retrieved. Cannot be null or empty.</param>
-    /// <param name="options">The options for the specified agent. Will never be null.</param>
-    /// <returns>Always true. Returns configured options if present; otherwise default fallback options.</returns>
+    /// <param name="options">
+    /// When this method returns <see langword="true"/>, contains the options for the specified agent;
+    /// otherwise, <see langword="null"/>.
+    /// </param>
+    /// <returns><see langword="true"/> if options were found for the agent; otherwise, <see langword="false"/>.</returns>
     public bool TryGet(string agentName, [NotNullWhen(true)] out FunctionsAgentOptions? options)
     {
         ArgumentException.ThrowIfNullOrEmpty(agentName);
-
-        if (this._functionsAgentOptions.TryGetValue(agentName, out FunctionsAgentOptions? existing))
-        {
-            options = existing;
-            return true;
-        }
-
-        // If not defined, return default options.
-        options = s_defaultOptions;
-        return true;
+        return this._functionsAgentOptions.TryGetValue(agentName, out options);
     }
 }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentFunctionMetadataTransformer.cs

@@ -6,9 +6,13 @@ using Microsoft.Extensions.Logging;
 namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
 
 /// <summary>
-/// Transforms function metadata by registering durable agent functions for each configured agent.
+/// Transforms function metadata by registering durable agent functions for each explicitly configured agent.
 /// </summary>
-/// <remarks>This transformer adds both entity trigger and HTTP trigger functions for every agent registered in the application.</remarks>
+/// <remarks>
+/// This transformer adds entity, HTTP, and MCP tool trigger functions for agents that have
+/// explicit <see cref="FunctionsAgentOptions"/>. Agents auto-registered by workflows
+/// (which lack explicit options) are handled by <see cref="DurableWorkflowsFunctionMetadataTransformer"/>.
+/// </remarks>
 internal sealed class DurableAgentFunctionMetadataTransformer : IFunctionMetadataTransformer
 {
     private readonly ILogger<DurableAgentFunctionMetadataTransformer> _logger;
@@ -38,24 +42,27 @@ internal sealed class DurableAgentFunctionMetadataTransformer : IFunctionMetadat
         {
             string agentName = kvp.Key;
 
-            this._logger.LogRegisteringTriggerForAgent(agentName, "entity");
+            // Only generate triggers for agents with explicit Functions agent options.
+            // Agents auto-registered by workflows are handled by DurableWorkflowsFunctionMetadataTransformer.
+            if (!this._functionsAgentOptionsProvider.TryGet(agentName, out FunctionsAgentOptions? agentTriggerOptions))
+            {
+                continue;
+            }
 
+            this._logger.LogRegisteringTriggerForAgent(agentName, "entity");
             original.Add(FunctionMetadataFactory.CreateEntityTrigger(agentName));
 
-            if (this._functionsAgentOptionsProvider.TryGet(agentName, out FunctionsAgentOptions? agentTriggerOptions))
+            if (agentTriggerOptions.HttpTrigger.IsEnabled)
             {
-                if (agentTriggerOptions.HttpTrigger.IsEnabled)
-                {
-                    this._logger.LogRegisteringTriggerForAgent(agentName, "http");
-                    original.Add(FunctionMetadataFactory.CreateHttpTrigger(agentName, $"agents/{agentName}/run", BuiltInFunctions.RunAgentHttpFunctionEntryPoint));
-                }
+                this._logger.LogRegisteringTriggerForAgent(agentName, "http");
+                original.Add(FunctionMetadataFactory.CreateHttpTrigger(agentName, $"agents/{agentName}/run", BuiltInFunctions.RunAgentHttpFunctionEntryPoint));
+            }
 
-                if (agentTriggerOptions.McpToolTrigger.IsEnabled)
-                {
-                    AIAgent agent = kvp.Value(this._serviceProvider);
-                    this._logger.LogRegisteringTriggerForAgent(agentName, "mcpTool");
-                    original.Add(CreateMcpToolTrigger(agentName, agent.Description));
-                }
+            if (agentTriggerOptions.McpToolTrigger.IsEnabled)
+            {
+                AIAgent agent = kvp.Value(this._serviceProvider);
+                this._logger.LogRegisteringTriggerForAgent(agentName, "mcpTool");
+                original.Add(CreateMcpToolTrigger(agentName, agent.Description));
             }
         }
     }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentsOptionsExtensions.cs

@@ -134,4 +134,17 @@ public static class DurableAgentsOptionsExtensions
     {
         return new Dictionary<string, FunctionsAgentOptions>(s_agentOptions, StringComparer.OrdinalIgnoreCase);
     }
+
+    /// <summary>
+    /// Ensures every agent in <paramref name="agentNames"/> has an entry in the
+    /// options registry. Agents that already have explicit options are left untouched.
+    /// New entries receive the default configuration (HTTP trigger enabled, MCP tool disabled).
+    /// </summary>
+    internal static void EnsureDefaultOptionsForAll(IEnumerable<string> agentNames)
+    {
+        foreach (string name in agentNames)
+        {
+            s_agentOptions.TryAdd(name, new FunctionsAgentOptions { HttpTrigger = { IsEnabled = true } });
+        }
+    }
 }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionMetadataFactory.cs

@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
+using System.Text.Json.Nodes;
 using Microsoft.Agents.AI.DurableTask;
 using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
 
@@ -98,4 +99,65 @@ internal static class FunctionMetadataFactory
             ScriptFile = BuiltInFunctions.ScriptFile,
         };
     }
+
+    /// <summary>
+    /// Creates function metadata for an MCP tool trigger function that starts a workflow.
+    /// </summary>
+    /// <param name="workflowName">The name of the workflow to expose as an MCP tool.</param>
+    /// <param name="description">An optional description for the MCP tool. If null, a default description is generated.</param>
+    /// <returns>A <see cref="DefaultFunctionMetadata"/> configured for an MCP tool trigger.</returns>
+    internal static DefaultFunctionMetadata CreateWorkflowMcpToolTrigger(
+        string workflowName,
+        string? description)
+    {
+        var functionName = $"{BuiltInFunctions.McpToolPrefix}{workflowName}";
+        var toolDescription = description ?? $"Run the {workflowName} workflow";
+
+        var toolProperties = new JsonArray(new JsonObject
+        {
+            ["propertyName"] = "input",
+            ["propertyType"] = "string",
+            ["description"] = "The input to the workflow.",
+            ["isRequired"] = true,
+            ["isArray"] = false,
+        });
+
+        var triggerBinding = new JsonObject
+        {
+            ["name"] = "context",
+            ["type"] = "mcpToolTrigger",
+            ["direction"] = "In",
+            ["toolName"] = workflowName,
+            ["description"] = toolDescription,
+            ["toolProperties"] = toolProperties.ToJsonString(),
+        };
+
+        var inputBinding = new JsonObject
+        {
+            ["name"] = "input",
+            ["type"] = "mcpToolProperty",
+            ["direction"] = "In",
+            ["propertyName"] = "input",
+            ["description"] = "The input to the workflow",
+            ["isRequired"] = true,
+            ["dataType"] = "String",
+            ["propertyType"] = "string",
+        };
+
+        var clientBinding = new JsonObject
+        {
+            ["name"] = "client",
+            ["type"] = "durableClient",
+            ["direction"] = "In",
+        };
+
+        return new DefaultFunctionMetadata
+        {
+            Name = functionName,
+            Language = "dotnet-isolated",
+            RawBindings = [triggerBinding.ToJsonString(), inputBinding.ToJsonString(), clientBinding.ToJsonString()],
+            EntryPoint = BuiltInFunctions.RunWorkflowMcpToolFunctionEntryPoint,
+            ScriptFile = BuiltInFunctions.ScriptFile,
+        };
+    }
 }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsApplicationBuilderExtensions.cs

@@ -27,9 +27,16 @@ public static class FunctionsApplicationBuilderExtensions
     {
         ArgumentNullException.ThrowIfNull(configure);
 
+        // Create/get shared options BEFORE the DurableTask library call so it can find them.
+        FunctionsDurableOptions sharedOptions = GetOrCreateSharedOptions(builder.Services);
+
         // The main agent services registration is done in Microsoft.DurableTask.Agents.
         builder.Services.ConfigureDurableAgents(configure);
 
+        // Ensure all agents registered through this path have default FunctionsAgentOptions.
+        // This distinguishes them from agents auto-registered by workflows.
+        DurableAgentsOptionsExtensions.EnsureDefaultOptionsForAll(sharedOptions.Agents.GetAgentFactories().Keys);
+
         builder.Services.TryAddSingleton<IFunctionsAgentOptionsProvider>(_ =>
             new DefaultFunctionsAgentOptionsProvider(DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot()));
 
@@ -67,6 +74,13 @@ public static class FunctionsApplicationBuilderExtensions
 
         builder.Services.ConfigureDurableOptions(configure);
 
+        if (DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot().Count > 0)
+        {
+            builder.Services.TryAddSingleton<IFunctionsAgentOptionsProvider>(_ =>
+                new DefaultFunctionsAgentOptionsProvider(DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot()));
+            builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<IFunctionMetadataTransformer, DurableAgentFunctionMetadataTransformer>());
+        }
+
         if (sharedOptions.Workflows.Workflows.Count > 0)
         {
             builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<IFunctionMetadataTransformer, DurableWorkflowsFunctionMetadataTransformer>());
@@ -102,12 +116,14 @@ public static class FunctionsApplicationBuilderExtensions
 
         builder.UseWhen<BuiltInFunctionExecutionMiddleware>(static context =>
             string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) ||
+            string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentMcpToolFunctionEntryPoint, StringComparison.Ordinal) ||
             string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal) ||
             string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrchestrationHttpFunctionEntryPoint, StringComparison.Ordinal) ||
             string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrchestrationFunctionEntryPoint, StringComparison.Ordinal) ||
             string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) ||
             string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.GetWorkflowStatusHttpFunctionEntryPoint, StringComparison.Ordinal) ||
-            string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RespondToWorkflowHttpFunctionEntryPoint, StringComparison.Ordinal)
+            string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RespondToWorkflowHttpFunctionEntryPoint, StringComparison.Ordinal) ||
+            string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowMcpToolFunctionEntryPoint, StringComparison.Ordinal)
         );
         builder.Services.TryAddSingleton<BuiltInFunctionExecutor>();
     }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsDurableOptions.cs

@@ -10,6 +10,7 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
 internal sealed class FunctionsDurableOptions : DurableOptions
 {
     private readonly HashSet<string> _statusEndpointWorkflows = new(StringComparer.OrdinalIgnoreCase);
+    private readonly HashSet<string> _mcpToolTriggerWorkflows = new(StringComparer.OrdinalIgnoreCase);
 
     /// <summary>
     /// Enables the status HTTP endpoint for the specified workflow.
@@ -26,4 +27,20 @@ internal sealed class FunctionsDurableOptions : DurableOptions
     {
         return this._statusEndpointWorkflows.Contains(workflowName);
     }
+
+    /// <summary>
+    /// Enables the MCP tool trigger for the specified workflow.
+    /// </summary>
+    internal void EnableMcpToolTrigger(string workflowName)
+    {
+        this._mcpToolTriggerWorkflows.Add(workflowName);
+    }
+
+    /// <summary>
+    /// Returns whether the MCP tool trigger is enabled for the specified workflow.
+    /// </summary>
+    internal bool IsMcpToolTriggerEnabled(string workflowName)
+    {
+        return this._mcpToolTriggerWorkflows.Contains(workflowName);
+    }
 }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowOptionsExtensions.cs

@@ -27,4 +27,31 @@ public static class DurableWorkflowOptionsExtensions
             functionsOptions.EnableStatusEndpoint(workflow.Name!);
         }
     }
+
+    /// <summary>
+    /// Adds a workflow and configures whether to expose a status HTTP endpoint and/or an MCP tool trigger.
+    /// </summary>
+    /// <param name="options">The workflow options to add the workflow to.</param>
+    /// <param name="workflow">The workflow instance to add.</param>
+    /// <param name="exposeStatusEndpoint">If <see langword="true"/>, a GET endpoint is generated at <c>workflows/{name}/status/{runId}</c>.</param>
+    /// <param name="exposeMcpToolTrigger">If <see langword="true"/>, an MCP tool trigger is generated for the workflow.</param>
+    public static void AddWorkflow(this DurableWorkflowOptions options, Workflow workflow, bool exposeStatusEndpoint, bool exposeMcpToolTrigger)
+    {
+        ArgumentNullException.ThrowIfNull(options);
+
+        options.AddWorkflow(workflow);
+
+        if (options.ParentOptions is FunctionsDurableOptions functionsOptions)
+        {
+            if (exposeStatusEndpoint)
+            {
+                functionsOptions.EnableStatusEndpoint(workflow.Name!);
+            }
+
+            if (exposeMcpToolTrigger)
+            {
+                functionsOptions.EnableMcpToolTrigger(workflow.Name!);
+            }
+        }
+    }
 }

dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowsFunctionMetadataTransformer.cs

@@ -50,8 +50,11 @@ internal sealed class DurableWorkflowsFunctionMetadataTransformer : IFunctionMet
         int initialCount = original.Count;
         this._logger.LogTransformingFunctionMetadata(initialCount);
 
-        // Track registered function names to avoid duplicates when workflows share executors.
-        HashSet<string> registeredFunctions = [];
+        // Seed with existing function names to avoid duplicates across transformers
+        // (e.g., when DurableAgentFunctionMetadataTransformer already registered entity triggers).
+        HashSet<string> registeredFunctions = new(
+            original.Select(f => f.Name!),
+            StringComparer.OrdinalIgnoreCase);
 
         DurableWorkflowOptions workflowOptions = this._options.Workflows;
         foreach (var workflow in workflowOptions.Workflows)
@@ -113,6 +116,17 @@ internal sealed class DurableWorkflowsFunctionMetadataTransformer : IFunctionMet
                 }
             }
 
+            // Register an MCP tool trigger if opted in via AddWorkflow(exposeMcpToolTrigger: true).
+            if (this._options.IsMcpToolTriggerEnabled(workflow.Key))
+            {
+                string mcpToolFunctionName = $"{BuiltInFunctions.McpToolPrefix}{workflow.Key}";
+                if (registeredFunctions.Add(mcpToolFunctionName))
+                {
+                    this._logger.LogRegisteringWorkflowTrigger(workflow.Key, mcpToolFunctionName, "mcpTool");
+                    original.Add(FunctionMetadataFactory.CreateWorkflowMcpToolTrigger(workflow.Key, workflow.Value.Description));
+                }
+            }
+
             // Register activity or entity functions for each executor in the workflow.
             // ReflectExecutors() returns all executors across the graph; no need to manually traverse edges.
             foreach (KeyValuePair<string, ExecutorBinding> entry in workflow.Value.ReflectExecutors())

dotnet/src/Microsoft.Agents.AI.Workflows.Generators/Generation/SourceBuilder.cs

@@ -38,6 +38,7 @@ internal static class SourceBuilder
         sb.AppendLine("using System.Collections.Generic;");
         sb.AppendLine("using Microsoft.Agents.AI.Workflows;");
         sb.AppendLine();
+        sb.AppendLine("using RouteBuilder = Microsoft.Agents.AI.Workflows.RouteBuilder;");
 
         // Namespace
         if (!string.IsNullOrWhiteSpace(info.Namespace))

dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/FileSystemJsonCheckpointStore.cs

@@ -5,11 +5,14 @@ using System.Collections.Generic;
 using System.IO;
 using System.Text;
 using System.Text.Json;
+using System.Text.Json.Serialization.Metadata;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace Microsoft.Agents.AI.Workflows.Checkpointing;
 
+internal record CheckpointFileIndexEntry(CheckpointInfo CheckpointInfo, string FileName);
+
 /// <summary>
 /// Provides a file system-based implementation of a JSON checkpoint store that persists checkpoint data and index
 /// information to disk using JSON files.
@@ -28,6 +31,8 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
     internal DirectoryInfo Directory { get; }
     internal HashSet<CheckpointInfo> CheckpointIndex { get; }
 
+    private static JsonTypeInfo<CheckpointFileIndexEntry> EntryTypeInfo => WorkflowsJsonUtilities.JsonContext.Default.CheckpointFileIndexEntry;
+
     /// <summary>
     /// Initializes a new instance of the <see cref="FileSystemJsonCheckpointStore"/> class that uses the specified directory
     /// </summary>
@@ -64,9 +69,11 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
             using StreamReader reader = new(this._indexFile, encoding: Encoding.UTF8, detectEncodingFromByteOrderMarks: false, BufferSize, leaveOpen: true);
             while (reader.ReadLine() is string line)
             {
-                if (JsonSerializer.Deserialize(line, KeyTypeInfo) is { } info)
+                if (JsonSerializer.Deserialize(line, EntryTypeInfo) is { } entry)
                 {
-                    this.CheckpointIndex.Add(info);
+                    // We never actually use the file names from the index entries since they can be derived from the CheckpointInfo, but it is useful to
+                    // have the UrlEncoded file names in the index file for human readability
+                    this.CheckpointIndex.Add(entry.CheckpointInfo);
                 }
             }
         }
@@ -93,8 +100,14 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
         }
     }
 
-    private string GetFileNameForCheckpoint(string sessionId, CheckpointInfo key)
-        => Path.Combine(this.Directory.FullName, $"{sessionId}_{key.CheckpointId}.json");
+    internal string GetFileNameForCheckpoint(string sessionId, CheckpointInfo key)
+    {
+        string protoPath = $"{sessionId}_{key.CheckpointId}.json";
+
+        // Escape the protoPath to ensure it is a valid file name, especially if sessionId or CheckpointId contain path separators, etc.
+        return Uri.EscapeDataString(protoPath) // This takes care of most of the invalid path characters
+                  .Replace(".", "%2E");        // This takes care of escaping the root folder, since EscapeDataString does not escape dots
+    }
 
     private CheckpointInfo GetUnusedCheckpointInfo(string sessionId)
     {
@@ -116,13 +129,16 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
 
         CheckpointInfo key = this.GetUnusedCheckpointInfo(sessionId);
         string fileName = this.GetFileNameForCheckpoint(sessionId, key);
+        string filePath = Path.Combine(this.Directory.FullName, fileName);
+
         try
         {
-            using Stream checkpointStream = File.Open(fileName, FileMode.Create, FileAccess.Write, FileShare.None);
+            using Stream checkpointStream = File.Open(filePath, FileMode.Create, FileAccess.Write, FileShare.None);
             using Utf8JsonWriter jsonWriter = new(checkpointStream, new JsonWriterOptions() { Indented = false });
             value.WriteTo(jsonWriter);
 
-            JsonSerializer.Serialize(this._indexFile!, key, KeyTypeInfo);
+            CheckpointFileIndexEntry entry = new(key, fileName);
+            JsonSerializer.Serialize(this._indexFile!, entry, EntryTypeInfo);
             byte[] bytes = Encoding.UTF8.GetBytes(Environment.NewLine);
             await this._indexFile!.WriteAsync(bytes, 0, bytes.Length, CancellationToken.None).ConfigureAwait(false);
             await this._indexFile!.FlushAsync(CancellationToken.None).ConfigureAwait(false);
@@ -136,7 +152,7 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
             try
             {
                 // try to clean up after ourselves
-                File.Delete(fileName);
+                File.Delete(filePath);
             }
             catch { }
 
@@ -149,6 +165,7 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
     {
         this.CheckDisposed();
         string fileName = this.GetFileNameForCheckpoint(sessionId, key);
+        string filePath = Path.Combine(this.Directory.FullName, fileName);
 
         if (!this.CheckpointIndex.Contains(key) ||
             !File.Exists(fileName))
@@ -156,7 +173,7 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
             throw new KeyNotFoundException($"Checkpoint '{key.CheckpointId}' not found in store at '{this.Directory.FullName}'.");
         }
 
-        using FileStream checkpointFileStream = File.Open(fileName, FileMode.Open, FileAccess.Read, FileShare.Read);
+        using FileStream checkpointFileStream = File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read);
         using JsonDocument document = await JsonDocument.ParseAsync(checkpointFileStream).ConfigureAwait(false);
 
         return document.RootElement.Clone();

dotnet/src/Microsoft.Agents.AI.Workflows/ConfigurationExtensions.cs

@@ -3,9 +3,9 @@
 namespace Microsoft.Agents.AI.Workflows;
 
 /// <summary>
-/// Provides extensions methods for creating <see cref="Configured{TSubject}"/> objects
+/// Provides extension methods for creating <see cref="Configured{TSubject}"/> objects
 /// </summary>
-public static class ConfigurationExtensions
+internal static class ConfigurationExtensions
 {
     /// <summary>
     /// Creates a new configuration that treats the subject as its base type, allowing configuration to be applied at

dotnet/src/Microsoft.Agents.AI.Workflows/Configured.cs

@@ -8,7 +8,7 @@ namespace Microsoft.Agents.AI.Workflows;
 /// <summary>
 /// Provides methods for creating <see cref="Configured{TSubject}"/> instances.
 /// </summary>
-public static class Configured
+internal static class Configured
 {
     /// <summary>
     /// Creates a <see cref="Configured{TSubject}"/> instance from an existing subject instance.
@@ -50,10 +50,10 @@ public static class Configured
 /// A representation of a preconfigured, lazy-instantiatable instance of <typeparamref name="TSubject"/>.
 /// </summary>
 /// <typeparam name="TSubject">The type of the preconfigured subject.</typeparam>
-/// <param name="factoryAsync">A factory to intantiate the subject when desired.</param>
+/// <param name="factoryAsync">A factory to instantiate the subject when desired.</param>
 /// <param name="id">The unique identifier for the configured subject.</param>
 /// <param name="raw"></param>
-public class Configured<TSubject>(Func<Config, string, ValueTask<TSubject>> factoryAsync, string id, object? raw = null)
+internal class Configured<TSubject>(Func<ExecutorConfig, string, ValueTask<TSubject>> factoryAsync, string id, object? raw = null)
 {
     /// <summary>
     /// Gets the raw representation of the configured object, if any.
@@ -66,14 +66,14 @@ public class Configured<TSubject>(Func<Config, string, ValueTask<TSubject>> fact
     public string Id => id;
 
     /// <summary>
-    /// Gets the factory function to create an instance of <typeparamref name="TSubject"/> given a <see cref="Config"/>.
+    /// Gets the factory function to create an instance of <typeparamref name="TSubject"/> given a <see cref="ExecutorConfig"/>.
     /// </summary>
-    public Func<Config, string, ValueTask<TSubject>> FactoryAsync => factoryAsync;
+    public Func<ExecutorConfig, string, ValueTask<TSubject>> FactoryAsync => factoryAsync;
 
     /// <summary>
     /// The configuration for this configured instance.
     /// </summary>
-    public Config Configuration => new(this.Id);
+    public ExecutorConfig Configuration => new(this.Id);
 
     /// <summary>
     /// Gets a "partially" applied factory function that only requires no parameters to create an instance of
@@ -87,11 +87,11 @@ public class Configured<TSubject>(Func<Config, string, ValueTask<TSubject>> fact
 /// </summary>
 /// <typeparam name="TSubject">The type of the preconfigured subject.</typeparam>
 /// <typeparam name="TOptions">The type of configuration options for the preconfigured subject.</typeparam>
-/// <param name="factoryAsync">A factory to intantiate the subject when desired.</param>
+/// <param name="factoryAsync">A factory to instantiate the subject when desired.</param>
 /// <param name="id">The unique identifier for the configured subject.</param>
 /// <param name="options">Additional configuration options for the subject.</param>
 /// <param name="raw"></param>
-public class Configured<TSubject, TOptions>(Func<Config<TOptions>, string, ValueTask<TSubject>> factoryAsync, string id, TOptions? options = default, object? raw = null)
+internal class Configured<TSubject, TOptions>(Func<ExecutorConfig<TOptions>, string, ValueTask<TSubject>> factoryAsync, string id, TOptions? options = default, object? raw = null)
 {
     /// <summary>
     /// The raw representation of the configured object, if any.
@@ -109,14 +109,14 @@ public class Configured<TSubject, TOptions>(Func<Config<TOptions>, string, Value
     public TOptions? Options => options;
 
     /// <summary>
-    /// Gets the factory function to create an instance of <typeparamref name="TSubject"/> given a <see cref="Config{TOptions}"/>.
+    /// Gets the factory function to create an instance of <typeparamref name="TSubject"/> given a <see cref="ExecutorConfig{TOptions}"/>.
     /// </summary>
-    public Func<Config<TOptions>, string, ValueTask<TSubject>> FactoryAsync => factoryAsync;
+    public Func<ExecutorConfig<TOptions>, string, ValueTask<TSubject>> FactoryAsync => factoryAsync;
 
     /// <summary>
     /// The configuration for this configured instance.
     /// </summary>
-    public Config<TOptions> Configuration => new(this.Id, this.Options);
+    public ExecutorConfig<TOptions> Configuration => new(this.Id, this.Options);
 
     /// <summary>
     /// Gets a "partially" applied factory function that only requires no parameters to create an instance of
@@ -124,11 +124,11 @@ public class Configured<TSubject, TOptions>(Func<Config<TOptions>, string, Value
     /// </summary>
     internal Func<string, ValueTask<TSubject>> BoundFactoryAsync => (sessionId) => this.CreateValidatingMemoizedFactory()(this.Configuration, sessionId);
 
-    private Func<Config, string, ValueTask<TSubject>> CreateValidatingMemoizedFactory()
+    private Func<ExecutorConfig, string, ValueTask<TSubject>> CreateValidatingMemoizedFactory()
     {
         return FactoryAsync;
 
-        async ValueTask<TSubject> FactoryAsync(Config configuration, string sessionId)
+        async ValueTask<TSubject> FactoryAsync(ExecutorConfig configuration, string sessionId)
         {
             if (this.Id != configuration.Id)
             {

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/AsyncRunHandle.cs

@@ -53,6 +53,9 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable
     public ValueTask<RunStatus> GetStatusAsync(CancellationToken cancellationToken = default)
         => this._eventStream.GetStatusAsync(cancellationToken);
 
+    internal bool TryGetResponsePortExecutorId(string portId, out string? executorId)
+        => this._stepRunner.TryGetResponsePortExecutorId(portId, out executorId);
+
     public async IAsyncEnumerable<WorkflowEvent> TakeEventStreamAsync(bool blockOnPendingRequest, [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         //Debug.Assert(breakOnHalt);

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/EdgeMap.cs

@@ -3,6 +3,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -95,6 +96,18 @@ internal sealed class EdgeMap
         return portRunner.ChaseEdgeAsync(new MessageEnvelope(response, ExecutorIdentity.None), this._stepTracer, cancellationToken);
     }
 
+    internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId)
+    {
+        if (this._portEdgeRunners.TryGetValue(portId, out ResponseEdgeRunner? portRunner))
+        {
+            executorId = portRunner.ExecutorId;
+            return true;
+        }
+
+        executorId = null;
+        return false;
+    }
+
     internal async ValueTask<Dictionary<EdgeId, PortableValue>> ExportStateAsync()
     {
         Dictionary<EdgeId, PortableValue> exportedStates = [];

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ISuperStepRunner.cs

@@ -19,6 +19,7 @@ internal interface ISuperStepRunner
     bool HasUnprocessedMessages { get; }
 
     ValueTask EnqueueResponseAsync(ExternalResponse response, CancellationToken cancellationToken = default);
+    bool TryGetResponsePortExecutorId(string portId, out string? executorId);
 
     ValueTask<bool> IsValidInputTypeAsync<T>(CancellationToken cancellationToken = default);
     ValueTask<bool> EnqueueMessageAsync<T>(T message, CancellationToken cancellationToken = default);

dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorBindingExtensions.cs

@@ -113,7 +113,7 @@ public static class ExecutorBindingExtensions
     /// <param name="id">An id for the executor to be instantiated.</param>
     /// <param name="options">An optional parameter specifying the options.</param>
     /// <returns>An <see cref="ExecutorBinding"/> instance that resolves to the result of the factory call when messages get sent to it.</returns>
-    public static ExecutorBinding BindExecutor<TExecutor, TOptions>(this Func<Config<TOptions>, string, ValueTask<TExecutor>> factoryAsync, string id, TOptions? options = null)
+    public static ExecutorBinding BindExecutor<TExecutor, TOptions>(this Func<ExecutorConfig<TOptions>, string, ValueTask<TExecutor>> factoryAsync, string id, TOptions? options = null)
         where TExecutor : Executor
         where TOptions : ExecutorOptions
     {
@@ -139,7 +139,7 @@ public static class ExecutorBindingExtensions
     /// <returns>An <see cref="ExecutorBinding"/> instance that resolves to the result of the factory call when messages get sent to it.</returns>
     [Obsolete("Use BindExecutor() instead")]
     [EditorBrowsable(EditorBrowsableState.Never)]
-    public static ExecutorBinding ConfigureFactory<TExecutor, TOptions>(this Func<Config<TOptions>, string, ValueTask<TExecutor>> factoryAsync, string id, TOptions? options = null)
+    public static ExecutorBinding ConfigureFactory<TExecutor, TOptions>(this Func<ExecutorConfig<TOptions>, string, ValueTask<TExecutor>> factoryAsync, string id, TOptions? options = null)
         where TExecutor : Executor
         where TOptions : ExecutorOptions
         => factoryAsync.BindExecutor(id, options);

dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorConfig.cs

@@ -6,7 +6,7 @@ namespace Microsoft.Agents.AI.Workflows;
 /// Represents a configuration for an object with a string identifier. For example, <see cref="IIdentified"/> object.
 /// </summary>
 /// <param name="id">A unique identifier for the configurable object.</param>
-public class Config(string id)
+public class ExecutorConfig(string id)
 {
     /// <summary>
     /// Gets a unique identifier for the configurable object.
@@ -23,7 +23,7 @@ public class Config(string id)
 /// <typeparam name="TOptions">The type of options for the configurable object.</typeparam>
 /// <param name="id">A unique identifier for the configurable object.</param>
 /// <param name="options">The options for the configurable object.</param>
-public class Config<TOptions>(string id, TOptions? options = default) : Config(id)
+public class ExecutorConfig<TOptions>(string id, TOptions? options = default) : ExecutorConfig(id)
 {
     /// <summary>
     /// Gets the options for the configured object.

dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs

@@ -160,6 +160,8 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
 
     bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests;
     bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions;
+    bool ISuperStepRunner.TryGetResponsePortExecutorId(string portId, out string? executorId)
+        => this.RunContext.TryGetResponsePortExecutorId(portId, out executorId);
 
     public bool IsCheckpointingEnabled => this.RunContext.IsCheckpointingEnabled;
 

dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs

@@ -296,6 +296,9 @@ internal sealed class InProcessRunnerContext : IRunnerContext
         return this._externalRequests.TryRemove(requestId, out _);
     }
 
+    internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId)
+        => this._edgeMap.TryGetResponsePortExecutorId(portId, out executorId);
+
     private IEventSink OutgoingEvents { get; }
 
     internal StateManager StateManager { get; } = new();

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs

@@ -68,10 +68,17 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
             throw new InvalidOperationException($"No pending ToolApprovalRequest found with id '{response.RequestId}'.");
         }
 
-        List<ChatMessage> implicitTurnMessages = [new ChatMessage(ChatRole.User, [response])];
+        // Merge the external response with any already-buffered regular messages so mixed-content
+        // resumes can be processed in one invocation.
+        return this.ProcessTurnMessagesAsync(async (pendingMessages, ctx, ct) =>
+        {
+            pendingMessages.Add(new ChatMessage(ChatRole.User, [response]));
 
-        // ContinueTurnAsync owns failing to emit a TurnToken if this response does not clear up all remaining outstanding requests.
-        return this.ContinueTurnAsync(implicitTurnMessages, context, this._currentTurnEmitEvents ?? false, cancellationToken);
+            await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false);
+
+            // Clear the buffered turn messages because they were consumed by ContinueTurnAsync.
+            return null;
+        }, context, cancellationToken);
     }
 
     private ValueTask HandleFunctionResultAsync(
@@ -84,8 +91,17 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
             throw new InvalidOperationException($"No pending FunctionCall found with id '{result.CallId}'.");
         }
 
-        List<ChatMessage> implicitTurnMessages = [new ChatMessage(ChatRole.Tool, [result])];
-        return this.ContinueTurnAsync(implicitTurnMessages, context, this._currentTurnEmitEvents ?? false, cancellationToken);
+        // Merge the external response with any already-buffered regular messages so mixed-content
+        // resumes can be processed in one invocation.
+        return this.ProcessTurnMessagesAsync(async (pendingMessages, ctx, ct) =>
+        {
+            pendingMessages.Add(new ChatMessage(ChatRole.Tool, [result]));
+
+            await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false);
+
+            // Clear the buffered turn messages because they were consumed by ContinueTurnAsync.
+            return null;
+        }, context, cancellationToken);
     }
 
     public bool ShouldEmitStreamingEvents(bool? emitEvents)
@@ -198,7 +214,7 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
             ExtractUnservicedRequests(response.Messages.SelectMany(message => message.Contents));
         }
 
-        if (this._options.EmitAgentResponseEvents == true)
+        if (this._options.EmitAgentResponseEvents)
         {
             await context.YieldOutputAsync(response, cancellationToken).ConfigureAwait(false);
         }

dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIContentExternalHandler.cs

@@ -16,10 +16,12 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
     where TResponseContent : AIContent
 {
     private readonly PortBinding? _portBinding;
+    private readonly string _portId;
     private ConcurrentDictionary<string, TRequestContent> _pendingRequests = new();
 
     public AIContentExternalHandler(ref ProtocolBuilder protocolBuilder, string portId, bool intercepted, Func<TResponseContent, IWorkflowContext, CancellationToken, ValueTask> handler)
     {
+        this._portId = portId;
         PortBinding? portBinding = null;
         protocolBuilder = protocolBuilder.ConfigureRoutes(routeBuilder => ConfigureRoutes(routeBuilder, out portBinding));
         this._portBinding = portBinding;
@@ -58,12 +60,14 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
     {
         if (!this._pendingRequests.TryAdd(id, requestContent))
         {
-            throw new InvalidOperationException($"A pending request with ID '{id}' already exists.");
+            // Request is already pending; treat as an idempotent re-emission.
+            // Do not repost to the sink because request IDs must remain unique while pending.
+            return default;
         }
 
         return this.IsIntercepted
              ? context.SendMessageAsync(requestContent, cancellationToken: cancellationToken)
-             : this._portBinding.PostRequestAsync(requestContent, id, cancellationToken);
+             : this._portBinding.PostRequestAsync(requestContent, this.CreateExternalRequestId(id), cancellationToken);
     }
 
     public bool MarkRequestAsHandled(string id)
@@ -74,6 +78,8 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
     [MemberNotNullWhen(false, nameof(_portBinding))]
     private bool IsIntercepted => this._portBinding == null;
 
+    private string CreateExternalRequestId(string requestId) => $"{this._portId.Length}:{this._portId}:{requestId}";
+
     private static string MakeKey(string id) => $"{id}_PendingRequests";
 
     public async ValueTask OnCheckpointingAsync(string id, IWorkflowContext context, CancellationToken cancellationToken = default)

dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs

@@ -60,6 +60,9 @@ public sealed class StreamingRun : CheckpointableRunBase, IAsyncDisposable
     internal ValueTask<bool> TrySendMessageUntypedAsync(object message, Type? declaredType = null)
         => this._runHandle.EnqueueMessageUntypedAsync(message, declaredType);
 
+    internal bool TryGetResponsePortExecutorId(string portId, out string? executorId)
+        => this._runHandle.TryGetResponsePortExecutorId(portId, out executorId);
+
     /// <summary>
     /// Asynchronously streams workflow events as they occur during workflow execution.
     /// </summary>

dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs

@@ -25,6 +25,25 @@ internal sealed class WorkflowSession : AgentSession
 
     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;
@@ -90,6 +109,7 @@ internal sealed class WorkflowSession : AgentSession
 
         this.LastCheckpoint = sessionState.LastCheckpoint;
         this.StateBag = sessionState.StateBag;
+        this._pendingRequests = sessionState.PendingRequests ?? [];
     }
 
     public CheckpointInfo? LastCheckpoint { get; set; }
@@ -101,7 +121,8 @@ internal sealed class WorkflowSession : AgentSession
             this.SessionId,
             this.LastCheckpoint,
             this._inMemoryCheckpointManager,
-            this.StateBag);
+            this.StateBag,
+            this._pendingRequests);
 
         return marshaller.Marshal(info);
     }
@@ -141,7 +162,7 @@ internal sealed class WorkflowSession : AgentSession
         return update;
     }
 
-    private async ValueTask<StreamingRun> CreateOrResumeRunAsync(List<ChatMessage> messages, CancellationToken cancellationToken = default)
+    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.
@@ -154,18 +175,155 @@ internal sealed class WorkflowSession : AgentSession
                                                cancellationToken)
                             .ConfigureAwait(false);
 
-            await run.TrySendMessageAsync(messages).ConfigureAwait(false);
-            return run;
+            // 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);
         }
 
-        return await this._executionEnvironment
+        StreamingRun newRun = await this._executionEnvironment
                             .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);
+                    }
+
+                    AIContent 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>
+    /// Creates the workflow-facing request content surfaced in response updates.
+    /// </summary>
+    private static AIContent CreateRequestContentForDelivery(ExternalRequest request) => 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 AIContent NormalizeResponseContentForDelivery(AIContent content, ExternalRequest request) => content switch
+    {
+        FunctionResultContent functionResultContent when request.TryGetDataAs(out FunctionCallContent? functionCallContent)
+            => CloneFunctionResultContent(functionResultContent, functionCallContent.CallId),
+        ToolApprovalResponseContent toolApprovalResponseContent when request.TryGetDataAs(out ToolApprovalRequestContent? toolApprovalRequestContent)
+            => CloneToolApprovalResponseContent(toolApprovalResponseContent, toolApprovalRequestContent.RequestId),
+        _ => 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)
@@ -175,12 +333,25 @@ internal sealed class WorkflowSession : AgentSession
             this.LastResponseId = Guid.NewGuid().ToString("N");
             List<ChatMessage> messages = this.ChatHistoryProvider.GetFromBookmark(this).ToList();
 
-#pragma warning disable CA2007 // Analyzer misfiring and not seeing .ConfigureAwait(false) below.
-            await using StreamingRun run =
+            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
 
-            await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
+            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))
@@ -192,8 +363,13 @@ internal sealed class WorkflowSession : AgentSession
                         break;
 
                     case RequestInfoEvent requestInfo:
-                        FunctionCallContent fcContent = requestInfo.Request.ToFunctionCall();
-                        AgentResponseUpdate update = this.CreateUpdate(this.LastResponseId, evt, fcContent);
+                        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;
 
@@ -267,15 +443,116 @@ internal sealed class WorkflowSession : AgentSession
     /// <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)
+        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;
     }
 }

dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowsJsonUtilities.cs

@@ -71,6 +71,7 @@ internal static partial class WorkflowsJsonUtilities
     [JsonSerializable(typeof(PortableValue))]
     [JsonSerializable(typeof(PortableMessageEnvelope))]
     [JsonSerializable(typeof(InMemoryCheckpointManager))]
+    [JsonSerializable(typeof(CheckpointFileIndexEntry))]
 
     // Runtime State Types
     [JsonSerializable(typeof(ScopeKey))]

dotnet/src/Microsoft.Agents.AI/AIContextProviderDecorators/AIContextProviderChatClient.cs

@@ -161,6 +161,8 @@ internal sealed class AIContextProviderChatClient : DelegatingChatClient
         }
 
         // Materialize the accumulated context back into messages and options.
+        // Clone options to avoid mutating the caller's instance across calls.
+        options = options?.Clone();
         var enrichedMessages = aiContext.Messages ?? [];
 
         var tools = aiContext.Tools as IList<AITool> ?? aiContext.Tools?.ToList();

dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgent.cs

@@ -138,6 +138,9 @@ public sealed partial class ChatClientAgent : AIAgent
         this._aiContextProviderStateKeys = ValidateAndCollectStateKeys(this._agentOptions?.AIContextProviders, this.ChatHistoryProvider);
 
         this._logger = (loggerFactory ?? chatClient.GetService<ILoggerFactory>() ?? NullLoggerFactory.Instance).CreateLogger<ChatClientAgent>();
+
+        // Warn if using a custom chat client stack with end-of-run persistence but no ChatHistoryPersistingChatClient.
+        this.WarnOnMissingPersistingClient();
     }
 
     /// <summary>
@@ -211,12 +214,14 @@ public sealed partial class ChatClientAgent : AIAgent
          ChatClientAgentContinuationToken? _) =
             await this.PrepareSessionAndMessagesAsync(session, inputMessages, options, cancellationToken).ConfigureAwait(false);
 
-        var chatClient = this.ChatClient;
+        // Update the run context with the resolved session so any downstream classes
+        // always have a valid session, even when the caller passed null.
+        EnsureRunContextHasSession(safeSession);
 
+        var chatClient = this.ChatClient;
         chatClient = ApplyRunOptionsTransformations(options, chatClient);
 
         var loggingAgentName = this.GetLoggingAgentName();
-
         this._logger.LogAgentChatClientInvokingAgent(nameof(RunAsync), this.Id, loggingAgentName, this._chatClientType);
 
         // Call the IChatClient and notify the AIContextProvider of any failures.
@@ -227,8 +232,7 @@ public sealed partial class ChatClientAgent : AIAgent
         }
         catch (Exception ex)
         {
-            await this.NotifyChatHistoryProviderOfFailureAsync(safeSession, ex, inputMessagesForChatClient, chatOptions, cancellationToken).ConfigureAwait(false);
-            await this.NotifyAIContextProviderOfFailureAsync(safeSession, ex, inputMessagesForChatClient, cancellationToken).ConfigureAwait(false);
+            await this.NotifyProvidersOfFailureAtEndOfRunAsync(safeSession, ex, inputMessagesForChatClient, chatOptions, cancellationToken).ConfigureAwait(false);
             throw;
         }
 
@@ -236,7 +240,8 @@ public sealed partial class ChatClientAgent : AIAgent
 
         // We can derive the type of supported session from whether we have a conversation id,
         // so let's update it and set the conversation id for the service session case.
-        this.UpdateSessionConversationId(safeSession, chatResponse.ConversationId, cancellationToken);
+        var forceEndOfRunPersistence = chatOptions?.ContinuationToken is not null || chatOptions?.AllowBackgroundResponses is true;
+        this.UpdateSessionConversationIdAtEndOfRun(safeSession, chatResponse.ConversationId, cancellationToken, forceUpdate: forceEndOfRunPersistence);
 
         // Ensure that the author name is set for each message in the response.
         foreach (ChatMessage chatResponseMessage in chatResponse.Messages)
@@ -244,11 +249,10 @@ public sealed partial class ChatClientAgent : AIAgent
             chatResponseMessage.AuthorName ??= this.Name;
         }
 
-        // Only notify the session of new messages if the chatResponse was successful to avoid inconsistent message state in the session.
-        await this.NotifyChatHistoryProviderOfNewMessagesAsync(safeSession, inputMessagesForChatClient, chatResponse.Messages, chatOptions, cancellationToken).ConfigureAwait(false);
-
-        // Notify the AIContextProvider of all new messages.
-        await this.NotifyAIContextProviderOfSuccessAsync(safeSession, inputMessagesForChatClient, chatResponse.Messages, cancellationToken).ConfigureAwait(false);
+        // Notify providers of all new messages unless persistence is handled per-service-call by the decorator.
+        // When background responses are allowed, force notification since per-service-call persistence
+        // is unreliable (the caller may stop consuming the stream before the decorator can persist).
+        await this.NotifyProvidersOfNewMessagesAtEndOfRunAsync(safeSession, inputMessagesForChatClient, chatResponse.Messages, chatOptions, cancellationToken, forceNotify: forceEndOfRunPersistence).ConfigureAwait(false);
 
         return new AgentResponse(chatResponse)
         {
@@ -296,6 +300,10 @@ public sealed partial class ChatClientAgent : AIAgent
          ChatClientAgentContinuationToken? continuationToken) =
             await this.PrepareSessionAndMessagesAsync(session, inputMessages, options, cancellationToken).ConfigureAwait(false);
 
+        // Update the run context with the resolved session so any downstream classes
+        // always have a valid session, even when the caller passed null.
+        EnsureRunContextHasSession(safeSession);
+
         var chatClient = this.ChatClient;
 
         chatClient = ApplyRunOptionsTransformations(options, chatClient);
@@ -315,8 +323,7 @@ public sealed partial class ChatClientAgent : AIAgent
         }
         catch (Exception ex)
         {
-            await this.NotifyChatHistoryProviderOfFailureAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), chatOptions, cancellationToken).ConfigureAwait(false);
-            await this.NotifyAIContextProviderOfFailureAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), cancellationToken).ConfigureAwait(false);
+            await this.NotifyProvidersOfFailureAtEndOfRunAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), chatOptions, cancellationToken).ConfigureAwait(false);
             throw;
         }
 
@@ -330,8 +337,7 @@ public sealed partial class ChatClientAgent : AIAgent
         }
         catch (Exception ex)
         {
-            await this.NotifyChatHistoryProviderOfFailureAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), chatOptions, cancellationToken).ConfigureAwait(false);
-            await this.NotifyAIContextProviderOfFailureAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), cancellationToken).ConfigureAwait(false);
+            await this.NotifyProvidersOfFailureAtEndOfRunAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), chatOptions, cancellationToken).ConfigureAwait(false);
             throw;
         }
 
@@ -353,27 +359,31 @@ public sealed partial class ChatClientAgent : AIAgent
 
             try
             {
+                // Re-ensure the run context has the resolved session before each MoveNextAsync.
+                // The base class RunStreamingAsync restores the original context (potentially with
+                // null session) after each yield, so we must re-establish it for the decorator.
+                EnsureRunContextHasSession(safeSession);
                 hasUpdates = await responseUpdatesEnumerator.MoveNextAsync().ConfigureAwait(false);
             }
             catch (Exception ex)
             {
-                await this.NotifyChatHistoryProviderOfFailureAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), chatOptions, cancellationToken).ConfigureAwait(false);
-                await this.NotifyAIContextProviderOfFailureAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), cancellationToken).ConfigureAwait(false);
+                await this.NotifyProvidersOfFailureAtEndOfRunAsync(safeSession, ex, GetInputMessages(inputMessagesForChatClient, continuationToken), chatOptions, cancellationToken).ConfigureAwait(false);
                 throw;
             }
         }
 
         var chatResponse = responseUpdates.ToChatResponse();
 
+        var forceEndOfRunPersistence = continuationToken is not null || chatOptions?.AllowBackgroundResponses is true;
+
         // We can derive the type of supported session from whether we have a conversation id,
         // so let's update it and set the conversation id for the service session case.
-        this.UpdateSessionConversationId(safeSession, chatResponse.ConversationId, cancellationToken);
+        this.UpdateSessionConversationIdAtEndOfRun(safeSession, chatResponse.ConversationId, cancellationToken, forceUpdate: forceEndOfRunPersistence);
 
-        // To avoid inconsistent state we only notify the session of the input messages if no error occurs after the initial request.
-        await this.NotifyChatHistoryProviderOfNewMessagesAsync(safeSession, GetInputMessages(inputMessagesForChatClient, continuationToken), chatResponse.Messages, chatOptions, cancellationToken).ConfigureAwait(false);
-
-        // Notify the AIContextProvider of all new messages.
-        await this.NotifyAIContextProviderOfSuccessAsync(safeSession, GetInputMessages(inputMessagesForChatClient, continuationToken), chatResponse.Messages, cancellationToken).ConfigureAwait(false);
+        // Notify providers of all new messages unless persistence is handled per-service-call by the decorator.
+        // When resuming from a continuation token or using background responses, force notification
+        // to send the combined data (per-service-call persistence is unreliable for these scenarios).
+        await this.NotifyProvidersOfNewMessagesAtEndOfRunAsync(safeSession, GetInputMessages(inputMessagesForChatClient, continuationToken), chatResponse.Messages, chatOptions, cancellationToken, forceNotify: forceEndOfRunPersistence).ConfigureAwait(false);
     }
 
     /// <inheritdoc/>
@@ -441,17 +451,29 @@ public sealed partial class ChatClientAgent : AIAgent
     #region Private
 
     /// <summary>
-    /// Notify the <see cref="AIContextProvider"/> when an agent run succeeded, if there is an <see cref="AIContextProvider"/>.
+    /// Notifies the <see cref="ChatHistoryProvider"/> and all <see cref="AIContextProviders"/> of successfully completed messages.
     /// </summary>
-    private async Task NotifyAIContextProviderOfSuccessAsync(
+    /// <remarks>
+    /// This method is also called by <see cref="ChatHistoryPersistingChatClient"/> to persist messages per-service-call.
+    /// </remarks>
+    internal async Task NotifyProvidersOfNewMessagesAsync(
         ChatClientAgentSession session,
-        IEnumerable<ChatMessage> inputMessages,
+        IEnumerable<ChatMessage> requestMessages,
         IEnumerable<ChatMessage> responseMessages,
+        ChatOptions? chatOptions,
         CancellationToken cancellationToken)
     {
+        ChatHistoryProvider? chatHistoryProvider = this.ResolveChatHistoryProvider(chatOptions, session);
+
+        if (chatHistoryProvider is not null)
+        {
+            var invokedContext = new ChatHistoryProvider.InvokedContext(this, session, requestMessages, responseMessages);
+            await chatHistoryProvider.InvokedAsync(invokedContext, cancellationToken).ConfigureAwait(false);
+        }
+
         if (this.AIContextProviders is { Count: > 0 } contextProviders)
         {
-            AIContextProvider.InvokedContext invokedContext = new(this, session, inputMessages, responseMessages);
+            AIContextProvider.InvokedContext invokedContext = new(this, session, requestMessages, responseMessages);
 
             foreach (var contextProvider in contextProviders)
             {
@@ -461,17 +483,29 @@ public sealed partial class ChatClientAgent : AIAgent
     }
 
     /// <summary>
-    /// Notify the <see cref="AIContextProvider"/> of any failure during an agent run, if there is an <see cref="AIContextProvider"/>.
+    /// Notifies the <see cref="ChatHistoryProvider"/> and all <see cref="AIContextProviders"/> of a failure during a service call.
     /// </summary>
-    private async Task NotifyAIContextProviderOfFailureAsync(
+    /// <remarks>
+    /// This method is also called by <see cref="ChatHistoryPersistingChatClient"/> to report failures per-service-call.
+    /// </remarks>
+    internal async Task NotifyProvidersOfFailureAsync(
         ChatClientAgentSession session,
         Exception ex,
-        IEnumerable<ChatMessage> inputMessages,
+        IEnumerable<ChatMessage> requestMessages,
+        ChatOptions? chatOptions,
         CancellationToken cancellationToken)
     {
+        ChatHistoryProvider? chatHistoryProvider = this.ResolveChatHistoryProvider(chatOptions, session);
+
+        if (chatHistoryProvider is not null)
+        {
+            var invokedContext = new ChatHistoryProvider.InvokedContext(this, session, requestMessages, ex);
+            await chatHistoryProvider.InvokedAsync(invokedContext, cancellationToken).ConfigureAwait(false);
+        }
+
         if (this.AIContextProviders is { Count: > 0 } contextProviders)
         {
-            AIContextProvider.InvokedContext invokedContext = new(this, session, inputMessages, ex);
+            AIContextProvider.InvokedContext invokedContext = new(this, session, requestMessages, ex);
 
             foreach (var contextProvider in contextProviders)
             {
@@ -667,6 +701,12 @@ public sealed partial class ChatClientAgent : AIAgent
             throw new InvalidOperationException("A session must be provided when continuing a background response with a continuation token.");
         }
 
+        if ((continuationToken is not null || chatOptions?.AllowBackgroundResponses is true) && this.PersistsChatHistoryPerServiceCall && this._logger.IsEnabled(LogLevel.Warning))
+        {
+            var warningAgentName = this.GetLoggingAgentName();
+            this._logger.LogAgentChatClientBackgroundResponseFallback(this.Id, warningAgentName);
+        }
+
         session ??= await this.CreateSessionAsync(cancellationToken).ConfigureAwait(false);
         if (session is not ChatClientAgentSession typedSession)
         {
@@ -754,7 +794,7 @@ public sealed partial class ChatClientAgent : AIAgent
         return (typedSession, chatOptions, messagesList, continuationToken);
     }
 
-    private void UpdateSessionConversationId(ChatClientAgentSession session, string? responseConversationId, CancellationToken cancellationToken)
+    internal void UpdateSessionConversationId(ChatClientAgentSession session, string? responseConversationId, CancellationToken cancellationToken)
     {
         if (string.IsNullOrWhiteSpace(responseConversationId) && !string.IsNullOrWhiteSpace(session.ConversationId))
         {
@@ -798,45 +838,162 @@ public sealed partial class ChatClientAgent : AIAgent
         }
     }
 
-    private Task NotifyChatHistoryProviderOfFailureAsync(
+    /// <summary>
+    /// Updates the session conversation ID at the end of an agent run.
+    /// </summary>
+    /// <remarks>
+    /// When a <see cref="ChatHistoryPersistingChatClient"/> in persist mode handles per-service-call
+    /// conversation ID updates, this end-of-run update is skipped. When the decorator is in mark-only
+    /// mode or absent, the update is performed here. When <paramref name="forceUpdate"/> is <see langword="true"/>
+    /// (continuation token scenarios), the update is always performed.
+    /// </remarks>
+    private void UpdateSessionConversationIdAtEndOfRun(ChatClientAgentSession session, string? responseConversationId, CancellationToken cancellationToken, bool forceUpdate = false)
+    {
+        if (!forceUpdate && this.PersistsChatHistoryPerServiceCall)
+        {
+            return;
+        }
+
+        this.UpdateSessionConversationId(session, responseConversationId, cancellationToken);
+    }
+
+    /// <summary>
+    /// Notifies providers of successfully completed messages at the end of an agent run.
+    /// </summary>
+    /// <remarks>
+    /// When a <see cref="ChatHistoryPersistingChatClient"/> in persist mode handles per-service-call
+    /// notification, this end-of-run notification is skipped. When the decorator is in mark-only mode,
+    /// only the marked messages are persisted. When no decorator is present (custom stack with
+    /// <see cref="ChatClientAgentOptions.PersistChatHistoryAtEndOfRun"/>), all messages are persisted.
+    /// When <paramref name="forceNotify"/> is <see langword="true"/> (continuation token or
+    /// background response scenarios), notification is always performed with all messages because
+    /// per-service-call persistence is unreliable in these scenarios.
+    /// </remarks>
+    private Task NotifyProvidersOfNewMessagesAtEndOfRunAsync(
+        ChatClientAgentSession session,
+        IEnumerable<ChatMessage> requestMessages,
+        IEnumerable<ChatMessage> responseMessages,
+        ChatOptions? chatOptions,
+        CancellationToken cancellationToken,
+        bool forceNotify = false)
+    {
+        if (!forceNotify && this.PersistsChatHistoryPerServiceCall)
+        {
+            return Task.CompletedTask;
+        }
+
+        if (!forceNotify && this.HasMarkOnlyChatHistoryPersistingClient)
+        {
+            // In mark-only mode, persist only messages that were marked by the decorator.
+            var markedRequestMessages = GetMarkedMessages(requestMessages);
+            var markedResponseMessages = GetMarkedMessages(responseMessages);
+            return this.NotifyProvidersOfNewMessagesAsync(session, markedRequestMessages, markedResponseMessages, chatOptions, cancellationToken);
+        }
+
+        return this.NotifyProvidersOfNewMessagesAsync(session, requestMessages, responseMessages, chatOptions, cancellationToken);
+    }
+
+    /// <summary>
+    /// Notifies providers of a failure at the end of an agent run.
+    /// </summary>
+    /// <remarks>
+    /// When a <see cref="ChatHistoryPersistingChatClient"/> in persist mode handles per-service-call
+    /// notification (including failure), this end-of-run notification is skipped to avoid
+    /// duplicate notification. In all other cases, failure is reported at the end of the run.
+    /// </remarks>
+    private Task NotifyProvidersOfFailureAtEndOfRunAsync(
         ChatClientAgentSession session,
         Exception ex,
         IEnumerable<ChatMessage> requestMessages,
         ChatOptions? chatOptions,
         CancellationToken cancellationToken)
     {
-        ChatHistoryProvider? provider = this.ResolveChatHistoryProvider(chatOptions, session);
-
-        // Only notify the provider if we have one.
-        // If we don't have one, it means that the chat history is service managed and the underlying service is responsible for storing messages.
-        if (provider is not null)
+        if (this.PersistsChatHistoryPerServiceCall)
         {
-            var invokedContext = new ChatHistoryProvider.InvokedContext(this, session, requestMessages, ex);
-
-            return provider.InvokedAsync(invokedContext, cancellationToken).AsTask();
+            return Task.CompletedTask;
         }
 
-        return Task.CompletedTask;
+        return this.NotifyProvidersOfFailureAsync(session, ex, requestMessages, chatOptions, cancellationToken);
     }
 
-    private Task NotifyChatHistoryProviderOfNewMessagesAsync(
-        ChatClientAgentSession session,
-        IEnumerable<ChatMessage> requestMessages,
-        IEnumerable<ChatMessage> responseMessages,
-        ChatOptions? chatOptions,
-        CancellationToken cancellationToken)
+    /// <summary>
+    /// Gets a value indicating whether the agent has a <see cref="ChatHistoryPersistingChatClient"/>
+    /// decorator in persist mode (not mark-only), which handles per-service-call persistence.
+    /// </summary>
+    private bool PersistsChatHistoryPerServiceCall
     {
-        ChatHistoryProvider? provider = this.ResolveChatHistoryProvider(chatOptions, session);
-
-        // Only notify the provider if we have one.
-        // If we don't have one, it means that the chat history is service managed and the underlying service is responsible for storing messages.
-        if (provider is not null)
+        get
         {
-            var invokedContext = new ChatHistoryProvider.InvokedContext(this, session, requestMessages, responseMessages);
-            return provider.InvokedAsync(invokedContext, cancellationToken).AsTask();
+            var persistingClient = this.ChatClient.GetService<ChatHistoryPersistingChatClient>();
+            return persistingClient?.MarkOnly == false;
+        }
+    }
+
+    /// <summary>
+    /// Gets a value indicating whether the agent has a <see cref="ChatHistoryPersistingChatClient"/>
+    /// decorator in mark-only mode, which marks messages for later persistence at the end of the run.
+    /// </summary>
+    private bool HasMarkOnlyChatHistoryPersistingClient
+    {
+        get
+        {
+            var persistingClient = this.ChatClient.GetService<ChatHistoryPersistingChatClient>();
+            return persistingClient?.MarkOnly == true;
+        }
+    }
+
+    /// <summary>
+    /// Returns only the messages that have been marked as persisted by a <see cref="ChatHistoryPersistingChatClient"/> in mark-only mode.
+    /// </summary>
+    private static List<ChatMessage> GetMarkedMessages(IEnumerable<ChatMessage> messages)
+    {
+        return messages.Where(m =>
+            m.AdditionalProperties?.TryGetValue(ChatHistoryPersistingChatClient.PersistedMarkerKey, out var value) == true && value is true).ToList();
+    }
+
+    /// <summary>
+    /// Ensures that <see cref="AIAgent.CurrentRunContext"/> contains the resolved session.
+    /// </summary>
+    /// <remarks>
+    /// The base class sets <see cref="AIAgent.CurrentRunContext"/> with the raw session parameter
+    /// (which may be null) and restores it after each yield in streaming scenarios. After
+    /// <see cref="PrepareSessionAndMessagesAsync"/> resolves or creates a session, we update the
+    /// context so the <see cref="ChatHistoryPersistingChatClient"/> decorator always has a valid session.
+    /// The original agent from the context is preserved to maintain the top-of-stack agent in
+    /// decorated agent scenarios.
+    /// </remarks>
+    private static void EnsureRunContextHasSession(ChatClientAgentSession safeSession)
+    {
+        var context = CurrentRunContext;
+        if (context is not null && context.Session != safeSession)
+        {
+            CurrentRunContext = new(context.Agent, safeSession, context.RequestMessages, context.RunOptions);
+        }
+    }
+
+    /// <summary>
+    /// Checks for potential misconfiguration when using a custom chat client stack and logs warnings.
+    /// </summary>
+    private void WarnOnMissingPersistingClient()
+    {
+        if (this._agentOptions?.UseProvidedChatClientAsIs is not true)
+        {
+            return;
         }
 
-        return Task.CompletedTask;
+        if (this._agentOptions?.PersistChatHistoryAtEndOfRun is not true)
+        {
+            return;
+        }
+
+        var persistingClient = this.ChatClient.GetService<ChatHistoryPersistingChatClient>();
+        if (persistingClient is null && this._logger.IsEnabled(LogLevel.Warning))
+        {
+            var loggingAgentName = this.GetLoggingAgentName();
+            this._logger.LogAgentChatClientMissingPersistingClient(
+                this.Id,
+                loggingAgentName);
+        }
     }
 
     private ChatHistoryProvider? ResolveChatHistoryProvider(ChatOptions? chatOptions, ChatClientAgentSession session)

dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgentLogMessages.cs

@@ -69,4 +69,32 @@ internal static partial class ChatClientAgentLogMessages
         string chatHistoryProviderName,
         string agentId,
         string agentName);
+
+    /// <summary>
+    /// Logs a warning when <see cref="ChatClientAgentOptions.UseProvidedChatClientAsIs"/> is <see langword="true"/>
+    /// and <see cref="ChatClientAgentOptions.PersistChatHistoryAtEndOfRun"/> is <see langword="true"/>,
+    /// but no <see cref="ChatHistoryPersistingChatClient"/> is found in the custom chat client stack.
+    /// </summary>
+    [LoggerMessage(
+        Level = LogLevel.Warning,
+        Message = "Agent {AgentId}/{AgentName}: PersistChatHistoryAtEndOfRun is enabled with a custom chat client stack (UseProvidedChatClientAsIs), but no ChatHistoryPersistingChatClient was found in the pipeline. All messages will be persisted at the end of the run without marking. This setup is not supported with some other features, e.g. handoffs. Consider adding a ChatHistoryPersistingChatClient to the pipeline using the UseChatHistoryPersisting extension method.")]
+    public static partial void LogAgentChatClientMissingPersistingClient(
+        this ILogger logger,
+        string agentId,
+        string agentName);
+
+    /// <summary>
+    /// Logs a warning when per-service-call persistence falls back to end-of-run persistence
+    /// because the run involves background responses (continuation token resumption or
+    /// <c>AllowBackgroundResponses</c>). Per-service-call persistence is
+    /// unreliable in these scenarios because the caller may stop consuming the stream before
+    /// the decorator's post-stream persistence code can execute.
+    /// </summary>
+    [LoggerMessage(
+        Level = LogLevel.Warning,
+        Message = "Agent {AgentId}/{AgentName}: Per-service-call persistence is falling back to end-of-run persistence because the run involves background responses. Messages will be marked during the run and persisted at the end.")]
+    public static partial void LogAgentChatClientBackgroundResponseFallback(
+        this ILogger logger,
+        string agentId,
+        string agentName);
 }

dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientAgentOptions.cs

@@ -1,7 +1,9 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using Microsoft.Extensions.AI;
+using Microsoft.Shared.DiagnosticIds;
 
 namespace Microsoft.Agents.AI;
 
@@ -89,6 +91,56 @@ public sealed class ChatClientAgentOptions
     /// </value>
     public bool ThrowOnChatHistoryProviderConflict { get; set; } = true;
 
+    /// <summary>
+    /// Gets or sets a value indicating whether to persist chat history only at the end of the full agent run
+    /// rather than after each individual service call.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// By default, <see cref="ChatClientAgent"/> persists request and response messages either via
+    /// a <see cref="ChatHistoryProvider"/>, or the underlying AI service's chat history storage.
+    /// Persistence is done immediately after each call to the AI service within the function invocation loop.
+    /// When storing in the underlying AI service, the session's <see cref="ChatClientAgentSession.ConversationId"/>
+    /// is also updated after each service call, keeping it in sync with the service-side conversation state.
+    /// </para>
+    /// <para>
+    /// Setting this property to <see langword="true"/> causes messages to be marked during the function
+    /// invocation loop but persisted only at the end of the full agent run, providing atomic run semantics.
+    /// Updating the <see cref="ChatClientAgentSession.ConversationId"/> is likewise deferred and
+    /// updated only at the end of the run, consistent with atomic run semantics.
+    /// A <see cref="ChatHistoryPersistingChatClient"/> decorator is inserted into the chat client pipeline
+    /// in mark-only mode, and the <see cref="ChatClientAgent"/> persists only the marked messages at the
+    /// end of the run.
+    /// </para>
+    /// <para>
+    /// When this option is <see langword="false"/> (the default), the <see cref="ChatHistoryPersistingChatClient"/>
+    /// decorator persists messages and updates the <see cref="ChatClientAgentSession.ConversationId"/>
+    /// immediately after each service call. This may leave chat history in a state where
+    /// <see cref="FunctionResultContent"/> is required to start a new run if the last successful service
+    /// call returned <see cref="FunctionCallContent"/>.
+    /// </para>
+    /// <para>
+    /// This option has no effect when <see cref="UseProvidedChatClientAsIs"/> is <see langword="true"/>.
+    /// When using a custom chat client stack, you can add a <see cref="ChatHistoryPersistingChatClient"/>
+    /// manually via the <see cref="ChatClientBuilderExtensions.UseChatHistoryPersisting"/>
+    /// extension method.
+    /// </para>
+    /// <para>
+    /// Note that when using single threaded service stored chat history, like OpenAI Conversations,
+    /// there is only one id, so even if the conversation id is not updated after each service call,
+    /// the chat history will still contain intermediate messages. Setting this property to <see langword="true"/>
+    /// in this case will therefore have no real effect. Setting this property to <see langword="true"/> when using
+    /// OpenAI Responses with response ids on the other hand, allows atomic run semantics, since
+    /// each service request produces a new response id, and if the run fails mid-loop, the session will
+    /// still contain the pre-run respnose id, allowing the next run to start with a clean slate.
+    /// </para>
+    /// </remarks>
+    /// <value>
+    /// Default is <see langword="false"/>.
+    /// </value>
+    [Experimental(DiagnosticIds.Experiments.AgentsAIExperiments)]
+    public bool PersistChatHistoryAtEndOfRun { get; set; }
+
     /// <summary>
     /// Creates a new instance of <see cref="ChatClientAgentOptions"/> with the same values as this instance.
     /// </summary>
@@ -105,5 +157,6 @@ public sealed class ChatClientAgentOptions
             ClearOnChatHistoryProviderConflict = this.ClearOnChatHistoryProviderConflict,
             WarnOnChatHistoryProviderConflict = this.WarnOnChatHistoryProviderConflict,
             ThrowOnChatHistoryProviderConflict = this.ThrowOnChatHistoryProviderConflict,
+            PersistChatHistoryAtEndOfRun = this.PersistChatHistoryAtEndOfRun,
         };
 }

dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientBuilderExtensions.cs

@@ -2,8 +2,10 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using Microsoft.Agents.AI;
 using Microsoft.Extensions.Logging;
+using Microsoft.Shared.DiagnosticIds;
 using Microsoft.Shared.Diagnostics;
 
 namespace Microsoft.Extensions.AI;
@@ -82,4 +84,46 @@ public static class ChatClientBuilderExtensions
             options: options,
             loggerFactory: loggerFactory,
             services: services);
+
+    /// <summary>
+    /// Adds a <see cref="ChatHistoryPersistingChatClient"/> to the chat client pipeline.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// This decorator should be positioned between the <see cref="FunctionInvokingChatClient"/> and the leaf
+    /// <see cref="IChatClient"/> in the pipeline. It intercepts service calls to either persist messages
+    /// immediately or mark them for later persistence, depending on the <paramref name="markOnly"/> parameter.
+    /// </para>
+    /// <para>
+    /// If <paramref name="markOnly"/> is set to <see langword="true"/>, the <see cref="ChatClientAgent"/>
+    /// should be configured with <see cref="ChatClientAgentOptions.PersistChatHistoryAtEndOfRun"/> set to <see langword="true"/>
+    /// as without this combination, messages will never be persisted when using a <see cref="ChatHistoryProvider"/> for
+    /// chat history persistence.
+    /// </para>
+    /// <para>
+    /// This extension method is intended for use with custom chat client stacks when
+    /// <see cref="ChatClientAgentOptions.UseProvidedChatClientAsIs"/> is <see langword="true"/>.
+    /// When <see cref="ChatClientAgentOptions.UseProvidedChatClientAsIs"/> is <see langword="false"/> (the default),
+    /// the <see cref="ChatClientAgent"/> automatically injects this decorator.
+    /// </para>
+    /// <para>
+    /// This decorator only works within the context of a running <see cref="ChatClientAgent"/> and will throw an
+    /// exception if used in any other stack.
+    /// </para>
+    /// </remarks>
+    /// <param name="builder">The <see cref="ChatClientBuilder"/> to add the decorator to.</param>
+    /// <param name="markOnly">
+    /// When <see langword="true"/>, messages are marked with metadata but not persisted immediately,
+    /// and the session's <see cref="ChatClientAgentSession.ConversationId"/> is not updated.
+    /// The <see cref="ChatClientAgent"/> will persist only the marked messages and update the
+    /// conversation ID at the end of the run.
+    /// When <see langword="false"/> (the default), messages are persisted and the conversation ID
+    /// is updated immediately after each service call.
+    /// </param>
+    /// <returns>The <paramref name="builder"/> for chaining.</returns>
+    [Experimental(DiagnosticIds.Experiments.AgentsAIExperiments)]
+    public static ChatClientBuilder UseChatHistoryPersisting(this ChatClientBuilder builder, bool markOnly = false)
+    {
+        return builder.Use(innerClient => new ChatHistoryPersistingChatClient(innerClient, markOnly));
+    }
 }

dotnet/src/Microsoft.Agents.AI/ChatClient/ChatClientExtensions.cs

@@ -63,6 +63,15 @@ public static class ChatClientExtensions
             });
         }
 
+        // ChatHistoryPersistingChatClient is registered after FunctionInvokingChatClient so that it sits
+        // between FIC and the leaf client. ChatClientBuilder.Build applies factories in reverse order,
+        // making the first Use() call outermost. By adding our decorator second, the resulting pipeline is:
+        //   FunctionInvokingChatClient → ChatHistoryPersistingChatClient → leaf IChatClient
+        // This allows the decorator to persist messages after each individual service call within
+        // FIC's function invocation loop, or to mark them for later persistence at the end of the run.
+        bool markOnly = options?.PersistChatHistoryAtEndOfRun is true;
+        chatBuilder.Use(innerClient => new ChatHistoryPersistingChatClient(innerClient, markOnly));
+
         var agentChatClient = chatBuilder.Build(services);
 
         if (options?.ChatOptions?.Tools is { Count: > 0 })

dotnet/src/Microsoft.Agents.AI/ChatClient/ChatHistoryPersistingChatClient.cs

@@ -0,0 +1,313 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.AI;
+
+namespace Microsoft.Agents.AI;
+
+/// <summary>
+/// A delegating chat client that notifies <see cref="ChatHistoryProvider"/> and <see cref="AIContextProvider"/>
+/// instances of request and response messages after each individual call to the inner chat client,
+/// or marks messages for later persistence depending on the configured mode.
+/// </summary>
+/// <remarks>
+/// <para>
+/// This decorator is intended to operate between the <see cref="FunctionInvokingChatClient"/> and the leaf
+/// <see cref="IChatClient"/> in a <see cref="ChatClientAgent"/> pipeline.
+/// </para>
+/// <para>
+/// In persist mode (the default), it ensures that providers are notified and the session's
+/// <see cref="ChatClientAgentSession.ConversationId"/> is updated after each service call, so that
+/// intermediate messages (e.g., tool calls and results) are saved even if the process is interrupted
+/// mid-loop.
+/// </para>
+/// <para>
+/// In mark-only mode (<see cref="MarkOnly"/> is <see langword="true"/>), it marks messages with metadata
+/// but does not notify providers or update the <see cref="ChatClientAgentSession.ConversationId"/>.
+/// Both are deferred to the <see cref="ChatClientAgent"/> at the end of the run, providing atomic
+/// run semantics.
+/// </para>
+/// <para>
+/// This chat client must be used within the context of a running <see cref="ChatClientAgent"/>. It retrieves the
+/// current agent and session from <see cref="AIAgent.CurrentRunContext"/>, which is set automatically when an agent's
+/// <see cref="AIAgent.RunAsync(IEnumerable{ChatMessage}, AgentSession?, AgentRunOptions?, CancellationToken)"/> or
+/// <see cref="AIAgent.RunStreamingAsync(IEnumerable{ChatMessage}, AgentSession?, AgentRunOptions?, CancellationToken)"/>
+/// method is called. The <see cref="ChatClientAgent"/> ensures the run context always contains a resolved session,
+/// even when the caller passes null. An <see cref="InvalidOperationException"/> is thrown if no run context is
+/// available or if the agent is not a <see cref="ChatClientAgent"/>.
+/// </para>
+/// </remarks>
+internal sealed class ChatHistoryPersistingChatClient : DelegatingChatClient
+{
+    /// <summary>
+    /// The key used in <see cref="ChatMessage.AdditionalProperties"/> and <see cref="AIContent.AdditionalProperties"/>
+    /// to mark messages and their content as already persisted to chat history.
+    /// </summary>
+    internal const string PersistedMarkerKey = "_chatHistoryPersisted";
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ChatHistoryPersistingChatClient"/> class.
+    /// </summary>
+    /// <param name="innerClient">The underlying chat client that will handle the core operations.</param>
+    /// <param name="markOnly">
+    /// When <see langword="true"/>, messages are marked with metadata but not persisted immediately,
+    /// and the session's <see cref="ChatClientAgentSession.ConversationId"/> is not updated.
+    /// The <see cref="ChatClientAgent"/> will persist only the marked messages and update the
+    /// conversation ID at the end of the run.
+    /// When <see langword="false"/> (the default), messages are persisted and the conversation ID
+    /// is updated immediately after each service call.
+    /// </param>
+    public ChatHistoryPersistingChatClient(IChatClient innerClient, bool markOnly = false)
+        : base(innerClient)
+    {
+        this.MarkOnly = markOnly;
+    }
+
+    /// <summary>
+    /// Gets a value indicating whether this decorator is in mark-only mode.
+    /// </summary>
+    /// <remarks>
+    /// When <see langword="true"/>, messages are marked with metadata but not persisted immediately,
+    /// and the session's <see cref="ChatClientAgentSession.ConversationId"/> is not updated.
+    /// Both are deferred to the <see cref="ChatClientAgent"/> at the end of the run.
+    /// When <see langword="false"/>, messages are persisted and the conversation ID is updated
+    /// after each service call.
+    /// </remarks>
+    public bool MarkOnly { get; }
+
+    /// <inheritdoc/>
+    public override async Task<ChatResponse> GetResponseAsync(
+        IEnumerable<ChatMessage> messages,
+        ChatOptions? options = null,
+        CancellationToken cancellationToken = default)
+    {
+        var (agent, session) = GetRequiredAgentAndSession();
+
+        ChatResponse response;
+        try
+        {
+            response = await base.GetResponseAsync(messages, options, cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            var newRequestMessagesOnFailure = GetNewRequestMessages(messages);
+            await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false);
+            throw;
+        }
+
+        var newRequestMessages = GetNewRequestMessages(messages);
+
+        if (this.ShouldDeferPersistence(options))
+        {
+            // In mark-only mode or when resuming from a continuation token, just mark messages
+            // for later persistence by ChatClientAgent. Conversation ID and provider notification
+            // are deferred to end-of-run. For continuation tokens, the end-of-run handler needs
+            // to send the combined data from both the previous and current runs.
+            MarkAsPersisted(newRequestMessages);
+            MarkAsPersisted(response.Messages);
+        }
+        else
+        {
+            // In persist mode, persist immediately and update conversation ID.
+            agent.UpdateSessionConversationId(session, response.ConversationId, cancellationToken);
+            await agent.NotifyProvidersOfNewMessagesAsync(session, newRequestMessages, response.Messages, options, cancellationToken).ConfigureAwait(false);
+            MarkAsPersisted(newRequestMessages);
+            MarkAsPersisted(response.Messages);
+        }
+
+        return response;
+    }
+
+    /// <inheritdoc/>
+    public override async IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(
+        IEnumerable<ChatMessage> messages,
+        ChatOptions? options = null,
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var (agent, session) = GetRequiredAgentAndSession();
+
+        List<ChatResponseUpdate> responseUpdates = [];
+
+        IAsyncEnumerator<ChatResponseUpdate> enumerator;
+        try
+        {
+            enumerator = base.GetStreamingResponseAsync(messages, options, cancellationToken).GetAsyncEnumerator(cancellationToken);
+        }
+        catch (Exception ex)
+        {
+            var newRequestMessagesOnFailure = GetNewRequestMessages(messages);
+            await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false);
+            throw;
+        }
+
+        bool hasUpdates;
+        try
+        {
+            hasUpdates = await enumerator.MoveNextAsync().ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            var newRequestMessagesOnFailure = GetNewRequestMessages(messages);
+            await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false);
+            throw;
+        }
+
+        while (hasUpdates)
+        {
+            var update = enumerator.Current;
+            responseUpdates.Add(update);
+            yield return update;
+
+            try
+            {
+                hasUpdates = await enumerator.MoveNextAsync().ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                var newRequestMessagesOnFailure = GetNewRequestMessages(messages);
+                await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false);
+                throw;
+            }
+        }
+
+        var chatResponse = responseUpdates.ToChatResponse();
+        var newRequestMessages = GetNewRequestMessages(messages);
+
+        if (this.ShouldDeferPersistence(options))
+        {
+            // In mark-only mode or when resuming from a continuation token, just mark messages
+            // for later persistence by ChatClientAgent. Conversation ID and provider notification
+            // are deferred to end-of-run. For continuation tokens, the end-of-run handler needs
+            // to send the combined data from both the previous and current runs.
+            MarkAsPersisted(newRequestMessages);
+            MarkAsPersisted(chatResponse.Messages);
+        }
+        else
+        {
+            // In persist mode, persist immediately and update conversation ID.
+            agent.UpdateSessionConversationId(session, chatResponse.ConversationId, cancellationToken);
+            await agent.NotifyProvidersOfNewMessagesAsync(session, newRequestMessages, chatResponse.Messages, options, cancellationToken).ConfigureAwait(false);
+            MarkAsPersisted(newRequestMessages);
+            MarkAsPersisted(chatResponse.Messages);
+        }
+    }
+
+    /// <summary>
+    /// Gets the current <see cref="ChatClientAgent"/> and <see cref="ChatClientAgentSession"/> from the run context.
+    /// </summary>
+    private static (ChatClientAgent Agent, ChatClientAgentSession Session) GetRequiredAgentAndSession()
+    {
+        var runContext = AIAgent.CurrentRunContext
+            ?? throw new InvalidOperationException(
+                $"{nameof(ChatHistoryPersistingChatClient)} can only be used within the context of a running AIAgent. " +
+                "Ensure that the chat client is being invoked as part of an AIAgent.RunAsync or AIAgent.RunStreamingAsync call.");
+
+        var chatClientAgent = runContext.Agent.GetService<ChatClientAgent>()
+            ?? throw new InvalidOperationException(
+                $"{nameof(ChatHistoryPersistingChatClient)} can only be used with a {nameof(ChatClientAgent)}. " +
+                $"The current agent is of type '{runContext.Agent.GetType().Name}'.");
+
+        if (runContext.Session is not ChatClientAgentSession chatClientAgentSession)
+        {
+            throw new InvalidOperationException(
+                $"{nameof(ChatHistoryPersistingChatClient)} requires a {nameof(ChatClientAgentSession)}. " +
+                $"The current session is of type '{runContext.Session?.GetType().Name ?? "null"}'.");
+        }
+
+        return (chatClientAgent, chatClientAgentSession);
+    }
+
+    /// <summary>
+    /// Determines whether persistence should be deferred to end-of-run instead of happening immediately.
+    /// </summary>
+    /// <returns>
+    /// <see langword="true"/> when in <see cref="MarkOnly"/> mode, when the call is resuming from
+    /// a continuation token (since the end-of-run handler needs to combine data from the previous
+    /// and current runs), or when background responses are allowed (since the caller may stop
+    /// consuming the stream mid-run, preventing the post-stream persistence code from executing).
+    /// </returns>
+    private bool ShouldDeferPersistence(ChatOptions? options)
+    {
+        return this.MarkOnly || options?.ContinuationToken is not null || options?.AllowBackgroundResponses is true;
+    }
+
+    /// <summary>
+    /// Returns only the request messages that have not yet been persisted to chat history.
+    /// </summary>
+    /// <remarks>
+    /// A message is considered already persisted if any of the following is true:
+    /// <list type="bullet">
+    /// <item>It has the <see cref="PersistedMarkerKey"/> in its <see cref="ChatMessage.AdditionalProperties"/>.</item>
+    /// <item>It has an <see cref="AgentRequestMessageSourceType"/> of <see cref="AgentRequestMessageSourceType.ChatHistory"/>
+    /// (indicating it was loaded from chat history and does not need to be re-persisted).</item>
+    /// <item>It has <see cref="ChatMessage.Contents"/> and all of its <see cref="AIContent"/> items have the
+    /// <see cref="PersistedMarkerKey"/> in their <see cref="AIContent.AdditionalProperties"/>. This handles the
+    /// streaming case where <see cref="FunctionInvokingChatClient"/> reconstructs <see cref="ChatMessage"/> objects
+    /// independently via <c>ToChatResponse()</c>, producing different object references that share the same
+    /// underlying <see cref="AIContent"/> instances.</item>
+    /// </list>
+    /// </remarks>
+    /// <returns>A list of request messages that have not yet been persisted.</returns>
+    /// <param name="messages">The full set of request messages to filter.</param>
+    private static List<ChatMessage> GetNewRequestMessages(IEnumerable<ChatMessage> messages)
+    {
+        return messages.Where(m => !IsAlreadyPersisted(m)).ToList();
+    }
+
+    /// <summary>
+    /// Determines whether a message has already been persisted to chat history by this decorator.
+    /// </summary>
+    private static bool IsAlreadyPersisted(ChatMessage message)
+    {
+        if (message.AdditionalProperties?.TryGetValue(PersistedMarkerKey, out var value) == true && value is true)
+        {
+            return true;
+        }
+
+        if (message.GetAgentRequestMessageSourceType() == AgentRequestMessageSourceType.ChatHistory)
+        {
+            return true;
+        }
+
+        // In streaming mode, FunctionInvokingChatClient reconstructs ChatMessage objects via ToChatResponse()
+        // independently, producing different ChatMessage instances. However, the underlying AIContent objects
+        // (e.g., FunctionCallContent, FunctionResultContent) are shared references. Checking for markers on
+        // AIContent handles dedup in this case.
+        if (message.Contents.Count > 0 && message.Contents.All(c => c.AdditionalProperties?.TryGetValue(PersistedMarkerKey, out var value) == true && value is true))
+        {
+            return true;
+        }
+
+        return false;
+    }
+
+    /// <summary>
+    /// Marks the given messages as persisted by setting a marker on both the <see cref="ChatMessage"/>
+    /// and each of its <see cref="AIContent"/> items.
+    /// </summary>
+    /// <remarks>
+    /// Both levels are marked because <see cref="FunctionInvokingChatClient"/> may reconstruct
+    /// <see cref="ChatMessage"/> objects in streaming mode (losing the message-level marker),
+    /// but the <see cref="AIContent"/> references are shared and retain their markers.
+    /// </remarks>
+    /// <param name="messages">The messages to mark as persisted.</param>
+    private static void MarkAsPersisted(IEnumerable<ChatMessage> messages)
+    {
+        foreach (var message in messages)
+        {
+            message.AdditionalProperties ??= new();
+            message.AdditionalProperties[PersistedMarkerKey] = true;
+
+            foreach (var content in message.Contents)
+            {
+                content.AdditionalProperties ??= new();
+                content.AdditionalProperties[PersistedMarkerKey] = true;
+            }
+        }
+    }
+}