From bba28a2b9fb91b5af8055eb070cd0559457c1dd0 Mon Sep 17 00:00:00 2001 From: Alb11747 Date: Fri, 13 Nov 2026 05:25:45 +0000 Subject: [PATCH 1/2] fix(server): runs no longer stay preparing after a full disk When workspace preparation fails, the launch service records the failure with one prepared-run.fail dispatch. On a full disk that write fails too and was only logged, so the run stayed "preparing" and held the thread's queue until a server restart, even after space was freed. Retry the write with capped backoff until storage recovers or the run settles another way, matching what run execution already does for failures after a run has started. Co-Authored-By: Claude Fable 5.2 (primary) --- .../ThreadLaunchService.test.ts | 80 +++++++++++++++- .../orchestration-v2/ThreadLaunchService.ts | 93 +++++++++++++------ 2 files changed, 145 insertions(+), 28 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 781054c969e98..163bc1b85ca6d 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -58,6 +58,7 @@ import * as TextGeneration from "../textGeneration/TextGeneration.ts"; import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as Orchestrator from "./Orchestrator.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; @@ -135,6 +136,10 @@ interface HarnessOptions { readonly getLaunchSettings?: ServerSettings.ServerSettingsService["Service"]["getSettings"]; readonly providers?: ReadonlyArray; readonly git?: Partial; + readonly dispatch?: ( + command: Parameters[0], + dispatch: ThreadManagement.ThreadManagementService["Service"]["dispatch"], + ) => ReturnType; } function makeHarness(options: HarnessOptions = {}) { @@ -238,11 +243,25 @@ function makeHarness(options: HarnessOptions = {}) { folderForThread: () => Effect.succeedNone, }), ); + const interceptDispatch = options.dispatch; + const layerLaunchThreads = + interceptDispatch === undefined + ? layerThreadManagement + : Layer.effect( + ThreadManagement.ThreadManagementService, + Effect.gen(function* () { + const threads = yield* ThreadManagement.ThreadManagementService; + return ThreadManagement.ThreadManagementService.of({ + ...threads, + dispatch: (command) => interceptDispatch(command, threads.dispatch), + }); + }), + ).pipe(Layer.provide(layerThreadManagement)); const layerLaunch = ThreadLaunch.layer.pipe( Layer.provide( Layer.mergeAll( layerExternalServices, - layerThreadManagement, + layerLaunchThreads, layerReceipts, layerOrchestrator, IdAllocator.layer, @@ -1539,6 +1558,65 @@ it.effect.each(["fails", "is interrupted"] as const)( }), ); +it.effect("fails a run whose preparation failure could not be stored until storage recovers", () => + Effect.gen(function* () { + const storageFull = yield* Deferred.make(); + let rejectedWrites = 0; + const harness = makeHarness({ + fetchWorktreeBaseCommit: () => + Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.fetchRemote", + command: "git", + cwd: project.workspaceRoot, + detail: "Git could not reach the remote.", + exitCode: 128, + }), + ), + dispatch: (command, dispatch) => + command.type !== "prepared-run.fail" + ? dispatch(command) + : Effect.suspend(() => { + if (rejectedWrites >= 2) return dispatch(command); + rejectedWrites += 1; + return Deferred.succeed(storageFull, undefined).pipe( + Effect.andThen( + Effect.fail( + new Orchestrator.OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: new Error("SQLITE_FULL: database or disk is full"), + }), + ), + ), + ); + }), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:storage-full", + thread: "thread:launch:storage-full", + message: "First message", + workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, + }), + ); + yield* Deferred.await(storageFull); + yield* TestClock.adjust("1 second"); + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && stored.event.payload.status === "failed", + ), + Stream.runHead, + ); + assert.equal(rejectedWrites, 2); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect.each(["failed", "interrupted"] as const)( "holds a message queued during setup that arrives after the setup %s", (ending) => { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index b354ba6737e30..ecd9aa0000f60 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -25,12 +25,14 @@ import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; +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 Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -932,36 +934,73 @@ const make = Effect.gen(function* () { threadId: ThreadId, runId: RunId | null, cause: unknown, - ) => - runId === null - ? Effect.logWarning("Thread workspace preparation failed", { - commandId: input.commandId, - threadId, - cause, - }) - : threads - .dispatch({ - type: "prepared-run.fail", - commandId: CommandId.make(`${input.commandId}:fail`), + ) => { + if (runId === null) { + return Effect.logWarning("Thread workspace preparation failed", { + commandId: input.commandId, + threadId, + cause, + }); + } + const persistFailure = threads.dispatch({ + type: "prepared-run.fail", + commandId: CommandId.make(`${input.commandId}:fail`), + threadId, + runId, + failure: makeProviderFailure({ + cause, + message: failureDetail(cause), + class: "validation_error", + retryable: false, + }), + }); + const runIsPreparing = threads.getThreadRecords(threadId, ["runs"], { runIds: [runId] }).pipe( + Effect.map((current) => + current.runs.some((run) => run.id === runId && run.status === "preparing"), + ), + // The read can fail in the same outage, which says nothing about the run. + Effect.catchCause(() => Effect.succeed(true)), + ); + return persistFailure.pipe( + Effect.catchCause((persistCause) => + Effect.gen(function* () { + yield* Effect.logWarning("Failed to persist thread workspace preparation failure", { + commandId: input.commandId, threadId, - runId, - failure: makeProviderFailure({ - cause, - message: failureDetail(cause), - class: "validation_error", - retryable: false, - }), - }) - .pipe( - Effect.catchCause((persistCause) => - Effect.logWarning("Failed to persist thread workspace preparation failure", { - commandId: input.commandId, - threadId, - cause, - persistCause, - }), + cause, + persistCause, + }); + if (Cause.hasInterruptsOnly(persistCause) || !(yield* runIsPreparing)) return; + // A full disk rejects this write as well, and nothing else moves the run out of + // "preparing", so it would hold the thread's queue until a restart. Retry until + // storage recovers or the run settles another way. + yield* persistFailure.pipe( + // Storage failures can be defects; retry both failure channels. + Effect.catchCauseIf( + (retryCause) => !Cause.hasInterruptsOnly(retryCause), + (retryCause) => + runIsPreparing.pipe( + Effect.flatMap((preparing) => + preparing ? Effect.fail(retryCause) : Effect.void, + ), + ), ), + Effect.retry( + Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + ), + ), + Effect.ignoreCause, + // Callers reach this from finalizers; shutdown must still be able to stop it. + Effect.interruptible, + Effect.forkIn(preparationScope), ); + }), + ), + ); + }; const reservePreparation = (commandId: CommandId) => Ref.modify(scheduledLaunches, (scheduled) => { From beba834b2ad5e821b41deb0e913caa5107723e5f Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 02:36:06 +0000 Subject: [PATCH 2/2] fix(server): retain preparation ownership during failure recovery --- .../ThreadLaunchService.test.ts | 40 +++++++++++-------- .../orchestration-v2/ThreadLaunchService.ts | 1 - 2 files changed, 24 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index f85883277a441..01e3eeff57286 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -1699,16 +1699,21 @@ it.effect("fails a run whose preparation failure could not be stored until stora Effect.gen(function* () { const storageFull = yield* Deferred.make(); let rejectedWrites = 0; + let preparations = 0; const harness = makeHarness({ fetchWorktreeBaseCommit: () => - Effect.fail( - new GitCommandError({ - operation: "GitVcsDriver.fetchRemote", - command: "git", - cwd: project.workspaceRoot, - detail: "Git could not reach the remote.", - exitCode: 128, - }), + Effect.sync(() => preparations++).pipe( + Effect.andThen( + Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.fetchRemote", + command: "git", + cwd: project.workspaceRoot, + detail: "Git could not reach the remote.", + exitCode: 128, + }), + ), + ), ), dispatch: (command, dispatch) => command.type !== "prepared-run.fail" @@ -1732,15 +1737,18 @@ it.effect("fails a run whose preparation failure could not be stored until stora yield* Effect.gen(function* () { const launches = yield* ThreadLaunch.ThreadLaunchService; const threads = yield* ThreadManagement.ThreadManagementService; - const launched = yield* launches.launch( - launchInput({ - command: "command:launch:storage-full", - thread: "thread:launch:storage-full", - message: "First message", - workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, - }), - ); + const input = launchInput({ + command: "command:launch:storage-full", + thread: "thread:launch:storage-full", + message: "First message", + workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, + }); + const launched = yield* launches.launch(input); yield* Deferred.await(storageFull); + yield* TestClock.adjust("0 seconds"); + yield* launches.launch(input); + yield* TestClock.adjust("0 seconds"); + assert.equal(preparations, 1); yield* TestClock.adjust("1 second"); yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( Stream.filter( diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index e8d87196a1c52..89179505840f8 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -1049,7 +1049,6 @@ const make = Effect.gen(function* () { Effect.ignoreCause, // Callers reach this from finalizers; shutdown must still be able to stop it. Effect.interruptible, - Effect.forkIn(preparationScope), ); }), ),