Merge branch 'feature-foundry-agents' of https://github.com/microsoft/agent-framework into feature-foundry-agents

Package descriptions
Merge branch 'main' into feature-foundry-agents
2026-06-16 21:04:09 +08:00 · 2025-11-15 01:05:10 +00:00 · 2025-11-15 01:05:00 +00:00 · 2025-11-15 01:04:31 +00:00 · 2025-11-14 15:44:00 -08:00 · 2025-11-14 23:20:47 +00:00
52 changed files with 928 additions and 1108 deletions
@@ -11,6 +11,9 @@ updates:
    schedule:
      interval: "cron"
      cronjob: "0 8 * * 4,0" # Every Thursday(4) and Sunday(0) at 8:00 UTC
+    experimental:
+      nuget-native-updater: false
+      enable-cooldown-metrics-collection: false
    ignore:
      # For all System.* and Microsoft.Extensions/Bcl.* packages, ignore all major version updates
      - dependency-name: "System.*"
@@ -25,14 +28,6 @@ updates:
      - "dependencies"

  # Maintain dependencies for python
-  - package-ecosystem: "pip"
-    directory: "python/"
-    schedule:
-      interval: "weekly"
-      day: "monday"
-    labels:
-      - "python"
-      - "dependencies"
  - package-ecosystem: "uv"
    directory: "python/"
    schedule:
@@ -15,10 +15,10 @@
    <PackageVersion Include="Aspire.Hosting.AppHost" Version="$(AspireAppHostSdkVersion)" />
    <PackageVersion Include="Aspire.Hosting.Azure.CognitiveServices" Version="$(AspireAppHostSdkVersion)" />
    <PackageVersion Include="Aspire.Microsoft.Azure.Cosmos" Version="$(AspireAppHostSdkVersion)" />
-    <PackageVersion Include="CommunityToolkit.Aspire.OllamaSharp" Version="13.0.0-beta.440" />
+    <PackageVersion Include="CommunityToolkit.Aspire.OllamaSharp" Version="13.0.0-beta.435" />
    <!-- Azure.* -->
-    <PackageVersion Include="Azure.AI.Projects" Version="1.2.0-beta.3" />
-    <PackageVersion Include="Azure.AI.Projects.OpenAI" Version="1.0.0-beta.3" />
+    <PackageVersion Include="Azure.AI.Projects" Version="1.2.0-beta.1" />
+    <PackageVersion Include="Azure.AI.Projects.OpenAI" Version="1.0.0-beta.1" />
    <PackageVersion Include="Azure.AI.Agents.Persistent" Version="1.2.0-beta.7" />
    <PackageVersion Include="Azure.AI.OpenAI" Version="2.5.0-beta.1" />
    <PackageVersion Include="Azure.Identity" Version="1.17.0" />
@@ -2,9 +2,9 @@
  <PropertyGroup>
    <!-- Central version prefix - applies to all nuget packages. -->
    <VersionPrefix>1.0.0</VersionPrefix>
-    <PackageVersion Condition="'$(VersionSuffix)' != ''">$(VersionPrefix)-$(VersionSuffix).251114.1</PackageVersion>
-    <PackageVersion Condition="'$(VersionSuffix)' == ''">$(VersionPrefix)-preview.251114.1</PackageVersion>
-    <GitTag>1.0.0-preview.251114.1</GitTag>
+    <PackageVersion Condition="'$(VersionSuffix)' != ''">$(VersionPrefix)-$(VersionSuffix).251113.1</PackageVersion>
+    <PackageVersion Condition="'$(VersionSuffix)' == ''">$(VersionPrefix)-preview.251113.1</PackageVersion>
+    <GitTag>1.0.0-preview.251113.1</GitTag>

    <Configurations>Debug;Release;Publish</Configurations>
    <IsPackable>true</IsPackable>
@@ -1,83 +0,0 @@
-# Getting started with Foundry Agents
-
-The getting started with Foundry Agents samples demonstrate the fundamental concepts and functionalities
-of Azure Foundry Agents and can be used with Azure Foundry as the AI provider.
-
-These samples showcase how to work with agents managed through Azure Foundry, including agent creation,
-versioning, multi-turn conversations, and advanced features like code interpretation and computer use.
-
-## Getting started with Foundry Agents prerequisites
-
-Before you begin, ensure you have the following prerequisites:
-
- .NET 8.0 SDK or later
- Azure Foundry service endpoint and project configured
- Azure CLI installed and authenticated (for Azure credential authentication)
-
-**Note**: These samples use Azure Foundry Agents. For more information, see [Azure AI Foundry documentation](https://learn.microsoft.com/en-us/azure/ai-foundry/).
-
-**Note**: These samples use Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
-
-## Samples
-
-|Sample|Description|
-|---|---|
-|[Basics](./FoundryAgents_Step01.1_Basics/)|This sample demonstrates how to create and manage AI agents with versioning|
-|[Running a simple agent](./FoundryAgents_Step01.2_Running/)|This sample demonstrates how to create and run a basic Foundry agent|
-|[Multi-turn conversation](./FoundryAgents_Step02_MultiturnConversation/)|This sample demonstrates how to implement a multi-turn conversation with a Foundry agent|
-|[Using function tools](./FoundryAgents_Step03.1_UsingFunctionTools/)|This sample demonstrates how to use function tools with a Foundry agent|
-|[Using OpenAPI function tools](./FoundryAgents_Step03.2_UsingFunctionTools_FromOpenAPI/)|This sample demonstrates how to create function tools from an OpenAPI spec and use them with a Foundry agent|
-|[Using function tools with approvals](./FoundryAgents_Step04_UsingFunctionToolsWithApprovals/)|This sample demonstrates how to use function tools where approvals require human in the loop approvals before execution|
-|[Structured output](./FoundryAgents_Step05_StructuredOutput/)|This sample demonstrates how to use structured output with a Foundry agent|
-|[Persisted conversations](./FoundryAgents_Step06_PersistedConversations/)|This sample demonstrates how to persist conversations and reload them later|
-|[Observability](./FoundryAgents_Step07_Observability/)|This sample demonstrates how to add telemetry to a Foundry agent|
-|[Dependency injection](./FoundryAgents_Step08_DependencyInjection/)|This sample demonstrates how to add and resolve a Foundry agent with a dependency injection container|
-|[Using MCP client as tools](./FoundryAgents_Step09_UsingMcpClientAsTools/)|This sample demonstrates how to use MCP clients as tools with a Foundry agent|
-|[Using images](./FoundryAgents_Step10_UsingImages/)|This sample demonstrates how to use image multi-modality with a Foundry agent|
-|[Exposing as a function tool](./FoundryAgents_Step11_AsFunctionTool/)|This sample demonstrates how to expose a Foundry agent as a function tool|
-|[Using middleware](./FoundryAgents_Step12_Middleware/)|This sample demonstrates how to use middleware with a Foundry agent|
-|[Using plugins](./FoundryAgents_Step13_Plugins/)|This sample demonstrates how to use plugins with a Foundry agent|
-|[Code interpreter](./FoundryAgents_Step14_CodeInterpreter/)|This sample demonstrates how to use the code interpreter tool with a Foundry agent|
-|[Computer use](./FoundryAgents_Step15_ComputerUse/)|This sample demonstrates how to use computer use capabilities with a Foundry agent|
-
-## Running the samples from the console
-
-To run the samples, navigate to the desired sample directory, e.g.
-
-```powershell
-cd FoundryAgents_Step01.2_Running
-```
-
-Set the following environment variables:
-
-```powershell
-$env:AZURE_FOUNDRY_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Azure Foundry resource endpoint
-$env:AZURE_FOUNDRY_PROJECT_DEPLOYMENT_NAME="gpt-4o-mini"  # Optional, defaults to gpt-4o-mini
-```
-
-If the variables are not set, you will be prompted for the values when running the samples.
-
-Execute the following command to build the sample:
-
-```powershell
-dotnet build
-```
-
-Execute the following command to run the sample:
-
-```powershell
-dotnet run --no-build
-```
-
-Or just build and run in one step:
-
-```powershell
-dotnet run
-```
-
-## Running the samples from Visual Studio
-
-Open the solution in Visual Studio and set the desired sample project as the startup project. Then, run the project using the built-in debugger or by pressing `F5`.
-
-You will be prompted for any required environment variables if they are not already set.
-
@@ -8,7 +8,6 @@ of the agent framework.
 |Sample|Description|
 |---|---|
 |[Agents](./Agents/README.md)|Step by step instructions for getting started with agents|
-|[Foundry Agents](./FoundryAgents/README.md)|Getting started with Azure Foundry Agents|
 |[Agent Providers](./AgentProviders/README.md)|Getting started with creating agents using various providers|
 |[Agents With Retrieval Augmented Generation (RAG)](./AgentWithRAG/README.md)|Adding Retrieval Augmented Generation (RAG) capabilities to your agents.|
 |[Agents With Memory](./AgentWithMemory/README.md)|Adding Memory capabilities to your agents.|
@@ -13,7 +13,6 @@ internal interface ISuperStepJoinContext

    ValueTask ForwardWorkflowEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default);
    ValueTask SendMessageAsync<TMessage>(string senderId, [DisallowNull] TMessage message, CancellationToken cancellationToken = default);
-    ValueTask YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken = default);

    ValueTask<string> AttachSuperstepAsync(ISuperStepRunner superStepRunner, CancellationToken cancellationToken = default);
    ValueTask<bool> DetachSuperstepAsync(string id);
@@ -429,21 +429,4 @@ internal sealed class InProcessRunnerContext : IRunnerContext

    ValueTask ISuperStepJoinContext.SendMessageAsync<TMessage>(string senderId, [DisallowNull] TMessage message, CancellationToken cancellationToken)
        => this.SendMessageAsync(senderId, Throw.IfNull(message), cancellationToken: cancellationToken);
-
-    async ValueTask ISuperStepJoinContext.YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken)
-    {
-        this.CheckEnded();
-        Throw.IfNull(output);
-
-        Executor sourceExecutor = await this.EnsureExecutorAsync(sourceId, tracer: null, cancellationToken).ConfigureAwait(false);
-        if (!sourceExecutor.CanOutput(output.GetType()))
-        {
-            throw new InvalidOperationException($"Cannot output object of type {output.GetType().Name}. Expecting one of [{string.Join(", ", sourceExecutor.OutputTypes)}].");
-        }
-
-        if (this._outputFilter.CanOutput(sourceId, output))
-        {
-            await this.AddEventAsync(new WorkflowOutputEvent(output, sourceId), cancellationToken).ConfigureAwait(false);
-        }
-    }
 }
@@ -192,16 +192,11 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
                    resultTask = this._joinContext?.ForwardWorkflowEventAsync(new SubworkflowErrorEvent(this.Id, errorEvent.Data as Exception)).AsTask() ?? Task.CompletedTask;
                    break;
                case WorkflowOutputEvent outputEvent:
-                    if (this._joinContext != null && outputEvent.Data != null)
+                    if (this._joinContext != null &&
+                        this._options.AutoSendMessageHandlerResultObject
+                        && outputEvent.Data != null)
                    {
-                        if (this._options.AutoSendMessageHandlerResultObject)
-                        {
-                            resultTask = this._joinContext.SendMessageAsync(this.Id, outputEvent.Data).AsTask();
-                        }
-                        else if (this._options.AutoYieldOutputHandlerResultObject)
-                        {
-                            resultTask = this._joinContext.YieldOutputAsync(this.Id, outputEvent.Data).AsTask();
-                        }
+                        resultTask = this._joinContext.SendMessageAsync(this.Id, outputEvent.Data).AsTask();
                    }
                    break;
                case RequestHaltEvent requestHaltEvent:
@@ -107,9 +107,6 @@ public class TestRunContext : IRunnerContext
    public ValueTask SendMessageAsync<TMessage>(string senderId, [System.Diagnostics.CodeAnalysis.DisallowNull] TMessage message, CancellationToken cancellationToken = default)
        => this.SendMessageAsync(senderId, message, cancellationToken);

-    public ValueTask YieldOutputAsync(string sourceId, object output, CancellationToken cancellationToken = default)
-        => this.AddEventAsync(new WorkflowOutputEvent(output, sourceId), cancellationToken);
-
    ValueTask<string> ISuperStepJoinContext.AttachSuperstepAsync(ISuperStepRunner superStepRunner, CancellationToken cancellationToken) => new(string.Empty);
    ValueTask<bool> ISuperStepJoinContext.DetachSuperstepAsync(string joinId) => new(false);
 }
