From ceafeb81d36cdb968f449c9f1f185a9c1d209559 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:08:40 +0000 Subject: [PATCH 01/15] test(server): reproduce stale delegated-task question wakes replacing results Refs pingdotgg/t3code#25637 Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 129 ++++++++++++++++++ .../Orchestrator.control-reads.test.ts | 106 ++++++++++++++ 2 files changed, 235 insertions(+) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 6ce02cbb32e2a..8f9200cd02860 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -401,6 +401,135 @@ describe("OrchestratorMcpService", () => { }), ); + it.effect("keeps a question-notification run from replacing the child's latest result", () => + Effect.gen(function* () { + const parentThreadId = ThreadId.make("thread:mcp-stale-question-parent"); + const childThreadId = ThreadId.make("thread:mcp-stale-question-child"); + const grandchildThreadId = ThreadId.make("thread:mcp-stale-question-grandchild"); + const taskId = NodeId.make("node:mcp-stale-question-task"); + const resultRunId = RunId.make("run:mcp-stale-question-result"); + const wakeRunId = RunId.make("run:mcp-stale-question-wake"); + const completedAt = DateTime.makeUnsafe("2026-10-26T03:06:53.000Z"); + const wakeCompletedAt = DateTime.makeUnsafe("2026-10-26T03:07:09.000Z"); + const parentProjection = { + thread: { id: parentThreadId }, + runs: [], + contextTransfers: [], + subagents: [ + { + id: taskId, + threadId: parentThreadId, + origin: "app_owned", + childThreadId, + driver: "codex", + model: "gpt-5.6-terra", + status: "completed", + result: "Opened the fix PR.", + completionDelivery: { state: "pending" }, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const childProjection = { + thread: { id: childThreadId, title: "Issue lead" }, + runs: [ + { + id: resultRunId, + ordinal: 1, + status: "completed", + delegatedTaskId: taskId, + startedAt: completedAt, + completedAt, + }, + { + id: wakeRunId, + ordinal: 2, + status: "completed", + startedAt: wakeCompletedAt, + completedAt: wakeCompletedAt, + }, + ], + contextTransfers: [], + messages: [ + { + id: MessageId.make("message:mcp-stale-question-result"), + runId: resultRunId, + role: "assistant", + text: "Opened the fix PR.", + createdAt: completedAt, + updatedAt: completedAt, + }, + { + id: MessageId.make("message:mcp-stale-question-wake"), + runId: wakeRunId, + role: "user", + text: "Delegated task asked a question and is still working.", + notification: { + source: { + kind: "delegated_task", + taskIds: [NodeId.make("node:mcp-stale-question-grandchild-task")], + childThreadId: grandchildThreadId, + }, + outcome: "updated", + summary: "Delegated task has a question", + }, + createdAt: wakeCompletedAt, + updatedAt: wakeCompletedAt, + }, + { + id: MessageId.make("message:mcp-stale-question-answer"), + runId: wakeRunId, + role: "assistant", + text: "Answered the delayed question.", + createdAt: wakeCompletedAt, + updatedAt: wakeCompletedAt, + }, + ], + subagents: [], + providerThreads: [], + runtimeRequests: [], + nodes: [], + turnItems: [], + } as unknown as OrchestrationV2ThreadProjection; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: (threadId) => + Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection), + getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), + dispatch: () => Effect.succeed({} as never), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-stale-question"), + requestNamespace: "provider-session:mcp-stale-question", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:mcp-stale-question", + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + + const result = yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.taskStatus(scope, taskId); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + assert.equal(result.summary, "Opened the fix PR."); + assert.equal(result.latestTerminalRunId, resultRunId); + assert.isNull(result.latestTerminalSummary); + }), + ); + it.effect("does not dispose delivery when a nonterminal task has no active child run", () => Effect.gen(function* () { const parentThreadId = ThreadId.make("thread:mcp-cancel-parent"); diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index c36dfc365af60..37f70e0ee50a5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1213,6 +1213,112 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); +it.effect("drops a queued question wake once the asking task completes", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; + const eventSink = yield* EventSink.EventSinkV2; + const parentThreadId = ThreadId.make("thread:stale-question-parent"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-stale-question-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:stale-question"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-stale-question-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:stale-question-parent"), + text: "Delegate the proof", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!; + yield* orchestrator.dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("delegate-stale-question-task"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Capture the proof", + completionWake: "settled_only", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "mcp", + }); + const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; + const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!; + const now = yield* DateTime.now; + const question = { + id: EventId.make("event:stale-question"), + type: "runtime-request.updated", + threadId: task.childThreadId!, + runId: childRun.id, + occurredAt: now, + payload: { + id: RuntimeRequestId.make("request:stale-question"), + nodeId: childRun.rootNodeId!, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: now, + resolvedAt: null, + }, + } satisfies OrchestrationV2DomainEvent; + yield* eventSink.write({ events: [question] }); + yield* reactor.handle(question); + const wake = (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.find( + (message) => message.notification !== undefined, + )!; + const wakeRun = () => + projections + .getThreadRecords(parentThreadId, ["runs"]) + .pipe(Effect.map(({ runs }) => runs.find((run) => run.userMessageId === wake.id))); + assert.equal((yield* wakeRun())?.status, "queued"); + + // The child finishes without ever resolving its message-capability question. + const childCompleted = { + id: EventId.make("event:stale-question-child-completed"), + type: "run.updated", + threadId: task.childThreadId!, + runId: childRun.id, + occurredAt: now, + payload: { ...childRun, status: "completed", completedAt: now }, + } satisfies OrchestrationV2DomainEvent; + const taskCompleted = { + id: EventId.make("event:stale-question-task-completed"), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { ...task, status: "completed", result: "Proof captured.", completedAt: now }, + } satisfies OrchestrationV2DomainEvent; + yield* eventSink.write({ + commandId: CommandId.make("command:runtime-reconcile:stale-question"), + events: [childCompleted, taskCompleted], + }); + yield* reactor.handle(childCompleted); + yield* reactor.handle(taskCompleted); + + assert.equal((yield* wakeRun())?.status, "cancelled"); + }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), +); + it.effect.each([false, true])("rejects thread.create for an existing id (deleted: %s)", (deleted) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 3083be58dfd1003c2de6b202021b1428ab059c34 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:11:09 +0000 Subject: [PATCH 02/15] fix(server): drop stale delegated-task question wakes and keep them out of task results A question wake queued on the parent stayed queued after the asking task settled, and a child's question-notification run could replace its real result in the task result. Cancel a settled task's queued question wakes, and skip question-wake runs when picking the latest result run. Fixes pingdotgg/t3code#25637 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 11 +++- .../DelegatedTaskQuestionReactor.ts | 57 ++++++++++++++----- 2 files changed, 51 insertions(+), 17 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 32bfa58c59cc4..a4d1cc8bfed0b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -636,16 +636,21 @@ function latestTerminalResultRun( projection: Pick, delegatedRun: OrchestrationV2Run | undefined, ): OrchestrationV2Run | undefined { - const monitorRunIds = new Set( + const notificationRunIds = new Set( projection.messages - .filter((message) => message.notification?.source.kind === "monitor") + .filter( + ({ notification }) => + notification?.source.kind === "monitor" || + // Question wakes report on a grandchild; they are not this task's result. + (notification?.source.kind === "delegated_task" && notification.outcome === "updated"), + ) .map((message) => message.runId), ); return projection.runs .filter( (run) => ThreadManagementService.isTerminalRunStatus(run.status) && - !monitorRunIds.has(run.id) && + !notificationRunIds.has(run.id) && run.status !== "rolled_back" && (run.id === delegatedRun?.id || run.startedAt !== null), ) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 50ae6c74678fa..f21ef3453990a 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -4,6 +4,8 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, + type RuntimeRequestId, + type ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; @@ -58,6 +60,27 @@ const make = Effect.gen(function* () { }); }); + const cancelQueuedWakes = Effect.fnUntraced(function* ( + parentThreadId: ThreadId, + requestIds: ReadonlyArray, + ) { + const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]); + for (const requestId of requestIds) { + const key = `${parentThreadId}:${requestId}`; + const messageId = MessageId.make(`message:delegated-task-question:${key}`); + const queued = parent.runs.find( + (run) => run.status === "queued" && run.userMessageId === messageId, + ); + if (queued === undefined) continue; + yield* engine.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`), + threadId: parentThreadId, + runId: queued.id, + }); + } + }); + // Runs for every domain event; most events return before any work is needed. const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = Effect.fnUntraced( function* (event) { @@ -100,6 +123,25 @@ const make = Effect.gen(function* () { } return; } + if (event.type === "subagent.updated") { + const task = event.payload; + if ( + task.origin !== "app_owned" || + task.childThreadId === null || + !["completed", "failed", "cancelled", "interrupted"].includes(task.status) + ) { + return; + } + // A settled task can no longer use an answer to its still-pending questions. + const child = yield* projections.getThreadRecords(task.childThreadId, ["runtimeRequests"]); + yield* cancelQueuedWakes( + task.threadId, + child.runtimeRequests + .filter((request) => request.kind === "user_input") + .map((request) => request.id), + ); + return; + } if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") { return; } @@ -107,21 +149,8 @@ const make = Effect.gen(function* () { const parentThreadId = child.lineage.parentThreadId; if (parentThreadId === null) return; const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id); - const key = `${parentThreadId}:${event.payload.id}`; - const messageId = MessageId.make(`message:delegated-task-question:${key}`); if (event.payload.status !== "pending" || request?.status !== "pending") { - const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]); - const queued = parent.runs.find( - (run) => run.status === "queued" && run.userMessageId === messageId, - ); - if (queued !== undefined) { - yield* engine.dispatch({ - type: "queued-run.cancel", - commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`), - threadId: parentThreadId, - runId: queued.id, - }); - } + yield* cancelQueuedWakes(parentThreadId, [event.payload.id]); return; } const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]); From f9bba831640a3f8d89d9db0f74d66d464b6e97dc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:21:14 +0000 Subject: [PATCH 03/15] fix(server): validate queued question lifetime and task ownership --- .../DelegatedTaskQuestionReactor.ts | 24 +- .../Orchestrator.control-reads.test.ts | 283 +++++++++++------- .../src/orchestration-v2/Orchestrator.ts | 53 +++- 3 files changed, 255 insertions(+), 105 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index f21ef3453990a..2a4b76e12fcd8 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -133,11 +133,31 @@ const make = Effect.gen(function* () { return; } // A settled task can no longer use an answer to its still-pending questions. - const child = yield* projections.getThreadRecords(task.childThreadId, ["runtimeRequests"]); + const child = yield* projections.getThreadRecords(task.childThreadId, [ + "runtimeRequests", + "runs", + "nodes", + ]); + const taskRunIds = new Set( + child.runs + .filter( + (run) => + run.delegatedTaskId === task.id || + (run.delegatedTaskId === undefined && + child.thread.forkedFrom?.type === "node" && + child.thread.forkedFrom.nodeId === task.id), + ) + .map((run) => run.id), + ); + const taskNodeIds = new Set( + child.nodes + .filter((node) => node.runId !== null && taskRunIds.has(node.runId)) + .map((node) => node.id), + ); yield* cancelQueuedWakes( task.threadId, child.runtimeRequests - .filter((request) => request.kind === "user_input") + .filter((request) => request.kind === "user_input" && taskNodeIds.has(request.nodeId)) .map((request) => request.id), ); return; diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 37f70e0ee50a5..4f7c715aa68f1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1213,110 +1213,189 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); -it.effect("drops a queued question wake once the asking task completes", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; - const eventSink = yield* EventSink.EventSinkV2; - const parentThreadId = ThreadId.make("thread:stale-question-parent"); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make("create-stale-question-parent"), - threadId: parentThreadId, - projectId: ProjectId.make("project:stale-question"), - title: "Parent", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - createdBy: "user", - creationSource: "web", - }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("start-stale-question-parent"), - threadId: parentThreadId, - messageId: MessageId.make("message:stale-question-parent"), - text: "Delegate the proof", - attachments: [], - dispatchMode: { type: "start_immediately" }, - createdBy: "user", - creationSource: "web", - }); - const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!; - yield* orchestrator.dispatch({ - type: "delegated_task.request", - commandId: CommandId.make("delegate-stale-question-task"), - parentThreadId, - parentRunId: parentRun.id, - parentNodeId: parentRun.rootNodeId!, - task: "Capture the proof", - completionWake: "settled_only", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - createdBy: "agent", - creationSource: "mcp", - }); - const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; - const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!; - const now = yield* DateTime.now; - const question = { - id: EventId.make("event:stale-question"), - type: "runtime-request.updated", - threadId: task.childThreadId!, - runId: childRun.id, - occurredAt: now, - payload: { - id: RuntimeRequestId.make("request:stale-question"), - nodeId: childRun.rootNodeId!, - providerTurnId: null, - nativeRequestRef: null, - kind: "user_input", - status: "pending", - responseCapability: { type: "message" }, - createdAt: now, - resolvedAt: null, - }, - } satisfies OrchestrationV2DomainEvent; - yield* eventSink.write({ events: [question] }); - yield* reactor.handle(question); - const wake = (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.find( - (message) => message.notification !== undefined, - )!; - const wakeRun = () => - projections - .getThreadRecords(parentThreadId, ["runs"]) - .pipe(Effect.map(({ runs }) => runs.find((run) => run.userMessageId === wake.id))); - assert.equal((yield* wakeRun())?.status, "queued"); +it.effect.each([false, true])( + "drops a stale question wake (missed terminal event: %s)", + (missedEvent) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; + const eventSink = yield* EventSink.EventSinkV2; + const parentThreadId = ThreadId.make("thread:stale-question-parent"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-stale-question-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:stale-question"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-stale-question-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:stale-question-parent"), + text: "Delegate the proof", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!; + yield* orchestrator.dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("delegate-stale-question-task"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Capture the proof", + completionWake: "settled_only", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "mcp", + }); + const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; + const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])) + .runs[0]!; + const now = yield* DateTime.now; + const question = { + id: EventId.make("event:stale-question"), + type: "runtime-request.updated", + threadId: task.childThreadId!, + runId: childRun.id, + occurredAt: now, + payload: { + id: RuntimeRequestId.make("request:stale-question"), + nodeId: childRun.rootNodeId!, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: now, + resolvedAt: null, + }, + } satisfies OrchestrationV2DomainEvent; + yield* eventSink.write({ events: [question] }); + yield* reactor.handle(question); + const wake = (yield* projections.getThreadRecords(parentThreadId, [ + "messages", + ])).messages.find((message) => message.notification !== undefined)!; + const wakeRun = () => + projections + .getThreadRecords(parentThreadId, ["runs"]) + .pipe(Effect.map(({ runs }) => runs.find((run) => run.userMessageId === wake.id))); + assert.equal((yield* wakeRun())?.status, "queued"); - // The child finishes without ever resolving its message-capability question. - const childCompleted = { - id: EventId.make("event:stale-question-child-completed"), - type: "run.updated", - threadId: task.childThreadId!, - runId: childRun.id, - occurredAt: now, - payload: { ...childRun, status: "completed", completedAt: now }, - } satisfies OrchestrationV2DomainEvent; - const taskCompleted = { - id: EventId.make("event:stale-question-task-completed"), - type: "subagent.updated", - threadId: parentThreadId, - occurredAt: now, - payload: { ...task, status: "completed", result: "Proof captured.", completedAt: now }, - } satisfies OrchestrationV2DomainEvent; - yield* eventSink.write({ - commandId: CommandId.make("command:runtime-reconcile:stale-question"), - events: [childCompleted, taskCompleted], - }); - yield* reactor.handle(childCompleted); - yield* reactor.handle(taskCompleted); + // The child finishes without ever resolving its message-capability question. + const childCompleted = { + id: EventId.make("event:stale-question-child-completed"), + type: "run.updated", + threadId: task.childThreadId!, + runId: childRun.id, + occurredAt: now, + payload: { ...childRun, status: "completed", completedAt: now }, + } satisfies OrchestrationV2DomainEvent; + const taskCompleted = { + id: EventId.make("event:stale-question-task-completed"), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { ...task, status: "completed", result: "Proof captured.", completedAt: now }, + } satisfies OrchestrationV2DomainEvent; + yield* eventSink.write({ + commandId: CommandId.make("command:runtime-reconcile:stale-question"), + events: [childCompleted, taskCompleted], + }); + yield* reactor.handle(childCompleted); + if (missedEvent) { + // Persisted terminal state can predate the reactor subscription after restart. + yield* projections.apply({ + id: EventId.make("event:stale-question-parent-finished"), + type: "run.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }); + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("resume-stale-question"), + threadId: parentThreadId, + }); + } else { + yield* reactor.handle(taskCompleted); + } + assert.equal((yield* wakeRun())?.status, "cancelled"); + if (missedEvent) return; - assert.equal((yield* wakeRun())?.status, "cancelled"); - }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("stale-question-followup"), + threadId: task.childThreadId!, + messageId: MessageId.make("message:stale-question-followup"), + text: "Capture another proof", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const followup = (yield* projections.getThreadRecords(task.childThreadId!, [ + "runs", + ])).runs.find((run) => run.userMessageId === "message:stale-question-followup")!; + const followupQuestion = { + ...question, + id: EventId.make("event:stale-question-followup"), + runId: followup.id, + payload: { + ...question.payload, + id: RuntimeRequestId.make("request:stale-question-followup"), + nodeId: followup.rootNodeId!, + }, + }; + yield* eventSink.write({ events: [followupQuestion] }); + yield* reactor.handle(followupQuestion); + const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ + "runs", + ])).runs.find( + (run) => + run.userMessageId === + `message:delegated-task-question:${parentThreadId}:${followupQuestion.payload.id}`, + )!; + assert.equal(followupWake.status, "queued"); + yield* reactor.handle(taskCompleted); + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( + (run) => run.id === followupWake.id, + )?.status, + "queued", + ); + yield* projections.apply({ + id: EventId.make("event:followup-question-parent-finished"), + type: "run.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }); + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("resume-followup-question"), + threadId: parentThreadId, + }); + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( + (run) => run.id === followupWake.id, + )?.status, + "starting", + ); + }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); it.effect.each([false, true])("rejects thread.create for an existing id (deleted: %s)", (deleted) => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0c6ca268f1e04..949f6c58f63c5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1465,6 +1465,54 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]); }); + const cancelStaleQuestionWakes = Effect.fn("Orchestrator.cancelStaleQuestionWakes")(function* ( + projection: OrchestrationV2ThreadProjection, + ) { + const events = yield* Ref.make>([]); + for (const run of projection.runs) { + if (run.status !== "queued") continue; + const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); + const notification = message?.notification; + if (notification?.source.kind !== "delegated_task" || notification.outcome !== "updated") + continue; + const source = notification.source; + const task = projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)); + let pending = false; + if ( + task !== undefined && + task.childThreadId !== null && + !["completed", "failed", "cancelled", "interrupted"].includes(task.status) + ) { + const child = yield* projectionStore.getThreadRecords(task.childThreadId, [ + "runtimeRequests", + ]); + pending = child.runtimeRequests.some( + (request) => + request.kind === "user_input" && + request.status === "pending" && + message?.id === `message:delegated-task-question:${projection.thread.id}:${request.id}`, + ); + } + if (pending) continue; + yield* emitQueuedRunCancellation({ + command: { + type: "queued-run.cancel", + commandId: CommandId.make(`command:system:stale-question:${run.id}`), + threadId: projection.thread.id, + runId: run.id, + }, + events, + projection, + run, + now: yield* DateTime.now, + }); + } + const cancellations = yield* Ref.get(events); + if (cancellations.length === 0) return false; + yield* writeSystemEvents(cancellations); + return true; + }); + const startNextQueuedRun = Effect.fn("Orchestrator.startNextQueuedRun")( function* ( threadId: ThreadId, @@ -1475,7 +1523,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ) { if (!(yield* projectionStore.canStartQueuedRun(threadId))) return; - const projection = yield* readCommandProjection(threadId); + let projection = yield* readCommandProjection(threadId); if ( projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || @@ -1485,6 +1533,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; } + if (yield* cancelStaleQuestionWakes(projection)) { + projection = yield* readCommandProjection(threadId); + } const storedQueuedRun = options?.queuedRunId === undefined ? nextQueuedRun(projection) From 6337ba450346643d5396b71b4b55b3aedcfbf6fa Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:25:47 +0000 Subject: [PATCH 04/15] fix(server): reject stale question notifications at admission --- .../Orchestrator.control-reads.test.ts | 16 +++++- .../src/orchestration-v2/Orchestrator.ts | 56 ++++++++++++------- 2 files changed, 50 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 4f7c715aa68f1..370b556291f83 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1334,7 +1334,21 @@ it.effect.each([false, true])( yield* reactor.handle(taskCompleted); } assert.equal((yield* wakeRun())?.status, "cancelled"); - if (missedEvent) return; + if (missedEvent) { + const before = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.length; + const delayedQuestion = { + ...question, + id: EventId.make("event:delayed-question"), + payload: { ...question.payload, id: RuntimeRequestId.make("request:delayed-question") }, + }; + yield* eventSink.write({ events: [delayedQuestion] }); + yield* reactor.handle(delayedQuestion); + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.length, + before, + ); + return; + } yield* orchestrator.dispatch({ type: "message.dispatch", diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 949f6c58f63c5..4d99f9a9c75a1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1465,6 +1465,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]); }); + const isStaleQuestionWake = Effect.fn("Orchestrator.isStaleQuestionWake")(function* ( + projection: Pick, + messageId: MessageId, + notification: OrchestrationV2ThreadProjection["messages"][number]["notification"], + ) { + if (notification?.source.kind !== "delegated_task" || notification.outcome !== "updated") + return false; + const source = notification.source; + const task = projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)); + if ( + task === undefined || + task.childThreadId === null || + ["completed", "failed", "cancelled", "interrupted"].includes(task.status) + ) + return true; + const child = yield* projectionStore + .getThreadRecords(task.childThreadId, ["runtimeRequests"]) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: task.childThreadId!, cause }), + ), + ); + return !child.runtimeRequests.some( + (request) => + request.kind === "user_input" && + request.status === "pending" && + messageId === `message:delegated-task-question:${projection.thread.id}:${request.id}`, + ); + }); + const cancelStaleQuestionWakes = Effect.fn("Orchestrator.cancelStaleQuestionWakes")(function* ( projection: OrchestrationV2ThreadProjection, ) { @@ -1472,28 +1502,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio for (const run of projection.runs) { if (run.status !== "queued") continue; const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); - const notification = message?.notification; - if (notification?.source.kind !== "delegated_task" || notification.outcome !== "updated") - continue; - const source = notification.source; - const task = projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)); - let pending = false; if ( - task !== undefined && - task.childThreadId !== null && - !["completed", "failed", "cancelled", "interrupted"].includes(task.status) - ) { - const child = yield* projectionStore.getThreadRecords(task.childThreadId, [ - "runtimeRequests", - ]); - pending = child.runtimeRequests.some( - (request) => - request.kind === "user_input" && - request.status === "pending" && - message?.id === `message:delegated-task-question:${projection.thread.id}:${request.id}`, - ); - } - if (pending) continue; + message === undefined || + !(yield* isStaleQuestionWake(projection, message.id, message.notification)) + ) + continue; yield* emitQueuedRunCancellation({ command: { type: "queued-run.cancel", @@ -5037,6 +5050,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio effects: Ref.Ref>, ) { let projection = yield* getProjectionWithPendingEvents(command.threadId, events); + if (yield* isStaleQuestionWake(projection, command.messageId, command.notification)) return; if (command.manualContinuationOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId); const limited = latestRootProviderFailure(source ?? null, projection.turnItems); From 819ff10e37bcba129aba47e6693c7d4d85d77338 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:28:26 +0000 Subject: [PATCH 05/15] fix(server): retain task results across question notification turns --- .../SubagentProjection.test.ts | 25 +++++++++++++++++++ .../src/server/subagentProjection.ts | 15 ++++++++++- 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index d3506649f6a62..25dea1fcf5c75 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -263,6 +263,31 @@ it("waits for nested work and retains the report across monitor acknowledgements assert.equal(progress.resultRun?.id, report.id); }); +it("retains the task result after answering a delegated question notification", () => { + const { projection, run } = taskFixture(); + const wake = { ...run, id: RunId.make("question-wake"), ordinal: 2 }; + const progress = delegatedTaskProgress({ + ...projection, + runs: [run, wake], + messages: [ + { + runId: wake.id, + notification: { + source: { + kind: "delegated_task", + taskIds: [NodeId.make("grandchild")], + childThreadId: ThreadId.make("grandchild-thread"), + }, + outcome: "updated", + summary: "Delegated task has a question", + }, + }, + ], + }); + assert.equal(progress.state, "result_available"); + assert.equal(progress.resultRun?.id, run.id); +}); + it("exposes the provider failure rather than a progress message from the failed run", () => { const { projection, run } = taskFixture(); const failedRun = { ...run, status: "failed" as const }; diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index f83a0b376233b..148f340ae3e67 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -255,8 +255,21 @@ export function delegatedTaskProgress(projection: { task.completionDelivery?.state === "claimed")), ) || projection.providerThreads.some((thread) => (thread.pendingBackgroundTasks?.length ?? 0) > 0); + const questionRuns = new Set( + projection.messages + .filter( + ({ notification }) => + notification?.source.kind === "delegated_task" && notification.outcome === "updated", + ) + .map((message) => message.runId), + ); const resultRun = workRuns - .filter((run) => terminal(run.status) && (run.startedAt !== null || run.ordinal === 1)) + .filter( + (run) => + !questionRuns.has(run.id) && + terminal(run.status) && + (run.startedAt !== null || run.ordinal === 1), + ) .toSorted((a, b) => (runRanAfter(a, b) ? -1 : runRanAfter(b, a) ? 1 : 0))[0]; return { state: From c8bac5f226244b676de1730ee6cd13df9f672d8e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:33:10 +0000 Subject: [PATCH 06/15] fix(server): preserve question result classification across restart --- .../src/mcp/OrchestratorMcpService.test.ts | 257 +++++++++--------- apps/server/src/mcp/OrchestratorMcpService.ts | 15 +- .../SubagentProjection.test.ts | 57 ++-- .../src/server/subagentProjection.ts | 44 ++- 4 files changed, 211 insertions(+), 162 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 8f9200cd02860..7732968711837 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -401,133 +401,144 @@ describe("OrchestratorMcpService", () => { }), ); - it.effect("keeps a question-notification run from replacing the child's latest result", () => - Effect.gen(function* () { - const parentThreadId = ThreadId.make("thread:mcp-stale-question-parent"); - const childThreadId = ThreadId.make("thread:mcp-stale-question-child"); - const grandchildThreadId = ThreadId.make("thread:mcp-stale-question-grandchild"); - const taskId = NodeId.make("node:mcp-stale-question-task"); - const resultRunId = RunId.make("run:mcp-stale-question-result"); - const wakeRunId = RunId.make("run:mcp-stale-question-wake"); - const completedAt = DateTime.makeUnsafe("2026-10-26T03:06:53.000Z"); - const wakeCompletedAt = DateTime.makeUnsafe("2026-10-26T03:07:09.000Z"); - const parentProjection = { - thread: { id: parentThreadId }, - runs: [], - contextTransfers: [], - subagents: [ - { - id: taskId, - threadId: parentThreadId, - origin: "app_owned", - childThreadId, - driver: "codex", - model: "gpt-5.6-terra", - status: "completed", - result: "Opened the fix PR.", - completionDelivery: { state: "pending" }, - }, - ], - } as unknown as OrchestrationV2ThreadProjection; - const childProjection = { - thread: { id: childThreadId, title: "Issue lead" }, - runs: [ - { - id: resultRunId, - ordinal: 1, - status: "completed", - delegatedTaskId: taskId, - startedAt: completedAt, - completedAt, - }, - { - id: wakeRunId, - ordinal: 2, - status: "completed", - startedAt: wakeCompletedAt, - completedAt: wakeCompletedAt, - }, - ], - contextTransfers: [], - messages: [ - { - id: MessageId.make("message:mcp-stale-question-result"), - runId: resultRunId, - role: "assistant", - text: "Opened the fix PR.", - createdAt: completedAt, - updatedAt: completedAt, - }, - { - id: MessageId.make("message:mcp-stale-question-wake"), - runId: wakeRunId, - role: "user", - text: "Delegated task asked a question and is still working.", - notification: { - source: { - kind: "delegated_task", - taskIds: [NodeId.make("node:mcp-stale-question-grandchild-task")], - childThreadId: grandchildThreadId, + it.effect.each([0, 2])( + "keeps a question wake with %s restarts from replacing the child's latest result", + (restarts) => + Effect.gen(function* () { + const parentThreadId = ThreadId.make("thread:mcp-stale-question-parent"); + const childThreadId = ThreadId.make("thread:mcp-stale-question-child"); + const grandchildThreadId = ThreadId.make("thread:mcp-stale-question-grandchild"); + const taskId = NodeId.make("node:mcp-stale-question-task"); + const resultRunId = RunId.make("run:mcp-stale-question-result"); + const wakeRunId = RunId.make("run:mcp-stale-question-wake"); + const completedAt = DateTime.makeUnsafe("2026-10-26T03:06:53.000Z"); + const wakeCompletedAt = DateTime.makeUnsafe("2026-10-26T03:07:09.000Z"); + const parentProjection = { + thread: { id: parentThreadId }, + runs: [], + contextTransfers: [], + subagents: [ + { + id: taskId, + threadId: parentThreadId, + origin: "app_owned", + childThreadId, + driver: "codex", + model: "gpt-5.6-terra", + status: "completed", + result: "Opened the fix PR.", + completionDelivery: { state: "pending" }, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const childProjection = { + thread: { id: childThreadId, title: "Issue lead" }, + runs: [ + { + id: resultRunId, + ordinal: 1, + status: "completed", + delegatedTaskId: taskId, + startedAt: completedAt, + completedAt, + }, + { + id: wakeRunId, + ordinal: 2, + status: "completed", + startedAt: wakeCompletedAt, + completedAt: wakeCompletedAt, + }, + ...Array.from({ length: restarts }, (_, index) => ({ + id: RunId.make(`run:mcp-question-restart:${index}`), + ordinal: 3 + index, + status: "completed", + startedAt: wakeCompletedAt, + completedAt: wakeCompletedAt, + restartContinuationOfRunId: + index === 0 ? wakeRunId : RunId.make(`run:mcp-question-restart:${index - 1}`), + })), + ], + contextTransfers: [], + messages: [ + { + id: MessageId.make("message:mcp-stale-question-result"), + runId: resultRunId, + role: "assistant", + text: "Opened the fix PR.", + createdAt: completedAt, + updatedAt: completedAt, + }, + { + id: MessageId.make("message:mcp-stale-question-wake"), + runId: wakeRunId, + role: "user", + text: "Delegated task asked a question and is still working.", + notification: { + source: { + kind: "delegated_task", + taskIds: [NodeId.make("node:mcp-stale-question-grandchild-task")], + childThreadId: grandchildThreadId, + }, + outcome: "updated", + summary: "Delegated task has a question", }, - outcome: "updated", - summary: "Delegated task has a question", + createdAt: wakeCompletedAt, + updatedAt: wakeCompletedAt, }, - createdAt: wakeCompletedAt, - updatedAt: wakeCompletedAt, - }, - { - id: MessageId.make("message:mcp-stale-question-answer"), - runId: wakeRunId, - role: "assistant", - text: "Answered the delayed question.", - createdAt: wakeCompletedAt, - updatedAt: wakeCompletedAt, + { + id: MessageId.make("message:mcp-stale-question-answer"), + runId: wakeRunId, + role: "assistant", + text: "Answered the delayed question.", + createdAt: wakeCompletedAt, + updatedAt: wakeCompletedAt, + }, + ], + subagents: [], + providerThreads: [], + runtimeRequests: [], + nodes: [], + turnItems: [], + } as unknown as OrchestrationV2ThreadProjection; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: (threadId) => + Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection), + getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), + dispatch: () => Effect.succeed({} as never), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-stale-question"), + requestNamespace: "provider-session:mcp-stale-question", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:mcp-stale-question", + providerInstanceId: ProviderInstanceId.make("codex"), }, - ], - subagents: [], - providerThreads: [], - runtimeRequests: [], - nodes: [], - turnItems: [], - } as unknown as OrchestrationV2ThreadProjection; - const layerDependencies = Layer.mergeAll( - NodeServices.layer, - Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadRecords: (threadId) => - Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection), - getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), - dispatch: () => Effect.succeed({} as never), - }), - ServerSettings.layerTest(), - Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), - Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ - list: () => Effect.succeed([]), - }), - Layer.mock(ProjectService.ProjectService)({}), - Layer.mock(SecretRequests.SecretRequests)({}), - Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), - ); - const scope: McpInvocationScope = { - environmentId: EnvironmentId.make("environment:mcp-stale-question"), - requestNamespace: "provider-session:mcp-stale-question", - thread: { - threadId: parentThreadId, - providerSessionId: "provider-session:mcp-stale-question", - providerInstanceId: ProviderInstanceId.make("codex"), - }, - client: undefined, - capabilities: new Set(["orchestration"]), - issuedAt: 1, - }; + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; - const result = yield* Effect.gen(function* () { - const service = yield* OrchestratorMcpService.OrchestratorMcpService; - return yield* service.taskStatus(scope, taskId); - }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); - assert.equal(result.summary, "Opened the fix PR."); - assert.equal(result.latestTerminalRunId, resultRunId); - assert.isNull(result.latestTerminalSummary); - }), + const result = yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.taskStatus(scope, taskId); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + assert.equal(result.summary, "Opened the fix PR."); + assert.equal(result.latestTerminalRunId, resultRunId); + assert.isNull(result.latestTerminalSummary); + }), ); it.effect("does not dispose delivery when a nonterminal task has no active child run", () => diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index a4d1cc8bfed0b..4965225666782 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -90,6 +90,7 @@ import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterReg import { subagentResultForRun, delegatedTaskProgress, + delegatedQuestionRunIds, } from "@t3tools/provider-core/server/subagentProjection"; import { DispatchModeLimit, @@ -636,16 +637,12 @@ function latestTerminalResultRun( projection: Pick, delegatedRun: OrchestrationV2Run | undefined, ): OrchestrationV2Run | undefined { - const notificationRunIds = new Set( - projection.messages - .filter( - ({ notification }) => - notification?.source.kind === "monitor" || - // Question wakes report on a grandchild; they are not this task's result. - (notification?.source.kind === "delegated_task" && notification.outcome === "updated"), - ) + const notificationRunIds = new Set([ + ...delegatedQuestionRunIds(projection), + ...projection.messages + .filter((message) => message.notification?.source.kind === "monitor") .map((message) => message.runId), - ); + ]); return projection.runs .filter( (run) => diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index 25dea1fcf5c75..fb1b1b401c638 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -263,30 +263,43 @@ it("waits for nested work and retains the report across monitor acknowledgements assert.equal(progress.resultRun?.id, report.id); }); -it("retains the task result after answering a delegated question notification", () => { - const { projection, run } = taskFixture(); - const wake = { ...run, id: RunId.make("question-wake"), ordinal: 2 }; - const progress = delegatedTaskProgress({ - ...projection, - runs: [run, wake], - messages: [ - { - runId: wake.id, - notification: { - source: { - kind: "delegated_task", - taskIds: [NodeId.make("grandchild")], - childThreadId: ThreadId.make("grandchild-thread"), +it.each([0, 2])( + "retains the task result after a question wake with %s restart continuations", + (restarts) => { + const { projection, run } = taskFixture(); + const wake = { ...run, id: RunId.make("question-wake"), ordinal: 2 }; + const progress = delegatedTaskProgress({ + ...projection, + runs: [ + run, + wake, + ...Array.from({ length: restarts }, (_, index) => ({ + ...wake, + id: RunId.make(`question-restart:${index}`), + ordinal: 3 + index, + restartContinuationOfRunId: + index === 0 ? wake.id : RunId.make(`question-restart:${index - 1}`), + })), + ], + messages: [ + { + runId: wake.id, + notification: { + source: { + kind: "delegated_task", + taskIds: [NodeId.make("grandchild")], + childThreadId: ThreadId.make("grandchild-thread"), + }, + outcome: "updated", + summary: "Delegated task has a question", }, - outcome: "updated", - summary: "Delegated task has a question", }, - }, - ], - }); - assert.equal(progress.state, "result_available"); - assert.equal(progress.resultRun?.id, run.id); -}); + ], + }); + assert.equal(progress.state, "result_available"); + assert.equal(progress.resultRun?.id, run.id); + }, +); it("exposes the provider failure rather than a progress message from the failed run", () => { const { projection, run } = taskFixture(); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 148f340ae3e67..491080c1ab41d 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -15,6 +15,7 @@ import type { ProviderInstanceId, ProviderThreadId, ProviderTurnId, + RunId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -219,6 +220,40 @@ export function subagentResultForRun( }; } +export function delegatedQuestionRunIds(projection: { + readonly runs: ReadonlyArray>; + readonly messages: ReadonlyArray< + Pick + >; +}) { + const questionRuns = new Set( + projection.messages.flatMap(({ runId, notification }) => + runId !== null && + notification?.source.kind === "delegated_task" && + notification.outcome === "updated" + ? [runId] + : [], + ), + ); + const continuations = new Map>(); + for (const run of projection.runs) { + if (run.restartContinuationOfRunId === undefined) continue; + const siblings = continuations.get(run.restartContinuationOfRunId) ?? []; + siblings.push(run.id); + continuations.set(run.restartContinuationOfRunId, siblings); + } + // Restart continuations carry system notifications instead of the original question. + const pending = [...questionRuns]; + for (const runId of pending) { + for (const continuationId of continuations.get(runId) ?? []) { + if (questionRuns.has(continuationId)) continue; + questionRuns.add(continuationId); + pending.push(continuationId); + } + } + return questionRuns; +} + export function delegatedTaskProgress(projection: { readonly runs: OrchestrationV2ThreadProjection["runs"]; readonly messages: ReadonlyArray< @@ -255,14 +290,7 @@ export function delegatedTaskProgress(projection: { task.completionDelivery?.state === "claimed")), ) || projection.providerThreads.some((thread) => (thread.pendingBackgroundTasks?.length ?? 0) > 0); - const questionRuns = new Set( - projection.messages - .filter( - ({ notification }) => - notification?.source.kind === "delegated_task" && notification.outcome === "updated", - ) - .map((message) => message.runId), - ); + const questionRuns = delegatedQuestionRunIds(projection); const resultRun = workRuns .filter( (run) => From ee1fd53c0d1354b4f31d3ff0f7365a698ddfcfcd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 22:39:04 +0000 Subject: [PATCH 07/15] fix(server): retain question turns that observe a child result --- .../src/mcp/OrchestratorMcpService.test.ts | 34 +++++++++++++++---- apps/server/src/mcp/OrchestratorMcpService.ts | 2 +- .../SubagentProjection.test.ts | 26 +++++++++++--- .../src/server/subagentProjection.ts | 20 +++++++++++ 4 files changed, 70 insertions(+), 12 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 7732968711837..b4c9217a51220 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -401,9 +401,14 @@ describe("OrchestratorMcpService", () => { }), ); - it.effect.each([0, 2])( - "keeps a question wake with %s restarts from replacing the child's latest result", - (restarts) => + it.effect.each([ + { restarts: 0, observed: false }, + { restarts: 2, observed: false }, + { restarts: 0, observed: true }, + { restarts: 2, observed: true }, + ])( + "selects the task result for question wake $restarts restarts, observed $observed", + ({ restarts, observed }) => Effect.gen(function* () { const parentThreadId = ThreadId.make("thread:mcp-stale-question-parent"); const childThreadId = ThreadId.make("thread:mcp-stale-question-child"); @@ -411,6 +416,11 @@ describe("OrchestratorMcpService", () => { const taskId = NodeId.make("node:mcp-stale-question-task"); const resultRunId = RunId.make("run:mcp-stale-question-result"); const wakeRunId = RunId.make("run:mcp-stale-question-wake"); + const expectedRunId = observed + ? restarts === 0 + ? wakeRunId + : RunId.make(`run:mcp-question-restart:${restarts - 1}`) + : resultRunId; const completedAt = DateTime.makeUnsafe("2026-10-26T03:06:53.000Z"); const wakeCompletedAt = DateTime.makeUnsafe("2026-10-26T03:07:09.000Z"); const parentProjection = { @@ -488,14 +498,21 @@ describe("OrchestratorMcpService", () => { }, { id: MessageId.make("message:mcp-stale-question-answer"), - runId: wakeRunId, + runId: observed ? expectedRunId : wakeRunId, role: "assistant", text: "Answered the delayed question.", createdAt: wakeCompletedAt, updatedAt: wakeCompletedAt, }, ], - subagents: [], + subagents: observed + ? [ + { + status: "completed", + completionDelivery: { state: "acknowledged", observedByRunId: wakeRunId }, + }, + ] + : [], providerThreads: [], runtimeRequests: [], nodes: [], @@ -536,8 +553,11 @@ describe("OrchestratorMcpService", () => { return yield* service.taskStatus(scope, taskId); }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); assert.equal(result.summary, "Opened the fix PR."); - assert.equal(result.latestTerminalRunId, resultRunId); - assert.isNull(result.latestTerminalSummary); + assert.equal(result.latestTerminalRunId, expectedRunId); + assert.equal( + result.latestTerminalSummary, + observed ? "Answered the delayed question." : null, + ); }), ); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 4965225666782..838267e880e8e 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -634,7 +634,7 @@ function pageIncludesTerminalTaskResult(input: { } function latestTerminalResultRun( - projection: Pick, + projection: Pick, delegatedRun: OrchestrationV2Run | undefined, ): OrchestrationV2Run | undefined { const notificationRunIds = new Set([ diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index fb1b1b401c638..65a3bd4dc78e2 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -263,13 +263,31 @@ it("waits for nested work and retains the report across monitor acknowledgements assert.equal(progress.resultRun?.id, report.id); }); -it.each([0, 2])( - "retains the task result after a question wake with %s restart continuations", - (restarts) => { +it.each([ + [0, false], + [2, false], + [0, true], + [2, true], +] as const)( + "selects the task result after a question wake with %s restarts (observed result: %s)", + (restarts, observed) => { const { projection, run } = taskFixture(); const wake = { ...run, id: RunId.make("question-wake"), ordinal: 2 }; + const expectedRunId = observed + ? restarts === 0 + ? wake.id + : RunId.make(`question-restart:${restarts - 1}`) + : run.id; const progress = delegatedTaskProgress({ ...projection, + subagents: observed + ? [ + { + status: "completed", + completionDelivery: { state: "acknowledged", observedByRunId: wake.id }, + }, + ] + : [], runs: [ run, wake, @@ -297,7 +315,7 @@ it.each([0, 2])( ], }); assert.equal(progress.state, "result_available"); - assert.equal(progress.resultRun?.id, run.id); + assert.equal(progress.resultRun?.id, expectedRunId); }, ); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 491080c1ab41d..c1c88178f4f4a 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -225,6 +225,9 @@ export function delegatedQuestionRunIds(projection: { readonly messages: ReadonlyArray< Pick >; + readonly subagents: ReadonlyArray< + Pick + >; }) { const questionRuns = new Set( projection.messages.flatMap(({ runId, notification }) => @@ -251,6 +254,23 @@ export function delegatedQuestionRunIds(projection: { pending.push(continuationId); } } + // A question turn can finish the task when it observes a child result. + const observed = new Set( + projection.subagents.flatMap((task) => + task.completionDelivery?.observedByRunId == null + ? [] + : [task.completionDelivery.observedByRunId], + ), + ); + const resultRuns = [...observed]; + for (const runId of resultRuns) { + questionRuns.delete(runId); + for (const continuationId of continuations.get(runId) ?? []) { + if (observed.has(continuationId)) continue; + observed.add(continuationId); + resultRuns.push(continuationId); + } + } return questionRuns; } From 159dce44f12d5ec87833a2b52fabc36ee545aefb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 28 Oct 2026 00:24:18 +0000 Subject: [PATCH 08/15] fix(server): keep stale question wakes from failing queued runs Starting a parent's next queued run reads each queued question wake's child thread; a failed read now logs and treats the wake as live instead of failing the user's unrelated queued run. A stale wake dispatch now records an accepted receipt with the no-op metadata event, as stale restart continuations do, and each queued wake cancel failure is logged without aborting the remaining ones. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 1 + .../DelegatedTaskQuestionReactor.ts | 22 ++++++++++++----- .../Orchestrator.control-reads.test.ts | 21 ++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 24 +++++++++++++++++-- 4 files changed, 60 insertions(+), 8 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index a7a2854e9cd8d..7f96c85998669 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -544,6 +544,7 @@ describe("OrchestratorMcpService", () => { }), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), ); const scope: McpInvocationScope = { diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 9c540e02afe5f..1ccf7c03736ad 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -72,12 +72,22 @@ const make = Effect.gen(function* () { (run) => run.status === "queued" && run.userMessageId === messageId, ); if (queued === undefined) continue; - yield* engine.dispatch({ - type: "queued-run.cancel", - commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`), - threadId: parentThreadId, - runId: queued.id, - }); + yield* engine + .dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`), + threadId: parentThreadId, + runId: queued.id, + }) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("cancelling a delegated task question wake failed", { + threadId: parentThreadId, + requestId, + cause, + }), + ), + ); } }); diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 370b556291f83..33f32f77b00d3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1343,6 +1343,27 @@ it.effect.each([false, true])( }; yield* eventSink.write({ events: [delayedQuestion] }); yield* reactor.handle(delayedQuestion); + // A stale delivery is accepted without starting work, not rejected. + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("stale-question-redelivery"), + threadId: parentThreadId, + messageId: MessageId.make("message:stale-question-redelivery"), + text: "Answer the delegated question", + notification: { + source: { + kind: "delegated_task", + taskIds: [task.id], + childThreadId: task.childThreadId!, + }, + outcome: "updated", + summary: "Delegated task has a question", + }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + }); assert.equal( (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.length, before, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcf1afebaf2b5..0aaa1622e83c8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1516,7 +1516,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); if ( message === undefined || - !(yield* isStaleQuestionWake(projection, message.id, message.notification)) + !(yield* isStaleQuestionWake(projection, message.id, message.notification).pipe( + // A child thread read must not fail the parent's unrelated next queued run. + Effect.catchTags({ + OrchestratorProjectionError: (cause) => + Effect.logWarning("Failed to check a queued task question wake", { + runId: run.id, + cause, + }).pipe(Effect.as(false)), + }), + )) ) continue; yield* emitQueuedRunCancellation({ @@ -5069,7 +5078,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio effects: Ref.Ref>, ) { let projection = yield* getProjectionWithPendingEvents(command.threadId, events); - if (yield* isStaleQuestionWake(projection, command.messageId, command.notification)) return; + if (yield* isStaleQuestionWake(projection, command.messageId, command.notification)) { + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + occurredAt: yield* DateTime.now, + payload: projection.thread, + }); + return; + } if (command.manualContinuationOfRunId !== undefined) { const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId); const limited = latestRootProviderFailure(source ?? null, projection.turnItems); From ae2aa414e35b244509d5b7020ea772a532ec7f8d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 28 Oct 2026 00:39:13 +0000 Subject: [PATCH 09/15] fix(server): question turns that receive a child result inline keep their report A delegated completion steered into a running question turn records delivery without observedByRunId, so result selection skipped that turn's final report. Any run holding a delegated completion message now counts as having observed a child result, along with its restart continuations. Co-Authored-By: Claude Opus 5.5 --- .../SubagentProjection.test.ts | 65 +++++++++++++------ .../src/server/subagentProjection.ts | 16 +++-- 2 files changed, 55 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index 65a3bd4dc78e2..a74ee15c72782 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -264,30 +264,42 @@ it("waits for nested work and retains the report across monitor acknowledgements }); it.each([ - [0, false], - [2, false], - [0, true], - [2, true], + [0, "unobserved"], + [2, "unobserved"], + [0, "acknowledged"], + [2, "acknowledged"], + [0, "delivered inline"], + [2, "delivered inline"], ] as const)( - "selects the task result after a question wake with %s restarts (observed result: %s)", - (restarts, observed) => { + "selects the task result after a question wake with %s restarts (child result: %s)", + (restarts, observation) => { const { projection, run } = taskFixture(); const wake = { ...run, id: RunId.make("question-wake"), ordinal: 2 }; - const expectedRunId = observed - ? restarts === 0 - ? wake.id - : RunId.make(`question-restart:${restarts - 1}`) - : run.id; + const grandchildId = NodeId.make("grandchild"); + const expectedRunId = + observation === "unobserved" + ? run.id + : restarts === 0 + ? wake.id + : RunId.make(`question-restart:${restarts - 1}`); const progress = delegatedTaskProgress({ ...projection, - subagents: observed - ? [ - { - status: "completed", - completionDelivery: { state: "acknowledged", observedByRunId: wake.id }, - }, - ] - : [], + subagents: + observation === "acknowledged" + ? [ + { + status: "completed", + completionDelivery: { state: "acknowledged", observedByRunId: wake.id }, + }, + ] + : observation === "delivered inline" + ? [ + { + status: "completed", + completionDelivery: { state: "delivered", observedByRunId: null }, + }, + ] + : [], runs: [ run, wake, @@ -305,13 +317,26 @@ it.each([ notification: { source: { kind: "delegated_task", - taskIds: [NodeId.make("grandchild")], + taskIds: [grandchildId], childThreadId: ThreadId.make("grandchild-thread"), }, outcome: "updated", summary: "Delegated task has a question", }, }, + // A completion steered into the running question turn joins that run. + ...(observation === "delivered inline" + ? [ + { + runId: wake.id, + delegatedCompletion: { + parentRunId: run.id, + generation: 1, + taskIds: [grandchildId], + }, + }, + ] + : []), ], }); assert.equal(progress.state, "result_available"); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index c1c88178f4f4a..c82b6f25eec59 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -223,7 +223,7 @@ export function subagentResultForRun( export function delegatedQuestionRunIds(projection: { readonly runs: ReadonlyArray>; readonly messages: ReadonlyArray< - Pick + Pick >; readonly subagents: ReadonlyArray< Pick @@ -254,14 +254,18 @@ export function delegatedQuestionRunIds(projection: { pending.push(continuationId); } } - // A question turn can finish the task when it observes a child result. - const observed = new Set( - projection.subagents.flatMap((task) => + // A question turn can finish the task when it observes a child result, by + // reading it or by receiving its completion steered into the turn. + const observed = new Set([ + ...projection.subagents.flatMap((task) => task.completionDelivery?.observedByRunId == null ? [] : [task.completionDelivery.observedByRunId], ), - ); + ...projection.messages.flatMap(({ runId, delegatedCompletion }) => + runId === null || delegatedCompletion === undefined ? [] : [runId], + ), + ]); const resultRuns = [...observed]; for (const runId of resultRuns) { questionRuns.delete(runId); @@ -277,7 +281,7 @@ export function delegatedQuestionRunIds(projection: { export function delegatedTaskProgress(projection: { readonly runs: OrchestrationV2ThreadProjection["runs"]; readonly messages: ReadonlyArray< - Pick + Pick >; readonly subagents: ReadonlyArray< Pick From 74000dfce30c7e7a894da9798340ea6fecbe21ee Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 28 Oct 2026 00:49:36 +0000 Subject: [PATCH 10/15] fix(server): disposing an observed task keeps its observing run A question turn that read a child's result through task_status stays the task's result after the agent cancels that finished child: disposal now keeps the recorded observer instead of clearing it. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/DelegatedCompletionDelivery.test.ts | 6 +++--- apps/server/src/orchestration-v2/Orchestrator.ts | 3 ++- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 77e4cd11ba071..b907226d160cd 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -949,7 +949,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { afterDispose.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery, { state: "disposed", - observedByRunId: null, + observedByRunId: runId, }, ); @@ -968,7 +968,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { } assert.deepEqual(acknowledgedTask.event.payload.completionDelivery, { state: "disposed", - observedByRunId: null, + observedByRunId: runId, }); assert.equal(acknowledgeAfterDispose.storedEvents.length, 1); @@ -978,7 +978,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ?.completionDelivery, { state: "disposed", - observedByRunId: null, + observedByRunId: runId, }, ); }), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0aaa1622e83c8..0e5bb31a8ddee 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2371,10 +2371,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...task, completionDelivery: { state, + // Disposal keeps the observer: that run stays eligible as the task's result. observedByRunId: command.type === "delegated_task.completion-delivery.acknowledge" ? command.observedByRunId - : null, + : (task.completionDelivery?.observedByRunId ?? null), }, updatedAt: now, }; From 5c8839eed580063bf94d30dae28ed6cb8407ae4b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 28 Oct 2026 00:53:07 +0000 Subject: [PATCH 11/15] fix(server): leave result selection to main now that stale question wakes cannot start Stale delegated-task question wakes are now cancelled when the task settles or the question resolves, and dropped if they reach dispatch or the queue, so any question turn that starts is real work. Excluding question turns from a task's result misselected the outcome when such a turn received a child result inline, observed a later-disposed child, or cancelled its sub-delegate and finished the task itself. Restore main's result selection and disposal. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 161 ------------------ apps/server/src/mcp/OrchestratorMcpService.ts | 12 +- .../DelegatedCompletionDelivery.test.ts | 6 +- .../src/orchestration-v2/Orchestrator.ts | 3 +- .../SubagentProjection.test.ts | 81 --------- .../src/server/subagentProjection.ts | 69 +------- 6 files changed, 11 insertions(+), 321 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 7f96c85998669..ead5229f86502 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -412,167 +412,6 @@ describe("OrchestratorMcpService", () => { }), ); - it.effect.each([ - { restarts: 0, observed: false }, - { restarts: 2, observed: false }, - { restarts: 0, observed: true }, - { restarts: 2, observed: true }, - ])( - "selects the task result for question wake $restarts restarts, observed $observed", - ({ restarts, observed }) => - Effect.gen(function* () { - const parentThreadId = ThreadId.make("thread:mcp-stale-question-parent"); - const childThreadId = ThreadId.make("thread:mcp-stale-question-child"); - const grandchildThreadId = ThreadId.make("thread:mcp-stale-question-grandchild"); - const taskId = NodeId.make("node:mcp-stale-question-task"); - const resultRunId = RunId.make("run:mcp-stale-question-result"); - const wakeRunId = RunId.make("run:mcp-stale-question-wake"); - const expectedRunId = observed - ? restarts === 0 - ? wakeRunId - : RunId.make(`run:mcp-question-restart:${restarts - 1}`) - : resultRunId; - const completedAt = DateTime.makeUnsafe("2026-10-26T03:06:53.000Z"); - const wakeCompletedAt = DateTime.makeUnsafe("2026-10-26T03:07:09.000Z"); - const parentProjection = { - thread: { id: parentThreadId }, - runs: [], - contextTransfers: [], - subagents: [ - { - id: taskId, - threadId: parentThreadId, - origin: "app_owned", - childThreadId, - driver: "codex", - model: "gpt-5.6-terra", - status: "completed", - result: "Opened the fix PR.", - completionDelivery: { state: "pending" }, - }, - ], - } as unknown as OrchestrationV2ThreadProjection; - const childProjection = { - thread: { id: childThreadId, title: "Issue lead" }, - runs: [ - { - id: resultRunId, - ordinal: 1, - status: "completed", - delegatedTaskId: taskId, - startedAt: completedAt, - completedAt, - }, - { - id: wakeRunId, - ordinal: 2, - status: "completed", - startedAt: wakeCompletedAt, - completedAt: wakeCompletedAt, - }, - ...Array.from({ length: restarts }, (_, index) => ({ - id: RunId.make(`run:mcp-question-restart:${index}`), - ordinal: 3 + index, - status: "completed", - startedAt: wakeCompletedAt, - completedAt: wakeCompletedAt, - restartContinuationOfRunId: - index === 0 ? wakeRunId : RunId.make(`run:mcp-question-restart:${index - 1}`), - })), - ], - contextTransfers: [], - messages: [ - { - id: MessageId.make("message:mcp-stale-question-result"), - runId: resultRunId, - role: "assistant", - text: "Opened the fix PR.", - createdAt: completedAt, - updatedAt: completedAt, - }, - { - id: MessageId.make("message:mcp-stale-question-wake"), - runId: wakeRunId, - role: "user", - text: "Delegated task asked a question and is still working.", - notification: { - source: { - kind: "delegated_task", - taskIds: [NodeId.make("node:mcp-stale-question-grandchild-task")], - childThreadId: grandchildThreadId, - }, - outcome: "updated", - summary: "Delegated task has a question", - }, - createdAt: wakeCompletedAt, - updatedAt: wakeCompletedAt, - }, - { - id: MessageId.make("message:mcp-stale-question-answer"), - runId: observed ? expectedRunId : wakeRunId, - role: "assistant", - text: "Answered the delayed question.", - createdAt: wakeCompletedAt, - updatedAt: wakeCompletedAt, - }, - ], - subagents: observed - ? [ - { - status: "completed", - completionDelivery: { state: "acknowledged", observedByRunId: wakeRunId }, - }, - ] - : [], - providerThreads: [], - runtimeRequests: [], - nodes: [], - turnItems: [], - } as unknown as OrchestrationV2ThreadProjection; - const layerDependencies = Layer.mergeAll( - NodeServices.layer, - Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadRecords: (threadId) => - Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection), - getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), - dispatch: () => Effect.succeed({} as never), - }), - ServerSettings.layerTest(), - Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), - Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ - list: () => Effect.succeed([]), - }), - Layer.mock(ProjectService.ProjectService)({}), - Layer.mock(SecretRequests.SecretRequests)({}), - Layer.mock(RemoteDelegation.RemoteDelegation)({}), - Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), - ); - const scope: McpInvocationScope = { - environmentId: EnvironmentId.make("environment:mcp-stale-question"), - requestNamespace: "provider-session:mcp-stale-question", - thread: { - threadId: parentThreadId, - providerSessionId: "provider-session:mcp-stale-question", - providerInstanceId: ProviderInstanceId.make("codex"), - }, - client: undefined, - capabilities: new Set(["orchestration"]), - issuedAt: 1, - }; - - const result = yield* Effect.gen(function* () { - const service = yield* OrchestratorMcpService.OrchestratorMcpService; - return yield* service.taskStatus(scope, taskId); - }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); - assert.equal(result.summary, "Opened the fix PR."); - assert.equal(result.latestTerminalRunId, expectedRunId); - assert.equal( - result.latestTerminalSummary, - observed ? "Answered the delayed question." : null, - ); - }), - ); - it.effect("does not dispose delivery when a nonterminal task has no active child run", () => Effect.gen(function* () { const parentThreadId = ThreadId.make("thread:mcp-cancel-parent"); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 8a51a446cd286..ea9e8a055641b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -87,7 +87,6 @@ import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterReg import { subagentResultForRun, delegatedTaskProgress, - delegatedQuestionRunIds, } from "@t3tools/provider-core/server/subagentProjection"; import { DispatchModeLimit, @@ -604,20 +603,19 @@ function pageIncludesTerminalTaskResult(input: { } function latestTerminalResultRun( - projection: Pick, + projection: Pick, delegatedRun: OrchestrationV2Run | undefined, ): OrchestrationV2Run | undefined { - const notificationRunIds = new Set([ - ...delegatedQuestionRunIds(projection), - ...projection.messages + const monitorRunIds = new Set( + projection.messages .filter((message) => message.notification?.source.kind === "monitor") .map((message) => message.runId), - ]); + ); return projection.runs .filter( (run) => ThreadManagementService.isTerminalRunStatus(run.status) && - !notificationRunIds.has(run.id) && + !monitorRunIds.has(run.id) && run.status !== "rolled_back" && (run.id === delegatedRun?.id || run.startedAt !== null), ) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index b907226d160cd..77e4cd11ba071 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -949,7 +949,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { afterDispose.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery, { state: "disposed", - observedByRunId: runId, + observedByRunId: null, }, ); @@ -968,7 +968,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { } assert.deepEqual(acknowledgedTask.event.payload.completionDelivery, { state: "disposed", - observedByRunId: runId, + observedByRunId: null, }); assert.equal(acknowledgeAfterDispose.storedEvents.length, 1); @@ -978,7 +978,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ?.completionDelivery, { state: "disposed", - observedByRunId: runId, + observedByRunId: null, }, ); }), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0e5bb31a8ddee..0aaa1622e83c8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2371,11 +2371,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...task, completionDelivery: { state, - // Disposal keeps the observer: that run stays eligible as the task's result. observedByRunId: command.type === "delegated_task.completion-delivery.acknowledge" ? command.observedByRunId - : (task.completionDelivery?.observedByRunId ?? null), + : null, }, updatedAt: now, }; diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index a74ee15c72782..d3506649f6a62 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -263,87 +263,6 @@ it("waits for nested work and retains the report across monitor acknowledgements assert.equal(progress.resultRun?.id, report.id); }); -it.each([ - [0, "unobserved"], - [2, "unobserved"], - [0, "acknowledged"], - [2, "acknowledged"], - [0, "delivered inline"], - [2, "delivered inline"], -] as const)( - "selects the task result after a question wake with %s restarts (child result: %s)", - (restarts, observation) => { - const { projection, run } = taskFixture(); - const wake = { ...run, id: RunId.make("question-wake"), ordinal: 2 }; - const grandchildId = NodeId.make("grandchild"); - const expectedRunId = - observation === "unobserved" - ? run.id - : restarts === 0 - ? wake.id - : RunId.make(`question-restart:${restarts - 1}`); - const progress = delegatedTaskProgress({ - ...projection, - subagents: - observation === "acknowledged" - ? [ - { - status: "completed", - completionDelivery: { state: "acknowledged", observedByRunId: wake.id }, - }, - ] - : observation === "delivered inline" - ? [ - { - status: "completed", - completionDelivery: { state: "delivered", observedByRunId: null }, - }, - ] - : [], - runs: [ - run, - wake, - ...Array.from({ length: restarts }, (_, index) => ({ - ...wake, - id: RunId.make(`question-restart:${index}`), - ordinal: 3 + index, - restartContinuationOfRunId: - index === 0 ? wake.id : RunId.make(`question-restart:${index - 1}`), - })), - ], - messages: [ - { - runId: wake.id, - notification: { - source: { - kind: "delegated_task", - taskIds: [grandchildId], - childThreadId: ThreadId.make("grandchild-thread"), - }, - outcome: "updated", - summary: "Delegated task has a question", - }, - }, - // A completion steered into the running question turn joins that run. - ...(observation === "delivered inline" - ? [ - { - runId: wake.id, - delegatedCompletion: { - parentRunId: run.id, - generation: 1, - taskIds: [grandchildId], - }, - }, - ] - : []), - ], - }); - assert.equal(progress.state, "result_available"); - assert.equal(progress.resultRun?.id, expectedRunId); - }, -); - it("exposes the provider failure rather than a progress message from the failed run", () => { const { projection, run } = taskFixture(); const failedRun = { ...run, status: "failed" as const }; diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index c82b6f25eec59..f83a0b376233b 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -15,7 +15,6 @@ import type { ProviderInstanceId, ProviderThreadId, ProviderTurnId, - RunId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -220,68 +219,10 @@ export function subagentResultForRun( }; } -export function delegatedQuestionRunIds(projection: { - readonly runs: ReadonlyArray>; - readonly messages: ReadonlyArray< - Pick - >; - readonly subagents: ReadonlyArray< - Pick - >; -}) { - const questionRuns = new Set( - projection.messages.flatMap(({ runId, notification }) => - runId !== null && - notification?.source.kind === "delegated_task" && - notification.outcome === "updated" - ? [runId] - : [], - ), - ); - const continuations = new Map>(); - for (const run of projection.runs) { - if (run.restartContinuationOfRunId === undefined) continue; - const siblings = continuations.get(run.restartContinuationOfRunId) ?? []; - siblings.push(run.id); - continuations.set(run.restartContinuationOfRunId, siblings); - } - // Restart continuations carry system notifications instead of the original question. - const pending = [...questionRuns]; - for (const runId of pending) { - for (const continuationId of continuations.get(runId) ?? []) { - if (questionRuns.has(continuationId)) continue; - questionRuns.add(continuationId); - pending.push(continuationId); - } - } - // A question turn can finish the task when it observes a child result, by - // reading it or by receiving its completion steered into the turn. - const observed = new Set([ - ...projection.subagents.flatMap((task) => - task.completionDelivery?.observedByRunId == null - ? [] - : [task.completionDelivery.observedByRunId], - ), - ...projection.messages.flatMap(({ runId, delegatedCompletion }) => - runId === null || delegatedCompletion === undefined ? [] : [runId], - ), - ]); - const resultRuns = [...observed]; - for (const runId of resultRuns) { - questionRuns.delete(runId); - for (const continuationId of continuations.get(runId) ?? []) { - if (observed.has(continuationId)) continue; - observed.add(continuationId); - resultRuns.push(continuationId); - } - } - return questionRuns; -} - export function delegatedTaskProgress(projection: { readonly runs: OrchestrationV2ThreadProjection["runs"]; readonly messages: ReadonlyArray< - Pick + Pick >; readonly subagents: ReadonlyArray< Pick @@ -314,14 +255,8 @@ export function delegatedTaskProgress(projection: { task.completionDelivery?.state === "claimed")), ) || projection.providerThreads.some((thread) => (thread.pendingBackgroundTasks?.length ?? 0) > 0); - const questionRuns = delegatedQuestionRunIds(projection); const resultRun = workRuns - .filter( - (run) => - !questionRuns.has(run.id) && - terminal(run.status) && - (run.startedAt !== null || run.ordinal === 1), - ) + .filter((run) => terminal(run.status) && (run.startedAt !== null || run.ordinal === 1)) .toSorted((a, b) => (runRanAfter(a, b) ? -1 : runRanAfter(b, a) ? 1 : 0))[0]; return { state: From e04174384972e4f1264bff6eb941c9ea5355f259 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 28 Oct 2026 01:16:16 +0000 Subject: [PATCH 12/15] perf(server): skip child run and node reads when a settled task asked nothing Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/DelegatedTaskQuestionReactor.ts | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 1ccf7c03736ad..9fd17580e333e 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -143,11 +143,12 @@ const make = Effect.gen(function* () { return; } // A settled task can no longer use an answer to its still-pending questions. - const child = yield* projections.getThreadRecords(task.childThreadId, [ + const { runtimeRequests } = yield* projections.getThreadRecords(task.childThreadId, [ "runtimeRequests", - "runs", - "nodes", ]); + const questions = runtimeRequests.filter((request) => request.kind === "user_input"); + if (questions.length === 0) return; + const child = yield* projections.getThreadRecords(task.childThreadId, ["runs", "nodes"]); const taskRunIds = new Set( child.runs .filter( @@ -166,8 +167,8 @@ const make = Effect.gen(function* () { ); yield* cancelQueuedWakes( task.threadId, - child.runtimeRequests - .filter((request) => request.kind === "user_input" && taskNodeIds.has(request.nodeId)) + questions + .filter((request) => taskNodeIds.has(request.nodeId)) .map((request) => request.id), ); return; From adfa4df667b1fb781d7a02670984ba0a6df2ce39 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 05:55:22 +0000 Subject: [PATCH 13/15] refactor(server): drop non-null assertion in stale question wake check AI-assisted; verified with the tests and tools listed above. --- apps/server/src/orchestration-v2/Orchestrator.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 099c6ec39aabd..114552187776b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1893,17 +1893,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return false; const source = notification.source; const task = projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)); + const childThreadId = task?.childThreadId ?? null; if ( task === undefined || - task.childThreadId === null || + childThreadId === null || ["completed", "failed", "cancelled", "interrupted"].includes(task.status) ) return true; const child = yield* projectionStore - .getThreadRecords(task.childThreadId, ["runtimeRequests"]) + .getThreadRecords(childThreadId, ["runtimeRequests"]) .pipe( Effect.mapError( - (cause) => new OrchestratorProjectionError({ threadId: task.childThreadId!, cause }), + (cause) => new OrchestratorProjectionError({ threadId: childThreadId, cause }), ), ); return !child.runtimeRequests.some( From 62e0657bfb111dbe6ced75cd4a98bd1a26282169 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 06:44:56 +0000 Subject: [PATCH 14/15] fix(server): keep question wakes for tasks the parent does not track AI-assisted; verified with the tests and tools listed above. --- apps/server/src/orchestration-v2/Orchestrator.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 114552187776b..e4f99e99eaef9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1893,9 +1893,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return false; const source = notification.source; const task = projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)); - const childThreadId = task?.childThreadId ?? null; + // An unknown task gives no evidence the question is obsolete. + if (task === undefined) return false; + const childThreadId = task.childThreadId; if ( - task === undefined || childThreadId === null || ["completed", "failed", "cancelled", "interrupted"].includes(task.status) ) From 7d53c49d1575ad286ce065fec2eeacec79cb0be6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 06:59:31 +0000 Subject: [PATCH 15/15] fix(server): validate question wakes of tasks whose result was handed over AI-assisted; verified with the tests and tools listed above. --- .../Orchestrator.control-reads.test.ts | 8 ++++++++ apps/server/src/orchestration-v2/Orchestrator.ts | 11 +++++++++-- 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index c6fcfad1d1133..32f66cbdfb911 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1490,6 +1490,14 @@ it.effect.each([false, true])( }); yield* reactor.handle(childCompleted); if (missedEvent) { + // An acknowledged result drops the task from the parent's command projection. + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("ack-stale-question-task"), + parentThreadId, + taskId: task.id, + observedByRunId: parentRun.id, + }); // Persisted terminal state can predate the reactor subscription after restart. yield* projections.apply({ id: EventId.make("event:stale-question-parent-finished"), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e4f99e99eaef9..e1d46939c91dc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1892,7 +1892,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (notification?.source.kind !== "delegated_task" || notification.outcome !== "updated") return false; const source = notification.source; - const task = projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)); + const threadId = projection.thread.id; + const task = + projection.subagents.find((candidate) => source.taskIds.includes(candidate.id)) ?? + // Command projections omit settled tasks whose result was already handed over. + (yield* projectionStore + .getThreadRecords(threadId, ["subagents"], { subagentIds: source.taskIds }) + .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause })))) + .subagents[0]; // An unknown task gives no evidence the question is obsolete. if (task === undefined) return false; const childThreadId = task.childThreadId; @@ -1912,7 +1919,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (request) => request.kind === "user_input" && request.status === "pending" && - messageId === `message:delegated-task-question:${projection.thread.id}:${request.id}`, + messageId === `message:delegated-task-question:${threadId}:${request.id}`, ); });