From cfab7a2d7900e7d0971c4b990588664984466517 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 09:48:31 -0400 Subject: [PATCH 1/2] feat(core): publish terminal session run failures --- packages/core/src/session/event.ts | 18 ++++++++++ packages/core/src/session/execution/local.ts | 35 ++++++++++++++++++-- packages/core/src/session/input.ts | 14 ++++++++ packages/core/src/session/message-updater.ts | 1 + packages/core/src/session/projector.ts | 1 + packages/core/src/session/run-coordinator.ts | 8 +++-- 6 files changed, 73 insertions(+), 4 deletions(-) diff --git a/packages/core/src/session/event.ts b/packages/core/src/session/event.ts index 5eaf037168..9d0ede00ff 100644 --- a/packages/core/src/session/event.ts +++ b/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", "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, @@ -475,6 +492,7 @@ const DurableDefinitions = [ Prompted, PromptLifecycle.Admitted, PromptLifecycle.Promoted, + Run.Failed, InterruptRequested, ContextUpdated, Synthetic, diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts index 8f1b1763a0..d44f47eead 100644 --- a/packages/core/src/session/execution/local.ts +++ b/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({ 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 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({ diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts index f8bc2b0e6b..e201274773 100644 --- a/packages/core/src/session/input.ts +++ b/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()("SessionInput.LifecycleConflict", { id: SessionMessage.ID, }) {} diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index cf1eb2cedf..343a49f461 100644 --- a/packages/core/src/session/message-updater.ts +++ b/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( diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index bffe4e74c6..ae46227037 100644 --- a/packages/core/src/session/projector.ts +++ b/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)) diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index d52b63e8f1..0f2ebf3587 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/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 = (options: { readonly drain: (key: Key, mode: Mode) => Effect.Effect - readonly onFailure?: (key: Key, cause: Cause.Cause) => Effect.Effect + readonly onFailure?: ( + key: Key, + cause: Cause.Cause, + context: { readonly mode: "wake"; readonly seq?: number }, + ) => Effect.Effect }): Effect.Effect, never, Scope.Scope> => Effect.gen(function* () { const active = new Map>() @@ -154,7 +158,7 @@ export const make = (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 }))) } } From aeaabc39287ae06dcc21883b6c7b18022c19dc48 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 21 Jun 2026 17:05:03 -0400 Subject: [PATCH 2/2] fix(core): align terminal failures with current runner --- packages/core/src/session/event.ts | 2 +- packages/core/src/session/execution/local.ts | 2 +- packages/core/test/session-run-coordinator.test.ts | 11 ++++++++--- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/packages/core/src/session/event.ts b/packages/core/src/session/event.ts index 9d0ede00ff..11a9bd6af7 100644 --- a/packages/core/src/session/event.ts +++ b/packages/core/src/session/event.ts @@ -124,7 +124,7 @@ export namespace Run { ...options, schema: { ...Base, - reason: Schema.Literals(["execution-failed", "step-limit-exceeded", "unknown"]), + reason: Schema.Literals(["execution-failed", "unknown"]), input: Schema.Struct({ messageID: SessionMessageID.ID, admittedSeq: NonNegativeInt, diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts index d44f47eead..2840810491 100644 --- a/packages/core/src/session/execution/local.ts +++ b/packages/core/src/session/execution/local.ts @@ -41,7 +41,7 @@ export const layer = Layer.effect( yield* events.publish(SessionEvent.Run.Failed, { sessionID, timestamp: yield* DateTime.now, - reason: error instanceof SessionRunner.StepLimitExceededError ? "step-limit-exceeded" : "execution-failed", + reason: error === undefined ? "unknown" : "execution-failed", ...(input === undefined ? {} : { diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index 39fb5779d3..4cfaf02311 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/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[] = [] + const contexts: Array<{ readonly mode: "wake"; readonly seq?: number }> = [] const reportedOnce = yield* Deferred.make() const coordinator = yield* SessionRunCoordinator.make({ 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 }]) }), ), )