From 85a9327f65e20cb643dc677e1447e0f41700baf1 Mon Sep 17 00:00:00 2001 From: David Kaya Date: Fri, 27 Mar 2026 20:49:52 +0100 Subject: [PATCH] fix: surface handoff specialist activity\n\nQueue per-agent thinking updates from synchronous Copilot SDK session\nevents, flush them through the workflow runner, and promote handoff\ntargets to thinking immediately after handoff activity is emitted.\nThis fixes the UI case where downstream agents stayed on 'No status\nyet' during handoff patterns.\n\nAlso add targeted handoff diagnostics and regression tests.\n\nCo-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Services/CopilotTurnExecutionState.cs | 57 ++++++-- .../Services/CopilotWorkflowRunner.cs | 116 ++++++++++++++-- .../CopilotTurnExecutionStateTests.cs | 65 +++++++++ .../CopilotWorkflowRunnerTests.cs | 128 ++++++++++++++++++ 4 files changed, 344 insertions(+), 22 deletions(-) diff --git a/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs b/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs index 5f834c2..2d3e0b7 100644 --- a/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs +++ b/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs @@ -9,6 +9,7 @@ internal sealed class CopilotTurnExecutionState { private readonly RunTurnCommandDto _command; private readonly HashSet _startedAgents = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentQueue _pendingActivityEvents = new(); private readonly ConcurrentQueue _pendingMcpOauthRequests = new(); private readonly ConcurrentDictionary _observedAgentsByMessageId = new(StringComparer.Ordinal); private readonly StreamingTranscriptBuffer _transcriptBuffer = new(); @@ -31,22 +32,22 @@ internal sealed class CopilotTurnExecutionState AgentIdentity agent, Func onActivity) { - ActiveAgent = agent; - - if (!_startedAgents.Add(agent.AgentId)) + AgentActivityEventDto? thinkingActivity = CreateThinkingActivityIfNeeded(agent); + if (thinkingActivity is null) { return; } - await onActivity(new AgentActivityEventDto + await onActivity(thinkingActivity).ConfigureAwait(false); + } + + public void QueueThinkingIfNeeded(AgentIdentity agent) + { + AgentActivityEventDto? thinkingActivity = CreateThinkingActivityIfNeeded(agent); + if (thinkingActivity is not null) { - Type = "agent-activity", - RequestId = _command.RequestId, - SessionId = _command.SessionId, - ActivityType = "thinking", - AgentId = agent.AgentId, - AgentName = agent.AgentName, - }).ConfigureAwait(false); + _pendingActivityEvents.Enqueue(thinkingActivity); + } } public void ApplyActivity(AgentActivityEventDto activity) @@ -70,12 +71,15 @@ internal sealed class CopilotTurnExecutionState { case AssistantMessageDeltaEvent messageDelta when !string.IsNullOrWhiteSpace(messageDelta.Data?.MessageId): RecordObservedAgentForMessage(agent, messageDelta.Data!.MessageId); + QueueThinkingIfNeeded(agent); break; case AssistantMessageEvent assistantMessage when !string.IsNullOrWhiteSpace(assistantMessage.Data?.MessageId): RecordObservedAgentForMessage(agent, assistantMessage.Data!.MessageId); + QueueThinkingIfNeeded(agent); break; case AssistantReasoningDeltaEvent: ActiveAgent = agent; + QueueThinkingIfNeeded(agent); break; case McpOauthRequiredEvent: ActiveAgent = agent; @@ -87,6 +91,17 @@ internal sealed class CopilotTurnExecutionState } } + public IReadOnlyList DrainPendingActivityEvents() + { + List pending = []; + while (_pendingActivityEvents.TryDequeue(out AgentActivityEventDto? activity)) + { + pending.Add(activity); + } + + return pending; + } + public void EnqueuePendingMcpOauthRequest(McpOauthRequiredEventDto request) { ArgumentNullException.ThrowIfNull(request); @@ -139,6 +154,26 @@ internal sealed class CopilotTurnExecutionState _observedAgentsByMessageId[messageId] = agent; } + private AgentActivityEventDto? CreateThinkingActivityIfNeeded(AgentIdentity agent) + { + ActiveAgent = agent; + + if (!_startedAgents.Add(agent.AgentId)) + { + return null; + } + + return new AgentActivityEventDto + { + Type = "agent-activity", + RequestId = _command.RequestId, + SessionId = _command.SessionId, + ActivityType = "thinking", + AgentId = agent.AgentId, + AgentName = agent.AgentName, + }; + } + public void UpdateCompletedMessages( IReadOnlyList allMessages, IReadOnlyList inputMessages) diff --git a/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs b/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs index 1b91bb9..280a31c 100644 --- a/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs +++ b/sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs @@ -1,3 +1,4 @@ +using System.Linq; using Aryx.AgentHost.Contracts; using GitHub.Copilot.SDK; using Microsoft.Agents.AI.Workflows; @@ -7,6 +8,7 @@ namespace Aryx.AgentHost.Services; public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner { + private const string HandoffFunctionPrefix = "handoff_to_"; private readonly PatternValidator _patternValidator; private readonly CopilotApprovalCoordinator _approvalCoordinator = new(); private readonly CopilotUserInputCoordinator _userInputCoordinator = new(); @@ -83,6 +85,7 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner { bool shouldEndTurn = await HandleWorkflowEventAsync(command, evt, inputMessages, state, onDelta, onActivity) .ConfigureAwait(false); + await EmitPendingActivityEventsAsync(state, onActivity).ConfigureAwait(false); await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false); if (shouldEndTurn) { @@ -90,11 +93,13 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner } } + await EmitPendingActivityEventsAsync(state, onActivity).ConfigureAwait(false); await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false); return state.FinalizeCompletedMessages(); } catch (OperationCanceledException) when (runCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested) { + await EmitPendingActivityEventsAsync(state, onActivity).ConfigureAwait(false); await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false); ExitPlanModeRequestedEventDto? exitPlanModeEvent = _exitPlanModeCoordinator.ConsumePendingRequest(command.RequestId); @@ -108,6 +113,16 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner } } + private static async Task EmitPendingActivityEventsAsync( + CopilotTurnExecutionState state, + Func onActivity) + { + foreach (AgentActivityEventDto activity in state.DrainPendingActivityEvents()) + { + await onActivity(activity).ConfigureAwait(false); + } + } + private static async Task EmitPendingMcpOauthRequestsAsync( CopilotTurnExecutionState state, Func onMcpOAuthRequired) @@ -140,13 +155,21 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner Func onDelta, Func onActivity) { - if (evt is ExecutorInvokedEvent invoked - && AgentIdentityResolver.TryResolveKnownAgentIdentity( + if (evt is ExecutorInvokedEvent invoked) + { + if (AgentIdentityResolver.TryResolveKnownAgentIdentity( command.Pattern, invoked.ExecutorId, out AgentIdentity invokedAgent)) - { - await state.EmitThinkingIfNeeded(invokedAgent, onActivity).ConfigureAwait(false); + { + TraceHandoff(command, $"Executor invoked: {invoked.ExecutorId} -> {invokedAgent.AgentName} ({invokedAgent.AgentId})."); + await state.EmitThinkingIfNeeded(invokedAgent, onActivity).ConfigureAwait(false); + } + else + { + TraceHandoff(command, $"Executor invoked without a known agent match: {invoked.ExecutorId}."); + } + return false; } @@ -160,11 +183,14 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner if (activity is null) { - return WorkflowRequestInfoInterpreter.RequiresUserInputTurnBoundary(command, requestInfo); + bool requiresBoundary = WorkflowRequestInfoInterpreter.RequiresUserInputTurnBoundary(command, requestInfo); + TraceHandoff( + command, + $"Request info produced no activity for data type '{requestInfo.Request.Data.TypeId}'. Requires boundary: {requiresBoundary}."); + return requiresBoundary; } - state.ApplyActivity(activity); - await onActivity(activity).ConfigureAwait(false); + await EmitActivityAsync(command, state, activity, onActivity).ConfigureAwait(false); return false; } @@ -174,14 +200,22 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner return false; } - if (evt is ExecutorCompletedEvent completed - && AgentIdentityResolver.TryResolveObservedAgentIdentity( + if (evt is ExecutorCompletedEvent completed) + { + if (AgentIdentityResolver.TryResolveObservedAgentIdentity( command.Pattern, completed.ExecutorId, state.ActiveAgent, out AgentIdentity completedAgent)) - { - state.ClearActiveAgentIfMatching(completedAgent); + { + TraceHandoff(command, $"Executor completed: {completed.ExecutorId} -> {completedAgent.AgentName} ({completedAgent.AgentId})."); + state.ClearActiveAgentIfMatching(completedAgent); + } + else + { + TraceHandoff(command, $"Executor completed without a known agent match: {completed.ExecutorId}."); + } + return false; } @@ -203,6 +237,12 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner { AgentIdentity? updateAgent = null; string authorName = update.ExecutorId; + string[] handoffFunctionCalls = update.Update.Contents + .OfType() + .Select(content => content.Name) + .Where(IsHandoffFunctionName) + .Distinct(StringComparer.Ordinal) + .ToArray(); if (state.TryResolveObservedAgentForMessage(update.Update.MessageId, out AgentIdentity observedMessageAgent)) { updateAgent = observedMessageAgent; @@ -220,8 +260,21 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner if (updateAgent.HasValue) { + if (handoffFunctionCalls.Length > 0) + { + TraceHandoff( + command, + $"Agent response update from {updateAgent.Value.AgentName} ({updateAgent.Value.AgentId}) requested handoff via {string.Join(", ", handoffFunctionCalls)}."); + } + await state.EmitThinkingIfNeeded(updateAgent.Value, onActivity).ConfigureAwait(false); } + else if (!string.IsNullOrEmpty(update.Update.Text) || handoffFunctionCalls.Length > 0) + { + TraceHandoff( + command, + $"Agent response update could not resolve agent for executor '{update.ExecutorId}' and message '{update.Update.MessageId ?? ""}'."); + } if (string.IsNullOrEmpty(update.Update.Text)) { @@ -245,4 +298,45 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner Content = currentContent, }).ConfigureAwait(false); } + + private static async Task EmitActivityAsync( + RunTurnCommandDto command, + CopilotTurnExecutionState state, + AgentActivityEventDto activity, + Func onActivity) + { + state.ApplyActivity(activity); + TraceHandoff( + command, + $"Activity emitted: {activity.ActivityType} -> {activity.AgentName ?? activity.AgentId ?? ""}."); + await onActivity(activity).ConfigureAwait(false); + + if (string.Equals(activity.ActivityType, "handoff", StringComparison.Ordinal) + && !string.IsNullOrWhiteSpace(activity.AgentId) + && !string.IsNullOrWhiteSpace(activity.AgentName)) + { + TraceHandoff( + command, + $"Promoting handoff target to thinking: {activity.AgentName} ({activity.AgentId})."); + await state.EmitThinkingIfNeeded( + new AgentIdentity(activity.AgentId, activity.AgentName), + onActivity).ConfigureAwait(false); + } + } + + private static bool IsHandoffFunctionName(string? candidate) + { + return !string.IsNullOrWhiteSpace(candidate) + && candidate.StartsWith(HandoffFunctionPrefix, StringComparison.Ordinal); + } + + private static void TraceHandoff(RunTurnCommandDto command, string message) + { + if (!string.Equals(command.Pattern.Mode, "handoff", StringComparison.OrdinalIgnoreCase)) + { + return; + } + + Console.Error.WriteLine($"[aryx handoff] {message}"); + } } diff --git a/sidecar/tests/Aryx.AgentHost.Tests/CopilotTurnExecutionStateTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/CopilotTurnExecutionStateTests.cs index 50d13e2..a6e6641 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/CopilotTurnExecutionStateTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/CopilotTurnExecutionStateTests.cs @@ -29,6 +29,71 @@ public sealed class CopilotTurnExecutionStateTests Assert.Equal("Primary", state.ActiveAgent.Value.AgentName); } + [Fact] + public void ObserveSessionEvent_AssistantMessageDelta_QueuesThinkingActivity() + { + RunTurnCommandDto command = CreateCommand(); + CopilotTurnExecutionState state = new(command); + + state.ObserveSessionEvent( + command.Pattern.Agents[0], + SessionEvent.FromJson( + """ + { + "type": "assistant.message_delta", + "data": { + "messageId": "msg-1", + "deltaContent": "Hello" + }, + "id": "11111111-1111-1111-1111-111111111111", + "timestamp": "2026-03-27T00:00:00Z" + } + """)); + + AgentActivityEventDto activity = Assert.Single(state.DrainPendingActivityEvents()); + Assert.Equal("thinking", activity.ActivityType); + Assert.Equal("agent-1", activity.AgentId); + Assert.Equal("Primary", activity.AgentName); + Assert.True(state.TryResolveObservedAgentForMessage("msg-1", out AgentIdentity observedAgent)); + Assert.Equal("agent-1", observedAgent.AgentId); + } + + [Fact] + public async Task EmitThinkingIfNeeded_DoesNotDuplicateQueuedThinkingActivity() + { + RunTurnCommandDto command = CreateCommand(); + CopilotTurnExecutionState state = new(command); + + state.ObserveSessionEvent( + command.Pattern.Agents[0], + SessionEvent.FromJson( + """ + { + "type": "assistant.reasoning_delta", + "data": { + "reasoningId": "reasoning-1", + "deltaContent": "Planning" + }, + "id": "22222222-2222-2222-2222-222222222222", + "timestamp": "2026-03-27T00:00:00Z" + } + """)); + + List activities = [.. state.DrainPendingActivityEvents()]; + + await state.EmitThinkingIfNeeded( + new AgentIdentity("agent-1", "Primary"), + activity => + { + activities.Add(activity); + return Task.CompletedTask; + }); + + AgentActivityEventDto thinking = Assert.Single(activities); + Assert.Equal("thinking", thinking.ActivityType); + Assert.Equal("agent-1", thinking.AgentId); + } + [Fact] public void DrainPendingMcpOauthRequests_ReturnsQueuedRequestsAndClearsQueue() { diff --git a/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs index 0d41a4a..b4ad8e0 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs @@ -1,6 +1,10 @@ +using System.Reflection; +using System.Runtime.CompilerServices; using Aryx.AgentHost.Contracts; using Aryx.AgentHost.Services; using GitHub.Copilot.SDK; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.Workflows; using Microsoft.Extensions.AI; namespace Aryx.AgentHost.Tests; @@ -645,6 +649,9 @@ public sealed class CopilotWorkflowRunnerTests Assert.Equal("agent-handoff-ux", observedAgent.AgentId); Assert.Equal("UX Specialist", observedAgent.AgentName); Assert.Equal("agent-handoff-ux", state.ActiveAgent?.AgentId); + AgentActivityEventDto activity = Assert.Single(state.DrainPendingActivityEvents()); + Assert.Equal("thinking", activity.ActivityType); + Assert.Equal("agent-handoff-ux", activity.AgentId); } [Fact] @@ -668,6 +675,71 @@ public sealed class CopilotWorkflowRunnerTests Assert.Equal("agent-handoff-ux", state.ActiveAgent?.AgentId); Assert.Equal("UX Specialist", state.ActiveAgent?.AgentName); + AgentActivityEventDto activity = Assert.Single(state.DrainPendingActivityEvents()); + Assert.Equal("thinking", activity.ActivityType); + Assert.Equal("agent-handoff-ux", activity.AgentId); + } + + [Fact] + public async Task HandleWorkflowEventAsync_EmitsThinkingForHandoffTargets() + { + RunTurnCommandDto command = CreateHandoffCommand(); + CopilotTurnExecutionState state = new(command); + state.ObserveSessionEvent( + CreateAgent("agent-handoff-triage", "Triage"), + SessionEvent.FromJson( + """ + { + "type": "assistant.reasoning_delta", + "data": { + "reasoningId": "reasoning-1", + "deltaContent": "Delegating." + }, + "id": "33333333-3333-3333-3333-333333333333", + "timestamp": "2026-03-27T00:00:00Z" + } + """)); + _ = state.DrainPendingActivityEvents(); + RequestInfoEvent requestInfo = CreateRequestInfoEvent( + CreateHandoffTarget("agent-handoff-ux", "UX Specialist")); + List activities = []; + + MethodInfo handleWorkflowEvent = typeof(CopilotWorkflowRunner).GetMethod( + "HandleWorkflowEventAsync", + BindingFlags.NonPublic | BindingFlags.Static)!; + Task handleTask = (Task)handleWorkflowEvent.Invoke( + null, + [ + command, + requestInfo, + Array.Empty(), + state, + (Func)(_ => Task.CompletedTask), + (Func)(activity => + { + activities.Add(activity); + return Task.CompletedTask; + }), + ])!; + + bool shouldEndTurn = await handleTask; + + Assert.False(shouldEndTurn); + Assert.Collection( + activities, + handoff => + { + Assert.Equal("handoff", handoff.ActivityType); + Assert.Equal("agent-handoff-ux", handoff.AgentId); + Assert.Equal("UX Specialist", handoff.AgentName); + Assert.Equal("agent-handoff-triage", handoff.SourceAgentId); + }, + thinking => + { + Assert.Equal("thinking", thinking.ActivityType); + Assert.Equal("agent-handoff-ux", thinking.AgentId); + Assert.Equal("UX Specialist", thinking.AgentName); + }); } [Fact] @@ -1227,6 +1299,33 @@ public sealed class CopilotWorkflowRunnerTests }; } + private static RequestInfoEvent CreateRequestInfoEvent(object payload) + { + RequestPort port = RequestPort.Create("test-port"); + ExternalRequest request = ExternalRequest.Create(port, payload, "request-1"); + return new RequestInfoEvent(request); + } + + private static object CreateHandoffTarget(string id, string name) + { + Type type = Type.GetType( + "Microsoft.Agents.AI.Workflows.Specialized.HandoffTarget, Microsoft.Agents.AI.Workflows", + throwOnError: true)!; + return Activator.CreateInstance(type, CreateChatClientAgent(id, name), "Handle the UX work.")!; + } + + private static ChatClientAgent CreateChatClientAgent(string id, string name) + { + return new ChatClientAgent( + new StubChatClient(), + id, + name, + "Stub agent for handoff tests.", + [], + null!, + null!); + } + private static RunTurnCommandDto CreateApprovalCommand() { return new RunTurnCommandDto @@ -1258,4 +1357,33 @@ public sealed class CopilotWorkflowRunnerTests }, }; } + + private sealed class StubChatClient : IChatClient + { + public Task GetResponseAsync( + IEnumerable messages, + ChatOptions? options, + CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + + public async IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, + ChatOptions? options, + [EnumeratorCancellation] + CancellationToken cancellationToken) + { + yield break; + } + + public object? GetService(Type serviceType, object? serviceKey) + { + return null; + } + + public void Dispose() + { + } + } }