refactor(server): narrow event feed scope

packages/core/src/event.ts

@@ -1,6 +1,6 @@
 export * as EventV2 from "./event"
 
-import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect"
+import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
 import type { EventLog } from "@opencode-ai/schema/event-log"
@@ -89,11 +89,6 @@ const decodeSerializedEvent = (event: SerializedEvent): Payload => {
   }
 }
 
-export class SubscriberOverflowError extends Schema.TaggedErrorClass<SubscriberOverflowError>()(
-  "EventV2.SubscriberOverflow",
-  { capacity: Schema.Int },
-) {}
-
 export const versionedType = Event.versionedType
 export const durable = Event.durable
 export const ephemeral = Event.ephemeral
@@ -167,27 +162,6 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Event") {}
 
-export const liveBounded = (
-  events: Interface,
-  options: { readonly capacity: number; readonly accept?: (event: Payload) => boolean },
-) =>
-  Effect.gen(function* () {
-    const queue = yield* Queue.dropping<Payload, SubscriberOverflowError>(options.capacity)
-    const unsubscribe = yield* events.listen((event) =>
-      options.accept && !options.accept(event)
-        ? Effect.void
-        : Queue.offer(queue, event).pipe(
-            Effect.flatMap((accepted) =>
-              accepted
-                ? Effect.void
-                : Queue.fail(queue, new SubscriberOverflowError({ capacity: options.capacity })).pipe(Effect.asVoid),
-            ),
-          ),
-    )
-    yield* Effect.addFinalizer(() => unsubscribe.pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid))
-    return Stream.fromQueue(queue)
-  })
-
 export interface LayerOptions {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
   /** Maximum durable rows read per page while replaying or tailing an aggregate log. */

packages/core/test/event.test.ts

@@ -2,8 +2,6 @@ import { describe, expect } from "bun:test"
 import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Ref, Schema, Stream } from "effect"
 import { EventV2 } from "@opencode-ai/core/event"
 import { Event } from "@opencode-ai/schema/event"
-import { EventManifest } from "@opencode-ai/schema/event-manifest"
-import { McpEvent } from "@opencode-ai/schema/mcp-event"
 import { Session } from "@opencode-ai/schema/session"
 import { SessionEvent } from "@opencode-ai/schema/session-event"
 import { SessionV1 } from "@opencode-ai/schema/session-v1"
@@ -363,51 +361,6 @@ describe("EventV2", () => {
     }),
   )
 
