From f8ae16f3d52b1fae0419b8a4f2d895b9bba296fb Mon Sep 17 00:00:00 2001 From: Dara Adedeji <76637177+SunkenInTime@users.noreply.github.com> Date: Fri, 2 Oct 2026 14:13:43 -0400 Subject: [PATCH 1/6] fix(server): V2 runs no longer look busy forever after their events stop saving When a run's event consumer hit a write error (for example SQLite "database is locked"), it tried once to record the run as failed. If that write also failed, the error was dropped in a detached fiber: the run stayed "running", the agent kept working unseen, and only a server restart cleared it. Now the consumer stops the provider turn and retries the failed terminal with capped backoff until it lands, and the failure explains what happened. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/ProviderFailure.ts | 5 + .../RunExecutionService.test.ts | 130 +++++++++++--- .../orchestration-v2/RunExecutionService.ts | 160 +++++++++++------- 3 files changed, 217 insertions(+), 78 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderFailure.ts b/apps/server/src/orchestration-v2/ProviderFailure.ts index b0c7e5fdd6270..0b82f605c87e9 100644 --- a/apps/server/src/orchestration-v2/ProviderFailure.ts +++ b/apps/server/src/orchestration-v2/ProviderFailure.ts @@ -46,6 +46,11 @@ function causeMessage(cause: unknown): string | undefined { message = "The provider event stream closed unexpectedly. Retry the turn; if it keeps failing, check the provider and server logs."; break; + // Run execution stops the provider turn when its events cannot be saved. + case "ProviderEventPublishError": + message = + "T3 Code could not save this turn's progress, so it stopped the agent. Send a message to continue."; + break; case "ProviderAdapterOpenSessionError": message = "The provider session could not be opened. Check that the provider is installed and signed in, then retry the turn."; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index ec292a02b86ff..1306c4dbcd96d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -32,10 +32,12 @@ import * as Exit from "effect/Exit"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -3200,7 +3202,10 @@ it.effect("refreshes pull requests after a provider stream exits with an error", }), ), }); - assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + assert.deepEqual( + observed.filter((entry) => !entry.startsWith("turn-interrupted:")), + ["run:failed", "pull-requests-refreshed"], + ); assert.deepEqual( written.map((item) => item.type), ["error"], @@ -3210,6 +3215,65 @@ it.effect("refreshes pull requests after a provider stream exits with an error", }), ); +it.effect("stops the agent and records the run failed when its events cannot be saved", () => + Effect.gen(function* () { + const finalizationWriteFailed = yield* Deferred.make(); + const ids = backgroundScenarioIds("lost-ingestion"); + const capture = yield* captureRootRunTermination({ + key: "lost-ingestion", + shouldFinalizeRun: () => Effect.succeed(true), + events: () => Stream.make(rootTerminalEvent(ids, "completed")), + ingestNormalized: (event) => + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: event.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + // The lock that broke ingestion is still held for the first failed write. + failedFinalizationWrites: 1, + onFinalizationWriteFailed: Deferred.succeed(finalizationWriteFailed, undefined), + }).pipe(Effect.forkChild); + + yield* Deferred.await(finalizationWriteFailed); + for (let step = 0; step < 12; step++) { + yield* Effect.yieldNow; + yield* TestClock.adjust("5 seconds"); + } + const { observed, written, turnInterrupted } = yield* Fiber.join(capture); + + assert.equal(yield* Deferred.await(turnInterrupted), ids.rootProviderTurnId); + assert.deepEqual( + observed.filter((entry) => !entry.startsWith("turn-interrupted:")), + ["run:failed", "pull-requests-refreshed"], + ); + const error = written.find((item) => item.type === "error"); + assert.include(error?.failure.message ?? "", "could not save this turn's progress"); + }), +); + +it.effect("does not record a lost run as failed after the run moved on", () => + Effect.gen(function* () { + const { observed, written } = yield* captureRootRunTermination({ + key: "lost-ingestion-superseded", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")), + ingestNormalized: (event) => + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: event.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + runIsCurrent: false, + }); + assert.notInclude(observed, "run:failed"); + assert.deepEqual(written, []); + }), +); + it.effect("refreshes pull requests only once when startup failure closes its event stream", () => Effect.gen(function* () { const ingestionStarted = yield* Deferred.make(); @@ -3262,6 +3326,12 @@ function captureRootRunTermination(input: { ) => Stream.Stream; readonly startTurn?: ProviderAdapterV2SessionRuntime["startTurn"]; readonly refreshAfterTurn?: Effect.Effect; + readonly ingestNormalized?: ProviderEventIngestor.ProviderEventIngestorV2Shape["ingestNormalized"]; + /** Fails this many run-finalization writes before letting one through. */ + readonly failedFinalizationWrites?: number; + readonly onFinalizationWriteFailed?: Effect.Effect; + /** False when another attempt or a Stop already moved the run on. */ + readonly runIsCurrent?: boolean; }) { return Effect.gen(function* () { const ids = backgroundScenarioIds(input.key); @@ -3276,8 +3346,30 @@ function captureRootRunTermination(input: { const writtenItems = yield* Ref.make>([]); const observed = yield* Ref.make>([]); const ingestionDone = yield* Deferred.make(); + const failedWrites = yield* Ref.make(input.failedFinalizationWrites ?? 0); + const turnInterrupted = yield* Deferred.make(); const captureTurnItem = (payload: OrchestrationV2TurnItem) => Ref.update(writtenItems, (current) => [...current, payload]); + const recordFinalization = ( + events: Parameters[0]["events"], + ) => + Effect.gen(function* () { + if ((yield* Ref.getAndUpdate(failedWrites, (count) => count - 1)) > 0) { + yield* input.onFinalizationWriteFailed ?? Effect.void; + return yield* new EventSink.EventSinkWriteError({ + eventCount: events.length, + cause: "database is locked", + }); + } + for (const event of events) { + if (event.type === "turn-item.updated") { + yield* captureTurnItem(event.payload); + } + if (event.type === "run.updated") { + yield* Ref.update(observed, (current) => [...current, `run:${event.payload.status}`]); + } + } + }); const testLayer = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -3292,26 +3384,17 @@ function captureRootRunTermination(input: { } return []; }), - writeWithEffects: (payload) => - Effect.gen(function* () { - for (const event of payload.events) { - if (event.type === "turn-item.updated") { - yield* captureTurnItem(event.payload); - } - if (event.type === "run.updated") { - yield* Ref.update(observed, (current) => [ - ...current, - `run:${event.payload.status}`, - ]); - } - } - return []; - }), - writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), + writeWithEffects: (payload) => recordFinalization(payload.events).pipe(Effect.as([])), + writeIfRunCurrent: (payload) => + input.runIsCurrent === false + ? Effect.succeed({ committed: false, storedEvents: [] }) + : recordFinalization(payload.events).pipe( + Effect.as({ committed: true, storedEvents: [] }), + ), }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ - ingestNormalized: () => Effect.succeed([]), + ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), }), ServerSettings.layerTest(), Layer.succeed(RunFinalizationService.RunFinalizationObserver, { @@ -3363,6 +3446,11 @@ function captureRootRunTermination(input: { close: Deferred.succeed(ingestionDone, undefined), }), startTurn: input.startTurn ?? (() => Effect.void), + interruptTurn: ({ providerTurnId }: { readonly providerTurnId: ProviderTurnId }) => + Ref.update(observed, (current) => [ + ...current, + `turn-interrupted:${providerTurnId}`, + ]).pipe(Effect.andThen(Deferred.succeed(turnInterrupted, providerTurnId))), } as unknown as ProviderAdapterV2SessionRuntime, run: { id: ids.runId, @@ -3416,7 +3504,11 @@ function captureRootRunTermination(input: { }).pipe(Effect.provide(testLayer)); 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), + turnInterrupted, + }; }); } diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 71211c1265459..085cb839d904a 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -27,10 +27,12 @@ import { import * as Context from "effect/Context"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -1166,6 +1168,102 @@ export const layer: Layer.Layer< } return true; }); + // This consumer is the only thing that settles the run. If it dies + // first, the agent keeps working while the thread shows it running + // forever. Stop the provider turn so the work really ends, then keep + // retrying the failed terminal until it lands. + const finalizeAfterLostIngestion = (cause: Cause.Cause) => + Effect.gen(function* () { + if (yield* Ref.get(rootRunFinalized)) { + yield* Effect.logWarning("orchestration V2 provider event ingestion failed", { + runId: input.run.id, + cause, + }); + return; + } + yield* Effect.logError( + "orchestration V2 lost provider event ingestion; stopping the provider turn", + { runId: input.run.id, cause }, + ); + const providerThread = yield* Ref.get(latestProviderThread); + const { rootProviderTurnId } = yield* Ref.get(eventRouting); + if (rootProviderTurnId !== null) { + // Stop the way the user's Stop does, background work included. + // Detached so the adapter finishes its own cancel cleanup on its + // own clock while the failure is recorded. + yield* input.session + .interruptTurn({ + providerThread, + providerTurnId: rootProviderTurnId, + requestRuntimeRestart: true, + }) + .pipe( + Effect.catchCause((stopCause) => + Effect.logWarning( + "orchestration V2 could not stop a provider turn after losing its events", + { runId: input.run.id, cause: stopCause }, + ), + ), + Effect.forkDetach, + ); + } + const latestItemOrdinal = yield* Ref.get(latestTurnItemOrdinal); + const openSubagents = yield* Ref.get(openRunOwnedSubagents); + yield* writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread, + attempt: input.attempt, + ...(input.shouldFinalizeRun === undefined + ? {} + : { shouldFinalizeRun: input.shouldFinalizeRun }), + ...(input.hasUnpairedRunInterruptRequest === undefined + ? {} + : { hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest }), + openRunOwnedSubagents: openSubagents, + terminal: makeFailedTerminalEvent( + makeProviderFailure({ + cause: Cause.squash(cause), + class: "unknown", + }), + latestItemOrdinal + 1, + ), + failureItemPersisted: false, + refreshAfterTurn, + // Retries can land after a steer or Stop moved the run on; only + // write while this attempt still owns a running run. + writeIfRunCurrent: { activeAttemptId: input.attempt.id, expectedStatus: "running" }, + }).pipe( + Effect.tapCause((writeCause) => + Effect.logError( + "orchestration V2 could not record a lost run as failed; retrying", + { + runId: input.run.id, + cause: writeCause, + }, + ), + ), + // Recording the failure is the last chance to show the truth, + // so it retries until it lands, backing off to every 30s. + Effect.retry({ + schedule: Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + Schedule.jittered, + ), + }), + ); + }).pipe( + Effect.mapError( + (writeCause) => + new RunExecutionIngestError({ + runId: input.run.id, + cause: { ingest: cause, write: writeCause }, + }), + ), + ); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); const providerEventFiber = yield* eventSubscription.events.pipe( Stream.filterEffect((event) => @@ -1264,65 +1362,9 @@ export const layer: Layer.Layer< yield* finalizeRootRun(terminal); }), ), - Effect.catchCause((cause) => - Ref.get(rootRunFinalized).pipe( - Effect.flatMap((finalized) => - Effect.logWarning("orchestration V2 provider event ingestion failed", { - runId: input.run.id, - cause, - }).pipe( - Effect.andThen( - finalized - ? Effect.void - : Ref.get(latestProviderThread).pipe( - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - ...(input.shouldFinalizeRun === undefined - ? {} - : { shouldFinalizeRun: input.shouldFinalizeRun }), - ...(input.hasUnpairedRunInterruptRequest === undefined - ? {} - : { - hasUnpairedRunInterruptRequest: - input.hasUnpairedRunInterruptRequest, - }), - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: "unknown", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, - }), - ), - ), - ), - ), - ), - ), - ), - Effect.mapError( - (writeCause) => - new RunExecutionIngestError({ - runId: input.run.id, - cause: { ingest: cause, write: writeCause }, - }), - ), - ), - ), - ), + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + finalizeAfterLostIngestion, ), Effect.ensuring(eventSubscription.close), Effect.forkDetach, From d155032cf63a3d4cfc8d91b355756f89d0d238c2 Mon Sep 17 00:00:00 2001 From: Dara Adedeji <76637177+SunkenInTime@users.noreply.github.com> Date: Fri, 2 Oct 2026 14:52:27 -0400 Subject: [PATCH 2/6] fix(server): stop the root turn even when saving fails before it is reported If the first failed write came before the provider reported its root turn, recovery had no turn id and could not stop the agent. The consumer now records the failure, stops writing, and keeps routing until the root turn is known, then recovers as before. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../RunExecutionService.test.ts | 63 ++++++++++++- .../orchestration-v2/RunExecutionService.ts | 88 ++++++++++++------- 2 files changed, 118 insertions(+), 33 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 1306c4dbcd96d..c0324985f1fba 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3253,6 +3253,62 @@ it.effect("stops the agent and records the run failed when its events cannot be }), ); +it.effect("stops the root turn even when saving fails before the turn is reported", () => + Effect.gen(function* () { + const ids = backgroundScenarioIds("lost-ingestion-early"); + const now = yield* DateTime.now; + const lateTurnId = ProviderTurnId.make("provider-turn:lost-ingestion-early:native"); + const firstWriteFailed = yield* Deferred.make(); + const providerThreadEvent = { + type: "provider_thread.updated", + driver, + providerThread: { id: ids.providerThreadId } as OrchestrationV2ProviderThread, + } as ProviderAdapterV2Event; + const rootTurnEvent: ProviderAdapterV2Event = { + type: "provider_turn.updated", + driver, + threadId: ids.threadId, + providerTurn: { + id: lateTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }; + const { observed, turnInterrupted } = yield* captureRootRunTermination({ + key: "lost-ingestion-early", + shouldFinalizeRun: () => Effect.succeed(true), + attemptProviderTurnId: null, + // The provider reports its turn only after the first write has failed. + events: () => + Stream.make(providerThreadEvent).pipe( + Stream.concat( + Stream.fromEffect(Deferred.await(firstWriteFailed).pipe(Effect.as(rootTurnEvent))), + ), + ), + ingestNormalized: (event) => + Deferred.succeed(firstWriteFailed, undefined).pipe( + Effect.andThen( + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: event.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + ), + ), + }); + assert.equal(yield* Deferred.await(turnInterrupted), lateTurnId); + assert.include(observed, "run:failed"); + }), +); + it.effect("does not record a lost run as failed after the run moved on", () => Effect.gen(function* () { const { observed, written } = yield* captureRootRunTermination({ @@ -3332,6 +3388,8 @@ function captureRootRunTermination(input: { readonly onFinalizationWriteFailed?: Effect.Effect; /** False when another attempt or a Stop already moved the run on. */ readonly runIsCurrent?: boolean; + /** Null when the provider has not reported the root turn yet. */ + readonly attemptProviderTurnId?: ProviderTurnId | null; }) { return Effect.gen(function* () { const ids = backgroundScenarioIds(input.key); @@ -3471,7 +3529,10 @@ function captureRootRunTermination(input: { } as OrchestrationV2ProviderThread, attempt: { id: ids.attemptId, - providerTurnId: ids.rootProviderTurnId, + providerTurnId: + input.attemptProviderTurnId === undefined + ? ids.rootProviderTurnId + : input.attemptProviderTurnId, } as OrchestrationV2RunAttempt, attemptId: ids.attemptId, providerTurnOrdinal: 1, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 085cb839d904a..b6894d0362815 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -948,6 +948,10 @@ export const layer: Layer.Layer< ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); + // Set when an event write fails. The stream then stops writing but + // keeps routing until it has seen the root provider turn, so + // recovery always knows which turn to stop. + const lostIngestion = yield* Ref.make | null>(null); const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Set()); @@ -1271,6 +1275,7 @@ export const layer: Layer.Layer< ), Stream.tap((event) => Effect.gen(function* () { + if ((yield* Ref.get(lostIngestion)) !== null) return; let storedEventCount = 0; const deliveredEvent = filterAssistantEvent( event, @@ -1286,37 +1291,45 @@ export const layer: Layer.Layer< const isRootProviderThreadUpdate = event.type === "provider_thread.updated" && event.providerThread.id === input.providerThread.id; - const storedEvents = yield* providerEventIngestor.ingestNormalized({ - analyticsContext: { - modelSelection: input.modelSelection, - runtimeMode: input.runtimePolicy.runtimeMode, - interactionMode: input.runtimePolicy.interactionMode, - }, - providerSessionId: input.providerSessionId, - providerInstanceId: input.run.providerInstanceId, - threadId: input.run.threadId, - runId: input.run.id, - nodeId: input.rootNode.id, - event: deliveredEvent, - ...(isRootProviderThreadUpdate - ? rootTerminalAlreadySeen - ? { - writeIfProviderThreadOwner: { - providerThreadId: input.providerThread.id, - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedLastRunOrdinal: input.run.ordinal, - }, - } - : { - writeIfRunCurrent: { - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - } - : {}), - }); + const storedEvents = yield* providerEventIngestor + .ingestNormalized({ + analyticsContext: { + modelSelection: input.modelSelection, + runtimeMode: input.runtimePolicy.runtimeMode, + interactionMode: input.runtimePolicy.interactionMode, + }, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + threadId: input.run.threadId, + runId: input.run.id, + nodeId: input.rootNode.id, + event: deliveredEvent, + ...(isRootProviderThreadUpdate + ? rootTerminalAlreadySeen + ? { + writeIfProviderThreadOwner: { + providerThreadId: input.providerThread.id, + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedLastRunOrdinal: input.run.ordinal, + }, + } + : { + writeIfRunCurrent: { + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + } + : {}), + }) + .pipe( + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + (cause) => Ref.set(lostIngestion, cause).pipe(Effect.as(null)), + ), + ); + if (storedEvents === null) return; storedEventCount = storedEvents.length; if ( isRootProviderThreadUpdate && @@ -1350,11 +1363,22 @@ export const layer: Layer.Layer< yield* trackChildLifecycle(event, deliveredEvent !== null); }), ), - Stream.takeUntilEffect(() => shouldStopProviderEventIngestion), + Stream.takeUntilEffect(() => + Effect.gen(function* () { + if ((yield* Ref.get(lostIngestion)) !== null) { + return (yield* Ref.get(eventRouting)).rootProviderTurnId !== null; + } + return yield* shouldStopProviderEventIngestion; + }), + ), Stream.runDrain, Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), Effect.flatMap(() => Effect.gen(function* () { + const lost = yield* Ref.get(lostIngestion); + if (lost !== null) { + return yield* new RunExecutionIngestError({ runId: input.run.id, cause: lost }); + } const terminal = yield* Ref.get(terminalEvent); if (terminal === null) { return; From 2d45a9b631bdc7afa86e64e4749710fffc0f16ff Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 18:51:10 +0000 Subject: [PATCH 3/6] test(server): cover recovery after provider event publication fails --- .../RunExecutionService.test.ts | 317 +++++++++++++++++- 1 file changed, 311 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 8cf9a4f28a439..ae60059641ff6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -32,10 +32,13 @@ import * as Exit from "effect/Exit"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Clock from "effect/Clock"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -3849,10 +3852,301 @@ it.effect("keeps completed runs completed when pull request refresh fails", () = }), ); +it.effect.each(["known", "late", "absent", "stop-error"] as const)( + "recovers publication failure (%s) and reports the actual stop outcome", + (scenario) => + Effect.gen(function* () { + const key = `publication-failure:${scenario}`; + const ids = backgroundScenarioIds(key); + const started = yield* Deferred.make(); + const publicationFailed = yield* Deferred.make(); + const stopStarted = yield* Deferred.make(); + const releaseStop = yield* Deferred.make(); + const finished = yield* Deferred.make(); + const interrupts = yield* Ref.make([]); + const stopped = yield* Ref.make(false); + const notices = yield* Ref.make>([]); + const now = yield* DateTime.now; + const capture = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + attemptProviderTurnId: + scenario === "late" || scenario === "absent" ? null : ids.rootProviderTurnId, + startTurn: () => Deferred.succeed(started, undefined).pipe(Effect.asVoid), + events: () => + Stream.unwrap( + Deferred.await(started).pipe( + Effect.as( + Stream.make({ + type: "provider_thread.updated", + driver, + providerThread: { id: ids.providerThreadId } as OrchestrationV2ProviderThread, + } as ProviderAdapter.ProviderAdapterV2Event).pipe( + Stream.concat( + scenario === "late" + ? Stream.fromEffect( + Deferred.await(publicationFailed).pipe( + Effect.as({ + type: "provider_turn.updated", + driver, + threadId: ids.threadId, + providerTurn: { + id: ids.rootProviderTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + } satisfies ProviderAdapter.ProviderAdapterV2Event), + ), + ) + : Stream.empty, + ), + Stream.concat(Stream.never), + ), + ), + ), + ), + ingestNormalized: (input) => + Deferred.succeed(publicationFailed, undefined).pipe( + Effect.andThen( + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: input.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + ), + ), + interruptTurn: (input) => + Effect.gen(function* () { + yield* Ref.update(interrupts, (calls) => [...calls, input]); + yield* Deferred.succeed(stopStarted, undefined); + yield* Deferred.await(releaseStop); + if (scenario === "stop-error") { + return yield* new ProviderAdapter.ProviderAdapterInterruptError({ + driver, + providerThreadId: ids.providerThreadId, + providerTurnId: input.providerTurnId, + cause: "provider refused cancellation", + }); + } + yield* Ref.set(stopped, true); + }), + beforeTerminalWrite: (events) => + Effect.gen(function* () { + for (const event of events) { + if (event.type === "turn-item.updated" && event.payload.type === "error") { + const confirmed = yield* Ref.get(stopped); + const message = event.payload.failure.message; + yield* Ref.update(notices, (current) => [ + ...current, + { + message, + stopped: confirmed, + }, + ]); + } + } + }), + }).pipe( + Effect.tap(() => Deferred.succeed(finished, undefined)), + Effect.forkChild, + ); + + yield* Deferred.await(publicationFailed); + if (scenario !== "absent") { + yield* Effect.raceFirst( + Deferred.await(stopStarted), + Deferred.await(finished).pipe( + Effect.andThen(Effect.die("consumer closed without stopping the known root turn")), + ), + ); + } + if (scenario === "stop-error") { + yield* Deferred.succeed(releaseStop, undefined); + } + // Exercise discovery/stop deadlines while the provider is silent or cancellation is pending. + yield* TestClock.adjust("1 minute"); + const pendingNotices = yield* Ref.get(notices); + for (const notice of pendingNotices) { + assert.notMatch( + notice.message, + /(?:stopped|cancelled) the agent|agent (?:was|has been|is) (?:stopped|cancelled)/i, + ); + } + yield* Deferred.succeed(releaseStop, undefined); + if (scenario === "absent") { + assert.isTrue( + yield* Deferred.isDone(finished), + "missing turn discovery must have a bounded fallback", + ); + } + const { written, observed } = yield* Fiber.join(capture); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + const error = written.find((item) => item.type === "error"); + assert.match(error?.failure.message ?? "", /(?:save|persist|record).*?(?:progress|events)/i); + const calls = yield* Ref.get(interrupts); + if (scenario === "absent") { + assert.deepEqual(calls, []); + } else { + assert.equal(calls.length, 1); + assert.equal(calls[0]?.providerTurnId, ids.rootProviderTurnId); + assert.equal(calls[0]?.providerThread.id, ids.providerThreadId); + assert.isTrue(calls[0]?.requestRuntimeRestart); + } + for (const notice of yield* Ref.get(notices)) { + if (!notice.stopped) { + assert.notMatch( + notice.message, + /(?:stopped|cancelled) the agent|agent (?:was|has been|is) (?:stopped|cancelled)/i, + ); + } + } + if (scenario === "stop-error" || scenario === "absent") { + assert.match( + error?.failure.message ?? "", + /(?:could not|cannot|unable|not confirmed|may still).*?(?:stop|running|agent)/i, + ); + } + }), +); + +it.effect.each(["current", "stopped", "new-attempt"] as const)( + "retries failed terminal persistence with backoff while the run is %s", + (scenario) => + Effect.gen(function* () { + const key = `publication-retry:${scenario}`; + const ids = backgroundScenarioIds(key); + const firstFailure = yield* Deferred.make(); + const releaseSecondFailure = yield* Deferred.make(); + const current = yield* Ref.make({ attemptId: ids.attemptId, status: "running" }); + const writeTimes = yield* Ref.make([]); + const interrupts = yield* Ref.make(0); + const capture = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + events: () => + Stream.make(backgroundTurnItemEvent(ids, "command_execution", "running", 1)).pipe( + Stream.concat(Stream.never), + ), + ingestNormalized: (input) => + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: input.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + interruptTurn: () => Ref.update(interrupts, (count) => count + 1), + terminalWriteIsCurrent: (guard) => + Effect.gen(function* () { + assert.equal(guard.runId, ids.runId); + assert.equal(guard.threadId, ids.threadId); + assert.equal(guard.activeAttemptId, ids.attemptId); + assert.equal(guard.expectedStatus, "running"); + const owner = yield* Ref.get(current); + return ( + guard.activeAttemptId === owner.attemptId && guard.expectedStatus === owner.status + ); + }), + beforeTerminalWrite: (events) => + Effect.gen(function* () { + const now = yield* Clock.currentTimeMillis; + const times = yield* Ref.updateAndGet(writeTimes, (times) => [...times, now]); + if (times.length <= 2) { + if (times.length === 1) { + yield* Deferred.succeed(firstFailure, undefined); + } else { + yield* Deferred.await(releaseSecondFailure); + } + return yield* new EventSink.EventSinkWriteError({ + eventCount: events.length, + cause: "database is locked", + }); + } + }), + }).pipe(Effect.forkChild); + + yield* Deferred.await(firstFailure); + yield* TestClock.adjust("1 minute"); + const times = yield* Ref.get(writeTimes); + assert.equal(times.length, 2, "terminal persistence must retry after the first failed write"); + assert.isAbove(times[1]!, times[0]!, "retries must back off"); + if (scenario !== "current") { + yield* Ref.set(current, { + attemptId: + scenario === "new-attempt" ? RunAttemptId.make(`${ids.attemptId}:next`) : ids.attemptId, + status: scenario === "stopped" ? "interrupted" : "running", + }); + } + yield* Deferred.succeed(releaseSecondFailure, undefined); + yield* TestClock.adjust("1 minute"); + const { observed, written, finalEvents, committedEffects } = yield* Fiber.join(capture); + assert.equal(yield* Ref.get(interrupts), 1, "terminal retries must not repeat cancellation"); + if (scenario === "current") { + const allTimes = yield* Ref.get(writeTimes); + assert.equal(allTimes.length, 3); + assert.isAbove(allTimes[2]!, allTimes[1]!); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + assert.equal(written.filter((item) => item.type === "error").length, 1); + } else { + assert.deepEqual(finalEvents, []); + assert.deepEqual(committedEffects, []); + assert.deepEqual(written, []); + assert.deepEqual(observed, []); + } + }), +); + +it.effect("does not cancel a newer attempt after publication failure loses ownership", () => + Effect.gen(function* () { + const current = yield* Ref.make(true); + const interrupts = yield* Ref.make(0); + const { written, finalEvents, observed } = yield* captureRootRunTermination({ + key: "publication-failure:lost-owner", + shouldFinalizeRun: () => Ref.get(current), + terminalWriteIsCurrent: () => Ref.get(current), + events: (ids) => Stream.make(backgroundTurnItemEvent(ids, "command_execution", "running", 1)), + ingestNormalized: (input) => + Ref.set(current, false).pipe( + Effect.andThen( + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: input.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + ), + ), + interruptTurn: () => Ref.update(interrupts, (count) => count + 1), + }); + assert.equal(yield* Ref.get(interrupts), 0); + assert.deepEqual(written, []); + assert.deepEqual(finalEvents, []); + assert.deepEqual(observed, []); + }), +); + function captureRootRunTermination(input: { readonly key: string; readonly shouldFinalizeRun: () => Effect.Effect; readonly rejectTerminalWrite?: boolean; + readonly attemptProviderTurnId?: ProviderTurnId | null; + readonly ingestNormalized?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["ingestNormalized"]; + readonly interruptTurn?: ProviderAdapter.ProviderAdapterV2SessionRuntime["interruptTurn"]; + readonly beforeTerminalWrite?: ( + events: ReadonlyArray, + ) => Effect.Effect; + readonly terminalWriteIsCurrent?: ( + input: Parameters[0], + ) => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly seedOpenSubagent?: boolean; readonly events?: ( @@ -3886,6 +4180,7 @@ function captureRootRunTermination(input: { Ref.update(writtenItems, (current) => [...current, payload]); const captureFinalEvents = (events: ReadonlyArray) => Effect.gen(function* () { + yield* input.beforeTerminalWrite?.(events) ?? Effect.void; yield* Ref.update(finalEvents, (current) => [...current, ...events]); for (const event of events) { if (event.type === "turn-item.updated") { @@ -3917,25 +4212,28 @@ function captureRootRunTermination(input: { 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); + yield* Ref.update(committedEffects, (current) => [...current, ...payload.effects]); return []; }), writeIfRunCurrent: (payload) => Effect.gen(function* () { const effects = payload.effects ?? []; yield* Ref.update(submittedEffects, (current) => [...current, ...effects]); - if (input.rejectTerminalWrite === true) { + if ( + input.rejectTerminalWrite === true || + !(yield* input.terminalWriteIsCurrent?.(payload) ?? Effect.succeed(true)) + ) { return { committed: false, storedEvents: [] }; } - yield* Ref.update(committedEffects, (current) => [...current, ...effects]); yield* captureFinalEvents(payload.events); + yield* Ref.update(committedEffects, (current) => [...current, ...effects]); return { committed: true, storedEvents: [] }; }), }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ - ingestNormalized: () => Effect.succeed([]), + ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), }), ServerSettings.layerTest(), Layer.succeed(RunFinalizationService.RunFinalizationObserver, { @@ -3987,6 +4285,7 @@ function captureRootRunTermination(input: { close: Deferred.succeed(ingestionDone, undefined), }), startTurn: input.startTurn ?? (() => Effect.void), + interruptTurn: input.interruptTurn ?? (() => Effect.void), } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, run: { id: ids.runId, @@ -3996,7 +4295,10 @@ function captureRootRunTermination(input: { } as OrchestrationV2Run, rootNode: { id: ids.rootNodeId, - providerTurnId: ids.rootProviderTurnId, + providerTurnId: + input.attemptProviderTurnId === undefined + ? ids.rootProviderTurnId + : input.attemptProviderTurnId, } as OrchestrationV2ExecutionNode, checkpointScope: { id: CheckpointScopeId.make(`checkpoint-scope:${input.key}`), @@ -4007,7 +4309,10 @@ function captureRootRunTermination(input: { } as OrchestrationV2ProviderThread, attempt: { id: ids.attemptId, - providerTurnId: ids.rootProviderTurnId, + providerTurnId: + input.attemptProviderTurnId === undefined + ? ids.rootProviderTurnId + : input.attemptProviderTurnId, } as OrchestrationV2RunAttempt, attemptId: ids.attemptId, providerTurnOrdinal: 1, From 67583d6db1a45e3ac12914f017ae9f2f2c461d34 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 19:00:25 +0000 Subject: [PATCH 4/6] fix(server): settle runs after provider event persistence fails --- .../orchestration-v2/RunExecutionService.ts | 260 ++++++++++++------ 1 file changed, 174 insertions(+), 86 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 9abe812c3d125..0cf79a56d766b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -27,11 +27,14 @@ import { import * as Context from "effect/Context"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -972,6 +975,8 @@ export const layer: Layer.Layer< ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); + const publicationFailed = + yield* Deferred.make(); const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Set()); @@ -1210,6 +1215,114 @@ export const layer: Layer.Layer< } return true; }); + const finalizeAfterIngestionFailure = Effect.fnUntraced(function* ( + cause: Cause.Cause, + ) { + yield* Effect.logWarning("orchestration V2 provider event ingestion failed", { + runId: input.run.id, + cause, + }); + if (yield* Ref.get(rootRunFinalized)) { + return; + } + const providerThread = yield* Ref.get(latestProviderThread); + const { rootProviderTurnId } = yield* Ref.get(eventRouting); + let failure = makeProviderFailure({ cause: Cause.squash(cause), class: "unknown" }); + let threadDisposition: ProviderTerminalEvent["threadDisposition"] = "reusable"; + if (yield* Deferred.isDone(publicationFailed)) { + const stopped = yield* Effect.exit( + Effect.gen(function* () { + // The start guard checks both the original attempt and running status. + const stillOwned = yield* ( + input.shouldStartProviderTurn ?? + input.shouldFinalizeRun ?? + (() => Effect.succeed(true)) + )(); + if (!stillOwned) { + return "stale" as const; + } + if (rootProviderTurnId === null) { + return "unknown" as const; + } + yield* input.session + .interruptTurn({ + providerThread, + providerTurnId: rootProviderTurnId, + requestRuntimeRestart: true, + }) + .pipe(Effect.timeout("10 seconds")); + return "stopped" as const; + }), + ); + if (Exit.isFailure(stopped)) { + if (Cause.hasInterruptsOnly(stopped.cause)) { + return yield* Effect.failCause(stopped.cause); + } + yield* Effect.logWarning( + "orchestration V2 could not confirm provider cancellation after event publication failed", + { + runId: input.run.id, + cause: stopped.cause, + }, + ); + } else if (stopped.value === "stale") { + return; + } + const confirmed = Exit.isSuccess(stopped) && stopped.value === "stopped"; + failure = makeProviderFailure({ + class: "unknown", + message: confirmed + ? "T3 Code could not save this turn's progress, so it stopped the agent. Send a message to continue." + : "T3 Code could not save this turn's progress and could not confirm that the agent stopped. It may still be running; check the provider before continuing.", + }); + threadDisposition = confirmed ? "reusable" : "broken"; + } + const latestItemOrdinal = yield* Ref.get(latestTurnItemOrdinal); + const openSubagents = yield* Ref.get(openRunOwnedSubagents); + yield* writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread, + attempt: input.attempt, + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running", + }, + openRunOwnedSubagents: openSubagents, + terminal: { + ...makeFailedTerminalEvent(failure, latestItemOrdinal + 1), + ...(rootProviderTurnId === null ? {} : { providerTurnId: rootProviderTurnId }), + threadDisposition, + }, + failureItemPersisted: false, + refreshAfterTurn, + }).pipe( + Effect.tapError((writeCause) => + Effect.logError( + "orchestration V2 terminal persistence failed; retrying with backoff", + { + runId: input.run.id, + cause: writeCause, + }, + ), + ), + Effect.retry({ + schedule: Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + ), + }), + Effect.mapError( + (writeCause) => + new RunExecutionIngestError({ + runId: input.run.id, + cause: { ingest: cause, write: writeCause }, + }), + ), + ); + }); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); // Held text writes nothing, so its item would take a position only // when finally written, behind every item that began after it. @@ -1220,6 +1333,9 @@ export const layer: Layer.Layer< ), Stream.tap((event) => Effect.gen(function* () { + if (yield* Deferred.isDone(publicationFailed)) { + return; + } let storedEventCount = 0; const deliveredEvent = filterAssistantEvent( event, @@ -1230,37 +1346,47 @@ export const layer: Layer.Layer< const isRootProviderThreadUpdate = event.type === "provider_thread.updated" && event.providerThread.id === input.providerThread.id; - const storedEvents = yield* providerEventIngestor.ingestNormalized({ - analyticsContext: { - modelSelection: input.modelSelection, - runtimeMode: input.runtimePolicy.runtimeMode, - interactionMode: input.runtimePolicy.interactionMode, - }, - providerSessionId: input.providerSessionId, - providerInstanceId: input.run.providerInstanceId, - threadId: input.run.threadId, - runId: input.run.id, - nodeId: input.rootNode.id, - event: deliveredEvent, - ...(isRootProviderThreadUpdate - ? rootTerminalAlreadySeen - ? { - writeIfProviderThreadOwner: { - providerThreadId: input.providerThread.id, - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedLastRunOrdinal: input.run.ordinal, - }, - } - : { - writeIfRunCurrent: { - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - } - : {}), - }); + const storedEvents = yield* providerEventIngestor + .ingestNormalized({ + analyticsContext: { + modelSelection: input.modelSelection, + runtimeMode: input.runtimePolicy.runtimeMode, + interactionMode: input.runtimePolicy.interactionMode, + }, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + threadId: input.run.threadId, + runId: input.run.id, + nodeId: input.rootNode.id, + event: deliveredEvent, + ...(isRootProviderThreadUpdate + ? rootTerminalAlreadySeen + ? { + writeIfProviderThreadOwner: { + providerThreadId: input.providerThread.id, + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedLastRunOrdinal: input.run.ordinal, + }, + } + : { + writeIfRunCurrent: { + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + } + : {}), + }) + .pipe( + Effect.catchTags({ + ProviderEventPublishError: (error) => + Deferred.succeed(publicationFailed, error).pipe(Effect.as(null)), + }), + ); + if (storedEvents === null) { + return; + } storedEventCount = storedEvents.length; if ( isRootProviderThreadUpdate && @@ -1305,11 +1431,26 @@ export const layer: Layer.Layer< yield* trackChildLifecycle(event, deliveredEvent !== null); }), ), - Stream.takeUntilEffect(() => shouldStopProviderEventIngestion), + Stream.takeUntilEffect(() => + Effect.gen(function* () { + if (yield* Deferred.isDone(publicationFailed)) { + return (yield* Ref.get(eventRouting)).rootProviderTurnId !== null; + } + return yield* shouldStopProviderEventIngestion; + }), + ), Stream.runDrain, + // Keep routing after a failed write to discover the actual native turn ID, + // but a silent provider must not leave the run running indefinitely. + Effect.raceFirst( + Deferred.await(publicationFailed).pipe(Effect.andThen(Effect.sleep("10 seconds"))), + ), Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), Effect.flatMap(() => Effect.gen(function* () { + if (yield* Deferred.isDone(publicationFailed)) { + return yield* Deferred.await(publicationFailed).pipe(Effect.flatMap(Effect.fail)); + } const terminal = yield* Ref.get(terminalEvent); if (terminal === null) { // A session release interrupts or fails the stream; a clean @@ -1330,60 +1471,7 @@ export const layer: Layer.Layer< // restart recovery; every other failure fails the run. Effect.catchCauseIf( (cause) => !Cause.hasInterruptsOnly(cause), - (cause) => - Ref.get(rootRunFinalized).pipe( - Effect.flatMap((finalized) => - Effect.logWarning("orchestration V2 provider event ingestion failed", { - runId: input.run.id, - cause, - }).pipe( - Effect.andThen( - finalized - ? Effect.void - : Ref.get(latestProviderThread).pipe( - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running", - }, - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: "unknown", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, - }), - ), - ), - ), - ), - ), - ), - ), - Effect.mapError( - (writeCause) => - new RunExecutionIngestError({ - runId: input.run.id, - cause: { ingest: cause, write: writeCause }, - }), - ), - ), - ), - ), + finalizeAfterIngestionFailure, ), Effect.ensuring(eventSubscription.close), Effect.forkDetach, From 5c2b13f24a6623161b2ab76d3cdbb7bcc04fbe3e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 19:14:57 +0000 Subject: [PATCH 5/6] test(server): cover recovery cancellation races and slow stops --- .../RunExecutionService.test.ts | 127 +++++++++++++++++- 1 file changed, 122 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index ae60059641ff6..62a1b9b4b89d5 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -53,6 +53,7 @@ import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; const driver = ProviderDriverKind.make("codex"); @@ -3972,8 +3973,16 @@ it.effect.each(["known", "late", "absent", "stop-error"] as const)( yield* Deferred.succeed(releaseStop, undefined); } // Exercise discovery/stop deadlines while the provider is silent or cancellation is pending. - yield* TestClock.adjust("1 minute"); + yield* TestClock.adjust(scenario === "known" ? "15 seconds" : "1 minute"); const pendingNotices = yield* Ref.get(notices); + if (scenario === "known") { + assert.deepEqual( + pendingNotices, + [], + "slow cancellation must finish before reporting failure", + ); + assert.isFalse(yield* Ref.get(stopped)); + } for (const notice of pendingNotices) { assert.notMatch( notice.message, @@ -3981,13 +3990,13 @@ it.effect.each(["known", "late", "absent", "stop-error"] as const)( ); } yield* Deferred.succeed(releaseStop, undefined); - if (scenario === "absent") { + if (scenario === "absent" || scenario === "late") { assert.isTrue( yield* Deferred.isDone(finished), - "missing turn discovery must have a bounded fallback", + "missing turn discovery and unconfirmed cancellation must have bounded fallbacks", ); } - const { written, observed } = yield* Fiber.join(capture); + const { written, observed, finalEvents } = yield* Fiber.join(capture); assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); const error = written.find((item) => item.type === "error"); assert.match(error?.failure.message ?? "", /(?:save|persist|record).*?(?:progress|events)/i); @@ -4008,7 +4017,18 @@ it.effect.each(["known", "late", "absent", "stop-error"] as const)( ); } } - if (scenario === "stop-error" || scenario === "absent") { + if (scenario === "known") { + assert.isTrue( + yield* Ref.get(stopped), + "allow adapter cleanup beyond its internal 10s wait", + ); + assert.match(error?.failure.message ?? "", /stopped the agent/i); + assert.isTrue( + finalEvents.some( + (event) => event.type === "provider-thread.updated" && event.payload.status === "idle", + ), + ); + } else { assert.match( error?.failure.message ?? "", /(?:could not|cannot|unable|not confirmed|may still).*?(?:stop|running|agent)/i, @@ -4134,8 +4154,102 @@ it.effect("does not cancel a newer attempt after publication failure loses owner }), ); +it.effect("keeps Stop and replacement work behind an in-flight recovery cancellation", () => + Effect.gen(function* () { + const key = "publication-failure:replacement-during-cancel"; + const ids = backgroundScenarioIds(key); + const replacementAttemptId = RunAttemptId.make(`${ids.attemptId}:replacement`); + const replacementTurnId = ProviderTurnId.make(`${ids.rootProviderTurnId}:replacement`); + const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const stopStarted = yield* Deferred.make(); + const releaseStop = yield* Deferred.make(); + const current = yield* Ref.make({ attemptId: ids.attemptId, status: "running" }); + const provider = yield* Ref.make({ + turnId: ids.rootProviderTurnId, + running: true, + backgroundRunning: true, + }); + const order = yield* Ref.make([]); + const isOriginalRunning = Ref.get(current).pipe( + Effect.map((run) => run.attemptId === ids.attemptId && run.status === "running"), + ); + const capture = yield* captureRootRunTermination({ + key, + threadCommands, + shouldStartProviderTurn: () => isOriginalRunning, + shouldFinalizeRun: () => isOriginalRunning, + terminalWriteIsCurrent: () => isOriginalRunning, + events: () => + Stream.make(backgroundTurnItemEvent(ids, "command_execution", "running", 1)).pipe( + Stream.concat(Stream.never), + ), + ingestNormalized: (input) => + Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: input.providerSessionId, + eventCount: 1, + cause: "database is locked", + }), + ), + interruptTurn: (input) => + Effect.gen(function* () { + assert.equal(input.providerTurnId, ids.rootProviderTurnId); + yield* Ref.update(order, (events) => [...events, "recovery:cancel-started"]); + yield* Deferred.succeed(stopStarted, undefined); + yield* Deferred.await(releaseStop); + // OpenCode stops thread background work before matching the requested foreground turn. + yield* Ref.update(provider, (state) => ({ + ...state, + backgroundRunning: input.requestRuntimeRestart ? false : state.backgroundRunning, + running: state.turnId === input.providerTurnId ? false : state.running, + })); + yield* Ref.update(order, (events) => [...events, "recovery:cancel-finished"]); + }), + }).pipe(Effect.forkChild); + + yield* Deferred.await(stopStarted); + // Dispatch and queue advancement both use this executor in Orchestrator. + const replacement = yield* threadCommands + .withLock( + ids.threadId, + Effect.gen(function* () { + yield* Ref.set(current, { attemptId: ids.attemptId, status: "interrupted" }); + yield* Ref.update(order, (events) => [...events, "stop:committed"]); + yield* Ref.set(current, { attemptId: replacementAttemptId, status: "running" }); + yield* Ref.set(provider, { + turnId: replacementTurnId, + running: true, + backgroundRunning: true, + }); + yield* Ref.update(order, (events) => [...events, "replacement:started"]); + }), + ) + .pipe(Effect.forkChild({ startImmediately: true })); + + yield* Deferred.succeed(releaseStop, undefined); + yield* Fiber.join(capture); + yield* Fiber.join(replacement); + assert.deepEqual(yield* Ref.get(provider), { + turnId: replacementTurnId, + running: true, + backgroundRunning: true, + }); + assert.deepEqual(yield* Ref.get(current), { + attemptId: replacementAttemptId, + status: "running", + }); + assert.deepEqual(yield* Ref.get(order), [ + "recovery:cancel-started", + "recovery:cancel-finished", + "stop:committed", + "replacement:started", + ]); + }).pipe(Effect.provide(ThreadCommandExecutor.layer)), +); + function captureRootRunTermination(input: { readonly key: string; + readonly threadCommands?: ThreadCommandExecutor.ThreadCommandExecutor["Service"]; readonly shouldFinalizeRun: () => Effect.Effect; readonly rejectTerminalWrite?: boolean; readonly attemptProviderTurnId?: ProviderTurnId | null; @@ -4199,6 +4313,9 @@ function captureRootRunTermination(input: { }), Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, + input.threadCommands === undefined + ? ThreadCommandExecutor.layer + : Layer.succeed(ThreadCommandExecutor.ThreadCommandExecutor, input.threadCommands), Layer.mock(EventSink.EventSinkV2)({ write: (payload) => Effect.gen(function* () { From efc1751fec29101c2a2c31ca3ce4baacdb27c95a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 19:20:28 +0000 Subject: [PATCH 6/6] fix(server): serialize recovery cancellation with thread commands --- .../RunExecutionService.test.ts | 14 +++++ .../orchestration-v2/RunExecutionService.ts | 51 +++++++++++-------- .../src/orchestration-v2/runtimeLayer.ts | 1 + .../testkit/ProviderReplayHarness.ts | 1 + 4 files changed, 45 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 62a1b9b4b89d5..cc662b48a5ee5 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -64,6 +64,7 @@ const layerTurnItemPositionsTest = Layer.mock(TurnItemPositionStore.TurnItemPosi const layerRunExecutionTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, @@ -748,6 +749,7 @@ it.effect("passes the thread's MCP app context to the provider under a safe key" RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, Layer.mock(McpAppModelContext.McpAppModelContext)({ forThread: () => Effect.succeed([ @@ -788,6 +790,7 @@ it.effect("fails the run when its ownership check cannot be read before calling const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, @@ -1060,6 +1063,7 @@ it.effect("starts the provider when checkpoint baseline capture fails", () => const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => @@ -1179,6 +1183,7 @@ it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as co const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => @@ -1349,6 +1354,7 @@ it.effect("ingests new member threads of an owned child after the root turn term const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, @@ -1579,6 +1585,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, @@ -1935,6 +1942,7 @@ it.effect( const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, @@ -2151,6 +2159,7 @@ it.effect("drops late root provider-thread writes from a superseded attempt", () const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, @@ -2351,6 +2360,7 @@ it.effect( const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, @@ -2521,6 +2531,7 @@ it.effect( const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, @@ -2684,6 +2695,7 @@ it.effect( const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, @@ -3125,6 +3137,7 @@ it.effect( const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void, @@ -4812,6 +4825,7 @@ function runBackgroundItemScenario( const layerTest = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), layerTurnItemPositionsTest, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 0cf79a56d766b..3effa3bb2831f 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -53,6 +53,7 @@ import { } from "@t3tools/provider-core/server/failure"; import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; export interface ProviderEventRoutingState { readonly ownedThreadIds: ReadonlySet; @@ -573,6 +574,7 @@ export const layer: Layer.Layer< | ServerSettings.ServerSettingsService | McpAppModelContext.McpAppModelContext | TurnItemPositionStore.TurnItemPositionStoreV2 + | ThreadCommandExecutor.ThreadCommandExecutor > = Layer.effect( RunExecutionServiceV2, Effect.gen(function* () { @@ -583,6 +585,7 @@ export const layer: Layer.Layer< const serverSettings = yield* ServerSettings.ServerSettingsService; const mcpAppModelContext = yield* McpAppModelContext.McpAppModelContext; const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2; + const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const finalizationObserver = yield* RunFinalizationService.RunFinalizationObserver; const writeFinalRunEvents = (input: { @@ -1231,28 +1234,32 @@ export const layer: Layer.Layer< let threadDisposition: ProviderTerminalEvent["threadDisposition"] = "reusable"; if (yield* Deferred.isDone(publicationFailed)) { const stopped = yield* Effect.exit( - Effect.gen(function* () { - // The start guard checks both the original attempt and running status. - const stillOwned = yield* ( - input.shouldStartProviderTurn ?? - input.shouldFinalizeRun ?? - (() => Effect.succeed(true)) - )(); - if (!stillOwned) { - return "stale" as const; - } - if (rootProviderTurnId === null) { - return "unknown" as const; - } - yield* input.session - .interruptTurn({ - providerThread, - providerTurnId: rootProviderTurnId, - requestRuntimeRestart: true, - }) - .pipe(Effect.timeout("10 seconds")); - return "stopped" as const; - }), + threadCommands.withLock( + input.run.threadId, + Effect.gen(function* () { + // Adapters can stop thread-wide work before matching the turn ID. + // Keep Stop and queue advancement behind this ownership check and stop. + const stillOwned = yield* ( + input.shouldStartProviderTurn ?? + input.shouldFinalizeRun ?? + (() => Effect.succeed(true)) + )(); + if (!stillOwned) { + return "stale" as const; + } + if (rootProviderTurnId === null) { + return "unknown" as const; + } + yield* input.session + .interruptTurn({ + providerThread, + providerTurnId: rootProviderTurnId, + requestRuntimeRestart: true, + }) + .pipe(Effect.timeout("30 seconds")); + return "stopped" as const; + }), + ), ); if (Exit.isFailure(stopped)) { if (Cause.hasInterruptsOnly(stopped.cause)) { diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index aa44a6171e557..8c8d64c5401a9 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -150,6 +150,7 @@ const layerProviderAuthServiceProvided = ProviderAuthService.layer.pipe( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layer, layerCheckpointServiceProvided, layerEventSinkProvided, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 0342a3ccc9d94..383d0998ce60c 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -379,6 +379,7 @@ export function layerWithRegistry( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, layerCheckpointServiceProvided, layerEventSinkProvided,