@@ -1,190 +0,0 @@
-// Copyright (c) Microsoft. All rights reserved.
-
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using FluentAssertions;
-
-namespace Microsoft.Agents.AI.Workflows.UnitTests;
-
-public sealed class WorkflowHostExecutorTests
-{
-    [Fact]
-    public async Task WorkflowHostExecutor_WithAutoYieldOutputHandlerResultObjectTrue_YieldsSubworkflowOutput()
-    {
-        // Arrange
-        const string outputData = "test output from subworkflow";
-
-        Func<string, IWorkflowContext, CancellationToken, ValueTask> processFunc = (input, context, cancellationToken) => context.YieldOutputAsync(input, cancellationToken);
-
-        ExecutorBinding subworkflowExecutor = processFunc.BindAsExecutor("SubworkflowExecutor", threadsafe: true);
-        Workflow subworkflow = new WorkflowBuilder(subworkflowExecutor)
-            .WithOutputFrom(subworkflowExecutor)
-            .Build();
-
-        ExecutorBinding workflowHostExecutor = subworkflow.BindAsExecutor(
-            "HostExecutor",
-            new ExecutorOptions
-            {
-                AutoSendMessageHandlerResultObject = false,
-                AutoYieldOutputHandlerResultObject = true
-            });
-
-        Func<string, string, ValueTask<Executor>> createOrchestrator = (id, _) => new(new OrchestratorExecutor(id));
-        ExecutorBinding orchestrator = createOrchestrator.BindExecutor();
-
-        Workflow workflow = new WorkflowBuilder(orchestrator)
-            .AddEdge(orchestrator, workflowHostExecutor)
-            .AddEdge(workflowHostExecutor, orchestrator)
-            .WithOutputFrom(workflowHostExecutor)
-            .Build();
-
-        // Act
-        Run workflowRun = await InProcessExecution.RunAsync(workflow, outputData);
-
-        // Assert
-        RunStatus status = await workflowRun.GetStatusAsync();
-        status.Should().Be(RunStatus.Idle);
-
-        List<WorkflowOutputEvent> outputEvents = workflowRun.OutgoingEvents
-            .OfType<WorkflowOutputEvent>()
-            .ToList();
-
-        outputEvents.Should().HaveCount(1, "the workflow should produce exactly one output event");
-        outputEvents[0].As<string>().Should().Be(outputData, "the output should match the input data");
-    }
-
-    [Fact]
-    public async Task WorkflowHostExecutor_WithAutoSendMessageHandlerResultObjectTrue_SendsMessageNotYield()
-    {
-        // Arrange
-        const string outputData = "test output from subworkflow";
-
-        Func<string, IWorkflowContext, CancellationToken, ValueTask> processFunc = (input, context, cancellationToken) => context.YieldOutputAsync(input, cancellationToken);
-
-        ExecutorBinding subworkflowExecutor = processFunc.BindAsExecutor("SubworkflowExecutor", threadsafe: true);
-        Workflow subworkflow = new WorkflowBuilder(subworkflowExecutor)
-            .WithOutputFrom(subworkflowExecutor)
-            .Build();
-
-        ExecutorBinding workflowHostExecutor = subworkflow.BindAsExecutor(
-            "HostExecutor",
-            new ExecutorOptions
-            {
-                AutoSendMessageHandlerResultObject = true,
-                AutoYieldOutputHandlerResultObject = false
-            });
-
-        Func<string, string, ValueTask<Executor>> createOrchestrator = (id, _) => new(new OrchestratorExecutor(id));
-        ExecutorBinding orchestrator = createOrchestrator.BindExecutor();
-
-        Workflow workflow = new WorkflowBuilder(orchestrator)
-            .AddEdge(orchestrator, workflowHostExecutor)
-            .AddEdge(workflowHostExecutor, orchestrator)
-            .WithOutputFrom(orchestrator)
-            .Build();
-
-        // Act
-        Run workflowRun = await InProcessExecution.RunAsync(workflow, outputData);
-
-        // Assert
-        RunStatus status = await workflowRun.GetStatusAsync();
-        status.Should().Be(RunStatus.Idle);
-
-        List<WorkflowOutputEvent> outputEvents = workflowRun.OutgoingEvents
-            .OfType<WorkflowOutputEvent>()
-            .ToList();
-
-        // With AutoSendMessageHandlerResultObject, the output is sent as a message back to orchestrator, which yields it
-        outputEvents.Should().HaveCount(1, "the workflow should produce exactly one output event");
-        outputEvents[0].As<string>().Should().Be(outputData, "the output should match the input data");
-    }
-
-    [Fact]
-    public async Task WorkflowHostExecutor_WithBothOptionsFalse_DoesNotPropagate()
-    {
-        // Arrange
-        const string outputData = "test output from subworkflow";
-
-        Func<string, IWorkflowContext, CancellationToken, ValueTask> processFunc = (input, context, cancellationToken) => context.YieldOutputAsync(input, cancellationToken);
-
-        ExecutorBinding subworkflowExecutor = processFunc.BindAsExecutor("SubworkflowExecutor", threadsafe: true);
-        Workflow subworkflow = new WorkflowBuilder(subworkflowExecutor)
-            .WithOutputFrom(subworkflowExecutor)
-            .Build();
-
-        ExecutorBinding workflowHostExecutor = subworkflow.BindAsExecutor(
-            "HostExecutor",
-            new ExecutorOptions
-            {
-                AutoSendMessageHandlerResultObject = false,
-                AutoYieldOutputHandlerResultObject = false
-            });
-
-        Func<string, string, ValueTask<Executor>> createOrchestrator = (id, _) => new(new OrchestratorExecutor(id));
-        ExecutorBinding orchestrator = createOrchestrator.BindExecutor();
-
-        Workflow workflow = new WorkflowBuilder(orchestrator)
-            .AddEdge(orchestrator, workflowHostExecutor)
-            .AddEdge(workflowHostExecutor, orchestrator)
-            .WithOutputFrom(orchestrator)
-            .Build();
-
-        // Act
-        Run workflowRun = await InProcessExecution.RunAsync(workflow, outputData);
-
-        // Assert
-        RunStatus status = await workflowRun.GetStatusAsync();
-        status.Should().Be(RunStatus.Idle);
-
-        List<WorkflowOutputEvent> outputEvents = workflowRun.OutgoingEvents
-            .OfType<WorkflowOutputEvent>()
-            .ToList();
-
-        // When both options are false, the subworkflow output is not propagated
-        outputEvents.Should().BeEmpty("no output should be yielded when both options are false");
-    }
-
-    private sealed class OrchestratorExecutor : StatefulExecutor<OrchestratorExecutor.State>
-    {
-        internal sealed class State
-        {
-            public bool ReceivedInput { get; set; }
-            public string? Result { get; set; }
-        }
-
-        public OrchestratorExecutor(string id)
-            : base(id, () => new State(), declareCrossRunShareable: false)
-        {
-        }
-
-        protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
-        {
-            return routeBuilder
-                .AddHandler<string>(this.HandleInputAsync);
-        }
-
-        private async ValueTask HandleInputAsync(string input, IWorkflowContext context, CancellationToken cancellationToken)
-        {
-            await this.InvokeWithStateAsync(ProcessInputAsync, context, cancellationToken: cancellationToken);
-
-            async ValueTask<State?> ProcessInputAsync(State state, IWorkflowContext context, CancellationToken cancellationToken)
-            {
-                if (!state.ReceivedInput)
-                {
-                    state.ReceivedInput = true;
-                    await context.SendMessageAsync(input, cancellationToken: cancellationToken);
-                }
-                else
-                {
-                    state.Result = input;
-                    await context.YieldOutputAsync(input, cancellationToken);
-                }
-
-                return state;
-            }
-        }
-    }
-}
@@ -18,7 +18,6 @@ from agent_framework import (
    FunctionCallContent,
    FunctionResultContent,
    HostedCodeInterpreterTool,
-    HostedFileContent,
    HostedMCPTool,
    HostedWebSearchTool,
    Role,
@@ -123,7 +122,6 @@ class AnthropicClient(BaseChatClient):
        api_key: str | None = None,
        model_id: str | None = None,
        anthropic_client: AsyncAnthropic | None = None,
-        additional_beta_flags: list[str] | None = None,
        env_file_path: str | None = None,
        env_file_encoding: str | None = None,
        **kwargs: Any,
@@ -136,8 +134,6 @@ class AnthropicClient(BaseChatClient):
            anthropic_client: An existing Anthropic client to use. If not provided, one will be created.
                This can be used to further configure the client before passing it in.
                For instance if you need to set a different base_url for testing or private deployments.
-            additional_beta_flags: Additional beta flags to enable on the client.
-                Default flags are: "mcp-client-2025-04-04", "code-execution-2025-08-25".
            env_file_path: Path to environment file for loading settings.
            env_file_encoding: Encoding of the environment file.
            kwargs: Additional keyword arguments passed to the parent class.
@@ -200,7 +196,6 @@ class AnthropicClient(BaseChatClient):

        # Initialize instance variables
        self.anthropic_client = anthropic_client
-        self.additional_beta_flags = additional_beta_flags or []
        self.model_id = anthropic_settings.chat_model_id
        # streaming requires tracking the last function call ID and name
        self._last_call_id_name: tuple[str, str] | None = None
@@ -251,16 +246,12 @@ class AnthropicClient(BaseChatClient):
        Returns:
            A dictionary of run options for the Anthropic client.
        """
-        if chat_options.additional_properties and "additional_beta_flags" in chat_options.additional_properties:
-            betas = chat_options.additional_properties.pop("additional_beta_flags")
-        else:
-            betas = []
        run_options: dict[str, Any] = {
            "model": chat_options.model_id or self.model_id,
            "messages": self._convert_messages_to_anthropic_format(messages),
            "max_tokens": chat_options.max_tokens or ANTHROPIC_DEFAULT_MAX_TOKENS,
            "extra_headers": {"User-Agent": AGENT_FRAMEWORK_USER_AGENT},
-            "betas": {*BETA_FLAGS, *self.additional_beta_flags, *betas},
+            "betas": BETA_FLAGS,
        }

        # Add any additional options from chat_options or kwargs
@@ -405,7 +396,7 @@ class AnthropicClient(BaseChatClient):
                case HostedCodeInterpreterTool():
                    code_tool: dict[str, Any] = {
                        "type": "code_execution_20250825",
-                        "name": "code_execution",
+                        "name": "code_interpreter",
                    }
                    tool_list.append(code_tool)
                case HostedMCPTool():
@@ -533,7 +524,17 @@ class AnthropicClient(BaseChatClient):
                            annotations=self._parse_citations(content_block),
                        )
                    )
-                case "tool_use" | "mcp_tool_use" | "server_tool_use":
+                case "tool_use":
+                    self._last_call_id_name = (content_block.id, content_block.name)
+                    contents.append(
+                        FunctionCallContent(
+                            call_id=content_block.id,
+                            name=content_block.name,
+                            arguments=content_block.input,
+                            raw_representation=content_block,
+                        )
+                    )
+                case "mcp_tool_use" | "server_tool_use":
                    self._last_call_id_name = (content_block.id, content_block.name)
                    contents.append(
                        FunctionCallContent(
@@ -571,19 +572,6 @@ class AnthropicClient(BaseChatClient):
                    | "text_editor_code_execution_tool_result"
                ):
                    call_id, name = self._last_call_id_name or (None, None)
-                    if (
-                        content_block.content
-                        and (
-                            content_block.content.type == "bash_code_execution_result"
-                            or content_block.content.type == "code_execution_result"
-                        )
-                        and content_block.content.content
-                    ):
-                        for result_content in content_block.content.content:
-                            if hasattr(result_content, "file_id"):
-                                contents.append(
-                                    HostedFileContent(file_id=result_content.file_id, raw_representation=result_content)
-                                )
                    contents.append(
                        FunctionResultContent(
                            call_id=content_block.tool_use_id,
@@ -50,9 +50,7 @@ def create_test_anthropic_client(
 ) -> AnthropicClient:
    """Helper function to create AnthropicClient instances for testing, bypassing normal validation."""
    if anthropic_settings is None:
-        anthropic_settings = AnthropicSettings(
-            api_key="test-api-key-12345", chat_model_id="claude-3-5-sonnet-20241022", env_file_path="test.env"
-        )
+        anthropic_settings = AnthropicSettings(api_key="test-api-key-12345", chat_model_id="claude-3-5-sonnet-20241022")

    # Create client instance directly
    client = object.__new__(AnthropicClient)
@@ -63,7 +61,6 @@ def create_test_anthropic_client(
    client._last_call_id_name = None
    client.additional_properties = {}
    client.middleware = None
-    client.additional_beta_flags = []

    return client

@@ -73,7 +70,7 @@ def create_test_anthropic_client(

 def test_anthropic_settings_init(anthropic_unit_test_env: dict[str, str]) -> None:
    """Test AnthropicSettings initialization."""
-    settings = AnthropicSettings(env_file_path="test.env")
+    settings = AnthropicSettings()

    assert settings.api_key is not None
    assert settings.api_key.get_secret_value() == anthropic_unit_test_env["ANTHROPIC_API_KEY"]
@@ -83,7 +80,8 @@ def test_anthropic_settings_init(anthropic_unit_test_env: dict[str, str]) -> Non
 def test_anthropic_settings_init_with_explicit_values() -> None:
    """Test AnthropicSettings initialization with explicit values."""
    settings = AnthropicSettings(
-        api_key="custom-api-key", chat_model_id="claude-3-opus-20240229", env_file_path="test.env"
+        api_key="custom-api-key",
+        chat_model_id="claude-3-opus-20240229",
    )

    assert settings.api_key is not None
@@ -116,7 +114,6 @@ def test_anthropic_client_init_auto_create_client(anthropic_unit_test_env: dict[
    client = AnthropicClient(
        api_key=anthropic_unit_test_env["ANTHROPIC_API_KEY"],
        model_id=anthropic_unit_test_env["ANTHROPIC_CHAT_MODEL_ID"],
-        env_file_path="test.env",
    )

    assert client.anthropic_client is not None
@@ -310,7 +307,7 @@ def test_convert_tools_to_anthropic_format_code_interpreter(mock_anthropic_clien
    assert "tools" in result
    assert len(result["tools"]) == 1
    assert result["tools"][0]["type"] == "code_execution_20250825"
-    assert result["tools"][0]["name"] == "code_execution"
+    assert result["tools"][0]["name"] == "code_interpreter"


 def test_convert_tools_to_anthropic_format_mcp_tool(mock_anthropic_client: MagicMock) -> None:
@@ -728,32 +725,6 @@ async def test_anthropic_client_integration_function_calling() -> None:
    assert has_function_call


-@pytest.mark.flaky
-@skip_if_anthropic_integration_tests_disabled
-async def test_anthropic_client_integration_hosted_tools() -> None:
-    """Integration test for hosted tools."""
-    client = AnthropicClient()
-
-    messages = [ChatMessage(role=Role.USER, text="What tools do you have available?")]
-    tools = [
-        HostedWebSearchTool(),
-        HostedCodeInterpreterTool(),
-        HostedMCPTool(
-            name="example-mcp",
-            url="https://learn.microsoft.com/api/mcp",
-            approval_mode="never_require",
-        ),
-    ]
-
-    response = await client.get_response(
-        messages=messages,
-        chat_options=ChatOptions(tools=tools, max_tokens=100),
-    )
-
-    assert response is not None
-    assert response.text is not None
-
-
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
 async def test_anthropic_client_integration_with_system_message() -> None:
@@ -37,8 +37,6 @@ from ._events import (
    ExecutorFailedEvent,
    ExecutorInvokedEvent,
    RequestInfoEvent,
-    SuperStepCompletedEvent,
-    SuperStepStartedEvent,
    WorkflowErrorDetails,
    WorkflowEvent,
    WorkflowEventSource,
@@ -154,8 +152,6 @@ __all__ = [
    "StandardMagenticManager",
    "SubWorkflowRequestMessage",
    "SubWorkflowResponseMessage",
-    "SuperStepCompletedEvent",
-    "SuperStepStartedEvent",
    "SwitchCaseEdgeGroup",
    "SwitchCaseEdgeGroupCase",
    "SwitchCaseEdgeGroupDefault",
@@ -35,8 +35,6 @@ from ._events import (
    ExecutorFailedEvent,
    ExecutorInvokedEvent,
    RequestInfoEvent,
-    SuperStepCompletedEvent,
-    SuperStepStartedEvent,
    WorkflowErrorDetails,
    WorkflowEvent,
    WorkflowEventSource,
@@ -150,8 +148,6 @@ __all__ = [
    "StandardMagenticManager",
    "SubWorkflowRequestMessage",
    "SubWorkflowResponseMessage",
-    "SuperStepCompletedEvent",
-    "SuperStepStartedEvent",
    "SwitchCaseEdgeGroup",
    "SwitchCaseEdgeGroupCase",
    "SwitchCaseEdgeGroupDefault",
@@ -1,7 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.

 import logging
-import sys
 from dataclasses import dataclass
 from typing import Any, cast

@@ -21,11 +20,6 @@ from ._message_utils import normalize_messages_input
 from ._request_info_mixin import response_handler
 from ._workflow_context import WorkflowContext

-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-
 logger = logging.getLogger(__name__)


@@ -185,8 +179,7 @@ class AgentExecutor(Executor):
            self._pending_responses_to_agent.clear()
            await self._run_agent_and_emit(ctx)

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
+    async def snapshot_state(self) -> dict[str, Any]:
        """Capture current executor state for checkpointing.

        NOTE: if the thread storage is on the server side, the full thread state
@@ -203,6 +196,9 @@ class AgentExecutor(Executor):
            client_module = self._agent.chat_client.__class__.__module__

            if client_class_name == "AzureAIAgentClient" and "azure_ai" in client_module:
+                # TODO(TaoChenOSU): update this warning when we surface the hooks for
+                # custom executor checkpointing.
+                # https://github.com/microsoft/agent-framework/issues/1816
                logger.warning(
                    "Checkpointing an AgentExecutor with AzureAIAgentClient that uses server-side threads. "
                    "Currently, checkpointing does not capture messages from server-side threads "
@@ -221,8 +217,7 @@ class AgentExecutor(Executor):
            "pending_responses_to_agent": encode_checkpoint_value(self._pending_responses_to_agent),
        }

-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+    async def restore_state(self, state: dict[str, Any]) -> None:
        """Restore executor state from checkpoint.

        Args:
@@ -4,7 +4,6 @@

 import inspect
 import logging
-import sys
 from abc import ABC, abstractmethod
 from collections.abc import Awaitable, Callable, Sequence
 from typing import Any
@@ -14,12 +13,6 @@ from ._executor import Executor
 from ._orchestrator_helpers import ParticipantRegistry
 from ._workflow_context import WorkflowContext

-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-
-
 logger = logging.getLogger(__name__)


@@ -217,12 +210,11 @@ class BaseGroupChatOrchestrator(Executor, ABC):

    # State persistence (shared across all patterns)

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
+    def snapshot_state(self) -> dict[str, Any]:
        """Capture current orchestrator state for checkpointing.

        Default implementation uses OrchestrationState to serialize common state.
-        Subclasses can override this method or _snapshot_pattern_metadata() to add pattern-specific data.
+        Subclasses should override _snapshot_pattern_metadata() to add pattern-specific data.

        Returns:
            Serialized state dict
@@ -246,12 +238,11 @@ class BaseGroupChatOrchestrator(Executor, ABC):
        """
        return {}

-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+    def restore_state(self, state: dict[str, Any]) -> None:
        """Restore orchestrator state from checkpoint.

        Default implementation uses OrchestrationState to deserialize common state.
-        Subclasses can override this method or _restore_pattern_metadata() to restore pattern-specific data.
+        Subclasses should override _restore_pattern_metadata() to restore pattern-specific data.

        Args:
            state: Serialized state dict
@@ -6,7 +6,9 @@ These utilities operate on standard `list[ChatMessage]` collections and simple
 dictionary snapshots so orchestrators can share logic without new mixins.
 """

-from collections.abc import Sequence
+import json
+from collections.abc import Mapping, Sequence
+from typing import Any

 from .._types import ChatMessage

@@ -24,3 +26,25 @@ def ensure_author(message: ChatMessage, fallback: str) -> ChatMessage:
    """Attach `fallback` author if message is missing `author_name`."""
    message.author_name = message.author_name or fallback
    return message
+
+
+def snapshot_state(conversation: Sequence[ChatMessage]) -> dict[str, Any]:
+    """Build an immutable snapshot for checkpoint storage."""
+    if hasattr(conversation, "to_dict"):
+        result = conversation.to_dict()  # type: ignore[attr-defined]
+        if isinstance(result, dict):
+            return result  # type: ignore[return-value]
+        if isinstance(result, Mapping):
+            return dict(result)  # type: ignore[arg-type]
+    serialisable: list[dict[str, Any]] = []
+    for message in conversation:
+        if hasattr(message, "to_dict") and callable(message.to_dict):  # type: ignore[attr-defined]
+            msg_dict = message.to_dict()  # type: ignore[attr-defined]
+            serialisable.append(dict(msg_dict) if isinstance(msg_dict, Mapping) else msg_dict)  # type: ignore[arg-type]
+        elif hasattr(message, "to_json") and callable(message.to_json):  # type: ignore[attr-defined]
+            json_payload = message.to_json()  # type: ignore[attr-defined]
+            parsed = json.loads(json_payload) if isinstance(json_payload, str) else json_payload
+            serialisable.append(dict(parsed) if isinstance(parsed, Mapping) else parsed)  # type: ignore[arg-type]
+        else:
+            serialisable.append(dict(getattr(message, "__dict__", {})))  # type: ignore[arg-type]
+    return {"messages": serialisable}
@@ -294,36 +294,6 @@ class WorkflowOutputEvent(WorkflowEvent):
        return f"{self.__class__.__name__}(data={self.data}, source_executor_id={self.source_executor_id})"


-class SuperStepEvent(WorkflowEvent):
-    """Event triggered when a superstep starts or ends."""
-
-    def __init__(self, iteration: int, data: Any | None = None):
-        """Initialize the superstep event.
-
-        Args:
-            iteration: The number of the superstep (1-based index).
-            data: Optional data associated with the superstep event.
-        """
-        super().__init__(data)
-        self.iteration = iteration
-
-    def __repr__(self) -> str:
-        """Return a string representation of the superstep event."""
-        return f"{self.__class__.__name__}(iteration={self.iteration}, data={self.data})"
-
-
-class SuperStepStartedEvent(SuperStepEvent):
-    """Event triggered when a superstep starts."""
-
-    ...
-
-
-class SuperStepCompletedEvent(SuperStepEvent):
-    """Event triggered when a superstep ends."""
-
-    ...
-
-
 class ExecutorEvent(WorkflowEvent):
    """Base class for executor events."""

@@ -340,13 +310,17 @@ class ExecutorEvent(WorkflowEvent):
 class ExecutorInvokedEvent(ExecutorEvent):
    """Event triggered when an executor handler is invoked."""

-    ...
+    def __repr__(self) -> str:
+        """Return a string representation of the executor handler invoke event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})"


 class ExecutorCompletedEvent(ExecutorEvent):
    """Event triggered when an executor handler is completed."""

