diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 2b2033779f210aa70c354cecfdf90f8e1b060906..5b426477c2409dd7693a6c9733022519a411fc5e 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -200,7 +200,16 @@ describe("OrchestratorMcpService", () => { } as unknown as OrchestrationV2ThreadProjection; const childProjection = { thread: { id: childThreadId }, - runs: [{ id: RunId.make("run:mcp-restart-child"), ordinal: 1, status: "cancelled" }], + runs: [ + { + id: RunId.make("run:mcp-restart-child"), + ordinal: 1, + status: "cancelled", + delegatedTaskId: taskId, + }, + // An unrelated later run cannot supply this follow-up's result. + { id: RunId.make("run:mcp-restart-later"), ordinal: 2, status: "running" }, + ], contextTransfers: [], messages: [], subagents: [], diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index b218a76d1ab5033f14e63b65ac1cfc2a642c3185..872163b660d1a4861d2a0d9b687c6d380ff470e5 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -351,6 +351,8 @@ export function delegatedTaskRun( childProjection: Pick, task: OrchestrationV2Subagent, ): OrchestrationV2Run | undefined { + const taskRun = childProjection.runs.find((run) => run.delegatedTaskId === task.id); + if (taskRun !== undefined) return taskRun; const spawnTransfer = childProjection.contextTransfers.find( (transfer) => transfer.type === "subagent_spawn" && @@ -390,22 +392,62 @@ function isTerminalTaskStatus( ); } -function directAppOwnedChildTask( +function directAppOwnedChildTasks( parent: Pick, - target: Pick, -): OrchestrationV2Subagent | undefined { + target: Pick, + returnedRunIds: ReadonlySet, +): ReadonlyArray { if ( target.thread.lineage.parentThreadId !== parent.thread.id || target.thread.lineage.relationshipToParent !== "subagent" ) { - return undefined; + return []; } - return parent.subagents.find( + const original = parent.subagents.find( (task) => task.origin === "app_owned" && task.threadId === parent.thread.id && + task.id === + (target.thread.forkedFrom?.type === "node" ? target.thread.forkedFrom.nodeId : undefined) && task.childThreadId === target.thread.id, ); + const taskIds = new Set( + target.runs + .filter((run) => returnedRunIds.has(run.id)) + .map((run) => run.delegatedTaskId ?? original?.id), + ); + return parent.subagents.filter( + (task) => + taskIds.has(task.id) && + task.origin === "app_owned" && + task.childThreadId === target.thread.id, + ); +} + +function taskResultTransfer( + parent: Pick, + child: Pick, + task: OrchestrationV2Subagent, +) { + const originalRun = delegatedTaskRun(child, task); + const resultRunIds = new Set( + child.runs + .filter( + (run) => + run.delegatedTaskId === task.id || + (originalRun?.delegatedTaskId === undefined && run.delegatedTaskId === undefined), + ) + .map((run) => run.id), + ); + if (originalRun !== undefined) resultRunIds.add(originalRun.id); + return parent.contextTransfers.find( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === task.childThreadId && + transfer.targetThreadId === parent.thread.id && + ((transfer.sourcePoint.runId !== undefined && resultRunIds.has(transfer.sourcePoint.runId)) || + (transfer.sourcePoint.runId === undefined && originalRun?.delegatedTaskId === undefined)), + ); } function pageIncludesTerminalTaskResult(input: { @@ -418,12 +460,7 @@ function pageIncludesTerminalTaskResult(input: { >; readonly maxChars: number; }): boolean { - const transfer = input.parent.contextTransfers.find( - (transfer) => - transfer.type === "subagent_result" && - transfer.sourceThreadId === input.target.thread.id && - transfer.targetThreadId === input.parent.thread.id, - ); + const transfer = taskResultTransfer(input.parent, input.target, input.task); if (transfer === undefined) return false; const run = transfer.sourcePoint.runId === undefined @@ -1241,9 +1278,21 @@ const make = Effect.gen(function* () { const childRun = delegatedTaskRun(childControls, task); const terminalRun = latestTerminalResultRun(childControls, childRun); const progress = delegatedTaskProgress(childControls); + const taskResultRun = + childRun?.delegatedTaskId === task.id + ? childControls.runs + .filter((run) => run.delegatedTaskId === task.id) + .toSorted((a, b) => b.ordinal - a.ordinal)[0] + : progress.resultRun; + const taskWorkState = + childRun?.delegatedTaskId === task.id + ? isTerminalTaskStatus(taskStatusForRun(taskResultRun)) + ? "result_available" + : "working" + : progress.state; const resultRunIds = [ ...new Set( - [progress.resultRun?.id, terminalRun?.id].filter((id): id is RunId => id !== undefined), + [taskResultRun?.id, terminalRun?.id].filter((id): id is RunId => id !== undefined), ), ]; const resultRecords = yield* threadManagement @@ -1263,12 +1312,12 @@ const make = Effect.gen(function* () { // the child started working again after this read. const heldForRestart = task.result === null && - progress.state === "result_available" && + taskWorkState === "result_available" && (yield* threadManagement - .delegatedTaskResultPending(task.childThreadId) + .delegatedTaskResultPending(task.childThreadId, childRun?.delegatedTaskId) .pipe(Effect.mapError(threadManagementFailure))); const workState = - task.result !== null ? "result_available" : heldForRestart ? "working" : progress.state; + task.result !== null ? "result_available" : heldForRestart ? "working" : taskWorkState; const status = task.result !== null ? taskStatusForRun( @@ -1280,15 +1329,15 @@ const make = Effect.gen(function* () { : childRun, ) : workState === "result_available" - ? taskStatusForRun(progress.resultRun ?? childRun) - : taskStatusForRun(childRun) === "queued" + ? taskStatusForRun(taskResultRun ?? childRun) + : taskStatusForRun(taskResultRun ?? childRun) === "queued" ? "queued" : "running"; const derivedResult = task.result !== null ? task.result - : progress.resultRun !== undefined && isTerminalTaskStatus(status) - ? subagentResultForRun(childProjection, progress.resultRun).text + : taskResultRun !== undefined && isTerminalTaskStatus(status) + ? subagentResultForRun(childProjection, taskResultRun).text : null; const resultTransfers = parentProjection.contextTransfers.filter( (transfer) => @@ -1304,7 +1353,7 @@ const make = Effect.gen(function* () { ? resultTransfers.find((transfer) => transfer.sourcePoint.runId === undefined) : undefined) ?? null); - const resultTransfer = resultTransfers[0] ?? null; + const resultTransfer = taskResultTransfer(parentProjection, childProjection, task) ?? null; const terminalStatus = terminalRun === undefined ? null : taskStatusForRun(terminalRun); const response = { taskId: task.id, @@ -2315,45 +2364,48 @@ const make = Effect.gen(function* () { { concurrency: 1 }, ); const messagesByThreadId = new Map(sourceMessages); - const task = parent === undefined ? undefined : directAppOwnedChildTask(parent, target); + const returnedRunIds = new Set( + page + .filter((row) => row.sourceThreadId === target.thread.id) + .map((row) => row.item.runId), + ); + const tasks = + parent === undefined ? [] : directAppOwnedChildTasks(parent, target, returnedRunIds); if ( parent !== undefined && scope.thread !== undefined && - task !== undefined && + tasks.length > 0 && (input.textOffset ?? 0) === 0 ) { - const transfer = parent.contextTransfers.find( - (transfer) => - transfer.type === "subagent_result" && - transfer.sourceThreadId === target.thread.id && - transfer.targetThreadId === parent.thread.id, - ); - const resultRunId = transfer?.sourcePoint.runId ?? delegatedTaskRun(target, task)?.id; - const resultRunIds = resultRunId === undefined ? [] : [resultRunId]; - const resultRecords = yield* threadManagement - .getThreadRecords(target.thread.id, ["messages", "turnItems"], { - messageRoles: ["assistant"], - messageRunIds: resultRunIds, - turnItemRunIds: resultRunIds, - turnItemTypes: ["assistant_message", "error"], - }) - .pipe(Effect.mapError(threadManagementFailure)); - if ( - pageIncludesTerminalTaskResult({ - parent, - page, - task, - target: { ...target, ...resultRecords }, - maxChars, - }) - ) { - yield* readTask( - scope as McpThreadInvocationScope, - task.id, - false, - true, - "thread-read-acknowledge", - ); + for (const task of tasks) { + const transfer = taskResultTransfer(parent, target, task); + const resultRunId = transfer?.sourcePoint.runId ?? delegatedTaskRun(target, task)?.id; + const resultRunIds = resultRunId === undefined ? [] : [resultRunId]; + const resultRecords = yield* threadManagement + .getThreadRecords(target.thread.id, ["messages", "turnItems"], { + messageRoles: ["assistant"], + messageRunIds: resultRunIds, + turnItemRunIds: resultRunIds, + turnItemTypes: ["assistant_message", "error"], + }) + .pipe(Effect.mapError(threadManagementFailure)); + if ( + pageIncludesTerminalTaskResult({ + parent, + page, + task, + target: { ...target, ...resultRecords }, + maxChars, + }) + ) { + yield* readTask( + scope as McpThreadInvocationScope, + task.id, + false, + true, + "thread-read-acknowledge", + ); + } } } return { @@ -2419,6 +2471,9 @@ const make = Effect.gen(function* () { threadId: input.threadId, messageId, runId: result.run.id, + ...(result.run.delegatedTaskId === undefined + ? {} + : { taskId: result.run.delegatedTaskId }), status: result.run.status, delivery: result.delivery, } satisfies OrchestratorMcpThreadSendResult; diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 8f709ecf86258af21a7abe6494cc02e645bd861b..e7e82eaddffbb213849d5e72a6127c671d434b73 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -1782,6 +1782,9 @@ describe("orchestrator MCP toolkit", () => { expect(delegatedStatus.resultContextTransferId).not.toBeNull(); expect(delegatedStatus.latestTerminalResultContextTransferId).not.toBeNull(); + // Wait-mode completion already returned its result while the parent was live. + yield* expectOffersToStay(0); + const childFollowupCall = yield* invoke("t3_thread_send", { threadId: delegated.childThreadId, message: "Confirm the delegated API boundary remains inspected.", @@ -1924,10 +1927,14 @@ describe("orchestrator MCP toolkit", () => { latestTerminalStatus: "interrupted", }); - // 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); + // Later requested work has its own delivery, including an interrupted follow-up. + expect(childFollowup.taskId).toBeDefined(); + expect(activeChildFollowup.taskId).toBeDefined(); + yield* waitForContinuationOffers(2); + yield* expectOffersToStay(2); + yield* invoke("task_status", { taskId: childFollowup.taskId }); + yield* invoke("task_status", { taskId: activeChildFollowup.taskId }); + yield* Ref.set(continuationOffers, []); const repeatedDelegatedCall = yield* invoke("delegate_task", { task: delegatedPrompt, @@ -4007,6 +4014,18 @@ describe("orchestrator MCP toolkit", () => { const pendingProjection = yield* orchestrator.getThreadProjection( delegated.childThreadId, ); + const followupTasks = (yield* orchestrator.getThreadProjection( + parentThreadId, + )).subagents.filter( + (task) => + task.childThreadId === delegated.childThreadId && task.id !== delegated.taskId, + ); + expect(followupTasks).toHaveLength(2); + expect(followupTasks.every((task) => task.result === null)).toBe(true); + expect(followupTasks.map((task) => task.id)).toEqual([ + runningFollowup.taskId, + queuedFollowup.taskId, + ]); expect( pendingProjection.runs.find((run) => run.id === delegated.childRunId)?.status, ).toBe("completed"); @@ -4082,6 +4101,47 @@ describe("orchestrator MCP toolkit", () => { expect(finalProjection.runs.find((run) => run.id === queuedFollowup.runId)?.status).toBe( "completed", ); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: finalSequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === followupTasks[1]!.id && + stored.event.payload.status === "completed", + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Follow-up task did not settle."), + onSome: () => Effect.void, + }), + ), + ); + const finishedParent = yield* orchestrator.getThreadProjection(parentThreadId); + expect( + finishedParent.subagents.find((task) => task.id === followupTasks[0]!.id)?.status, + ).toBe("interrupted"); + expect( + finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id), + ).toMatchObject({ + status: "completed", + result: queuedFollowupResult, + }); + expect( + finishedParent.contextTransfers.filter( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === delegated.childThreadId, + ), + ).toHaveLength(3); + expect( + finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id) + ?.completionDelivery?.state, + ).toBe("claimed"); const finalStatusCall = yield* invoke("task_status", { taskId: delegated.taskId, }); @@ -4097,8 +4157,18 @@ describe("orchestrator MCP toolkit", () => { latestTerminalRunId: queuedFollowup.runId, latestTerminalStatus: "completed", latestTerminalSummary: queuedFollowupResult, - latestTerminalResultContextTransferId: null, }); + expect(finalStatus.latestTerminalResultContextTransferId).not.toBeNull(); + const read = yield* invoke("t3_thread_read", { + threadId: delegated.childThreadId, + limit: 100, + }); + expect(read.isError).not.toBe(true); + const afterRead = yield* orchestrator.getThreadProjection(parentThreadId); + expect( + afterRead.subagents.find((task) => task.id === queuedFollowup.taskId) + ?.completionDelivery?.state, + ).toBe("acknowledged"); }).pipe(Effect.provide(layerTest)); }), ), diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts index e95639d308dc707cc5383108d39741f2825c394c..b8f735069d31c78b5fe371fd3eb235bcf8a5a542 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts @@ -22,7 +22,7 @@ describe("orchestrator MCP tool guidance", () => { assert.include(DelegateTaskTool.description ?? "", "childThreadId is backing storage"); assert.include( OrchestratorToolkit.tools.t3_thread_send.description ?? "", - "Do not use a delegated task's childThreadId to start another review round", + "a requested follow-up creates a fresh task", ); assert.include( OrchestratorToolkit.tools.task_cancel.description ?? "", diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index ac0ebb63b7ddf04fa6dfde6a0cf3de4fa3ef261e..33be4d24cb1f3cd3b40daaf8e7c9b89b5af0020d 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -219,7 +219,7 @@ export const ThreadUpdateTool = Tool.make("t3_thread_update", { const ThreadSendTool = Tool.make("t3_thread_send", { description: - "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Do not use a delegated task's childThreadId to start another review round here; use delegate_task with the full review context and a new clientRequestId for that round. Thread messages do not create a new delegated task or reopen a completed task. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", + "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId; its completion notifies the parent, and the original result stays unchanged. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", parameters: OrchestratorMcpThreadSendInput, success: OrchestratorMcpThreadSendResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 5a41e5fd08e26e8d65ecd4f6c089771bf48d320f..ad0c6703c1d9ba1f765e079a1e89bb578ae94fa1 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -41,6 +41,7 @@ import { continueRestartedRun } from "./RestartContinuation.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; const layerPlatformTest = Layer.merge( NodeServices.layer, @@ -904,6 +905,318 @@ const seedRestartCancelledChild = (input: { return { taskId, childThreadId, childRunId }; }); +it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( + "subagent follow-ups", + (it) => { + const outcomes = [ + "completed", + "failed", + "interrupted", + "restarted", + "stopped", + "overlapping", + ] as const; + it.effect.each(outcomes)( + "delivers a %s follow-up independently of the original task", + (outcome) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make(`thread:followup-${outcome}`); + const projectId = ProjectId.make(`project:followup-${outcome}`); + const parentRunId = RunId.make(`run:followup-${outcome}`); + const rootNodeId = NodeId.make(`node:followup-${outcome}`); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make(`node:seed-${outcome}`), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: `followup-child-${outcome}`, + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + let firstWake = yield* requests.take; + while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take; + assert.equal(firstWake.delegatedCompletion?.parentRunId, parentRunId); + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:followup-ack-${outcome}`), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + const original = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === child.taskId, + )!; + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find( + (run) => run.id === parentRunId, + )!; + yield* eventSink.write({ + commandId: reconcileCommandId(`followup-parent-finished-${outcome}`), + events: [ + { + id: EventId.make(`event:followup-parent-finished-${outcome}`), + type: "run.updated", + threadId: parentThreadId, + runId: parentRunId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); + const messageId = MessageId.make(`message:followup-${outcome}`); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: outcome === "failed" ? "user" : "agent", + creationSource: outcome === "failed" ? "web" : "mcp", + commandId: CommandId.make(`command:followup-send-${outcome}`), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId, + text: "Check the same work again.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); + let run = childProjection.runs.find((row) => row.userMessageId === messageId)!; + assert.isDefined(run.delegatedTaskId); + const workingParent = yield* orchestrator.getShellSnapshot(); + const backgroundTask = workingParent.threads + .find((thread) => thread.id === parentThreadId) + ?.pendingBackgroundTasks?.find((task) => task.kind === "subagent"); + assert.equal(backgroundTask?.childThreadId, child.childThreadId); + if (outcome === "overlapping") { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:overlapping-followup"), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: MessageId.make("message:overlapping-followup"), + text: "Keep working on another follow-up.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + } + if (outcome === "restarted") { + const cut = { ...run, status: "cancelled" as const, startedAt: now, completedAt: now }; + yield* eventSink.writeWithEffects({ + commandId: reconcileCommandId("followup-restart-cut"), + events: [ + { + id: EventId.make("event:followup-restart-cut"), + type: "run.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: cut, + }, + ], + effects: [ + { + id: `effect:restart-continuation:${run.id}`, + commandId: reconcileCommandId("followup-restart-cut"), + threadId: child.childThreadId, + request: { type: "provider-runtime.continue", sourceRunId: run.id }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + assert.isNull( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === run.delegatedTaskId, + )!.result, + ); + const continuationMessageId = MessageId.make("message:followup-restart-continuation"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make("command:followup-restart-continuation"), + threadId: child.childThreadId, + restartContinuationOfRunId: run.id, + messageId: continuationMessageId, + text: "Continue where you left off.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + const continuation = (yield* orchestrator.getThreadProjection( + child.childThreadId, + )).runs.find((row) => row.userMessageId === continuationMessageId)!; + assert.equal(continuation.delegatedTaskId, run.delegatedTaskId); + run = continuation; + } + if (outcome === "stopped") { + yield* orchestrator.dispatch({ + type: "thread.stop", + threadId: parentThreadId, + commandId: CommandId.make(`command:followup-stop-${outcome}`), + }); + } + // Reconciliation skips the live terminal listener; recovery must finish this recorded task. + yield* eventSink.write({ + commandId: reconcileCommandId(`followup-finish-${outcome}`), + events: [ + { + id: EventId.make(`event:followup-finish-${outcome}`), + type: "run.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: { + ...run, + status: + outcome === "stopped" + ? "interrupted" + : outcome === "restarted" || outcome === "overlapping" + ? "completed" + : outcome, + startedAt: now, + completedAt: now, + }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const finished = yield* orchestrator.getThreadProjection(parentThreadId); + assert.deepEqual( + finished.subagents.find((task) => task.id === child.taskId), + original, + ); + const followup = finished.subagents.find((task) => task.id === run.delegatedTaskId)!; + if (outcome === "overlapping") { + const later = finished.subagents.find( + (task) => + task.childThreadId === child.childThreadId && + task.id !== child.taskId && + task.id !== followup.id, + )!; + assert.equal(later.status, "pending"); + assert.isNull(later.result); + } + assert.equal( + followup.status, + outcome === "stopped" + ? "interrupted" + : outcome === "restarted" || outcome === "overlapping" + ? "completed" + : outcome, + ); + assert.equal( + followup.completionDelivery?.state, + outcome === "stopped" ? "disposed" : "claimed", + ); + if (outcome !== "stopped") { + let wake = yield* requests.take; + while ( + wake.threadId !== parentThreadId || + wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId + ) { + wake = yield* requests.take; + } + assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); + assert.notEqual( + wake.delegatedCompletion?.messageId, + firstWake.delegatedCompletion?.messageId, + ); + } + assert.equal( + finished.contextTransfers.filter( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === child.childThreadId, + ).length, + 2, + ); + yield* orchestrator.recoverDelegatedTasks; + const recovered = yield* orchestrator.getThreadProjection(parentThreadId); + assert.equal(recovered.contextTransfers.length, finished.contextTransfers.length); + if (outcome === "completed") { + const automaticMessageId = MessageId.make("message:followup-automatic-wake"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make("command:followup-automatic-wake"), + threadId: child.childThreadId, + messageId: automaticMessageId, + text: "A monitor reported new output.", + attachments: [], + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: "Monitor updated", + }, + dispatchMode: { type: "queue_after_active" }, + }); + const automatic = (yield* orchestrator.getThreadProjection( + child.childThreadId, + )).runs.find((run) => run.userMessageId === automaticMessageId)!; + assert.isUndefined(automatic.delegatedTaskId); + yield* eventSink.write({ + commandId: reconcileCommandId("followup-automatic-finished"), + events: [ + { + id: EventId.make("event:followup-automatic-finished"), + type: "run.updated", + threadId: child.childThreadId, + runId: automatic.id, + occurredAt: now, + payload: { ...automatic, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const afterAutomatic = yield* orchestrator.getThreadProjection(parentThreadId); + assert.deepEqual(afterAutomatic.subagents, recovered.subagents); + assert.equal(afterAutomatic.contextTransfers.length, recovered.contextTransfers.length); + const control = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "followup-control-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + const afterControl = yield* orchestrator.getThreadProjection(parentThreadId); + assert.equal( + afterControl.subagents.find((task) => task.id === control.taskId)?.completionDelivery + ?.state, + "claimed", + ); + assert.equal( + afterControl.contextTransfers.filter( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === control.childThreadId, + ).length, + 1, + ); + } + }), + ); + }, +); + it.layer(layerTest)("delegated tasks across a server restart", (it) => { it.effect("holds a restart-cancelled child for its continuation's result", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 977b225f525664d839969520902969f77b31e684..475c6bea0a94041ecefd309652a99b4dd676d6d4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -22,6 +22,7 @@ import { CommandId, isProviderNativeSubagentThread, MessageId, + type NodeId, type ModelSelection, OrchestrationV2Command, type OrchestrationV2InternalCommand, @@ -279,9 +280,11 @@ export interface OrchestratorV2Shape { /** * Whether a delegated child's apparent result is not final yet: a restart * continuation is pending, or the child is working again. + * A taskId limits the check to that follow-up's runs. */ readonly delegatedTaskResultPending: ( childThreadId: ThreadId, + taskId?: NodeId, ) => Effect.Effect; readonly dispatch: ( command: OrchestrationV2ServerCommand, @@ -4561,6 +4564,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } } + const continuedTaskId = projection.runs.find( + (run) => run.id === command.restartContinuationOfRunId, + )?.delegatedTaskId; + if (projection.thread.settledOverride !== null) { const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { @@ -4980,6 +4987,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }), }; const attempt: OrchestrationV2RunAttempt = { id: attemptId, @@ -5322,6 +5330,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(dispatchMode.type === "defer_start" && dispatchMode.workspaceStrategy !== undefined ? { workspacePreparation: dispatchMode.workspaceStrategy } : {}), + ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }), ...wakeWorkStartedAt(projection.runs, command), }; const attempt: OrchestrationV2RunAttempt = { @@ -6015,6 +6024,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }), ...wakeWorkStartedAt(projection.runs, command), }; const attempt: OrchestrationV2RunAttempt = { @@ -6738,6 +6748,165 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ); + // A follow-up is new requested work in the same child conversation. Give it + // its own task so results, acknowledgments, Stop, and recovery use the same + // lifecycle as a fresh delegation without replacing the original result. + const recordSubagentFollowup = Effect.fnUntraced(function* ( + command: Extract, + events: Ref.Ref>, + ) { + if ( + command.scheduledTaskId !== undefined || + command.notification !== undefined || + command.restartContinuationOfRunId !== undefined || + (command.createdBy !== "user" && command.creationSource !== "mcp") + ) + return; + const pending = yield* Ref.get(events); + const message = pending + .flatMap((event) => + event.type === "message.updated" && event.payload.id === command.messageId + ? [event.payload] + : [], + ) + .at(-1); + if (message?.runId == null) return; + const child = yield* projectionStore.getThread(command.threadId); + const parentThreadId = child.lineage.parentThreadId; + if ( + child.lineage.relationshipToParent !== "subagent" || + parentThreadId === null || + (command.createdBy !== "user" && command.senderThreadId !== parentThreadId) + ) + return; + const run = + pending + .flatMap((event) => + (event.type === "run.created" || event.type === "run.updated") && + event.payload.id === message.runId + ? [event.payload] + : [], + ) + .at(-1) ?? + (yield* projectionStore.getThreadRecords(command.threadId, ["runs"])).runs.find( + (row) => row.id === message.runId, + ); + if (run === undefined || run.delegatedTaskId !== undefined) return; + const parent = yield* projectionStore + .getThreadRecords(parentThreadId, ["runs", "subagents", "providerTurns", "attempts"]) + .pipe(mapDispatchError(command)); + const original = parent.subagents.find( + (task) => + task.origin === "app_owned" && + task.childThreadId === child.id && + task.id === (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined), + ); + // Before first publication, follow-ups remain part of the original task. + if (original === undefined || original.result === null) return; + const owner = + parent.runs.filter(isBlockingRun).toSorted((a, b) => b.ordinal - a.ordinal)[0] ?? + latestExecutedRun(parent.runs); + if (owner === null || owner.rootNodeId === null) return; + const now = message.createdAt; + const adapter = yield* providerAdapters.get(run.providerInstanceId); + const taskId = idAllocator.derive.delegatedTaskNode({ commandId: command.commandId }); + const task: OrchestrationV2Subagent = { + ...original, + createdBy: command.createdBy, + driver: adapter.driver, + id: taskId, + runId: owner.id, + parentNodeId: owner.rootNodeId, + providerInstanceId: run.providerInstanceId, + providerThreadId: run.providerThreadId, + model: run.modelSelection.model, + modelSelection: run.modelSelection, + prompt: command.text, + title: child.title, + completionWake: "always", + completionDelivery: { state: "pending", observedByRunId: null }, + status: run.status === "queued" ? "pending" : "running", + progress: undefined, + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }; + const node: OrchestrationV2ExecutionNode = { + id: taskId, + threadId: parentThreadId, + runId: owner.id, + parentNodeId: owner.rootNodeId, + rootNodeId: owner.rootNodeId, + kind: "subagent", + status: task.status, + countsForRun: false, + providerThreadId: run.providerThreadId, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }; + const turnItem: OrchestrationV2TurnItem = { + id: idAllocator.derive.delegatedTaskTurnItem({ commandId: command.commandId }), + threadId: parentThreadId, + runId: owner.id, + nodeId: taskId, + providerThreadId: owner.providerThreadId, + providerTurnId: providerTurnForRun(parent, owner)?.id ?? null, + nativeItemRef: null, + parentItemId: null, + ordinal: yield* nextTurnItemOrdinal(parent), + status: task.status, + title: child.title, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "subagent", + subagentId: taskId, + origin: "app_owned", + driver: task.driver, + providerInstanceId: task.providerInstanceId, + childThreadId: child.id, + prompt: command.text, + result: null, + }; + const emitEvent = emit(events, command); + yield* emitEvent({ + type: "run.updated", + threadId: child.id, + runId: run.id, + occurredAt: now, + payload: { ...run, delegatedTaskId: taskId }, + }); + yield* emitEvent({ + type: "node.updated", + threadId: parentThreadId, + runId: owner.id, + nodeId: taskId, + occurredAt: now, + payload: node, + }); + yield* emitEvent({ + type: "subagent.updated", + threadId: parentThreadId, + runId: owner.id, + nodeId: taskId, + occurredAt: now, + payload: task, + }); + yield* emitEvent({ + type: "turn-item.updated", + threadId: parentThreadId, + runId: owner.id, + nodeId: taskId, + occurredAt: now, + payload: turnItem, + }); + }); + // 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. @@ -9581,16 +9750,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio * running). `settledContinuationOf` names the restart continuation that just * declined or failed, so its own still-running effect does not hold the child. */ - const finalizeAppOwnedSubagent = ( - childThreadId: ThreadId, - options?: { readonly settledContinuationOf?: RunId }, + const finalizeAppOwnedSubagentTask = ( + childControls: ProjectionRecords< + "runs" | "messages" | "subagents" | "providerThreads" | "providerTurns" | "attempts" + >, + options: { + readonly settledContinuationOf?: RunId; + readonly taskId: NodeId; + readonly resultRun?: OrchestrationV2Run; + }, ) => Effect.gen(function* () { - const childControls = yield* projectionStore.getThreadRecords( - childThreadId, - ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"], - { messageRoles: ["user"] }, - ); + const childThreadId = childControls.thread.id; const forkedFrom = childControls.thread.forkedFrom; if ( childControls.thread.lineage.relationshipToParent !== "subagent" || @@ -9599,9 +9770,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { return; } - const progress = delegatedTaskProgress(childControls); - if (progress.state !== "result_available") return; - const childRun = progress.resultRun; + const taskRuns = childControls.runs.filter((run) => run.delegatedTaskId === options.taskId); + const childRun = + taskRuns.length === 0 + ? options.taskId === forkedFrom.nodeId + ? options.resultRun + : undefined + : taskRuns.toSorted((a, b) => b.ordinal - a.ordinal)[0]; if (childRun === undefined) return; const terminalStatus = delegatedTaskTerminalStatus(childRun.status); if (terminalStatus === null) { @@ -9609,7 +9784,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } if ( yield* childAwaitsRestartContinuation( - childControls.runs, + taskRuns.length === 0 ? childControls.runs : taskRuns, childRun, options?.settledContinuationOf, ) @@ -9649,7 +9824,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const task = parentProjection.subagents.find( (candidate) => - candidate.id === forkedFrom.nodeId && + candidate.id === options.taskId && candidate.origin === "app_owned" && candidate.childThreadId === childThreadId, ); @@ -9660,7 +9835,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (transfer) => transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId && - transfer.targetThreadId === parentThreadId, + transfer.targetThreadId === parentThreadId && + (transfer.sourcePoint.runId === childRun.id || + (options.taskId === forkedFrom.nodeId && transfer.sourcePoint.runId === undefined)), ); if (existingResultTransfer !== undefined) { return; @@ -9873,6 +10050,49 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + const finalizeAppOwnedSubagent = ( + childThreadId: ThreadId, + options?: { readonly settledContinuationOf?: RunId }, + ) => + Effect.gen(function* () { + const childControls = yield* projectionStore.getThreadRecords( + childThreadId, + ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"], + { messageRoles: ["user"] }, + ); + const parentThreadId = childControls.thread.lineage.parentThreadId; + if ( + childControls.thread.lineage.relationshipToParent !== "subagent" || + parentThreadId === null || + childControls.thread.forkedFrom?.type !== "node" + ) + return; + const progress = delegatedTaskProgress(childControls); + const parent = yield* projectionStore.getThreadRecords(parentThreadId, [ + "subagents", + "contextTransfers", + ]); + for (const task of parent.subagents) { + if ( + task.origin !== "app_owned" || + task.childThreadId !== childThreadId || + (task.id === childControls.thread.forkedFrom.nodeId && + progress.state !== "result_available") || + (task.result !== null && + parent.contextTransfers.some( + (transfer) => + transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId, + )) + ) + continue; + yield* finalizeAppOwnedSubagentTask(childControls, { + ...options, + taskId: task.id, + ...(progress.resultRun === undefined ? {} : { resultRun: progress.resultRun }), + }); + } + }); + const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) => Effect.gen(function* () { // Ordinary turns do not own a delegated delivery. Read just their input @@ -10266,6 +10486,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } yield* dispatchMessage(command, events, effects); + yield* recordSubagentFollowup(command, events).pipe(mapDispatchError(command)); break; } case "notification.delivery.accept": @@ -10627,7 +10848,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); const dispatchWithReceipt = (command: OrchestrationV2ServerCommand) => - threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + Effect.gen(function* () { + const dispatch = threadDispatch.withLock( + commandThreadId(command), + dispatchWithReceiptEffect(command), + ); + if ( + command.type !== "message.dispatch" || + (command.createdBy !== "user" && command.creationSource !== "mcp") + ) + return yield* dispatch; + const parentThreadId = yield* appOwnedSubagentParentThreadId(command.threadId).pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + // Follow-up intake and Stop serialize on the parent before taking the child lock. + return yield* parentThreadId === undefined + ? dispatch + : threadDispatch.withLock(parentThreadId, dispatch); + }); const handleTerminalRun = (stored: OrchestrationV2StoredEvent) => Effect.gen(function* () { @@ -10792,13 +11032,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); }); - const delegatedTaskResultPending = (childThreadId: ThreadId) => + const delegatedTaskResultPending = (childThreadId: ThreadId, taskId?: NodeId) => Effect.gen(function* () { const child = yield* projectionStore.getThreadRecords( childThreadId, ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"], { messageRoles: ["user"] }, ); + if (taskId !== undefined) { + const runs = child.runs.filter((run) => run.delegatedTaskId === taskId); + const run = runs.toSorted((a, b) => b.ordinal - a.ordinal)[0]; + if (run === undefined || delegatedTaskTerminalStatus(run.status) === null) return true; + return yield* childAwaitsRestartContinuation(runs, run); + } const progress = delegatedTaskProgress(child); // A caller's older read saw a result; newer work since then means it is not final. if (progress.state !== "result_available") return true; diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index c7ea0530d3512bbdefa520d14247b4a037b03a75..2ec0533e4b8ab78fc4951b7e3614f56089404752 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -299,6 +299,89 @@ it.effect("recovers terminal subagent results until their cross-thread transfer }).pipe(Effect.provide(layerTest)), ); +it.effect.each([ + { name: "SQLite", layer: layerTest }, + { name: "memory", layer: ProjectionStore.layerMemory }, +])("recovers only recorded follow-up results with $name projections", ({ layer }) => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const parent = yield* createThread("followup-parent"); + const child = yield* createThread("followup-child", { + lineage: { parentThreadId: parent, relationshipToParent: "subagent", rootThreadId: parent }, + forkedFrom: { type: "node", nodeId: NodeId.make("node:original") }, + }); + const originalRunId = yield* createRun(child, "completed"); + yield* projections.apply({ + id: EventId.make("event:original-result"), + type: "context-transfer.created", + threadId: parent, + occurredAt: now, + payload: { + id: ContextTransferId.make("transfer:original-result"), + type: "subagent_result", + sourceThreadId: child, + targetThreadId: parent, + sourcePoint: { threadId: child, runId: originalRunId }, + basePoint: null, + sourceProviderInstanceId: providerInstanceId, + targetProviderInstanceId: providerInstanceId, + targetRunId: null, + status: "pending", + resolution: null, + createdBy: "system", + error: null, + createdAt: now, + updatedAt: now, + consumedAt: null, + }, + }); + // Historical untracked work does not acquire a notification on restart. + yield* createRun(child, "completed", { ordinal: 2 }); + assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []); + const taskId = NodeId.make("node:followup"); + yield* createRun(child, "completed", { ordinal: 3, delegatedTaskId: taskId }); + yield* createRun(child, "running", { ordinal: 4 }); + const task = { + id: taskId, + threadId: parent, + runId: null, + parentNodeId: NodeId.make("node:parent"), + origin: "app_owned" as const, + createdBy: "agent" as const, + driver, + providerInstanceId, + providerThreadId: null, + childThreadId: child, + nativeTaskRef: null, + prompt: "Check the follow-up.", + title: null, + model: null, + status: "running" as const, + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }; + yield* projections.apply({ + id: EventId.make("event:followup-task"), + type: "subagent.updated", + threadId: parent, + occurredAt: now, + payload: task, + }); + assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), [child]); + yield* projections.apply({ + id: EventId.make("event:followup-settled"), + type: "subagent.updated", + threadId: parent, + occurredAt: now, + payload: { ...task, status: "completed", result: "Follow-up result.", completedAt: now }, + }); + assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []); + }).pipe(Effect.provide(layer)), +); + it.effect("includes shared sessions and provider-owned background rosters in recovery", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 8c33faef808b21c982720ad43ae8c97da0484257..51343fd8af916e93c2f3a705d34e86f70055aa2c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -537,6 +537,7 @@ export const ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION = 2; function needsRecovery( projection: OrchestrationV2ThreadProjection, kind: ProjectionRecoveryKind, + parent?: Pick, ): boolean { if (projection.thread.deletedAt !== null) return false; switch (kind) { @@ -556,15 +557,28 @@ function needsRecovery( projection.thread.lineage.relationshipToParent === "subagent" && parentThreadId !== null && projection.thread.forkedFrom?.type === "node" && - ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( - latestUnheldRun(projection.runs)?.status ?? "idle", - ) && - !projection.contextTransfers.some( - (transfer) => - transfer.type === "subagent_result" && - transfer.sourceThreadId === projection.thread.id && - transfer.targetThreadId === parentThreadId, - ) + ((parent?.subagents ?? []).some( + (task) => + task.origin === "app_owned" && + task.result === null && + task.childThreadId === projection.thread.id && + projection.runs.some( + (run) => + run.delegatedTaskId === task.id && + ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( + run.status, + ), + ), + ) || + (["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( + latestUnheldRun(projection.runs)?.status ?? "idle", + ) && + ![...projection.contextTransfers, ...(parent?.contextTransfers ?? [])].some( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === projection.thread.id && + transfer.targetThreadId === parentThreadId, + ))) ); } case "runtime": @@ -3576,23 +3590,28 @@ 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 ( + AND (EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS requested + JOIN orchestration_v2_projection_subagents AS task + ON task.subagent_id = json_extract(requested.payload_json, '$.delegatedTaskId') + WHERE requested.thread_id = child.thread_id + AND requested.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + AND task.child_thread_id = child.thread_id + AND json_extract(task.payload_json, '$.origin') = 'app_owned' + AND json_extract(task.payload_json, '$.result') IS NULL + ) OR ( + -- The original task still waits for the latest unheld work. + (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 ( SELECT 1 FROM orchestration_v2_projection_context_transfers WHERE source_thread_id = child.thread_id AND target_thread_id = json_extract(child.payload_json, '$.lineage.parentThreadId') AND type = 'subagent_result' - ) + ))) ELSE 0 END `; case "runtime": @@ -6121,7 +6140,15 @@ export const layerMemory: Layer.Layer = Layer.effect( Effect.gen(function* () { const projections = (yield* Ref.get(replayState)).projections; return [...projections.values()] - .filter((projection) => needsRecovery(projection, kind)) + .filter((projection) => + needsRecovery( + projection, + kind, + projection.thread.lineage.parentThreadId === null + ? undefined + : projections.get(projection.thread.lineage.parentThreadId), + ), + ) .toSorted( (left, right) => DateTime.toEpochMillis(left.thread.updatedAt) - diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx index 328a14575faa2dfa78e82b83a3987f7b199a277b..4c637d13c05d75b865a9b4dd8a29572c2b3a1b3a 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx @@ -170,23 +170,35 @@ function relationshipThreadTitle(input: { } /** - * A delegated task settles with its first run, but the parent can keep sending - * the child follow-ups. While the child thread has a live run, the row's timer - * and hover card follow that run instead of the settled task. + * The row's timer and hover card follow current child work, including queued + * follow-ups whose provider turn has not started yet. */ -function liveSubagent( +function currentSubagent( agent: Agent | undefined, childThread: OrchestrationV2ThreadShell | null | undefined, ): Agent | undefined { - const liveStatus = childThread?.activityRunStatus; - if (!agent || !liveStatus) return agent; - const startedAt = childThread.activityRunStartedAt; + const liveStatus = + childThread?.activityRunStatus ?? (childThread?.status === "queued" ? "queued" : null); + if (!agent || !childThread) return agent; + const newerRun = + childThread.latestRunRequestedAt && + DateTime.toEpochMillis(childThread.latestRunRequestedAt) > + Date.parse(agent.completedAt ?? agent.startedAt ?? agent.updatedAt); + if (!liveStatus && !newerRun) return agent; + const status = liveStatus ?? childThread.status; + const startedAt = liveStatus ? childThread.activityRunStartedAt : childThread.latestRunStartedAt; + const completedAt = liveStatus ? null : childThread.latestRunCompletedAt; return { ...agent, - status: liveStatus === "running" || liveStatus === "waiting" ? liveStatus : "pending", + status: + status === "preparing" || status === "starting" || status === "queued" + ? "pending" + : status === "rolled_back" + ? "interrupted" + : status, startedAt: startedAt ? DateTime.formatIso(startedAt) : null, - completedAt: null, - // The settled task's output belongs to its first run, not this one. + completedAt: completedAt ? DateTime.formatIso(completedAt) : null, + // The task's output belongs to its recorded run, not newer child work. progress: null, result: null, error: null, @@ -380,7 +392,7 @@ export function ThreadRelationshipsPanel(props: { ? BotIcon : GitForkIcon; const relationship = relationshipLabel(edge, props.threadId); - const agent = liveSubagent( + const agent = currentSubagent( isSubagent && !isParent ? subagentsByThreadId.get(threadId) : undefined, node?.thread, ); diff --git a/packages/client-runtime/src/state/threadRelationships.test.ts b/packages/client-runtime/src/state/threadRelationships.test.ts index d96fba5aab2251ab6fcfd2994d85f658061bf53d..5077dfbb1ca7b5527b4236e8f5126781c96571f4 100644 --- a/packages/client-runtime/src/state/threadRelationships.test.ts +++ b/packages/client-runtime/src/state/threadRelationships.test.ts @@ -188,11 +188,17 @@ describe("thread relationships", () => { }); it.each([ - ["running", "running"], - [null, "completed"], + ["running", "running", "running"], + ["waiting", "completed", "waiting"], + [null, "queued", "queued"], + [null, "failed", "failed"], + [null, "interrupted", "interrupted"], + [null, "cancelled", "cancelled"], + [null, "completed", "completed"], + [null, "rolled_back", "interrupted"], ])( - "shows a subagent whose child activity is %s as %s after its delegated task settled", - (childActivity, expected) => { + "shows child activity %s and latest status %s as %s after its delegated task settled", + (childActivity, childStatus, expected) => { const parent = ThreadId.make("thread-parent"); const child = ThreadId.make("thread-child"); const graph = deriveThreadRelationshipGraph({ @@ -200,8 +206,9 @@ describe("thread relationships", () => { { id: parent, status: "completed", forkedFrom: null, lineage: { parentThreadId: null } }, { id: child, - status: childActivity ?? "completed", + status: childStatus, activityRunStatus: childActivity, + latestRunId: "run-child-followup", forkedFrom: null, lineage: { parentThreadId: parent, relationshipToParent: "subagent" }, }, diff --git a/packages/client-runtime/src/state/threadRelationships.ts b/packages/client-runtime/src/state/threadRelationships.ts index 4ef792a6ada79dd1898a15e0ce263ad234993f5b..74f8893684f0321bea4e2b7c892758861cf6d915 100644 --- a/packages/client-runtime/src/state/threadRelationships.ts +++ b/packages/client-runtime/src/state/threadRelationships.ts @@ -71,7 +71,12 @@ export function deriveThreadRelationshipGraph(input: { const addEdge = (edge: ThreadRelationshipEdge) => { ensureNode(edge.sourceThreadId); ensureNode(edge.targetThreadId); - edgesByKey.set(edgeKey(edge), edge); + edgesByKey.set( + edgeKey(edge), + edge.kind === "subagent" && edge.status === "rolled_back" + ? { ...edge, status: "interrupted" } + : edge, + ); }; for (const thread of threads) { @@ -92,14 +97,15 @@ export function deriveThreadRelationshipGraph(input: { const ownerThreadId = input.projection.thread.id; for (const subagent of input.projection.subagents) { if (subagent.childThreadId === null) continue; - // The subagent record settles with the delegated task's first run, but the - // parent can keep sending the child follow-ups. A live run on the child - // thread outranks that settled status. + const child = threadsById.get(subagent.childThreadId); + // The row describes the child conversation, including queued follow-ups + // and their outcomes. A child without an app run still uses its task. addEdge({ sourceThreadId: ownerThreadId, targetThreadId: subagent.childThreadId, kind: "subagent", - status: threadsById.get(subagent.childThreadId)?.activityRunStatus ?? subagent.status, + status: + child?.activityRunStatus ?? (child?.latestRunId != null ? child.status : subagent.status), }); } for (const transfer of input.projection.contextTransfers) { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 591420ce73c660d25db01dd709e53695653b3f22..423f8234178b902a4515d00bfb5ea7927f0b2b74 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -561,6 +561,8 @@ export const OrchestrationV2Run = Schema.Struct({ contextHandoffId: Schema.NullOr(ContextHandoffId), /** Links server-generated restart continuations to the interrupted run. */ restartContinuationOfRunId: Schema.optional(RunId), + /** The delegated task this run answers, including a follow-up in an existing child thread. */ + delegatedTaskId: Schema.optional(NodeId), /** * Set on wake runs (background notifications, delegated task results, * restart continuations): when the work they continue started. Read it diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 0ab3e0b9b86ce85d3f799fb1822ae9472eeb5947..f0ffa07bd401c667971e04d8e820833afbc19fdf 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -434,6 +434,8 @@ export const OrchestratorMcpThreadSendResult = Schema.Struct({ threadId: ThreadId, messageId: MessageId, runId: RunId, + /** Fresh task for a follow-up to an app-owned child; the previous task stays terminal. */ + taskId: Schema.optional(NodeId), status: OrchestrationV2RunStatus, delivery: Schema.Literals(["started", "queued", "steered", "restarted"]), });