.NET: Track and update A2A task state - #7998
Conversation
Use the streaming agent API for new A2A messages so task responses continue receiving status and artifact updates. Aggregate updates for non-immediate requests and preserve response metadata in completed task artifacts. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 82ccdf82-97a9-4d31-a66a-9641edbccd35
There was a problem hiding this comment.
Pull request overview
Updates A2A hosting to stream agent responses into task state and artifacts.
Changes:
- Routes new messages through streaming execution.
- Supports immediate streaming and aggregated task/message responses.
- Adds coverage for response modes, empty responses, and metadata.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
dotnet/src/Microsoft.Agents.AI.Hosting.A2A/A2AAgentHandler.cs |
Implements streaming and task aggregation. |
dotnet/tests/Microsoft.Agents.AI.Hosting.A2A.UnitTests/A2AAgentHandlerTests.cs |
Tests routing and task lifecycle behavior. |
dotnet/tests/Microsoft.Agents.AI.Hosting.A2A.UnitTests/A2AServerServiceCollectionExtensionsTests.cs |
Adds end-to-end server response tests. |
Suppressed comments (1)
dotnet/src/Microsoft.Agents.AI.Hosting.A2A/A2AAgentHandler.cs:102
- Routing the non-streaming endpoint through this helper drops the run-mode value that the previous path passed as
AgentRunOptions.AllowBackgroundResponses: the shared helper callsCreateRunOptions(context)without the result of_runMode. Implementations such asFoundryHostedRequestAgentandChatClientAgentuse this option to activate background/resumable execution, soAllowBackgroundIfSupportednow only wraps a foreground run in a task andDisallowBackgroundno longer explicitly disables it. Evaluate the run mode before creating the options and pass that result into the streaming run.
await this.HandleNewMessageStreamingAsync(context, eventQueue, aggregateTaskUpdates, cancellationToken).ConfigureAwait(false);
💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.
| // Aggregate task updates unless the client requests an immediate response. | ||
| bool aggregateTaskUpdates = context.Configuration?.ReturnImmediately is not true; | ||
|
|
||
| // AIAgent does not support resuming from arbitrary prior tasks. | ||
| // Throw explicitly so the client gets a clear error rather than a response | ||
| // that silently ignores the referenced task context. | ||
| if (context.Message?.ReferenceTaskIds is { Count: > 0 }) | ||
| { | ||
| throw new NotSupportedException("ReferenceTaskIds is not supported. AIAgent cannot resume from arbitrary prior task context."); | ||
| } | ||
|
|
||
| List<ChatMessage> chatMessages = context.Message is not null ? [context.Message.ToChatMessage()] : []; | ||
|
|
||
| // Decide whether to run in background based on user preferences and agent capabilities | ||
| var decisionContext = new A2ARunDecisionContext(context); | ||
| var allowBackgroundResponses = await this._runMode.ShouldRunInBackgroundAsync(decisionContext, cancellationToken).ConfigureAwait(false); | ||
|
|
||
| var options = CreateRunOptions(context, allowBackgroundResponses); | ||
|
|
||
| AgentResponse response; | ||
| try | ||
| { | ||
| response = await this._hostAgent.RunAsync( | ||
| chatMessages, | ||
| session: session, | ||
| options: options, | ||
| cancellationToken: cancellationToken).ConfigureAwait(false); | ||
| } | ||
| finally | ||
| { | ||
| await this._hostAgent.SaveSessionAsync(contextId, session, CancellationToken.None).ConfigureAwait(false); | ||
| } | ||
|
|
||
| if (response.ContinuationToken is null) | ||
| { | ||
| // Return a lightweight message response (no task lifecycle needed). | ||
| var message = CreateMessageFromResponse(contextId, response); | ||
| await eventQueue.EnqueueMessageAsync(message, cancellationToken).ConfigureAwait(false); | ||
| } | ||
| else | ||
| { | ||
| // Long-running operation: emit task lifecycle events. | ||
| var taskUpdater = new TaskUpdater(eventQueue, context.TaskId, contextId); | ||
| await taskUpdater.SubmitAsync(cancellationToken).ConfigureAwait(false); | ||
|
|
||
| Message? progressMessage = response.Messages.Count > 0 | ||
| ? CreateMessageFromResponse(contextId, response) | ||
| : null; | ||
|
|
||
| await taskUpdater.StartWorkAsync(progressMessage, cancellationToken).ConfigureAwait(false); | ||
| } | ||
| await this.HandleNewMessageStreamingAsync(context, eventQueue, aggregateTaskUpdates, cancellationToken).ConfigureAwait(false); |
| AgentResponse response = await updates.ToAgentResponseAsync(cancellationToken).ConfigureAwait(false); | ||
|
|
||
| if (response.Messages.Count == 0) | ||
| await updater.SubmitAsync(cancellationToken).ConfigureAwait(false); |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
MAF Automated Review — Iteration 1
Result: Findings reported
Scope: full PR (1 commit(s)): b811d678ea46
Model: gpt-5.6-sol
Overview
The review found 4 verified inline finding(s).
Reviewed the supplied pull-request change set across correctness, security/reliability, architecture, and failure behavior.
4 verified findings remained after source verification (1 high, 3 medium) across 1 file. Details are attached to the affected lines below.
Affected areas: dotnet/src/Microsoft.Agents.AI.Hosting.A2A/A2AAgentHandler.cs
|
|
||
| await taskUpdater.StartWorkAsync(progressMessage, cancellationToken).ConfigureAwait(false); | ||
| } | ||
| await this.HandleNewMessageStreamingAsync(context, eventQueue, aggregateTaskUpdates, cancellationToken).ConfigureAwait(false); |
There was a problem hiding this comment.
A2AgentHandler.cs:165-194 leaves AllowBackgroundResponses unset but treats AllowBackgroundIfSupported as an unconditional task decision, so foreground completions become persisted terminal tasks instead of fallback messages; any fix must preserve streamed task updates and ReturnImmediately aggregation semantics.
| { | ||
| // The server allows background responses and this is either a streaming request or a | ||
| // non-streaming request with ReturnImmediately enabled, so emit task updates as they arrive. | ||
| await StreamTaskUpdatesAsync(updates, taskUpdater, cancellationToken).ConfigureAwait(false); |
There was a problem hiding this comment.
For a non-streaming request with ReturnImmediately = true, this new route writes artifacts through ArtifactStreamWriter, which does not copy AgentResponseUpdate.AdditionalProperties. The previous non-streaming response exposed those properties as message metadata, and the new aggregate-task branch explicitly preserves them, so only the immediate task mode silently loses response metadata. Please carry the response metadata onto the streamed artifact updates as well.
| if (response.Messages.Count == 0) | ||
| await updater.SubmitAsync(cancellationToken).ConfigureAwait(false); | ||
|
|
||
| if (response.Messages.ToParts() is { Count: > 0 } parts) |
There was a problem hiding this comment.
When the stream contains multiple assistant messages with different MessageId values, flattening all messages to one parts list produces a single generated artifact and discards both the message boundaries and IDs. The equivalent streaming path deliberately creates a separate artifact for each message boundary, so ReturnImmediately = false changes the task's logical result shape. Please emit one artifact per aggregated message and retain its message ID where available.
| AgentResponse response = await updates.ToAgentResponseAsync(cancellationToken).ConfigureAwait(false); | ||
|
|
||
| if (response.Messages.Count == 0) | ||
| await updater.SubmitAsync(cancellationToken).ConfigureAwait(false); |
There was a problem hiding this comment.
After this task is submitted, metadata conversion, artifact enqueueing, or completion can still throw or be canceled, but this aggregate path has no failure/cancellation transition. The request then fails after the server has persisted a permanently Submitted task with no worker left to advance it. Please either prepare the complete terminal event before submission or mirror the streaming path's catch handling so every post-submit exit terminalizes the task.
Motivation & Context
A2A hosting can return either a message or a task. Previously, when it returned a background task, subsequent agent updates were not applied, leaving the task indefinitely in the
Workingstate.This change uses the agent streaming API for both response types. The server either aggregates updates into a message, returns an initial task and continues updating it, or waits for the stream to finish and returns the completed task.
Description & Review Guide
ReturnImmediately. Background-disabled registrations aggregate the stream into a message. The task aggregation path also preserves response metadata on artifacts, and the four server/client response combinations have dedicated coverage.AgentRunModeandReturnImmediately, and on task artifact construction in the aggregate path.Related Issue
Fixes #5362
Contribution Checklist
breaking changelabel (or add "[BREAKING]" to the title prefix, before or after any language prefix) — a workflow keeps the label and title prefix in sync automatically.