-    ...
+    def __repr__(self) -> str:
+        """Return a string representation of the executor handler complete event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})"


 class ExecutorFailedEvent(ExecutorEvent):
@@ -155,11 +155,6 @@ class Executor(RequestInfoMixin, DictConvertible):
    that parent workflows can intercept. See WorkflowExecutor documentation for details on
    workflow composition patterns and request/response handling.

-    ## State Management
-    Executors can contain states that persist across workflow runs and checkpoints. Override the
-    `on_checkpoint_save` and `on_checkpoint_restore` methods to implement custom state
-    serialization and restoration logic.
-
    ## Implementation Notes
    - Do not call `execute()` directly - it's invoked by the workflow engine
    - Do not override `execute()` - define handlers using decorators instead
@@ -465,32 +460,6 @@ class Executor(RequestInfoMixin, DictConvertible):
                return self._handlers[message_type]
        raise RuntimeError(f"Executor {self.__class__.__name__} cannot handle message of type {type(message)}.")

-    async def on_checkpoint_save(self) -> dict[str, Any]:
-        """Hook called when the workflow is being saved to a checkpoint.
-
-        Override this method in subclasses to implement custom logic that should
-        return state to be saved in the checkpoint.
-
-        The returned state dictionary will be passed to `on_checkpoint_restore`
-        when the workflow is restored from the checkpoint. The dictionary should
-        only contain JSON-serializable data.
-
-        Returns:
-            A state dictionary to be saved during checkpointing.
-        """
-        return {}
-
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        """Hook called when the workflow is restored from a checkpoint.
-
-        Override this method in subclasses to implement custom logic that should
-        run when the workflow is restored from a checkpoint.
-
-        Args:
-            state: The state dictionary that was saved during checkpointing.
-        """
-        ...
-

 # endregion: Executor

@@ -16,7 +16,6 @@ Key properties:

 import logging
 import re
-import sys
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Any
@@ -51,12 +50,6 @@ from ._workflow import Workflow
 from ._workflow_builder import WorkflowBuilder
 from ._workflow_context import WorkflowContext

-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-
-
 logger = logging.getLogger(__name__)


@@ -314,6 +307,15 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
        ctx: WorkflowContext[AgentExecutorRequest | list[ChatMessage], list[ChatMessage] | _ConversationForUserInput],
    ) -> None:
        """Process an agent's response and determine whether to route, request input, or terminate."""
+        # Hydrate coordinator state (and detect new run) using checkpointable executor state
+        state = await ctx.get_executor_state()
+        if not state:
+            self._clear_conversation()
+        elif not self._get_conversation():
+            restored = self._restore_conversation_from_state(state)
+            if restored:
+                self._conversation = list(restored)
+
        source = ctx.get_source_executor_id()
        is_starting_agent = source == self._starting_agent_id

@@ -341,7 +343,7 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
            # Update current agent when handoff occurs
            self._current_agent_id = target
            logger.info(f"Handoff detected: {source} -> {target}. Routing control to specialist '{target}'.")
-
+            await self._persist_state(ctx)
            # Clean tool-related content before sending to next agent
            cleaned = clean_conversation_for_handoff(conversation)
            request = AgentExecutorRequest(messages=cleaned, should_respond=True)
@@ -358,6 +360,7 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
            f"Agent '{source}' responded without handoff. "
            f"Requesting user input. Return-to-previous: {self._return_to_previous}"
        )
+        await self._persist_state(ctx)

        if await self._check_termination():
            # Clean the output conversation for display
@@ -385,6 +388,7 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
        """Receive full conversation with new user input from gateway, update history, trim for agent."""
        # Update authoritative conversation
        self._conversation = list(message.full_conversation)
+        await self._persist_state(ctx)

        # Check termination before sending to agent
        if await self._check_termination():
@@ -469,7 +473,11 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
            )
        return list(conversation)

-    @override
+    async def _persist_state(self, ctx: WorkflowContext[Any, Any]) -> None:
+        """Store authoritative conversation snapshot without losing rich metadata."""
+        state_payload = self.snapshot_state()
+        await ctx.set_executor_state(state_payload)
+
    def _snapshot_pattern_metadata(self) -> dict[str, Any]:
        """Serialize pattern-specific state.

@@ -484,7 +492,6 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
            }
        return {}

-    @override
    def _restore_pattern_metadata(self, metadata: dict[str, Any]) -> None:
        """Restore pattern-specific state.

@@ -496,6 +503,17 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
        if self._return_to_previous and "current_agent_id" in metadata:
            self._current_agent_id = metadata["current_agent_id"]

+    def _restore_conversation_from_state(self, state: Mapping[str, Any]) -> list[ChatMessage]:
+        """Rehydrate the coordinator's conversation history from checkpointed state.
+
+        DEPRECATED: Use restore_state() instead. Kept for backward compatibility.
+        """
+        from ._orchestration_state import OrchestrationState
+
+        orch_state_dict = {"conversation": state.get("full_conversation", state.get("conversation", []))}
+        temp_state = OrchestrationState.from_dict(orch_state_dict)
+        return list(temp_state.conversation)
+
    def _apply_response_metadata(self, conversation: list[ChatMessage], agent_response: AgentRunResponse) -> None:
        """Merge top-level response metadata into the latest assistant message."""
        if not agent_response.additional_properties:
@@ -45,15 +45,9 @@ from ._workflow import Workflow, WorkflowRunResult
 from ._workflow_context import WorkflowContext

 if sys.version_info >= (3, 11):
-    from typing import Self
+    from typing import Self  # pragma: no cover
 else:
-    from typing_extensions import Self
-
-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-
+    from typing_extensions import Self  # pragma: no cover

 logger = logging.getLogger(__name__)

@@ -679,11 +673,11 @@ class MagenticManagerBase(ABC):
        """Prepare the final answer."""
        ...

-    def on_checkpoint_save(self) -> dict[str, Any]:
+    def snapshot_state(self) -> dict[str, Any]:
        """Serialize runtime state for checkpointing."""
        return {}

-    def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+    def restore_state(self, state: dict[str, Any]) -> None:
        """Restore runtime state from checkpoint data."""
        return

@@ -701,6 +695,22 @@ class StandardMagenticManager(MagenticManagerBase):

    task_ledger: _MagenticTaskLedger | None

+    def snapshot_state(self) -> dict[str, Any]:
+        state = super().snapshot_state()
+        if self.task_ledger is not None:
+            state = dict(state)
+            state["task_ledger"] = self.task_ledger.to_dict()
+        return state
+
+    def restore_state(self, state: dict[str, Any]) -> None:
+        super().restore_state(state)
+        ledger = state.get("task_ledger")
+        if ledger is not None:
+            try:
+                self.task_ledger = _MagenticTaskLedger.from_dict(ledger)
+            except Exception:  # pragma: no cover - defensive
+                logger.warning("Failed to restore manager task ledger from checkpoint state")
+
    def __init__(
        self,
        chat_client: ChatClientProtocol,
@@ -930,22 +940,6 @@ class StandardMagenticManager(MagenticManagerBase):
            author_name=response.author_name or MAGENTIC_MANAGER_NAME,
        )

-    @override
-    def on_checkpoint_save(self) -> dict[str, Any]:
-        state: dict[str, Any] = {}
-        if self.task_ledger is not None:
-            state["task_ledger"] = self.task_ledger.to_dict()
-        return state
-
-    @override
-    def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        ledger = state.get("task_ledger")
-        if ledger is not None:
-            try:
-                self.task_ledger = _MagenticTaskLedger.from_dict(ledger)
-            except Exception:  # pragma: no cover - defensive
-                logger.warning("Failed to restore manager task ledger from checkpoint state")
-

 # endregion Magentic Manager

@@ -1003,6 +997,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
        # Terminal state marker to stop further processing after completion/limits
        self._terminated = False
        # Tracks whether checkpoint state has been applied for this run
+        self._state_restored = False

    def _get_author_name(self) -> str:
        """Get the magentic manager name for orchestrator-generated messages."""
@@ -1041,8 +1036,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
        )
        await ctx.add_event(event)

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
+    def snapshot_state(self) -> dict[str, Any]:
        """Capture current orchestrator state for checkpointing.

        Uses OrchestrationState for structure but maintains Magentic's complex metadata
@@ -1061,16 +1055,14 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
            state["magentic_context"] = self._context.to_dict()
        if self._task_ledger is not None:
            state["task_ledger"] = _message_to_payload(self._task_ledger)
-
-        try:
-            state["manager_state"] = self._manager.on_checkpoint_save()
-        except Exception as exc:
-            logger.warning("Failed to save manager state for checkpoint: %s\nSkipping...", exc)
-
+        manager_state: dict[str, Any] | None = None
+        with contextlib.suppress(Exception):
+            manager_state = self._manager.snapshot_state()
+        if manager_state:
+            state["manager_state"] = manager_state
        return state

-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+    def restore_state(self, state: dict[str, Any]) -> None:
        """Restore orchestrator state from checkpoint.

        Maintains backward compatibility with existing Magentic checkpoints
@@ -1120,7 +1112,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
        manager_state = state.get("manager_state")
        if manager_state is not None:
            try:
-                self._manager.on_checkpoint_restore(manager_state)
+                self._manager.restore_state(manager_state)
            except Exception as exc:  # pragma: no cover
                logger.warning("Failed to restore manager state: %s", exc)

@@ -1150,6 +1142,49 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
        for name, description in expected.items():
            restored[name] = description

+    def _snapshot_pattern_metadata(self) -> dict[str, Any]:
+        """Serialize pattern-specific state.
+
+        Magentic uses custom snapshot_state() instead of base class hooks.
+        This method exists to satisfy the base class contract.
+
+        Returns:
+            Empty dict (Magentic manages its own state)
+        """
+        return {}
+
+    def _restore_pattern_metadata(self, metadata: dict[str, Any]) -> None:
+        """Restore pattern-specific state.
+
+        Magentic uses custom restore_state() instead of base class hooks.
+        This method exists to satisfy the base class contract.
+
+        Args:
+            metadata: Pattern-specific state dict (ignored)
+        """
+        pass
+
+    async def _ensure_state_restored(
+        self,
+        context: WorkflowContext[Any, Any],
+    ) -> None:
+        if self._state_restored and self._context is not None:
+            return
+        state = await context.get_executor_state()
+        if not state:
+            self._state_restored = True
+            return
+        if not isinstance(state, dict):
+            self._state_restored = True
+            return
+        try:
+            self.restore_state(state)
+        except Exception as exc:  # pragma: no cover
+            logger.warning("Magentic Orchestrator: Failed to apply checkpoint state: %s", exc, exc_info=True)
+            raise
+        else:
+            self._state_restored = True
+
    @handler
    async def handle_start_message(
        self,
@@ -1169,7 +1204,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
        )
        if message.messages:
            self._context.chat_history.extend(message.messages)
-
+        self._state_restored = True
        # Non-streaming callback for the orchestrator receipt of the task
        await self._emit_orchestrator_message(context, message.task, ORCH_MSG_KIND_USER_TASK)

@@ -1234,7 +1269,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
        """Handle responses from agents."""
        if getattr(self, "_terminated", False):
            return
-
+        await self._ensure_state_restored(context)
        if self._context is None:
            raise RuntimeError("Magentic Orchestrator: Received response but not initialized")

@@ -1266,7 +1301,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
    ) -> None:
        if getattr(self, "_terminated", False):
            return
-
+        await self._ensure_state_restored(context)
        if self._context is None:
            return

@@ -1601,9 +1636,9 @@ class MagenticAgentExecutor(Executor):
        self._agent = agent
        self._agent_id = agent_id
        self._chat_history: list[ChatMessage] = []
+        self._state_restored = False

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
+    def snapshot_state(self) -> dict[str, Any]:
        """Capture current executor state for checkpointing.

        Returns:
@@ -1615,8 +1650,7 @@ class MagenticAgentExecutor(Executor):
            "chat_history": encode_chat_messages(self._chat_history),
        }

-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+    def restore_state(self, state: dict[str, Any]) -> None:
        """Restore executor state from checkpoint.

        Args:
@@ -1634,6 +1668,24 @@ class MagenticAgentExecutor(Executor):
        else:
            self._chat_history = []

