chore: prepare temporary Frank production validation

CONTEXT.md

@@ -162,12 +162,14 @@ _Avoid_: Response envelope
 - SDK executes Server's assembled `HttpRouter` in memory. It opens no listener and performs no network I/O, while preserving Server routing, middleware, codecs, handlers, and errors.
 - The Effect Client and SDK re-export their decoded datatype facade from Schema so callers do not depend on internal package locations or Core's versioned names.
 - A capability intended for both networked and **Embedded OpenCode** belongs in the authoritative public `HttpApi`; embedded-only same-process capabilities extend **Embedded OpenCode** separately.
-- `sessions.events({ sessionID, after })` is a public durable Session event stream. It verifies the Session, replays durable events after the optional aggregate sequence, continues with newly committed durable events, excludes live-only fragments, and is transported as SSE in both networked and embedded modes.
+- `sessions.events({ sessionID, after })` is one public Session-scoped event stream. It verifies the Session, captures a fixed durable cutoff after registering observation, replays durable events in `(after, cutoff]`, emits one authoritative process-local `session.activity` value, then continues with committed durable events and live-only Session output fragments. Only durable events carry aggregate-sequence cursor metadata.
 - `events.subscribe()` is a distinct public instance-wide live stream for Session and non-Session activity. It has no replay guarantee and includes connection, heartbeat, and instance-disposal lifecycle events; consumers recover from disconnection by refreshing authoritative state.
 - A Session ID is not an optional filter on `events.subscribe()`: instance-wide live events and durable Session events have different schemas, replay guarantees, cursors, lifecycle events, and failure behavior.
 - The initial common OpenCode Client does not expose server-global event aggregation. `events.subscribe()` is bounded to the connected OpenCode instance or workspace; any future cross-instance administrative stream requires a separately designed API.
 - `events.subscribe()` does not automatically reconnect after transport loss. The live-only stream fails with `ClientError`; consumers refresh authoritative state before explicitly opening a new subscription because events missed during disconnection cannot be replayed.
-- `sessions.events({ sessionID, after })` returns the generated HTTP client's cold durable event stream and does not build reconnection policy into the endpoint or client constructor. Transport loss fails the stream with `ClientError`. Callers may compose an explicit resuming stream above it by retaining the last observed durable sequence and opening a new subscription with `after`; any reusable resume helper remains a separate API design question.
+- `sessions.events({ sessionID, after })` returns the generated HTTP client's cold Session event stream and does not build reconnection policy into the endpoint or client constructor. Transport loss fails the stream with `ClientError`. Callers resume with the last observed durable sequence; live-only fragments are not replayed, and every connection receives a fresh authoritative activity value.
+- The Session event stream treats database rows as authoritative and process notifications as bounded, non-blocking wakeups. Before emitting an observed live-only fragment it drains later durable rows, preserving causal durable start boundaries without exposing an additional fence field.
+- Aggregate deletion currently removes the Session's durable event rows and sequence, including deletion history. The Session event stream therefore cannot promise replay after deletion until retention or a typed history-expired outcome is designed.
 - The stable `sessions.list(...)` design returns a **Page** in both networked and **Embedded OpenCode**; embedded execution does not define a separate unbounded array-returning list operation. The beta client currently preserves the existing HTTP `{ data, cursor }` envelope until emitter-level Page projection is implemented.
 - Session list cursors are opaque branded values carrying continuation query and ordering state. Consumers pass them back unchanged and do not inspect storage anchors or encoded filter fields.
 - A Session list continuation accepts only its opaque cursor. Scope, filters, ordering, and page size are fixed by the initial query and carried by that cursor.

packages/client/src/generated/types.ts

@@ -581,7 +581,26 @@ export type SessionsContextOutput = {
           readonly reasoning: number
           readonly cache: { readonly read: number; readonly write: number }
         }
-        readonly error?: { readonly type: "unknown"; readonly message: string }
+        readonly error?:
+          | { readonly type: "unknown"; readonly message: string }
+          | {
+              readonly type: "provider"
+              readonly category:
+                | "invalid-request"
+                | "no-route"
+                | "authentication"
+                | "rate-limit"
+                | "quota-exceeded"
+                | "content-policy"
+                | "provider-internal"
+                | "transport"
+                | "invalid-provider-output"
+                | "unknown"
+              readonly message: string
+              readonly status?: number | null
+              readonly retryable: boolean
+              readonly retryAfterMs?: number | null
+            }
       }
     | {
         readonly type: "compaction"
@@ -601,6 +620,14 @@ export type SessionsEventsInput = {
 }
 
 export type SessionsEventsOutput =
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.activity"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: { readonly sessionID: string; readonly active: boolean }
+    }
   | {
       readonly id: string
       readonly metadata?: { readonly [x: string]: unknown }
@@ -794,7 +821,26 @@ export type SessionsEventsOutput =
         readonly timestamp: number
         readonly sessionID: string
         readonly assistantMessageID: string
-        readonly error: { readonly type: "unknown"; readonly message: string }
+        readonly error:
+          | { readonly type: "unknown"; readonly message: string }
+          | {
+              readonly type: "provider"
+              readonly category:
+                | "invalid-request"
+                | "no-route"
+                | "authentication"
+                | "rate-limit"
+                | "quota-exceeded"
+                | "content-policy"
+                | "provider-internal"
+                | "transport"
+                | "invalid-provider-output"
+                | "unknown"
+              readonly message: string
+              readonly status?: number | undefined
+              readonly retryable: boolean
+              readonly retryAfterMs?: number | undefined
+            }
       }
     }
   | {
@@ -810,6 +856,20 @@ export type SessionsEventsOutput =
         readonly textID: string
       }
     }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.next.text.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly textID: string
+        readonly delta: string
+      }
+    }
   | {
       readonly id: string
       readonly metadata?: { readonly [x: string]: unknown }
@@ -824,6 +884,49 @@ export type SessionsEventsOutput =
         readonly text: string
       }
     }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.next.reasoning.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly reasoningID: string
+        readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.next.reasoning.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly reasoningID: string
+        readonly delta: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.next.reasoning.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly reasoningID: string
+        readonly text: string
+        readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } }
+      }
+    }
   | {
       readonly id: string
       readonly metadata?: { readonly [x: string]: unknown }
@@ -838,6 +941,20 @@ export type SessionsEventsOutput =
         readonly name: string
       }
     }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.next.tool.input.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly delta: string
+      }
+    }
   | {
       readonly id: string
       readonly metadata?: { readonly [x: string]: unknown }
@@ -932,35 +1049,6 @@ export type SessionsEventsOutput =
         }
       }
     }
-  | {
-      readonly id: string
-      readonly metadata?: { readonly [x: string]: unknown }
-      readonly type: "session.next.reasoning.started"
-      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
-      readonly location?: { readonly directory: string; readonly workspaceID?: string }
-      readonly data: {
-        readonly timestamp: number
-        readonly sessionID: string
-        readonly assistantMessageID: string
-        readonly reasoningID: string
-        readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } }
-      }
-    }
-  | {
-      readonly id: string
-      readonly metadata?: { readonly [x: string]: unknown }
-      readonly type: "session.next.reasoning.ended"
-      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
-      readonly location?: { readonly directory: string; readonly workspaceID?: string }
-      readonly data: {
-        readonly timestamp: number
-        readonly sessionID: string
-        readonly assistantMessageID: string
-        readonly reasoningID: string
-        readonly text: string
-        readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } }
-      }
-    }
   | {
       readonly id: string
       readonly metadata?: { readonly [x: string]: unknown }
@@ -994,6 +1082,19 @@ export type SessionsEventsOutput =
         readonly reason: "auto" | "manual"
       }
     }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown }
+      readonly type: "session.next.compaction.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number }
+      readonly location?: { readonly directory: string; readonly workspaceID?: string }
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly text: string
+      }
+    }
   | {
       readonly id: string
       readonly metadata?: { readonly [x: string]: unknown }
@@ -1198,7 +1299,26 @@ export type SessionsMessageOutput = {
           readonly reasoning: number
           readonly cache: { readonly read: number; readonly write: number }
         }
-        readonly error?: { readonly type: "unknown"; readonly message: string }
+        readonly error?:
+          | { readonly type: "unknown"; readonly message: string }
+          | {
+              readonly type: "provider"
+              readonly category:
+                | "invalid-request"
+                | "no-route"
+                | "authentication"
+                | "rate-limit"
+                | "quota-exceeded"
+                | "content-policy"
+                | "provider-internal"
+                | "transport"
+                | "invalid-provider-output"
+                | "unknown"
+              readonly message: string
+              readonly status?: number | null
+              readonly retryable: boolean
+              readonly retryAfterMs?: number | null
+            }
       }
     | {
         readonly type: "compaction"

packages/client/test/effect.test.ts

@@ -22,7 +22,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
       return Effect.succeed(
         HttpClientResponse.fromWeb(
           request,
-          new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\n`, {
+          new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(activityEvent)}\n\n`, {
             headers: { "content-type": "text/event-stream" },
           }),
         ),
@@ -93,6 +93,8 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
   expect(DateTime.toEpochMillis(result.admitted.timeCreated)).toBe(1_717_171_717_000)
   expect(result.context).toEqual([])
   expect(DateTime.toEpochMillis(result.events[0].data.timestamp)).toBe(1_717_171_717_000)
+  expect(result.events[1]).toEqual(activityEvent)
+  expect(result.events[1]).not.toHaveProperty("durable")
   expect(result.message).toEqual(expect.objectContaining({ id: "msg_model", type: "model-switched" }))
 })
 
@@ -145,3 +147,9 @@ const modelSwitchedEvent = {
     model: { id: "claude", providerID: "anthropic" },
   },
 }
+
+const activityEvent = {
+  id: "evt_activity",
+  type: "session.activity" as const,
+  data: { sessionID: "ses_test", active: false },
+}

packages/client/test/promise.test.ts

