From 048115772b136ee90485031c7c7b47fc27f78312 Mon Sep 17 00:00:00 2001 From: juan Date: Sun, 11 Oct 2026 22:15:09 -0500 Subject: [PATCH] fix(server): keep a resumed subagent's thread live after a restart Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- .../RunExecutionService.test.ts | 147 ++++++++++++++++++ .../orchestration-v2/RunExecutionService.ts | 23 ++- 2 files changed, 162 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 0e68d9f562e29..fd52dd42258c5 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -165,6 +165,153 @@ it.effect("routes shared-runtime events only to their owning root run", () => }), ); +it("claims a resumed run-owned subagent thread before routing its child events", () => { + const threadId = ThreadId.make("thread:resumed-child-routing"); + const childThreadId = ThreadId.make("thread:resumed-child-routing:child"); + const runId = RunId.make("run:resumed-child-routing"); + const identity: RunExecutionService.ProviderEventRouteIdentity = { + threadId, + runId, + attemptId: RunAttemptId.make("attempt:resumed-child-routing"), + providerThreadId: ProviderThreadId.make("provider-thread:resumed-child-routing"), + }; + const initial = RunExecutionService.makeProviderEventRoutingState({ + identity, + providerTurnId: null, + }); + const subagentEvent = { + type: "subagent.updated", + driver, + subagent: { + id: NodeId.make("node:resumed-child-routing"), + threadId, + runId, + childThreadId, + status: "running", + }, + } as ProviderAdapter.ProviderAdapterV2Event; + const childNodeEvent = { + type: "node.updated", + driver, + node: { + id: NodeId.make("node:resumed-child-routing:root"), + threadId: childThreadId, + runId: null, + kind: "root_turn", + status: "running", + completedAt: null, + providerThreadId: null, + }, + } as ProviderAdapter.ProviderAdapterV2Event; + const childItemEvent = { + type: "turn_item.updated", + driver, + turnItem: { + id: TurnItemId.make("turn-item:resumed-child-routing"), + threadId: childThreadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "running", + title: null, + startedAt: null, + completedAt: null, + updatedAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"), + type: "assistant_message", + messageId: MessageId.make("message:resumed-child-routing"), + text: "Child event", + streaming: false, + }, + } as ProviderAdapter.ProviderAdapterV2Event; + const [subagentAccepted, claimed] = RunExecutionService.routeProviderEvent( + subagentEvent, + identity, + initial, + ); + + assert.isTrue(subagentAccepted); + assert.isTrue(claimed.ownedThreadIds.has(childThreadId)); + assert.isTrue(RunExecutionService.routeProviderEvent(childNodeEvent, identity, claimed)[0]); + assert.isTrue(RunExecutionService.routeProviderEvent(childItemEvent, identity, claimed)[0]); +}); + +it("does not claim an unowned subagent's child thread", () => { + const threadId = ThreadId.make("thread:unowned-child-routing"); + const childThreadId = ThreadId.make("thread:unowned-child-routing:child"); + const identity: RunExecutionService.ProviderEventRouteIdentity = { + threadId, + runId: RunId.make("run:unowned-child-routing"), + attemptId: RunAttemptId.make("attempt:unowned-child-routing"), + providerThreadId: ProviderThreadId.make("provider-thread:unowned-child-routing"), + }; + const initial = RunExecutionService.makeProviderEventRoutingState({ + identity, + providerTurnId: null, + }); + const subagentEvent = { + type: "subagent.updated", + driver, + subagent: { + id: NodeId.make("node:unowned-child-routing"), + threadId: ThreadId.make("thread:unowned-child-routing:other-parent"), + runId: RunId.make("run:unowned-child-routing:other"), + childThreadId, + status: "running", + }, + } as ProviderAdapter.ProviderAdapterV2Event; + const childNodeEvent = { + type: "node.updated", + driver, + node: { + id: NodeId.make("node:unowned-child-routing:root"), + threadId: childThreadId, + runId: null, + kind: "root_turn", + status: "running", + completedAt: null, + providerThreadId: null, + }, + } as ProviderAdapter.ProviderAdapterV2Event; + const childItemEvent = { + type: "turn_item.updated", + driver, + turnItem: { + id: TurnItemId.make("turn-item:unowned-child-routing"), + threadId: childThreadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "running", + title: null, + startedAt: null, + completedAt: null, + updatedAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"), + type: "assistant_message", + messageId: MessageId.make("message:unowned-child-routing"), + text: "Child event", + streaming: false, + }, + } as ProviderAdapter.ProviderAdapterV2Event; + const [subagentAccepted, unchanged] = RunExecutionService.routeProviderEvent( + subagentEvent, + identity, + initial, + ); + + assert.isFalse(subagentAccepted); + assert.isFalse(unchanged.ownedThreadIds.has(childThreadId)); + assert.isFalse(RunExecutionService.routeProviderEvent(childNodeEvent, identity, unchanged)[0]); + assert.isFalse(RunExecutionService.routeProviderEvent(childItemEvent, identity, unchanged)[0]); +}); + it("routes only enrolled workflow members after the root turn ends", () => { const threadId = ThreadId.make("thread:late-child"); const rootProviderTurnId = ProviderTurnId.make("provider-turn:late-child"); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 61ab0a3f09543..eca8fb8b5029b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -430,20 +430,27 @@ export function routeProviderEvent( ownedThreadIds: new Set([...state.ownedThreadIds, coordinator.childThreadId]), } : state; - if ( - !ownsRun(coordinator.runId) || - coordinator.workflow === undefined || - coordinator.childThreadId === null - ) { + if (!ownsRun(coordinator.runId) || coordinator.childThreadId === null) { return [belongs, adopted]; } + const childThreadId = coordinator.childThreadId; const workflowMemberParents = new Map(state.workflowMemberParents); - for (const member of coordinator.workflow.agents) { + for (const member of coordinator.workflow?.agents ?? []) { if (member.childThreadId !== undefined) { - workflowMemberParents.set(member.childThreadId, coordinator.childThreadId); + workflowMemberParents.set(member.childThreadId, childThreadId); } } - return [true, { ...adopted, workflowMemberParents }]; + return [ + true, + { + ...adopted, + workflowMemberParents, + ownedThreadIds: + childThreadId === input.threadId + ? adopted.ownedThreadIds + : new Set([...adopted.ownedThreadIds, childThreadId]), + }, + ]; } case "message.updated": return [ownsRun(event.message.runId) || ownsChildThread(event.message.threadId), state];