feat(core): add execution lifecycle and retries

Author: kitlangton

packages/client/src/promise/generated/types.ts

@@ -0,0 +1,14 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260703210000_reset_v2_execution_errors",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`DELETE FROM \`session_input\`;`)
+      yield* tx.run(`DELETE FROM \`session_message\`;`)
+      yield* tx.run(`DELETE FROM \`event\`;`)
+      yield* tx.run(`DELETE FROM \`event_sequence\`;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration.gen.ts

@@ -59,7 +59,14 @@ export type AskResult = typeof AskResult.Type
 
 export const Event = Permission.Event
 
-export class RejectedError extends Schema.TaggedErrorClass<RejectedError>()("PermissionV2.RejectedError", {}) {}
+export class RejectedError extends Schema.TaggedErrorClass<RejectedError>()("PermissionV2.RejectedError", {
+  permission: Schema.String,
+  resources: Schema.Array(Schema.String),
+}) {
+  override get message() {
+    return `Permission rejected: ${this.permission}`
+  }
+}
 
 export class CorrectedError extends Schema.TaggedErrorClass<CorrectedError>()("PermissionV2.CorrectedError", {
   feedback: Schema.String,
@@ -119,9 +126,17 @@ const layer = Layer.effect(
     const pending = new Map<ID, Pending>()
 
     yield* Effect.addFinalizer(() =>
-      Effect.forEach(pending.values(), (item) => Deferred.fail(item.deferred, new RejectedError()), {
-        discard: true,
-      }).pipe(
+      Effect.forEach(
+        pending.values(),
+        (item) =>
+          Deferred.fail(
+            item.deferred,
+            new RejectedError({ permission: item.request.action, resources: [...item.request.resources] }),
+          ),
+        {
+          discard: true,
+        },
+      ).pipe(
         Effect.ensuring(
           Effect.sync(() => {
             pending.clear()
@@ -230,7 +245,12 @@ const layer = Layer.effect(
           if (input.reply === "reject") {
             yield* Deferred.fail(
               existing.deferred,
-              input.message ? new CorrectedError({ feedback: input.message }) : new RejectedError(),
+              input.message
+                ? new CorrectedError({ feedback: input.message })
+                : new RejectedError({
+                    permission: existing.request.action,
+                    resources: [...existing.request.resources],
+                  }),
             )
             pending.delete(input.requestID)
             for (const [id, item] of pending) {
@@ -240,7 +260,10 @@ const layer = Layer.effect(
                 requestID: item.request.id,
                 reply: "reject",
               })
-              yield* Deferred.fail(item.deferred, new RejectedError())
+              yield* Deferred.fail(
+                item.deferred,
+                new RejectedError({ permission: item.request.action, resources: [...item.request.resources] }),
+              )
               pending.delete(id)
             }
             return

packages/core/src/database/migration/20260703210000_reset_v2_execution_errors.ts

@@ -1,6 +1,7 @@
 import { Schema } from "effect"
 import { SessionMessage } from "./message"
 import { SessionSchema } from "./schema"
+import { SessionError } from "@opencode-ai/schema/session-error"
 
 export class MessageDecodeError extends Schema.TaggedErrorClass<MessageDecodeError>()("Session.MessageDecodeError", {
   sessionID: SessionSchema.ID,
@@ -10,3 +11,20 @@ export class MessageDecodeError extends Schema.TaggedErrorClass<MessageDecodeErr
     return `Failed to decode message ${this.messageID} in session ${this.sessionID}`
   }
 }
+
+export class StepFailedError extends Schema.TaggedErrorClass<StepFailedError>()("Session.StepFailedError", {
+  error: SessionError.Error,
+}) {
+  override get message() {
+    return this.error.message
+  }
+}
+
+export class UserInterruptedError extends Schema.TaggedErrorClass<UserInterruptedError>()(
+  "Session.UserInterruptedError",
+  {},
+) {
+  override get message() {
+    return "Session interrupted by user"
+  }
+}

packages/core/src/permission.ts

@@ -1,4 +1,4 @@
-import { Cause, DateTime, Effect, Exit, Layer } from "effect"
+import { Cause, Effect, Exit, Layer } from "effect"
 import { EventV2 } from "../../event"
 import { LocationServiceMap } from "../../location-service-map"
 import { makeGlobalNode } from "../../effect/app-node"
@@ -8,6 +8,16 @@ import { SessionRunner } from "../runner"
 import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { SessionExecution } from "../execution"
+import { toSessionError } from "../to-session-error"
+import { UserInterruptedError } from "../error"
+
+export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?: "user" | "shutdown" | "superseded") {
+  if (Exit.isSuccess(exit)) return { type: "succeeded" as const }
+  if (Cause.hasInterrupts(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" }
+  const failure = Cause.squash(exit.cause)
+  if (failure instanceof UserInterruptedError) return { type: "interrupted" as const, reason: "user" as const }
+  return { type: "failed" as const, error: toSessionError(failure) }
+}
 
 /** Current-process routing for implicit-local Locations. Future remote placement belongs here. */
 const layer = Layer.effect(
@@ -16,7 +26,23 @@ const layer = Layer.effect(
     const store = yield* SessionStore.Service
     const locations = yield* LocationServiceMap.Service
     const events = yield* EventV2.Service
-    const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError>({
+    const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
+      effect.pipe(
+        Effect.tapCause((cause) =>
+          Cause.hasInterruptsOnly(cause)
+            ? Effect.void
+            : Effect.logError("Failed to publish Session execution lifecycle", cause).pipe(
+                Effect.annotateLogs({ sessionID }),
+              ),
+        ),
+        Effect.asVoid,
+      )
+    const coordinator = yield* SessionRunCoordinator.make<
+      SessionSchema.ID,
+      SessionRunner.RunError,
+      "user" | "shutdown" | "superseded"
+    >({
+      started: (sessionID) => reportLifecycle(sessionID, events.publish(SessionEvent.Execution.Started, { sessionID })),
       drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) {
         const session = yield* store.get(sessionID)
         if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
@@ -29,28 +55,31 @@ 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,
-            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,
+      // One terminal observation per busy period, covering every coalesced drain.
+      settled: (sessionID, exit, reason) =>
+        reportLifecycle(
+          sessionID,
+          Effect.gen(function* () {
+            const outcome = terminal(exit, reason)
+            if (outcome.type === "succeeded") {
+              yield* events.publish(SessionEvent.Execution.Succeeded, { sessionID })
+              return
+            }
+            if (outcome.type === "interrupted") {
+              yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason })
+              return
+            }
+            yield* events.publish(SessionEvent.Execution.Failed, {
+              sessionID,
+              error: outcome.error,
+            })
+          }),
         ),
     })
 
     return SessionExecution.Service.of({
       active: coordinator.active,
-      interrupt: coordinator.interrupt,
+      interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"),
       resume: coordinator.run,
       wake: coordinator.wake,
       awaitIdle: coordinator.awaitIdle,

packages/core/src/session/error.ts

@@ -1,5 +1,5 @@
 import { castDraft, produce, type WritableDraft } from "immer"
-import { Effect } from "effect"
+import { DateTime, Effect } from "effect"
 import { SessionEvent } from "./event"
 import { SessionMessage } from "./message"
 
@@ -111,6 +111,17 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       if (assistant) yield* adapter.updateAssistant(produce(assistant, recipe))
     })
 
+  const clearCurrentRetry = Effect.gen(function* () {
+    const assistant = yield* adapter.getCurrentAssistant()
+    if (assistant?.retry) {
+      yield* adapter.updateAssistant(
+        produce(assistant, (draft) => {
+          draft.retry = undefined
+        }),
+      )
+    }
+  })
+
   return Effect.gen(function* () {
     yield* SessionEvent.All.match(event, {
       "session.agent.selected": (event) => {
@@ -144,7 +155,10 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       "session.forked": () => Effect.void,
       "session.prompt.promoted": () => Effect.void,
       "session.prompt.admitted": () => Effect.void,
-      "session.execution.settled": () => Effect.void,
+      "session.execution.started": () => Effect.void,
+      "session.execution.succeeded": () => clearCurrentRetry,
+      "session.execution.failed": () => clearCurrentRetry,
+      "session.execution.interrupted": () => clearCurrentRetry,
       "session.context.updated": (event) =>
         adapter.appendMessage(
           SessionMessage.System.make({
@@ -205,10 +219,26 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       },
       "session.step.started": (event) => {
         return Effect.gen(function* () {
+          const existing = yield* adapter.getAssistant(event.data.assistantMessageID)
+          if (existing) {
+            yield* adapter.updateAssistant(
+              produce(existing, (draft) => {
+                draft.agent = event.data.agent
+                draft.model = castDraft(event.data.model)
+                draft.retry = undefined
+                draft.error = undefined
+                draft.finish = undefined
+                draft.time.completed = undefined
+                if (event.data.snapshot) draft.snapshot = { ...draft.snapshot, start: event.data.snapshot }
+              }),
+            )
+            return
+          }
           const currentAssistant = yield* adapter.getCurrentAssistant()
           if (currentAssistant) {
             yield* adapter.updateAssistant(
               produce(currentAssistant, (draft) => {
+                draft.retry = undefined
                 draft.time.completed = event.created
               }),
             )
@@ -244,7 +274,8 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
         return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
           draft.time.completed = event.created
           draft.finish = "error"
-          draft.error = event.data.error
+          draft.error = castDraft(event.data.error)
+          draft.retry = undefined
         })
       },
       "session.text.started": (event) => {
@@ -383,7 +414,15 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
           }
         })
       },
-      "session.retried": () => Effect.void,
+      "session.retry.scheduled": (event) => {
+        return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
+          draft.retry = {
+            attempt: event.data.attempt,
+            at: DateTime.makeUnsafe(event.data.at),
+            error: castDraft(event.data.error),
+          }
+        })
+      },
       "session.compaction.started": () => Effect.void,
       "session.compaction.delta": () => Effect.void,
       "session.compaction.ended": (event) => {

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

@@ -634,6 +634,9 @@ const layer = Layer.effectDiscard(
         })
       }),
     )
+    yield* events.project(SessionEvent.Execution.Succeeded, (event) => run(db, event))
+    yield* events.project(SessionEvent.Execution.Failed, (event) => run(db, event))
+    yield* events.project(SessionEvent.Execution.Interrupted, (event) => run(db, event))
     yield* events.project(SessionEvent.ContextUpdated, (event) => run(db, event))
     yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
     yield* events.project(SessionEvent.Skill.Activated, (event) =>
@@ -660,7 +663,7 @@ const layer = Layer.effectDiscard(
     yield* events.project(SessionEvent.Tool.Failed, (event) => run(db, event))
     yield* events.project(SessionEvent.Reasoning.Started, (event) => run(db, event))
     yield* events.project(SessionEvent.Reasoning.Ended, (event) => run(db, event))
-    // yield* events.project(SessionEvent.Retried, (event) => run(db, event))
+    yield* events.project(SessionEvent.RetryScheduled, (event) => run(db, event))
     yield* events.project(SessionEvent.Compaction.Ended, (event) => run(db, event))
     yield* events.project(SessionEvent.RevertEvent.Staged, (event) =>
       db