feat(plugin): add session retry hook

packages/core/src/session/model-request.ts

@@ -48,6 +48,7 @@ const declineDefect = (cause: Cause.Cause<Tool.Error>) => {
 export interface Prepared {
   readonly request: LLMRequest
   readonly options: StreamOptions
+  readonly retry: (event: PluginHooks.Domains["session"]["retry"]) => Effect.Effect<void>
   /**
    * One request-scoped execution operation. Unknown and hook-removed calls
    * fail individually through the same seam.
@@ -364,9 +365,11 @@ export const layer = Layer.effect(
         tools
           .execute({ ...input, definitions: hooked })
           .pipe(Effect.catchCauseFilter(declineDefect, (decline) => Effect.fail(decline)))
+      const retry: Prepared["retry"] = (event) => hooks.trigger("session", "retry", event).pipe(Effect.asVoid)
       return {
         request,
         options,
+        retry,
         executeTool,
       }
     })

packages/core/src/session/runner/llm.ts

@@ -1,7 +1,7 @@
 export * as SessionRunnerLLM from "./llm.js"
 
 import { Message } from "@opencode-ai/ai"
-import { Cause, Effect, Exit, FiberMap, Layer, Pull, Schedule } from "effect"
+import { Cause, Effect, Exit, FiberMap, Layer } from "effect"
 import { Database } from "../../database/database.js"
 import { Bus } from "../../bus.js"
 import { InstructionState } from "../instruction-state.js"
@@ -176,7 +176,7 @@ const layer = Layer.effect(
     const runStep = Effect.fn("SessionRunner.runStep")(function* (first: SessionContext.Loaded, step: number) {
       const sessionID = first.session.id
       let assistantMessageID = SessionMessage.ID.create()
-      const retry = yield* Schedule.toStepWithSleep(SessionRunnerRetry.schedule(bus, sessionID))
+      const retry = yield* SessionRunnerRetry.make(bus, sessionID)
       let initial: SessionContext.Loaded | undefined = first
       let recoverOverflow = true
       let recoverContinuation = true
@@ -222,6 +222,15 @@ const layer = Layer.effect(
           agent: loaded.agent.id,
           model: loaded.model,
           prepared,
+          retry: (cause, error, proposed) =>
+            retry.decide({
+              cause,
+              error,
+              agent: loaded.agent.id,
+              model: loaded.model.ref,
+              hook: prepared.retry,
+              retry: proposed,
+            }),
           recoverContinuation,
           recoverOverflow: Effect.suspend(() =>
             recoverOverflow && compaction.enabled()
@@ -232,18 +241,17 @@ const layer = Layer.effect(
         const completed = yield* SessionStep.Outcome.$match(outcome, {
           Completed: (outcome) => Effect.succeed(outcome.needsContinuation),
           Retry: (outcome) =>
-            retry({ cause: outcome.cause, error: outcome.error, assistantMessageID }).pipe(
-              Pull.catchDone(() =>
-                bus
-                  .publish(SessionEvent.Step.Failed, { sessionID, assistantMessageID, error: outcome.error })
-                  .pipe(Effect.andThen(outcome.cause)),
-              ),
-              Effect.asVoid,
-            ),
+            retry.wait({
+              decision: outcome.decision,
+              error: outcome.error,
+              assistantMessageID,
+            }),
           Continue: Effect.fnUntraced(function* (outcome) {
-            yield* retry({ cause: outcome.cause, error: outcome.error, assistantMessageID }).pipe(
-              Pull.catchDone(() => outcome.cause),
-            )
+            yield* retry.wait({
+              decision: outcome.decision,
+              error: outcome.error,
+              assistantMessageID,
+            })
             yield* bus.publish(SessionEvent.Synthetic, { sessionID, text: CONTINUE_AFTER_INCOMPLETE_STREAM })
             assistantMessageID = SessionMessage.ID.create()
           }),

packages/core/src/session/runner/retry.ts

@@ -1,17 +1,29 @@
 export * as SessionRunnerRetry from "./retry.js"
 
 import { AIError } from "@opencode-ai/ai"
+import { Agent } from "@opencode-ai/schema/agent"
+import { Model } from "@opencode-ai/schema/model"
 import { SessionError } from "@opencode-ai/schema/session-error"
-import { Duration, Effect, Schedule } from "effect"
+import { Clock, Duration, Effect, Pull, Schedule } from "effect"
 import { Bus } from "../../bus.js"
+import type { PluginHooks } from "../../plugin/hooks.js"
 import { SessionEvent } from "../event.js"
 import { SessionMessage } from "../message.js"
 import { SessionSchema } from "../schema.js"
 
-export interface Input {
+interface Input {
   readonly cause: AIError
   readonly error: SessionError.Error
-  readonly assistantMessageID: SessionMessage.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly hook: (event: PluginHooks.Domains["session"]["retry"]) => Effect.Effect<void>
+  readonly retry: boolean
+}
+
+export interface Decision {
+  readonly retry: true
+  readonly attempt: number
+  readonly delay: number
 }
 
 export function isRetryable(error: AIError) {
@@ -55,22 +67,58 @@ const retryAfter = (input: Input) => {
   return undefined
 }
 
-export const schedule = (bus: Bus.Interface, sessionID: SessionSchema.ID) =>
-  Schedule.max([Schedule.exponential("2 seconds"), Schedule.recurs(4)]).pipe(
-    Schedule.jittered,
-    Schedule.setInputType<Input>(),
-    Schedule.modifyDelay(({ input, duration: delay }) => {
-      const minimum = retryAfter(input)
-      const duration = minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum))
-      return Effect.succeed(Duration.millis(Math.ceil(Duration.toMillis(duration))))
-    }),
-    Schedule.tap((metadata) =>
-      bus.publish(SessionEvent.RetryScheduled, {
-        sessionID,
-        assistantMessageID: metadata.input.assistantMessageID,
-        attempt: metadata.attempt + 1,
-        at: metadata.now + Duration.toMillis(metadata.duration),
-        error: metadata.input.error,
-      }),
-    ),
-  )
+const schedule = Schedule.max([Schedule.exponential("2 seconds"), Schedule.recurs(4)]).pipe(
+  Schedule.jittered,
+  Schedule.setInputType<Input>(),
+  Schedule.modifyDelay(({ input, duration: delay }) => {
+    const minimum = retryAfter(input)
+    const duration = minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum))
+    return Effect.succeed(Duration.millis(Math.ceil(Duration.toMillis(duration))))
+  }),
+)
+
+export const make = (bus: Bus.Interface, sessionID: SessionSchema.ID) =>
+  Effect.gen(function* () {
+    const step = yield* Schedule.toStep(schedule)
+    let attempt = 1
+    const decide = (input: Input) =>
+      Effect.gen(function* () {
+        const now = yield* Clock.currentTimeMillis
+        const next = yield* step(now, input).pipe(Pull.catchDone(() => Effect.succeed(undefined)))
+        if (!next) return { retry: false as const }
+        const [, duration] = next
+        attempt++
+        const delay = Math.ceil(Duration.toMillis(duration))
+        const event: PluginHooks.Domains["session"]["retry"] = {
+          sessionID,
+          agent: input.agent,
+          model: input.model,
+          error: input.error,
+          attempt,
+          decision: input.retry ? { retry: true, delay } : { retry: false },
+        }
+        yield* input.hook(event)
+        if (!event.decision.retry) return event.decision
+        const normalized =
+          Number.isFinite(event.decision.delay) && event.decision.delay >= 0 ? Math.ceil(event.decision.delay) : delay
+        return { retry: true as const, attempt, delay: normalized }
+      })
+    const wait = (input: {
+      readonly decision: Decision
+      readonly assistantMessageID: SessionMessage.ID
+      readonly error: SessionError.Error
+    }) =>
+      Effect.gen(function* () {
+        const scheduled = yield* Clock.currentTimeMillis
+        yield* bus.publish(SessionEvent.RetryScheduled, {
+          sessionID,
+          assistantMessageID: input.assistantMessageID,
+          attempt: input.decision.attempt,
+          at: scheduled + input.decision.delay,
+          error: input.error,
+        })
+        const remaining = Math.max(0, scheduled + input.decision.delay - (yield* Clock.currentTimeMillis))
+        yield* Effect.sleep(Duration.millis(remaining))
+      })
+    return { decide, wait }
+  })

packages/core/src/session/runner/step.ts

@@ -31,8 +31,11 @@ import { SessionRunnerRetry } from "./retry.js"
 
 export type Outcome = Data.TaggedEnum<{
   Completed: { readonly needsContinuation: boolean }
-  Retry: { readonly cause: AIError; readonly error: SessionError.Error }
-  Continue: { readonly cause: AIError; readonly error: SessionError.Error }
+  Retry: { readonly error: SessionError.Error; readonly decision: SessionRunnerRetry.Decision }
+  Continue: {
+    readonly error: SessionError.Error
+    readonly decision: SessionRunnerRetry.Decision
+  }
   RecoverFull: {}
   Compacted: {}
 }>
@@ -44,6 +47,11 @@ interface Input {
   readonly agent: Agent.ID
   readonly model: SessionRunnerModel.Resolved
   readonly prepared: SessionModelRequest.Prepared
+  readonly retry: (
+    cause: AIError,
+    error: SessionError.Error,
+    retry: boolean,
+  ) => Effect.Effect<{ readonly retry: false } | SessionRunnerRetry.Decision>
   readonly recoverContinuation: boolean
   /** The runner owns compaction policy; the attempt invokes it only before durable output. */
   readonly recoverOverflow: Effect.Effect<boolean>
@@ -161,10 +169,21 @@ export const make = Effect.gen(function* () {
           !recorded.outputStarted
         )
           return Outcome.RecoverFull()
-        if (llmFailure && llmError && SessionRunnerRetry.isRetryable(llmFailure) && !recorded.outputStarted) {
+        const retry =
+          llmFailure && llmError && !isContextOverflowFailure(llmFailure)
+            ? yield* restore(
+                input.retry(
+                  llmFailure,
+                  llmError,
+                  SessionRunnerRetry.isRetryable(llmFailure) ||
+                    (recorded.outputStarted && isInterruptedStream(llmFailure)),
+                ),
+              )
+            : undefined
+        if (llmFailure && llmError && retry?.retry && !recorded.outputStarted) {
           // Retry state projects onto the existing assistant, even before it has produced output.
           yield* publisher.startAssistant()
-          return Outcome.Retry({ cause: llmFailure, error: llmError })
+          return Outcome.Retry({ error: llmError, decision: retry })
         }
         if (llmError) yield* publisher.failAssistant(llmError)
 
@@ -221,20 +240,15 @@ export const make = Effect.gen(function* () {
             })
         }
 
-        // After durable output, recovery continues instead of replaying: the
-        // partial assistant message is already persisted history. Any failure
-        // the pre-output gate would retry is continued here, plus interrupted
-        // streams, whose read failures may carry delivery states the retry
-        // policy rejects for full resends.
         if (
           llmFailure &&
           llmError &&
-          (isInterruptedStream(llmFailure) || SessionRunnerRetry.isRetryable(llmFailure)) &&
+          retry?.retry &&
           record.outputStarted &&
           tools.declines.length === 0 &&
           !tools.interrupted
         )
-          return Outcome.Continue({ cause: llmFailure, error: llmError })
+          return Outcome.Continue({ error: llmError, decision: retry })
 
         if (Exit.isFailure(stream)) return yield* Effect.failCause(stream.cause)
         if (tools.declines.length > 0) return yield* Effect.interrupt

packages/core/test/session-runner.test.ts

@@ -552,6 +552,7 @@ const setup = Effect.gen(function* () {
     admit,
     resume,
     context: session.context(sessionID),
+    hooks,
     messages: session.messages({ sessionID }),
     inbox: session.inbox(sessionID),
     runPrompt: Effect.fnUntraced(function* (text: string) {
@@ -4356,9 +4357,14 @@ describe("SessionRunnerLLM", () => {
     yield* s.admit("Retry transport")
     yield* s.llm.push(Stream.fail(providerUnavailable()))
     yield* s.llm.push(TestLLM.text("Recovered", "retry-success"))
+    const scheduled = yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
+      Stream.filter((event) => event.data.sessionID === sessionID),
+      Stream.runHead,
+      Effect.forkScoped({ startImmediately: true }),
+    )
 
     const run = yield* s.resume.pipe(Effect.forkChild)
-    yield* s.llm.wait(1)
+    yield* Fiber.join(scheduled)
     yield* TestClock.adjust("1599 millis")
     expect(s.requests).toHaveLength(1)
     yield* TestClock.adjust("801 millis")
@@ -4376,6 +4382,81 @@ describe("SessionRunnerLLM", () => {
     expect((yield* s.context).filter((message) => message.type === "assistant")).toHaveLength(1)
   })
 
+  scenario("allows session retry hooks to veto a proposed retry", function* (s) {
+    const failure = providerUnavailable()
+    let observed: PluginHooks.Domains["session"]["retry"] | undefined
+    yield* s.hooks.register("session", "retry", (event) =>
+      Effect.sync(() => {
+        observed = event
+        event.decision = { retry: false }
+      }),
+    )
+    yield* s.llm.push(Stream.fail(failure))
+
+    expect(yield* s.runPrompt("Do not retry transport").pipe(Effect.flip)).toBe(failure)
+    expect(s.requests).toHaveLength(1)
+    expect(observed).toMatchObject({
+      sessionID,
+      agent: "build",
+      model: { providerID: "fake", id: "fake-model" },
+      error: { type: "provider.transport", message: "Provider unavailable" },
+      attempt: 2,
+      decision: { retry: false },
+    })
+    expect(yield* recordedEventTypes(sessionID)).not.toContain("session.retry.scheduled.1")
+  })
+
+  scenario("allows session retry hooks to retry a terminal provider failure", function* (s) {
+    yield* s.hooks.register("session", "retry", (event) =>
+      Effect.sync(() => {
+        expect(event.decision).toEqual({ retry: false })
+        event.decision = { retry: true, delay: 0 }
+      }),
+    )
+    yield* s.admit("Retry invalid request")
+    yield* s.llm.push(Stream.fail(invalidRequest()), TestLLM.text("Recovered", "forced-retry-success"))
+
+    yield* s.resume
+
+    expect(s.requests).toHaveLength(2)
+    expect(yield* s.context).toMatchObject([
+      Expected.user("Retry invalid request"),
+      Expected.assistant({ finish: "stop" }, [Expected.text("Recovered")]),
+    ])
+  })
+
+  scenario("uses the final session retry hook delay", function* (s) {
+    yield* s.hooks.register("session", "retry", (event) =>
+      Effect.sync(() => {
+        event.decision = { retry: true, delay: 10_000 }
+      }),
+    )
+    yield* s.hooks.register("session", "retry", (event) =>
+      Effect.sync(() => {
+        expect(event.decision).toEqual({ retry: true, delay: 10_000 })
+        event.decision = { retry: true, delay: 5_000 }
+      }),
+    )
+    yield* s.admit("Use custom retry delay")
+    yield* s.llm.push(Stream.fail(providerUnavailable()), TestLLM.text("Recovered", "hook-delay-success"))
+    const scheduled = yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
+      Stream.filter((event) => event.data.sessionID === sessionID),
+      Stream.runHead,
+      Effect.forkScoped({ startImmediately: true }),
+    )
+
+    const run = yield* s.resume.pipe(Effect.forkChild)
+    yield* Fiber.join(scheduled)
+    yield* TestClock.adjust("4999 millis")
+    expect(s.requests).toHaveLength(1)
+    yield* TestClock.adjust("1 millis")
+    yield* Fiber.join(run)
+
+    expect(s.requests).toHaveLength(2)
+    const assistant = requireAssistant(yield* s.context)
+    expect(assistant.retry).toBeUndefined()
+  })
+
   scenario("does not start another physical attempt after interruption during retry backoff", function* (s) {
     yield* s.admit("Interrupt retry backoff")
     yield* s.llm.push(Stream.fail(providerUnavailable()), TestLLM.text("Must not run", "unused-retry"))

packages/core/test/session-step.test.ts

@@ -105,6 +105,7 @@ for (const fixture of [
           agent: Agent.defaultID,
           model,
           prepared: {
+            retry: () => Effect.void,
             request: LLM.request({ model: model.model, prompt: "Run one tool", toolChoice: fixture.toolChoice }),
             options: {},
             executeTool: () =>
@@ -113,6 +114,8 @@ for (const fixture of [
                 return { content: "Completed tool" }
               }),
           },
+          retry: (_cause, _error, retry) =>
+            Effect.succeed(retry ? { retry: true, attempt: 2, delay: 0 } : { retry: false }),
           recoverContinuation: true,
           recoverOverflow: Effect.succeed(false),
         })