refactor(core): simplify v2 prompt lifecycle and execution coordination (#35047)

packages/core/src/session/execution/local.ts

@@ -1,6 +1,8 @@
-import { Cause, Effect, Layer } from "effect"
+import { Cause, DateTime, Effect, Exit, Layer } from "effect"
+import { EventV2 } from "../../event"
 import { LocationServiceMap } from "../../location-service-map"
 import { makeGlobalNode } from "../../effect/app-node"
+import { SessionEvent } from "../event"
 import { SessionRunCoordinator } from "../run-coordinator"
 import { SessionRunner } from "../runner"
 import { SessionSchema } from "../schema"
@@ -13,6 +15,7 @@ const layer = Layer.effect(
   Effect.gen(function* () {
     const store = yield* SessionStore.Service
     const locations = yield* LocationServiceMap.Service
+    const events = yield* EventV2.Service
     const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError>({
       drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) {
         const session = yield* store.get(sessionID)
@@ -26,6 +29,24 @@ const layer = Layer.effect(
           ),
         )
       }),
+      // One ExecutionSettled per execution (busy period), covering every coalesced drain.
+      settled: (sessionID, exit) =>
+        Effect.gen(function* () {
+          const failure =
+            Exit.isFailure(exit) && !Cause.hasInterrupts(exit.cause) ? Cause.squash(exit.cause) : undefined
+          yield* events.publish(SessionEvent.ExecutionSettled, {
+            sessionID,
+            timestamp: yield* DateTime.now,
+            outcome: Exit.isSuccess(exit) ? "success" : Cause.hasInterrupts(exit.cause) ? "interrupted" : "failure",
+            error:
+              failure !== undefined
+                ? { type: "unknown", message: failure instanceof Error ? failure.message : String(failure) }
+                : undefined,
+          })
+        }).pipe(
+          Effect.catchCause(() => Effect.void),
+          Effect.asVoid,
+        ),
     })
 
     return SessionExecution.Service.of({
@@ -41,7 +62,7 @@ const layer = Layer.effect(
 export const node = makeGlobalNode({
   service: SessionExecution.Service,
   layer,
-  deps: [SessionStore.node, LocationServiceMap.node],
+  deps: [SessionStore.node, LocationServiceMap.node, EventV2.node],
 })
 
 export * as SessionExecutionLocal from "./local"

packages/core/src/session/input.ts

@@ -1,6 +1,6 @@
 export * as SessionInput from "./input"
 
-import { and, asc, eq, isNull, lte } from "drizzle-orm"
+import { and, asc, eq, isNull } from "drizzle-orm"
 import { DateTime, Effect, Schema } from "effect"
 import { Admitted, Delivery } from "@opencode-ai/schema/session-input"
 import type { Database } from "../database/database"
@@ -145,26 +145,11 @@ export const projectPrompted = Effect.fn("SessionInput.projectPrompted")(functio
     return
   }
 
+  // Every Prompted event is published from an admitted inbox row, so a missing or
+  // divergent row on replay is an invariant violation.
   const stored = yield* find(db, input.id)
-  if (stored) {
-    if (!matchesProjection(stored, input) || stored.promotedSeq !== input.promotedSeq)
-      return yield* Effect.die(new LifecycleConflict({ id: input.id }))
-    return
-  }
-
-  yield* db
-    .insert(SessionInputTable)
-    .values({
-      id: input.id,
-      session_id: input.sessionID,
-      prompt: encodePrompt(input.prompt),
-      delivery: input.delivery,
-      admitted_seq: input.promotedSeq,
-      promoted_seq: input.promotedSeq,
-      time_created: DateTime.toEpochMillis(input.timeCreated),
-    })
-    .run()
-    .pipe(Effect.orDie)
+  if (!stored || !matchesProjection(stored, input) || stored.promotedSeq !== input.promotedSeq)
+    return yield* Effect.die(new LifecycleConflict({ id: input.id }))
 })
 
 export const hasPending = Effect.fn("SessionInput.hasPending")(function* (
@@ -195,9 +180,8 @@ export const equivalent = (
     readonly prompt: Prompt
     readonly delivery: Delivery
   },
-) => input.delivery === expected.delivery && matchesPrompt(input, expected)
-
-const matchesPrompt = (input: Admitted, expected: { readonly sessionID: SessionSchema.ID; readonly prompt: Prompt }) =>
+) =>
+  input.delivery === expected.delivery &&
   input.sessionID === expected.sessionID &&
   JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt))
 
@@ -246,7 +230,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* (
   db: DatabaseService,
   events: EventV2.Interface,
   sessionID: SessionSchema.ID,
-  cutoff: number,
 ) {
   const rows = yield* db
     .select()
@@ -256,7 +239,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* (
         eq(SessionInputTable.session_id, sessionID),
         isNull(SessionInputTable.promoted_seq),
         eq(SessionInputTable.delivery, "steer"),
-        lte(SessionInputTable.admitted_seq, cutoff),
       ),
     )
     .orderBy(asc(SessionInputTable.admitted_seq))

