From cf467bed0fa781bc5ba3ececaf9a2623363bc491 Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Fri, 24 Jul 2026 23:10:03 -0400 Subject: [PATCH] fix(orchestrator): Wake settled parents when delegated children finish An app-owned delegated child wrote its result into the parent projection but never started a parent run, so an async delegation's result sat unread until the user's next message. Offer a provider continuation request when a delegated child terminalizes, with a per-task completionWake policy separating async delegations from blocking waits. The wake must also reach the model. ProviderContinuationRequests is a provider-native mechanism whose dispatch only triggers ingestion of output the adapter already buffered, and both ClaudeAdapterV2 and AcpAdapterV2 discard the message text when it is marked creationSource provider. An app-owned child buffers nothing, so a wake marked that way settled instantly having prompted nothing. App-owned wakes now dispatch as message_text via a shared delegatedTaskWakeRequest helper used by both producers. --- apps/server/src/mcp/OrchestratorMcpService.ts | 55 +++- ...OrchestratorMcpToolkit.integration.test.ts | 227 ++++++++++++++++- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../src/orchestration-v2/Orchestrator.ts | 234 +++++++++++++++++- .../ProviderContinuationRequests.ts | 13 + .../ProviderContinuationService.test.ts | 49 ++++ .../ProviderContinuationService.ts | 6 +- .../src/orchestration-v2/runtimeLayer.ts | 3 + .../testkit/OrchestratorScenario.ts | 1 + apps/server/src/ws.ts | 1 + .../contracts/src/orchestrationV2.test.ts | 70 ++++++ packages/contracts/src/orchestrationV2.ts | 16 ++ 12 files changed, 661 insertions(+), 16 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 291062c00a6..a2220c99c13 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1053,6 +1053,10 @@ const make = Effect.gen(function* () { modelSelection: target.modelSelection, runtimeMode, interactionMode, + // Async delegations wake the parent on every child terminal; wait + // delegations deliver through the blocking tool call, so a wake is + // only needed if the parent settled first (timeout, disconnect). + completionWake: input.mode === "wait" ? "settled_only" : "always", }) .pipe( Effect.mapError((error) => @@ -1072,18 +1076,59 @@ const make = Effect.gen(function* () { "Delegated task command did not produce a task projection.", ); } + const taskId = taskEvent.event.payload.id; if (input.mode !== "wait") { - return yield* readTask(scope, taskEvent.event.payload.id); + return yield* readTask(scope, taskId); } const timeoutMs = Math.min( MAX_WAIT_TIMEOUT_MS, Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), ); - const waited = yield* waitForTask(scope, taskEvent.event.payload.id, timeoutMs); - return Option.isSome(waited) - ? waited.value - : yield* readTask(scope, taskEvent.event.payload.id, true); + const waited = yield* waitForTask(scope, taskId, timeoutMs); + if (Option.isSome(waited)) { + return waited.value; + } + // The blocking wait timed out, so it no longer owns delivery: upgrade + // the task so a later terminal wakes the parent even mid-turn. Best + // effort; on failure the settled_only policy still wakes a settled + // parent. + yield* threadManagement + .dispatch({ + type: "delegated_task.wake-policy", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "delegate-task-wake-policy", + }), + parentThreadId: scope.threadId, + taskId, + completionWake: "always", + }) + .pipe( + // The tool result is the timed-out task either way, so failures + // stay warnings. Keep the two shapes apart: a rejected receipt + // means this exact command id already failed (a replay of a + // no-op upgrade), while anything else is a fresh dispatch fault. + Effect.catch((error) => + Effect.logWarning("orchestrator-mcp.delegate-task.wake-policy-failed", { + taskId, + outcome: + error._tag === "OrchestratorCommandPreviouslyRejectedError" + ? "previously_rejected" + : "dispatch_failed", + error, + }), + ), + Effect.catchCause((cause) => + Effect.logWarning("orchestrator-mcp.delegate-task.wake-policy-failed", { + taskId, + outcome: "defect", + cause, + }), + ), + ); + return yield* readTask(scope, taskId, true); }), taskStatus: (scope, taskId) => readTask(scope, taskId), cancelTask: (scope, input) => diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 191aee8c219..78b4d86edb0 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -52,6 +52,10 @@ import { type ProviderAdapterV2TurnInput, } from "../orchestration-v2/ProviderAdapter.ts"; import { makeLayer as makeProviderAdapterRegistryLayer } from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import { + type ProviderContinuationRequest, + ProviderContinuationRequests, +} from "../orchestration-v2/ProviderContinuationRequests.ts"; import { checkpointWorkspace } from "../orchestration-v2/testkit/ReplayFixtureWorkspace.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts"; @@ -411,6 +415,38 @@ describe("orchestrator MCP toolkit", () => { : `Claude completed: ${turn.message.text}`, }), ]); + // Captures parent-wake offers made when a delegated child + // terminalizes after the parent run settled. + const continuationOffers = yield* Ref.make>( + [], + ); + const continuationProbeLayer = Layer.succeed(ProviderContinuationRequests, { + offer: (request) => + Ref.update(continuationOffers, (existing) => [...existing, request]), + take: Effect.never, + }); + // Offers land after the finalize projection writes, so poll briefly + // instead of asserting counts immediately. + const waitForContinuationOffers = (count: number) => + Effect.gen(function* () { + for (let attempt = 0; attempt < 1_000; attempt += 1) { + const current = yield* Ref.get(continuationOffers); + if (current.length >= count) { + return current; + } + yield* Effect.sleep("5 millis"); + } + return yield* Ref.get(continuationOffers); + }); + // Absence has no event to await, so sample repeatedly instead of + // trusting a single sleep to outlast a late offer. + const expectOffersToStay = (count: number) => + Effect.gen(function* () { + for (let sample = 0; sample < 4; sample += 1) { + yield* Effect.sleep("50 millis"); + expect(yield* Ref.get(continuationOffers)).toHaveLength(count); + } + }); const orchestratorLayer = makeOrchestratorV2ReplayLayerWithRegistry( { name: "orchestrator-mcp-toolkit", @@ -425,7 +461,7 @@ describe("orchestrator MCP toolkit", () => { }, }, registryLayer, - ); + ).pipe(Layer.provide(continuationProbeLayer)); const orchestrationLayer = Layer.merge( orchestratorLayer, threadManagementServiceLayer.pipe(Layer.provide(orchestratorLayer)), @@ -781,6 +817,11 @@ describe("orchestrator MCP toolkit", () => { expect(delegatedStatus.status).toBe("completed"); expect(delegatedStatus.resultContextTransferId).not.toBeNull(); + // A wait-mode child (completionWake settled_only) that completes + // while the parent run is live does not offer a wake: the + // blocking delegate_task call above already returned the result. + yield* expectOffersToStay(0); + const repeatedDelegatedCall = yield* invoke("delegate_task", { task: delegatedPrompt, target: { @@ -886,6 +927,18 @@ describe("orchestrator MCP toolkit", () => { ).pipe(Effect.orDie); expect(cancelledStatus.status).toBe("interrupted"); + // The cancelled async child carries completionWake "always", so + // its terminal offers a wake even though the parent run is live; + // queue_after_active sequences the continuation behind it. + const offersAfterCancel = yield* waitForContinuationOffers(1); + expect(offersAfterCancel).toHaveLength(1); + expect(offersAfterCancel[0]).toMatchObject({ + threadId: parentThreadId, + delivery: "message_text", + }); + expect(offersAfterCancel[0]?.detail).toContain(cancellable.taskId); + expect(offersAfterCancel[0]?.detail).toContain("interrupted"); + const createInput = { clientRequestId: "create-thread-batch-1", threads: [ @@ -1232,6 +1285,178 @@ describe("orchestrator MCP toolkit", () => { expect( listed.threads.some((thread) => thread.relationshipToParent === "subagent"), ).toBe(false); + + // A wait-mode delegation whose blocking wait times out no longer + // owns delivery, so delegate_task upgrades the task to "always". + // Its terminal then wakes the parent even mid-turn. + const upgradedCall = yield* invoke("delegate_task", { + task: cancellationPrompt, + target: { + providerInstanceId: codexInstanceId, + model: codexModel, + }, + mode: "wait", + timeoutMs: 1, + clientRequestId: "delegate-wait-upgrade-1", + }); + const upgradedDelegated = yield* decodeDelegateTaskResult( + upgradedCall.structuredContent, + ).pipe(Effect.orDie); + expect(upgradedDelegated.status).toBe("running"); + yield* waitForProjection(orchestrator, upgradedDelegated.childThreadId, (projection) => + projection.providerTurns.some((turn) => turn.status === "running"), + ); + expect( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === upgradedDelegated.taskId, + )?.completionWake, + ).toBe("always"); + const upgradeCancelCall = yield* invoke("task_cancel", { + taskId: upgradedDelegated.taskId, + reason: "Terminalize while the parent run is live.", + clientRequestId: "cancel-wait-upgrade-1", + }); + expect(upgradeCancelCall.isError).toBe(false); + yield* waitForProjection(orchestrator, parentThreadId, (projection) => + projection.subagents.some( + (task) => task.id === upgradedDelegated.taskId && task.status === "interrupted", + ), + ); + const offersAfterUpgrade = yield* waitForContinuationOffers(2); + expect(offersAfterUpgrade).toHaveLength(2); + expect(offersAfterUpgrade[1]).toMatchObject({ + threadId: parentThreadId, + delivery: "message_text", + }); + expect(offersAfterUpgrade[1]?.detail).toContain(upgradedDelegated.taskId); + + // The MCP tool cannot force the reverse interleaving (child + // terminal before the upgrade lands), so dispatch the command + // directly. The first wait-mode delegation completed while the + // parent run was live, so finalize skipped its offer under + // settled_only; the upgrade must accept, persist the policy, and + // deliver the wake finalize declined. + const terminalUpgrade = yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("command:mcp-parent:wake-policy-terminal"), + parentThreadId, + taskId: delegated.taskId, + completionWake: "always", + }); + expect( + terminalUpgrade.storedEvents.some( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === delegated.taskId && + stored.event.payload.completionWake === "always", + ), + ).toBe(true); + expect( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === delegated.taskId, + )?.completionWake, + ).toBe("always"); + const offersAfterTerminalUpgrade = yield* waitForContinuationOffers(3); + expect(offersAfterTerminalUpgrade).toHaveLength(3); + expect(offersAfterTerminalUpgrade[2]).toMatchObject({ + threadId: parentThreadId, + delivery: "message_text", + }); + expect(offersAfterTerminalUpgrade[2]?.detail).toContain(delegated.taskId); + + // Legacy records omit completionWake and stay settled_only. The + // MCP service always sets the field now, so dispatch the request + // directly to cover the legacy shape. + if (parentRun === undefined || parentRun.rootNodeId === null) { + return yield* Effect.die(new Error("Parent run missing.")); + } + const legacyDispatch = yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:mcp-parent:delegate-legacy"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId, + task: cancellationPrompt, + modelSelection: codexSelection, + runtimeMode: "full-access", + interactionMode: "default", + }); + const legacyTaskEvent = legacyDispatch.storedEvents.find( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.origin === "app_owned", + ); + if (legacyTaskEvent?.event.type !== "subagent.updated") { + return yield* Effect.die(new Error("Legacy delegated task projection missing.")); + } + const legacyTask = legacyTaskEvent.event.payload; + expect(legacyTask.completionWake).toBeUndefined(); + if (legacyTask.childThreadId === null) { + return yield* Effect.die(new Error("Legacy delegated task child thread missing.")); + } + const legacyChildThreadId = legacyTask.childThreadId; + yield* waitForProjection(orchestrator, legacyChildThreadId, (projection) => + projection.providerTurns.some((turn) => turn.status === "running"), + ); + // Nothing terminalized here, so the offer count must hold. + yield* expectOffersToStay(3); + + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("command:mcp-parent:interrupt-wake"), + threadId: parentThreadId, + runId: parentRun.id, + reason: "Settle the parent before the legacy child terminalizes.", + }); + yield* waitForProjection(orchestrator, parentThreadId, (projection) => + projection.runs.every( + (run) => + run.status !== "preparing" && + run.status !== "starting" && + run.status !== "running", + ), + ); + const legacyCancelCall = yield* invoke("task_cancel", { + taskId: legacyTask.id, + reason: "Terminalize the legacy child after the parent settled.", + clientRequestId: "cancel-legacy-1", + }); + expect(legacyCancelCall.isError).toBe(false); + yield* waitForProjection(orchestrator, legacyChildThreadId, (projection) => + projection.runs.some((run) => run.status === "interrupted"), + ); + const offers = yield* waitForContinuationOffers(4); + expect(offers).toHaveLength(4); + expect(offers[3]).toMatchObject({ threadId: parentThreadId, delivery: "message_text" }); + expect(offers[3]?.detail).toContain(legacyTask.id); + expect(offers[3]?.detail).toContain("task_status"); + + // Same command against a terminal task whose finalize already + // offered (the parent was settled then, and still is): the policy + // must persist without a second offer, or the parent wakes twice. + const settledUpgrade = yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("command:mcp-parent:wake-policy-settled"), + parentThreadId, + taskId: legacyTask.id, + completionWake: "always", + }); + expect( + settledUpgrade.storedEvents.some( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === legacyTask.id && + stored.event.payload.completionWake === "always", + ), + ).toBe(true); + expect( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === legacyTask.id, + )?.completionWake, + ).toBe("always"); + yield* expectOffersToStay(4); }).pipe(Effect.provide(testLayer)); }), ), diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 73b0be83bdd..8cb692a182d 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -49,7 +49,7 @@ export const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities export const DelegateTaskTool = Tool.make("delegate_task", { description: - "Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Use this whenever the user asks for an agent, subagent, worker, delegated task, or parallel help—including cross-provider work. The childThreadId is backing storage, not an ordinary top-level thread. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. Prefer mode='async' and poll task_status for long work; mode='wait' blocks until completion or timeout.", + "Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Use this whenever the user asks for an agent, subagent, worker, delegated task, or parallel help—including cross-provider work. The childThreadId is backing storage, not an ordinary top-level thread. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. An async child's completion wakes this thread with a continuation message naming the task (queued behind any turn in progress), so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.", parameters: OrchestratorMcpDelegateTaskInput, success: OrchestratorMcpDelegateTaskResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3de91906479..b76a10a5e88 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -45,6 +45,10 @@ import { makeKeyedSerialExecutor } from "./KeyedSerialExecutor.ts"; import { applyToProjection, emptyProjection, ProjectionStoreV2 } from "./ProjectionStore.ts"; import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; +import { + type ProviderContinuationRequest, + ProviderContinuationRequests, +} from "./ProviderContinuationRequests.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts"; import { RuntimePolicyV2 } from "./RuntimePolicy.ts"; @@ -197,6 +201,7 @@ function commandThreadId(command: OrchestrationV2Command): ThreadId { case "provider.switch": return command.threadId; case "delegated_task.request": + case "delegated_task.wake-policy": case "thread.created.record": return command.parentThreadId; case "thread.fork": @@ -220,6 +225,60 @@ function isBlockingRun(run: OrchestrationV2Run): boolean { ); } +/** + * A parent thread is "live" for wake purposes while a run is still producing + * agent output. A run parked at "waiting" is post-terminal drain, so its agent + * turn is over and a wake is still needed. + */ +function hasLiveRun(projection: OrchestrationV2ThreadProjection): boolean { + return projection.runs.some( + (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", + ); +} + +function delegatedTaskWakeDetail( + task: Pick, +): string { + const label = task.title === null ? task.id : `"${task.title}"`; + return task.status === "completed" + ? `Delegated task ${label} completed. Use task_status with taskId ${task.id} to read the result.` + : `Delegated task ${label} ended with status ${task.status}. Use task_status with taskId ${task.id} for details.`; +} + +/** + * Both app-owned wake producers must go through this. An app-owned child leaves + * nothing buffered in the adapter, so the detail text is the entire wake and has + * to reach the provider as a real prompt. Omitting `delivery` here would mark + * the dispatch as an adapter-buffered wake, which ClaudeAdapterV2 and + * AcpAdapterV2 both answer by discarding the text and settling the turn against + * an empty buffer. + */ +function delegatedTaskWakeRequest(input: { + readonly threadId: ThreadId; + readonly providerThread: Pick< + OrchestrationV2ThreadProjection["providerThreads"][number], + "id" | "driver" + >; + readonly task: Pick; +}): ProviderContinuationRequest { + return { + threadId: input.threadId, + providerThreadId: input.providerThread.id, + driver: input.providerThread.driver, + detail: delegatedTaskWakeDetail(input.task), + delivery: "message_text", + }; +} + +function isTerminalDelegatedTaskStatus(status: OrchestrationV2Subagent["status"]): boolean { + return ( + status === "completed" || + status === "failed" || + status === "cancelled" || + status === "interrupted" + ); +} + function delegatedTaskTerminalStatus( status: OrchestrationV2Run["status"], ): OrchestrationV2Subagent["status"] | null { @@ -410,6 +469,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const idAllocator = yield* IdAllocatorV2; const projectionStore = yield* ProjectionStoreV2; const providerAdapters = yield* ProviderAdapterRegistryV2; + const continuationRequests = yield* ProviderContinuationRequests; const providerSessions = yield* ProviderSessionManagerV2; const providerSwitchService = yield* ProviderSwitchServiceV2; const runtimePolicy = yield* RuntimePolicyV2; @@ -3702,6 +3762,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio prompt: command.task, title: command.title ?? null, model: command.modelSelection.model, + ...(command.completionWake === undefined ? {} : { completionWake: command.completionWake }), status: "running", result: null, startedAt: now, @@ -3853,6 +3914,98 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ); + // Rewrites a delegated task's completionWake after creation. The wait path + // uses this when its blocking window ends without a terminal (timeout), so + // a child that later terminalizes mid-parent-turn still wakes the parent. + // Runs under the parent thread's dispatch lock, which is also what finalize + // takes for its parent-side writes, so the two never interleave on this row. + const dispatchDelegatedTaskWakePolicy = Effect.fn( + "orchestrationV2.dispatch.delegatedTaskWakePolicy", + )(function* ( + command: Extract, + events: Ref.Ref>, + ) { + const parentProjection = yield* projectionStore + .getThreadProjection(command.parentThreadId) + .pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: command.parentThreadId, + cause, + }), + ), + ); + const task = parentProjection.subagents.find( + (candidate) => candidate.id === command.taskId && candidate.origin === "app_owned", + ); + // No-op commands reject with a descriptive cause, matching the thread + // mutation handlers ("already archived", "not archived"). + if (task === undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Delegated task ${command.taskId} is not an app-owned task of thread ${command.parentThreadId}.`, + }); + } + if (task.completionWake === command.completionWake) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Delegated task ${command.taskId} already wakes the parent with completionWake ${command.completionWake}.`, + }); + } + const now = yield* DateTime.now; + const emitEvent = emit(events, command); + yield* emitEvent({ + type: "subagent.updated", + threadId: command.parentThreadId, + ...(task.runId === null ? {} : { runId: task.runId }), + nodeId: task.id, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + occurredAt: now, + payload: { ...task, completionWake: command.completionWake, updatedAt: now }, + }); + // A non-terminal task needs no offer here: finalize reads the upgraded + // policy when the child terminalizes. Both writers hold this parent lock, + // so a terminal task means finalize already committed the terminal row and + // already made its offer decision under the pre-upgrade policy: under + // settled_only it offered iff the parent had no live run. Offer here only + // when the parent has a live run now, which is precisely the case where + // finalize skipped; queue_after_active then sequences the wake behind that + // run. When the parent is not live, finalize already offered and a second + // offer would wake the parent twice. (If the parent settled in between, + // this skips a wake that finalize also skipped; a missed wake is cheaper + // than a duplicate one, and the result is already in the projection.) + if (command.completionWake !== "always" || !isTerminalDelegatedTaskStatus(task.status)) { + return; + } + if (!hasLiveRun(parentProjection)) { + return; + } + const parentRun = + task.runId === null + ? undefined + : parentProjection.runs.find((candidate) => candidate.id === task.runId); + const parentProviderThread = + parentRun?.providerThreadId === null || parentRun?.providerThreadId === undefined + ? undefined + : parentProjection.providerThreads.find( + (candidate) => candidate.id === parentRun.providerThreadId, + ); + if (parentProviderThread === undefined) { + return; + } + yield* continuationRequests.offer( + delegatedTaskWakeRequest({ + threadId: command.parentThreadId, + providerThread: parentProviderThread, + task, + }), + ); + }); + const dispatchCreatedThreadRecord = Effect.fn("orchestrationV2.dispatch.createdThreadRecord")( function* ( command: Extract, @@ -4893,6 +5046,31 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]); }); + /** + * Parent thread of an app-owned delegated child, or undefined when the + * thread is not one. Thread lineage and fork origin are immutable, so this + * is safe to read without holding either thread's dispatch lock. + */ + const appOwnedSubagentParentThreadId = (childThreadId: ThreadId) => + Effect.gen(function* () { + const childProjection = yield* projectionStore.getThreadProjection(childThreadId); + const lineage = childProjection.thread.lineage; + return lineage.relationshipToParent === "subagent" && + lineage.parentThreadId !== null && + childProjection.thread.forkedFrom?.type === "node" + ? lineage.parentThreadId + : undefined; + }); + + /** + * Transfers a terminal child's result into its parent and offers the parent + * wake. Every mutation here targets the PARENT thread, so callers must hold + * the parent thread's dispatch lock rather than the child's: the + * delegated_task.wake-policy handler rewrites the same subagent row under + * that lock with a full-row payload, and unserialized writers clobber each + * other (stale policy on the terminal row, or a terminal row regressed to + * running). + */ const finalizeAppOwnedSubagent = (childThreadId: ThreadId) => Effect.gen(function* () { const childProjection = yield* projectionStore.getThreadProjection(childThreadId); @@ -5094,6 +5272,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: resultTransfer, }, ]); + + // Nothing else re-invokes a parent that already settled: the child's + // result lands in the projection above, but no parent run starts to let + // the agent act on it. Offer a continuation (dispatched + // queue_after_active) so the parent wakes and collects the result. + // Policy comes from the task's completionWake: "always" (async + // delegations) offers on every terminal, sequencing behind a live + // parent run like a provider task notification; "settled_only" (wait + // delegations, and legacy records without the field) skips while the + // parent has a run in preparing/starting/running, because the blocking + // tool call already returns the result there. A run parked at + // "waiting" is post-terminal drain, so its agent turn is over and the + // wake is still needed. The result-transfer guard above makes this a + // single offer per child, replay included. + if (parentProviderThread === undefined) { + return; + } + if ((task.completionWake ?? "settled_only") === "settled_only") { + // Re-read under the parent lock so the gate sees the run states left + // by the writes above rather than the pre-write snapshot. + if (hasLiveRun(yield* projectionStore.getThreadProjection(parentThreadId))) { + return; + } + } + yield* continuationRequests.offer( + delegatedTaskWakeRequest({ + threadId: parentThreadId, + providerThread: parentProviderThread, + task: updatedTask, + }), + ); }); const dispatchUnsupported = (command: OrchestrationV2Command) => @@ -5188,6 +5397,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "delegated_task.request": yield* dispatchDelegatedTaskRequest(command, events, effects); break; + case "delegated_task.wake-policy": + yield* dispatchDelegatedTaskWakePolicy(command, events); + break; case "thread.created.record": yield* dispatchCreatedThreadRecord(command, events); break; @@ -5339,14 +5551,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio stored.event.payload.status === "rolled_back"), ), Stream.runForEach((stored) => - threadDispatch - .withLock( - stored.event.threadId, - finalizeAppOwnedSubagent(stored.event.threadId).pipe( - Effect.andThen(startNextQueuedRun(stored.event.threadId)), - ), - ) - .pipe(Effect.catchCause(() => Effect.void)), + Effect.gen(function* () { + const threadId = stored.event.threadId; + // finalize writes the parent thread and startNextQueuedRun writes this + // thread, so each takes its own thread's lock, sequentially and never + // nested: dispatchDelegatedTaskRequest already writes child events + // while holding the parent lock, so nesting the parent lock inside the + // child lock here would invert that order, and the keyed executor's + // semaphores are neither reentrant nor deadlock-aware. + const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId); + if (parentThreadId !== undefined) { + yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId)); + } + yield* threadDispatch.withLock(threadId, startNextQueuedRun(threadId)); + }).pipe(Effect.catchCause(() => Effect.void)), ), Effect.forkDetach, ); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts b/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts index 275c14bf220..5c6260baaa0 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts @@ -10,6 +10,19 @@ export interface ProviderContinuationRequest { readonly providerThreadId: ProviderThreadId; readonly driver: ProviderDriverKind; readonly detail: string | null; + /** + * How the continuation turn gets its content. + * + * `adapter_buffered` (default) is the provider-native wake: the adapter has + * already buffered the CLI's wake output, and the dispatched message only + * triggers ingestion. `ClaudeAdapterV2` deliberately discards the message + * text on that path. + * + * `message_text` is for app-owned work with no buffered provider output, such + * as a delegated child finishing. The text is the entire wake, so it must + * reach the provider as a real prompt. + */ + readonly delivery?: "adapter_buffered" | "message_text"; readonly dispatchIfCurrent?: ( effect: Effect.Effect, ) => Effect.Effect, E, R>; diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts index f3bbe364385..36e0384043d 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts @@ -82,6 +82,55 @@ function testLayer(input: { } describe("ProviderContinuationService", () => { + it.effect("marks an adapter-buffered wake with the provider creation source", () => { + return Effect.gen(function* () { + const dispatched = yield* Queue.unbounded(); + yield* Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests; + yield* requests.offer(request()); + const command = (yield* Queue.take(dispatched)) as { readonly creationSource: string }; + // ClaudeAdapterV2 keys on this to attach buffered CLI output. + assert.equal(command.creationSource, "provider"); + }).pipe( + Effect.provide( + testLayer({ dispatched, getThreadProjection: () => Effect.succeed(projection) }), + ), + Effect.scoped, + ); + }); + }); + + it.effect("delivers a message_text wake as a real prompt, not a buffered wake", () => { + return Effect.gen(function* () { + const dispatched = yield* Queue.unbounded(); + yield* Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests; + yield* requests.offer({ + threadId, + providerThreadId, + driver, + detail: "Delegated task completed.", + delivery: "message_text", + }); + const command = (yield* Queue.take(dispatched)) as { + readonly creationSource: string; + readonly text: string; + }; + // An app-owned child buffers nothing in the adapter, so this text is + // the whole wake. Marking it "provider" would make ClaudeAdapterV2 + // drop it and settle the turn having prompted nothing. + assert.notEqual(command.creationSource, "provider"); + assert.equal(command.creationSource, "server"); + assert.equal(command.text, "Delegated task completed."); + }).pipe( + Effect.provide( + testLayer({ dispatched, getThreadProjection: () => Effect.succeed(projection) }), + ), + Effect.scoped, + ); + }); + }); + it.effect("dispatches a current request exactly once", () => { return Effect.gen(function* () { const dispatched = yield* Queue.unbounded(); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index b37ec52b522..b580bec3420 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -58,7 +58,11 @@ export const workerLive = Layer.effectDiscard( attachments: [], dispatchMode: { type: "queue_after_active" }, createdBy: "agent", - creationSource: "provider", + // "provider" marks an adapter-buffered wake, which ClaudeAdapterV2 + // detects to attach the buffered CLI output and drop this text. A + // message_text wake has no buffered output, so it must not carry that + // marker or the turn settles immediately having prompted nothing. + creationSource: request.delivery === "message_text" ? "server" : "provider", }); if (request.dispatchIfCurrent === undefined) { yield* dispatch; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index c2a4a3fd5a9..565add32b84 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -190,6 +190,9 @@ const orchestratorProvided = orchestratorLayer.pipe( contextHandoffServiceProvided, idAllocatorLayer, providerAdapterRegistryProvided, + // Same layer reference as the continuation worker and the adapter + // infrastructure so layer memoization yields one shared request queue. + providerContinuationRequestsLayer, providerEventIngestorProvided, runtimePolicyProvided, providerSessionManagerProvided, diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index 8d1a791b6e4..cb348f6287a 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -122,6 +122,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray { expect(command.parentNodeId).toBe(NodeId.make("node-parent-1")); }); + it("decodes delegated task wake-policy commands", () => { + const command = decodeOrchestrationV2Command({ + type: "delegated_task.wake-policy", + commandId: "command-delegate-wake-policy-1", + parentThreadId: "thread-parent-1", + taskId: "node-subagent-1", + completionWake: "always", + }); + + expect(command.type).toBe("delegated_task.wake-policy"); + if (command.type !== "delegated_task.wake-policy") { + throw new Error("expected delegated_task.wake-policy"); + } + expect(command.parentThreadId).toBe(ThreadId.make("thread-parent-1")); + expect(command.taskId).toBe(NodeId.make("node-subagent-1")); + expect(command.completionWake).toBe("always"); + + // The policy carries the whole decision, so it is required and closed. + expect(() => + decodeOrchestrationV2Command({ + type: "delegated_task.wake-policy", + commandId: "command-delegate-wake-policy-2", + parentThreadId: "thread-parent-1", + taskId: "node-subagent-1", + }), + ).toThrow(); + expect(() => + decodeOrchestrationV2Command({ + type: "delegated_task.wake-policy", + commandId: "command-delegate-wake-policy-3", + parentThreadId: "thread-parent-1", + taskId: "node-subagent-1", + completionWake: "sometimes", + }), + ).toThrow(); + }); + it("decodes durable created-thread timeline records", () => { const command = decodeOrchestrationV2Command({ type: "thread.created.record", @@ -414,6 +451,39 @@ describe("orchestration V2 contracts", () => { expect(turnItem.progress).toBe("Inspecting package metadata"); }); + it("decodes app-owned subagent parent-wake policies", () => { + const appOwnedSubagent = { + id: "node-subagent-2", + threadId: "thread-1", + runId: "run-1", + parentNodeId: "node-root-1", + origin: "app_owned", + createdBy: "agent", + driver: "codex", + providerInstanceId: "codex", + providerThreadId: null, + childThreadId: "thread-child-1", + nativeTaskRef: null, + prompt: "Inspect the API boundary.", + title: "API inspection", + model: "gpt-5.4", + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }; + const decode = Schema.decodeUnknownSync(OrchestrationV2Subagent); + + // Legacy records predate the field and must behave as settled_only. + expect(decode(appOwnedSubagent).completionWake).toBeUndefined(); + expect(decode({ ...appOwnedSubagent, completionWake: "always" }).completionWake).toBe("always"); + expect(decode({ ...appOwnedSubagent, completionWake: "settled_only" }).completionWake).toBe( + "settled_only", + ); + expect(() => decode({ ...appOwnedSubagent, completionWake: "sometimes" })).toThrow(); + }); + it("decodes thread projections with an ordered turn item rendering stream", () => { const projection = Schema.decodeUnknownSync(OrchestrationV2ThreadProjection)({ thread: { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 186e0b36fb0..e5a419453c0 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -442,6 +442,12 @@ export const OrchestrationV2Subagent = Schema.Struct({ prompt: Schema.String, title: Schema.NullOr(Schema.String), model: Schema.NullOr(Schema.String), + // Parent-wake policy for app-owned tasks: "always" offers a continuation on + // every terminal (async delegations; queue_after_active sequences it behind + // a live parent run), "settled_only" offers only when the parent has no + // live run (wait-mode delegations, whose result returns through the + // blocking tool call). Absent on legacy records; treated as settled_only. + completionWake: Schema.optional(Schema.Literals(["always", "settled_only"])), status: Schema.Literals([ "pending", "running", @@ -1969,8 +1975,18 @@ export const OrchestrationV2Command = Schema.Union([ modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, + // Omitted behaves as "settled_only" (no wake while the parent has a live + // run); producers that want fire-and-forget wakes must set "always". + completionWake: Schema.optional(Schema.Literals(["always", "settled_only"])), createdAt: Schema.optional(Schema.DateTimeUtc), }), + Schema.Struct({ + type: Schema.Literal("delegated_task.wake-policy"), + commandId: CommandId, + parentThreadId: ThreadId, + taskId: NodeId, + completionWake: Schema.Literals(["always", "settled_only"]), + }), Schema.Struct({ type: Schema.Literal("thread.created.record"), commandId: CommandId,