@@ -25,9 +25,12 @@ test("session methods use the public HTTP contract", async () => {
       const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url
       requests.push({ url, init })
       if (url.includes("/event")) {
-        return new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\n`, {
-          headers: { "content-type": "text/event-stream" },
-        })
+        return new Response(
+          `data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(activityEvent)}\n\n`,
+          {
+            headers: { "content-type": "text/event-stream" },
+          },
+        )
       }
       if (url.includes("/prompt")) return Response.json(admission)
       if (url.includes("/context")) return Response.json({ data: [] })
@@ -65,7 +68,8 @@ test("session methods use the public HTTP contract", async () => {
   expect(created.id).toBe("ses_test")
   expect(admitted.id).toBe("msg_test")
   expect(context).toEqual([])
-  expect(events).toEqual([modelSwitchedEvent])
+  expect(events).toEqual([modelSwitchedEvent, activityEvent])
+  expect(events[1]).not.toHaveProperty("durable")
   expect(message).toEqual(modelSwitchedMessage)
   expect(requests.map((request) => [request.init?.method, request.url])).toEqual([
     ["GET", "http://localhost:3000/api/session?limit=10&order=desc"],
@@ -153,3 +157,9 @@ const modelSwitchedEvent = {
     model: { id: "claude", providerID: "anthropic" },
   },
 }
+
+const activityEvent = {
+  id: "evt_activity",
+  type: "session.activity",
+  data: { sessionID: "ses_test", active: false },
+}

packages/core/src/event.ts

@@ -1,9 +1,9 @@
 export * as EventV2 from "./event"
 
-import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
+import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Scope, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
-import { and, asc, eq, gt } from "drizzle-orm"
+import { and, asc, eq, gt, lte } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
@@ -67,6 +67,19 @@ export interface Interface {
   readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
   readonly all: () => Stream.Stream<Payload>
   readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream<Payload>
+  readonly observeAggregate: (input: {
+    readonly aggregateID: string
+    readonly after?: number
+    readonly live: (event: Payload) => boolean
+  }) => Effect.Effect<
+    {
+      readonly replay: ReadonlyArray<Payload>
+      readonly updates: Stream.Stream<Payload>
+      readonly offer: (event: Payload, position?: "after" | "before") => boolean
+    },
+    never,
+    Scope.Scope
+  >
   /** @deprecated Use `all()` and consume the returned stream. */
   readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
   readonly project: <D extends Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
@@ -94,12 +107,12 @@ export const layerWith = (options?: LayerOptions) =>
     Effect.gen(function* () {
       const pubsub = {
         all: yield* PubSub.unbounded<Payload>(),
-        durable: new Map<string, Set<PubSub.PubSub<void>>>(),
+        durable: new Map<string, Set<PubSub.PubSub<number>>>(),
         typed: new Map<string, PubSub.PubSub<Payload>>(),
       }
       const projectors = new Map<string, Subscriber[]>()
       // TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads.
-      const listeners = new Array<Subscriber>()
+      const listeners = new Set<Subscriber>()
       const { db } = yield* Database.Service
 
       const getOrCreate = (definition: Definition) =>
@@ -275,7 +288,7 @@ export const layerWith = (options?: LayerOptions) =>
                   if (committed) {
                     yield* Effect.forEach(
                       pubsub.durable.get(committed.aggregateID) ?? [],
-                      (wake) => PubSub.publish(wake, undefined),
+                      (wake) => PubSub.publish(wake, committed.seq),
                       { discard: true },
                     )
                   }
@@ -472,13 +485,19 @@ export const layerWith = (options?: LayerOptions) =>
         }
       }
 
-      const readAfter = (aggregateID: string, after: number) =>
+      const readAfter = (aggregateID: string, after: number, through?: number) =>
         (options?.beforeAggregateRead?.(aggregateID) ?? Effect.void).pipe(
           Effect.andThen(
             db
               .select()
               .from(EventTable)
-              .where(and(eq(EventTable.aggregate_id, aggregateID), gt(EventTable.seq, after)))
+              .where(
+                and(
+                  eq(EventTable.aggregate_id, aggregateID),
+                  gt(EventTable.seq, after),
+                  through === undefined ? undefined : lte(EventTable.seq, through),
+                ),
+              )
               .orderBy(asc(EventTable.seq))
               .all(),
           ),
@@ -498,7 +517,7 @@ export const layerWith = (options?: LayerOptions) =>
 
       const subscribeDurable = (aggregateID: string) =>
         Effect.gen(function* () {
-          const wake = yield* PubSub.sliding<void>(1)
+          const wake = yield* PubSub.sliding<number>(1)
           const subscription = yield* PubSub.subscribe(wake)
           yield* Effect.acquireRelease(
             Effect.sync(() => {
@@ -516,34 +535,83 @@ export const layerWith = (options?: LayerOptions) =>
           return subscription
         })
 
+      const aggregateDrain = (
+        aggregateID: string,
+        after: number,
+      ): ((through?: number) => Effect.Effect<ReadonlyArray<Payload>>) => {
+        let sequence = after
+        return (through?: number) =>
+          through !== undefined && through <= sequence
+            ? Effect.succeed<ReadonlyArray<Payload>>([])
+            : Effect.suspend(() => readAfter(aggregateID, sequence, through)).pipe(
+                Effect.tap((events) =>
+                  Effect.sync(() => {
+                    sequence = events.at(-1)?.durable?.seq ?? sequence
+                  }),
+                ),
+              )
+      }
+
       const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream<Payload> =>
         Stream.unwrap(
           Effect.gen(function* () {
             const wakes = yield* subscribeDurable(input.aggregateID)
-            let sequence = input.after ?? -1
-            const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
-              Effect.tap((events) =>
-                Effect.sync(() => {
-                  sequence = events.at(-1)?.durable?.seq ?? sequence
-                }),
-              ),
-            )
-            const historical = yield* read
-            const live = Stream.fromSubscription(wakes).pipe(
-              Stream.mapEffect(() => read),
-              Stream.flattenIterable,
-            )
+            const drain = aggregateDrain(input.aggregateID, input.after ?? -1)
+            const historical = yield* drain()
+            const live = Stream.fromSubscription(wakes).pipe(Stream.mapEffect(drain), Stream.flattenIterable)
             return Stream.concat(Stream.fromIterable(historical), live)
           }),
         )
 
+      const observeAggregate: Interface["observeAggregate"] = (input) =>
+        Effect.gen(function* () {
+          type Signal =
+            | { readonly _tag: "durable" }
+            | { readonly _tag: "transient"; readonly event: Payload; readonly position: "after" | "before" }
+          const signals = yield* Queue.dropping<Signal, Cause.Done>(256)
+          const wakes = yield* subscribeDurable(input.aggregateID)
+          let durableQueued = false
+          let durableThrough = -1
+          const offer = (event: Payload, position: "after" | "before" = "after") => {
+            const offered = Queue.offerUnsafe(signals, { _tag: "transient", event, position })
+            if (!offered) Queue.endUnsafe(signals)
+            return offered
+          }
+          const unsubscribe = yield* listen((event) =>
+            Effect.sync(() => {
+              if (input.live(event)) offer(event)
+            }),
+          )
+          yield* Effect.addFinalizer(() => unsubscribe.pipe(Effect.andThen(Queue.shutdown(signals))))
+          yield* Stream.runForEach(Stream.fromSubscription(wakes), (sequence) =>
+            Effect.sync(() => {
+              durableThrough = Math.max(durableThrough, sequence)
+              if (durableQueued) return
+              durableQueued = Queue.offerUnsafe(signals, { _tag: "durable" })
+            }),
+          ).pipe(Effect.forkScoped)
+
+          const cutoff = yield* latestSequence(db, input.aggregateID)
+          const replay = yield* readAfter(input.aggregateID, input.after ?? -1, cutoff)
+          const drain = aggregateDrain(input.aggregateID, cutoff)
+          const updates = Stream.fromQueue(signals).pipe(
+            Stream.mapEffect((signal) => {
+              if (signal._tag === "durable") {
+                durableQueued = false
+                return drain(durableThrough)
+              }
+              if (signal.position === "before") return Effect.succeed([signal.event])
+              return drain().pipe(Effect.map((events) => [...events, signal.event]))
+            }),
+            Stream.flattenIterable,
+          )
+          return { replay, updates, offer }
+        })
+
       const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
         Effect.sync(() => {
-          listeners.push(listener)
-          return Effect.sync(() => {
-            const index = listeners.indexOf(listener)
-            if (index >= 0) listeners.splice(index, 1)
-          })
+          listeners.add(listener)
+          return Effect.sync(() => listeners.delete(listener)).pipe(Effect.asVoid)
         })
 
       const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
@@ -558,6 +626,7 @@ export const layerWith = (options?: LayerOptions) =>
         subscribe,
         all: streamAll,
         durable,
+        observeAggregate,
         listen,
         project,
         replay,

packages/core/src/session.ts

@@ -128,7 +128,7 @@ export interface Interface {
   readonly events: (input: {
     sessionID: SessionSchema.ID
     after?: number
-  }) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
+  }) => Stream.Stream<SessionEvent.StreamEvent, NotFoundError>
   readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect<void, NotFoundError>
   readonly switchModel: (input: {
     sessionID: SessionSchema.ID
@@ -185,7 +185,11 @@ export const layer = Layer.unwrap(
           const store = yield* SessionStore.Service
           const locations = yield* LocationServiceMap
           const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
-          const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
+          const sessionEventTypes = new Set<string>(SessionEvent.Definitions.map((definition) => definition.type))
+          const isSessionEvent = (event: EventV2.Payload): event is SessionEvent.Event =>
+            sessionEventTypes.has(event.type)
+          const isDurableSessionEvent = (event: EventV2.Payload): event is SessionEvent.DurableEvent =>
+            event.durable !== undefined && isSessionEvent(event)
           const decode = (row: typeof SessionMessageTable.$inferSelect) =>
             decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
               Effect.mapError(
@@ -341,10 +345,34 @@ export const layer = Layer.unwrap(
             }),
             events: (input) =>
               Stream.unwrap(
-                result
-                  .get(input.sessionID)
-                  .pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
-              ).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
+                Effect.gen(function* () {
+                  yield* result.get(input.sessionID)
+                  const activity = yield* execution.activity(input.sessionID)
+                  const observed = yield* events.observeAggregate({
+                    aggregateID: input.sessionID,
+                    after: input.after,
+                    live: (event) =>
+                      event.durable === undefined && isSessionEvent(event) && event.data.sessionID === input.sessionID,
+                  })
+                  const initialActivity = SessionEvent.makeActivity(
+                    input.sessionID,
+                    yield* activity.attach((active) =>
+                      observed.offer(SessionEvent.makeActivity(input.sessionID, active), active ? "before" : "after"),
+                    ),
+                  )
+                  return Stream.fromIterable(observed.replay.filter(isDurableSessionEvent)).pipe(
+                    Stream.concat(Stream.make(initialActivity)),
+                    Stream.concat(
+                      observed.updates.pipe(
+                        Stream.filter(
+                          (event): event is SessionEvent.StreamEvent =>
+                            event.type === SessionEvent.Activity.type || isSessionEvent(event),
+                        ),
+                      ),
+                    ),
+                  )
+                }),
+              ),
             prompt: Effect.fn("V2Session.prompt")((input) =>
               Effect.uninterruptible(
                 Effect.gen(function* () {

packages/core/src/session/execution.ts

@@ -1,12 +1,15 @@
 export * as SessionExecution from "./execution"
 
-import { Context, Effect, Layer } from "effect"
+import { Context, Effect, Layer, Scope } from "effect"
 import { SessionRunner } from "./runner/index"
+import { SessionRunCoordinator } from "./run-coordinator"
 import { SessionSchema } from "./schema"
 
 export interface Interface {
   /** Snapshots active execution owned by this process. */
   readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
+  /** Observes foreground ownership with an authoritative initial snapshot. */
+  readonly activity: (sessionID: SessionSchema.ID) => Effect.Effect<SessionRunCoordinator.Activity, never, Scope.Scope>
   /** Starts execution while idle or joins the active execution. */
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
   /** Registers newly recorded work. Repeated wakeups may coalesce. */
@@ -23,6 +26,7 @@ export const noopLayer = Layer.succeed(
   Service,
   Service.of({
     active: Effect.succeed(new Set()),
+    activity: () => Effect.succeed({ attach: () => Effect.succeed(false) }),
     resume: () => Effect.void,
     wake: () => Effect.void,
     interrupt: () => Effect.void,

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

@@ -29,6 +29,7 @@ export const layer = Layer.effect(
 
     return SessionExecution.Service.of({
       active: coordinator.active,
+      activity: coordinator.activity,
       interrupt: coordinator.interrupt,
       resume: coordinator.run,
       wake: coordinator.wake,

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

@@ -2,10 +2,17 @@ export * as SessionRunCoordinator from "./run-coordinator"
 
 import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
 
+export interface Activity {
+  /** Installs a synchronous observer and snapshots current ownership atomically. */
+  readonly attach: (observer: (active: boolean) => void) => Effect.Effect<boolean>
+}
+
 /** Serializes execution for each key while allowing different keys to run concurrently. */
 export interface Coordinator<Key, E> {
   /** Snapshots keys with an execution owned by this coordinator. */
   readonly active: Effect.Effect<ReadonlySet<Key>>
+  /** Registers transition observation before taking its authoritative snapshot. */
+  readonly activity: (key: Key) => Effect.Effect<Activity, never, Scope.Scope>
   /** Starts execution while idle or joins the active execution. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Registers one coalesced follow-up after newly recorded work. */
@@ -26,6 +33,7 @@ export const make = <Key, E>(options: {
 }): Effect.Effect<Coordinator<Key, E>, never, Scope.Scope> =>
   Effect.gen(function* () {
     const active = new Map<Key, Entry<E>>()
+    const activityObservers = new Map<Key, Set<(active: boolean) => void>>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
 
     const makeEntry = (): Entry<E> => ({
@@ -34,6 +42,10 @@ export const make = <Key, E>(options: {
       stopping: false,
     })
 
+    const notifyActivity = (key: Key, value: boolean) => {
+      for (const observer of activityObservers.get(key) ?? []) observer(value)
+    }
+
     const start = (key: Key, entry: Entry<E>, force: boolean, successor = false) => {
       const ready = Deferred.makeUnsafe<void>()
       const owner = fork(
@@ -56,8 +68,10 @@ export const make = <Key, E>(options: {
       }
 
       const successor = entry.pendingWake ? makeEntry() : undefined
-      if (successor === undefined) active.delete(key)
-      else {
+      if (successor === undefined) {
+        active.delete(key)
+        notifyActivity(key, false)
+      } else {
         active.set(key, successor)
         start(key, successor, false, true)
       }
@@ -74,6 +88,7 @@ export const make = <Key, E>(options: {
 
         const next = makeEntry()
         active.set(key, next)
+        notifyActivity(key, true)
         start(key, next, true)
         return restore(Deferred.await(next.done))
       })
@@ -88,6 +103,7 @@ export const make = <Key, E>(options: {
 
         const next = makeEntry()
         active.set(key, next)
+        notifyActivity(key, true)
         start(key, next, false)
       })
 
@@ -100,5 +116,33 @@ export const make = <Key, E>(options: {
         return Fiber.interrupt(entry.owner)
       })
 
-    return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt }
+    const activity = (key: Key) =>
+      Effect.gen(function* () {
+        let attached: ((active: boolean) => void) | undefined
+        const observer = (value: boolean) => {
+          attached?.(value)
+        }
+        yield* Effect.acquireRelease(
+          Effect.sync(() => {
+            const observers = activityObservers.get(key) ?? new Set()
+            observers.add(observer)
+            activityObservers.set(key, observers)
+          }),
+          () =>
+            Effect.sync(() => {
+              const observers = activityObservers.get(key)
+              observers?.delete(observer)
+              if (observers?.size === 0) activityObservers.delete(key)
+            }),
+        )
+        return {
+          attach: (next: (active: boolean) => void) =>
+            Effect.sync(() => {
+              attached = next
+              return active.has(key)
+            }),
+        }
+      })
+
+    return { active: Effect.sync(() => new Set(active.keys())), activity, run, wake, interrupt }
   })

packages/core/src/session/runner/llm.ts

@@ -6,6 +6,7 @@ import {
   Message,
   SystemPart,
   isContextOverflowFailure,
+  type LLMErrorReason,
   type ProviderErrorEvent,
 } from "@opencode-ai/llm"
 import { Cause, DateTime, Effect, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
@@ -32,7 +33,7 @@ import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { type RunError, Service } from "./index"
 import { SessionRunnerModel } from "./model"
-import { createLLMEventPublisher } from "./publish-llm-event"
+import { createLLMEventPublisher, providerError } from "./publish-llm-event"
 import { toLLMMessages } from "./to-llm-message"
 import { MAX_STEPS_PROMPT } from "./max-steps"
 import { Snapshot } from "../../snapshot"
@@ -87,6 +88,29 @@ import { Snapshot } from "../../snapshot"
  * explicit loop starts the next provider turn after local settlement. Configured agent step limits bound the loop.
  */
 
+const providerCategories = {
+  InvalidRequest: "invalid-request",
+  NoRoute: "no-route",
+  Authentication: "authentication",
+  RateLimit: "rate-limit",
+  QuotaExceeded: "quota-exceeded",
+  ContentPolicy: "content-policy",
+  ProviderInternal: "provider-internal",
+  Transport: "transport",
+  InvalidProviderOutput: "invalid-provider-output",
+  UnknownProvider: "unknown",
+} as const satisfies Record<LLMErrorReason["_tag"], Parameters<typeof providerError>[0]["category"]>
+
+const toSessionProviderError = (reason: LLMErrorReason) =>
+  providerError({
+    category: providerCategories[reason._tag],
+    status:
+      ("http" in reason ? reason.http?.response?.status : undefined) ??
+      ("status" in reason ? reason.status : undefined),
+    retryable: reason.retryable,
+    retryAfterMs: "retryAfterMs" in reason ? reason.retryAfterMs : undefined,
+  })
+
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
@@ -283,7 +307,7 @@ export const layer = Layer.effect(
           const llmFailure = failure instanceof LLMError ? failure : undefined
           if (llmFailure && !publisher.hasProviderError()) {
             yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true))
-            yield* withPublication(publisher.failAssistant(llmFailure.reason.message))
+            yield* withPublication(publisher.failAssistant(toSessionProviderError(llmFailure.reason)))
           }
           if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers)
           const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit)
@@ -299,7 +323,7 @@ export const layer = Layer.effect(
             yield* FiberSet.clear(toolFibers)
             yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted"))
             if (publisher.hasActiveAssistant())
-              yield* withPublication(publisher.failAssistant("Provider turn interrupted"))
+              yield* withPublication(publisher.failAssistant({ type: "unknown", message: "Provider turn interrupted" }))
           }
           if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) {
             const failure = Cause.squash(settled.cause)

packages/core/src/session/runner/model.ts

@@ -3,7 +3,7 @@ export * as SessionRunnerModel from "./model"
 import { type Model } from "@opencode-ai/llm"
 import * as AnthropicMessages from "@opencode-ai/llm/protocols/anthropic-messages"
 import * as OpenAICompatibleChat from "@opencode-ai/llm/protocols/openai-compatible-chat"
-import * as OpenAIResponses from "@opencode-ai/llm/protocols/openai-responses"
+import * as OpenAI from "@opencode-ai/llm/providers/openai"
 import { Auth, type AnyRoute } from "@opencode-ai/llm/route"
 import { Context, Effect, Layer, Schema } from "effect"
 import { produce } from "immer"
@@ -139,8 +139,12 @@ export const fromCatalogModel = (
         })
   const key = apiKey(resolved, credential)
   if (resolved.api.type === "aisdk" && resolved.api.package === "@ai-sdk/openai") {
+    const store = resolved.request.body.store
+    const route = OpenAI.configure({
+      providerOptions: typeof store === "boolean" ? { openai: { store } } : undefined,
+    }).responses(resolved.api.id).route
     return Effect.succeed(
-      withDefaults(resolved, OpenAIResponses.route)
+      withDefaults(resolved, route)
         .with({ auth: key === undefined ? Auth.none : Auth.bearer(key) })
         .model({ id: resolved.api.id }),
     )

packages/core/src/session/runner/publish-llm-event.ts

@@ -50,6 +50,33 @@ const settledOutput = (value: ToolOutput | undefined, result: ToolResultValue):
   return { structured: record(settled.structured), content: settled.content }
 }
 
+const providerMessages: Record<SessionMessage.ProviderErrorCategory, string> = {
+  "invalid-request": "Provider rejected the request",
+  "no-route": "Provider unavailable: no provider route",
+  authentication: "Provider authentication failed",
+  "rate-limit": "Provider rate limit exceeded",
+  "quota-exceeded": "Provider quota exceeded",
+  "content-policy": "Provider rejected the invalid request due to content policy",
+  "provider-internal": "Provider service unavailable",
+  transport: "Provider connection failed",
+  "invalid-provider-output": "Provider returned an invalid response",
+  unknown: "Provider request failed",
+}
+
+export const providerError = (input: {
+  readonly category: SessionMessage.ProviderErrorCategory
+  readonly status?: number
+  readonly retryable?: boolean
+  readonly retryAfterMs?: number
+}): SessionMessage.ProviderError => ({
+  type: "provider",
+  category: input.category,
+  message: providerMessages[input.category],
+  status: input.status,
+  retryable: input.retryable ?? (input.category === "rate-limit" || input.category === "provider-internal"),
+  retryAfterMs: input.retryAfterMs,
+})
+
 /** Persist one provider turn without executing tools or starting a continuation turn. */
 export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) => {
   const tools = new Map<
@@ -67,7 +94,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
   const timestamp = DateTime.now
   let assistantMessageID: SessionMessage.ID | undefined
   let assistantActive = false
-  let assistantFailed = false
   let providerFailed = false
   let stepSettlement: { readonly finish: string; readonly tokens: ReturnType<typeof tokens> } | undefined
 
@@ -196,17 +222,17 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
     yield* flushFragments()
   })
 
-  const failAssistant = Effect.fnUntraced(function* (message: string) {
-    if (assistantFailed) return
+  const failAssistant = Effect.fnUntraced(function* (error: SessionMessage.Error) {
+    if (providerFailed) return
+    providerFailed = true
     yield* flush()
     const assistantMessageID = yield* startAssistant()
     assistantActive = false
-    assistantFailed = true
     yield* events.publish(SessionEvent.Step.Failed, {
       sessionID: input.sessionID,
       timestamp: yield* timestamp,
       assistantMessageID,
-      error: { type: "unknown", message },
+      error,
     })
   })
 
@@ -402,8 +428,13 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
       case "finish":
         return
       case "provider-error":
-        providerFailed = true
-        yield* failAssistant(event.message)
+        yield* failAssistant(
+          providerError({
+            category: event.category ?? (event.classification === "context-overflow" ? "invalid-request" : "unknown"),
+            status: event.status,
+            retryable: event.retryable,
+          }),
+        )
         return
     }
   })

packages/core/test/event.test.ts

@@ -418,6 +418,127 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("observes a fixed replay cutoff and delivers commits during replay as updates", () =>
+    Effect.gen(function* () {
+      const readStarted = yield* Deferred.make<void>()
+      const continueRead = yield* Deferred.make<void>()
+      let pause = true
+      const eventLayer = EventV2.layerWith({
+        beforeAggregateRead: () =>
+          pause
+            ? Deferred.succeed(readStarted, undefined).pipe(Effect.andThen(Deferred.await(continueRead)))
+            : Effect.void,
+      }).pipe(Layer.provide(Database.defaultLayer))
+
+      yield* Effect.gen(function* () {
+        const events = yield* EventV2.Service
+        const aggregateID = Session.ID.create()
+        const observer = yield* events.observeAggregate({ aggregateID, live: () => false }).pipe(Effect.forkScoped)
+        yield* Deferred.await(readStarted)
+
+        pause = false
+        yield* events.publish(DurableMessage, durableData(aggregateID, "after cutoff"))
+        yield* Deferred.succeed(continueRead, undefined)
+        const observed = yield* Fiber.join(observer)
+        const update = yield* observed.updates.pipe(Stream.take(1), Stream.runCollect)
+
+        expect(observed.replay).toEqual([])
+        expect(Array.from(update).map((event) => [event.durable?.seq, event.data])).toEqual([
+          [0, durableData(aggregateID, "after cutoff")],
+        ])
+      }).pipe(Effect.provide(Layer.mergeAll(Database.defaultLayer, eventLayer)))
+    }),
+  )
+
+  it.effect("observes replay commits that are not republished", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = Session.ID.create()
+      const observed = yield* events.observeAggregate({ aggregateID, live: () => false })
+      const update = yield* observed.updates.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+
+      yield* events.replay(
+        {
+          id: EventV2.ID.create(),
+          aggregateID,
+          seq: 0,
+          type: EventV2.versionedType(DurableMessage.type, 1),
+          data: durableData(aggregateID, "replayed"),
+        },
+        { publish: false },
+      )
+
+      expect(Array.from(yield* Fiber.join(update)).map((event) => event.data)).toEqual([
+        durableData(aggregateID, "replayed"),
+      ])
+    }),
+  )
+
+  it.effect("drains causally preceding durable rows before a live payload", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = Session.ID.create()
+      const observed = yield* events.observeAggregate({
+        aggregateID,
+        live: (event) => event.type === Message.type,
+      })
+      const updates = yield* observed.updates.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+
+      yield* events.publish(DurableMessage, durableData(aggregateID, "start"))
+      const live = yield* events.publish(Message, { text: "delta" })
+
+      expect(Array.from(yield* Fiber.join(updates))).toEqual([
+        expect.objectContaining({ durable: expect.objectContaining({ seq: 0 }) }),
+        live,
+      ])
+    }),
+  )
+
+  it.effect("coalesces saturated observer signals without losing durable rows", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = Session.ID.create()
+      const count = 300
+      const observed = yield* events.observeAggregate({ aggregateID, live: () => false })
+
+      for (let index = 0; index < count; index++) {
+        yield* events.publish(DurableMessage, durableData(aggregateID, String(index)))
+      }
+      const updates = yield* observed.updates.pipe(Stream.take(count), Stream.runCollect)
+
+      expect(Array.from(updates, (event) => event.durable?.seq)).toEqual(
+        Array.from({ length: count }, (_, index) => index),
+      )
+    }),
+  )
+
+  it.effect("coalesces durable notifications without repeated empty reads", () =>
+    Effect.gen(function* () {
+      let reads = 0
+      const eventLayer = EventV2.layerWith({
+        beforeAggregateRead: () =>
+          Effect.sync(() => {
+            reads++
+          }),
+      }).pipe(Layer.provide(Database.defaultLayer))
+
+      yield* Effect.gen(function* () {
+        const events = yield* EventV2.Service
+        const aggregateID = Session.ID.create()
+        const count = 20
+        const observed = yield* events.observeAggregate({ aggregateID, live: () => false })
+
+        for (let index = 0; index < count; index++) {
+          yield* events.publish(DurableMessage, durableData(aggregateID, String(index)))
+        }
+        const updates = yield* observed.updates.pipe(Stream.take(count), Stream.runCollect)
+
+        expect(updates).toHaveLength(count)
+        expect(reads).toBe(2)
+      }).pipe(Effect.provide(Layer.mergeAll(Database.defaultLayer, eventLayer)))
+    }),
+  )
+
   it.effect("coalesces durable aggregate wakes while draining every committed event", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service

packages/core/test/session-prompt.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { DateTime, Effect, Fiber, Layer, Stream } from "effect"
+import { DateTime, Deferred, Effect, Fiber, Layer, Stream } from "effect"
 import { eq } from "drizzle-orm"
 import { Database } from "@opencode-ai/core/database/database"
 import { EventV2 } from "@opencode-ai/core/event"
@@ -23,10 +23,29 @@ const executionCalls: SessionV2.ID[] = []
 const interruptCalls: SessionV2.ID[] = []
 const wakeCalls: SessionV2.ID[] = []
 const activeSessions = new Set<SessionV2.ID>()
+let activityObserver: ((active: boolean) => void) | undefined
+let activityState = false
+const setActivity = (active: boolean) =>
+  Effect.sync(() => {
+    activityState = active
+    activityObserver?.(active)
+  })
 const execution = Layer.succeed(
   SessionExecution.Service,
   SessionExecution.Service.of({
     active: Effect.sync(() => new Set(activeSessions)),
+    activity: () =>
+      Effect.sync(() => {
+        activityState = false
+        activityObserver = undefined
+        return {
+          attach: (observer: (active: boolean) => void) =>
+            Effect.sync(() => {
+              activityObserver = observer
+              return activityState
+            }),
+        }
+      }),
     resume: (sessionID) =>
       Effect.sync(() => {
         executionCalls.push(sessionID)
@@ -179,7 +198,7 @@ describe("SessionV2.prompt", () => {
       const session = yield* SessionV2.Service
       const events = yield* EventV2.Service
       const { db } = yield* Database.Service
-      const fiber = yield* session.events({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
+      const fiber = yield* session.events({ sessionID }).pipe(Stream.take(5), Stream.runCollect, Effect.forkScoped)
       yield* Effect.yieldNow
 
       yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false })
@@ -188,6 +207,7 @@ describe("SessionV2.prompt", () => {
       const streamed = Array.from(yield* Fiber.join(fiber))
 
       expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([
+        [undefined, "session.activity"],
         [0, "session.next.prompt.admitted"],
         [1, "session.next.prompt.admitted"],
         [2, "session.next.prompted"],
@@ -196,10 +216,122 @@ describe("SessionV2.prompt", () => {
       expect(
         Array.from(
           yield* session
-            .events({ sessionID, after: streamed[0]!.durable?.seq })
-            .pipe(Stream.take(1), Stream.runCollect),
+            .events({ sessionID, after: streamed[2]?.durable?.seq })
+            .pipe(Stream.take(3), Stream.runCollect),
         ).map((event) => [event.durable?.seq, event.type]),
-      ).toEqual([[1, "session.next.prompt.admitted"]])
+      ).toEqual([
+        [2, "session.next.prompted"],
+        [3, "session.next.prompted"],
+        [undefined, "session.activity"],
+      ])
+    }),
+  )
+
+  it.effect("replays history, emits activity, then fences live deltas behind durable starts", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      const assistantMessageID = SessionMessage.ID.create()
+      yield* events.publish(SessionEvent.Text.Started, {
+        sessionID,
+        assistantMessageID,
+        textID: "text-1",
+        timestamp: yield* DateTime.now,
+      })
+      const streamed = yield* session.events({ sessionID }).pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* events.publish(SessionEvent.Text.Delta, {
+        sessionID,
+        assistantMessageID,
+        textID: "text-1",
+        delta: "hello",
+        timestamp: yield* DateTime.now,
+      })
+
+      expect(Array.from(yield* Fiber.join(streamed)).map((event) => event.type)).toEqual([
+        "session.next.text.started",
+        "session.activity",
+        "session.next.text.delta",
+      ])
+    }),
+  )
+
+  it.effect("orders activity around the durable rows owned by a run", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      const assistantMessageID = SessionMessage.ID.create()
+      const streamed = yield* session.events({ sessionID }).pipe(Stream.take(5), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+
+      yield* setActivity(true)
+      yield* events.publish(SessionEvent.Text.Started, {
+        sessionID,
+        assistantMessageID,
+        textID: "text-activity",
+        timestamp: yield* DateTime.now,
+      })
+      yield* events.publish(SessionEvent.Text.Ended, {
+        sessionID,
+        assistantMessageID,
+        textID: "text-activity",
+        text: "done",
+        timestamp: yield* DateTime.now,
+      })
+      yield* setActivity(false)
+
+      expect(Array.from(yield* Fiber.join(streamed)).map((event) => [event.type, event.data])).toEqual([
+        ["session.activity", { sessionID, active: false }],
+        ["session.activity", { sessionID, active: true }],
+        ["session.next.text.started", expect.objectContaining({ sessionID })],
+        ["session.next.text.ended", expect.objectContaining({ sessionID })],
+        ["session.activity", { sessionID, active: false }],
+      ])
+    }),
+  )
+
+  it.effect("terminates the Session stream when live event buffering saturates", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      const initial = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      let seenInitial = false
+      const streamed = yield* session.events({ sessionID }).pipe(
+        Stream.mapEffect((event) => {
+          if (!seenInitial) {
+            seenInitial = true
+            return Deferred.succeed(initial, undefined).pipe(Effect.as(event))
+          }
+          return Deferred.await(release).pipe(Effect.as(event))
+        }),
+        Stream.runCollect,
+        Effect.forkScoped,
+      )
+      yield* Deferred.await(initial)
+
+      const timestamp = yield* DateTime.now
+      const assistantMessageID = SessionMessage.ID.create()
+      yield* Effect.forEach(
+        Array.from({ length: 1_000 }, (_, index) => index),
+        (index) =>
+          events.publish(SessionEvent.Text.Delta, {
+            sessionID,
+            assistantMessageID,
+            textID: "text-saturation",
+            delta: String(index),
+            timestamp,
+          }),
+        { concurrency: "unbounded", discard: true },
+      )
+      yield* Deferred.succeed(release, undefined)
+
+      const result = Array.from(yield* Fiber.join(streamed))
+      expect(result[0]?.type).toBe("session.activity")
+      expect(result.length).toBeLessThan(1_001)
     }),
   )
 

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

@@ -99,6 +99,27 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("reflects activity races in the snapshot or subsequent transitions", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const gate = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate))),
+        })
+        const activity = yield* coordinator.activity("session")
+
+        const run = yield* coordinator.run("session").pipe(Effect.forkChild)
+        yield* Deferred.await(started)
+        const transitions = new Array<boolean>()
+        expect(yield* activity.attach((active) => transitions.push(active))).toBeTrue()
+        yield* Deferred.succeed(gate, undefined)
+        yield* Fiber.join(run)
+        expect(transitions).toEqual([false])
+      }),
+    ),
+  )
+
   it.effect("cleans active executions after failure and defect", () =>
     Effect.scoped(
       Effect.gen(function* () {

packages/core/test/session-runner-model.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect } from "bun:test"
-import { LLM } from "@opencode-ai/llm"
+import { LLM, Message, ToolCallPart, ToolResultPart } from "@opencode-ai/llm"
+import * as OpenAIResponses from "@opencode-ai/llm/protocols/openai-responses"
 import { LLMClient } from "@opencode-ai/llm/route"
 import { DateTime, Effect } from "effect"
 import { Headers } from "effect/unstable/http"
@@ -72,6 +73,82 @@ describe("SessionRunnerModel", () => {
     }),
   )
 
+  it.effect("replays encrypted reasoning for stateless catalog OpenAI tool continuations", () =>
+    Effect.gen(function* () {
+      const catalog = ModelV2.Info.make({
+        ...model({ type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }),
+        api: {
+          type: "aisdk",
+          package: "@ai-sdk/openai",
+          id: ModelV2.ID.make("custom-reasoning-model"),
+          url: "https://openai.example/v1",
+        },
+        request: {
+          headers: {},
+          body: { include: ["reasoning.encrypted_content"] },
+        },
+      })
+      const resolved = yield* SessionRunnerModel.fromCatalogModel(catalog)
+      const continuation = LLM.request({
+        model: resolved,
+        messages: [
+          Message.user("Inspect the fixture."),
+          Message.assistant([
+            {
+              type: "reasoning",
+              text: "I should use the fixture tool.",
+              providerMetadata: {
+                openai: {
+                  itemId: "rs_fixture_1",
+                  reasoningEncryptedContent: "encrypted-fixture-state",
+                },
+              },
+            },
+            ToolCallPart.make({ id: "call_fixture_1", name: "inspect_fixture", input: { id: "fixture" } }),
+          ]),
+          Message.tool(
+            ToolResultPart.make({
+              id: "call_fixture_1",
+              name: "inspect_fixture",
+              result: { status: "ok" },
+            }),
+          ),
+        ],
+      })
+      const prepared = yield* LLMClient.prepare<OpenAIResponses.OpenAIResponsesBody>(continuation)
+
+      expect(prepared.body.input.some((item) => "type" in item && item.type === "item_reference")).toBe(false)
+      expect(prepared.body.store).toBe(false)
+      expect(resolved.route.defaults.http?.body).toMatchObject({ include: ["reasoning.encrypted_content"] })
+      expect(prepared.body.input).toContainEqual({
+        type: "reasoning",
+        id: "rs_fixture_1",
+        summary: [{ type: "summary_text", text: "I should use the fixture tool." }],
+        encrypted_content: "encrypted-fixture-state",
+      })
+      expect(prepared.body.input).toContainEqual({
+        type: "function_call_output",
+        call_id: "call_fixture_1",
+        output: '{"status":"ok"}',
+      })
+
+      const stateful = yield* SessionRunnerModel.fromCatalogModel(
+        ModelV2.Info.make({
+          ...catalog,
+          request: { ...catalog.request, body: { ...catalog.request.body, store: true } },
+        }),
+      )
+      const statefulPrepared = yield* LLMClient.prepare<OpenAIResponses.OpenAIResponsesBody>(
+        LLM.updateRequest(continuation, {
+          model: stateful,
+        }),
+      )
+
+      expect(statefulPrepared.body.store).toBe(true)
+      expect(statefulPrepared.body.input).toContainEqual({ type: "item_reference", id: "rs_fixture_1" })
+    }),
+  )
+
   it.effect("uses merged API settings for OpenAI-compatible auth and request defaults", () =>
     Effect.gen(function* () {
       const resolved = yield* SessionRunnerModel.fromCatalogModel(

packages/core/test/session-runner-recorded.test.ts

@@ -96,6 +96,7 @@ const execution = Layer.effect(
     })
     return SessionExecution.Service.of({
       active: coordinator.active,
+      activity: coordinator.activity,
       resume: coordinator.run,
       wake: coordinator.wake,
       interrupt: coordinator.interrupt,

packages/core/test/session-runner-tool-events.test.ts

@@ -28,6 +28,7 @@ const capture = () => {
       }),
     subscribe: () => Stream.empty,
     all: () => Stream.empty,
+    observeAggregate: () => Effect.die("not implemented"),
     durable: () => Stream.empty,
     listen: () => Effect.succeed(Effect.void),
     project: () => Effect.void,

packages/core/test/session-runner.test.ts

@@ -4,6 +4,11 @@ import {
   LLMError,
   LLMEvent,
   Model,
+  HttpContext,
+  HttpRateLimitDetails,
+  HttpRequestDetails,
+  HttpResponseDetails,
+  RateLimitReason,
   TransportReason,
   InvalidRequestReason,
   type LLMClientShape,
@@ -255,6 +260,7 @@ const execution = Layer.effect(
     })
     return SessionExecution.Service.of({
       active: coordinator.active,
+      activity: coordinator.activity,
       resume: coordinator.run,
       wake: coordinator.wake,
       interrupt: coordinator.interrupt,
@@ -515,13 +521,21 @@ const verifyPartialFlushOnFailure = (kind: FragmentKind) =>
     responseStream = Stream.concat(Stream.fromIterable(fixture.partialEvents), Stream.fail(failure))
 
     expect(yield* session.resume(sessionID).pipe(Effect.flip)).toBe(failure)
+    const expectedContent =
+      kind === "tool input"
+        ? {
+            type: "tool",
+            id: fragmentID(kind, "partial"),
+            state: { status: "error", error: { message: "Tool execution interrupted" } },
+          }
+        : fixture.expectedContent
     expect(yield* session.context(sessionID)).toMatchObject([
       { type: "user", text: prompt },
       {
         type: "assistant",
         finish: "error",
-        error: { type: "unknown", message: "Provider unavailable" },
-        content: [fixture.expectedContent],
+        error: { type: "provider", category: "transport", message: "Provider connection failed", retryable: false },
+        content: [expectedContent],
       },
     ])
   })
@@ -1196,7 +1210,16 @@ describe("SessionRunnerLLM", () => {
       expect(requests).toHaveLength(3)
       expect(yield* session.context(sessionID)).toMatchObject([
         { type: "compaction" },
-        { type: "assistant", finish: "error", error: { message: "prompt too long" } },
+        {
+          type: "assistant",
+          finish: "error",
+          error: {
+            type: "provider",
+            category: "invalid-request",
+            message: "Provider rejected the request",
+            retryable: false,
+          },
+        },
       ])
     }),
   )
@@ -1244,7 +1267,16 @@ describe("SessionRunnerLLM", () => {
       expect(context.some((message) => message.type === "compaction")).toBe(false)
       expect(context.slice(-2)).toMatchObject([
         { type: "user", text: "Continue" },
-        { type: "assistant", finish: "error", error: { message: "prompt too long" } },
+        {
+          type: "assistant",
+          finish: "error",
+          error: {
+            type: "provider",
+            category: "invalid-request",
+            message: "Provider rejected the request",
+            retryable: false,
+          },
+        },
       ])
     }),
   )
@@ -2920,7 +2952,16 @@ describe("SessionRunnerLLM", () => {
       expect(requests).toHaveLength(1)
       expect(yield* session.context(sessionID)).toMatchObject([
         { type: "user", text: "Fail durably" },
-        { type: "assistant", finish: "error", error: { type: "unknown", message: "Provider unavailable" } },
+        {
+          type: "assistant",
+          finish: "error",
+          error: {
+            type: "provider",
+            category: "unknown",
+            message: "Provider request failed",
+            retryable: false,
+          },
+        },
       ])
     }),
   )
@@ -2939,7 +2980,16 @@ describe("SessionRunnerLLM", () => {
       expect(requests).toHaveLength(1)
       expect(yield* session.context(sessionID)).toMatchObject([
         { type: "user", text: "Fail before step" },
-        { type: "assistant", finish: "error", error: { type: "unknown", message: "Provider unavailable" } },
+        {
+          type: "assistant",
+          finish: "error",
+          error: {
+            type: "provider",
+            category: "unknown",
+            message: "Provider request failed",
+            retryable: false,
+          },
+        },
       ])
     }),
   )
@@ -2966,7 +3016,12 @@ describe("SessionRunnerLLM", () => {
         {
           type: "assistant",
           finish: "error",
-          error: { message: "prompt too long" },
+          error: {
+            type: "provider",
+            category: "invalid-request",
+            message: "Provider rejected the request",
+            retryable: false,
+          },
           content: [{ type: "text", text: "Partial" }],
         },
       ])
@@ -2985,11 +3040,138 @@ describe("SessionRunnerLLM", () => {
       yield* replaySessionProjection(sessionID)
       expect(yield* session.context(sessionID)).toMatchObject([
         { type: "user", text: "Fail raw stream durably" },
-        { type: "assistant", finish: "error", error: { type: "unknown", message: "Provider unavailable" } },
+        {
+          type: "assistant",
+          finish: "error",
+          error: { type: "provider", category: "transport", message: "Provider connection failed", retryable: false },
+        },
       ])
     }),
   )
 
+  it.effect("does not publish step ended when the provider throws after step finish", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const db = yield* Database.Service
+      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Fail after finish" }), resume: false })
+      const failure = providerUnavailable()
+      responseStream = Stream.concat(
+        Stream.fromIterable([LLMEvent.stepFinish({ index: 0, reason: "stop" })]),
+        Stream.fail(failure),
+      )
+
+      expect(yield* session.resume(sessionID).pipe(Effect.flip)).toBe(failure)
+
+      const types = yield* db.db.select({ type: EventTable.type }).from(EventTable).all().pipe(Effect.orDie)
+      expect(types).toContainEqual({
+        type: EventV2.versionedType(SessionEvent.Step.Failed.type, 2),
+      })
+      expect(types).not.toContainEqual({
+        type: EventV2.versionedType(SessionEvent.Step.Ended.type, 2),
+      })
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Fail after finish" },
+        {
+          type: "assistant",
+          finish: "error",
+          error: { type: "provider", category: "transport", message: "Provider connection failed" },
+        },
+      ])
+    }),
+  )
+
+  it.effect("preserves sanitized HTTP rate-limit details in the durable event and projection", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const db = yield* Database.Service
+      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Fail with rate limit" }), resume: false })
+      const failure = new LLMError({
+        module: "RequestExecutor",
+        method: "execute",
+        reason: new RateLimitReason({
+          message: "secret provider response message",
+          retryAfterMs: 12_000,
+          rateLimit: new HttpRateLimitDetails({ retryAfterMs: 12_000, limit: { requests: "secret-limit" } }),
+          http: new HttpContext({
+            request: new HttpRequestDetails({
+              method: "POST",
+              url: "https://secret.example/v1/responses?api_key=credential",
+              headers: { authorization: "Bearer credential" },
+            }),
+            response: new HttpResponseDetails({ status: 429, headers: { "x-secret": "secret-header" } }),
+            body: '{"secret":"provider body"}',
+            requestId: "secret-request-id",
+          }),
+        }),
+      })
+      responseStream = Stream.fail(failure)
+
+      expect(yield* session.resume(sessionID).pipe(Effect.flip)).toBe(failure)
+
+      const event = yield* db.db
+        .select({ data: EventTable.data })
+        .from(EventTable)
+        .where(eq(EventTable.type, EventV2.versionedType(SessionEvent.Step.Failed.type, 2)))
+        .get()
+        .pipe(Effect.orDie)
+      const expected = {
+        type: "provider",
+        category: "rate-limit",
+        message: "Provider rate limit exceeded",
+        status: 429,
+        retryable: true,
+        retryAfterMs: 12_000,
+      }
+      expect(event?.data.error).toEqual(expected)
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Fail with rate limit" },
+        { type: "assistant", finish: "error", error: expected },
+      ])
+      expect(JSON.stringify(event)).not.toMatch(
+        /secret provider|secret\.example|credential|secret-header|provider body|secret-request-id|secret-limit/,
+      )
+      expect(JSON.stringify(yield* session.context(sessionID))).not.toMatch(
+        /secret provider|secret\.example|credential|secret-header|provider body|secret-request-id|secret-limit/,
+      )
+    }),
+  )
+
+  it.effect("projects categorized in-band failures without fabricating an HTTP status", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Fail in band" }), resume: false })
+      response = [
+        LLMEvent.providerError({
+          message: "rate_limit_exceeded: secret provider message",
+          category: "rate-limit",
+          retryable: false,
+        }),
+      ]
+
+      yield* session.resume(sessionID)
+
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Fail in band" },
+        {
+          type: "assistant",
+          finish: "error",
+          error: {
+            type: "provider",
+            category: "rate-limit",
+            message: "Provider rate limit exceeded",
+            retryable: false,
+          },
+        },
+      ])
+      const assistant = (yield* session.context(sessionID))[1]
+      expect(assistant?.type === "assistant" ? assistant.error : undefined).not.toHaveProperty("status")
+      expect(JSON.stringify(assistant)).not.toContain("secret provider message")
+    }),
+  )
+
   it.effect("does not continue automatically after a provider error follows a local tool call", () =>
     Effect.gen(function* () {
       yield* setup
@@ -3005,13 +3187,25 @@ describe("SessionRunnerLLM", () => {
       response = [
         LLMEvent.stepStart({ index: 0 }),
         LLMEvent.toolCall({ id: "call-before-provider-error", name: "echo", input: { text: "settled" } }),
-        LLMEvent.providerError({ message: "Provider unavailable" }),
+        LLMEvent.providerError({ message: "secret provider failure" }),
       ]
 
       yield* session.resume(sessionID)
 
       expect(requests).toHaveLength(1)
       expect(executions.slice(executionCount)).toEqual(["settled"])
+      const assistant = (yield* session.context(sessionID))[1]
+      expect(assistant).toMatchObject({
+        type: "assistant",
+        finish: "error",
+        error: {
+          type: "provider",
+          category: "unknown",
+          message: "Provider request failed",
+          retryable: false,
+        },
+      })
+      expect(JSON.stringify(assistant)).not.toContain("secret provider failure")
     }),
   )
 
@@ -3101,7 +3295,12 @@ describe("SessionRunnerLLM", () => {
         {
           type: "assistant",
           finish: "error",
-          error: { type: "unknown", message: "Provider unavailable" },
+          error: {
+            type: "provider",
+            category: "transport",
+            message: "Provider connection failed",
+            retryable: false,
+          },
           content: [{ type: "tool", id: "call-hosted-raw-failure", state: { status: "error" } }],
         },
       ])

packages/core/test/shared-schema.test.ts

@@ -144,6 +144,8 @@ test("Core reuses the canonical shared schemas", async () => {
     [coreSessionInput.Admitted, SessionInput.Admitted],
     [coreSessionMessage.ID, SessionMessage.ID],
     [coreSessionMessage.UnknownError, SessionMessage.UnknownError],
+    [coreSessionMessage.ProviderError, SessionMessage.ProviderError],
+    [coreSessionMessage.Error, SessionMessage.Error],
     [coreSessionMessage.AgentSwitched, SessionMessage.AgentSwitched],
     [coreSessionMessage.ModelSwitched, SessionMessage.ModelSwitched],
     [coreSessionMessage.User, SessionMessage.User],
@@ -204,3 +206,17 @@ test("shared record schemas construct and decode plain objects", () => {
   expect(Prompt.fromUserMessage({ text: "hello" })).toEqual(made)
   expect(Workspace.ID.ascending("")).toStartWith("wrk_")
 })
+
+test("assistant errors retain legacy unknown decode compatibility", () => {
+  const assistant = Schema.decodeUnknownSync(SessionMessage.Assistant)({
+    id: "msg_legacy_error",
+    type: "assistant",
+    agent: "build",
+    model: { id: "model", providerID: "provider" },
+    content: [],
+    error: { type: "unknown", message: "Legacy failure" },
+    time: { created: 0 },
+  })
+
+  expect(assistant.error).toEqual({ type: "unknown", message: "Legacy failure" })
+})

packages/llm/src/schema/errors.ts

@@ -4,6 +4,20 @@ import { ModelID, ProviderID, ProviderMetadata, RouteID } from "./ids"
 export const ProviderFailureClassification = Schema.Literal("context-overflow")
 export type ProviderFailureClassification = typeof ProviderFailureClassification.Type
 
+export const ProviderFailureCategory = Schema.Literals([
+  "invalid-request",
+  "no-route",
+  "authentication",
+  "rate-limit",
+  "quota-exceeded",
+  "content-policy",
+  "provider-internal",
+  "transport",
+  "invalid-provider-output",
+  "unknown",
+])
+export type ProviderFailureCategory = typeof ProviderFailureCategory.Type
+
 export class HttpRequestDetails extends Schema.Class<HttpRequestDetails>("LLM.HttpRequestDetails")({
   method: Schema.String,
   url: Schema.String,

packages/llm/src/schema/events.ts

@@ -2,7 +2,7 @@ import { Schema } from "effect"
 import { ContentBlockID, FinishReason, ProtocolID, ProviderMetadata, RouteID, ToolCallID } from "./ids"
 import { ModelSchema } from "./options"
 import { ToolOutput, ToolResultValue } from "./messages"
-import { ProviderFailureClassification } from "./errors"
+import { ProviderFailureCategory, ProviderFailureClassification } from "./errors"
 
 /**
  * Token usage reported by an LLM provider.
@@ -201,6 +201,8 @@ export const ProviderErrorEvent = Schema.Struct({
   type: Schema.tag("provider-error"),
   message: Schema.String,
   classification: Schema.optional(ProviderFailureClassification),
+  category: Schema.optional(ProviderFailureCategory),
+  status: Schema.optional(Schema.Number),
   retryable: Schema.optional(Schema.Boolean),
   providerMetadata: Schema.optional(ProviderMetadata),
 }).annotate({ identifier: "LLM.Event.ProviderError" })

packages/protocol/src/groups/session.ts

@@ -295,7 +295,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
         query: {
           after: Schema.NumberFromString.pipe(Schema.decodeTo(NonNegativeInt), Schema.optional),
         },
-        success: HttpApiSchema.StreamSse({ data: SessionEvent.Durable }),
+        success: HttpApiSchema.StreamSse({ data: SessionEvent.Stream }),
         error: SessionNotFoundError,
       })
         .middleware(sessionLocationMiddleware)
@@ -303,7 +303,8 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
           OpenApi.annotations({
             identifier: "v2.session.events",
             summary: "Subscribe to session events",
-            description: "Replay durable events after an aggregate sequence, then continue with new durable events.",
+            description:
+              "Replay durable events after an aggregate sequence, emit current process-local activity, then continue with durable and live-only Session events.",
           }),
         ),
     )

packages/schema/src/session-event.ts

@@ -50,6 +50,25 @@ const stepSettlementOptions = {
 
 export const UnknownError = SessionMessage.UnknownError
 export type UnknownError = SessionMessage.UnknownError
+export const ProviderError = SessionMessage.ProviderError
+export type ProviderError = SessionMessage.ProviderError
+export const Error = SessionMessage.Error
+export type Error = SessionMessage.Error
+
+export const Activity = Event.define({
+  type: "session.activity",
+  schema: {
+    sessionID: SessionID,
+    active: Schema.Boolean,
+  },
+})
+export type Activity = typeof Activity.Type
+
+export const makeActivity = (sessionID: SessionID, active: boolean): Activity => ({
+  id: Event.ID.create(),
+  type: Activity.type,
+  data: { sessionID, active },
+})
 
 export const AgentSwitched = Event.define({
   type: "session.next.agent.switched",
@@ -188,7 +207,7 @@ export namespace Step {
     schema: {
       ...Base,
       assistantMessageID: SessionMessage.ID,
-      error: UnknownError,
+      error: Error,
     },
   })
   export type Failed = typeof Failed.Type
@@ -517,3 +536,6 @@ export type DurableEvent = typeof Durable.Type
 export const All = Schema.Union(Definitions, { mode: "oneOf" }).pipe(Schema.toTaggedUnion("type"))
 export type Event = typeof All.Type
 export type Type = Event["type"]
+
+export const Stream = Schema.Union([Activity, All], { mode: "oneOf" }).pipe(Schema.toTaggedUnion("type"))
+export type StreamEvent = typeof Stream.Type

packages/schema/src/session-message.ts

@@ -21,6 +21,33 @@ export const UnknownError = Schema.Struct({
   message: Schema.String,
 }).annotate({ identifier: "Session.Error.Unknown" })
 
+export const ProviderErrorCategory = Schema.Literals([
+  "invalid-request",
+  "no-route",
+  "authentication",
+  "rate-limit",
+  "quota-exceeded",
+  "content-policy",
+  "provider-internal",
+  "transport",
+  "invalid-provider-output",
+  "unknown",
+])
+export type ProviderErrorCategory = typeof ProviderErrorCategory.Type
+
+export interface ProviderError extends Schema.Schema.Type<typeof ProviderError> {}
+export const ProviderError = Schema.Struct({
+  type: Schema.Literal("provider"),
+  category: ProviderErrorCategory,
+  message: Schema.String,
+  status: Schema.Finite.pipe(Schema.optional),
+  retryable: Schema.Boolean,
+  retryAfterMs: Schema.Finite.pipe(Schema.optional),
+}).annotate({ identifier: "Session.Error.Provider" })
+
+export const Error = Schema.Union([UnknownError, ProviderError]).pipe(Schema.toTaggedUnion("type"))
+export type Error = UnknownError | ProviderError
+
 const Base = {
   id: ID,
   metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional),
@@ -177,7 +204,7 @@ export const Assistant = Schema.Struct({
     reasoning: Schema.Finite,
     cache: Schema.Struct({ read: Schema.Finite, write: Schema.Finite }),
   }).pipe(optional),
-  error: UnknownError.pipe(optional),
+  error: Error.pipe(optional),
   time: Schema.Struct({
     created: DateTimeUtcFromMillis,
     completed: DateTimeUtcFromMillis.pipe(optional),

packages/sdk-next/test/embedded.test.ts

@@ -3,7 +3,7 @@ import { mkdtemp, rm } from "node:fs/promises"
 import { tmpdir } from "node:os"
 import { join } from "node:path"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { Effect, Option, Schema, Stream } from "effect"
+import { Effect, Fiber, Option, Schema, Stream } from "effect"
 
 test("embedded client uses the real router and handlers", async () => {
   const directory = await mkdtemp(join(tmpdir(), "opencode-embedded-"))
@@ -103,6 +103,167 @@ test("embedded client uses the real router and handlers", async () => {
   }
 })
 
+test("embedded session events live-tail stateless tool continuation through provider failure", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "opencode-embedded-live-tail-"))
+  const database = Flag.OPENCODE_DB
+  Flag.OPENCODE_DB = join(directory, "opencode.sqlite")
+  let requests = 0
+  const requestBodies: unknown[] = []
+  const server = Bun.serve({
+    port: 0,
+    async fetch(request) {
+      const body = await request.json()
+      if (JSON.stringify(body).includes("Generate a title for this conversation")) {
+        return new Response(
+          [
+            'data: {"type":"response.output_text.delta","item_id":"title","delta":"Live tail"}',
+            'data: {"type":"response.completed","response":{}}',
+            "",
+          ].join("\n\n"),
+          { headers: { "content-type": "text/event-stream" } },
+        )
+      }
+      requests++
+      requestBodies.push(body)
+      if (requests > 1) {
+        return Response.json({ error: { message: "Provider failed on turn B", type: "server_error" } }, { status: 500 })
+      }
+      return new Response(
+        [
+          'data: {"type":"response.output_item.added","item":{"type":"reasoning","id":"rs_live_tail","encrypted_content":null}}',
+          'data: {"type":"response.reasoning_summary_part.added","item_id":"rs_live_tail","summary_index":0}',
+          'data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_live_tail","summary_index":0,"delta":"Run the tool."}',
+          'data: {"type":"response.reasoning_summary_part.done","item_id":"rs_live_tail","summary_index":0}',
+          'data: {"type":"response.output_item.done","item":{"type":"reasoning","id":"rs_live_tail","encrypted_content":"encrypted-live-tail"}}',
+          'data: {"type":"response.output_item.added","item":{"type":"function_call","id":"item_live_tail","call_id":"call_live_tail","name":"live_tail_tool","arguments":""}}',
+          'data: {"type":"response.function_call_arguments.delta","item_id":"item_live_tail","delta":"{\\"value\\":\\"A\\"}"}',
+          'data: {"type":"response.output_item.done","item":{"type":"function_call","id":"item_live_tail","call_id":"call_live_tail","name":"live_tail_tool","arguments":"{\\"value\\":\\"A\\"}"}}',
+          'data: {"type":"response.completed","response":{}}',
+          "",
+        ].join("\n\n"),
+        { headers: { "content-type": "text/event-stream" } },
+      )
+    },
+  })
+
+  try {
+    const git = Bun.spawn(["git", "init"], { cwd: directory, stdout: "ignore", stderr: "ignore" })
+    expect(await git.exited).toBe(0)
+    await Bun.write(
+      join(directory, "opencode.json"),
+      JSON.stringify({
+        formatter: false,
+        lsp: false,
+        providers: {
+          test: {
+            name: "Test",
+            api: {
+              type: "aisdk",
+              package: "@ai-sdk/openai",
+              url: `http://127.0.0.1:${server.port}/v1`,
+              settings: {},
+            },
+            request: {
+              body: { apiKey: "test-key", store: false, include: ["reasoning.encrypted_content"] },
+            },
+            models: {
+              "test-model": {
+                name: "Test Model",
+                capabilities: { tools: true, input: ["text"], output: ["text"] },
+                limit: { context: 100_000, output: 10_000 },
+              },
+            },
+          },
+        },
+      }),
+    )
+    const { AbsolutePath, Agent, Location, Model, OpenCode, Prompt, Provider, Session, Tool } = await import("../src")
+    const sessionID = Session.ID.make(`ses_embedded_${crypto.randomUUID()}`)
+    const executions: string[] = []
+
+    await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const opencode = yield* OpenCode.create()
+          yield* opencode.tools.register({
+            live_tail_tool: Tool.make({
+              description: "Record step A",
+              input: Schema.Struct({ value: Schema.String }),
+              output: Schema.Struct({ value: Schema.String }),
+              execute: ({ value }) =>
+                Effect.sync(() => {
+                  executions.push(value)
+                  return { value }
+                }),
+            }),
+          })
+          yield* opencode.sessions.create({
+            id: sessionID,
+            agent: Agent.ID.make("build"),
+            model: Model.Ref.make({ id: Model.ID.make("test-model"), providerID: Provider.ID.make("test") }),
+            location: Location.Ref.make({ directory: AbsolutePath.make(directory) }),
+          })
+
+          let observedActive = false
+          const streamed = yield* opencode.sessions.events({ sessionID }).pipe(
+            Stream.takeUntil((event) => {
+              if (event.type !== "session.activity") return false
+              if (event.data.active) {
+                observedActive = true
+                return false
+              }
+              return observedActive
+            }),
+            Stream.runCollect,
+            Effect.timeout("10 seconds"),
+            Effect.forkScoped,
+          )
+          yield* Effect.yieldNow
+          yield* opencode.sessions.prompt({ sessionID, prompt: Prompt.make({ text: "Run step A, then fail B" }) })
+          const received = Array.from(yield* Fiber.join(streamed))
+          const durable = received.filter((event) => event.durable !== undefined)
+
+          expect(executions).toEqual(["A"])
+          expect(requests).toBeGreaterThanOrEqual(2)
+          expect(requestBodies.slice(0, 2)).toMatchObject([
+            { store: false },
+            {
+              store: false,
+              input: expect.arrayContaining([
+                {
+                  type: "reasoning",
+                  id: "rs_live_tail",
+                  summary: [{ type: "summary_text", text: "Run the tool." }],
+                  encrypted_content: "encrypted-live-tail",
+                },
+                { type: "function_call_output", call_id: "call_live_tail", output: '{"value":"A"}' },
+              ]),
+            },
+          ])
+          expect(durable.at(-1)?.type).toBe("session.next.step.failed")
+          expect(durable.map((event) => event.durable!.seq)).toEqual(
+            Array.from({ length: durable.length }, (_, index) => durable[0]!.durable!.seq + index),
+          )
+          expect(durable.map((event) => event.type)).toEqual(
+            expect.arrayContaining([
+              "session.next.tool.called",
+              "session.next.tool.success",
+              "session.next.step.ended",
+              "session.next.step.failed",
+            ]),
+          )
+          expect(received.at(-2)?.type).toBe("session.next.step.failed")
+          expect(received.at(-1)).toMatchObject({ type: "session.activity", data: { active: false } })
+        }),
+      ),
+    )
+  } finally {
+    server.stop(true)
+    Flag.OPENCODE_DB = database
+    await rm(directory, { recursive: true, force: true })
+  }
+}, 20_000)
+
 test("embedded client is available as a Layer service", async () => {
   const directory = await mkdtemp(join(tmpdir(), "opencode-embedded-layer-"))
   const database = Flag.OPENCODE_DB

packages/sdk/js/src/v2/gen/sdk.gen.ts

@@ -5713,7 +5713,7 @@ export class Session3 extends HeyApiClient {
   /**
    * Subscribe to session events
    *
-   * Replay durable events after an aggregate sequence, then continue with new durable events.
+   * Replay durable events after an aggregate sequence, emit current process-local activity, then continue with durable and live-only Session events.
    */
   public events<ThrowOnError extends boolean = false>(
     parameters: {

packages/sdk/js/src/v2/gen/types.gen.ts

@@ -952,7 +952,7 @@ export type GlobalEvent = {
           timestamp: number
           sessionID: string
           assistantMessageID: string
-          error: SessionErrorUnknown
+          error: SessionErrorUnknown | SessionErrorProvider
         }
       }
     | {
@@ -2955,6 +2955,25 @@ export type SessionErrorUnknown = {
   message: string
 }
 
+export type SessionErrorProvider = {
+  type: "provider"
+  category:
+    | "invalid-request"
+    | "no-route"
+    | "authentication"
+    | "rate-limit"
+    | "quota-exceeded"
+    | "content-policy"
+    | "provider-internal"
+    | "transport"
+    | "invalid-provider-output"
+    | "unknown"
+  message: string
+  status?: number
+  retryable: boolean
+  retryAfterMs?: number
+}
+
 export type LlmProviderMetadata = {
   [key: string]: {
     [key: string]: unknown
@@ -3399,7 +3418,7 @@ export type SyncEventSessionNextStepFailed = {
       timestamp: number
       sessionID: string
       assistantMessageID: string
-      error: SessionErrorUnknown
+      error: SessionErrorUnknown | SessionErrorProvider
     }
   }
 }
@@ -4009,7 +4028,7 @@ export type SessionMessageAssistant = {
       write: number
     }
   }
-  error?: SessionErrorUnknown
+  error?: SessionErrorUnknown | SessionErrorProvider
 }
 
 export type SessionMessageCompaction = {
@@ -4036,6 +4055,24 @@ export type SessionMessage =
   | SessionMessageAssistant
   | SessionMessageCompaction
 
+export type SessionActivity = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.activity"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    sessionID: string
+    active: boolean
+  }
+}
+
 export type SessionNextAgentSwitched = {
   id: string
   metadata?: {
@@ -4289,7 +4326,7 @@ export type SessionNextStepFailed = {
     timestamp: number
     sessionID: string
     assistantMessageID: string
-    error: SessionErrorUnknown
+    error: SessionErrorUnknown | SessionErrorProvider
   }
 }
 
@@ -4313,6 +4350,27 @@ export type SessionNextTextStarted = {
   }
 }
 
+export type SessionNextTextDelta = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.next.text.delta"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    timestamp: number
+    sessionID: string
+    assistantMessageID: string
+    textID: string
+    delta: string
+  }
+}
+
 export type SessionNextTextEnded = {
   id: string
   metadata?: {
@@ -4334,6 +4392,70 @@ export type SessionNextTextEnded = {
   }
 }
 
+export type SessionNextReasoningStarted = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.next.reasoning.started"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    timestamp: number
+    sessionID: string
+    assistantMessageID: string
+    reasoningID: string
+    providerMetadata?: LlmProviderMetadata
+  }
+}
+
+export type SessionNextReasoningDelta = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.next.reasoning.delta"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    timestamp: number
+    sessionID: string
+    assistantMessageID: string
+    reasoningID: string
+    delta: string
+  }
+}
+
+export type SessionNextReasoningEnded = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.next.reasoning.ended"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    timestamp: number
+    sessionID: string
+    assistantMessageID: string
+    reasoningID: string
+    text: string
+    providerMetadata?: LlmProviderMetadata
+  }
+}
+
 export type SessionNextToolInputStarted = {
   id: string
   metadata?: {
@@ -4355,6 +4477,27 @@ export type SessionNextToolInputStarted = {
   }
 }
 
+export type SessionNextToolInputDelta = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.next.tool.input.delta"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    timestamp: number
+    sessionID: string
+    assistantMessageID: string
+    callID: string
+    delta: string
+  }
+}
+
 export type SessionNextToolInputEnded = {
   id: string
   metadata?: {
@@ -4484,49 +4627,6 @@ export type SessionNextToolFailed = {
   }
 }
 
-export type SessionNextReasoningStarted = {
-  id: string
-  metadata?: {
-    [key: string]: unknown
-  }
-  type: "session.next.reasoning.started"
-  durable?: {
-    aggregateID: string
-    seq: number | "NaN" | "Infinity" | "-Infinity"
-    version: number | "NaN" | "Infinity" | "-Infinity"
-  }
-  location?: LocationRef
-  data: {
-    timestamp: number
-    sessionID: string
-    assistantMessageID: string
-    reasoningID: string
-    providerMetadata?: LlmProviderMetadata
-  }
-}
-
-export type SessionNextReasoningEnded = {
-  id: string
-  metadata?: {
-    [key: string]: unknown
-  }
-  type: "session.next.reasoning.ended"
-  durable?: {
-    aggregateID: string
-    seq: number | "NaN" | "Infinity" | "-Infinity"
-    version: number | "NaN" | "Infinity" | "-Infinity"
-  }
-  location?: LocationRef
-  data: {
-    timestamp: number
-    sessionID: string
-    assistantMessageID: string
-    reasoningID: string
-    text: string
-    providerMetadata?: LlmProviderMetadata
-  }
-}
-
 export type SessionNextRetried = {
   id: string
   metadata?: {
@@ -4567,6 +4667,26 @@ export type SessionNextCompactionStarted = {
   }
 }
 
+export type SessionNextCompactionDelta = {
+  id: string
+  metadata?: {
+    [key: string]: unknown
+  }
+  type: "session.next.compaction.delta"
+  durable?: {
+    aggregateID: string
+    seq: number | "NaN" | "Infinity" | "-Infinity"
+    version: number | "NaN" | "Infinity" | "-Infinity"
+  }
+  location?: LocationRef
+  data: {
+    timestamp: number
+    sessionID: string
+    messageID: string
+    text: string
+  }
+}
+
 export type SessionNextCompactionEnded = {
   id: string
   metadata?: {
@@ -5344,7 +5464,7 @@ export type V2EventSessionNextStepFailed = {
     timestamp: number
     sessionID: string
     assistantMessageID: string
-    error: SessionErrorUnknown
+    error: SessionErrorUnknown | SessionErrorProvider
   }
 }
 
@@ -6931,7 +7051,7 @@ export type EventSessionNextStepFailed = {
     timestamp: number
     sessionID: string
     assistantMessageID: string
-    error: SessionErrorUnknown
+    error: SessionErrorUnknown | SessionErrorProvider
   }
 }
 

specs/v2/session.md

@@ -172,11 +172,11 @@ Inbox promotion coalesces pending steers in durable admission order. Once contin
 
 Eager local-tool execution is intentionally unbounded in the current local slice. This minimizes tool latency but does not increase SQLite settlement throughput: Session-event publication remains serialized per provider turn. Before broadening exposure, revisit per-turn call limits, output truncation, and operational backpressure using observed workloads. The `session.next.*` event schemas remain experimental and unshipped; databases created by earlier experimental builds are disposable rather than compatibility targets.
 
-The synchronized `session.next.*` event family and projected Session-message model predate this branch. This slice refines their replay contract: projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers can use `sessions.events({ sessionID, after? })` to replay durable `session.next.*` events after an aggregate sequence cursor, then tail durable events without a race. Live-only text, reasoning, and tool-input fragments remain available through EventV2 subscriptions for connected renderers; they are intentionally absent from the replayable Session stream.
+The synchronized `session.next.*` event family and projected Session-message model predate this branch. Projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers use `sessions.events({ sessionID, after? })` for one Session-scoped stream containing replayable lifecycle/output events, process-local activity, and connected live-only output fragments.
 
-The first `sessions.events(...)` contract is durable-only during both replay and live tailing. This keeps one cursor equal to one persisted aggregate sequence and is sufficient for reconnect-safe consumers. A later UI-facing API may optionally interleave live-only deltas while connected, but those fragments must remain explicitly ephemeral: they cannot advance the durable cursor, replay after reconnect, or be mistaken for publication boundaries.
+The stream registers observation, captures one fixed durable cutoff `B`, emits durable rows in `(after, B]` in aggregate-sequence order, and then emits exactly one authoritative `session.activity` value. Only events carrying `durable` metadata advance the cursor. Activity and text, reasoning, compaction, or tool-input deltas are process-local and never replay after reconnect. Activity means foreground execution ownership in this process; `false` does not describe a terminal outcome, and background work does not activate its parent Session.
 
-Durable event tail wakeups are advisory and edge-triggered. Each active tail owns one sliding-capacity-1 dirty signal for its aggregate and re-queries SQLite after a wake. Repeated commits coalesce while the tail is busy because durable rows, not in-memory notifications, preserve every event and sequence. Subscribe and register the dirty signal before historical replay, then remove it when the tail closes, so replay handoff cannot miss a commit and inactive aggregates retain no wake state.
+Durable event tail notifications are advisory. Each active tail uses bounded non-blocking observation and re-queries SQLite after a wake. Repeated commits may coalesce because durable rows, not in-memory notifications, preserve every event and sequence. A live-only fragment is emitted only after draining later durable rows, which fences dependent deltas behind their committed start events. If a live consumer cannot keep up, the stream closes rather than blocking Session execution or SQLite writers. Aggregate deletion still removes durable history and remains a documented replay limitation rather than being redesigned here.
 
 Event replay owner claims are separate from clustered Session execution ownership. The former already fences synchronized projection reconstruction; the latter still needs distributed active-run acquisition, stale-runtime rejection, interruption, and placement orchestration.