+    async def _ensure_state_restored(self, context: WorkflowContext[Any, Any]) -> None:
+        if self._state_restored and self._chat_history:
+            return
+        state = await context.get_executor_state()
+        if not state:
+            self._state_restored = True
+            return
+        if not isinstance(state, dict):
+            self._state_restored = True
+            return
+        try:
+            self.restore_state(state)
+        except Exception as exc:  # pragma: no cover
+            logger.warning("Agent %s: Failed to apply checkpoint state: %s", self._agent_id, exc, exc_info=True)
+            raise
+        else:
+            self._state_restored = True
+
    @handler
    async def handle_response_message(
        self, message: _MagenticResponseMessage, context: WorkflowContext[_MagenticResponseMessage]
@@ -1641,6 +1693,8 @@ class MagenticAgentExecutor(Executor):
        """Handle response message (task ledger broadcast)."""
        logger.debug("Agent %s: Received response message", self._agent_id)

+        await self._ensure_state_restored(context)
+
        # Check if this message is intended for this agent
        if message.target_agent is not None and message.target_agent != self._agent_id and not message.broadcast:
            # Message is targeted to a different agent, ignore it
@@ -1681,6 +1735,8 @@ class MagenticAgentExecutor(Executor):

        logger.info("Agent %s: Received request to respond", self._agent_id)

+        await self._ensure_state_restored(context)
+
        # Add persona adoption message with appropriate role
        persona_role = self._get_persona_adoption_role()
        persona_msg = ChatMessage(
@@ -1727,6 +1783,7 @@ class MagenticAgentExecutor(Executor):
        """Reset the internal chat history of the agent (internal operation)."""
        logger.debug("Agent %s: Resetting chat history", self._agent_id)
        self._chat_history.clear()
+        self._state_restored = True

    async def _emit_agent_delta_event(
        self,
@@ -11,7 +11,7 @@ from ._checkpoint_encoding import DATACLASS_MARKER, MODEL_MARKER, decode_checkpo
 from ._const import EXECUTOR_STATE_KEY
 from ._edge import EdgeGroup
 from ._edge_runner import EdgeRunner, create_edge_runner
-from ._events import SuperStepCompletedEvent, SuperStepStartedEvent, WorkflowEvent
+from ._events import WorkflowEvent
 from ._executor import Executor
 from ._runner_context import (
    Message,
@@ -92,7 +92,6 @@ class Runner:

            while self._iteration < self._max_iterations:
                logger.info(f"Starting superstep {self._iteration + 1}")
-                yield SuperStepStartedEvent(iteration=self._iteration + 1)

                # Run iteration concurrently with live event streaming: we poll
                # for new events while the iteration coroutine progresses.
@@ -127,9 +126,6 @@ class Runner:
                # Create checkpoint after each superstep iteration
                await self._create_checkpoint_if_enabled(f"superstep_{self._iteration}")

-                yield SuperStepCompletedEvent(iteration=self._iteration)
-
-                # Check for convergence: no more messages to process
                if not await self._ctx.has_messages():
                    break

@@ -187,8 +183,8 @@ class Runner:
            return None

        try:
-            # Snapshot executor states
-            await self._save_executor_states()
+            # Auto-snapshot executor states
+            await self._auto_snapshot_executor_states()
            checkpoint_category = "initial" if checkpoint_type == "after_initial_execution" else "superstep"
            metadata = {
                "superstep": self._iteration,
@@ -207,6 +203,41 @@ class Runner:
            logger.warning(f"Failed to create {checkpoint_type} checkpoint: {e}")
            return None

+    async def _auto_snapshot_executor_states(self) -> None:
+        """Populate executor state by calling snapshot hooks on executors if available.
+
+        TODO(@taochen#1614): this method is potentially problematic if executors also call
+        set_executor_state on the context directly. We should clarify the intended usage
+        pattern for executor state management.
+
+        Convention:
+          - If an executor defines an async or sync method `snapshot_state(self) -> dict`, use it.
+          - Else if it has a plain attribute `state` that is a dict, use that.
+        Only JSON-serializable dicts should be provided by executors.
+        """
+        for exec_id, executor in self._executors.items():
+            state_dict: dict[str, Any] | None = None
+            snapshot = getattr(executor, "snapshot_state", None)
+            try:
+                if callable(snapshot):
+                    maybe = snapshot()
+                    if asyncio.iscoroutine(maybe):  # type: ignore[arg-type]
+                        maybe = await maybe  # type: ignore[assignment]
+                    if isinstance(maybe, dict):
+                        state_dict = maybe  # type: ignore[assignment]
+                else:
+                    state_attr = getattr(executor, "state", None)
+                    if isinstance(state_attr, dict):
+                        state_dict = state_attr  # type: ignore[assignment]
+            except Exception as ex:  # pragma: no cover
+                logger.debug(f"Executor {exec_id} snapshot_state failed: {ex}")
+
+            if state_dict is not None:
+                try:
+                    await self._set_executor_state(exec_id, state_dict)
+                except Exception as ex:  # pragma: no cover
+                    logger.debug(f"Failed to persist state for executor {exec_id}: {ex}")
+
    async def restore_from_checkpoint(
        self,
        checkpoint_id: str,
@@ -269,65 +300,7 @@ class Runner:
            logger.error(f"Failed to restore from checkpoint {checkpoint_id}: {e}")
            return False

-    async def _save_executor_states(self) -> None:
-        """Populate executor state by calling checkpoint hooks on executors.
-
-        Backward compatibility behavior:
-          - If an executor defines an async or sync method `snapshot_state(self) -> dict`, use it.
-          - Else if it has a plain attribute `state` that is a dict, use that.
-
-        Updated behavior:
-          - Executors should implement `on_checkpoint_save(self) -> dict` to provide state.
-
-        This method will try the backward compatibility behavior first; if that does not yield state,
-        it falls back to the updated behavior.
-
-        Only JSON-serializable dicts should be provided by executors.
-        """
-        for exec_id, executor in self._executors.items():
-            state_dict: dict[str, Any] | None = None
-            # Try backward compatibility behavior first
-            # TODO(@taochen): Remove backward compatibility
-            snapshot = getattr(executor, "snapshot_state", None)
-            try:
-                if callable(snapshot):
-                    maybe = snapshot()
-                    if asyncio.iscoroutine(maybe):  # type: ignore[arg-type]
-                        maybe = await maybe  # type: ignore[assignment]
-                    if isinstance(maybe, dict):
-                        state_dict = maybe  # type: ignore[assignment]
-                else:
-                    state_attr = getattr(executor, "state", None)
-                    if isinstance(state_attr, dict):
-                        state_dict = state_attr  # type: ignore[assignment]
-            except Exception as ex:  # pragma: no cover
-                logger.debug(f"Executor {exec_id} snapshot_state failed: {ex}")
-
-            if state_dict is None:
-                # Try the updated behavior only if backward compatibility did not yield state
-                try:
-                    state_dict = await executor.on_checkpoint_save()
-                except Exception as ex:  # pragma: no cover
-                    raise ValueError(f"Executor {exec_id} on_checkpoint_save failed: {ex}") from ex
-
-            try:
-                await self._set_executor_state(exec_id, state_dict)
-            except Exception as ex:  # pragma: no cover
-                logger.debug(f"Failed to persist state for executor {exec_id}: {ex}")
-
    async def _restore_executor_states(self) -> None:
-        """Restore executor state by calling restore hooks on executors.
-
-        Backward compatibility behavior:
-            - If an executor defines an async or sync method `restore_state(self, state: dict)`, use it.
-            - Else, skip restoration for that executor.
-
-        Updated behavior:
-            - Executors should implement `on_checkpoint_restore(self, state: dict)` to restore state.
-
-        This method will try the backward compatibility behavior first; if that does not restore state,
-        it falls back to the updated behavior.
-        """
        has_executor_states = await self._shared_state.has(EXECUTOR_STATE_KEY)
        if not has_executor_states:
            return
@@ -336,18 +309,16 @@ class Runner:
        if not isinstance(executor_states, dict):
            raise ValueError("Executor states in shared state is not a dictionary. Unable to restore.")

-        for executor_id, state in executor_states.items():  # pyright: ignore[reportUnknownVariableType]
+        for executor_id, state in executor_states.items():
            if not isinstance(executor_id, str):
                raise ValueError("Executor ID in executor states is not a string. Unable to restore.")
-            if not isinstance(state, dict) or not all(isinstance(k, str) for k in state):  # pyright: ignore[reportUnknownVariableType]
-                raise ValueError(f"Executor state for {executor_id} is not a dict[str, Any]. Unable to restore.")
+            if not isinstance(state, dict):
+                raise ValueError(f"Executor state for {executor_id} is not a dictionary. Unable to restore.")

            executor = self._executors.get(executor_id)
            if not executor:
                raise ValueError(f"Executor {executor_id} not found during state restoration.")

-            # Try backward compatibility behavior first
-            # TODO(@taochen): Remove backward compatibility
            restored = False
            restore_method = getattr(executor, "restore_state", None)
            try:
@@ -359,14 +330,6 @@ class Runner:
            except Exception as ex:  # pragma: no cover - defensive
                raise ValueError(f"Executor {executor_id} restore_state failed: {ex}") from ex

-            if not restored:
-                # Try the updated behavior only if backward compatibility did not restore
-                try:
-                    await executor.on_checkpoint_restore(state)  # pyright: ignore[reportUnknownArgumentType]
-                    restored = True
-                except Exception as ex:  # pragma: no cover - defensive
-                    raise ValueError(f"Executor {executor_id} on_checkpoint_restore failed: {ex}") from ex
-
            if not restored:
                logger.debug(f"Executor {executor_id} does not support state restoration; skipping.")

@@ -109,9 +109,9 @@ class Workflow(DictConvertible):
    """A graph-based execution engine that orchestrates connected executors.

    ## Overview
-    A workflow executes a directed graph of executors connected via edge groups using a
-    Pregel-like model, running in supersteps until the graph becomes idle. Workflows
-    are created using the WorkflowBuilder class - do not instantiate this class directly.
+    A workflow executes a directed graph of executors connected via edge groups using a Pregel-like model,
+    running in supersteps until the graph becomes idle. Workflows are created using the
+    WorkflowBuilder class - do not instantiate this class directly.

    ## Execution Model
    Executors run in synchronized supersteps where each executor:
@@ -142,10 +142,6 @@ class Workflow(DictConvertible):
    - HIL continuation: Provide `responses` to continue after RequestInfoExecutor requests
    - Runtime checkpointing: Provide `checkpoint_storage` to enable/override checkpointing for this run

-    ## State Management
-    Workflow instances contain states and states are preserved across calls to `run` and `run_stream`.
-    To execute multiple independent runs, create separate Workflow instances via WorkflowBuilder.
-
    ## External Input Requests
    Executors within a workflow can request external input using `ctx.request_info()`:
    1. Executor calls `ctx.request_info()` to request input
@@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any, Generic, Union, cast, get_args, get_origi

 from opentelemetry.propagate import inject
 from opentelemetry.trace import SpanKind
-from typing_extensions import Never, TypeVar, deprecated
+from typing_extensions import Never, TypeVar

 from ..observability import OtelAttr, create_workflow_span
 from ._const import EXECUTOR_STATE_KEY
@@ -410,11 +410,6 @@ class WorkflowContext(Generic[T_Out, T_W_Out]):
        """Get the shared state."""
        return self._shared_state

-    @deprecated(
-        "Override `on_checkpoint_save()` methods instead. "
-        "For cross-executor state sharing, use set_shared_state() instead. "
-        "This API will be removed after 12/01/2025."
-    )
    async def set_executor_state(self, state: dict[str, Any]) -> None:
        """Store executor state in shared state under a reserved key.

@@ -433,11 +428,6 @@ class WorkflowContext(Generic[T_Out, T_W_Out]):
        existing_states[self._executor_id] = state
        await self._shared_state.set(EXECUTOR_STATE_KEY, existing_states)

-    @deprecated(
-        "Override `on_checkpoint_restore()` methods instead. "
-        "For cross-executor state sharing, use get_shared_state() instead. "
-        "This API will be removed after 12/01/2025."
-    )
    async def get_executor_state(self) -> dict[str, Any] | None:
        """Retrieve previously persisted state for this executor, if any."""
        has_existing_states = await self._shared_state.has(EXECUTOR_STATE_KEY)
@@ -1,8 +1,8 @@
 # Copyright (c) Microsoft. All rights reserved.

 import asyncio
+import contextlib
 import logging
-import sys
 import uuid
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
@@ -26,12 +26,6 @@ from ._typing_utils import is_instance_of
 from ._workflow import WorkflowRunResult
 from ._workflow_context import WorkflowContext

-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-
-
 logger = logging.getLogger(__name__)


@@ -187,7 +181,8 @@ class WorkflowExecutor(Executor):

        # Includes all sub-workflow output types
        # Plus SubWorkflowRequestMessage if sub-workflow can make requests
-        output_types = workflow.output_types + [SubWorkflowRequestMessage]  # if applicable
+    output_types = workflow.output_types + [SubWorkflowRequestMessage]  # if applicable
+    ```

    ## Error Handling
    WorkflowExecutor propagates sub-workflow failures:
@@ -226,10 +221,23 @@ class WorkflowExecutor(Executor):

    ### Important Considerations
    **Shared Workflow Instance**: All concurrent executions use the same underlying workflow instance.
-    For proper isolation, ensure that the wrapped workflow and its executors are stateless.
+    For proper isolation, ensure that:
+    - The wrapped workflow and its executors are stateless
+    - Executors use WorkflowContext state management instead of instance variables
+    - Any shared state is managed through WorkflowContext.get_shared_state/set_shared_state

    .. code-block:: python

+        # Good: Stateless executor using context state
+        class StatelessExecutor(Executor):
+            @handler
+            async def process(self, data: str, ctx: WorkflowContext[str]) -> None:
+                # Use context state instead of instance variables
+                state = await ctx.get_executor_state() or {}
+                state["processed"] = data
+                await ctx.set_executor_state(state)
+
+
        # Avoid: Stateful executor with instance variables
        class StatefulExecutor(Executor):
            def __init__(self):
@@ -238,23 +246,23 @@ class WorkflowExecutor(Executor):

    ## Integration with Parent Workflows
    Parent workflows can intercept sub-workflow requests:
-
-    .. code-block:: python
-        class ParentExecutor(Executor):
-            @handler
-            async def handle_subworkflow_request(
-                self,
-                request: SubWorkflowRequestMessage,
-                ctx: WorkflowContext[SubWorkflowResponseMessage],
-            ) -> None:
-                # Handle request locally or forward to external source
-                if self.can_handle_locally(request):
-                    # Send response back to sub-workflow
-                    response = request.create_response(data="local response data")
-                    await ctx.send_message(response, target_id=request.source_executor_id)
-                else:
-                    # Forward to external handler
-                    await ctx.request_info(request.source_event, response_type=request.source_event.response_type)
+    ```python
+    class ParentExecutor(Executor):
+        @handler
+        async def handle_subworkflow_request(
+            self,
+            request: SubWorkflowRequestMessage,
+            ctx: WorkflowContext[SubWorkflowResponseMessage],
+        ) -> None:
+            # Handle request locally or forward to external source
+            if self.can_handle_locally(request):
+                # Send response back to sub-workflow
+                response = request.create_response(data="local response data")
+                await ctx.send_message(response, target_id=request.source_executor_id)
+            else:
+                # Forward to external handler
+                await ctx.request_info(request.source_event, response_type=request.source_event.response_type)
+    ```

    ## Implementation Notes
    - Sub-workflows run to completion before processing their results
@@ -288,6 +296,7 @@ class WorkflowExecutor(Executor):
        self._execution_contexts: dict[str, ExecutionContext] = {}  # execution_id -> ExecutionContext
        # Map request_id to execution_id for response routing
        self._request_to_execution: dict[str, str] = {}  # request_id -> execution_id
+        self._state_loaded: bool = False

    @property
    def input_types(self) -> list[type[Any]]:
@@ -353,6 +362,8 @@ class WorkflowExecutor(Executor):
            input_data: The input data to send to the sub-workflow.
            ctx: The workflow context from the parent.
        """
+        await self._ensure_state_loaded(ctx)
+
        # Create execution context for this sub-workflow run
        execution_id = str(uuid.uuid4())
        execution_context = ExecutionContext(
@@ -394,6 +405,8 @@ class WorkflowExecutor(Executor):
            response: The response to a previous request.
            ctx: The workflow context.
        """
+        await self._ensure_state_loaded(ctx)
+
        # Find the execution context for this request
        original_request = response.source_event
        execution_id = self._request_to_execution.get(original_request.request_id)
@@ -421,6 +434,8 @@ class WorkflowExecutor(Executor):
        # Accumulate the response in this execution's context
        execution_context.collected_responses[original_request.request_id] = response.data

+        await self._persist_execution_state(ctx)
+
        # Check if we have all expected responses for this execution
        if len(execution_context.collected_responses) < execution_context.expected_response_count:
            logger.debug(
@@ -444,20 +459,25 @@ class WorkflowExecutor(Executor):
            if not execution_context.pending_requests:
                del self._execution_contexts[execution_id]

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
-        """Get the current state of the WorkflowExecutor for checkpointing purposes."""
-        return {
-            "execution_contexts": {
-                execution_id: encode_checkpoint_value(execution_context)
-                for execution_id, execution_context in self._execution_contexts.items()
-            },
-            "request_to_execution": dict(self._request_to_execution),
-        }
+    async def _ensure_state_loaded(self, ctx: WorkflowContext[Any]) -> None:
+        if self._state_loaded:
+            return

-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        """Restore the WorkflowExecutor state from a checkpoint snapshot."""
+        state: dict[str, Any] | None = None
+        try:
+            state = await ctx.get_executor_state()
+        except Exception:
+            state = None
+
+        if isinstance(state, dict) and state:
+            with contextlib.suppress(Exception):
+                await self.restore_state(state)
+                self._state_loaded = True
+        else:
+            self._state_loaded = True
+
+    async def restore_state(self, state: dict[str, Any]) -> None:
+        """Restore pending request bookkeeping from a checkpoint snapshot."""
        # Validate the state contains the right keys
        if "execution_contexts" not in state:
            raise KeyError("Missing 'execution_contexts' in WorkflowExecutor state.")
@@ -509,6 +529,23 @@ class WorkflowExecutor(Executor):
            for event in request_info_events
        ])

+        self._state_loaded = True
+
+    async def _persist_execution_state(self, ctx: WorkflowContext) -> None:
+        """Persist the state of the WorkflowExecutor for checkpointing purposes."""
+        state = {
+            "execution_contexts": {
+                execution_id: encode_checkpoint_value(execution_context)
+                for execution_id, execution_context in self._execution_contexts.items()
+            },
+            "request_to_execution": dict(self._request_to_execution),
+        }
+
+        try:
+            await ctx.set_executor_state(state)
+        except Exception as exc:  # pragma: no cover - transport specific
+            logger.warning(f"WorkflowExecutor {self.id} failed to persist state: {exc}")
+
    async def _process_workflow_result(
        self,
        result: WorkflowRunResult,
@@ -598,3 +635,5 @@ class WorkflowExecutor(Executor):
            )
        else:
            raise RuntimeError(f"Unexpected workflow run state: {workflow_run_state}")
+
+        await self._persist_execution_state(ctx)
@@ -158,8 +158,8 @@ async def test_agent_executor_checkpoint_stores_and_restores_state() -> None:
    assert thread_messages[1].text == "Initial response 1"


-async def test_agent_executor_save_and_restore_state_directly() -> None:
-    """Test AgentExecutor's on_checkpoint_save and on_checkpoint_restore methods directly."""
+async def test_agent_executor_snapshot_and_restore_state_directly() -> None:
+    """Test AgentExecutor's snapshot_state and restore_state methods directly."""
    # Create agent with thread containing messages
    agent = _CountingAgent(id="direct_test_agent", name="DirectTestAgent")
    thread = AgentThread(message_store=ChatMessageStore())
@@ -182,7 +182,7 @@ async def test_agent_executor_save_and_restore_state_directly() -> None:
    executor._cache = list(cache_messages)  # type: ignore[reportPrivateUsage]

    # Snapshot the state
-    state = await executor.on_checkpoint_save()
+    state = await executor.snapshot_state()  # type: ignore[reportUnknownMemberType]

    # Verify snapshot contains both cache and thread
    assert "cache" in state
@@ -206,7 +206,7 @@ async def test_agent_executor_save_and_restore_state_directly() -> None:
    assert len(initial_thread_msgs) == 0

    # Restore state
-    await new_executor.on_checkpoint_restore(state)
+    await new_executor.restore_state(state)  # type: ignore[reportUnknownMemberType]

    # Verify cache is restored
    restored_cache = new_executor._cache  # type: ignore[reportPrivateUsage]
@@ -288,6 +288,57 @@ def test_build_fails_without_participants():
        HandoffBuilder().build()


+async def test_multiple_runs_dont_leak_conversation():
+    """Verify that running the same workflow multiple times doesn't leak conversation history."""
+    triage = _RecordingAgent(name="triage", handoff_to="specialist")
+    specialist = _RecordingAgent(name="specialist")
+
+    workflow = (
+        HandoffBuilder(participants=[triage, specialist])
+        .set_coordinator("triage")
+        .with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 2)
+        .build()
+    )
+
+    # First run
+    events = await _drain(workflow.run_stream("First run message"))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Second message"}))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "First run should emit output"
+
+    first_run_conversation = outputs[-1].data
+    assert isinstance(first_run_conversation, list)
+    first_run_conv_list = cast(list[ChatMessage], first_run_conversation)
+    first_run_user_messages = [msg for msg in first_run_conv_list if msg.role == Role.USER]
+    assert len(first_run_user_messages) == 2
+    assert any("First run message" in msg.text for msg in first_run_user_messages if msg.text)
+
+    # Second run - should start fresh, not include first run's messages
+    triage.calls.clear()
+    specialist.calls.clear()
+
+    events = await _drain(workflow.run_stream("Second run different message"))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Another message"}))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Second run should emit output"
+
+    second_run_conversation = outputs[-1].data
+    assert isinstance(second_run_conversation, list)
+    second_run_conv_list = cast(list[ChatMessage], second_run_conversation)
+    second_run_user_messages = [msg for msg in second_run_conv_list if msg.role == Role.USER]
+    assert len(second_run_user_messages) == 2, (
+        "Second run should have exactly 2 user messages, not accumulate first run"
+    )
+    assert any("Second run different message" in msg.text for msg in second_run_user_messages if msg.text)
+    assert not any("First run message" in msg.text for msg in second_run_user_messages if msg.text), (
+        "Second run should NOT contain first run's messages"
+    )
+
+
 async def test_handoff_async_termination_condition() -> None:
    """Test that async termination conditions work correctly."""
    termination_call_count = 0
@@ -534,7 +585,7 @@ async def test_return_to_previous_state_serialization():
    coordinator._current_agent_id = "specialist_a"  # type: ignore[reportPrivateUsage]

    # Snapshot the state
-    state = await coordinator.on_checkpoint_save()
+    state = coordinator.snapshot_state()

    # Verify pattern metadata includes current_agent_id
    assert "metadata" in state
@@ -552,7 +603,7 @@ async def test_return_to_previous_state_serialization():
    )

    # Restore state
-    await coordinator2.on_checkpoint_restore(state)
+    coordinator2.restore_state(state)

    # Verify current_agent_id was restored
    assert coordinator2._current_agent_id == "specialist_a", "Current agent should be restored from checkpoint"  # type: ignore[reportPrivateUsage]
@@ -1,6 +1,5 @@
 # Copyright (c) Microsoft. All rights reserved.

-import sys
 from collections.abc import AsyncIterable
 from dataclasses import dataclass
 from typing import Any, cast
@@ -43,11 +42,6 @@ from agent_framework._workflows._magentic import (  # type: ignore[reportPrivate
    _MagenticStartMessage,  # type: ignore
 )

-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-

 def test_magentic_start_message_from_string():
    msg = _MagenticStartMessage.from_string("Do the thing")
@@ -107,9 +101,8 @@ class FakeManager(MagenticManagerBase):
    next_speaker_name: str = "agentA"
    instruction_text: str = "Proceed with step 1"

-    @override
-    def on_checkpoint_save(self) -> dict[str, Any]:
-        state = super().on_checkpoint_save()
+    def snapshot_state(self) -> dict[str, Any]:
+        state = super().snapshot_state()
        if self.task_ledger is not None:
            state = dict(state)
            state["task_ledger"] = {
@@ -118,9 +111,8 @@ class FakeManager(MagenticManagerBase):
            }
        return state

-    @override
-    def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        super().on_checkpoint_restore(state)
+    def restore_state(self, state: dict[str, Any]) -> None:
+        super().restore_state(state)
        ledger_state = state.get("task_ledger")
        if isinstance(ledger_state, dict):
            ledger_dict = cast(dict[str, Any], ledger_state)
@@ -193,6 +185,7 @@ async def test_standard_manager_progress_ledger_and_fallback():
    assert ledger2.is_request_satisfied.answer is False


+@pytest.mark.skip(reason="Response handling refactored - responses no longer passed to run_stream()")
 async def test_magentic_workflow_plan_review_approval_to_completion():
    manager = FakeManager(max_round_count=10)
    wf = (
@@ -211,7 +204,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():

    completed = False
    output: ChatMessage | None = None
-    async for ev in wf.send_responses_streaming(
+    async for ev in wf.run_stream(
        responses={req_event.request_id: MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)}
    ):
        if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
@@ -225,6 +218,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
    assert isinstance(output, ChatMessage)


+@pytest.mark.skip(reason="Response handling refactored - responses no longer passed to run_stream()")
 async def test_magentic_plan_review_approve_with_comments_replans_and_proceeds():
    class CountingManager(FakeManager):
        # Declare as a model field so assignment is allowed under Pydantic
@@ -256,7 +250,7 @@ async def test_magentic_plan_review_approve_with_comments_replans_and_proceeds()
    # Reply APPROVE with comments (no edited text). Expect one replan and no second review round.
    saw_second_review = False
    completed = False
-    async for ev in wf.send_responses_streaming(
+    async for ev in wf.run_stream(
        responses={
            req_event.request_id: MagenticPlanReviewReply(
                decision=MagenticPlanReviewDecision.APPROVE,
@@ -304,6 +298,7 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
    assert data.role == Role.ASSISTANT


+@pytest.mark.skip(reason="Response handling refactored - send_responses_streaming no longer exists")
 async def test_magentic_checkpoint_resume_round_trip():
    storage = InMemoryCheckpointStorage()

@@ -374,7 +369,7 @@ class _DummyExec(Executor):
        pass


-async def test_magentic_agent_executor_on_checkpoint_save_and_restore_roundtrip():
+def test_magentic_agent_executor_snapshot_roundtrip():
    backing_executor = _DummyExec("backing")
    agent_exec = MagenticAgentExecutor(backing_executor, "agentA")
    agent_exec._chat_history.extend([  # type: ignore[reportPrivateUsage]
@@ -382,10 +377,10 @@ async def test_magentic_agent_executor_on_checkpoint_save_and_restore_roundtrip(
        ChatMessage(role=Role.ASSISTANT, text="world", author_name="agentA"),
    ])

-    state = await agent_exec.on_checkpoint_save()
+    state = agent_exec.snapshot_state()

    restored_executor = MagenticAgentExecutor(_DummyExec("backing2"), "agentA")
-    await restored_executor.on_checkpoint_restore(state)
+    restored_executor.restore_state(state)

    assert len(restored_executor._chat_history) == 2  # type: ignore[reportPrivateUsage]
    assert restored_executor._chat_history[0].text == "hello"  # type: ignore[reportPrivateUsage]
@@ -199,10 +199,7 @@ async def test_fan_out():

    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
    # executor_b will also emit a WorkflowOutputEvent (no WorkflowCompletedEvent anymore)
-    # Each superstep will emit also emit a WorkflowStartedEvent and WorkflowCompletedEvent
-    # This workflow will converge in 2 supersteps because executor_c will send one more message
-    # after executor_b completes
-    assert len(events) == 11
+    assert len(events) == 7

    assert events.get_final_state() == WorkflowRunState.IDLE
    outputs = events.get_outputs()
@@ -223,9 +220,7 @@ async def test_fan_out_multiple_completed_events():

    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
    # executor_b and executor_c will also emit a WorkflowOutputEvent (no WorkflowCompletedEvent anymore)
-    # Each superstep will emit also emit a WorkflowStartedEvent and WorkflowCompletedEvent
-    # This workflow will converge in 1 superstep because executor_a and executor_b will not send further messages
-    assert len(events) == 10
+    assert len(events) == 8

    # Multiple outputs are expected from both executors
    outputs = events.get_outputs()
@@ -251,8 +246,7 @@ async def test_fan_in():

    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
    # aggregator will also emit a WorkflowOutputEvent (no WorkflowCompletedEvent anymore)
-    # Each superstep will emit also emit a WorkflowStartedEvent and WorkflowCompletedEvent
-    assert len(events) == 13
+    assert len(events) == 9

    assert events.get_final_state() == WorkflowRunState.IDLE
    outputs = events.get_outputs()
@@ -37,14 +37,10 @@ class WorkflowHILRequest:
 class WorkflowTestExecutor(Executor):
    """Test executor with HIL."""

-    def __init__(self, id: str) -> None:
-        super().__init__(id=id)
-        self._data_value: str | None = None
-
    @handler
    async def process(self, data: WorkflowTestData, ctx: WorkflowContext) -> None:
        """Process data and request approval."""
-        self._data_value = data.value
+        await ctx.set_executor_state({"data_value": data.value})

        # Request HIL (checkpoint created here)
        await ctx.request_info(request_data=WorkflowHILRequest(question=f"Approve {data.value}?"), response_type=str)
@@ -54,7 +50,8 @@ class WorkflowTestExecutor(Executor):
        self, original_request: WorkflowHILRequest, response: str, ctx: WorkflowContext[str]
    ) -> None:
        """Handle HIL response."""
-        value = self._data_value or ""
+        state = await ctx.get_executor_state() or {}
+        value = state.get("data_value", "")
        await ctx.send_message(f"{value}_approved" if response.lower() == "yes" else f"{value}_rejected")


@@ -17,11 +17,11 @@ This directory contains samples demonstrating the capabilities of Microsoft Agen
 | [`getting_started/agents/anthropic/anthropic_basic.py`](./getting_started/agents/anthropic/anthropic_basic.py) | Agent with Anthropic Client |
 | [`getting_started/agents/anthropic/anthropic_advanced.py`](./getting_started/agents/anthropic/anthropic_advanced.py) | Advanced sample with `thinking` and hosted tools. |

-### Azure AI (based on `azure-ai-agents` V1 package)
+### Azure AI

 | File | Description |
 |------|-------------|
-| [`getting_started/agents/azure_ai_agent/azure_ai_basic.py`](./getting_started/agents/azure_ai_agent/azure_ai_basic.py) | Azure AI Agent Basic Example |
+| [`getting_started/agents/azure_ai_agent/azure_ai_basic.py`](./getting_started/agents/azure_ai/azure_ai_basic.py) | Azure AI Agent Basic Example |
 | [`getting_started/agents/azure_ai_agent/azure_ai_with_azure_ai_search.py`](./getting_started/agents/azure_ai_agent/azure_ai_with_azure_ai_search.py) | Azure AI Agent with Azure AI Search Example |
 | [`getting_started/agents/azure_ai_agent/azure_ai_with_bing_grounding.py`](./getting_started/agents/azure_ai_agent/azure_ai_with_bing_grounding.py) | Azure AI agent with Bing Grounding search for real-time web information |
 | [`getting_started/agents/azure_ai_agent/azure_ai_with_code_interpreter.py`](./getting_started/agents/azure_ai_agent/azure_ai_with_code_interpreter.py) | Azure AI Agent with Code Interpreter Example |
@@ -36,28 +36,6 @@ This directory contains samples demonstrating the capabilities of Microsoft Agen
 | [`getting_started/agents/azure_ai_agent/azure_ai_with_openapi_tools.py`](./getting_started/agents/azure_ai_agent/azure_ai_with_openapi_tools.py) | Azure AI agent with OpenAPI tools |
 | [`getting_started/agents/azure_ai_agent/azure_ai_with_thread.py`](./getting_started/agents/azure_ai_agent/azure_ai_with_thread.py) | Azure AI Agent with Thread Management Example |

-### Azure AI (based on `azure-ai-projects` V2 package)
-
-| File | Description |
-|------|-------------|
-| [`getting_started/agents/azure_ai/azure_ai_basic.py`](./getting_started/agents/azure_ai/azure_ai_basic.py) | Azure AI Agent Basic Example |
-| [`getting_started/agents/azure_ai/azure_ai_use_latest_version.py`](./getting_started/agents/azure_ai/azure_ai_use_latest_version.py) | Azure AI Agent latest version reuse example |
-| [`getting_started/agents/azure_ai/azure_ai_with_azure_ai_search.py`](./getting_started/agents/azure_ai/azure_ai_with_azure_ai_search.py) | Azure AI Agent with Azure AI Search Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_bing_grounding.py`](./getting_started/agents/azure_ai/azure_ai_with_bing_grounding.py) | Azure AI Agent with Bing Grounding Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_bing_custom_search.py`](./getting_started/agents/azure_ai/azure_ai_with_bing_custom_search.py) | Azure AI Agent with Bing Custom Search Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_browser_automation.py`](./getting_started/agents/azure_ai/azure_ai_with_browser_automation.py) | Azure AI Agent with Browser Automation Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_code_interpreter.py`](./getting_started/agents/azure_ai/azure_ai_with_code_interpreter.py) | Azure AI Agent with Code Interpreter Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_existing_agent.py`](./getting_started/agents/azure_ai/azure_ai_with_existing_agent.py) | Azure AI Agent with Existing Agent Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_existing_conversation.py`](./getting_started/agents/azure_ai/azure_ai_with_existing_conversation.py) | Azure AI Agent with Existing Conversation Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_explicit_settings.py`](./getting_started/agents/azure_ai/azure_ai_with_explicit_settings.py) | Azure AI Agent with Explicit Settings Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_file_search.py`](./getting_started/agents/azure_ai/azure_ai_with_file_search.py) | Azure AI Agent with File Search Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_hosted_mcp.py`](./getting_started/agents/azure_ai/azure_ai_with_hosted_mcp.py) | Azure AI Agent with Hosted MCP Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_response_format.py`](./getting_started/agents/azure_ai/azure_ai_with_response_format.py) | Azure AI Agent with Structured Output Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_thread.py`](./getting_started/agents/azure_ai/azure_ai_with_thread.py) | Azure AI Agent with Thread Management Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_image_generation.py`](./getting_started/agents/azure_ai/azure_ai_with_image_generation.py) | Azure AI Agent with Image Generation Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_microsoft_fabric.py`](./getting_started/agents/azure_ai/azure_ai_with_microsoft_fabric.py) | Azure AI Agent with Microsoft Fabric Example |
-| [`getting_started/agents/azure_ai/azure_ai_with_web_search.py`](./getting_started/agents/azure_ai/azure_ai_with_web_search.py) | Azure AI Agent with Web Search Example |
-
 ### Azure OpenAI

 | File | Description |
