fix(core): align terminal failures with current runner

packages/core/src/session/event.ts

@@ -118,6 +118,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", "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,
@@ -475,6 +492,7 @@ const DurableDefinitions = [
   Prompted,
   PromptLifecycle.Admitted,
   PromptLifecycle.Promoted,
+  Run.Failed,
   InterruptRequested,
   ContextUpdated,
   Synthetic,

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

@@ -1,4 +1,7 @@
-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"
@@ -6,6 +9,8 @@ import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { SessionExecution } from "../execution"
 import { logFailure } from "../logging"
+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(
@@ -13,6 +18,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)
@@ -21,7 +28,31 @@ export const layer = Layer.effect(
           Effect.provide(locations.get(session.location)),
         )
       }),
-      onFailure: (sessionID, cause) => logFailure("Failed to drain Session", sessionID, cause),
+      onFailure: (sessionID, cause, context) =>
+        Effect.gen(function* () {
+          yield* logFailure("Failed to drain Session", sessionID, 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 === undefined ? "unknown" : "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

@@ -138,6 +138,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

@@ -406,6 +406,7 @@ export const layer = Layer.effectDiscard(
         )
       }),
     )
+    yield* events.project(SessionEvent.Run.Failed, () => Effect.void)
     yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
     // TODO: Reconstruct context epoch replacement state during replay without adding replay state to every EventV2 payload.
     yield* events.project(SessionEvent.ContextUpdated, (event) => run(db, event))

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

@@ -64,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>>()
@@ -154,7 +158,7 @@ export const make = <Key, A, E>(options: {
         demand._tag === "wake" &&
         options.onFailure !== undefined
       ) {
-        report(Effect.suspend(() => options.onFailure!(key, exit.cause)))
+        report(Effect.suspend(() => options.onFailure!(key, exit.cause, { mode: "wake", seq: demand.seq })))
       }
     }
 

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

@@ -915,19 +915,24 @@ describe("SessionRunCoordinator", () => {
       Effect.gen(function* () {
         const failure = new Error("wake failed")
         const reported: Cause.Cause<Error>[] = []
+        const contexts: Array<{ readonly mode: "wake"; readonly seq?: number }> = []
         const reportedOnce = yield* Deferred.make<void>()
         const coordinator = yield* SessionRunCoordinator.make<string, void, Error>({
           drain: () => Effect.fail(failure),
-          onFailure: (_key, cause) =>
-            Effect.sync(() => reported.push(cause)).pipe(Effect.andThen(Deferred.succeed(reportedOnce, undefined))),
+          onFailure: (_key, cause, context) =>
+            Effect.sync(() => {
+              reported.push(cause)
+              contexts.push(context)
+            }).pipe(Effect.andThen(Deferred.succeed(reportedOnce, undefined))),
         })
 
-        yield* coordinator.wake("session")
+        yield* coordinator.wake("session", 7)
         yield* Deferred.await(reportedOnce)
         yield* Effect.yieldNow
 
         expect(reported).toHaveLength(1)
         expect(Cause.squash(reported[0]!)).toBe(failure)
+        expect(contexts).toEqual([{ mode: "wake", seq: 7 }])
       }),
     ),
   )