starred/aryx
1
0
Fork 0
mirror of https://github.com/davidkaya/aryx.git synced 2026-07-24 13:38:43 +02:00
Code Issues 3 Packages Projects Releases 10 Wiki Activity

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>

Browse Source
This commit is contained in:
David Kaya
2026-03-27 20:49:52 +01:00
parent 2a952dfebe
commit 85a9327f65
4 changed files with 344 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs
+46 -11
View File
@@ -9,6 +9,7 @@ internal sealed class CopilotTurnExecutionState
{
private readonly RunTurnCommandDto _command;
private readonly HashSet<string> _startedAgents = new(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentQueue<AgentActivityEventDto> _pendingActivityEvents = new();
private readonly ConcurrentQueue<McpOauthRequiredEventDto> _pendingMcpOauthRequests = new();
private readonly ConcurrentDictionary<string, AgentIdentity> _observedAgentsByMessageId = new(StringComparer.Ordinal);
private readonly StreamingTranscriptBuffer _transcriptBuffer = new();
@@ -31,22 +32,22 @@ internal sealed class CopilotTurnExecutionState
AgentIdentity agent,
Func<AgentActivityEventDto, Task> 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<AgentActivityEventDto> DrainPendingActivityEvents()
{
List<AgentActivityEventDto> 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<ChatMessage> allMessages,
IReadOnlyList<ChatMessage> inputMessages)
sidecar/src/Aryx.AgentHost/Services/CopilotWorkflowRunner.cs
+105 -11
View File
@@ -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<AgentActivityEventDto, Task> onActivity)
{
foreach (AgentActivityEventDto activity in state.DrainPendingActivityEvents())
{
await onActivity(activity).ConfigureAwait(false);
}
}
private static async Task EmitPendingMcpOauthRequestsAsync(
CopilotTurnExecutionState state,
Func<McpOauthRequiredEventDto, Task> onMcpOAuthRequired)
@@ -140,13 +155,21 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
Func<TurnDeltaEventDto, Task> onDelta,
Func<AgentActivityEventDto, Task> 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<FunctionCallContent>()
.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 ?? "<none>"}'.");
}
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<AgentActivityEventDto, Task> onActivity)
{
state.ApplyActivity(activity);
TraceHandoff(
command,
$"Activity emitted: {activity.ActivityType} -> {activity.AgentName ?? activity.AgentId ?? "<unknown>"}.");
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}");
}
}
sidecar/tests/Aryx.AgentHost.Tests/CopilotTurnExecutionStateTests.cs
+65
View File
@@ -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<AgentActivityEventDto> 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()
{
sidecar/tests/Aryx.AgentHost.Tests/CopilotWorkflowRunnerTests.cs
+128
View File
@@ -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<AgentActivityEventDto> activities = [];
MethodInfo handleWorkflowEvent = typeof(CopilotWorkflowRunner).GetMethod(
"HandleWorkflowEventAsync",
BindingFlags.NonPublic | BindingFlags.Static)!;
Task<bool> handleTask = (Task<bool>)handleWorkflowEvent.Invoke(
null,
[
command,
requestInfo,
Array.Empty<ChatMessage>(),
state,
(Func<TurnDeltaEventDto, Task>)(_ => Task.CompletedTask),
(Func<AgentActivityEventDto, Task>)(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<object, object>("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<ChatResponse> GetResponseAsync(
IEnumerable<ChatMessage> messages,
ChatOptions? options,
CancellationToken cancellationToken)
{
throw new NotSupportedException();
}
public async IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(
IEnumerable<ChatMessage> messages,
ChatOptions? options,
[EnumeratorCancellation]
CancellationToken cancellationToken)
{
yield break;
}
public object? GetService(Type serviceType, object? serviceKey)
{
return null;
}
public void Dispose()
{
}
}
}