refactor(core): simplify step recovery guards

packages/core/src/session/context.ts

@@ -2,6 +2,7 @@ export * as SessionContext from "./context.js"
 
 import { Context, Effect, Layer } from "effect"
 import { Agent } from "../agent.js"
+import { Bus } from "../bus.js"
 import { Catalog } from "../catalog.js"
 import { CodeModeInstructions } from "../codemode/instructions.js"
 import { Database } from "../database/database.js"
@@ -20,6 +21,7 @@ import { Tool } from "../tool.js"
 import { AgentNotFoundError } from "./error.js"
 import { SessionHistory } from "./history.js"
 import { InstructionEntry } from "./instruction-entry.js"
+import { InstructionState } from "./instruction-state.js"
 import { SessionMessage } from "./message.js"
 import { SessionModelRequest } from "./model-request.js"
 import { SessionRunnerModel } from "./runner/model.js"
@@ -51,6 +53,10 @@ export interface Loaded {
 export interface Interface {
   /** Selects the Session, agent, instructions, and tools used by subsequent work. */
   readonly select: (sessionID: SessionSchema.ID) => Effect.Effect<Selection, AgentNotFoundError>
+  /** Prepares the instruction baseline before delivery, or refreshes it without delivering input. */
+  readonly preflight: (
+    sessionID: SessionSchema.ID,
+  ) => Effect.Effect<Selection, AgentNotFoundError | Instructions.InitializationBlocked>
   /** Resolves the model and active history for that selection. */
   readonly load: (selection: Selection) => Effect.Effect<Loaded, SessionRunnerModel.Error>
   readonly resolveModel: (
@@ -75,6 +81,7 @@ const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
     const agents = yield* Agent.Service
+    const bus = yield* Bus.Service
     const builtins = yield* InstructionBuiltIns.Service
     const catalog = yield* Catalog.Service
     const db = (yield* Database.Service).db
@@ -157,6 +164,13 @@ const layer = Layer.effect(
       }
     })
 
+    const preflight = Effect.fn("SessionContext.preflight")(function* (sessionID: SessionSchema.ID) {
+      const selected = yield* select(sessionID)
+      // A blocked initial instruction baseline must leave admitted input pending.
+      yield* InstructionState.prepare(db, bus, selected.instructions, sessionID)
+      return selected
+    })
+
     const load = Effect.fn("SessionContext.load")(function* (selection: Selection) {
       const model = yield* resolveModel(selection.session)
       const history = yield* SessionHistory.entriesForRunner(db, selection.session.id, selection.instructions)
@@ -170,7 +184,7 @@ const layer = Layer.effect(
       }
     })
 
-    return Service.of({ select, load, resolveModel, selectTitle, prepare: modelRequests.prepare })
+    return Service.of({ select, preflight, load, resolveModel, selectTitle, prepare: modelRequests.prepare })
   }),
 )
 
