feat(sdk): restore session runtime operations

AGENTS.md

@@ -1,4 +1,6 @@
-- To regenerate the JavaScript SDK, run `./packages/sdk/js/script/build.ts`.
+- To regenerate the legacy JavaScript SDK, run `./packages/sdk/js/script/build.ts`.
+- After changing the public Protocol or Server `HttpApi`, run `bun run generate` from `packages/client`. Do not edit `src/generated` or `src/generated-effect` directly.
+- Keep runtime dependencies directed from Schema to Core and Protocol, then from Core and Protocol to Server. Client runtime code may depend on Schema and Protocol but never Core or Server; `sdk-next` composes Client, Core, and Server.
 - The default branch in this repo is `dev`.
 - Local `main` ref may not exist; use `dev` or `origin/dev` for diffs.
 

CONTEXT.md

@@ -61,7 +61,7 @@ A temporary file created under OpenCode's shared tool-output directory to retain
 The host-supplied environment overlay applied by the server when creating a PTY, observed for the request Location and resolved PTY working directory.
 
 **OpenCode Client**:
-The generated Effect API shared by networked and in-process consumers, executed through an `HttpClient` against the same `HttpApi` router and handlers.
+The generated Promise and Effect APIs derived from the public `HttpApi`; **Embedded OpenCode** shares the Effect API through an in-memory `HttpClient` against the same router and handlers.
 _Avoid_: Remote client
 
 **SDK Contract IR**:
@@ -149,17 +149,19 @@ _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 a public raw Session event stream transported as SSE in both networked and embedded modes. It verifies the Session, replays durable events after the optional aggregate sequence, then continues with raw live events for that Session, including durable events and ephemeral text, reasoning, and tool-input fragments.
+- A `sessions.events(...)` recovery cursor is the greatest observed `event.durable.seq`. Durable events carry their aggregate sequence in the existing raw event envelope; ephemeral events omit `durable` and never advance the cursor. Reconnecting with `after` replays only later durable events because ephemeral events are not recoverable.
+- The replay-to-live handoff subscribes to live events before reading durable history, emits historical durable events first, buffers concurrent live events during catch-up, deduplicates buffered durable events already covered by replay, then flushes the remaining buffer in publication order before continuing live.
 - `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 mixed 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`; ephemeral events observed after that sequence are intentionally not replayed. Any reusable resume helper remains a separate API design question.
 - 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 `SessionMessageNotFoundError` without disclosing cross-Session ownership. Absence is not represented as `undefined` across the public HTTP boundary.
