fix(server): enforce event stream ownership and diagnostics

packages/cli/src/mini/noninteractive.ts

@@ -55,7 +55,7 @@ const GLOBAL_FORM_SESSION_ID = "global"
 
 export async function runNonInteractivePrompt(input: Input) {
   const controller = new AbortController()
-  const stream = input.client.event.subscribe({ signal: controller.signal })[Symbol.asyncIterator]()
+  const stream = input.client.event.subscribe(undefined, { signal: controller.signal })[Symbol.asyncIterator]()
   const connected = await stream.next()
   if (connected.done) throw new Error("Event stream disconnected before prompt admission")
 

packages/cli/src/mini/stream-v2.transport.ts

@@ -832,7 +832,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
         const connection = new AbortController()
         const abortConnection = () => connection.abort()
         controller.signal.addEventListener("abort", abortConnection, { once: true })
-        const stream = input.sdk.event.subscribe({ signal: connection.signal })[Symbol.asyncIterator]()
+        const stream = input.sdk.event.subscribe(undefined, { signal: connection.signal })[Symbol.asyncIterator]()
         try {
           const first = await stream.next()
           if (first.done || first.value.type !== "server.connected") throw new Error("Event stream disconnected")

packages/client/src/effect/api/api.ts

@@ -767,8 +767,13 @@ export interface SkillApi<E = never> {
   readonly list: SkillListOperation<E>
 }
 
+type Endpoint19_0Request = Parameters<RawClient["server.event"]["event.subscribe"]>[0]
+export type Endpoint19_0Input = {
+  readonly location?: Endpoint19_0Request["query"]["location"]
+  readonly session?: Endpoint19_0Request["query"]["session"]
+}
 export type Endpoint19_0Output = StreamValue<EffectValue<ReturnType<RawClient["server.event"]["event.subscribe"]>>>
-export type EventSubscribeOperation<E = never> = () => Stream.Stream<Endpoint19_0Output, E>
+export type EventSubscribeOperation<E = never> = (input?: Endpoint19_0Input) => Stream.Stream<Endpoint19_0Output, E>
 
 export interface EventApi<E = never> {
   readonly subscribe: EventSubscribeOperation<E>
@@ -992,8 +997,12 @@ export type Endpoint26_1Input = { readonly location?: Endpoint26_1Request["query
 export type Endpoint26_1Output = EffectValue<ReturnType<RawClient["server.debug"]["debug.location.evict"]>>
 export type DebugLocationEvictOperation<E = never> = (input?: Endpoint26_1Input) => Effect.Effect<Endpoint26_1Output, E>
 
+export type Endpoint26_2Output = EffectValue<ReturnType<RawClient["server.debug"]["debug.event-feed"]>>
+export type DebugEventFeedOperation<E = never> = () => Effect.Effect<Endpoint26_2Output, E>
+
 export interface DebugApi<E = never> {
   readonly location: { readonly list: DebugLocationListOperation<E>; readonly evict: DebugLocationEvictOperation<E> }
+  readonly "event-feed": DebugEventFeedOperation<E>
 }
 
 export interface AppApi<E = never> {

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

@@ -908,9 +908,14 @@ const Endpoint18_0 = (raw: RawClient["server.skill"]) => (input?: Endpoint18_0In
 
 const adaptGroup18 = (raw: RawClient["server.skill"]) => ({ list: Endpoint18_0(raw) })
 
-const Endpoint19_0 = (raw: RawClient["server.event"]) => () =>
+type Endpoint19_0Request = Parameters<RawClient["server.event"]["event.subscribe"]>[0]
+type Endpoint19_0Input = {
+  readonly location?: Endpoint19_0Request["query"]["location"]
+  readonly session?: Endpoint19_0Request["query"]["session"]
+}
+const Endpoint19_0 = (raw: RawClient["server.event"]) => (input?: Endpoint19_0Input) =>
   Stream.unwrap(
-    raw["event.subscribe"]({}).pipe(
+    raw["event.subscribe"]({ query: { location: input?.["location"], session: input?.["session"] } }).pipe(
       Effect.mapError(mapClientError),
       Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))),
     ),
@@ -1181,8 +1186,12 @@ type Endpoint26_1Input = { readonly location?: Endpoint26_1Request["query"]["loc
 const Endpoint26_1 = (raw: RawClient["server.debug"]) => (input?: Endpoint26_1Input) =>
   raw["debug.location.evict"]({ query: { location: input?.["location"] } }).pipe(Effect.mapError(mapClientError))
 
+const Endpoint26_2 = (raw: RawClient["server.debug"]) => () =>
+  raw["debug.event-feed"]({}).pipe(Effect.mapError(mapClientError))
+
 const adaptGroup26 = (raw: RawClient["server.debug"]) => ({
   location: { list: Endpoint26_0(raw), evict: Endpoint26_1(raw) },
+  "event-feed": Endpoint26_2(raw),
 })
 
 const adaptClient = (raw: RawClient) => ({

packages/client/src/promise/generated/client.ts

@@ -151,6 +151,7 @@ import type {
   CommandListOutput,
   SkillListInput,
   SkillListOutput,
+  EventSubscribeInput,
   EventSubscribeOutput,
   PtyListInput,
   PtyListOutput,
@@ -197,6 +198,7 @@ import type {
   DebugLocationListOutput,
   DebugLocationEvictInput,
   DebugLocationEvictOutput,
+  DebugEventFeedOutput,
 } from "./types"
 import { ClientError } from "./client-error"
 
@@ -292,6 +294,12 @@ export function make(options: ClientOptions) {
       const reader = response.body.getReader()
       const decoder = new TextDecoder()
       let buffer = ""
+      const signal = requestOptions?.signal
+      const onAbort = () => {
+        void reader.cancel(signal?.reason).catch(() => undefined)
+      }
+      if (signal?.aborted) onAbort()
+      else signal?.addEventListener("abort", onAbort, { once: true })
       try {
         while (true) {
           let next
@@ -327,10 +335,16 @@ export function make(options: ClientOptions) {
           if (next.done) return
         }
       } finally {
+        signal?.removeEventListener("abort", onAbort)
         try {
           await reader.cancel()
         } catch {}
-        reader.releaseLock()
+        try {
+          reader.releaseLock()
+        } catch {}
+        try {
+          await response.body?.cancel()
+        } catch {}
       }
     },
   })
@@ -1350,9 +1364,16 @@ export function make(options: ClientOptions) {
         ),
     },
     event: {
-      subscribe: (requestOptions?: RequestOptions): AsyncIterable<EventSubscribeOutput> =>
+      subscribe: (input?: EventSubscribeInput, requestOptions?: RequestOptions): AsyncIterable<EventSubscribeOutput> =>
         sse<EventSubscribeOutput>(
-          { method: "GET", path: `/api/event`, successStatus: 200, declaredStatuses: [401, 400], empty: false },
+          {
+            method: "GET",
+            path: `/api/event`,
+            query: { location: input?.["location"], session: input?.["session"] },
+            successStatus: 200,
+            declaredStatuses: [401, 400],
+            empty: false,
+          },
           requestOptions,
         ),
     },
@@ -1663,6 +1684,17 @@ export function make(options: ClientOptions) {
             requestOptions,
           ),
       },
+      "event-feed": (requestOptions?: RequestOptions) =>
+        request<DebugEventFeedOutput>(
+          {
+            method: "GET",
+            path: `/api/debug/event-feed`,
+            successStatus: 200,
+            declaredStatuses: [401, 400],
+            empty: false,
+          },
+          requestOptions,
+        ),
     },
   }
 }

packages/client/src/promise/generated/types.ts

@@ -4542,6 +4542,17 @@ export type SkillListOutput = {
   data: Array<SkillInfo>
 }
 
+export type EventSubscribeInput = {
+  readonly location?: {
+    readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined
+    readonly session?: string | ReadonlyArray<string> | ReadonlyArray<string> | undefined
+  }["location"]
+  readonly session?: {
+    readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined
+    readonly session?: string | ReadonlyArray<string> | ReadonlyArray<string> | undefined
+  }["session"]
+}
+
 export type EventSubscribeOutput = V2Event
 
 export type PtyListInput = {
@@ -4859,3 +4870,12 @@ export type DebugLocationEvictInput = {
 }
 
 export type DebugLocationEvictOutput = void
+
+export type DebugEventFeedOutput = {
+  active: number
+  opens: number
+  closes: number
+  serializedEvents: number
+  serializedBytes: number
+  overflows: number
+}