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() + { + } + } }