@@ -182,6 +196,7 @@ export const node = makeLocationNode({
   layer,
   deps: [
     Agent.node,
+    Bus.node,
     Catalog.node,
     Database.node,
     InstructionBuiltIns.node,

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

@@ -0,0 +1,292 @@
+export * as SessionAttempt from "./attempt.js"
+
+import {
+  AIError,
+  InvalidProviderOutputError,
+  LLMClient,
+  LLMEvent,
+  isContextOverflowFailure,
+  type ProviderErrorEvent,
+  type ToolCall,
+} from "@opencode-ai/ai"
+import { Cause, Data, Effect, Exit, Fiber, Option, Stream } from "effect"
+import { SessionError } from "@opencode-ai/schema/session-error"
+import { Agent } from "../../agent.js"
+import { Bus } from "../../bus.js"
+import { Permission } from "../../permission.js"
+import { Snapshot } from "../../snapshot.js"
+import { Tool } from "../../tool.js"
+import { ToolOutput } from "../../tool-output.js"
+import { QuestionTool } from "../../tool/plugin/question.js"
+import { StepFailedError } from "../error.js"
+import { SessionEvent } from "../event.js"
+import { SessionMessage } from "../message.js"
+import { SessionModelRequest } from "../model-request.js"
+import { SessionSchema } from "../schema.js"
+import { toSessionError } from "../to-session-error.js"
+import { SessionUsage } from "../usage.js"
+import { SessionRunnerModel } from "./model.js"
+import { createLLMEventPublisher } from "./publish-llm-event.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 }
+  RecoverFull: {}
+  Compacted: {}
+}>
+export const Outcome = Data.taggedEnum<Outcome>()
+
+/** Inspection only; the publisher and joined tool exits stay inside the attempt scope. */
+interface Result {
+  readonly outputStarted: boolean
+  readonly overflowBeforeOutput: boolean
+  /** Raw AI stream failure or synthesized unknown-finish failure; the full Exit stays private. */
+  readonly failure: AIError | undefined
+  /** Absent when a provider failure is already recorded or held, even if the stream also failed. */
+  readonly error: SessionError.Error | undefined
+}
+
+type Decision = Exclude<Outcome, { readonly _tag: "Completed" | "Continue" }>
+type Restore = <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
+
+interface Input {
+  readonly sessionID: SessionSchema.ID
+  readonly assistantMessageID: SessionMessage.ID
+  readonly agent: Agent.ID
+  readonly model: SessionRunnerModel.Resolved
+  readonly prepared: SessionModelRequest.Prepared
+}
+
+const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const
+const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const
+const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const
+
+/** Owns one provider invocation, its tools, and publication until the Step chooses a disposition. */
+export const make = Effect.gen(function* () {
+  const bus = yield* Bus.Service
+  const llm = yield* LLMClient.Service
+  const snapshots = yield* Snapshot.Service
+  const toolOutput = yield* ToolOutput.Service
+
+  // Returning no decision accepts the attempt's existing settlement and Cause handling.
+  const use = Effect.fn("SessionAttempt.use")(function* (
+    input: Input,
+    decide: (result: Result, restore: Restore) => Effect.Effect<Decision | undefined>,
+  ) {
+    const startSnapshot = yield* snapshots.capture()
+    const publisher = createLLMEventPublisher(bus, {
+      sessionID: input.sessionID,
+      assistantMessageID: input.assistantMessageID,
+      agent: input.agent,
+      model: input.model.ref,
+      providerMetadataKey: input.model.model.route.providerMetadataKey ?? input.model.model.provider,
+      snapshot: startSnapshot,
+    })
+    const toolRuns: Array<{
+      readonly call: ToolCall
+      readonly fiber: Fiber.Fiber<void, Permission.DeclinedError | QuestionTool.CancelledError>
+    }> = []
+    const interruptTools = Effect.suspend(() => Fiber.interruptAll(toolRuns.map((run) => run.fiber)))
+    const executeTool = (call: ToolCall) => {
+      if (input.prepared.request.toolChoice?.type === "none")
+        return new Tool.Error({ message: "Tools are disabled after the maximum agent steps" })
+      return input.prepared.executeTool({
+        sessionID: input.sessionID,
+        agent: input.agent,
+        messageID: input.assistantMessageID,
+        call,
+        progress: (update) => publisher.progress(call.id, update),
+      })
+    }
+
+    // Provider and tool fibers retain per-source order without a shared writer queue.
+    // A local execution starts only after its Tool.Called publication completes.
+    let overflowFailure: ProviderErrorEvent | undefined
+    // Read to the end, not just the finish event, so the next request can reuse this response.
+    const providerStream = llm.stream(input.prepared.request, input.prepared.options).pipe(
+      Stream.runForEach((event) =>
+        Effect.gen(function* () {
+          if (overflowFailure || publisher.hasProviderError()) return
+          if (
+            LLMEvent.is.providerError(event) &&
+            isContextOverflowFailure(event) &&
+            !publisher.record().outputStarted
+          ) {
+            overflowFailure = event
+            return
+          }
+          yield* publisher.publish(event)
+          if (event.type !== "tool-call" || event.providerExecuted) return
+          toolRuns.push({
+            call: event,
+            fiber: yield* Effect.uninterruptibleMask((restore) =>
+              restore(executeTool(event)).pipe(
+                Effect.flatMap(toolOutput.truncate),
+                Effect.flatMap((outcome) => publisher.toolExecution(event.id, event.name, outcome)),
+                Effect.catchTag("Tool.Error", (error) =>
+                  publisher.failTool(event.id, toSessionError(error), error.metadata).pipe(Effect.asVoid),
+                ),
+              ),
+            ).pipe(Effect.forkScoped),
+          })
+        }),
+      ),
+      Effect.ensuring(publisher.flush()),
+    )
+
+    // Keep the final tool and Step events uninterruptible, even when the work itself is cancelled.
+    return yield* Effect.uninterruptibleMask((restore) =>
+      Effect.gen(function* () {
+        const stream = yield* restore(providerStream).pipe(Effect.exit)
+        const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream))
+        const streamInterrupted = Exit.hasInterrupts(stream)
+        if (!overflowFailure && publisher.hasStarted()) yield* publisher.streamed()
+        if (streamInterrupted) yield* interruptTools
+        const joined = yield* restore(Fiber.awaitAll(toolRuns.map((run) => run.fiber))).pipe(Effect.exit)
+        if (Exit.isFailure(joined)) yield* interruptTools
+        const tools = classifyToolExits(joined, toolRuns)
+
+        const recorded = publisher.record()
+        const unknownFinish =
+          Exit.isSuccess(stream) && recorded.finish?.finish === "unknown"
+            ? new AIError({
+                reason: new InvalidProviderOutputError({
+                  message: "The provider response ended with an unknown finish reason.",
+                  classification: "incomplete-stream",
+                }),
+              })
+            : undefined
+        const llmFailure = streamFailure instanceof AIError ? streamFailure : unknownFinish
+        // A held overflow will record provider failure unless recovery replaces this attempt.
+        // Include that gate before publication so Step policy cannot transparently retry it.
+        const providerFailed = recorded.providerFailed || overflowFailure !== undefined
+        const llmError = llmFailure && !providerFailed ? toSessionError(llmFailure) : undefined
+        const decision = yield* decide(
+          {
+            outputStarted: recorded.outputStarted,
+            overflowBeforeOutput: !recorded.outputStarted && isContextOverflowFailure(overflowFailure ?? streamFailure),
+            failure: llmFailure,
+            error: llmError,
+          },
+          restore,
+        )
+        if (decision?._tag === "Compacted") return decision
+        if (overflowFailure) yield* publisher.publish(overflowFailure)
+        if (decision?._tag === "RecoverFull") return decision
+        if (decision?._tag === "Retry") {
+          // Retry state projects onto the existing assistant, even before it has produced output.
+          yield* publisher.startAssistant()
+          return decision
+        }
+        if (llmError) yield* publisher.failAssistant(llmError)
+
+        for (const decline of tools.declines)
+          yield* publisher.failTool(decline.call.id, {
+            type: "aborted",
+            message:
+              decline.reason._tag === "QuestionTool.CancelledError"
+                ? decline.reason.message
+                : "The user declined this tool call",
+          })
+        const interrupted = tools.declines.length > 0 || streamInterrupted || tools.interrupted
+        const toolFailure = interrupted
+          ? TOOLS_INTERRUPTED
+          : tools.failure !== undefined
+            ? toSessionError(Cause.squash(tools.failure))
+            : providerFailed
+              ? TOOLS_INTERRUPTED
+              : undefined
+        if (toolFailure) yield* publisher.failUnsettledTools(toolFailure)
+        if (interrupted) yield* publisher.failAssistant(STEP_INTERRUPTED)
+
+        // All local fibers have joined; only provider-hosted results can still be missing.
+        if (llmError || (Exit.isSuccess(stream) && !providerFailed)) {
+          const missing = yield* publisher.failUnsettledTools(RESULT_MISSING, "hosted")
+          if (missing && !llmError && !recorded.finish) yield* publisher.failAssistant(RESULT_MISSING)
+        }
+
+        const record = publisher.record()
+        if (record.finish || record.failure) {
+          const snapshot = yield* snapshots.capture()
+          const files =
+            startSnapshot && snapshot
+              ? startSnapshot === snapshot
+                ? []
+                : yield* snapshots
+                    .files({ from: startSnapshot, to: snapshot })
+                    .pipe(Effect.orElseSucceed(() => undefined))
+              : undefined
+          const usage = record.finish
+            ? { cost: SessionUsage.calculateCost(input.model.cost, record.finish.tokens), tokens: record.finish.tokens }
+            : undefined
+          if (record.failure) yield* publisher.publishStepFailure({ ...usage, snapshot, files })
+          if (record.finish && usage && !record.failure)
+            yield* bus.publish(SessionEvent.Step.Ended, {
+              sessionID: input.sessionID,
+              assistantMessageID: yield* publisher.startAssistant(),
+              finish: record.finish.finish,
+              rawFinish: record.finish.rawFinish,
+              providerState: record.finish.providerState,
+              ...usage,
+              snapshot,
+              files,
+            })
+        }
+
+        if (
+          llmFailure &&
+          llmError &&
+          isInterruptedStream(llmFailure) &&
+          record.outputStarted &&
+          tools.declines.length === 0 &&
+          !tools.interrupted
+        )
+          return Outcome.Continue({ cause: llmFailure, error: llmError })
+
+        if (Exit.isFailure(stream)) return yield* Effect.failCause(stream.cause)
+        if (tools.declines.length > 0) return yield* Effect.interrupt
+        if (tools.interrupted && tools.failure) return yield* Effect.failCause(tools.failure)
+        if (tools.interrupted && Exit.isFailure(joined)) return yield* Effect.failCause(joined.cause)
+        if (record.failure) return yield* new StepFailedError({ error: record.failure })
+        return Outcome.Completed({
+          needsContinuation: input.prepared.request.toolChoice?.type !== "none" && record.needsContinuation,
+        })
+      }),
+    )
+  }, Effect.scoped)
+
+  return { use }
+})
+
+const isInterruptedStream = (failure: AIError) => {
+  if (failure.reason._tag === "InvalidProviderOutput") return failure.reason.classification === "incomplete-stream"
+  if (failure.reason._tag === "Transport") return failure.reason.operation === "read"
+  return false
+}
+
+/** Tool.Error settles in each fiber; only user declines remain in the typed error channel. */
+const classifyToolExits = (
+  settled: Exit.Exit<Array<Exit.Exit<void, Permission.DeclinedError | QuestionTool.CancelledError>>>,
+  runs: ReadonlyArray<{ readonly call: ToolCall }>,
+) => {
+  const exits = Exit.isSuccess(settled) ? settled.value : []
+  const declines = exits.flatMap((exit, index) =>
+    Exit.isFailure(exit)
+      ? exit.cause.reasons.flatMap((reason) =>
+          Cause.isFailReason(reason) ? [{ call: runs[index].call, reason: reason.error }] : [],
+        )
+      : [],
+  )
+  const causes = Exit.isFailure(settled)
+    ? [settled.cause]
+    : exits.flatMap((exit) => (Exit.isFailure(exit) ? [exit.cause] : []))
+  const failure = causes
+    .flatMap((cause) => {
+      if (Cause.hasInterrupts(cause)) return []
+      const reasons = cause.reasons.filter(Cause.isDieReason)
+      return reasons.length > 0 ? [Cause.fromReasons<never>(reasons)] : []
+    })
+    .at(0)
+  return { interrupted: causes.some(Cause.hasInterrupts), declines, failure }
+}

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