@@ -0,0 +1,2 @@
+OPENAI_CHAT_MODEL_ID="gpt-4o-2024-08-06"
+OPENAI_API_KEY="your-openai-api-key"
@@ -0,0 +1,47 @@
+# Hosted Agents with Hosted MCP Demo
+
+This demo showcases an agent that has access to a MCP tool that can talk to the Microsoft Learn documentation platform, hosted as an agent endpoint running locally in a Docker container.
+
+## What the Project Does
+
+This project demonstrates how to:
+
+- Create an agent with a hosted MCP tool using the Agent Framework
+- Host the agent as an agent endpoint running in a Docker container
+
+## Prerequisites
+
+- OpenAI API access and credentials
+- Required environment variables (see Configuration section)
+
+## Configuration
+
+Follow the `.env.example` file to set up the necessary environment variables for OpenAI.
+
+## Docker Deployment
+
+Build and run using Docker:
+
+```bash
+# Build the Docker image
+docker build -t hosted-agent-mcp .
+
+# Run the container
+docker run -p 8088:8088 hosted-agent-mcp
+```
+
+> If you update the environment variables in the `.env` file or change the code or the dockerfile, make sure to rebuild the Docker image to apply the changes.
+
+## Testing the Agent
+
+Once the agent is running, you can test it by sending queries that contain the trigger keywords. For example:
+
+```bash
+curl -sS -H "Content-Type: application/json" -X POST http://localhost:8088/responses -d '{"input": "How to create an Azure storage account using az cli?","stream":false}'
+```
+
+Expected response:
+
+```bash
+{"object":"response","metadata":{},"agent":null,"conversation":{"id":"conv_6Y7osWAQ1ASyUZ7Ze0LL6dgPubmQv52jHb7G9QDqpV5yakc3ay"},"type":"message","role":"assistant","temperature":1.0,"top_p":1.0,"user":"","id":"resp_Vfd6mdmnmTZ2RNirwfldfqldWLhaxD6fO2UkXsVUg1jYJgftL9","created_at":1763075575,"output":[{"id":"msg_6Y7osWAQ1ASyUZ7Ze0PwiK2V4Bb7NOPaaEpQoBvFRZ5h6OfW4u","type":"message","status":"completed","role":"assistant","content":[{"type":"output_text","text":"To create an Azure Storage account using the Azure CLI, you'll need to follow these steps:\n\n1. **Install Azure CLI**: Make sure the Azure CLI is installed on your machine. You can download it from [here](https://docs.microsoft.com/en-us/cli/azure/install-azure-cli).\n\n2. **Log in to Azure**: Open your terminal or command prompt and use the following command to log in to your Azure account:\n\n    ```bash\n    az login\n    ```\n\n   This command will open a web browser where you can log in with your Azure account credentials. If you're using a service principal, you would use `az login --service-principal ...` with the appropriate parameters.\n\n3. **Select the Subscription**: If you have multiple Azure subscriptions, set the default subscription that you want to use:\n\n    ```bash\n    az account set --subscription \"Your Subscription Name\"\n    ```\n\n4. **Create a Resource Group**: If you don’t already have a resource group, create one using:\n\n    ```bash\n    az group create --name myResourceGroup --location eastus\n    ```\n\n   Replace `myResourceGroup` and `eastus` with your desired resource group name and location.\n\n5. **Create the Storage Account**: Use the following command to create the storage account:\n\n    ```bash\n    az storage account create --name mystorageaccount --resource-group myResourceGroup --location eastus --sku Standard_LRS\n    ```\n\n   Replace `mystorageaccount` with a unique name for your storage account. The storage account name must be between 3 and 24 characters in length, and may contain numbers and lowercase letters only. You can also choose other `--sku` options like `Standard_GRS`, `Standard_RAGRS`, `Standard_ZRS`, `Premium_LRS`, based on your redundancy and performance needs.\n\nBy following these steps, you'll create a new Azure Storage account in the specified resource group and location with the specified SKU.","annotations":[],"logprobs":[]}]}],"parallel_tool_calls":true,"status":"completed"}
+```
@@ -1,30 +0,0 @@
-# Unique identifier/name for this agent
-name: agent-with-hosted-mcp
-# Brief description of what this agent does
-description: >
-  An AI agent that uses Azure OpenAI with a Hosted Model Context Protocol (MCP) server.
-  The agent answers questions by searching Microsoft Learn documentation using MCP tools.
-metadata:
-  # Categorization tags for organizing and discovering agents
-  authors:
-    - Microsoft Agent Framework Team
-  tags:
-    - Azure AI AgentServer
-    - Microsoft Agent Framework
-    - Model Context Protocol
-    - MCP
-template:
-  name: agent-with-hosted-mcp
-  # The type of agent - "hosted" for HOBO, "container" for COBO
-  kind: hosted
-  protocols:
-    - protocol: responses
-  environment_variables:
-    - name: AZURE_OPENAI_ENDPOINT
-      value: ${AZURE_OPENAI_ENDPOINT}
-    - name: AZURE_OPENAI_CHAT_DEPLOYMENT_NAME
-      value: "{{chat}}"
-resources:
-  - kind: model
-    id: gpt-4o-mini
-    name: chat
@@ -1,14 +1,14 @@
 # Copyright (c) Microsoft. All rights reserved.

+
 from agent_framework import HostedMCPTool
-from agent_framework.azure import AzureOpenAIChatClient
+from agent_framework.openai import OpenAIChatClient
 from azure.ai.agentserver.agentframework import from_agent_framework  # pyright: ignore[reportUnknownVariableType]
-from azure.identity import DefaultAzureCredential


 def main():
-    # Create an Agent using the Azure OpenAI Chat Client with a MCP Tool that connects to Microsoft Learn MCP
-    agent = AzureOpenAIChatClient(credential=DefaultAzureCredential()).create_agent(
+    # Create an Agent using the OpenAI Chat Client with a MCP Tool that connects to Microsoft Learn MCP
+    agent = OpenAIChatClient().create_agent(
        name="DocsAgent",
        instructions="You are a helpful assistant that can help with microsoft documentation questions.",
        tools=HostedMCPTool(
@@ -0,0 +1,2 @@
+OPENAI_CHAT_MODEL_ID="gpt-4o-2024-08-06"
+OPENAI_API_KEY="your-openai-api-key"
@@ -0,0 +1,54 @@
+# Hosted Agents with Text Search RAG Demo
+
+This demo showcases an agent that uses Retrieval-Augmented Generation (RAG) with text search capabilities that will be hosted as an agent endpoint running locally in a Docker container.
+
+## What the Project Does
+
+This project demonstrates how to:
+
+- Build a customer support agent using the Agent Framework
+- Implement a custom `TextSearchContextProvider` that simulates document retrieval
+- Host the agent as an agent endpoint running in a Docker container
+
+The agent responds to customer inquiries about:
+
+- **Return & Refund Policies** - Triggered by keywords: "return", "refund"
+- **Shipping Information** - Triggered by keyword: "shipping"
+- **Product Care Instructions** - Triggered by keywords: "tent", "fabric"
+
+## Prerequisites
+
+- OpenAI API access and credentials
+- Required environment variables (see Configuration section)
+
+## Configuration
+
+Follow the `.env.example` file to set up the necessary environment variables for OpenAI.
+
+## Docker Deployment
+
+Build and run using Docker:
+
+```bash
+# Build the Docker image
+docker build -t hosted-agent-rag .
+
+# Run the container
+docker run -p 8088:8088 hosted-agent-rag
+```
+
+> If you update the environment variables in the `.env` file or change the code or the dockerfile, make sure to rebuild the Docker image to apply the changes.
+
+## Testing the Agent
+
+Once the agent is running, you can test it by sending queries that contain the trigger keywords. For example:
+
+```bash
+curl -sS -H "Content-Type: application/json" -X POST http://localhost:8088/responses -d '{"input": "What is the return policy","stream":false}'
+```
+
+Expected response:
+
+```bash
+{"object":"response","metadata":{},"agent":null,"conversation":{"id":"conv_2GbSxDpJJ89B6N4FQkKhrHaz78Hjtxy9b30JEPuY9YFjJM0uw3"},"type":"message","role":"assistant","temperature":1.0,"top_p":1.0,"user":"","id":"resp_Bvffxq0iIzlVkx2I8x7hV4fglm9RBPWfMCpNtEpDT6ciV2IG6z","created_at":1763071467,"output":[{"id":"msg_2GbSxDpJJ89B6N4FQknLsnxkwwFS2FULJqRV9jMey2BOXljqUz","type":"message","status":"completed","role":"assistant","content":[{"type":"output_text","text":"As of the most recent update, Contoso Outdoors' return policy allows customers to return products within 30 days of purchase for a full refund or exchange, provided the items are in their original condition and packaging. However, make sure to check your purchase receipt or the company's website for the most updated and specific details, as policies can vary by location and may change over time.","annotations":[],"logprobs":[]}]}],"parallel_tool_calls":true,"status":"completed"}
+```
@@ -1,33 +0,0 @@
-# Unique identifier/name for this agent
-name: agent-with-text-search-rag
-# Brief description of what this agent does
-description: >
-  An AI agent that uses a ContextProvider for retrieval augmented generation (RAG) capabilities.
-  The agent runs searches against an external knowledge base before each model invocation and
-  injects the results into the model context. It can answer questions about Contoso Outdoors
-  policies and products, including return policies, refunds, shipping options, and product care
-  instructions such as tent maintenance.
-metadata:
-  # Categorization tags for organizing and discovering agents
-  authors:
-    - Microsoft Agent Framework Team
-  tags:
-    - Azure AI AgentServer
-    - Microsoft Agent Framework
-    - Retrieval-Augmented Generation
-    - RAG
-template:
-  name: agent-with-text-search-rag
-  # The type of agent - "hosted" for HOBO, "container" for COBO
-  kind: hosted
-  protocols:
-    - protocol: responses
-  environment_variables:
-    - name: AZURE_OPENAI_ENDPOINT
-      value: ${AZURE_OPENAI_ENDPOINT}
-    - name: AZURE_OPENAI_CHAT_DEPLOYMENT_NAME
-      value: "{{chat}}"
-resources:
-  - kind: model
-    id: gpt-4o-mini
-    name: chat
@@ -7,9 +7,8 @@ from dataclasses import dataclass
 from typing import Any

 from agent_framework import ChatMessage, Context, ContextProvider, Role
-from agent_framework.azure import AzureOpenAIChatClient
+from agent_framework.openai import OpenAIChatClient
 from azure.ai.agentserver.agentframework import from_agent_framework  # pyright: ignore[reportUnknownVariableType]
-from azure.identity import DefaultAzureCredential

 if sys.version_info >= (3, 12):
    from typing import override
@@ -92,8 +91,8 @@ class TextSearchContextProvider(ContextProvider):


 def main():
-    # Create an Agent using the Azure OpenAI Chat Client
-    agent = AzureOpenAIChatClient(credential=DefaultAzureCredential()).create_agent(
+    # Create an Agent using the OpenAI Chat Client
+    agent = OpenAIChatClient().create_agent(
        name="SupportSpecialist",
        instructions=(
            "You are a helpful support specialist for Contoso Outdoors. "
@@ -0,0 +1,2 @@
+OPENAI_CHAT_MODEL_ID="gpt-4o-2024-08-06"
+OPENAI_API_KEY="your-openai-api-key"
@@ -0,0 +1,49 @@
+# Hosted Workflow Agents Demo
+
+This demo showcases an agent that is backed by a workflow of multiple agents running concurrently, hosted as an agent endpoint in a Docker container.
+
+## What the Project Does
+
+This project demonstrates how to:
+
+- Build a workflow of agents using the Agent Framework
+- Host the workflow agent as an agent endpoint running in a Docker container
+
+The agent responds to product launch strategy inquiries by concurrently leveraging insights from three specialized agents:
+
+- **Researcher Agent** - Provides market research insights
+- **Marketer Agent** - Crafts marketing value propositions and messaging
+- **Legal Agent** - Reviews for compliance and legal considerations
+
+## Prerequisites
+
+- OpenAI API access and credentials
+- Required environment variables (see Configuration section)
+
+## Configuration
+
+Follow the `.env.example` file to set up the necessary environment variables for OpenAI.
+
+## Docker Deployment
+
+Build and run using Docker:
+
+```bash
+# Build the Docker image
+docker build -t hosted-agent-workflow .
+
+# Run the container
+docker run -p 8088:8088 hosted-agent-workflow
+```
+
+> If you update the environment variables in the `.env` file or change the code or the dockerfile, make sure to rebuild the Docker image to apply the changes.
+
+## Testing the Agent
+
+Once the agent is running, you can test it by sending queries that contain the trigger keywords. For example:
+
+```bash
+curl -sS -H "Content-Type: application/json" -X POST http://localhost:8088/responses -d '{"input": "We are launching a new budget-friendly electric bike for urban commuters.","stream":false}'
+```
+
+> Expected response is not shown here for brevity. The response will include insights from the researcher, marketer, and legal agents based on the input prompt.
@@ -1,28 +0,0 @@
-# Unique identifier/name for this agent
-name: agents-in-workflow
-# Brief description of what this agent does
-description: >
-  A workflow agent that responds to product launch strategy inquiries by concurrently leveraging insights from three specialized agents.
-metadata:
-  # Categorization tags for organizing and discovering agents
-  authors:
-    - Microsoft Agent Framework Team
-  tags:
-    - Azure AI AgentServer
-    - Microsoft Agent Framework
-    - Workflows
-template:
-  name: agents-in-workflow
-  # The type of agent - "hosted" for HOBO, "container" for COBO
-  kind: hosted
-  protocols:
-    - protocol: responses
-  environment_variables:
-    - name: AZURE_OPENAI_ENDPOINT
-      value: ${AZURE_OPENAI_ENDPOINT}
-    - name: AZURE_OPENAI_CHAT_DEPLOYMENT_NAME
-      value: "{{chat}}"
-resources:
-  - kind: model
-    id: gpt-4o-mini
-    name: chat
@@ -1,28 +1,27 @@
 # Copyright (c) Microsoft. All rights reserved.

 from agent_framework import ConcurrentBuilder
-from agent_framework.azure import AzureOpenAIChatClient
-from azure.ai.agentserver.agentframework import from_agent_framework
-from azure.identity import DefaultAzureCredential  # pyright: ignore[reportUnknownVariableType]
+from agent_framework.openai import OpenAIChatClient
+from azure.ai.agentserver.agentframework import from_agent_framework  # pyright: ignore[reportUnknownVariableType]


 def main():
    # Create agents
-    researcher = AzureOpenAIChatClient(credential=DefaultAzureCredential()).create_agent(
+    researcher = OpenAIChatClient().create_agent(
        instructions=(
            "You're an expert market and product researcher. "
            "Given a prompt, provide concise, factual insights, opportunities, and risks."
        ),
        name="researcher",
    )
-    marketer = AzureOpenAIChatClient(credential=DefaultAzureCredential()).create_agent(
+    marketer = OpenAIChatClient().create_agent(
        instructions=(
            "You're a creative marketing strategist. "
            "Craft compelling value propositions and target messaging aligned to the prompt."
        ),
        name="marketer",
    )
-    legal = AzureOpenAIChatClient(credential=DefaultAzureCredential()).create_agent(
+    legal = OpenAIChatClient().create_agent(
        instructions=(
            "You're a cautious legal/compliance reviewer. "
            "Highlight constraints, disclaimers, and policy concerns based on the prompt."
@@ -8,8 +8,7 @@ This folder contains examples demonstrating how to create and use agents with di

 | Folder | Description |
 |--------|-------------|
-| **[`azure_ai_agent/`](azure_ai_agent/)** | Create agents using Azure AI Agent Service (based on `azure-ai-agents` V1 package) including function tools, code interpreter, MCP integration, thread management, and more. |
-| **[`azure_ai/`](azure_ai/)** | Create agents using Azure AI Agent Service (based on `azure-ai-projects` [V2](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-projects/CHANGELOG.md#200b1-2025-11-11) package) including function tools, code interpreter, MCP integration, thread management, and more. |
+| **[`azure_ai/`](azure_ai/)** | Create agents using Azure AI Foundry Agent Service with various tools including function tools, code interpreter, MCP integration, and thread management |

 ### Microsoft Copilot Studio Examples

@@ -8,7 +8,6 @@ This folder contains examples demonstrating how to use Anthropic's Claude models
 |------|-------------|
 | [`anthropic_basic.py`](anthropic_basic.py) | Demonstrates how to setup a simple agent using the AnthropicClient, with both streaming and non-streaming responses. |
 | [`anthropic_advanced.py`](anthropic_advanced.py) | Shows advanced usage of the AnthropicClient, including hosted tools and `thinking`. |
-| [`anthropic_skills.py`](anthropic_skills.py) | Illustrates how to use Anthropic-managed Skills with an agent, including the Code Interpreter tool and file generation and saving. |

 ## Environment Variables

@@ -1,88 +0,0 @@
-# Copyright (c) Microsoft. All rights reserved.
-
-import asyncio
-import logging
-from pathlib import Path
-
-from agent_framework import HostedCodeInterpreterTool, HostedFileContent
-from agent_framework.anthropic import AnthropicClient
-
-logger = logging.getLogger(__name__)
-"""
-Anthropic Skills Agent Example
-
-This sample demonstrates using Anthropic with:
- Listing and using Anthropic-managed Skills.
- One approach to add additional beta flags.
-    You can also set additonal_chat_options with "additional_beta_flags" per request.
- Creating an agent with the Code Interpreter tool and a Skill.
- Catching and downloading generated files from the agent.
-"""
-
-
-async def main() -> None:
-    """Example of streaming response (get results as they are generated)."""
-    client = AnthropicClient(additional_beta_flags=["skills-2025-10-02"])
-
-    # List Anthropic-managed Skills
-    skills = await client.anthropic_client.beta.skills.list(source="anthropic", betas=["skills-2025-10-02"])
-    for skill in skills.data:
-        print(f"{skill.source}: {skill.id} (version: {skill.latest_version})")
-
-    # Create a agent with the pptx skill enabled
-    # Skills also need the code interpreter tool to function
-    agent = client.create_agent(
-        name="DocsAgent",
-        instructions="You are a helpful agent for creating powerpoint presentations.",
-        tools=HostedCodeInterpreterTool(),
-        max_tokens=20000,
-        additional_chat_options={
-            "thinking": {"type": "enabled", "budget_tokens": 10000},
-            "container": {"skills": [{"type": "anthropic", "skill_id": "pptx", "version": "latest"}]},
-        },
-    )
-
-    print(
-        "The agent output will use the following colors:\n"
-        "\033[0mUser: (default)\033[0m\n"
-        "\033[0mAgent: (default)\033[0m\n"
-        "\033[32mAgent Reasoning: (green)\033[0m\n"
-        "\033[34mUsage: (blue)\033[0m\n"
-    )
-    query = "Create a presentation about renewable energy with 5 slides"
-    print(f"User: {query}")
-    print("Agent: ", end="", flush=True)
-    files: list[HostedFileContent] = []
-    async for chunk in agent.run_stream(query):
-        for content in chunk.contents:
-            match content.type:
-                case "text":
-                    print(content.text, end="", flush=True)
-                case "text_reasoning":
-                    print(f"\033[32m{content.text}\033[0m", end="", flush=True)
-                case "usage":
-                    print(f"\n\033[34m[Usage so far: {content.details}]\033[0m\n", end="", flush=True)
-                case "hosted_file":
-                    # Catch generated files
-                    files.append(content)
-                case _:
-                    logger.debug("Unhandled content type: %s", content.type)
-                    pass
-
-    print("\n")
-    if files:
-        # Save to a new file (will be in the folder where you are running this script)
-        # When running this sample multiple times, the files will be overritten
-        # Since I'm using the pptx skill, the files will be PowerPoint presentations
-        print("Generated files:")
-        for idx, file in enumerate(files):
-            file_content = await client.anthropic_client.beta.files.download(
-                file_id=file.file_id, betas=["files-api-2025-04-14"]
-            )
-            with open(Path(__file__).parent / f"renewable_energy-{idx}.pptx", "wb") as f:
-                await file_content.write_to_file(f.name)
-            print(f"File {idx}: renewable_energy-{idx}.pptx saved to disk.")
-
-
-if __name__ == "__main__":
-    asyncio.run(main())
@@ -1,6 +1,6 @@
 # Azure AI Agent Examples

-This folder contains examples demonstrating different ways to create and use agents with the Azure AI client from the `agent_framework.azure` package. These examples use the `AzureAIClient` with the `azure-ai-projects` 2.x (V2) API surface (see [changelog](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-projects/CHANGELOG.md#200b1-2025-11-11)). For V1 (`azure-ai-agents` 1.x) samples using `AzureAIAgentClient`, see the [Azure AI V1 examples folder](../azure_ai_agent/).
+This folder contains examples demonstrating different ways to create and use agents with the Azure AI client from the `agent_framework.azure` package.

 ## Examples

@@ -1,6 +1,6 @@
 # Azure AI Agent Examples

-This folder contains examples demonstrating different ways to create and use agents with the Azure AI chat client from the `agent_framework.azure` package. These examples use the `AzureAIAgentClient` with the `azure-ai-agents` 1.x (V1) API surface. For updated V2 (`azure-ai-projects` 2.x) samples, see the [Azure AI V2 examples folder](../azure_ai/).
+This folder contains examples demonstrating different ways to create and use agents with the Azure AI chat client from the `agent_framework.azure` package.

 ## Examples

@@ -17,7 +17,7 @@ Workflow Steps:
 import asyncio
 import logging
 from dataclasses import dataclass
-from typing import Literal
+from typing import Literal, Annotated

 from agent_framework import (
    Case,
@@ -31,11 +31,9 @@ from agent_framework import (
 from pydantic import BaseModel, Field
 from typing_extensions import Never

-
 # Define response model with clear user guidance
 class SpamDecision(BaseModel):
    """User's decision on whether the email is spam."""
-
    decision: Literal["spam", "not spam"] = Field(
        description="Enter 'spam' to mark as spam, or 'not spam' to mark as legitimate"
    )
@@ -73,11 +71,10 @@ class SpamDetectorResponse:
 class SpamApprovalRequest:
    """Human-in-the-loop approval request for spam classification."""

-    email_message: str
-    detected_as_spam: bool
-    confidence: float
-    reasons: list[str]
-    full_email_content: EmailContent
+    email_message: str = ""
+    detected_as_spam: bool = False
+    confidence: float = 0.0
+    reasons: str = ""


@dataclass
@@ -131,6 +128,8 @@ class EmailPreprocessor(Executor):
        await ctx.send_message(result)


+
+
 class SpamDetector(Executor):
    """Step 2: An executor that analyzes content and determines if a message is spam."""

@@ -140,9 +139,7 @@ class SpamDetector(Executor):
        self._spam_keywords = spam_keywords

    @handler
-    async def handle_email_content(
-        self, email_content: EmailContent, ctx: WorkflowContext[SpamApprovalRequest]
-    ) -> None:
+    async def handle_email_content(self, email_content: EmailContent, ctx: WorkflowContext[SpamApprovalRequest]) -> None:
        """Analyze email content and determine if the message is spam, then request human approval."""
        await asyncio.sleep(2.0)  # Simulate analysis and detection time

@@ -189,13 +186,25 @@ class SpamDetector(Executor):

        is_spam = spam_score >= 0.5

+        # Store detection result in executor state for later use
+        # Store minimal data needed (not complex objects that don't serialize well)
+        await ctx.set_executor_state({
+            "original_message": email_content.original_message,
+            "cleaned_message": email_content.cleaned_message,
+            "word_count": email_content.word_count,
+            "has_suspicious_patterns": email_content.has_suspicious_patterns,
+            "is_spam": is_spam,
+            "ai_original_classification": is_spam,  # Store original AI decision
+            "confidence_score": spam_score,
+            "spam_reasons": spam_reasons
+        })
+
        # Request human approval before proceeding using new API
        approval_request = SpamApprovalRequest(
            email_message=email_text[:200],  # First 200 chars
            detected_as_spam=is_spam,
            confidence=spam_score,
-            reasons=spam_reasons,
-            full_email_content=email_content,
+            reasons=", ".join(spam_reasons) if spam_reasons else "no specific reasons"
        )

        await ctx.request_info(
@@ -205,15 +214,20 @@ class SpamDetector(Executor):

    @response_handler
    async def handle_human_response(
-        self, original_request: SpamApprovalRequest, response: SpamDecision, ctx: WorkflowContext[SpamDetectorResponse]
+        self,
+        original_request: SpamApprovalRequest,
+        response: SpamDecision,
+        ctx: WorkflowContext[SpamDetectorResponse]
    ) -> None:
        """Process human approval response and continue workflow."""
        print(f"[SpamDetector] handle_human_response called with response: {response}")

        # Get stored detection result
-        ai_original = original_request.detected_as_spam
-        confidence_score = original_request.confidence
-        spam_reasons = original_request.reasons
+        state = await ctx.get_executor_state() or {}
+        print(f"[SpamDetector] Retrieved state: {state}")
+        ai_original = state.get("ai_original_classification", False)
+        confidence_score = state.get("confidence_score", 0.0)
+        spam_reasons = state.get("spam_reasons", [])

        # Parse human decision from the response model
        human_decision = response.decision.strip().lower()
@@ -227,21 +241,27 @@ class SpamDetector(Executor):
            # Default to AI decision if unclear
            is_spam = ai_original

+        # Reconstruct EmailContent from stored primitives
+        email_content = EmailContent(
+            original_message=state.get("original_message", ""),
+            cleaned_message=state.get("cleaned_message", ""),
+            word_count=state.get("word_count", 0),
+            has_suspicious_patterns=state.get("has_suspicious_patterns", False)
+        )
+
        result = SpamDetectorResponse(
-            email_content=original_request.full_email_content,
+            email_content=email_content,
            is_spam=is_spam,
            confidence_score=confidence_score,
            spam_reasons=spam_reasons,
            human_reviewed=True,
            human_decision=response.decision,
-            ai_original_classification=ai_original,
+            ai_original_classification=ai_original
        )

-        print(
-            f"[SpamDetector] Sending SpamDetectorResponse: is_spam={is_spam}, confidence={confidence_score}, human_reviewed=True"
-        )
+        print(f"[SpamDetector] Sending SpamDetectorResponse: is_spam={is_spam}, confidence={confidence_score}, human_reviewed=True")
        await ctx.send_message(result)
-        print("[SpamDetector] Message sent successfully")
+        print(f"[SpamDetector] Message sent successfully")


 class SpamHandler(Executor):
@@ -407,9 +427,7 @@ workflow = (
        spam_detector,
        [
            Case(condition=lambda x: isinstance(x, SpamDetectorResponse) and x.is_spam, target=spam_handler),
-            Default(
-                target=legitimate_message_handler
-            ),  # Default handles non-spam and non-SpamDetectorResponse messages
+            Default(target=legitimate_message_handler),  # Default handles non-spam and non-SpamDetectorResponse messages
        ],
    )
    .add_edge(spam_handler, final_processor)
@@ -3,7 +3,6 @@
 import asyncio
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, override

 # NOTE: the Azure client imports above are real dependencies. When running this
 # sample outside of Azure-enabled environments you may wish to swap in the
@@ -117,19 +116,19 @@ class ReviewGateway(Executor):
    def __init__(self, id: str, writer_id: str) -> None:
        super().__init__(id=id)
        self._writer_id = writer_id
-        self._iteration = 0

    @handler
    async def on_agent_response(self, response: AgentExecutorResponse, ctx: WorkflowContext) -> None:
        # Capture the agent output so we can surface it to the reviewer and persist iterations.
-        self._iteration += 1
-
+        draft = response.agent_run_response.text or ""
+        iteration = int((await ctx.get_executor_state() or {}).get("iteration", 0)) + 1
+        await ctx.set_executor_state({"iteration": iteration, "last_draft": draft})
        # Emit a human approval request.
        await ctx.request_info(
            request_data=HumanApprovalRequest(
                prompt="Review the draft. Reply 'approve' or provide edit instructions.",
-                draft=response.agent_run_response.text,
-                iteration=self._iteration,
+                draft=draft,
+                iteration=iteration,
            ),
            response_type=str,
        )
@@ -143,33 +142,28 @@ class ReviewGateway(Executor):
    ) -> None:
        # The `original_request` is the request we sent earlier that is now being answered.
        reply = feedback.strip()
+        state = await ctx.get_executor_state() or {}
+        draft = state.get("last_draft") or (original_request.draft or "")

-        if len(reply) == 0 or reply.lower() == "approve":
+        if reply.lower() == "approve":
            # Workflow is completed when the human approves.
-            await ctx.yield_output(original_request.draft)
+            await ctx.yield_output(draft)
            return

        # Any other response loops us back to the writer with fresh guidance.
+        guidance = reply or "Tighten the copy and emphasise customer benefit."
+        iteration = int(state.get("iteration", 1)) + 1
+        await ctx.set_executor_state({"iteration": iteration, "last_draft": draft})
        prompt = (
            "Revise the launch note. Respond with the new copy only.\n\n"
-            f"Previous draft:\n{original_request.draft}\n\n"
-            f"Human guidance: {reply}"
+            f"Previous draft:\n{draft}\n\n"
+            f"Human guidance: {guidance}"
        )
        await ctx.send_message(
            AgentExecutorRequest(messages=[ChatMessage(Role.USER, text=prompt)], should_respond=True),
            target_id=self._writer_id,
        )

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
-        # Save the current iteration count in executor state for checkpointing.
-        return {"iteration": self._iteration}
-
-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        # Restore the iteration count from executor state during checkpoint recovery.
-        self._iteration = state.get("iteration", 0)
-

 def create_workflow(checkpoint_storage: FileCheckpointStorage) -> Workflow:
    """Assemble the workflow graph used by both the initial run and resume."""
@@ -253,10 +247,10 @@ async def run_interactive_session(
        else:
            if initial_message:
                print(f"\nStarting workflow with brief: {initial_message}\n")
-                event_stream = workflow.run_stream(message=initial_message)
+                event_stream = workflow.run_stream(initial_message)
            elif checkpoint_id:
                print("\nStarting workflow from checkpoint...\n")
-                event_stream = workflow.run_stream(checkpoint_id=checkpoint_id)
+                event_stream = workflow.run_stream(checkpoint_id)
            else:
                raise ValueError("Either initial_message or checkpoint_id must be provided")

@@ -1,157 +1,322 @@
 # Copyright (c) Microsoft. All rights reserved.

-"""
-Sample: Checkpointing and Resuming a Workflow
-
-Purpose:
-This sample shows how to enable checkpointing for a long-running workflow
-that can be paused and resumed.
-
-What you learn:
- How to configure checkpointing storage (InMemoryCheckpointStorage for testing)
- How to resume a workflow from a checkpoint after interruption
- How to implement executor state management with checkpoint hooks
- How to handle workflow interruptions and automatic recovery
-
-Pipeline:
-This sample shows a workflow that computes factor pairs for numbers up to a given limit:
-1) A start executor that receives the upper limit and creates the initial task
-2) A worker executor that processes each number to find its factor pairs
-3) The worker uses checkpoint hooks to save/restore its internal state
-
-Prerequisites:
- Basic understanding of workflow concepts, including executors, edges, events, etc.
-"""
-
 import asyncio
-from dataclasses import dataclass
-from random import random
-from typing import Any, override
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Any

 from agent_framework import (
+    AgentExecutor,
+    AgentExecutorRequest,
+    AgentExecutorResponse,
+    ChatMessage,
    Executor,
-    InMemoryCheckpointStorage,
-    SuperStepCompletedEvent,
+    FileCheckpointStorage,
+    Role,
    WorkflowBuilder,
-    WorkflowCheckpoint,
    WorkflowContext,
-    WorkflowOutputEvent,
+    get_checkpoint_summary,
    handler,
 )
+from agent_framework.azure import AzureOpenAIChatClient
+from azure.identity import AzureCliCredential
+
+if TYPE_CHECKING:
+    from agent_framework import Workflow
+    from agent_framework._workflows._checkpoint import WorkflowCheckpoint
+
+"""
+Sample: Checkpointing and Resuming a Workflow (with an Agent stage)
+
+Purpose:
+This sample shows how to enable checkpointing at superstep boundaries, persist both
+executor-local state and shared workflow state, and then resume execution from a specific
+checkpoint. The workflow demonstrates a simple text-processing pipeline that includes
+an LLM-backed AgentExecutor stage.
+
+Pipeline:
+1) UpperCaseExecutor converts input to uppercase and records state.
+2) ReverseTextExecutor reverses the string.
+3) SubmitToLowerAgent prepares an AgentExecutorRequest for the lowercasing agent.
+4) lower_agent (AgentExecutor) converts text to lowercase via Azure OpenAI.
+5) FinalizeFromAgent yields the final result.
+
+What you learn:
+- How to persist executor state using ctx.get_executor_state and ctx.set_executor_state.
+- How to persist shared workflow state using ctx.set_shared_state for cross-executor visibility.
+- How to configure FileCheckpointStorage and call with_checkpointing on WorkflowBuilder.
+- How to list and inspect checkpoints programmatically.
+- How to interactively choose a checkpoint to resume from (instead of always resuming
+    from the most recent or a hard-coded one) using run_stream.
+- How workflows complete by yielding outputs when idle, not via explicit completion events.
+
+Prerequisites:
+- Azure AI or Azure OpenAI available for AzureOpenAIChatClient.
+- Authentication with azure-identity via AzureCliCredential. Run az login locally.
+- Filesystem access for writing JSON checkpoint files in a temp directory.
+"""
+
+# Define the temporary directory for storing checkpoints.
+# These files allow the workflow to be resumed later.
+DIR = os.path.dirname(__file__)
+TEMP_DIR = os.path.join(DIR, "tmp", "checkpoints")
+os.makedirs(TEMP_DIR, exist_ok=True)


-@dataclass
-class ComputeTask:
-    """Task containing the list of numbers remaining to be processed."""
-
-    remaining_numbers: list[int]
-
-
-class StartExecutor(Executor):
-    """Initiates the workflow by providing the upper limit for factor pair computation."""
+class UpperCaseExecutor(Executor):
+    """Uppercases the input text and persists both local and shared state."""

    @handler
-    async def start(self, upper_limit: int, ctx: WorkflowContext[ComputeTask]) -> None:
-        """Start the workflow with a list of numbers to process."""
-        print(f"StartExecutor: Starting factor pair computation up to {upper_limit}")
-        await ctx.send_message(ComputeTask(remaining_numbers=list(range(1, upper_limit + 1))))
+    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
+        result = text.upper()
+        print(f"UpperCaseExecutor: '{text}' -> '{result}'")
+
+        # Persist executor-local state so it is captured in checkpoints
+        # and available after resume for observability or logic.
+        prev = await ctx.get_executor_state() or {}
+        count = int(prev.get("count", 0)) + 1
+        await ctx.set_executor_state({
+            "count": count,
+            "last_input": text,
+            "last_output": result,
+        })
+
+        # Write to shared_state so downstream executors and any resumed runs can read it.
+        await ctx.set_shared_state("original_input", text)
+        await ctx.set_shared_state("upper_output", result)
+
+        # Send transformed text to the next executor.
+        await ctx.send_message(result)


-class WorkerExecutor(Executor):
-    """Processes numbers to compute their factor pairs and manages executor state for checkpointing."""
+class SubmitToLowerAgent(Executor):
+    """Builds an AgentExecutorRequest to send to the lowercasing agent while keeping shared-state visibility."""

-    def __init__(self, id: str) -> None:
+    def __init__(self, id: str, agent_id: str):
        super().__init__(id=id)
-        self._composite_number_pairs: dict[int, list[tuple[int, int]]] = {}
+        self._agent_id = agent_id

    @handler
-    async def compute(
-        self,
-        task: ComputeTask,
-        ctx: WorkflowContext[ComputeTask, dict[int, list[tuple[int, int]]]],
-    ) -> None:
-        """Process the next number in the task, computing its factor pairs."""
-        next_number = task.remaining_numbers.pop(0)
+    async def submit(self, text: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
+        # Demonstrate reading shared_state written by UpperCaseExecutor.
+        # Shared state survives across checkpoints and is visible to all executors.
+        orig = await ctx.get_shared_state("original_input")
+        upper = await ctx.get_shared_state("upper_output")
+        print(f"LowerAgent (shared_state): original_input='{orig}', upper_output='{upper}'")

-        print(f"WorkerExecutor: Computing factor pairs for {next_number}")
-        pairs: list[tuple[int, int]] = []
-        for i in range(1, next_number):
-            if next_number % i == 0:
-                pairs.append((i, next_number // i))
-        self._composite_number_pairs[next_number] = pairs
+        # Build a minimal, deterministic prompt for the AgentExecutor.
+        prompt = f"Convert the following text to lowercase. Return ONLY the transformed text.\n\nText: {text}"

-        if not task.remaining_numbers:
-            # All numbers processed - output the results
-            await ctx.yield_output(self._composite_number_pairs)
-        else:
-            # More numbers to process - continue with remaining task
-            await ctx.send_message(task)
+        # Send to the AgentExecutor. should_respond=True instructs the agent to produce a reply.
+        await ctx.send_message(
+            AgentExecutorRequest(messages=[ChatMessage(Role.USER, text=prompt)], should_respond=True),
+            target_id=self._agent_id,
+        )

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
-        """Save the executor's internal state for checkpointing."""
-        return {"composite_number_pairs": self._composite_number_pairs}

-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        """Restore the executor's internal state from a checkpoint."""
-        self._composite_number_pairs = state.get("composite_number_pairs", {})
+class FinalizeFromAgent(Executor):
+    """Consumes the AgentExecutorResponse and yields the final result."""
+
+    @handler
+    async def finalize(self, response: AgentExecutorResponse, ctx: WorkflowContext[Any, str]) -> None:
+        result = response.agent_run_response.text or ""
+
+        # Persist executor-local state for auditability when inspecting checkpoints.
+        prev = await ctx.get_executor_state() or {}
+        count = int(prev.get("count", 0)) + 1
+        await ctx.set_executor_state({
+            "count": count,
+            "last_output": result,
+            "final": True,
+        })
+
+        # Yield the final result so external consumers see the final value.
+        await ctx.yield_output(result)
+
+
+class ReverseTextExecutor(Executor):
+    """Reverses the input text and persists local state."""
+
+    @handler
+    async def reverse_text(self, text: str, ctx: WorkflowContext[str]) -> None:
+        result = text[::-1]
+        print(f"ReverseTextExecutor: '{text}' -> '{result}'")
+
+        # Persist executor-local state so checkpoint inspection can reveal progress.
+        prev = await ctx.get_executor_state() or {}
+        count = int(prev.get("count", 0)) + 1
+        await ctx.set_executor_state({
+            "count": count,
+            "last_input": text,
+            "last_output": result,
+        })
+
+        # Forward the reversed string to the next stage.
+        await ctx.send_message(result)
+
+
+def create_workflow(checkpoint_storage: FileCheckpointStorage) -> "Workflow":
+    # Instantiate the pipeline executors.
+    upper_case_executor = UpperCaseExecutor(id="upper-case")
+    reverse_text_executor = ReverseTextExecutor(id="reverse-text")
+
+    # Configure the agent stage that lowercases the text.
+    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
+    lower_agent = AgentExecutor(
+        chat_client.create_agent(
+            instructions=("You transform text to lowercase. Reply with ONLY the transformed text.")
+        ),
+        id="lower_agent",
+    )
+
+    # Bridge to the agent and terminalization stage.
+    submit_lower = SubmitToLowerAgent(id="submit_lower", agent_id=lower_agent.id)
+    finalize = FinalizeFromAgent(id="finalize")
+
+    # Build the workflow with checkpointing enabled.
+    return (
+        WorkflowBuilder(max_iterations=5)
+        .add_edge(upper_case_executor, reverse_text_executor)  # Uppercase -> Reverse
+        .add_edge(reverse_text_executor, submit_lower)  # Reverse -> Build Agent request
+        .add_edge(submit_lower, lower_agent)  # Submit to AgentExecutor
+        .add_edge(lower_agent, finalize)  # Agent output -> Finalize
+        .set_start_executor(upper_case_executor)  # Entry point
+        .with_checkpointing(checkpoint_storage=checkpoint_storage)  # Enable persistence
+        .build()
+    )
+
+
+def _render_checkpoint_summary(checkpoints: list["WorkflowCheckpoint"]) -> None:
+    """Display human-friendly checkpoint metadata using framework summaries."""
+
+    if not checkpoints:
+        return
+
+    print("\nCheckpoint summary:")
+    for cp in sorted(checkpoints, key=lambda c: c.timestamp):
+        summary = get_checkpoint_summary(cp)
+        msg_count = sum(len(v) for v in cp.messages.values())
+        state_keys = sorted(summary.executor_ids)
+        orig = cp.shared_state.get("original_input")
+        upper = cp.shared_state.get("upper_output")
+
+        line = (
+            f"- {summary.checkpoint_id} | iter={summary.iteration_count} | messages={msg_count} | states={state_keys}"
+        )
+        if summary.status:
+            line += f" | status={summary.status}"
+        line += f" | shared_state: original_input='{orig}', upper_output='{upper}'"
+        print(line)


 async def main():
-    # Create workflow executors
-    start_executor = StartExecutor(id="start")
-    worker_executor = WorkerExecutor(id="worker")
+    # Clear existing checkpoints in this sample directory for a clean run.
+    checkpoint_dir = Path(TEMP_DIR)
+    for file in checkpoint_dir.glob("*.json"):  # noqa: ASYNC240
+        file.unlink()

-    # Build workflow with checkpointing enabled
-    workflow_builder = (
-        WorkflowBuilder()
-        .set_start_executor(start_executor)
-        .add_edge(start_executor, worker_executor)
-        .add_edge(worker_executor, worker_executor)  # Self-loop for iterative processing
-    )
-    checkpoint_storage = InMemoryCheckpointStorage()
-    workflow_builder = workflow_builder.with_checkpointing(checkpoint_storage=checkpoint_storage)
+    # Backing store for checkpoints written by with_checkpointing.
+    checkpoint_storage = FileCheckpointStorage(storage_path=TEMP_DIR)

-    # Run workflow with automatic checkpoint recovery
-    latest_checkpoint: WorkflowCheckpoint | None = None
-    while True:
-        workflow = workflow_builder.build()
+    workflow = create_workflow(checkpoint_storage=checkpoint_storage)

-        # Start from checkpoint or fresh execution
-        print(f"\n** Workflow {workflow.id} started **")
-        event_stream = (
-            workflow.run_stream(message=10)
-            if latest_checkpoint is None
-            else workflow.run_stream(checkpoint_id=latest_checkpoint.checkpoint_id)
-        )
+    # Run the full workflow once and observe events as they stream.
+    print("Running workflow with initial message...")
+    async for event in workflow.run_stream(message="hello world"):
+        print(f"Event: {event}")

-        output: str | None = None
-        async for event in event_stream:
-            if isinstance(event, WorkflowOutputEvent):
-                output = event.data
-                break
-            if isinstance(event, SuperStepCompletedEvent) and random() < 0.5:
-                # Randomly simulate system interruptions
-                # The `SuperStepCompletedEvent` ensures we only interrupt after
-                # the current super-step is fully complete and checkpointed.
-                # If we interrupt mid-step, the workflow may resume from an earlier point.
-                print("\n** Simulating workflow interruption. Stopping execution. **")
+    # Inspect checkpoints written during the run.
+    all_checkpoints = await checkpoint_storage.list_checkpoints()
+    if not all_checkpoints:
+        print("No checkpoints found!")
+        return
+
+    # All checkpoints created by this run share the same workflow_id.
+    workflow_id = all_checkpoints[0].workflow_id
+
+    _render_checkpoint_summary(all_checkpoints)
+
+    # Offer an interactive selection of checkpoints to resume from.
+    sorted_cps = sorted([cp for cp in all_checkpoints if cp.workflow_id == workflow_id], key=lambda c: c.timestamp)
+
+    print("\nAvailable checkpoints to resume from:")
+    for idx, cp in enumerate(sorted_cps):
+        summary = get_checkpoint_summary(cp)
+        line = f"  [{idx}] id={summary.checkpoint_id} iter={summary.iteration_count}"
+        if summary.status:
+            line += f" status={summary.status}"
+        msg_count = sum(len(v) for v in cp.messages.values())
+        line += f" messages={msg_count}"
+        print(line)
+
+    user_input = input(  # noqa: ASYNC250
+        "\nEnter checkpoint index (or paste checkpoint id) to resume from, or press Enter to skip resume: "
+    ).strip()
+
+    if not user_input:
+        print("No checkpoint selected. Exiting without resuming.")
+        return
+
+    chosen_cp_id: str | None = None
+
+    # Try as index first
+    if user_input.isdigit():
+        idx = int(user_input)
+        if 0 <= idx < len(sorted_cps):
+            chosen_cp_id = sorted_cps[idx].checkpoint_id
+    # Fall back to direct id match
+    if chosen_cp_id is None:
+        for cp in sorted_cps:
+            if cp.checkpoint_id.startswith(user_input):  # allow prefix match for convenience
+                chosen_cp_id = cp.checkpoint_id
                break

-        # Find the latest checkpoint to resume from
-        all_checkpoints = await checkpoint_storage.list_checkpoints()
-        if not all_checkpoints:
-            raise RuntimeError("No checkpoints available to resume from.")
-        latest_checkpoint = all_checkpoints[-1]
-        print(
-            f"Checkpoint {latest_checkpoint.checkpoint_id}: "
-            f"(iter={latest_checkpoint.iteration_count}, messages={latest_checkpoint.messages})"
-        )
+    if chosen_cp_id is None:
+        print("Input did not match any checkpoint. Exiting without resuming.")
+        return

-        if output is not None:
-            print(f"\nWorkflow completed successfully with output: {output}")
-            break
+    # You can reuse the same workflow graph definition and resume from a prior checkpoint.
+    # This second workflow instance does not enable checkpointing to show that resumption
+    # reads from stored state but need not write new checkpoints.
+    new_workflow = create_workflow(checkpoint_storage=checkpoint_storage)
+
+    print(f"\nResuming from checkpoint: {chosen_cp_id}")
+    async for event in new_workflow.run_stream(checkpoint_id=chosen_cp_id, checkpoint_storage=checkpoint_storage):
+        print(f"Resumed Event: {event}")
+
+    """
+    Sample Output:
+
+    Running workflow with initial message...
+    UpperCaseExecutor: 'hello world' -> 'HELLO WORLD'
+    Event: ExecutorInvokeEvent(executor_id=upper_case_executor)
+    Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
+    ReverseTextExecutor: 'HELLO WORLD' -> 'DLROW OLLEH'
+    Event: ExecutorInvokeEvent(executor_id=reverse_text_executor)
+    Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
+    LowerAgent (shared_state): original_input='hello world', upper_output='HELLO WORLD'
+    Event: ExecutorInvokeEvent(executor_id=submit_lower)
+    Event: ExecutorInvokeEvent(executor_id=lower_agent)
+    Event: ExecutorInvokeEvent(executor_id=finalize)
+
+    Checkpoint summary:
+    - dfc63e72-8e8d-454f-9b6d-0d740b9062e6 | label='after_initial_execution' | iter=0 | messages=1 | states=['upper_case_executor'] | shared_state: original_input='hello world', upper_output='HELLO WORLD'
+    - a78c345a-e5d9-45ba-82c0-cb725452d91b | label='superstep_1' | iter=1 | messages=1 | states=['reverse_text_executor', 'upper_case_executor'] | shared_state: original_input='hello world', upper_output='HELLO WORLD'
+    - 637c1dbd-a525-4404-9583-da03980537a2 | label='superstep_2' | iter=2 | messages=0 | states=['finalize', 'lower_agent', 'reverse_text_executor', 'submit_lower', 'upper_case_executor'] | shared_state: original_input='hello world', upper_output='HELLO WORLD'
+
+    Available checkpoints to resume from:
+        [0] id=dfc63e72-... iter=0 messages=1 label='after_initial_execution'
+        [1] id=a78c345a-... iter=1 messages=1 label='superstep_1'
+        [2] id=637c1dbd-... iter=2 messages=0 label='superstep_2'
+
+    Enter checkpoint index (or paste checkpoint id) to resume from, or press Enter to skip resume: 1
+
+    Resuming from checkpoint: a78c345a-e5d9-45ba-82c0-cb725452d91b
+    LowerAgent (shared_state): original_input='hello world', upper_output='HELLO WORLD'
+    Resumed Event: ExecutorInvokeEvent(executor_id=submit_lower)
+    Resumed Event: ExecutorInvokeEvent(executor_id=lower_agent)
+    Resumed Event: ExecutorInvokeEvent(executor_id=finalize)
+    """  # noqa: E501


 if __name__ == "__main__":
@@ -7,7 +7,6 @@ import uuid
 from dataclasses import dataclass, field, replace
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Any, override

 from agent_framework import (
    Executor,
@@ -206,8 +205,6 @@ class LaunchCoordinator(Executor):

    def __init__(self) -> None:
        super().__init__(id="launch_coordinator")
-        # Track pending requests to match responses
-        self._pending_requests: dict[str, SubWorkflowRequestMessage] = {}

    @handler
    async def kick_off(self, topic: str, ctx: WorkflowContext[DraftTask]) -> None:
@@ -247,9 +244,11 @@ class LaunchCoordinator(Executor):
        if not isinstance(request.source_event.data, ReviewRequest):
            raise TypeError(f"Expected 'ReviewRequest', got {type(request.source_event.data)}")

-        # Record the request for response matching
+        # Record the request to response matching
        review_request = request.source_event.data
-        self._pending_requests[review_request.id] = request
+        executor_state = await ctx.get_executor_state() or {}
+        executor_state[review_request.id] = request
+        await ctx.set_executor_state(executor_state)

        # Send the request without modification
        await ctx.request_info(request_data=review_request, response_type=str)
@@ -266,25 +265,17 @@ class LaunchCoordinator(Executor):
        Note that the response must be sent back using SubWorkflowResponseMessage to route
        the response back to the sub-workflow.
        """
-        request_message = self._pending_requests.pop(original_request.id, None)
+        executor_state = await ctx.get_executor_state() or {}
+        request_message = executor_state.pop(original_request.id, None)
+
+        # Save the executor state back to the context
+        await ctx.set_executor_state(executor_state)

        if request_message is None:
            raise ValueError("No matching pending request found for the resource response")

        await ctx.send_message(request_message.create_response(response))

-    @override
-    async def on_checkpoint_save(self) -> dict[str, Any]:
-        """Capture any additional state needed for checkpointing."""
-        return {
-            "pending_requests": self._pending_requests,
-        }
-
-    @override
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        """Restore any additional state needed from checkpointing."""
-        self._pending_requests = state.get("pending_requests", {})
-

 # ---------------------------------------------------------------------------
 # Workflow construction helpers
@@ -365,7 +356,9 @@ async def main() -> None:
    workflow2 = build_parent_workflow(storage)

    request_info_event: RequestInfoEvent | None = None
-    async for event in workflow2.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
+    async for event in workflow2.run_stream(
+        resume_checkpoint.checkpoint_id,
+    ):
        if isinstance(event, RequestInfoEvent):
            request_info_event = event