test(sdk): cover embedded session live tail

fix(sdk): preserve session stream ordering
feat(sdk): unify session event stream
2026-10-28 03:14:44 +08:00 · 2026-06-26 00:00:38 -04:00 · 2026-06-25 23:50:59 -04:00 · 2026-06-25 23:50:42 -04:00 · 2026-06-26 03:35:39 +00:00 · 2026-06-26 03:33:35 +00:00
38 changed files with 1281 additions and 502 deletions
@@ -162,18 +162,21 @@ _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.
 - `sessions.messages(...)` returns a **Page** and uses the same cursor discipline as `sessions.list(...)`: the initial request supplies `sessionID`, ordering, and page size; continuation supplies `sessionID` plus only an opaque branded message cursor carrying ordering, page size, direction, and message anchor. Using a cursor with another Session is invalid.
 - `sessions.message({ sessionID, messageID })` is a required resource lookup. An unknown Session fails with `SessionNotFoundError`; a known Session with an absent or differently owned message fails with `MessageNotFoundError` without disclosing cross-Session ownership. Absence is not represented as `undefined` across the public HTTP boundary.
 - `sessions.interrupt({ sessionID })` first verifies that the durable Session exists, failing with `SessionNotFoundError` otherwise. For a known Session, interruption is idempotent: idle, already-settled, or locally unowned execution is a no-op.
+- `sessions.active()` snapshots the current process's foreground Session drain registry as a record of Session IDs to `{ type: "running" }`. Missing IDs are inactive; background subagents and tasks do not make their parent Session active, and process restart clears the registry.
 - `sessions.context({ sessionID })` preserves the existing message-only operation. It returns projected conversational messages selected as Session context; it does not include or represent the complete provider request context, whose baseline system context and other contributions remain separate.
 - **Open question**: Should a future, separately named operation expose the complete provider request context, including baseline system context, selected source contributions, and context-epoch metadata?
 - `sessions.prompt(...)` exposes `resume?: boolean`. Omitting it preserves durable admission followed by an advisory execution wake; `resume: false` requests durable admit-only behavior.
@@ -55,43 +55,49 @@ const Endpoint0_1 = (raw: RawClient["server.session"]) => (input?: Endpoint0_1In
    Effect.map((value) => value.data),
  )

-type Endpoint0_2Request = Parameters<RawClient["server.session"]["session.get"]>[0]
-type Endpoint0_2Input = { readonly sessionID: Endpoint0_2Request["params"]["sessionID"] }
-const Endpoint0_2 = (raw: RawClient["server.session"]) => (input: Endpoint0_2Input) =>
+const Endpoint0_2 = (raw: RawClient["server.session"]) => () =>
+  raw["session.active"]({}).pipe(
+    Effect.mapError(mapClientError),
+    Effect.map((value) => value.data),
+  )
+
+type Endpoint0_3Request = Parameters<RawClient["server.session"]["session.get"]>[0]
+type Endpoint0_3Input = { readonly sessionID: Endpoint0_3Request["params"]["sessionID"] }
+const Endpoint0_3 = (raw: RawClient["server.session"]) => (input: Endpoint0_3Input) =>
  raw["session.get"]({ params: { sessionID: input.sessionID } }).pipe(
    Effect.mapError(mapClientError),
    Effect.map((value) => value.data),
  )

-type Endpoint0_3Request = Parameters<RawClient["server.session"]["session.switchAgent"]>[0]
-type Endpoint0_3Input = {
-  readonly sessionID: Endpoint0_3Request["params"]["sessionID"]
-  readonly agent: Endpoint0_3Request["payload"]["agent"]
+type Endpoint0_4Request = Parameters<RawClient["server.session"]["session.switchAgent"]>[0]
+type Endpoint0_4Input = {
+  readonly sessionID: Endpoint0_4Request["params"]["sessionID"]
+  readonly agent: Endpoint0_4Request["payload"]["agent"]
 }
-const Endpoint0_3 = (raw: RawClient["server.session"]) => (input: Endpoint0_3Input) =>
+const Endpoint0_4 = (raw: RawClient["server.session"]) => (input: Endpoint0_4Input) =>
  raw["session.switchAgent"]({ params: { sessionID: input.sessionID }, payload: { agent: input.agent } }).pipe(
    Effect.mapError(mapClientError),
  )

-type Endpoint0_4Request = Parameters<RawClient["server.session"]["session.switchModel"]>[0]
-type Endpoint0_4Input = {
-  readonly sessionID: Endpoint0_4Request["params"]["sessionID"]
-  readonly model: Endpoint0_4Request["payload"]["model"]
+type Endpoint0_5Request = Parameters<RawClient["server.session"]["session.switchModel"]>[0]
+type Endpoint0_5Input = {
+  readonly sessionID: Endpoint0_5Request["params"]["sessionID"]
+  readonly model: Endpoint0_5Request["payload"]["model"]
 }
-const Endpoint0_4 = (raw: RawClient["server.session"]) => (input: Endpoint0_4Input) =>
+const Endpoint0_5 = (raw: RawClient["server.session"]) => (input: Endpoint0_5Input) =>
  raw["session.switchModel"]({ params: { sessionID: input.sessionID }, payload: { model: input.model } }).pipe(
    Effect.mapError(mapClientError),
  )

-type Endpoint0_5Request = Parameters<RawClient["server.session"]["session.prompt"]>[0]
-type Endpoint0_5Input = {
-  readonly sessionID: Endpoint0_5Request["params"]["sessionID"]
-  readonly id?: Endpoint0_5Request["payload"]["id"]
-  readonly prompt: Endpoint0_5Request["payload"]["prompt"]
-  readonly delivery?: Endpoint0_5Request["payload"]["delivery"]
-  readonly resume?: Endpoint0_5Request["payload"]["resume"]
+type Endpoint0_6Request = Parameters<RawClient["server.session"]["session.prompt"]>[0]
+type Endpoint0_6Input = {
+  readonly sessionID: Endpoint0_6Request["params"]["sessionID"]
+  readonly id?: Endpoint0_6Request["payload"]["id"]
+  readonly prompt: Endpoint0_6Request["payload"]["prompt"]
+  readonly delivery?: Endpoint0_6Request["payload"]["delivery"]
+  readonly resume?: Endpoint0_6Request["payload"]["resume"]
 }
-const Endpoint0_5 = (raw: RawClient["server.session"]) => (input: Endpoint0_5Input) =>
+const Endpoint0_6 = (raw: RawClient["server.session"]) => (input: Endpoint0_6Input) =>
  raw["session.prompt"]({
    params: { sessionID: input.sessionID },
    payload: { id: input.id, prompt: input.prompt, delivery: input.delivery, resume: input.resume },
@@ -100,23 +106,23 @@ const Endpoint0_5 = (raw: RawClient["server.session"]) => (input: Endpoint0_5Inp
    Effect.map((value) => value.data),
  )

-type Endpoint0_6Request = Parameters<RawClient["server.session"]["session.compact"]>[0]
-type Endpoint0_6Input = { readonly sessionID: Endpoint0_6Request["params"]["sessionID"] }
-const Endpoint0_6 = (raw: RawClient["server.session"]) => (input: Endpoint0_6Input) =>
-  raw["session.compact"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))
-
-type Endpoint0_7Request = Parameters<RawClient["server.session"]["session.wait"]>[0]
+type Endpoint0_7Request = Parameters<RawClient["server.session"]["session.compact"]>[0]
 type Endpoint0_7Input = { readonly sessionID: Endpoint0_7Request["params"]["sessionID"] }
 const Endpoint0_7 = (raw: RawClient["server.session"]) => (input: Endpoint0_7Input) =>
+  raw["session.compact"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))
+
+type Endpoint0_8Request = Parameters<RawClient["server.session"]["session.wait"]>[0]
+type Endpoint0_8Input = { readonly sessionID: Endpoint0_8Request["params"]["sessionID"] }
+const Endpoint0_8 = (raw: RawClient["server.session"]) => (input: Endpoint0_8Input) =>
  raw["session.wait"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))

-type Endpoint0_8Request = Parameters<RawClient["server.session"]["session.revert.stage"]>[0]
-type Endpoint0_8Input = {
-  readonly sessionID: Endpoint0_8Request["params"]["sessionID"]
-  readonly messageID: Endpoint0_8Request["payload"]["messageID"]
-  readonly files?: Endpoint0_8Request["payload"]["files"]
+type Endpoint0_9Request = Parameters<RawClient["server.session"]["session.revert.stage"]>[0]
+type Endpoint0_9Input = {
+  readonly sessionID: Endpoint0_9Request["params"]["sessionID"]
+  readonly messageID: Endpoint0_9Request["payload"]["messageID"]
+  readonly files?: Endpoint0_9Request["payload"]["files"]
 }
-const Endpoint0_8 = (raw: RawClient["server.session"]) => (input: Endpoint0_8Input) =>
+const Endpoint0_9 = (raw: RawClient["server.session"]) => (input: Endpoint0_9Input) =>
  raw["session.revert.stage"]({
    params: { sessionID: input.sessionID },
    payload: { messageID: input.messageID, files: input.files },
@@ -125,30 +131,30 @@ const Endpoint0_8 = (raw: RawClient["server.session"]) => (input: Endpoint0_8Inp
    Effect.map((value) => value.data),
  )

-type Endpoint0_9Request = Parameters<RawClient["server.session"]["session.revert.clear"]>[0]
-type Endpoint0_9Input = { readonly sessionID: Endpoint0_9Request["params"]["sessionID"] }
-const Endpoint0_9 = (raw: RawClient["server.session"]) => (input: Endpoint0_9Input) =>
-  raw["session.revert.clear"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))
-
-type Endpoint0_10Request = Parameters<RawClient["server.session"]["session.revert.commit"]>[0]
+type Endpoint0_10Request = Parameters<RawClient["server.session"]["session.revert.clear"]>[0]
 type Endpoint0_10Input = { readonly sessionID: Endpoint0_10Request["params"]["sessionID"] }
 const Endpoint0_10 = (raw: RawClient["server.session"]) => (input: Endpoint0_10Input) =>
-  raw["session.revert.commit"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))
+  raw["session.revert.clear"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))

-type Endpoint0_11Request = Parameters<RawClient["server.session"]["session.context"]>[0]
+type Endpoint0_11Request = Parameters<RawClient["server.session"]["session.revert.commit"]>[0]
 type Endpoint0_11Input = { readonly sessionID: Endpoint0_11Request["params"]["sessionID"] }
 const Endpoint0_11 = (raw: RawClient["server.session"]) => (input: Endpoint0_11Input) =>
+  raw["session.revert.commit"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))
+
+type Endpoint0_12Request = Parameters<RawClient["server.session"]["session.context"]>[0]
+type Endpoint0_12Input = { readonly sessionID: Endpoint0_12Request["params"]["sessionID"] }
+const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12Input) =>
  raw["session.context"]({ params: { sessionID: input.sessionID } }).pipe(
    Effect.mapError(mapClientError),
    Effect.map((value) => value.data),
  )