-  it.effect("ends only an overflowing bounded subscriber without blocking other listeners", () =>
-    Effect.gen(function* () {
-      const events = yield* EventV2.Service
-      const consuming = yield* Deferred.make<void>()
-      const release = yield* Deferred.make<void>()
-      const slowStream = yield* EventV2.liveBounded(events, { capacity: 1 })
-      const fastStream = yield* EventV2.liveBounded(events, { capacity: 8 })
-      const slow = yield* slowStream.pipe(
-        Stream.runForEach(() => Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Deferred.await(release)))),
-        Effect.forkScoped,
-      )
-      const fast = yield* fastStream.pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
-
-      yield* events.publish(Message, { text: "one" })
-      yield* Deferred.await(consuming)
-      yield* events.publish(Message, { text: "two" })
-      yield* events.publish(Message, { text: "overflow" })
-      const last = yield* events.publish(Message, { text: "still delivered" })
-      yield* Deferred.succeed(release, undefined)
-
-      const slowExit = yield* Fiber.await(slow)
-      expect(Exit.findErrorOption(slowExit).pipe(Option.getOrUndefined)).toBeInstanceOf(EventV2.SubscriberOverflowError)
-      expect(Array.from(yield* Fiber.join(fast))).toEqual([
-        expect.objectContaining({ data: { text: "one" } }),
-        expect.objectContaining({ data: { text: "two" } }),
-        expect.objectContaining({ data: { text: "overflow" } }),
-        last,
-      ])
-    }),
-  )
-
-  it.effect("filters internal events before they enter a bounded server stream", () =>
-    Effect.gen(function* () {
-      const events = yield* EventV2.Service
-      const stream = yield* EventV2.liveBounded(events, { capacity: 1, accept: EventManifest.isServer })
-      const received = yield* stream.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
-
-      yield* events.publish(McpEvent.ToolsChanged, { server: "one" })
-      yield* events.publish(McpEvent.ToolsChanged, { server: "two" })
-      const published = yield* events.publish(McpEvent.StatusChanged, { server: "example" })
-
-      expect(Array.from(yield* Fiber.join(received))).toEqual([published])
-    }),
-  )
-
   it.effect("preserves observer interruption", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service

packages/server/package.json

@@ -9,6 +9,7 @@
     "./*": "./src/*.ts"
   },
   "scripts": {
+    "test": "bun test --only-failures",
     "typecheck": "tsgo --noEmit"
   },
   "dependencies": {

packages/server/src/event-feed.ts

@@ -0,0 +1,95 @@
+export * as EventFeed from "./event-feed"
+
+import { EventV2 } from "@opencode-ai/core/event"
+import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
+import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "effect"
+import { Sse } from "effect/unstable/encoding"
+
+export const SubscriberCapacity = 4_096
+
+export class SubscriberOverflowError extends Schema.TaggedErrorClass<SubscriberOverflowError>()(
+  "EventFeed.SubscriberOverflow",
+  { capacity: Schema.Int },
+) {}
+
+export class EncodingError extends Schema.TaggedErrorClass<EncodingError>()("EventFeed.EncodingError", {
+  eventID: EventV2.ID,
+  eventType: Schema.String,
+  cause: Schema.Defect(),
+}) {}
+
+export type Error = SubscriberOverflowError | EncodingError
+
+export interface Interface {
+  readonly subscribe: Effect.Effect<Stream.Stream<string, Error>, never, Scope.Scope>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/server/EventFeed") {}
+
+const encode = Schema.encodeUnknownSync(OpenCodeEvent)
+
+export function frame(event: OpenCodeEvent) {
+  return Sse.encoder.write({
+    _tag: "Event",
+    event: "message",
+    id: undefined,
+    data: JSON.stringify(encode(event)),
+  })
+}
+
+export const make = Effect.fn("EventFeed.make")(function* (
+  observe: (subscriber: EventV2.Subscriber) => Effect.Effect<EventV2.Unsubscribe>,
+  options?: { readonly capacity?: number; readonly encode?: (event: OpenCodeEvent) => string },
+) {
+  const capacity = options?.capacity ?? SubscriberCapacity
+  const subscribers = new Set<Queue.Queue<string, Error>>()
+
+  const fail = (error: Error) =>
+    Effect.sync(() => {
+      const current = Array.from(subscribers)
+      subscribers.clear()
+      for (const subscriber of current) Queue.failCauseUnsafe(subscriber, Cause.fail(error))
+    })
+
+  const publish = Effect.fnUntraced(function* (event: EventV2.Payload) {
+    if (!isOpenCodeEvent(event)) return
+    if (subscribers.size === 0) return
+    const encoded = yield* Effect.try({
+      try: () => (options?.encode ?? frame)(event),
+      catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }),
+    }).pipe(
+      Effect.catch((error) =>
+        Effect.logError("Failed to encode public event", {
+          eventID: error.eventID,
+          eventType: error.eventType,
+          cause: error.cause,
+        }).pipe(Effect.andThen(fail(error)), Effect.as(undefined)),
+      ),
+    )
+    if (encoded === undefined) return
+    for (const subscriber of subscribers) {
+      if (Queue.offerUnsafe(subscriber, encoded)) continue
+      subscribers.delete(subscriber)
+      Queue.failCauseUnsafe(subscriber, Cause.fail(new SubscriberOverflowError({ capacity })))
+    }
+  })
+
+  const unsubscribe = yield* observe(publish)
+  yield* Effect.addFinalizer(() => unsubscribe)
+
+  return Service.of({
+    subscribe: Effect.acquireRelease(
+      Queue.dropping<string, Error>(capacity).pipe(Effect.tap((queue) => Effect.sync(() => subscribers.add(queue)))),
+      (queue) =>
+        Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid),
+    ).pipe(Effect.map(Stream.fromQueue)),
+  })
+})
+
+export const layer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const events = yield* EventV2.Service
+    return yield* make(events.listen)
+  }),
+)

packages/server/src/handlers.ts

@@ -26,6 +26,7 @@ import { CredentialHandler } from "./handlers/credential"
 import { ProjectHandler } from "./handlers/project"
 import { ProjectCopyHandler } from "./handlers/project-copy"
 import { VcsHandler } from "./handlers/vcs"
