diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index ebb1aa9e4c1a3f5f8b420e6eac1b3c41f5b4fb95..78d4ee48e66645e040c6de3c9cfa9a03d1d278a1 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -370,9 +370,11 @@ export function hasPendingChildRuns( childProjection: Pick, delegatedRun: OrchestrationV2Run | undefined, ): boolean { + // Held queued runs wait for the user to resume the child; task_cancel holds them. return childProjection.runs.some( (run) => !ThreadManagementService.isTerminalRunStatus(run.status) && + !(run.status === "queued" && run.queueHeld === true) && (delegatedRun === undefined || run.ordinal > delegatedRun.ordinal), ); } diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index e59f8fff9a056265b042a0b1b7b62c65f20c4d9f..0dbb2f5716837822d7c274a2ae614c11e0f6b0b2 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -1873,6 +1873,20 @@ describe("orchestrator MCP toolkit", () => { legacyDelegatedRun, ), ).toBe(true); + // A queue Stop held waits for the user; it is not pending child work. + expect( + hasPendingChildRuns( + { + ...legacyChildProjection, + runs: legacyChildProjection.runs.map((run) => + run.id === activeChildFollowup.runId + ? { ...run, status: "queued", queueHeld: true } + : run, + ), + }, + legacyDelegatedRun, + ), + ).toBe(false); const completedTaskCancelCall = yield* invoke("task_cancel", { taskId: delegated.taskId, reason: "Stop the child's later work too.", diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index e7ef07101a39ec86763aa9b0a85bf3d215815800..ac0ebb63b7ddf04fa6dfde6a0cf3de4fa3ef261e 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -76,7 +76,7 @@ export const DelegateTaskTool = Tool.make("delegate_task", { const TaskStatusTool = Tool.make("task_status", { description: - "Needs an agent running inside a T3 thread. Read a T3-owned delegated task created by this parent thread. childRunId identifies the original delegated run. workState distinguishes working, waiting_for_children, and result_available; a completed turn with live nested work is not a completed task. summary is the final task result, including provider errors on failure, and remains stable after publication. hasPendingChildRuns reports later queued or executing turns in the backing child thread, even after the task is terminal; it does not reopen the task, and task_cancel stops those turns too. latestTerminal* provides later non-monitor turn results. Reading a terminal result acknowledges its automatic parent delivery.", + "Needs an agent running inside a T3 thread. Read a T3-owned delegated task created by this parent thread. childRunId identifies the original delegated run. workState distinguishes working, waiting_for_children, and result_available; a completed turn with live nested work is not a completed task. summary is the final task result, including provider errors on failure, and remains stable after publication. hasPendingChildRuns reports later queued or executing turns in the backing child thread, even after the task is terminal; it does not reopen the task, and task_cancel stops those turns too. Turns held in a stopped queue wait for the user and do not count. latestTerminal* provides later non-monitor turn results. Reading a terminal result acknowledges its automatic parent delivery.", parameters: OrchestratorMcpTaskStatusInput, success: OrchestratorMcpDelegateTaskResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 6f1819a672b79ccbce24a8edbbc6f1e514818fe5..0344e3aed4cfe162bbc8741f302496c75be6c854 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -14,6 +14,7 @@ import { ProviderThreadId, RunId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -1204,6 +1205,197 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + it.effect("settles a cancelled child whose held wakes wait behind it", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:held-wake-parent"); + const projectId = ProjectId.make("project:held-wake-parent"); + const runId = RunId.make("run:held-wake-parent"); + const rootNodeId = NodeId.make("node:held-wake-parent-root"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:held-wake-parent-settled"), + deliveryState: "delivered", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "held-wake-child", + completionWake: "always", + continuationPending: false, + now, + }); + // Pull request watch wakes queued while the child ran; the restart held them. + const held = runEvent({ + threadId: child.childThreadId, + runId: RunId.make("run:held-wake-child:2"), + ordinal: 2, + status: "queued", + now, + }); + yield* eventSink.write({ + commandId: CommandId.make("command:held-wake-child:held"), + events: [ + { + ...held, + payload: { ...held.payload, startedAt: null, completedAt: null, queueHeld: true }, + }, + ], + }); + + yield* orchestrator.recoverDelegatedTasks; + + const recovered = yield* orchestrator.getThreadProjection(threadId); + const task = recovered.subagents.find((row) => row.id === child.taskId); + assert.equal(task?.status, "cancelled"); + assert.isNotNull(task?.result ?? null); + assert.isTrue( + recovered.contextTransfers.some( + (transfer) => + transfer.type === "subagent_result" && transfer.sourceThreadId === child.childThreadId, + ), + ); + assert.isFalse(yield* orchestrator.delegatedTaskResultPending(child.childThreadId)); + // The held wake stays held for the user to resume or discard. + const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); + assert.deepEqual( + childProjection.runs.map((run) => [run.status, run.queueHeld ?? false]), + [ + ["cancelled", false], + ["queued", true], + ], + ); + }), + ); + + it.effect("settles a child whose provider failure holds its queued wakes", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:failed-hold-parent"); + const projectId = ProjectId.make("project:failed-hold-parent"); + const runId = RunId.make("run:failed-hold-parent"); + const rootNodeId = NodeId.make("node:failed-hold-parent-root"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:failed-hold-parent-settled"), + deliveryState: "delivered", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "failed-hold-child", + completionWake: "always", + continuationPending: false, + now, + }); + // The child is resumed, and a wake queues behind its running turn. + const activeRun = runEvent({ + threadId: child.childThreadId, + runId: RunId.make("run:failed-hold-child:2"), + ordinal: 2, + status: "running", + now, + }); + const queued = runEvent({ + threadId: child.childThreadId, + runId: RunId.make("run:failed-hold-child:3"), + ordinal: 3, + status: "queued", + now, + }); + yield* eventSink.write({ + commandId: CommandId.make("command:failed-hold-child:resumed"), + events: [ + activeRun, + { ...queued, payload: { ...queued.payload, startedAt: null, completedAt: null } }, + ], + }); + const afterSequence = yield* eventSink.latestSequence(); + const errorItemId = TurnItemId.make("turn-item:failed-hold-child:error"); + yield* eventSink.write({ + commandId: CommandId.make("command:failed-hold-child:failed"), + events: [ + { + id: EventId.make("event:failed-hold-child:error"), + type: "turn-item.updated", + threadId: child.childThreadId, + runId: activeRun.runId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + id: errorItemId, + type: "error", + threadId: child.childThreadId, + runId: activeRun.runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "failed", + title: "Provider failure", + startedAt: now, + completedAt: now, + updatedAt: now, + failure: { + class: "provider_error", + message: "Provider failed.", + code: "provider_failed", + retryable: null, + }, + }, + }, + { + ...activeRun, + id: EventId.make("event:failed-hold-child:failed"), + payload: { ...activeRun.payload, status: "failed", completedAt: now }, + }, + ], + }); + + // The failure holds the queue, which leaves the failure as the task's result. + const settled = yield* eventSink + .stream({ afterSequence, eventType: "subagent.updated" }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && stored.event.payload.id === child.taskId, + ), + Stream.take(1), + Stream.runHead, + ); + assert.isTrue(settled._tag === "Some"); + const recovered = yield* orchestrator.getThreadProjection(threadId); + assert.equal(recovered.subagents.find((row) => row.id === child.taskId)?.status, "failed"); + const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); + assert.deepEqual( + childProjection.runs.map((run) => [run.status, run.queueHeld ?? false]), + [ + ["cancelled", false], + ["failed", false], + ["queued", true], + ], + ); + }), + ); + it.effect("wakes the parent for a settled_only task once its spawning turn ended", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5d4b1cfda15f3fa4d696ebab128c75811ef5cc9b..e0d288282b1ab1b48d404b8dca89cc3ac547a011 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10613,25 +10613,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // 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)); - } if (stored.event.type === "run.updated") { yield* threadDispatch.withLock( threadId, finalizeDelegatedCompletionDelivery(threadId, stored.event.payload.id), ); } - yield* threadDispatch.withLock( - threadId, - startNextQueuedRun( + yield* threadDispatch + .withLock( threadId, - stored.event.type === "run.updated" && stored.event.payload.status === "failed" - ? { failedRunId: stored.event.payload.id } - : undefined, - ), - ); + startNextQueuedRun( + threadId, + stored.event.type === "run.updated" && stored.event.payload.status === "failed" + ? { failedRunId: stored.event.payload.id } + : undefined, + ), + ) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to start the next queued V2 run", { threadId, cause }), + ), + ); + // After the queue decision: a provider failure holds the child's queued + // wakes, and only then is the failed run the task's result. + const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId); + if (parentThreadId !== undefined) { + yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId)); + } }).pipe( Effect.catchCause((cause) => Effect.logWarning("Failed to react to terminal V2 run", { diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6bde9f596e2ef3fbc11eedcf2943302f4ad34a6c..c7ea0530d3512bbdefa520d14247b4a037b03a75 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -231,7 +231,7 @@ it.effect("recovers terminal subagent results until their cross-thread transfer const now = yield* DateTime.now; const parent = yield* createThread("subagent-parent"); const children: Array = []; - for (const name of ["terminal", "archived", "deleted", "running"]) { + for (const name of ["terminal", "archived", "deleted", "running", "held", "queued"]) { const child = yield* createThread(`subagent-${name}`, { lineage: { parentThreadId: parent, relationshipToParent: "subagent", rootThreadId: parent }, forkedFrom: { type: "node", nodeId: NodeId.make(`node:${name}`) }, @@ -239,12 +239,22 @@ it.effect("recovers terminal subagent results until their cross-thread transfer deletedAt: name === "deleted" ? now : null, }); yield* createRun(child, name === "running" ? "running" : "completed"); + // A wake queued behind the result: held by Stop or a restart, or still deliverable. + if (name === "held" || name === "queued") { + yield* createRun(child, "queued", { + ordinal: 2, + startedAt: null, + ...(name === "held" ? { queueHeld: true } : {}), + }); + } children.push(child); } + const terminalChildren = new Set([children[0]!, children[1]!, children[4]!]); assert.deepEqual( new Set(yield* projections.getRecoveryThreadIds("subagent-results")), - new Set(children.slice(0, 2)), + terminalChildren, ); + const completed = children[0]!; const transferId = ContextTransferId.make("transfer:recovery:subagent-result"); yield* projections.apply({ @@ -273,7 +283,10 @@ it.effect("recovers terminal subagent results until their cross-thread transfer }); const archivedChild = children[1]; assert.isDefined(archivedChild); - assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), [archivedChild]); + assert.deepEqual( + new Set(yield* projections.getRecoveryThreadIds("subagent-results")), + new Set([archivedChild, children[4]!]), + ); assert.deepEqual(yield* projections.getUnreadableThreadIds(), []); yield* sql` UPDATE orchestration_v2_projection_context_transfers SET payload_json = '{}' diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4da19774075b7ff90fcd361dfb4aed03d37b5e65..11f98f2330b5d45546de4ea02bdcda94fff85faa 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -514,7 +514,7 @@ function needsRecovery( parentThreadId !== null && projection.thread.forkedFrom?.type === "node" && ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( - projection.runs.at(-1)?.status ?? "idle", + latestUnheldRun(projection.runs)?.status ?? "idle", ) && !projection.contextTransfers.some( (transfer) => @@ -3505,9 +3505,15 @@ export const layer: Layer.Layer = json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' AND json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL AND json_extract(child.payload_json, '$.forkedFrom.type') = 'node' + -- A held queue waits for the user, so the newest unheld run + -- decides, matching latestUnheldRun. AND ( SELECT status FROM orchestration_v2_projection_runs WHERE thread_id = child.thread_id + AND NOT ( + status = 'queued' + AND json_extract(payload_json, '$.queueHeld') IS 1 + ) ORDER BY ordinal DESC LIMIT 1 ) IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') AND NOT EXISTS ( diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index 23d180102cd3e66a6dd6b59267a174fa6b5d90e1..e2691e08e0db6680099c3957821cbab6ae9f4def 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -237,6 +237,14 @@ it("waits for nested work and retains the report across monitor acknowledgements startedAt: null, }; assert.equal(delegatedTaskProgress({ ...projection, runs: [run, pending] }).state, "working"); + // Stop or a restart holds queued wakes for the user; they are not work the task owes. + const cancelled = { ...run, status: "cancelled" as const }; + const held = delegatedTaskProgress({ + ...projection, + runs: [cancelled, { ...pending, queueHeld: true }], + }); + assert.equal(held.state, "result_available"); + assert.equal(held.resultRun?.id, cancelled.id); const report = { ...pending, status: "completed" as const, startedAt: parentCreatedAt }; const monitor = { ...report, id: RunId.make("monitor"), ordinal: 3 }; const artifacts = makeSubagentConversationArtifacts({ diff --git a/apps/server/src/orchestration-v2/SubagentProjection.ts b/apps/server/src/orchestration-v2/SubagentProjection.ts index f28dc1aa4be5ff8d1a2f157a79428020e6714b0b..7ee8d7c1aa8d9d0554697c6e7b5138991c51e9d5 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.ts @@ -229,7 +229,11 @@ export function delegatedTaskProgress(projection: { const workRuns = projection.runs.filter( (run) => !monitorRuns.has(run.id) && run.status !== "rolled_back", ); - const active = workRuns.some((run) => !terminal(run.status)); + // A held queue waits for the user to resume it (after Stop, a restart, or a + // provider failure), so its runs are not work the task still owes. + const active = workRuns.some( + (run) => !terminal(run.status) && !(run.status === "queued" && run.queueHeld === true), + ); const children = projection.subagents.some( (task) =>