-type Endpoint0_12Request = Parameters<RawClient["server.session"]["session.events"]>[0]
-type Endpoint0_12Input = {
-  readonly sessionID: Endpoint0_12Request["params"]["sessionID"]
-  readonly after?: Endpoint0_12Request["query"]["after"]
+type Endpoint0_13Request = Parameters<RawClient["server.session"]["session.events"]>[0]
+type Endpoint0_13Input = {
+  readonly sessionID: Endpoint0_13Request["params"]["sessionID"]
+  readonly after?: Endpoint0_13Request["query"]["after"]
 }
-const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12Input) =>
+const Endpoint0_13 = (raw: RawClient["server.session"]) => (input: Endpoint0_13Input) =>
  Stream.unwrap(
    raw["session.events"]({ params: { sessionID: input.sessionID }, query: { after: input.after } }).pipe(
      Effect.mapError(mapClientError),
@@ -156,17 +162,17 @@ const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12I
    ),
  )

-type Endpoint0_13Request = Parameters<RawClient["server.session"]["session.interrupt"]>[0]
-type Endpoint0_13Input = { readonly sessionID: Endpoint0_13Request["params"]["sessionID"] }
-const Endpoint0_13 = (raw: RawClient["server.session"]) => (input: Endpoint0_13Input) =>
+type Endpoint0_14Request = Parameters<RawClient["server.session"]["session.interrupt"]>[0]
+type Endpoint0_14Input = { readonly sessionID: Endpoint0_14Request["params"]["sessionID"] }
+const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14Input) =>
  raw["session.interrupt"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError))

-type Endpoint0_14Request = Parameters<RawClient["server.session"]["session.message"]>[0]
-type Endpoint0_14Input = {
-  readonly sessionID: Endpoint0_14Request["params"]["sessionID"]
-  readonly messageID: Endpoint0_14Request["params"]["messageID"]
+type Endpoint0_15Request = Parameters<RawClient["server.session"]["session.message"]>[0]
+type Endpoint0_15Input = {
+  readonly sessionID: Endpoint0_15Request["params"]["sessionID"]
+  readonly messageID: Endpoint0_15Request["params"]["messageID"]
 }
-const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14Input) =>
+const Endpoint0_15 = (raw: RawClient["server.session"]) => (input: Endpoint0_15Input) =>
  raw["session.message"]({ params: { sessionID: input.sessionID, messageID: input.messageID } }).pipe(
    Effect.mapError(mapClientError),
    Effect.map((value) => value.data),
@@ -175,19 +181,20 @@ const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14I
 const adaptGroup0 = (raw: RawClient["server.session"]) => ({
  list: Endpoint0_0(raw),
  create: Endpoint0_1(raw),
-  get: Endpoint0_2(raw),
-  switchAgent: Endpoint0_3(raw),
-  switchModel: Endpoint0_4(raw),
-  prompt: Endpoint0_5(raw),
-  compact: Endpoint0_6(raw),
-  wait: Endpoint0_7(raw),
-  stage: Endpoint0_8(raw),
-  clear: Endpoint0_9(raw),
-  commit: Endpoint0_10(raw),
-  context: Endpoint0_11(raw),
-  events: Endpoint0_12(raw),
-  interrupt: Endpoint0_13(raw),
-  message: Endpoint0_14(raw),
+  active: Endpoint0_2(raw),
+  get: Endpoint0_3(raw),
+  switchAgent: Endpoint0_4(raw),
+  switchModel: Endpoint0_5(raw),
+  prompt: Endpoint0_6(raw),
+  compact: Endpoint0_7(raw),
+  wait: Endpoint0_8(raw),
+  stage: Endpoint0_9(raw),
+  clear: Endpoint0_10(raw),
+  commit: Endpoint0_11(raw),
+  context: Endpoint0_12(raw),
+  events: Endpoint0_13(raw),
+  interrupt: Endpoint0_14(raw),
+  message: Endpoint0_15(raw),
 })

 const adaptClient = (raw: RawClient) => ({ sessions: adaptGroup0(raw["server.session"]) })
@@ -3,6 +3,7 @@ import type {
  SessionsListOutput,
  SessionsCreateInput,
  SessionsCreateOutput,
+  SessionsActiveOutput,
  SessionsGetInput,
  SessionsGetOutput,
  SessionsSwitchAgentInput,
@@ -198,6 +199,17 @@ export function make(options: ClientOptions) {
          },
          requestOptions,
        ).then((value) => value.data),
+      active: (requestOptions?: RequestOptions) =>
+        request<{ readonly data: SessionsActiveOutput }>(
+          {
+            method: "GET",
+            path: `/api/session/active`,
+            successStatus: 200,
+            declaredStatuses: [401, 400],
+            empty: false,
+          },
+          requestOptions,
+        ).then((value) => value.data),
      get: (input: SessionsGetInput, requestOptions?: RequestOptions) =>
        request<{ readonly data: SessionsGetOutput }>(
          {
@@ -243,6 +243,8 @@ export type SessionsCreateOutput = {
  }
 }["data"]

+export type SessionsActiveOutput = { readonly data: { readonly [x: string]: { readonly type: "running" } } }["data"]
+
 export type SessionsGetInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] }

 export type SessionsGetOutput = {
@@ -572,7 +574,6 @@ export type SessionsContextOutput = {
        >
        readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray<string> }
        readonly finish?: string
-        readonly settlement?: "completed" | "failed" | "interrupted"
        readonly cost?: number
        readonly tokens?: {
          readonly input: number
@@ -600,6 +601,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 }
@@ -796,14 +805,6 @@ export type SessionsEventsOutput =
        readonly error: { readonly type: "unknown"; readonly message: string }
      }
    }
