diff --git a/dotnet/Directory.Packages.props b/dotnet/Directory.Packages.props index 7d8852c2659..53458b7ac1e 100644 --- a/dotnet/Directory.Packages.props +++ b/dotnet/Directory.Packages.props @@ -39,6 +39,7 @@ + diff --git a/dotnet/samples/02-agents/Harness/BuildYourOwnClaw/Claw_Step04_ProductionReady/ClawAgent.Hosted/.gitignore b/dotnet/samples/02-agents/Harness/BuildYourOwnClaw/Claw_Step04_ProductionReady/ClawAgent.Hosted/.gitignore index 5fe72b02045..bf405d0a839 100644 --- a/dotnet/samples/02-agents/Harness/BuildYourOwnClaw/Claw_Step04_ProductionReady/ClawAgent.Hosted/.gitignore +++ b/dotnet/samples/02-agents/Harness/BuildYourOwnClaw/Claw_Step04_ProductionReady/ClawAgent.Hosted/.gitignore @@ -1,2 +1,3 @@ .azure -azure.yaml \ No newline at end of file +azure.yaml +Properties/launchSettings.json \ No newline at end of file diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/.env.example b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/.env.example index 1104c2d65ba..e415ccc83a3 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/.env.example +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/.env.example @@ -1,5 +1,4 @@ -# Optional local countdown delay -COUNTDOWN_DELAY_SECONDS=1 - -# Local development only +# Local development only ASPNETCORE_URLS=http://+:8088 +IDEMPOTENT_SERVICE_ENDPOINT=http://localhost:8089/ +IDEMPOTENT_OPERATION_SCOPE=local-countdown diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/HostedWorkflowResilientLongRunning.csproj b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/HostedWorkflowResilientLongRunning.csproj index 57d2acbe972..cd1bfe36bff 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/HostedWorkflowResilientLongRunning.csproj +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/HostedWorkflowResilientLongRunning.csproj @@ -19,6 +19,10 @@ true + + + + diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/Program.cs b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/Program.cs index 0928eac8cb6..fea5c679ec1 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/Program.cs +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/Program.cs @@ -2,11 +2,11 @@ // Sample: a long-running countdown workflow hosted as a resilient background response. // Each completed superstep is paired with an AgentServer response checkpoint so a restarted -// process resumes with ordered output and without losing or duplicating countdown items. +// process resumes without losing confirmed output. An interrupted in-flight step can run again. using System.Globalization; -using System.Text.RegularExpressions; using DotNetEnv; +using Hosted_Shared_Contributor_Setup; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Foundry.Hosting; using Microsoft.Agents.AI.Workflows; @@ -16,20 +16,16 @@ var agentName = System.Environment.GetEnvironmentVariable("AGENT_NAME") ?? "hosted-workflow-resilient-long-running"; -var delaySeconds = int.TryParse( - System.Environment.GetEnvironmentVariable("COUNTDOWN_DELAY_SECONDS"), - NumberStyles.None, - CultureInfo.InvariantCulture, - out int configuredDelaySeconds) - ? configuredDelaySeconds - : 1; -if (delaySeconds < 0) -{ - throw new InvalidOperationException("COUNTDOWN_DELAY_SECONDS must be zero or greater."); -} +var idempotentServiceEndpoint = System.Environment.GetEnvironmentVariable("IDEMPOTENT_SERVICE_ENDPOINT") + ?? throw new InvalidOperationException("IDEMPOTENT_SERVICE_ENDPOINT is not set."); +var operationScope = System.Environment.GetEnvironmentVariable("IDEMPOTENT_OPERATION_SCOPE") + ?? throw new InvalidOperationException("IDEMPOTENT_OPERATION_SCOPE is not set."); + +using var idempotentServiceHttpClient = new HttpClient { BaseAddress = new Uri(idempotentServiceEndpoint) }; +var idempotentService = new IdempotentServiceClient(idempotentServiceHttpClient); var start = new CountdownStartExecutor(); -var countdown = new CountdownExecutor(TimeSpan.FromSeconds(delaySeconds)); +var countdown = new CountdownExecutor(idempotentService, operationScope); var complete = new CountdownCompleteExecutor(); Workflow workflow = new WorkflowBuilder(start) @@ -46,60 +42,98 @@ includeWorkflowOutputsInResponse: true); var builder = WebApplication.CreateBuilder(args); -builder.Services.AddFoundryResponses( - agent, - configure: options => options.ResilientBackground = true); +builder.Services.AddFoundryResponses(agent, configure: options => options.ResilientBackground = true); var app = builder.Build(); app.MapFoundryResponses(); +Task? requestedShutdown = null; if (app.Environment.IsDevelopment()) { app.MapFoundryResponses("openai/v1"); } +// This configuration is for local development demonstration purposes only. +// When hosted in Foundry the lifetime of the agent process is managed and shutdowns are handled gracefully. +if (app.Environment.IsDevelopment() + && string.Equals( + System.Environment.GetEnvironmentVariable("ENABLE_E2E_SHUTDOWN_ENDPOINT"), + "true", + StringComparison.OrdinalIgnoreCase)) +{ + app.MapPost("/shutdown", (IEnumerable hostedServices) => + { + // The E2E closes its client stream first, then signals the resilient task service before + // stopping HTTP. This reproduces the hosted-service shutdown path used by AgentServer tests. + IHostedService taskDurabilityService = hostedServices.SingleOrDefault( + service => string.Equals( + service.GetType().FullName, + "Azure.AI.AgentServer.Core.Tasks.Engine.TaskDurabilityService", + StringComparison.Ordinal)) + ?? throw new InvalidOperationException("The AgentServer resilient task service is not registered."); +#pragma warning disable CA2025 // Awaited after app.RunAsync before top-level disposable resources are disposed. + requestedShutdown ??= StopServerAsync(app, taskDurabilityService); +#pragma warning restore CA2025 + return Results.Accepted(); + }); +} + Console.WriteLine($"Process ID: {System.Environment.ProcessId}"); -app.Run(); +await app.RunAsync(); +if (requestedShutdown is not null) +{ + await requestedShutdown; +} +static async Task StopServerAsync(WebApplication app, IHostedService taskDurabilityService) +{ + await Task.Delay(TimeSpan.FromMilliseconds(100)); + await taskDurabilityService.StopAsync(CancellationToken.None); + await app.StopAsync(); +} + +/// +/// Starts the countdown ten above the numeric input so the E2E can interrupt it after some progress. +/// [SendsMessage(typeof(int))] -[YieldsOutput(typeof(string))] -internal sealed partial class CountdownStartExecutor() : ChatProtocolExecutor( - "start", - new ChatProtocolExecutorOptions { AutoSendTurnToken = false }) +internal sealed class CountdownStartExecutor() : ChatProtocolExecutor( + "start", new ChatProtocolExecutorOptions { AutoSendTurnToken = false }) { protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => base.ConfigureProtocol(protocolBuilder).SendsMessage(); - protected override async ValueTask TakeTurnAsync( + protected override ValueTask TakeTurnAsync( List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) { - string input = string.Join( - System.Environment.NewLine, - messages.Select(message => message.Text).Where(text => !string.IsNullOrWhiteSpace(text))); - Match match = PositiveIntegerRegex().Match(input); - if (!match.Success - || !int.TryParse(match.Value, NumberStyles.None, CultureInfo.InvariantCulture, out int target) - || target <= 0) - { - await context.YieldOutputAsync( - "The message must contain a positive integer counter target.", - cancellationToken); - return; - } + // The first workflow turn contains one message; add 10 to its countdown start value. + var maxNumberOfMessages = 10 + int.Parse(messages.Single().Text, CultureInfo.InvariantCulture); - await context.SendMessageAsync(target, cancellationToken: cancellationToken); + return context.SendMessageAsync(maxNumberOfMessages, cancellationToken: cancellationToken); } - - [GeneratedRegex(@"(? +/// Calls the idempotent service for each count, yields its result, and schedules the next count. +/// +/// +/// +/// For demonstration purposes only. The workflow and its backing service should not be used as-is in production. +/// +/// +/// A service call can finish before the workflow and response checkpoints are confirmed. +/// If the process stops during that interval, recovery can call the service again for the same count. +/// Reusing the scope and operation ID lets the service return the stored result without repeating its effect. +/// Checkpoint recovery and stream replay do not undo effects already performed by downstream services. +/// +/// [SendsMessage(typeof(int))] [SendsMessage(typeof(string))] [YieldsOutput(typeof(string))] -internal sealed class CountdownExecutor(TimeSpan delay) : Executor("countdown") +internal sealed class CountdownExecutor( + IdempotentServiceClient idempotentService, + string operationScope) : Executor("countdown") { public override async ValueTask HandleAsync( int message, @@ -108,30 +142,22 @@ public override async ValueTask HandleAsync( { if (message <= 0) { - await context.SendMessageAsync( - "Countdown complete.", - targetId: "complete", - cancellationToken: cancellationToken); + await context.SendMessageAsync(string.Empty, targetId: "complete", cancellationToken: cancellationToken); return; } - await Task.Delay(delay, cancellationToken); - await context.YieldOutputAsync( - message.ToString(CultureInfo.InvariantCulture), - cancellationToken); - await context.SendMessageAsync( - message - 1, - targetId: "countdown", - cancellationToken: cancellationToken); + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + string result = await idempotentService.ExecuteOperationAsync(operationScope, message, cancellationToken); + await context.YieldOutputAsync(result, cancellationToken); + await context.SendMessageAsync(message - 1, targetId: this.Id, cancellationToken: cancellationToken); } } -[YieldsOutput(typeof(string))] -internal sealed class CountdownCompleteExecutor() : Executor("complete") +internal sealed class CountdownCompleteExecutor() : Executor("complete") { - public override ValueTask HandleAsync( + public override ValueTask HandleAsync( string message, IWorkflowContext context, CancellationToken cancellationToken = default) => - context.YieldOutputAsync(message, cancellationToken); + ValueTask.FromResult(message); } diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/README.md b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/README.md index 3b2b20498b1..af42d0aeb30 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/README.md +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/README.md @@ -1,40 +1,79 @@ # Hosted-Workflow-Resilient-Long-Running A deterministic countdown workflow that demonstrates resilient background execution. Each number is -one workflow output item. If the process stops, AgentServer restores the last response snapshot and -the workflow resumes from the exact workflow checkpoint ID recorded in that snapshot. +one workflow output item. The workflow calls a separate idempotent HTTP service before emitting each +number. If the process stops, AgentServer supplies its saved response and MAF resumes the workflow +from the checkpoint referenced by that response. -For an input such as `Count down from 6`, the final message outputs are: +> **Local demonstration only.** This sample was prepared for +> [`Using-E2E-Resilience`](../Using-E2E-Resilience/) to demonstrate workflow recovery after a crash or +> graceful shutdown. The countdown only makes step recovery visible, and the SQLite service simulates +> an operation to illustrate how repeated calls avoid duplicate records. Neither represents a real +> business scenario. This workflow, its HTTP client, and the service are not intended for production +> use or deployment to Foundry. + +The input must be a single message containing integer text. The start executor adds ten so the E2E +has some progress to interrupt. With input `10`, a normal run emits: ```text -6 -5 -4 -3 -2 +20 +19 +... +11 +10 +... 1 -Countdown complete. ``` -The exact list makes recovery errors visible. A missing item means response state advanced beyond -workflow state. A repeated item means the workflow resumed before the response snapshot boundary. +At zero, the workflow terminates without a visible completion message. A recovered stream can +include repeated text; this is separate from whether the service created duplicate operation records. ## Workflow | Executor | Behavior | | --- | --- | -| `start` | Reads the first positive integer from the request. | -| `countdown` | Waits, yields the current number, decrements it, and sends it back to itself. | -| `complete` | Yields `Countdown complete.` after the counter reaches zero. | +| `start` | Parses the single text input as an integer, adds ten, and sends it to `countdown`. | +| `countdown` | Waits one second, calls the idempotent service, yields the returned number, and sends the decremented number to itself. | +| `complete` | Receives an empty string when the count reaches zero and ends the workflow. | All executor IDs and the workflow agent ID are stable so a replacement process reconstructs the same workflow topology. +## Idempotent operations + +The countdown calls `IdempotentServiceClient` from `Hosted_Shared_Contributor_Setup`. The +[`Using-E2E-Resilience`](../Using-E2E-Resilience/) executable hosts the backing Kestrel service in a +separate process when started with `--idempotent-service`. Only that service accesses SQLite. + +The service uses `(scope, operation_id)` as the primary key. The count is the operation ID. The first +call creates a row containing its result: + +```text +Operation Crash/10 executed. +``` + +If recovery executes the same countdown step again, the insert leaves that row unchanged and the +service reads and returns its stored result: + +```text +Duplicate operation Crash/10 ignored. +``` + +Configure the service with `IDEMPOTENT_SERVICE_ENDPOINT` and identify the logical operation group +with `IDEMPOTENT_OPERATION_SCOPE`. Keep the scope unchanged across recovery attempts. Use a different +scope for an unrelated run, otherwise the same counts intentionally reuse the previous results. + +The service can complete an operation before the workflow's progress is confirmed. A crash or +shutdown during that interval can cause the step to be invoked again. The stable key protects the +database operation, not the execution of the workflow step or the text displayed by a client. +For a real email, payment, or API write, the downstream service must enforce equivalent idempotency. +Simply placing that external call beside a SQLite insert would not make the two actions atomic. + ## Recovery boundary -At every completed workflow superstep, Foundry Hosting: +After the workflow finishes a batch of work and creates its execution checkpoint, MAF Hosting: -1. Closes the response output item produced by that superstep. +1. Closes any remaining response output for that work. 2. Saves the matching AgentSession. 3. Writes the workflow checkpoint ID to AgentServer internal response metadata as `_last_checkpoint_id`. @@ -43,7 +82,14 @@ At every completed workflow superstep, Foundry Hosting: 5. Yields `ResponseEventStream.Checkpoint()`. On recovery, the handler reads `_last_checkpoint_id` from `PersistedResponse` and selects that exact -workflow checkpoint before execution continues. +workflow checkpoint before execution continues. `PersistedResponse` is the saved response supplied +by AgentServer, not the client's accumulated text. + +Workflow checkpoints, AgentSession storage, response storage, and the replay log are not one +transaction. Output can be published before its matching checkpoint is confirmed. Recovery can +therefore rerun unconfirmed work. A later `response.in_progress` event carries replacement response +state for clients that reconstruct the current result; it does not reverse an external service call. +See [streaming with reconnect](https://learn.microsoft.com/azure/foundry/agents/how-to/stream-with-reconnect). ## Local development @@ -53,57 +99,29 @@ The easiest local demonstration is the automated E2E console: dotnet run --project dotnet\samples\04-hosting\FoundryHostedAgents\responses\Using-E2E-Resilience ``` -It starts this server, prints countdown outputs, force-kills the process, restarts it, prints replay -and recovery outputs, and validates the final sequence. - -To run only the server, copy `.env.example` to `.env`, then run: - -```powershell -dotnet run --tl:off -``` - -Set `COUNTDOWN_DELAY_SECONDS=0` to make a normal run complete immediately. - -## Deploy from source - -Create an empty working directory outside the repository: - -```powershell -$work = Join-Path $env:TEMP "hosted-workflow-resilient-long-running-work" -New-Item -ItemType Directory -Path $work -Force | Out-Null -Set-Location $work - -$sample = "/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/azure.yaml" -azd auth login -azd ai agent init -m $sample -``` - -### Contributors testing framework changes - -Skip this section unless the current framework changes have not been released. Pack the repository -source into the scaffolded upload before provisioning: - -```powershell -/dotnet/samples/04-hosting/FoundryHostedAgents/scripts/Add-LocalFrameworkFeed.ps1 ` - -Path ./hosted-workflow-resilient-long-running -``` +It runs both an abrupt process crash and a host shutdown after a countdown operation, starts a +replacement process for each scenario, displays replayed output, and queries the idempotent service +for its completed operation count. The current input is `10`, so 20 stored operations are expected. +The E2E reports the count rather than asserting it. The shutdown path exercises `IsShutdownRequested` +and recovery deferral. -Then deploy: +To run the components manually, start the service from the repository root in a separate terminal: ```powershell -Set-Location hosted-workflow-resilient-long-running -azd provision -azd deploy +$env:ASPNETCORE_URLS = "http://localhost:8089" +$env:IDEMPOTENT_SERVICE_DATABASE_PATH = Join-Path $env:TEMP "countdown-idempotency.db" +dotnet run --project dotnet\samples\04-hosting\FoundryHostedAgents\responses\Using-E2E-Resilience -- --idempotent-service ``` -Grant the hosted agent identity `Foundry User` on the Foundry project so it can write workflow -checkpoints and AgentSession state. +In this hosted sample's directory, copy `.env.example` to `.env`, configure the service endpoint and +scope, and run `dotnet run`. Send integer text such as `"10"` to the Responses endpoint. Closing the +HTTP stream of an accepted background response does not cancel its server-side execution. ## Automated coverage `ResilientTwoLifetimeIntegrationTests.StoppedHost_RecoversWorkflowWithCompleteOrderedOutputAsync` -starts the Responses host twice over shared durable state. It interrupts the first host while the -counter is processing `3`, then verifies that the recovered response contains exactly: +uses its own test workflow, not these sample executors. It starts the Responses host twice over +shared durable state, interrupts the first host while its counter is processing `3`, and verifies: ```text 6, 5, 4, 3, 2, 1, Countdown complete. @@ -111,7 +129,7 @@ counter is processing `3`, then verifies that the recovered response contains ex ## Related samples -- [Using-E2E-Resilience](../Using-E2E-Resilience/README.md): automated local crash-recovery console. +- [Using-E2E-Resilience](../Using-E2E-Resilience/README.md): automated crash and shutdown recovery console. - [Hosted-Workflow-Resilient](../Hosted-Workflow-Resilient/README.md): resilient model-backed translation workflow. - [Hosted-Workflow-Simple](../Hosted-Workflow-Simple/README.md): workflow hosting without resilient background execution. - [Hosted-Steering](../Hosted-Steering/README.md): mid-turn steering. diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/azure.yaml b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/azure.yaml index 55281bc1c08..83b59499147 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/azure.yaml +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-Workflow-Resilient-Long-Running/azure.yaml @@ -16,7 +16,8 @@ services: runtime: dotnet_10 env: ASPNETCORE_URLS: http://+:8088 - COUNTDOWN_DELAY_SECONDS: "1" + IDEMPOTENT_SERVICE_ENDPOINT: ${IDEMPOTENT_SERVICE_ENDPOINT} + IDEMPOTENT_OPERATION_SCOPE: ${IDEMPOTENT_OPERATION_SCOPE} container: resources: cpu: "0.5" diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted_Shared_Contributor_Setup/IdempotentServiceClient.cs b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted_Shared_Contributor_Setup/IdempotentServiceClient.cs new file mode 100644 index 00000000000..db8cd970e31 --- /dev/null +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted_Shared_Contributor_Setup/IdempotentServiceClient.cs @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Net.Http.Json; +using System.Text.Json; + +namespace Hosted_Shared_Contributor_Setup; + +/// +/// Calls the idempotent operation service used by the resilience samples. +/// +/// +/// +/// For demonstration purposes only. This sample client and its backing service should not be used as-is in production. +/// +/// +/// Recovery may execute a workflow step again if its checkpoint was not saved before a crash. +/// Services called by that step must handle repeated requests without repeating unintended downstream effects. +/// This sample reuses the same scope and operation ID so the service can return the previously stored result. +/// Idempotency is enforced by the service, not by this client. +/// +/// +public sealed class IdempotentServiceClient(HttpClient httpClient) +{ + /// + /// Executes an operation or returns its previously stored result. + /// + /// The operation scope. + /// The operation identifier within the scope. + /// The cancellation token. + /// The operation result. + public async Task ExecuteOperationAsync( + string scope, + int operationId, + CancellationToken cancellationToken = default) + { + using HttpResponseMessage response = await httpClient.PostAsync( + new Uri($"operations/{Uri.EscapeDataString(scope)}/{operationId}", UriKind.Relative), + content: null, + cancellationToken); + response.EnsureSuccessStatusCode(); + await using Stream content = await response.Content.ReadAsStreamAsync(cancellationToken); + using JsonDocument document = await JsonDocument.ParseAsync(content, cancellationToken: cancellationToken); + return document.RootElement.GetProperty("result").GetString() + ?? throw new InvalidOperationException("The idempotent service returned an empty result."); + } + + /// + /// Gets the number of completed operations in a scope. + /// + /// The operation scope. + /// The cancellation token. + /// The number of completed operations. + public async Task GetOperationCountAsync(string scope, CancellationToken cancellationToken = default) + { + int? count = await httpClient.GetFromJsonAsync( + new Uri($"operations/{Uri.EscapeDataString(scope)}/count", UriKind.Relative), + cancellationToken); + return count + ?? throw new InvalidOperationException("The idempotent service returned an empty operation count."); + } +} diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/IdempotentService.cs b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/IdempotentService.cs new file mode 100644 index 00000000000..d58032f5fa6 --- /dev/null +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/IdempotentService.cs @@ -0,0 +1,140 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Globalization; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; + +/// +/// Hosts the SQLite-backed operation service used by the resilience demonstration. +/// +/// +/// +/// For demonstration purposes only. This service and its client should not be used as-is in production. +/// The service runs in a separate process so its operation records survive the hosted workflow process being replaced. +/// +/// +/// Recovery may execute an unconfirmed workflow step again. The same scope and operation ID return the stored result +/// instead of adding another row. Reusing the result prevents a repeated service effect, not repeated stream text. +/// The database insert is the simulated effect; this does not make an external email or payment call transactional. +/// +/// +internal static class IdempotentService +{ + /// + /// Runs the service when this executable is started with --idempotent-service. + /// + /// Host arguments after removing --idempotent-service. + public static async Task RunAsync(string[] args) + { + string databasePath = System.Environment.GetEnvironmentVariable("IDEMPOTENT_SERVICE_DATABASE_PATH") + ?? Path.Combine(AppContext.BaseDirectory, "operations.db"); + var store = new SqliteOperationStore(databasePath); + + var builder = WebApplication.CreateBuilder(args); + builder.Services.AddSingleton(store); + + await using var app = builder.Build(); + app.MapGet("/readiness", () => Results.Ok()); + + app.MapPost("/operations/{scope}/{operationId:int}", + async ( + string scope, + int operationId, + SqliteOperationStore operationStore, + CancellationToken cancellationToken) => + { + OperationResult operation = await operationStore.ExecuteAsync(scope, operationId, cancellationToken); + return Results.Ok(operation); + }); + + app.MapGet("/operations/{scope}/count", + async (string scope, SqliteOperationStore operationStore, CancellationToken cancellationToken) + => Results.Ok(await operationStore.GetCountAsync(scope, cancellationToken))); + + await app.RunAsync(); + } + + private sealed record OperationResult(string Result, bool Created); + + private sealed class SqliteOperationStore + { + private readonly string _connectionString; + + public SqliteOperationStore(string databasePath) + { + Directory.CreateDirectory(Path.GetDirectoryName(Path.GetFullPath(databasePath))!); + this._connectionString = new SqliteConnectionStringBuilder + { + DataSource = databasePath, + }.ToString(); + + using var connection = new SqliteConnection(this._connectionString); + connection.Open(); + using SqliteCommand command = connection.CreateCommand(); + command.CommandText = + """ + CREATE TABLE IF NOT EXISTS operations ( + scope TEXT NOT NULL, + operation_id INTEGER NOT NULL, + result TEXT NOT NULL, + PRIMARY KEY (scope, operation_id) + ); + """; + command.ExecuteNonQuery(); + } + + public async Task ExecuteAsync(string scope, int operationId, CancellationToken cancellationToken) + { + string result = operationId.ToString(CultureInfo.InvariantCulture); + await using var connection = new SqliteConnection(this._connectionString); + await connection.OpenAsync(cancellationToken); + + await using SqliteCommand insert = connection.CreateCommand(); + insert.CommandText = + """ + INSERT OR IGNORE INTO operations (scope, operation_id, result) + VALUES ($scope, $operationId, $result); + """; + insert.Parameters.AddWithValue("$scope", scope); + insert.Parameters.AddWithValue("$operationId", operationId); + insert.Parameters.AddWithValue("$result", result); + bool created = await insert.ExecuteNonQueryAsync(cancellationToken) == 1; + if (created) + { + Console.WriteLine($"Operation {scope}/{operationId} executed."); + return new(result, Created: true); + } + + await using SqliteCommand select = connection.CreateCommand(); + select.CommandText = + """ + SELECT result + FROM operations + WHERE scope = $scope AND operation_id = $operationId; + """; + select.Parameters.AddWithValue("$scope", scope); + select.Parameters.AddWithValue("$operationId", operationId); + string storedResult = (string?)await select.ExecuteScalarAsync(cancellationToken) + ?? throw new InvalidOperationException($"Operation {scope}/{operationId} exists without a stored result."); + Console.WriteLine($"Duplicate operation {scope}/{operationId} ignored."); + return new(storedResult, Created: false); + } + + public async Task GetCountAsync(string scope, CancellationToken cancellationToken) + { + await using var connection = new SqliteConnection(this._connectionString); + await connection.OpenAsync(cancellationToken); + await using SqliteCommand command = connection.CreateCommand(); + command.CommandText = + """ + SELECT COUNT(*) + FROM operations + WHERE scope = $scope; + """; + command.Parameters.AddWithValue("$scope", scope); + return Convert.ToInt32(await command.ExecuteScalarAsync(cancellationToken), CultureInfo.InvariantCulture); + } + } +} diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Program.cs b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Program.cs index 65136cce510..88aec8ecd26 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Program.cs +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Program.cs @@ -1,221 +1,31 @@ // Copyright (c) Microsoft. All rights reserved. -using System.ClientModel; -using System.ClientModel.Primitives; -using System.Diagnostics; -using System.Globalization; -using System.Net; -using System.Net.Sockets; -using System.Text; -using System.Text.Json; -using Azure.AI.Projects; +// Demonstrates workflow recovery and service-side idempotency across process interruptions. +// Sample only, not a production implementation. Repeated stream text is displayed, not used to execute operations. + using Hosted_Shared_Contributor_Setup; using Microsoft.Agents.AI; using Microsoft.Extensions.AI; -using OpenAI.Responses; +using static ServerManager; -const string AgentName = "hosted-workflow-resilient-long-running"; -VerificationOptions options = VerificationOptions.Parse(args); -string repositoryRoot = FindRepositoryRoot(); -string serverProject = Path.Combine( - repositoryRoot, - "dotnet", - "samples", - "04-hosting", - "FoundryHostedAgents", - "responses", - "Hosted-Workflow-Resilient-Long-Running", - "HostedWorkflowResilientLongRunning.csproj"); -string workingRoot = Path.Combine( - Path.GetTempPath(), - $"maf-resilient-workflow-{Guid.NewGuid():N}"); -string serverOutput = Path.Combine(workingRoot, "server"); -string serverAssembly = Path.Combine( - serverOutput, - "HostedWorkflowResilientLongRunning.dll"); -string stateRoot = Path.Combine(workingRoot, "state"); -string logPath = Path.Combine( - Path.GetTempPath(), - $"maf-resilient-workflow-{Guid.NewGuid():N}.log"); -int port = GetAvailablePort(); -var baseAddress = new Uri($"http://127.0.0.1:{port}"); -bool succeeded = false; +if (args is ["--idempotent-service", .. var serviceArgs]) +{ + await IdempotentService.RunAsync(serviceArgs); + return; +} -Directory.CreateDirectory(workingRoot); +using var cancellationSource = new CancellationTokenSource(TimeSpan.FromMinutes(6)); +string interruptNumberMessage = "10"; -var cancellationSource = new CancellationTokenSource(TimeSpan.FromMinutes(3)); -var client = new HttpClient -{ - BaseAddress = baseAddress, - Timeout = Timeout.InfiniteTimeSpan, -}; -await using var logWriter = new StreamWriter(logPath, append: false, new UTF8Encoding(false)) -{ - AutoFlush = true, -}; -ServerProcess? server = null; -Task? createStream = null; -AgentStreamObserver? streamObserver = null; -LocalAgentClient? localAgentClient = null; -CancellationTokenSource? initialStreamCancellation = null; try { - PrintHeader(options, stateRoot, logPath); - - Console.WriteLine("Preparing isolated Debug server binaries..."); - await BuildServerAsync( - serverProject, - serverOutput, - logWriter, - cancellationSource.Token); - Console.WriteLine(" server build complete"); - Console.WriteLine(); - - Console.WriteLine("[1/7] Starting the first server process..."); - Console.WriteLine($" endpoint: {baseAddress}"); - server = StartServer( - serverAssembly, - stateRoot, - port, - options.DelaySeconds, - logWriter); - Console.WriteLine($" process tree root: {server.Id}"); - await WaitForReadinessAsync(client, cancellationSource.Token); - Console.WriteLine(" server ready"); - Console.WriteLine(); - - localAgentClient = CreateClientAgent(baseAddress, AgentName); - AIAgent agent = localAgentClient.Agent; - AgentSession session = await agent.CreateSessionAsync(cancellationSource.Token); - AgentRunOptions runOptions = new() { AllowBackgroundResponses = true }; - - Console.WriteLine("[2/7] Starting the background countdown..."); - streamObserver = new AgentStreamObserver(options.CrashAfterCount); - initialStreamCancellation = - CancellationTokenSource.CreateLinkedTokenSource(cancellationSource.Token); -#pragma warning disable CA2025 // The stream must run concurrently until the server is killed; finally awaits it before disposing resources. - createStream = WatchInitialAgentStreamAsync( - agent, - session, - runOptions, - options.Target, - streamObserver, - initialStreamCancellation.Token); -#pragma warning restore CA2025 - - string responseId = await WaitForResponseIdAsync( - streamObserver, - createStream, - cancellationSource.Token); - Console.WriteLine($" response id: {responseId}"); - Console.WriteLine(); - - Console.WriteLine( - $"[3/7] Waiting for {options.CrashAfterCount} countdown items and their response checkpoint..."); - await streamObserver.CrashPointReached.Task.WaitAsync(cancellationSource.Token); - await WaitForPersistedResponseCheckpointAsync( - stateRoot, - responseId, - streamObserver.CompletedTexts, - cancellationSource.Token); - Console.WriteLine(" checkpoint persisted"); - Console.WriteLine(); - - Console.WriteLine("[4/7] Force-killing the first server process..."); - initialStreamCancellation.Cancel(); - await IgnoreExpectedDisconnectAsync(createStream); - createStream = null; - initialStreamCancellation.Dispose(); - initialStreamCancellation = null; - await server.KillAsync(); - server = null; - DeleteStaleStreamLocks(stateRoot); - Console.WriteLine(" process terminated"); - Console.WriteLine(); - - Console.WriteLine("[5/7] Starting a replacement server over the same durable state..."); - server = StartServer( - serverAssembly, - stateRoot, - port, - options.DelaySeconds, - logWriter); - Console.WriteLine($" process tree root: {server.Id}"); - await WaitForReadinessAsync(client, cancellationSource.Token); - Console.WriteLine(" recovery scan completed"); - Console.WriteLine(); - Console.WriteLine("[6/7] Reconnecting with the sequence-aware continuation token..."); - streamObserver.BeginRecovery(); - runOptions.ContinuationToken = streamObserver.ContinuationToken - ?? throw new InvalidOperationException( - "The initial stream did not provide a continuation token."); - await WatchRecoveredAgentStreamAsync( - agent, - session, - runOptions, - streamObserver, - cancellationSource.Token); - - List actual = streamObserver.CompletedTexts; - List expected = - [ - .. Enumerable.Range(1, options.Target) - .Reverse() - .Select(value => value.ToString(CultureInfo.InvariantCulture)), - "Countdown complete.", - ]; - - if (!streamObserver.ResponseCompleted) - { - throw new InvalidOperationException( - "The recovered stream ended without response.completed."); - } - - if (!actual.SequenceEqual(expected)) - { - throw new InvalidOperationException( - "Recovered output did not match the expected countdown." + - $"{Environment.NewLine}Expected: {string.Join(", ", expected)}" + - $"{Environment.NewLine}Actual: {string.Join(", ", actual)}"); - } - - Console.WriteLine(); - Console.WriteLine("[7/7] Replaying from the start without a sequence cursor..."); - AgentRunOptions replayOptions = new() - { - AllowBackgroundResponses = true, - ContinuationToken = CreateReplayFromStartToken(responseId), - }; - var replayObserver = new AgentStreamObserver(int.MaxValue); - await WatchReplayedAgentStreamAsync( - agent, - session, - replayOptions, - replayObserver, - cancellationSource.Token); - if (!replayObserver.ResponseCompleted - || !replayObserver.CompletedTexts.SequenceEqual(expected)) - { - throw new InvalidOperationException( - "The cursor-free replay did not return the complete countdown."); - } - int retainedCountdownUpdates = - actual.Count(text => text != "Countdown complete."); - int replayedCountdownUpdates = - replayObserver.CompletedTexts.Count( - text => text != "Countdown complete."); - Console.WriteLine(); - Console.WriteLine( - $"Client retained countdown updates: {retainedCountdownUpdates}"); - Console.WriteLine( - $"Replay countdown updates: {replayedCountdownUpdates}"); + await RunScenarioAsync(interruptNumberMessage, InterruptionKind.Crash, cancellationSource.Token); + await RunScenarioAsync(interruptNumberMessage, InterruptionKind.Shutdown, cancellationSource.Token); Console.WriteLine(); Console.ForegroundColor = ConsoleColor.Green; - Console.WriteLine( - "PASS: crash recovery completed with ordered output and no missing or duplicated items."); + Console.WriteLine("PASS: both recovery paths completed with every countdown operation stored once in SQLite."); Console.ResetColor(); - succeeded = true; } catch (Exception exception) { @@ -223,749 +33,116 @@ await WatchReplayedAgentStreamAsync( Console.ForegroundColor = ConsoleColor.Red; Console.Error.WriteLine($"FAIL: {exception.Message}"); Console.ResetColor(); - Console.Error.WriteLine($"Server log: {logPath}"); System.Environment.ExitCode = 1; } -finally -{ - if (server is not null) - { - await server.KillAsync(); - } - - if (createStream is not null) - { - initialStreamCancellation?.Cancel(); - await IgnoreExpectedDisconnectAsync(createStream); - } - - initialStreamCancellation?.Dispose(); - client.Dispose(); - localAgentClient?.Dispose(); - cancellationSource.Dispose(); - - if (succeeded) - { - TryDeleteDirectory(workingRoot); - } - else - { - Console.Error.WriteLine($"E2E working directory retained at: {workingRoot}"); - } -} -static void PrintHeader( - VerificationOptions options, - string stateRoot, - string logPath) -{ - Console.ForegroundColor = ConsoleColor.Cyan; - Console.WriteLine("============================================================"); - Console.WriteLine("Resilient long-running workflow E2E demonstration"); - Console.WriteLine("============================================================"); - Console.ResetColor(); - Console.WriteLine($"Countdown target: {options.Target}"); - Console.WriteLine($"Crash after: {options.CrashAfterCount} message items"); - Console.WriteLine($"Step delay: {options.DelaySeconds} second(s)"); - Console.WriteLine($"Durable state: {stateRoot}"); - Console.WriteLine($"Server log: {logPath}"); - Console.WriteLine(); -} - -static ServerProcess StartServer( - string serverAssembly, - string stateRoot, - int port, - int delaySeconds, - TextWriter logWriter) -{ - var startInfo = new ProcessStartInfo - { - FileName = "dotnet", - WorkingDirectory = Path.GetDirectoryName(serverAssembly)!, - RedirectStandardOutput = true, - RedirectStandardError = true, - UseShellExecute = false, - CreateNoWindow = true, - }; - startInfo.ArgumentList.Add("exec"); - startInfo.ArgumentList.Add(serverAssembly); - startInfo.Environment["AGENTSERVER_STATE_ROOT"] = stateRoot; - startInfo.Environment["FOUNDRY_AGENT_SESSION_ID"] = "using-e2e-resilience"; - startInfo.Environment["AGENT_NAME"] = AgentName; - startInfo.Environment["ASPNETCORE_URLS"] = $"http://127.0.0.1:{port}"; - startInfo.Environment["ASPNETCORE_ENVIRONMENT"] = "Development"; - startInfo.Environment["COUNTDOWN_DELAY_SECONDS"] = - delaySeconds.ToString(CultureInfo.InvariantCulture); - startInfo.Environment["DOTNET_NOLOGO"] = "true"; - startInfo.Environment.Remove("FOUNDRY_HOSTING_ENVIRONMENT"); - - return ServerProcess.Start(startInfo, logWriter); -} - -static LocalAgentClient CreateClientAgent(Uri baseAddress, string agentName) -{ - Uri httpsProjectEndpoint = new UriBuilder(baseAddress) - { - Scheme = Uri.UriSchemeHttps, - Port = baseAddress.Port, - }.Uri; - - var transportClient = new HttpClient( - new LocalHttpSchemeRewriteHandler(baseAddress)); - var clientOptions = new AIProjectClientOptions - { - Transport = new HttpClientPipelineTransport(transportClient), - }; - - AIAgent agent = new AIProjectClient( - httpsProjectEndpoint, - new LocalDevelopmentTokenCredential(), - clientOptions) - .AsAIAgent( - model: agentName, - instructions: "Invoke the local hosted countdown workflow."); - return new LocalAgentClient(agent, transportClient); -} - -static ResponseContinuationToken CreateReplayFromStartToken( - string responseId) -{ - ResponseContinuationToken innerToken = - ResponseContinuationToken.FromBytes( - JsonSerializer.SerializeToUtf8Bytes( - new { responseId })); - string serializedInnerToken = JsonSerializer.Serialize( - innerToken, - AgentAbstractionsJsonUtilities.DefaultOptions.GetTypeInfo( - typeof(ResponseContinuationToken))); - byte[] bytes = JsonSerializer.SerializeToUtf8Bytes( - new - { - type = "chatClientAgentContinuationToken", - innerToken = serializedInnerToken, - }); - return ResponseContinuationToken.FromBytes(bytes); -} - -static async Task BuildServerAsync( - string serverProject, - string serverOutput, - TextWriter logWriter, +static async Task RunScenarioAsync( + string interruptNumberMessage, + InterruptionKind interruptionKind, CancellationToken cancellationToken) { - var startInfo = new ProcessStartInfo - { - FileName = "dotnet", - WorkingDirectory = Path.GetDirectoryName(serverProject)!, - RedirectStandardOutput = true, - RedirectStandardError = true, - UseShellExecute = false, - CreateNoWindow = true, - }; - startInfo.ArgumentList.Add("build"); - startInfo.ArgumentList.Add(serverProject); - startInfo.ArgumentList.Add("--configuration"); - startInfo.ArgumentList.Add("Debug"); - startInfo.ArgumentList.Add("--output"); - startInfo.ArgumentList.Add(serverOutput); - startInfo.ArgumentList.Add("--tl:off"); - startInfo.Environment["DOTNET_NOLOGO"] = "true"; + await using var serverManager = new ServerManager(interruptionKind); - using Process process = Process.Start(startInfo) - ?? throw new InvalidOperationException("Could not start the server build."); - TextWriter synchronizedLogWriter = TextWriter.Synchronized(logWriter); - process.OutputDataReceived += (_, eventArgs) => - { - if (eventArgs.Data is not null) - { - synchronizedLogWriter.WriteLine($"[build stdout] {eventArgs.Data}"); - } - }; - process.ErrorDataReceived += (_, eventArgs) => - { - if (eventArgs.Data is not null) - { - synchronizedLogWriter.WriteLine($"[build stderr] {eventArgs.Data}"); - } - }; - process.BeginOutputReadLine(); - process.BeginErrorReadLine(); + PrintHeader(interruptionKind, serverManager); - await process.WaitForExitAsync(cancellationToken); - process.WaitForExit(); - if (process.ExitCode != 0) - { - throw new InvalidOperationException( - $"Server build failed with exit code {process.ExitCode}."); - } -} + Console.WriteLine($"[{interruptionKind} 1/6] Building the hosted server..."); + await serverManager.BuildServerAsync(cancellationToken); -static async Task WaitForReadinessAsync( - HttpClient client, - CancellationToken cancellationToken) -{ - var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(30); - while (DateTimeOffset.UtcNow < deadline) - { - try - { - using var requestCancellation = - CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - requestCancellation.CancelAfter(TimeSpan.FromSeconds(2)); - using HttpResponseMessage response = await client.GetAsync( - new Uri("readiness", UriKind.Relative), - requestCancellation.Token); - if (response.StatusCode == HttpStatusCode.OK) - { - return; - } - } - catch (Exception exception) - when (exception is HttpRequestException or TaskCanceledException) - { - } + Console.WriteLine($"[{interruptionKind} 2/6] Starting the idempotent service and hosted server..."); + Console.WriteLine($" Idempotent service Process ID: {await serverManager.StartIdempotentServiceAsync(cancellationToken)}"); + Console.WriteLine($" Hosted server Process ID: {await serverManager.StartHostedAgentServerAsync(cancellationToken)}"); - await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken); - } - - throw new TimeoutException("Server did not become ready within 30 seconds."); -} - -static async Task WatchInitialAgentStreamAsync( - AIAgent agent, - AgentSession session, - AgentRunOptions options, - int target, - AgentStreamObserver observer, - CancellationToken cancellationToken) -{ - await foreach (AgentResponseUpdate update in agent.RunStreamingAsync( - $"Count down from {target}", - session, - options, - cancellationToken)) - { - observer.ObserveInitial(update); - } -} + AIAgent agent = serverManager.GetAIAgent(); + AgentSession session = await agent.CreateSessionAsync(cancellationToken); + IdempotentServiceClient idempotentService = new(new HttpClient { BaseAddress = serverManager.IdempotentServiceBaseAddress }); -static async Task WatchRecoveredAgentStreamAsync( - AIAgent agent, - AgentSession session, - AgentRunOptions options, - AgentStreamObserver observer, - CancellationToken cancellationToken) -{ - await foreach (AgentResponseUpdate update in agent.RunStreamingAsync( - session, - options, - cancellationToken)) - { - observer.ObserveRecovered(update); - } -} + Console.WriteLine($"[{interruptionKind} 3/6] Starting the background response..."); -static async Task WatchReplayedAgentStreamAsync( - AIAgent agent, - AgentSession session, - AgentRunOptions options, - AgentStreamObserver observer, - CancellationToken cancellationToken) -{ - await foreach (AgentResponseUpdate update in agent.RunStreamingAsync( - session, - options, - cancellationToken)) + ResponseContinuationToken responseToken; + await using (var response = agent.RunStreamingAsync( + interruptNumberMessage, session, new AgentRunOptions { AllowBackgroundResponses = true }, cancellationToken) + .GetAsyncEnumerator(cancellationToken)) { - observer.ObserveReplayed(update); - } -} - -static async Task WaitForResponseIdAsync( - AgentStreamObserver observer, - Task createStream, - CancellationToken cancellationToken) -{ - Task completed = await Task.WhenAny( - observer.ResponseId.Task, - createStream, - Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken)); - if (completed == createStream) - { - await createStream; - throw new InvalidOperationException( - "The initial stream ended before returning a response ID."); - } - - cancellationToken.ThrowIfCancellationRequested(); - return await observer.ResponseId.Task; -} - -static async Task WaitForPersistedResponseCheckpointAsync( - string stateRoot, - string responseId, - IReadOnlyList expectedPrefix, - CancellationToken cancellationToken) -{ - string path = Path.Combine( - stateRoot, - "responses", - "envelopes", - $"{responseId}.json"); - var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(15); - while (DateTimeOffset.UtcNow < deadline) - { - try - { - using FileStream file = new( - path, - FileMode.Open, - FileAccess.Read, - FileShare.ReadWrite | FileShare.Delete); - using JsonDocument document = await JsonDocument.ParseAsync( - file, - cancellationToken: cancellationToken); - JsonElement response = - document.RootElement.GetProperty("envelope"); - List persistedTexts = GetPersistedMessageTexts(response); - bool hasCheckpointMetadata = - response.TryGetProperty("metadata", out JsonElement metadata) - && metadata.TryGetProperty("_internal_metadata", out JsonElement internalMetadata) - && !string.IsNullOrWhiteSpace(internalMetadata.GetString()); - if (hasCheckpointMetadata - && persistedTexts.Count >= expectedPrefix.Count - && persistedTexts - .Take(expectedPrefix.Count) - .SequenceEqual(expectedPrefix)) - { - return; - } - } - catch (Exception exception) - when (exception is IOException - or JsonException - or KeyNotFoundException) - { - } - - await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); - } - - throw new TimeoutException( - "The response checkpoint was not persisted within 15 seconds."); -} - -static List GetPersistedMessageTexts(JsonElement response) -{ - List texts = []; - foreach (JsonElement item in response.GetProperty("output").EnumerateArray()) - { - if (item.GetProperty("type").GetString() != "message") + if (!await response.MoveNextAsync()) { - continue; + throw new InvalidOperationException("The background response ended before it was accepted."); } - foreach (JsonElement content in item.GetProperty("content").EnumerateArray()) - { - if (content.GetProperty("type").GetString() == "output_text") - { - texts.Add(content.GetProperty("text").GetString() ?? string.Empty); - } - } + responseToken = response.Current.ContinuationToken + ?? throw new InvalidOperationException("The accepted response did not provide a continuation token."); } - return texts; -} + Console.WriteLine($"[{interruptionKind} 4/6] Waiting for operation {interruptNumberMessage}..."); -static async Task IgnoreExpectedDisconnectAsync(Task streamTask) -{ - try - { - await streamTask; - } - catch (Exception exception) - when (IsExpectedDisconnect(exception)) + var followOptions = new AgentRunOptions { - } - - static bool IsExpectedDisconnect(Exception exception) - { - if (exception is AggregateException aggregate) - { - return aggregate - .Flatten() - .InnerExceptions - .All(IsExpectedDisconnect); - } - - return exception is ClientResultException - or HttpRequestException - or IOException - or OperationCanceledException; - } -} - -static void DeleteStaleStreamLocks(string stateRoot) -{ - string streamsPath = Path.Combine(stateRoot, "streams"); - if (!Directory.Exists(streamsPath)) - { - return; - } - - foreach (string lockPath in Directory.EnumerateFiles( - streamsPath, - "*.jsonl.lock", - SearchOption.TopDirectoryOnly)) - { - for (int attempt = 1; attempt <= 10; attempt++) - { - try - { - File.Delete(lockPath); - break; - } - catch (UnauthorizedAccessException) when (attempt < 10) - { - Thread.Sleep(TimeSpan.FromMilliseconds(250)); - } - catch (IOException) when (attempt < 10) - { - Thread.Sleep(TimeSpan.FromMilliseconds(250)); - } - } - } -} - -static int GetAvailablePort() -{ - var listener = new TcpListener(IPAddress.Loopback, 0); - listener.Start(); - int port = ((IPEndPoint)listener.LocalEndpoint).Port; - listener.Stop(); - return port; -} - -static string FindRepositoryRoot() -{ - foreach (string start in new[] { Environment.CurrentDirectory, AppContext.BaseDirectory }) - { - DirectoryInfo? directory = new(start); - while (directory is not null) - { - if (File.Exists(Path.Combine( - directory.FullName, - "dotnet", - "agent-framework-dotnet.slnx"))) - { - return directory.FullName; - } - - directory = directory.Parent; - } - } - - throw new InvalidOperationException( - "Could not find the Agent Framework repository root."); -} + AllowBackgroundResponses = true, + ContinuationToken = responseToken, + }; -static void TryDeleteDirectory(string path) -{ try { - Directory.Delete(path, recursive: true); - } - catch (IOException) - { - } - catch (UnauthorizedAccessException) - { - } -} - -internal sealed class AgentStreamObserver(int crashAfterCount) -{ - private readonly Dictionary _messageBuffers = - new(StringComparer.Ordinal); - private readonly HashSet _completedMessageIds = - new(StringComparer.Ordinal); - private List? _preCrashTexts; - private bool? _recoveryIncludesSnapshot; - private int _recoverySnapshotIndex; - private int _messageCount; - - public TaskCompletionSource ResponseId { get; } = - new(TaskCreationOptions.RunContinuationsAsynchronously); - - public TaskCompletionSource CrashPointReached { get; } = - new(TaskCreationOptions.RunContinuationsAsynchronously); - - public List CompletedTexts { get; } = []; - - public ResponseContinuationToken? ContinuationToken { get; private set; } - - public bool ResponseCompleted { get; private set; } - - public void BeginRecovery() - { - this._preCrashTexts = [.. this.CompletedTexts]; - this._recoveryIncludesSnapshot = null; - this._recoverySnapshotIndex = 0; - } - - public void ObserveInitial(AgentResponseUpdate update) => - this.Observe(update, "before", trackCheckpoint: true); - - public void ObserveRecovered(AgentResponseUpdate update) => - this.Observe(update, "recovered", trackCheckpoint: false); - - public void ObserveReplayed(AgentResponseUpdate update) => - this.Observe(update, "replayed", trackCheckpoint: false); - - private void Observe( - AgentResponseUpdate update, - string phase, - bool trackCheckpoint) - { - object? rawRepresentation = - update.RawRepresentation is ChatResponseUpdate chatResponseUpdate - ? chatResponseUpdate.RawRepresentation - : update.RawRepresentation; - - if (update.ContinuationToken is { } continuationToken) - { - this.ContinuationToken = continuationToken; - } - - if (!string.IsNullOrWhiteSpace(update.ResponseId)) - { - this.ResponseId.TrySetResult(update.ResponseId); - } - - if (!string.IsNullOrWhiteSpace(update.MessageId) - && !string.IsNullOrEmpty(update.Text)) - { - if (!this._messageBuffers.TryGetValue( - update.MessageId, - out StringBuilder? buffer)) - { - buffer = new StringBuilder(); - this._messageBuffers[update.MessageId] = buffer; - } - - buffer.Append(update.Text); - } - - if (rawRepresentation is StreamingResponseOutputItemDoneUpdate - { - Item: MessageResponseItem message - } - && this._completedMessageIds.Add(message.Id)) + await foreach (AgentResponseUpdate update in agent.RunStreamingAsync(session, followOptions, cancellationToken)) { - string text = this._messageBuffers.TryGetValue( - message.Id, - out StringBuilder? buffer) - ? buffer.ToString() - : string.Empty; - if (phase != "before" && text.Length == 0) - { - return; - } - - if (phase == "recovered" - && this.TryHandleRecoverySnapshot(text)) + if (!string.IsNullOrEmpty(update.Text)) { - return; + Console.WriteLine($" {update.Text}"); } - this.CompletedTexts.Add(text); - WriteOutput(phase, text); - - if (trackCheckpoint && ++this._messageCount >= crashAfterCount) + if (update.Text.Contains(interruptNumberMessage, StringComparison.OrdinalIgnoreCase)) { - this.CrashPointReached.TrySetResult(); + if (interruptionKind == InterruptionKind.Crash) + { + Console.WriteLine(" Killing / crashing the server ..."); + await serverManager.CrashServerAsync(); + serverManager.DeleteStaleStreamLocks(); + } + else + { + Console.WriteLine(" Shutting down the server ..."); + await serverManager.RequestShutdownAsync(cancellationToken); + await serverManager.WaitForServerExitAsync(cancellationToken); + } } } - - if (rawRepresentation is StreamingResponseCompletedUpdate) - { - this.ResponseCompleted = true; - } } - - private bool TryHandleRecoverySnapshot(string text) + catch { - if (this._preCrashTexts is not { Count: > 0 } preCrashTexts) - { - return false; - } - - this._recoveryIncludesSnapshot ??= - string.Equals(text, preCrashTexts[0], StringComparison.Ordinal); - if (this._recoveryIncludesSnapshot is not true) - { - return false; - } - - if (this._recoverySnapshotIndex >= preCrashTexts.Count) - { - return false; - } - - if (!string.Equals( - text, - preCrashTexts[this._recoverySnapshotIndex], - StringComparison.Ordinal)) - { - throw new InvalidOperationException( - "The response snapshot returned during reconnection did not match the pre-crash output."); - } - - this._recoverySnapshotIndex++; - WriteOutput("restored", text); - return true; - } - - private static void WriteOutput(string phase, string text) - { - Console.ForegroundColor = phase == "recovered" - ? ConsoleColor.Green - : ConsoleColor.DarkGray; - Console.WriteLine($" {phase,-9} > {text}"); - Console.ResetColor(); - } -} - -internal sealed class ServerProcess -{ - private readonly Process _process; - private readonly Task _outputPump; - private readonly Task _errorPump; - - private ServerProcess(Process process, TextWriter logWriter) - { - this._process = process; - this._outputPump = PumpAsync(process.StandardOutput, logWriter, "stdout"); - this._errorPump = PumpAsync(process.StandardError, logWriter, "stderr"); + Console.WriteLine(" The connection was interrupted."); } - public int Id => this._process.Id; + Console.WriteLine($"[{interruptionKind} 5/6] Starting the replacement server..."); + Console.WriteLine($" Process ID: {await serverManager.StartHostedAgentServerAsync(cancellationToken)}"); - public static ServerProcess Start( - ProcessStartInfo startInfo, - TextWriter logWriter) + Console.WriteLine($"[{interruptionKind} 6/6] Reading the recovered response..."); + var recoveryOptions = new AgentRunOptions { - Process process = Process.Start(startInfo) - ?? throw new InvalidOperationException("Could not start the server process."); - return new ServerProcess(process, TextWriter.Synchronized(logWriter)); - } - - public async Task KillAsync() - { - if (!this._process.HasExited) - { - this._process.Kill(entireProcessTree: true); - } - - await this._process.WaitForExitAsync(); - await Task.WhenAll(this._outputPump, this._errorPump) - .WaitAsync(TimeSpan.FromSeconds(5)); - this._process.Dispose(); - } + AllowBackgroundResponses = true, + ContinuationToken = responseToken, + }; - private static async Task PumpAsync( - StreamReader reader, - TextWriter writer, - string source) + List textUpdates = []; + await foreach (AgentResponseUpdate update in agent.RunStreamingAsync(session, recoveryOptions, cancellationToken)) { - while (await reader.ReadLineAsync() is { } line) + if (!string.IsNullOrEmpty(update.Text)) { - await writer.WriteLineAsync($"[{source}] {line}"); + Console.WriteLine($" {update.Text}{(textUpdates.Contains(update.Text) ? " > Repeated due to abrupt crash and recovery replay (idempotency matters here)" : "")}"); + textUpdates.Add(update.Text); } } -} -internal sealed class LocalAgentClient( - AIAgent agent, - HttpClient transportClient) : IDisposable -{ - public AIAgent Agent { get; } = agent; - - public void Dispose() => transportClient.Dispose(); + int operationCount = await idempotentService.GetOperationCountAsync(serverManager.OperationScope, cancellationToken); + Console.WriteLine($" Idempotent service contains {operationCount} completed operations."); + serverManager.MarkSucceeded(); } -internal sealed record VerificationOptions( - int Target, - int CrashAfterCount, - int DelaySeconds) +static void PrintHeader(InterruptionKind interruptionKind, ServerManager harness) { - public static VerificationOptions Parse(string[] args) - { - int target = 20; - int? crashAfterCount = null; - int delaySeconds = 1; - - for (int index = 0; index < args.Length; index++) - { - string argument = args[index]; - switch (argument) - { - case "--target": - target = ReadInteger(args, ref index, argument); - break; - case "--crash-after-count": - crashAfterCount = ReadInteger(args, ref index, argument); - break; - case "--delay-seconds": - delaySeconds = ReadInteger(args, ref index, argument); - break; - default: - throw new ArgumentException($"Unknown argument '{argument}'."); - } - } - - int resolvedCrashAfterCount = crashAfterCount ?? Math.Max(1, target / 2); - if (target < 2) - { - throw new ArgumentOutOfRangeException( - nameof(args), - "Target must be at least 2."); - } - - if (resolvedCrashAfterCount < 1 || resolvedCrashAfterCount >= target) - { - throw new ArgumentOutOfRangeException( - nameof(args), - "Crash count must be greater than zero and less than the target."); - } - - if (delaySeconds < 0) - { - throw new ArgumentOutOfRangeException( - nameof(args), - "Delay seconds must be zero or greater."); - } - - return new(target, resolvedCrashAfterCount, delaySeconds); - } - - private static int ReadInteger( - string[] args, - ref int index, - string argument) - { - if (++index >= args.Length - || !int.TryParse( - args[index], - NumberStyles.None, - CultureInfo.InvariantCulture, - out int value)) - { - throw new ArgumentException( - $"Argument '{argument}' requires an integer value."); - } - - return value; - } + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Cyan; + Console.WriteLine("============================================================"); + Console.WriteLine($"Running {(interruptionKind == InterruptionKind.Crash ? "Abrupt process crash" : "Host shutdown")} scenario ... "); + Console.WriteLine("============================================================"); + Console.ResetColor(); + Console.WriteLine(); } diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/README.md b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/README.md index 1741878c4db..69c44bb459a 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/README.md +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/README.md @@ -1,103 +1,123 @@ -# Using-E2E-Resilience - -A self-contained local E2E demonstration for -[`Hosted-Workflow-Resilient-Long-Running`](../Hosted-Workflow-Resilient-Long-Running/). -It owns both server process lifetimes, consumes their response stream, and prints every countdown -output in one console. - -The E2E creates a MAF client agent through `AIProjectClient.AsAIAgent(model, instructions)`. It -enables `AgentRunOptions.AllowBackgroundResponses`, consumes `AgentResponseUpdate` values, saves the -latest non-null `ResponseContinuationToken`, and supplies that token after the replacement server -starts. It does not implement the Responses HTTP or SSE protocol itself. - -The demonstration uses one MAF client agent and one agent session for three calls: - -1. Starts the hosted workflow server as a child process. -2. Creates a stored background streaming response through the MAF agent. -3. Prints countdown messages as MAF streaming updates arrive. -4. Waits until the matching workflow and response checkpoint is durable. -5. Force-kills the server process tree. -6. Starts a replacement server over the same AgentServer state. -7. The second call reconnects with the sequence-aware continuation token and prints only newly - recovered messages. -8. The third call uses the same agent and session with the same response ID but no sequence cursor, - replaying the entire stream from the start. -9. The E2E verifies that the client accumulator and cursor-free replay contain the same complete - countdown. +# Using-E2E-Resilience -## Run +This local E2E runs +[`Hosted-Workflow-Resilient-Long-Running`](../Hosted-Workflow-Resilient-Long-Running/) +through two recovery scenarios: -Run from the repository root: +1. `Crash`, which terminates the hosted workflow process abruptly. +2. `Shutdown`, which follows the graceful shutdown path and defers the response for recovery. -```powershell -dotnet run --project dotnet\samples\04-hosting\FoundryHostedAgents\responses\Using-E2E-Resilience -``` +> **Local demonstration only.** The E2E, countdown workflow, HTTP client, and SQLite service exist only +> to demonstrate recovery after a crash or graceful shutdown. The countdown makes step recovery +> visible, and the database insert is a simulated operation; no real email, payment, or other external +> action is performed. This is not a real business scenario or a template for production use or +> deployment to Foundry. + +## Three processes + +The E2E uses three processes while each scenario is running: + +| Process | Responsibility | +| --- | --- | +| `Using-E2E-Resilience` | Drives the scenario and consumes the recovered stream. | +| `Hosted-Workflow-Resilient-Long-Running` | Runs the resilient countdown workflow. | +| `Using-E2E-Resilience --idempotent-service` | Runs `IdempotentService` on Kestrel and stores operations in SQLite. | + +The idempotent service is a class inside this sample, not a separate project or demo. The E2E +launches another instance of its own executable with `--idempotent-service`. That process runs only +the service, not the Crash and Shutdown scenarios, and remains running while the hosted workflow +process is replaced. -The E2E program starts the first server, ends it abruptly, starts the replacement server with the -same durable state, and ends the replacement when the verification completes. A separately running -local server may remain open: the E2E uses a random port, isolated Debug binaries, and an isolated -AgentServer state directory. +Both the hosted workflow and the E2E use `IdempotentServiceClient` from +`Hosted_Shared_Contributor_Setup`: -No Azure project, model deployment, credentials, or second terminal is required. -The E2E builds the server in Debug into an isolated temporary directory, so it does not reuse or -overwrite the binaries of a separately running local server. +1. The hosted workflow posts operations to the idempotent service. +2. The idempotent service stores them in SQLite. +3. The E2E queries the same service for the completed operation count. -`AIProjectClient` requires an HTTPS endpoint before its bearer-token policy will run. The shared -`LocalHttpSchemeRewriteHandler` presents HTTPS to that pipeline, then routes the request to the -random loopback HTTP port at transport time. The handler rejects non-loopback targets. +The workflow calls the service for every countdown value. The service stores each operation once +using `(scope, operation_id)` as the SQLite primary key. If recovery repeats a workflow step, the +service returns the existing result instead of creating another row. -Example: +## Current flow + +`Program.cs` sets `interruptNumberMessage` to `"10"`. The hosted workflow adds ten to that numeric +input, so the countdown starts at 20 and the E2E interrupts it when it receives `10`. + +1. Build the hosted server and start the idempotent service and hosted server on separate loopback ports. +2. Start a streaming background response. Read its first update and save the continuation token. +3. Dispose that initial stream enumerator. This closes the client stream, not the accepted background operation. +4. Open a second stream with the token and interrupt the hosted process at the selected count. +5. Start a replacement hosted process with the same workflow state, service endpoint, and operation scope. +6. Open a third stream using the original token, then query the service for the completed operation count. + +The recovered stream is consumed independently of the second request. The client displays all text +updates and labels repeated text within that third request. It does not execute operations or +remove duplicates. With the current input, the expected database count is 20. The E2E prints the +returned count; it does not assert the count or the exact stream sequence. + +## Why idempotency matters + +An operation can finish at the service before the workflow's progress is durably confirmed. If the +hosted process stops during that interval, recovery may execute the step and call the service again. +The service's primary key makes the repeated call return the saved result without adding another row: ```text -[1/7] Starting the first server process... -[2/7] Starting the background countdown... - before > 20 - before > 19 - before > 18 -... -[4/7] Force-killing the first server process... -[5/7] Starting a replacement server over the same durable state... -[6/7] Reconnecting to the response stream... - recovered > 10 - recovered > 9 -... - recovered > Countdown complete. - -[7/7] Replaying from the start without a sequence cursor... - replayed > 20 - replayed > 19 -... - replayed > Countdown complete. - -Client retained countdown updates: 20 -Replay countdown updates: 20 - -PASS: crash recovery completed with ordered output and no missing or duplicated items. +Operation Crash/10 executed. +Duplicate operation Crash/10 ignored. ``` -## Options +Those are service log messages. Repeated text in the client stream can come from replaying old events +or running an unconfirmed step again, so repeated text alone does not prove a second service effect. +The timing of interruption determines whether the step must run again. -```powershell -dotnet run --project dotnet\samples\04-hosting\FoundryHostedAgents\responses\Using-E2E-Resilience -- ` - --target 30 ` - --crash-after-count 12 ` - --delay-seconds 1 -``` +**Workflow recovery, stream replay, and service idempotency have different jobs.** Checkpoints restore +workflow progress. Replay lets a client receive stored events again. Idempotency protects the +service's operation from being applied twice. Neither replay nor a checkpoint undoes an email or +payment already performed. Real downstream services must enforce that protection themselves. -| Option | Default | Meaning | +## Internal service mode + +The E2E starts service mode automatically with a random loopback address in `ASPNETCORE_URLS` and an +isolated SQLite file in `IDEMPOTENT_SERVICE_DATABASE_PATH`. Both are preserved while the hosted +workflow restarts. At the end, the E2E stops the service and removes temporary state on success. + +Crash and Shutdown use separate temporary databases. Within each scenario, the service and its +database remain alive across both hosted process lifetimes. Failed scenarios retain their files and +print the paths for investigation; successful cleanup is best effort. + +| Method | Path | Purpose | | --- | --- | --- | -| `--target` | `20` | First countdown value. Must be at least 2. | -| `--crash-after-count` | Half the target | Number of completed countdown messages before the crash. | -| `--delay-seconds` | `1` | Delay between countdown steps. | +| `GET` | `/readiness` | Reports that the service is ready. | +| `POST` | `/operations/{scope}/{operationId}` | Creates an operation or returns its stored result. | +| `GET` | `/operations/{scope}/count` | Returns the number of completed operations in the scope. | + +## What normally triggers each path + +Foundry sends `SIGTERM` when it intentionally stops a hosted agent container and can provide a +graceful shutdown window. This can happen during managed lifecycle operations such as session +compute deprovisioning, scale-in, or redeployment. + +An abrupt crash provides no shutdown window. Typical examples include an application process crash, +a forced process termination, and an out-of-memory kill. + +The local Shutdown scenario requests `StopAsync()` on the AgentServer task service, then stops the +web host. It exercises the shutdown signal without sending an OS signal on Windows. It does not +guarantee that a partially completed workflow step will never run again. -Server output is redirected to a temporary log whose path is printed at startup. Each run uses a -random local port and an isolated AgentServer state directory. Successful runs delete their durable -state. Failed runs retain state and print its path for investigation. +See [the hosted agent runtime contract](https://learn.microsoft.com/azure/foundry/agents/concepts/hosted-agent-contract) +and [recovery guidance](https://learn.microsoft.com/azure/foundry/agents/how-to/recover-long-running-work). -The second call's continuation token resumes after the last update consumed before the crash. -Previously consumed countdown messages are retained in the client accumulator and are not streamed -again. Only work after the durable checkpoint appears as `recovered`. +## Run + +No Azure project, model deployment, credentials, or second terminal is required. The E2E builds the +hosted workflow server and starts both server processes automatically. + +From the repository root: + +```powershell +dotnet run --project dotnet\samples\04-hosting\FoundryHostedAgents\responses\Using-E2E-Resilience +``` -For the third call, the E2E derives another valid `ChatClientAgent` continuation token whose inner -Responses token contains the same response ID without a sequence number. That call prints every -persisted stream item as `replayed`. +To change the demonstration, edit `interruptNumberMessage` in `Program.cs`. `--idempotent-service` +selects the internal service process instead of running the E2E scenarios. diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/ServerManager.cs b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/ServerManager.cs new file mode 100644 index 00000000000..d4c3b8056e8 --- /dev/null +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/ServerManager.cs @@ -0,0 +1,380 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.ClientModel.Primitives; +using System.Diagnostics; +using System.Net; +using System.Net.Sockets; +using System.Text; +using Azure.AI.Projects; +using Hosted_Shared_Contributor_Setup; +using Microsoft.Agents.AI; +using Microsoft.Extensions.AI; + +internal sealed class ServerManager(ServerManager.InterruptionKind interruptionKind) : IAsyncDisposable +{ + private const string AgentName = "hosted-workflow-resilient-long-running"; + private readonly string _repositoryRoot = FindRepositoryRoot(); + private readonly string _workingRoot = Path.Combine(Path.GetTempPath(), $"maf-resilience-{interruptionKind}-{Guid.NewGuid():N}"); + private readonly int _port = GetAvailablePort(); + private readonly int _servicePort = GetAvailablePort(); + private readonly StreamWriter _logWriter = new( + Path.Combine(Path.GetTempPath(), $"maf-resilience-{interruptionKind}-{Guid.NewGuid():N}.log"), + append: false, + new UTF8Encoding(false)) + { + AutoFlush = true, + }; + + private ServerProcess? _server; + private ServerProcess? _idempotentService; + private LocalAgentClient? _agentClient; + private bool _succeeded; + public string StateRoot => Path.Combine(this._workingRoot, "state"); + + public string LogPath => ((FileStream)this._logWriter.BaseStream).Name; + + public Uri HostedAgentBaseAddress => new($"http://127.0.0.1:{this._port}"); + + public Uri IdempotentServiceBaseAddress => new($"http://127.0.0.1:{this._servicePort}"); + + public string OperationScope => interruptionKind.ToString(); + + public HttpClient ControlClient { get; } = new() + { + Timeout = Timeout.InfiniteTimeSpan, + }; + + /// + /// Gets the instance pointing to the hosted server. + /// + /// The instance. + public AIAgent GetAIAgent() => (this._agentClient ??= CreateClientAgent(this.HostedAgentBaseAddress, AgentName)).Agent; + + public async Task BuildServerAsync(CancellationToken cancellationToken) + { + Directory.CreateDirectory(this._workingRoot); + string serverProject = Path.Combine( + this._repositoryRoot, + "dotnet", + "samples", + "04-hosting", + "FoundryHostedAgents", + "responses", + "Hosted-Workflow-Resilient-Long-Running", + "HostedWorkflowResilientLongRunning.csproj"); + await this.BuildProjectAsync(serverProject, Path.Combine(this._workingRoot, "server"), cancellationToken); + } + + public async Task StartIdempotentServiceAsync(CancellationToken cancellationToken) + { + if (this._idempotentService is not null) + { + throw new InvalidOperationException("The idempotent service is already running."); + } + + string serviceAssembly = Path.Combine(AppContext.BaseDirectory, "using-e2e-resilience.dll"); + var startInfo = new ProcessStartInfo + { + FileName = "dotnet", + WorkingDirectory = Path.GetDirectoryName(serviceAssembly)!, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + CreateNoWindow = true, + }; + startInfo.ArgumentList.Add("exec"); + startInfo.ArgumentList.Add(serviceAssembly); + startInfo.ArgumentList.Add("--idempotent-service"); + startInfo.Environment["ASPNETCORE_URLS"] = $"http://127.0.0.1:{this._servicePort}"; + startInfo.Environment["IDEMPOTENT_SERVICE_DATABASE_PATH"] = Path.Combine(this._workingRoot, "idempotent-service.db"); + startInfo.Environment["DOTNET_NOLOGO"] = "true"; + + this._idempotentService = ServerProcess.Start(startInfo, this._logWriter); + await WaitForReadinessAsync(this.ControlClient, this.IdempotentServiceBaseAddress, cancellationToken); + return this._idempotentService.Id; + } + + public async Task StartHostedAgentServerAsync(CancellationToken cancellationToken) + { + if (this._server is not null) + { + throw new InvalidOperationException("The server is already running."); + } + + string serverAssembly = Path.Combine(this._workingRoot, "server", "HostedWorkflowResilientLongRunning.dll"); + var startInfo = new ProcessStartInfo + { + FileName = "dotnet", + WorkingDirectory = Path.GetDirectoryName(serverAssembly)!, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + CreateNoWindow = true, + }; + startInfo.ArgumentList.Add("exec"); + startInfo.ArgumentList.Add(serverAssembly); + startInfo.Environment["AGENTSERVER_STATE_ROOT"] = this.StateRoot; + startInfo.Environment["FOUNDRY_AGENT_SESSION_ID"] = $"using-e2e-resilience-{interruptionKind}"; + startInfo.Environment["AGENT_NAME"] = AgentName; + startInfo.Environment["ASPNETCORE_URLS"] = $"http://127.0.0.1:{this._port}"; + startInfo.Environment["ASPNETCORE_ENVIRONMENT"] = "Development"; + startInfo.Environment["IDEMPOTENT_SERVICE_ENDPOINT"] = this.IdempotentServiceBaseAddress.AbsoluteUri; + startInfo.Environment["IDEMPOTENT_OPERATION_SCOPE"] = this.OperationScope; + startInfo.Environment["ENABLE_E2E_SHUTDOWN_ENDPOINT"] = + string.Equals(interruptionKind.ToString(), InterruptionKind.Shutdown.ToString(), StringComparison.OrdinalIgnoreCase) ? "true" : "false"; + startInfo.Environment["DOTNET_NOLOGO"] = "true"; + startInfo.Environment.Remove("FOUNDRY_HOSTING_ENVIRONMENT"); + + this._server = ServerProcess.Start(startInfo, this._logWriter); + await WaitForReadinessAsync(this.ControlClient, this.HostedAgentBaseAddress, cancellationToken); + return this._server.Id; + } + + public async Task CrashServerAsync() + { + ServerProcess server = this._server ?? throw new InvalidOperationException("The server is not running."); + await server.KillAsync(); + this._server = null; + } + + public async Task RequestShutdownAsync(CancellationToken cancellationToken) + { + using HttpResponseMessage response = await this.ControlClient.PostAsync( + new Uri(this.HostedAgentBaseAddress, "shutdown"), content: null, cancellationToken); + response.EnsureSuccessStatusCode(); + } + + public async Task WaitForServerExitAsync(CancellationToken cancellationToken) + { + ServerProcess server = this._server ?? throw new InvalidOperationException("The server is not running."); + await server.WaitForExitAsync(cancellationToken); + this._server = null; + } + + public void DeleteStaleStreamLocks() + { + string streamsPath = Path.Combine(this.StateRoot, "streams"); + if (!Directory.Exists(streamsPath)) + { + return; + } + + foreach (string lockPath in Directory.EnumerateFiles(streamsPath, "*.jsonl.lock", SearchOption.TopDirectoryOnly)) + { + for (int attempt = 1; attempt <= 10; attempt++) + { + try + { + File.Delete(lockPath); + break; + } + catch (UnauthorizedAccessException) when (attempt < 10) + { + Thread.Sleep(TimeSpan.FromMilliseconds(250)); + } + catch (IOException) when (attempt < 10) + { + Thread.Sleep(TimeSpan.FromMilliseconds(250)); + } + } + } + } + + public void MarkSucceeded() => this._succeeded = true; + + public async ValueTask DisposeAsync() + { + string logPath = this.LogPath; + if (this._server is not null) + { + await this._server.KillAsync(); + } + + if (this._idempotentService is not null) + { + await this._idempotentService.KillAsync(); + } + + this.ControlClient.Dispose(); + this._agentClient?.Dispose(); + await this._logWriter.DisposeAsync(); + + if (this._succeeded) + { + TryDeleteDirectory(this._workingRoot); + TryDeleteFile(logPath); + } + else + { + Console.Error.WriteLine($"E2E working directory retained at: {this._workingRoot}"); + Console.Error.WriteLine($"Server log: {logPath}"); + } + } + + private async Task BuildProjectAsync(string projectPath, string outputPath, CancellationToken cancellationToken) + { + var startInfo = new ProcessStartInfo + { + FileName = "dotnet", + WorkingDirectory = Path.GetDirectoryName(projectPath)!, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + CreateNoWindow = true, + }; + startInfo.ArgumentList.Add("build"); + startInfo.ArgumentList.Add(projectPath); + startInfo.ArgumentList.Add("--configuration"); + startInfo.ArgumentList.Add("Debug"); + startInfo.ArgumentList.Add("--output"); + startInfo.ArgumentList.Add(outputPath); + startInfo.ArgumentList.Add("--tl:off"); + startInfo.Environment["DOTNET_NOLOGO"] = "true"; + + using Process process = Process.Start(startInfo) + ?? throw new InvalidOperationException($"Could not build '{projectPath}'."); + TextWriter synchronizedLogWriter = TextWriter.Synchronized(this._logWriter); + process.OutputDataReceived += (_, eventArgs) => + { + if (eventArgs.Data is not null) + { + synchronizedLogWriter.WriteLine($"[build stdout] {eventArgs.Data}"); + } + }; + process.ErrorDataReceived += (_, eventArgs) => + { + if (eventArgs.Data is not null) + { + synchronizedLogWriter.WriteLine($"[build stderr] {eventArgs.Data}"); + } + }; + process.BeginOutputReadLine(); + process.BeginErrorReadLine(); + + await process.WaitForExitAsync(cancellationToken); + process.WaitForExit(); + if (process.ExitCode != 0) + { + throw new InvalidOperationException($"Build failed with exit code {process.ExitCode} for '{projectPath}'."); + } + } + + private static LocalAgentClient CreateClientAgent(Uri baseAddress, string agentName) + { + Uri httpsProjectEndpoint = new UriBuilder(baseAddress) + { + Scheme = Uri.UriSchemeHttps, + Port = baseAddress.Port, + }.Uri; + + var transportClient = new HttpClient(new LocalHttpSchemeRewriteHandler(baseAddress)); + var clientOptions = new AIProjectClientOptions + { + Transport = new HttpClientPipelineTransport(transportClient), + }; + + AIAgent agent = new AIProjectClient( + httpsProjectEndpoint, + new LocalDevelopmentTokenCredential(), + clientOptions) + .AsAIAgent( + model: agentName, + instructions: "Invoke the local hosted countdown workflow."); + return new LocalAgentClient(agent, transportClient); + } + + private static async Task WaitForReadinessAsync( + HttpClient client, Uri baseAddress, CancellationToken cancellationToken) + { + var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(30); + while (DateTimeOffset.UtcNow < deadline) + { + try + { + using var requestCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + requestCancellation.CancelAfter(TimeSpan.FromSeconds(2)); + using HttpResponseMessage response = await client.GetAsync(new Uri(baseAddress, "readiness"), requestCancellation.Token); + if (response.StatusCode == HttpStatusCode.OK) + { + return; + } + } + catch (Exception exception) when (exception is HttpRequestException or TaskCanceledException) + { + } + + await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken); + } + + throw new TimeoutException("Server did not become ready within 30 seconds."); + } + + private static int GetAvailablePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + int port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static string FindRepositoryRoot() + { + foreach (string start in new[] { Environment.CurrentDirectory, AppContext.BaseDirectory }) + { + DirectoryInfo? directory = new(start); + while (directory is not null) + { + if (File.Exists(Path.Combine(directory.FullName, "dotnet", "agent-framework-dotnet.slnx"))) + { + return directory.FullName; + } + + directory = directory.Parent; + } + } + + throw new InvalidOperationException("Could not find the Agent Framework repository root."); + } + + private static void TryDeleteDirectory(string path) + { + try + { + Directory.Delete(path, recursive: true); + } + catch (IOException) + { + } + catch (UnauthorizedAccessException) + { + } + } + + private static void TryDeleteFile(string path) + { + try + { + File.Delete(path); + } + catch (IOException) + { + } + catch (UnauthorizedAccessException) + { + } + } + + internal enum InterruptionKind + { + Crash, + Shutdown, + } + + private sealed class LocalAgentClient(AIAgent agent, HttpClient transportClient) : IDisposable + { + public AIAgent Agent { get; } = agent; + + public void Dispose() => transportClient.Dispose(); + } +} diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/ServerProcess.cs b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/ServerProcess.cs new file mode 100644 index 00000000000..951f6fe56a6 --- /dev/null +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/ServerProcess.cs @@ -0,0 +1,65 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics; + +internal sealed class ServerProcess +{ + private readonly Process _process; + private readonly Task _outputPump; + private readonly Task _errorPump; + private bool _disposed; + + private ServerProcess(Process process, TextWriter logWriter) + { + this._process = process; + this._outputPump = PumpAsync(process.StandardOutput, logWriter, "stdout"); + this._errorPump = PumpAsync(process.StandardError, logWriter, "stderr"); + } + + public int Id => this._process.Id; + + public static ServerProcess Start(ProcessStartInfo startInfo, TextWriter logWriter) + { + Process process = Process.Start(startInfo) + ?? throw new InvalidOperationException("Could not start the server process."); + return new ServerProcess(process, TextWriter.Synchronized(logWriter)); + } + + public async Task KillAsync() + { + if (!this._process.HasExited) + { + this._process.Kill(entireProcessTree: true); + } + + await this.CompleteAsync(); + } + + public async Task WaitForExitAsync(CancellationToken cancellationToken) + { + await this._process.WaitForExitAsync(cancellationToken); + await this.CompleteAsync(); + } + + private async Task CompleteAsync() + { + if (this._disposed) + { + return; + } + + await this._process.WaitForExitAsync(); + await Task.WhenAll(this._outputPump, this._errorPump) + .WaitAsync(TimeSpan.FromSeconds(5)); + this._process.Dispose(); + this._disposed = true; + } + + private static async Task PumpAsync(StreamReader reader, TextWriter writer, string source) + { + while (await reader.ReadLineAsync() is { } line) + { + await writer.WriteLineAsync($"[{source}] {line}"); + } + } +} diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Using-E2E-Resilience.csproj b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Using-E2E-Resilience.csproj index f450479e8e5..ebf75c79b5c 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Using-E2E-Resilience.csproj +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-E2E-Resilience/Using-E2E-Resilience.csproj @@ -14,10 +14,14 @@ - + + + + + diff --git a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-Samples/README.md b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-Samples/README.md index 543132c0c5b..78d518670bb 100644 --- a/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-Samples/README.md +++ b/dotnet/samples/04-hosting/FoundryHostedAgents/responses/Using-Samples/README.md @@ -48,8 +48,8 @@ never hits the TLS check. | [`Hosted-Toolbox-AuthPaths-Client/`](./Hosted-Toolbox-AuthPaths-Client/) | Hosted toolbox agents | Handles OAuth consent, function-tool approvals, and native MCP approvals. Use it with `Hosted-Toolbox-AuthPaths` or `Hosted-ToolboxMcpSkills`. | | [`SessionFilesClient/`](./SessionFilesClient/) | [`Hosted-Files`](../Hosted-Files/) | Same shape as `SimpleAgent`, framed around the bundled-files demo. | -For a self-contained crash-recovery demonstration that starts, interrupts, and restarts its own -local server, see [`Using-E2E-Resilience`](../Using-E2E-Resilience/). +For a self-contained demonstration that covers both crash and shutdown recovery, see +[`Using-E2E-Resilience`](../Using-E2E-Resilience/). ## Configuration (common to all clients) diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs index 32248cf30b1..b9983ed6f05 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs @@ -212,11 +212,10 @@ public override async IAsyncEnumerable CreateAsync( ? new ResponseEventStream(context, persistedResponse) : new ResponseEventStream(context, request); - WorkflowSessionCheckpointRecovery? workflowCheckpointRecovery = - session?.GetService(); if (context.IsRecovery && sessionRestoredFromStore - && workflowCheckpointRecovery is not null) + && session?.GetService() + is { } workflowCheckpointRecovery) { string? checkpointId = stream.InternalMetadata.TryGetValue(LatestWorkflowCheckpointIdMetadataKey, out string? persistedCheckpointId) @@ -462,59 +461,12 @@ await this._toolboxService bool steeringDetected = false; bool deferredForRecovery = false; - async ValueTask PersistWorkflowCheckpointAsync( - CheckpointInfo checkpoint, - CancellationToken checkpointCancellationToken) - { - if (!isResilientTurn - || workflowCheckpointRecovery is null - || session is null - || string.IsNullOrWhiteSpace(agentSessionId) - || (stream.InternalMetadata.TryGetValue(LatestWorkflowCheckpointIdMetadataKey, out string? lastCheckpointId) - && string.Equals(lastCheckpointId, checkpoint.CheckpointId, StringComparison.Ordinal))) - { - return null; - } - - try - { - await sessionStore.SaveSessionAsync( - agent, - agentSessionId, - session, - resolvedUserId, - checkpointCancellationToken).ConfigureAwait(false); - } - catch (Exception ex) when (ex is not OperationCanceledException) - { - if (this._logger.IsEnabled(LogLevel.Debug)) - { - this._logger.LogDebug( - ex, - "Workflow checkpoint {CheckpointId} was not paired with response {ResponseId} because its AgentSession could not be saved.", - checkpoint.CheckpointId, - context.ResponseId); - } - - return null; - } - - stream.InternalMetadata[LatestWorkflowCheckpointIdMetadataKey] = checkpoint.CheckpointId; - return stream.EmitInProgress(); - } - - // Check whenever the agent is storing messages when it should not. - bool CheckNotAllowedStoreUsage() => - // For IChatClients implementations when the backend is set to not store (store = false) the returned responseMessage.ConversationId comes null. - // If for any reason this property is set it means that the storage setting was enabled when it shouldn't. - !allowStoredOutputEnabled && session is ChatClientAgentSession { ConversationId: not null }; - - var enumerator = OutputConverter.ConvertUpdatesToEventsAsync( + var enumerator = OutputConverter.ConvertUpdatesToItemsAsync( agent.RunStreamingAsync(messages, session, options: options, cancellationToken: consentCts.Token), stream, session?.StateBag, - persistWorkflowCheckpointHandler: PersistWorkflowCheckpointAsync, cancellationToken: cancellationToken).GetAsyncEnumerator(cancellationToken); + var pendingEvents = new Queue(); try { while (true) @@ -530,12 +482,41 @@ bool CheckNotAllowedStoreUsage() => ResponseStreamEvent? evt = null; try { - if (!await enumerator.MoveNextAsync().ConfigureAwait(false)) + if (!pendingEvents.TryDequeue(out evt)) { - break; + if (!await enumerator.MoveNextAsync().ConfigureAwait(false)) + { + break; + } + + switch (enumerator.Current) + { + case OutputConverterItem.ResponseEvent responseEvent: + evt = responseEvent.Event; + break; + + case OutputConverterItem.WorkflowCheckpoint workflowCheckpoint: + ResponseStreamEvent? checkpointStateEvent = + await PersistWorkflowCheckpointAsync( + workflowCheckpoint.Checkpoint, + cancellationToken).ConfigureAwait(false); + + if (checkpointStateEvent is not null) + { + // AgentServer persists its orchestrator-owned response snapshot. + // Apply the updated internal metadata first, then request persistence. + pendingEvents.Enqueue(checkpointStateEvent); + pendingEvents.Enqueue(stream.Checkpoint()); + } + + continue; + + default: + throw new InvalidOperationException( + "The output converter returned an unsupported item."); + } } - evt = enumerator.Current; if (evt is ResponseCompletedEvent) { consentCts.Token.ThrowIfCancellationRequested(); @@ -665,8 +646,8 @@ bool CheckNotAllowedStoreUsage() => // remains authoritative for a turn that reaches normal completion. if (isResilientTurn && evt is ResponseOutputItemDoneEvent - && workflowCheckpointRecovery is null && session is not null + && session.GetService() is null && !string.IsNullOrWhiteSpace(agentSessionId) && !turnFailed) { @@ -732,6 +713,53 @@ await sessionStore.SaveSessionAsync( { yield return completedEvent; } + + async ValueTask PersistWorkflowCheckpointAsync( + CheckpointInfo checkpoint, + CancellationToken checkpointCancellationToken) + { + if (!isResilientTurn + || session is null + || session.GetService() is null + || string.IsNullOrWhiteSpace(agentSessionId) + || (stream.InternalMetadata.TryGetValue(LatestWorkflowCheckpointIdMetadataKey, out string? lastCheckpointId) + && string.Equals(lastCheckpointId, checkpoint.CheckpointId, StringComparison.Ordinal))) + { + return null; + } + + try + { + await sessionStore.SaveSessionAsync( + agent, + agentSessionId, + session, + resolvedUserId, + checkpointCancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + if (this._logger.IsEnabled(LogLevel.Debug)) + { + this._logger.LogDebug( + ex, + "Workflow checkpoint {CheckpointId} was not paired with response {ResponseId} because its AgentSession could not be saved.", + checkpoint.CheckpointId, + context.ResponseId); + } + + return null; + } + + stream.InternalMetadata[LatestWorkflowCheckpointIdMetadataKey] = checkpoint.CheckpointId; + return stream.EmitInProgress(); + } + + // Check whenever the agent is storing messages when it should not. + bool CheckNotAllowedStoreUsage() => + // For IChatClients implementations when the backend is set to not store (store = false) the returned responseMessage.ConversationId comes null. + // If for any reason this property is set it means that the storage setting was enabled when it shouldn't. + !allowStoredOutputEnabled && session is ChatClientAgentSession { ConversationId: not null }; } /// diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/OutputConverter.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/OutputConverter.cs index 27be4dcdea4..8327ebad4b5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/OutputConverter.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/OutputConverter.cs @@ -18,6 +18,16 @@ namespace Microsoft.Agents.AI.Foundry.Hosting; +internal abstract record OutputConverterItem +{ + public sealed record ResponseEvent(ResponseStreamEvent Event) : OutputConverterItem; + + public sealed record WorkflowCheckpoint(CheckpointInfo Checkpoint) : OutputConverterItem; + + public static implicit operator OutputConverterItem(ResponseStreamEvent value) => + new ResponseEvent(value); +} + /// /// Converts agent-framework streams into /// Responses Server SDK sequences using the @@ -32,18 +42,44 @@ internal static class OutputConverter /// The agent response updates to convert. /// The SDK event stream builder. /// Optional session state bag used to persist tool-approval id mappings across turns. - /// - /// Optional callback invoked after all output from a completed workflow superstep has been closed. - /// /// Cancellation token. - /// An async enumerable of SDK response stream events (excluding lifecycle events). + /// + /// An async enumerable of SDK response stream events. Workflow checkpoint boundaries are omitted. + /// + public static async IAsyncEnumerable ConvertUpdatesToEventsAsync( + IAsyncEnumerable updates, + ResponseEventStream stream, + AgentSessionStateBag? stateBag = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await foreach (OutputConverterItem item in ConvertUpdatesToItemsAsync( + updates, + stream, + stateBag, + cancellationToken).ConfigureAwait(false)) + { + if (item is OutputConverterItem.ResponseEvent responseEvent) + { + yield return responseEvent.Event; + } + } + } + + /// + /// Converts a stream of into response events and workflow + /// checkpoint boundaries. + /// + /// The agent response updates to convert. + /// The SDK event stream builder. + /// Optional session state bag used to persist tool-approval id mappings across turns. + /// Cancellation token. + /// Converted response events and workflow checkpoint boundaries in source order. [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing function call arguments dictionary.")] [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing function call arguments dictionary.")] - public static async IAsyncEnumerable ConvertUpdatesToEventsAsync( + public static async IAsyncEnumerable ConvertUpdatesToItemsAsync( IAsyncEnumerable updates, ResponseEventStream stream, AgentSessionStateBag? stateBag = null, - Func>? persistWorkflowCheckpointHandler = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { ResponseUsage? accumulatedUsage = null; @@ -82,19 +118,12 @@ public static async IAsyncEnumerable ConvertUpdatesToEvents yield return evt; } - if (workflowEvent is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } checkpoint } - && persistWorkflowCheckpointHandler is not null) - { - ResponseStreamEvent? checkpointStateEvent = - await persistWorkflowCheckpointHandler(checkpoint, cancellationToken).ConfigureAwait(false); - if (checkpointStateEvent is not null) + if (workflowEvent is SuperStepCompletedEvent { - // AgentServer persists its orchestrator-owned response snapshot. Emit the - // updated response state first so internal metadata becomes part of that - // authoritative snapshot, then persist it with the control event. - yield return checkpointStateEvent; - yield return stream.Checkpoint(); - } + CompletionInfo.Checkpoint: { } checkpoint + }) + { + yield return new OutputConverterItem.WorkflowCheckpoint(checkpoint); } continue; diff --git a/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/OutputConverterWorkflowTests.cs b/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/OutputConverterWorkflowTests.cs index 5cd73404f86..9264c3341a4 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/OutputConverterWorkflowTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/OutputConverterWorkflowTests.cs @@ -19,6 +19,33 @@ namespace Microsoft.Agents.AI.Foundry.Hosting.UnitTests; /// public class OutputConverterWorkflowTests { + [Fact] + public async Task ConvertUpdatesToItemsAsync_SuperStepCheckpoint_EmitsCheckpointBoundaryAsync() + { + // Arrange + var (stream, _) = CreateTestStream(); + var checkpoint = new CheckpointInfo("workflow-session", "checkpoint-1"); + var completion = new SuperStepCompletionInfo([]) { Checkpoint = checkpoint }; + var update = new AgentResponseUpdate + { + RawRepresentation = new SuperStepCompletedEvent(1, completion) + }; + + // Act + var items = new List(); + await foreach (OutputConverterItem item in OutputConverter.ConvertUpdatesToItemsAsync( + ToAsync([update]), + stream)) + { + items.Add(item); + } + + // Assert + OutputConverterItem.WorkflowCheckpoint boundary = + Assert.Single(items.OfType()); + Assert.Same(checkpoint, boundary.Checkpoint); + } + [Fact] public async Task SequentialWorkflowPattern_ProducesCorrectEventsAsync() {