packages/core/src/session/run-coordinator.ts

@@ -6,111 +6,117 @@ import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
 export interface Coordinator<Key, E> {
   /** Snapshots keys with an execution owned by this coordinator. */
   readonly active: Effect.Effect<ReadonlySet<Key>>
-  /** Starts execution while idle or joins the active execution. */
+  /** Starts an execution while idle or joins the active execution. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Registers one coalesced follow-up after newly recorded work. */
   readonly wake: (key: Key) => Effect.Effect<void>
-  /** Stops active execution and waits for its cleanup. */
+  /** Stops the active execution and waits for its cleanup. */
   readonly interrupt: (key: Key) => Effect.Effect<void>
   /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
   readonly awaitIdle: (key: Key) => Effect.Effect<void>
 }
 
-type Entry<E> = {
+/**
+ * One execution is a busy period for one key: one fiber that drains from the first wake
+ * until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
+ * execution rings it, and the execution loop drains again instead of ending. The doorbell
+ * closes the gap between a drain's last eligibility check and the idle transition, since
+ * those cannot be one atomic step. `done` resolves joiners with this execution's exit.
+ */
+type Execution<E> = {
   readonly done: Deferred.Deferred<void, E>
-  owner?: Fiber.Fiber<void, never>
+  owner?: Fiber.Fiber<void>
   pendingWake: boolean
   stopping: boolean
 }
 
 export const make = <Key, E>(options: {
   readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
+  /**
+   * Runs in the execution fiber for every exit, including interruption, after the final
+   * drain and before the execution settles (waiters resolve after it completes).
+   */
+  readonly settled?: (key: Key, exit: Exit.Exit<void, E>) => Effect.Effect<void>
 }): Effect.Effect<Coordinator<Key, E>, never, Scope.Scope> =>
   Effect.gen(function* () {
-    const active = new Map<Key, Entry<E>>()
+    const executions = new Map<Key, Execution<E>>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
 
-    const makeEntry = (): Entry<E> => ({
-      done: Deferred.makeUnsafe<void, E>(),
-      pendingWake: false,
-      stopping: false,
-    })
+    const loop = (key: Key, execution: Execution<E>, force: boolean): Effect.Effect<void, E> =>
+      Effect.suspend(() => options.drain(key, force)).pipe(
+        Effect.flatMap(() =>
+          Effect.suspend(() => {
+            if (execution.stopping || !execution.pendingWake) return Effect.void
+            execution.pendingWake = false
+            // Trampoline so drains that complete synchronously cannot grow the stack.
+            return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
+          }),
+        ),
+      )
 
-    const start = (key: Key, entry: Entry<E>, force: boolean, successor = false) => {
-      const ready = Deferred.makeUnsafe<void>()
-      const owner = fork(
-        (successor ? Effect.yieldNow : Deferred.await(ready)).pipe(
-          Effect.andThen(Effect.suspend(() => options.drain(key, force))),
-          Effect.onExit((exit) => Effect.sync(() => settle(key, entry, exit))),
+    const start = (key: Key, force: boolean) => {
+      const execution: Execution<E> = { done: Deferred.makeUnsafe<void, E>(), pendingWake: false, stopping: false }
+      executions.set(key, execution)
+      // The leading yield lets `owner` be assigned before the drain can settle, and keeps
+      // failing self-waking executions from growing the stack across successor starts.
+      // Drains start one tick after wake; callers observe progress through events or run.
+      execution.owner = fork(
+        Effect.yieldNow.pipe(
+          Effect.andThen(loop(key, execution, force)),
+          Effect.onExit((exit) => options.settled?.(key, exit) ?? Effect.void),
+          Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
           Effect.exit,
           Effect.asVoid,
         ),
       )
-      entry.owner = owner
-      if (!successor) Deferred.doneUnsafe(ready, Effect.void)
+      return execution
     }
 
-    const settle = (key: Key, entry: Entry<E>, exit: Exit.Exit<void, E>) => {
-      if (Exit.isSuccess(exit) && !entry.stopping && entry.pendingWake) {
-        entry.pendingWake = false
-        start(key, entry, false, true)
-        return
-      }
-
-      const successor = entry.pendingWake ? makeEntry() : undefined
-      if (successor === undefined) active.delete(key)
-      else {
-        active.set(key, successor)
-        start(key, successor, false, true)
-      }
-      Deferred.doneUnsafe(entry.done, exit)
+    // A doorbell that survives the execution loop (rung after the loop decided to end, or
+    // during failure or interruption cleanup) starts a fresh execution for the remaining work.
+    const settle = (key: Key, execution: Execution<E>, exit: Exit.Exit<void, E>) => {
+      if (execution.pendingWake) start(key, false)
+      else executions.delete(key)
+      Deferred.doneUnsafe(execution.done, exit)
     }
 
     const run = (key: Key): Effect.Effect<void, E> =>
       Effect.uninterruptibleMask((restore) => {
-        const entry = active.get(key)
-        if (entry !== undefined) {
-          if (entry.stopping) return restore(Deferred.await(entry.done).pipe(Effect.andThen(run(key))))
-          return restore(Deferred.await(entry.done))
+        const execution = executions.get(key)
+        if (execution !== undefined) {
+          if (execution.stopping) return restore(Deferred.await(execution.done).pipe(Effect.andThen(run(key))))
+          return restore(Deferred.await(execution.done))
         }
-
-        const next = makeEntry()
-        active.set(key, next)
-        start(key, next, true)
-        return restore(Deferred.await(next.done))
+        return restore(Deferred.await(start(key, true).done))
       })
 
     const wake = (key: Key) =>
       Effect.sync(() => {
-        const entry = active.get(key)
-        if (entry !== undefined) {
-          entry.pendingWake = true
+        const execution = executions.get(key)
+        if (execution !== undefined) {
+          execution.pendingWake = true
           return
         }
-
-        const next = makeEntry()
-        active.set(key, next)
-        start(key, next, false)
+        start(key, false)
       })
 
     const interrupt = (key: Key): Effect.Effect<void> =>
       Effect.suspend(() => {
-        const entry = active.get(key)
-        if (entry?.owner === undefined) return Effect.void
-        entry.stopping = true
-        entry.pendingWake = false
-        return Fiber.interrupt(entry.owner)
+        const execution = executions.get(key)
+        if (execution?.owner === undefined) return Effect.void
+        execution.stopping = true
+        execution.pendingWake = false
+        return Fiber.interrupt(execution.owner)
       })
 
-    // Each successful drain reuses its entry.done across coalesced wakes, so one await
-    // already spans steered and queued continuation. Re-check after it settles to cover a
-    // fresh wake (or a failure/stopping successor) that installs a new entry.
+    // One execution's `done` already spans coalesced continuations; re-check after it
+    // settles to cover a successor execution started by a late doorbell.
     const awaitIdle = (key: Key): Effect.Effect<void> =>
       Effect.suspend(() => {
-        const entry = active.get(key)
-        if (entry === undefined) return Effect.void
-        return Deferred.await(entry.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
+        const execution = executions.get(key)
+        if (execution === undefined) return Effect.void
+        return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
       })
 
-    return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt, awaitIdle }
+    return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
   })

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

@@ -34,7 +34,7 @@ import { SessionInput } from "../input"
 import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { SessionTitle } from "../title"
-import { type RunError, Service } from "./index"
+import { Service } from "./index"
 import { SessionRunnerModel } from "./model"
 import { createLLMEventPublisher } from "./publish-llm-event"
 import { toLLMMessages } from "./to-llm-message"
@@ -157,22 +157,6 @@ const layer = Layer.effect(
     const isQuestionRejected = (cause: Cause.Cause<unknown>) =>
       cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionTool.RejectedError)
 
-    type TurnTransition =
-      // Automatic compaction completed; rebuild the request from compacted history.
-      | { readonly _tag: "ContinueAfterCompaction"; readonly step: number }
-      // Overflow compaction completed; rebuild once through the path without overflow recovery.
-      | { readonly _tag: "ContinueAfterOverflowCompaction"; readonly step: number }
-
-    class TurnTransitionError extends Error {
-      constructor(readonly transition: TurnTransition) {
-        super()
-      }
-    }
-
-    const continueAfterCompaction = (step: number) => new TurnTransitionError({ _tag: "ContinueAfterCompaction", step })
-    const continueAfterOverflowCompaction = (step: number) =>
-      new TurnTransitionError({ _tag: "ContinueAfterOverflowCompaction", step })
-
     const loadSystemContext = (agent: AgentV2.Selection, sessionID: SessionSchema.ID) =>
       Effect.all(
         [
@@ -208,12 +192,11 @@ const layer = Layer.effect(
       let needsContinuation = false
       let currentStep = step
       if (promotion) {
-        const cutoff = yield* EventV2.latestSequence(db, session.id)
         let promoted = 0
-        if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
+        if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id)
         if (promotion === "queue") {
           promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id))
-          promoted += yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
+          promoted += yield* SessionInput.promoteSteers(db, events, session.id)
         }
         if (promoted > 0) currentStep = 1
       }
@@ -239,8 +222,9 @@ const layer = Layer.effect(
         tools: toolMaterialization?.definitions ?? [],
         toolChoice: isLastStep ? "none" : undefined,
       })
+      // Automatic compaction completed; rebuild the request from compacted history.
       if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request }))
-        return yield* Effect.die(continueAfterCompaction(currentStep))
+        return { _tag: "RestartAfterCompaction", step: currentStep } as const
       const startSnapshot = yield* snapshots.capture()
       const publisher = createLLMEventPublisher(events, {
         sessionID: session.id,
@@ -303,44 +287,71 @@ const layer = Layer.effect(
 
       return yield* Effect.uninterruptibleMask((restore) =>
         Effect.gen(function* () {
+          // Gather the evidence: how did the provider stream end?
           const stream = yield* restore(providerStream).pipe(Effect.exit)
-          const failure =
-            stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined
+          const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream))
+          // Note: Exit.hasInterrupts is a type guard whose false branch unsoundly narrows
+          // away non-interrupt failures, so both interrupt checks stay Cause-based.
+          const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)
+
+          // A context overflow before any assistant output is recoverable: compact and
+          // restart the turn instead of surfacing the provider error.
           if (
             recoverOverflow &&
             !publisher.hasAssistantStarted() &&
-            isContextOverflowFailure(overflowFailure ?? failure) &&
+            isContextOverflowFailure(overflowFailure ?? streamFailure) &&
             (yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request })))
           )
-            return yield* Effect.die(continueAfterOverflowCompaction(currentStep))
+            return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
+
+          // An unrecovered held-back overflow becomes the turn's durable provider error. A
+          // thrown LLM failure fails hosted tool calls and the assistant unless a provider
+          // error was already recorded from the stream.
           if (overflowFailure) yield* publish(overflowFailure)
-          const llmFailure = failure instanceof LLMError ? failure : undefined
+          const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined
           if (llmFailure && !publisher.hasProviderError()) {
             yield* publication.withPermit(publisher.failUnsettledTools("Provider did not return a tool result", true))
             yield* publication.withPermit(publisher.failAssistant(llmFailure.reason.message))
           }
-          if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers)
+          // Provider error events only arrive from the stream, so the flag is final here.
+          const providerFailed = publisher.hasProviderError()
+
+          // Settle tool fibers: an interrupted stream abandons unstarted tool work first.
+          if (streamInterrupted) yield* FiberSet.clear(toolFibers)
           const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit)
-          const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)
           const toolsInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)
