Merge branch 'main' into dev/dotnet_workflow/add_handoff_sample

docs: Add Handoff sample to readme
test: Add Handoff composability test (#5208 )
2026-06-16 21:04:09 +08:00 · 2026-04-16 13:04:51 -04:00 · 2026-04-16 12:42:28 -04:00 · 2026-04-16 16:36:09 +00:00 · 2026-04-16 12:01:19 -04:00 · 2026-04-16 15:22:39 +00:00
37 changed files with 2116 additions and 59 deletions
@@ -171,7 +171,7 @@ jobs:
          -m integration
          -n logical --dist worksteal
          --timeout=120 --session-timeout=900 --timeout_method thread
-          --retries 2 --retry-delay 5
+          --retries 2 --retry-delay 30
      - name: Stop local MCP server
        if: always()
        shell: bash
@@ -287,7 +287,7 @@ jobs:
          -m integration
          -n logical --dist worksteal
          --timeout=120 --session-timeout=900 --timeout_method thread
-          --retries 2 --retry-delay 5
+          --retries 2 --retry-delay 30
          --junitxml=pytest.xml
        working-directory: ./python
      - name: Stop local MCP server
@@ -4,8 +4,9 @@
  <!-- https://learn.microsoft.com/en-us/nuget/consume-packages/Central-Package-Management -->
  <Sdk Name="Microsoft.Build.CentralPackageVersions" Version="2.1.3" />
  <!-- Only run 'dotnet format' on dev machines, Release builds. Skip on GitHub Actions -->
-  <!-- as this runs in its own Actions job. -->
-  <Target Name="DotnetFormatOnBuild" BeforeTargets="Build" Condition=" '$(Configuration)' == 'Release' AND '$(GITHUB_ACTIONS)' == '' ">
+  <!-- as this runs in its own Actions job. Only run for net10.0 target frameworks since the dotnet format command -->
+  <!-- already formats all target frameworks in project. Otherwise it will run format x times x where x is the number of target frameworks -->
+  <Target Name="DotnetFormatOnBuild" BeforeTargets="Build" Condition=" '$(Configuration)' == 'Release' AND '$(GITHUB_ACTIONS)' == '' AND '$(TargetFramework)' == 'net10.0' ">
    <Message Text="Running dotnet format" Importance="high" />
    <Exec Command="dotnet format --no-restore -v diag $(ProjectFileName)" />
  </Target>
@@ -11,8 +11,8 @@
  </PropertyGroup>
  <ItemGroup>
    <!-- Aspire.* -->
-    <PackageVersion Include="Anthropic" Version="12.11.0" />
-    <PackageVersion Include="Anthropic.Foundry" Version="0.4.2" />
+    <PackageVersion Include="Anthropic" Version="12.13.0" />
+    <PackageVersion Include="Anthropic.Foundry" Version="0.5.0" />
    <PackageVersion Include="Aspire.Azure.AI.OpenAI" Version="13.0.0-preview.1.25560.3" />
    <PackageVersion Include="Aspire.Hosting.AppHost" Version="$(AspireAppHostSdkVersion)" />
    <PackageVersion Include="Aspire.Hosting.Azure.CognitiveServices" Version="$(AspireAppHostSdkVersion)" />
@@ -32,19 +32,19 @@
    <!-- Newtonsoft.Json -->
    <PackageVersion Include="Newtonsoft.Json" Version="13.0.4" />
    <!-- System.* -->
-    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.4" />
+    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.6" />
    <PackageVersion Include="Microsoft.Bcl.HashCode" Version="6.0.0" />
    <PackageVersion Include="Microsoft.Bcl.Memory" Version="10.0.4" />
    <PackageVersion Include="System.ClientModel" Version="1.10.0" />
    <PackageVersion Include="System.CodeDom" Version="10.0.0" />
    <PackageVersion Include="System.Collections.Immutable" Version="10.0.1" />
    <PackageVersion Include="System.CommandLine" Version="2.0.0-rc.2.25502.107" />
-    <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="10.0.4" />
+    <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="10.0.6" />
    <PackageVersion Include="System.Linq.AsyncEnumerable" Version="10.0.4" />
    <PackageVersion Include="System.Net.Http.Json" Version="10.0.0" />
    <PackageVersion Include="System.Net.ServerSentEvents" Version="10.0.4" />
-    <PackageVersion Include="System.Text.Json" Version="10.0.4" />
-    <PackageVersion Include="System.Threading.Channels" Version="10.0.4" />
+    <PackageVersion Include="System.Text.Json" Version="10.0.6" />
+    <PackageVersion Include="System.Threading.Channels" Version="10.0.6" />
    <PackageVersion Include="System.Threading.Tasks.Extensions" Version="4.6.3" />
    <PackageVersion Include="System.Net.Security" Version="4.3.2" />
    <!-- OpenTelemetry -->
@@ -63,37 +63,28 @@
    <PackageVersion Include="Microsoft.AspNetCore.OpenApi" Version="10.0.0" />
    <PackageVersion Include="Swashbuckle.AspNetCore.SwaggerUI" Version="10.0.0" />
    <!-- Microsoft.Extensions.* -->
-    <PackageVersion Include="Microsoft.Extensions.AI" Version="10.4.0" />
-    <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="10.4.0" />
-    <PackageVersion Include="Microsoft.Extensions.AI.Evaluation" Version="10.4.0" />
-    <PackageVersion Include="Microsoft.Extensions.AI.Evaluation.Quality" Version="10.4.0" />
-    <PackageVersion Include="Microsoft.Extensions.AI.Evaluation.Safety" Version="10.3.0-preview.1.26109.11" />
-    <PackageVersion Include="Microsoft.Extensions.AI.OpenAI" Version="10.4.0" />
+    <PackageVersion Include="Microsoft.Extensions.AI" Version="10.5.0" />
+    <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="10.5.0" />
+    <PackageVersion Include="Microsoft.Extensions.AI.OpenAI" Version="10.5.0" />
    <PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="10.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.Compliance.Abstractions" Version="10.4.0" />
+    <PackageVersion Include="Microsoft.Extensions.Compliance.Abstractions" Version="10.5.0" />
    <PackageVersion Include="Microsoft.Extensions.Configuration" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.Configuration.Binder" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.Configuration.UserSecrets" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.4" />
+    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.6" />
    <PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.4" />
+    <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.6" />
    <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.ServiceDiscovery" Version="10.0.0" />
    <PackageVersion Include="Microsoft.Extensions.VectorData.Abstractions" Version="9.7.0" />
    <!-- Vector Stores -->
    <PackageVersion Include="Microsoft.SemanticKernel.Connectors.InMemory" Version="1.67.0-preview" />
    <PackageVersion Include="Microsoft.SemanticKernel.Connectors.Qdrant" Version="1.67.0-preview" />
-    <!-- Semantic Kernel -->
-    <PackageVersion Include="Microsoft.SemanticKernel" Version="1.67.0" />
-    <PackageVersion Include="Microsoft.SemanticKernel.Agents.Core" Version="1.67.0" />
-    <PackageVersion Include="Microsoft.SemanticKernel.Agents.OpenAI" Version="1.67.0-preview" />
-    <PackageVersion Include="Microsoft.SemanticKernel.Agents.AzureAI" Version="1.67.0-preview" />
-    <PackageVersion Include="Microsoft.SemanticKernel.Plugins.OpenApi" Version="1.67.0" />
    <!-- Agent SDKs -->
    <PackageVersion Include="GitHub.Copilot.SDK" Version="0.1.29" />
    <PackageVersion Include="Microsoft.Agents.CopilotStudio.Client" Version="1.3.171-beta" />
@@ -107,11 +98,10 @@
    <!-- MCP -->
    <PackageVersion Include="ModelContextProtocol" Version="1.1.0" />
    <!-- Inference SDKs -->
-    <PackageVersion Include="AWSSDK.Extensions.Bedrock.MEAI" Version="4.0.5.1" />
    <PackageVersion Include="Microsoft.ML.OnnxRuntimeGenAI" Version="0.10.0" />
    <PackageVersion Include="Microsoft.ML.Tokenizers" Version="2.0.0" />
    <PackageVersion Include="OllamaSharp" Version="5.4.8" />
-    <PackageVersion Include="OpenAI" Version="2.9.1" />
+    <PackageVersion Include="OpenAI" Version="2.10.0" />
    <!-- Identity -->
    <PackageVersion Include="Microsoft.Identity.Client.Extensions.Msal" Version="4.83.1" />
    <!-- Workflows -->
@@ -126,7 +116,6 @@
    <PackageVersion Include="Microsoft.DurableTask.Worker.AzureManaged" Version="1.18.0" />
    <!-- Azure Functions -->
    <PackageVersion Include="Microsoft.Azure.Functions.Worker" Version="2.50.0" />
-    <PackageVersion Include="Microsoft.Azure.Functions.Worker.ApplicationInsights" Version="2.50.0" />
    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.12.1" />
    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" Version="1.0.1" />
    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http" Version="3.3.0" />
@@ -243,6 +243,9 @@
  <Folder Name="/Samples/03-workflows/HumanInTheLoop/">
    <Project Path="samples/03-workflows/HumanInTheLoop/HumanInTheLoopBasic/HumanInTheLoopBasic.csproj" />
  </Folder>
+  <Folder Name="/Samples/03-workflows/Orchestration/">
+    <Project Path="samples/03-workflows/Orchestration/Handoff/Handoff.csproj" />
+  </Folder>
  <Folder Name="/Samples/03-workflows/Observability/">
    <Project Path="samples/03-workflows/Observability/ApplicationInsights/ApplicationInsights.csproj" />
    <Project Path="samples/03-workflows/Observability/AspireDashboard/AspireDashboard.csproj" />
@@ -288,7 +291,7 @@
    <File Path="samples/04-hosting/A2A/README.md" />
    <Project Path="samples/04-hosting/A2A/A2AAgent_AsFunctionTools/A2AAgent_AsFunctionTools.csproj" />
    <Project Path="samples/04-hosting/A2A/A2AAgent_PollingForTaskCompletion/A2AAgent_PollingForTaskCompletion.csproj" />
-  </Folder>  
+  </Folder>
  <Folder Name="/Samples/05-end-to-end/">
    <Project Path="samples/05-end-to-end/AgentWithPurview/AgentWithPurview.csproj" />
    <Project Path="samples/05-end-to-end/M365Agent/M365Agent.csproj" />
@@ -0,0 +1,72 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Microsoft.Agents.AI;
+using Microsoft.Agents.AI.Workflows;
+using Microsoft.Extensions.AI;
+
+/// <summary>
+/// The registry of agents used in the workflow.
+/// </summary>
+/// <param name="chatClient">The <see cref="IChatClient"/> to use as the agent backend.</param>
+internal sealed class AgentRegistry(IChatClient chatClient)
+{
+    internal const string IntakeAgentName = "Assistant";
+    public AIAgent IntakeAgent { get; } = chatClient.AsAIAgent(
+        instructions:
+            """
+                You receive a user request and are responsible for routing to the correct initial expert agent.
+                """,
+        IntakeAgentName
+        );
+
+    internal const string LiquidityAnalysisAgentName = "Liquidity Analysis";
+    public AIAgent LiquidityAnalysisAgent { get; } = chatClient.AsAIAgent(
+        instructions:
+            """
+                You are responsible for Liquidity Analysis.
+                """,
+        LiquidityAnalysisAgentName
+        );
+
+    internal const string TaxAnalysisAgentName = "Tax Analysis";
+    public AIAgent TaxAnalysisAgent { get; } = chatClient.AsAIAgent(
+        instructions:
+            """
+                You are responsible for Tax Analysis. 
+                """,
+        TaxAnalysisAgentName
+        );
+
+    internal const string ForeignExchangeAgentName = "Foreign Exchange Analysis";
+    public AIAgent ForeignExchangeAgent { get; } = chatClient.AsAIAgent(
+        instructions:
+            """
+                You are responsible for Foreign Exchange Analysis. 
+                """,
+        ForeignExchangeAgentName
+        );
+
+    internal const string EquityAgentName = "Equity Analysis";
+    public AIAgent EquityAgent { get; } = chatClient.AsAIAgent(
+        instructions:
+            """
+                You are responsible for Equity Analysis. 
+                """,
+        EquityAgentName
+        );
+
+    public IEnumerable<AIAgent> Experts => [this.LiquidityAnalysisAgent, this.TaxAnalysisAgent, this.ForeignExchangeAgent, this.EquityAgent];
+
+    public HashSet<AIAgent> All
+    {
+        get
+        {
+            if (field == null)
+            {
+                field = [this.IntakeAgent, .. this.Experts];
+            }
+
+            return field;
+        }
+    }
+}
@@ -0,0 +1,29 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFrameworks>net10.0</TargetFrameworks>
+
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+
+    <NoWarn>MAAIW001</NoWarn>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Azure.AI.Projects" />
+    <PackageReference Include="Azure.Identity" />
+    <PackageReference Include="Microsoft.Extensions.AI.OpenAI" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
+    <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI\Microsoft.Agents.AI.csproj" />
+    <!-- Include Workflows source generator when using [MessageHandler] attribute -->
+    <ProjectReference Include="$(RepoRoot)/dotnet/src/Microsoft.Agents.AI.Workflows.Generators/Microsoft.Agents.AI.Workflows.Generators.csproj"
+                      OutputItemType="Analyzer"
+                      ReferenceOutputAssembly="false"
+                      GlobalPropertiesToRemove="TargetFramework" />
+  </ItemGroup>
+
+</Project>
@@ -0,0 +1,125 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Azure.AI.Projects;
+using Azure.Identity;
+using Microsoft.Agents.AI;
+using Microsoft.Agents.AI.Workflows;
+using Microsoft.Extensions.AI;
+
+string endpoint = Environment.GetEnvironmentVariable("AZURE_AI_PROJECT_ENDPOINT")
+            ?? throw new InvalidOperationException("AZURE_AI_PROJECT_ENDPOINT is not set.");
+string deploymentName = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "gpt-5.4-mini";
+
+// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production.
+// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid
+// latency issues, unintended credential probing, and potential security risks from fallback mechanisms.
+AIProjectClient projectClient = new(new Uri(endpoint), new DefaultAzureCredential());
+
+IChatClient chatClient = projectClient.ProjectOpenAIClient
+                                      .GetChatClient(deploymentName)
+                                      .AsIChatClient();
+
+Workflow workflow = CreateWorkflow(chatClient);
+
+await RunWorkflowAsync(workflow).ConfigureAwait(false);
+
+static Workflow CreateWorkflow(IChatClient chatClient)
+{
+    AgentRegistry agents = new(chatClient);
+
+    HandoffWorkflowBuilder handoffBuilder = AgentWorkflowBuilder.CreateHandoffBuilderWith(agents.IntakeAgent);
+
+    // Add a handoff to each of the experts from every agent in the registry (experts + Intake)
+    foreach (AIAgent expert in agents.Experts)
+    {
+        handoffBuilder.WithHandoffs(agents.All.Except([expert]), expert);
+    }
+
+    // Let agents request more user information and return to the asking agent (rather than going back to the intake agent)
+    handoffBuilder.EnableReturnToPrevious();
+
+    return handoffBuilder.Build();
+}
+
+static async Task RunWorkflowAsync(Workflow workflow)
+{
+    using CancellationTokenSource cts = CreateConsoleCancelKeySource();
+    await using StreamingRun run = await InProcessExecution.OpenStreamingAsync(workflow, cancellationToken: cts.Token)
+                                                           .ConfigureAwait(false);
+
+    bool hadError = false;
+    do
+    {
+        Console.Write("> ");
+        string userInput = Console.ReadLine() ?? string.Empty;
+
+        if (userInput.Equals("exit", StringComparison.OrdinalIgnoreCase))
+        {
+            break;
+        }
+
+        await run.TrySendMessageAsync(userInput);
+        string? speakingAgent = null;
+        await foreach (WorkflowEvent evt in run.WatchStreamAsync(cts.Token))
+        {
+            switch (evt)
+            {
+                case AgentResponseUpdateEvent update:
+                {
+                    if (speakingAgent == null || speakingAgent != update.Update.AuthorName)
+                    {
+                        speakingAgent = update.Update.AuthorName;
+                        Console.Write($"\n{speakingAgent}: ");
+                    }
+
+                    Console.Write(update.Update.Text);
+                    break;
+                }
+
+                case WorkflowErrorEvent workflowError:
+                {
+                    Console.ForegroundColor = ConsoleColor.Red;
+
+                    if (workflowError.Exception != null)
+                    {
+                        Console.WriteLine($"\nWorkflow error: {workflowError.Exception}");
+                    }
+                    else
+                    {
+                        Console.WriteLine("\nUnknown workflow error occurred.");
+                    }
+
+                    Console.ResetColor();
+
+                    hadError = true;
+                    break;
+                }
+
+                case WorkflowWarningEvent workflowWarning when workflowWarning.Data is string message:
+                {
+                    Console.ForegroundColor = ConsoleColor.Yellow;
+                    Console.WriteLine(message);
+                    Console.ResetColor();
+                    break;
+                }
+            }
+        }
+    } while (!hadError);
+}
+
+static CancellationTokenSource CreateConsoleCancelKeySource()
+{
+    CancellationTokenSource cts = new();
+
+    // Normally, support a way to detach events, but in this case this is a termination signal, so cleanup will happen
+    // as part of application shutdown.
+    Console.CancelKeyPress += (s, args) =>
+    {
+        cts.Cancel();
+
+        // We handle cleanup + termination ourselves
+        args.Cancel = true;
+    };
+
+    return cts;
+}
@@ -56,3 +56,9 @@ Once completed, please proceed to the other samples listed below.
 | [Edge Conditions](./ConditionalEdges/01_EdgeCondition) | Introduces conditional edges for dynamic routing based on executor outputs |
 | [Switch-Case Routing](./ConditionalEdges/02_SwitchCase) | Extends conditional edges with switch-case routing for multiple paths |
 | [Multi-Selection Routing](./ConditionalEdges/03_MultiSelection) | Demonstrates multi-selection routing where one executor can trigger multiple downstream executors |
+
+### Orchestration Patterns
+
+| Sample | Concepts |
+|--------|----------|
+| [Handoff Orchestration](./Orchestration/Handoff) | Introduces the Handoff Orchestration pattern |
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <IsReleaseCandidate>true</IsReleaseCandidate>
+    <IsReleaseCandidate>false</IsReleaseCandidate>
    <ImplicitUsings>enable</ImplicitUsings>
    <InjectSharedThrow>true</InjectSharedThrow>
  </PropertyGroup>
@@ -419,6 +419,12 @@ internal sealed class InProcessRunnerContext : IRunnerContext
                                                      .Select(id => this.EnsureExecutorAsync(id, tracer: null).AsTask())
                                                      .ToArray();

+        // Discard queued external deliveries from the superseded timeline so a runtime
+        // restore cannot apply stale responses after importing the checkpoint state.
+        while (this._queuedExternalDeliveries.TryDequeue(out _))
+        {
+        }
+
        this._nextStep = new StepContext();
        this._nextStep.ImportMessages(importedState.QueuedMessages);

@@ -483,6 +483,14 @@ public sealed class AnthropicBetaServiceExtensionsTests

            public IBetaMessageService Messages => new Mock<IBetaMessageService>().Object;

+            public global::Anthropic.Services.Beta.IAgentService Agents => throw new NotImplementedException();
+
+            public global::Anthropic.Services.Beta.IEnvironmentService Environments => throw new NotImplementedException();
+
+            public global::Anthropic.Services.Beta.ISessionService Sessions => throw new NotImplementedException();
+
+            public global::Anthropic.Services.Beta.IVaultService Vaults => throw new NotImplementedException();
+
            public IBetaService WithOptions(Func<ClientOptions, ClientOptions> modifier)
            {
                throw new NotImplementedException();
@@ -279,6 +279,48 @@ public class CheckpointResumeTests
            "the workflow should be able to continue after the runtime restore replay");
    }

