diff --git a/CONTEXT.md b/CONTEXT.md
index 5e5955d344..e6e5977998 100644
--- a/CONTEXT.md
+++ b/CONTEXT.md
@@ -162,12 +162,14 @@ _Avoid_: Response envelope
 - SDK executes Server's assembled `HttpRouter` in memory. It opens no listener and performs no network I/O, while preserving Server routing, middleware, codecs, handlers, and errors.
 - The Effect Client and SDK re-export their decoded datatype facade from Schema so callers do not depend on internal package locations or Core's versioned names.
 - A capability intended for both networked and **Embedded OpenCode** belongs in the authoritative public `HttpApi`; embedded-only same-process capabilities extend **Embedded OpenCode** separately.
-- `sessions.events({ sessionID, after })` is a public durable Session event stream. It verifies the Session, replays durable events after the optional aggregate sequence, continues with newly committed durable events, excludes live-only fragments, and is transported as SSE in both networked and embedded modes.
+- `sessions.events({ sessionID, after })` is one public Session-scoped event stream. It verifies the Session, captures a fixed durable cutoff after registering observation, replays durable events in `(after, cutoff]`, emits one authoritative process-local `session.activity` value, then continues with committed durable events and live-only Session output fragments. Only durable events carry aggregate-sequence cursor metadata.
 - `events.subscribe()` is a distinct public instance-wide live stream for Session and non-Session activity. It has no replay guarantee and includes connection, heartbeat, and instance-disposal lifecycle events; consumers recover from disconnection by refreshing authoritative state.
 - A Session ID is not an optional filter on `events.subscribe()`: instance-wide live events and durable Session events have different schemas, replay guarantees, cursors, lifecycle events, and failure behavior.
 - The initial common OpenCode Client does not expose server-global event aggregation. `events.subscribe()` is bounded to the connected OpenCode instance or workspace; any future cross-instance administrative stream requires a separately designed API.
 - `events.subscribe()` does not automatically reconnect after transport loss. The live-only stream fails with `ClientError`; consumers refresh authoritative state before explicitly opening a new subscription because events missed during disconnection cannot be replayed.
-- `sessions.events({ sessionID, after })` returns the generated HTTP client's cold durable event stream and does not build reconnection policy into the endpoint or client constructor. Transport loss fails the stream with `ClientError`. Callers may compose an explicit resuming stream above it by retaining the last observed durable sequence and opening a new subscription with `after`; any reusable resume helper remains a separate API design question.
+- `sessions.events({ sessionID, after })` returns the generated HTTP client's cold Session event stream and does not build reconnection policy into the endpoint or client constructor. Transport loss fails the stream with `ClientError`. Callers resume with the last observed durable sequence; live-only fragments are not replayed, and every connection receives a fresh authoritative activity value.
+- The Session event stream treats database rows as authoritative and process notifications as bounded, non-blocking wakeups. Before emitting an observed live-only fragment it drains later durable rows, preserving causal durable start boundaries without exposing an additional fence field.
+- Aggregate deletion currently removes the Session's durable event rows and sequence, including deletion history. The Session event stream therefore cannot promise replay after deletion until retention or a typed history-expired outcome is designed.
 - The stable `sessions.list(...)` design returns a **Page** in both networked and **Embedded OpenCode**; embedded execution does not define a separate unbounded array-returning list operation. The beta client currently preserves the existing HTTP `{ data, cursor }` envelope until emitter-level Page projection is implemented.
 - Session list cursors are opaque branded values carrying continuation query and ordering state. Consumers pass them back unchanged and do not inspect storage anchors or encoded filter fields.
 - A Session list continuation accepts only its opaque cursor. Scope, filters, ordering, and page size are fixed by the initial query and carried by that cursor.
diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts
index f426a50f87..fbc9d47b84 100644
--- a/packages/client/src/generated/types.ts
+++ b/packages/client/src/generated/types.ts
@@ -601,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 }
@@ -810,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 }
@@ -824,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 }
@@ -838,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 }
@@ -932,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 }
@@ -994,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 }
diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts
index 6ca73c22f4..12d1316c17 100644
--- a/packages/client/test/effect.test.ts
+++ b/packages/client/test/effect.test.ts
@@ -22,7 +22,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
       return Effect.succeed(
         HttpClientResponse.fromWeb(
           request,
-          new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\n`, {
+          new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(activityEvent)}\n\n`, {
             headers: { "content-type": "text/event-stream" },
           }),
         ),
@@ -93,6 +93,8 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
   expect(DateTime.toEpochMillis(result.admitted.timeCreated)).toBe(1_717_171_717_000)
   expect(result.context).toEqual([])
   expect(DateTime.toEpochMillis(result.events[0].data.timestamp)).toBe(1_717_171_717_000)
+  expect(result.events[1]).toEqual(activityEvent)
+  expect(result.events[1]).not.toHaveProperty("durable")
   expect(result.message).toEqual(expect.objectContaining({ id: "msg_model", type: "model-switched" }))
 })
 
@@ -145,3 +147,9 @@ const modelSwitchedEvent = {
     model: { id: "claude", providerID: "anthropic" },
   },
 }