+- `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.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?

packages/client/src/generated-effect/.httpapi-codegen.json

@@ -1 +1,5 @@
-["client-error.ts", "client.ts", "index.ts"]
+[
+  "client-error.ts",
+  "client.ts",
+  "index.ts"
+]

packages/client/src/generated-effect/client.ts

@@ -1,5 +1,5 @@
 // Generated by @opencode-ai/httpapi-codegen. Do not edit.
-import { Effect, Schema } from "effect"
+import { Effect, Stream, Schema } from "effect"
 import { Sse } from "effect/unstable/encoding"
 import { HttpClientError } from "effect/unstable/http"
 import { HttpApi, HttpApiClient } from "effect/unstable/httpapi"
@@ -143,6 +143,35 @@ const Endpoint0_11 = (raw: RawClient["server.session"]) => (input: Endpoint0_11I
     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"]
+}
+const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12Input) =>
+  Stream.unwrap(
+    raw["session.events"]({ params: { sessionID: input.sessionID }, query: { after: input.after } }).pipe(
+      Effect.mapError(mapClientError),
+      Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))),
+    ),
+  )
+
+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) =>
+  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"]
+}
+const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14Input) =>
+  raw["session.message"]({ params: { sessionID: input.sessionID, messageID: input.messageID } }).pipe(
+    Effect.mapError(mapClientError),
+    Effect.map((value) => value.data),
+  )
+
 const adaptGroup0 = (raw: RawClient["server.session"]) => ({
   list: Endpoint0_0(raw),
   create: Endpoint0_1(raw),
@@ -156,6 +185,9 @@ const adaptGroup0 = (raw: RawClient["server.session"]) => ({
   clear: Endpoint0_9(raw),
   commit: Endpoint0_10(raw),
   context: Endpoint0_11(raw),
+  events: Endpoint0_12(raw),
+  interrupt: Endpoint0_13(raw),
+  message: Endpoint0_14(raw),
 })
 
 const adaptClient = (raw: RawClient) => ({ sessions: adaptGroup0(raw["server.session"]) })

packages/client/src/generated/.httpapi-codegen.json

@@ -1 +1,6 @@
-["client-error.ts", "client.ts", "index.ts", "types.ts"]
+[
+  "client-error.ts",
+  "client.ts",
+  "index.ts",
+  "types.ts"
+]

packages/client/src/generated/client.ts

@@ -23,6 +23,12 @@ import type {
   SessionsCommitOutput,
   SessionsContextInput,
   SessionsContextOutput,
+  SessionsEventsInput,
+  SessionsEventsOutput,
+  SessionsInterruptInput,
+  SessionsInterruptOutput,
+  SessionsMessageInput,
+  SessionsMessageOutput,
 } from "./types"
 import { ClientError } from "./client-error"
 
@@ -306,6 +312,40 @@ export function make(options: ClientOptions) {
           },
           requestOptions,
         ).then((value) => value.data),
+      events: (input: SessionsEventsInput, requestOptions?: RequestOptions): AsyncIterable<SessionsEventsOutput> =>
+        sse<SessionsEventsOutput>(
+          {
+            method: "GET",
+            path: `/api/session/${encodeURIComponent(input.sessionID)}/event`,
+            query: { after: input.after },
+            successStatus: 200,
+            declaredStatuses: [404, 400, 401],
+            empty: false,
+          },
+          requestOptions,
+        ),
+      interrupt: (input: SessionsInterruptInput, requestOptions?: RequestOptions) =>
+        request<SessionsInterruptOutput>(
+          {
+            method: "POST",
+            path: `/api/session/${encodeURIComponent(input.sessionID)}/interrupt`,
+            successStatus: 204,
+            declaredStatuses: [404, 400, 401],
+            empty: true,
+          },
+          requestOptions,
+        ),
+      message: (input: SessionsMessageInput, requestOptions?: RequestOptions) =>
+        request<{ readonly data: SessionsMessageOutput }>(
+          {
+            method: "GET",
+            path: `/api/session/${encodeURIComponent(input.sessionID)}/message/${encodeURIComponent(input.messageID)}`,
+            successStatus: 200,
+            declaredStatuses: [404, 400, 401],
+            empty: false,
+          },
+          requestOptions,
+        ).then((value) => value.data),
     },
   }
 }

packages/client/src/generated/types.ts

@@ -629,3 +629,767 @@ export type SessionsContextOutput = {
       }
   >
 }["data"]
+
+export type SessionsEventsInput = {
+  readonly sessionID: { readonly sessionID: string }["sessionID"]
+  readonly after?: { readonly after?: string | undefined }["after"]
+}
+
+export type SessionsEventsOutput =
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.agent.switched"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly agent: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.model.switched"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string | undefined }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.moved"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly location: { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        readonly subdirectory?: string | undefined
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.prompted"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly prompt: {
+          readonly text: string
+          readonly files?:
+            | ReadonlyArray<{
+                readonly uri: string
+                readonly mime: string
+                readonly name?: string | undefined
+                readonly description?: string | undefined
+                readonly source?: { readonly start: number; readonly end: number; readonly text: string } | undefined
+              }>
+            | undefined
+          readonly agents?:
+            | ReadonlyArray<{
+                readonly name: string
+                readonly source?: { readonly start: number; readonly end: number; readonly text: string } | undefined
+              }>
+            | undefined
+        }
+        readonly delivery: "steer" | "queue"
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.prompt.admitted"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly prompt: {
+          readonly text: string
+          readonly files?:
+            | ReadonlyArray<{
+                readonly uri: string
+                readonly mime: string
+                readonly name?: string | undefined
+                readonly description?: string | undefined
+                readonly source?: { readonly start: number; readonly end: number; readonly text: string } | undefined
+              }>
+            | undefined
+          readonly agents?:
+            | ReadonlyArray<{
+                readonly name: string
+                readonly source?: { readonly start: number; readonly end: number; readonly text: string } | undefined
+              }>
+            | undefined
+        }
+        readonly delivery: "steer" | "queue"
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.context.updated"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly text: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.synthetic"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly text: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.shell.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly callID: string
+        readonly command: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.shell.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly callID: string
+        readonly output: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.step.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly agent: string
+        readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string | undefined }
+        readonly snapshot?: string | undefined
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.step.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly finish: string
+        readonly cost: number
+        readonly tokens: {
+          readonly input: number
+          readonly output: number
+          readonly reasoning: number
+          readonly cache: { readonly read: number; readonly write: number }
+        }
+        readonly snapshot?: string | undefined
+        readonly files?: ReadonlyArray<string> | undefined
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.step.failed"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly error: { readonly type: "unknown"; readonly message: string }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.text.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly textID: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.text.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      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 } | undefined
+      readonly type: "session.next.text.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly textID: string
+        readonly text: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.reasoning.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly reasoningID: string
+        readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } } | undefined
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.reasoning.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      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 } | undefined
+      readonly type: "session.next.reasoning.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      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 } } | undefined
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.tool.input.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly name: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.tool.input.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      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 } | undefined
+      readonly type: "session.next.tool.input.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly text: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.tool.called"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly tool: string
+        readonly input: { readonly [x: string]: unknown }
+        readonly provider: {
+          readonly executed: boolean
+          readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } | undefined
+        }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.tool.progress"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly structured: { readonly [x: string]: any }
+        readonly content: ReadonlyArray<
+          | { readonly type: "text"; readonly text: string }
+          | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string | undefined }
+        >
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.tool.success"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly structured: { readonly [x: string]: any }
+        readonly content: ReadonlyArray<
+          | { readonly type: "text"; readonly text: string }
+          | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string | undefined }
+        >
+        readonly outputPaths?: ReadonlyArray<string> | undefined
+        readonly result?: unknown | undefined
+        readonly provider: {
+          readonly executed: boolean
+          readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } | undefined
+        }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.tool.failed"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly assistantMessageID: string
+        readonly callID: string
+        readonly error: { readonly type: "unknown"; readonly message: string }
+        readonly result?: unknown | undefined
+        readonly provider: {
+          readonly executed: boolean
+          readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } | undefined
+        }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.retried"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly attempt: number
+        readonly error: {
+          readonly message: string
+          readonly statusCode?: number | undefined
+          readonly isRetryable: boolean
+          readonly responseHeaders?: { readonly [x: string]: string } | undefined
+          readonly responseBody?: string | undefined
+          readonly metadata?: { readonly [x: string]: string } | undefined
+        }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.compaction.started"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly reason: "auto" | "manual"
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.compaction.delta"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly text: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.compaction.ended"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly messageID: string
+        readonly reason: "auto" | "manual"
+        readonly text: string
+        readonly recent: string
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.revert.staged"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: {
+        readonly timestamp: number
+        readonly sessionID: string
+        readonly revert: {
+          readonly messageID: string
+          readonly partID?: string | undefined
+          readonly snapshot?: string | undefined
+          readonly diff?: string | undefined
+          readonly files?:
+            | ReadonlyArray<{
+                readonly path: string
+                readonly status: "added" | "modified" | "deleted"
+                readonly additions: number
+                readonly deletions: number
+                readonly patch: string
+              }>
+            | undefined
+        }
+      }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.revert.cleared"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: { readonly timestamp: number; readonly sessionID: string }
+    }
+  | {
+      readonly id: string
+      readonly metadata?: { readonly [x: string]: unknown } | undefined
+      readonly type: "session.next.revert.committed"
+      readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } | undefined
+      readonly location?:
+        | { readonly directory: string; readonly workspaceID?: string | undefined | undefined }
+        | undefined
+      readonly data: { readonly timestamp: number; readonly sessionID: string; readonly messageID: string }
+    }
+
+export type SessionsInterruptInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] }
+
+export type SessionsInterruptOutput = void
+
+export type SessionsMessageInput = {
+  readonly sessionID: { readonly sessionID: string; readonly messageID: string }["sessionID"]
+  readonly messageID: { readonly sessionID: string; readonly messageID: string }["messageID"]
+}
+
+export type SessionsMessageOutput = {
+  readonly data:
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number }
+        readonly type: "agent-switched"
+        readonly agent: string
+      }
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number }
+        readonly type: "model-switched"
+        readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string | null }
+      }
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number }
+        readonly text: string
+        readonly files?: ReadonlyArray<{
+          readonly uri: string
+          readonly mime: string
+          readonly name?: string | null
+          readonly description?: string | null
+          readonly source?: { readonly start: number; readonly end: number; readonly text: string } | null
+        }> | null
+        readonly agents?: ReadonlyArray<{
+          readonly name: string
+          readonly source?: { readonly start: number; readonly end: number; readonly text: string } | null
+        }> | null
+        readonly type: "user"
+      }
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number }
+        readonly sessionID: string
+        readonly text: string
+        readonly type: "synthetic"
+      }
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number }
+        readonly type: "system"
+        readonly text: string
+      }
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number; readonly completed?: number | null }
+        readonly type: "shell"
+        readonly callID: string
+        readonly command: string
+        readonly output: string
+      }
+    | {
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number; readonly completed?: number | null }
+        readonly type: "assistant"
+        readonly agent: string
+        readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string | null }
+        readonly content: ReadonlyArray<
+          | { readonly type: "text"; readonly id: string; readonly text: string }
+          | {
+              readonly type: "reasoning"
+              readonly id: string
+              readonly text: string
+              readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } | null
+            }
+          | {
+              readonly type: "tool"
+              readonly id: string
+              readonly name: string
+              readonly provider?: {
+                readonly executed: boolean
+                readonly metadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } | null
+                readonly resultMetadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } | null
+              } | null
+              readonly state:
+                | { readonly status: "pending"; readonly input: string }
+                | {
+                    readonly status: "running"
+                    readonly input: { readonly [x: string]: JsonValue }
+                    readonly structured: { readonly [x: string]: any }
+                    readonly content: ReadonlyArray<
+                      | { readonly type: "text"; readonly text: string }
+                      | {
+                          readonly type: "file"
+                          readonly uri: string
+                          readonly mime: string
+                          readonly name?: string | null
+                        }
+                    >
+                  }
+                | {
+                    readonly status: "completed"
+                    readonly input: { readonly [x: string]: JsonValue }
+                    readonly attachments?: ReadonlyArray<{
+                      readonly uri: string
+                      readonly mime: string
+                      readonly name?: string | null
+                      readonly description?: string | null
+                      readonly source?: { readonly start: number; readonly end: number; readonly text: string } | null
+                    }> | null
+                    readonly content: ReadonlyArray<
+                      | { readonly type: "text"; readonly text: string }
+                      | {
+                          readonly type: "file"
+                          readonly uri: string
+                          readonly mime: string
+                          readonly name?: string | null
+                        }
+                    >
+                    readonly outputPaths?: ReadonlyArray<string> | null
+                    readonly structured: { readonly [x: string]: any }
+                    readonly result?: JsonValue | null
+                  }
+                | {
+                    readonly status: "error"
+                    readonly input: { readonly [x: string]: JsonValue }
+                    readonly content: ReadonlyArray<
+                      | { readonly type: "text"; readonly text: string }
+                      | {
+                          readonly type: "file"
+                          readonly uri: string
+                          readonly mime: string
+                          readonly name?: string | null
+                        }
+                    >
+                    readonly structured: { readonly [x: string]: any }
+                    readonly error: { readonly type: "unknown"; readonly message: string }
+                    readonly result?: JsonValue | null
+                  }
+              readonly time: {
+                readonly created: number
+                readonly ran?: number | null
+                readonly completed?: number | null
+                readonly pruned?: number | null
+              }
+            }
+        >
+        readonly snapshot?: {
+          readonly start?: string | null
+          readonly end?: string | null
+          readonly files?: ReadonlyArray<string> | null
+        } | null
+        readonly finish?: string | null
+        readonly cost?: number | null
+        readonly tokens?: {
+          readonly input: number
+          readonly output: number
+          readonly reasoning: number
+          readonly cache: { readonly read: number; readonly write: number }
+        } | null
+        readonly error?: { readonly type: "unknown"; readonly message: string } | null
+      }
+    | {
+        readonly type: "compaction"
+        readonly reason: "auto" | "manual"
+        readonly summary: string
+        readonly recent: string
+        readonly id: string
+        readonly metadata?: { readonly [x: string]: JsonValue } | null
+        readonly time: { readonly created: number }
+      }
+}["data"]

packages/client/test/effect.test.ts

@@ -1,7 +1,7 @@
 import { expect, test } from "bun:test"
-import { DateTime, Effect } from "effect"
+import { DateTime, Effect, Stream } from "effect"
 import { HttpClient, HttpClientResponse } from "effect/unstable/http"
-import { AbsolutePath, Agent, Location, Model, OpenCode, Prompt, Session } from "../src/effect"
+import { AbsolutePath, Agent, Location, Model, OpenCode, Prompt, Session, SessionMessage } from "../src/effect"
 
 test("sessions.get returns the decoded Effect projection", async () => {
   const httpClient = HttpClient.make((request) =>
@@ -18,12 +18,25 @@ test("sessions.get returns the decoded Effect projection", async () => {
 test("session methods retain decoded Effect inputs and outputs", async () => {
   const httpClient = HttpClient.make((request) => {
     const url = request.url
+    if (url.includes("/event")) {
+      return Effect.succeed(
+        HttpClientResponse.fromWeb(
+          request,
+          new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\n`, {
+            headers: { "content-type": "text/event-stream" },
+          }),
+        ),
+      )
+    }
     if (url.includes("/prompt")) {
       return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(admission)))
     }
     if (url.includes("/context")) {
       return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ data: [] })))
     }