-          if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) {
-            yield* FiberSet.clear(toolFibers)
-            yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
-            yield* publication.withPermit(publisher.failAssistant("Provider turn interrupted"))
-            return yield* Effect.interrupt
-          }
-          if (streamInterrupted || toolsInterrupted) {
+          const questionDismissed = settled._tag === "Failure" && isQuestionRejected(settled.cause)
+
+          if (questionDismissed || streamInterrupted || toolsInterrupted) {
             yield* FiberSet.clear(toolFibers)
             yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
             yield* publication.withPermit(publisher.failAssistant("Provider turn interrupted"))
+            // Match V1: dismissing a question halts the loop like an interruption.
+            if (questionDismissed) return yield* Effect.interrupt
           }
-          if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) {
-            const failure = Cause.squash(settled.cause)
+          // A settled tool fiber failure is one of two things. A defect from a tool
+          // implementation becomes a failed tool call the model can read, and the turn still
+          // settles so the model may recover. A typed infrastructure failure (tool output
+          // could not be persisted) also fails the assistant and then fails the drain.
+          const settledFailure = settled._tag === "Failure" && !toolsInterrupted ? settled.cause : undefined
+          const infraError =
+            settledFailure === undefined ? undefined : Option.getOrUndefined(Cause.findErrorOption(settledFailure))
+          if (settledFailure !== undefined) {
+            const failure = infraError ?? Cause.squash(settledFailure)
             const message = failure instanceof Error ? failure.message : String(failure)
             yield* publication.withPermit(publisher.failUnsettledTools(`Tool execution failed: ${message}`))
+            if (infraError !== undefined)
+              yield* publication.withPermit(publisher.failAssistant(`Tool execution failed: ${message}`))
           }
+
           const stepSettlement = publisher.stepSettlement()
-          if (stepSettlement && !streamInterrupted && !toolsInterrupted && !publisher.hasProviderError()) {
+          if (
+            stepSettlement &&
+            !streamInterrupted &&
+            !toolsInterrupted &&
+            infraError === undefined &&
+            !providerFailed
+          ) {
             const endSnapshot = yield* snapshots.capture()
             const files =
               startSnapshot && endSnapshot
@@ -361,49 +372,43 @@ const layer = Layer.effect(
               }),
             )
           }