+
+const activityEvent = {
+  id: "evt_activity",
+  type: "session.activity" as const,
+  data: { sessionID: "ses_test", active: false },
+}
diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts
index 952a3fc0d7..953dc73733 100644
--- a/packages/client/test/promise.test.ts
+++ b/packages/client/test/promise.test.ts
@@ -25,9 +25,12 @@ test("session methods use the public HTTP contract", async () => {
       const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url
       requests.push({ url, init })
       if (url.includes("/event")) {
-        return new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\n`, {
-          headers: { "content-type": "text/event-stream" },
-        })
+        return new Response(
+          `data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(activityEvent)}\n\n`,
+          {
+            headers: { "content-type": "text/event-stream" },
+          },
+        )
       }
       if (url.includes("/prompt")) return Response.json(admission)
       if (url.includes("/context")) return Response.json({ data: [] })
@@ -65,7 +68,8 @@ test("session methods use the public HTTP contract", async () => {
   expect(created.id).toBe("ses_test")
   expect(admitted.id).toBe("msg_test")
   expect(context).toEqual([])
-  expect(events).toEqual([modelSwitchedEvent])
+  expect(events).toEqual([modelSwitchedEvent, activityEvent])
+  expect(events[1]).not.toHaveProperty("durable")
   expect(message).toEqual(modelSwitchedMessage)
   expect(requests.map((request) => [request.init?.method, request.url])).toEqual([
     ["GET", "http://localhost:3000/api/session?limit=10&order=desc"],
@@ -153,3 +157,9 @@ const modelSwitchedEvent = {
     model: { id: "claude", providerID: "anthropic" },
   },
 }
+
+const activityEvent = {
+  id: "evt_activity",
+  type: "session.activity",
+  data: { sessionID: "ses_test", active: false },
+}
diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts
index 5324c319b0..1a1c08c3c9 100644
--- a/packages/core/src/event.ts
+++ b/packages/core/src/event.ts
@@ -1,9 +1,9 @@
 export * as EventV2 from "./event"
 
-import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
+import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Scope, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
-import { and, asc, eq, gt } from "drizzle-orm"
+import { and, asc, eq, gt, lte } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
@@ -67,6 +67,15 @@ 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> },
+    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>
@@ -472,13 +481,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(),
           ),
@@ -537,6 +552,42 @@ export const layerWith = (options?: LayerOptions) =>
           }),
         )
 
+      const observeAggregate: Interface["observeAggregate"] = (input) =>
+        Effect.gen(function* () {
+          type Signal = { readonly _tag: "durable" } | { readonly _tag: "live"; readonly event: Payload }
+          const signals = yield* Queue.dropping<Signal, Cause.Done>(256)
+          const unsubscribe = yield* listen((event) =>
+            Effect.sync(() => {
+              if (event.durable?.aggregateID === input.aggregateID) {
+                // Durable payloads are only wakeups; a full queue already contains work that will drain the database.
+                Queue.offerUnsafe(signals, { _tag: "durable" })
+                return
+              }
+              if (!input.live(event)) return
+              if (!Queue.offerUnsafe(signals, { _tag: "live", event })) Queue.endUnsafe(signals)
+            }),
+          )
+          yield* Effect.addFinalizer(() => unsubscribe.pipe(Effect.andThen(Queue.shutdown(signals))))
+
+          const cutoff = yield* latestSequence(db, input.aggregateID)
+          const replay = yield* readAfter(input.aggregateID, input.after ?? -1, cutoff)
+          let sequence = cutoff
+          const drain = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
+            Effect.tap((events) =>
+              Effect.sync(() => {
+                sequence = events.at(-1)?.durable?.seq ?? sequence
+              }),
+            ),
+          )
+          const updates = Stream.fromQueue(signals).pipe(
+            Stream.mapEffect((signal) =>
+              drain.pipe(Effect.map((events) => (signal._tag === "live" ? [...events, signal.event] : events))),
+            ),
+            Stream.flattenIterable,
+          )
+          return { replay, updates }
+        })
+
       const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
         Effect.sync(() => {
           listeners.push(listener)
@@ -558,6 +609,7 @@ export const layerWith = (options?: LayerOptions) =>
         subscribe,
         all: streamAll,
         durable,
+        observeAggregate,
         listen,
         project,
         replay,
diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts
index 97745b41bf..d422dca323 100644
--- a/packages/core/src/session.ts
+++ b/packages/core/src/session.ts
@@ -106,6 +106,7 @@ export type MessageNotFoundError = SessionRevert.MessageNotFoundError
 export type Error = NotFoundError | MessageDecodeError | OperationUnavailableError | PromptConflictError
 
 export interface Interface {
+  readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
   readonly list: (input?: ListInput) => Effect.Effect<SessionSchema.Info[]>
   readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info>
   readonly get: (sessionID: SessionSchema.ID) => Effect.Effect<SessionSchema.Info, NotFoundError>
@@ -128,7 +129,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
@@ -186,6 +187,7 @@ export const layer = Layer.unwrap(
           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 decode = (row: typeof SessionMessageTable.$inferSelect) =>
             decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
               Effect.mapError(
@@ -198,6 +200,7 @@ export const layer = Layer.unwrap(
             )
 
           const result = Service.of({
+            active: execution.active,
             create: Effect.fn("V2Session.create")(function* (input) {
               const sessionID = input.id ?? SessionSchema.ID.create()
               const recorded = yield* store.get(sessionID)
@@ -341,10 +344,46 @@ 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 &&
+                      sessionEventTypes.has(event.type) &&
+                      typeof event.data === "object" &&
+                      event.data !== null &&
+                      "sessionID" in event.data &&
+                      event.data.sessionID === input.sessionID,
+                  })
+                  const initialActivity: SessionEvent.Activity = {
+                    id: EventV2.ID.create(),
+                    type: SessionEvent.Activity.type,
+                    data: { sessionID: input.sessionID, active: yield* activity.snapshot },
+                  }
+                  const replay = observed.replay.filter((event): event is SessionEvent.DurableEvent =>
+                    isDurableSessionEvent(event),
+                  )
+                  const updates = observed.updates.pipe(
+                    Stream.filter((event): event is SessionEvent.Event => Schema.is(SessionEvent.All)(event)),
+                  )
+                  const activityChanges = activity.changes.pipe(
+                    Stream.map(
+                      (active): SessionEvent.Activity => ({
+                        id: EventV2.ID.create(),
+                        type: SessionEvent.Activity.type,
+                        data: { sessionID: input.sessionID, active },
+                      }),
+                    ),
+                  )
+                  return Stream.fromIterable(replay).pipe(
+                    Stream.concat(Stream.make(initialActivity)),
+                    Stream.concat(Stream.merge(updates, activityChanges)),
+                  )
+                }),
+              ),
             prompt: Effect.fn("V2Session.prompt")((input) =>
               Effect.uninterruptible(
                 Effect.gen(function* () {
diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts
index f92929e6ee..13eb8de860 100644
--- a/packages/core/src/session/execution.ts
+++ b/packages/core/src/session/execution.ts
@@ -1,12 +1,20 @@
 export * as SessionExecution from "./execution"
 
-import { Context, Effect, Layer } from "effect"
+import { Context, Effect, Layer, Scope, Stream } from "effect"
 import { SessionRunner } from "./runner/index"
 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<
+    { readonly snapshot: Effect.Effect<boolean>; readonly changes: Stream.Stream<boolean> },
+    never,
+    Scope.Scope
+  >
   /** Starts execution while idle or joins the active execution. */
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
   /** Registers newly recorded work. Repeated wakeups may coalesce. */
@@ -23,6 +31,7 @@ export const noopLayer = Layer.succeed(
   Service,
   Service.of({
     active: Effect.succeed(new Set()),
+    activity: () => Effect.succeed({ snapshot: Effect.succeed(false), changes: Stream.never }),
     resume: () => Effect.void,
     wake: () => Effect.void,
     interrupt: () => Effect.void,
diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts
index bd1be74539..0ee2b50a54 100644
--- a/packages/core/src/session/execution/local.ts
+++ b/packages/core/src/session/execution/local.ts
@@ -29,6 +29,7 @@ export const layer = Layer.effect(
 
     return SessionExecution.Service.of({
       active: coordinator.active,
+      activity: coordinator.activity,
       interrupt: coordinator.interrupt,
       resume: coordinator.run,
       wake: coordinator.wake,
diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts
index 2f89aff9e3..00738147eb 100644
--- a/packages/core/src/session/run-coordinator.ts
+++ b/packages/core/src/session/run-coordinator.ts
@@ -1,11 +1,19 @@
 export * as SessionRunCoordinator from "./run-coordinator"
 
-import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, FiberSet, Queue, Scope, Stream } from "effect"
+
+export interface Activity {
+  /** Discards earlier transitions and snapshots current ownership. */
+  readonly snapshot: Effect.Effect<boolean>
+  readonly changes: Stream.Stream<boolean>
+}
 
 /** Serializes execution for each key while allowing different keys to run concurrently. */
 export interface Coordinator<Key, E> {
   /** Snapshots keys with an execution owned by this coordinator. */
   readonly active: Effect.Effect<ReadonlySet<Key>>
+  /** Registers transition observation before taking its authoritative snapshot. */
+  readonly activity: (key: Key) => Effect.Effect<Activity, never, Scope.Scope>
   /** Starts execution while idle or joins the active execution. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Registers one coalesced follow-up after newly recorded work. */
@@ -26,6 +34,7 @@ export const make = <Key, E>(options: {
 }): Effect.Effect<Coordinator<Key, E>, never, Scope.Scope> =>
   Effect.gen(function* () {
     const active = new Map<Key, Entry<E>>()
+    const activityObservers = new Map<Key, Set<(active: boolean) => void>>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
 
     const makeEntry = (): Entry<E> => ({
@@ -34,6 +43,10 @@ export const make = <Key, E>(options: {
       stopping: false,
     })
 
+    const notifyActivity = (key: Key, value: boolean) => {
+      for (const observer of activityObservers.get(key) ?? []) observer(value)
+    }
+
     const start = (key: Key, entry: Entry<E>, force: boolean, successor = false) => {
       const ready = Deferred.makeUnsafe<void>()
       const owner = fork(
@@ -56,8 +69,10 @@ export const make = <Key, E>(options: {
       }
 
       const successor = entry.pendingWake ? makeEntry() : undefined
-      if (successor === undefined) active.delete(key)
-      else {
+      if (successor === undefined) {
+        active.delete(key)
+        notifyActivity(key, false)
+      } else {
         active.set(key, successor)
         start(key, successor, false, true)
       }
@@ -74,6 +89,7 @@ export const make = <Key, E>(options: {
 
         const next = makeEntry()
         active.set(key, next)
+        notifyActivity(key, true)
         start(key, next, true)
         return restore(Deferred.await(next.done))
       })
@@ -88,6 +104,7 @@ export const make = <Key, E>(options: {
 
         const next = makeEntry()
         active.set(key, next)
+        notifyActivity(key, true)
         start(key, next, false)
       })
 
@@ -100,5 +117,30 @@ export const make = <Key, E>(options: {
         return Fiber.interrupt(entry.owner)
       })
 
-    return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt }
+    const activity = (key: Key) =>
+      Effect.gen(function* () {
+        const queue = yield* Queue.dropping<boolean, Cause.Done>(256)
+        const observer = (value: boolean) => {
+          if (!Queue.offerUnsafe(queue, value)) Queue.endUnsafe(queue)
+        }
+        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)
+            }).pipe(Effect.andThen(Queue.shutdown(queue))),
+        )
+        return {
+          snapshot: Queue.clear(queue).pipe(Effect.andThen(Effect.sync(() => active.has(key)))),
+          changes: Stream.fromQueue(queue),
+        }
+      })
+
+    return { active: Effect.sync(() => new Set(active.keys())), activity, run, wake, interrupt }
   })
diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts
index e2b2a5df04..0c6280fe54 100644
--- a/packages/core/test/event.test.ts
+++ b/packages/core/test/event.test.ts
@@ -418,6 +418,76 @@ 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("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 aggregate wakes while draining every committed event", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service
diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts
index 9ff0e22490..1fc1301321 100644
--- a/packages/core/test/session-prompt.test.ts
+++ b/packages/core/test/session-prompt.test.ts
@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { DateTime, Effect, Fiber, Layer, Stream } from "effect"
+import { DateTime, Deferred, Effect, Fiber, Layer, Stream } from "effect"
 import { eq } from "drizzle-orm"
 import { Database } from "@opencode-ai/core/database/database"
 import { EventV2 } from "@opencode-ai/core/event"
@@ -23,10 +23,29 @@ const executionCalls: SessionV2.ID[] = []
 const interruptCalls: SessionV2.ID[] = []
 const wakeCalls: SessionV2.ID[] = []
 const activeSessions = new Set<SessionV2.ID>()
+let activityObserver: ((active: boolean) => void) | undefined
+let activityState = false
+const setActivity = (active: boolean) =>
+  Effect.sync(() => {
+    activityState = active
+    activityObserver?.(active)
+  })
 const execution = Layer.succeed(
   SessionExecution.Service,
   SessionExecution.Service.of({
     active: Effect.sync(() => new Set(activeSessions)),
+    activity: () =>
+      Effect.sync(() => {
+        activityState = false
+        activityObserver = undefined
+        return {
+          attach: (observer: (active: boolean) => void) =>
+            Effect.sync(() => {
+              activityObserver = observer
+              return activityState
+            }),
+        }
+      }),
     resume: (sessionID) =>
       Effect.sync(() => {
         executionCalls.push(sessionID)
@@ -179,7 +198,7 @@ describe("SessionV2.prompt", () => {
       const session = yield* SessionV2.Service
       const events = yield* EventV2.Service
       const { db } = yield* Database.Service
-      const fiber = yield* session.events({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
+      const fiber = yield* session.events({ sessionID }).pipe(Stream.take(5), Stream.runCollect, Effect.forkScoped)
       yield* Effect.yieldNow
 
       yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false })
@@ -188,6 +207,7 @@ describe("SessionV2.prompt", () => {
       const streamed = Array.from(yield* Fiber.join(fiber))
 
       expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([
+        [undefined, "session.activity"],
         [0, "session.next.prompt.admitted"],
         [1, "session.next.prompt.admitted"],
         [2, "session.next.prompted"],
@@ -196,10 +216,44 @@ 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",
+      ])
     }),
   )
 
diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts
index dfbeda664c..302a8956fe 100644
--- a/packages/core/test/session-run-coordinator.test.ts
+++ b/packages/core/test/session-run-coordinator.test.ts
@@ -99,6 +99,27 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("reflects activity races in the snapshot or subsequent transitions", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const gate = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate))),
+        })
+        const activity = yield* coordinator.activity("session")
+
+        const run = yield* coordinator.run("session").pipe(Effect.forkChild)
+        yield* Deferred.await(started)
+        const transitions = new Array<boolean>()
+        expect(yield* activity.attach((active) => transitions.push(active))).toBeTrue()
+        yield* Deferred.succeed(gate, undefined)
+        yield* Fiber.join(run)
+        expect(transitions).toEqual([false])
+      }),
+    ),
+  )
+
   it.effect("cleans active executions after failure and defect", () =>
     Effect.scoped(
       Effect.gen(function* () {
diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts
index da8699fbd2..2ea3a1c55c 100644
--- a/packages/core/test/session-runner-recorded.test.ts
+++ b/packages/core/test/session-runner-recorded.test.ts
@@ -96,6 +96,7 @@ const execution = Layer.effect(
     })
     return SessionExecution.Service.of({
       active: coordinator.active,
+      activity: coordinator.activity,
       resume: coordinator.run,
       wake: coordinator.wake,
       interrupt: coordinator.interrupt,
diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts
index f96ea4dea2..f1b252caed 100644
--- a/packages/core/test/session-runner-tool-events.test.ts
+++ b/packages/core/test/session-runner-tool-events.test.ts
@@ -28,6 +28,7 @@ const capture = () => {
       }),
     subscribe: () => Stream.empty,
     all: () => Stream.empty,
+    observeAggregate: () => Effect.die("not implemented"),
     durable: () => Stream.empty,
     listen: () => Effect.succeed(Effect.void),
     project: () => Effect.void,
diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts
index c9594e8244..8b77d10d0f 100644
--- a/packages/core/test/session-runner.test.ts
+++ b/packages/core/test/session-runner.test.ts
@@ -255,6 +255,7 @@ const execution = Layer.effect(
     })
     return SessionExecution.Service.of({
       active: coordinator.active,
+      activity: coordinator.activity,
       resume: coordinator.run,
       wake: coordinator.wake,
       interrupt: coordinator.interrupt,
diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts
index f1f908e6ac..f623d1bdb8 100644
--- a/packages/protocol/src/groups/session.ts
+++ b/packages/protocol/src/groups/session.ts
@@ -295,7 +295,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
         query: {
           after: Schema.NumberFromString.pipe(Schema.decodeTo(NonNegativeInt), Schema.optional),
         },
-        success: HttpApiSchema.StreamSse({ data: SessionEvent.Durable }),
+        success: HttpApiSchema.StreamSse({ data: SessionEvent.Stream }),
         error: SessionNotFoundError,
       })
         .middleware(sessionLocationMiddleware)
@@ -303,7 +303,8 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
           OpenApi.annotations({
             identifier: "v2.session.events",
             summary: "Subscribe to session events",
-            description: "Replay durable events after an aggregate sequence, then continue with new durable events.",
+            description:
+              "Replay durable events after an aggregate sequence, emit current process-local activity, then continue with durable and live-only Session events.",
           }),
         ),
     )
diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts
index f58c3dcce0..7667b79378 100644
--- a/packages/schema/src/session-event.ts
+++ b/packages/schema/src/session-event.ts
@@ -51,6 +51,15 @@ 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 AgentSwitched = Event.define({
   type: "session.next.agent.switched",
   ...options,
@@ -517,3 +526,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
diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts
index 9beb809922..8b974d99f2 100644
--- a/packages/sdk/js/src/v2/gen/sdk.gen.ts
+++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts
@@ -5713,7 +5713,7 @@ export class Session3 extends HeyApiClient {
   /**
    * Subscribe to session events
    *
-   * Replay durable events after an aggregate sequence, then continue with new durable events.
+   * Replay durable events after an aggregate sequence, emit current process-local activity, then continue with durable and live-only Session events.
    */
   public events<ThrowOnError extends boolean = false>(
     parameters: {
diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts
index ae610a3f12..2f61039abd 100644
--- a/packages/sdk/js/src/v2/gen/types.gen.ts
+++ b/packages/sdk/js/src/v2/gen/types.gen.ts
@@ -4036,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?: {
@@ -4313,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?: {
@@ -4334,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?: {
@@ -4355,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?: {
@@ -4484,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?: {
@@ -4567,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?: {
diff --git a/specs/v2/session.md b/specs/v2/session.md
index 54daa69713..00a16790f4 100644
--- a/specs/v2/session.md
+++ b/specs/v2/session.md
@@ -172,11 +172,11 @@ Inbox promotion coalesces pending steers in durable admission order. Once contin
 
 Eager local-tool execution is intentionally unbounded in the current local slice. This minimizes tool latency but does not increase SQLite settlement throughput: Session-event publication remains serialized per provider turn. Before broadening exposure, revisit per-turn call limits, output truncation, and operational backpressure using observed workloads. The `session.next.*` event schemas remain experimental and unshipped; databases created by earlier experimental builds are disposable rather than compatibility targets.
 
-The synchronized `session.next.*` event family and projected Session-message model predate this branch. This slice refines their replay contract: projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers can use `sessions.events({ sessionID, after? })` to replay durable `session.next.*` events after an aggregate sequence cursor, then tail durable events without a race. Live-only text, reasoning, and tool-input fragments remain available through EventV2 subscriptions for connected renderers; they are intentionally absent from the replayable Session stream.
+The synchronized `session.next.*` event family and projected Session-message model predate this branch. Projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers use `sessions.events({ sessionID, after? })` for one Session-scoped stream containing replayable lifecycle/output events, process-local activity, and connected live-only output fragments.
 
-The first `sessions.events(...)` contract is durable-only during both replay and live tailing. This keeps one cursor equal to one persisted aggregate sequence and is sufficient for reconnect-safe consumers. A later UI-facing API may optionally interleave live-only deltas while connected, but those fragments must remain explicitly ephemeral: they cannot advance the durable cursor, replay after reconnect, or be mistaken for publication boundaries.
+The stream registers observation, captures one fixed durable cutoff `B`, emits durable rows in `(after, B]` in aggregate-sequence order, and then emits exactly one authoritative `session.activity` value. Only events carrying `durable` metadata advance the cursor. Activity and text, reasoning, compaction, or tool-input deltas are process-local and never replay after reconnect. Activity means foreground execution ownership in this process; `false` does not describe a terminal outcome, and background work does not activate its parent Session.
 
-Durable event tail wakeups are advisory and edge-triggered. Each active tail owns one sliding-capacity-1 dirty signal for its aggregate and re-queries SQLite after a wake. Repeated commits coalesce while the tail is busy because durable rows, not in-memory notifications, preserve every event and sequence. Subscribe and register the dirty signal before historical replay, then remove it when the tail closes, so replay handoff cannot miss a commit and inactive aggregates retain no wake state.
+Durable event tail notifications are advisory. Each active tail uses bounded non-blocking observation and re-queries SQLite after a wake. Repeated commits may coalesce because durable rows, not in-memory notifications, preserve every event and sequence. A live-only fragment is emitted only after draining later durable rows, which fences dependent deltas behind their committed start events. If a live consumer cannot keep up, the stream closes rather than blocking Session execution or SQLite writers. Aggregate deletion still removes durable history and remains a documented replay limitation rather than being redesigned here.
 
 Event replay owner claims are separate from clustered Session execution ownership. The former already fences synchronized projection reconstruction; the latter still needs distributed active-run acquisition, stale-runtime rejection, interruption, and placement orchestration.