+import { EventFeed } from "./event-feed"
 
 export const handlers = Layer.mergeAll(
   HealthHandler,
@@ -48,7 +49,7 @@ export const handlers = Layer.mergeAll(
   FileSystemHandler,
   CommandHandler,
   SkillHandler,
-  EventHandler,
+  EventHandler.pipe(Layer.provide(EventFeed.layer)),
   PtyHandler,
   ShellHandler,
   QuestionHandler,

packages/server/src/handlers/event.ts

@@ -1,44 +1,23 @@
 import { EventV2 } from "@opencode-ai/core/event"
-import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
-import { Effect, Schema, Stream } from "effect"
-import { Sse } from "effect/unstable/encoding"
+import { Effect, Stream } from "effect"
 import { HttpServerResponse } from "effect/unstable/http"
 import { HttpApiBuilder } from "effect/unstable/httpapi"
 import { Api } from "../api"
-
-// Session execution emits dense event bursts; allow healthy SSE clients enough
-// time to absorb one without weakening the bounded slow-subscriber failure.
-const subscriberCapacity = 4_096
-
-function eventData(data: unknown): Sse.Event {
-  return {
-    _tag: "Event",
-    event: "message",
-    id: undefined,
-    data: JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(data)),
-  }
-}
+import { EventFeed } from "../event-feed"
 
 export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) =>
   Effect.gen(function* () {
-    const events = yield* EventV2.Service
+    const feed = yield* EventFeed.Service
     return handlers.handleRaw("event.subscribe", () =>
       Effect.gen(function* () {
         const connected = {
           id: EventV2.ID.create(),
           type: "server.connected",
           data: {},
-        }
+        } as const
         const output = Stream.unwrap(
-          Effect.gen(function* () {
-            // Acquiring the bounded stream installs its listener before readiness is observable.
-            const live = yield* EventV2.liveBounded(events, {
-              capacity: subscriberCapacity,
-              accept: isOpenCodeEvent,
-            })
-            return Stream.make(connected).pipe(Stream.concat(live))
-          }),
-        ).pipe(Stream.map(eventData), Stream.pipeThroughChannel(Sse.encode()))
+          feed.subscribe.pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))),
+        )
         const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n"))
         return HttpServerResponse.stream(
           output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText),

packages/server/test/event-feed.test.ts