+    /// <summary>
+    /// Verifies that restoring a live run clears any queued external responses from the
+    /// superseded timeline before importing checkpoint state.
+    /// </summary>
+    [Fact]
+    internal async Task Checkpoint_Restore_ClearsQueuedExternalResponsesBeforeImportAsync()
+    {
+        Workflow workflow = CreateSimpleRequestWorkflow();
+        CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
+        InProcessExecutionEnvironment env = ExecutionEnvironment.InProcess_Lockstep.ToWorkflowExecutionEnvironment();
+
+        await using StreamingRun run = await env.WithCheckpointing(checkpointManager)
+                                                .RunStreamingAsync(workflow, "Hello");
+
+        (ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(run);
+
+        await run.SendResponseAsync(pendingRequest.CreateResponse("World"));
+        await run.RestoreCheckpointAsync(checkpoint);
+
+        List<WorkflowEvent> restoredEvents = await ReadToHaltAsync(run);
+        ExternalRequest replayedRequest = restoredEvents.OfType<RequestInfoEvent>()
+                                                        .Select(evt => evt.Request)
+                                                        .Should()
+                                                        .ContainSingle("the restored run should still be waiting for the checkpointed request")
+                                                        .Subject;
+
+        restoredEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
+            "a queued response from the superseded timeline should not be processed after restore");
+        RunStatus statusAfterRestore = await run.GetStatusAsync();
+        statusAfterRestore.Should().Be(RunStatus.PendingRequests,
+            "the restored run should remain pending until a post-restore response is sent");
+
+        await run.SendResponseAsync(replayedRequest.CreateResponse("Again"));
+
+        List<WorkflowEvent> completionEvents = await ReadToHaltAsync(run);
+        completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
+            "the restored request should complete cleanly once a new response is provided");
+        RunStatus finalStatus = await run.GetStatusAsync();
+        finalStatus.Should().Be(RunStatus.Idle,
+            "the workflow should finish once the replayed request receives a fresh response");
+    }
+
    /// <summary>
    /// Verifies that a resumed parent workflow re-emits pending requests that originated in a subworkflow.
    /// </summary>
@@ -1,8 +1,14 @@
 // 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 FluentAssertions;
 using Microsoft.Agents.AI.Workflows.Specialized;
+using Microsoft.Extensions.AI;

 namespace Microsoft.Agents.AI.Workflows.UnitTests;

@@ -68,4 +74,98 @@ public class HandoffAgentExecutorTests : AIAgentHostingExecutorTestsBase
        AgentResponseEvent[] updates = testContext.Events.OfType<AgentResponseEvent>().ToArray();
        CheckResponseEventsAgainstTestMessages(updates, expectingResponse: executorSetting, agent.GetDescriptiveId());
    }
+
+    [Fact]
+    public async Task Test_HandoffAgentExecutor_PreservesExistingInstructionsAndToolsAsync()
+    {
+        // Arrange
+        const string BaseInstructions = "BaseInstructions";
+        const string HandoffInstructions = "HandoffInstructions";
+
+        AITool someTool = AIFunctionFactory.CreateDeclaration("BaseTool", null, AIFunctionFactory.Create(() => { }).JsonSchema);
+
+        OptionValidatingChatClient chatClient = new(BaseInstructions, HandoffInstructions, someTool);
+        AIAgent handoffAgent = chatClient.AsAIAgent(BaseInstructions, tools: [someTool]);
+        AIAgent targetAgent = new TestEchoAgent();
+
+        HandoffAgentExecutorOptions options = new(HandoffInstructions, false, null, HandoffToolCallFilteringBehavior.None);
+        HandoffTarget handoff = new(targetAgent);
+        HandoffAgentExecutor executor = new(handoffAgent, [handoff], options);
+
+        TestWorkflowContext testContext = new(executor.Id);
+        HandoffState state = new(new(false), null, [], null);
+
+        // Act / Assert
+        Func<Task> runStreamingAsync = async () => await executor.HandleAsync(state, testContext);
+        await runStreamingAsync.Should().NotThrowAsync();
+    }
+
+    private sealed class OptionValidatingChatClient(string baseInstructions, string handoffInstructions, AITool baseTool) : IChatClient
+    {
+        public void Dispose()
+        {
+        }
+
+        private void CheckOptions(ChatOptions? options)
+        {
+            options.Should().NotBeNull();
+
+            options.Instructions.Should().NotBeNullOrEmpty("Handoff orchestration should preserve and augment instructions.")
+                                     .And.Contain(baseInstructions, because: "Handoff orchestration should preserve existing instructions.")
+                                     .And.Contain(handoffInstructions, because: "Handoff orchestration should inject handoff instructions.");
+
+            options.Tools.Should().NotBeNullOrEmpty("Handoff orchestration should preserve and augment tools.")
+                                  .And.Contain(tool => tool.Name == baseTool.Name, "Handoff orchestration should preserve existing tools.")
+                                  .And.Contain(tool => tool.Name.StartsWith(HandoffWorkflowBuilder.FunctionPrefix, StringComparison.Ordinal),
+                                               because: "Handoff orchestration should inject handoff tools.");
+        }
+
+        private List<ChatMessage> ResponseMessages =>
+            [
+                new ChatMessage(ChatRole.Assistant, "Ok")
+                {
+                    MessageId = Guid.NewGuid().ToString(),
+                    AuthorName = nameof(OptionValidatingChatClient)
+                }
+            ];
+
+        public Task<ChatResponse> GetResponseAsync(IEnumerable<ChatMessage> messages, ChatOptions? options = null, CancellationToken cancellationToken = default)
+        {
+            this.CheckOptions(options);
+
+            ChatResponse response = new(this.ResponseMessages)
+            {
+                ResponseId = Guid.NewGuid().ToString("N"),
+                CreatedAt = DateTimeOffset.Now
+            };
+
+            return Task.FromResult(response);
+        }
+
+        public object? GetService(Type serviceType, object? serviceKey = null)
+        {
+            if (serviceType == typeof(OptionValidatingChatClient))
+            {
+                return this;
+            }
+
+            return null;
+        }
+
+        public async IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(IEnumerable<ChatMessage> messages, ChatOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        {
+            this.CheckOptions(options);
+
+            string responseId = Guid.NewGuid().ToString("N");
+            foreach (ChatMessage message in this.ResponseMessages)
+            {
+                yield return new(message.Role, message.Contents)
+                {
+                    ResponseId = responseId,
+                    MessageId = message.MessageId,
+                    CreatedAt = DateTimeOffset.Now
+                };
+            }
+        }
+    }
 }
@@ -374,6 +374,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
                    contents=contents,
                    role="assistant" if item.role == A2ARole.agent else "user",
                    response_id=str(getattr(item, "message_id", uuid.uuid4())),
+                    additional_properties={"a2a_metadata": item.metadata} if item.metadata else None,
                    raw_representation=item,
                )
                all_updates.append(update)
@@ -452,13 +453,24 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
                        role=message.role,
                        response_id=task.id,
                        message_id=getattr(message.raw_representation, "artifact_id", None),
+                        additional_properties={"a2a_metadata": merged}
+                        if (merged := {**message.additional_properties, **(task.metadata or {})})
+                        else None,
                        raw_representation=task,
                    )
                    for message in task_messages
                ]
            if task.artifacts is not None:
                return []
-            return [AgentResponseUpdate(contents=[], role="assistant", response_id=task.id, raw_representation=task)]
+            return [
+                AgentResponseUpdate(
+                    contents=[],
+                    role="assistant",
+                    response_id=task.id,
+                    additional_properties={"a2a_metadata": task.metadata} if task.metadata else None,
+                    raw_representation=task,
+                )
+            ]

        if background and status.state in IN_PROGRESS_TASK_STATES:
            token = self._build_continuation_token(task)
@@ -468,6 +480,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
                    role="assistant",
                    response_id=task.id,
                    continuation_token=token,
+                    additional_properties={"a2a_metadata": task.metadata} if task.metadata else None,
                    raw_representation=task,
                )
            ]
@@ -488,6 +501,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
                        contents=contents,
                        role="assistant" if status.message.role == A2ARole.agent else "user",
                        response_id=task.id,
