diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index aae5d1b68db83..3a9cb4facbd38 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9112,7 +9112,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } childThreadIds.add(childThreadId); const child = yield* projectionStore - .getThreadRecords(childThreadId, ["nodes", "subagents", "turnItems"], { + .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], { // Runless items are the subagent's; any run there is the user's own. turnItemRunIds: [null], turnItemStatuses: ["pending", "running", "waiting"], @@ -9671,7 +9671,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* Ref.update(events, (current) => [...current, ...ended]); if (subagent.childThreadId !== null) { const child = yield* projectionStore - .getThreadRecords(subagent.childThreadId, ["nodes", "turnItems"], { + .getThreadRecords(subagent.childThreadId, ["nodes", "providerTurns", "turnItems"], { turnItemRunIds: [null], turnItemStatuses: ["pending", "running", "waiting"], }) diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce388..6ed24558f01dc 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -12,6 +12,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderThreadId, + ProviderTurnId, RunId, RuntimeRequestId, ThreadId, @@ -966,3 +967,102 @@ it.effect("marks fork descendants unreadable when their source is missing or cor ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("startup ends the provider turn of a native child root it cancels", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const parentThreadId = yield* createThread("native-child-root-turn-parent"); + const threadId = yield* createThread("native-child-root-turn"); + const rootNodeId = NodeId.make(`node:${threadId}:thread-root`); + const providerThreadId = ProviderThreadId.make(`provider-thread:${threadId}`); + const providerTurnId = ProviderTurnId.make(`provider-turn:${threadId}`); + yield* projections.apply({ + id: EventId.make(`event:${threadId}:root`), + type: "node.updated", + threadId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: rootNodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId, + kind: "root_turn", + status: "running", + countsForRun: false, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }); + // A native subagent's turn belongs to no run attempt. + yield* projections.apply({ + id: EventId.make(`event:${threadId}:provider-turn`), + type: "provider-turn.updated", + threadId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId, + nodeId: rootNodeId, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + + const subagentId = NodeId.make(`node:${parentThreadId}:subagent`); + yield* projections.apply({ + id: EventId.make(`event:${parentThreadId}:subagent`), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { + id: subagentId, + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make(`node:${parentThreadId}:root`), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId, + childThreadId: threadId, + nativeTaskRef: null, + title: null, + model: null, + prompt: "sleep 300", + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }); + + yield* recovery.reconcile("startup"); + + const projection = yield* projections.getThreadProjection(threadId); + assert.deepEqual( + { + root: projection.nodes.find((node) => node.id === rootNodeId)?.status, + providerTurn: projection.providerTurns.find((turn) => turn.id === providerTurnId)?.status, + }, + { root: "cancelled", providerTurn: "cancelled" }, + "a root startup recovery cancelled left its provider turn running", + ); + }).pipe(Effect.provide(layerRecoveryTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 700395facd79d..3c26ae3747179 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4605,14 +4605,26 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_provider_turns AS provider_turn WHERE provider_turn.thread_id = ${threadId} AND provider_turn.status IN ('pending', 'starting', 'running', 'waiting') - AND provider_turn.run_attempt_id IN ( - SELECT attempt_id FROM orchestration_v2_projection_run_attempts - WHERE thread_id = ${threadId} - AND run_id IN ( - SELECT run_id FROM orchestration_v2_projection_runs + AND ( + provider_turn.run_attempt_id IN ( + SELECT attempt_id FROM orchestration_v2_projection_run_attempts + WHERE thread_id = ${threadId} + AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + ) + ) + OR ( + provider_turn.run_attempt_id IS NULL + AND provider_turn.node_id IN ( + SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId} - AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + AND run_id IS NULL + AND kind = 'root_turn' + AND status IN ('pending', 'running', 'waiting') ) + ) ) ORDER BY provider_turn.provider_thread_id ASC, provider_turn.ordinal ASC `, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 03bbbf0506b50..f25d03bc7f527 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -11,6 +11,7 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, + type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2Subagent, @@ -2624,6 +2625,106 @@ it.effect.each(["failure", "defect", "interrupt-failure", "ownership-lost"] as c }), ); +it.effect.each([ + { status: "completed", failure: "later-stream", interrupt: "succeeds" }, + { status: "failed", failure: "later-stream", interrupt: "succeeds" }, + { status: "completed", failure: "terminal-ingest", interrupt: "succeeds" }, + { status: "completed", failure: "terminal-ingest", interrupt: "fails" }, +] as const)( + "stores a $status root provider turn snapshot after a $failure failure only if the interrupt succeeds (it $interrupt)", + ({ status, failure, interrupt }) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const terminalAt = DateTime.subtract(now, { seconds: 1 }); + const key = `terminal-snapshot:${status}:${failure}:${interrupt}`; + const durableTurns = yield* Ref.make< + ReadonlyMap + >(new Map()); + const persist = (turn: OrchestrationV2ProviderTurn) => + Ref.update(durableTurns, (current) => new Map(current).set(turn.id, turn)); + const interrupts = yield* Ref.make< + ReadonlyArray + >([]); + const ids = backgroundScenarioIds(key); + const rootTurn = ( + turnStatus: OrchestrationV2ProviderTurn["status"], + completedAt: DateTime.Utc | null, + ) => ({ + type: "provider_turn.updated" as const, + driver, + providerTurn: { + id: ids.rootProviderTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: turnStatus, + startedAt: terminalAt, + completedAt, + }, + }); + const { observed, finalEvents } = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + events: () => + Stream.make(rootTurn("running", null), rootTurn(status, terminalAt)).pipe( + Stream.concat( + failure === "later-stream" + ? Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make(`session:${key}`), + cause: "provider process exited", + }), + ) + : Stream.empty, + ), + ), + ingestNormalized: ({ event }) => + event.type !== "provider_turn.updated" + ? Effect.succeed([]) + : failure === "terminal-ingest" && event.providerTurn.status !== "running" + ? Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: ProviderSessionId.make(`session:${key}`), + eventCount: 1, + cause: new Error("SQLITE_FULL: database or disk is full"), + }), + ) + : persist(event.providerTurn).pipe(Effect.as([])), + interruptTurn: (input) => + Ref.update(interrupts, (current) => [...current, input]).pipe( + Effect.andThen( + interrupt === "fails" + ? Effect.fail( + new ProviderAdapter.ProviderAdapterInterruptError({ + driver, + providerThreadId: input.providerThread.id, + providerTurnId: input.providerTurnId, + cause: "provider already exited", + }), + ) + : Effect.void, + ), + ), + beforeTerminalWrite: (request) => + Effect.forEach(request.events, (event) => + event.type === "provider-turn.updated" ? persist(event.payload) : Effect.void, + ).pipe(Effect.as(true)), + }); + const calls = yield* Ref.get(interrupts); + assert.equal(calls.length, 1); + assert.isTrue(calls[0]?.requestRuntimeRestart); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + assert.isTrue(finalEvents.some((event) => event.type === "run.updated")); + const durable = (yield* Ref.get(durableTurns)).get(ids.rootProviderTurnId); + // A failed interrupt keeps the stored turn running for the next start to retry. + assert.equal(durable?.status, interrupt === "fails" ? "running" : status); + assert.deepEqual(durable?.completedAt, interrupt === "fails" ? null : terminalAt); + }), +); + it.effect("does not interrupt an ended provider when its terminal write fails", () => Effect.gen(function* () { const interrupts = yield* Ref.make(0); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 9ec61cc85b994..d7789df380b68 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -620,7 +620,8 @@ export const layer: Layer.Layer< readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly loadRunInterruptReason?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; - readonly interruptedProviderTurn?: OrchestrationV2ProviderTurn; + // Root turn snapshot written with the run when T3 ends the run itself. + readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; readonly completedAt?: DateTime.Utc; readonly failureItemPersisted: boolean; @@ -747,7 +748,7 @@ export const layer: Layer.Layer< : [], events: [ ...cascadedSubagentEvents, - ...(input.interruptedProviderTurn === undefined + ...(input.abandonedProviderTurn === undefined ? [] : [ { @@ -758,11 +759,7 @@ export const layer: Layer.Layer< nodeId: input.rootNode.id, providerInstanceId: input.run.providerInstanceId, occurredAt: completedAt, - payload: { - ...input.interruptedProviderTurn, - status: "interrupted" as const, - completedAt, - }, + payload: input.abandonedProviderTurn, }, ]), ...(finalizedAttempt === null @@ -1482,8 +1479,9 @@ export const layer: Layer.Layer< if (yield* Ref.get(rootRunFinalized)) return; const providerThread = yield* Ref.get(latestProviderThread); const routing = yield* Ref.get(eventRouting); + const rootProviderTurn = yield* Ref.get(latestRootProviderTurn); // Stop background work too; the failed consumer cannot persist terminal events. - const interruptedProviderTurn = + const abandonedProviderTurn = routing.rootProviderTurnId !== null && !routing.rootTurnEnded ? yield* input.session .interruptTurn({ @@ -1492,14 +1490,28 @@ export const layer: Layer.Layer< requestRuntimeRestart: true, }) .pipe( - Effect.as(yield* Ref.get(latestRootProviderTurn)), + Effect.as(true), Effect.catchCauseIf( (interruptCause) => !Cause.hasInterruptsOnly(interruptCause), (interruptCause) => Effect.logWarning( "orchestration V2 failed to interrupt turn after ingestion failure", { runId: input.run.id, cause: interruptCause }, - ).pipe(Effect.as(null)), + ).pipe(Effect.as(false)), + ), + // A failed interrupt leaves the stored turn running so the next start + // retries it. The snapshot is cached before it is published, so it may + // not be stored: replay a terminal one as-is. + Effect.map((interrupted) => + !interrupted || rootProviderTurn === null + ? null + : isTerminalProviderTurnStatus(rootProviderTurn.status) + ? rootProviderTurn + : { + ...rootProviderTurn, + status: "interrupted" as const, + completedAt, + }, ), ) : null; @@ -1517,7 +1529,7 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, - ...(interruptedProviderTurn === null ? {} : { interruptedProviderTurn }), + ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), completedAt, terminal: makeFailedTerminalEvent( makeProviderFailure({ cause: Cause.squash(cause), class: "unknown" }), diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index e87fd7d9c6eea..d3506649f6a62 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -375,6 +375,7 @@ it.effect("ends a runless root turn's items when its node already ended", () => providerInstanceId: parentProviderInstanceId, projection: { nodes: [root], + providerTurns: [], turnItems: [item("turn-item:command"), item("turn-item:reasoning")], } as never, status: "cancelled", diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 0c38faa5929f1..23a52f39d33d3 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1335,6 +1335,25 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) occurredAt: now, payload: childRoot, }); + const childTurnId = ProviderTurnId.make("turn:native-stop-child"); + yield* projections.apply({ + id: EventId.make("event:native-child-stop:child-turn"), + type: "provider-turn.updated", + threadId: childRoot.threadId, + nodeId: childRoot.id, + occurredAt: now, + payload: { + id: childTurnId, + providerThreadId: ProviderThreadId.make("provider-thread:native-stop-child"), + nodeId: childRoot.id, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); const providerThread = before.providerThreads.find( (thread) => thread.id === run.providerThreadId, )!; @@ -1499,6 +1518,10 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) ); const child = yield* orchestrator.getThreadProjection(childRoot.threadId); assert.equal(child.nodes.find((node) => node.id === childRoot.id)?.status, confirmedStatus); + assert.equal( + child.providerTurns.find((turn) => turn.id === childTurnId)?.status, + confirmedStatus, + ); assert.isNotNull(child.nodes.find((node) => node.id === childRoot.id)?.completedAt); assert.equal(settled.subagents.find((task) => task.id === sibling.id)?.status, "running"); assert.deepEqual(yield* threadState(threadId), stateBefore); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 9393d3a551878..f83a0b376233b 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -271,7 +271,7 @@ export function delegatedTaskProgress(projection: { function writtenIds( emitted: ReadonlyArray, - type: "subagent.updated" | "node.updated" | "turn-item.updated", + type: "subagent.updated" | "node.updated" | "turn-item.updated" | "provider-turn.updated", ): ReadonlySet { return new Set(emitted.flatMap((event) => (event.type === type ? [event.payload.id] : []))); } @@ -359,7 +359,10 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT >(input: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; - readonly projection: Pick; + readonly projection: Pick< + OrchestrationV2ThreadProjection, + "nodes" | "providerTurns" | "turnItems" + >; readonly status: "completed" | "failed" | "interrupted" | "cancelled"; readonly completedAt?: DateTime.Utc; readonly now: DateTime.Utc; @@ -369,6 +372,7 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT const { now, status, completedAt = now } = input; const endedNodeIds = writtenIds(input.emitted, "node.updated"); const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); + const endedProviderTurnIds = writtenIds(input.emitted, "provider-turn.updated"); const scope = { threadId: input.threadId, providerInstanceId: input.providerInstanceId, @@ -421,6 +425,23 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT }, }); } + for (const providerTurn of input.projection.providerTurns) { + if ( + providerTurn.nodeId !== node.id || + providerTurn.runAttemptId !== null || + !isOrchestrationV2WorkActive(providerTurn.status) || + endedProviderTurnIds.has(providerTurn.id) + ) { + continue; + } + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "provider-turn.updated", + nodeId: node.id, + payload: { ...providerTurn, status, completedAt }, + }); + } } return events; });