@@ -0,0 +1,176 @@
+import { describe, expect, test } from "bun:test"
+import { AgentV2 } from "@opencode-ai/core/agent"
+import { EventV2 } from "@opencode-ai/core/event"
+import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
+import { DateTime, Deferred, Effect, Exit, Fiber, Option, Schema, Stream } from "effect"
+import { EventFeed } from "../src/event-feed"
+
+const Internal = EventV2.ephemeral({ type: "test.internal", schema: { value: Schema.String } })
+
+const event = (id: string): EventV2.Payload<typeof AgentV2.Event.Updated> => ({
+  id: EventV2.ID.make(`evt_${id}`),
+  created: DateTime.makeUnsafe(Date.now()),
+  type: AgentV2.Event.Updated.type,
+  data: {},
+})
+
+const internal = (value: string): EventV2.Payload<typeof Internal> => ({
+  id: EventV2.ID.create(),
+  created: DateTime.makeUnsafe(Date.now()),
+  type: Internal.type,
+  data: { value },
+})
+
+function makeSource() {
+  let subscriber: EventV2.Subscriber | undefined
+  return {
+    observe: (next: EventV2.Subscriber) =>
+      Effect.sync(() => {
+        subscriber = next
+        return Effect.sync(() => {
+          if (subscriber === next) subscriber = undefined
+        })
+      }),
+    publish: (event: EventV2.Payload) => Effect.suspend(() => (subscriber ? subscriber(event) : Effect.void)),
+  }
+}
+
+describe("EventFeed", () => {
+  test("preserves the public SSE frame encoding", () => {
+    const payload = event("wire")
+    expect(EventFeed.frame(payload)).toBe(
+      `data: ${JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(payload))}\n\n`,
+    )
+  })
+
+  test("encodes once and delivers the same frame to every subscriber", async () => {
+    let encodes = 0
+    const result = await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const source = makeSource()
+          const feed = yield* EventFeed.make(source.observe, {
+            encode: (event) => {
+              encodes += 1
+              return event.type
+            },
+          })
+          const first = yield* feed.subscribe
+          const second = yield* feed.subscribe
+          const left = yield* first.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+          const right = yield* second.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+
+          yield* source.publish(event("example"))
+          return [Array.from(yield* Fiber.join(left)), Array.from(yield* Fiber.join(right))] as const
+        }),
+      ),
+    )
+
+    expect(result).toEqual([[AgentV2.Event.Updated.type], [AgentV2.Event.Updated.type]])
+    expect(encodes).toBe(1)
+  })
+
+  test("fails only the subscriber that exceeds its lag capacity", async () => {
+    const result = await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const source = makeSource()
+          const feed = yield* EventFeed.make(source.observe, {
+            capacity: 1,
+            encode: (event) => event.id,
+          })
+          const slow = yield* feed.subscribe
+          const fast = yield* feed.subscribe
+          const first = yield* Deferred.make<void>()
+          const second = yield* Deferred.make<void>()
+          const received = new Array<string>()
+          const fastFiber = yield* fast.pipe(
+            Stream.take(3),
+            Stream.runForEach((frame) =>
+              Effect.sync(() => received.push(frame)).pipe(
+                Effect.andThen(
+                  frame === "evt_one"
+                    ? Deferred.succeed(first, undefined)
+                    : frame === "evt_two"
+                      ? Deferred.succeed(second, undefined)
+                      : Effect.void,
+                ),
+              ),
+            ),
+            Effect.forkScoped,
+          )
+
+          yield* source.publish(event("one"))
+          yield* Deferred.await(first)
+          yield* source.publish(event("two"))
+          yield* Deferred.await(second)
+          yield* source.publish(event("three"))
+
+          yield* Fiber.join(fastFiber)
+
+          return {
+            slow: yield* slow.pipe(Stream.runCollect, Effect.exit),
+            fast: received,
+          }
+        }),
+      ),
+    )
+
+    expect(result.fast).toEqual(["evt_one", "evt_two", "evt_three"])
+    expect(Exit.isFailure(result.slow)).toBeTrue()
+    if (Exit.isSuccess(result.slow)) return
+    expect(Option.getOrUndefined(Exit.findErrorOption(result.slow))).toBeInstanceOf(EventFeed.SubscriberOverflowError)
+  })
+
+  test("filters internal events before they consume subscriber capacity", async () => {
+    const result = await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const source = makeSource()
+          const feed = yield* EventFeed.make(source.observe, { capacity: 1, encode: (event) => event.type })
+          const stream = yield* feed.subscribe
+
+          yield* source.publish(internal("one"))
+          yield* source.publish(internal("two"))
+          yield* source.publish(event("public"))
+
+          return Array.from(yield* stream.pipe(Stream.take(1), Stream.runCollect))
+        }),
+      ),
+    )
+
+    expect(result).toEqual([AgentV2.Event.Updated.type])
+  })
+
+  test("disconnects current subscribers after an encoding failure and continues for later subscribers", async () => {
+    const result = await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const source = makeSource()
+          const feed = yield* EventFeed.make(source.observe, {
+            encode: (event) => {
+              if (event.id === EventV2.ID.make("evt_bad")) throw new Error("invalid event")
+              return event.id
+            },
+          })
+          const current = yield* feed.subscribe
+          const failed = yield* current.pipe(Stream.runCollect, Effect.exit, Effect.forkScoped)
+
+          yield* source.publish(event("bad"))
+          const exit = yield* Fiber.join(failed)
+
+          const next = yield* feed.subscribe
+          const received = yield* next.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+          yield* source.publish(event("good"))
+
+          return { exit, received: Array.from(yield* Fiber.join(received)) }
+        }),
+      ),
+    )
+
+    expect(Exit.isFailure(result.exit)).toBeTrue()
+    if (Exit.isSuccess(result.exit)) return
+    expect(Option.getOrUndefined(Exit.findErrorOption(result.exit))).toBeInstanceOf(EventFeed.EncodingError)
+    expect(result.received).toEqual(["evt_good"])
+  })
+})

specs/v2/event-stream-architecture.md