@@ -1,17 +1,13 @@
 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"
 import { SessionCompaction } from "../compaction.js"
 import { SessionContext } from "../context.js"
 import { SessionEvent } from "../event.js"
 import { SessionInbox } from "../inbox.js"
-import { SessionModelRequest } from "../model-request.js"
 import { SessionModelTransport } from "../model-transport.js"
-import { SessionMessage } from "../message.js"
 import { SessionSchema } from "../schema.js"
 import { SessionStore } from "../store.js"
 import { SessionTitle } from "../title.js"
@@ -19,15 +15,9 @@ import { DrainResult, Service, type Continuation } from "./index.js"
 import { Snapshot } from "../../snapshot.js"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import { llmClient } from "../../effect/app-node-platform.js"
-import { StepFailedError } from "../error.js"
-import { SessionRunnerRetry } from "./retry.js"
 import { SessionStep } from "./step.js"
 import { ToolOutput } from "../../tool-output.js"
 import { PluginSupervisor } from "../../plugin/supervisor.js"
-import { MAX_STEPS_PROMPT } from "./max-steps.js"
-
-const CONTINUE_AFTER_INCOMPLETE_STREAM =
-  "The previous response was interrupted. Continue from where you left off without repeating completed content."
 
 const layer = Layer.effect(
   Service,
@@ -135,7 +125,7 @@ const layer = Layer.effect(
                 return DrainResult.Complete()
               return yield* restore(
                 Effect.gen(function* () {
-                  const selected = yield* prepareContext(sessionID)
+                  const selected = yield* context.preflight(sessionID)
                   const promoted = yield* SessionInbox.promote(
                     db,
                     bus,
@@ -158,107 +148,13 @@ const layer = Layer.effect(
       while (true) {
         const next = yield* advanceToStep()
         if (next._tag !== "Ready") return next
-        continuing = yield* runStep(next.context, step)
+        continuing = yield* steps.run({ first: next.context, number: step })
         step++
         force = false
         entering = false
       }
     })
 
-    const prepareContext = Effect.fn("SessionRunner.prepareContext")(function* (sessionID: SessionSchema.ID) {
-      const selected = yield* context.select(sessionID)
-      // A blocked initial instruction baseline must leave admitted input pending.
-      yield* InstructionState.prepare(db, bus, selected.instructions, sessionID)
-      return selected
-    })
-
-    /** Owns logical Step policy; each attempt owns its streaming, tools, and durable settlement. */
-    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))
-      let initial: SessionContext.Loaded | undefined = first
-      let recoverOverflow = true
-      let recoverContinuation = true
-      while (true) {
-        // Reuse boundary preparation once; retries refresh context without delivering more input.
-        const loaded = initial ?? (yield* prepareContext(sessionID).pipe(Effect.flatMap(context.load)))
-        initial = undefined
-        const compactionInput = {
-          session: loaded.session,
-          messages: loaded.messages,
-          resolved: loaded.model,
-          prepare: context.prepare,
-        }
-        if (compaction.required(compactionInput)) {
-          const compacted = yield* compaction.compact(compactionInput)
-          if (compacted.status !== "completed") return yield* new StepFailedError({ error: compacted.error })
-          assistantMessageID = SessionMessage.ID.create()
-          continue
-        }
-        const stepLimitReached = loaded.agent.info.steps !== undefined && step >= loaded.agent.info.steps
-        const transcript = SessionModelRequest.baseTranscript({
-          agent: loaded.agent.info,
-          model: loaded.model,
-          tools: loaded.tools,
-          initial: loaded.initial,
-          messages: loaded.messages,
-        })
-        const prepared = yield* context.prepare({
-          scope: { session: loaded.session, agentID: loaded.agent.id, model: loaded.model, tools: loaded.tools },
-          transcript: {
-            system: transcript.system,
-            messages: stepLimitReached
-              ? [...transcript.messages, Message.assistant(MAX_STEPS_PROMPT)]
-              : transcript.messages,
-          },
-          // Keep tool definitions on the final Step to preserve the provider's cached prefix.
-          toolChoice: stepLimitReached ? "none" : undefined,
-          webSocket: "session",
-        })
-        const outcome = yield* steps.attempt({
-          sessionID,
-          assistantMessageID,
-          agent: loaded.agent.id,
-          model: loaded.model,
-          prepared,
-          recoverContinuation,
-          recoverOverflow: Effect.suspend(() =>
-            recoverOverflow && compaction.enabled()
-              ? compaction.compact(compactionInput).pipe(Effect.map((result) => result.status === "completed"))
-              : Effect.succeed(false),
-          ),
-        })
-        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,
-            ),
-          Continue: Effect.fnUntraced(function* (outcome) {
-            yield* retry({ cause: outcome.cause, error: outcome.error, assistantMessageID }).pipe(
-              Pull.catchDone(() => outcome.cause),
-            )
-            yield* bus.publish(SessionEvent.Synthetic, { sessionID, text: CONTINUE_AFTER_INCOMPLETE_STREAM })
-            assistantMessageID = SessionMessage.ID.create()
-          }),
-          Compacted: Effect.fnUntraced(function* () {
-            recoverOverflow = false
-            assistantMessageID = SessionMessage.ID.create()
-          }),
-          RecoverFull: Effect.fnUntraced(function* () {
-            recoverContinuation = false
-          }),
-        })
-        if (completed !== undefined) return completed
-      }
-    })
-
     const settleStaleToolCalls = Effect.fn("SessionRunner.settleStaleToolCalls")(function* (
       sessionID: SessionSchema.ID,
     ) {

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

@@ -1,279 +1,137 @@
 export * as SessionStep from "./step.js"
 
-import {
-  AIError,
-  InvalidProviderOutputError,
-  LLMClient,
-  LLMEvent,
-  isContextOverflowFailure,
-  type ProviderErrorEvent,
-  type ToolCall,
-} from "@opencode-ai/ai"
-import { Cause, Data, Effect, Exit, Fiber, Option, Stream } from "effect"
-import { SessionError } from "@opencode-ai/schema/session-error"
-import { Agent } from "../../agent.js"
+import { Message } from "@opencode-ai/ai"
+import { Effect, Pull, Schedule } from "effect"
 import { Bus } from "../../bus.js"
-import { Permission } from "../../permission.js"
-import { Snapshot } from "../../snapshot.js"
-import { Tool } from "../../tool.js"
-import { ToolOutput } from "../../tool-output.js"
-import { QuestionTool } from "../../tool/plugin/question.js"
+import { SessionCompaction } from "../compaction.js"
+import { SessionContext } from "../context.js"
 import { StepFailedError } from "../error.js"
 import { SessionEvent } from "../event.js"
 import { SessionMessage } from "../message.js"
 import { SessionModelRequest } from "../model-request.js"
-import { SessionSchema } from "../schema.js"
-import { toSessionError } from "../to-session-error.js"
-import { SessionUsage } from "../usage.js"
-import { SessionRunnerModel } from "./model.js"
-import { createLLMEventPublisher } from "./publish-llm-event.js"
+import { SessionAttempt } from "./attempt.js"
+import { MAX_STEPS_PROMPT } from "./max-steps.js"
 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 }
-  RecoverFull: {}
-  Compacted: {}
-}>
-export const Outcome = Data.taggedEnum<Outcome>()
+const CONTINUE_AFTER_INCOMPLETE_STREAM =
+  "The previous response was interrupted. Continue from where you left off without repeating completed content."
 
