diff --git a/apps/server/src/orchestration-v2/CheckpointService.test.ts b/apps/server/src/orchestration-v2/CheckpointService.test.ts index 250a079c08b70..2571bcceef2c6 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.test.ts @@ -96,3 +96,85 @@ it.effect( } }), ); + +it.effect("backs off only baseline retries after repeated workspace capture timeouts", () => + Effect.gen(function* () { + const scope: OrchestrationV2CheckpointScope = { + id: CheckpointScopeId.make("checkpoint-scope:slow-capture"), + threadId: ThreadId.make("thread:slow-capture"), + runId: RunId.make("run:slow-capture:1"), + nodeId: NodeId.make("node:slow-capture:1"), + parentScopeId: null, + providerThreadId: ProviderThreadId.make("provider-thread:slow-capture"), + kind: "root_run", + ordinalWithinParent: 0, + advancesAppRunCount: true, + cwd: "/repo/slow", + createdAt: DateTime.makeUnsafe("2026-07-28T00:00:00.000Z"), + }; + const timeout = new VcsProcessTimeoutError({ + operation: "GitVcsDriver.checkpoints.captureCheckpoint", + command: "git", + cwd: scope.cwd, + timeoutMs: 30_000, + }); + let attempts = 0; + let healthy = false; + const layerTest = CheckpointService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + Layer.mock(CheckpointStore.CheckpointStore)({ + isGitRepository: () => Effect.succeed(true), + hasCheckpointRef: () => Effect.succeed(false), + captureCheckpoint: () => + Effect.suspend(() => { + attempts++; + return healthy ? Effect.void : Effect.fail(timeout); + }), + }), + ), + ), + Layer.provideMerge(NodeCrypto.layer), + ); + yield* Effect.gen(function* () { + const checkpoints = yield* CheckpointService.CheckpointServiceV2; + const baseline = { scope, ordinalWithinScope: 0 }; + const turn = { + scope, + ordinalWithinScope: 1, + runId: scope.runId, + nodeId: scope.nodeId, + appRunOrdinal: 1, + capturedAt: scope.createdAt, + }; + yield* Effect.result(checkpoints.captureBaseline(baseline)); + assert.strictEqual((yield* checkpoints.capture(turn)).status, "error"); + assert.strictEqual(attempts, 2); + + const otherThread = { ...scope, id: CheckpointScopeId.make("checkpoint-scope:other") }; + yield* checkpoints.captureBaseline({ scope: otherThread, ordinalWithinScope: 0 }); + assert.strictEqual(attempts, 2, "baseline retried a repeatedly timing-out workspace"); + + yield* Effect.result( + checkpoints.captureBaseline({ + scope: { ...otherThread, cwd: "/repo/other" }, + ordinalWithinScope: 0, + }), + ); + assert.strictEqual(attempts, 3, "another workspace must still attempt capture"); + + const failedTurn = yield* checkpoints.capture(turn); + assert.strictEqual(failedTurn.status, "error"); + assert.strictEqual(attempts, 4, "post-turn capture must not back off"); + + yield* checkpoints.captureBaseline(baseline); + assert.strictEqual(attempts, 4, "baseline retries must wait for post-turn recovery"); + healthy = true; + assert.strictEqual((yield* checkpoints.capture(turn)).status, "ready"); + assert.strictEqual(attempts, 5); + yield* checkpoints.captureBaseline(baseline); + assert.strictEqual(attempts, 6, "a successful post-turn capture must clear backoff"); + }).pipe(Effect.provide(layerTest)); + }), +); diff --git a/apps/server/src/orchestration-v2/CheckpointService.ts b/apps/server/src/orchestration-v2/CheckpointService.ts index 09d6f83faef1e..2672db309521e 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.ts @@ -9,6 +9,7 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import * as Cache from "effect/Cache"; import * as Context from "effect/Context"; import * as Digest from "effect/crypto/Digest"; import * as DateTime from "effect/DateTime"; @@ -17,6 +18,7 @@ import * as Base64Url from "effect/encoding/Base64Url"; import * as Hex from "effect/encoding/Hex"; import * as KeyedLock from "@t3tools/shared/KeyedLock"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { isGitImport, parseTurnDiffFilesFromNumstat } from "../checkpointing/Diffs.ts"; @@ -247,6 +249,27 @@ export const layer: Layer.Layer< const withWorkspaceLock = (cwd: string, effect: Effect.Effect) => workspaceLocks.withLock(cwd, effect); + const captureTimeouts = yield* Cache.make({ + capacity: 128, + lookup: () => Effect.succeed(0), + }); + // Both callers hold the workspace lock; post-turn captures always probe for recovery. + const captureWorkspaceCheckpoint = Effect.fn("CheckpointService.captureWorkspaceCheckpoint")( + function* (input: CheckpointStore.CaptureCheckpointInput) { + yield* checkpointStore.captureCheckpoint(input).pipe( + Effect.tap(() => Cache.invalidate(captureTimeouts, input.cwd)), + Effect.tapError((error) => + error._tag === "VcsProcessTimeoutError" + ? Effect.gen(function* () { + const previous = yield* Cache.get(captureTimeouts, input.cwd); + yield* Cache.set(captureTimeouts, input.cwd, previous + 1); + }) + : Cache.invalidate(captureTimeouts, input.cwd), + ), + ); + }, + ); + const isGitCheckpointable = (cwd: string) => checkpointStore.isGitRepository(cwd).pipe(Effect.orElseSucceed(() => false)); @@ -270,16 +293,22 @@ export const layer: Layer.Layer< return; } - yield* checkpointStore - .captureCheckpoint({ - cwd: input.scope.cwd, - checkpointRef, - }) - .pipe( - Effect.catchTags({ - VcsUnsupportedOperationError: () => Effect.void, - }), + const timeouts = yield* Cache.getOption(captureTimeouts, input.scope.cwd); + if (Option.isSome(timeouts) && timeouts.value >= 2) { + return yield* Effect.logWarning( + "orchestration V2 checkpoint baseline skipped after repeated capture timeouts", + { scopeId: input.scope.id, ordinalWithinScope: input.ordinalWithinScope }, ); + } + + yield* captureWorkspaceCheckpoint({ + cwd: input.scope.cwd, + checkpointRef, + }).pipe( + Effect.catchTags({ + VcsUnsupportedOperationError: () => Effect.void, + }), + ); }), ).pipe( Effect.mapError( @@ -387,24 +416,22 @@ export const layer: Layer.Layer< }); } - const captureStatus = yield* checkpointStore - .captureCheckpoint({ - cwd: input.scope.cwd, - checkpointRef, - }) - .pipe( - Effect.as("ready" as const), - Effect.catchTags({ - VcsUnsupportedOperationError: () => Effect.succeed("missing" as const), - }), - Effect.catch((cause) => - Effect.logWarning("orchestration V2 checkpoint capture failed", { - scopeId: input.scope.id, - checkpointRef, - cause: String(cause), - }).pipe(Effect.as("error" as const)), - ), - ); + const captureStatus = yield* captureWorkspaceCheckpoint({ + cwd: input.scope.cwd, + checkpointRef, + }).pipe( + Effect.as("ready" as const), + Effect.catchTags({ + VcsUnsupportedOperationError: () => Effect.succeed("missing" as const), + }), + Effect.catch((cause) => + Effect.logWarning("orchestration V2 checkpoint capture failed", { + scopeId: input.scope.id, + checkpointRef, + cause: String(cause), + }).pipe(Effect.as("error" as const)), + ), + ); if (captureStatus !== "ready") { return makeCheckpoint({