From 686572fa1ed0c29441d6ca513777e2747e94ef15 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 14:40:56 +0800 Subject: [PATCH 1/7] feat(session): retry safe recovery up to 3 times with exponential backoff Safe recovery handles a clean stream disconnect before any visible output or tool side effect, where it is safe to silently replay the assistant turn. Previously it retried only once with a fixed 1s delay and then errored the session out. Raise SAFE_RECOVERY_MAX_ATTEMPTS from 1 to 3 and reuse the existing API-path delay() backoff (2s -> 4s -> 8s, capped at 30s) instead of the fixed 1s wait. The budget gate is already generic (safeRecoveryAttempt < maxAttempts); recoveryFor now reports max_attempts: 3, backoff_ms: 2000. The API-error retry path (policy(), 10 attempts) is unchanged. Closes #1006 --- packages/opencode/src/session/retry.ts | 9 ++-- .../src/session/run-incident/policy.ts | 6 +-- .../src/session/run-incident/types.ts | 2 +- .../test/session/processor-effect.test.ts | 48 +++++++++++++------ packages/opencode/test/session/retry.test.ts | 44 +++++++++++++---- .../session/run-incident-safety-gate.test.ts | 22 +++++++++ 6 files changed, 99 insertions(+), 32 deletions(-) diff --git a/packages/opencode/src/session/retry.ts b/packages/opencode/src/session/retry.ts index 0275c972e..fc6b9428f 100644 --- a/packages/opencode/src/session/retry.ts +++ b/packages/opencode/src/session/retry.ts @@ -15,8 +15,7 @@ export const RETRY_BACKOFF_FACTOR = 2 export const RETRY_MAX_DELAY_NO_HEADERS = 30_000 // 30 seconds export const RETRY_MAX_DELAY = 2_147_483_647 // max 32-bit signed integer for setTimeout export const RETRY_MAX_ATTEMPTS = 10 -export const SAFE_RECOVERY_REPLAY_DELAY = 1_000 -export const SAFE_RECOVERY_MAX_ATTEMPTS = 1 +export const SAFE_RECOVERY_MAX_ATTEMPTS = 3 function cap(ms: number) { return Math.min(ms, RETRY_MAX_DELAY) @@ -190,15 +189,17 @@ export function safeRecoveryPolicy(opts: { Effect.succeed((meta: Schedule.InputMetadata) => { if (meta.attempt > SAFE_RECOVERY_MAX_ATTEMPTS) return Cause.done(meta.attempt) return Effect.gen(function* () { + // Reuse the API-path backoff schedule: 2s -> 4s -> 8s, capped at 30s. + const wait = delay(meta.attempt) const now = yield* Clock.currentTimeMillis yield* opts.set({ attempt: meta.attempt, message: "", - next: now + SAFE_RECOVERY_REPLAY_DELAY, + next: now + wait, presentation: "recovery", reason: "network_connection_dropped", }) - return [meta.attempt, Duration.millis(SAFE_RECOVERY_REPLAY_DELAY)] as [number, Duration.Duration] + return [meta.attempt, Duration.millis(wait)] as [number, Duration.Duration] }) }), ) diff --git a/packages/opencode/src/session/run-incident/policy.ts b/packages/opencode/src/session/run-incident/policy.ts index e56f4a984..5c2e0f70f 100644 --- a/packages/opencode/src/session/run-incident/policy.ts +++ b/packages/opencode/src/session/run-incident/policy.ts @@ -38,7 +38,7 @@ export function recoveryFor(input: { recommendation: "auto_retry_once", confidence: "high", reason: "no_visible_output_or_tool_execution", - auto_retry: { max_attempts: 1, backoff_ms: 1_000 }, + auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, } } if (isBeforeFirstProviderProgressCause(input.cause) && beforeProgressBoundaryEvidenceBlocksRetry(terminalFacts)) { @@ -63,7 +63,7 @@ export function recoveryFor(input: { recommendation: "auto_retry_once", confidence: "high", reason: "reasoning_only_without_final_text_or_tool_activity", - auto_retry: { max_attempts: 1, backoff_ms: 1_000 }, + auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, } } if (!terminalFacts.side_effect_facts_complete) { @@ -150,7 +150,7 @@ export function recoveryFor(input: { recommendation: "auto_retry_once", confidence: "medium", reason: "no_visible_output_or_tool_execution", - auto_retry: { max_attempts: 1, backoff_ms: 1_000 }, + auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, } } return { ...base, recommendation: "unknown", confidence: "low", reason: "unknown" } diff --git a/packages/opencode/src/session/run-incident/types.ts b/packages/opencode/src/session/run-incident/types.ts index 511c5c3dc..f78860ef1 100644 --- a/packages/opencode/src/session/run-incident/types.ts +++ b/packages/opencode/src/session/run-incident/types.ts @@ -188,7 +188,7 @@ export type RecoveryDecision = { | "local_lifecycle_close" | "user_cancel" | "unknown" - auto_retry?: { max_attempts: 1; backoff_ms: number; attempted_at?: number } + auto_retry?: { max_attempts: number; backoff_ms: number; attempted_at?: number } user_action?: { kind: "continue" | "resume" | "retry" | "confirm_continue" | "dismiss"; idempotency_key: string } safety_scope: "visible_output_and_tool_side_effects" } diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 4395315bf..42c7d86b3 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -387,6 +387,8 @@ attemptTimeoutIt.live("reasoning connect watchdog is attempt-scoped for before-p expect(capturedAttemptConnectTimeouts).toEqual([ SessionProcessor.REASONING_FIRST_ATTEMPT_CONNECT_TIMEOUT_MS, SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, ]) expect(retryStatus).toMatchObject({ type: "retry", @@ -403,6 +405,14 @@ attemptTimeoutIt.live("reasoning connect watchdog is attempt-scoped for before-p attempt_index: 2, connect_timeout_ms: SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, }, + { + attempt_index: 3, + connect_timeout_ms: SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + }, + { + attempt_index: 4, + connect_timeout_ms: SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + }, ]) expect(parts.some((part) => part.type === "notice" && part.kind === "safe_retry_failed")).toBe(true) expect(handle.message.error).toBeUndefined() @@ -435,8 +445,12 @@ attemptTimeoutIt.live("reasoning connect watchdog is attempt-scoped for before-p expect(capturedAttemptConnectTimeouts).toEqual([ SessionProcessor.REASONING_FIRST_ATTEMPT_CONNECT_TIMEOUT_MS, SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, SessionProcessor.REASONING_FIRST_ATTEMPT_CONNECT_TIMEOUT_MS, SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, ]) }), { git: true, config: providerCfg("http://localhost:1/v1") }, @@ -545,6 +559,8 @@ attemptTimeoutIt.live("reasoning first attempt uses fast timeout with unclassifi expect(capturedAttemptConnectTimeouts).toEqual([ SessionProcessor.REASONING_FIRST_ATTEMPT_CONNECT_TIMEOUT_MS, SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, + SessionProcessor.REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS, ]) expect(parts.some((part) => part.type === "notice" && part.kind === "safe_retry_failed")).toBe(true) expect(handle.message.error).toBeUndefined() @@ -1268,7 +1284,7 @@ it.live("session.processor effect tests retry recognized structured json errors" ), ) -it.live("retryable API errors write a safe retry notice after one recovery retry", () => +it.live("retryable API errors write a safe retry notice after the recovery retries are exhausted", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { @@ -1276,7 +1292,9 @@ it.live("retryable API errors write a safe retry notice after one recovery retry yield* llm.error(503, { error: "temporarily unavailable" }) yield* llm.error(503, { error: "still unavailable" }) - yield* llm.text("third attempt should not run") + yield* llm.error(503, { error: "still unavailable" }) + yield* llm.error(503, { error: "still unavailable" }) + yield* llm.text("recovered attempt should not run") const chat = yield* session.create({}) const parent = yield* user(chat.id, "retry api twice") @@ -1308,12 +1326,12 @@ it.live("retryable API errors write a safe retry notice after one recovery retry const parts = MessageV2.parts(msg.id) expect(value).toBe("stop") - expect(yield* llm.calls).toBe(2) - expect(parts.some((part) => part.type === "text" && part.text === "third attempt should not run")).toBe(false) + expect(yield* llm.calls).toBe(4) + expect(parts.some((part) => part.type === "text" && part.text === "recovered attempt should not run")).toBe(false) expect(parts.some((part) => part.type === "notice" && part.kind === "safe_retry_failed")).toBe(true) expect(handle.message.error).toBeUndefined() - expect(handle.message.diagnostics?.run_observability?.attempts).toHaveLength(2) - expect(handle.message.diagnostics?.run_observability?.recovered_incidents).toHaveLength(1) + expect(handle.message.diagnostics?.run_observability?.attempts).toHaveLength(4) + expect(handle.message.diagnostics?.run_observability?.recovered_incidents).toHaveLength(3) }), { git: true, config: (url) => providerCfg(url) }, ), @@ -1505,7 +1523,7 @@ it.live("connect timeout auto retry stops if lifecycle closes during backoff", ( }) yield* Effect.promise(() => withLifecycleCloseAction([path.resolve(dir)], action, async () => { - await Bun.sleep(1_200) + await Bun.sleep(3_000) }), ) const value = yield* Fiber.join(run) @@ -1955,13 +1973,13 @@ it.live("reasoning-only failure with a provider-executed tool does not auto retr ), ) -it.live("reasoning-only retry writes a notice after the one safe retry is exhausted", () => +it.live("reasoning-only retry writes a notice after the safe retries are exhausted", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { const { processors, session, provider } = yield* boot() - for (const suffix of ["first", "second"]) { + for (const suffix of ["first", "second", "third", "fourth"]) { yield* llm.push( raw({ head: [ @@ -1988,7 +2006,7 @@ it.live("reasoning-only retry writes a notice after the one safe retry is exhaus }), ) } - yield* llm.text("third attempt should not run") + yield* llm.text("recovered attempt should not run") const chat = yield* session.create({}) const parent = yield* user(chat.id, "reasoning retry fails twice") @@ -2025,12 +2043,12 @@ it.live("reasoning-only retry writes a notice after the one safe retry is exhaus const parts = MessageV2.parts(msg.id) expect(value).toBe("stop") - expect(yield* llm.calls).toBe(2) + expect(yield* llm.calls).toBe(4) expect(parts.some((part) => part.type === "reasoning")).toBe(false) - expect(parts.some((part) => part.type === "text" && part.text === "third attempt should not run")).toBe(false) + expect(parts.some((part) => part.type === "text" && part.text === "recovered attempt should not run")).toBe(false) expect(parts.some((part) => part.type === "notice" && part.kind === "safe_retry_failed")).toBe(true) expect(handle.message.error).toBeUndefined() - expect(handle.message.diagnostics?.run_observability?.recovered_incidents).toHaveLength(1) + expect(handle.message.diagnostics?.run_observability?.recovered_incidents).toHaveLength(3) }), { git: true, config: (url) => providerCfg(url) }, ), @@ -2677,6 +2695,8 @@ it.live("connect timeout writes a safe retry notice and flips session_status idl const { processors, session, provider } = yield* boot() const sts = yield* SessionStatus.Service + yield* llm.hang + yield* llm.hang yield* llm.hang yield* llm.hang @@ -2714,7 +2734,7 @@ it.live("connect timeout writes a safe retry notice and flips session_status idl const state = yield* sts.get(chat.id) expect(result).toBe("stop") - expect(yield* llm.calls).toBe(2) + expect(yield* llm.calls).toBe(4) expect(handle.message.error).toBeUndefined() expect(parts.some((part) => part.type === "notice" && part.kind === "safe_retry_failed")).toBe(true) expect(stored.info.role).toBe("assistant") diff --git a/packages/opencode/test/session/retry.test.ts b/packages/opencode/test/session/retry.test.ts index 637d70ba3..2dbe8eab0 100644 --- a/packages/opencode/test/session/retry.test.ts +++ b/packages/opencode/test/session/retry.test.ts @@ -2,7 +2,7 @@ import { describe, expect, test } from "bun:test" import type { NamedError } from "@opencode-ai/util/error" import { APICallError } from "ai" import { setTimeout as sleep } from "node:timers/promises" -import { Effect, Exit, Pull, Schedule } from "effect" +import { Clock, Effect, Exit, Pull, Schedule } from "effect" import { SessionRetry } from "../../src/session/retry" import { MessageV2 } from "../../src/session/message-v2" import { ProviderID } from "../../src/provider/schema" @@ -171,7 +171,7 @@ describe("session.retry.delay", () => { }) }) - test("safe recovery policy stops after the one replay budget is exhausted", async () => { + test("safe recovery policy uses exponential backoff across the replay budget then stops", async () => { const statuses: Array<{ attempt: number message: string @@ -180,6 +180,8 @@ describe("session.retry.delay", () => { reason: "network_connection_dropped" }> = [] + const observedWaits: number[] = [] + const exit = await Effect.runPromise( Effect.gen(function* () { const step = yield* Schedule.toStepWithMetadata( @@ -187,7 +189,15 @@ describe("session.retry.delay", () => { set: (info) => Effect.sync(() => statuses.push(info)), }), ) - yield* step(undefined) + // Exhaust the full budget of automatic replays, capturing the scheduled + // wait for each one. info.next is `now + delay(attempt)`, so subtracting + // the clock reading taken just before the step recovers the backoff. + for (let attempt = 0; attempt < SessionRetry.SAFE_RECOVERY_MAX_ATTEMPTS; attempt++) { + const before = yield* Clock.currentTimeMillis + yield* step(undefined) + observedWaits.push(statuses[statuses.length - 1]!.next - before) + } + // The next step exceeds the budget and must terminate the schedule. return yield* Effect.exit(step(undefined)) }), ) @@ -196,14 +206,28 @@ describe("session.retry.delay", () => { if (Exit.isFailure(exit)) { expect(Pull.isDoneCause(exit.cause)).toBe(true) } - expect(statuses).toHaveLength(1) - expect(statuses[0]).toMatchObject({ - attempt: 1, - message: "", - presentation: "recovery", - reason: "network_connection_dropped", + expect(statuses).toHaveLength(SessionRetry.SAFE_RECOVERY_MAX_ATTEMPTS) + expect(statuses.map((status) => status.attempt)).toEqual([1, 2, 3]) + // Backoff reuses the API-path delay(): 2s -> 4s -> 8s. The monotonic clock can + // only add to the gap, so each observed wait is at least the scheduled backoff + // and comfortably under the next step up. + const expectedBackoff = [SessionRetry.delay(1), SessionRetry.delay(2), SessionRetry.delay(3)] + expect(expectedBackoff).toEqual([2000, 4000, 8000]) + observedWaits.forEach((wait, index) => { + expect(wait).toBeGreaterThanOrEqual(expectedBackoff[index]!) + expect(wait).toBeLessThan(expectedBackoff[index]! + 1000) }) - }) + expect( + statuses.every( + (status) => + status.message === "" && + status.presentation === "recovery" && + status.reason === "network_connection_dropped", + ), + ).toBe(true) + // This exercises the real schedule, so it waits the full 2s + 4s + 8s backoff. + // Override Bun's 5s default timeout to keep it green on focused runs and CI. + }, 20_000) test("policy stops retrying after the configured max attempts", async () => { const attempts: number[] = [] diff --git a/packages/opencode/test/session/run-incident-safety-gate.test.ts b/packages/opencode/test/session/run-incident-safety-gate.test.ts index d34b2b569..1d49b2698 100644 --- a/packages/opencode/test/session/run-incident-safety-gate.test.ts +++ b/packages/opencode/test/session/run-incident-safety-gate.test.ts @@ -43,6 +43,28 @@ describe("run incident safety gate", () => { }) }) + test("allows replays across a multi-attempt budget and blocks once exhausted", () => { + const recovery = { + ...base, + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", + auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, + } as const + + for (const safeRecoveryAttempt of [0, 1, 2]) { + expect(RunIncident.evaluateReplaySafety({ recovery, safeRecoveryAttempt })).toMatchObject({ + canReplay: true, + recoveryMode: "replay", + }) + } + + expect(RunIncident.evaluateReplaySafety({ recovery, safeRecoveryAttempt: 3 })).toMatchObject({ + canReplay: false, + recoveryMode: "auto_replay_blocked", + blockedReason: "safe_recovery_budget_exhausted", + }) + }) + test("keeps visible-output recovery as continuation instead of replay", () => { const decision = RunIncident.evaluateReplaySafety({ recovery: { From fe9721a91beadd441105e2432c5c5761c3616864 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 15:52:27 +0800 Subject: [PATCH 2/7] refactor(session): rename auto_retry_once to auto_retry The recommendation value carried a misleading `_once` suffix now that safe recovery retries up to 3 times. Rename across type definition, policy, safety gate, presentation, and all test fixtures. --- .../src/session/run-incident/policy.ts | 6 ++--- .../src/session/run-incident/presentation.ts | 2 +- .../src/session/run-incident/safety-gate.ts | 2 +- .../src/session/run-incident/types.ts | 2 +- packages/opencode/test/session/export.test.ts | 4 +-- .../test/session/processor-effect.test.ts | 8 +++--- .../test/session/retry-decision.test.ts | 2 +- .../session/run-incident-safety-gate.test.ts | 6 ++--- .../test/session/run-observability.test.ts | 26 +++++++++---------- 9 files changed, 29 insertions(+), 29 deletions(-) diff --git a/packages/opencode/src/session/run-incident/policy.ts b/packages/opencode/src/session/run-incident/policy.ts index 5c2e0f70f..9c717b3a1 100644 --- a/packages/opencode/src/session/run-incident/policy.ts +++ b/packages/opencode/src/session/run-incident/policy.ts @@ -35,7 +35,7 @@ export function recoveryFor(input: { ) { return { ...base, - recommendation: "auto_retry_once", + recommendation: "auto_retry", confidence: "high", reason: "no_visible_output_or_tool_execution", auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, @@ -60,7 +60,7 @@ export function recoveryFor(input: { ) { return { ...base, - recommendation: "auto_retry_once", + recommendation: "auto_retry", confidence: "high", reason: "reasoning_only_without_final_text_or_tool_activity", auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, @@ -147,7 +147,7 @@ export function recoveryFor(input: { if (retryableTransport) { return { ...base, - recommendation: "auto_retry_once", + recommendation: "auto_retry", confidence: "medium", reason: "no_visible_output_or_tool_execution", auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, diff --git a/packages/opencode/src/session/run-incident/presentation.ts b/packages/opencode/src/session/run-incident/presentation.ts index 8ac38a6e7..f67c4b03d 100644 --- a/packages/opencode/src/session/run-incident/presentation.ts +++ b/packages/opencode/src/session/run-incident/presentation.ts @@ -30,7 +30,7 @@ export function plainSummary(input: { cause: TerminalCause; facts: IncidentFacts } function actionKey(recovery: RecoveryDecision) { - if (recovery.recommendation === "auto_retry_once") return "run_incident.action.retry" + if (recovery.recommendation === "auto_retry") return "run_incident.action.retry" if (recovery.recommendation === "offer_continue") return "run_incident.action.continue" if (recovery.recommendation === "offer_resume_with_confirmation") return "run_incident.action.confirm_continue" if (recovery.recommendation === "ask_user_before_retry") return "run_incident.action.confirm_retry" diff --git a/packages/opencode/src/session/run-incident/safety-gate.ts b/packages/opencode/src/session/run-incident/safety-gate.ts index f97defdc8..ad4621041 100644 --- a/packages/opencode/src/session/run-incident/safety-gate.ts +++ b/packages/opencode/src/session/run-incident/safety-gate.ts @@ -12,7 +12,7 @@ export function evaluateReplaySafety(input: { }): ReplaySafetyDecision { const safety = input.recovery - if (safety.recommendation === "auto_retry_once") { + if (safety.recommendation === "auto_retry") { const maxAttempts = safety.auto_retry?.max_attempts ?? 1 if (input.safeRecoveryAttempt < maxAttempts) { return { diff --git a/packages/opencode/src/session/run-incident/types.ts b/packages/opencode/src/session/run-incident/types.ts index f78860ef1..d5480ec6e 100644 --- a/packages/opencode/src/session/run-incident/types.ts +++ b/packages/opencode/src/session/run-incident/types.ts @@ -168,7 +168,7 @@ export type MaterializedToolBoundary = { export type RecoveryDecision = { recommendation: - | "auto_retry_once" + | "auto_retry" | "offer_continue" | "offer_resume_with_confirmation" | "ask_user_before_retry" diff --git a/packages/opencode/test/session/export.test.ts b/packages/opencode/test/session/export.test.ts index 7860fd767..38d2d8983 100644 --- a/packages/opencode/test/session/export.test.ts +++ b/packages/opencode/test/session/export.test.ts @@ -2264,7 +2264,7 @@ describe("redactPart", () => { monotonicMs: 125, technical_retryable: true, safety_gate_decision: { - recommendation: "auto_retry_once", + recommendation: "auto_retry", confidence: "high", reason: "no_visible_output_or_tool_execution", safety_scope: "visible_output_and_tool_side_effects", @@ -2321,7 +2321,7 @@ describe("redactPart", () => { }) expect(sanitized.diagnostics.run_observability?.[0]?.recovery_decision).toMatchObject({ technical_retryable: true, - safety_gate_recommendation: "auto_retry_once", + safety_gate_recommendation: "auto_retry", recovery_mode: "replay", attempt_kind: "safe_recovery_replay", timeout_policy: "reasoning_first_attempt", diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 42c7d86b3..45cb3f47d 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -565,7 +565,7 @@ attemptTimeoutIt.live("reasoning first attempt uses fast timeout with unclassifi expect(parts.some((part) => part.type === "notice" && part.kind === "safe_retry_failed")).toBe(true) expect(handle.message.error).toBeUndefined() expect(handle.message.diagnostics?.run_observability?.recovered_incidents?.[0]?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }), @@ -1449,7 +1449,7 @@ it.live("connect timeout before provider progress auto retries once and succeeds subcategory: "connect", }) expect(observability?.recovered_incidents?.[0]?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) expect(observability?.attempts).toHaveLength(2) @@ -1682,7 +1682,7 @@ it.live("disabled unknown tools do not block safe connect-timeout auto retry", ( proof_result: "complete", }) expect(stored.info.diagnostics?.run_observability?.recovered_incidents?.[0]?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) } @@ -1784,7 +1784,7 @@ it.live("reasoning-only retry removes failed reasoning before replaying the assi if (stored?.info.role === "assistant") { expect(stored.info.error).toBeUndefined() expect(stored.info.diagnostics?.run_observability?.recovered_incidents?.[0]?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "reasoning_only_without_final_text_or_tool_activity", }) } diff --git a/packages/opencode/test/session/retry-decision.test.ts b/packages/opencode/test/session/retry-decision.test.ts index 386a7c085..3c0314213 100644 --- a/packages/opencode/test/session/retry-decision.test.ts +++ b/packages/opencode/test/session/retry-decision.test.ts @@ -3,7 +3,7 @@ import { buildModelRetryDecision, selectRetryTimeoutPolicy } from "../../src/ses import type { RunIncident } from "../../src/session/run-incident" const safeReplayGate: RunIncident.Recovery = { - recommendation: "auto_retry_once", + recommendation: "auto_retry", confidence: "high", reason: "reasoning_only_without_final_text_or_tool_activity", auto_retry: { max_attempts: 1, backoff_ms: 1_000 }, diff --git a/packages/opencode/test/session/run-incident-safety-gate.test.ts b/packages/opencode/test/session/run-incident-safety-gate.test.ts index 1d49b2698..b37516cd3 100644 --- a/packages/opencode/test/session/run-incident-safety-gate.test.ts +++ b/packages/opencode/test/session/run-incident-safety-gate.test.ts @@ -11,7 +11,7 @@ describe("run incident safety gate", () => { const decision = RunIncident.evaluateReplaySafety({ recovery: { ...base, - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "reasoning_only_without_final_text_or_tool_activity", auto_retry: { max_attempts: 1, backoff_ms: 1_000 }, }, @@ -29,7 +29,7 @@ describe("run incident safety gate", () => { const decision = RunIncident.evaluateReplaySafety({ recovery: { ...base, - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", auto_retry: { max_attempts: 1, backoff_ms: 1_000 }, }, @@ -46,7 +46,7 @@ describe("run incident safety gate", () => { test("allows replays across a multi-attempt budget and blocks once exhausted", () => { const recovery = { ...base, - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, } as const diff --git a/packages/opencode/test/session/run-observability.test.ts b/packages/opencode/test/session/run-observability.test.ts index 8b9f969da..23fd49035 100644 --- a/packages/opencode/test/session/run-observability.test.ts +++ b/packages/opencode/test/session/run-observability.test.ts @@ -131,7 +131,7 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 131, monotonicMs: 231 }) expect(decision).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) expect(summary.incident?.terminal_cause).toMatchObject({ @@ -240,7 +240,7 @@ describe("RunObservability", () => { expect(summary.recovery_decision).toMatchObject({ technical_retryable: true, - safety_gate_recommendation: "auto_retry_once", + safety_gate_recommendation: "auto_retry", safety_gate_reason: "no_visible_output_or_tool_execution", recovery_mode: "replay", attempt_kind: "safe_recovery_replay", @@ -1137,7 +1137,7 @@ describe("RunObservability", () => { }) expect(decision).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }) @@ -1181,7 +1181,7 @@ describe("RunObservability", () => { tool_input_started: true, }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) expect(summary.retry_safety).toMatchObject({ @@ -1273,11 +1273,11 @@ describe("RunObservability", () => { proof_reason: "unknown_tool_boundary", }) expect(decision).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }) @@ -1307,7 +1307,7 @@ describe("RunObservability", () => { proof_reason: "all_boundaries_classified", }) expect(decision).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }) @@ -1402,7 +1402,7 @@ describe("RunObservability", () => { for (const [name, overrides] of cases) { const decision = recoveryForBeforeProgress(overrides) - expect(decision, name).not.toMatchObject({ recommendation: "auto_retry_once" }) + expect(decision, name).not.toMatchObject({ recommendation: "auto_retry" }) } }) @@ -1993,11 +1993,11 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 16, monotonicMs: 160 }) expect(decision).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "reasoning_only_without_final_text_or_tool_activity", }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "reasoning_only_without_final_text_or_tool_activity", }) expect(summary.retry_safety).toMatchObject({ @@ -2276,7 +2276,7 @@ describe("RunObservability", () => { subcategory: "before_first_provider_progress", }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }) @@ -2317,7 +2317,7 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 22, monotonicMs: 220 }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }) @@ -2353,7 +2353,7 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 23, monotonicMs: 230 }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "auto_retry_once", + recommendation: "auto_retry", reason: "no_visible_output_or_tool_execution", }) }) From 01c464b243f83afa46c2f6a754e38154ad8d4bca Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 16:14:44 +0800 Subject: [PATCH 3/7] fix(session): let budget-exhausted decision overwrite intermediate replay in recorder When safe recovery retries exhaust the budget, the final auto_replay_blocked decision now overwrites the earlier replay decision in the recorder diagnostics. Previously the recorder skipped all subsequent recordRecoveryDecision calls after the first retry_attempted, so the diagnostic summary never reflected the budget exhaustion reason. --- .../src/session/run-observability/recorder.ts | 6 +- .../test/session/run-observability.test.ts | 149 +++++++++++++++++- 2 files changed, 148 insertions(+), 7 deletions(-) diff --git a/packages/opencode/src/session/run-observability/recorder.ts b/packages/opencode/src/session/run-observability/recorder.ts index 2bad70947..8bb234c94 100644 --- a/packages/opencode/src/session/run-observability/recorder.ts +++ b/packages/opencode/src/session/run-observability/recorder.ts @@ -534,7 +534,11 @@ export function createRecorder(input: RecorderInput): Recorder { return deriveCurrentIncident(next.at, { includeRecoveredTerminal: true })?.recovery ?? unknownRecovery() }, recordRecoveryDecision(next) { - if (recoveryDecision?.retry_attempted && next.safe_recovery_attempt > recoveryDecision.safe_recovery_attempt) { + if ( + recoveryDecision?.retry_attempted && + next.safe_recovery_attempt > recoveryDecision.safe_recovery_attempt && + next.recovery_mode !== "auto_replay_blocked" + ) { rememberEvent(next.monotonicMs) return } diff --git a/packages/opencode/test/session/run-observability.test.ts b/packages/opencode/test/session/run-observability.test.ts index 23fd49035..6bc78b371 100644 --- a/packages/opencode/test/session/run-observability.test.ts +++ b/packages/opencode/test/session/run-observability.test.ts @@ -256,7 +256,7 @@ describe("RunObservability", () => { }) }) - test("does not mark replay recovered before the recovery attempt succeeds", () => { + test("exhausted budget overwrites replay decision with blocked state", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_recovery_decision_failed_attempt"), traceID: MessageID.make("msg_recovery_decision_failed_attempt"), @@ -334,11 +334,148 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 10, monotonicMs: 190 }) expect(summary.recovery_decision).toMatchObject({ - retry_attempted: true, - failed_attempt_provider_progress_seen: false, - recovery_attempt_id: second.attemptID, - recovery_attempt_provider_progress_seen: true, - outcome: "failed", + recovery_mode: "auto_replay_blocked", + blocked_reason: "safe_recovery_budget_exhausted", + safe_recovery_attempt: 1, + presentation: "safe_recovery_failed", + outcome: "blocked", + }) + }) + + test("budget-exhausted recovery decision overwrites intermediate replay decision", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_budget_exhausted_overwrites"), + traceID: MessageID.make("msg_budget_exhausted_overwrites"), + sessionID: SessionID.make("ses_budget_exhausted_overwrites"), + messageID: MessageID.make("msg_budget_exhausted_overwrites"), + providerID: "test", + modelID: "test-model", + createdAt: 1, + monotonicStartMs: 100, + }) + + const a1 = recorder.beginAttempt({ attemptIndex: 1, at: 2, monotonicMs: 110, connectTimeoutMs: 60_000 }) + recorder.recordSideEffectBoundarySnapshot({ + attemptID: a1.attemptID, + at: 2, + monotonicMs: 112, + snapshot: { + exposed_tool_count: 0, + unknown_tool_count: 0, + unclassified_effect_count: 0, + provider_executed_capability_present: false, + external_boundary_present: false, + proof_result: "complete", + proof_reason: "all_boundaries_classified", + }, + }) + const r1 = recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: a1.attemptID, + at: 3, + monotonicMs: 120, + error: new Error("LLM stream connection timed out after 60000ms without provider progress"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + retryable: true, + }) + recorder.recordRecoveryDecision({ + attemptID: a1.attemptID, + at: 4, + monotonicMs: 130, + technical_retryable: true, + safety_gate_decision: r1, + recovery_mode: "replay", + attempt_kind: "safe_recovery_replay", + model_stream_attempt: 1, + safe_recovery_attempt: 0, + timeout_policy: "reasoning_first_attempt", + presentation: "recovery", + }) + recorder.recordAutoRetryAttempted({ attemptID: a1.attemptID, at: 5, monotonicMs: 140 }) + + const a2 = recorder.beginAttempt({ attemptIndex: 2, at: 6, monotonicMs: 150, connectTimeoutMs: 60_000 }) + const r2 = recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: a2.attemptID, + at: 7, + monotonicMs: 160, + error: new Error("LLM stream connection timed out"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + retryable: true, + }) + recorder.recordRecoveryDecision({ + attemptID: a2.attemptID, + at: 8, + monotonicMs: 170, + technical_retryable: true, + safety_gate_decision: r2, + recovery_mode: "replay", + attempt_kind: "safe_recovery_replay", + model_stream_attempt: 2, + safe_recovery_attempt: 1, + timeout_policy: "reasoning_safe_recovery", + presentation: "recovery", + }) + recorder.recordAutoRetryAttempted({ attemptID: a2.attemptID, at: 9, monotonicMs: 180 }) + + const a3 = recorder.beginAttempt({ attemptIndex: 3, at: 10, monotonicMs: 190, connectTimeoutMs: 60_000 }) + const r3 = recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: a3.attemptID, + at: 11, + monotonicMs: 200, + error: new Error("LLM stream connection timed out"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + retryable: true, + }) + recorder.recordRecoveryDecision({ + attemptID: a3.attemptID, + at: 12, + monotonicMs: 210, + technical_retryable: true, + safety_gate_decision: r3, + recovery_mode: "replay", + attempt_kind: "safe_recovery_replay", + model_stream_attempt: 3, + safe_recovery_attempt: 2, + timeout_policy: "reasoning_safe_recovery", + presentation: "recovery", + }) + recorder.recordAutoRetryAttempted({ attemptID: a3.attemptID, at: 13, monotonicMs: 220 }) + + const a4 = recorder.beginAttempt({ attemptIndex: 4, at: 14, monotonicMs: 230, connectTimeoutMs: 60_000 }) + const r4 = recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: a4.attemptID, + at: 15, + monotonicMs: 240, + error: new Error("LLM stream connection timed out"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + retryable: true, + }) + recorder.recordRecoveryDecision({ + attemptID: a4.attemptID, + at: 16, + monotonicMs: 250, + technical_retryable: true, + safety_gate_decision: r4, + recovery_mode: "auto_replay_blocked", + blocked_reason: "safe_recovery_budget_exhausted", + attempt_kind: "safe_recovery_replay", + model_stream_attempt: 4, + safe_recovery_attempt: 3, + timeout_policy: "reasoning_safe_recovery", + presentation: "safe_recovery_failed", + }) + + const summary = recorder.finalize({ completedAt: 17, monotonicMs: 260 }) + + expect(summary.recovery_decision).toMatchObject({ + recovery_mode: "auto_replay_blocked", + blocked_reason: "safe_recovery_budget_exhausted", + safe_recovery_attempt: 3, + presentation: "safe_recovery_failed", + outcome: "blocked", }) }) From c0b9f4461f8952228ad2406c4c3de6fcf28aa20b Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 16:43:37 +0800 Subject: [PATCH 4/7] fix(session): interrupt safe-recovery backoff on maintenance lifecycle close Race the backoff sleep against a 500ms lifecycle-close polling loop so that beginLifecycleClose (maintenance reload/dispose) is detected during the sleep, not only at the before/after_backoff checkpoints. Also check isLifecycleClosing in retryStillAllowed so the polling path and the checkpoint path both cover maintenance closes. --- .../src/session/lifecycle-provenance.ts | 4 ++ packages/opencode/src/session/processor.ts | 64 +++++++++++++----- .../test/session/processor-effect.test.ts | 67 ++++++++++++++++++- 3 files changed, 117 insertions(+), 18 deletions(-) diff --git a/packages/opencode/src/session/lifecycle-provenance.ts b/packages/opencode/src/session/lifecycle-provenance.ts index ea0c5273e..40a7b9d5b 100644 --- a/packages/opencode/src/session/lifecycle-provenance.ts +++ b/packages/opencode/src/session/lifecycle-provenance.ts @@ -142,6 +142,10 @@ function hasLifecycleClose(directories: readonly string[]): boolean { return directories.some((directory) => (closingByDirectory.get(directory) ?? 0) > 0) } +export function isLifecycleClosing(directory: string): boolean { + return (closingByDirectory.get(directory) ?? 0) > 0 +} + function notifyCloseWaiters() { for (const waiter of [...closeWaiters]) { if (hasLifecycleClose([waiter.directory])) continue diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 5750ecb22..2ac2dadbe 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -27,7 +27,7 @@ import { InstanceState } from "@/effect/instance-state" import { TurnChange } from "./turn-change" import { LLMTrace } from "./llm-trace" import { RunObservability } from "./run-observability" -import { currentLifecycleCloseAction, lifecycleCloseActionMeta } from "./lifecycle-provenance" +import { currentLifecycleCloseAction, isLifecycleClosing, lifecycleCloseActionMeta } from "./lifecycle-provenance" const log = Log.create({ service: "session.processor" }) const TOOL_CLEANUP_TIMEOUT_MS = 1_000 @@ -1231,19 +1231,34 @@ export const layer: Layer.Layer< const retryStillAllowed = Effect.fn("SessionProcessor.retryStillAllowed")(function* (stage: string) { const lifecycleAction = currentLifecycleCloseAction(ctx.directory) - if (!lifecycleAction) return { allowed: true as const } - ctx.runTrace.recordScopeClosed({ - at: Date.now(), - monotonicMs: performance.now(), - source: `session.processor.safe_recovery.${stage}`, - reason: "lifecycle_close_before_auto_retry", - propagationPoint: "session.processor.safe_recovery", - ...lifecycleCloseActionMeta(lifecycleAction), - }) - return { - allowed: false as const, - interruptionMessage: LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE, + if (lifecycleAction) { + ctx.runTrace.recordScopeClosed({ + at: Date.now(), + monotonicMs: performance.now(), + source: `session.processor.safe_recovery.${stage}`, + reason: "lifecycle_close_before_auto_retry", + propagationPoint: "session.processor.safe_recovery", + ...lifecycleCloseActionMeta(lifecycleAction), + }) + return { + allowed: false as const, + interruptionMessage: LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE, + } } + if (isLifecycleClosing(ctx.directory)) { + ctx.runTrace.recordScopeClosed({ + at: Date.now(), + monotonicMs: performance.now(), + source: `session.processor.safe_recovery.${stage}`, + reason: "maintenance_lifecycle_close_pending", + propagationPoint: "session.processor.safe_recovery", + }) + return { + allowed: false as const, + interruptionMessage: LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE, + } + } + return { allowed: true as const } }) const retrySignalFor = (error: unknown) => { @@ -1428,15 +1443,30 @@ export const layer: Layer.Layer< if (beforeRetry.allowed) { automaticStreamRetriesUsed += 1 yield* removeReasoningForAttempt(attemptID) - const safeRecoveryScheduled = yield* safeRecoveryStep(undefined).pipe( - Effect.as(true), - Effect.catchCause(() => Effect.succeed(false)), + const lifecycleCloseWatch = Effect.gen(function* () { + while (!isLifecycleClosing(ctx.directory) && !currentLifecycleCloseAction(ctx.directory)) { + yield* Effect.sleep("500 millis") + } + }).pipe(Effect.as("lifecycle_close" as const)) + const backoffResult = yield* Effect.race( + safeRecoveryStep(undefined).pipe(Effect.as("scheduled" as const)), + lifecycleCloseWatch, + ).pipe( + Effect.catchCause(() => Effect.succeed("exhausted" as const)), Effect.onInterrupt(() => recordProcessInterrupt(attemptID)), ) - if (!safeRecoveryScheduled) { + if (backoffResult === "exhausted") { yield* writeSafeRetryFailedNotice(attemptID) break } + if (backoffResult === "lifecycle_close") { + const closeCheck = yield* retryStillAllowed("during_backoff") + yield* halt(result.error, attemptID, { + recordFailure: false, + interruptionMessage: closeCheck.allowed ? LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE : closeCheck.interruptionMessage, + }) + break + } const afterRetry = yield* retryStillAllowed("after_backoff") if (afterRetry.allowed) { ctx.runTrace.recordAutoRetryAttempted({ diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 45cb3f47d..1dbfc80f6 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -18,7 +18,7 @@ import { LLM } from "../../src/session/llm" import { MessageV2 } from "../../src/session/message-v2" import { SessionProcessor } from "../../src/session/processor" import { SessionDiagnostics } from "../../src/session/diagnostics" -import { createLifecycleCloseAction, withLifecycleCloseAction } from "../../src/session/lifecycle-provenance" +import { beginLifecycleClose, createLifecycleCloseAction, withLifecycleCloseAction } from "../../src/session/lifecycle-provenance" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SessionStatus } from "../../src/session/status" import { SessionSummary } from "../../src/session/summary" @@ -1549,6 +1549,71 @@ it.live("connect timeout auto retry stops if lifecycle closes during backoff", ( ), ) +it.live("maintenance lifecycle close during backoff interrupts safe recovery", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const retrySeen = defer() + const { processors, session, provider } = yield* boot() + const bus = yield* Bus.Service + + yield* llm.hang + yield* llm.text("should not run") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "maintenance close during backoff") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const off = yield* bus.subscribeCallback(SessionStatus.Event.Status, (evt) => { + if (evt.properties.sessionID !== chat.id) return + if (evt.properties.status.type === "retry") retrySeen.resolve() + }) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const run = yield* handle + .process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "maintenance close during backoff" }], + tools: {}, + connectTimeoutMs: 20, + streamTimeoutMs: 1_000, + }) + .pipe(Effect.forkChild) + + yield* Effect.promise(() => retrySeen.promise) + const releaseClose = beginLifecycleClose([path.resolve(dir)]) + yield* Fiber.join(run) + releaseClose() + off() + + const stored = (yield* session.messages({ sessionID: chat.id })).find( + (message) => message.info.role === "assistant" && message.info.id === msg.id, + ) + expect(yield* llm.calls).toBe(1) + expect(stored?.info.role).toBe("assistant") + if (stored?.info.role === "assistant") { + expect(stored.info.error?.data.message).toContain("lifecycle close") + } + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + it.live("connect timeout auto retry records abort if interrupted during backoff", () => provideTmpdirServer( ({ dir, llm }) => From 63e7f6392ccc6a05dd27d2f56b847fde826bf515 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 17:42:04 +0800 Subject: [PATCH 5/7] refactor(session): single-source retry constants, semantic lifecycle signal, injectable delay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P3-1: policy.ts now imports SAFE_RECOVERY_MAX_ATTEMPTS and RETRY_INITIAL_DELAY from retry.ts instead of hardcoding 3 / 2_000. P3-2: Replace 500ms polling loop in processor with whenLifecycleCloseBegins() — a promise-based signal from lifecycle-provenance that resolves when beginLifecycleClose or withLifecycleCloseAction fires for the directory. P3-3: safeRecoveryPolicy accepts an optional delay function. Processor threads it from its Input type. Tests inject a 10ms delay, dropping processor-effect from ~118s to ~21s and retry.test.ts from ~25s to ~11s. --- .../src/session/lifecycle-provenance.ts | 20 ++++++ packages/opencode/src/session/processor.ts | 17 +++-- packages/opencode/src/session/retry.ts | 5 +- .../src/session/run-incident/policy.ts | 12 +++- .../test/session/processor-effect.test.ts | 71 ++++++++++--------- packages/opencode/test/session/retry.test.ts | 34 ++++----- 6 files changed, 94 insertions(+), 65 deletions(-) diff --git a/packages/opencode/src/session/lifecycle-provenance.ts b/packages/opencode/src/session/lifecycle-provenance.ts index 40a7b9d5b..7df1d0af6 100644 --- a/packages/opencode/src/session/lifecycle-provenance.ts +++ b/packages/opencode/src/session/lifecycle-provenance.ts @@ -47,6 +47,7 @@ const activeRunsByDirectory = new Map() const idleWaiters = new Set<{ directories: readonly string[]; resolve: () => void }>() const closingByDirectory = new Map() const closeWaiters = new Set<{ directory: string; resolve: (release: () => void) => void }>() +const closingStartWaiters = new Set<{ directory: string; resolve: () => void }>() export function directoryKey(directory: string): string { const digest = createHash("sha256").update(directory).digest("hex").slice(0, 16) @@ -103,6 +104,7 @@ export async function withLifecycleCloseAction( stack.push(action) activeByDirectory.set(directory, stack) } + notifyClosingStartWaiters(directories) try { return await fn() } finally { @@ -146,6 +148,23 @@ export function isLifecycleClosing(directory: string): boolean { return (closingByDirectory.get(directory) ?? 0) > 0 } +export function whenLifecycleCloseBegins(directory: string): Promise { + if (isLifecycleClosing(directory) || currentLifecycleCloseAction(directory)) { + return Promise.resolve() + } + return new Promise((resolve) => { + closingStartWaiters.add({ directory, resolve }) + }) +} + +function notifyClosingStartWaiters(directories: readonly string[]) { + for (const waiter of [...closingStartWaiters]) { + if (!directories.includes(waiter.directory)) continue + closingStartWaiters.delete(waiter) + waiter.resolve() + } +} + function notifyCloseWaiters() { for (const waiter of [...closeWaiters]) { if (hasLifecycleClose([waiter.directory])) continue @@ -229,6 +248,7 @@ export function beginLifecycleClose(directories: readonly string[]): () => void for (const directory of uniqueDirectories) { closingByDirectory.set(directory, (closingByDirectory.get(directory) ?? 0) + 1) } + notifyClosingStartWaiters(uniqueDirectories) let released = false return () => { if (released) return diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 2ac2dadbe..4575e132f 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -27,7 +27,12 @@ import { InstanceState } from "@/effect/instance-state" import { TurnChange } from "./turn-change" import { LLMTrace } from "./llm-trace" import { RunObservability } from "./run-observability" -import { currentLifecycleCloseAction, isLifecycleClosing, lifecycleCloseActionMeta } from "./lifecycle-provenance" +import { + currentLifecycleCloseAction, + isLifecycleClosing, + lifecycleCloseActionMeta, + whenLifecycleCloseBegins, +} from "./lifecycle-provenance" const log = Log.create({ service: "session.processor" }) const TOOL_CLEANUP_TIMEOUT_MS = 1_000 @@ -98,6 +103,7 @@ type Input = { assistantMessage: MessageV2.Assistant sessionID: SessionID model: Provider.Model + safeRecoveryDelay?: (attempt: number) => number } export interface Interface { @@ -1326,6 +1332,7 @@ export const layer: Layer.Layer< presentation: info.presentation, reason: info.reason, }), + delay: input.safeRecoveryDelay, }), ) @@ -1443,11 +1450,9 @@ export const layer: Layer.Layer< if (beforeRetry.allowed) { automaticStreamRetriesUsed += 1 yield* removeReasoningForAttempt(attemptID) - const lifecycleCloseWatch = Effect.gen(function* () { - while (!isLifecycleClosing(ctx.directory) && !currentLifecycleCloseAction(ctx.directory)) { - yield* Effect.sleep("500 millis") - } - }).pipe(Effect.as("lifecycle_close" as const)) + const lifecycleCloseWatch = Effect.promise(() => whenLifecycleCloseBegins(ctx.directory)).pipe( + Effect.as("lifecycle_close" as const), + ) const backoffResult = yield* Effect.race( safeRecoveryStep(undefined).pipe(Effect.as("scheduled" as const)), lifecycleCloseWatch, diff --git a/packages/opencode/src/session/retry.ts b/packages/opencode/src/session/retry.ts index fc6b9428f..1865aa347 100644 --- a/packages/opencode/src/session/retry.ts +++ b/packages/opencode/src/session/retry.ts @@ -184,13 +184,14 @@ export function safeRecoveryPolicy(opts: { presentation: "recovery" reason: "network_connection_dropped" }) => Effect.Effect + delay?: (attempt: number) => number }) { + const delayFn = opts.delay ?? delay return Schedule.fromStepWithMetadata( Effect.succeed((meta: Schedule.InputMetadata) => { if (meta.attempt > SAFE_RECOVERY_MAX_ATTEMPTS) return Cause.done(meta.attempt) return Effect.gen(function* () { - // Reuse the API-path backoff schedule: 2s -> 4s -> 8s, capped at 30s. - const wait = delay(meta.attempt) + const wait = delayFn(meta.attempt) const now = yield* Clock.currentTimeMillis yield* opts.set({ attempt: meta.attempt, diff --git a/packages/opencode/src/session/run-incident/policy.ts b/packages/opencode/src/session/run-incident/policy.ts index 9c717b3a1..677a1a91d 100644 --- a/packages/opencode/src/session/run-incident/policy.ts +++ b/packages/opencode/src/session/run-incident/policy.ts @@ -1,6 +1,12 @@ import { allowsBeforeProgressRetry as boundaryAllowsBeforeProgressRetry } from "../run-observability/boundary" +import { RETRY_INITIAL_DELAY, SAFE_RECOVERY_MAX_ATTEMPTS } from "../retry" import type { IncidentFacts, RecoveryDecision, TerminalCause } from "./types" +const SAFE_RECOVERY_AUTO_RETRY = { + max_attempts: SAFE_RECOVERY_MAX_ATTEMPTS, + backoff_ms: RETRY_INITIAL_DELAY, +} as const + export function recoveryFor(input: { cause: TerminalCause facts: IncidentFacts @@ -38,7 +44,7 @@ export function recoveryFor(input: { recommendation: "auto_retry", confidence: "high", reason: "no_visible_output_or_tool_execution", - auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, + auto_retry: SAFE_RECOVERY_AUTO_RETRY, } } if (isBeforeFirstProviderProgressCause(input.cause) && beforeProgressBoundaryEvidenceBlocksRetry(terminalFacts)) { @@ -63,7 +69,7 @@ export function recoveryFor(input: { recommendation: "auto_retry", confidence: "high", reason: "reasoning_only_without_final_text_or_tool_activity", - auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, + auto_retry: SAFE_RECOVERY_AUTO_RETRY, } } if (!terminalFacts.side_effect_facts_complete) { @@ -150,7 +156,7 @@ export function recoveryFor(input: { recommendation: "auto_retry", confidence: "medium", reason: "no_visible_output_or_tool_execution", - auto_retry: { max_attempts: 3, backoff_ms: 2_000 }, + auto_retry: SAFE_RECOVERY_AUTO_RETRY, } } return { ...base, recommendation: "unknown", confidence: "low", reason: "unknown" } diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 1dbfc80f6..0c48e20f0 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -32,6 +32,8 @@ import { raw, reply, TestLLMServer } from "../lib/llm-server" void Log.init({ print: false }) +const FAST_SAFE_RECOVERY_DELAY = () => 10 + const summary = Layer.succeed( SessionSummary.Service, SessionSummary.Service.of({ @@ -356,7 +358,7 @@ attemptTimeoutIt.live("reasoning connect watchdog is attempt-scoped for before-p if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") retrySeen.resolve(evt.properties.status) }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -420,6 +422,7 @@ attemptTimeoutIt.live("reasoning connect watchdog is attempt-scoped for before-p const manualParent = yield* user(chat.id, "manual retry starts fresh") const manualMsg = yield* assistant(chat.id, manualParent.id, path.resolve(dir)) const manualHandle = yield* processors.create({ + safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: manualMsg, sessionID: chat.id, model: mdl, @@ -467,7 +470,7 @@ attemptTimeoutIt.live("reasoning first attempt keeps global timeout when active const parent = yield* user(chat.id, "reasoning external boundary timeout policy") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(reasoningRef.providerID, reasoningRef.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -526,7 +529,7 @@ attemptTimeoutIt.live("reasoning first attempt uses fast timeout with unclassifi const parent = yield* user(chat.id, "reasoning unclassified local tool timeout policy") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(reasoningRef.providerID, reasoningRef.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -644,7 +647,7 @@ it.live("session.processor keeps late tool execution diagnostics on the bound to const parent = yield* user(chat.id, "late tool attempt binding") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -779,7 +782,7 @@ it.live("session.processor records late transport failures against the failing s const parent = yield* user(chat.id, "late transport attempt binding") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -862,7 +865,7 @@ it.live("session.processor effect tests capture llm input cleanly", () => const parent = yield* user(chat.id, "hi") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -951,7 +954,7 @@ it.live("session.processor effect tests preserve text start time", () => const parent = yield* user(chat.id, "hi") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1015,7 +1018,7 @@ it.live("session.processor effect tests stop after token overflow requests compa const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const base = yield* provider.getModel(ref.providerID, ref.modelID) const mdl = { ...base, limit: { context: 20, output: 10 } } - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1060,7 +1063,7 @@ it.live("session.processor effect tests flag empty completions", () => const parent = yield* user(chat.id, "empty") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1104,7 +1107,7 @@ it.live("session.processor effect tests capture reasoning from http mock", () => const parent = yield* user(chat.id, "reason") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1154,7 +1157,7 @@ it.live("session.processor effect tests reset reasoning state across retries", ( const parent = yield* user(chat.id, "reason") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1205,7 +1208,7 @@ it.live("session.processor effect tests do not retry unknown json errors", () => const parent = yield* user(chat.id, "json") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1250,7 +1253,7 @@ it.live("session.processor effect tests retry recognized structured json errors" const parent = yield* user(chat.id, "retry json") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1300,7 +1303,7 @@ it.live("retryable API errors write a safe retry notice after the recovery retri const parent = yield* user(chat.id, "retry api twice") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1356,7 +1359,7 @@ it.live("session.processor effect tests publish retry status updates", () => if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") states.push(evt.properties.status.attempt) }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1402,7 +1405,7 @@ it.live("connect timeout before provider progress auto retries once and succeeds const parent = yield* user(chat.id, "auto retry connect timeout") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1489,7 +1492,7 @@ it.live("connect timeout auto retry stops if lifecycle closes during backoff", ( if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") retrySeen.resolve() }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1568,7 +1571,7 @@ it.live("maintenance lifecycle close during backoff interrupts safe recovery", ( if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") retrySeen.resolve() }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1633,7 +1636,7 @@ it.live("connect timeout auto retry records abort if interrupted during backoff" if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") retrySeen.resolve() }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1697,7 +1700,7 @@ it.live("disabled unknown tools do not block safe connect-timeout auto retry", ( const parent = yield* user(chat.id, "disabled unknown tool retry") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1799,7 +1802,7 @@ it.live("reasoning-only retry removes failed reasoning before replaying the assi if (evt.properties.sessionID !== chat.id) return if (evt.properties.status.type === "retry") retrySeen.resolve(evt.properties.status) }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1895,7 +1898,7 @@ it.live("reasoning-only failure with an external-result tool does not auto retry const parent = yield* user(chat.id, "external boundary retry") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -1986,7 +1989,7 @@ it.live("reasoning-only failure with a provider-executed tool does not auto retr const parent = yield* user(chat.id, "provider boundary retry") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2077,7 +2080,7 @@ it.live("reasoning-only retry writes a notice after the safe retries are exhaust const parent = yield* user(chat.id, "reasoning retry fails twice") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2156,7 +2159,7 @@ it.live("retryable stream error after visible output does not replay the assista const parent = yield* user(chat.id, "visible output retry guard") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2214,7 +2217,7 @@ it.live("session.processor effect tests compact on structured context overflow", const parent = yield* user(chat.id, "compact json") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2257,7 +2260,7 @@ it.live("session.processor effect tests mark pending tools as aborted on cleanup const parent = yield* user(chat.id, "tool abort") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2388,7 +2391,7 @@ it.live("session.processor effect tests mark materialized tools as prepared but const parent = yield* user(chat.id, "materialized tool abort") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2488,7 +2491,7 @@ it.live("session.processor effect tests execute Responses args-done-only tool ca const parent = yield* user(chat.id, "responses args done tool") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(copilotResponsesRef.providerID, copilotResponsesRef.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2563,7 +2566,7 @@ it.live("test LLM server preserves cached input tokens when converting chat chun const parent = yield* user(chat.id, "responses cache usage") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(copilotResponsesRef.providerID, copilotResponsesRef.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2628,7 +2631,7 @@ it.live("session.processor effect tests rewrite aborted question tool error to f const parent = yield* user(chat.id, "question abort") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2702,7 +2705,7 @@ it.live("session.processor effect tests record aborted errors and idle state", ( errs.push(evt.properties.error.name) seen.resolve() }) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2769,7 +2772,7 @@ it.live("connect timeout writes a safe retry notice and flips session_status idl const parent = yield* user(chat.id, "bad model") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, @@ -2825,7 +2828,7 @@ it.live("session.processor effect tests mark interruptions aborted without manua const parent = yield* user(chat.id, "interrupt") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const mdl = yield* provider.getModel(ref.providerID, ref.modelID) - const handle = yield* processors.create({ + const handle = yield* processors.create({ safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, assistantMessage: msg, sessionID: chat.id, model: mdl, diff --git a/packages/opencode/test/session/retry.test.ts b/packages/opencode/test/session/retry.test.ts index 2dbe8eab0..0ab38263a 100644 --- a/packages/opencode/test/session/retry.test.ts +++ b/packages/opencode/test/session/retry.test.ts @@ -171,7 +171,11 @@ describe("session.retry.delay", () => { }) }) - test("safe recovery policy uses exponential backoff across the replay budget then stops", async () => { + test("delay() produces the expected exponential backoff values", () => { + expect([SessionRetry.delay(1), SessionRetry.delay(2), SessionRetry.delay(3)]).toEqual([2000, 4000, 8000]) + }) + + test("safe recovery policy uses injected delay across the replay budget then stops", async () => { const statuses: Array<{ attempt: number message: string @@ -180,24 +184,24 @@ describe("session.retry.delay", () => { reason: "network_connection_dropped" }> = [] - const observedWaits: number[] = [] + const requestedWaits: number[] = [] + const fastDelay = (attempt: number) => { + const wait = attempt * 10 + requestedWaits.push(wait) + return wait + } const exit = await Effect.runPromise( Effect.gen(function* () { const step = yield* Schedule.toStepWithMetadata( SessionRetry.safeRecoveryPolicy({ set: (info) => Effect.sync(() => statuses.push(info)), + delay: fastDelay, }), ) - // Exhaust the full budget of automatic replays, capturing the scheduled - // wait for each one. info.next is `now + delay(attempt)`, so subtracting - // the clock reading taken just before the step recovers the backoff. for (let attempt = 0; attempt < SessionRetry.SAFE_RECOVERY_MAX_ATTEMPTS; attempt++) { - const before = yield* Clock.currentTimeMillis yield* step(undefined) - observedWaits.push(statuses[statuses.length - 1]!.next - before) } - // The next step exceeds the budget and must terminate the schedule. return yield* Effect.exit(step(undefined)) }), ) @@ -208,15 +212,7 @@ describe("session.retry.delay", () => { } expect(statuses).toHaveLength(SessionRetry.SAFE_RECOVERY_MAX_ATTEMPTS) expect(statuses.map((status) => status.attempt)).toEqual([1, 2, 3]) - // Backoff reuses the API-path delay(): 2s -> 4s -> 8s. The monotonic clock can - // only add to the gap, so each observed wait is at least the scheduled backoff - // and comfortably under the next step up. - const expectedBackoff = [SessionRetry.delay(1), SessionRetry.delay(2), SessionRetry.delay(3)] - expect(expectedBackoff).toEqual([2000, 4000, 8000]) - observedWaits.forEach((wait, index) => { - expect(wait).toBeGreaterThanOrEqual(expectedBackoff[index]!) - expect(wait).toBeLessThan(expectedBackoff[index]! + 1000) - }) + expect(requestedWaits).toEqual([10, 20, 30]) expect( statuses.every( (status) => @@ -225,9 +221,7 @@ describe("session.retry.delay", () => { status.reason === "network_connection_dropped", ), ).toBe(true) - // This exercises the real schedule, so it waits the full 2s + 4s + 8s backoff. - // Override Bun's 5s default timeout to keep it green on focused runs and CI. - }, 20_000) + }) test("policy stops retrying after the configured max attempts", async () => { const attempts: number[] = [] From 817b3471b786e9bf120ed08d1dcf0b17e669c55d Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 18:18:57 +0800 Subject: [PATCH 6/7] fix(session): clean up lifecycle close waiter when backoff wins the race whenLifecycleCloseBegins now returns { promise, cancel } instead of a bare Promise. The processor wraps it in Effect.callback with a cleanup finalizer that calls cancel() on interrupt, so the waiter is removed from closingStartWaiters when backoff completes normally. Adds closingStartWaiterCount() for test observability and a test that verifies no waiter leaks after a full safe-recovery exhaust cycle without any lifecycle close. --- .../src/session/lifecycle-provenance.ts | 20 ++++++-- packages/opencode/src/session/processor.ts | 8 +-- .../test/session/processor-effect.test.ts | 50 ++++++++++++++++++- 3 files changed, 70 insertions(+), 8 deletions(-) diff --git a/packages/opencode/src/session/lifecycle-provenance.ts b/packages/opencode/src/session/lifecycle-provenance.ts index 7df1d0af6..a59648092 100644 --- a/packages/opencode/src/session/lifecycle-provenance.ts +++ b/packages/opencode/src/session/lifecycle-provenance.ts @@ -148,13 +148,25 @@ export function isLifecycleClosing(directory: string): boolean { return (closingByDirectory.get(directory) ?? 0) > 0 } -export function whenLifecycleCloseBegins(directory: string): Promise { +export function whenLifecycleCloseBegins(directory: string): { promise: Promise; cancel: () => void } { if (isLifecycleClosing(directory) || currentLifecycleCloseAction(directory)) { - return Promise.resolve() + return { promise: Promise.resolve(), cancel: () => {} } } - return new Promise((resolve) => { - closingStartWaiters.add({ directory, resolve }) + const waiter = { directory, resolve: () => {} } + const promise = new Promise((resolve) => { + waiter.resolve = resolve }) + closingStartWaiters.add(waiter) + return { + promise, + cancel: () => { + closingStartWaiters.delete(waiter) + }, + } +} + +export function closingStartWaiterCount(): number { + return closingStartWaiters.size } function notifyClosingStartWaiters(directories: readonly string[]) { diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 4575e132f..b7afb0fa6 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -1450,9 +1450,11 @@ export const layer: Layer.Layer< if (beforeRetry.allowed) { automaticStreamRetriesUsed += 1 yield* removeReasoningForAttempt(attemptID) - const lifecycleCloseWatch = Effect.promise(() => whenLifecycleCloseBegins(ctx.directory)).pipe( - Effect.as("lifecycle_close" as const), - ) + const lifecycleCloseWatch = Effect.callback<"lifecycle_close">((resume) => { + const handle = whenLifecycleCloseBegins(ctx.directory) + handle.promise.then(() => resume(Effect.succeed("lifecycle_close" as const))) + return Effect.sync(() => handle.cancel()) + }) const backoffResult = yield* Effect.race( safeRecoveryStep(undefined).pipe(Effect.as("scheduled" as const)), lifecycleCloseWatch, diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 0c48e20f0..0cdcc2d54 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -18,7 +18,7 @@ import { LLM } from "../../src/session/llm" import { MessageV2 } from "../../src/session/message-v2" import { SessionProcessor } from "../../src/session/processor" import { SessionDiagnostics } from "../../src/session/diagnostics" -import { beginLifecycleClose, createLifecycleCloseAction, withLifecycleCloseAction } from "../../src/session/lifecycle-provenance" +import { beginLifecycleClose, closingStartWaiterCount, createLifecycleCloseAction, withLifecycleCloseAction } from "../../src/session/lifecycle-provenance" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SessionStatus } from "../../src/session/status" import { SessionSummary } from "../../src/session/summary" @@ -1617,6 +1617,54 @@ it.live("maintenance lifecycle close during backoff interrupts safe recovery", ( ), ) +it.live("lifecycle close waiter is cleaned up after normal backoff completion", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + yield* llm.hang + yield* llm.hang + yield* llm.hang + yield* llm.hang + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "waiter cleanup check") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + safeRecoveryDelay: FAST_SAFE_RECOVERY_DELAY, + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const before = closingStartWaiterCount() + yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "waiter cleanup check" }], + tools: {}, + connectTimeoutMs: 20, + streamTimeoutMs: 1_000, + }) + + expect(closingStartWaiterCount()).toBe(before) + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + it.live("connect timeout auto retry records abort if interrupted during backoff", () => provideTmpdirServer( ({ dir, llm }) => From ebd917aee031b1c0efaed0b58104ff1315084234 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 31 May 2026 18:33:54 +0800 Subject: [PATCH 7/7] fix(session): only catch schedule-done cause, re-interrupt on pure interrupts catchCause on the backoff race now checks Cause.hasInterruptsOnly: pure interrupts are re-raised as Effect.interrupt so onInterrupt still fires, while schedule exhaustion maps to "exhausted". --- packages/opencode/src/session/processor.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index b7afb0fa6..624abc003 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -1459,8 +1459,12 @@ export const layer: Layer.Layer< safeRecoveryStep(undefined).pipe(Effect.as("scheduled" as const)), lifecycleCloseWatch, ).pipe( - Effect.catchCause(() => Effect.succeed("exhausted" as const)), Effect.onInterrupt(() => recordProcessInterrupt(attemptID)), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.succeed("exhausted" as const), + ), ) if (backoffResult === "exhausted") { yield* writeSafeRetryFailedNotice(attemptID)