+                        additional_properties={"a2a_metadata": task.metadata} if task.metadata else None,
                        raw_representation=task,
                    )
                ]
@@ -502,12 +516,17 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
            contents = self._parse_contents_from_a2a(update_event.artifact.parts)
            if not contents:
                return []
+            merged_metadata = {
+                **(update_event.artifact.metadata or {}),
+                **(update_event.metadata or {}),
+            } or None
            return [
                AgentResponseUpdate(
                    contents=contents,
                    role="assistant",
                    response_id=update_event.task_id,
                    message_id=update_event.artifact.artifact_id,
+                    additional_properties={"a2a_metadata": merged_metadata} if merged_metadata else None,
                    raw_representation=update_event,
                )
            ]
@@ -523,11 +542,16 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
        if not contents:
            return []

+        merged_metadata = {
+            **(message.metadata or {}),
+            **(update_event.metadata or {}),
+        } or None
        return [
            AgentResponseUpdate(
                contents=contents,
                role="assistant" if message.role == A2ARole.agent else "user",
                response_id=update_event.task_id,
+                additional_properties={"a2a_metadata": merged_metadata} if merged_metadata else None,
                raw_representation=update_event,
            )
        ]
@@ -642,9 +666,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
                case _:
                    raise ValueError(f"Unknown content type: {content.type}")

-        # Exclude framework-internal keys (e.g. attribution) from wire metadata
-        internal_keys = {"_attribution", "context_id"}
-        metadata = {k: v for k, v in message.additional_properties.items() if k not in internal_keys} or None
+        metadata = message.additional_properties.get("a2a_metadata")

        return A2AMessage(
            role=A2ARole("user"),
@@ -718,6 +740,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
                Message(
                    role="assistant" if history_item.role == A2ARole.agent else "user",
                    contents=contents,
+                    additional_properties=history_item.metadata,
                    raw_representation=history_item,
                )
            )
@@ -730,5 +753,6 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
        return Message(
            role="assistant",
            contents=contents,
+            additional_properties=artifact.metadata,
            raw_representation=artifact,
        )
@@ -530,7 +530,7 @@ def test_prepare_message_for_a2a_forwards_context_id() -> None:
    message = Message(
        role="user",
        contents=[Content.from_text(text="Continue the task")],
-        additional_properties={"context_id": "ctx-123", "trace_id": "trace-456"},
+        additional_properties={"context_id": "ctx-123", "a2a_metadata": {"trace_id": "trace-456"}},
    )

    result = agent._prepare_message_for_a2a(message)
