Roger Barreto
1f1da1bddb
* .NET: Add OpenAI Responses protocol helpers and optional execution state (ADR-0032)
* Fix netstandard2.0/net472 build; harden helpers and workflow checkpoint key per review
* .NET: Migrate hosting Responses samples to Azure.AI.Projects and fix workflow resume
Migrate HostingResponsesAgent and HostingResponsesWorkflow samples from
Azure.AI.OpenAI to Azure.AI.Projects (AIProjectClient.AsAIAgent), using the
FOUNDRY_PROJECT_ENDPOINT/FOUNDRY_MODEL convention.
Fix HostedWorkflowState.RunOrResumeAsync: on subsequent turns, restore the
session's latest checkpoint and run the workflow forward with the new turn's
input (mirroring the Python hosting host's restore-then-run semantics) instead
of resuming a halted run with no input, which waited on input indefinitely.
Add round-trip resume tests and update ADR-0032/spec-003 wording.
* .NET: Fix HostedWorkflowState resume hang on unserviced external requests
On resume, HostedWorkflowState.RunOrResumeAsync drained the workflow with the
blocking WatchStreamAsync overload, so a workflow that halts at an unserviced
RequestInfoEvent (human-in-the-loop / approval) blocked forever — asymmetric
with the first-turn RunAsync path, which returns at the same halt. Break the
drain when a superstep completes with HasPendingRequests, restoring symmetry
with turn 1. Add a HITL approval-gate workflow and a resume-does-not-block test.
* .NET: Warn when a HostedWorkflowState resume makes no progress
Add an optional ILoggerFactory to HostedWorkflowState and log a warning when a
resumed turn produces no events, mirroring the Python host's zero-event restore
warning (a stale checkpoint or an input that does not match the workflow's
expected type leaves session state unprogressed). Add a non-chat string workflow
helper, a capturing logger, and a red/green test.
* .NET: Resume HostedWorkflowState from durable checkpoint on cursor miss
Add CheckpointManager.GetLatestCheckpointAsync(sessionId) and have
HostedWorkflowState fall back to it when its in-memory head cursor misses, so a
durable CheckpointManager resumes a session across a process restart or a new
holder instead of restarting from the workflow's start executor. Mirrors the
Python host's per-turn get_latest read-through. Add a counting workflow that
proves resume-vs-fresh via accumulated state, plus a red/green test, and update
ADR-0032/spec-003 and the XML remarks.
* .NET: Serialize HostedWorkflowState turns through a workflow lock
A single workflow instance backs the holder and workflow instances do not
support concurrent runs (the runner throws "already owned by another runner"),
so concurrent turns could fault or race the head cursor. Serialize all turns
through one SemaphoreSlim (mirroring the Python host's workflow lock) and make
HostedWorkflowState IDisposable to own it. Add a gated workflow and a
deterministic concurrency red/green test.
* .NET: Cover non-chat resume and multi-turn checkpoint advance
Add tests for HostedWorkflowState resuming a non-chat-protocol workflow (no
TurnToken) and for a third turn continuing to advance the head checkpoint,
closing the coverage gaps the parity review flagged.
* .NET: Add streaming workflow resume path and stream the workflow sample
Add HostedWorkflowState.RunOrResumeStreamingAsync, which yields the turn's
WorkflowEvents as they occur (fresh run or checkpoint resume) under the same
serialization lock and records the head checkpoint after the stream drains,
keeping the blocking and streaming workflow paths in lockstep with the Python
host. Honor stream:true in the HostingResponsesWorkflow sample by projecting
AgentResponseUpdateEvent updates over the Responses SSE wire. Add a streaming
resume test and update the README/spec.
* .NET: Cover Responses input adaptation to a typed workflow start executor
Demonstrate that HostedWorkflowState's generic RunOrResumeAsync<TInput> is the
input-adaptation seam (parity with Python's ResponsesChannel run hook): the app
adapts the Responses input into the workflow start executor's own type at the
call site. Add a typed-brief workflow and a test, and note the seam in spec-003.
* .NET: Drain workflow resume non-blocking to prevent hang and truncation
The resume drain used a SuperStepCompletedEvent{HasPendingRequests} proxy over
the blocking public WatchStreamAsync. That proxy (a) truncated a resumed turn
when a superstep both emitted a request and queued downstream work, and (b)
could fail to fire at all — re-introducing the indefinite hang — when a resume
input drove no superstep (e.g. a rejected non-chat input).
Make StreamingRun.WatchStreamAsync(bool blockOnPendingRequest, CancellationToken)
public and drain both the blocking and streaming resume paths with
blockOnPendingRequest:false, exactly matching the first-turn RunAsync semantics
(Run.RunToNextHaltAsync). Add guard tests: resume with a rejected input does not
hang, and a resume superstep with a request plus downstream work is not
truncated (verified red against the old proxy).
* .NET: Return file-store checkpoint index in commit order
CheckpointManager.GetLatestCheckpointAsync takes the last entry of a store's
index as the head checkpoint. FileSystemJsonCheckpointStore backed its index
with a HashSet, whose enumeration order is not contractual: after a rollback
frees and reuses a slot, enumeration can diverge from commit order, so the
durable read-through could resume a stale checkpoint. Mirror the HashSet with an
insertion-ordered list and enumerate it from RetrieveIndexAsync so 'latest' is
reliable. Add a CheckpointManager.GetLatestCheckpointAsync contract test over the
file store.
Note: the HashSet disorder is only reachable via the internal rollback path, so
the test locks the ordering contract rather than reproducing the rare disorder.
* .NET: Advance cursor when a streaming resume is abandoned
RunOrResumeStreamingAsync recorded the head checkpoint only after the stream was
fully enumerated. If an SSE consumer disconnected mid-turn after supersteps had
committed, the in-memory cursor kept the previous turn's head; because the next
turn is then a cursor hit, durable read-through could not self-heal, so it
resumed pre-disconnect state. Record the run's last committed checkpoint in a
finally so an abandoned stream still advances the cursor. Add a red/green test.
* .NET: Stream only the final agent's updates in the workflow sample
ExtractUpdates streamed every agent's updates, so the sequential Writer->Reviewer
sample streamed the intermediate draft and the final answer over SSE, differing
from the non-streaming response (final message only). Filter the streamed updates
to the final agent so streaming and non-streaming produce the same response.
Live-verified against Foundry: one output item streamed instead of two.
* .NET: Isolate the holder lock in the concurrency test
The concurrency test asserted the second same-session turn did not enter the
workflow, which also passes via the engine's concurrent-run ownership guard
(which faults) rather than the holder lock (which waits). Assert instead that the
second turn is not completed while the first holds the lock: a fault would
complete the task, so a pending task isolates the holder lock from the engine
guard. Verified red with the lock removed.
* Fix IDE1006 naming in tests; address review feedback and add hosting/live tests
* Document commit-order contract for ICheckpointStore.RetrieveIndexAsync
* Restructure hosting samples under af-hosting with client/server split matching Python parity
* Clarify hosting sample README wording and drop Python comparisons
* Make AgentSessionStore.DeleteSessionAsync abstract and rename session id parameter to sessionStoreId
* Rename OpenAIResponses id helpers and parse the request once for id extraction
* Reclaim per-session locks in HostedAgentState and demonstrate session locking in the agent sample
* Internalize per-session locking in HostedAgentState (automatic, on by default) and remove mirroring-Python wording from code and spec
* Remove HostedAgentState; app-owned routes use AgentSessionStore directly
HostedAgentState only bundled an AIAgent with an AgentSessionStore and, after
the per-session lock was removed, its GetOrCreateSessionAsync/SaveSessionAsync/
DeleteSessionAsync were pass-throughs that just bound the agent argument.
Create-on-miss already lives in the store (unlike Python, whose get/set-only
SessionStore justifies its AgentState holder), so the type earned its place
only via the lock.
Each AgentSessionStore.GetSessionAsync now returns an independent session
instance per call, so concurrent gets fork the same stored state (e.g.
branching from previous_response_id or managing several conversation ids)
without sharing an instance. The store does no cross-call locking; serializing
concurrent runs against the same id is the application's concern.
- Delete HostedAgentState and its unit tests.
- Rewire the local_responses sample and the OpenAI hosting unit/integration
tests to call AgentSessionStore (GetSessionAsync/SaveSessionAsync) directly.
- Update ADR-0032, spec-003, and the af-hosting sample READMEs.
* Isolate hosted session snapshots and distinguish conversation vs response continuation
Mirrors the Python hosted-session isolation work: a hosted session read must be
an independent copy, and the app-owned route must persist under the right
continuation key depending on how the caller continued the thread.
- AgentSessionStore.GetSessionAsync: document the isolation invariant (each
call returns an independent AgentSession so concurrent branches from one
previous_response_id do not observe each other's mutations or alter stored
state); fix the stale "or null if not found" wording (in-box stores return a
fresh created session on miss). The in-box stores already satisfy this via a
serialize/deserialize snapshot round-trip.
- local_responses sample + hosting unit-test route: choose the save key by
channel. A stable conversation id is a mutable head (write back under the
same id; app owns single-writer coordination). A previous_response_id
continuation or first turn is an immutable snapshot (save under the new
response id so branches from the same prior response stay independent).
- Add regression tests: independent get returns a distinct instance
(InMemoryAgentSessionStore); previous_response_id supports independent
branches ([1,2,2,3,3]); conversation id advances the mutable head ([1,2]).
- Update the sample README and ADR-0032 wording.
* Add workflow-factory support to HostedWorkflowState for concurrent sessions
HostedWorkflowState backed every session with one shared Workflow instance and
serialized all turns through a lock, so independent sessions could not run
concurrently. Add a workflow-factory constructor and remove the run lock.
- New constructor HostedWorkflowState(Func<CancellationToken, ValueTask<Workflow>>
workflowFactory, ..., bool cacheWorkflow = false):
- cacheWorkflow: false (default) builds a fresh instance per run, so independent
sessions run in parallel. A resume rehydrates a fresh instance from the
session's checkpoint in the shared store.
- cacheWorkflow: true builds the workflow once, lazily on first use, and reuses
it (a deferred, cached target that, like a shared instance, cannot run
concurrent turns).
- Remove the internal SemaphoreSlim run lock and IDisposable; the instance
constructor is unchanged in behaviour (one shared instance still cannot run
concurrent turns). Turns are no longer serialized by the holder; a single
writer per session is the application's responsibility.
- Switch the local_responses_workflow sample to the factory constructor with an
explicit cacheWorkflow: false, and document the option.
- Add tests: parallel independent sessions (factory), fresh-instance resume,
cached factory builds once and reuses, uncached factory builds per run.
- Update ADR-0032, spec-003, and the sample README.
* Clarify in ADR-0032 how .NET covers AgentState factory and async-setup via DI
* Rebuild cached workflow after a faulted build and add checkpoint index dedup tests
45 lines
1.7 KiB
C#
45 lines
1.7 KiB
C#
// 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;
|
|
|
|
/// <summary>
|
|
/// Tests for <see cref="CheckpointManager.GetLatestCheckpointAsync"/>, which must return the most recently
|
|
/// committed checkpoint for a session regardless of the backing store implementation.
|
|
/// </summary>
|
|
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<CheckpointInfo> 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<CheckpointInfo> 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]);
|
|
}
|
|
}
|