-interface Input {
-  readonly sessionID: SessionSchema.ID
-  readonly assistantMessageID: SessionMessage.ID
-  readonly agent: Agent.ID
-  readonly model: SessionRunnerModel.Resolved
-  readonly prepared: SessionModelRequest.Prepared
-  readonly recoverContinuation: boolean
-  /** The runner owns compaction policy; the attempt invokes it only before durable output. */
-  readonly recoverOverflow: Effect.Effect<boolean>
-}
-
-const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const
-const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const
-const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const
-
-/** Captures Location-scoped dependencies without introducing another service or execution loop. */
+/** A logical Step owns request preparation and recovery, without promoting inbox input. */
 export const make = Effect.gen(function* () {
   const bus = yield* Bus.Service
-  const llm = yield* LLMClient.Service
-  const snapshots = yield* Snapshot.Service
-  const toolOutput = yield* ToolOutput.Service
+  const context = yield* SessionContext.Service
+  const compaction = yield* SessionCompaction.Service
+  const attempts = yield* SessionAttempt.make
 
-  const attempt = Effect.fn("SessionStep.attempt")(function* (input: Input) {
-    const startSnapshot = yield* snapshots.capture()
-    const publisher = createLLMEventPublisher(bus, {
-      sessionID: input.sessionID,
-      assistantMessageID: input.assistantMessageID,
-      agent: input.agent,
-      model: input.model.ref,
-      providerMetadataKey: input.model.model.route.providerMetadataKey ?? input.model.model.provider,
-      snapshot: startSnapshot,
-    })
-    const toolRuns: Array<{
-      readonly call: ToolCall
-      readonly fiber: Fiber.Fiber<void, Permission.DeclinedError | QuestionTool.CancelledError>
-    }> = []
-    const interruptTools = Effect.suspend(() => Fiber.interruptAll(toolRuns.map((run) => run.fiber)))
-    const executeTool = (call: ToolCall) => {
-      if (input.prepared.request.toolChoice?.type === "none")
-        return new Tool.Error({ message: "Tools are disabled after the maximum agent steps" })
-      return input.prepared.executeTool({
-        sessionID: input.sessionID,
-        agent: input.agent,
-        messageID: input.assistantMessageID,
-        call,
-        progress: (update) => publisher.progress(call.id, update),
+  const run = Effect.fn("SessionStep.run")(function* (input: {
+    readonly first: SessionContext.Loaded
+    readonly number: number
+  }) {
+    const sessionID = input.first.session.id
+    let assistantMessageID = SessionMessage.ID.create()
+    const retry = yield* Schedule.toStepWithSleep(SessionRunnerRetry.schedule(bus, sessionID))
+    let initial: SessionContext.Loaded | undefined = input.first
+    let recoverOverflow = true
+    let recoverContinuation = true
+    while (true) {
+      // Reuse boundary preparation once; retries refresh context without delivering more input.
+      const loaded = initial ?? (yield* context.preflight(sessionID).pipe(Effect.flatMap(context.load)))
+      initial = undefined
+      const compactionInput = {
+        session: loaded.session,
+        messages: loaded.messages,
+        resolved: loaded.model,
+        prepare: context.prepare,
+      }
+      if (compaction.required(compactionInput)) {
+        const compacted = yield* compaction.compact(compactionInput)
+        if (compacted.status !== "completed") return yield* new StepFailedError({ error: compacted.error })
+        assistantMessageID = SessionMessage.ID.create()
+        continue
+      }
+      const stepLimitReached = loaded.agent.info.steps !== undefined && input.number >= loaded.agent.info.steps
+      const transcript = SessionModelRequest.baseTranscript({
+        agent: loaded.agent.info,
+        model: loaded.model,
+        tools: loaded.tools,
+        initial: loaded.initial,
+        messages: loaded.messages,
       })
+      const prepared = yield* context.prepare({
+        scope: { session: loaded.session, agentID: loaded.agent.id, model: loaded.model, tools: loaded.tools },
+        transcript: {
+          system: transcript.system,
+          messages: stepLimitReached
+            ? [...transcript.messages, Message.assistant(MAX_STEPS_PROMPT)]
+            : transcript.messages,
+        },
+        // Keep tool definitions on the final Step to preserve the provider's cached prefix.
+        toolChoice: stepLimitReached ? "none" : undefined,
+        webSocket: "session",
+      })
+      const outcome = yield* attempts.use(
+        {
+          sessionID,
+          assistantMessageID,
+          agent: loaded.agent.id,
+          model: loaded.model,
+          prepared,
+        },
+        (result, restore) =>
+          Effect.gen(function* () {
+            if (result.outputStarted) return undefined
+            // The attempt retains its pending terminal while interruptible summarization runs.
+            if (result.overflowBeforeOutput) {
+              // Even skipped recovery must observe pending interruption before publishing the held error.
+              const compacted = yield* restore(
+                recoverOverflow && compaction.enabled()
+                  ? compaction.compact(compactionInput).pipe(Effect.map((result) => result.status === "completed"))
+                  : Effect.succeed(false),
+              )
+              if (compacted) return SessionAttempt.Outcome.Compacted()
+            }
+            if (
+              recoverContinuation &&
+              result.failure?.reason._tag === "Transport" &&
+              (result.failure.reason.recovery === "retry-full" ||
+                result.failure.reason.recovery === "rotate-and-retry-full")
+            )
+              return SessionAttempt.Outcome.RecoverFull()
+            if (result.failure && result.error && SessionRunnerRetry.isRetryable(result.failure))
+              return SessionAttempt.Outcome.Retry({ cause: result.failure, error: result.error })
+            return undefined
+          }),
+      )
+      switch (outcome._tag) {
+        case "Completed":
+          return outcome.needsContinuation
+        case "Retry":
+          yield* 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)),
+            ),
+          )
+          continue
+        case "Continue":
+          // The partial span is already settled; share backoff before committing continuation.
+          yield* retry({ cause: outcome.cause, error: outcome.error, assistantMessageID }).pipe(
+            Pull.catchDone(() => outcome.cause),
+          )
+          yield* bus.publish(SessionEvent.Synthetic, { sessionID, text: CONTINUE_AFTER_INCOMPLETE_STREAM })
+          assistantMessageID = SessionMessage.ID.create()
+          continue
+        case "Compacted":
+          recoverOverflow = false
+          assistantMessageID = SessionMessage.ID.create()
+          continue
+        case "RecoverFull":
+          recoverContinuation = false
+          continue
+      }
     }
+  })
 
-    // Provider and tool fibers retain per-source order without a shared writer queue.
-    // A local execution starts only after its Tool.Called publication completes.
-    let overflowFailure: ProviderErrorEvent | undefined
-    // Read to the end, not just the finish event, so the next request can reuse this response.
-    const providerStream = llm.stream(input.prepared.request, input.prepared.options).pipe(
-      Stream.runForEach((event) =>
-        Effect.gen(function* () {
-          if (overflowFailure || publisher.hasProviderError()) return
-          if (
-            LLMEvent.is.providerError(event) &&
-            isContextOverflowFailure(event) &&
-            !publisher.record().outputStarted
-          ) {
-            overflowFailure = event
-            return
-          }
-          yield* publisher.publish(event)
-          if (event.type !== "tool-call" || event.providerExecuted) return
-          toolRuns.push({
-            call: event,
-            fiber: yield* Effect.uninterruptibleMask((restore) =>
-              restore(executeTool(event)).pipe(
-                Effect.flatMap(toolOutput.truncate),
-                Effect.flatMap((outcome) => publisher.toolExecution(event.id, event.name, outcome)),
-                Effect.catchTag("Tool.Error", (error) =>
-                  publisher.failTool(event.id, toSessionError(error), error.metadata).pipe(Effect.asVoid),
-                ),
-              ),
-            ).pipe(Effect.forkScoped),
-          })
-        }),
-      ),
-      Effect.ensuring(publisher.flush()),
-    )
-
-    // Keep the final tool and Step events uninterruptible, even when the work itself is cancelled.
-    return yield* Effect.uninterruptibleMask((restore) =>
-      Effect.gen(function* () {
-        const stream = yield* restore(providerStream).pipe(Effect.exit)
-        const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream))
-        const streamInterrupted = Exit.hasInterrupts(stream)
-        if (!overflowFailure && publisher.hasStarted()) yield* publisher.streamed()
-        if (streamInterrupted) yield* interruptTools
-        const joined = yield* restore(Fiber.awaitAll(toolRuns.map((run) => run.fiber))).pipe(Effect.exit)
-        if (Exit.isFailure(joined)) yield* interruptTools
-        const tools = classifyToolExits(joined, toolRuns)
-
-        if (
-          !publisher.record().outputStarted &&
-          isContextOverflowFailure(overflowFailure ?? streamFailure) &&
-          (yield* restore(input.recoverOverflow))
-        )
-          return Outcome.Compacted()
-
-        if (overflowFailure) yield* publisher.publish(overflowFailure)
-        const recorded = publisher.record()
-        const unknownFinish =
-          Exit.isSuccess(stream) && recorded.finish?.finish === "unknown"
-            ? new AIError({
-                reason: new InvalidProviderOutputError({
-                  message: "The provider response ended with an unknown finish reason.",
-                  classification: "incomplete-stream",
-                }),
-              })
-            : undefined
-        const llmFailure = streamFailure instanceof AIError ? streamFailure : unknownFinish
-        const llmError = llmFailure && !recorded.providerFailed ? toSessionError(llmFailure) : undefined
-        if (
-          input.recoverContinuation &&
-          llmFailure?.reason._tag === "Transport" &&
-          (llmFailure.reason.recovery === "retry-full" || llmFailure.reason.recovery === "rotate-and-retry-full") &&
-          !recorded.outputStarted
-        )
-          return Outcome.RecoverFull()
-        if (llmFailure && llmError && SessionRunnerRetry.isRetryable(llmFailure) && !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 })
-        }
-        if (llmError) yield* publisher.failAssistant(llmError)
-
-        for (const decline of tools.declines)
-          yield* publisher.failTool(decline.call.id, {
-            type: "aborted",
-            message:
-              decline.reason._tag === "QuestionTool.CancelledError"
-                ? decline.reason.message
-                : "The user declined this tool call",
-          })
-        const interrupted = tools.declines.length > 0 || streamInterrupted || tools.interrupted
-        const toolFailure = interrupted
-          ? TOOLS_INTERRUPTED
-          : tools.failure !== undefined
-            ? toSessionError(Cause.squash(tools.failure))
-            : recorded.providerFailed
-              ? TOOLS_INTERRUPTED
-              : undefined
-        if (toolFailure) yield* publisher.failUnsettledTools(toolFailure)
-        if (interrupted) yield* publisher.failAssistant(STEP_INTERRUPTED)
-
-        // All local fibers have joined; only provider-hosted results can still be missing.
-        if (llmError || (Exit.isSuccess(stream) && !recorded.providerFailed)) {
-          const missing = yield* publisher.failUnsettledTools(RESULT_MISSING, "hosted")
-          if (missing && !llmError && !recorded.finish) yield* publisher.failAssistant(RESULT_MISSING)
-        }
-
-        const record = publisher.record()
-        if (record.finish || record.failure) {
-          const snapshot = yield* snapshots.capture()
-          const files =
-            startSnapshot && snapshot
-              ? startSnapshot === snapshot
-                ? []
-                : yield* snapshots
-                    .files({ from: startSnapshot, to: snapshot })
-                    .pipe(Effect.orElseSucceed(() => undefined))
-              : undefined
-          const usage = record.finish
-            ? { cost: SessionUsage.calculateCost(input.model.cost, record.finish.tokens), tokens: record.finish.tokens }
-            : undefined
-          if (record.failure) yield* publisher.publishStepFailure({ ...usage, snapshot, files })
-          if (record.finish && usage && !record.failure)
-            yield* bus.publish(SessionEvent.Step.Ended, {
-              sessionID: input.sessionID,
-              assistantMessageID: yield* publisher.startAssistant(),
-              finish: record.finish.finish,
-              rawFinish: record.finish.rawFinish,
-              providerState: record.finish.providerState,
-              ...usage,
-              snapshot,
-              files,
-            })
-        }
-
-        if (
-          llmFailure &&
-          llmError &&
-          isInterruptedStream(llmFailure) &&
-          record.outputStarted &&
-          tools.declines.length === 0 &&
-          !tools.interrupted
-        )
-          return Outcome.Continue({ cause: llmFailure, error: llmError })
-
-        if (Exit.isFailure(stream)) return yield* Effect.failCause(stream.cause)
-        if (tools.declines.length > 0) return yield* Effect.interrupt
-        if (tools.interrupted && tools.failure) return yield* Effect.failCause(tools.failure)
-        if (tools.interrupted && Exit.isFailure(joined)) return yield* Effect.failCause(joined.cause)
-        if (record.failure) return yield* new StepFailedError({ error: record.failure })
-        return Outcome.Completed({
-          needsContinuation: input.prepared.request.toolChoice?.type !== "none" && record.needsContinuation,
-        })
-      }),
-    )
-  }, Effect.scoped)
-
-  return { attempt }
+  return { run }
 })