+    if (url.includes("/message/")) {
+      return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ data: modelSwitchedMessage })))
+    }
     if (request.method === "POST" && url.endsWith("/api/session")) {
       return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(session)))
     }
@@ -53,7 +66,15 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
     yield* client.sessions.compact({ sessionID: Session.ID.make("ses_test") })
     yield* client.sessions.wait({ sessionID: Session.ID.make("ses_test") })
     const context = yield* client.sessions.context({ sessionID: Session.ID.make("ses_test") })
-    return { page, created, admitted, context }
+    const events = yield* client.sessions
+      .events({ sessionID: Session.ID.make("ses_test"), after: 0 })
+      .pipe(Stream.runCollect)
+    yield* client.sessions.interrupt({ sessionID: Session.ID.make("ses_test") })
+    const message = yield* client.sessions.message({
+      sessionID: Session.ID.make("ses_test"),
+      messageID: SessionMessage.ID.make("msg_model"),
+    })
+    return { page, 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)
@@ -64,6 +85,8 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
   expect(Object.getPrototypeOf(result.admitted.prompt)).toBe(Object.prototype)
   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.message).toEqual(expect.objectContaining({ id: "msg_model", type: "model-switched" }))
 })
 
 const session = {
@@ -96,3 +119,22 @@ const admission = {
     timeCreated: 1_717_171_717_000,
   },
 }
+
+const modelSwitchedMessage = {
+  id: "msg_model",
+  type: "model-switched",
+  time: { created: 1_717_171_717_000 },
+  model: { id: "claude", providerID: "anthropic" },
+}
+
+const modelSwitchedEvent = {
+  id: "evt_model",
+  type: "session.next.model.switched",
+  durable: { aggregateID: "ses_test", seq: 1, version: 1 },
+  data: {
+    timestamp: 1_717_171_717_000,
+    sessionID: "ses_test",
+    messageID: "msg_model",
+    model: { id: "claude", providerID: "anthropic" },
+  },
+}

packages/client/test/promise.test.ts

