diff --git a/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts b/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..04c862940ac8bfd9e1980fcf61090a1725bcfbb7 --- /dev/null +++ b/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts @@ -0,0 +1,357 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EnvironmentId, + MessageId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + ThreadId, + TurnItemId, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; +import * as EffectWorker from "../orchestration-v2/EffectWorker.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; + +const driver = ProviderDriverKind.make("claudeAgent"); +const instanceId = ProviderInstanceId.make("claudeAgent"); +const modelSelection = { instanceId, model: "claude-test" }; + +// A delegated Claude child ends its turn while a background native subagent it +// started keeps running (#17154). Cancelling the task stops the child, which +// ends that subagent's process, so the task must end instead of waiting on it. +it.effect("task_cancel ends a delegated child's background native subagent", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("task-cancel-native-subagent"); + const turns: Array<{ + readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput; + readonly events: Queue.Queue; + }> = []; + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const events = yield* Queue.unbounded(); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd, + model: modelSelection.model, + capabilities: ClaudeProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: input.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + turns.push({ turn, events }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: null, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: () => Effect.die("unused"), + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), + }; + const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter); + const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( + { name: "task-cancel-native-subagent" }, + layerAdapters, + { runEffectWorker: false }, + ); + const layerOrchestration = Layer.merge( + layerOrchestrator, + ThreadManagementService.layer.pipe(Layer.provide(layerOrchestrator)), + ); + const layerTest = OrchestratorMcpService.layer.pipe( + Layer.provideMerge(layerOrchestration), + Layer.provide( + Layer.mergeAll( + layerAdapters, + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + NodeServices.layer, + ), + ), + ); + + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const parentThreadId = ThreadId.make("thread:task-cancel-parent"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const turnRunning = () => + watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:task-cancel"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const parentRunning = yield* turnRunning(); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:start-parent"), + text: "Delegate a review", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(parentRunning); + + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; + const childRunning = yield* turnRunning(); + yield* orchestrator.dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("delegate"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Review the change", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + createdBy: "agent", + creationSource: "mcp", + }); + yield* worker.drain(); + yield* Fiber.join(childRunning); + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + const child = turns.find((entry) => entry.turn.threadId === task.childThreadId)!; + const childTurn = (yield* orchestrator.getThreadProjection(task.childThreadId!)) + .providerTurns[0]!; + + // The child starts a background native subagent, then ends its turn. + const now = yield* DateTime.now; + const subagentId = NodeId.make("node:native-subagent"); + const native = { + threadId: task.childThreadId!, + runId: child.turn.runId, + origin: "provider_native", + driver, + providerInstanceId: instanceId, + providerThreadId: childTurn.providerThreadId, + childThreadId: null, + prompt: "Check the tests", + title: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + } as const; + const waiting = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === child.turn.runId && + event.payload.status === "waiting", + ); + for (const event of [ + { + type: "node.updated", + driver, + node: { + id: subagentId, + threadId: native.threadId, + runId: native.runId, + parentNodeId: child.turn.rootNodeId, + rootNodeId: child.turn.rootNodeId, + kind: "subagent", + status: "running", + countsForRun: false, + providerThreadId: childTurn.providerThreadId, + providerTurnId: childTurn.id, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + { + type: "subagent.updated", + driver, + subagent: { + ...native, + id: subagentId, + parentNodeId: child.turn.rootNodeId, + createdBy: "agent", + nativeTaskRef: null, + model: null, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...native, + id: TurnItemId.make("turn-item:native-subagent"), + nodeId: subagentId, + providerTurnId: childTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + type: "subagent", + subagentId, + }, + }, + { + type: "provider_turn.updated", + driver, + providerTurn: { ...childTurn, status: "completed", completedAt: now }, + }, + { + type: "turn.terminal", + driver, + providerThreadId: childTurn.providerThreadId, + providerTurnId: childTurn.id, + runOrdinal: child.turn.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }, + ] satisfies ReadonlyArray) { + yield* Queue.offer(child.events, event); + } + yield* Fiber.join(waiting); + yield* worker.drain(); + + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:task-cancel"), + requestNamespace: "provider-session:task-cancel", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:task-cancel", + providerInstanceId: instanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + const before = yield* service.taskStatus(scope, task.id); + assert.equal(before.status, "running"); + assert.equal(before.workState, "waiting_for_children"); + + const cancelled = yield* service.cancelTask(scope, { + taskId: task.id, + clientRequestId: "cancel-task", + }); + assert.equal(cancelled.status, "cancel_requested"); + yield* worker.drain(); + + const childAfter = yield* orchestrator.getThreadProjection(task.childThreadId!); + assert.equal( + childAfter.subagents.find((entry) => entry.id === subagentId)?.status, + "interrupted", + ); + assert.equal( + childAfter.nodes.find((entry) => entry.id === subagentId)?.status, + "interrupted", + ); + assert.equal( + childAfter.turnItems.find((item) => item.type === "subagent")?.status, + "interrupted", + ); + // The child's own turn had completed, so the task keeps that result. + const after = yield* service.taskStatus(scope, task.id); + assert.equal(after.status, "completed"); + assert.notEqual(after.workState, "waiting_for_children"); + }).pipe(Effect.provide(layerTest)); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 6d5e0b92d87f8b6315cec15f09a581ba10ccbb4f..d1e62aad91bbcb74e23fadfddb00186cf2c7749d 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -414,6 +414,113 @@ const stopEarlierBackgroundWork = ({ const codexProviderThread = (yield* orchestrator.getThreadProjection(threadId)) .providerThreads[0]!; + // A native subagent's own thread, still showing its runless root turn as working. + const subagentThread = (name: string) => + Effect.gen(function* () { + const subagentThreadId = ThreadId.make(`thread:${name}`); + const rootId = NodeId.make(`node:${name}-root`); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`create-${name}`), + threadId: subagentThreadId, + projectId: ProjectId.make("project:background-work-stop"), + title: name, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "agent", + creationSource: "provider", + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`${name}-root`), + type: "node.updated", + threadId: subagentThreadId, + occurredAt: now, + payload: { + id: rootId, + threadId: subagentThreadId, + runId: null, + parentNodeId: null, + rootNodeId: rootId, + kind: "root_turn", + status: "running", + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + ], + }); + return { threadId: subagentThreadId, rootId }; + }); + const { threadId: reviewerThreadId, rootId: reviewerRootId } = + yield* subagentThread("reviewer"); + // The reviewer starts a tester, which runs on a thread of its own. + const tester = yield* subagentThread("tester"); + const nestedSubagentId = NodeId.make("subagent:nested"); + const nestedItemId = TurnItemId.make("turn-item:nested-subagent"); + const nested = { + threadId: reviewerThreadId, + runId: null, + origin: "provider_native", + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: tester.threadId, + prompt: "Check the tests", + title: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + } as const; + yield* sink.write({ + events: [ + // A subagent the reviewer started, which the provider records on the + // reviewer's own thread. + { + id: EventId.make("nested-subagent-item"), + type: "turn-item.updated", + threadId: reviewerThreadId, + occurredAt: now, + payload: { + ...nested, + id: nestedItemId, + nodeId: nestedSubagentId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + type: "subagent", + subagentId: nestedSubagentId, + }, + }, + { + id: EventId.make("nested-subagent"), + type: "subagent.updated", + threadId: reviewerThreadId, + occurredAt: now, + payload: { + ...nested, + id: nestedSubagentId, + parentNodeId: reviewerRootId, + createdBy: "agent", + nativeTaskRef: null, + model: null, + }, + }, + ], + }); // A settled run with its root node, attempt, provider turn and, // optionally, a command or native subagent it left running. const settledRun = (input: { @@ -550,11 +657,35 @@ const stopEarlierBackgroundWork = ({ origin: "provider_native", driver, providerInstanceId: instanceId, - childThreadId: null, + childThreadId: reviewerThreadId, prompt: "Review the change", result: null, }, }); + if (item.kind === "subagent") { + events.push({ + id: EventId.make(`subagent:${input.ordinal}`), + type: "subagent.updated", + threadId, + runId, + occurredAt: now, + payload: { + ...base, + id: NodeId.make(`subagent:${input.ordinal}`), + parentNodeId: nodeId, + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: reviewerThreadId, + nativeTaskRef: null, + prompt: "Review the change", + model: null, + result: null, + }, + }); + } } return { runId, providerTurnId, events }; }; @@ -747,6 +878,17 @@ const stopEarlierBackgroundWork = ({ ? ["interrupted", "interrupted"] : ["interrupted", "interrupted", "interrupted"], ); + // The reviewer died with its provider process, so it no longer reads as running. + assert.equal(after.subagents[0]?.status, "interrupted"); + const reviewerThread = yield* orchestrator.getThreadProjection(reviewerThreadId); + assert.equal(reviewerThread.nodes[0]?.status, "interrupted"); + assert.equal(reviewerThread.subagents[0]?.status, "interrupted"); + assert.equal( + reviewerThread.turnItems.find((item) => item.id === nestedItemId)?.status, + "interrupted", + ); + const testerThread = yield* orchestrator.getThreadProjection(tester.threadId); + assert.equal(testerThread.nodes[0]?.status, "interrupted"); }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry( diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 9a1e43a8b44b42d156934cd31a0f466d90424c60..f791977235c95f8afcc4e1c90e75f050ad62822d 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -3052,9 +3052,9 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { completedAt: null, }); // The parent run settled while its background subagent kept working, - // then the server died. Recovery already cancels the parent's subagent - // item, entity, and node; the child's runless root turn lives on another - // thread and must be settled too. + // Stop ended the subagent's item, then the server died. Recovery cancels + // the parent's subagent entity and node; the child's runless root turn + // lives on another thread and must be settled too. yield* eventSink.commitCommand({ commandId: CommandId.make("command:foundation-native-subagent"), threadId: parentId, @@ -3186,7 +3186,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { parentItemId: null, ordinal: 1, type: "subagent", - status: "running", + status: "interrupted", title: null, startedAt: now, completedAt: null, @@ -3222,6 +3222,10 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { const parentProjection = yield* projectionStore.getThreadProjection(parentId); assert.equal(parentProjection.subagents[0]?.status, "cancelled"); + assert.equal( + parentProjection.nodes.find((node) => node.id === subagentId)?.status, + "cancelled", + ); const childProjection = yield* projectionStore.getThreadProjection(childId); const childRoot = childProjection.nodes.find((candidate) => candidate.id === childRootId); assert.equal(childRoot?.status, "cancelled"); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d1e1dd1e7a27b0a097860b39797dfd4ecac770f6..908fe09467748fdf06162163b6fac7b37697ea06 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -117,6 +117,8 @@ import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts"; import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; import { RuntimePolicyV2 } from "./RuntimePolicy.ts"; import { + endOrphanedNativeSubagent, + endRunlessRootTurns, makeSubagentChildThread, subagentResultForRun, delegatedTaskProgress, @@ -8308,7 +8310,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly events: Ref.Ref>; readonly projection: Pick< OrchestrationV2ThreadProjection, - "runs" | "turnItems" | "providerThreads" + "runs" | "nodes" | "subagents" | "turnItems" | "providerThreads" >; readonly stoppedProviderThreadId: OrchestrationV2ProviderThread["id"] | null; readonly throughRunOrdinal: number; @@ -8317,6 +8319,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.gen(function* () { const emitEvent = emit(input.events, input.command); const runOrdinals = new Map(input.projection.runs.map((run) => [run.id, run.ordinal])); + const allocateEventId = () => + idAllocator.allocate.event({ + threadId: input.command.threadId, + commandId: input.command.commandId, + }); + const childThreadIds = new Set(); const liveness = new Map(); const hasLiveSession = (providerThreadId: OrchestrationV2ProviderThread["id"]) => Effect.gen(function* () { @@ -8366,6 +8374,61 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: input.now, }, }); + if (item.type !== "subagent") continue; + // A native subagent died with the process too: end its record, its own + // thread's runless root turn, and the subagents it started there, which + // a provider can record on that thread rather than this one. + const dying: Array<{ + readonly projection: Pick< + OrchestrationV2ThreadProjection, + "nodes" | "subagents" | "turnItems" + >; + readonly subagent: Extract; + }> = [{ projection: input.projection, subagent: item }]; + for (const { projection, subagent } of dying) { + const ended = yield* endOrphanedNativeSubagent({ + projection, + subagentId: subagent.subagentId, + status: "interrupted", + now: input.now, + emitted: yield* Ref.get(input.events), + allocateEventId, + }).pipe(mapDispatchError(input.command)); + yield* Ref.update(input.events, (events) => [...events, ...ended]); + const childThreadId = subagent.childThreadId; + if ( + subagent.origin !== "provider_native" || + childThreadId === null || + childThreadIds.has(childThreadId) + ) { + continue; + } + childThreadIds.add(childThreadId); + const child = yield* projectionStore + .getThreadRecords(childThreadId, ["nodes", "subagents", "turnItems"], { + // Runless items are the subagent's; any run there is the user's own. + turnItemRunIds: [null], + turnItemStatuses: ["pending", "running", "waiting"], + }) + .pipe( + Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), + mapDispatchError(input.command), + ); + if (child === null) continue; + const childEnded = yield* endRunlessRootTurns({ + threadId: childThreadId, + providerInstanceId: subagent.providerInstanceId, + projection: child, + status: "interrupted", + now: input.now, + emitted: yield* Ref.get(input.events), + allocateEventId, + }).pipe(mapDispatchError(input.command)); + yield* Ref.update(input.events, (events) => [...events, ...childEnded]); + for (const nested of child.turnItems) { + if (nested.type === "subagent") dying.push({ projection: child, subagent: nested }); + } + } } for (const providerThread of input.projection.providerThreads) { // A live process owns its roster and reports clearing it. diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 51343fd8af916e93c2f3a705d34e86f70055aa2c..200f1548d700a79f2a872bc8e881ba4a6ca736cc 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -589,6 +589,11 @@ function needsRecovery( (run.status === "queued" && run.queueHeld !== true), ) || projection.runtimeRequests.some((request) => request.status === "pending") || + projection.subagents.some( + (subagent) => + subagent.origin === "provider_native" && + ["pending", "running", "waiting"].includes(subagent.status), + ) || projection.providerSessions.some( (session) => session.status !== "stopped" && session.status !== "error", ) || @@ -3664,6 +3669,11 @@ export const layer: Layer.Layer = AND node.status IN ('pending', 'running', 'waiting') ) UNION + -- No native subagent outlives the provider process that ran it. + SELECT thread_id FROM orchestration_v2_projection_subagents + WHERE origin = 'provider_native' + AND status IN ('pending', 'running', 'waiting') + UNION SELECT item.thread_id FROM orchestration_v2_projection_turn_items AS item WHERE NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS run @@ -4067,6 +4077,11 @@ export const layer: Layer.Layer = WHERE item.thread_id = ${threadId} AND item.type = 'subagent' AND item.status IN ('pending', 'running', 'waiting') ) + OR node.node_id IN ( + SELECT subagent_id FROM orchestration_v2_projection_subagents + WHERE thread_id = ${threadId} AND origin = 'provider_native' + AND status IN ('pending', 'running', 'waiting') + ) ) ORDER BY COALESCE(node.started_at, ''), node.node_id ASC `, @@ -4075,7 +4090,8 @@ export const layer: Layer.Layer = WHERE subagent.thread_id = ${threadId} AND subagent.status IN ('pending', 'starting', 'running', 'waiting') AND ( - subagent.run_id IN ( + subagent.origin = 'provider_native' + OR subagent.run_id IN ( SELECT run_id FROM orchestration_v2_projection_runs WHERE thread_id = ${threadId} AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index da764d41d73011cd93e853327a9753348e5bc215..c8802a30f4d4248c6993e8f39b9400fd8c3d19f6 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -1227,6 +1227,7 @@ it.effect( { id: staleSubagentNodeId, runId: settledRunId, + origin: "provider_native", driver: ProviderDriverKind.make("claude"), providerInstanceId: claudeInstanceId, status: "running", @@ -1235,6 +1236,7 @@ it.effect( // Already finished with a real result: must never be overwritten. id: doneSubagentNodeId, runId: settledRunId, + origin: "provider_native", driver: ProviderDriverKind.make("claude"), providerInstanceId: claudeInstanceId, status: "completed", diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 7a063bec065bc60e11cbb6023526b4c9c5265842..b3c74621302e2f4f203c498df934ad969442a3c9 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -22,6 +22,10 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; +import { + endOrphanedNativeSubagent, + endRunlessRootTurns, +} from "@t3tools/provider-core/server/subagentProjection"; import { cancelledRosterTaskWork, cancelledTurnItemWork, @@ -88,10 +92,6 @@ function isNonterminalTurnItemStatus(status: string): boolean { return status === "pending" || status === "running" || status === "waiting"; } -function isNonterminalSubagentStatus(status: string): boolean { - return status === "pending" || status === "running" || status === "waiting"; -} - function isNonterminalNodeStatus(status: string): boolean { return status === "pending" || status === "running" || status === "waiting"; } @@ -362,26 +362,6 @@ export const make = Effect.gen(function* () { payload: { ...node, status: "cancelled", completedAt: now }, }); } - for (const subagent of projection.subagents.filter( - (candidate) => - candidate.runId === run.id && - !isAppOwnedDelegation(candidate) && - (candidate.status === "pending" || - candidate.status === "running" || - candidate.status === "waiting"), - )) { - events.push({ - id: yield* allocateEventId(), - type: "subagent.updated", - threadId: projection.thread.id, - runId: run.id, - nodeId: subagent.id, - driver: subagent.driver, - providerInstanceId: subagent.providerInstanceId, - occurredAt: now, - payload: { ...subagent, status: "cancelled", completedAt: now, updatedAt: now }, - }); - } for (const providerTurn of projection.providerTurns.filter( (candidate) => candidate.runAttemptId !== null && @@ -486,103 +466,34 @@ export const make = Effect.gen(function* () { }); } } - if (item.type !== "subagent") { - continue; - } - // Cancelling only the turn item would leave the linked subagent entity - // non-terminal forever, since the dead provider process can no longer - // emit its terminal event. Match the exact linked id so a subagent - // that already finished is never overwritten. - const staleSubagent = projection.subagents.find( - (candidate) => - candidate.id === item.subagentId && isNonterminalSubagentStatus(candidate.status), - ); - if (staleSubagent !== undefined) { - events.push({ - id: yield* allocateEventId(), - type: "subagent.updated", - threadId: projection.thread.id, - ...(item.runId === null ? {} : { runId: item.runId }), - nodeId: staleSubagent.id, - driver: staleSubagent.driver, - providerInstanceId: staleSubagent.providerInstanceId, - occurredAt: now, - payload: { ...staleSubagent, status: "cancelled", completedAt: now, updatedAt: now }, - }); - } - const staleSubagentNode = projection.nodes.find( - (candidate) => - candidate.id === item.subagentId && isNonterminalNodeStatus(candidate.status), + } + // No provider process outlives the server, so a native subagent still open + // here never reports its end, whatever state its run and item are in. + for (const subagent of projection.subagents ?? []) { + events.push( + ...(yield* endOrphanedNativeSubagent({ + projection, + subagentId: subagent.id, + status: "cancelled", + now, + emitted: events, + allocateEventId, + })), ); - if (staleSubagentNode !== undefined && !cancelledStaleNodeIds.has(staleSubagentNode.id)) { - cancelledStaleNodeIds.add(staleSubagentNode.id); - events.push({ - id: yield* allocateEventId(), - type: "node.updated", - threadId: projection.thread.id, - ...(item.runId === null ? {} : { runId: item.runId }), - nodeId: staleSubagentNode.id, - providerInstanceId, - occurredAt: now, - payload: { ...staleSubagentNode, status: "cancelled", completedAt: now }, - }); - } } - // A provider-native subagent thread has no runs: its work is a runless - // root turn, plus items under it (Claude's live progress item), that - // only the dead provider process could settle. Left running, the child + // A provider-native subagent thread has no runs; left running, the child // would show as working forever. - const cancelledStaleItemIds = new Set( - events.flatMap((event) => (event.type === "turn-item.updated" ? [event.payload.id] : [])), - ); - for (const node of projection.nodes) { - if ( - node.kind !== "root_turn" || - node.runId !== null || - !isNonterminalNodeStatus(node.status) || - cancelledStaleNodeIds.has(node.id) - ) { - continue; - } - cancelledStaleNodeIds.add(node.id); - events.push({ - id: yield* allocateEventId(), - type: "node.updated", + events.push( + ...(yield* endRunlessRootTurns({ threadId: projection.thread.id, - nodeId: node.id, providerInstanceId: projection.thread.providerInstanceId, - occurredAt: now, - payload: { ...node, status: "cancelled", completedAt: now }, - }); - for (const item of projection.turnItems) { - if ( - item.nodeId !== node.id || - item.runId !== null || - !isNonterminalTurnItemStatus(item.status) || - cancelledStaleItemIds.has(item.id) - ) { - continue; - } - cancelledStaleItemIds.add(item.id); - events.push({ - id: yield* allocateEventId(), - type: "turn-item.updated", - threadId: projection.thread.id, - nodeId: node.id, - providerInstanceId: projection.thread.providerInstanceId, - occurredAt: now, - payload: { - ...item, - status: "cancelled", - completedAt: now, - updatedAt: now, - ...(item.type === "reasoning" || item.type === "assistant_message" - ? { streaming: false } - : {}), - }, - }); - } - } + projection, + status: "cancelled", + now, + emitted: events, + allocateEventId, + })), + ); // All provider processes are gone on startup/shutdown: clear any // persisted Waiting roster (including idle threads from settled roots) // and idle active threads without resurrecting active status. diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index 34b3d8352c68cdc71c4bbec75e7aa9c58fdc5d96..34b608f961e5fa607be0319d5cf4cf046245444e 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -481,6 +481,7 @@ it.effect.each([ { id: "subagent:background", runId, + origin: "provider_native", driver, providerInstanceId: instanceId, status: "running", diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 70820566378718117c0d981d51495ea848ca9f91..c42c402003d40d176564a20fda3994bbd415ea0f 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1432,6 +1432,137 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", }), ); +it.effect("ends a nested native subagent and its thread when the root run is interrupted", () => + Effect.gen(function* () { + const key = "run-execution-nested-native"; + const threadId = ThreadId.make(`thread:${key}`); + const childThreadId = ThreadId.make(`thread:${key}:child`); + const grandchildThreadId = ThreadId.make(`thread:${key}:grandchild`); + const runId = RunId.make(`run:${key}`); + const attemptId = RunAttemptId.make(`attempt:${key}`); + const providerInstanceId = ProviderInstanceId.make("codex"); + const providerThreadId = ProviderThreadId.make(`provider-thread:${key}`); + const rootProviderTurnId = ProviderTurnId.make(`provider-turn:${key}`); + const nestedId = NodeId.make(`node:${key}:nested`); + const grandchildRootId = NodeId.make(`node:${key}:grandchild-root`); + const finalized = yield* Deferred.make>(); + const layerTest = RunExecutionService.layer.pipe( + Layer.provide( + Layer.mergeAll( + McpAppModelContext.layerEmpty, + Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + Layer.mock(EventSink.EventSinkV2)({ + write: () => Effect.succeed([]), + writeWithEffects: (input) => + Effect.as( + input.events.some((event) => event.type === "run.updated") + ? Deferred.succeed(finalized, input.events) + : Effect.void, + [], + ), + }), + IdAllocator.layer, + Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + ingestNormalized: () => Effect.succeed([]), + }), + ServerSettings.layerTest(), + ), + ), + ); + const subagentThread = (id: ThreadId, parentThreadId: ThreadId) => + ({ + type: "app_thread.created", + driver, + appThread: { + id, + lineage: { parentThreadId, relationshipToParent: "subagent", rootThreadId: threadId }, + }, + }) as ProviderAdapter.ProviderAdapterV2Event; + const events: ReadonlyArray = [ + subagentThread(childThreadId, threadId), + subagentThread(grandchildThreadId, childThreadId), + { + type: "subagent.updated", + driver, + subagent: { + id: nestedId, + threadId: childThreadId, + runId: null, + origin: "provider_native", + childThreadId: grandchildThreadId, + status: "running", + }, + } as ProviderAdapter.ProviderAdapterV2Event, + { + type: "node.updated", + driver, + node: { + id: grandchildRootId, + threadId: grandchildThreadId, + runId: null, + kind: "root_turn", + status: "running", + }, + } as ProviderAdapter.ProviderAdapterV2Event, + { + type: "turn.terminal", + driver, + providerThreadId, + providerTurnId: rootProviderTurnId, + runOrdinal: 1, + status: "interrupted", + failure: null, + threadDisposition: "reusable", + }, + ]; + + yield* Effect.gen(function* () { + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; + yield* runExecution.startRootRun({ + commandId: CommandId.make(`command:${key}`), + appThread: { id: threadId } as OrchestrationV2AppThread, + providerSessionId: ProviderSessionId.make(`session:${key}`), + session: { + events: Stream.fromIterable(events), + startTurn: () => Effect.void, + } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, + run: { id: runId, threadId, ordinal: 1, providerInstanceId } as OrchestrationV2Run, + rootNode: { id: NodeId.make(`node:${key}`) } as OrchestrationV2ExecutionNode, + checkpointScope: { + id: CheckpointScopeId.make(`checkpoint-scope:${key}`), + } as OrchestrationV2CheckpointScope, + providerThread: { id: providerThreadId, driver } as OrchestrationV2ProviderThread, + attempt: { id: attemptId, providerTurnId: rootProviderTurnId } as OrchestrationV2RunAttempt, + attemptId, + providerTurnOrdinal: 1, + message: { + messageId: MessageId.make(`message:${key}:user`), + text: "Spawn a nested subagent.", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" }, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false }, + }, + }); + }).pipe(Effect.provide(layerTest)); + + const ended = (yield* Deferred.await(finalized)).flatMap((event) => + event.type === "subagent.updated" || event.type === "node.updated" + ? [[event.type, event.payload.id, event.payload.status]] + : [], + ); + assert.deepInclude(ended, ["subagent.updated", nestedId, "interrupted"]); + assert.deepInclude(ended, ["node.updated", grandchildRootId, "interrupted"]); + }), +); + it.effect("keeps ingesting a late background command item completion after root terminal", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-command", (ids) => [ diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index a316ce0ccd6212fa90f2bca7df1274ab8cee5f42..fa6d0fbd7f125d9ed7cb5c137894dab043c63327 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1044,7 +1044,14 @@ export const layer: Layer.Layer< // Preserve childThreadId linkage for the root-run lifetime even // after the subagent row terminalizes, so open child-thread // nodes can still be proven linked on a later root interrupt. - if (belongsToRootRun) { + // A provider records nested native subagents on the child + // thread, runless; they die with this run too. + if ( + belongsToRootRun || + (belongsToOwnedChildThread && + event.subagent.runId === null && + event.subagent.origin === "provider_native") + ) { yield* Ref.update(openRunOwnedSubagents, (current) => { const withLink = withLinkedChildThreadId(current, event.subagent.childThreadId); const subagents = new Map(withLink.subagents); diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index 59eca69cf35a17dcb1686e1b72a35f2d4ba2e6ea..65b64000b4709cdfda37e39da7c1534763ed87a0 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -15,12 +15,15 @@ import { ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; import { makeSubagentChildThread, delegatedTaskProgress, subagentResultForRun, makeSubagentConversationArtifacts, + endOrphanedNativeSubagent, + endRunlessRootTurns, } from "@t3tools/provider-core/server/subagentProjection"; import { emptyProjection } from "./ProjectionStore.ts"; @@ -314,3 +317,64 @@ it("exposes the provider failure rather than a progress message from the failed assert.equal(result.turnItemId, artifacts.turnItem.id); assert.isNull(result.messageId); }); + +it.effect("ends an orphaned native subagent's node when only its record already ended", () => + Effect.gen(function* () { + const id = NodeId.make("node:orphaned-native-subagent"); + const subagent = { id, origin: "provider_native", status: "running", runId: null }; + const events = yield* endOrphanedNativeSubagent({ + projection: { + subagents: [subagent], + nodes: [{ id, status: "running", runId: null }], + turnItems: [], + } as never, + subagentId: id, + status: "cancelled", + now: yield* DateTime.now, + emitted: [ + { type: "subagent.updated", payload: { ...subagent, status: "cancelled" } }, + ] as never, + allocateEventId: () => Effect.succeed(EventId.make("event:orphaned-native-subagent")), + }); + assert.deepEqual( + events.map((event) => [event.type, event.type === "node.updated" && event.payload.status]), + [["node.updated", "cancelled"]], + ); + }), +); + +it.effect("ends a runless root turn's items when its node already ended", () => + Effect.gen(function* () { + const root = { + id: NodeId.make("node:runless-root"), + kind: "root_turn", + status: "running", + runId: null, + }; + const item = (id: string) => ({ id, nodeId: root.id, runId: null, status: "running" }); + const events = yield* endRunlessRootTurns({ + threadId: childThreadId, + providerInstanceId: parentProviderInstanceId, + projection: { + nodes: [root], + turnItems: [item("turn-item:command"), item("turn-item:reasoning")], + } as never, + status: "cancelled", + now: yield* DateTime.now, + emitted: [ + { + type: "turn-item.updated", + payload: { ...item("turn-item:command"), status: "cancelled" }, + }, + { type: "node.updated", payload: { ...root, status: "cancelled" } }, + ] as never, + allocateEventId: () => Effect.succeed(EventId.make("event:runless-root")), + }); + assert.deepEqual( + events.map( + (event) => `${event.type} ${event.type === "turn-item.updated" && event.payload.id}`, + ), + ["turn-item.updated turn-item:reasoning"], + ); + }), +); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 7ee8d7c1aa8d9d0554697c6e7b5138991c51e9d5..1a6b31a886e52cb40d6e2a4a5ea5b8acf2692165 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -1,4 +1,5 @@ import type { + EventId, MessageId, ModelSelection, NodeId, @@ -6,6 +7,7 @@ import type { OrchestrationV2Actor, OrchestrationV2ConversationMessage, OrchestrationV2CreationSource, + OrchestrationV2DomainEvent, OrchestrationV2ProviderRef, OrchestrationV2Run, OrchestrationV2ThreadProjection, @@ -18,6 +20,7 @@ import type { } from "@t3tools/contracts"; import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; import { isOrchestrationV2WorkActive } from "@t3tools/contracts"; function trimmed(value: string | null | undefined): string | undefined { @@ -257,3 +260,151 @@ export function delegatedTaskProgress(projection: { resultRun, }; } + +/** Ids that `emitted` already wrote with an event of `type`. */ +function writtenIds( + emitted: ReadonlyArray, + type: "subagent.updated" | "node.updated" | "turn-item.updated", +): ReadonlySet { + return new Set(emitted.flatMap((event) => (event.type === type ? [event.payload.id] : []))); +} + +/** + * Ends a provider-native subagent, its node and its turn item once the provider + * process that ran it is gone, since no terminal event will come from it. Each + * record is skipped if already settled or already ended by `emitted`. + */ +export const endOrphanedNativeSubagent = (input: { + readonly projection: Pick; + readonly subagentId: NodeId; + readonly status: "interrupted" | "cancelled"; + readonly now: DateTime.Utc; + readonly emitted: ReadonlyArray; + readonly allocateEventId: () => Effect.Effect; +}) => + Effect.gen(function* () { + const { now, status, subagentId } = input; + const subagent = input.projection.subagents.find((entry) => entry.id === subagentId); + if (subagent?.origin !== "provider_native") return []; + const node = input.projection.nodes.find((entry) => entry.id === subagentId); + const item = input.projection.turnItems.find( + (entry) => entry.type === "subagent" && entry.subagentId === subagentId, + ); + const scope = { + threadId: subagent.threadId, + ...(subagent.runId === null ? {} : { runId: subagent.runId }), + nodeId: subagentId, + providerInstanceId: subagent.providerInstanceId, + occurredAt: now, + }; + const events: Array = []; + if ( + isOrchestrationV2WorkActive(subagent.status) && + !writtenIds(input.emitted, "subagent.updated").has(subagentId) + ) { + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "subagent.updated", + driver: subagent.driver, + payload: { ...subagent, status, completedAt: now, updatedAt: now }, + }); + } + if ( + node !== undefined && + isOrchestrationV2WorkActive(node.status) && + !writtenIds(input.emitted, "node.updated").has(subagentId) + ) { + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "node.updated", + payload: { ...node, status, completedAt: now }, + }); + } + if ( + item !== undefined && + isOrchestrationV2WorkActive(item.status) && + !writtenIds(input.emitted, "turn-item.updated").has(item.id) + ) { + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "turn-item.updated", + payload: { ...item, status, completedAt: now, updatedAt: now }, + }); + } + return events; + }); + +/** + * Ends a provider-native subagent thread's work, a runless root turn plus the + * items under it (Claude's live progress), which only the provider process that + * ran it could settle. Skips records `emitted` already ended. + */ +export const endRunlessRootTurns = (input: { + readonly threadId: ThreadId; + readonly providerInstanceId: ProviderInstanceId; + readonly projection: Pick; + readonly status: "interrupted" | "cancelled"; + readonly now: DateTime.Utc; + readonly emitted: ReadonlyArray; + readonly allocateEventId: () => Effect.Effect; +}) => + Effect.gen(function* () { + const { now, status } = input; + const endedNodeIds = writtenIds(input.emitted, "node.updated"); + const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); + const scope = { + threadId: input.threadId, + providerInstanceId: input.providerInstanceId, + occurredAt: now, + }; + const events: Array = []; + for (const node of input.projection.nodes) { + if ( + node.kind !== "root_turn" || + node.runId !== null || + !isOrchestrationV2WorkActive(node.status) + ) { + continue; + } + // Recovery may have ended this node through one of its items; its other + // items still need ending. + if (!endedNodeIds.has(node.id)) { + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "node.updated", + nodeId: node.id, + payload: { ...node, status, completedAt: now }, + }); + } + for (const item of input.projection.turnItems) { + if ( + item.nodeId !== node.id || + item.runId !== null || + !isOrchestrationV2WorkActive(item.status) || + endedItemIds.has(item.id) + ) { + continue; + } + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "turn-item.updated", + nodeId: node.id, + payload: { + ...item, + status, + completedAt: now, + updatedAt: now, + ...(item.type === "reasoning" || item.type === "assistant_message" + ? { streaming: false } + : {}), + }, + }); + } + } + return events; + });