diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 4dcfaa069f..690d3598e6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -11,9 +11,11 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, + type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2Subagent, + type OrchestrationV2StoredEvent, type OrchestrationV2TurnItem, ProviderDriverKind, ProviderInstanceId, @@ -3089,6 +3091,93 @@ function rootTerminalEvent( : { ...common, status, failure: null }; } +it.effect.each([false, true])( + "retains queued child output when the adapter is already idle and root ownership lost=%s", + (rootOwnershipLost) => + Effect.gen(function* () { + const childEvents = yield* Ref.make>([]); + yield* runBackgroundItemScenario( + "ordered-background-continuation", + (ids) => { + const childProviderThreadId = ProviderThreadId.make( + "provider-thread:ordered-background-continuation:child", + ); + const snapshot = (populated: boolean, child = false) => ({ + type: "provider_thread.updated" as const, + driver, + providerThread: { + id: child ? childProviderThreadId : ids.providerThreadId, + driver, + providerInstanceId: ProviderInstanceId.make("codex"), + providerSessionId: null, + appThreadId: child ? ids.childThreadId : ids.threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "idle" as const, + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + pendingBackgroundTasks: populated + ? [{ taskId: "child", kind: "subagent" as const, childThreadId: ids.childThreadId }] + : [], + createdAt: DateTime.makeUnsafe(0), + updatedAt: DateTime.makeUnsafe(0), + } satisfies OrchestrationV2ProviderThread, + }); + const childTurn = (status: "running" | "completed") => ({ + type: "provider_turn.updated" as const, + driver, + threadId: ids.childThreadId, + providerTurn: { + id: ProviderTurnId.make("provider-turn:ordered-background-continuation:child"), + providerThreadId: ProviderThreadId.make( + "provider-thread:ordered-background-continuation:child", + ), + runAttemptId: null, + ordinal: 2, + status, + } as OrchestrationV2ProviderTurn, + }); + return [ + childThreadCreatedEvent(ids), + snapshot(true), + snapshot(true, true), + subagentEvent(ids, "running"), + rootTerminalEvent(ids, "completed"), + subagentEvent(ids, "completed"), + snapshot(false), + childTurn("running"), + { + type: "message.updated", + driver, + message: { threadId: ids.childThreadId, runId: null, text: "GRANDCHILD_OK" }, + } as ProviderAdapter.ProviderAdapterV2Event, + childTurn("completed"), + snapshot(false, true), + snapshot(false), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + rejectRootProviderThreadUpdates: rootOwnershipLost, + onIngest: (event) => + Ref.update(childEvents, (events) => + event.type === "provider_turn.updated" + ? [...events, event.providerTurn.status] + : event.type === "message.updated" + ? [...events, event.message.text] + : events, + ), + }, + ); + assert.deepEqual(yield* Ref.get(childEvents), ["running", "GRANDCHILD_OK", "completed"]); + }), +); + + function runBackgroundItemScenario( key: string, makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray, @@ -3098,6 +3187,11 @@ function runBackgroundItemScenario( ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; readonly onSubscribe?: Effect.Effect; + readonly onIngest?: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; + readonly hasPendingBackgroundWorkForThread?: ( + thread: OrchestrationV2ProviderThread, + ) => Effect.Effect; + readonly rejectRootProviderThreadUpdates?: boolean; readonly onEventsConsumed?: Effect.Effect; readonly onFinalized?: ( events: ReadonlyArray, @@ -3136,6 +3230,8 @@ function runBackgroundItemScenario( ingestNormalized: (input) => Effect.gen(function* () { const event = input.event; + yield* options?.onIngest?.(event) ?? Effect.void; + if (event.type === "turn_item.updated") { yield* Ref.update(observed, (current) => [ ...current, @@ -3148,6 +3244,18 @@ function runBackgroundItemScenario( `subagent:${event.subagent.status}`, ]); } + if (event.type === "provider_thread.updated") { + if ( + options?.rejectRootProviderThreadUpdates === true && + input.writeIfProviderThreadOwner !== undefined + ) + return []; + return [ + { + event: { type: "provider-thread.updated", payload: event.providerThread }, + } as OrchestrationV2StoredEvent, + ]; + } return []; }), }), @@ -3164,6 +3272,12 @@ function runBackgroundItemScenario( providerSessionId: ProviderSessionId.make(`session:${key}`), session: { events: Stream.empty, + ...(options?.hasPendingBackgroundWorkForThread === undefined + ? {} + : { + hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread, + }), + subscribeEvents: Effect.gen(function* () { yield* options?.onSubscribe ?? Effect.void; const events = Stream.fromIterable(makeEvents(ids)).pipe(