feat(core): publish terminal session run failures

This reverts commit fd07e58384.

packages/core/src/session/event.ts

@@ -119,6 +119,23 @@ export namespace PromptLifecycle {
   export type Promoted = typeof Promoted.Type
 }
 
+export namespace Run {
+  export const Failed = EventV2.define({
+    type: "session.next.run.failed",
+    ...options,
+    schema: {
+      ...Base,
+      reason: Schema.Literals(["execution-failed", "step-limit-exceeded", "unknown"]),
+      input: Schema.Struct({
+        messageID: SessionMessageID.ID,
+        admittedSeq: NonNegativeInt,
+        promotedSeq: NonNegativeInt.pipe(Schema.optional),
+      }).pipe(Schema.optional),
+    },
+  })
+  export type Failed = typeof Failed.Type
+}
+
 export const InterruptRequested = EventV2.define({
   type: "session.next.interrupt.requested",
   ...options,
@@ -476,6 +493,7 @@ const DurableDefinitions = [
   Prompted,
   PromptLifecycle.Admitted,
   PromptLifecycle.Promoted,
+  Run.Failed,
   InterruptRequested,
   ContextUpdated,
   Synthetic,

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

@@ -1,10 +1,15 @@
-import { Effect, Layer } from "effect"
+import { Cause, DateTime, Effect, Layer, Option } from "effect"
+import { LLMError } from "@opencode-ai/llm"
+import { Database } from "../../database/database"
+import { EventV2 } from "../../event"
 import { LocationServiceMap } from "../../location-layer"
 import { SessionRunCoordinator } from "../run-coordinator"
 import { SessionRunner } from "../runner"
 import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { SessionExecution } from "../execution"
+import { SessionEvent } from "../event"
+import { SessionInput } from "../input"
 
 /** Current-process routing for implicit-local Locations. Future remote placement belongs here. */
 export const layer = Layer.effect(
@@ -12,6 +17,8 @@ export const layer = Layer.effect(
   Effect.gen(function* () {
     const store = yield* SessionStore.Service
     const locations = yield* LocationServiceMap
+    const database = yield* Database.Service
+    const events = yield* EventV2.Service
     const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, void, SessionRunner.RunError>({
       drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, mode) {
         const session = yield* store.get(sessionID)
@@ -20,11 +27,34 @@ export const layer = Layer.effect(
           Effect.provide(locations.get(session.location)),
         )
       }),
-      onFailure: (sessionID, cause) =>
-        Effect.logError("Failed to drain Session").pipe(
-          Effect.annotateLogs("sessionID", sessionID),
-          Effect.annotateLogs("cause", cause),
-        ),
+      onFailure: (sessionID, cause, context) =>
+        Effect.gen(function* () {
+          yield* Effect.logError("Failed to drain Session").pipe(
+            Effect.annotateLogs("sessionID", sessionID),
+            Effect.annotateLogs("cause", cause),
+          )
+          if (Cause.hasInterruptsOnly(cause)) return
+          const error = Option.getOrUndefined(Cause.findErrorOption(cause))
+          // Provider failures already publish Step.Failed before escaping the runner.
+          if (error instanceof LLMError) return
+          const input = context.seq === undefined
+            ? undefined
+            : yield* SessionInput.findByAdmittedSeq(database.db, sessionID, context.seq)
+          yield* events.publish(SessionEvent.Run.Failed, {
+            sessionID,
+            timestamp: yield* DateTime.now,
+            reason: error instanceof SessionRunner.StepLimitExceededError ? "step-limit-exceeded" : "execution-failed",
+            ...(input === undefined
+              ? {}
+              : {
+                  input: {
+                    messageID: input.id,
+                    admittedSeq: input.admittedSeq,
+                    ...(input.promotedSeq === undefined ? {} : { promotedSeq: input.promotedSeq }),
+                  },
+                }),
+          })
+        }),
     })
 
     return SessionExecution.Service.of({

packages/core/src/session/input.ts

@@ -47,6 +47,20 @@ export const find = Effect.fn("SessionInput.find")(function* (db: DatabaseServic
   return row === undefined ? undefined : fromRow(row)
 })
 
+export const findByAdmittedSeq = Effect.fn("SessionInput.findByAdmittedSeq")(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  admittedSeq: number,
+) {
+  const row = yield* db
+    .select()
+    .from(SessionInputTable)
+    .where(and(eq(SessionInputTable.session_id, sessionID), eq(SessionInputTable.admitted_seq, admittedSeq)))
+    .get()
+    .pipe(Effect.orDie)
+  return row === undefined ? undefined : fromRow(row)
+})
+
 export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()("SessionInput.LifecycleConflict", {
   id: SessionMessage.ID,
 }) {}

packages/core/src/session/message-updater.ts

@@ -139,6 +139,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       },
       "session.next.prompt.admitted": () => Effect.void,
       "session.next.prompt.promoted": () => Effect.void,
+      "session.next.run.failed": () => Effect.void,
       "session.next.interrupt.requested": () => Effect.void,
       "session.next.context.updated": (event) =>
         adapter.appendMessage(

packages/core/src/session/projector.ts

@@ -410,6 +410,7 @@ export const layer = Layer.effectDiscard(
         )
       }),
     )
+    yield* events.project(SessionEvent.Run.Failed, () => Effect.void)
     yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
     yield* events.project(SessionEvent.ContextUpdated, (event) => {
       if (!event.replay || event.seq === undefined) return run(db, event)

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

@@ -39,13 +39,14 @@ export interface Coordinator<Key, A, E> {
 /** One Session's process-local execution lane: one active demand and at most one coalesced follow-up. */
 type Entry<A, E> = {
   readonly done: Deferred.Deferred<A, E>
-  readonly settled: Deferred.Deferred<Exit.Exit<A, E>>
+  readonly settled: Deferred.Deferred<Exit.Exit<A, E> | undefined>
   current: Demand
   pending?: Demand
   explicitWaiter?: Deferred.Deferred<A, E>
   interruptSeq?: number
   owner?: Fiber.Fiber<void, never>
   stopping: boolean
+  advisoryRetryAvailable: boolean
 }
 
 /** Combines follow-up demand: runs dominate, while wakes retain the newest durable admission sequence. */
@@ -63,7 +64,11 @@ const maxSeq = (left: number | undefined, right: number | undefined) => {
 /** Constructs a scoped coordinator. Every in-memory transition is synchronous. */
 export const make = <Key, A, E>(options: {
   readonly drain: (key: Key, mode: Mode) => Effect.Effect<A, E>
-  readonly onFailure?: (key: Key, cause: Cause.Cause<E>) => Effect.Effect<void>
+  readonly onFailure?: (
+    key: Key,
+    cause: Cause.Cause<E>,
+    context: { readonly mode: "wake"; readonly seq?: number },
+  ) => Effect.Effect<void>
 }): Effect.Effect<Coordinator<Key, A, E>, never, Scope.Scope> =>
   Effect.gen(function* () {
     const active = new Map<Key, Entry<A, E>>()
@@ -81,12 +86,16 @@ export const make = <Key, A, E>(options: {
       }),
     )
 
-    const makeEntry = (current: Demand, explicitWaiter?: Deferred.Deferred<A, E>): Entry<A, E> => ({
+    const makeEntry = (current: Demand, options?: {
+      readonly explicitWaiter?: Deferred.Deferred<A, E>
+      readonly advisoryRetryAvailable?: boolean
+    }): Entry<A, E> => ({
       done: Deferred.makeUnsafe<A, E>(),
-      settled: Deferred.makeUnsafe<Exit.Exit<A, E>>(),
+      settled: Deferred.makeUnsafe<Exit.Exit<A, E> | undefined>(),
       current,
-      explicitWaiter,
+      explicitWaiter: options?.explicitWaiter,
       stopping: false,
+      advisoryRetryAvailable: options?.advisoryRetryAvailable ?? current._tag === "wake",
     })
 
     const start = (key: Key, entry: Entry<A, E>, demand: Demand, successor = false) => {
@@ -132,6 +141,7 @@ export const make = <Key, A, E>(options: {
           const pending = entry.pending
           entry.pending = undefined
           entry.current = pending
+          entry.advisoryRetryAvailable = pending._tag === "wake"
           start(key, entry, pending, true)
           return
         }
@@ -141,19 +151,26 @@ export const make = <Key, A, E>(options: {
         return
       }
 
-      const successor = entry.pending !== undefined ? makeEntry(entry.pending, entry.explicitWaiter) : undefined
+      const successor =
+        entry.pending !== undefined
+          ? makeEntry(entry.pending, { explicitWaiter: entry.explicitWaiter })
+          : exit._tag === "Failure" && demand._tag === "wake" && !entry.stopping && entry.advisoryRetryAvailable
+            ? makeEntry(demand, { explicitWaiter: entry.explicitWaiter, advisoryRetryAvailable: false })
+            : undefined
+      const retrying = successor !== undefined && entry.pending === undefined
       if (successor === undefined) active.delete(key)
       else active.set(key, successor)
       if (successor !== undefined) start(key, successor, successor.current, true)
       Deferred.doneUnsafe(entry.done, exit)
-      Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
+      Deferred.doneUnsafe(entry.settled, Effect.succeed(retrying ? undefined : exit))
       if (
         exit._tag === "Failure" &&
         !(entry.stopping && Cause.hasInterruptsOnly(exit.cause)) &&
         demand._tag === "wake" &&
+        successor === undefined &&
         options.onFailure !== undefined
       ) {
-        report(Effect.suspend(() => options.onFailure!(key, exit.cause)))
+        report(Effect.suspend(() => options.onFailure!(key, exit.cause, { mode: "wake", seq: demand.seq })))
       }
     }
 
@@ -184,7 +201,7 @@ export const make = <Key, A, E>(options: {
             Deferred.await(shutdown).pipe(Effect.as(Exit.void)),
           )
           if (closed) break
-          if (exit._tag === "Failure" && firstFailure === undefined) firstFailure = exit.cause
+          if (exit?._tag === "Failure" && firstFailure === undefined) firstFailure = exit.cause
         }
         if (firstFailure !== undefined) return yield* Effect.failCause(firstFailure)
       })

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

@@ -33,7 +33,9 @@ describe("SessionRunCoordinator", () => {
     Effect.scoped(
       Effect.gen(function* () {
         const drained = yield* Deferred.make<void>()
-        const coordinator = yield* SessionRunCoordinator.make({ drain: () => Deferred.succeed(drained, undefined) })
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () => Deferred.succeed(drained, undefined),
+        })
 
         yield* coordinator.wake("session")
         yield* Deferred.await(drained)
@@ -44,7 +46,9 @@ describe("SessionRunCoordinator", () => {
   it.effect("does nothing when interrupted while idle", () =>
     Effect.scoped(
       Effect.gen(function* () {
-        const coordinator = yield* SessionRunCoordinator.make({ drain: () => Effect.void })
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () => Effect.void,
+        })
 
         yield* coordinator.interrupt("session")
       }),
@@ -55,7 +59,9 @@ describe("SessionRunCoordinator", () => {
     Effect.scoped(
       Effect.gen(function* () {
         let runs = 0
-        const coordinator = yield* SessionRunCoordinator.make({ drain: () => Effect.sync(() => runs++) })
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () => Effect.sync(() => runs++),
+        })
 
         yield* coordinator.interrupt("session", 2)
         yield* coordinator.wake("session", 1)
@@ -722,6 +728,36 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("retries a failed advisory successor once without another wake", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstGate = yield* Deferred.make<void>()
+        let runs = 0
+        const failure = new Error("transient wake failure")
+        const coordinator = yield* SessionRunCoordinator.make<string, void, Error>({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.await(firstGate).pipe(Effect.andThen(Effect.fail(failure)))
+                  : run === 2
+                    ? Effect.fail(failure)
+                    : Effect.void,
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session", 1)
+        yield* coordinator.wake("session", 2)
+        yield* Deferred.succeed(firstGate, undefined)
+        const idle = yield* coordinator.awaitIdle("session").pipe(Effect.exit)
+
+        expect(runs).toBe(3)
+        expect(Exit.isSuccess(idle)).toBeTrue()
+      }),
+    ),
+  )
+
   it.effect("upgrades an active wake when an explicit run joins it", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -916,8 +952,9 @@ describe("SessionRunCoordinator", () => {
         const failure = new Error("wake failed")
         const reported: Cause.Cause<Error>[] = []
         const reportedOnce = yield* Deferred.make<void>()
+        let runs = 0
         const coordinator = yield* SessionRunCoordinator.make<string, void, Error>({
-          drain: () => Effect.fail(failure),
+          drain: () => Effect.sync(() => runs++).pipe(Effect.andThen(Effect.fail(failure))),
           onFailure: (_key, cause) =>
             Effect.sync(() => reported.push(cause)).pipe(Effect.andThen(Deferred.succeed(reportedOnce, undefined))),
         })
@@ -927,6 +964,7 @@ describe("SessionRunCoordinator", () => {
         yield* Effect.yieldNow
 
         expect(reported).toHaveLength(1)
+        expect(runs).toBe(2)
         expect(Cause.squash(reported[0]!)).toBe(failure)
       }),
     ),