diff --git a/packages/opencode/src/session/checkpoint.ts b/packages/opencode/src/session/checkpoint.ts index 8b39ef138..980896688 100644 --- a/packages/opencode/src/session/checkpoint.ts +++ b/packages/opencode/src/session/checkpoint.ts @@ -513,7 +513,11 @@ export class Service extends Context.Service()("@opencode/Se // Writer state per session // --------------------------------------------------------------------------- -export type WriterOutcome = "success" | "failure" +// "timeout" means the caller's bounded wait expired while the writer was STILL +// IN FLIGHT — it is deliberately distinct from "failure" (the writer settled +// unsuccessfully). See waitForWriter for why conflating the two silently +// disables checkpointing for a session whose writers are merely slow. +export type WriterOutcome = "success" | "failure" | "timeout" interface WriterState { // Holds the AgentOutcome Deferred returned by Actor.spawn so callers can @@ -982,10 +986,23 @@ export const layer: Layer.Layer< // 5min so a long-but-honest writer is not mistaken for a failure by // the prune retry watcher. AgentOutcome → WriterOutcome translation: // success → "success", failure / cancelled → "failure". + // + // The bound expiring is NOT a writer failure. This timeout does not + // cancel the writer, and the settle watcher that owns the watermark + // advance (see tryStartCheckpointWriter) awaits the SAME Deferred with no + // bound — so a slow-but-successful writer still advances + // last_checkpoint_message_id after we stop waiting. Reporting "failure" + // here made the prune retry watcher count a working writer as broken, + // and MAX_WRITER_FAILURES such waits then tripped "gave up after max + // consecutive failures", permanently disabling checkpointing for a + // session whose every writer had actually succeeded. Report "timeout" so + // callers can distinguish "still in flight" from "settled unsuccessfully" + // (prune's `result !== "failure"` guard already skips the counter). const outcome = yield* Deferred.await(state.writing).pipe( Effect.timeout(300_000), - Effect.catch(() => Effect.succeed({ status: "failure", error: "timeout" })), + Effect.catch(() => Effect.succeed("timeout" as const)), ) + if (outcome === "timeout") return "timeout" as const return outcome.status === "success" ? ("success" as const) : ("failure" as const) }) diff --git a/packages/opencode/src/session/prune.ts b/packages/opencode/src/session/prune.ts index e294b3d57..b1effce0f 100644 --- a/packages/opencode/src/session/prune.ts +++ b/packages/opencode/src/session/prune.ts @@ -314,6 +314,10 @@ export const layer: Layer.Layer< writerFailures.delete(input.sessionID) return } + // "no-writer" and "timeout" both mean "not a settled failure". A + // timed-out wait leaves the writer in flight (and still able to + // advance the watermark), so counting it would retire a + // merely-slow-but-working writer. Only a real failure ticks below. if (result !== "failure") return const next = (writerFailures.get(input.sessionID) ?? 0) + 1 writerFailures.set(input.sessionID, next) diff --git a/packages/opencode/test/session/checkpoint-writer-wait-timeout.test.ts b/packages/opencode/test/session/checkpoint-writer-wait-timeout.test.ts new file mode 100644 index 000000000..906b8078f --- /dev/null +++ b/packages/opencode/test/session/checkpoint-writer-wait-timeout.test.ts @@ -0,0 +1,127 @@ +import { describe, expect } from "bun:test" +import { Deferred, Effect, Fiber, Layer } from "effect" +import * as TestClock from "effect/testing/TestClock" +import { Bus } from "../../src/bus" +import { Config } from "../../src/config" +import { Agent } from "../../src/agent/agent" +import { Memory } from "../../src/memory" +import { ActorRegistry } from "../../src/actor/registry" +import { Actor, type AgentOutcome } from "../../src/actor/spawn" +import { spawnRef } from "../../src/actor/spawn-ref" +import { TaskRegistry } from "../../src/task/registry" +import { SessionCheckpoint } from "../../src/session/checkpoint" +import { Log } from "../../src/util" +import { Plugin } from "../../src/plugin" +import { provideTmpdirInstance } from "../fixture/fixture" +import { Session as SessionNs } from "../../src/session" +import { MessageID, PartID } from "../../src/session/schema" +import { ModelID, ProviderID } from "../../src/provider/schema" +import { ProviderTest } from "../fake/provider" +import { testEffect } from "../lib/effect" +import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner" + +void Log.init({ print: false }) + +const ref = { + providerID: ProviderID.make("test"), + modelID: ModelID.make("test-model"), +} + +// Actor stub whose outcome Deferred never resolves — a writer still grinding +// through LLM round-trips when the caller's bounded wait expires. +const hangingActor = Layer.effect( + Actor.Service, + Effect.gen(function* () { + const prevSpawnRef = spawnRef.current + let counter = 0 + const impl = Actor.Service.of({ + spawn: (input) => + Effect.gen(function* () { + counter += 1 + const outcome = yield* Deferred.make() + return { actorID: `${input.agentType}-${counter}`, sessionID: input.sessionID, outcome } + }), + cancel: () => Effect.void, + getForkContext: () => Effect.succeed(undefined), + }) + spawnRef.current = impl + yield* Effect.addFinalizer( + () => + Effect.sync(() => { + if (spawnRef.current === impl) spawnRef.current = prevSpawnRef + }), + ) + return impl + }), +) + +const deps = Layer.mergeAll( + ProviderTest.fake().layer, + Agent.defaultLayer, + Plugin.defaultLayer, + Bus.layer, + Config.defaultLayer, + Memory.defaultLayer, + TaskRegistry.defaultLayer, + ActorRegistry.defaultLayer, + hangingActor, +) + +const env = Layer.mergeAll( + SessionNs.defaultLayer, + CrossSpawnSpawner.defaultLayer, + SessionCheckpoint.layer.pipe(Layer.provide(SessionNs.defaultLayer), Layer.provideMerge(deps)), +) + +const it = testEffect(env) + +describe("SessionCheckpoint.waitForWriter", () => { + it.effect( + "in-flight writer past the wait bound reports 'timeout', never 'failure'", + provideTmpdirInstance(() => + Effect.gen(function* () { + const svc = yield* SessionCheckpoint.Service + const ssn = yield* SessionNs.Service + const info = yield* ssn.create({}) + + // Writer needs at least one message to get past the empty-delta guard. + const user = yield* ssn.updateMessage({ + id: MessageID.ascending(), + role: "user", + sessionID: info.id, + agent: "build", + model: ref, + time: { created: Date.now() }, + }) + yield* ssn.updatePart({ + id: PartID.ascending(), + messageID: user.id, + sessionID: info.id, + type: "text", + text: "seed", + }) + + const started = yield* svc.tryStartCheckpointWriter({ + sessionID: info.id, + model: { providerID: "test", modelID: "test-model" }, + promptOps: {} as never, + }) + expect(started).toBe("started") + + // Drive past the 5-minute internal bound on the TestClock. The writer's + // Deferred is still unresolved, so the wait expires while the writer is + // genuinely in flight. + const fiber = yield* Effect.forkChild(svc.waitForWriter(info.id)) + yield* TestClock.adjust("6 minutes") + const result = yield* Fiber.join(fiber) + + // Regression: this used to be "failure", which made the prune retry + // watcher tick writerFailures and (after MAX_WRITER_FAILURES) trip + // "gave up after max consecutive failures" — permanently disabling + // checkpointing for a session whose writers were only slow. + expect(result).toBe("timeout") + expect(result).not.toBe("failure") + }), + ), + ) +})