diff --git a/apps/server/src/mcp/OrchestratorMcpService.queued-tasks.test.ts b/apps/server/src/mcp/OrchestratorMcpService.queued-tasks.test.ts new file mode 100644 index 000000000000..0b621817e591 --- /dev/null +++ b/apps/server/src/mcp/OrchestratorMcpService.queued-tasks.test.ts @@ -0,0 +1,251 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + EnvironmentId, + MessageId, + NodeId, + type OrchestrationV2ThreadProjection, + type OrchestrationV2ServerCommand, + ProjectId, + ProviderInstanceId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; + +const parentThreadId = ThreadId.make("thread:queued-task-parent"); +const childThreadId = ThreadId.make("thread:queued-task-child"); +const taskId = NodeId.make("node:queued-task"); +const now = DateTime.makeUnsafe("2026-10-03T11:00:00Z"); +const providerInstanceId = ProviderInstanceId.make("claude_two"); +const scope: McpInvocationScope = { + requestNamespace: "test", + client: undefined, + environmentId: EnvironmentId.make("environment:queued-task"), + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:queued-task", + providerInstanceId, + }, + capabilities: new Set(["orchestration"]), + issuedAt: 1, +}; + +function fixture( + options: { + published?: boolean; + executing?: boolean; + failCancellation?: boolean; + secondQueued?: boolean; + deliveryPending?: boolean; + } = {}, +) { + const projectId = ProjectId.make("project:queued-task"); + const thread = { + id: childThreadId, + projectId, + title: "Child", + modelSelection: { instanceId: providerInstanceId, model: "claude-opus-5-5" }, + lineage: { parentThreadId, relationshipToParent: "subagent", rootThreadId: parentThreadId }, + createdAt: now, + updatedAt: now, + deletedAt: null, + archivedAt: null, + settledOverride: null, + settledAt: null, + runtimeMode: "full-access", + interactionMode: "default", + }; + const run = (ordinal: number, status: string) => ({ + id: RunId.make(`run:queued-task:${ordinal}`), + threadId: childThreadId, + ordinal, + status, + userMessageId: MessageId.make(`message:queued-task:${ordinal}`), + requestedAt: now, + startedAt: status === "queued" ? null : now, + completedAt: status === "failed" || status === "completed" ? now : null, + queueHeld: status === "queued", + modelSelection: thread.modelSelection, + providerInstanceId, + }); + const child = { + thread, + runs: [ + run(1, "failed"), + run(2, "queued"), + run(3, "completed"), + run(4, "completed"), + ...(options.executing ? [run(5, "running")] : []), + ...(options.secondQueued ? [run(6, "queued")] : []), + ], + messages: [ + { + id: MessageId.make("result:queued-task:4"), + runId: RunId.make("run:queued-task:4"), + role: "assistant", + text: "Latest completed follow-up result", + updatedAt: now, + }, + ], + turnItems: [], + providerThreads: [{ status: "idle", pendingBackgroundTasks: [] }], + runtimeRequests: [], + contextTransfers: [], + subagents: [], + } as unknown as OrchestrationV2ThreadProjection; + const parent = { + thread: { + ...thread, + id: parentThreadId, + lineage: { parentThreadId: null, relationshipToParent: null }, + }, + runs: [], + contextTransfers: [], + subagents: [ + { + id: taskId, + threadId: parentThreadId, + childThreadId, + origin: "app_owned", + status: options.published ? "completed" : "running", + providerInstanceId, + model: "claude-opus-5-5", + result: options.published ? "Published original result" : null, + completionDelivery: { state: options.deliveryPending ? "pending" : "disposed" }, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const commands: OrchestrationV2ServerCommand[] = []; + const layer = OrchestratorMcpService.layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(child.thread as never), + getThreadRecords: (id) => Effect.succeed(id === parentThreadId ? parent : child), + getProjectThreadRecords: () => Effect.succeed(child), + stopDelegatedTasks: () => Effect.void, + delegatedTaskResultPending: () => Effect.succeed(true), + getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }), + dispatch: (command) => + Effect.suspend(() => { + commands.push(command); + return options.failCancellation && command.type === "thread.stop" + ? Effect.fail(new Error("Cancellation failed") as never) + : Effect.succeed({} as never); + }), + }), + ServerSettings.layerTest({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ProjectService.ProjectService)({}), + ), + ), + ); + return { child, parent, commands, layer }; +} + +it.effect("reports an old held continuation as queued and keeps the latest result readable", () => { + const { commands, layer } = fixture(); + return Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const task = yield* service.taskStatus(scope, taskId); + assert.equal(task.status, "queued"); + assert.equal(task.workState, "working"); + // Held input remains visible in the queue counts, but does not count as active child work. + assert.equal(task.hasPendingChildRuns, false); + assert.equal(task.latestTerminalRunId, RunId.make("run:queued-task:4")); + assert.equal(task.latestTerminalSummary, "Latest completed follow-up result"); + assert.isNull(task.summary); + const read = yield* service.readThread(scope, { threadId: childThreadId, runLimit: 1 }); + assert.equal(read.thread.status, "completed"); + assert.isNull(read.thread.activeRunId); + assert.equal(read.thread.pendingRequestCount, 0); + assert.equal(read.thread.queuedRunCount, 1); + assert.equal(read.thread.heldQueuedRunCount, 1); + assert.equal(read.recentRuns.length, 1); + assert.deepEqual(commands, []); + }).pipe(Effect.provide(layer)); +}); + +it.effect("stops queued-only unpublished work while preserving held input", () => { + const { commands, child, layer } = fixture({ secondQueued: true }); + return Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const result = yield* service.cancelTask(scope, { taskId, clientRequestId: "cancel-held" }); + assert.equal(result.status, "cancel_requested"); + assert.deepEqual( + commands.map((command) => command.type), + ["thread.stop"], + ); + const command = commands[0]!; + if (command.type !== "thread.stop") return yield* Effect.die("Expected thread stop"); + assert.equal(command.threadId, childThreadId); + assert.deepEqual( + child.runs.filter((run) => run.status === "queued").map((run) => run.queueHeld), + [true, true], + ); + }).pipe(Effect.provide(layer)); +}); + +it.effect("keeps truly executing child work running ahead of held queued work", () => { + const { layer } = fixture({ executing: true }); + return Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const result = yield* service.taskStatus(scope, taskId); + assert.equal(result.status, "running"); + assert.equal(result.hasPendingChildRuns, true); + }).pipe(Effect.provide(layer)); +}); + +it.effect("keeps a published result terminal while stopping later backing-thread work", () => { + const { commands, layer } = fixture({ published: true, executing: true }); + return Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const status = yield* service.taskStatus(scope, taskId); + assert.equal(status.status, "completed"); + assert.equal(status.summary, "Published original result"); + assert.equal(status.hasPendingChildRuns, true); + const cancel = yield* service.cancelTask(scope, { + taskId, + clientRequestId: "cancel-published", + }); + assert.equal(cancel.status, "completed"); + assert.deepEqual( + commands.map((command) => command.type), + ["thread.stop"], + ); + }).pipe(Effect.provide(layer)); +}); + +it.effect("does not dispose completion delivery when stopping the child fails", () => { + const { commands, layer } = fixture({ failCancellation: true, deliveryPending: true }); + return Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const error = yield* service + .cancelTask(scope, { taskId, clientRequestId: "cancel-fails" }) + .pipe(Effect.flip); + assert.equal(error.code, "task_not_cancellable"); + assert.deepEqual( + commands.map((command) => command.type), + ["thread.stop"], + ); + }).pipe(Effect.provide(layer)); +}); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index f9bb61fa810e..185454dbf2b4 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -766,6 +766,10 @@ function threadDetail( pendingRequestCount: projection.runtimeRequests.filter( (request) => request.status === "pending", ).length, + queuedRunCount: projection.runs.filter((run) => run.status === "queued").length, + heldQueuedRunCount: projection.runs.filter( + (run) => run.status === "queued" && run.queueHeld === true, + ).length, archived: projection.thread.archivedAt !== null, ...threadSettlement(projection.thread), // From the shell, like the list, so read and list agree on snooze state. @@ -1352,6 +1356,10 @@ const make = Effect.gen(function* () { const childRun = delegatedTaskRun(childControls, task); const terminalRun = latestTerminalResultRun(childControls, childRun); const progress = delegatedTaskProgress(childControls); + const pendingRuns = childControls.runs + .filter((run) => !ThreadManagementService.isTerminalRunStatus(run.status)) + .toSorted((a, b) => b.ordinal - a.ordinal); + const pendingRun = pendingRuns.find((run) => run.status !== "queued") ?? pendingRuns[0]; const taskResultRun = childRun?.delegatedTaskId === task.id ? childControls.runs @@ -1404,7 +1412,9 @@ const make = Effect.gen(function* () { ) : workState === "result_available" ? taskStatusForRun(taskResultRun ?? childRun) - : taskStatusForRun(taskResultRun ?? childRun) === "queued" + : pendingRun !== undefined + ? taskStatusForRun(pendingRun) + : taskStatusForRun(taskResultRun ?? childRun) === "queued" ? "queued" : "running"; const derivedResult = @@ -2222,11 +2232,12 @@ const make = Effect.gen(function* () { const child = yield* loadProjection(current.childThreadId); if ( current.workState !== "waiting_for_children" && - ThreadManagementService.latestActiveRun(child) === undefined + ThreadManagementService.latestActiveRun(child) === undefined && + !child.runs.some((run) => run.status === "queued") ) { return yield* failure( "task_not_cancellable", - `Delegated task ${input.taskId} has no interruptible child run.`, + `Delegated task ${input.taskId} has no interruptible or queued child run.`, ); } yield* stopWithinLimit; diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index f2c34bb22cbe..b7122178a84e 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -206,7 +206,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "Read durable state and a paginated timeline from any T3 thread in this environment. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", + "Read durable state and a paginated timeline from any T3 thread in this environment. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. pendingRequestCount counts provider approval/input requests; queuedRunCount and heldQueuedRunCount report unstarted message runs independently of the latest run status. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index ad0c6703c1d9..693e14b0a432 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -37,6 +37,7 @@ import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryService.ts"; import { continueRestartedRun } from "./RestartContinuation.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index d6bd2b1b823f..6d6bccc7be35 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -2781,6 +2781,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { providerTurns: [{ providerThreadId, runAttemptId: attemptId, status: "running" }], runtimeRequests: [], turnItems: [], + subagents: [], } as unknown as OrchestrationV2ThreadProjection; const recovery = yield* ProviderRuntimeRecovery.make.pipe( Effect.provide( diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 05c2b41c0116..c109341ea87c 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -147,6 +147,7 @@ it.effect("expires orphaned runtime requests before command readiness", () => { providerThreads: [], runs: [], nodes: [], + subagents: [], } as unknown as OrchestrationV2ThreadProjection; const layer = ProviderRuntimeRecovery.layer.pipe( Layer.provide(ServerSettings.layerTest()), @@ -198,6 +199,7 @@ it.effect("preserves async questions across startup and shutdown", () => { runs: [], nodes: [], turnItems: [], + subagents: [], } as unknown as OrchestrationV2ThreadProjection; const commitCommand = vi.fn(() => Effect.die("an async question needs no process-loss write")); const layer = ProviderRuntimeRecovery.layer.pipe( @@ -247,6 +249,7 @@ it.effect("uses the same reconciliation path to cancel runtime requests during s providerThreads: [], runs: [], nodes: [], + subagents: [], } as unknown as OrchestrationV2ThreadProjection; const layer = ProviderRuntimeRecovery.layer.pipe( Layer.provide(ServerSettings.layerTest()), @@ -1155,9 +1158,9 @@ it.effect( }, ); -it.effect( - "terminalizes a leftover nonpersistent dynamic_tool on a settled run after process loss", - () => { +it.effect.each(["completed", "failed", "interrupted", "cancelled"] as const)( + "terminalizes a leftover nonpersistent dynamic_tool on a %s run after process loss", + (runStatus) => { const threadId = ThreadId.make("thread_recovery_orphan_wait"); const settledRunId = RunId.make("run_recovery_orphan_wait_settled"); const providerThreadId = ProviderThreadId.make("provider_thread_recovery_orphan_wait"); @@ -1186,7 +1189,7 @@ it.effect( }, ], providerTurns: [], - runs: [{ id: settledRunId, status: "completed", providerInstanceId: codexInstanceId }], + runs: [{ id: settledRunId, status: runStatus, providerInstanceId: codexInstanceId }], attempts: [], nodes: [ { id: orphanWaitNodeId, runId: settledRunId, status: "running", kind: "tool_call" }, @@ -1280,9 +1283,9 @@ it.effect( }, ); -it.effect( - "terminalizes the linked subagent and node for a stale subagent item on a settled run", - () => { +it.effect.each(["completed", "running", "waiting"] as const)( + "cancels native tasks on a %s parent run while preserving app-owned task ownership", + (runStatus) => { const threadId = ThreadId.make("thread_recovery_subagent"); const settledRunId = RunId.make("run_recovery_subagent_settled"); const providerThreadId = ProviderThreadId.make("provider_thread_recovery_subagent"); @@ -1291,6 +1294,18 @@ it.effect( const staleItemId = TurnItemId.make("turn_item_recovery_subagent_stale"); const doneItemId = TurnItemId.make("turn_item_recovery_subagent_done"); const claudeInstanceId = ProviderInstanceId.make("claude"); + const appOwnedTasks = (["pending", "running", "waiting", "completed"] as const).map( + (status) => ({ + id: NodeId.make(`node_recovery_app_owned_${status}`), + runId: settledRunId, + driver: ProviderDriverKind.make("claude"), + providerInstanceId: claudeInstanceId, + origin: "app_owned" as const, + childThreadId: ThreadId.make(`thread_recovery_app_owned_${status}`), + status, + result: status === "completed" ? "Published child result" : null, + }), + ); let committedInput: Parameters[0] | null = null; const projection = { @@ -1307,12 +1322,12 @@ it.effect( }, ], providerTurns: [], - // Settled run: the stale-item loop owns it, not the nonterminal loop. - runs: [{ id: settledRunId, status: "completed", providerInstanceId: claudeInstanceId }], + runs: [{ id: settledRunId, status: runStatus, providerInstanceId: claudeInstanceId }], attempts: [], nodes: [ { id: staleSubagentNodeId, runId: settledRunId, status: "running" }, { id: doneSubagentNodeId, runId: settledRunId, status: "completed" }, + ...appOwnedTasks.map((task) => ({ id: task.id, runId: settledRunId, status: task.status })), ], subagents: [ { @@ -1333,6 +1348,7 @@ it.effect( status: "completed", result: "done", }, + ...appOwnedTasks, ], messages: [], turnItems: [ @@ -1346,6 +1362,16 @@ it.effect( subagentId: staleSubagentNodeId, providerInstanceId: claudeInstanceId, }, + ...appOwnedTasks.map((task) => ({ + id: TurnItemId.make(`turn_item_${task.id}`), + runId: settledRunId, + nodeId: task.id, + providerThreadId, + type: "subagent", + status: task.status === "completed" ? "running" : task.status, + subagentId: task.id, + providerInstanceId: claudeInstanceId, + })), { id: doneItemId, runId: settledRunId, @@ -1388,6 +1414,20 @@ it.effect( yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup"); const events = committedInput?.events ?? []; + // App-owned children recover from their own runs, including a held queue. + // Neither parent process loss nor a stale item may rewrite their results. + for (const task of appOwnedTasks) { + assert.isFalse( + events.some( + (event) => + (event.type === "subagent.updated" && event.payload.id === task.id) || + (event.type === "node.updated" && event.payload.id === task.id) || + (event.type === "turn-item.updated" && event.payload.nodeId === task.id), + ), + ); + } + assert.equal(appOwnedTasks[3]?.result, "Published child result"); + // Only the nonterminal subagent item is cancelled. const turnItemCancels = events.filter( (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled", diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index ad57247e0384..aaf03d6d3d30 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -3,6 +3,7 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { CommandId, type OrchestrationV2DomainEvent, + type NodeId, type ProviderThreadId, type OrchestrationV2RestartCancelledBackgroundWork, type OrchestrationV2Subagent, @@ -113,6 +114,19 @@ function isAppOwnedDelegationItem( return item.type === "subagent" && isAppOwnedDelegation(item); } +function appOwnedSubagentIds(projection: ProjectionStore.ProjectionRuntimeRecoveryState) { + return new Set( + projection.subagents.filter((task) => task.origin === "app_owned").map((task) => task.id), + ); +} + +function isAppOwnedSubagentItem( + item: OrchestrationV2ThreadProjection["turnItems"][number], + taskIds: ReadonlySet, +): boolean { + return item.type === "subagent" && taskIds.has(item.subagentId); +} + function providerThreadHasPendingBackgroundTasks( providerThread: OrchestrationV2ThreadProjection["providerThreads"][number], ): boolean { @@ -192,6 +206,9 @@ export const make = Effect.gen(function* () { continueAfterRestart: boolean, ) { const now = yield* DateTime.now; + // App-owned children have their own durable runs and recovery. Losing + // the parent provider session does not cancel the delegated task. + const taskIds = appOwnedSubagentIds(projection); const runs = [] as Array; for (const run of nonterminalRuns(projection)) { if (run.status === "waiting") { @@ -272,6 +289,7 @@ export const make = Effect.gen(function* () { const recordCancelledBackgroundItem = ( item: OrchestrationV2ThreadProjection["turnItems"][number], ) => { + if (isAppOwnedSubagentItem(item, taskIds)) return; if (!isBackgroundCapableTurnItemType(item.type)) return; const work = cancelledTurnItemWork(item); if (work === undefined) return; @@ -345,6 +363,7 @@ export const make = Effect.gen(function* () { for (const node of projection.nodes.filter( (candidate) => candidate.runId === run.id && + !taskIds.has(candidate.id) && !messageRequestNodeIds.has(candidate.id) && !delegatedTaskNodeIds.has(candidate.id) && (candidate.status === "pending" || @@ -398,6 +417,7 @@ export const make = Effect.gen(function* () { for (const item of projection.turnItems.filter( (candidate) => candidate.runId === run.id && + !isAppOwnedSubagentItem(candidate, taskIds) && (candidate.nodeId === null || !messageRequestNodeIds.has(candidate.nodeId)) && !isAppOwnedDelegationItem(candidate) && (candidate.status === "pending" || @@ -426,6 +446,7 @@ export const make = Effect.gen(function* () { const recoveredNonterminalRunIds = new Set(runs.map((run) => run.id)); const cancelledStaleNodeIds = new Set(); for (const item of projection.turnItems ?? []) { + if (isAppOwnedSubagentItem(item, taskIds)) continue; if (item.runId !== null && recoveredNonterminalRunIds.has(item.runId)) { continue; } diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 0738426a4449..70467f55ad77 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -371,6 +371,8 @@ export const OrchestratorMcpThreadDetail = Schema.Struct({ runCount: NonNegativeInt, itemCount: NonNegativeInt, pendingRequestCount: NonNegativeInt, + queuedRunCount: Schema.optionalKey(NonNegativeInt), + heldQueuedRunCount: Schema.optionalKey(NonNegativeInt), archived: Schema.Boolean, settled: Schema.Boolean, settledAt: Schema.NullOr(IsoDateTime), diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.test.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.test.ts index 013a49f0c125..5b9668364aa4 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.test.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.test.ts @@ -3,6 +3,7 @@ import type { OrchestrationV2PendingBackgroundTask } from "@t3tools/contracts"; import { backgroundWorkHoldsCompletion, derivePendingBackgroundWork, + pendingBackgroundTurnItems, turnItemUpdateCanEndBackgroundWork, } from "./orchestrationV2PendingBackgroundWork.ts"; @@ -41,6 +42,72 @@ describe("backgroundWorkHoldsCompletion", () => { }); describe("derivePendingBackgroundWork", () => { + it.each(["failed", "interrupted", "cancelled"] as const)( + "excludes a %s run's orphan handoff while preserving live background work and queued history", + (status) => { + const runs = [ + { id: "run-1" as never, ordinal: 1, status }, + { id: "run-2" as never, ordinal: 2, status: "queued" as const, queueHeld: true }, + { id: "run-3" as never, ordinal: 3, status: "completed" as const }, + { id: "run-4" as never, ordinal: 4, status: "completed" as const }, + ]; + const turnItems = [ + { + id: "handoff", + type: "dynamic_tool" as const, + status: "running" as const, + runId: "run-1", + title: "mcp__t3-code__t3_worktree_handoff", + }, + { + id: "completed-run-background", + type: "dynamic_tool" as const, + status: "running" as const, + runId: "run-3", + title: "Await background result", + }, + { + id: "child", + type: "subagent" as const, + status: "running" as const, + runId: "run-1", + title: "Independent child", + }, + { + id: "command", + type: "command_execution" as const, + status: "running" as const, + runId: "run-1", + title: "Dev server", + }, + ]; + const tasks = derivePendingBackgroundWork({ + latestRun: runs[3], + runs, + turnItems, + providerThreads: [ + { + id: "provider" as never, + pendingBackgroundTasks: [{ taskId: "live-roster", kind: "background_task" as const }], + }, + ], + }); + expect(tasks.map((task) => task.taskId)).toEqual([ + "live-roster", + "completed-run-background", + "child", + "command", + ]); + expect(pendingBackgroundTurnItems({ turnItems, runs }).map((item) => item.id)).toEqual([ + "completed-run-background", + "child", + "command", + ]); + expect(runs[1]).toMatchObject({ status: "queued", queueHeld: true }); + expect(turnItems[0]?.status).toBe("running"); + }, + ); + it("returns empty while the latest run is not settled", () => { const tasks = derivePendingBackgroundWork({ latestRun: { id: "run-1" as never, ordinal: 1, status: "running" }, diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 5a17bad321ed..4234a4a82446 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -104,7 +104,7 @@ type PendingBackgroundWorkTurnItem = { readonly type: OrchestrationV2TurnItem["type"]; readonly status: OrchestrationV2TurnItem["status"]; readonly title: string | null; - /** When present and the run is rolled_back, the item is abandoned, not pending. */ + /** Run ownership distinguishes abandoned foreground tools from live background work. */ readonly runId?: OrchestrationV2Run["id"] | string | null; readonly nativeItemRef?: { readonly nativeId: string | null; @@ -190,15 +190,22 @@ export function pendingBackgroundTurnItems run.status === "rolled_back").map((run) => String(run.id)), ); + const abandonedToolRunIds = new Set( + (input.runs ?? []) + .filter((run) => ["failed", "interrupted", "cancelled"].includes(run.status)) + .map((run) => String(run.id)), + ); return input.turnItems.filter( (item) => BACKGROUND_TURN_ITEM_TYPES.has(item.type) && isOrchestrationV2WorkActive(item.status) && !(item.type === "dynamic_tool" && isPersistentDynamicToolInput(item.input)) && - // Null/absent run id stays eligible; only known rolled_back runs drop. + // A failed foreground MCP call cannot finish after its run has ended. + // Commands and native children can outlive the root; their state remains authoritative. (item.runId === undefined || item.runId === null || - !rolledBackRunIds.has(String(item.runId))), + (!rolledBackRunIds.has(String(item.runId)) && + !(item.type === "dynamic_tool" && abandonedToolRunIds.has(String(item.runId))))), ); } @@ -216,7 +223,7 @@ export function pendingBackgroundTurnItems