@@ -24,8 +24,14 @@ test("session methods use the public HTTP contract", async () => {
     fetch: async (input, init) => {
       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" },
+        })
+      }
       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 (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" } })
@@ -47,11 +53,17 @@ test("session methods use the public HTTP contract", async () => {
   await client.sessions.compact({ sessionID: "ses_test" })
   await client.sessions.wait({ sessionID: "ses_test" })
   const context = await client.sessions.context({ sessionID: "ses_test" })
+  const events = []
+  for await (const event of client.sessions.events({ sessionID: "ses_test", after: "0" })) events.push(event)
+  await client.sessions.interrupt({ sessionID: "ses_test" })
+  const message = await client.sessions.message({ sessionID: "ses_test", messageID: "msg_model" })
 
   expect(page.cursor.next).toBe("next")
   expect(created.id).toBe("ses_test")
   expect(admitted.id).toBe("msg_test")
   expect(context).toEqual([])
+  expect(events).toEqual([modelSwitchedEvent])
+  expect(message).toEqual(modelSwitchedMessage)
   expect(requests.map((request) => [request.init?.method, request.url])).toEqual([
     ["GET", "http://localhost:3000/api/session?limit=10&order=desc"],
     ["POST", "http://localhost:3000/api/session"],
@@ -61,6 +73,9 @@ test("session methods use the public HTTP contract", async () => {
     ["POST", "http://localhost:3000/api/session/ses_test/compact"],
     ["POST", "http://localhost:3000/api/session/ses_test/wait"],
     ["GET", "http://localhost:3000/api/session/ses_test/context"],
+    ["GET", "http://localhost:3000/api/session/ses_test/event?after=0"],
+    ["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
   if (typeof body !== "string") throw new Error("Expected JSON request body")
@@ -115,3 +130,22 @@ const admission = {
     timeCreated: 1_717_171_717_000,
   },
 }
+
+const modelSwitchedMessage = {
+  id: "msg_model",
+  type: "model-switched",
+  time: { created: 1_717_171_717_000 },
+  model: { id: "claude", providerID: "anthropic" },
+}
+
+const modelSwitchedEvent = {
+  id: "evt_model",
+  type: "session.next.model.switched",
+  durable: { aggregateID: "ses_test", seq: 1, version: 1 },
+  data: {
+    timestamp: 1_717_171_717_000,
+    sessionID: "ses_test",
+    messageID: "msg_model",
+    model: { id: "claude", providerID: "anthropic" },
+  },
+}

packages/core/src/event.ts

@@ -67,6 +67,11 @@ 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 follow: <A extends Payload>(input: {
+    readonly aggregateID: string
+    readonly after?: number
+    readonly matches: (event: Payload) => event is A
+  }) => Stream.Stream<A>
   /** @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>
@@ -86,6 +91,7 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Ev
 
 export interface LayerOptions {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
+  readonly followerCapacity?: number
 }
 
 export const layerWith = (options?: LayerOptions) =>
@@ -546,6 +552,48 @@ export const layerWith = (options?: LayerOptions) =>
           })
         })
 
+      const follow = <A extends Payload>(input: {
+        readonly aggregateID: string
+        readonly after?: number
+        readonly matches: (event: Payload) => event is A
+      }): Stream.Stream<A> =>
+        Stream.unwrap(
+          Effect.gen(function* () {
+            const pubsub = yield* PubSub.dropping<A>(options?.followerCapacity ?? 1024)
+            const subscription = yield* PubSub.subscribe(pubsub)
+            const listener = (event: Payload) =>
+              input.matches(event)
+                ? PubSub.publish(pubsub, event).pipe(
+                    Effect.flatMap((published) => (published ? Effect.void : PubSub.shutdown(pubsub))),
+                  )
+                : Effect.void
+            yield* Effect.acquireRelease(listen(listener), (unsubscribe) =>
+              unsubscribe.pipe(Effect.andThen(PubSub.shutdown(pubsub))),
+            )
+            let sequence = input.after ?? -1
+            const read: Effect.Effect<ReadonlyArray<A>> = Effect.suspend(() =>
+              readAfter(input.aggregateID, sequence),
+            ).pipe(
+              Effect.map((events) => events.flatMap((event) => (input.matches(event) ? [event] : []))),
+              Effect.tap((events) =>
+                Effect.sync(() => {
+                  sequence = events.at(-1)?.durable?.seq ?? sequence
+                }),
+              ),
+            )
+            const historical = yield* read
+            const live = Stream.fromSubscription(subscription).pipe(
+              Stream.filter((event) => {
+                if (event.durable?.aggregateID !== input.aggregateID) return true
+                if (event.durable.seq <= sequence) return false
+                sequence = event.durable.seq
+                return true
+              }),
+            )
+            return Stream.fromIterable(historical).pipe(Stream.concat(live))
+          }),
+        )
+
       const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
         Effect.sync(() => {
           const list = projectors.get(definition.type) ?? []
@@ -558,6 +606,7 @@ export const layerWith = (options?: LayerOptions) =>
         subscribe,
         all: streamAll,
         durable,
+        follow,
         listen,
         project,
         replay,

packages/core/src/session.ts

@@ -128,7 +128,7 @@ export interface Interface {
   readonly events: (input: {
     sessionID: SessionSchema.ID
     after?: number
-  }) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
+  }) => Stream.Stream<SessionEvent.Event, NotFoundError>
   readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect<void, NotFoundError>
   readonly switchModel: (input: {
     sessionID: SessionSchema.ID
@@ -184,7 +184,10 @@ 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 isSessionEvent =
+            (sessionID: SessionSchema.ID) =>
+            (event: EventV2.Payload): event is SessionEvent.Event =>
+              Schema.is(SessionEvent.All)(event) && event.data.sessionID === sessionID
           const decode = (row: typeof SessionMessageTable.$inferSelect) =>
             decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
               Effect.mapError(
@@ -340,10 +343,16 @@ 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))),
+                result.get(input.sessionID).pipe(
+                  Effect.as(
+                    events.follow({
+                      aggregateID: input.sessionID,
+                      after: input.after,
+                      matches: isSessionEvent(input.sessionID),
+                    }),
+                  ),
+                ),
+              ),
             prompt: Effect.fn("V2Session.prompt")((input) =>
               Effect.uninterruptible(
                 Effect.gen(function* () {

packages/core/test/event.test.ts

@@ -58,6 +58,14 @@ const GlobalMessage = EventV2.define({
   },
 })
 
+const LiveMessage = EventV2.define({
+  type: "test.live",
+  schema: {
+    sessionID: Session.ID,
+    text: Schema.String,
+  },
+})
+
 const VersionedMessage = EventV2.define({
   type: "test.versioned",
   durable: {
@@ -418,6 +426,135 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("replays durable history before buffered matching live events", () =>
+    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()
+        yield* events.publish(DurableMessage, durableData(aggregateID, "before"))
+        const matches = (
+          event: EventV2.Payload,
+        ): event is EventV2.Payload & { readonly data: { readonly sessionID: Session.ID } } =>
+          (event.type === DurableMessage.type || event.type === LiveMessage.type) &&
+          typeof event.data === "object" &&
+          event.data !== null &&
+          "sessionID" in event.data &&
+          event.data.sessionID === aggregateID
+        const fiber = yield* events
+          .follow({ aggregateID, matches })
+          .pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped)
+        yield* Deferred.await(readStarted)
+
+        pause = false
+        yield* events.publish(DurableMessage, durableData(aggregateID, "during"))
+        yield* events.publish(LiveMessage, { sessionID: aggregateID, text: "live" })
+        yield* Deferred.succeed(continueRead, undefined)
+
+        expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.type])).toEqual([
+          [0, "message.removed"],
+          [1, "message.removed"],
+          [undefined, "test.live"],
+        ])
+      }).pipe(Effect.provide(Layer.mergeAll(Database.defaultLayer, eventLayer)))
+    }),
+  )
+
+  it.effect("preserves raw live publication order around durable events", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = Session.ID.create()
+      const matches = (
+        event: EventV2.Payload,
+      ): event is EventV2.Payload & { readonly data: { readonly sessionID: Session.ID } } =>
+        (event.type === DurableMessage.type || event.type === LiveMessage.type) &&
+        typeof event.data === "object" &&
+        event.data !== null &&
+        "sessionID" in event.data &&
+        event.data.sessionID === aggregateID
+      const fiber = yield* events
+        .follow({ aggregateID, matches })
+        .pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+
+      yield* events.publish(DurableMessage, durableData(aggregateID, "first"))
+      yield* events.publish(LiveMessage, { sessionID: aggregateID, text: "between" })
+      yield* events.publish(DurableMessage, durableData(aggregateID, "second"))
+
+      expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.type])).toEqual([
+        [0, "message.removed"],
+        [undefined, "test.live"],
+        [1, "message.removed"],
+      ])
+    }),
+  )
+
+  it.effect("terminates a lagging follower without blocking publishers", () =>
+    Effect.gen(function* () {
+      const readStarted = yield* Deferred.make<void>()
+      const continueRead = yield* Deferred.make<void>()
+      const eventLayer = EventV2.layerWith({
+        followerCapacity: 1,
+        beforeAggregateRead: () =>
+          Deferred.succeed(readStarted, undefined).pipe(Effect.andThen(Deferred.await(continueRead))),
+      }).pipe(Layer.provide(Database.defaultLayer))
+
+      yield* Effect.gen(function* () {
+        const events = yield* EventV2.Service
+        const aggregateID = Session.ID.create()
+        const matches = (
+          event: EventV2.Payload,
+        ): event is EventV2.Payload & { readonly data: { readonly sessionID: Session.ID } } =>
+          event.type === LiveMessage.type &&
+          typeof event.data === "object" &&
+          event.data !== null &&
+          "sessionID" in event.data &&
+          event.data.sessionID === aggregateID
+        const fiber = yield* events.follow({ aggregateID, matches }).pipe(Stream.runCollect, Effect.forkScoped)
+        yield* Deferred.await(readStarted)
+
+        yield* events.publish(LiveMessage, { sessionID: aggregateID, text: "fills buffer" })
+        yield* events.publish(LiveMessage, { sessionID: aggregateID, text: "closes follower" })
+        yield* events.publish(LiveMessage, { sessionID: aggregateID, text: "publisher remains live" })
+        yield* Deferred.succeed(continueRead, undefined)
+
+        expect(Array.from(yield* Fiber.join(fiber)).length).toBeLessThanOrEqual(1)
+      }).pipe(Effect.provide(Layer.mergeAll(Database.defaultLayer, eventLayer)))
+    }),
+  )
+
+  it.effect("removes a follower listener when its stream is interrupted", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = Session.ID.create()
+      let matches = 0
+      const fiber = yield* events
+        .follow({
+          aggregateID,
+          matches: (_event): _event is EventV2.Payload => {
+            matches++
+            return false
+          },
+        })
+        .pipe(Stream.runDrain, Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* Fiber.interrupt(fiber)
+
+      yield* events.publish(LiveMessage, { sessionID: aggregateID, text: "after interruption" })
+
+      expect(matches).toBe(0)
+    }),
+  )
+
   it.effect("coalesces durable aggregate wakes while draining every committed event", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service

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

@@ -164,18 +164,25 @@ describe("SessionV2.prompt", () => {
     }),
   )
 
-  it.effect("streams durable Session events after an aggregate sequence", () =>
+  it.effect("replays durable Session events then streams raw live events", () =>
     Effect.gen(function* () {
       yield* setup
       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 })
       yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false })
       yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER)
+      yield* events.publish(SessionEvent.Text.Delta, {
+        sessionID,
+        assistantMessageID: SessionMessage.ID.create(),
+        textID: "text_1",
+        timestamp: yield* DateTime.now,
+        delta: "live",
+      })
       const streamed = Array.from(yield* Fiber.join(fiber))
 
       expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([
@@ -183,12 +190,11 @@ describe("SessionV2.prompt", () => {
         [1, "session.next.prompt.admitted"],
         [2, "session.next.prompted"],
         [3, "session.next.prompted"],
+        [undefined, "session.next.text.delta"],
       ])
       expect(
         Array.from(
-          yield* session
-            .events({ sessionID, after: streamed[0]!.durable?.seq })
-            .pipe(Stream.take(1), Stream.runCollect),
+          yield* session.events({ sessionID, after: streamed[0].durable?.seq }).pipe(Stream.take(1), Stream.runCollect),
         ).map((event) => [event.durable?.seq, event.type]),
       ).toEqual([[1, "session.next.prompt.admitted"]])
     }),

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

@@ -29,6 +29,7 @@ const capture = () => {
     subscribe: () => Stream.empty,
     all: () => Stream.empty,
     durable: () => Stream.empty,
+    follow: () => Stream.empty,
     listen: () => Effect.succeed(Effect.void),
     project: () => Effect.void,
     replay: () => Effect.void,

packages/httpapi-codegen/README.md

@@ -37,6 +37,6 @@ The existing public `generate(Api, { directory })` operation writes the rich Eff
 
 Generation formats TypeScript with Prettier before writing. Output paths are flat, unique, and checked against traversal, reserved manifest names, and existing symbolic links.
 
-Generated source starts with one self-contained module per `HttpApiGroup`, plus root client and index modules. Schema dependencies may be duplicated across group modules. Cross-group schema partitioning is deferred until measured output or bundle cost requires it.
+Portable Effect output uses one self-contained module per `HttpApiGroup`, plus root client and index modules. Promise output uses shared type and client modules, while imported Effect output keeps adapters in the root client module. Schema dependencies may be duplicated across portable Effect group modules. Cross-group schema partitioning is deferred until measured output or bundle cost requires it.
 
-Codegen preserves group and endpoint identifiers exactly. The composed remote `HttpApi` owns public names such as `session` and `get`; the generator performs no prefix stripping, casing conversion, or public-name annotation mapping.
+Codegen preserves transport identifiers internally. `compile` may explicitly map consumer-facing group names, and endpoint operation IDs are projected to their final dot-delimited segment. The generator performs no other implicit product-specific naming or public-name annotation mapping.

packages/httpapi-codegen/src/index.ts

@@ -69,6 +69,7 @@ type Slot = {
 
 const resolveHttpApiStatus = SchemaAST.resolveAt<number>("httpApiStatus")
 const resolveHttpApiEncoding = SchemaAST.resolveAt<HttpApiSchema.Encoding>("~httpApiEncoding")
+const resolveContentSchema = SchemaAST.resolveAt<SchemaAST.AST>("contentSchema")
 const Manifest = Schema.fromJsonString(Schema.Array(Schema.String))
 const manifestName = ".httpapi-codegen.json"
 
@@ -125,9 +126,10 @@ export function compile<Id extends string, Groups extends HttpApiGroup.Any>(
         ...responseSchemas(success.schema, `${name}.success`),
         ...errorSchemas.map((item) => [`${name}.error`, item.schema] as const),
       ]
-      const effectPortable = [params, query, headers, ...payloads, success, ...errorSchemas].every(
-        (item) => item?.effectPortable !== false,
-      )
+      const effectPortable =
+        [params, query, headers, ...payloads, success, ...errorSchemas].every(
+          (item) => item?.effectPortable !== false,
+        ) && streamEffectPortable(success.schema)
       if (effectPortable) {
         for (const [path, schema] of schemaPaths) assertPortable(schema, path, portable)
       }
@@ -454,7 +456,7 @@ function renderPromiseTypes(groups: ReadonlyArray<Group>) {
         const success = typeOf(
           isStreamSchema(successSchema) && successSchema._tag === "StreamSse"
             ? successSchema.sseMode === "data"
-              ? streamDataSchema(successSchema)
+              ? streamEncodedDataSchema(successSchema)
               : successSchema.events
             : successSchema,
         )
@@ -782,17 +784,6 @@ function responseSchemas(schema: Schema.Top, path: string): Array<readonly [stri
   if (!isStreamSchema(schema)) return [[path, schema]]
   if (schema._tag === "StreamUint8Array") return []
   const value = schema.sseMode === "data" ? streamDataSchema(schema) : schema.events
-  const rebuilt =
-    schema.sseMode === "data"
-      ? HttpApiSchema.StreamSse({ data: value, error: schema.error, contentType: schema.contentType })
-      : HttpApiSchema.StreamSse({
-          events: schema.events,
-          error: schema.error,
-          contentType: schema.contentType,
-        })
-  if (!sameEncoding(schema.events.ast, rebuilt.events.ast)) {
-    throw new GenerationError({ reason: `Unportable schema: ${path}.${schema.sseMode}` })
-  }
   return [
     [`${path}.${schema.sseMode}`, value],
     [`${path}.error`, schema.error],
@@ -964,11 +955,33 @@ function isStreamSchema(schema: Schema.Top): schema is HttpApiSchema.StreamSchem
 }
 
 function streamDataSchema(schema: Extract<HttpApiSchema.StreamSchema, { readonly _tag: "StreamSse" }>) {
-  const ast = Schema.toType(schema.events).ast
+  return Schema.make(streamDataAst(Schema.toType(schema.events).ast))
+}
+
+function streamEncodedDataSchema(schema: Extract<HttpApiSchema.StreamSchema, { readonly _tag: "StreamSse" }>) {
+  const data = streamDataAst(schema.events.ast)
+  const encodedAst = data.encoding?.at(-1)?.to
+  if (encodedAst === undefined) throw new GenerationError({ reason: "Invalid SSE data schema" })
+  const encoded = resolveContentSchema(encodedAst)
+  if (!SchemaAST.isAST(encoded)) throw new GenerationError({ reason: "Invalid SSE data schema" })
+  return Schema.make(encoded)
+}
+
+function streamDataAst(ast: SchemaAST.AST) {
   if (!SchemaAST.isObjects(ast)) throw new GenerationError({ reason: "Invalid SSE data schema" })
   const data = ast.propertySignatures.find((field) => field.name === "data")?.type
   if (data === undefined) throw new GenerationError({ reason: "Invalid SSE data schema" })
-  return Schema.make(data)
+  return data
+}
+
+function streamEffectPortable(schema: Schema.Top) {
+  if (!isStreamSchema(schema) || schema._tag === "StreamUint8Array" || schema.sseMode === "events") return true
+  const rebuilt = HttpApiSchema.StreamSse({
+    data: streamDataSchema(schema),
+    error: schema.error,
+    contentType: schema.contentType,
+  })
+  return sameEncoding(schema.events.ast, rebuilt.events.ast)
 }
 
 function renderGroup(group: Group, groupIndex: number) {

packages/httpapi-codegen/test/generate.test.ts

@@ -395,7 +395,9 @@ describe("HttpApiCodegen.generate", () => {
         api(
           HttpApiEndpoint.get("subscribe", "/event", {
             query: { after: Schema.optional(Schema.Number) },
-            success: HttpApiSchema.StreamSse({ data: Schema.Struct({ type: Schema.String }) }),
+            success: HttpApiSchema.StreamSse({
+              data: Schema.Struct({ type: Schema.String, count: Schema.NumberFromString }),
+            }),
           }),
         ),
       ),
@@ -416,7 +418,7 @@ describe("HttpApiCodegen.generate", () => {
           return new Response(
             new ReadableStream({
               start(controller) {
-                controller.enqueue(encoder.encode('data: {"type":"ready"}\r'))
+                controller.enqueue(encoder.encode('data: {"type":"ready","count":"1"}\r'))
                 controller.enqueue(encoder.encode("\n\r\n"))
                 controller.close()
               },
@@ -430,7 +432,7 @@ describe("HttpApiCodegen.generate", () => {
       expect(requests).toBe(0)
       const received = []
       for await (const event of events) received.push(event)
-      expect(received).toEqual([{ type: "ready" }])
+      expect(received).toEqual([{ type: "ready", count: "1" }])
       expect(requests).toBe(1)
       expect(url).toBe("https://example.com/event?after=2")
     } finally {

packages/protocol/src/groups/session.ts

@@ -3,7 +3,7 @@ import { SessionInput } from "@opencode-ai/schema/session-input"
 import { Prompt } from "@opencode-ai/schema/prompt"
 import { Session } from "@opencode-ai/schema/session"
 import { Project } from "@opencode-ai/schema/project"
-import { AbsolutePath, PositiveInt, RelativePath, withStatics } from "@opencode-ai/schema/schema"
+import { AbsolutePath, NonNegativeInt, PositiveInt, RelativePath, withStatics } from "@opencode-ai/schema/schema"
 import { Workspace } from "@opencode-ai/schema/workspace"
 import { Context, Encoding, Result, Schema, Struct } from "effect"
 import { HttpApiEndpoint, HttpApiGroup, HttpApiMiddleware, HttpApiSchema, OpenApi } from "effect/unstable/httpapi"
@@ -20,6 +20,7 @@ import { Agent } from "@opencode-ai/schema/agent"
 import { Model } from "@opencode-ai/schema/model"
 import { Location } from "@opencode-ai/schema/location"
 import { Revert } from "@opencode-ai/schema/revert"
+import { SessionEvent } from "@opencode-ai/schema/session-event"
 
 const SessionsQueryFields = {
   workspace: Workspace.ID.pipe(Schema.optional),
@@ -32,6 +33,10 @@ const SessionsQueryFields = {
   search: Schema.optional(Schema.String),
 }
 
+const SessionEventSchema: Schema.Codec<typeof SessionEvent.All.Type, typeof SessionEvent.All.Encoded> = Schema.make(
+  SessionEvent.All.ast,
+)
+
 const SessionsDirectoryQuery = Schema.Struct({
   ...SessionsQueryFields,
   directory: AbsolutePath,
@@ -272,6 +277,55 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
           }),
         ),
     )
+    .add(
+      HttpApiEndpoint.get("session.events", "/api/session/:sessionID/event", {
+        params: { sessionID: Session.ID },
+        query: {
+          after: Schema.NumberFromString.pipe(Schema.decodeTo(NonNegativeInt), Schema.optional),
+        },
+        success: HttpApiSchema.StreamSse({ data: SessionEventSchema }),
+        error: SessionNotFoundError,
+      })
+        .middleware(sessionLocationMiddleware)
+        .annotateMerge(
+          OpenApi.annotations({
+            identifier: "v2.session.events",
+            summary: "Subscribe to session events",
+            description:
+              "Replay durable events after an aggregate sequence, then continue with raw live Session events.",
+          }),
+        ),
+    )
+    .add(
+      HttpApiEndpoint.post("session.interrupt", "/api/session/:sessionID/interrupt", {
+        params: { sessionID: Session.ID },
+        success: HttpApiSchema.NoContent,
+        error: SessionNotFoundError,
+      })
+        .middleware(sessionLocationMiddleware)
+        .annotateMerge(
+          OpenApi.annotations({
+            identifier: "v2.session.interrupt",
+            summary: "Interrupt session execution",
+            description: "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op.",
+          }),
+        ),
+    )
+    .add(
+      HttpApiEndpoint.get("session.message", "/api/session/:sessionID/message/:messageID", {
+        params: { sessionID: Session.ID, messageID: SessionMessage.ID },
+        success: Schema.Struct({ data: SessionMessage.Message }),
+        error: [SessionNotFoundError, MessageNotFoundError],
+      })
+        .middleware(sessionLocationMiddleware)
+        .annotateMerge(
+          OpenApi.annotations({
+            identifier: "v2.session.message",
+            summary: "Get session message",
+            description: "Retrieve one projected message owned by the Session.",
+          }),
+        ),
+    )
     .annotateMerge(
       OpenApi.annotations({
         title: "sessions",

packages/sdk-next/README.md

@@ -11,7 +11,9 @@ const opencode = yield * OpenCode.create()
 const session = yield * opencode.sessions.get({ sessionID })
 ```
 
-It also exposes local-only `tools.register(...)`. Closing the owning Effect Scope releases router resources, location services, fibers, and scoped tool registrations.
+It also exports `Tool` and exposes local-only `tools.register(...)`, replacing the former `@opencode-ai/core/public` facade. Registration uses Core's host-level `ApplicationTools` service shared by the host's Locations; each Location retains its own `ToolRegistry` for overlay, lookup, and settlement. Closing the owning Effect Scope releases router resources, location services, fibers, and scoped tool registrations.
+
+`sessions.events({ sessionID, after })` first replays durable events after the optional aggregate sequence, then emits raw live Session events, including ephemeral deltas. Only `event.durable.seq` advances the recovery cursor. `sessions.interrupt(...)` targets execution owned by this host, and `sessions.message(...)` retrieves one projected Session message.
 
 The same constructor is available as a service Layer:
 

packages/sdk-next/src/opencode.ts

@@ -2,40 +2,24 @@ import { OpenCode } from "@opencode-ai/client/effect"
 import { PermissionSaved } from "@opencode-ai/core/permission/saved"
 import { ApplicationTools } from "@opencode-ai/core/tool/application-tools"
 import { createEmbeddedRoutes } from "@opencode-ai/server/routes"
-import { Cause, Context, Effect, Layer } from "effect"
-import {
-  HttpClient,
-  HttpRouter,
-  HttpServer,
-  HttpServerError,
-  HttpServerRequest,
-  HttpServerResponse,
-} from "effect/unstable/http"
+import { Context, Effect, Layer } from "effect"
+import { HttpClient, HttpClientRequest, HttpClientResponse, HttpRouter, HttpServer } from "effect/unstable/http"
 
 export const create = Effect.fn("OpenCode.create")(function* () {
   const applicationTools = ApplicationTools.layer
-  const { handler, permissions, tools } = yield* Effect.all({
-    // Reusing this Layer value lets registration and every Location share one memoized host-level registry.
-    handler: HttpRouter.toHttpEffect(
-      createEmbeddedRoutes().pipe(Layer.provide(applicationTools), Layer.provide(HttpServer.layerServices)),
-    ),
-    permissions: PermissionSaved.Service,
-    tools: ApplicationTools.Service,
-  }).pipe(Effect.provide(Layer.merge(applicationTools, PermissionSaved.defaultLayer)))
-  const httpClient = HttpClient.make(
-    Effect.fnUntraced(function* (request) {
-      const response = yield* handler.pipe(
-        Effect.provideService(HttpServerRequest.HttpServerRequest, HttpServerRequest.fromClientRequest(request)),
-        Effect.provideService(ApplicationTools.Service, tools),
-        Effect.provideService(PermissionSaved.Service, permissions),
-        Effect.catchCause((cause) =>
-          Cause.hasInterruptsOnly(cause)
-            ? Effect.interrupt
-            : HttpServerError.causeResponse(cause).pipe(Effect.map(([response]) => response)),
-        ),
-      )
-      return HttpServerResponse.toClientResponse(response, { request })
-    }, Effect.scoped),
+  const services = Layer.merge(applicationTools, PermissionSaved.defaultLayer)
+  const context = yield* Layer.build(services)
+  const web = HttpRouter.toWebHandler(
+    createEmbeddedRoutes().pipe(Layer.provide(Layer.succeedContext(context)), Layer.provide(HttpServer.layerServices)),
+    { disableLogger: true },
+  )
+  yield* Effect.addFinalizer(() => Effect.promise(web.dispose))
+  const tools = Context.get(context, ApplicationTools.Service)
+  const httpClient = HttpClient.make((request, _url, signal) =>
+    Effect.gen(function* () {
+      const input = yield* HttpClientRequest.toWeb(request, { signal }).pipe(Effect.orDie)
+      return HttpClientResponse.fromWeb(request, yield* Effect.promise(() => web.handler(input, context)))
+    }),
   )
   const client = yield* OpenCode.make({ baseUrl: "http://opencode.local" }).pipe(
     Effect.provideService(HttpClient.HttpClient, httpClient),

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

@@ -3,7 +3,7 @@ import { mkdtemp, rm } from "node:fs/promises"
 import { tmpdir } from "node:os"
 import { join } from "node:path"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { Effect, Schema } from "effect"
+import { Effect, Option, Schema, Stream } from "effect"
 
 test("embedded client uses the real router and handlers", async () => {
   const directory = await mkdtemp(join(tmpdir(), "opencode-embedded-"))
@@ -39,8 +39,31 @@ test("embedded client uses the real router and handlers", async () => {
         resume: false,
       })
       const context = yield* opencode.sessions.context({ sessionID })
-      const missing = yield* Effect.flip(
-        opencode.sessions.get({ sessionID: Session.ID.make(`ses_missing_${crypto.randomUUID()}`) }),
+      const event = yield* opencode.sessions
+        .events({ sessionID })
+        .pipe(Stream.take(1), Stream.runHead, Effect.map(Option.getOrUndefined))
+      const modelMessage = Option.fromNullishOr(context.find((message) => message.type === "model-switched")).pipe(
+        Option.getOrThrow,
+      )
+      const message = yield* opencode.sessions.message({ sessionID, messageID: modelMessage.id })
+      yield* opencode.sessions.interrupt({ sessionID })
+      const other = yield* opencode.sessions.create({
+        location: Location.Ref.make({ directory: AbsolutePath.make(directory) }),
+      })
+      const missingSessionID = Session.ID.make(`ses_missing_${crypto.randomUUID()}`)
+      const missing = yield* Effect.all(
+        [
+          opencode.sessions.events({ sessionID: missingSessionID }).pipe(Stream.runHead, Effect.flip),
+          opencode.sessions.interrupt({ sessionID: missingSessionID }).pipe(Effect.flip),
+          opencode.sessions.message({ sessionID: missingSessionID, messageID: modelMessage.id }).pipe(Effect.flip),
+        ],
+        { concurrency: "unbounded" },
+      )
+      const missingMessage = yield* Effect.flip(
+        opencode.sessions.message({
+          sessionID: other.id,
+          messageID: modelMessage.id,
+        }),
       )
 
       expect(created.id).toBe(sessionID)
@@ -49,7 +72,14 @@ test("embedded client uses the real router and handlers", async () => {
       expect(page.data.some((session) => session.id === sessionID)).toBe(true)
       expect(admitted.sessionID).toBe(sessionID)
       expect(context.some((message) => message.type === "model-switched")).toBe(true)
-      expect(missing._tag).toBe("SessionNotFoundError")
+      expect(event).toMatchObject({ type: "session.next.model.switched", durable: { seq: 1 } })
+      expect(message).toEqual(modelMessage)
+      expect(missing.map((error) => error._tag)).toEqual([
+        "SessionNotFoundError",
+        "SessionNotFoundError",
+        "SessionNotFoundError",
+      ])
+      expect(missingMessage._tag).toBe("MessageNotFoundError")
     })
     await Effect.runPromise(Effect.scoped(program))
   } finally {

packages/server/src/handlers/session.ts

@@ -1,5 +1,5 @@
 import { SessionV2 } from "@opencode-ai/core/session"
-import { DateTime, Effect } from "effect"
+import { DateTime, Effect, Stream } from "effect"
 import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi"
 import { Api } from "../api"
 import { SessionsCursor } from "@opencode-ai/protocol/groups/session"
@@ -318,5 +318,32 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
           }
         }),
       )
+      .handle(
+        "session.events",
+        Effect.fn((ctx) =>
+          Effect.succeed(
+            session.events({ sessionID: ctx.params.sessionID, after: ctx.query.after }).pipe(Stream.orDie),
+          ),
+        ),
+      )
+      .handle(
+        "session.interrupt",
+        Effect.fn(function* (ctx) {
+          yield* session.interrupt(ctx.params.sessionID)
+          return HttpApiSchema.NoContent.make()
+        }),
+      )
+      .handle(
+        "session.message",
+        Effect.fn(function* (ctx) {
+          const message = yield* session.message(ctx.params)
+          if (message) return { data: message }
+          return yield* new MessageNotFoundError({
+            sessionID: ctx.params.sessionID,
+            messageID: ctx.params.messageID,
+            message: `Message not found: ${ctx.params.messageID}`,
+          })
+        }),
+      )
   }),
 )

specs/v2/schema-changelog.md

@@ -583,7 +583,7 @@ Reason:
 Compatibility:
 
 - No migration, synchronized event version, OpenAPI, or SDK regeneration is required.
-- `sessions.events({ sessionID, after? })` remains a replay-and-tail stream of every durable event in aggregate sequence order.
+- `sessions.events({ sessionID, after? })` replays every later durable event in aggregate sequence order, then tails raw live Session events. Ephemeral events never advance the recovery cursor.
 
 ## 2026-06-03: Sequential V2 Apply Patch Tool
 

specs/v2/session.md

@@ -165,17 +165,17 @@ Inbox promotion coalesces pending steers in durable admission order. Once contin
 
 Eager local-tool execution is intentionally unbounded in the current local slice. This minimizes tool latency but does not increase SQLite settlement throughput: Session-event publication remains serialized per provider turn. Before broadening exposure, revisit per-turn call limits, output truncation, and operational backpressure using observed workloads. The `session.next.*` event schemas remain experimental and unshipped; databases created by earlier experimental builds are disposable rather than compatibility targets.
 
-The synchronized `session.next.*` event family and projected Session-message model predate this branch. This slice refines their replay contract: projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers can use `sessions.events({ sessionID, after? })` to replay durable `session.next.*` events after an aggregate sequence cursor, then tail durable events without a race. Live-only text, reasoning, and tool-input fragments remain available through EventV2 subscriptions for connected renderers; they are intentionally absent from the replayable Session stream.
+The synchronized `session.next.*` event family and projected Session-message model predate this branch. 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 raw live Session events without a race. Live-only text, reasoning, and tool-input fragments appear only after the stream reaches its live boundary and never advance the recovery cursor.
 
-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 such as Discord publication. 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. Until that contract is designed, connected renderers can combine `sessions.events(...)` with direct EventV2 delta subscriptions.
+The first `sessions.events(...)` contract replays durable history and then tails the raw live Session stream, including ephemeral deltas while connected. The recovery cursor remains the greatest observed durable aggregate sequence: ephemeral fragments cannot advance it, replay after reconnect, or be mistaken for durable publication boundaries. The replay-to-live handoff subscribes before reading history, buffers concurrent live events until catch-up completes, deduplicates durable overlap, then flushes the remaining buffer in publication order.
 
-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.
+The internal durable-only `EventV2.durable(...)` tail uses advisory edge-triggered wakeups. 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. The mixed `sessions.events(...)` follower instead buffers matching raw live events in publication order during replay; its bounded buffer terminates lagging consumers so they can reconnect from their last durable cursor rather than silently losing events.
 
 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.
 
 ## Current Tool Registry Slice
 
-`ToolRegistry` is Location-scoped. Contributions are scoped replayable transforms: closing a contribution scope removes its definition and rebuilds the advertised catalog. Execution decodes input, optionally authorizes the call, invokes the retained handler, validates output, and settles failures as typed tool-result errors.
+`ApplicationTools` stores process-scoped application registrations shared by all Locations. Each Location-scoped `ToolRegistry` overlays Location registrations, materializes definitions, and owns lookup and settlement. Closing a contribution scope removes its definition and rebuilds the advertised catalog. Trusted tool executors capture and perform authorization; the registry applies catalog visibility filtering, decodes input, invokes the retained handler, validates output, and settles failures as typed tool-result errors.
 
 When a Session omits `agent`, both execution and permission evaluation use the default `build` agent. A caller must not observe `build` model behavior while permission checks silently evaluate an empty no-agent policy.