-          if (publisher.hasProviderError())
-            yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
-          if (stream._tag === "Success" && !publisher.hasProviderError())
+          // A provider error orphans recorded local calls; a clean stream can still leave
+          // hosted calls without results.
+          if (providerFailed) yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
+          if (stream._tag === "Success" && !providerFailed)
             yield* publication.withPermit(publisher.failUnsettledTools("Provider did not return a tool result", true))
+
           if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
-          if (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause))
+          if (settled._tag === "Failure" && (toolsInterrupted || infraError !== undefined))
             return yield* Effect.failCause(settled.cause)
-          return { needsContinuation: !publisher.hasProviderError() && needsContinuation, step: currentStep }
+          return {
+            _tag: "Completed",
+            needsContinuation: !providerFailed && needsContinuation,
+            step: currentStep,
+          } as const
         }),
       )
     }, Effect.scoped)
-    type RunTurn = (
+
+    const runTurn = Effect.fnUntraced(function* (
       sessionID: SessionSchema.ID,
       promotion: SessionInput.Delivery | undefined,
       step: number,
-    ) => Effect.Effect<{ readonly needsContinuation: boolean; readonly step: number }, RunError>
-
-    const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) {
-      return yield* runTurnAttempt(sessionID, promotion, step).pipe(
-        Effect.catchDefect(
-          Effect.fnUntraced(function* (defect) {
-            if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect)
-            if (defect.transition._tag === "ContinueAfterOverflowCompaction")
-              return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow")
-            yield* Effect.yieldNow
-            return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step)
-          }),
-        ),
-      )
-    })
-
-    const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) {
-      return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe(
-        Effect.catchDefect(
-          Effect.fnUntraced(function* (defect) {
-            if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect)
-            yield* Effect.yieldNow
-            if (defect.transition._tag === "ContinueAfterOverflowCompaction")
-              return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step)
-            return yield* runTurn(sessionID, undefined, defect.transition.step)
-          }),
-        ),
-      )
+    ) {
+      // Compaction restarts rebuild the request from compacted history without re-promoting.
+      // Overflow recovery is one-shot: a post-compaction attempt must not recover another
+      // overflow, so the recovery hook is dropped after it fires.
+      let recoverOverflow: typeof compaction.compactAfterOverflow | undefined = compaction.compactAfterOverflow
+      let currentPromotion = promotion
+      let currentStep = step
+      while (true) {
+        const attempt = yield* runTurnAttempt(sessionID, currentPromotion, currentStep, recoverOverflow)
+        if (attempt._tag === "Completed") return { needsContinuation: attempt.needsContinuation, step: attempt.step }
+        if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined
+        yield* Effect.yieldNow
+        currentPromotion = undefined
+        currentStep = attempt.step
+      }
     })
 
     const drain = Effect.fnUntraced(function* (input: {
@@ -437,32 +442,10 @@ const layer = Layer.effect(
       }
     })
 
-    const run = Effect.fn("SessionRunner.run")(
-      (input: { readonly sessionID: SessionSchema.ID; readonly force: boolean }) =>
-        drain(input).pipe(
-          Effect.onExit((exit) =>
-            Effect.gen(function* () {
-              const failure =
-                Exit.isFailure(exit) && !Cause.hasInterrupts(exit.cause) ? Cause.squash(exit.cause) : undefined
-              yield* events.publish(SessionEvent.ExecutionSettled, {
-                sessionID: input.sessionID,
-                timestamp: yield* DateTime.now,
-                outcome: Exit.isSuccess(exit) ? "success" : Cause.hasInterrupts(exit.cause) ? "interrupted" : "failure",
-                error:
-                  failure !== undefined
-                    ? { type: "unknown", message: failure instanceof Error ? failure.message : String(failure) }
-                    : undefined,
-              })
-            }).pipe(
-              Effect.catchCause(() => Effect.void),
-              Effect.asVoid,
-            ),
-          ),
-        ),
-    )
-
     return Service.of({
-      run,
+      // ExecutionSettled is published per execution (busy period) by SessionExecution,
+      // not per drain here.
+      run: Effect.fn("SessionRunner.run")(drain),
     })
   }),
 )

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