-
-const isInterruptedStream = (failure: AIError) => {
-  if (failure.reason._tag === "InvalidProviderOutput") return failure.reason.classification === "incomplete-stream"
-  if (failure.reason._tag === "Transport") return failure.reason.operation === "read"
-  return false
-}
-
-/** Tool.Error settles in each fiber; only user declines remain in the typed error channel. */
-const classifyToolExits = (
-  settled: Exit.Exit<Array<Exit.Exit<void, Permission.DeclinedError | QuestionTool.CancelledError>>>,
-  runs: ReadonlyArray<{ readonly call: ToolCall }>,
-) => {
-  const exits = Exit.isSuccess(settled) ? settled.value : []
-  const declines = exits.flatMap((exit, index) =>
-    Exit.isFailure(exit)
-      ? exit.cause.reasons.flatMap((reason) =>
-          Cause.isFailReason(reason) ? [{ call: runs[index].call, reason: reason.error }] : [],
-        )
-      : [],
-  )
-  const causes = Exit.isFailure(settled)
-    ? [settled.cause]
-    : exits.flatMap((exit) => (Exit.isFailure(exit) ? [exit.cause] : []))
-  const failure = causes
-    .flatMap((cause) => {
-      if (Cause.hasInterrupts(cause)) return []
-      const reasons = cause.reasons.filter(Cause.isDieReason)
-      return reasons.length > 0 ? [Cause.fromReasons<never>(reasons)] : []
-    })
-    .at(0)
-  return { interrupted: causes.some(Cause.hasInterrupts), declines, failure }
-}

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