@@ -0,0 +1,239 @@
+# V2 Event Stream Architecture
+
+## Decision
+
+The public HTTP event stream uses one Server-scoped encoded feed with one independently bounded queue per connection.
+
+```text
+Core EventV2.listen()
+          |
+          | one global subscription
+          v
+Server EventFeed
+  public filter
+  schema encode
+  JSON encode
+  SSE frame once
+          |
+          | nonblocking offer of one shared immutable string
+          v
+ Queue A   Queue B   Queue C
+    |         |         |
+ HTTP A    HTTP B    HTTP C
+```
+
+Core owns event meaning, publication, persistence, typed observation, durable logs, replay, and transactional projection.
+
+Server owns public event selection, wire encoding, bounded connection delivery, and subscriber lifecycle.
+
+Protocol continues to own the `OpenCodeEvent` SSE contract. Generated Promise and Effect clients remain unchanged.
+
+## Context
+
+Before this change, every `/api/event` connection called `EventV2.liveBounded`. Each call registered a Core callback listener and allocated a dropping queue of raw event payloads. Every HTTP connection then independently performed:
+
+1. Public-event filtering.
+2. `OpenCodeEvent` schema encoding.
+3. `JSON.stringify`.
+4. SSE framing.
+5. UTF-8 encoding.
+
+With `N` connected TUIs, schema and wire encoding therefore ran `N` times for every accepted event.
+
+`liveBounded` was introduced before zero-argument `events.subscribe()` became the unified live interface. It stayed on deprecated `listen` because it provided a stronger contract than the shared unbounded Core PubSub: one slow subscriber could overflow and fail without blocking healthy subscribers.
+
+The global cross-location event stream is intentional. The endpoint is outside `LocationMiddleware`, and the TUI uses event location metadata to update state for multiple locations. The feed must not add request-location filtering.
+
+## Delivery Law
+
+The Server feed preserves this law:
+
+> A connection has an independent finite lag budget. Exceeding it terminates only that connection while publication and healthy connections continue in order.
+
+Each connection receives a `Queue.dropping` with capacity 4,096 accepted public frames.
+
+When an offer returns `false`:
+
+1. The queue is removed from the active subscriber registry immediately.
+2. The queue is failed with `SubscriberOverflowError`.
+3. The same frame is still offered to every other active queue.
+4. Core publication and the Server observer never suspend on that connection.
+
+Previously accepted frames drain before the queue failure surfaces. The overflow-causing frame is not accepted by that connection.
+
+Internal Core events, `server.connected`, and heartbeats do not consume the queue capacity.
+
+## Why Independent Queues
+
+The design was reviewed twice, including explicit consideration of one shared Effect PubSub of encoded frames.
+
+### Shared PubSub benefits
+
+A shared PubSub stores each frame once and gives each subscriber a cursor. Its retained feed storage is proportional to maximum lag rather than the sum of every subscriber's lag.
+
+### Shared PubSub costs
+
+Effect's bounded PubSub strategies do not directly express independent subscriber failure:
+
+- `bounded` can suspend the shared publisher behind the slowest subscriber;
+- `dropping` rejects one publication for every subscriber when shared capacity is full;
+- `sliding` silently skips events while leaving the stale subscriber connected;
+- `unbounded` removes the structural memory bound.
+
+Independent eviction can be built on a dropping PubSub, but requires:
+
+- retaining every subscription's child scope;
+- a separate typed overflow signal because subscription closure appears as interruption/completion;
+- serialization of registration, removal, eviction, and publication;
+- lag scans at capacity;
+- waiting for scope closure to release shared ring slots;
+- terminal handling if a supposedly impossible shared publish returns `false`;
+- immediate discard of the stale subscriber's previously accepted unread backlog.
+
+That is a custom multicast protocol layered over PubSub.
+
+The incremental benefit is queue-slot references, not encoded frame copies: every independent queue stores the same immutable encoded string reference. At 50 clients each retaining 4,096 frames, raw references are roughly 1.6 MiB before array overhead. HTTP runtime, TLS, kernel, proxy, and client buffers may dominate that cost.
+
+The chosen queue design captures the dominant optimization, encode once, while retaining direct queue-local overflow semantics and a smaller failure domain.
+
+Revisit shared PubSub storage only if measurements after shared encoding show queue reference retention or per-queue offers are material.
+
+## Capacity
+
+The migration preserves the existing 4,096-event capacity.
+
+This is compatibility, not a claim that 4,096 is optimal. It is an event-count lag threshold, not a complete memory bound:
+
+- frames vary in size;
+- stream pulls may move batches into HTTP buffers before queue lag reflects them;
+- kernel and client buffers are outside Server accounting.
+
+Do not raise capacity merely because frames are encoded once. A larger threshold retains stale clients longer.
+
+Tune capacity separately using observed:
+
+- public event rates and burst sizes;
+- healthy subscriber queue high-water marks;
+- encoded frame-size distribution;
+- overflow and reconnect frequency;
+- retained heap and RSS;
+- downstream drain duration under a stalled reader.
+
+Add a byte budget only if measurements show event count is an inadequate memory safeguard.
+
+## Feed Lifecycle
+
+### Server scope
+
+`EventFeed.layer` is built once with the Server handler graph. It registers one global Core listener outside request location middleware.
+
+The listener is installed synchronously before the feed service is exposed. Public filtering, encoding, and nonblocking queue offers happen inline once per Core event. This avoids both a startup gap and an unbounded asynchronous ingress backlog.
+
+Core invokes the one observer sequentially. It does not fork encoding or fan-out per event, so every healthy subscriber observes the same order.
+
+When no HTTP subscribers are registered, the observer returns before wire encoding, so headless and idle servers do not pay serialization cost.
+
+### Connection scope
+
+Each `feed.subscribe` acquisition:
+
+1. Allocates one dropping queue.
+2. Registers it synchronously.
+3. Returns `Stream.fromQueue(queue)`.
+4. Removes and shuts down the queue when the request scope closes.
+
+The raw handler acquires and registers the queue before prepending its connection-specific `server.connected` frame:
+
+```text
+register queue
+  -> emit server.connected
+  -> drain queued live frames
+```
+
+Events before registration may be missed, consistent with a volatile stream. Events after registration queue behind `server.connected`.
+
+Heartbeats remain connection-local and outside the feed.
+
+### Encoding failure
+
+If one accepted public event cannot be encoded:
+
+1. Log its ID, type, and cause.
+2. Fail every currently connected queue with `EncodingError`.
+3. Skip the malformed volatile event.
+4. Keep the feed available for later connections and valid events.
+
+Keeping current clients connected would create a silent gap. Permanently terminating the feed would poison future connections.
+
+## HTTP And Code Generation
+
+Protocol remains unchanged:
+
+```ts
+HttpApiSchema.StreamSse({ data: OpenCodeEvent })
+```
+
+The raw handler continues to own:
+
+- the unique `server.connected` event;
+- the 15-second heartbeat;
+- SSE response headers;
+- `HttpServerResponse.stream` construction.
+
+The feed supplies complete immutable SSE frame strings for ordinary public events. The handler merges connection-local frames and performs text-to-byte encoding.
+
+Because method, path, schema, and wire representation do not change:
+
+- OpenAPI does not change;
+- generated Promise clients do not change;
+- generated Effect clients do not change;
+- TUI decoding and reconnect behavior do not change;
+- client regeneration is not required.
+
+## Core Cleanup
+
+The Server no longer uses `EventV2.liveBounded`, so Core removes that dead helper and its transport-specific overflow error. The feed registers one observer through the existing `listen` interface; other listeners are unchanged.
+
+Transactional projector registration is unrelated and remains unchanged.
+
+## Benchmark
+
+The disposable benchmark reproduced the previous per-connection schema/JSON/SSE encoding path with a representative 8 KiB public event. It used one warmup and nine measured runs; median was the primary metric and median absolute deviation was reported. The benchmark was intentionally not committed because it isolated the removed encoding boundary rather than exercising the complete HTTP stack.
+
+Results on Apple Silicon with Bun 1.3.14:
+
+| Clients | Current median | Shared median | Change |
+| ------: | -------------: | ------------: | -----: |
+|       1 |       9.488 ms |      9.554 ms |  +0.7% |
+|      10 |      96.312 ms |     10.352 ms | -89.3% |
+|      50 |     553.928 ms |     12.389 ms | -97.8% |
+
+The benchmark isolates the repeated encoding boundary. It does not claim to measure socket throughput, client decoding, or downstream HTTP buffering. Queue offers and socket writes remain proportional to connected clients.
+
+An experiment replacing direct schema encoding plus `JSON.stringify` with `Schema.fromJsonString(OpenCodeEvent)` was discarded: the one-client median regressed from approximately 9.5 ms to 38.8 ms with substantially higher variance.
+
+## Verification
+
+Behavioral tests cover:
+
+- one encoding operation for multiple subscribers;
+- identical frame delivery to healthy subscribers;
+- independent slow-subscriber overflow;
+- healthy delivery of events after another subscriber overflows;
+- filtering internal events before capacity;
+- failure of current subscribers after malformed public encoding;
+- continued delivery to later subscribers after an encoding failure.
+
+Package typechecks and the existing Core event/event-logger suites protect the Core interface migration.
+
+## Future Measurements
+
+Before further transport work, measure the complete HTTP path with real generated clients:
+
+- 1, 10, and 50 connected clients;
+- dense session bursts;
+- small and large event payloads;
+- one stalled socket reader;
+- CPU, allocation rate, heap/RSS, publisher latency, observer time, queue occupancy, and reconnect timing.
+
+Only those measurements should motivate shared PubSub storage, shared UTF-8 bytes, byte budgets, drain timeouts, or a different capacity.