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 }))) } }