@@ -14,7 +14,7 @@ import { Session } from "@opencode-ai/core/session"
 import { SessionMessage } from "@opencode-ai/core/session/message"
 import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
-import { SessionStep } from "@opencode-ai/core/session/runner/step"
+import { SessionAttempt } from "@opencode-ai/core/session/runner/attempt"
 import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
 import { Snapshot } from "@opencode-ai/core/snapshot"
 import { ToolOutput } from "@opencode-ai/core/tool-output"
@@ -49,7 +49,7 @@ for (const fixture of [
       const files = [RelativePath.make("changed.ts")]
       let captures = 0
       let executions = 0
-      const steps = yield* SessionStep.make.pipe(
+      const attempts = yield* SessionAttempt.make.pipe(
         Effect.provide(
           Layer.mock(Snapshot.Service)({
             capture: () => Effect.sync(() => (captures++ === 0 ? start : end)),
@@ -98,30 +98,31 @@ for (const fixture of [
           LLMEvent.toolCall({ id: "call-test", name: "test", input: {} }),
         ),
       )
-      const result = yield* steps
-        .attempt({
-          sessionID,
-          assistantMessageID,
-          agent: Agent.defaultID,
-          model,
-          prepared: {
-            request: LLM.request({ model: model.model, prompt: "Run one tool", toolChoice: fixture.toolChoice }),
-            options: {},
-            executeTool: () =>
-              Effect.sync(() => {
-                executions++
-                return { content: "Completed tool" }
-              }),
+      const result = yield* attempts
+        .use(
+          {
+            sessionID,
+            assistantMessageID,
+            agent: Agent.defaultID,
+            model,
+            prepared: {
+              request: LLM.request({ model: model.model, prompt: "Run one tool", toolChoice: fixture.toolChoice }),
+              options: {},
+              executeTool: () =>
+                Effect.sync(() => {
+                  executions++
+                  return { content: "Completed tool" }
+                }),
+            },
           },
-          recoverContinuation: true,
-          recoverOverflow: Effect.succeed(false),
-        })
+          () => Effect.succeed(undefined),
+        )
         .pipe(Effect.exit)
       expect(Exit.isSuccess(result)).toBe(fixture.finish === "stop")
       expect(executions).toBe(fixture.toolChoice === "none" ? 0 : 1)
       if (Exit.isSuccess(result))
         expect(result.value).toEqual(
-          SessionStep.Outcome.Completed({ needsContinuation: fixture.toolChoice !== "none" }),
+          SessionAttempt.Outcome.Completed({ needsContinuation: fixture.toolChoice !== "none" }),
         )
       expect(yield* llm.requests()).toHaveLength(1)
       expect(captures).toBe(2)

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

@@ -205,6 +205,7 @@ const makeRunnerState = () => {
     systemLoadHook: Effect.void,
     skillBaselines: new Map<Agent.ID, string>(),
     pluginFlushHook: Effect.void,
+    compactionEndedHook: Effect.void,
     authorizations: new Array<Tool.Context>(),
     executions: new Array<string>(),
     closedTransports: new Array<Session.ID>(),
@@ -405,6 +406,25 @@ const layer = Layer.unwrap(
         small: () => Effect.undefined,
       },
     })
+    const compaction = makeLocationNode({
+      service: SessionCompaction.Service,
+      layer: SessionCompaction.layer.pipe(
+        Layer.updateService(Bus.Service, (bus) =>
+          Bus.Service.of({
+            ...bus,
+            publish: (definition, data, options) =>
+              bus
+                .publish(definition, data, options)
+                .pipe(
+                  Effect.tap(() =>
+                    definition.type === SessionEvent.Compaction.Ended.type ? state.compactionEndedHook : Effect.void,
+                  ),
+                ),
+          }),
+        ),
+      ),
+      deps: [Bus.node, LayerNodePlatform.llmClient],
+    })
     const replacements: LayerNode.Replacements = [
       [Snapshot.node, Snapshot.noopLayer],
       [LayerNodePlatform.llmClient, TestLLM.clientLayer],
@@ -418,6 +438,7 @@ const layer = Layer.unwrap(
       [Config.node, config],
       [PluginSupervisor.node, pluginSupervisor],
       [SessionModelTransport.node, modelTransport],
+      [SessionCompaction.node, compaction],
     ]
     const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
       ...replacements,
@@ -2501,6 +2522,29 @@ describe("SessionRunnerLLM", () => {
     )
   })
 
+  scenario("does not publish a held overflow when interrupted with automatic compaction disabled", function* (s) {
+    const compaction = yield* SessionCompaction.Service
+    yield* compaction.transform((draft) => draft.configure({ auto: false }))
+    const tail = yield* Deferred.make<void>()
+    yield* s.admit("Interrupt held overflow")
+    yield* s.llm.push(
+      Stream.concat(
+        Stream.make(LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })),
+        Stream.fromEffect(Deferred.succeed(tail, undefined)).pipe(Stream.flatMap(() => Stream.never)),
+      ),
+    )
+    const run = yield* s.resume.pipe(Effect.forkChild)
+    yield* Deferred.await(tail)
+
+    yield* s.session.interrupt(sessionID)
+
+    const exit = yield* Fiber.await(run)
+    expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+    expect(s.requests).toHaveLength(1)
+    expect(yield* s.context).toMatchObject([Expected.user("Interrupt held overflow")])
+    expect((yield* recordedEventTypes(sessionID)).filter((type) => type.startsWith("session.step."))).toEqual([])
+  })
+
   scenario("recovers from provider context overflow without a configured context limit", function* (s) {
     yield* setupOverflowRecovery(s)
     s.currentModel = model
@@ -2631,6 +2675,39 @@ describe("SessionRunnerLLM", () => {
     )
   })
 
+  scenario("interrupts after overflow compaction commits before recovery hands off", function* (s) {
+    yield* setupOverflowRecovery(s)
+    const committed = yield* Deferred.make<void>()
+    // The real Bus publication has returned, but the recovery outcome has not.
+    s.compactionEndedHook = Deferred.succeed(committed, undefined).pipe(Effect.andThen(Effect.never))
+    yield* s.llm.push(
+      [
+        LLMEvent.stepStart({ index: 0 }),
+        LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" }),
+      ],
+      TestLLM.text("Committed overflow summary", "text-summary"),
+      TestLLM.text("Must not retry", "text-unexpected-retry"),
+    )
+    yield* s.admit("Continue")
+    const run = yield* s.resume.pipe(Effect.forkChild)
+    yield* Deferred.await(committed)
+    const assistant = requireAssistant(yield* s.messages)
+
+    yield* s.session.interrupt(sessionID)
+
+    const exit = yield* Fiber.await(run)
+    expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+    expect(s.requests).toHaveLength(2)
+    expect(yield* s.context).toMatchObject([
+      { type: "compaction", status: "completed", reason: "auto", summary: "Committed overflow summary" },
+    ])
+    expect(yield* recordedStepSettlementTypes(sessionID, assistant.id)).toEqual(["session.step.started.1"])
+    expect((yield* recordedEventTypes(sessionID)).filter((type) => type.startsWith("session.compaction."))).toEqual([
+      "session.compaction.started.1",
+      "session.compaction.ended.1",
+    ])
+  })
+
   scenario("uses epoch values after compaction while a source is unavailable", function* (s) {
     yield* s.runPrompt("First")
     s.systemBaseline = "Changed context"
@@ -3626,6 +3703,32 @@ describe("SessionRunnerLLM", () => {
     expect(userTexts(s.requests[0])).toEqual(["Recover promoted input"])
   })
 
+  scenario("does not execute a local tool when durable call projection rolls back", function* (s) {
+    const defect = new Error("Tool.Called projection failed")
+    yield* s.bus.project(SessionEvent.Tool.Called, () =>
+      Effect.gen(function* () {
+        // The production projector has updated this transaction; the failure must roll it back.
+        expect(requireAssistant(yield* s.messages.pipe(Effect.orDie)).content).toMatchObject([
+          { type: "tool", id: "call-rollback", state: { status: "running", input: { text: "Must not execute" } } },
+        ])
+        return yield* Effect.die(defect)
+      }),
+    )
+    yield* s.admit("Call echo")
+    yield* s.llm.push(TestLLM.tool("call-rollback", "echo", { text: "Must not execute" }))
+
+    expect(yield* s.resume.pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
+
+    expect(s.requests).toHaveLength(1)
+    expect(s.executions).toEqual([])
+    expect(s.authorizations).toEqual([])
+    expect(requireAssistant(yield* s.messages).content).toMatchObject([
+      { type: "tool", id: "call-rollback", state: { status: "streaming" } },
+    ])
+    expect(yield* recordedEventTypes(sessionID)).not.toContain("session.tool.called.1")
+    expect(yield* recordedEventTypes(sessionID)).not.toContain("session.tool.success.2")
+  })
+
   scenario("does not strand a committed promotion when a post-commit listener defects", function* (s) {
     yield* s.bus.listen((event) =>
       event.type === SessionEvent.InboxDelivered.type ? Effect.die("fail after prompt promotion commits") : Effect.void,