diff --git a/docs/decisions/0032-dotnet-hosting-protocol-helpers.md b/docs/decisions/0032-dotnet-hosting-protocol-helpers.md new file mode 100644 index 00000000000..a2a87b606e7 --- /dev/null +++ b/docs/decisions/0032-dotnet-hosting-protocol-helpers.md @@ -0,0 +1,152 @@ +--- +status: accepted +contact: rogerbarreto +date: 2026-07-08 +deciders: rogerbarreto +consulted: eavanvalkenburg +informed: [] +--- + +# .NET hosting: OpenAI Responses protocol helpers for app-owned routing + +Realizes the helper-first direction of [ADR-0027](0027-hosting-channels.md) for .NET. + +## Context and Problem Statement + +[ADR-0027](0027-hosting-channels.md) refocused the (Python) hosting design away from a channel +framework toward **protocol conversion helpers plus optional execution state**: Agent Framework owns +protocol-native <-> run conversion, while the application owns HTTP routing, authentication, +middleware, storage, and native SDK calls. + +.NET already ships `Microsoft.Agents.AI.Hosting.OpenAI`, a route-owning server that **exposes an +`AIAgent` (or workflow) as the OpenAI Responses API** (`MapOpenAIResponses` + `IResponsesService`). It +owns the routes, an in-memory response/conversation store, streaming, and lifecycle. The question is +what, if anything, .NET must add to satisfy the ADR-0027 boundary. + +## Decision Drivers + +- Do not reinvent conversion logic that already exists and is battle-tested in `Hosting.OpenAI`. +- Give applications a way to own their own route/auth/middleware/storage while reusing Agent Framework + conversion (the ADR-0027 boundary). +- Keep the released public surface small. +- Stay consistent with the existing .NET hosting stack, which deliberately does **not** use the OpenAI + SDK Responses types server-side (it hand-rolled its own wire model). + +## Considered Options + +1. Self-contained new package that reimplements conversion using the OpenAI SDK Responses types + (mirrors the Python `agent-framework-hosting-responses` lineage). +2. New package that reuses `Hosting.OpenAI`'s internal converters (via `InternalsVisibleTo` or by + moving the conversion core out). +3. Thin public helper facade **inside** `Hosting.OpenAI` over the existing internal converters, plus + protocol-neutral execution-state holders in `Microsoft.Agents.AI.Hosting`. + +### First-principles gap analysis + +A capability comparison of the ADR-0027 / PR #6891 helper surface against the existing .NET stack: + +| Python helper capability | .NET today | Status | +| --- | --- | --- | +| `responses_to_run` | `ResponseInput.GetInputMessages` + `InputMessage.ToChatMessage` + `OpenAIResponsesMapOptions.RunOptionsFactory` | exists, internal | +| `responses_from_run` | `AgentResponseExtensions.ToResponse` | exists, internal | +| `responses_from_streaming_run` | `AgentResponseUpdateExtensions.ToStreamingResponseAsync` + `SseJsonResult` (also renders workflow events) | exists, internal, richer | +| `responses_session_id` | continuity resolved inside `InMemoryResponsesService` | exists, internal, not standalone | +| `create_response_id` | `IdGenerator` | exists, internal | +| `AgentState` (target + store, get-or-create) | `AgentSessionStore` (get-or-create + save + serialize + isolation) | richer; missing `Delete` + per-session lock | +| `SessionStore` (get/set/delete) | `AgentSessionStore` + `InMemoryAgentSessionStore` | richer; no `Delete` | +| `WorkflowState` + checkpoint resume | `WorkflowCatalog`/`HostedWorkflowBuilder`; workflow events already render over Responses; `CheckpointManager` is session-keyed | partial; no per-session checkpoint cursor | +| App owns routing/auth/middleware/storage | `MapOpenAIResponses`/`IResponsesService` own routing + storage | **the one real gap** | + +.NET already covers ~90% of the capability, and more richly (its streaming renderer even emits workflow +events; its session store serializes and supports per-principal isolation, neither of which Python's +in-memory `SessionStore` does). The single genuine gap is the **ownership model**: every conversion +primitive is bundled behind the route-owning server, so an application cannot own its own route and +call just the conversion. + +Note on lineage: Python's Responses offering was introduced *as a channel* (PR #6580) and always used +the `openai` SDK Responses types. .NET's `Hosting.OpenAI` predates and is independent of channels and +hand-rolled its own server-side wire DTOs (the SDK's Responses types are client-shaped and awkward +server-side). So Option 1 would both reinvent a working asset and contradict the .NET codebase's own +precedent. + +## Decision Outcome + +Chosen option: **3. Thin public helper facade inside `Hosting.OpenAI` plus neutral state holders**, +because the only real gap is the ownership model, so the work is to *un-bundle* the existing +converters, not to rebuild them or add a package. + +### Public surface + +`Microsoft.Agents.AI.Hosting.OpenAI` gains a single public static facade, `OpenAIResponses`, whose +boundary is `System.Text.Json` (`JsonElement`/streamed events), matching Python's dict boundary and +keeping the hand-rolled wire DTOs internal: + +- `OpenAIResponses.ToAgentRunRequest(JsonElement body)` -> messages + `AgentRunOptions?`. +- `OpenAIResponses.WriteResponse(AgentRunResponse response, string responseId, string? sessionId = null)` + -> a Responses-shaped `JsonElement`. +- `OpenAIResponses.WriteResponseStreamAsync(IAsyncEnumerable updates, string responseId, ...)` + -> Responses SSE `data:` frames. +- `OpenAIResponses.GetSessionId(JsonElement body)` -> `previous_response_id` or `conversation` id, or + `null`. Kept **separate** from `ToAgentRunRequest` so the trust boundary is visible: choosing to use + a request-derived key is an explicit application decision. +- `OpenAIResponses.CreateResponseId()` -> a `resp_*` id. + +All helpers are side-effect-free and delegate to the existing internal converters. `MapOpenAIResponses` +public behavior is unchanged; it and the facade share one internal conversion core (an internal +`ToResponse` overload with an optional originating request is added so the facade can render without a +request object). + +### Optional execution state (neutral package) + +`Microsoft.Agents.AI.Hosting` gains: + +- `AgentSessionStore.DeleteSessionAsync(...)` (+ `InMemoryAgentSessionStore` implementation and + isolation-decorator passthrough): the one missing store operation. +- `HostedAgentState`: a thin holder bundling an `AIAgent` with an `AgentSessionStore`, exposing + `GetOrCreateSessionAsync`, `SaveSessionAsync` (including under a newly minted id), and + `DeleteSessionAsync`, with optional per-session locking. It exists because only the holder has both + the target and the store; it does not replace `AgentSessionStore`, which already provides + serialization and isolation that Python's `AgentState`/`SessionStore` lack. +- `HostedWorkflowState`: a thin holder bundling a workflow target with a `CheckpointManager` and an + internal `sessionId -> CheckpointInfo` head cursor, exposing `RunOrResumeAsync`. .NET's checkpoint + store is already `sessionId`-keyed (unlike Python's workflow-name keying), but `CheckpointInfo` has + no ordering, so the holder remembers the head checkpoint per session to resume. On subsequent turns it + restores that checkpoint and runs the workflow forward with the new turn's input (mirroring the Python + host's restore-then-run semantics), rather than continuing a halted run with no input. When the + in-memory cursor misses (new holder / process restart) it reads the session's latest checkpoint from the + `CheckpointManager`, so a durable manager resumes across restarts. + +### Scope + +Responses only for v1; the facade is named so a parallel `OpenAIChatCompletions` facade can follow. +No new package, no OpenAI-SDK-typed reimplementation, no change to `MapOpenAIResponses` public +behavior. + +### Security responsibilities + +Consistent with ADR-0027, the application owns the trust boundary. `GetSessionId(...)` returns an +untrusted candidate key; the application must authenticate the caller and authorize/bind the id before +using it as an `AgentSessionStore` key or workflow checkpoint session id. Multi-user hosts must scope +the session store per principal (`IsolationKeyScopedAgentSessionStore`). Helpers stay side-effect-free; +persistence happens only after the run completes. + +## Consequences + +Positive: + +- Smallest possible surface: the released addition is one facade type plus two thin state holders and + one new store method. +- No duplicated conversion; the app-owned-routing path and the route-owning server share one core. +- `MapOpenAIResponses` users are unaffected. + +Negative: + +- The facade's `JsonElement` boundary is less strongly typed than the internal DTOs (accepted to keep + the wire model internal and mirror Python's dict boundary). +- Workflow resume relies on an in-memory head cursor by default; durable multi-replica hosts must + supply their own cursor persistence. + +## More Information + +- Parent ADR: [ADR-0027](0027-hosting-channels.md). +- Spec: `docs/specs/003-dotnet-hosting-protocol-helpers.md`. diff --git a/docs/specs/003-dotnet-hosting-protocol-helpers.md b/docs/specs/003-dotnet-hosting-protocol-helpers.md new file mode 100644 index 00000000000..c768a323591 --- /dev/null +++ b/docs/specs/003-dotnet-hosting-protocol-helpers.md @@ -0,0 +1,242 @@ +--- +status: accepted +contact: rogerbarreto +date: 2026-07-08 +deciders: rogerbarreto +consulted: eavanvalkenburg +informed: [] +--- + +# .NET hosting: OpenAI Responses protocol helpers and optional execution state + +Implements [ADR-0032](../decisions/0032-dotnet-hosting-protocol-helpers.md), which realizes the +helper-first direction of [ADR-0027](../decisions/0027-hosting-channels.md) for .NET. + +## What is the goal of this feature? + +Let application developers expose an `AIAgent` or workflow over the OpenAI Responses protocol **while +owning their own ASP.NET Core route, authentication, middleware, and storage**, by calling small, +side-effect-free Agent Framework conversion helpers instead of adopting the batteries-included, +route-owning `MapOpenAIResponses` server. + +Success: an application can implement a working `POST /responses` endpoint (sync + streaming) in its +own minimal-API handler using only the public helpers plus its own auth/storage, with no dependency on +`MapOpenAIResponses` or `IResponsesService`. + +## What is the problem being solved? + +.NET already exposes agents as the OpenAI Responses API, but only through the route-owning +`MapOpenAIResponses`/`IResponsesService`, which also owns routing, response/conversation storage, +streaming, and lifecycle. An application that wants its own routing (custom auth, middleware, status +codes, durable storage, or a different framework surface) currently has no supported way to reuse the +framework's Responses<->agent conversion. Every conversion primitive that would make this possible +already exists in `Microsoft.Agents.AI.Hosting.OpenAI` but is `internal`. + +This feature un-bundles that conversion into a public, app-callable surface, and adds the minimal +execution-state helpers an app needs for session continuity and workflow checkpoint resume. + +## API Changes + +### `Microsoft.Agents.AI.Hosting.OpenAI` (new public static facade `OpenAIResponses`) + +Boundary is `System.Text.Json`; the wire DTOs stay internal. All members are side-effect-free. + +```csharp +namespace Microsoft.Agents.AI.Hosting.OpenAI; + +public static class OpenAIResponses +{ + // Wire -> Agent Framework run input. + public static OpenAIResponsesRunRequest ToAgentRunRequest( + JsonElement body, + OpenAIResponsesMapOptions? mapOptions = null); + + // Agent Framework result -> Responses payload (no originating request required). + public static JsonElement WriteResponse( + AgentResponse response, + string responseId, + string? sessionId = null); + + // Agent Framework stream -> Responses SSE `data:` frames. + public static IAsyncEnumerable WriteResponseStreamAsync( + IAsyncEnumerable updates, + string responseId, + string? sessionId = null, + CancellationToken cancellationToken = default); + + // Untrusted candidate continuation key: previous_response_id or conversation id (or null). + // Kept SEPARATE from ToAgentRunRequest so using a request-derived key is an explicit decision. + public static string? GetSessionId(JsonElement body); + + // Mint a `resp_*` id. + public static string CreateResponseId(); +} + +// Result of ToAgentRunRequest. +public sealed class OpenAIResponsesRunRequest +{ + public IList Messages { get; } + public AgentRunOptions? Options { get; } +} +``` + +`ToAgentRunRequest` honors `OpenAIResponsesMapOptions.RunOptionsFactory` exactly as the route model +does (by default no request setting is mapped onto the run; unsupported settings surface as a +`NotSupportedException`). `WriteResponse`/`WriteResponseStreamAsync` reuse the existing internal +`AgentResponseExtensions.ToResponse` / `AgentResponseUpdateExtensions.ToStreamingResponseAsync` +converters (an internal `ToResponse` overload with an optional originating request is added so the +facade can render without one). The streaming renderer's existing workflow-event support is preserved. + +### `Microsoft.Agents.AI.Hosting` (execution state, protocol-neutral) + +```csharp +namespace Microsoft.Agents.AI.Hosting; + +public abstract class AgentSessionStore +{ + // ... existing members ... + + // New: the one missing store operation. Virtual (not abstract) with a default that throws + // NotSupportedException, so existing external stores (e.g. the Foundry hosting stores) keep + // compiling; the in-box Hosting stores override it. In-box overrides treat deleting a missing + // session as a no-op. + public virtual ValueTask DeleteSessionAsync( + AIAgent agent, string conversationId, CancellationToken cancellationToken = default); +} + +// Thin holder: pairs an agent target with a session store. +public sealed class HostedAgentState +{ + public HostedAgentState(AIAgent agent, AgentSessionStore? sessionStore = null); + + public ValueTask GetOrCreateSessionAsync(string sessionId, CancellationToken ct = default); + public ValueTask SaveSessionAsync(string sessionId, AgentSession session, CancellationToken ct = default); + public ValueTask DeleteSessionAsync(string sessionId, CancellationToken ct = default); +} + +// Thin holder: pairs a workflow target with checkpointing + a per-session head cursor. +public sealed class HostedWorkflowState +{ + public HostedWorkflowState(Workflow workflow, CheckpointManager? checkpointManager = null); + + // First turn runs forward from the start; subsequent turns restore the session's latest + // checkpoint and run forward with the new turn's input, then record the new head checkpoint. + public ValueTask RunOrResumeAsync( + string sessionId, object input, CancellationToken ct = default); +} +``` + +`HostedAgentState.GetOrCreateSessionAsync` delegates to `AgentSessionStore.GetSessionAsync(agent, id)` +(which already creates on miss). `SaveSessionAsync(id, session)` persists post-run, including under a +newly minted `resp_*` id when the protocol mints a new continuation id. Optional per-session locking +serializes concurrent first-touch of the same id. `DeleteSessionAsync` uses the new store method. + +`HostedWorkflowState` defaults to `CheckpointManager.CreateInMemory()` and an in-memory +`sessionId -> CheckpointInfo` cursor. Because the checkpoint store is already `sessionId`-keyed but +`CheckpointInfo` carries no ordering, the holder remembers the head checkpoint per session so +`RunOrResumeAsync` can resume the correct one. On subsequent turns it restores that checkpoint to +rehydrate accumulated workflow state and then runs the workflow forward with the new turn's input — +mirroring the Python hosting host's restore-then-run semantics (`agent_framework_hosting`'s +`_invoke_workflow`), rather than continuing a halted run with no input (which would wait for input +indefinitely). For agent (chat-protocol) workflows the new input is accompanied by a `TurnToken` so the +turn is driven. When the in-memory cursor misses (a new holder or a process restart), the holder falls +back to `CheckpointManager.GetLatestCheckpointAsync(sessionId)`, so a durable `CheckpointManager` resumes +correctly across restarts (the default in-memory manager does not persist, so a restart starts fresh). A +resume that produces no events is logged as a warning (possible stale checkpoint or mismatched input), +mirroring the Python host's zero-event restore warning. Because a single workflow instance backs the +holder and workflow instances do not support concurrent runs, `RunOrResumeAsync` serializes turns through +one lock (mirroring the Python host's workflow lock); `HostedWorkflowState` is therefore `IDisposable`. A +streaming counterpart, `RunOrResumeStreamingAsync`, yields the turn's `WorkflowEvent`s as they occur (for +example to render agent updates over the Responses SSE wire) and records the head checkpoint once the +stream is fully enumerated, keeping the blocking and streaming workflow paths in lockstep as Python does. +Because `RunOrResumeAsync`/`RunOrResumeStreamingAsync` are generic over the input type, the application +adapts the Responses input into the workflow's start-executor input type at the call site (for example +parsing a structured payload into a typed record) — the .NET counterpart of Python's `ResponsesChannel` +run hook, without coupling the holder to a specific wire type. + +## Non-goals for v1 + +- ChatCompletions / Conversations helper surfaces (the facade is named so `OpenAIChatCompletions` can + follow). +- Changing `MapOpenAIResponses` public behavior. +- A new package or an OpenAI-SDK-typed reimplementation. +- Durable/pluggable workflow checkpoint-cursor storage (in-memory default only for v1). + +## Security responsibilities (application-owned) + +- Authenticate the caller before using any `GetSessionId(...)` result. +- Authorize and bind the candidate id to the authenticated principal/tenant before using it as an + `AgentSessionStore` key or a workflow checkpoint session id. +- For multi-user hosts, wrap the store with `IsolationKeyScopedAgentSessionStore` (for example via + `UseClaimsBasedSessionIsolation(...)`), so the session namespace is scoped per principal. +- Persist session/checkpoint state only after the run or stream has completed. + +## E2E Code Samples + +### Agent over Responses, app-owned route (non-streaming + SSE) + +```csharp +var agent = /* an AIAgent */; +var state = new HostedAgentState(agent); // in-memory session store by default + +app.MapPost("/responses", async (HttpContext http, CancellationToken ct) => +{ + using var doc = await JsonDocument.ParseAsync(http.Request.Body, cancellationToken: ct); + JsonElement body = doc.RootElement; + + // App owns auth + id trust decisions. + string? candidate = OpenAIResponses.GetSessionId(body); + string sessionId = Authorize(http.User, candidate) ?? OpenAIResponses.CreateResponseId(); + + var run = OpenAIResponses.ToAgentRunRequest(body); + var session = await state.GetOrCreateSessionAsync(sessionId, ct); + + string responseId = OpenAIResponses.CreateResponseId(); + + if (body.TryGetProperty("stream", out var s) && s.GetBoolean()) + { + http.Response.ContentType = "text/event-stream"; + var updates = agent.RunStreamingAsync(run.Messages, session, run.Options, ct); + await foreach (var frame in OpenAIResponses.WriteResponseStreamAsync(updates, responseId, sessionId, ct)) + { + await http.Response.WriteAsync(frame, ct); + await http.Response.Body.FlushAsync(ct); + } + await state.SaveSessionAsync(responseId, session, ct); + return Results.Empty; + } + + var result = await agent.RunAsync(run.Messages, session, run.Options, ct); + await state.SaveSessionAsync(responseId, session, ct); + return Results.Json(OpenAIResponses.WriteResponse(result, responseId, sessionId)); +}); +``` + +### Workflow over Responses with checkpoint resume + +Workflow checkpoint resume requires a **stable** session key across turns. `previous_response_id` changes +every turn, so it is not a valid checkpoint key; use the `conversation` id (constant for the conversation). +Because `GetSessionId(...)` prefers `previous_response_id`, a workflow route reads the conversation id +directly rather than calling `GetSessionId(...)`. + +```csharp +var state = new HostedWorkflowState(workflow); // in-memory checkpoints + cursor + +app.MapPost("/responses", async (HttpContext http, CancellationToken ct) => +{ + using var doc = await JsonDocument.ParseAsync(http.Request.Body, cancellationToken: ct); + JsonElement body = doc.RootElement; + + // Stable, authorized checkpoint key. GetConversationId(...) reads the conversation id (string or object). + string sessionId = Authorize(http.User, GetConversationId(body)) + ?? OpenAIResponses.CreateResponseId(); + + var run = OpenAIResponses.ToAgentRunRequest(body); + + // Runs forward on first call, resumes from the session's head checkpoint thereafter. + var result = await state.RunOrResumeAsync(sessionId, run.Messages, ct); + + return Results.Json(OpenAIResponses.WriteResponse(result.AsAgentResponse(), + OpenAIResponses.CreateResponseId(), sessionId)); +}); +``` diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index 27cf8605740..e639de13e63 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -313,7 +313,19 @@ - + + + + + + + + + + + + + @@ -639,7 +651,9 @@ - + + + diff --git a/dotnet/samples/04-hosting/af-hosting/README.md b/dotnet/samples/04-hosting/af-hosting/README.md new file mode 100644 index 00000000000..2a2f7845913 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/README.md @@ -0,0 +1,38 @@ +# Agent Framework hosting helper samples + +End-to-end samples for exposing Agent Framework targets through app-owned hosting routes. + +The helper-first hosting packages provide protocol conversion and optional execution state. The +application still owns the web framework, native SDK clients, authentication, response construction, and +deployment shape. + +| Sample | What it shows | +|---|---| +| [`local_responses/`](./local_responses) | One agent behind an app-owned ASP.NET Core route using the `OpenAIResponses` helper functions plus `HostedAgentState` / `AgentSessionStore`. Start here to learn the helper seam. | +| [`local_responses_workflow/`](./local_responses_workflow) | A workflow target behind an app-owned ASP.NET Core route using the `OpenAIResponses` helper functions, `HostedWorkflowState`, an explicit `CheckpointManager`, and an app-owned checkpoint cursor. | + +Each sample is a **client/server pair**. Unlike the Python samples (which run a server `app.py` and calling +scripts as loose files), .NET samples are projects, so each pair is split into two projects: + +``` +local_responses/ +├── Server/ # exposes POST /responses using the OpenAIResponses helpers +└── Client/ # consumes it two ways: CC (IChatClient) and MAF (AIAgent) +``` + +The `Client` shows the two idiomatic ways to consume the endpoint from .NET, both against the same server: + +- **CC** — a plain `Microsoft.Extensions.AI.IChatClient` (the lower-level chat-client path). +- **MAF** — a Microsoft Agent Framework `AIAgent` (the higher-level agent path). + +## Relationship to `../foundry-hosted-agents/` + +The sibling [`../FoundryHostedAgents/`](../FoundryHostedAgents) directory contains samples for agents that +run inside the Foundry Hosted Agents platform. Those samples use the Foundry-managed protocol surface with +no app-owned hosting route involved. + +| Aspect | `af-hosting/` (this directory) | `FoundryHostedAgents/` | +|---|---|---| +| Server stack | App-owned ASP.NET Core + hosting protocol helpers | Foundry Hosted Agents runtime | +| Protocol surface | The app exposes the route and calls helpers | The platform exposes Responses + Invocations | +| When to pick this | You need custom hosting code or want to learn the helper seam | You want the Foundry-managed hosting surface | diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/Client/Client.csproj b/dotnet/samples/04-hosting/af-hosting/local_responses/Client/Client.csproj new file mode 100644 index 00000000000..966bcba0de8 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/Client/Client.csproj @@ -0,0 +1,19 @@ + + + + Exe + net10.0 + enable + enable + + + + + + + + + + + + diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/Client/Program.cs b/dotnet/samples/04-hosting/af-hosting/local_responses/Client/Program.cs new file mode 100644 index 00000000000..8c92187d3ce --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/Client/Program.cs @@ -0,0 +1,79 @@ +// Copyright (c) Microsoft. All rights reserved. + +// Client for the HostingResponsesAgent server sample. It shows the two idiomatic ways to consume an +// OpenAI Responses endpoint from .NET, both pointed at the same app-owned server route: +// +// 1. CC - a plain Microsoft.Extensions.AI IChatClient (the lower-level chat-client path). +// 2. MAF - a Microsoft Agent Framework AIAgent + AgentSession (the higher-level agent path). +// +// Both run the same three-turn conversation. The third turn only makes sense if the server remembered +// the first turn, so it also proves multi-turn session continuity across the rotating response-id chain. + +using System.ClientModel; +using Microsoft.Agents.AI; +using Microsoft.Extensions.AI; +using OpenAI; +using OpenAI.Responses; + +string serverUrl = Environment.GetEnvironmentVariable("RESPONSES_SERVER_URL") ?? "http://localhost:5000"; + +// The server ignores the model id (it runs its own configured agent), but the OpenAI SDK requires one to +// shape the request. Reuse FOUNDRY_MODEL for parity with the server sample. +string model = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-5.4-mini"; + +string[] prompts = +[ + "What is the weather in Tokyo?", + "And what about Amsterdam?", + "Which of the two cities we just discussed is warmer?", +]; + +// A single ResponsesClient pointed at the local server backs both consumption paths. The api key is unused +// by the sample server, but the SDK requires a credential. +ResponsesClient responseClient = new OpenAIClient( + new ApiKeyCredential("not-needed"), + new OpenAIClientOptions { Endpoint = new Uri(serverUrl) }) + .GetResponsesClient(); + +Console.WriteLine($"Connecting to {serverUrl}\n"); + +await RunWithChatClientAsync(responseClient, model, prompts).ConfigureAwait(false); +await RunWithAgentAsync(responseClient, model, prompts).ConfigureAwait(false); + +// CC path: consume the endpoint through a Microsoft.Extensions.AI IChatClient. Continuity is threaded by +// hand: each response carries the server's response id as ChatResponse.ConversationId, which we pass back +// as the next turn's ChatOptions.ConversationId. Because it is a "resp_" id, the SDK sends it as +// previous_response_id, exactly what the server's GetSessionId reads. +static async Task RunWithChatClientAsync(ResponsesClient responseClient, string model, string[] prompts) +{ + Console.WriteLine("== CC: Microsoft.Extensions.AI IChatClient =="); + IChatClient chatClient = responseClient.AsIChatClient(model); + + string? previousResponseId = null; + foreach (string prompt in prompts) + { + Console.WriteLine($"User: {prompt}"); + ChatResponse response = await chatClient.GetResponseAsync( + prompt, + new ChatOptions { ConversationId = previousResponseId }).ConfigureAwait(false); + Console.WriteLine($"Agent: {response.Text}"); + previousResponseId = response.ConversationId; + Console.WriteLine($"Response ID: {previousResponseId}\n"); + } +} + +// MAF path: consume the same endpoint through an Agent Framework AIAgent. A single AgentSession threads the +// rotating response-id chain automatically, so the caller only sends the new user message each turn. +static async Task RunWithAgentAsync(ResponsesClient responseClient, string model, string[] prompts) +{ + Console.WriteLine("== MAF: Agent Framework AIAgent + AgentSession =="); + AIAgent agent = responseClient.AsAIAgent(model: model, name: "HostedResponsesClient"); + AgentSession session = await agent.CreateSessionAsync().ConfigureAwait(false); + + foreach (string prompt in prompts) + { + Console.WriteLine($"User: {prompt}"); + AgentResponse response = await agent.RunAsync(prompt, session).ConfigureAwait(false); + Console.WriteLine($"Agent: {response.Text}\n"); + } +} diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/Client/README.md b/dotnet/samples/04-hosting/af-hosting/local_responses/Client/README.md new file mode 100644 index 00000000000..27c130017e6 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/Client/README.md @@ -0,0 +1,20 @@ +# Client (Hosting Responses Agent) + +Client half of the [Hosting Responses Agent](../README.md) sample. + +Runs the same three-turn conversation twice against the server's `POST /responses` route, once per +consumption path: + +- **CC** — a plain `Microsoft.Extensions.AI.IChatClient` from `ResponsesClient.AsIChatClient(model)`. + Continuity is threaded by hand: each response's `ChatResponse.ConversationId` (a `resp_` id) is passed + back as the next turn's `ChatOptions.ConversationId`, which the SDK sends as `previous_response_id`. +- **MAF** — a Microsoft Agent Framework `AIAgent` from `ResponsesClient.AsAIAgent(...)`. A single + `AgentSession` threads the rotating response-id chain automatically. + +The third turn asks about the first turn, so a correct answer proves multi-turn session continuity. + +```bash +dotnet run +``` + +Defaults to `http://localhost:5000`; override with `RESPONSES_SERVER_URL`. Start the server first. diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/README.md b/dotnet/samples/04-hosting/af-hosting/local_responses/README.md new file mode 100644 index 00000000000..eb3385229fb --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/README.md @@ -0,0 +1,57 @@ +# Hosting Responses Agent (client / server) + +A client/server pair showing how to expose an `AIAgent` over the OpenAI Responses protocol from an +app-owned ASP.NET Core route, and how to consume it from .NET two different ways. + +``` +local_responses/ +├── Server/ # exposes POST /responses using the OpenAIResponses helpers +└── Client/ # consumes it two ways: CC (IChatClient) and MAF (AIAgent) +``` + +## Server + +The server owns routing, authentication, and session storage. The framework provides only the protocol +conversion via `OpenAIResponses` (`GetSessionId`, `ToAgentRunRequest`, `WriteResponse` / +`WriteResponseStreamAsync`), instead of the batteries-included `MapOpenAIResponses` server. The agent has a +deterministic `lookup_weather` tool (mirroring the Python sample). Session continuity uses `HostedAgentState` +over an in-memory `AgentSessionStore`. It binds to `http://localhost:5000`. + +See [Server/README.md](Server/README.md). + +## Client + +A single program that runs the same three-turn conversation twice, once per consumption path: + +- **CC** — a plain `Microsoft.Extensions.AI.IChatClient` (the lower-level chat-client path). +- **MAF** — a Microsoft Agent Framework `AIAgent` + `AgentSession` (the higher-level agent path). + +Both point at the same server. The third turn asks about the first turn, proving multi-turn session +continuity across the rotating response-id chain. + +See [Client/README.md](Client/README.md). + +## Run + +Start the server in one shell: + +```bash +export FOUNDRY_PROJECT_ENDPOINT="https://.services.ai.azure.com/api/projects/" +export FOUNDRY_MODEL="gpt-5.4-mini" # optional, defaults to gpt-5.4-mini +dotnet run --project Server +``` + +Then run the client in another shell: + +```bash +dotnet run --project Client +``` + +The client defaults to `http://localhost:5000`; override with `RESPONSES_SERVER_URL`. + +## Security note + +`OpenAIResponses.GetSessionId(...)` returns an untrusted candidate key. The server's `Authorize(...)` is a +placeholder; a real application must authenticate the caller and authorize/bind the id to the authenticated +principal before using it as a session key. For multi-user hosts, scope the store with +`IsolationKeyScopedAgentSessionStore`. diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/Server/Program.cs b/dotnet/samples/04-hosting/af-hosting/local_responses/Server/Program.cs new file mode 100644 index 00000000000..899b49f2bbf --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/Server/Program.cs @@ -0,0 +1,99 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample shows how an application can own its own ASP.NET Core route and expose an AIAgent over the +// OpenAI Responses protocol by calling the Agent Framework OpenAIResponses conversion helpers, instead of +// using the batteries-included MapOpenAIResponses server. The application keeps control of routing, auth, +// and session storage; the helpers provide only the protocol <-> agent conversion. + +using System.ComponentModel; +using System.Text.Json; +using Azure.AI.Projects; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.Hosting; +using Microsoft.Agents.AI.Hosting.OpenAI; +using Microsoft.Extensions.AI; + +var builder = WebApplication.CreateBuilder(args); + +// Configuration via environment variables (never hardcode secrets). +string endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT") + ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set."); +string model = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-5.4-mini"; + +// A deterministic weather tool (mirrors the Python sample's lookup_weather @tool). +[Description("Return a deterministic weather report for a city.")] +static string LookupWeather([Description("The city to look up weather for.")] string location) +{ + int highTemp = 5 + (System.Text.Encoding.UTF8.GetBytes(location).Sum(b => b) % 21); + return location switch + { + "Seattle" => $"Seattle is rainy with a high of {highTemp}°C.", + "Amsterdam" => $"Amsterdam is cloudy with a high of {highTemp}°C.", + "Tokyo" => $"Tokyo is clear with a high of {highTemp}°C.", + _ => $"{location} is sunny with a high of {highTemp}°C.", + }; +} + +// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production. +// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid +// latency issues, unintended credential probing, and potential security risks from fallback mechanisms. +AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential()) + .AsAIAgent( + model: model, + instructions: "You are a friendly weather assistant. Use the lookup_weather tool for any weather " + + "question and answer in one short sentence.", + name: "WeatherAgent", + tools: [AIFunctionFactory.Create(LookupWeather, name: "lookup_weather")]); + +// Optional shared execution state: pairs the agent with a session store (in-memory by default). +var state = new HostedAgentState(agent); + +var app = builder.Build(); + +// The application owns this route. It parses the OpenAI Responses body with the helpers, runs the agent +// itself, and renders the response with the helpers. Binding the body as JsonElement lets ASP.NET Core +// deserialize the JSON request body directly, so there is no JsonDocument to own or dispose. +app.MapPost("/responses", async (JsonElement body, HttpContext http, CancellationToken cancellationToken) => +{ + // The candidate continuation id is untrusted. A real app authenticates the caller and authorizes/binds + // this key to the principal before using it. This sample simply falls back to a fresh id. + string? candidateSessionId = OpenAIResponses.GetSessionId(body); + string sessionId = Authorize(http, candidateSessionId) ?? OpenAIResponses.CreateResponseId(); + + OpenAIResponsesRunRequest run = OpenAIResponses.ToAgentRunRequest(body); + AgentSession session = await state.GetOrCreateSessionAsync(sessionId, cancellationToken).ConfigureAwait(false); + string responseId = OpenAIResponses.CreateResponseId(); + + bool stream = body.TryGetProperty("stream", out JsonElement streamProp) && streamProp.ValueKind == JsonValueKind.True; + + if (stream) + { + http.Response.ContentType = "text/event-stream"; + var updates = agent.RunStreamingAsync(run.Messages, session, run.Options, cancellationToken); + await foreach (string frame in OpenAIResponses.WriteResponseStreamAsync(updates, responseId, responseId, cancellationToken).ConfigureAwait(false)) + { + await http.Response.WriteAsync(frame, cancellationToken).ConfigureAwait(false); + await http.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + // Persist the post-run session under the new response id so the next turn can continue from it. + await state.SaveSessionAsync(responseId, session, cancellationToken).ConfigureAwait(false); + + // The SSE body was already written straight to http.Response above, so return an empty result: + // this returns from the handler (the non-streaming code below does not run) without writing a body. + return Results.Empty; + } + + AgentResponse result = await agent.RunAsync(run.Messages, session, run.Options, cancellationToken).ConfigureAwait(false); + await state.SaveSessionAsync(responseId, session, cancellationToken).ConfigureAwait(false); + return Results.Json(OpenAIResponses.WriteResponse(result, responseId, responseId)); +}); + +// Bind to a fixed local URL so the paired client sample has a deterministic default. +// Override with the ASPNETCORE_URLS environment variable when needed. +app.Run("http://localhost:5000"); + +// Application-owned trust decision. Replace with real authentication + authorization: verify the caller, +// then authorize/bind the candidate id to the authenticated principal before returning it. +static string? Authorize(HttpContext http, string? candidateSessionId) => candidateSessionId; diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/Server/README.md b/dotnet/samples/04-hosting/af-hosting/local_responses/Server/README.md new file mode 100644 index 00000000000..66e0e9a1499 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/Server/README.md @@ -0,0 +1,30 @@ +# Server (Hosting Responses Agent) + +Server half of the [Hosting Responses Agent](../README.md) sample. + +Exposes an `AIAgent` over the OpenAI Responses protocol on an app-owned `POST /responses` route: + +- `OpenAIResponses.GetSessionId(body)` extracts the untrusted continuation-id candidate. +- `OpenAIResponses.ToAgentRunRequest(body)` parses the request into messages + run options. +- `OpenAIResponses.WriteResponse(...)` / `WriteResponseStreamAsync(...)` render the agent output back to the + Responses wire shape (non-streaming JSON and SSE). + +Session continuity uses `HostedAgentState` over an in-memory `AgentSessionStore`. The agent has a +deterministic `lookup_weather` tool (mirroring the Python sample). Binds to `http://localhost:5000` (override +with `ASPNETCORE_URLS`). + +```bash +export FOUNDRY_PROJECT_ENDPOINT="https://.services.ai.azure.com/api/projects/" +export FOUNDRY_MODEL="gpt-5.4-mini" +dotnet run +``` + +You can also call it directly with curl: + +```bash +curl -s http://localhost:5000/responses -H "content-type: application/json" \ + -d '{ "input": "What is the weather in Tokyo?" }' + +curl -N http://localhost:5000/responses -H "content-type: application/json" \ + -d '{ "input": "What is the weather in Tokyo?", "stream": true }' +``` diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses/Server/Server.csproj b/dotnet/samples/04-hosting/af-hosting/local_responses/Server/Server.csproj new file mode 100644 index 00000000000..232aede78d2 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses/Server/Server.csproj @@ -0,0 +1,20 @@ + + + + net10.0 + enable + enable + + + + + + + + + + + + + + diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/Client.csproj b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/Client.csproj new file mode 100644 index 00000000000..966bcba0de8 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/Client.csproj @@ -0,0 +1,19 @@ + + + + Exe + net10.0 + enable + enable + + + + + + + + + + + + diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/Program.cs b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/Program.cs new file mode 100644 index 00000000000..29b9852669a --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/Program.cs @@ -0,0 +1,69 @@ +// Copyright (c) Microsoft. All rights reserved. + +// Client for the local_responses_workflow server sample. Like the agent client, it shows the two idiomatic +// ways to consume an OpenAI Responses endpoint from .NET, both pointed at the same app-owned workflow route: +// +// 1. CC - a plain Microsoft.Extensions.AI IChatClient (the lower-level chat-client path). +// 2. MAF - a Microsoft Agent Framework AIAgent + AgentSession (the higher-level agent path). +// +// The server implements previous_response_id continuation only (it rejects conversation-id continuity), so +// both paths follow the rotating response-id chain: the first turn sends a JSON brief, the follow-up turn +// continues from the first turn's response id. The workflow resumes its checkpoint across that chain. + +using System.ClientModel; +using Microsoft.Agents.AI; +using Microsoft.Extensions.AI; +using OpenAI; +using OpenAI.Responses; + +string serverUrl = Environment.GetEnvironmentVariable("RESPONSES_SERVER_URL") ?? "http://localhost:5001"; +string model = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-5.4-mini"; + +const string Brief = """{ "topic": "electric SUV", "style": "playful", "audience": "young families" }"""; +const string FollowUp = "Make it a little more premium, but still family friendly."; + +ResponsesClient responseClient = new OpenAIClient( + new ApiKeyCredential("not-needed"), + new OpenAIClientOptions { Endpoint = new Uri(serverUrl) }) + .GetResponsesClient(); + +Console.WriteLine($"Connecting to {serverUrl}\n"); + +await RunWithChatClientAsync(responseClient, model).ConfigureAwait(false); +await RunWithAgentAsync(responseClient, model).ConfigureAwait(false); + +// CC path: consume the endpoint through a Microsoft.Extensions.AI IChatClient. Continuity is threaded by +// hand: each response's ChatResponse.ConversationId (a "resp_" id) is passed back as the next turn's +// ChatOptions.ConversationId, which the SDK sends as previous_response_id. +static async Task RunWithChatClientAsync(ResponsesClient responseClient, string model) +{ + Console.WriteLine("== CC: Microsoft.Extensions.AI IChatClient =="); + IChatClient chatClient = responseClient.AsIChatClient(model); + + Console.WriteLine($"User: {Brief}"); + ChatResponse first = await chatClient.GetResponseAsync(Brief).ConfigureAwait(false); + Console.WriteLine($"Workflow: {first.Text}\n"); + + Console.WriteLine($"User: {FollowUp}"); + ChatResponse second = await chatClient.GetResponseAsync( + FollowUp, + new ChatOptions { ConversationId = first.ConversationId }).ConfigureAwait(false); + Console.WriteLine($"Workflow: {second.Text}\n"); +} + +// MAF path: consume the same endpoint through an Agent Framework AIAgent. A single AgentSession threads the +// rotating previous_response_id chain automatically, so the caller only sends the new input each turn. +static async Task RunWithAgentAsync(ResponsesClient responseClient, string model) +{ + Console.WriteLine("== MAF: Agent Framework AIAgent + AgentSession =="); + AIAgent agent = responseClient.AsAIAgent(model: model, name: "HostedWorkflowClient"); + AgentSession session = await agent.CreateSessionAsync().ConfigureAwait(false); + + Console.WriteLine($"User: {Brief}"); + AgentResponse first = await agent.RunAsync(Brief, session).ConfigureAwait(false); + Console.WriteLine($"Workflow: {first.Text}\n"); + + Console.WriteLine($"User: {FollowUp}"); + AgentResponse second = await agent.RunAsync(FollowUp, session).ConfigureAwait(false); + Console.WriteLine($"Workflow: {second.Text}\n"); +} diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/README.md b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/README.md new file mode 100644 index 00000000000..f4c9d66da0e --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Client/README.md @@ -0,0 +1,20 @@ +# Client (Hosting Responses Workflow) + +Client half of the [Hosting Responses Workflow](../README.md) sample. + +Runs the same two-turn conversation twice against the server's `POST /responses` route, once per consumption +path: + +- **CC** — a plain `Microsoft.Extensions.AI.IChatClient` from `ResponsesClient.AsIChatClient(model)`. +- **MAF** — a Microsoft Agent Framework `AIAgent` + `AgentSession` from `ResponsesClient.AsAIAgent(...)`. + +Both send a JSON brief on the first turn and a refinement on the second, following the rotating +`previous_response_id` chain (the CC path threads it by hand via `ChatOptions.ConversationId`; the MAF path +lets `AgentSession` do it). The follow-up only makes sense if the workflow resumed the first turn's +checkpoint, so it proves checkpoint continuity. + +```bash +dotnet run +``` + +Defaults to `http://localhost:5001`; override with `RESPONSES_SERVER_URL`. Start the server first. \ No newline at end of file diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/README.md b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/README.md new file mode 100644 index 00000000000..b6719528309 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/README.md @@ -0,0 +1,64 @@ +# Hosting Responses Workflow (client / server) + +A client/server pair showing how to expose a **workflow** over the OpenAI Responses protocol from an +app-owned ASP.NET Core route with per-session checkpoint resume, and how to consume it from .NET two +different ways. This mirrors the Python `local_responses_workflow` sample. + +``` +local_responses_workflow/ +├── Server/ # exposes POST /responses; previous_response_id continuation with checkpoint resume +└── Client/ # consumes it two ways: CC (IChatClient) and MAF (AIAgent) +``` + +## Server + +The server owns routing, authentication, and checkpoint storage. It uses the `OpenAIResponses` conversion +helpers for the wire protocol and `HostedWorkflowState` for per-session checkpoint resume. The workflow is a +brief adapter, a slogan-writer agent, and a formatter that renders one slogan line. The first turn runs the +workflow forward; later turns restore the latest checkpoint and run forward with the new brief. It binds to +`http://localhost:5001`. + +Like the Python sample, the server supports **`previous_response_id` continuation only** and **rejects +`conversation` continuity with HTTP 400**. Because `previous_response_id` rotates every turn, the app owns a +cursor that maps each response id to the stable workflow session id (the .NET equivalent of the Python +sample's `CheckpointCursorStore`). + +See [Server/README.md](Server/README.md). + +## Client + +A single program that runs the same two-turn conversation twice, once per consumption path: + +- **CC** — a plain `Microsoft.Extensions.AI.IChatClient` (the lower-level chat-client path). +- **MAF** — a Microsoft Agent Framework `AIAgent` + `AgentSession` (the higher-level agent path). + +Both send a JSON brief on the first turn and a refinement on the second, following the rotating +`previous_response_id` chain (the CC path threads it by hand; the MAF path lets `AgentSession` do it). The +follow-up only makes sense if the workflow resumed the first turn's checkpoint. + +See [Client/README.md](Client/README.md). + +## Run + +Start the server in one shell: + +```bash +export FOUNDRY_PROJECT_ENDPOINT="https://.services.ai.azure.com/api/projects/" +export FOUNDRY_MODEL="gpt-5.4-mini" # optional, defaults to gpt-5.4-mini +dotnet run --project Server +``` + +Then run the client in another shell: + +```bash +dotnet run --project Client +``` + +The client defaults to `http://localhost:5001`; override with `RESPONSES_SERVER_URL`. + +## Why previous_response_id needs a cursor + +`previous_response_id` changes every turn, so it cannot key checkpoint storage directly. The app maps each +response id to the stable workflow session id, so every id in the rotating chain resumes the same +checkpointed run. Sending `conversation` is rejected with HTTP 400 to keep this sample focused on one +continuation mode. \ No newline at end of file diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/Program.cs b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/Program.cs new file mode 100644 index 00000000000..94b21fa8bb6 --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/Program.cs @@ -0,0 +1,196 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample shows how an application can own its own ASP.NET Core route and expose a workflow over the +// OpenAI Responses protocol. It uses the OpenAIResponses conversion helpers for the wire protocol and +// HostedWorkflowState for per-session checkpoint resume. The application keeps control of routing, auth, +// and checkpoint storage. +// +// Mirroring the Python local_responses_workflow sample, this server demonstrates previous_response_id +// continuation ONLY. It rejects conversation-id continuity with HTTP 400. Because previous_response_id +// rotates every turn, the app owns a cursor store that maps each response id to the stable workflow +// session id, so the whole rotating chain resumes the same checkpointed run (the .NET equivalent of the +// Python sample's file-backed CheckpointCursorStore). + +using System.Collections.Concurrent; +using System.Text.Json; +using Azure.AI.Projects; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.Hosting; +using Microsoft.Agents.AI.Hosting.OpenAI; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Extensions.AI; + +var builder = WebApplication.CreateBuilder(args); + +// Configuration via environment variables (never hardcode secrets). +string endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT") + ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set."); +string model = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-5.4-mini"; + +// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production. +// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid +// latency issues, unintended credential probing, and potential security risks from fallback mechanisms. +var projectClient = new AIProjectClient(new Uri(endpoint), new AzureCliCredential()); +AIAgent writer = projectClient.AsAIAgent( + model: model, + instructions: "You are an excellent slogan writer. Create one short slogan from the given brief.", + name: "writer"); + +// Workflow shape (mirrors the Python sample): a brief adapter turns the Responses input into the writer's +// prompt and drives the agent turn, then a formatter renders the writer's output as a single slogan line. +var briefExecutor = new BriefExecutor(); +var formatterExecutor = new SloganFormatterExecutor(); + +Workflow workflow = new WorkflowBuilder(briefExecutor) + .AddEdge(briefExecutor, writer) + .AddEdge(writer, formatterExecutor) + .WithOutputFrom(formatterExecutor) + .Build(); + +// Optional shared execution state: pairs the workflow with an in-memory CheckpointManager and a per-session +// sessionId -> CheckpointInfo head cursor so a session can resume from its last checkpoint. +var state = new HostedWorkflowState(workflow); + +// App-owned response-id -> workflow-session-id cursor. previous_response_id rotates each turn, so every id in +// a conversation's chain maps to the same workflow session, and resuming any of them restores that session's +// latest checkpoint. In-memory for this local sample; a real app persists this per tenant/user. +var responseToSession = new ConcurrentDictionary(StringComparer.Ordinal); + +var app = builder.Build(); + +// The application owns this route. Binding the body as JsonElement lets ASP.NET Core deserialize the JSON +// request body directly, so there is no JsonDocument to own or dispose. +app.MapPost("/responses", async (JsonElement body, CancellationToken cancellationToken) => +{ + OpenAIResponsesRunRequest run; + try + { + run = OpenAIResponses.ToAgentRunRequest(body); + } + catch (ArgumentException) + { + return Results.BadRequest(); + } + + // This sample supports previous_response_id continuation only. GetSessionId returns previous_response_id + // first, otherwise the conversation id; anything that is not a "resp_" id is a conversation id, which this + // server does not implement. The candidate is untrusted: a real app authenticates the caller and + // authorizes/binds it before use. + string? candidate = OpenAIResponses.GetSessionId(body); + if (candidate?.StartsWith("resp_", StringComparison.Ordinal) == false) + { + return Results.Problem( + detail: "This server supports previous_response_id continuation only; conversation is not implemented.", + statusCode: StatusCodes.Status400BadRequest); + } + + string? previousResponseId = candidate; + string responseId = OpenAIResponses.CreateResponseId(); + + // Resolve the workflow session: continue the chain's session when previous_response_id is known, otherwise + // start a fresh workflow continuation. + string sessionId = previousResponseId is not null && responseToSession.TryGetValue(previousResponseId, out string? existing) + ? existing + : Guid.NewGuid().ToString("N"); + + // Runs the workflow forward on the first call for this session, or restores the session's latest checkpoint + // and runs forward with this turn's brief thereafter, then records the new head checkpoint. + string brief = ExtractBrief(run.Messages); + HostedWorkflowRunResult result = await state.RunOrResumeAsync(sessionId, brief, cancellationToken).ConfigureAwait(false); + + // Map this response id onto the workflow session so the next previous_response_id continues the same run. + responseToSession[responseId] = sessionId; + + AgentResponse response = BuildWorkflowResponse(result); + return Results.Json(OpenAIResponses.WriteResponse(response, responseId, previousResponseId)); +}); + +// Bind to a fixed local URL so the paired client sample has a deterministic default. +// Override with the ASPNETCORE_URLS environment variable when needed. +app.Run("http://localhost:5001"); + +// Flattens the Responses input messages into a single brief string for the workflow's start executor. +static string ExtractBrief(IEnumerable messages) + => string.Join("\n", messages.Select(m => m.Text).Where(t => !string.IsNullOrWhiteSpace(t))).Trim(); + +// Extracts the workflow's final string output (the formatted slogan) from its output events, falling back to a +// short run summary when the workflow emitted no string output this turn. +static AgentResponse BuildWorkflowResponse(HostedWorkflowRunResult result) +{ + string? slogan = null; + foreach (WorkflowEvent evt in result.Events) + { + if (evt is WorkflowOutputEvent output && output.Data is string text && !string.IsNullOrWhiteSpace(text)) + { + slogan = text; + } + } + + return new AgentResponse(new ChatMessage( + ChatRole.Assistant, + slogan ?? $"{result.Events.Count} workflow event(s) processed.")); +} + +/// +/// Adapts the Responses brief into the writer agent's turn. It builds the writer prompt from the brief (a plain +/// topic string or a JSON object with topic/style/audience), sends it as a user message, and emits the +/// that drives the downstream agent. This keeps the workflow non-chat-protocol (its +/// output is a plain string) while still driving the agent, mirroring the Python sample's prompt builder. +/// +[SendsMessage(typeof(ChatMessage))] +[SendsMessage(typeof(TurnToken))] +internal sealed class BriefExecutor() : Executor("brief") +{ + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + string topic = message.Trim(); + string style = "modern"; + string audience = "general"; + + if (topic.StartsWith('{')) + { + try + { + using JsonDocument doc = JsonDocument.Parse(topic); + JsonElement root = doc.RootElement; + if (root.ValueKind == JsonValueKind.Object && root.TryGetProperty("topic", out JsonElement topicElement)) + { + topic = topicElement.GetString() ?? topic; + style = root.TryGetProperty("style", out JsonElement styleElement) ? styleElement.GetString() ?? style : style; + audience = root.TryGetProperty("audience", out JsonElement audienceElement) ? audienceElement.GetString() ?? audience : audience; + } + } + catch (JsonException) + { + // Not a JSON brief; treat the whole text as the topic. + } + } + + if (string.IsNullOrWhiteSpace(topic)) + { + topic = "a generic product"; + } + + string prompt = + $"Topic: {topic}\n" + + $"Style: {style}\n" + + $"Audience: {audience}\n\n" + + "Write a single short slogan that fits the topic, style, and audience."; + + await context.SendMessageAsync(new ChatMessage(ChatRole.User, prompt), cancellationToken: cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(new TurnToken(emitEvents: true), cancellationToken: cancellationToken).ConfigureAwait(false); + } +} + +/// +/// Formats the writer agent's output as the workflow's final response: one terminal-friendly slogan line. +/// +internal sealed class SloganFormatterExecutor() : Executor, string>("terminal_formatter") +{ + public override ValueTask HandleAsync(List message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + string slogan = string.Join("\n", message.Select(m => m.Text ?? string.Empty)).Trim().Trim('"'); + return ValueTask.FromResult($"Slogan: \"{slogan}\""); + } +} diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/README.md b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/README.md new file mode 100644 index 00000000000..dcbae86b6aa --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/README.md @@ -0,0 +1,31 @@ +# Server (Hosting Responses Workflow) + +Server half of the [Hosting Responses Workflow](../README.md) sample. + +Exposes a workflow over the OpenAI Responses protocol on an app-owned `POST /responses` route. The workflow +is a brief adapter, a slogan-writer agent, and a formatter that renders one slogan line. It uses the +`OpenAIResponses` conversion helpers for the wire protocol and `HostedWorkflowState` for per-session +checkpoint resume. + +Like the Python sample, this server supports **`previous_response_id` continuation only** and **rejects +`conversation` continuity with HTTP 400**. Because `previous_response_id` rotates every turn, the app owns a +cursor that maps each response id to the stable workflow session id, so the whole rotating chain resumes the +same checkpointed run. The first turn runs the workflow forward; later turns restore the latest checkpoint +and run forward with the new brief. Binds to `http://localhost:5001` (override with `ASPNETCORE_URLS`). + +```bash +export FOUNDRY_PROJECT_ENDPOINT="https://.services.ai.azure.com/api/projects/" +export FOUNDRY_MODEL="gpt-5.4-mini" +dotnet run +``` + +Call it directly, following the response-id chain across turns (the second call continues the first): + +```bash +curl -s http://localhost:5001/responses -H "content-type: application/json" \ + -d '{ "input": "{\"topic\": \"electric SUV\", \"style\": \"playful\", \"audience\": \"young families\"}" }' + +# Take the "id" (resp_...) from the response above and pass it as previous_response_id: +curl -s http://localhost:5001/responses -H "content-type: application/json" \ + -d '{ "input": "Make it a little more premium.", "previous_response_id": "resp_..." }' +``` \ No newline at end of file diff --git a/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/Server.csproj b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/Server.csproj new file mode 100644 index 00000000000..3831a5c20cc --- /dev/null +++ b/dotnet/samples/04-hosting/af-hosting/local_responses_workflow/Server/Server.csproj @@ -0,0 +1,21 @@ + + + + net10.0 + enable + enable + + + + + + + + + + + + + + + diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/OpenAIResponses.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/OpenAIResponses.cs new file mode 100644 index 00000000000..8f27b7ef227 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/OpenAIResponses.cs @@ -0,0 +1,165 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Hosting.OpenAI.Responses; +using Microsoft.Agents.AI.Hosting.OpenAI.Responses.Models; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Hosting.OpenAI; + +/// +/// Side-effect-free helpers that convert between the OpenAI Responses wire protocol and Agent Framework +/// run values, for applications that own their own HTTP route, authentication, middleware, and storage. +/// +/// +/// +/// These helpers are the app-owned-routing counterpart to MapOpenAIResponses. +/// MapOpenAIResponses owns routing and storage; these helpers let an application own those concerns +/// and reuse only the protocol conversion. Both share the same internal conversion logic. +/// +/// +/// Trust boundary. returns an untrusted candidate +/// continuation key. The application must authenticate the caller and authorize/bind the id to the +/// authenticated principal before using it as a session or checkpoint key. The helpers never perform I/O. +/// +/// +public static class OpenAIResponses +{ + /// + /// Converts an OpenAI Responses request body into Agent Framework run values (messages and options). + /// + /// The OpenAI Responses-shaped request body. + /// + /// Optional options controlling how request settings are mapped onto the run. By default no request + /// setting is mapped onto the run. + /// + /// The parsed messages and mapped run options. + /// The body could not be parsed as an OpenAI Responses request. + /// A request setting is not supported by the configured mapping. + public static OpenAIResponsesRunRequest ToAgentRunRequest(JsonElement body, OpenAIResponsesMapOptions? mapOptions = null) + { + CreateResponse request; + try + { + request = body.Deserialize(OpenAIHostingJsonContext.Default.CreateResponse) + ?? throw new ArgumentException("The request body could not be parsed as an OpenAI Responses request.", nameof(body)); + } + catch (JsonException ex) + { + throw new ArgumentException("The request body could not be parsed as an OpenAI Responses request.", nameof(body), ex); + } + + if (request.Input is null) + { + throw new ArgumentException("The request body is missing the required 'input' field.", nameof(body)); + } + + AgentRunOptions? options = (mapOptions ?? new OpenAIResponsesMapOptions()).RunOptionsFactory(request.ToRequestInfo()); + + var messages = new List(); + foreach (InputMessage inputMessage in request.Input.GetInputMessages()) + { + messages.Add(inputMessage.ToChatMessage()); + } + + return new OpenAIResponsesRunRequest(messages, options); + } + + /// + /// Converts a final into an OpenAI Responses-shaped payload. + /// + /// The agent response to render. + /// The id to assign to the rendered response (see ). + /// + /// The optional continuation/session id to surface as the response's conversation id. + /// + /// An OpenAI Responses-shaped . + public static JsonElement WriteResponse(AgentResponse response, string responseId, string? sessionId = null) + { + ArgumentNullException.ThrowIfNull(response); + ArgumentException.ThrowIfNullOrEmpty(responseId); + + AgentInvocationContext context = CreateContext(responseId, sessionId); + Response wire = response.ToResponse(EmptyRequest(), context); + return JsonSerializer.SerializeToElement(wire, OpenAIHostingJsonContext.Default.Response); + } + + /// + /// Converts a stream of into OpenAI Responses Server-Sent-Event frames. + /// + /// The agent streaming updates. + /// The id to assign to the rendered response (see ). + /// The optional continuation/session id to surface as the response's conversation id. + /// The to monitor for cancellation requests. + /// An async sequence of SSE frame strings, each terminated by a blank line. + public static async IAsyncEnumerable WriteResponseStreamAsync( + IAsyncEnumerable updates, + string responseId, + string? sessionId = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(updates); + ArgumentException.ThrowIfNullOrEmpty(responseId); + + AgentInvocationContext context = CreateContext(responseId, sessionId); + await foreach (StreamingResponseEvent streamingEvent in updates + .ToStreamingResponseAsync(EmptyRequest(), context, cancellationToken) + .ConfigureAwait(false)) + { + string json = JsonSerializer.Serialize(streamingEvent, OpenAIHostingJsonContext.Default.StreamingResponseEvent); + yield return $"event: {streamingEvent.Type}\ndata: {json}\n\n"; + } + } + + /// + /// Extracts the continuation/session id candidate from an OpenAI Responses request body. + /// + /// The OpenAI Responses-shaped request body. + /// + /// The previous_response_id when present; otherwise the conversation id when present; + /// otherwise . Returns for a body that cannot be parsed. + /// + /// + /// This is kept separate from so the + /// trust boundary stays visible: using a request-derived key is an explicit application decision. The returned + /// value is an untrusted candidate key until the application has authorized it for the caller. + /// + /// The Responses protocol treats previous_response_id and conversation as mutually exclusive; if a + /// payload sets both, this helper prefers previous_response_id (the response-chain pointer). Note that + /// previous_response_id changes each turn and is therefore not a stable partition key; prefer the + /// conversation id when a stable key is required (for example a workflow checkpoint cursor key). + /// + /// + public static string? GetSessionId(JsonElement body) + { + CreateResponse? request; + try + { + request = body.Deserialize(OpenAIHostingJsonContext.Default.CreateResponse); + } + catch (JsonException) + { + return null; + } + + return request?.PreviousResponseId ?? request?.Conversation?.Id; + } + + /// + /// Creates a new OpenAI Responses-shaped response id (a resp_* value). + /// + /// A new response id. + public static string CreateResponseId() => IdGenerator.NewId("resp"); + + private static AgentInvocationContext CreateContext(string responseId, string? sessionId) + => new(new IdGenerator(responseId, sessionId)); + + // The rendering converters never read the request input; a minimal request lets the facade render + // a response without requiring the caller to supply the originating request object. + private static CreateResponse EmptyRequest() => new() { Input = ResponseInput.FromText(string.Empty) }; +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/OpenAIResponsesRunRequest.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/OpenAIResponsesRunRequest.cs new file mode 100644 index 00000000000..8a75fd45044 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/OpenAIResponsesRunRequest.cs @@ -0,0 +1,36 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Hosting.OpenAI; + +/// +/// The result of converting an OpenAI Responses request body into Agent Framework run values via +/// . +/// +/// +/// This type carries the values an application passes to +/// (or the streaming equivalent) when it owns its own hosting route. It does not run the agent; the +/// application remains in control of when and how the run happens. +/// +public sealed class OpenAIResponsesRunRequest +{ + internal OpenAIResponsesRunRequest(IList messages, AgentRunOptions? options) + { + this.Messages = messages; + this.Options = options; + } + + /// + /// Gets the chat messages parsed from the request body, ready to pass to an run. + /// + public IList Messages { get; } + + /// + /// Gets the run options mapped from the request, or when no request setting is + /// mapped onto the run. The mapping is controlled by ; + /// by default no request setting is mapped. + /// + public AgentRunOptions? Options { get; } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/AgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/AgentSessionStore.cs index 7c0539fe511..4fe31164def 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/AgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/AgentSessionStore.cs @@ -76,6 +76,24 @@ public abstract ValueTask GetSessionAsync( string conversationId, CancellationToken cancellationToken = default); + /// + /// Deletes a stored agent session, if present. + /// + /// The agent that owns this session. + /// The unique identifier for the conversation/session to delete. + /// The to monitor for cancellation requests. + /// A task that represents the asynchronous delete operation. + /// + /// The default implementation throws . Stores that support removal + /// override this method. Deleting a missing session is a no-op for stores that override it. + /// + /// The store does not support deletion. + public virtual ValueTask DeleteSessionAsync( + AIAgent agent, + string conversationId, + CancellationToken cancellationToken = default) + => throw new NotSupportedException($"{this.GetType().Name} does not support session deletion."); + /// Asks the for an object of the specified type . /// The type of object being requested. /// An optional key that can be used to help identify the target service. diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/DelegatingAgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/DelegatingAgentSessionStore.cs index e80d3b907a3..faa4e9bd1cf 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/DelegatingAgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/DelegatingAgentSessionStore.cs @@ -60,6 +60,10 @@ public override ValueTask GetSessionAsync(AIAgent agent, string co public override ValueTask SaveSessionAsync(AIAgent agent, string conversationId, AgentSession session, CancellationToken cancellationToken = default) => this.InnerStore.SaveSessionAsync(agent, conversationId, session, cancellationToken); + /// + public override ValueTask DeleteSessionAsync(AIAgent agent, string conversationId, CancellationToken cancellationToken = default) + => this.InnerStore.DeleteSessionAsync(agent, conversationId, cancellationToken); + /// /// /// This implementation first checks if this instance satisfies the service request. diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/HostedAgentState.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedAgentState.cs new file mode 100644 index 00000000000..fef9a0bf58d --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedAgentState.cs @@ -0,0 +1,140 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Hosting; + +/// +/// Optional shared execution state for applications that own their own hosting route and want to reuse +/// Agent Framework session continuity. Pairs a single target with an +/// . +/// +/// +/// +/// This holder exists because only an object that has both the resolved agent target and the session +/// store can offer a target-aware get-or-create. It does not own routing, authentication, middleware, or +/// storage policy; those remain with the application. It does not replace , +/// which already provides serialization and per-principal isolation. +/// +/// +/// Trust boundary. The sessionId values passed to these methods are +/// application-selected partition keys. When a key originates from the wire (for example via +/// OpenAIResponses.GetSessionId(...)), the application must authenticate the caller and authorize +/// the key before using it here. For multi-user hosts, scope the underlying store per principal with +/// . +/// +/// +public sealed class HostedAgentState +{ + private readonly AIAgent _agent; + private readonly AgentSessionStore _sessionStore; + private readonly ConcurrentDictionary? _sessionLocks; + + /// + /// Initializes a new instance of the class. + /// + /// The agent target used by route code. + /// + /// The session store to use. Defaults to a fresh when not provided. + /// + /// + /// When , serializes access per session id. Defaults + /// to . + /// + /// is . + public HostedAgentState(AIAgent agent, AgentSessionStore? sessionStore = null, bool enableSessionLocking = false) + { + _ = Throw.IfNull(agent); + + this._agent = agent; + this._sessionStore = sessionStore ?? new InMemoryAgentSessionStore(); + this._sessionLocks = enableSessionLocking ? new ConcurrentDictionary(StringComparer.Ordinal) : null; + } + + /// + /// Returns the stored session for , creating a new session on first use. + /// + /// The application-selected session id. + /// The to monitor for cancellation requests. + /// The resolved or newly created . + public ValueTask GetOrCreateSessionAsync(string sessionId, CancellationToken cancellationToken = default) + { + _ = Throw.IfNullOrEmpty(sessionId); + return this._sessionStore.GetSessionAsync(this._agent, sessionId, cancellationToken); + } + + /// + /// Persists under . Call this after the run + /// completes, including under a newly minted continuation id when the protocol mints one. + /// + /// The application-selected session id (may be a newly minted id). + /// The session to persist. + /// The to monitor for cancellation requests. + /// A task representing the asynchronous save operation. + public ValueTask SaveSessionAsync(string sessionId, AgentSession session, CancellationToken cancellationToken = default) + { + _ = Throw.IfNullOrEmpty(sessionId); + _ = Throw.IfNull(session); + return this._sessionStore.SaveSessionAsync(this._agent, sessionId, session, cancellationToken); + } + + /// + /// Deletes the stored session for , if present. + /// + /// The application-selected session id. + /// The to monitor for cancellation requests. + /// A task representing the asynchronous delete operation. + /// The underlying store does not support deletion. + public ValueTask DeleteSessionAsync(string sessionId, CancellationToken cancellationToken = default) + { + _ = Throw.IfNullOrEmpty(sessionId); + return this._sessionStore.DeleteSessionAsync(this._agent, sessionId, cancellationToken); + } + + /// + /// Acquires an exclusive lock for so concurrent requests for the same + /// session serialize their get-run-save cycle. Dispose the returned value to release the lock. + /// + /// The application-selected session id. + /// The to monitor for cancellation requests. + /// An that releases the lock when disposed. + /// + /// When session locking is not enabled, this returns immediately with a no-op releaser. + /// + public async ValueTask LockSessionAsync(string sessionId, CancellationToken cancellationToken = default) + { + _ = Throw.IfNullOrEmpty(sessionId); + + if (this._sessionLocks is null) + { + return NoopReleaser.Instance; + } + + SemaphoreSlim gate = this._sessionLocks.GetOrAdd(sessionId, static _ => new SemaphoreSlim(1, 1)); + await gate.WaitAsync(cancellationToken).ConfigureAwait(false); + return new SemaphoreReleaser(gate); + } + + private sealed class SemaphoreReleaser(SemaphoreSlim gate) : IAsyncDisposable + { + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "The semaphore is owned by the per-session dictionary and reused across callers; the releaser only releases it.")] + private SemaphoreSlim? _gate = gate; + + public ValueTask DisposeAsync() + { + Interlocked.Exchange(ref this._gate, null)?.Release(); + return default; + } + } + + private sealed class NoopReleaser : IAsyncDisposable + { + public static readonly NoopReleaser Instance = new(); + + public ValueTask DisposeAsync() => default; + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowRunResult.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowRunResult.cs new file mode 100644 index 00000000000..4a71bf33005 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowRunResult.cs @@ -0,0 +1,34 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; + +namespace Microsoft.Agents.AI.Hosting; + +/// +/// The result of a run or resume. +/// +public sealed class HostedWorkflowRunResult +{ + internal HostedWorkflowRunResult(string sessionId, IReadOnlyList events, Workflows.CheckpointInfo? checkpoint) + { + this.SessionId = sessionId; + this.Events = events; + this.Checkpoint = checkpoint; + } + + /// + /// Gets the application-selected session id this run was executed under. + /// + public string SessionId { get; } + + /// + /// Gets the workflow events emitted during this run. + /// + public IReadOnlyList Events { get; } + + /// + /// Gets the head checkpoint recorded for the session after this run, or when + /// checkpointing produced no checkpoint. + /// + public Workflows.CheckpointInfo? Checkpoint { get; } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowState.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowState.cs new file mode 100644 index 00000000000..61ffdab0b20 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowState.cs @@ -0,0 +1,303 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Hosting; + +/// +/// Optional shared execution state for applications that own their own hosting route and want to expose a +/// workflow with per-session checkpoint resume. Pairs a target with a +/// and an application-scoped sessionId -> CheckpointInfo head cursor. +/// +/// +/// +/// The .NET workflow checkpoint store is already keyed by session id, but carries +/// no ordering, so this holder remembers the head checkpoint per session to resume the correct one. It does not +/// own routing, authentication, or storage policy. +/// +/// +/// The in-memory head cursor accelerates the common case, but when it misses (for example a new holder or a +/// process restart) the holder falls back to . A durable +/// therefore resumes correctly across restarts; the default in-memory manager does +/// not persist, so with it a restart starts the session fresh. +/// +/// +/// Trust boundary. sessionId is an application-selected partition key. When it originates +/// from the wire, the application must authenticate the caller and authorize the key before using it here. The +/// checkpoint boundary must be at least as specific as the authorized session boundary. +/// +/// +public sealed class HostedWorkflowState : IDisposable +{ + private readonly CheckpointManager _checkpointManager; + private readonly IWorkflowExecutionEnvironment _executionEnvironment; + private readonly Workflow _workflow; + private readonly ILogger _logger; + private readonly ConcurrentDictionary _cursor = new(StringComparer.Ordinal); + + // A single workflow instance backs every session on this holder, and workflow instances do not support + // concurrent runs, so all turns are serialized through one lock (mirroring the Python host's workflow lock). + private readonly SemaphoreSlim _workflowLock = new(1, 1); + + /// + /// Initializes a new instance of the class. + /// + /// The workflow target. + /// + /// The checkpoint manager to use. Defaults to when not provided. + /// + /// + /// The workflow execution environment used to run and resume the workflow. Defaults to an in-process environment + /// () configured with . Supplying a + /// custom environment (for example a future durable/out-of-process environment) is supported; the supplied + /// environment must be configured to checkpoint into the same store as , since + /// the holder reads that manager directly to recover the head checkpoint when its in-memory cursor misses. + /// + /// + /// The logger factory used to report resume diagnostics (for example, a resume turn that made no progress). + /// Defaults to when not provided. + /// + /// is . + public HostedWorkflowState( + Workflow workflow, + CheckpointManager? checkpointManager = null, + IWorkflowExecutionEnvironment? executionEnvironment = null, + ILoggerFactory? loggerFactory = null) + { + _ = Throw.IfNull(workflow); + + this._workflow = workflow; + this._checkpointManager = checkpointManager ?? CheckpointManager.CreateInMemory(); + this._executionEnvironment = executionEnvironment ?? InProcessExecution.Default.WithCheckpointing(this._checkpointManager); + this._logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger(typeof(HostedWorkflowState)); + } + + /// + /// Runs the workflow forward for with checkpointing on the first turn, or, on + /// subsequent turns, restores the session's recorded head checkpoint and then runs the workflow forward with + /// the new turn's . The new head checkpoint is recorded for the session afterwards. + /// + /// + /// The resume semantics mirror the Python hosting host (agent_framework_hosting's _invoke_workflow): + /// each turn restores the latest checkpoint to rehydrate accumulated workflow state and then applies the new + /// input, rather than continuing a halted run with no input (which would leave the run waiting for input + /// indefinitely). For agent (chat-protocol) workflows the new input is accompanied by a + /// so the turn is driven, matching the fresh-run path. + /// + /// The workflow input type. + /// The application-selected session id. + /// The input to run on this turn (used both when starting a new run and when resuming). + /// The to monitor for cancellation requests. + /// The run result, including the events emitted on this turn and the recorded head checkpoint. + public async ValueTask RunOrResumeAsync(string sessionId, TInput input, CancellationToken cancellationToken = default) + where TInput : notnull + { + _ = Throw.IfNullOrEmpty(sessionId); + _ = Throw.IfNull(input); + + // Serialize turns: the shared workflow instance cannot be run by two runners at once, and concurrent + // same-session turns would otherwise race the head cursor. + await this._workflowLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + return await this.RunOrResumeCoreAsync(sessionId, input, cancellationToken).ConfigureAwait(false); + } + finally + { + this._workflowLock.Release(); + } + } + + private async ValueTask RunOrResumeCoreAsync(string sessionId, TInput input, CancellationToken cancellationToken) + where TInput : notnull + { + if (!this._cursor.TryGetValue(sessionId, out CheckpointInfo? head)) + { + // The in-memory cursor is empty for this session. Fall back to the checkpoint manager so a durable + // manager still resumes after the cursor is lost (for example a process restart or a new holder over + // the same store), mirroring the Python host's per-turn get_latest read-through. + head = await this._checkpointManager.GetLatestCheckpointAsync(sessionId, cancellationToken).ConfigureAwait(false); + } + + if (head is null) + { + // First turn for this session: run the workflow forward from its start executor with the input. + Run freshRun = await this._executionEnvironment.RunAsync(this._workflow, input, sessionId, cancellationToken).ConfigureAwait(false); + await using (freshRun.ConfigureAwait(false)) + { + return this.Record(sessionId, freshRun.OutgoingEvents.ToList(), freshRun.LastCheckpoint); + } + } + + // Subsequent turn: restore the session's latest checkpoint to rehydrate accumulated workflow state, then + // run the workflow forward with the new turn's input. Agent workflows use the chat protocol, which requires + // a TurnToken to drive the turn (mirroring how the fresh-run path seeds one). + // + // The streaming resume restores state without draining to a halt first; the non-streaming resume would + // block waiting for input immediately after restore (before we can deliver the new input). + ProtocolDescriptor descriptor = await this._workflow.DescribeProtocolAsync(cancellationToken).ConfigureAwait(false); + + StreamingRun resumed = await this._executionEnvironment.ResumeStreamingAsync(this._workflow, head, cancellationToken).ConfigureAwait(false); + await using (resumed.ConfigureAwait(false)) + { + await resumed.TrySendMessageAsync(input).ConfigureAwait(false); + if (descriptor.IsChatProtocol() && input is not TurnToken) + { + await resumed.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); + } + + List events = []; + // Drain non-blocking on pending requests, matching the first-turn RunAsync path + // (Run.RunToNextHaltAsync also uses blockOnPendingRequest: false): the workflow may halt awaiting an + // external response, and blocking there would wait indefinitely. + await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken).ConfigureAwait(false)) + { + events.Add(evt); + } + + if (events.Count == 0) + { + this.WarnOnNoProgress(sessionId); + } + + return this.Record(sessionId, events, resumed.LastCheckpoint); + } + } + + /// + /// Streams the events of a run-or-resume turn as they occur, applying the same restore-then-run semantics as + /// : the first turn runs the workflow + /// forward from its start executor, and subsequent turns restore the session's latest checkpoint and run + /// forward with . The session's head checkpoint is recorded when the stream ends, + /// including when the consumer abandons enumeration early. + /// + /// + /// Turns are serialized through the holder's workflow lock, which is held for the lifetime of the returned + /// enumerator. The caller must enumerate the stream to completion (or dispose it) to release the lock. The head + /// checkpoint is recorded from the run's last committed checkpoint when the stream ends — whether it completes + /// normally or the consumer disposes it early — so an interrupted turn still advances the session cursor. + /// + /// The workflow input type. + /// The application-selected session id. + /// The input to run on this turn. + /// The to monitor for cancellation requests. + /// An asynchronous stream of the s emitted during this turn. + public async IAsyncEnumerable RunOrResumeStreamingAsync(string sessionId, TInput input, [EnumeratorCancellation] CancellationToken cancellationToken = default) + where TInput : notnull + { + _ = Throw.IfNullOrEmpty(sessionId); + _ = Throw.IfNull(input); + + await this._workflowLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + if (!this._cursor.TryGetValue(sessionId, out CheckpointInfo? head)) + { + head = await this._checkpointManager.GetLatestCheckpointAsync(sessionId, cancellationToken).ConfigureAwait(false); + } + + ProtocolDescriptor descriptor = await this._workflow.DescribeProtocolAsync(cancellationToken).ConfigureAwait(false); + + // The fresh streaming run enqueues the input itself; the streaming resume restores state and needs the + // input delivered explicitly. Neither streaming entry point seeds a TurnToken, so drive chat-protocol + // workflows with one on both paths. + StreamingRun run = head is null + ? await this._executionEnvironment.RunStreamingAsync(this._workflow, input, sessionId, cancellationToken).ConfigureAwait(false) + : await this._executionEnvironment.ResumeStreamingAsync(this._workflow, head, cancellationToken).ConfigureAwait(false); + + await using (run.ConfigureAwait(false)) + { + if (head is not null) + { + await run.TrySendMessageAsync(input).ConfigureAwait(false); + } + + if (descriptor.IsChatProtocol() && input is not TurnToken) + { + await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); + } + + int eventCount = 0; + try + { + // Drain non-blocking on pending requests (see RunOrResumeCoreAsync) so a workflow that halts + // awaiting an external response ends the stream instead of blocking indefinitely. + await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellationToken).ConfigureAwait(false)) + { + eventCount++; + yield return evt; + } + + if (eventCount == 0 && head is not null) + { + this.WarnOnNoProgress(sessionId); + } + } + finally + { + // Record the head checkpoint even when the consumer abandons the stream (for example an SSE + // client disconnect), so an interrupted turn still advances the session cursor to the last + // committed checkpoint and a later turn resumes from there rather than re-running prior work. + this.UpdateCursor(sessionId, run.LastCheckpoint); + } + } + } + finally + { + this._workflowLock.Release(); + } + } + + private HostedWorkflowRunResult Record(string sessionId, List events, CheckpointInfo? checkpoint) + { + this.UpdateCursor(sessionId, checkpoint); + return new HostedWorkflowRunResult(sessionId, events, checkpoint); + } + + private void UpdateCursor(string sessionId, CheckpointInfo? checkpoint) + { + if (checkpoint is not null) + { + this._cursor[sessionId] = checkpoint; + } + } + + private void WarnOnNoProgress(string sessionId) + // The resumed turn drove no work. This mirrors the Python host's zero-event restore warning: the + // checkpoint may be stale or the input may not match the workflow's expected type, so the session's + // state may not have progressed. + => this._logger.LogWarning( + "Resuming workflow session '{SessionId}' produced no events; the checkpoint may be stale or the input may not match the workflow's expected input type. Session state may not have progressed.", + sessionId); + + /// + /// Gets the recorded head checkpoint for , if any. + /// + /// The application-selected session id. + /// When this method returns, the recorded head checkpoint, or . + /// when a checkpoint is recorded for the session; otherwise . + /// + /// Internal cursor-inspection helper used by tests; not part of the public surface. + /// + internal bool TryGetCheckpoint(string sessionId, out CheckpointInfo? checkpoint) + { + _ = Throw.IfNullOrEmpty(sessionId); + return this._cursor.TryGetValue(sessionId, out checkpoint); + } + + /// + /// Releases the resources used by this instance, including the workflow serialization lock. + /// + public void Dispose() => this._workflowLock.Dispose(); +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/IsolationKeyScopedAgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/IsolationKeyScopedAgentSessionStore.cs index f379d625fd1..1ee4c46b44a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/IsolationKeyScopedAgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/IsolationKeyScopedAgentSessionStore.cs @@ -106,4 +106,11 @@ public override async ValueTask SaveSessionAsync(AIAgent agent, string conversat string scopedConversationId = await this.GetScopedConversationIdAsync(conversationId, cancellationToken).ConfigureAwait(false); await this.InnerStore.SaveSessionAsync(agent, scopedConversationId, session, cancellationToken).ConfigureAwait(false); } + + /// + public override async ValueTask DeleteSessionAsync(AIAgent agent, string conversationId, CancellationToken cancellationToken = default) + { + string scopedConversationId = await this.GetScopedConversationIdAsync(conversationId, cancellationToken).ConfigureAwait(false); + await this.InnerStore.DeleteSessionAsync(agent, scopedConversationId, cancellationToken).ConfigureAwait(false); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/Local/InMemoryAgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/Local/InMemoryAgentSessionStore.cs index 832c411977f..d9a8f3613d6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/Local/InMemoryAgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/Local/InMemoryAgentSessionStore.cs @@ -63,5 +63,12 @@ public override async ValueTask GetSessionAsync(AIAgent agent, str }; } + /// + public override ValueTask DeleteSessionAsync(AIAgent agent, string conversationId, CancellationToken cancellationToken = default) + { + this._threads.TryRemove(GetKey(conversationId, agent.Id), out _); + return default; + } + private static string GetKey(string conversationId, string agentId) => $"{agentId}:{conversationId}"; } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/NoopAgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/NoopAgentSessionStore.cs index 163c1ea867e..cc1df02a33b 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/NoopAgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/NoopAgentSessionStore.cs @@ -14,7 +14,7 @@ public sealed class NoopAgentSessionStore : AgentSessionStore /// public override ValueTask SaveSessionAsync(AIAgent agent, string conversationId, AgentSession session, CancellationToken cancellationToken = default) { - return new ValueTask(); + return default; } /// @@ -22,4 +22,10 @@ public override ValueTask GetSessionAsync(AIAgent agent, string co { return agent.CreateSessionAsync(cancellationToken); } + + /// + public override ValueTask DeleteSessionAsync(AIAgent agent, string conversationId, CancellationToken cancellationToken = default) + { + return default; + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/CheckpointManager.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/CheckpointManager.cs index 2b5bb5b0347..efaf4f8abd7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/CheckpointManager.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/CheckpointManager.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Text.Json; +using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Checkpointing; @@ -58,4 +59,29 @@ ValueTask ICheckpointManager.LookupCheckpointAsync(string sessionId, ValueTask> ICheckpointManager.RetrieveIndexAsync(string sessionId, CheckpointInfo? withParent) => this._impl.RetrieveIndexAsync(sessionId, withParent); + + /// + /// Retrieves the most recently committed checkpoint for the specified session, or + /// when the session has no checkpoints. + /// + /// The session identifier whose latest checkpoint should be retrieved. + /// The to monitor for cancellation requests. + /// + /// The latest for , or when no + /// checkpoint has been committed for that session. + /// + public async ValueTask GetLatestCheckpointAsync(string sessionId, CancellationToken cancellationToken = default) + { + // ICheckpointStore.RetrieveIndexAsync is contractually required to return checkpoints in commit order + // (oldest first, most recently committed last), so the last enumerated entry is the latest checkpoint. + IEnumerable index = await this._impl.RetrieveIndexAsync(sessionId, withParent: null).ConfigureAwait(false); + + CheckpointInfo? latest = null; + foreach (CheckpointInfo info in index) + { + latest = info; + } + + return latest; + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/FileSystemJsonCheckpointStore.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/FileSystemJsonCheckpointStore.cs index 0c3d57976d2..cb0891d6179 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/FileSystemJsonCheckpointStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/FileSystemJsonCheckpointStore.cs @@ -34,7 +34,19 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos private FileStream? _indexFile; internal DirectoryInfo Directory { get; } + + // O(1) membership set used to allocate unique checkpoint ids (GetUnusedCheckpointInfo) and to guard + // RetrieveCheckpointAsync. It intentionally coexists with OrderedCheckpointIndex below: this set answers + // "does this checkpoint exist?" in O(1), while the list preserves commit order. A single HashSet cannot do + // both because HashSet enumeration order is not a contract. internal HashSet CheckpointIndex { get; } + + // Insertion-ordered mirror of CheckpointIndex. HashSet enumeration order is not a contract (it can diverge + // from insertion order once a slot is freed by a rollback and reused), so RetrieveIndexAsync enumerates this + // list to return checkpoints in commit order, which callers such as CheckpointManager.GetLatestCheckpointAsync + // rely on to identify the head checkpoint. + private List OrderedCheckpointIndex { get; } = []; + private Dictionary CheckpointParents { get; } = []; private HashSet CheckpointsWithKnownParent { get; } = []; @@ -80,7 +92,11 @@ public FileSystemJsonCheckpointStore(DirectoryInfo directory) { // We never actually use the file names from the index entries since they can be derived from the CheckpointInfo, but it is useful to // have the UrlEncoded file names in the index file for human readability - this.CheckpointIndex.Add(entry.CheckpointInfo); + if (this.CheckpointIndex.Add(entry.CheckpointInfo)) + { + this.OrderedCheckpointIndex.Add(entry.CheckpointInfo); + } + this.CheckpointParents[entry.CheckpointInfo] = entry.ParentCheckpointId; if (entry.HasParentMetadata) { @@ -129,6 +145,8 @@ private CheckpointInfo GetUnusedCheckpointInfo(string sessionId) key = new(sessionId); } while (!this.CheckpointIndex.Add(key)); + this.OrderedCheckpointIndex.Add(key); + return key; } @@ -164,6 +182,7 @@ public override async ValueTask CreateCheckpointAsync(string ses catch (Exception ex) { this.CheckpointIndex.Remove(key); + this.OrderedCheckpointIndex.Remove(key); this.CheckpointParents.Remove(key); this.CheckpointsWithKnownParent.Remove(key); @@ -202,7 +221,7 @@ public override ValueTask> RetrieveIndexAsync(string { this.CheckDisposed(); - return new(this.CheckpointIndex + return new(this.OrderedCheckpointIndex .Where(checkpoint => checkpoint.SessionId == sessionId && (withParent is null || !this.CheckpointsWithKnownParent.Contains(checkpoint) || diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointStore.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointStore.cs index af2fa5423e1..b4a65b70abc 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/ICheckpointStore.cs @@ -19,8 +19,14 @@ public interface ICheckpointStore /// An optional parent checkpoint to filter the results. If specified, only checkpoints with the given parent are /// returned; otherwise, all checkpoints for the session are included. /// A value task representing the asynchronous operation. The result contains a collection of objects associated with the specified session. The collection is empty if no checkpoints are - /// found. + /// cref="CheckpointInfo"/> objects associated with the specified session, ordered by commit time from the oldest to the + /// most recently committed. The collection is empty if no checkpoints are found. + /// + /// Implementations must return checkpoints in the order they were committed, with the most recently committed checkpoint + /// last. This ordering is a contract of the store: callers such as rely on it to identify + /// the latest checkpoint for a session, so a store that returns checkpoints unordered (or newest-first) will cause the + /// wrong checkpoint to be resumed. + /// ValueTask> RetrieveIndexAsync(string sessionId, CheckpointInfo? withParent = null); /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs index c7a19380b2b..db8e449177e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRun.cs @@ -77,7 +77,24 @@ public IAsyncEnumerable WatchStreamAsync( CancellationToken cancellationToken = default) => this.WatchStreamAsync(blockOnPendingRequest: true, cancellationToken); - internal IAsyncEnumerable WatchStreamAsync( + /// + /// Asynchronously streams workflow events as they occur, with control over how the stream behaves when the + /// workflow halts awaiting an external response. + /// + /// + /// When is (the default behavior of + /// ), the stream pauses at a pending-request halt and resumes + /// once a response is supplied via . When it is , the + /// stream instead ends at that halt, returning control to the caller — matching the non-streaming + /// behavior, which is useful for hosts that drive their own turn loop. + /// + /// to block at a pending-request halt awaiting a + /// response; to end the stream at that halt. + /// A that can be used to cancel the streaming + /// operation. If cancellation is requested, the stream will end and no further events will be yielded, but this + /// will not cancel the workflow execution. + /// An asynchronous stream of objects representing significant workflow state changes. + public IAsyncEnumerable WatchStreamAsync( bool blockOnPendingRequest, CancellationToken cancellationToken = default) => this._runHandle.TakeEventStreamAsync(blockOnPendingRequest, cancellationToken); diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests.csproj b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests.csproj new file mode 100644 index 00000000000..94fc6720017 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests.csproj @@ -0,0 +1,19 @@ + + + + $(TargetFrameworksCore) + True + $(NoWarn);OPENAI001; + + + + + + + + + + + + + diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests/OpenAIResponsesHostingLiveTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests/OpenAIResponsesHostingLiveTests.cs new file mode 100644 index 00000000000..04819bc65ef --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests/OpenAIResponsesHostingLiveTests.cs @@ -0,0 +1,96 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Linq; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; +using OpenAI; +using Shared.IntegrationTests; + +namespace Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests; + +/// +/// Live integration tests for the app-owned routing helper surface ( plus +/// ) exercised against a real OpenAI model. These confirm the crucial +/// consumption paths — request conversion, an agent run, response rendering, and multi-turn session +/// continuity — behave correctly end to end with a real chat client. +/// +/// +/// Skipped unless the OpenAI configuration is present (OPENAI_API_KEY), so runs without secrets stay +/// green. The in-process, in-memory (no live model) coverage lives in the +/// Microsoft.Agents.AI.Hosting.OpenAI.UnitTests project (OpenAIResponsesHostingTests). +/// +public sealed class OpenAIResponsesHostingLiveTests +{ + private static string? ApiKey => Environment.GetEnvironmentVariable(TestSettings.OpenAIApiKey); + private static string ModelName => Environment.GetEnvironmentVariable(TestSettings.OpenAIChatModelName) ?? "gpt-4o-mini"; + + [Fact] + public async Task NonStreamingRun_RendersResponsesShapedPayloadAsync() + { + // Arrange + Assert.SkipWhen(string.IsNullOrEmpty(ApiKey), "OPENAI_API_KEY is not configured; skipping live hosting test."); + AIAgent agent = CreateAgent(); + var state = new HostedAgentState(agent); + JsonElement body = ParseBody("""{ "input": "Reply with exactly the word: apple" }"""); + + // Act + string sessionId = OpenAIResponses.GetSessionId(body) ?? OpenAIResponses.CreateResponseId(); + OpenAIResponsesRunRequest run = OpenAIResponses.ToAgentRunRequest(body); + AgentSession session = await state.GetOrCreateSessionAsync(sessionId); + string responseId = OpenAIResponses.CreateResponseId(); + AgentResponse result = await agent.RunAsync(run.Messages, session, run.Options); + JsonElement payload = OpenAIResponses.WriteResponse(result, responseId, responseId); + + // Assert + Assert.Equal(responseId, payload.GetProperty("id").GetString()); + Assert.Equal("response", payload.GetProperty("object").GetString()); + Assert.Contains("output", payload.EnumerateObject().Select(p => p.Name)); + } + + [Fact] + public async Task MultiTurn_ContinuesSessionAcrossTurnsAsync() + { + // Arrange + Assert.SkipWhen(string.IsNullOrEmpty(ApiKey), "OPENAI_API_KEY is not configured; skipping live hosting test."); + AIAgent agent = CreateAgent(); + var state = new HostedAgentState(agent); + + // Act: first turn establishes context, second turn continues from the first response id. + string firstResponseId = await RunTurnAsync(agent, state, """{ "input": "Remember the number 7." }"""); + JsonElement secondBody = ParseBody($$"""{ "input": "What number did I ask you to remember?", "previous_response_id": "{{firstResponseId}}" }"""); + string secondSessionId = OpenAIResponses.GetSessionId(secondBody)!; + OpenAIResponsesRunRequest secondRun = OpenAIResponses.ToAgentRunRequest(secondBody); + AgentSession session = await state.GetOrCreateSessionAsync(secondSessionId); + AgentResponse secondResult = await agent.RunAsync(secondRun.Messages, session, secondRun.Options); + + // Assert: continuation succeeded and the model produced a textual answer. + Assert.Equal(secondSessionId, firstResponseId); + Assert.False(string.IsNullOrWhiteSpace(secondResult.Text)); + } + + private static async Task RunTurnAsync(AIAgent agent, HostedAgentState state, string bodyJson) + { + JsonElement body = ParseBody(bodyJson); + string sessionId = OpenAIResponses.GetSessionId(body) ?? OpenAIResponses.CreateResponseId(); + OpenAIResponsesRunRequest run = OpenAIResponses.ToAgentRunRequest(body); + AgentSession session = await state.GetOrCreateSessionAsync(sessionId); + string responseId = OpenAIResponses.CreateResponseId(); + _ = await agent.RunAsync(run.Messages, session, run.Options); + await state.SaveSessionAsync(responseId, session); + return responseId; + } + + private static ChatClientAgent CreateAgent() => + new( + new OpenAIClient(ApiKey).GetChatClient(ModelName).AsIChatClient(), + instructions: "You are a concise assistant.", + name: "assistant"); + + private static JsonElement ParseBody(string json) + { + using JsonDocument doc = JsonDocument.Parse(json); + return doc.RootElement.Clone(); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIResponsesHostingTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIResponsesHostingTests.cs new file mode 100644 index 00000000000..2712127daa3 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIResponsesHostingTests.cs @@ -0,0 +1,254 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Hosting.OpenAI.UnitTests; +using Microsoft.Agents.AI.Workflows; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; + +namespace Microsoft.Agents.AI.Hosting.OpenAI.Tests; + +/// +/// In-process, in-memory hosting tests for : a request travels through an +/// app-owned ASP.NET Core route (hosted in-memory via ) that wires +/// plus / , +/// exactly like the HostingResponsesAgent / HostingResponsesWorkflow samples. These run fully +/// in-process against a deterministic mock chat client — there is no external server process and no live +/// model. Live-model coverage lives in the separate Microsoft.Agents.AI.Hosting.OpenAI.IntegrationTests +/// project. This mirrors the Python test_http_round_trip.py coverage. +/// +public sealed class OpenAIResponsesHostingTests : IAsyncDisposable +{ + private WebApplication? _app; + private HttpClient? _client; + + [Fact] + public async Task AgentRoute_NonStreaming_ReturnsResponsesShapedJsonAsync() + { + // Arrange + HttpClient client = await this.StartAgentHostAsync(new TestHelpers.ConversationMemoryMockChatClient("Hello from the agent")); + + // Act + HttpResponseMessage response = await client.PostAsync( + new Uri("/responses", UriKind.Relative), + JsonContent("""{ "input": "Hi there" }""")); + + // Assert + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + using JsonDocument doc = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); + JsonElement root = doc.RootElement; + Assert.StartsWith("resp_", root.GetProperty("id").GetString()); + Assert.Equal("response", root.GetProperty("object").GetString()); + Assert.Contains("Hello from the agent", root.GetRawText()); + } + + [Fact] + public async Task AgentRoute_Streaming_ReturnsServerSentEventsAsync() + { + // Arrange + HttpClient client = await this.StartAgentHostAsync(new TestHelpers.ConversationMemoryMockChatClient("Streamed answer")); + + // Act + HttpResponseMessage response = await client.PostAsync( + new Uri("/responses", UriKind.Relative), + JsonContent("""{ "input": "Hi there", "stream": true }""")); + + // Assert + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + Assert.Equal("text/event-stream", response.Content.Headers.ContentType?.MediaType); + string body = await response.Content.ReadAsStringAsync(); + Assert.Contains("event: response.created", body, StringComparison.Ordinal); + Assert.Contains("event: response.completed", body, StringComparison.Ordinal); + Assert.Contains("Streamed answer", body, StringComparison.Ordinal); + } + + [Fact] + public async Task AgentRoute_MultiTurn_ReusesSessionAcrossTurnsAsync() + { + // Arrange: a recording mock so we can prove the second turn saw the first turn's history. + var recorder = new TestHelpers.ConversationMemoryMockChatClient("ok"); + HttpClient client = await this.StartAgentHostAsync(recorder); + + // Act: first turn, then continue using the returned response id as previous_response_id. + using JsonDocument first = JsonDocument.Parse(await (await client.PostAsync( + new Uri("/responses", UriKind.Relative), JsonContent("""{ "input": "first turn" }"""))).Content.ReadAsStringAsync()); + string responseId = first.RootElement.GetProperty("id").GetString()!; + + HttpResponseMessage secondResponse = await client.PostAsync( + new Uri("/responses", UriKind.Relative), + JsonContent($$"""{ "input": "second turn", "previous_response_id": "{{responseId}}" }""")); + + // Assert + Assert.Equal(HttpStatusCode.OK, secondResponse.StatusCode); + Assert.Equal(2, recorder.CallHistory.Count); + + // The second call must include the first turn's user message (continuity through HostedAgentState), + // whereas the first call must not contain the second turn's text. + string firstCallText = string.Join("\n", recorder.CallHistory[0].Select(m => m.Text)); + string secondCallText = string.Join("\n", recorder.CallHistory[1].Select(m => m.Text)); + Assert.Contains("first turn", secondCallText, StringComparison.Ordinal); + Assert.DoesNotContain("second turn", firstCallText, StringComparison.Ordinal); + } + + [Fact] + public async Task AgentRoute_MalformedBody_ReturnsBadRequestAsync() + { + // Arrange + HttpClient client = await this.StartAgentHostAsync(new TestHelpers.ConversationMemoryMockChatClient("ok")); + + // Act (missing the required "input" field) + HttpResponseMessage response = await client.PostAsync( + new Uri("/responses", UriKind.Relative), + JsonContent("""{ "model": "x" }""")); + + // Assert + Assert.Equal(HttpStatusCode.BadRequest, response.StatusCode); + } + + [Fact] + public async Task WorkflowRoute_RunThenResume_AdvancesCheckpointAcrossTurnsAsync() + { + // Arrange: a two-agent sequential workflow behind an app-owned route with checkpoint resume keyed by + // the stable conversation id. + HttpClient client = await this.StartWorkflowHostAsync(); + const string ConversationId = "conv_it_1"; + + // Act: first turn runs the workflow forward; second turn (same conversation) resumes from the checkpoint. + HttpResponseMessage firstResponse = await client.PostAsync( + new Uri("/responses", UriKind.Relative), + JsonContent($$"""{ "input": "draft this", "conversation": "{{ConversationId}}" }""")); + HttpResponseMessage secondResponse = await client.PostAsync( + new Uri("/responses", UriKind.Relative), + JsonContent($$"""{ "input": "and again", "conversation": "{{ConversationId}}" }""")); + + // Assert: both turns succeed and produce Responses-shaped payloads. + Assert.Equal(HttpStatusCode.OK, firstResponse.StatusCode); + Assert.Equal(HttpStatusCode.OK, secondResponse.StatusCode); + using JsonDocument doc = JsonDocument.Parse(await secondResponse.Content.ReadAsStringAsync()); + Assert.StartsWith("resp_", doc.RootElement.GetProperty("id").GetString()); + Assert.Equal(ConversationId, doc.RootElement.GetProperty("conversation").GetProperty("id").GetString()); + } + + private async Task StartAgentHostAsync(IChatClient chatClient) + { + WebApplicationBuilder builder = WebApplication.CreateBuilder(); + builder.WebHost.UseTestServer(); + + this._app = builder.Build(); + + var agent = new ChatClientAgent(chatClient, instructions: "You are a helpful assistant.", name: "assistant"); + var state = new HostedAgentState(agent); + + this._app.MapPost("/responses", async (JsonElement body, HttpContext http, CancellationToken ct) => + { + OpenAIResponsesRunRequest run; + try + { + run = OpenAIResponses.ToAgentRunRequest(body); + } + catch (ArgumentException) + { + return Results.BadRequest(); + } + + string sessionId = OpenAIResponses.GetSessionId(body) ?? OpenAIResponses.CreateResponseId(); + AgentSession session = await state.GetOrCreateSessionAsync(sessionId, ct); + string responseId = OpenAIResponses.CreateResponseId(); + + if (body.TryGetProperty("stream", out JsonElement s) && s.ValueKind == JsonValueKind.True) + { + http.Response.ContentType = "text/event-stream"; + var updates = agent.RunStreamingAsync(run.Messages, session, run.Options, ct); + await foreach (string frame in OpenAIResponses.WriteResponseStreamAsync(updates, responseId, responseId, ct)) + { + await http.Response.WriteAsync(frame, ct); + } + + await state.SaveSessionAsync(responseId, session, ct); + return Results.Empty; + } + + AgentResponse result = await agent.RunAsync(run.Messages, session, run.Options, ct); + await state.SaveSessionAsync(responseId, session, ct); + return Results.Json(OpenAIResponses.WriteResponse(result, responseId, responseId)); + }); + + await this._app.StartAsync(); + this._client = this.ResolveTestClient(); + return this._client; + } + + private async Task StartWorkflowHostAsync() + { + WebApplicationBuilder builder = WebApplication.CreateBuilder(); + builder.WebHost.UseTestServer(); + + this._app = builder.Build(); + + AIAgent writer = new ChatClientAgent(new TestHelpers.ConversationMemoryMockChatClient("draft"), name: "Writer"); + AIAgent reviewer = new ChatClientAgent(new TestHelpers.ConversationMemoryMockChatClient("final"), name: "Reviewer"); + Workflow workflow = AgentWorkflowBuilder.BuildSequential(workflowName: "WriteAndReview", agents: [writer, reviewer]); + var state = new HostedWorkflowState(workflow); + + this._app.MapPost("/responses", async (JsonElement body, CancellationToken ct) => + { + OpenAIResponsesRunRequest run = OpenAIResponses.ToAgentRunRequest(body); + string sessionId = GetConversationId(body) ?? OpenAIResponses.CreateResponseId(); + + HostedWorkflowRunResult result = await state.RunOrResumeAsync(sessionId, run.Messages.ToList(), ct); + + var response = new AgentResponse(new ChatMessage(ChatRole.Assistant, $"{result.Events.Count} event(s)")); + return Results.Json(OpenAIResponses.WriteResponse(response, OpenAIResponses.CreateResponseId(), sessionId)); + }); + + await this._app.StartAsync(); + this._client = this.ResolveTestClient(); + return this._client; + } + + private HttpClient ResolveTestClient() + { + TestServer server = this._app!.Services.GetRequiredService() as TestServer + ?? throw new InvalidOperationException("TestServer not found"); + return server.CreateClient(); + } + + private static StringContent JsonContent(string json) => new(json, Encoding.UTF8, "application/json"); + + private static string? GetConversationId(JsonElement body) + { + if (!body.TryGetProperty("conversation", out JsonElement conversation)) + { + return null; + } + + return conversation.ValueKind switch + { + JsonValueKind.String => conversation.GetString(), + JsonValueKind.Object when conversation.TryGetProperty("id", out JsonElement id) => id.GetString(), + _ => null, + }; + } + + public async ValueTask DisposeAsync() + { + this._client?.Dispose(); + if (this._app is not null) + { + await this._app.DisposeAsync(); + } + + GC.SuppressFinalize(this); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIResponsesTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIResponsesTests.cs new file mode 100644 index 00000000000..36885af8f2e --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIResponsesTests.cs @@ -0,0 +1,105 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Text.Json; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Hosting.OpenAI.UnitTests; + +/// +/// Unit tests for the helper facade. +/// +public class OpenAIResponsesTests +{ + [Fact] + public void ToAgentRunRequest_StringInput_ProducesUserMessage() + { + // Arrange + using var doc = JsonDocument.Parse("""{ "input": "Hello there" }"""); + + // Act + var request = OpenAIResponses.ToAgentRunRequest(doc.RootElement); + + // Assert + var message = Assert.Single(request.Messages); + Assert.Equal(ChatRole.User, message.Role); + Assert.Equal("Hello there", message.Text); + Assert.Null(request.Options); + } + + [Fact] + public void GetSessionId_PreviousResponseId_IsReturned() + { + // Arrange + using var doc = JsonDocument.Parse("""{ "input": "hi", "previous_response_id": "resp_abc" }"""); + + // Act & Assert + Assert.Equal("resp_abc", OpenAIResponses.GetSessionId(doc.RootElement)); + } + + [Fact] + public void GetSessionId_ConversationId_IsReturned() + { + // Arrange + using var doc = JsonDocument.Parse("""{ "input": "hi", "conversation": "conv_xyz" }"""); + + // Act & Assert + Assert.Equal("conv_xyz", OpenAIResponses.GetSessionId(doc.RootElement)); + } + + [Fact] + public void GetSessionId_NoContinuationKey_ReturnsNull() + { + // Arrange + using var doc = JsonDocument.Parse("""{ "input": "hi" }"""); + + // Act & Assert + Assert.Null(OpenAIResponses.GetSessionId(doc.RootElement)); + } + + [Fact] + public void ToAgentRunRequest_InvalidBody_ThrowsArgumentException() + { + // Arrange (missing the required "input" field) + using var doc = JsonDocument.Parse("""{ "model": "x" }"""); + + // Act & Assert + Assert.Throws("body", () => OpenAIResponses.ToAgentRunRequest(doc.RootElement)); + } + + [Fact] + public void GetSessionId_MalformedBody_ReturnsNull() + { + // Arrange (an array is not a valid Responses body) + using var doc = JsonDocument.Parse("""[ 1, 2, 3 ]"""); + + // Act & Assert + Assert.Null(OpenAIResponses.GetSessionId(doc.RootElement)); + } + + [Fact] + public void CreateResponseId_HasResponsePrefix() + { + // Act + string id = OpenAIResponses.CreateResponseId(); + + // Assert + Assert.StartsWith("resp_", id); + } + + [Fact] + public void WriteResponse_RendersIdAndOutputText() + { + // Arrange + var response = new AgentResponse(new ChatMessage(ChatRole.Assistant, "Hello from the agent")); + const string ResponseId = "resp_test123"; + + // Act + JsonElement payload = OpenAIResponses.WriteResponse(response, ResponseId, sessionId: "conv_1"); + + // Assert + Assert.Equal(ResponseId, payload.GetProperty("id").GetString()); + Assert.Equal("conv_1", payload.GetProperty("conversation").GetProperty("id").GetString()); + Assert.Contains("Hello from the agent", payload.GetRawText()); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/ApprovalGateWorkflow.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/ApprovalGateWorkflow.cs new file mode 100644 index 00000000000..9348bd3d18c --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/ApprovalGateWorkflow.cs @@ -0,0 +1,52 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Builds a minimal human-in-the-loop workflow whose start executor forwards its input to a +/// , so the workflow emits a and halts awaiting an +/// external response. Used to verify that resuming such a workflow does not block indefinitely. +/// +internal static class ApprovalGateWorkflow +{ + internal const string RequestPortId = "approval"; + + internal static Workflow Build() + { + var gate = new ApprovalGateExecutor("gate", RequestPortId); + RequestPort port = RequestPort.Create(RequestPortId); + var finalize = new FinalizeExecutor("finalize"); + + return new WorkflowBuilder(gate) + .AddEdge(gate, port) + .AddEdge(port, finalize) + .WithOutputFrom(finalize) + .Build(); + } + + internal sealed record ApprovalRequest(string Prompt); + + private sealed class ApprovalGateExecutor(string id, string requestPortId) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .SendsMessage(); + + private ValueTask HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default) + => context.SendMessageAsync(new ApprovalRequest(input), requestPortId, cancellationToken); + } + + private sealed class FinalizeExecutor(string id) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .YieldsOutput(); + + private ValueTask HandleAsync(string approval, IWorkflowContext context, CancellationToken cancellationToken = default) + => context.YieldOutputAsync($"approved:{approval}", cancellationToken); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/BriefWorkflow.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/BriefWorkflow.cs new file mode 100644 index 00000000000..24ecb003d40 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/BriefWorkflow.cs @@ -0,0 +1,36 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Builds a workflow whose start executor accepts a typed (not +/// List<ChatMessage>) and yields a formatted string. Used to demonstrate that an application can +/// adapt Responses input into a workflow's native start-executor input via the generic +/// RunOrResumeAsync<TInput> (the .NET counterpart of Python's ResponsesChannel run hook). +/// +internal static class BriefWorkflow +{ + internal sealed record WriterBrief(string Topic, string Style); + + internal static Workflow Build() + { + var writer = new BriefExecutor("brief"); + return new WorkflowBuilder(writer) + .WithOutputFrom(writer) + .Build(); + } + + private sealed class BriefExecutor(string id) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .YieldsOutput(); + + private ValueTask HandleAsync(WriterBrief brief, IWorkflowContext context, CancellationToken cancellationToken = default) + => context.YieldOutputAsync($"[{brief.Style}] {brief.Topic}", cancellationToken); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/CapturingLoggerFactory.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/CapturingLoggerFactory.cs new file mode 100644 index 00000000000..1e7b306bd1d --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/CapturingLoggerFactory.cs @@ -0,0 +1,38 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// A minimal that captures every log entry it produces, for asserting on +/// diagnostics emitted by the system under test. +/// +internal sealed class CapturingLoggerFactory : ILoggerFactory +{ + public List<(LogLevel Level, string Message)> Entries { get; } = []; + + public ILogger CreateLogger(string categoryName) => new CapturingLogger(this.Entries); + + public void AddProvider(ILoggerProvider provider) + { + // No-op: this factory always produces capturing loggers. + } + + public void Dispose() + { + // No-op. + } + + private sealed class CapturingLogger(List<(LogLevel Level, string Message)> entries) : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + => entries.Add((logLevel, formatter(state, exception))); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/CountingWorkflow.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/CountingWorkflow.cs new file mode 100644 index 00000000000..c80b8b843e3 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/CountingWorkflow.cs @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Builds a non-chat-protocol workflow whose single executor keeps a running count in workflow state and +/// yields count:N. Because the count is checkpointed, a genuine resume observes the accumulated value +/// (for example count:2 on the second turn), whereas a fresh run restarts at count:1. Used to +/// prove that a turn actually resumed from a prior checkpoint rather than starting over. +/// +internal static class CountingWorkflow +{ + internal static Workflow Build() + { + var counter = new CountingExecutor("counter"); + return new WorkflowBuilder(counter) + .WithOutputFrom(counter) + .Build(); + } + + private sealed class CountingExecutor(string id) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .YieldsOutput(); + + private async ValueTask HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default) + { + int count = await context.ReadOrInitStateAsync("count", () => 0, cancellationToken).ConfigureAwait(false); + count++; + await context.QueueStateUpdateAsync("count", count, cancellationToken: cancellationToken).ConfigureAwait(false); + await context.YieldOutputAsync($"count:{count}", cancellationToken).ConfigureAwait(false); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/FanOutRequestWorkflow.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/FanOutRequestWorkflow.cs new file mode 100644 index 00000000000..dce4af95bc9 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/FanOutRequestWorkflow.cs @@ -0,0 +1,59 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Builds a workflow whose start executor, in a single superstep, both emits an external request (to a +/// ) and queues a message to a downstream executor that yields output. Used to +/// verify that resuming does not truncate the turn at the request-bearing superstep before the downstream +/// executor runs. +/// +internal static class FanOutRequestWorkflow +{ + internal const string RequestPortId = "approval"; + internal const string DownstreamPrefix = "downstream:"; + + internal static Workflow Build() + { + var start = new FanOutExecutor("start", RequestPortId, "downstream"); + RequestPort port = RequestPort.Create(RequestPortId); + var downstream = new DownstreamExecutor("downstream"); + + return new WorkflowBuilder(start) + .AddEdge(start, port) + .AddEdge(start, downstream) + .WithOutputFrom(downstream) + .Build(); + } + + internal sealed record ApprovalRequest(string Prompt); + + private sealed class FanOutExecutor(string id, string requestPortId, string downstreamId) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .SendsMessage() + .SendsMessage(); + + private async ValueTask HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Same superstep: emit an external request AND queue downstream work. + await context.SendMessageAsync(new ApprovalRequest(input), requestPortId, cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(input, downstreamId, cancellationToken).ConfigureAwait(false); + } + } + + private sealed class DownstreamExecutor(string id) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .YieldsOutput(); + + private ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + => context.YieldOutputAsync($"{DownstreamPrefix}{message}", cancellationToken); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/GatedCountingWorkflow.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/GatedCountingWorkflow.cs new file mode 100644 index 00000000000..2b5c8802c64 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/GatedCountingWorkflow.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Builds a non-chat-protocol workflow whose executor signals when it starts and then blocks on a +/// test-controlled gate before finishing. This lets a test hold a turn "inside" the workflow and observe +/// whether a second, concurrent turn for the same holder is allowed to run at the same time. +/// +internal static class GatedCountingWorkflow +{ + internal static Workflow Build(SemaphoreSlim entered, Task release) + { + var gated = new GatedExecutor("gated", entered, release); + return new WorkflowBuilder(gated) + .WithOutputFrom(gated) + .Build(); + } + + private sealed class GatedExecutor(string id, SemaphoreSlim entered, Task release) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .YieldsOutput(); + + private async ValueTask HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Signal that this turn has started running inside the workflow, then wait for the test to release. + entered.Release(); + await release.ConfigureAwait(false); + + int count = await context.ReadOrInitStateAsync("count", () => 0, cancellationToken).ConfigureAwait(false); + count++; + await context.QueueStateUpdateAsync("count", count, cancellationToken: cancellationToken).ConfigureAwait(false); + await context.YieldOutputAsync($"count:{count}", cancellationToken).ConfigureAwait(false); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/HostedAgentStateTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/HostedAgentStateTests.cs new file mode 100644 index 00000000000..2adccb2a145 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/HostedAgentStateTests.cs @@ -0,0 +1,137 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Threading; +using System.Threading.Tasks; +using Moq; +using Moq.Protected; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Unit tests for the class. +/// +public class HostedAgentStateTests +{ + private readonly Mock _agentMock = new(); + private readonly Mock _storeMock = new(); + private readonly AgentSession _session = new TestAgentSession(); + + [Fact] + public void Constructor_NullAgent_Throws() => + // Act & Assert + Assert.Throws("agent", () => new HostedAgentState(null!)); + + [Fact] + public async Task Constructor_NullStore_UsesInMemoryStoreAsync() + { + // Arrange: a null store must fall back to an in-memory store, whose create-on-miss path calls the agent. + this._agentMock + .Protected() + .Setup>("CreateSessionCoreAsync", ItExpr.IsAny()) + .Returns(new ValueTask(this._session)); + var state = new HostedAgentState(this._agentMock.Object); + + // Act + var session = await state.GetOrCreateSessionAsync("session-1"); + + // Assert + Assert.Same(this._session, session); + } + + [Fact] + public async Task GetOrCreateSessionAsync_DelegatesToStoreWithAgentAndIdAsync() + { + // Arrange + this._storeMock + .Setup(x => x.GetSessionAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(this._session); + var state = new HostedAgentState(this._agentMock.Object, this._storeMock.Object); + + // Act + var result = await state.GetOrCreateSessionAsync("session-1"); + + // Assert + Assert.Same(this._session, result); + this._storeMock.Verify(x => x.GetSessionAsync(this._agentMock.Object, "session-1", It.IsAny()), Times.Once); + } + + [Fact] + public async Task SaveSessionAsync_DelegatesToStoreWithAgentAndIdAsync() + { + // Arrange + this._storeMock + .Setup(x => x.SaveSessionAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(ValueTask.CompletedTask); + var state = new HostedAgentState(this._agentMock.Object, this._storeMock.Object); + + // Act + await state.SaveSessionAsync("resp-new", this._session); + + // Assert + this._storeMock.Verify(x => x.SaveSessionAsync(this._agentMock.Object, "resp-new", this._session, It.IsAny()), Times.Once); + } + + [Fact] + public async Task DeleteSessionAsync_DelegatesToStoreWithAgentAndIdAsync() + { + // Arrange + this._storeMock + .Setup(x => x.DeleteSessionAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(ValueTask.CompletedTask); + var state = new HostedAgentState(this._agentMock.Object, this._storeMock.Object); + + // Act + await state.DeleteSessionAsync("session-1"); + + // Assert + this._storeMock.Verify(x => x.DeleteSessionAsync(this._agentMock.Object, "session-1", It.IsAny()), Times.Once); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + public async Task GetOrCreateSessionAsync_InvalidId_ThrowsAsync(string? sessionId) + { + // Arrange + var state = new HostedAgentState(this._agentMock.Object, this._storeMock.Object); + + // Act & Assert + await Assert.ThrowsAnyAsync(() => state.GetOrCreateSessionAsync(sessionId!).AsTask()); + } + + [Fact] + public async Task LockSessionAsync_LockingDisabled_ReturnsNoopReleaserAsync() + { + // Arrange + var state = new HostedAgentState(this._agentMock.Object, this._storeMock.Object); + + // Act + var releaser = await state.LockSessionAsync("session-1"); + + // Assert (a second acquire does not block when locking is disabled) + var second = await state.LockSessionAsync("session-1"); + await releaser.DisposeAsync(); + await second.DisposeAsync(); + } + + [Fact] + public async Task LockSessionAsync_LockingEnabled_SerializesSameSessionAsync() + { + // Arrange + var state = new HostedAgentState(this._agentMock.Object, this._storeMock.Object, enableSessionLocking: true); + var first = await state.LockSessionAsync("session-1"); + + // Act: a second acquire for the same session must not complete until the first is released. + var secondTask = state.LockSessionAsync("session-1").AsTask(); + var completedBeforeRelease = secondTask.IsCompleted; + await first.DisposeAsync(); + var second = await secondTask; + await second.DisposeAsync(); + + // Assert + Assert.False(completedBeforeRelease); + } + + private sealed class TestAgentSession : AgentSession; +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/HostedWorkflowStateTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/HostedWorkflowStateTests.cs new file mode 100644 index 00000000000..781888c3fbe --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/HostedWorkflowStateTests.cs @@ -0,0 +1,403 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging; +using Moq; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Unit tests for the class. +/// +public class HostedWorkflowStateTests +{ + [Fact] + public void Constructor_NullWorkflow_Throws() => + // Act & Assert + Assert.Throws("workflow", () => new HostedWorkflowState(null!)); + + [Fact] + public void TryGetCheckpoint_UnknownSession_ReturnsFalse() + { + // Arrange + var state = new HostedWorkflowState(CreateTestWorkflow()); + + // Act + bool found = state.TryGetCheckpoint("unknown", out var checkpoint); + + // Assert + Assert.False(found); + Assert.Null(checkpoint); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + public async Task RunOrResumeAsync_InvalidSessionId_ThrowsAsync(string? sessionId) + { + // Arrange + var state = new HostedWorkflowState(CreateTestWorkflow()); + + // Act & Assert + await Assert.ThrowsAnyAsync(() => state.RunOrResumeAsync(sessionId!, "input").AsTask()); + } + + [Fact] + public async Task RunOrResumeAsync_NullInput_ThrowsAsync() + { + // Arrange + var state = new HostedWorkflowState(CreateTestWorkflow()); + + // Act & Assert + await Assert.ThrowsAsync("input", () => state.RunOrResumeAsync("s1", null!).AsTask()); + } + + [Fact] + public async Task RunOrResumeAsync_FirstTurn_RunsAndRecordsCheckpointAsync() + { + // Arrange + var state = new HostedWorkflowState(CreateEchoWorkflow()); + + // Act + HostedWorkflowRunResult result = await state.RunOrResumeAsync("s1", InputMessages("hello")); + + // Assert + Assert.NotEmpty(result.Events); + Assert.NotNull(result.Checkpoint); + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? checkpoint)); + Assert.Same(result.Checkpoint, checkpoint); + Assert.Contains("hello", OutputText(result)); + } + + [Fact] + public async Task RunOrResumeAsync_SecondTurn_ResumesWithNewInputAndCompletesAsync() + { + // Arrange + var state = new HostedWorkflowState(CreateEchoWorkflow()); + HostedWorkflowRunResult first = await state.RunOrResumeAsync("s1", InputMessages("hello")); + CheckpointInfo? firstCheckpoint = first.Checkpoint; + + // Act: the second turn must restore the checkpoint and run forward with the NEW input. + // A regression here (resuming with no input) would hang, so guard with a timeout. + HostedWorkflowRunResult second = await state.RunOrResumeAsync("s1", InputMessages("world")) + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: the resumed turn processed the new input and advanced the checkpoint. + Assert.NotEmpty(second.Events); + Assert.Contains("world", OutputText(second)); + Assert.NotNull(second.Checkpoint); + Assert.NotSame(firstCheckpoint, second.Checkpoint); + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? head)); + Assert.Same(second.Checkpoint, head); + } + + [Fact] + public async Task RunOrResumeAsync_ResumeWithPendingRequest_DoesNotBlockAsync() + { + // Arrange: a human-in-the-loop workflow whose start executor forwards its input to a request port, + // so the workflow emits a RequestInfoEvent and halts awaiting an external response. + var state = new HostedWorkflowState(ApprovalGateWorkflow.Build()); + + // First turn halts at the pending request (the non-blocking baseline). + HostedWorkflowRunResult first = await state.RunOrResumeAsync("s1", "approve deploy") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Contains(first.Events, e => e is RequestInfoEvent); + Assert.NotNull(first.Checkpoint); + + // Act: resuming a workflow that halts at a pending request must also return instead of blocking + // forever. A regression (blocking drain) hangs here, so guard with a timeout. + HostedWorkflowRunResult second = await state.RunOrResumeAsync("s1", "approve deploy again") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: the resumed turn surfaced the pending request and returned. + Assert.Contains(second.Events, e => e is RequestInfoEvent); + } + + [Fact] + public async Task RunOrResumeAsync_ResumeMakesNoProgress_LogsWarningAsync() + { + // Arrange: a non-chat-protocol workflow that completes on the first turn. + var loggerFactory = new CapturingLoggerFactory(); + var state = new HostedWorkflowState(StringEchoWorkflow.Build(), loggerFactory: loggerFactory); + HostedWorkflowRunResult first = await state.RunOrResumeAsync("s1", "hello"); + Assert.NotEmpty(first.Events); + Assert.NotNull(first.Checkpoint); + + // Act: resume with an input the start executor cannot handle, so the turn drives no work. + HostedWorkflowRunResult second = await state.RunOrResumeAsync("s1", 42); + + // Assert: a resume that produced no events is surfaced as a warning (possible stale checkpoint / + // mismatched input), mirroring the Python host's zero-event restore warning. + Assert.Empty(second.Events); + Assert.Contains(loggerFactory.Entries, e => e.Level == LogLevel.Warning); + } + + [Fact] + public async Task RunOrResumeAsync_CursorMiss_ResumesFromManagerLatestCheckpointAsync() + { + // Arrange: a shared checkpoint manager stands in for durable storage that outlives the in-memory + // cursor. The first holder runs one turn; a counting workflow records count:1 in the checkpoint. + var manager = CheckpointManager.CreateInMemory(); + var first = new HostedWorkflowState(CountingWorkflow.Build(), manager); + HostedWorkflowRunResult firstResult = await first.RunOrResumeAsync("s1", "go"); + Assert.Contains("count:1", StringOutput(firstResult)); + + // Act: a NEW holder over the SAME manager (fresh cursor, e.g. after a process restart) runs the + // session again. With durable read-through it resumes from the manager's latest checkpoint. + var second = new HostedWorkflowState(CountingWorkflow.Build(), manager); + HostedWorkflowRunResult resumed = await second.RunOrResumeAsync("s1", "go") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: the count advanced to 2, proving it resumed from the prior checkpoint rather than + // restarting from scratch (which would yield count:1 again). + Assert.Contains("count:2", StringOutput(resumed)); + Assert.True(second.TryGetCheckpoint("s1", out _)); + } + + [Fact] + public async Task RunOrResumeAsync_ConcurrentSameSessionTurns_AreSerializedAsync() + { + // Arrange: a workflow that signals when a turn enters and then blocks on a gate, so the test can + // hold the first turn "inside" the workflow while it starts a second turn for the same session. + using var entered = new SemaphoreSlim(0, 2); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var state = new HostedWorkflowState(GatedCountingWorkflow.Build(entered, release.Task), CheckpointManager.CreateInMemory()); + + // Act: start the first turn and wait until it is running inside the workflow. + Task first = state.RunOrResumeAsync("s1", "go").AsTask(); + Assert.True(await entered.WaitAsync(TimeSpan.FromSeconds(10)), "the first turn should enter the workflow"); + + // Start a second same-session turn while the first is still inside the workflow. + Task second = state.RunOrResumeAsync("s1", "go").AsTask(); + + // Assert: the second turn must WAIT on the holder lock, not fault. Without the lock it would reach the + // engine's concurrent-run ownership guard and fault (completing the task); the lock instead leaves it + // pending until the first turn releases. Checking the task is not completed isolates the holder lock + // from the engine guard, and the entered gate confirms it did not run concurrently. + await Task.Delay(TimeSpan.FromMilliseconds(500)); + Assert.False(second.IsCompleted, "the second turn must wait on the holder lock rather than fault or run concurrently"); + Assert.False(await entered.WaitAsync(TimeSpan.FromMilliseconds(200)), "the second turn must not enter the workflow while the first holds it"); + + // Release both turns and let them run to completion. + release.SetResult(); + HostedWorkflowRunResult[] results = await Task.WhenAll(first, second).WaitAsync(TimeSpan.FromSeconds(30)); + + // Both turns completed successfully (proving the lock serialized rather than faulted them), advancing + // the count from 1 to 2. + string combined = string.Concat(results.Select(StringOutput)); + Assert.Contains("count:1", combined); + Assert.Contains("count:2", combined); + } + + [Fact] + public async Task RunOrResumeAsync_NonChatWorkflow_ResumesWithNewInputAsync() + { + // Arrange: a non-chat-protocol workflow (string start executor), so the resume path sends the input + // without a TurnToken. + var state = new HostedWorkflowState(CountingWorkflow.Build()); + HostedWorkflowRunResult first = await state.RunOrResumeAsync("s1", "go"); + Assert.Contains("count:1", StringOutput(first)); + + // Act + HostedWorkflowRunResult second = await state.RunOrResumeAsync("s1", "go") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: the non-chat resume carried state and advanced the checkpoint. + Assert.Contains("count:2", StringOutput(second)); + Assert.NotNull(second.Checkpoint); + Assert.NotSame(first.Checkpoint, second.Checkpoint); + } + + [Fact] + public async Task RunOrResumeAsync_ThirdTurn_KeepsAdvancingCheckpointAsync() + { + // Arrange + var state = new HostedWorkflowState(CreateEchoWorkflow()); + + // Act: three turns on the same session. + HostedWorkflowRunResult r1 = await state.RunOrResumeAsync("s1", InputMessages("a")); + HostedWorkflowRunResult r2 = await state.RunOrResumeAsync("s1", InputMessages("b")) + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + HostedWorkflowRunResult r3 = await state.RunOrResumeAsync("s1", InputMessages("c")) + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: the cursor keeps advancing past the second turn, and the head reflects the latest turn. + Assert.Contains("c", OutputText(r3)); + Assert.NotNull(r1.Checkpoint); + Assert.NotNull(r3.Checkpoint); + Assert.NotSame(r1.Checkpoint, r2.Checkpoint); + Assert.NotSame(r2.Checkpoint, r3.Checkpoint); + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? head)); + Assert.Same(r3.Checkpoint, head); + } + + [Fact] + public async Task RunOrResumeStreamingAsync_StreamsEventsAndResumesAsync() + { + // Arrange + var state = new HostedWorkflowState(CreateEchoWorkflow()); + + // Act: first turn streamed. + List firstEvents = []; + await foreach (WorkflowEvent evt in state.RunOrResumeStreamingAsync("s1", InputMessages("hello"))) + { + firstEvents.Add(evt); + } + + // Assert: events streamed and the checkpoint was recorded after the stream completed. + Assert.NotEmpty(firstEvents); + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? firstCheckpoint)); + Assert.NotNull(firstCheckpoint); + + // Act: second turn streamed via the resume path with new input. + List secondEvents = []; + await foreach (WorkflowEvent evt in state.RunOrResumeStreamingAsync("s1", InputMessages("world"))) + { + secondEvents.Add(evt); + } + + // Assert: the resumed stream processed the new input and advanced the checkpoint. + string output = string.Concat( + secondEvents + .OfType() + .Select(e => e.Data) + .OfType>() + .SelectMany(messages => messages) + .Select(m => m.Text)); + Assert.Contains("world", output); + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? secondCheckpoint)); + Assert.NotSame(firstCheckpoint, secondCheckpoint); + } + + [Fact] + public async Task RunOrResumeAsync_AdaptsResponsesInputToTypedStartExecutorAsync() + { + // Arrange: a workflow whose start executor takes a typed WriterBrief rather than List. + // The application adapts the Responses input into that type before calling RunOrResumeAsync — the + // generic TInput is the .NET counterpart of Python's ResponsesChannel run hook. + var state = new HostedWorkflowState(BriefWorkflow.Build()); + + // Simulate parsing a structured Responses text payload into the start executor's input type. + const string ResponsesText = "{\"topic\":\"electric SUV\",\"style\":\"playful\"}"; + using JsonDocument doc = JsonDocument.Parse(ResponsesText); + var brief = new BriefWorkflow.WriterBrief( + doc.RootElement.GetProperty("topic").GetString()!, + doc.RootElement.GetProperty("style").GetString()!); + + // Act + HostedWorkflowRunResult result = await state.RunOrResumeAsync("s1", brief); + + // Assert: the adapted input drove the typed start executor. + Assert.Contains("[playful] electric SUV", StringOutput(result)); + } + + [Fact] + public async Task RunOrResumeAsync_ResumeWithRejectedInput_DoesNotHangAsync() + { + // Arrange: a non-chat human-in-the-loop workflow whose first turn emits a request and halts. + var state = new HostedWorkflowState(ApprovalGateWorkflow.Build()); + HostedWorkflowRunResult first = await state.RunOrResumeAsync("s1", "approve") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Contains(first.Events, e => e is RequestInfoEvent); + + // Act: resume with an input the start executor cannot handle (wrong type), so no superstep runs. + // A drain that blocks on the restored pending request would hang here; guard with a timeout. + HostedWorkflowRunResult second = await state.RunOrResumeAsync("s1", 42) + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: it returned (surfacing the restored pending request) rather than blocking indefinitely. + Assert.Contains(second.Events, e => e is RequestInfoEvent); + } + + [Fact] + public async Task RunOrResumeAsync_ResumeSuperstepWithRequestAndDownstream_DoesNotTruncateAsync() + { + // Arrange: a workflow whose start executor, in one superstep, emits a request AND queues a message to + // a downstream executor that yields output. The first turn establishes a checkpoint. + var state = new HostedWorkflowState(FanOutRequestWorkflow.Build()); + HostedWorkflowRunResult first = await state.RunOrResumeAsync("s1", "one") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Contains(first.Events, e => e is RequestInfoEvent); + + // Act: resume with new input, which again fans out to the request port and the downstream executor. + HostedWorkflowRunResult second = await state.RunOrResumeAsync("s1", "two") + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(30)); + + // Assert: the resumed turn drained past the request-bearing superstep so the downstream output is + // present (a drain that broke at the request would truncate it). + Assert.Contains(second.Events, e => e is RequestInfoEvent); + Assert.Contains(FanOutRequestWorkflow.DownstreamPrefix, StringOutput(second)); + } + + [Fact] + public async Task RunOrResumeStreamingAsync_AbandonedAfterCheckpoint_AdvancesCursorAsync() + { + // Arrange + var state = new HostedWorkflowState(CreateEchoWorkflow()); + await foreach (WorkflowEvent _ in state.RunOrResumeStreamingAsync("s1", InputMessages("a"))) + { + // Enumerate the first turn to completion so the cursor holds its head checkpoint. + } + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? cp1)); + + // Act: abandon the second turn after a superstep has committed a checkpoint. + await foreach (WorkflowEvent evt in state.RunOrResumeStreamingAsync("s1", InputMessages("b"))) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: not null }) + { + break; + } + } + + // Assert: the abandoned turn still advanced the cursor to the last committed checkpoint, so a later + // turn resumes from there rather than re-running from the previous head. + Assert.True(state.TryGetCheckpoint("s1", out CheckpointInfo? cp2)); + Assert.NotEqual(cp1, cp2); + } + + private static List InputMessages(string text) => [new(ChatRole.User, text)]; + + private static string OutputText(HostedWorkflowRunResult result) => + string.Concat( + result.Events + .OfType() + .Select(e => e.Data) + .OfType>() + .SelectMany(messages => messages) + .Select(m => m.Text)); + + private static string StringOutput(HostedWorkflowRunResult result) => + string.Concat( + result.Events + .OfType() + .Select(e => e.Data) + .OfType()); + + private static Workflow CreateEchoWorkflow() => + AgentWorkflowBuilder.BuildSequential(workflowName: "echo", agents: [new TestEchoAgent(name: "echo")]); + + private static Workflow CreateTestWorkflow() + { + var mockAgent = new Mock(); + mockAgent.Setup(a => a.Name).Returns("testAgent"); + return AgentWorkflowBuilder.BuildSequential(workflowName: "wf", agents: [mockAgent.Object]); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/InMemoryAgentSessionStoreTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/InMemoryAgentSessionStoreTests.cs new file mode 100644 index 00000000000..dff8864fb34 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/InMemoryAgentSessionStoreTests.cs @@ -0,0 +1,85 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Moq; +using Moq.Protected; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Unit tests for across the in-box stores. +/// +public class InMemoryAgentSessionStoreTests +{ + [Fact] + public async Task DeleteSessionAsync_RemovesStoredSession_SoNextGetCreatesAsync() + { + // Arrange + var stored = JsonSerializer.SerializeToElement(new { marker = "stored" }); + var restoredSession = new TestAgentSession(); + var createdSession = new TestAgentSession(); + var agent = new Mock(); + agent.Protected() + .Setup>("SerializeSessionCoreAsync", ItExpr.IsAny(), ItExpr.IsAny(), ItExpr.IsAny()) + .Returns(new ValueTask(stored)); + agent.Protected() + .Setup>("DeserializeSessionCoreAsync", ItExpr.IsAny(), ItExpr.IsAny(), ItExpr.IsAny()) + .Returns(new ValueTask(restoredSession)); + agent.Protected() + .Setup>("CreateSessionCoreAsync", ItExpr.IsAny()) + .Returns(new ValueTask(createdSession)); + + var store = new InMemoryAgentSessionStore(); + + // Act & Assert + await store.SaveSessionAsync(agent.Object, "s1", new TestAgentSession()); + Assert.Same(restoredSession, await store.GetSessionAsync(agent.Object, "s1")); + + await store.DeleteSessionAsync(agent.Object, "s1"); + Assert.Same(createdSession, await store.GetSessionAsync(agent.Object, "s1")); + } + + [Fact] + public async Task DeleteSessionAsync_UnknownId_DoesNotThrowAsync() + { + // Arrange + var store = new InMemoryAgentSessionStore(); + + // Act & Assert (no exception) + await store.DeleteSessionAsync(new Mock().Object, "missing"); + } + + [Fact] + public async Task DeleteSessionAsync_NoopStore_CompletesAsync() + { + // Arrange + var store = new NoopAgentSessionStore(); + + // Act & Assert (no exception) + await store.DeleteSessionAsync(new Mock().Object, "any"); + } + + [Fact] + public async Task DeleteSessionAsync_BaseDefault_ThrowsNotSupportedAsync() + { + // Arrange + AgentSessionStore store = new ConcreteAgentSessionStore(); + + // Act & Assert + await Assert.ThrowsAsync(() => store.DeleteSessionAsync(new Mock().Object, "any").AsTask()); + } + + private sealed class TestAgentSession : AgentSession; + + private sealed class ConcreteAgentSessionStore : AgentSessionStore + { + public override ValueTask SaveSessionAsync(AIAgent agent, string conversationId, AgentSession session, CancellationToken cancellationToken = default) + => default; + + public override ValueTask GetSessionAsync(AIAgent agent, string conversationId, CancellationToken cancellationToken = default) + => new(new TestAgentSession()); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/StringEchoWorkflow.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/StringEchoWorkflow.cs new file mode 100644 index 00000000000..7d7436ac069 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/StringEchoWorkflow.cs @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Builds a minimal non-chat-protocol workflow whose single executor accepts a and yields +/// it back as output. Used to exercise the resume path for workflows whose start executor does not accept +/// List<ChatMessage> (so no is sent). +/// +internal static class StringEchoWorkflow +{ + internal static Workflow Build() + { + var echo = new StringEchoExecutor("echo"); + return new WorkflowBuilder(echo) + .WithOutputFrom(echo) + .Build(); + } + + private sealed class StringEchoExecutor(string id) : Executor(id) + { + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(this.HandleAsync)) + .YieldsOutput(); + + private ValueTask HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default) + => context.YieldOutputAsync($"echo:{input}", cancellationToken); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/TestEchoAgent.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/TestEchoAgent.cs new file mode 100644 index 00000000000..4f19997bcf3 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.UnitTests/TestEchoAgent.cs @@ -0,0 +1,113 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Text.Json; +using System.Text.Json.Serialization; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Hosting.UnitTests; + +/// +/// Deterministic used by workflow-hosting tests: it echoes each user message back as +/// an assistant message (optionally prefixed) and supports session serialization so it can participate in +/// checkpointed workflows. +/// +internal sealed class TestEchoAgent(string? id = null, string? name = null, string? prefix = null) : AIAgent +{ + protected override string? IdCore => id; + public override string? Name => name ?? base.Name; + + public InMemoryChatHistoryProvider ChatHistoryProvider { get; } = new(); + + protected override async ValueTask DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + { + return serializedState.Deserialize(jsonSerializerOptions) ?? await this.CreateSessionAsync(cancellationToken); + } + + protected override ValueTask SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + { + if (session is not EchoAgentSession typedSession) + { + throw new InvalidOperationException($"The provided session type '{session.GetType().Name}' is not compatible with this agent. Only sessions of type '{nameof(EchoAgentSession)}' can be serialized by this agent."); + } + + return new(JsonSerializer.SerializeToElement(typedSession, jsonSerializerOptions)); + } + + protected override ValueTask CreateSessionCoreAsync(CancellationToken cancellationToken = default) => + new(new EchoAgentSession()); + + private ChatMessage UpdateSession(ChatMessage message, AgentSession? session = null) + { + this.ChatHistoryProvider.GetMessages(session).Add(message); + + return message; + } + + private List EchoMessages(IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null) + { + List echoed = []; + foreach (ChatMessage message in messages) + { + this.UpdateSession(message, session); + + if (message.Role == ChatRole.User && !string.IsNullOrEmpty(message.Text)) + { + echoed.Add(this.UpdateSession(new ChatMessage(ChatRole.Assistant, $"{prefix}{message.Text}") + { + AuthorName = this.Name ?? this.Id, + CreatedAt = DateTimeOffset.Now, + MessageId = Guid.NewGuid().ToString("N") + }, session)); + } + } + + return echoed; + } + + protected override Task RunCoreAsync(IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + { + AgentResponse result = + new(this.EchoMessages(messages, session, options).ToList()) + { + AgentId = this.Id, + CreatedAt = DateTimeOffset.Now, + ResponseId = Guid.NewGuid().ToString("N"), + }; + + return Task.FromResult(result); + } + + protected override async IAsyncEnumerable RunCoreStreamingAsync(IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + string responseId = Guid.NewGuid().ToString("N"); + + foreach (ChatMessage message in this.EchoMessages(messages, session, options).ToList()) + { + yield return + new(message.Role, message.Contents) + { + AgentId = this.Id, + AuthorName = message.AuthorName, + ResponseId = responseId, + MessageId = message.MessageId, + CreatedAt = message.CreatedAt + }; + } + + await Task.CompletedTask; + } + + private sealed class EchoAgentSession : AgentSession + { + internal EchoAgentSession() { } + + [JsonConstructor] + internal EchoAgentSession(AgentSessionStateBag stateBag) : base(stateBag) { } + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointManagerLatestTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointManagerLatestTests.cs new file mode 100644 index 00000000000..2bc8b7393e8 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointManagerLatestTests.cs @@ -0,0 +1,44 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Text.Json; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.Checkpointing; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +/// +/// Tests for , which must return the most recently +/// committed checkpoint for a session regardless of the backing store implementation. +/// +public class CheckpointManagerLatestTests +{ + [Fact] + public async Task GetLatestCheckpointAsync_FileStore_ReturnsLastCommittedAsync() + { + // Arrange: commit a chain of checkpoints to a durable file store in a known order. + using TempDirectory dir = new(); + using FileSystemJsonCheckpointStore store = new(dir); + CheckpointManager manager = CheckpointManager.CreateJson(store); + + const string SessionId = "session-latest"; + List committed = []; + CheckpointInfo? parent = null; + for (int i = 0; i < 8; i++) + { + JsonElement value = JsonSerializer.SerializeToElement($"checkpoint-{i}"); + CheckpointInfo info = await store.CreateCheckpointAsync(SessionId, value, parent); + committed.Add(info); + parent = info; + } + + // Act + IEnumerable index = await store.RetrieveIndexAsync(SessionId); + CheckpointInfo? latest = await manager.GetLatestCheckpointAsync(SessionId); + + // Assert: the durable index preserves commit order, so the latest checkpoint is the last committed. + index.Should().Equal(committed, "the file-store index should be returned in commit order"); + latest.Should().Be(committed[^1]); + } +}