refactor(session): route safe recovery through retry policy

packages/opencode/src/session/processor.ts

@@ -1,4 +1,4 @@
-import { Cause, Deferred, Effect, Layer, Context, Scope } from "effect"
+import { Cause, Deferred, Effect, Layer, Context, Scope, Schedule } from "effect"
 import * as Stream from "effect/Stream"
 import { Bus } from "@/bus"
 import { Config } from "@/config"
@@ -31,7 +31,6 @@ import { currentLifecycleCloseAction, lifecycleCloseActionMeta } from "./lifecyc
 
 const log = Log.create({ service: "session.processor" })
 const TOOL_CLEANUP_TIMEOUT_MS = 1_000
-const SAFE_RECOVERY_AUTO_RETRY_BACKOFF_MS = 1_000
 export const REASONING_FIRST_ATTEMPT_CONNECT_TIMEOUT_MS = 60_000
 export const REASONING_SAFE_RETRY_CONNECT_TIMEOUT_MS = 120_000
 const LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE = "The run was interrupted by a local lifecycle close."
@@ -1301,6 +1300,20 @@ export const layer: Layer.Layer<
           safeRetryNoticeWritten = true
         })
 
+        const safeRecoveryStep = yield* Schedule.toStepWithMetadata(
+          SessionRetry.safeRecoveryPolicy({
+            set: (info) =>
+              status.set(ctx.sessionID, {
+                type: "retry",
+                attempt: info.attempt,
+                message: info.message,
+                next: info.next,
+                presentation: info.presentation,
+                reason: info.reason,
+              }),
+          }),
+        )
+
         const runAttempt = Effect.fn("SessionProcessor.runAttempt")(function* () {
           ctx.currentText = undefined
           ctx.reasoningMap = {}
@@ -1398,22 +1411,15 @@ export const layer: Layer.Layer<
               if (beforeRetry.allowed) {
                 automaticStreamRetriesUsed += 1
                 yield* removeReasoningForAttempt(attemptID)
-                const next = Date.now() + SAFE_RECOVERY_AUTO_RETRY_BACKOFF_MS
-                yield* status.set(ctx.sessionID, {
-                  type: "retry",
-                  attempt: ctx.attemptCount,
-                  message:
-                    retryDecision.presentation === "safe_recovery"
-                      ? ""
-                      : (retrySignal.message ?? "Retrying interrupted stream"),
-                  next,
-                  ...(retryDecision.presentation === "safe_recovery"
-                    ? { presentation: "safe_recovery" as const, reason: "network_connection_dropped" as const }
-                    : {}),
-                })
-                yield* Effect.sleep(`${SAFE_RECOVERY_AUTO_RETRY_BACKOFF_MS} millis`).pipe(
+                const safeRecoveryScheduled = yield* safeRecoveryStep(undefined).pipe(
+                  Effect.as(true),
+                  Effect.catchCause(() => Effect.succeed(false)),
                   Effect.onInterrupt(() => recordProcessInterrupt(attemptID)),
                 )
+                if (!safeRecoveryScheduled) {
+                  yield* writeSafeRetryFailedNotice(attemptID)
+                  break
+                }
                 const afterRetry = yield* retryStillAllowed("after_backoff")
                 if (afterRetry.allowed) {
                   ctx.runTrace.recordAutoRetryAttempted({

packages/opencode/src/session/retry.ts

@@ -15,6 +15,8 @@ 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
 
 function cap(ms: number) {
   return Math.min(ms, RETRY_MAX_DELAY)
@@ -175,4 +177,31 @@ export function policy(opts: {
   )
 }
 
+export function safeRecoveryPolicy(opts: {
+  set: (input: {
+    attempt: number
+    message: string
+    next: number
+    presentation: "safe_recovery"
+    reason: "network_connection_dropped"
+  }) => Effect.Effect<void>
+}) {
+  return Schedule.fromStepWithMetadata(
+    Effect.succeed((meta: Schedule.InputMetadata<unknown>) => {
+      if (meta.attempt > SAFE_RECOVERY_MAX_ATTEMPTS) return Cause.done(meta.attempt)
+      return Effect.gen(function* () {
+        const now = yield* Clock.currentTimeMillis
+        yield* opts.set({
+          attempt: meta.attempt,
+          message: "",
+          next: now + SAFE_RECOVERY_REPLAY_DELAY,
+          presentation: "safe_recovery",
+          reason: "network_connection_dropped",
+        })
+        return [meta.attempt, Duration.millis(SAFE_RECOVERY_REPLAY_DELAY)] as [number, Duration.Duration]
+      })
+    }),
+  )
+}
+
 export * as SessionRetry from "./retry"

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, Schedule } from "effect"
+import { 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"
@@ -128,6 +128,83 @@ describe("session.retry.delay", () => {
     })
   })
 
+  test("safe recovery policy emits lightweight retry presentation with separate attempt metadata", async () => {
+    await using tmp = await tmpdir()
+    await Instance.provide({
+      directory: tmp.path,
+      fn: async () => {
+        const sessionID = SessionID.make("session-safe-recovery-retry-test")
+
+        await Effect.runPromise(
+          Effect.gen(function* () {
+            const step = yield* Schedule.toStepWithMetadata(
+              SessionRetry.safeRecoveryPolicy({
+                set: (info) =>
+                  Effect.promise(() =>
+                    AppRuntime.runPromise(
+                      SessionStatus.Service.use((svc) =>
+                        svc.set(sessionID, {
+                          type: "retry",
+                          attempt: info.attempt,
+                          message: info.message,
+                          next: info.next,
+                          presentation: info.presentation,
+                          reason: info.reason,
+                        }),
+                      ),
+                    ),
+                  ),
+              }),
+            )
+            yield* step(undefined)
+          }),
+        )
+
+        expect(await AppRuntime.runPromise(SessionStatus.Service.use((svc) => svc.get(sessionID)))).toMatchObject({
+          type: "retry",
+          attempt: 1,
+          message: "",
+          presentation: "safe_recovery",
+          reason: "network_connection_dropped",
+        })
+      },
+    })
+  })
+
+  test("safe recovery policy stops after the one replay budget is exhausted", async () => {
+    const statuses: Array<{
+      attempt: number
+      message: string
+      next: number
+      presentation: "safe_recovery"
+      reason: "network_connection_dropped"
+    }> = []
+
+    const exit = await Effect.runPromise(
+      Effect.gen(function* () {
+        const step = yield* Schedule.toStepWithMetadata(
+          SessionRetry.safeRecoveryPolicy({
+            set: (info) => Effect.sync(() => statuses.push(info)),
+          }),
+        )
+        yield* step(undefined)
+        return yield* Effect.exit(step(undefined))
+      }),
+    )
+
+    expect(Exit.isFailure(exit)).toBe(true)
+    if (Exit.isFailure(exit)) {
+      expect(Pull.isDoneCause(exit.cause)).toBe(true)
+    }
+    expect(statuses).toHaveLength(1)
+    expect(statuses[0]).toMatchObject({
+      attempt: 1,
+      message: "",
+      presentation: "safe_recovery",
+      reason: "network_connection_dropped",
+    })
+  })
+
   test("policy stops retrying after the configured max attempts", async () => {
     const attempts: number[] = []
     let runs = 0