-  | {
-      readonly id: string
-      readonly metadata?: { readonly [x: string]: unknown }
-      readonly type: "session.next.step.interrupted"
-      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 id: string
      readonly metadata?: { readonly [x: string]: unknown }
@@ -817,6 +818,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 }
@@ -831,6 +846,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 }
@@ -845,6 +903,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 }
@@ -939,35 +1011,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 }
@@ -1001,6 +1044,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 +1254,6 @@ export type SessionsMessageOutput = {
        >
        readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray<string> }
        readonly finish?: string
-        readonly settlement?: "completed" | "failed" | "interrupted"
        readonly cost?: number
        readonly tokens?: {
          readonly input: number
@@ -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" },
          }),
        ),
@@ -37,6 +37,11 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
    if (url.includes("/message/")) {
      return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ data: modelSwitchedMessage })))
    }
+    if (url.endsWith("/api/session/active")) {
+      return Effect.succeed(
+        HttpClientResponse.fromWeb(request, Response.json({ data: { ses_test: { type: "running" } } })),
+      )
+    }
    if (request.method === "POST" && url.endsWith("/api/session")) {
      return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(session)))
    }
@@ -50,6 +55,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
  const result = await Effect.gen(function* () {
    const client = yield* OpenCode.make({ baseUrl: "http://localhost:3000" })
    const page = yield* client.sessions.list({ limit: 10 })
+    const active = yield* client.sessions.active()
    const created = yield* client.sessions.create({
      location: Location.Ref.make({ directory: AbsolutePath.make("/tmp/project") }),
    })
@@ -74,10 +80,11 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
      sessionID: Session.ID.make("ses_test"),
      messageID: SessionMessage.ID.make("msg_model"),
    })
-    return { page, created, admitted, context, events, message }
+    return { page, active, created, admitted, context, events, message }
  }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise)

  expect(DateTime.toEpochMillis(result.page.data[0].time.created)).toBe(1_717_171_717_000)
+  expect(result.active).toEqual({ ses_test: { type: "running" } })
  expect(Object.getPrototypeOf(result.page.data[0])).toBe(Object.prototype)
  expect(Object.getPrototypeOf(result.created)).toBe(Object.prototype)
  expect(result.created.id).toBe("ses_test")
@@ -86,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" }))
 })

@@ -138,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 },
+}
@@ -25,13 +25,17 @@ 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: [] })
      if (url.includes("/message/")) return Response.json({ data: modelSwitchedMessage })
+      if (url.endsWith("/api/session/active")) return Response.json({ data: { ses_test: { type: "running" } } })
      if (init?.method === "POST" && url.endsWith("/api/session")) return Response.json(session)
      if (init?.method === "POST") return new Response(null, { status: 204 })
      return Response.json({ data: [session.data], cursor: { next: "next" } })
@@ -39,6 +43,7 @@ test("session methods use the public HTTP contract", async () => {
  })

  const page = await client.sessions.list({ limit: "10", order: "desc" })
