diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 820b5ff78125..c1a72d05f685 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -31,6 +31,7 @@ import type { ProviderAdapterV2TurnInput, } from "./ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; @@ -45,10 +46,17 @@ const stopEarlierBackgroundWork = ({ failedStart = false, stopWithQueue, olderStart = false, + stalledRun, }: { readonly failedStart?: boolean; readonly stopWithQueue?: "thread.stop" | "run.interrupt"; readonly olderStart?: boolean; + readonly stalledRun?: + | "missing-session" + | "missing-session-terminal" + | "returned-interrupt" + | "returned-interrupt-terminal" + | "superseded-attempt"; }) => Effect.scoped( Effect.gen(function* () { @@ -210,6 +218,181 @@ const stopEarlierBackgroundWork = ({ }, ], }); + if (stalledRun !== undefined) { + const before = yield* orchestrator.getThreadProjection(threadId); + const run = before.runs[0]!; + const node = before.nodes.find((candidate) => candidate.id === run.rootNodeId)!; + const attempt = before.attempts[0]!; + if (stalledRun === "missing-session" || stalledRun === "missing-session-terminal") { + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + const failed = yield* watch( + (event) => event.type === "run.updated" && event.payload.status === "failed", + ); + yield* sessions.release({ + providerSessionId: first.providerThread.providerSessionId!, + reason: "runtime_error", + }); + yield* Fiber.join(failed); + // Disk exhaustion can lose these terminal writes. Restore that stale state. + yield* sink.write({ + events: [ + { + id: EventId.make("stale-run"), + type: "run.updated", + threadId, + occurredAt: now, + payload: run, + }, + { + id: EventId.make("stale-node"), + type: "node.updated", + threadId, + occurredAt: now, + payload: node, + }, + { + id: EventId.make("stale-attempt"), + type: "run-attempt.updated", + threadId, + occurredAt: now, + payload: attempt, + }, + ], + }); + } + const messageId = MessageId.make("partial-output"); + const item = before.turnItems.find((candidate) => candidate.id === devServerId)!; + assert.ok(item.type === "command_execution"); + const terminalProviderTurn = stalledRun.endsWith("-terminal"); + if (terminalProviderTurn) { + yield* sink.write({ + events: [ + { + id: EventId.make("terminal-provider-turn"), + type: "provider-turn.updated", + threadId, + occurredAt: now, + payload: { ...codexTurn, status: "completed", completedAt: now }, + }, + ...(stalledRun === "missing-session-terminal" + ? [ + { + id: EventId.make("completed-dev-server"), + type: "turn-item.updated" as const, + threadId, + occurredAt: now, + payload: { ...item, status: "completed" as const, completedAt: now }, + }, + ] + : []), + ], + }); + } + yield* sink.write({ + events: [ + { + id: EventId.make("partial-message"), + type: "message.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId: run.id, + nodeId: node.id, + role: "assistant", + text: "Partial output", + attachments: [], + streaming: true, + createdBy: "agent", + creationSource: "provider", + createdAt: now, + updatedAt: now, + }, + }, + { + id: EventId.make("partial-item"), + type: "turn-item.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + ...item, + providerThreadId: codexTurn.providerThreadId, + providerTurnId: codexTurn.id, + id: TurnItemId.make("partial-output"), + type: "assistant_message", + messageId, + text: "Partial output", + streaming: true, + }, + }, + ], + }); + if (stalledRun === "superseded-attempt") { + yield* sink.write({ + events: [ + { + id: EventId.make("new-attempt"), + type: "run.updated", + threadId, + occurredAt: now, + payload: { ...run, activeAttemptId: RunAttemptId.make("new-attempt") }, + }, + ], + }); + } + yield* orchestrator.dispatch( + stalledRun === "superseded-attempt" + ? { + type: "thread.background-work.settle", + commandId: CommandId.make("late-settle"), + threadId, + providerThreadId: codexTurn.providerThreadId, + providerTurnId: codexTurn.id, + } + : { + type: "thread.stop", + commandId: CommandId.make("stop-stalled-run"), + threadId, + }, + ); + yield* worker.drain(); + const after = yield* orchestrator.getThreadProjection(threadId); + const interrupted = stalledRun !== "superseded-attempt"; + assert.equal(after.runs[0]?.status, interrupted ? "interrupted" : "running"); + assert.equal(after.attempts[0]?.status, interrupted ? "interrupted" : "running"); + assert.equal( + after.providerTurns[0]?.status, + terminalProviderTurn ? "completed" : interrupted ? "interrupted" : "running", + ); + assert.equal( + after.nodes.find((candidate) => candidate.id === node.id)?.status, + interrupted ? "interrupted" : "running", + ); + const output = after.messages.find((message) => message.id === messageId)!; + assert.equal(output.streaming, !interrupted); + assert.equal(output.text, "Partial output"); + assert.equal( + after.turnItems.find((candidate) => candidate.id === devServerId)?.status, + stalledRun === "missing-session-terminal" + ? "completed" + : interrupted + ? "interrupted" + : "running", + ); + assert.equal( + after.turnItems.find((candidate) => candidate.type === "assistant_message")?.status, + interrupted ? "interrupted" : "running", + ); + assert.equal( + after.turnItems.filter((candidate) => candidate.type === "run_interrupt_result").length, + interrupted ? 1 : 0, + ); + assert.isEmpty(after.runs.filter((candidate) => candidate.status === "waiting")); + return; + } const settled = yield* watch( (event) => event.type === "run.updated" && @@ -599,3 +782,13 @@ it.effect.each(["thread.stop", "run.interrupt"] as const)( "%s reaches later background work when an older run is starting", (stopType) => stopEarlierBackgroundWork({ stopWithQueue: stopType, olderStart: true }), ); + +it.effect.each([ + "missing-session", + "missing-session-terminal", + "returned-interrupt", + "returned-interrupt-terminal", + "superseded-attempt", +] as const)("Stop recovers a stalled run after %s without changing a newer attempt", (stalledRun) => + stopEarlierBackgroundWork({ stalledRun }), +); diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index c2d21ed00aa0..d9bd8689f092 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -81,13 +81,14 @@ function layerExecutorFor(input: { readonly failFirstStart?: Ref.Ref; readonly threads?: Partial; readonly continueAfterRestart?: boolean; + readonly interrupt?: ProviderTurnControlService.ProviderTurnControlServiceV2Shape["interrupt"]; }) { const record = (event: string) => Ref.update(input.events, (events) => [...events, event]); const layerDependencies = Layer.mergeAll( Layer.succeed( ProviderTurnControlService.ProviderTurnControlServiceV2, ProviderTurnControlService.ProviderTurnControlServiceV2.of({ - interrupt: () => Effect.void, + interrupt: input.interrupt ?? (() => Effect.void), steer: () => Effect.void, interruptAndAwaitTerminal: (request) => record( @@ -193,6 +194,42 @@ it("does not retry pure interrupt races where the turn is already gone", () => { ); }); +it.effect("settles a stopped run when its adapter has already lost the native turn", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const events = yield* Ref.make>([]); + const layer = layerExecutorFor({ + events, + interrupt: () => + new ProviderTurnControlService.ProviderTurnControlError({ + threadId, + operation: "interrupt", + providerTurnId, + cause: "Provider turn is not active.", + }), + threads: { + dispatch: (command) => + Ref.update(events, (current) => [...current, command.type]).pipe( + Effect.as({ sequence: 1, storedEvents: [] }), + ), + }, + }); + yield* Effect.gen(function* () { + const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; + yield* executor.execute({ + ...restartEffect(now, { type: "detach" }), + request: { + type: "provider-turn.interrupt", + providerSessionId: oldSessionId, + providerThreadId, + providerTurnId, + }, + }); + }).pipe(Effect.provide(layer)); + assert.deepEqual(yield* Ref.get(events), ["thread.background-work.settle"]); + }), +); + it.effect("requeues a claim when a pre-execution worker check fails", () => Effect.gen(function* () { const now = DateTime.formatIso(yield* DateTime.now); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 79dff0a3eb6c..b3ee780334d3 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -171,6 +171,14 @@ export const layerExecutor: Layer.Layer< providerTurnId: effect.request.providerTurnId, }) .pipe( + Effect.catch((cause) => + isNonRetryableProviderTurnControlFailure( + effect.request.type, + Cause.pretty(Cause.fail(cause)), + ) + ? Effect.void + : Effect.fail(cause), + ), // The provider has stopped what it still ran and reported it. // Whatever the thread still shows on that provider thread is // work no process will report on, so the Stop ends it too. diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index d544763d2c77..039d1872da6f 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -90,6 +90,7 @@ export interface EventSinkV2Shape { readonly activeAttemptId: RunAttemptId; readonly expectedStatus: OrchestrationV2Run["status"]; readonly events: ReadonlyArray; + readonly effects?: ReadonlyArray; }) => Effect.Effect< { readonly committed: boolean; @@ -437,9 +438,17 @@ const layerBase: Layer.Layer< events: normalized, }); yield* applyStoredEvents(storedEvents); + yield* effectOutbox.enqueue(input.effects ?? []); return { committed: true as const, storedEvents }; }), - (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void), + (result) => + Effect.gen(function* () { + if (!result.committed) return; + if (input.effects !== undefined && input.effects.length > 0) { + yield* effectOutbox.notifyAvailable(input.effects.length); + } + yield* publishStoredEvents(result.storedEvents); + }), ); }, ); diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 894df7c8d367..5b70ae8b7fd5 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1525,6 +1525,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; const now = yield* DateTime.now; const threadId = ThreadId.make("thread:foundation-stale-provider-start"); const runId = RunId.make("run:foundation-stale-provider-start"); @@ -1567,6 +1568,30 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { ], }); + const captureEffect = { + id: "effect:foundation-current-capture", + commandId: CommandId.make("command:foundation-current-capture"), + threadId, + request: { + type: "checkpoint.capture" as const, + runId, + scopeId: CheckpointScopeId.make("scope:foundation-current-capture"), + }, + }; + assert.isTrue( + (yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: attemptId, + expectedStatus: "starting", + events: [], + effects: [captureEffect], + })).committed, + ); + assert.isTrue(Option.isSome(yield* outbox.get(captureEffect.id))); + yield* outbox.awaitAvailable; + const staleCaptureEffect = { ...captureEffect, id: "effect:foundation-stale-capture" }; + const reachedPrecommitGap = yield* Deferred.make(); const releaseStaleStart = yield* Deferred.make(); const providerStartCount = yield* Ref.make(0); @@ -1578,6 +1603,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { runId, activeAttemptId: attemptId, expectedStatus: "starting", + effects: [staleCaptureEffect], events: [ { id: EventId.make("event:foundation-stale-provider-start:running"), @@ -1619,11 +1645,12 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { const staleResult = yield* Fiber.join(staleStartFiber); assert.isFalse(staleResult.committed); + assert.isTrue(Option.isNone(yield* outbox.get(staleCaptureEffect.id))); assert.deepEqual(staleResult.storedEvents, []); assert.equal(yield* Ref.get(providerStartCount), 0); const projection = yield* projectionStore.getThreadProjection(threadId); assert.equal(projection.runs[0]?.status, "cancelled"); - }), + }).pipe(Effect.provide(Layer.fresh(layerTest))), ); it.effect("guards post-terminal provider-thread writes by attempt and run ordinal", () => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 9a8bac907d0c..de100d291b8f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -102,6 +102,7 @@ import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; import { ProviderContinuationRequests } from "./ProviderContinuationRequests.ts"; import { makeProviderFailure } from "./ProviderFailure.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts"; import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; @@ -8116,9 +8117,204 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio driver: providerThread.driver, providerInstanceId: providerThread.providerInstanceId, occurredAt: input.now, - payload: { ...providerThread, pendingBackgroundTasks: [], updatedAt: input.now }, + payload: { + ...providerThread, + status: + providerThread.status === "active" && + providerThread.lastRunOrdinal !== null && + providerThread.lastRunOrdinal <= input.throughRunOrdinal + ? "idle" + : providerThread.status, + pendingBackgroundTasks: [], + updatedAt: input.now, + }, + }); + } + }); + + // A successful Stop may have no terminal event to ingest: the session or + // native turn already died, or its final write failed when storage filled. + const settleInterruptedRun = (input: { + readonly command: Extract< + OrchestrationV2ServerCommand, + { readonly type: "run.interrupt" | "thread.background-work.settle" } + >; + readonly projection: Pick< + OrchestrationV2ThreadProjection, + | "runs" + | "attempts" + | "nodes" + | "subagents" + | "runtimeRequests" + | "messages" + | "turnItems" + | "providerThreads" + >; + readonly providerTurn: OrchestrationV2ProviderTurn; + readonly events: Ref.Ref>; + readonly effects: Ref.Ref>; + readonly now: DateTime.Utc; + }) => + Effect.gen(function* () { + const attempt = input.projection.attempts.find( + (candidate) => candidate.id === input.providerTurn.runAttemptId, + ); + const run = input.projection.runs.find((candidate) => candidate.id === attempt?.runId); + const rootNode = input.projection.nodes.find((candidate) => candidate.id === run?.rootNodeId); + if ( + attempt === undefined || + run === undefined || + rootNode === undefined || + run.activeAttemptId !== attempt.id || + run.status !== "running" || + attempt.status !== "running" + ) + return; + const emitEvent = emit(input.events, input.command); + const base = { + threadId: input.command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: input.now, + }; + if (input.providerTurn.status === "running") { + yield* emitEvent({ + ...base, + type: "provider-turn.updated", + payload: { + ...input.providerTurn, + status: "interrupted", + completedAt: input.now, + }, + }); + } + yield* emitEvent({ + ...base, + type: "run-attempt.updated", + payload: { + ...attempt, + status: "interrupted", + completedAt: input.now, + }, + }); + const providerThread = input.projection.providerThreads.find( + (candidate) => candidate.id === input.providerTurn.providerThreadId, + ); + if (providerThread?.status === "active" && providerThread.lastRunOrdinal === run.ordinal) { + yield* emitEvent({ + ...base, + type: "provider-thread.updated", + payload: { ...providerThread, status: "idle", updatedAt: input.now }, + }); + } + for (const message of input.projection.messages) { + if (message.runId !== run.id || !message.streaming) continue; + yield* emitEvent({ + ...base, + ...(message.nodeId === null ? {} : { nodeId: message.nodeId }), + type: "message.updated", + payload: { ...message, streaming: false, updatedAt: input.now }, + }); + } + const delegatedNodeIds = new Set( + input.projection.subagents + .filter((subagent) => subagent.origin === "app_owned") + .map((subagent) => subagent.id), + ); + const cascaded = yield* RunExecutionService.cascadeTerminalizeRunOwnedSubagents({ + run, + status: "interrupted", + completedAt: input.now, + allocateEventId: () => + idAllocator.allocate.event({ + threadId: input.command.threadId, + commandId: input.command.commandId, + }), + open: { + subagents: new Map( + input.projection.subagents + .filter((task) => task.runId === run.id && task.origin !== "app_owned") + .map((task) => [task.id, task]), + ), + nodes: new Map( + input.projection.nodes + .filter((node) => node.runId === run.id && !delegatedNodeIds.has(node.id)) + .map((node) => [node.id, node]), + ), + turnItems: new Map(), + childTurnItems: new Map( + input.projection.turnItems + .filter( + (item) => + item.runId === run.id && + (item.type === "assistant_message" || item.type === "reasoning"), + ) + .map((item) => [item.id, item]), + ), + linkedChildThreadIds: new Set([run.threadId]), + }, + }).pipe(mapDispatchError(input.command)); + yield* Ref.update(input.events, (events) => [...events, ...cascaded]); + for (const request of input.projection.runtimeRequests) { + if ( + request.status !== "pending" || + !input.projection.nodes.some( + (node) => node.id === request.nodeId && node.runId === run.id, + ) + ) + continue; + yield* emitEvent({ + ...base, + nodeId: request.nodeId, + type: "runtime-request.updated", + payload: { + ...request, + status: "cancelled", + resolvedAt: input.now, + responseCapability: { type: "not_resumable", reason: "The run was interrupted." }, + }, + }); + } + if (providerThread !== undefined) { + yield* emitEvent({ + ...base, + type: "turn-item.updated", + payload: RunExecutionService.makeInterruptResultTurnItem({ + idAllocator, + run, + rootNode, + providerThread, + completedAt: input.now, + }), }); } + const { delegatedCompletion: _delegatedCompletion, ...runWithoutDelegatedCompletion } = run; + yield* emitEvent({ + ...base, + type: "run.updated", + payload: { + ...runWithoutDelegatedCompletion, + status: "interrupted", + completedAt: input.now, + }, + }); + const scopeId = rootNode.checkpointScopeId; + if (scopeId !== null) { + yield* Ref.update(input.effects, (effects) => [ + ...effects, + { + id: `effect:checkpoint.capture:${run.id}`, + commandId: CommandId.make(`command:effect:checkpoint.capture:${run.id}`), + threadId: input.command.threadId, + request: { + type: "checkpoint.capture", + runId: run.id, + scopeId, + }, + } satisfies PendingOrchestrationEffectV2, + ]); + } }); /** @@ -8280,14 +8476,29 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio { readonly type: "thread.background-work.settle" } >, events: Ref.Ref>, + effects: Ref.Ref>, ) => Effect.gen(function* () { const [projection, stopped] = yield* Effect.all([ projectionStore.getThreadRecords( command.threadId, - ["runs", "attempts", "turnItems", "providerThreads"], + [ + "runs", + "attempts", + "nodes", + "subagents", + "runtimeRequests", + "turnItems", + "providerThreads", + ], { - turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], + turnItemTypes: [ + "command_execution", + "dynamic_tool", + "subagent", + "assistant_message", + "reasoning", + ], turnItemStatuses: ["pending", "running", "waiting"], }, ), @@ -8304,12 +8515,35 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (attempt) => attempt.id === stopped.providerTurn?.runAttemptId, )?.runId; const stoppedRun = projection.runs.find((run) => run.id === stoppedRunId); - // A new turn may have started since Stop; its work is not this Stop's. if ( stoppedRun === undefined || + (["preparing", "starting", "running"].includes(stoppedRun.status) && + stoppedRun.activeAttemptId !== stopped.providerTurn?.runAttemptId) + ) + return; + const now = yield* DateTime.now; + if (stopped.providerTurn !== undefined && stoppedRun.status === "running") { + const output = yield* projectionStore + .getThreadRecords(command.threadId, ["messages"], { + messageRunIds: [stoppedRun.id], + messageRoles: ["assistant"], + }) + .pipe(mapDispatchError(command)); + yield* settleInterruptedRun({ + command, + projection: { ...projection, messages: output.messages }, + providerTurn: stopped.providerTurn, + events, + effects, + now, + }); + } + // A new turn may have started since Stop; its work is not this Stop's. + if ( projection.runs.some( (run) => - run.status === "preparing" || run.status === "starting" || run.status === "running", + run.id !== stoppedRun.id && + (run.status === "preparing" || run.status === "starting" || run.status === "running"), ) ) { return; @@ -8320,7 +8554,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection, stoppedProviderThreadId: command.providerThreadId, throughRunOrdinal: stoppedRun.ordinal, - now: yield* DateTime.now, + now, }); }); @@ -8341,11 +8575,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "attempts", "turnItems", "subagents", + "runtimeRequests", ], // Thread-wide, like the Waiting strip: a settled thread's background // work can belong to an earlier run than the one Stop targets. { - turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], + turnItemTypes: [ + "command_execution", + "dynamic_tool", + "subagent", + "assistant_message", + "reasoning", + ], turnItemStatuses: ["pending", "running", "waiting"], }, ); @@ -8373,7 +8614,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection.providerTurns.findLast( (candidate) => candidate.runAttemptId === run?.activeAttemptId && - (candidate.status === "running" || hasBackgroundWork), + (run?.status === "running" || candidate.status === "running" || hasBackgroundWork), ) ?? (hasBackgroundWork ? projection.providerTurns.findLast( @@ -8623,18 +8864,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Run ${command.runId} is not interruptible.`, }); } - // Stop on a settled thread's background work. Its process may be gone - // (released, restarted) and only the projection still shows the work; - // the settle follow-up ends whatever no provider reports ending. - const settleOnly = - providerTurn.status !== "running" && - (providerThread.providerSessionId === null || - Option.isNone( - yield* providerSessions - .get(providerThread.providerSessionId) - .pipe(Effect.orElseSucceed(() => Option.none())), - )); - if (settleOnly) { + // A dead session cannot report a terminal event for either a running + // turn or a settled turn's background work. Stop ends both locally. + const sessionOption = + providerThread.providerSessionId === null + ? Option.none() + : yield* providerSessions.get(providerThread.providerSessionId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProviderAdapterError({ + commandId: command.commandId, + providerInstanceId: run.providerInstanceId, + cause, + }), + ), + ); + if (Option.isNone(sessionOption)) { yield* emitEvent({ type: "turn-item.updated", threadId: command.threadId, @@ -8645,6 +8890,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: interruptRequestItem, }); yield* stopRemainingWork(); + yield* settleInterruptedRun({ command, projection, providerTurn, events, effects, now }); yield* settleBackgroundWork({ command, events, @@ -8664,23 +8910,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } const providerSessionId = providerThread.providerSessionId; - const sessionOption = yield* providerSessions.get(providerSessionId).pipe( - Effect.mapError( - (cause) => - new OrchestratorProviderAdapterError({ - commandId: command.commandId, - providerInstanceId: run.providerInstanceId, - cause, - }), - ), - ); - if (Option.isNone(sessionOption)) { - return yield* new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: `Provider session ${providerThread.providerSessionId} is not active.`, - }); - } yield* enforceCommandPolicy(command)( commandPolicy.ensureInterrupt({ commandId: command.commandId, @@ -10079,7 +10308,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* dispatchCheckpointRollbackFail(command, events); break; case "thread.background-work.settle": - yield* dispatchBackgroundWorkSettle(command, events); + yield* dispatchBackgroundWorkSettle(command, events, effects); break; case "thread.fork": yield* dispatchThreadFork(command, events); diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index 47f7aa80bdc4..659922b3a876 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -12,6 +12,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as NodeTimersPromises from "node:timers/promises"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; @@ -185,6 +186,19 @@ export const layer: Layer.Layer< providerTurnId: loaded.providerTurn.id, requestRuntimeRestart: true, }); + // Give native terminal ingestion time to finish before the Stop + // follow-up repairs a run whose provider no longer reports on it. + const deadline = performance.now() + 2_000; + while (loaded.providerTurn.status === "running" && performance.now() < deadline) { + const current = yield* projections.getProviderControlContext(input.threadId, input); + if ( + current.providerTurn?.status !== "running" && + current.attempt?.status !== "running" + ) { + break; + } + yield* Effect.promise(() => NodeTimersPromises.setTimeout(10)); + } }).pipe( Effect.mapError((cause) => isProviderTurnControlError(cause) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 0df053898c90..9ce1e0536415 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3243,9 +3243,27 @@ it.effect("omits run_interrupt_result when superseded attempt request is already }), ); +it.effect("does not overwrite Stop when ownership changes after the finalization read", () => + Effect.gen(function* () { + const { written, observed, submittedEffects, committedEffects } = + yield* captureRootRunTermination({ + key: "stop-wins-finalization-gap", + shouldFinalizeRun: () => Effect.succeed(true), + rejectTerminalWrite: true, + }); + assert.deepEqual(written, []); + assert.deepEqual(observed, []); + assert.deepEqual( + submittedEffects.map((effect) => effect.request.type), + ["checkpoint.capture"], + ); + assert.deepEqual(committedEffects, []); + }), +); + it.effect("emits run_interrupt_result when hard-stop finalizes the active attempt", () => Effect.gen(function* () { - const { written, observed } = yield* captureRootRunTermination({ + const { written, observed, committedEffects } = yield* captureRootRunTermination({ key: "hard-stop", shouldFinalizeRun: () => Effect.succeed(true), }); @@ -3254,6 +3272,10 @@ it.effect("emits run_interrupt_result when hard-stop finalizes the active attemp ["run_interrupt_result"], ); assert.deepEqual(observed, ["run:interrupted", "pull-requests-refreshed"]); + assert.deepEqual( + committedEffects.map((effect) => effect.request.type), + ["checkpoint.capture"], + ); }), ); @@ -3382,6 +3404,7 @@ it.effect("keeps completed runs completed when pull request refresh fails", () = function captureRootRunTermination(input: { readonly key: string; readonly shouldFinalizeRun: () => Effect.Effect; + readonly rejectTerminalWrite?: boolean; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly seedOpenSubagent?: boolean; readonly events?: ( @@ -3402,6 +3425,8 @@ function captureRootRunTermination(input: { }); const writtenItems = yield* Ref.make>([]); const observed = yield* Ref.make>([]); + const submittedEffects = yield* Ref.make>([]); + const committedEffects = yield* Ref.make>([]); const ingestionDone = yield* Deferred.make(); const captureTurnItem = (payload: OrchestrationV2TurnItem) => Ref.update(writtenItems, (current) => [...current, payload]); @@ -3430,11 +3455,24 @@ function captureRootRunTermination(input: { } return []; }), - writeWithEffects: (payload) => captureFinalEvents(payload.events).pipe(Effect.as([])), + writeWithEffects: (payload) => + Effect.gen(function* () { + yield* Ref.update(submittedEffects, (current) => [...current, ...payload.effects]); + yield* Ref.update(committedEffects, (current) => [...current, ...payload.effects]); + yield* captureFinalEvents(payload.events); + return []; + }), writeIfRunCurrent: (payload) => - captureFinalEvents(payload.events).pipe( - Effect.as({ committed: true, storedEvents: [] }), - ), + Effect.gen(function* () { + const effects = payload.effects ?? []; + yield* Ref.update(submittedEffects, (current) => [...current, ...effects]); + if (input.rejectTerminalWrite === true) { + return { committed: false, storedEvents: [] }; + } + yield* Ref.update(committedEffects, (current) => [...current, ...effects]); + yield* captureFinalEvents(payload.events); + return { committed: true, storedEvents: [] }; + }), }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ @@ -3543,7 +3581,12 @@ function captureRootRunTermination(input: { }).pipe(Effect.provide(layerTest)); yield* Deferred.await(ingestionDone); - return { written: yield* Ref.get(writtenItems), observed: yield* Ref.get(observed) }; + return { + written: yield* Ref.get(writtenItems), + observed: yield* Ref.get(observed), + submittedEffects: yield* Ref.get(submittedEffects), + committedEffects: yield* Ref.get(committedEffects), + }; }); } diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 7ab2c87cdef4..254b810c6daf 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -783,6 +783,7 @@ export const layer: Layer.Layer< activeAttemptId: input.writeIfRunCurrent.activeAttemptId, expectedStatus: input.writeIfRunCurrent.expectedStatus, events: finalization.events, + effects: finalization.effects, }); if (!result.committed) { return; @@ -971,7 +972,15 @@ export const layer: Layer.Layer< attempt: input.attempt, ...(input.shouldFinalizeRun === undefined ? {} - : { shouldFinalizeRun: input.shouldFinalizeRun }), + : { + shouldFinalizeRun: input.shouldFinalizeRun, + // Stop may commit between the ownership read and this + // terminal write. Gate the events and checkpoint together. + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + }), ...(input.hasUnpairedRunInterruptRequest === undefined ? {} : { @@ -1437,7 +1446,7 @@ export const layer: Layer.Layer< }), ); -function makeInterruptResultTurnItem(input: { +export function makeInterruptResultTurnItem(input: { readonly idAllocator: IdAllocator.IdAllocatorV2Shape; readonly run: OrchestrationV2Run; readonly rootNode: OrchestrationV2ExecutionNode; diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index ba6dc2d284ea..687e881c4656 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -261,7 +261,10 @@ export function layerWithRegistry( readonly continueThreadsAfterServerUpdate?: boolean; } = {}, ): Layer.Layer< - Orchestrator.OrchestratorV2 | EffectWorker.OrchestrationEffectWorkerV2 | EventSink.EventSinkV2, + | Orchestrator.OrchestratorV2 + | EffectWorker.OrchestrationEffectWorkerV2 + | EventSink.EventSinkV2 + | ProviderSessionManager.ProviderSessionManagerV2, Error | MigrationError | PlatformError.PlatformError | SqlError > { const layerServerConfig = Layer.effect( @@ -474,6 +477,7 @@ export function layerWithRegistry( ); const layerReplayRuntime = Layer.mergeAll( layerOrchestratorProvided, + layerProviderSessionManagerProvided, layerEffectWorkerProvided, layerEventSinkProvided, layerContinuationWorkerProvided,