@@ -1385,3 +1385,210 @@ async def test_streaming_terminal_task_only_emits_unstreamed_artifacts(


 # endregion
+
+# region Metadata propagation tests
+
+
+async def test_message_metadata_propagated(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """A2AMessage.metadata should appear on response.additional_properties."""
+    msg = A2AMessage(
+        message_id="msg-meta",
+        role=A2ARole.agent,
+        parts=[Part(root=TextPart(text="hi"))],
+        metadata={"source": "server", "trace_id": "abc"},
+    )
+    mock_a2a_client.responses.append(msg)
+
+    response = await a2a_agent.run("hello")
+    assert response.additional_properties["a2a_metadata"]["source"] == "server"
+    assert response.additional_properties["a2a_metadata"]["trace_id"] == "abc"
+
+
+async def test_artifact_metadata_propagated(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """Artifact.metadata should appear on response.additional_properties."""
+    task = Task(
+        id="task-art-meta",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        artifacts=[
+            Artifact(
+                artifact_id="a1",
+                parts=[Part(root=TextPart(text="result"))],
+                metadata={"artifact_key": "artifact_value"},
+            ),
+        ],
+    )
+    mock_a2a_client.responses.append((task, None))
+
+    response = await a2a_agent.run("go")
+    assert response.additional_properties["a2a_metadata"]["artifact_key"] == "artifact_value"
+
+
+async def test_task_metadata_propagated_to_response(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """Task.metadata should appear on response.additional_properties for terminal tasks."""
+    task = Task(
+        id="task-meta",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        artifacts=[
+            Artifact(artifact_id="a1", parts=[Part(root=TextPart(text="done"))]),
+        ],
+        metadata={"task_key": "task_value"},
+    )
+    mock_a2a_client.responses.append((task, None))
+
+    response = await a2a_agent.run("go")
+    assert response.additional_properties["a2a_metadata"]["task_key"] == "task_value"
+
+
+async def test_task_artifact_update_event_metadata_merged(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """TaskArtifactUpdateEvent and Artifact metadata should both appear on the streaming update."""
+    artifact_event = TaskArtifactUpdateEvent(
+        task_id="task-ae",
+        context_id="ctx",
+        artifact=Artifact(
+            artifact_id="a1",
+            parts=[Part(root=TextPart(text="chunk"))],
+            metadata={"from_artifact": True},
+        ),
+        metadata={"from_event": True},
+    )
+    working_task = Task(
+        id="task-ae",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.working),
+    )
+    terminal_task = Task(
+        id="task-ae",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        artifacts=[
+            Artifact(artifact_id="a1", parts=[Part(root=TextPart(text="chunk"))]),
+        ],
+    )
+    terminal_event = TaskStatusUpdateEvent(
+        task_id="task-ae",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        final=True,
+    )
+    mock_a2a_client.responses.extend([
+        (working_task, artifact_event),
+        (terminal_task, terminal_event),
+    ])
+
+    stream = a2a_agent.run("hello", stream=True)
+    updates: list[AgentResponseUpdate] = []
+    async for update in stream:
+        updates.append(update)
+
+    artifact_update = updates[0]
+    assert artifact_update.additional_properties["a2a_metadata"]["from_artifact"] is True
+    assert artifact_update.additional_properties["a2a_metadata"]["from_event"] is True
+
+
+async def test_task_status_update_event_metadata_merged(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """TaskStatusUpdateEvent and its message metadata should both appear on the streaming update."""
+    status_event = TaskStatusUpdateEvent(
+        task_id="task-se",
+        context_id="ctx",
+        status=TaskStatus(
+            state=TaskState.working,
+            message=A2AMessage(
+                message_id="m1",
+                role=A2ARole.agent,
+                parts=[Part(root=TextPart(text="working..."))],
+                metadata={"msg_key": "msg_val"},
+            ),
+        ),
+        final=False,
+        metadata={"event_key": "event_val"},
+    )
+    working_task = Task(
+        id="task-se",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.working),
+    )
+    terminal_task = Task(
+        id="task-se",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        artifacts=[
+            Artifact(artifact_id="a1", parts=[Part(root=TextPart(text="done"))]),
+        ],
+    )
+    terminal_event = TaskStatusUpdateEvent(
+        task_id="task-se",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        final=True,
+    )
+    mock_a2a_client.responses.extend([
+        (working_task, status_event),
+        (terminal_task, terminal_event),
+    ])
+
+    stream = a2a_agent.run("hello", stream=True)
+    updates: list[AgentResponseUpdate] = []
+    async for update in stream:
+        updates.append(update)
+
+    status_update = updates[0]
+    assert status_update.additional_properties["a2a_metadata"]["msg_key"] == "msg_val"
+    assert status_update.additional_properties["a2a_metadata"]["event_key"] == "event_val"
+
+
+async def test_history_message_metadata_propagated(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """Metadata on a history Message should appear on response.additional_properties."""
+    task = Task(
+        id="task-hist",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.completed),
+        history=[
+            A2AMessage(
+                message_id="h1",
+                role=A2ARole.agent,
+                parts=[Part(root=TextPart(text="reply"))],
+                metadata={"history_key": "history_value"},
+            ),
+        ],
+    )
+    mock_a2a_client.responses.append((task, None))
+
+    response = await a2a_agent.run("go")
+    assert response.additional_properties["a2a_metadata"]["history_key"] == "history_value"
+
+
+async def test_continuation_token_update_carries_task_metadata(
+    a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient
+) -> None:
+    """In-progress tasks with background=True should propagate task metadata."""
+    task = Task(
+        id="task-cont",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.working),
+        metadata={"bg_key": "bg_value"},
+    )
+    mock_a2a_client.responses.append((task, None))
+
+    response = await a2a_agent.run("go", background=True)
+    assert response.continuation_token is not None
+    assert response.additional_properties["a2a_metadata"]["bg_key"] == "bg_value"
+
+
+async def test_none_metadata_leaves_additional_properties_empty(
+    a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient
+) -> None:
+    """When A2A types have no metadata, additional_properties should remain empty/default."""
+    msg = A2AMessage(
+        message_id="msg-none",
+        role=A2ARole.agent,
+        parts=[Part(root=TextPart(text="no meta"))],
+    )
+    mock_a2a_client.responses.append(msg)
+
+    response = await a2a_agent.run("hello")
+    assert not response.additional_properties
+
+
+# endregion
@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 import os
+import re
 from pathlib import Path
 from typing import Annotated, Any
 from unittest.mock import MagicMock, patch
@@ -1503,6 +1504,8 @@ async def test_anthropic_client_integration_function_calling() -> None:
@skip_if_anthropic_integration_tests_disabled
 async def test_anthropic_client_integration_hosted_tools() -> None:
    """Integration test for hosted tools."""
+    import anthropic
+
    client = AnthropicClient()

    messages = [Message(role="user", contents=["What tools do you have available?"])]
@@ -1515,10 +1518,18 @@ async def test_anthropic_client_integration_hosted_tools() -> None:
        ),
    ]

-    response = await client.get_response(
-        messages=messages,
-        options={"tools": tools, "max_tokens": 100},
-    )
+    try:
+        response = await client.get_response(
+            messages=messages,
+            options={"tools": tools, "max_tokens": 100},
+        )
+    except (
+        anthropic.BadRequestError,
+        anthropic.InternalServerError,
+        anthropic.APIConnectionError,
+        anthropic.APITimeoutError,
+    ) as e:
+        pytest.skip(f"Upstream MCP server unavailable: {e}")

    assert response is not None
    assert response.text is not None
@@ -1607,7 +1618,8 @@ async def test_anthropic_client_integration_images() -> None:

    assert response is not None
    assert response.messages[0].text is not None
-    assert "house" in response.messages[0].text.lower()
+    text = response.messages[0].text.lower()
+    assert re.search(r"\b(house|home|building|cottage|mansion|villa)\b", text)


 # Response Format Tests
@@ -63,6 +63,8 @@ agent_framework/
 - **`SessionContext`** - Context object for session-scoped data during agent runs
 - **`ContextProvider`** - Base class for context providers (RAG, memory systems)
 - **`HistoryProvider`** - Base class for conversation history storage
+- **`InMemoryHistoryProvider`** - Built-in session-state history provider for local runs
+- **`FileHistoryProvider`** - JSON Lines file-backed history provider storing one file per session with one message record per line

 ### Skills (`_skills.py`)

@@ -103,6 +103,7 @@ from ._middleware import (
 from ._sessions import (
    AgentSession,
    ContextProvider,
+    FileHistoryProvider,
    HistoryProvider,
    InMemoryHistoryProvider,
    SessionContext,
@@ -318,6 +319,7 @@ __all__ = [
    "FanInEdgeGroup",
    "FanOutEdgeGroup",
    "FileCheckpointStorage",
+    "FileHistoryProvider",
    "FinalT",
    "FinishReason",
    "FinishReasonLiteral",
@@ -47,6 +47,7 @@ class ExperimentalFeature(str, Enum):
    """

    EVALS = "EVALS"
+    FILE_HISTORY = "FILE_HISTORY"
    SKILLS = "SKILLS"


@@ -8,16 +8,24 @@ This module provides the core types for the context provider pipeline:
 - HistoryProvider: Base class for history storage providers
 - AgentSession: Lightweight session state container
 - InMemoryHistoryProvider: Built-in in-memory history provider
+- FileHistoryProvider: Built-in JSON Lines file history provider
 """

 from __future__ import annotations

+import asyncio
 import copy
+import json
+import threading
 import uuid
+import weakref
 from abc import abstractmethod
+from base64 import urlsafe_b64encode
 from collections.abc import Awaitable, Callable, Mapping, Sequence
-from typing import TYPE_CHECKING, Any, ClassVar, TypeGuard, cast
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias, TypeGuard, cast

+from ._feature_stage import ExperimentalFeature, experimental
 from ._middleware import ChatContext, ChatMiddleware
 from ._types import AgentResponse, ChatResponse, Message, ResponseStream
 from .exceptions import ChatClientInvalidResponseException
@@ -30,6 +38,17 @@ if TYPE_CHECKING:
 # Registry of known types for state deserialization
 _STATE_TYPE_REGISTRY: dict[str, type] = {}

+JsonDumps: TypeAlias = Callable[[Any], str | bytes]
+JsonLoads: TypeAlias = Callable[[str | bytes], Any]
+
+
+def _default_json_dumps(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False)
+
+
+def _default_json_loads(value: str | bytes) -> Any:
+    return json.loads(value)
+

 def _is_middleware_sequence(
    middleware: MiddlewareTypes | Sequence[MiddlewareTypes],
@@ -837,3 +856,247 @@ class InMemoryHistoryProvider(HistoryProvider):
            return
        existing = state.get("messages", [])
        state["messages"] = [*existing, *messages]
+
+
+@experimental(feature_id=ExperimentalFeature.FILE_HISTORY)
+class FileHistoryProvider(HistoryProvider):
+    """File-backed history provider that stores one JSON Lines file per session.
+
+    Each persisted message is written as a single JSON object per line. The
+    provider does not serialize full session snapshots into the file. By default
+    it uses the standard library ``json`` module, but callers can inject
+    alternative ``dumps`` and ``loads`` callables compatible with the JSON
+    Lines format.
+
+    Security posture:
+        Persisted history is stored as plaintext JSONL on the local filesystem.
+        Treat ``storage_path`` as trusted application storage, not as a secret
+        store. Encoded fallback filenames and resolved-path validation help
+        prevent path traversal via ``session_id``, but they do not encrypt file
+        contents or provide cross-process / cross-host locking. Use OS-level
+        file permissions, trusted directories, and carefully review what agent
+        or tool output is allowed to be persisted.
+    """
+
+    DEFAULT_SOURCE_ID: ClassVar[str] = "file_history"
+    DEFAULT_SESSION_FILE_STEM: ClassVar[str] = "default"
+    FILE_EXTENSION: ClassVar[str] = ".jsonl"
+    _FILE_LOCK_STRIPE_COUNT: ClassVar[int] = 64
+    _ENCODED_SESSION_PREFIX: ClassVar[str] = "~session-"
+    _FILE_WRITE_LOCKS: ClassVar[tuple[threading.Lock, ...]] = tuple(
+        threading.Lock() for _ in range(_FILE_LOCK_STRIPE_COUNT)
+    )
+    _WINDOWS_RESERVED_FILE_STEMS: ClassVar[frozenset[str]] = frozenset({
+        "CON",
+        "PRN",
+        "AUX",
+        "NUL",
+        "COM1",
+        "COM2",
+        "COM3",
+        "COM4",
+        "COM5",
+        "COM6",
+        "COM7",
+        "COM8",
+        "COM9",
+        "LPT1",
+        "LPT2",
+        "LPT3",
+        "LPT4",
+        "LPT5",
+        "LPT6",
+        "LPT7",
+        "LPT8",
+        "LPT9",
+    })
+
+    def __init__(
+        self,
+        storage_path: str | Path,
+        *,
+        source_id: str = DEFAULT_SOURCE_ID,
+        load_messages: bool = True,
+        store_inputs: bool = True,
+        store_context_messages: bool = False,
+        store_context_from: set[str] | None = None,
+        store_outputs: bool = True,
+        skip_excluded: bool = False,
+        dumps: JsonDumps | None = None,
+        loads: JsonLoads | None = None,
+    ) -> None:
+        """Initialize the file history provider.
+
+        Args:
+            storage_path: Directory path where session history files will be stored.
+
+        Keyword Args:
+            source_id: Unique identifier for this provider instance.
+            load_messages: Whether to load messages before invocation.
+            store_inputs: Whether to store input messages.
+            store_context_messages: Whether to store context from other providers.
+            store_context_from: If set, only store context from these source_ids.
+            store_outputs: Whether to store response messages.
+            skip_excluded: When True, ``get_messages`` omits messages whose
+                ``additional_properties["_excluded"]`` is truthy.
+            dumps: Callable that serializes a message payload dict to JSON text
+                or UTF-8 bytes. The returned JSON must fit on a single line.
+            loads: Callable that deserializes JSON text or bytes back to a
+                message payload dict.
+        """
+        super().__init__(
+            source_id=source_id,
+            load_messages=load_messages,
+            store_inputs=store_inputs,
+            store_context_messages=store_context_messages,
+            store_context_from=store_context_from,
+            store_outputs=store_outputs,
+        )
+        self.storage_path = Path(storage_path)
+        self.storage_path.mkdir(parents=True, exist_ok=True)
+        self._storage_root = self.storage_path.resolve()
+        self.skip_excluded = skip_excluded
+        self.dumps = dumps or _default_json_dumps
+        self.loads = loads or _default_json_loads
+        self._async_write_locks_by_loop: weakref.WeakKeyDictionary[
+            asyncio.AbstractEventLoop,
+            tuple[asyncio.Lock, ...],
+        ] = weakref.WeakKeyDictionary()
+
+    async def get_messages(
+        self,
+        session_id: str | None,
+        *,
+        state: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[Message]:
+        """Retrieve messages from the session's JSON Lines file."""
+        del state, kwargs
+        file_path = self._session_file_path(session_id)
+        async_lock = self._session_async_write_lock(file_path)
+        thread_lock = self._session_write_lock(file_path)
+
+        def _read_messages() -> list[Message]:
+            with thread_lock:
+                if not file_path.exists():
+                    return []
+
+                messages: list[Message] = []
+                with file_path.open(encoding="utf-8") as file_handle:
+                    for line_number, line in enumerate(file_handle, start=1):
+                        serialized = line.strip()
+                        if not serialized:
+                            continue
+                        try:
+                            payload = self.loads(serialized)
+                        except (TypeError, ValueError) as exc:
+                            raise ValueError(
+                                f"Failed to deserialize history line {line_number} from '{file_path}'."
+                            ) from exc
+                        if not isinstance(payload, Mapping):
+                            raise ValueError(
+                                f"History line {line_number} in '{file_path}' did not deserialize to a mapping."
+                            )
+
+                        try:
+                            message = Message.from_dict(dict(cast(Mapping[str, Any], payload)))
+                        except ValueError as exc:
+                            raise ValueError(
+                                f"History line {line_number} in '{file_path}' is not a valid Message payload."
+                            ) from exc
+                        messages.append(message)
+                return messages
+
+        async with async_lock:
+            messages = await asyncio.to_thread(_read_messages)
+        if self.skip_excluded:
+            messages = [m for m in messages if not m.additional_properties.get("_excluded", False)]
+        return messages
+
+    async def save_messages(
+        self,
+        session_id: str | None,
+        messages: Sequence[Message],
+        *,
+        state: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        """Append messages to the session's JSON Lines file."""
+        del state, kwargs
+        if not messages:
+            return
+
+        file_path = self._session_file_path(session_id)
+        async_lock = self._session_async_write_lock(file_path)
+        file_lock = self._session_write_lock(file_path)
+
+        def _append_messages() -> None:
+            with file_lock, file_path.open("a", encoding="utf-8") as file_handle:
+                for message in messages:
+                    file_handle.write(f"{self._serialize_message(message)}\n")
+
+        async with async_lock:
+            await asyncio.to_thread(_append_messages)
+
+    def _serialize_message(self, message: Message) -> str:
+        """Serialize a message payload to a single JSON Lines record."""
+        serialized = self.dumps(message.to_dict())
+        if isinstance(serialized, bytes):
+            serialized_text = serialized.decode("utf-8")
+        elif isinstance(serialized, str):
+            serialized_text = serialized
+        else:
+            raise TypeError("FileHistoryProvider.dumps must return str or bytes.")
+
+        if "\n" in serialized_text or "\r" in serialized_text:
+            raise ValueError("FileHistoryProvider.dumps must return single-line JSON for JSON Lines storage.")
+        return serialized_text
+
+    def _session_file_path(self, session_id: str | None) -> Path:
+        """Resolve the on-disk history file path for a session."""
+        file_path = (self._storage_root / f"{self._session_file_stem(session_id)}{self.FILE_EXTENSION}").resolve()
+        if not file_path.is_relative_to(self._storage_root):
+            raise ValueError(f"Session history path escaped storage directory: {session_id!r}")
+        return file_path
+
+    def _session_file_stem(self, session_id: str | None) -> str:
+        """Return the filename stem for a session."""
+        raw_session_id = session_id or self.DEFAULT_SESSION_FILE_STEM
+        if self._is_literal_session_file_stem_safe(raw_session_id):
+            return raw_session_id
+
+        encoded_session_id = urlsafe_b64encode(raw_session_id.encode("utf-8")).decode("ascii").rstrip("=")
+        return f"{self._ENCODED_SESSION_PREFIX}{encoded_session_id or self.DEFAULT_SESSION_FILE_STEM}"
+
+    def _session_async_write_lock(self, file_path: Path) -> asyncio.Lock:
+        """Return the event-loop-local async lock for a session history file."""
+        loop = asyncio.get_running_loop()
+        locks = self._async_write_locks_by_loop.get(loop)
+        if locks is None:
+            locks = tuple(asyncio.Lock() for _ in range(self._FILE_LOCK_STRIPE_COUNT))
+            self._async_write_locks_by_loop[loop] = locks
+        return locks[self._lock_index(file_path)]
+
+    @classmethod
+    def _session_write_lock(cls, file_path: Path) -> threading.Lock:
+        """Return the process-local thread lock for a session history file."""
+        return cls._FILE_WRITE_LOCKS[cls._lock_index(file_path)]
+
+    @classmethod
+    def _lock_index(cls, file_path: Path) -> int:
+        """Map a session history file to a bounded lock stripe."""
+        return hash(file_path) % cls._FILE_LOCK_STRIPE_COUNT
+
+    @classmethod
+    def _is_literal_session_file_stem_safe(cls, session_id: str) -> bool:
+        """Return whether the session ID can be used directly as a filename stem."""
+        if (
+            not session_id
+            or session_id.startswith(".")
+            or session_id.endswith((" ", "."))
+            or session_id.upper() in cls._WINDOWS_RESERVED_FILE_STEMS
+        ):
+            return False
+        if any(ord(character) < 32 for character in session_id):
+            return False
+        return all(character.isalnum() or character in "._-" for character in session_id)
@@ -59,6 +59,62 @@ class AgentExecutorResponse:
    agent_response: AgentResponse
    full_conversation: list[Message]

+    def with_text(self, text: str) -> "AgentExecutorResponse":
+        """Create a new AgentExecutorResponse with replaced text, preserving the conversation history.
+
+        Use this in custom executors that transform agent output text (e.g. upper-casing, summarising)
+        when you need downstream AgentExecutors to still have access to the full prior conversation.
+
+        Without this helper, sending a plain ``str`` from a custom executor breaks the context chain:
+        the downstream ``AgentExecutor.from_str`` handler only adds that one string to its cache and
+        loses all prior messages.  By using ``with_text`` the response type stays
+        ``AgentExecutorResponse``, so ``AgentExecutor.from_response`` is invoked instead and the full
+        conversation is preserved.
+
+        Args:
+            text: The replacement assistant message text.
+
+        Returns:
+            A new ``AgentExecutorResponse`` whose ``agent_response`` contains a single assistant
+            message with ``text``, and whose ``full_conversation`` is the prior conversation
+            (everything before the original agent turn) followed by the new assistant message.
+
+        Example:
+            .. code-block:: python
+
+                from agent_framework import AgentExecutorResponse, WorkflowContext, executor
+
+
+                @executor(
+                    id="upper_case_executor",
+                    input=AgentExecutorResponse,
+                    output=AgentExecutorResponse,
+                    workflow_output=str,
+                )
+                async def upper_case(
+                    response: AgentExecutorResponse,
+                    ctx: WorkflowContext[AgentExecutorResponse, str],
+                ) -> None:
+                    upper_text = response.agent_response.text.upper()
+                    await ctx.send_message(response.with_text(upper_text))
+                    await ctx.yield_output(upper_text)
+        """
+        new_message = Message("assistant", [text])
+        new_agent_response = AgentResponse(messages=[new_message])
+
+        # Strip off the original agent turn and replace with the new text.
+        n_agent_messages = len(self.agent_response.messages)
+        prior_messages = (
+            self.full_conversation[:-n_agent_messages] if n_agent_messages else list(self.full_conversation)
+        )
+        new_full_conversation = [*prior_messages, new_message]
+
+        return AgentExecutorResponse(
+            executor_id=self.executor_id,
+            agent_response=new_agent_response,
+            full_conversation=new_full_conversation,
+        )
+

 class AgentExecutor(Executor):
    """built-in executor that wraps an agent for handling messages.
@@ -183,7 +239,25 @@ class AgentExecutor(Executor):
        """Accept a raw user prompt string and run the agent.

        The new string input will be added to the cache which is used as the conversation context for the agent run.
+
+        Warning:
+            If the upstream executor received an ``AgentExecutorResponse`` but emits a plain
+            ``str``, this handler will be invoked instead of ``from_response``. This resets
+            the conversation context because only the new string is added to the cache and
+            all prior messages from the upstream agent are lost.
+
+            To preserve the full conversation when transforming agent output in a custom
+            executor, use ``AgentExecutorResponse.with_text(...)`` so that the message type
+            stays ``AgentExecutorResponse`` and ``from_response`` is called instead.
        """
+        if not self._cache and ctx.source_executor_ids != ["Workflow"]:
+            logger.warning(
+                "AgentExecutor '%s': from_str handler invoked with an empty cache. "
+                "If you are chaining from an AgentExecutor, the upstream custom executor may be "
+                "emitting a plain str instead of using AgentExecutorResponse.with_text(...), "
+                "which causes the full conversation context to be lost.",
+                self.id,
+            )
        self._cache.extend(normalize_messages_input(text))
        await self._run_agent_and_emit(ctx)

@@ -244,10 +244,10 @@ class FileCheckpointStorage:
    is serialized using pickle and embedded as base64-encoded strings within the JSON. This allows
    for human-readable checkpoint files while preserving the ability to store complex Python objects.

-    By default, checkpoint deserialization is restricted to a built-in set of safe
-    Python types (primitives, datetime, uuid, ...) and all ``agent_framework``
-    internal types.  To allow additional application-specific types, pass them via
-    the ``allowed_checkpoint_types`` parameter using ``"module:qualname"`` format.
+    By default, checkpoint deserialization is restricted to a built-in set of safe Python types
+    (primitives, datetime, uuid, ...), all ``agent_framework`` internal types, and OpenAI SDK types
+    (``openai.types``). To allow additional application-specific types, pass them via the
+    ``allowed_checkpoint_types`` parameter using ``"module:qualname"`` format.

    Example::

@@ -10,9 +10,9 @@ This hybrid approach provides:
 When ``allowed_types`` is supplied to :func:`decode_checkpoint_value`, a
 ``RestrictedUnpickler`` is used that limits which classes may be instantiated
 during deserialization.  The default built-in safe set covers common Python
-value types (primitives, datetime, uuid, ...) and all ``agent_framework``
-internal types.  Callers can extend the set by passing additional
-``"module:qualname"`` strings.
+value types (primitives, datetime, uuid, ...), all ``agent_framework`` internal
+types, and all ``openai.types`` types.  Callers can extend the set by passing
+additional ``"module:qualname"`` strings.
 """

 from __future__ import annotations
@@ -37,6 +37,9 @@ _JSON_NATIVE_TYPES = (str, int, float, bool, type(None))
 # Module prefix for framework-internal types that are always allowed
 _FRAMEWORK_MODULE_PREFIX = "agent_framework."

+# Module prefix for OpenAI SDK types that are always allowed
+_OPENAI_MODULE_PREFIX = "openai.types."
+
 # Built-in types considered safe for checkpoint deserialization.
 # Each entry is a ``module:qualname`` string matching the format produced by
 # :func:`_type_to_key`.  These are the classes for which pickle's
@@ -84,8 +87,9 @@ class _RestrictedUnpickler(pickle.Unpickler):  # noqa: S301
    """Unpickler that restricts which classes may be instantiated.

    Only classes whose ``module:qualname`` key appears in the combined allow
-    set (built-in safe types + framework types + caller-specified extras) are
-    permitted.  All other classes raise :class:`pickle.UnpicklingError`.
+    set (built-in safe types + framework types + OpenAI SDK types +
+    caller-specified extras) are permitted.  All other classes raise
+    :class:`pickle.UnpicklingError`.
    """

    def __init__(self, data: bytes, allowed_types: frozenset[str]) -> None:
@@ -99,6 +103,7 @@ class _RestrictedUnpickler(pickle.Unpickler):  # noqa: S301
            type_key in _BUILTIN_ALLOWED_TYPE_KEYS
            or type_key in self._allowed_types
            or module.startswith(_FRAMEWORK_MODULE_PREFIX)
+            or module.startswith(_OPENAI_MODULE_PREFIX)
        ):
            return super().find_class(module, name)  # type: ignore[no-any-return]  # nosec

@@ -268,6 +268,19 @@ def executor(
            forward references. When provided, takes precedence over introspection from the
            ``WorkflowContext`` second generic parameter (W_OutT).

+    Warning:
+        When placing a custom ``@executor`` **between** two ``AgentExecutor`` nodes, be
+        careful about the output type.  If the custom executor receives an
+        ``AgentExecutorResponse`` but emits a plain ``str``, the downstream
+        ``AgentExecutor.from_str`` handler is invoked instead of ``from_response``.
+        This resets the conversation context because only the new string is added to
+        the cache and all prior messages from the upstream agent are lost.
+
+        To preserve the full conversation, use
+        ``AgentExecutorResponse.with_text(new_text)`` to create a new response that
+        keeps the prior history, and set ``output=AgentExecutorResponse`` on the
+        decorator.
+
    Returns:
        A FunctionExecutor instance that can be wired into a Workflow.

@@ -11,11 +11,11 @@ import logging
 import types
 import uuid
 from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence
-from typing import Any, Literal, overload
+from typing import TYPE_CHECKING, Any, Literal, overload

+from .._sessions import ContextProvider
 from .._types import ResponseStream
 from ..observability import OtelAttr, capture_exception, create_workflow_span
-from ._agent import WorkflowAgent
 from ._checkpoint import CheckpointStorage
 from ._const import DEFAULT_MAX_ITERATIONS, GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
 from ._edge import (
@@ -35,6 +35,9 @@ from ._runner_context import RunnerContext
 from ._state import State
 from ._typing_utils import is_instance_of, try_coerce_to_type

+if TYPE_CHECKING:
+    from ._agent import WorkflowAgent
+
 logger = logging.getLogger(__name__)


@@ -910,7 +913,14 @@ class Workflow(DictConvertible):

        return list(output_types)

-    def as_agent(self, name: str | None = None) -> WorkflowAgent:
+    def as_agent(
+        self,
+        name: str | None = None,
+        *,
+        description: str | None = None,
+        context_providers: Sequence[ContextProvider] | None = None,
+        **kwargs: Any,
+    ) -> WorkflowAgent:
        """Create a WorkflowAgent that wraps this workflow.

        The returned agent converts standard agent inputs (strings, Message, or lists of these)
@@ -924,7 +934,10 @@ class Workflow(DictConvertible):
        initialization will fail with a ValueError.

        Args:
-            name: Optional name for the agent. If None, a default name will be generated.
+            name: Optional name for the agent. Defaults to workflow name.
+            description: Optional description of the agent. Defaults to workflow description.
+            context_providers: Optional sequence of context providers for the agent.
+            **kwargs: Additional keyword arguments passed to BaseAgent.

        Returns:
            A WorkflowAgent instance that wraps this workflow.
@@ -935,4 +948,10 @@ class Workflow(DictConvertible):
        # Import here to avoid circular imports
        from ._agent import WorkflowAgent

-        return WorkflowAgent(workflow=self, name=name)
+        return WorkflowAgent(
+            workflow=self,
+            name=name if name is not None else self.name,
+            description=description if description is not None else self.description,
+            context_providers=context_providers,
+            **kwargs,
+        )
@@ -1,7 +1,12 @@
 # Copyright (c) Microsoft. All rights reserved.

+import asyncio
 import json
+import threading
+import time
 from collections.abc import Awaitable, Callable, Sequence
+from pathlib import Path
+from typing import Any

 import pytest

@@ -10,6 +15,8 @@ from agent_framework import (
    AgentSession,
    ChatContext,
    ContextProvider,
+    ExperimentalFeature,
+    FileHistoryProvider,
    HistoryProvider,
    InMemoryHistoryProvider,
    Message,
@@ -505,3 +512,217 @@ class TestInMemoryHistoryProvider:
        ctx = SessionContext(session_id="s1", input_messages=[])
        ctx.extend_messages("custom-source", [Message(role="user", contents=["test"])])
        assert "custom-source" in ctx.context_messages
+
+
+class TestFileHistoryProvider:
+    def test_is_marked_experimental(self) -> None:
+        assert FileHistoryProvider.__feature_stage__ == "experimental"
+        assert FileHistoryProvider.__feature_id__ == ExperimentalFeature.FILE_HISTORY.value
+        assert FileHistoryProvider.__doc__ is not None
+        assert ".. warning:: Experimental" in FileHistoryProvider.__doc__
+
+    async def test_stores_and_loads_messages(self, tmp_path: Path) -> None:
+        from agent_framework import AgentResponse
+
+        provider = FileHistoryProvider(tmp_path)
+        session = AgentSession(session_id="s1")
+
+        input_message = Message(role="user", contents=["hello"])
+        response_message = Message(role="assistant", contents=["hi there"])
+        first_context = SessionContext(session_id=session.session_id, input_messages=[input_message])
+
+        await provider.before_run(  # type: ignore[arg-type]
+            agent=None,
+            session=session,
+            context=first_context,
+            state={},
+        )
+        first_context._response = AgentResponse(messages=[response_message])
+        await provider.after_run(  # type: ignore[arg-type]
+            agent=None,
+            session=session,
+            context=first_context,
+            state={},
+        )
+
+        session_file = provider._session_file_path(session.session_id)
+        assert session_file.name == "s1.jsonl"
+        assert session_file.exists()
+        raw_lines = (await asyncio.to_thread(session_file.read_text, encoding="utf-8")).splitlines()
+        assert len(raw_lines) == 2
+        payloads = [json.loads(line) for line in raw_lines]
+        assert all(payload["type"] == "message" for payload in payloads)
+        assert all("session_id" not in payload for payload in payloads)
+
+        second_context = SessionContext(
+            session_id=session.session_id, input_messages=[Message(role="user", contents=["again"])]
+        )
+        await provider.before_run(  # type: ignore[arg-type]
+            agent=None,
+            session=session,
+            context=second_context,
+            state={},
+        )
+        loaded = second_context.context_messages.get(provider.source_id, [])
+        assert len(loaded) == 2
+        assert loaded[0].text == "hello"
+        assert loaded[1].text == "hi there"
+
+    def test_creates_storage_directory(self, tmp_path: Path) -> None:
+        nested_path = tmp_path / "nested" / "history"
+        provider = FileHistoryProvider(nested_path)
+        assert provider.storage_path == nested_path
+        assert nested_path.exists()
+        assert nested_path.is_dir()
+
+    async def test_uses_encoded_filename_for_unsafe_session_id(self, tmp_path: Path) -> None:
+        provider = FileHistoryProvider(tmp_path)
+        unsafe_session_id = "../unsafe/session"
+
+        await provider.save_messages(unsafe_session_id, [Message(role="user", contents=["hello"])])
+
+        session_file = provider._session_file_path(unsafe_session_id)
+        assert session_file.parent == provider.storage_path
+        assert session_file.name.startswith("~session-")
+        assert session_file.suffix == ".jsonl"
+        assert session_file.exists()
+        jsonl_files = await asyncio.to_thread(
+            lambda: sorted(path.name for path in provider.storage_path.glob("*.jsonl"))
+        )
+        assert jsonl_files == [session_file.name]
+
+    async def test_allows_custom_serializers_returning_bytes(self, tmp_path: Path) -> None:
+        calls: list[str] = []
+
+        def dumps(payload: object) -> bytes:
+            calls.append("dumps")
+            return json.dumps(payload).encode("utf-8")
+
+        def loads(payload: str | bytes) -> object:
+            calls.append("loads")
+            if isinstance(payload, bytes):
+                payload = payload.decode("utf-8")
+            return json.loads(payload)
+
+        provider = FileHistoryProvider(tmp_path, dumps=dumps, loads=loads)
+
+        await provider.save_messages("custom-serializer", [Message(role="user", contents=["hello"])])
+        loaded = await provider.get_messages("custom-serializer")
+
+        assert calls == ["dumps", "loads"]
+        assert len(loaded) == 1
+        assert loaded[0].text == "hello"
+
+    async def test_invalid_jsonl_line_raises(self, tmp_path: Path) -> None:
+        provider = FileHistoryProvider(tmp_path)
+        await asyncio.to_thread(provider._session_file_path("broken").write_text, "{not-json}\n", encoding="utf-8")
+
+        with pytest.raises(ValueError, match="Failed to deserialize history line 1"):
+            await provider.get_messages("broken")
+
+    async def test_missing_session_file_returns_empty_messages(self, tmp_path: Path) -> None:
+        provider = FileHistoryProvider(tmp_path)
+
+        loaded = await provider.get_messages("missing")
+
+        assert loaded == []
+
+    async def test_none_session_id_uses_default_jsonl_file(self, tmp_path: Path) -> None:
+        provider = FileHistoryProvider(tmp_path)
+
+        await provider.save_messages(None, [Message(role="user", contents=["hello"])])
+
+        session_file = provider._session_file_path(None)
+        assert session_file.name == "default.jsonl"
+        loaded = await provider.get_messages(None)
+        assert [message.text for message in loaded] == ["hello"]
+
+    async def test_non_mapping_jsonl_line_raises(self, tmp_path: Path) -> None:
+        provider = FileHistoryProvider(tmp_path)
+        await asyncio.to_thread(provider._session_file_path("non-mapping").write_text, "[1, 2, 3]\n", encoding="utf-8")
+
+        with pytest.raises(ValueError, match="did not deserialize to a mapping"):
+            await provider.get_messages("non-mapping")
+
+    async def test_skip_excluded_omits_excluded_messages(self, tmp_path: Path) -> None:
+        provider = FileHistoryProvider(tmp_path, skip_excluded=True)
+
+        await provider.save_messages(
+            "skip-excluded",
+            [
+                Message(role="user", contents=["keep"]),
+                Message(role="assistant", contents=["skip"], additional_properties={"_excluded": True}),
+            ],
+        )
+
+        loaded = await provider.get_messages("skip-excluded")
+
+        assert [message.text for message in loaded] == ["keep"]
+
+    async def test_serializer_must_return_single_line_json(self, tmp_path: Path) -> None:
+        def dumps(payload: object) -> str:
+            return json.dumps(payload, indent=2)
+
+        provider = FileHistoryProvider(tmp_path, dumps=dumps)
+
+        with pytest.raises(ValueError, match="single-line JSON"):
+            await provider.save_messages("pretty-json", [Message(role="user", contents=["hello"])])
+
+    async def test_concurrent_writes_for_same_session_are_locked(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        provider = FileHistoryProvider(tmp_path)
+        session_id = "shared-session"
+        file_path = provider._session_file_path(session_id)
+        real_open = Path.open
+        write_started = threading.Event()
+        active_writes = 0
+        overlap_detected = False
+
+        class _TrackingFile:
+            def __init__(self, wrapped: Any) -> None:
+                self._wrapped = wrapped
+
+            def __enter__(self) -> "_TrackingFile":
+                self._wrapped.__enter__()
+                return self
+
+            def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+                self._wrapped.__exit__(exc_type, exc_val, exc_tb)
+
+            def write(self, data: str) -> int:
+                nonlocal active_writes, overlap_detected
+                write_started.set()
+                active_writes += 1
+                overlap_detected = overlap_detected or active_writes > 1
+                try:
+                    time.sleep(0.05)
+                    return int(self._wrapped.write(data))
+                finally:
+                    active_writes -= 1
+
+            def __getattr__(self, name: str) -> Any:
+                return getattr(self._wrapped, name)
+
+        def tracked_open(path: Path, *args: Any, **kwargs: Any) -> Any:
+            handle = real_open(path, *args, **kwargs)
+            if path == file_path and args and args[0] == "a":
+                return _TrackingFile(handle)
+            return handle
+
+        monkeypatch.setattr(Path, "open", tracked_open)
+
+        first_save = asyncio.create_task(provider.save_messages(session_id, [Message(role="user", contents=["first"])]))
+        started = await asyncio.to_thread(write_started.wait, 1.0)
+        assert started
+
+        second_save = asyncio.create_task(
+            provider.save_messages(session_id, [Message(role="assistant", contents=["second"])])
+        )
+        await asyncio.gather(first_save, second_save)
+
+        assert not overlap_detected
+        loaded = await provider.get_messages(session_id)
+        assert [message.text for message in loaded] == ["first", "second"]
@@ -216,3 +216,50 @@ def test_restricted_unpickler_raises_pickle_error():
    unpickler = _RestrictedUnpickler(pickled, frozenset())
    with pytest.raises(pickle.UnpicklingError, match="deserialization blocked"):
        unpickler.load()
+
+
+def test_restricted_decode_allows_openai_types():
+    """OpenAI SDK types are always allowed during restricted deserialization."""
+    from openai.types.chat.chat_completion import ChatCompletion, Choice
+    from openai.types.chat.chat_completion_message import ChatCompletionMessage
+    from openai.types.completion_usage import CompletionUsage
+
+    completion = ChatCompletion(
+        id="chatcmpl-test",
+        choices=[
+            Choice(
+                finish_reason="stop",
+                index=0,
+                message=ChatCompletionMessage(role="assistant", content="hello"),
+            )
+        ],
+        created=1700000000,
+        model="gpt-4",
+        object="chat.completion",
+        usage=CompletionUsage(completion_tokens=1, prompt_tokens=1, total_tokens=2),
+    )
+    encoded = encode_checkpoint_value(completion)
+    decoded = decode_checkpoint_value(encoded, allowed_types=frozenset())
+
+    assert isinstance(decoded, ChatCompletion)
+    assert decoded.id == "chatcmpl-test"
+    assert decoded.choices[0].message.content == "hello"
+
+
+def test_restricted_decode_allows_openai_response_types():
+    """OpenAI Responses API types are always allowed during restricted deserialization."""
+    from openai.types.responses.response_usage import InputTokensDetails, OutputTokensDetails, ResponseUsage
+
+    usage = ResponseUsage(
+        input_tokens=10,
+        output_tokens=20,
+        total_tokens=30,
+        input_tokens_details=InputTokensDetails(cached_tokens=0),
+        output_tokens_details=OutputTokensDetails(reasoning_tokens=0),
+    )
+    encoded = encode_checkpoint_value(usage)
+    decoded = decode_checkpoint_value(encoded, allowed_types=frozenset())
+
+    assert isinstance(decoded, ResponseUsage)
+    assert decoded.input_tokens == 10
+    assert decoded.output_tokens == 20
@@ -23,6 +23,7 @@ from agent_framework import (
    WorkflowBuilder,
    WorkflowContext,
    WorkflowRunState,
+    executor,
    handler,
 )
 from agent_framework.orchestrations import SequentialBuilder
@@ -478,3 +479,90 @@ async def test_from_response_preserves_service_session_id() -> None:
    assert result.get_outputs() is not None

    assert spy_agent._captured_service_session_id == "resp_PREVIOUS_RUN"  # pyright: ignore[reportPrivateUsage]
+
+
+@executor(
+    id="upper_case_executor",
+    input=AgentExecutorResponse,
+    output=AgentExecutorResponse,
+    workflow_output=str,
+)
+async def _upper_case_executor(
+    response: AgentExecutorResponse,
+    ctx: WorkflowContext[AgentExecutorResponse, str],
+) -> None:
+    upper_text = response.agent_response.text.upper()
+    await ctx.send_message(response.with_text(upper_text))
+    await ctx.yield_output(upper_text)
+
+
+async def test_with_text_preserves_full_conversation_through_custom_executor() -> None:
+    """Custom executor using with_text must preserve the full conversation chain."""
+    # Mirrors the reproduction from issue #5246:
+    # agent1 ("User likes sky red") -> agent2 ("User likes sky blue") -> upper_case -> agent3 ("User likes sky green")
+    agent1 = AgentExecutor(
+        _SimpleAgent(id="agent1", name="ContextAgent1", reply_text="User likes sky red"), id="agent1"
+    )
+    agent2 = AgentExecutor(
+        _SimpleAgent(id="agent2", name="ContextAgent2", reply_text="User likes sky blue"), id="agent2"
+    )
+    agent3 = AgentExecutor(
+        _SimpleAgent(id="agent3", name="ContextAgent3", reply_text="User likes sky green"), id="agent3"
+    )
+    capturer = _CaptureFullConversation(id="capture")
+
+    wf = (
+        WorkflowBuilder(start_executor=agent1, output_executors=[capturer])
+        .add_chain([agent1, agent2, _upper_case_executor, agent3, capturer])
+        .build()
+    )
+
+    result = await wf.run("")
+    payload = next(o for o in result.get_outputs() if isinstance(o, dict))
+
+    # The final agent must see the full conversation: user, agent1, UPPER(agent2), agent3
+    assert payload["roles"] == ["user", "assistant", "assistant", "assistant"]
+    assert payload["texts"][1] == "User likes sky red"
+    assert payload["texts"][2] == "USER LIKES SKY BLUE"
+    assert payload["texts"][3] == "User likes sky green"
+
+
+async def test_with_text_does_not_mutate_original() -> None:
+    """with_text returns a new instance; the original must be unmodified."""
+    original = AgentExecutorResponse(
+        executor_id="test_exec",
+        agent_response=AgentResponse(messages=[Message("assistant", ["original reply"])]),
+        full_conversation=[Message("user", ["prompt"]), Message("assistant", ["original reply"])],
+    )
+
+    new = original.with_text("transformed reply")
+
+    assert new is not original
+    assert new.agent_response.text == "transformed reply"
+    assert new.full_conversation[-1].text == "transformed reply"
+    assert new.full_conversation[-1].role == "assistant"
+    # Original unchanged
+    assert original.agent_response.text == "original reply"
+    assert original.full_conversation[-1].text == "original reply"
+
+
+async def test_with_text_strips_multi_message_agent_turn() -> None:
+    """When the agent turn has multiple messages (tool calls), with_text strips all of them."""
+    tool_call = Message("assistant", ["<tool_call>"])
+    tool_result = Message("tool", ["<result>"])
+    final_reply = Message("assistant", ["actual answer"])
+    user_msg = Message("user", ["question"])
+
+    original = AgentExecutorResponse(
+        executor_id="exec",
+        agent_response=AgentResponse(messages=[tool_call, tool_result, final_reply]),
+        full_conversation=[user_msg, tool_call, tool_result, final_reply],
+    )
+
+    new = original.with_text("summarised answer")
+
+    # Only the pre-agent-turn messages should remain, plus the replacement
+    assert len(new.full_conversation) == 2
+    assert new.full_conversation[0].text == "question"
+    assert new.full_conversation[1].text == "summarised answer"
+    assert new.agent_response.text == "summarised answer"
@@ -313,6 +313,37 @@ class TestWorkflowAgent:
        assert isinstance(agent_no_name, WorkflowAgent)
        assert agent_no_name.workflow is workflow

+    def test_workflow_as_agent_with_description_and_context_providers(self) -> None:
+        """Test that Workflow.as_agent() forwards description and context_providers."""
+        executor = SimpleExecutor(id="executor1", response_text="Response")
+        workflow = WorkflowBuilder(start_executor=executor).build()
+
+        history_provider = InMemoryHistoryProvider()
+        agent = workflow.as_agent(
+            name="MyAgent",
+            description="A test agent",
+            context_providers=[history_provider],
+        )
+
+        assert isinstance(agent, WorkflowAgent)
+        assert agent.name == "MyAgent"
+        assert agent.description == "A test agent"
+        assert history_provider in agent.context_providers
+
+    def test_workflow_as_agent_defaults_name_and_description_from_workflow(self) -> None:
+        """Test that as_agent() defaults name and description to the workflow's own values."""
+        executor = SimpleExecutor(id="executor1", response_text="Response")
+        workflow = WorkflowBuilder(
+            start_executor=executor,
+            name="my-workflow",
+            description="Workflow description",
+        ).build()
+
+        agent = workflow.as_agent()
+
+        assert agent.name == "my-workflow"
+        assert agent.description == "Workflow description"
+
    def test_workflow_as_agent_cannot_handle_agent_inputs(self) -> None:
        """Test that Workflow.as_agent() raises an error if the start executor cannot handle agent inputs."""

@@ -31,7 +31,7 @@ from agent_framework.exceptions import AgentException
 try:
    from copilot import CopilotClient, CopilotSession, SubprocessConfig
    from copilot.generated.session_events import PermissionRequest, SessionEvent, SessionEventType
-    from copilot.session import MCPServerConfig, PermissionRequestResult, SystemMessageConfig
+    from copilot.session import MCPServerConfig, PermissionRequestResult, ProviderConfig, SystemMessageConfig
    from copilot.tools import Tool as CopilotTool
    from copilot.tools import ToolInvocation, ToolResult
 except ImportError as _copilot_import_error:
@@ -120,6 +120,12 @@ class GitHubCopilotOptions(TypedDict, total=False):
    Supports both local (stdio) and remote (HTTP/SSE) servers.
    """

+    provider: ProviderConfig
+    """Custom API provider configuration for BYOK (Bring Your Own Key) scenarios.
+    Allows routing requests through your own OpenAI, Azure, or Anthropic endpoint
+    instead of the default GitHub Copilot backend.
+    """
+

 OptionsT = TypeVar(
    "OptionsT",
@@ -232,6 +238,7 @@ class GitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
        log_level = opts.pop("log_level", None)
        on_permission_request: PermissionHandlerType | None = opts.pop("on_permission_request", None)
        mcp_servers: dict[str, MCPServerConfig] | None = opts.pop("mcp_servers", None)
+        provider: ProviderConfig | None = opts.pop("provider", None)

        self._settings = load_settings(
            GitHubCopilotSettings,
@@ -247,6 +254,7 @@ class GitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
        self._tools = normalize_tools(tools)
        self._permission_handler = on_permission_request
        self._mcp_servers = mcp_servers
+        self._provider = provider
        self._default_options = opts
        self._started = False

@@ -730,6 +738,7 @@ class GitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
            opts.get("on_permission_request") or self._permission_handler or _deny_all_permissions
        )
        mcp_servers = opts.get("mcp_servers") or self._mcp_servers or None
+        provider = opts.get("provider") or self._provider or None
        tools = self._prepare_tools(self._tools) if self._tools else None

        return await self._client.create_session(
@@ -739,6 +748,7 @@ class GitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
            system_message=system_message or None,
            tools=tools or None,
            mcp_servers=mcp_servers or None,
+            provider=provider or None,
        )

    async def _resume_session(self, session_id: str, streaming: bool) -> CopilotSession:
@@ -755,4 +765,5 @@ class GitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
            streaming=streaming,
            tools=tools or None,
            mcp_servers=self._mcp_servers or None,
+            provider=self._provider or None,
        )
@@ -861,6 +861,7 @@ class TestGitHubCopilotAgentSessionManagement:
            streaming=unittest.mock.ANY,
            tools=unittest.mock.ANY,
            mcp_servers=unittest.mock.ANY,
+            provider=unittest.mock.ANY,
        )

    async def test_session_config_includes_model(
@@ -1084,6 +1085,198 @@ class TestGitHubCopilotAgentMCPServers:
        assert config["mcp_servers"] is None


+class TestGitHubCopilotAgentProvider:
+    """Test cases for provider configuration (BYOK / Managed Identity)."""
+
+    async def test_provider_passed_to_create_session(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that provider config is passed through to create_session."""
+        from copilot.session import ProviderConfig
+
+        provider: ProviderConfig = {
+            "type": "azure",
+            "base_url": "https://my-resource.openai.azure.com",
+            "bearer_token": "test-token",
+        }
+
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            client=mock_client,
+            default_options={"provider": provider},
+        )
+        await agent.start()
+
+        await agent._get_or_create_session(AgentSession())  # type: ignore
+
+        call_args = mock_client.create_session.call_args
+        config = call_args.kwargs
+        assert config["provider"]["type"] == "azure"
+        assert config["provider"]["base_url"] == "https://my-resource.openai.azure.com"
+        assert config["provider"]["bearer_token"] == "test-token"
+
+    async def test_provider_passed_to_resume_session(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that provider config is passed through to resume_session."""
+        from copilot.session import ProviderConfig
+
+        provider: ProviderConfig = {
+            "type": "azure",
+            "base_url": "https://my-resource.openai.azure.com",
+            "bearer_token": "test-token",
+        }
+
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            client=mock_client,
+            default_options={"provider": provider},
+        )
+        await agent.start()
+
+        session = AgentSession()
+        session.service_session_id = "existing-session-id"
+
+        await agent._get_or_create_session(session)  # type: ignore
+
+        mock_client.resume_session.assert_called_once()
+        call_args = mock_client.resume_session.call_args
+        config = call_args.kwargs
+        assert config["provider"]["type"] == "azure"
+
+    async def test_session_config_excludes_provider_when_not_set(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that provider is None in session config when not set."""
+        agent = GitHubCopilotAgent(client=mock_client)
+        await agent.start()
+
+        await agent._get_or_create_session(AgentSession())  # type: ignore
+
+        call_args = mock_client.create_session.call_args
+        config = call_args.kwargs
+        assert config["provider"] is None
+
+    async def test_resume_session_excludes_provider_when_not_set(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that provider is None in resume session config when not set."""
+        agent = GitHubCopilotAgent(client=mock_client)
+        await agent.start()
+
+        session = AgentSession()
+        session.service_session_id = "existing-session-id"
+
+        await agent._get_or_create_session(session)  # type: ignore
+
+        call_args = mock_client.resume_session.call_args
+        config = call_args.kwargs
+        assert config["provider"] is None
+
+    async def test_runtime_provider_takes_precedence(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that runtime provider options override default_options provider."""
+        from copilot.session import ProviderConfig
+
+        default_provider: ProviderConfig = {
+            "type": "azure",
+            "base_url": "https://default.openai.azure.com",
+            "bearer_token": "default-token",
+        }
+        runtime_provider: ProviderConfig = {
+            "type": "openai",
+            "base_url": "https://runtime.openai.com",
+            "api_key": "runtime-key",
+        }
+
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            client=mock_client,
+            default_options={"provider": default_provider},
+        )
+        await agent.start()
+
+        await agent._get_or_create_session(  # type: ignore
+            AgentSession(),
+            runtime_options={"provider": runtime_provider},
+        )
+
+        call_args = mock_client.create_session.call_args
+        config = call_args.kwargs
+        assert config["provider"]["type"] == "openai"
+        assert config["provider"]["base_url"] == "https://runtime.openai.com"
+
+    async def test_provider_not_leaked_into_default_options(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that provider is popped from opts and not left in _default_options."""
+        from copilot.session import ProviderConfig
+
+        provider: ProviderConfig = {
+            "type": "azure",
+            "base_url": "https://my-resource.openai.azure.com",
+            "bearer_token": "test-token",
+        }
+
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            client=mock_client,
+            default_options={"provider": provider, "model": "gpt-5"},
+        )
+
+        assert "provider" not in agent._default_options
+        assert agent._provider is not None
+        assert agent._provider["type"] == "azure"
+
+    async def test_provider_coexists_with_other_options(
+        self,
+        mock_client: MagicMock,
+    ) -> None:
+        """Test that provider works alongside model, tools, and mcp_servers."""
+        from copilot.session import MCPServerConfig, ProviderConfig
+
+        provider: ProviderConfig = {
+            "type": "azure",
+            "base_url": "https://my-resource.openai.azure.com",
+            "bearer_token": "test-token",
+        }
+        mcp_servers: dict[str, MCPServerConfig] = {
+            "test-server": {
+                "type": "stdio",
+                "command": "echo",
+                "args": ["hello"],
+                "tools": ["*"],
+            },
+        }
+
+        def my_tool(arg: str) -> str:
+            """A test tool."""
+            return arg
+
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            client=mock_client,
+            tools=[my_tool],
+            default_options={
+                "model": "gpt-5",
+                "provider": provider,
+                "mcp_servers": mcp_servers,
+            },
+        )
+        await agent.start()
+
+        await agent._get_or_create_session(AgentSession())  # type: ignore
+
+        call_args = mock_client.create_session.call_args
+        config = call_args.kwargs
+        assert config["provider"]["type"] == "azure"
+        assert config["model"] == "gpt-5"
+        assert config["mcp_servers"] is not None
+        assert config["tools"] is not None
+
+
 class TestGitHubCopilotAgentToolConversion:
    """Test cases for tool conversion."""

@@ -2474,6 +2474,29 @@ class RawOpenAIChatClient(  # type: ignore[misc]
                                raw_representation=event,
                            )
                        )
+                elif ann_type == "url_citation":
+                    ann_url = _get_ann_value("url")
+                    if ann_url:
+                        ann_start = _get_ann_value("start_index")
+                        ann_end = _get_ann_value("end_index")
+                        annotation_obj = Annotation(
+                            type="citation",
+                            title=_get_ann_value("title") or "",
+                            url=str(ann_url),
+                            additional_properties={"annotation_index": event.annotation_index},
+                            raw_representation=annotation,
+                        )
+                        if ann_start is not None and ann_end is not None:
+                            annotation_obj["annotated_regions"] = [
+                                TextSpanRegion(
+                                    type="text_span",
+                                    start_index=ann_start,
+                                    end_index=ann_end,
+                                )
+                            ]
+                        contents.append(
+                            Content.from_text(text="", annotations=[annotation_obj], raw_representation=event)
+                        )
                else:
                    logger.debug("Unparsed annotation type in streaming: %s", ann_type)
            case "response.output_item.done":
@@ -2570,8 +2570,65 @@ def test_streaming_annotation_added_with_container_file_citation() -> None:
    assert content.additional_properties.get("end_index") == 50


-def test_streaming_annotation_added_with_unknown_type() -> None:
-    """Test streaming annotation added event with unknown type is ignored."""
+def test_streaming_annotation_added_with_url_citation() -> None:
+    """Test streaming annotation added event with url_citation type produces citation annotation."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    chat_options = ChatOptions()
+    function_call_ids: dict[int, tuple[str, str]] = {}
+
+    mock_event = MagicMock()
+    mock_event.type = "response.output_text.annotation.added"
+    mock_event.annotation_index = 0
+    mock_event.annotation = {
+        "type": "url_citation",
+        "url": "https://example.sharepoint.com/sites/my-site/doc.pdf",
+        "title": "doc.pdf",
+        "start_index": 100,
+        "end_index": 112,
+    }
+
+    response = client._parse_chunk_from_openai(mock_event, chat_options, function_call_ids)
+
+    assert len(response.contents) == 1
+    content = response.contents[0]
+    assert content.type == "text"
+    assert content.annotations is not None
+    assert len(content.annotations) == 1
+    annotation = content.annotations[0]
+    assert annotation["type"] == "citation"
+    assert annotation["title"] == "doc.pdf"
+    assert annotation["url"] == "https://example.sharepoint.com/sites/my-site/doc.pdf"
+    assert annotation["additional_properties"]["annotation_index"] == 0
+    assert annotation["raw_representation"] == mock_event.annotation
+    assert annotation["annotated_regions"] is not None
+    assert len(annotation["annotated_regions"]) == 1
+    region = annotation["annotated_regions"][0]
+    assert region["type"] == "text_span"
+    assert region["start_index"] == 100
+    assert region["end_index"] == 112
+
+
+def test_streaming_annotation_added_with_url_citation_no_url() -> None:
+    """Test streaming annotation added event with url_citation but missing url is ignored."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    chat_options = ChatOptions()
+    function_call_ids: dict[int, tuple[str, str]] = {}
+
+    mock_event = MagicMock()
+    mock_event.type = "response.output_text.annotation.added"
+    mock_event.annotation_index = 0
+    mock_event.annotation = {
+        "type": "url_citation",
+        "title": "doc.pdf",
+    }
+
+    response = client._parse_chunk_from_openai(mock_event, chat_options, function_call_ids)
+
+    assert len(response.contents) == 0
+
+
+def test_streaming_annotation_added_with_url_citation_no_indices() -> None:
+    """Test streaming annotation with url_citation that has url but no start_index/end_index."""
    client = OpenAIChatClient(model="test-model", api_key="test-key")
    chat_options = ChatOptions()
    function_call_ids: dict[int, tuple[str, str]] = {}
@@ -2582,11 +2639,36 @@ def test_streaming_annotation_added_with_unknown_type() -> None:
    mock_event.annotation = {
        "type": "url_citation",
        "url": "https://example.com",
+        "title": "Example",
+    }
+
+    response = client._parse_chunk_from_openai(mock_event, chat_options, function_call_ids)
+
+    assert len(response.contents) == 1
+    annotation = response.contents[0].annotations[0]
+    assert annotation["type"] == "citation"
+    assert annotation["title"] == "Example"
+    assert annotation["url"] == "https://example.com"
+    assert annotation["additional_properties"]["annotation_index"] == 0
+    assert "annotated_regions" not in annotation
+
+
+def test_streaming_annotation_added_with_unknown_type() -> None:
+    """Test streaming annotation added event with unknown type is ignored."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    chat_options = ChatOptions()
+    function_call_ids: dict[int, tuple[str, str]] = {}
+
+    mock_event = MagicMock()
+    mock_event.type = "response.output_text.annotation.added"
+    mock_event.annotation_index = 0
+    mock_event.annotation = {
+        "type": "some_future_annotation_type",
+        "data": "test",
    }

    response = client._parse_chunk_from_openai(mock_event, chat_options, function_call_ids)

-    # url_citation should not produce HostedFileContent
    assert len(response.contents) == 0


@@ -8,6 +8,8 @@ These samples demonstrate different approaches to managing conversation history
 |------|-------------|
 | [`suspend_resume_session.py`](suspend_resume_session.py) | Suspend and resume conversation sessions, comparing service-managed sessions (Azure AI Foundry) with in-memory sessions (OpenAI). |
 | [`custom_history_provider.py`](custom_history_provider.py) | Implement a custom history provider by extending `HistoryProvider`, enabling conversation persistence in your preferred storage backend. |
+| [`file_history_provider.py`](file_history_provider.py) | Use the experimental `FileHistoryProvider` with `FoundryChatClient` and a function tool so the local JSON Lines file shows the full tool-calling loop. |
+| [`file_history_provider_conversation_persistence.py`](file_history_provider_conversation_persistence.py) | Persist a tool-driven weather conversation with `FileHistoryProvider`, inspect the stored JSONL records, and continue with another city. |
 | [`cosmos_history_provider.py`](cosmos_history_provider.py) | Use Azure Cosmos DB as a history provider for durable conversation storage with `CosmosHistoryProvider`. |
 | [`cosmos_history_provider_conversation_persistence.py`](cosmos_history_provider_conversation_persistence.py) | Persist and resume conversations across application restarts using `CosmosHistoryProvider` — serialize session state, restore it, and continue with full Cosmos DB history. |
 | [`cosmos_history_provider_messages.py`](cosmos_history_provider_messages.py) | Direct message history operations — retrieve stored messages as a transcript, clear session history, and verify data deletion. |
@@ -25,6 +27,20 @@ These samples demonstrate different approaches to managing conversation history
 **For `custom_history_provider.py`:**
 - `OPENAI_API_KEY`: Your OpenAI API key

+**For `file_history_provider.py`:**
+- `FOUNDRY_PROJECT_ENDPOINT`: Your Azure AI Foundry project endpoint
+- `FOUNDRY_MODEL`: The Foundry model deployment name
+- Azure CLI authentication (`az login`)
+- The sample writes plaintext JSONL conversation logs to disk; use a trusted
+  local directory and avoid treating the history files as secure secret storage
+
+**For `file_history_provider_conversation_persistence.py`:**
+- `FOUNDRY_PROJECT_ENDPOINT`: Your Azure AI Foundry project endpoint
+- `FOUNDRY_MODEL`: The Foundry model deployment name
+- Azure CLI authentication (`az login`)
+- The sample writes plaintext JSONL conversation logs to disk; use a trusted
+  local directory and avoid treating the history files as secure secret storage
+
 **For Cosmos DB samples (`cosmos_history_provider*.py`):**
 - `FOUNDRY_PROJECT_ENDPOINT`: Your Azure AI Foundry project endpoint
 - `FOUNDRY_MODEL`: The Foundry model deployment name
@@ -0,0 +1,157 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import asyncio
+import os
+import tempfile
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Annotated
+
+# Uncomment this filter to suppress the experimental FileHistoryProvider warning
+# before running the sample.
+# import warnings  # isort: skip
+# warnings.filterwarnings("ignore", message=r"\[FILE_HISTORY\].*", category=FutureWarning)
+from agent_framework import Agent, FileHistoryProvider, tool
+from agent_framework.foundry import FoundryChatClient
+from azure.identity import AzureCliCredential
+from dotenv import load_dotenv
+from pydantic import Field
+
+try:
+    import orjson
+except ImportError:
+    orjson = None
+
+
+# Load environment variables from .env file.
+load_dotenv()
+
+"""
+File History Provider
+
+This sample demonstrates how to use the experimental `FileHistoryProvider` with
+`FoundryChatClient` and a function tool so the persisted JSON Lines file shows
+the tool-calling loop as well as the regular chat turns.
+
+Environment variables:
+    FOUNDRY_PROJECT_ENDPOINT: Azure AI Foundry project endpoint.
+    FOUNDRY_MODEL: Foundry model deployment name.
+
+Key components:
+- `FileHistoryProvider`: Stores one message JSON object per line in a local
+  `.jsonl` file for each session.
+- `lookup_weather`: A function tool that makes the persisted file show the
+  assistant function call and tool result lines.
+- `json.dumps(..., indent=2)`: Pretty-prints selected records in the sample
+  output while keeping the on-disk JSONL file compact and valid.
+- `USE_TEMP_DIRECTORY`: Toggle between a temporary directory and a persistent
+  `sessions/` folder next to this sample file.
+
+Security posture:
+- The history files are plaintext JSONL on disk, so use a trusted storage
+  directory and treat the files as conversation logs, not as secure secret
+  storage.
+- Path safety checks protect the filename derived from the session id, but they
+  do not redact message contents or encrypt the file.
+"""
+
+USE_TEMP_DIRECTORY = False
+"""When True, store JSONL files in a temporary directory for this run only."""
+
+LOCAL_SESSIONS_DIRECTORY_NAME = "sessions"
+"""Folder name used when persisting history next to this sample file."""
+
+
+@tool(approval_mode="never_require")
+def lookup_weather(
+    location: Annotated[str, Field(description="The city to look up weather for.")],
+) -> str:
+    """Return a deterministic weather report for a city."""
+    weather_reports = {
+        "Seattle": "Seattle is rainy with a high of 13C.",
+        "Amsterdam": "Amsterdam is cloudy with a high of 16C.",
+    }
+    return weather_reports.get(location, f"{location} is sunny with a high of 20C.")
+
+
+@contextmanager
+def _resolve_storage_directory() -> Iterator[Path]:
+    """Yield the configured storage directory for the sample run."""
+    if USE_TEMP_DIRECTORY:
+        with tempfile.TemporaryDirectory(prefix="af-file-history-") as temp_directory:
+            yield Path(temp_directory)
+        return
+
+    storage_directory = Path(__file__).resolve().parent / LOCAL_SESSIONS_DIRECTORY_NAME
+    storage_directory.mkdir(parents=True, exist_ok=True)
+    yield storage_directory
+
+
+async def main() -> None:
+    """Run the file history provider sample."""
+
+    with _resolve_storage_directory() as storage_directory:
+        print(f"Using temporary directory: {USE_TEMP_DIRECTORY}")
+        print(f"Storage directory: {storage_directory}\n")
+
+        # 2. Create the agent with a tool so the JSONL file includes tool-calling messages.
+        agent = Agent(
+            client=FoundryChatClient(
+                project_endpoint=os.getenv("FOUNDRY_PROJECT_ENDPOINT"),
+                model=os.getenv("FOUNDRY_MODEL"),
+                credential=AzureCliCredential(),
+            ),
+            name="FileHistoryAgent",
+            instructions=(
+                "You are a helpful assistant, use the lookup_weather tool for weather questions and "
+                "answer with the tool result in one sentence."
+            ),
+            tools=[lookup_weather],
+            # if orjson is available, use it for faster JSON serialization in the FileHistoryProvider,
+            # otherwise fall back to the default json module.
+            context_providers=[
+                FileHistoryProvider(
+                    storage_directory,
+                    dumps=orjson.dumps if orjson else None,
+                    loads=orjson.loads if orjson else None,
+                )
+            ],
+            default_options={"store": False},
+        )
+
+        # 3. Let Agent create the default UUID session id for this conversation.
+        session = agent.create_session()
+
+        # 4. Ask a question that triggers the weather tool.
+        print("=== Run with tool calling ===")
+        query = "Use the lookup_weather tool for Seattle and tell me the weather."
+        response = await agent.run(query, session=session)
+        print(f"User:      {query}")
+        print(f"Assistant: {response.text}\n")
+
+        # 5. Ask a follow-up question that triggers the weather tool as well
+        print("=== Follow-up question ===")
+        query = "And what about Amsterdam?"
+        response = await agent.run(query, session=session)
+        print(f"User:      {query}")
+        print(f"Assistant: {response.text}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+"""
+Sample output:
+Using temporary directory: False
+Storage directory: /path/to/samples/02-agents/conversations/sessions
+
+=== Run with tool calling ===
+User:      Use the lookup_weather tool for Seattle and tell me the weather.
+Assistant: <model response varies>
+=== Follow-up question ===
+User:      And what about Amsterdam?
+Assistant: <model response varies>
+"""
@@ -0,0 +1,185 @@
+# Copyright (c) Microsoft. All rights reserved.
+# ruff: noqa: T201
+
+from __future__ import annotations
+
+import asyncio
+import json
+import tempfile
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Annotated
+
+# Uncomment this filter to suppress the experimental FileHistoryProvider warning
+# before running the sample.
+# import warnings  # isort: skip
+# warnings.filterwarnings("ignore", message=r"\[FILE_HISTORY\].*", category=FutureWarning)
+from agent_framework import Agent, FileHistoryProvider, tool
+from agent_framework.foundry import FoundryChatClient
+from azure.identity.aio import AzureCliCredential
+from dotenv import load_dotenv
+from pydantic import Field
+
+try:
+    import orjson
+except ImportError:
+    orjson = None
+
+
+load_dotenv()
+
+"""
+File History Provider Conversation Persistence
+
+This sample demonstrates persisting a tool-driven conversation with the
+experimental `FileHistoryProvider`, reading the stored JSONL file back from
+disk, and then continuing the same conversation with another city.
+
+Environment variables:
+    FOUNDRY_PROJECT_ENDPOINT: Azure AI Foundry project endpoint.
+    FOUNDRY_MODEL: Foundry model deployment name.
+
+Key components:
+- `FileHistoryProvider`: Stores one message JSON object per line in a local
+  `.jsonl` file for each session.
+- `get_weather`: A function tool that makes the persisted file show the
+  assistant function call and tool result records.
+- `json.dumps(..., indent=2)`: Pretty-prints a few persisted JSONL records
+  while keeping the on-disk file compact and valid.
+- `load_dotenv()`: Loads `.env` values up front so the sample can stay focused
+  on history persistence instead of manual environment variable plumbing.
+- Optional `orjson`: Uses `orjson.dumps` / `orjson.loads` automatically when
+  available, otherwise falls back to the standard library `json` module.
+
+Security posture:
+- The history file is plaintext JSONL on disk, so use a trusted storage
+  directory and treat it as conversation logging, not as secure secret storage.
+- Path safety checks protect the filename derived from the session id, but they
+  do not redact message contents or encrypt the file.
+"""
+
+USE_TEMP_DIRECTORY = False
+"""When True, store JSONL files in a temporary directory for this run only."""
+
+LOCAL_SESSIONS_DIRECTORY_NAME = "sessions"
+"""Folder name used when persisting history next to this sample file."""
+
+
+@tool(approval_mode="never_require")
+def get_weather(
+    city: Annotated[str, Field(description="The city to get the weather for.")],
+) -> str:
+    """Return a deterministic weather report for a city."""
+    weather_reports = {
+        "Seattle": "Seattle is rainy with a high of 13C.",
+        "Amsterdam": "Amsterdam is cloudy with a high of 16C.",
+    }
+    return weather_reports.get(city, f"{city} is sunny with a high of 20C.")
+
+
+@contextmanager
+def _resolve_storage_directory() -> Iterator[Path]:
+    """Yield the configured storage directory for the sample run."""
+    if USE_TEMP_DIRECTORY:
+        with tempfile.TemporaryDirectory(prefix="af-file-history-resume-") as temp_directory:
+            yield Path(temp_directory)
+        return
+
+    storage_directory = Path(__file__).resolve().parent / LOCAL_SESSIONS_DIRECTORY_NAME
+    storage_directory.mkdir(parents=True, exist_ok=True)
+    yield storage_directory
+
+
+async def main() -> None:
+    """Run the file history provider conversation persistence sample."""
+
+    with _resolve_storage_directory() as storage_directory:
+        print(f"Using temporary directory: {USE_TEMP_DIRECTORY}")
+        print(f"Storage directory: {storage_directory}\n")
+
+        # 1. Create the client, history provider, and tool-enabled agent.
+        agent = Agent(
+            client=FoundryChatClient(
+                credential=AzureCliCredential(),
+            ),
+            name="WeatherHistoryAgent",
+            instructions=(
+                "You are a helpful assistant. Use the get_weather tool for weather questions "
+                "and answer in one sentence using the tool result."
+            ),
+            tools=[get_weather],
+            context_providers=[
+                FileHistoryProvider(
+                    storage_directory,
+                    dumps=orjson.dumps if orjson else None,
+                    loads=orjson.loads if orjson else None,
+                )
+            ],
+            default_options={"store": False},
+        )
+
+        # 2. Ask about the first city so the JSONL file is created on disk.
+        session = agent.create_session()
+        history_file = storage_directory / f"{session.session_id}.jsonl"
+        print("=== First weather question ===\n")
+        first_query = "Use the get_weather tool and tell me the weather in Seattle."
+        first_response = await agent.run(first_query, session=session)
+        print(f"User:      {first_query}")
+        print(f"Assistant: {first_response.text}\n")
+
+        # 3. Read the stored JSONL records back from disk and pretty-print a few of them.
+        raw_lines = (await asyncio.to_thread(history_file.read_text, encoding="utf-8")).splitlines()
+        print(f"Stored message lines after first question: {len(raw_lines)}")
+        print(f"History file: {history_file}\n")
+        print("=== JSONL preview from disk ===\n")
+        for index, line in enumerate(raw_lines[:4], start=1):
+            print(f"Record {index}:")
+            print(json.dumps(json.loads(line), indent=2))
+            print()
+
+        # 4. Continue the same persisted conversation with another city.
+        print("=== Second weather question ===\n")
+        second_query = "Now use the get_weather tool for Amsterdam."
+        second_response = await agent.run(second_query, session=session)
+        print(f"User:      {second_query}")
+        print(f"Assistant: {second_response.text}\n")
+
+        updated_lines = (await asyncio.to_thread(history_file.read_text, encoding="utf-8")).splitlines()
+        print(f"Stored message lines after second question: {len(updated_lines)}")
+        print(f"History file: {history_file}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+"""
+Sample output:
+Using temporary directory: False
+Storage directory: /path/to/samples/02-agents/conversations/sessions
+
+=== First weather question ===
+
+User:      Use the get_weather tool and tell me the weather in Seattle.
+Assistant: <model response varies>
+
+Stored message lines after first question: 4
+History file: /path/to/samples/02-agents/conversations/sessions/<session-uuid>.jsonl
+
+=== JSONL preview from disk ===
+
+Record 1:
+{
+  "type": "message",
+  "role": "user",
+  ...
+}
+
+=== Second weather question ===
+
+User:      Now use the get_weather tool for Amsterdam.
+Assistant: <model response varies>
+
+Stored message lines after second question: 8
+History file: /path/to/samples/02-agents/conversations/sessions/<session-uuid>.jsonl
+"""
Author	SHA1	Message	Date
Jacob AlberandGitHub	fee6f1b862	Merge branch 'main' into dev/dotnet_workflow/add_handoff_sample	2026-04-16 13:04:51 -04:00
Jacob Alber	f853370621	docs: Add Handoff sample to readme	2026-04-16 12:42:28 -04:00
Jacob AlberandGitHub	c14beedb3a	test: Add Handoff composability test (#5208 )	2026-04-16 16:36:09 +00:00
Jacob Alber	2c9f543100	feat: Add Handoff sample	2026-04-16 12:01:19 -04:00
Kartik MadanandGitHub	43d98974d3	fix: propagate A2A metadata with namespaced key in additional_properties (#5240 ) (#5256 )	2026-04-16 15:22:39 +00:00
westey GitHub Copilot	60da0ffb48	.NET: Improve local release build perf by only formatting for one build target framework (#5266 ) * Improve local release build perf by only formatting for one build target framework * Update dotnet/Directory.Build.targets Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>	2026-04-16 15:21:33 +00:00
westeyandGitHub	a2044829b1	.NET: Update Microsoft.Extensions.AI to 10.5.0 and OpenAI to 2.10.0 and remove unused refs (#5269 ) * Update versions of System, Microsoft.Extensions and OpenAI packages * Remove unused package references * Remove further unused references	2026-04-16 11:03:51 +00:00
Giles Odigwe GitHub Copilot Copilot Evan Mattson	435c66e9c9	Python: Handle url_citation annotations in `FoundryChatClient` streaming responses (#5071 ) * Fix url_citation annotations dropped in streaming (#5029) Add url_citation branch to the streaming annotation handler in _parse_chunk_from_openai, mirroring the existing non-streaming path. The handler creates an Annotation with type='citation', title, url, and annotated_regions (TextSpanRegion), wrapped in Content.from_text. Update test_streaming_annotation_added_with_unknown_type to use a truly unknown type, and add new tests for url_citation (with and without url). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Address review feedback for #5029: Python: [Bug]: url_citation annotations silently dropped in Foundry streaming (SharePoint grounding citations lost) --------- Co-authored-by: Copilot <copilot@github.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>	2026-04-16 09:33:04 +00:00
Roger BarretoandGitHub	52d50be9e0	Bump Anthropic SDK to 12.13.0 and Anthropic.Foundry to 0.5.0 (#5279 ) - Update Anthropic from 12.11.0 to 12.13.0 - Update Anthropic.Foundry from 0.4.2 to 0.5.0 - Change Anthropic project from release candidate to preview - Add new IBetaService members (Agents, Environments, Sessions, Vaults) to test mock	2026-04-16 09:19:36 +00:00
Kartik Madan GitHub Evan Mattson	d20f9b5f97	Add AgentExecutorResponse.with_text() to preserve conversation history through custom executors (#5255 ) Fixes #5246 When a custom @executor transforms agent output and sends a plain str, the downstream AgentExecutor.from_str handler loses the full conversation context. This adds a with_text() helper that creates a new AgentExecutorResponse with replaced text while preserving the prior conversation chain, so AgentExecutor.from_response is invoked instead. - Add with_text(text) method to AgentExecutorResponse dataclass - Add 3 regression tests in test_full_conversation.py Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>	2026-04-16 08:39:19 +00:00
Peter IbekweandGitHub	87a8fa2a9d	.NET: Fix intermittent checkpoint-restore race in in-process workflow runs (#5134 ) * Improve workflow unit tests * Update test name prefix for clarity. * Update tests to surface any errors. * fix check-point restore-time race in off-thread workflow event stream * Fixes an intermittent checkpoint-restore race in in-process workflow runs.	2026-04-16 04:20:45 +00:00
Tao ChenandGitHub	8f7fd9525d	Python: Add OpenAI types to default checkpoint encoding allow list (#5297 ) * Add OpenAI types to default checkpoint encoding allow list * Address comments	2026-04-16 12:58:28 +09:00
Chinedum Echeta GitHub Tao Chen Evan Mattson	69697065ab	Python: Add context_providers and description to `workflow.as_agent()` (#4651 ) * Add context_providers and description to `workflow.as_agent()` * Add default workflow name and description * Positional * Move import --------- Co-authored-by: Tao Chen <taochen@microsoft.com> Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>	2026-04-16 02:47:29 +00:00
Evan MattsonandGitHub	fe4cd3cddc	Revert to public MCP server and skip on transient upstream errors (#5296 ) The local MCP server can't be used for hosted tools tests because Anthropic's backend needs to reach the MCP URL from their infrastructure (not localhost on the CI runner). Revert to learn.microsoft.com/api/mcp but catch BadRequestError, InternalServerError, APIConnectionError, and APITimeoutError and pytest.skip so upstream outages don't block the merge queue.	2026-04-16 11:46:49 +09:00
Evan MattsonandGitHub	611230cc8e	Python: improve misc-integration test robustness (#5295 ) * Python: use local MCP server for hosted tools test and broaden image assertion The hosted tools integration test was hitting rate limits on the external learn.microsoft.com MCP server, causing persistent failures that retries couldn't recover from. Switch to the local MCP server already spun up in CI via LOCAL_MCP_URL, skipping when the env var isn't set. Also broaden the image description assertion to accept common synonyms (cottage, mansion, villa, etc.) instead of just "house", since the model legitimately uses varied vocabulary for the same image. * Address review feedback: validate LOCAL_MCP_URL scheme and use word boundaries - Skip hosted tools test when LOCAL_MCP_URL lacks http/https scheme, matching the pattern used in test_mcp.py. - Use regex word boundaries for image assertion to avoid false matches like "villain" matching "villa".	2026-04-16 11:34:28 +09:00
Evan MattsonandGitHub	f112150cfb	Python: bump misc-integration retry delay to 30s (#5293 ) The misc-integration job (Anthropic, Ollama, MCP) frequently fails on merge to main when the upstream MCP server (e.g. learn.microsoft.com/api/mcp) returns a transient rate-limit error. The previous 5s retry delay is too short to ride out the upstream backoff window, so all retries fail and the merge queue is blocked. Bumping to 30s gives the upstream a chance to recover before pytest-retry re-runs the test.	2026-04-16 10:03:00 +09:00
Eduard van Valkenburg GitHub Copilot	ff05c22c58	Python: add experimental file history provider (#5248 ) * add experimental file history provider * Improve file history provider writes Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * typo * cleanup * cleanup * fix in readme * added security messages * Refine file history provider locking Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * added additional sample --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-04-15 22:23:37 +00:00
S3rj GitHub Sergey Borisov	eab7f09d03	Forward provider config to SessionConfig in GitHubCopilotAgent (fixes #5190 ) (#5195 ) Co-authored-by: Sergey Borisov <sergey.borisov@dataimpact.io>	2026-04-15 22:08:01 +00:00