+  const active = await client.sessions.active()
  const created = await client.sessions.create({ location: { directory: "/tmp/project" } })
  await client.sessions.switchAgent({ sessionID: "ses_test", agent: "build" })
  await client.sessions.switchModel({
@@ -59,13 +64,16 @@ test("session methods use the public HTTP contract", async () => {
  const message = await client.sessions.message({ sessionID: "ses_test", messageID: "msg_model" })

  expect(page.cursor.next).toBe("next")
+  expect(active).toEqual({ ses_test: { type: "running" } })
  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"],
+    ["GET", "http://localhost:3000/api/session/active"],
    ["POST", "http://localhost:3000/api/session"],
    ["POST", "http://localhost:3000/api/session/ses_test/agent"],
    ["POST", "http://localhost:3000/api/session/ses_test/model"],
@@ -77,7 +85,7 @@ test("session methods use the public HTTP contract", async () => {
    ["POST", "http://localhost:3000/api/session/ses_test/interrupt"],
    ["GET", "http://localhost:3000/api/session/ses_test/message/msg_model"],
  ])
-  const body = requests[4]?.init?.body
+  const body = requests.find((request) => request.url.endsWith("/api/session/ses_test/prompt"))?.init?.body
  if (typeof body !== "string") throw new Error("Expected JSON request body")
  expect(JSON.parse(body)).toEqual({
    prompt: { text: "Hello" },
@@ -149,3 +157,9 @@ const modelSwitchedEvent = {
    model: { id: "claude", providerID: "anthropic" },
  },
 }
+
+const activityEvent = {
+  id: "evt_activity",
+  type: "session.activity",
+  data: { sessionID: "ses_test", active: false },
+}
@@ -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,
@@ -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
@@ -155,6 +155,7 @@ export interface Interface {
  }) => Effect.Effect<void, OperationUnavailableError>
  readonly compact: (input: CompactInput) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
  readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
+  readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
  readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
  readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
  readonly revert: {
@@ -184,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(
@@ -340,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* () {
@@ -402,6 +431,7 @@ export const layer = Layer.unwrap(
              yield* result.get(sessionID)
              return yield* new OperationUnavailableError({ operation: "wait" })
            }),
+            active: execution.active,
            resume: Effect.fn("V2Session.resume")(function* (sessionID) {
              yield* result.get(sessionID)
              yield* execution.resume(sessionID)
@@ -1,10 +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. */
@@ -19,5 +24,11 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
 /** Low-level compatibility layer for callers that only need durable Session recording. */
 export const noopLayer = Layer.succeed(
  Service,
-  Service.of({ resume: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.void }),
+  Service.of({
+    active: Effect.succeed(new Set()),
+    activity: () => Effect.succeed({ attach: () => Effect.succeed(false) }),
+    resume: () => Effect.void,
+    wake: () => Effect.void,
+    interrupt: () => Effect.void,
+  }),
 )
@@ -28,6 +28,8 @@ export const layer = Layer.effect(
    })

    return SessionExecution.Service.of({
+      active: coordinator.active,
+      activity: coordinator.activity,
      interrupt: coordinator.interrupt,
      resume: coordinator.run,
      wake: coordinator.wake,
@@ -210,7 +210,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
        return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
          draft.time.completed = event.data.timestamp
          draft.finish = event.data.finish
-          draft.settlement = "completed"
          draft.cost = event.data.cost
          draft.tokens = event.data.tokens
          if (event.data.snapshot || event.data.files)
@@ -225,16 +224,9 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
        return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
          draft.time.completed = event.data.timestamp
          draft.finish = "error"
-          draft.settlement = "failed"
          draft.error = event.data.error
        })
      },
-      "session.next.step.interrupted": (event) => {
-        return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
-          draft.time.completed = event.data.timestamp
-          draft.settlement = "interrupted"
-        })
-      },
      "session.next.text.started": (event) => {
        return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
          draft.content.push(
@@ -381,7 +381,6 @@ export const layer = Layer.effectDiscard(
    yield* events.project(SessionEvent.Step.Started, (event) => run(db, event))
    yield* events.project(SessionEvent.Step.Ended, (event) => run(db, event))
    yield* events.project(SessionEvent.Step.Failed, (event) => run(db, event))
-    yield* events.project(SessionEvent.Step.Interrupted, (event) => run(db, event))
    yield* events.project(SessionEvent.Text.Started, (event) => run(db, event))
    yield* events.project(SessionEvent.Text.Ended, (event) => run(db, event))
    yield* events.project(SessionEvent.Tool.Input.Started, (event) => run(db, event))
@@ -2,8 +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. */
@@ -24,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> => ({
@@ -32,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(
@@ -54,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)
      }
@@ -72,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))
      })
@@ -86,6 +103,7 @@ export const make = <Key, E>(options: {

        const next = makeEntry()
        active.set(key, next)
+        notifyActivity(key, true)
        start(key, next, false)
      })

@@ -98,5 +116,33 @@ export const make = <Key, E>(options: {
        return Fiber.interrupt(entry.owner)
      })

-    return { 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 }
  })
@@ -290,7 +290,6 @@ export const layer = Layer.effect(
          if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) {
            yield* FiberSet.clear(toolFibers)
            yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted"))
-            yield* withPublication(publisher.interruptAssistant())
            return yield* Effect.interrupt
          }
          if (
@@ -299,7 +298,8 @@ export const layer = Layer.effect(
          ) {
            yield* FiberSet.clear(toolFibers)
            yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted"))
-            yield* withPublication(publisher.interruptAssistant())
+            if (publisher.hasActiveAssistant())
+              yield* withPublication(publisher.failAssistant("Provider turn interrupted"))
          }
          if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) {
            const failure = Cause.squash(settled.cause)
@@ -307,7 +307,7 @@ export const layer = Layer.effect(
            yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`))
          }
          const stepSettlement = publisher.stepSettlement()
-          if (stepSettlement && !publisher.hasProviderError() && !publisher.hasAssistantSettled()) {
+          if (stepSettlement && !publisher.hasProviderError()) {
            const endSnapshot = yield* snapshots.capture()
            const files =
              startSnapshot && endSnapshot
@@ -66,13 +66,15 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
  >()
  const timestamp = DateTime.now
  let assistantMessageID: SessionMessage.ID | undefined
-  let assistantSettled = false
+  let assistantActive = false
+  let assistantFailed = false
  let providerFailed = false
-  let stepSettlement: { readonly finish: SessionMessage.Finish; readonly tokens: ReturnType<typeof tokens> } | undefined
+  let stepSettlement: { readonly finish: string; readonly tokens: ReturnType<typeof tokens> } | undefined

  const startAssistant = Effect.fnUntraced(function* () {
    if (assistantMessageID !== undefined) return assistantMessageID
    assistantMessageID = SessionMessage.ID.create()
+    assistantActive = true
    yield* events.publish(SessionEvent.Step.Started, {
      ...input,
      assistantMessageID,
@@ -194,39 +196,20 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
    yield* flushFragments()
  })

-  const settleAssistant = Effect.fnUntraced(function* (
-    publish: (assistantMessageID: SessionMessage.ID) => Effect.Effect<void>,
-  ) {
-    if (assistantSettled) return
+  const failAssistant = Effect.fnUntraced(function* (message: string) {
+    if (assistantFailed) return
    yield* flush()
    const assistantMessageID = yield* startAssistant()
-    assistantSettled = true
-    yield* publish(assistantMessageID)
+    assistantActive = false
+    assistantFailed = true
+    yield* events.publish(SessionEvent.Step.Failed, {
+      sessionID: input.sessionID,
+      timestamp: yield* timestamp,
+      assistantMessageID,
+      error: { type: "unknown", message },
+    })
  })

-  const failAssistant = (message: string) =>
-    settleAssistant((assistantMessageID) =>
-      Effect.gen(function* () {
-        yield* events.publish(SessionEvent.Step.Failed, {
-          sessionID: input.sessionID,
-          timestamp: yield* timestamp,
-          assistantMessageID,
-          error: { type: "unknown", message },
-        })
-      }),
-    )
-
-  const interruptAssistant = () =>
-    settleAssistant((assistantMessageID) =>
-      Effect.gen(function* () {
-        yield* events.publish(SessionEvent.Step.Interrupted, {
-          sessionID: input.sessionID,
-          timestamp: yield* timestamp,
-          assistantMessageID,
-        })
-      }),
-    )
-
  const failUnsettledTools = Effect.fn("SessionRunner.failUnsettledTools")(function* (
    message: string,
    hostedOnly = false,
@@ -412,6 +395,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
      }
      case "step-finish":
        yield* flush()
+        assistantActive = false
        if (stepSettlement) return yield* Effect.die("Duplicate step finish")
        stepSettlement = { finish: event.reason, tokens: tokens(event.usage) }
        return
@@ -428,10 +412,9 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
    publish,
    flush,
    failAssistant,
-    interruptAssistant,
    failUnsettledTools,
+    hasActiveAssistant: () => assistantActive,
    hasAssistantStarted: () => assistantMessageID !== undefined,
-    hasAssistantSettled: () => assistantSettled,
    hasProviderError: () => providerFailed,
    stepSettlement: () => stepSettlement,
    startAssistant,
@@ -70,7 +70,7 @@ const toolResult = (tool: SessionMessage.AssistantTool, providerMetadata: Provid
 const assistant = (message: SessionMessage.Assistant, model: Model) => {
  const sameModel =
    String(message.model.providerID) === String(model.provider) && String(message.model.id) === String(model.id)
-  const reuseProviderMetadata = sameModel && message.error === undefined && message.settlement !== "interrupted"
+  const reuseProviderMetadata = sameModel && message.error === undefined
  const content = message.content.flatMap((item): ContentPart[] => {
    if (item.type === "text") return [{ type: "text", text: item.text }]
    if (item.type === "reasoning")
@@ -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
@@ -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"
@@ -22,9 +22,30 @@ import { testEffect } from "./lib/effect"
 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)
@@ -108,6 +129,13 @@ const eventCount = (type: string) =>
  )

 describe("SessionV2.prompt", () => {
+  it.effect("exposes the execution registry", () =>
+    Effect.gen(function* () {
+      activeSessions.add(sessionID)
+      expect(Array.from(yield* (yield* SessionV2.Service).active)).toEqual([sessionID])
+    }).pipe(Effect.ensuring(Effect.sync(() => activeSessions.clear()))),
+  )
+
  it.effect("delegates execution continuation through SessionExecution", () =>
    Effect.gen(function* () {
      yield* setup
@@ -170,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 })
@@ -179,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"],
@@ -187,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)
    }),
  )

@@ -66,6 +66,99 @@ describe("SessionRunCoordinator", () => {
    ),
  )

+  it.effect("snapshots only active executions", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        const firstGate = yield* Deferred.make<void>()
+        const secondGate = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (key: string) =>
+            Deferred.succeed(key === "first" ? firstStarted : secondStarted, undefined).pipe(
+              Effect.andThen(Deferred.await(key === "first" ? firstGate : secondGate)),
+            ),
+        })
+
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+        const first = yield* coordinator.run("first").pipe(Effect.forkChild)
+        yield* Deferred.await(firstStarted)
+        expect(Array.from(yield* coordinator.active)).toEqual(["first"])
+
+        const second = yield* coordinator.run("second").pipe(Effect.forkChild)
+        yield* Deferred.await(secondStarted)
+        expect(Array.from(yield* coordinator.active)).toEqual(["first", "second"])
+
+        yield* Deferred.succeed(firstGate, undefined)
+        yield* Fiber.join(first)
+        expect(Array.from(yield* coordinator.active)).toEqual(["second"])
+        yield* Deferred.succeed(secondGate, undefined)
+        yield* Fiber.join(second)
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+      }),
+    ),
+  )
+
+  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* () {
+        const failure = new Error("failed")
+        const defect = new Error("defect")
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (key: string) => (key === "failure" ? Effect.fail(failure) : Effect.die(defect)),
+        })
+
+        const failed = yield* coordinator.run("failure").pipe(Effect.exit)
+        expect(Exit.isFailure(failed) && Cause.hasFails(failed.cause)).toBeTrue()
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+
+        const died = yield* coordinator.run("defect").pipe(Effect.exit)
+        expect(Exit.isFailure(died) && Cause.hasDies(died.cause)).toBeTrue()
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+      }),
+    ),
+  )
+
+  it.effect("cleans active executions when its scope closes", () =>
+    Effect.gen(function* () {
+      const started = yield* Deferred.make<void>()
+      const coordinator = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const coordinator = yield* SessionRunCoordinator.make({
+            drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)),
+          })
+          yield* coordinator.wake("session")
+          yield* Deferred.await(started)
+          expect(Array.from(yield* coordinator.active)).toEqual(["session"])
+          return coordinator
+        }),
+      )
+
+      expect(Array.from(yield* coordinator.active)).toEqual([])
+    }),
+  )
+
  it.effect("coalesces wakes received during active execution", () =>
    Effect.scoped(
      Effect.gen(function* () {
@@ -166,6 +259,7 @@ describe("SessionRunCoordinator", () => {

        const exit = yield* Fiber.await(resumed)
        expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+        expect(Array.from(yield* coordinator.active)).toEqual([])
        expect(runs).toBe(1)
      }),
    ),
@@ -327,7 +327,7 @@ Recent work
    ])
  })

-  test("drops provider-native continuation metadata from interrupted assistant turns", () => {
+  test("drops provider-native continuation metadata from failed assistant turns", () => {
    const messages = toLLMMessages(
      [
        SessionMessage.Assistant.make({
@@ -361,7 +361,8 @@ Recent work
              time: { created, completed: created },
            }),
          ],
-          settlement: "interrupted",
+          finish: "error",
+          error: { type: "unknown", message: "Provider turn interrupted" },
          time: { created, completed: created },
        }),
      ],
@@ -95,6 +95,8 @@ const execution = Layer.effect(
      drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
    })
    return SessionExecution.Service.of({
+      active: coordinator.active,
+      activity: coordinator.activity,
      resume: coordinator.run,
      wake: coordinator.wake,
      interrupt: coordinator.interrupt,
@@ -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,
@@ -254,6 +254,8 @@ const execution = Layer.effect(
      drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
    })
    return SessionExecution.Service.of({
+      active: coordinator.active,
+      activity: coordinator.activity,
      resume: coordinator.run,
      wake: coordinator.wake,
      interrupt: coordinator.interrupt,
@@ -519,7 +521,6 @@ const verifyPartialFlushOnFailure = (kind: FragmentKind) =>
      {
        type: "assistant",
        finish: "error",
-        settlement: "failed",
        error: { type: "unknown", message: "Provider unavailable" },
        content: [fixture.expectedContent],
      },
@@ -543,28 +544,12 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) =>
    const fiber = yield* runner.run({ sessionID, force: true }).pipe(Effect.forkChild)
    yield* Deferred.await(streamed)
    yield* Fiber.interrupt(fiber)
-    const { db } = yield* Database.Service
-    const interruptedVersion = SessionEvent.Step.Interrupted.durable?.version
-    expect(interruptedVersion).toBe(2)
-    if (interruptedVersion === undefined) return yield* Effect.die("Step.Interrupted must be durable")
-    const settlements = yield* db
-      .select({ type: EventTable.type })
-      .from(EventTable)
-      .where(eq(EventTable.aggregate_id, sessionID))
-      .all()
-      .pipe(Effect.orDie)
-    expect(
-      settlements.filter(({ type }) =>
-        [SessionEvent.Step.Ended.type, SessionEvent.Step.Failed.type, SessionEvent.Step.Interrupted.type].some((settled) =>
-          type.startsWith(settled),
-        ),
-      ),
-    ).toEqual([{ type: EventV2.versionedType(SessionEvent.Step.Interrupted.type, interruptedVersion) }])
    expect(yield* session.context(sessionID)).toMatchObject([
      { type: "user", text: prompt },
      {
        type: "assistant",
-        settlement: "interrupted",
+        finish: "error",
+        error: { type: "unknown", message: "Provider turn interrupted" },
        content: [
          kind === "tool input"
            ? { type: "tool", id: fragmentID(kind, "interrupted"), state: { status: "error" } }
@@ -2670,7 +2655,6 @@ describe("SessionRunnerLLM", () => {
              state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } },
            },
          ],
-          settlement: "interrupted",
        },
      ])
    }),
@@ -2815,7 +2799,6 @@ describe("SessionRunnerLLM", () => {
              state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } },
            },
          ],
-          settlement: "interrupted",
        },
      ])
    }),
@@ -9,7 +9,7 @@ describe("public event manifest", () => {
    expect(EventManifest.Definitions).toBe(SchemaEventManifest.Definitions)
    expect(EventManifest.Latest).toBe(SchemaEventManifest.Latest)
    expect(EventManifest.Durable).toBe(SchemaEventManifest.Durable)
-    expect(EventManifest.Latest.size).toBe(89)
+    expect(EventManifest.Latest.size).toBe(88)
    expect(EventManifest.Latest.get("session.next.step.ended")).toBe(SessionEvent.Step.Ended)
    expect(EventManifest.Latest.get("todo.updated")).toBe(Todo.Event.Updated)
    expect(EventManifest.Latest.has("ide.installed")).toBe(false)
@@ -964,6 +964,7 @@ const scenarios: Scenario[] = [
      headers: ctx.headers(),
    }))
    .status(400, undefined, "none"),
+  http.protected.get("/api/session/active", "v2.session.active").json(200, data(object), "none"),
  http.protected
    .post("/api/session", "v2.session.create")
    .at((ctx) => ({
@@ -73,6 +73,10 @@ export const SessionsCursor = Schema.String.pipe(
 )
 export type SessionsCursor = typeof SessionsCursor.Type

+const SessionActive = Schema.Struct({
+  type: Schema.Literal("running"),
+}).annotate({ identifier: "SessionActive" })
+
 const SessionsQueryCursor = SessionsCursor.annotate({
  description: "Opaque pagination cursor returned as cursor.previous or cursor.next in the previous response.",
 })
@@ -124,6 +128,18 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
        }),
      ),
    )
+    .add(
+      HttpApiEndpoint.get("session.active", "/api/session/active", {
+        success: Schema.Struct({ data: Schema.Record(Session.ID, SessionActive) }),
+      }).annotateMerge(
+        OpenApi.annotations({
+          identifier: "v2.session.active",
+          summary: "List active sessions",
+          description:
+            "Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.",
+        }),
+      ),
+    )
    .add(
      HttpApiEndpoint.get("session.get", "/api/session/:sessionID", {
        params: { sessionID: Session.ID },
@@ -279,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)
@@ -287,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.",
          }),
        ),
    )
@@ -51,6 +51,21 @@ const stepSettlementOptions = {
 export const UnknownError = SessionMessage.UnknownError
 export type UnknownError = SessionMessage.UnknownError

+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",
  ...options,
@@ -165,7 +180,6 @@ export namespace Step {
    schema: {
      ...Base,
      assistantMessageID: SessionMessage.ID,
-      // Step.Ended v2 was originally persisted with an open string schema.
      finish: Schema.String,
      cost: Schema.Finite,
      tokens: Schema.Struct({
@@ -193,16 +207,6 @@ export namespace Step {
    },
  })
  export type Failed = typeof Failed.Type
-
-  export const Interrupted = Event.define({
-    type: "session.next.step.interrupted",
-    ...stepSettlementOptions,
-    schema: {
-      ...Base,
-      assistantMessageID: SessionMessage.ID,
-    },
-  })
-  export type Interrupted = typeof Interrupted.Type
 }

 export namespace Text {
@@ -469,7 +473,6 @@ export const DurableDefinitions = Event.inventory(
  Step.Started,
  Step.Ended,
  Step.Failed,
-  Step.Interrupted,
  Text.Started,
  Text.Ended,
  Tool.Input.Started,
@@ -501,7 +504,6 @@ export const Definitions = Event.inventory(
  Step.Started,
  Step.Ended,
  Step.Failed,
-  Step.Interrupted,
  Text.Started,
  Text.Delta,
  Text.Ended,
@@ -530,3 +532,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
@@ -21,19 +21,6 @@ export const UnknownError = Schema.Struct({
  message: Schema.String,
 }).annotate({ identifier: "Session.Error.Unknown" })

-export const Finish = Schema.Literals([
-  "stop",
-  "length",
-  "tool-calls",
-  "content-filter",
-  "error",
-  "unknown",
-])
-export type Finish = typeof Finish.Type
-
-export const Settlement = Schema.Literals(["completed", "failed", "interrupted"])
-export type Settlement = typeof Settlement.Type
-
 const Base = {
  id: ID,
  metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional),
@@ -182,9 +169,7 @@ export const Assistant = Schema.Struct({
    end: Schema.String.pipe(optional),
    files: Schema.Array(RelativePath).pipe(optional),
  }).pipe(optional),
-  // Projected histories predate the typed provider finish model and may contain arbitrary values.
  finish: Schema.String.pipe(optional),
-  settlement: Settlement.pipe(optional),
  cost: Schema.Finite.pipe(optional),
  tokens: Schema.Struct({
    input: Schema.Finite,
@@ -1,5 +1,4 @@
 import { describe, expect, test } from "bun:test"
-import { Schema } from "effect"
 import { FileSystem, Integration, Permission, Project, Reference, Session, Workspace } from "../src"
 import { EventManifest } from "../src/event-manifest"
 import { IdeEvent } from "../src/ide-event"
@@ -10,8 +9,8 @@ import { WorkspaceEvent } from "../src/workspace-event"

 describe("public event manifest", () => {
  test("owns the complete public event surface", () => {
-    expect(EventManifest.ServerDefinitions.length).toBe(59)
-    expect(EventManifest.Definitions.length).toBe(89)
+    expect(EventManifest.ServerDefinitions.length).toBe(55)
+    expect(EventManifest.Definitions.length).toBe(85)
    expect(SessionV1.Event.Definitions).toEqual([
      SessionV1.Event.Created,
      SessionV1.Event.Updated,
@@ -24,8 +23,8 @@ describe("public event manifest", () => {
      SessionV1.Event.Diff,
      SessionV1.Event.Error,
    ])
-    expect(EventManifest.Latest.size).toBe(89)
-    expect(EventManifest.Durable.size).toBe(36)
+    expect(EventManifest.Latest.size).toBe(85)
+    expect(EventManifest.Durable.size).toBe(32)
  })

  test("uses canonical definitions for current public events", () => {
@@ -43,28 +42,12 @@ describe("public event manifest", () => {
    expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated])
    expect(EventManifest.Latest.has("ide.installed")).toBe(false)
    expect(IdeEvent.Definitions).toEqual([IdeEvent.Installed])
-    const partDelta = EventManifest.Definitions.indexOf(SessionV1.Event.PartDelta)
-    expect(partDelta).toBeGreaterThanOrEqual(0)
-    expect(EventManifest.Definitions.slice(partDelta, partDelta + 3)).toEqual([
+    expect(EventManifest.Definitions.slice(40, 43)).toEqual([
      SessionV1.Event.PartDelta,
      SessionV1.Event.Diff,
      SessionV1.Event.Error,
    ])
-    expect(EventManifest.Latest.get("session.next.step.interrupted")).toBe(SessionEvent.Step.Interrupted)
    expect(EventManifest.Durable.has("session.next.step.ended.1")).toBe(false)
    expect(EventManifest.Durable.get("session.next.step.ended.2")).toBe(SessionEvent.Step.Ended)
  })
-
-  test("decodes legacy Step.Ended v2 finish strings", () => {
-    const event = Schema.decodeUnknownSync(SessionEvent.Step.Ended.data)({
-      sessionID: "ses_legacy",
-      timestamp: 0,
-      assistantMessageID: "msg_legacy",
-      finish: "legacy-provider-reason",
-      cost: 0,
-      tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
-    })
-
-    expect(event.finish).toBe("legacy-provider-reason")
-  })
 })
@@ -1,22 +0,0 @@
-import { expect, test } from "bun:test"
-import { Schema } from "effect"
-import { SessionMessage } from "../src/session-message"
-
-test("does not model interruption as a provider finish reason", () => {
-  expect(() => Schema.decodeUnknownSync(SessionMessage.Finish)("interrupted")).toThrow()
-  expect(Schema.decodeUnknownSync(SessionMessage.Finish)("error")).toBe("error")
-})
-
-test("decodes projected assistant histories with arbitrary finish strings", () => {
-  const message = Schema.decodeUnknownSync(SessionMessage.Message)({
-    id: "msg_legacy",
-    type: "assistant",
-    agent: "build",
-    model: { id: "model", providerID: "provider" },
-    content: [],
-    finish: "legacy-provider-reason",
-    time: { created: 0, completed: 1 },
-  })
-
-  expect(message).toMatchObject({ type: "assistant", finish: "legacy-provider-reason" })
-})
@@ -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-"))
@@ -33,6 +33,7 @@ test("embedded client uses the real router and handlers", async () => {
      yield* opencode.sessions.switchModel({ sessionID, model })
      const selected = yield* opencode.sessions.get({ sessionID })
      const page = yield* opencode.sessions.list({ directory: AbsolutePath.make(directory) })
+      const active = yield* opencode.sessions.active()
      const admitted = yield* opencode.sessions.prompt({
        sessionID,
        prompt: Prompt.make({ text: "Do not run" }),
@@ -81,6 +82,7 @@ test("embedded client uses the real router and handlers", async () => {
      expect(selected.model?.id).toBe(model.id)
      expect(selected.model?.providerID).toBe(model.providerID)
      expect(page.data.some((session) => session.id === sessionID)).toBe(true)
+      expect(active).toEqual({})
      expect(admitted.sessionID).toBe(sessionID)
      expect(prompted.type).toBe("session.next.prompted")
      expect(wakeContext).toContainEqual(expect.objectContaining({ id: wake.id, type: "user" }))
@@ -101,6 +103,146 @@ test("embedded client uses the real router and handlers", async () => {
  }
 })

+test("embedded session events live-tail location-owned durable rows 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 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: {"choices":[{"delta":{"role":"assistant"}}]}',
+            'data: {"choices":[{"delta":{"content":"Live tail"}}]}',
+            'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}',
+            "data: [DONE]",
+            "",
+          ].join("\n\n"),
+          { headers: { "content-type": "text/event-stream" } },
+        )
+      }
+      requests++
+      if (requests > 1) {
+        return Response.json({ error: { message: "Provider failed on turn B", type: "server_error" } }, { status: 500 })
+      }
+      return new Response(
+        [
+          'data: {"choices":[{"delta":{"role":"assistant"}}]}',
+          'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_live_tail","type":"function","function":{"name":"live_tail_tool","arguments":""}}]}}]}',
+          'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"value\\":\\"A\\"}"}}]}}]}',
+          'data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}',
+          "data: [DONE]",
+          "",
+        ].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-compatible",
+              url: `http://127.0.0.1:${server.port}/v1`,
+              settings: {},
+            },
+            request: { body: { apiKey: "test-key" } },
+            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(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
@@ -333,6 +333,8 @@ import type {
  V2QuestionRequestListResponses,
  V2ReferenceListErrors,
  V2ReferenceListResponses,
+  V2SessionActiveErrors,
+  V2SessionActiveResponses,
  V2SessionCompactErrors,
  V2SessionCompactResponses,
  V2SessionContextErrors,
@@ -5501,6 +5503,18 @@ export class Session3 extends HeyApiClient {
    })
  }

+  /**
+   * List active sessions
+   *
+   * Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.
+   */
+  public active<ThrowOnError extends boolean = false>(options?: Options<never, ThrowOnError>) {
+    return (options?.client ?? this.client).get<V2SessionActiveResponses, V2SessionActiveErrors, ThrowOnError>({
+      url: "/api/session/active",
+      ...options,
+    })
+  }
+
  /**
   * Get session
   *
@@ -5699,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: {
@@ -28,7 +28,6 @@ export type Event =
  | EventSessionNextStepStarted
  | EventSessionNextStepEnded
  | EventSessionNextStepFailed
-  | EventSessionNextStepInterrupted
  | EventSessionNextTextStarted
  | EventSessionNextTextDelta
  | EventSessionNextTextEnded
@@ -956,15 +955,6 @@ export type GlobalEvent = {
          error: SessionErrorUnknown
        }
      }
-    | {
-        id: string
-        type: "session.next.step.interrupted"
-        properties: {
-          timestamp: number
-          sessionID: string
-          assistantMessageID: string
-        }
-      }
    | {
        id: string
        type: "session.next.text.started"
@@ -1630,7 +1620,6 @@ export type GlobalEvent = {
    | SyncEventSessionNextStepStarted
    | SyncEventSessionNextStepEnded
    | SyncEventSessionNextStepFailed
-    | SyncEventSessionNextStepInterrupted
    | SyncEventSessionNextTextStarted
    | SyncEventSessionNextTextEnded
    | SyncEventSessionNextReasoningStarted
@@ -2701,6 +2690,10 @@ export type InvalidCursorError = {
  message: string
 }

+export type SessionActive = {
+  type: "running"
+}
+
 export type SessionNotFoundError = {
  _tag: "SessionNotFoundError"
  sessionID: string
@@ -2770,7 +2763,6 @@ export type V2Event =
  | V2EventSessionNextStepStarted
  | V2EventSessionNextStepEnded
  | V2EventSessionNextStepFailed
-  | V2EventSessionNextStepInterrupted
  | V2EventSessionNextTextStarted
  | V2EventSessionNextTextDelta
  | V2EventSessionNextTextEnded
@@ -3412,22 +3404,6 @@ export type SyncEventSessionNextStepFailed = {
  }
 }

-export type SyncEventSessionNextStepInterrupted = {
-  type: "sync"
-  id: string
-  syncEvent: {
-    type: "session.next.step.interrupted.2"
-    id: string
-    seq: number
-    aggregateID: string
-    data: {
-      timestamp: number
-      sessionID: string
-      assistantMessageID: string
-    }
-  }
-}
-
 export type SyncEventSessionNextTextStarted = {
  type: "sync"
  id: string
@@ -4023,7 +3999,6 @@ export type SessionMessageAssistant = {
    files?: Array<string>
  }
  finish?: string
-  settlement?: "completed" | "failed" | "interrupted"
  cost?: number
  tokens?: {
    input: number
@@ -4061,6 +4036,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?: {
@@ -4318,25 +4311,6 @@ export type SessionNextStepFailed = {
  }
 }

-export type SessionNextStepInterrupted = {
-  id: string
-  metadata?: {
-    [key: string]: unknown
-  }
-  type: "session.next.step.interrupted"
-  durable?: {
-    aggregateID: string
-    seq: number | "NaN" | "Infinity" | "-Infinity"
-    version: number | "NaN" | "Infinity" | "-Infinity"
-  }
-  location?: LocationRef
-  data: {
-    timestamp: number
-    sessionID: string
-    assistantMessageID: string
-  }
-}
-
 export type SessionNextTextStarted = {
  id: string
  metadata?: {
@@ -4357,6 +4331,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?: {
@@ -4378,6 +4373,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?: {
@@ -4399,6 +4458,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?: {
@@ -4528,49 +4608,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?: {
@@ -4611,6 +4648,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?: {
@@ -5392,25 +5449,6 @@ export type V2EventSessionNextStepFailed = {
  }
 }

-export type V2EventSessionNextStepInterrupted = {
-  id: string
-  metadata?: {
-    [key: string]: unknown
-  }
-  durable?: {
-    aggregateID: string
-    seq: number
-    version: number
-  }
-  location?: LocationRef
-  type: "session.next.step.interrupted"
-  data: {
-    timestamp: number
-    sessionID: string
-    assistantMessageID: string
-  }
-}
-
 export type V2EventSessionNextTextStarted = {
  id: string
  metadata?: {
@@ -6998,16 +7036,6 @@ export type EventSessionNextStepFailed = {
  }
 }

-export type EventSessionNextStepInterrupted = {
-  id: string
-  type: "session.next.step.interrupted"
-  properties: {
-    timestamp: number
-    sessionID: string
-    assistantMessageID: string
-  }
-}
-
 export type EventSessionNextTextStarted = {
  id: string
  type: "session.next.text.started"
@@ -12017,6 +12045,39 @@ export type V2SessionCreateResponses = {

 export type V2SessionCreateResponse = V2SessionCreateResponses[keyof V2SessionCreateResponses]

+export type V2SessionActiveData = {
+  body?: never
+  path?: never
+  query?: never
+  url: "/api/session/active"
+}
+
+export type V2SessionActiveErrors = {
+  /**
+   * InvalidRequestError
+   */
+  400: InvalidRequestError
+  /**
+   * UnauthorizedError
+   */
+  401: UnauthorizedError
+}
+
+export type V2SessionActiveError = V2SessionActiveErrors[keyof V2SessionActiveErrors]
+
+export type V2SessionActiveResponses = {
+  /**
+   * Success
+   */
+  200: {
+    data: {
+      [key: string]: unknown | SessionActive
+    }
+  }
+}
+
+export type V2SessionActiveResponse = V2SessionActiveResponses[keyof V2SessionActiveResponses]
+
 export type V2SessionGetData = {
  body?: never
  path: {
@@ -10187,6 +10187,66 @@
        ]
      }
    },
+    "/api/session/active": {
+      "get": {
+        "tags": ["sessions"],
+        "operationId": "v2.session.active",
+        "parameters": [],
+        "security": [],
+        "responses": {
+          "200": {
+            "description": "Success",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "properties": {
+                    "data": {
+                      "type": "object",
+                      "patternProperties": {
+                        "^ses": {
+                          "$ref": "#/components/schemas/SessionActive"
+                        }
+                      }
+                    }
+                  },
+                  "required": ["data"],
+                  "additionalProperties": false
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "InvalidRequestError",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/InvalidRequestError"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "UnauthorizedError",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/UnauthorizedError"
+                }
+              }
+            }
+          }
+        },
+        "description": "Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.",
+        "summary": "List active sessions",
+        "x-codeSamples": [
+          {
+            "lang": "js",
+            "source": "import { createOpencodeClient } from \"@opencode-ai/sdk\n\nconst client = createOpencodeClient()\nawait client.v2.session.active({\n  ...\n})"
+          }
+        ]
+      }
+    },
    "/api/session/{sessionID}": {
      "get": {
        "tags": ["sessions"],
@@ -23445,6 +23505,17 @@
        "required": ["_tag", "message"],
        "additionalProperties": false
      },
+      "SessionActive": {
+        "type": "object",
+        "properties": {
+          "type": {
+            "type": "string",
+            "enum": ["running"]
+          }
+        },
+        "required": ["type"],
+        "additionalProperties": false
+      },
      "SessionNotFoundError": {
        "type": "object",
        "properties": {
@@ -76,6 +76,16 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
          }
        }),
      )