@@ -81,11 +81,20 @@ describe("SessionV2.compact", () => {
       const events = yield* EventV2.Service
       const created = yield* session.create({ location })
 
+      const messageID = SessionMessage.ID.create()
+      const prompt = Prompt.make({ text: "Please compact this session history." })
+      yield* events.publish(SessionEvent.PromptAdmitted, {
+        sessionID: created.id,
+        messageID,
+        timestamp: DateTime.makeUnsafe(0),
+        prompt,
+        delivery: "steer",
+      })
       yield* events.publish(SessionEvent.Prompted, {
         sessionID: created.id,
-        messageID: SessionMessage.ID.create(),
+        messageID,
         timestamp: DateTime.makeUnsafe(0),
-        prompt: Prompt.make({ text: "Please compact this session history." }),
+        prompt,
         delivery: "steer",
       })
 

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

@@ -187,7 +187,7 @@ describe("SessionV2.create", () => {
         prompt: Prompt.make({ text: "First" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, parent.id)
       yield* events.publish(SessionEvent.Synthetic, {
         sessionID: parent.id,
         messageID: SessionMessage.ID.create(),
@@ -219,9 +219,9 @@ describe("SessionV2.create", () => {
       })
 
       yield* session.prompt({ sessionID: parent.id, prompt: Prompt.make({ text: "Parent changed" }), resume: false })
-      yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, parent.id)
       yield* session.prompt({ sessionID: forked.id, prompt: Prompt.make({ text: "Child continues" }), resume: false })
-      yield* SessionInput.promoteSteers(db, events, forked.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, forked.id)
 
       expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
       expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
@@ -246,13 +246,13 @@ describe("SessionV2.create", () => {
         prompt: Prompt.make({ text: "First" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, parent.id)
       const second = yield* session.prompt({
         sessionID: parent.id,
         prompt: Prompt.make({ text: "Second" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, parent.id)
 
       const forked = yield* session.fork({ sessionID: parent.id, messageID: second.id })
 
@@ -368,7 +368,7 @@ describe("SessionV2.create", () => {
       const { db } = yield* Database.Service
       const created = yield* session.create({ location })
       yield* session.prompt({ sessionID: created.id, prompt: Prompt.make({ text: "Hello" }), resume: false })
-      yield* SessionInput.promoteSteers(db, events, created.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, created.id)
 
       expect(
         Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)),
@@ -390,7 +390,7 @@ describe("SessionV2.create", () => {
         prompt: Prompt.make({ text: "Replay lifecycle" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id)
       const serialized = (yield* sourceDb
         .select()
         .from(EventTable)

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

@@ -129,6 +129,13 @@ describe("SessionProjector", () => {
         .pipe(Effect.orDie)
       const events = yield* EventV2.Service
 
+      yield* events.publish(SessionEvent.PromptAdmitted, {
+        sessionID,
+        messageID: SessionMessage.ID.make("msg_first"),
+        timestamp: created,
+        prompt: Prompt.make({ text: "first" }),
+        delivery: "steer",
+      })
       yield* events.publish(
         SessionEvent.Prompted,
         {
@@ -140,6 +147,13 @@ describe("SessionProjector", () => {
         },
         { id: EventV2.ID.make("evt_z") },
       )
+      yield* events.publish(SessionEvent.PromptAdmitted, {
+        sessionID,
+        messageID: SessionMessage.ID.make("msg_second"),
+        timestamp: created,
+        prompt: Prompt.make({ text: "second" }),
+        delivery: "steer",
+      })
       yield* events.publish(
         SessionEvent.Prompted,
         {

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

@@ -196,7 +196,7 @@ describe("SessionV2.prompt", () => {
         prompt: Prompt.make({ text: "boundary" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, sessionID)
       const stale = SessionMessage.ID.make("msg_stale_assistant")
       yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie)
       yield* events.publish(SessionEvent.RevertEvent.Staged, {
@@ -253,7 +253,7 @@ describe("SessionV2.prompt", () => {
 
       yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false })
       yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false })
-      yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers(db, events, sessionID)
       const streamed = Array.from(yield* Fiber.join(fiber))
 
       expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([
@@ -425,10 +425,7 @@ describe("SessionV2.prompt", () => {
       yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "Promote once" }), resume: false })
 
       yield* Effect.all(
-        [
-          SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER),
-          SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER),
-        ],
+        [SessionInput.promoteSteers(db, events, sessionID), SessionInput.promoteSteers(db, events, sessionID)],
         { concurrency: "unbounded" },
       )
 
@@ -440,23 +437,6 @@ describe("SessionV2.prompt", () => {
     }),
   )
 
-  it.effect("promotes steers only through the captured inbox cutoff", () =>
-    Effect.gen(function* () {
-      yield* setup
-      const { db } = yield* Database.Service
-      const session = yield* SessionV2.Service
-      const events = yield* EventV2.Service
-      const first = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Before cutoff" }), resume: false })
-      const cutoff = first.admittedSeq
-      const second = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "After cutoff" }), resume: false })
-
-      yield* SessionInput.promoteSteers(db, events, sessionID, cutoff)
-
-      expect(yield* admitted(first.id)).toHaveProperty("promotedSeq")
-      expect(yield* admitted(second.id)).not.toHaveProperty("promotedSeq")
-    }),
-  )
-
   it.effect("reprojects pending inbox input without scheduling execution", () =>
     Effect.gen(function* () {
       yield* setup
@@ -500,48 +480,6 @@ describe("SessionV2.prompt", () => {
     }),
   )
 
-  it.effect("returns an exact retry of a legacy projected prompt", () =>
-    Effect.gen(function* () {
-      yield* setup
-      const session = yield* SessionV2.Service
-      const events = yield* EventV2.Service
-      const prompt = Prompt.make({ text: "Historical prompt" })
-      yield* events.publish(SessionEvent.Prompted, {
-        sessionID,
-        messageID,
-        timestamp: yield* DateTime.now,
-        prompt,
-        delivery: "steer",
-      })
-
-      const retried = yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
-
-      expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical prompt" } })
-      expect(yield* admitted(messageID)).toHaveProperty("promotedSeq")
-    }),
-  )
-
-  it.effect("returns an exact retry of a legacy projected queued prompt", () =>
-    Effect.gen(function* () {
-      yield* setup
-      const session = yield* SessionV2.Service
-      const events = yield* EventV2.Service
-      const prompt = Prompt.make({ text: "Historical queued prompt" })
-      yield* events.publish(SessionEvent.Prompted, {
-        sessionID,
-        messageID,
-        timestamp: yield* DateTime.now,
-        prompt,
-        delivery: "queue",
-      })
-
-      const retried = yield* session.prompt({ id: messageID, sessionID, prompt, delivery: "queue", resume: false })
-
-      expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical queued prompt" } })
-      expect(yield* admitted(messageID)).toMatchObject({ delivery: "queue" })
-    }),
-  )
-
   it.effect("rejects reuse of one globally unique message ID across sessions", () =>
     Effect.gen(function* () {
       yield* setup

packages/core/test/session-run-coordinator.test.ts

@@ -392,6 +392,68 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("settles once per execution across coalesced drains", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const gate = yield* Deferred.make<void>()
+        const idle = yield* Deferred.make<void>()
+        let drains = 0
+        const settled: Exit.Exit<void, never>[] = []
+        const coordinator = yield* SessionRunCoordinator.make<string, never>({
+          drain: () =>
+            Effect.sync(() => ++drains).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate)))
+                  : Effect.void,
+              ),
+              Effect.asVoid,
+            ),
+          settled: (_key, exit) =>
+            Effect.sync(() => void settled.push(exit)).pipe(
+              Effect.andThen(Deferred.succeed(idle, undefined)),
+              Effect.asVoid,
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        yield* coordinator.wake("session")
+        yield* Deferred.succeed(gate, undefined)
+        yield* Deferred.await(idle)
+
+        expect(drains).toBe(2)
+        expect(settled).toHaveLength(1)
+        expect(Exit.isSuccess(settled[0]!)).toBe(true)
+      }),
+    ),
+  )
+
+  it.effect("settles interrupted executions before waiters resolve", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const gate = yield* Deferred.make<void>()
+        const settled: Exit.Exit<void, never>[] = []
+        const coordinator = yield* SessionRunCoordinator.make<string, never>({
+          drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate))),
+          settled: (_key, exit) => Effect.sync(() => void settled.push(exit)),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        yield* coordinator.interrupt("session")
+
+        expect(settled).toHaveLength(1)
+        expect(settled[0] !== undefined && Exit.isFailure(settled[0]) && Cause.hasInterrupts(settled[0].cause)).toBe(
+          true,
+        )
+        expect(yield* coordinator.active).toEqual(new Set())
+      }),
+    ),
+  )
+
   it.effect("trampolines synchronous self-waking execution", () =>
     Effect.scoped(
       Effect.gen(function* () {

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

@@ -156,6 +156,14 @@ const echo = Layer.effectDiscard(
         output: Schema.Struct({}),
         execute: () => Effect.die("unexpected tool defect"),
       }),
+      // BigInt output with no model content forces ToolOutputStore.bound onto its
+      // JSON.stringify encode path, which fails with a typed StorageError.
+      storefail: Tool.make({
+        description: "Produce output that cannot be persisted",
+        input: Schema.Struct({}),
+        output: Schema.Any,
+        execute: () => Effect.succeed({ big: 1n }),
+      }),
     }),
   ),
 )
@@ -653,6 +661,7 @@ describe("SessionRunnerLLM", () => {
       response = []
 
       const message = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run automatically" }) })
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(1)
       expect(yield* session.messages({ sessionID })).toMatchObject([
@@ -677,7 +686,7 @@ describe("SessionRunnerLLM", () => {
 
       expect(requests).toHaveLength(1)
       expect(requests[0]?.model).toBe(model)
-      expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect"])
+      expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect", "storefail"])
       expect(requests[0]?.messages.map((message) => ({ role: message.role, content: message.content }))).toEqual([
         { role: "user", content: [{ type: "text", text: "First" }] },
         { role: "user", content: [{ type: "text", text: "Second" }] },
@@ -712,6 +721,7 @@ describe("SessionRunnerLLM", () => {
 
       systemUnavailable = false
       yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "First" }) })
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(1)
       expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user"])
@@ -1553,7 +1563,7 @@ describe("SessionRunnerLLM", () => {
       yield* session.resume(sessionID)
 
       expect(requests).toHaveLength(1)
-      expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect"])
+      expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect", "storefail"])
       expect(yield* session.context(sessionID)).toMatchObject([
         { type: "user", text: "Use tools" },
         {
@@ -2389,7 +2399,7 @@ describe("SessionRunnerLLM", () => {
       const session = yield* SessionV2.Service
       const events = yield* EventV2.Service
       yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Recover interrupted tool" }), resume: false })
-      yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
       const assistantMessageID = SessionMessage.ID.create()
       yield* events.publish(SessionEvent.Step.Started, {
         sessionID,
@@ -2453,7 +2463,7 @@ describe("SessionRunnerLLM", () => {
         prompt: Prompt.make({ text: "Recover interrupted hosted tool" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
       const assistantMessageID = SessionMessage.ID.create()
       yield* events.publish(SessionEvent.Step.Started, {
         sessionID,
@@ -2513,7 +2523,7 @@ describe("SessionRunnerLLM", () => {
         prompt: Prompt.make({ text: "Recover interrupted tool input" }),
         resume: false,
       })
-      yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER)
+      yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
       const assistantMessageID = SessionMessage.ID.create()
       yield* events.publish(SessionEvent.Step.Started, {
         sessionID,
@@ -2626,8 +2636,9 @@ describe("SessionRunnerLLM", () => {
 
       const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
       yield* Deferred.await(streamStarted)
+      streamStarted = yield* Deferred.make<void>()
       const second = yield* session.resume(otherSessionID).pipe(Effect.forkChild)
-      yield* Effect.yieldNow
+      yield* Deferred.await(streamStarted)
 
       expect(requests).toHaveLength(2)
       expect(requests.map((request) => request.providerOptions?.openai?.promptCacheKey)).toEqual([
@@ -2797,6 +2808,49 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("fails the drain when tool output persistence fails", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Call storefail" }), resume: false })
+
+      requests.length = 0
+      responses = [
+        [
+          LLMEvent.stepStart({ index: 0 }),
+          LLMEvent.toolCall({ id: "call-storefail", name: "storefail", input: {} }),
+          LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
+          LLMEvent.finish({ reason: "tool-calls" }),
+        ],
+        [],
+      ]
+
+      const exit = yield* session.resume(sessionID).pipe(Effect.exit)
+
+      expect(Exit.isFailure(exit)).toBe(true)
+      expect(requests).toHaveLength(1)
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Call storefail" },
+        {
+          type: "assistant",
+          content: [
+            {
+              type: "tool",
+              id: "call-storefail",
+              state: {
+                status: "error",
+                error: {
+                  type: "unknown",
+                  message: expect.stringContaining("Tool execution failed: Failed to encode tool output"),
+                },
+              },
+            },
+          ],
+        },
+      ])
+    }),
+  )
+
   it.effect("interrupts runner continuation when a question is dismissed", () =>
     Effect.gen(function* () {
       yield* setup

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

@@ -41,7 +41,14 @@ const models = Layer.mock(SessionRunnerModel.Service)({
 })
 const it = testEffect(
   AppNodeBuilder.build(
-    LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, AgentV2.node, SessionTitle.node]),
+    LayerNode.group([
+      Database.node,
+      EventV2.node,
+      SessionProjector.node,
+      SessionStore.node,
+      AgentV2.node,
+      SessionTitle.node,
+    ]),
     [
       [llmClient, client],
       [SessionRunnerModel.node, models],
@@ -76,9 +83,17 @@ const insertSession = (id: SessionV2.ID) =>
 const prompt = (sessionID: SessionV2.ID, text: string) =>
   Effect.gen(function* () {
     const events = yield* EventV2.Service
+    const messageID = SessionMessage.ID.create()
+    yield* events.publish(SessionEvent.PromptAdmitted, {
+      sessionID,
+      messageID,
+      timestamp: DateTime.makeUnsafe(0),
+      prompt: Prompt.make({ text }),
+      delivery: "steer",
+    })
     yield* events.publish(SessionEvent.Prompted, {
       sessionID,
-      messageID: SessionMessage.ID.create(),
+      messageID,
       timestamp: DateTime.makeUnsafe(0),
       prompt: Prompt.make({ text }),
       delivery: "steer",
@@ -101,9 +116,9 @@ it.effect("generates a title from the sole user message and renames the session"
     yield* prompt(sessionID, "Help me debug the failing build")
 
     const store = yield* SessionStore.Service
-    const session = yield* store.get(sessionID).pipe(
-      Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))),
-    )
+    const session = yield* store
+      .get(sessionID)
+      .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))))
     const title = yield* SessionTitle.Service
     yield* title.generateForFirstPrompt(session)
 
@@ -131,9 +146,9 @@ it.effect("does not generate once a second user message exists", () =>
     yield* prompt(sessionID, "Second message")
 
     const store = yield* SessionStore.Service
-    const session = yield* store.get(sessionID).pipe(
-      Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))),
-    )
+    const session = yield* store
+      .get(sessionID)
+      .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))))
     const title = yield* SessionTitle.Service
     yield* title.generateForFirstPrompt(session)
 
@@ -179,9 +194,9 @@ it.effect("does not generate for a child session", () =>
     yield* prompt(sessionID, "Do this subtask")
 
     const store = yield* SessionStore.Service
-    const session = yield* store.get(sessionID).pipe(
-      Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))),
-    )
+    const session = yield* store
+      .get(sessionID)
+      .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))))
     const title = yield* SessionTitle.Service
     yield* title.generateForFirstPrompt(session)
 
@@ -197,9 +212,9 @@ it.effect("does not generate when the title agent is removed", () =>
     yield* prompt(sessionID, "Help me debug the failing build")
 
     const store = yield* SessionStore.Service
-    const session = yield* store.get(sessionID).pipe(
-      Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))),
-    )
+    const session = yield* store
+      .get(sessionID)
+      .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))))
     const title = yield* SessionTitle.Service
     yield* title.generateForFirstPrompt(session)