+      .handle(
+        "session.active",
+        Effect.fn(function* () {
+          return {
+            data: Object.fromEntries(
+              Array.from(yield* session.active, (sessionID) => [sessionID, { type: "running" as const }]),
+            ),
+          }
+        }),
+      )
      .handle(
        "session.get",
        Effect.fn(function* (ctx) {
@@ -227,7 +227,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
            if (!currentAssistant) return
            currentAssistant.time.completed = event.data.timestamp
            currentAssistant.finish = event.data.finish
-            currentAssistant.settlement = "completed"
            currentAssistant.cost = event.data.cost
            currentAssistant.tokens = event.data.tokens
            if (event.data.snapshot)
@@ -240,18 +239,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
            if (!currentAssistant) return
            currentAssistant.time.completed = event.data.timestamp
            currentAssistant.finish = "error"
-            currentAssistant.settlement = "failed"
            currentAssistant.error = event.data.error
          })
          break
-        case "session.next.step.interrupted":
-          message.update(event.data.sessionID, (draft) => {
-            const currentAssistant = message.assistant(draft, event.data.assistantMessageID)
-            if (!currentAssistant) return
-            currentAssistant.time.completed = event.data.timestamp
-            currentAssistant.settlement = "interrupted"
-          })
-          break
        case "session.next.text.started":
          message.update(event.data.sessionID, (draft) => {
            message.assistant(draft, event.data.assistantMessageID)?.content.push({
@@ -370,72 +370,6 @@ test("settles pending tools when a live failure arrives", async () => {
  }
 })

-test("marks an interrupted assistant inactive without a finish reason", async () => {
-  const events = createEventSource()
-  const calls = createFetch(undefined, events)
-  let sync!: ReturnType<typeof useData>
-  let ready!: () => void
-  const mounted = new Promise<void>((resolve) => {
-    ready = resolve
-  })
-
-  function Probe() {
-    sync = useData()
-    onMount(ready)
-    return <box />
-  }
-
-  const app = await testRender(() => (
-    <TestTuiContexts>
-      <SDKProvider url="http://test" directory={directory} events={events.source} fetch={calls.fetch}>
-        <ProjectProvider>
-          <DataProvider>
-            <Probe />
-          </DataProvider>
-        </ProjectProvider>
-      </SDKProvider>
-    </TestTuiContexts>
-  ))
-
-  try {
-    await mounted
-    emitEvent(events, {
-      id: "evt_step_started_interrupted",
-      type: "session.next.step.started",
-      properties: {
-        sessionID: "session-interrupted",
-        assistantMessageID: "msg_interrupted",
-        timestamp: 1,
-        agent: "build",
-        model: { id: "model-1", providerID: "provider-1" },
-      },
-    })
-    emitEvent(events, {
-      id: "evt_step_interrupted",
-      type: "session.next.step.interrupted",
-      properties: {
-        sessionID: "session-interrupted",
-        assistantMessageID: "msg_interrupted",
-        timestamp: 2,
-      },
-    })
-
-    await wait(() => {
-      const message = sync.session.message.list("session-interrupted")?.[0]
-      return message?.type === "assistant" && message.time.completed === 2
-    })
-    const assistant = sync.session.message.list("session-interrupted")?.[0]
-    expect(assistant).toMatchObject({
-      type: "assistant",
-      settlement: "interrupted",
-      time: { completed: 2 },
-    })
-    expect(assistant).not.toHaveProperty("finish")
-  } finally {
-    app.renderer.destroy()
-  }
-})
-
 test("renders admitted prompts only after they become model-visible", async () => {
  const events = createEventSource()
  const calls = createFetch(undefined, events)
@@ -25,6 +25,11 @@ sessions.interrupt(sessionID)
  -> clears a coalesced follow-up wake already registered with this coordinator
  -> preserves durable inbox rows for a later wake or resume
  -> idle or missing Session is a no-op
+
+sessions.active()
+  -> snapshots foreground Session drains owned by this process
+  -> returns only active Session IDs with { type: "running" }
+  -> absence means inactive; activity is not durable across process restarts
 ```

 `session_input` is the durable admission inbox. `PromptAdmitted` records and projects accepted input so pending queue state can be replayed, replicated, and observed by clients. Admitted inputs remain outside model-visible Session history until the serialized runner publishes `Prompted`. Its projector atomically writes the visible user message and marks the inbox row promoted in the same event transaction. The V1-to-V2 shadow bridge publishes the same `Prompted` event for already-visible V1 prompts.
@@ -161,15 +166,17 @@ Post-crash continuation recovery is intentionally deferred. A wake does not infe

 A process-global `SessionRunCoordinator` serializes execution for each local Session while allowing different Sessions to run concurrently. Resumes join active execution, overlapping wakes coalesce into one follow-up, and interruption stops current process-local execution without deleting durable inbox work. The runner enters the Session's current Location when execution starts and fences each new provider turn against that Location.

+The coordinator's active registry is also the source for `sessions.active()`. It represents only foreground Session drains owned by the current process; background subagents and tasks do not add parent Sessions to this registry. The snapshot is runtime state and is empty after a process restart.
+
 Inbox promotion coalesces pending steers in durable admission order. Once continuation would otherwise end, it promotes one queued input at a time in FIFO order. Add explicit inbox backlog and steering-batch limits before exposing broad multi-caller admission or untrusted queue growth.

 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.
Author	SHA1	Message	Date
Kit Langton	2ce8dfda5f	test(sdk): cover embedded session live tail	2026-06-26 00:00:38 -04:00
Kit Langton	5d27fac711	fix(sdk): preserve session stream ordering	2026-06-25 23:50:59 -04:00
Kit Langton	605fb0b6c5	feat(sdk): unify session event stream	2026-06-25 23:50:42 -04:00
opencode-agent[bot]	41f2705e72	chore: generate	2026-06-26 03:35:39 +00:00
Kit Langton	ef5c9f4931	feat(sdk): expose active sessions (#33991 )	2026-06-26 03:33:35 +00:00