feat(server): narrow event subscriptions by session interest

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>

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))),
     ),

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

@@ -151,6 +151,7 @@ import type {
   CommandListOutput,
   SkillListInput,
   SkillListOutput,
+  EventSubscribeInput,
   EventSubscribeOutput,
   PtyListInput,
   PtyListOutput,
@@ -1350,9 +1351,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,
         ),
     },

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 = {

packages/protocol/src/groups/event.ts

@@ -4,6 +4,20 @@ import { Location } from "@opencode-ai/schema/location"
 import type { Definition } from "@opencode-ai/schema/event"
 import { Schema } from "effect"
 import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi"
+import { locationQueryOpenApi } from "./location.js"
+
+export const sessionIDOf = Event.sessionIDOf
+
+export const EventSubscribeQuery = Schema.Struct({
+  location: Schema.optional(
+    Schema.Struct({
+      /** Required when `location` is present; workspace alone is not a valid scope. */
+      directory: Schema.String,
+      workspace: Schema.optional(Schema.String),
+    }),
+  ),
+  session: Schema.optional(Schema.Union([Schema.String, Schema.Array(Schema.String)])),
+}).annotate({ identifier: "EventSubscribeQuery" })
 
 const fields = {
   id: Event.ID,
@@ -32,15 +46,18 @@ const make = <const Definitions extends ReadonlyArray<Definition>>(definitions:
     group: HttpApiGroup.make("server.event")
       .add(
         HttpApiEndpoint.get("event.subscribe", "/api/event", {
+          query: EventSubscribeQuery,
           success: HttpApiSchema.StreamSse({ data: EventSchema }),
-        }).annotateMerge(
-          OpenApi.annotations({
-            identifier: "v2.event.subscribe",
-            summary: "Subscribe to events",
-            description:
-              "Subscribe to native event payloads for the server. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.",
-          }),
-        ),
+        })
+          .annotateMerge(locationQueryOpenApi)
+          .annotateMerge(
+            OpenApi.annotations({
+              identifier: "v2.event.subscribe",
+              summary: "Subscribe to events",
+              description:
+                "Subscribe to native event payloads for the server. Omit query params for the global public feed. Pass location to narrow by directory/workspace, and optional repeated session IDs to further narrow session-scoped events. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.",
+            }),
+          ),
       )
       .annotateMerge(OpenApi.annotations({ title: "event", description: "Experimental event stream routes." })),
   }

packages/schema/src/event.ts

@@ -1,6 +1,6 @@
 export * as Event from "./event.js"
 
-import { Schema, SchemaTransformation } from "effect"
+import { Option, Schema, SchemaTransformation } from "effect"
 import { optional } from "./schema.js"
 import { ascending } from "./identifier.js"
 import { Location } from "./location.js"
@@ -196,3 +196,15 @@ function readonlyMap<Key, Value>(map: Map<Key, Value>): ReadonlyMap<Key, Value>
   })
   return result
 }
+
+const SessionData = Schema.Struct({ sessionID: Schema.String })
+const decodeSessionData = Schema.decodeUnknownOption(SessionData)
+
+/** Session identity from a durable aggregate or an ephemeral `data.sessionID` field. */
+export function sessionIDOf(event: {
+  readonly durable?: { readonly aggregateID: string }
+  readonly data?: unknown
+}): string | undefined {
+  if (event.durable?.aggregateID) return event.durable.aggregateID
+  return Option.getOrUndefined(decodeSessionData(event.data))?.sessionID
+}

packages/schema/test/event.test.ts

@@ -49,4 +49,14 @@ describe("public event schemas", () => {
       aggregateID: "ses_test",
     })
   })
+
+  test("sessionIDOf reads durable aggregate or ephemeral data.sessionID", () => {
+    expect(Event.sessionIDOf({ durable: { aggregateID: "ses_durable" }, data: { sessionID: "ses_other" } })).toBe(
+      "ses_durable",
+    )
+    expect(Event.sessionIDOf({ data: { sessionID: "ses_ephemeral" } })).toBe("ses_ephemeral")
+    expect(Event.sessionIDOf({ data: {} })).toBeUndefined()
+    expect(Event.sessionIDOf({ data: { sessionID: 1 } })).toBeUndefined()
+    expect(Event.sessionIDOf({ data: null })).toBeUndefined()
+  })
 })

packages/server/src/event-feed.ts

@@ -1,11 +1,23 @@
 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 {
+  EventSubscribeQuery,
+  isOpenCodeEvent,
+  OpenCodeEvent,
+  sessionIDOf,
+} from "@opencode-ai/protocol/groups/event"
+import { Cause, Context, Effect, Layer, Option, Queue, Schema, Scope, Stream } from "effect"
 
 export const SubscriberCapacity = 4_096
 
+/** Core-published types that always pass interest filters (location and session). */
+export const GlobalEventTypes = new Set<string>([
+  "installation.updated",
+  "installation.update-available",
+  "global.disposed",
+])
+
 export class SubscriberOverflowError extends Schema.TaggedErrorClass<SubscriberOverflowError>()(
   "EventFeed.SubscriberOverflow",
   { capacity: Schema.Int },
@@ -19,8 +31,19 @@ export class EncodingError extends Schema.TaggedErrorClass<EncodingError>()("Eve
 
 export type Error = SubscriberOverflowError | EncodingError
 
+export type LocationInterest = {
+  readonly directory: string
+  readonly workspace?: string
+}
+
+/** Omit or leave empty for the global public feed; set location/sessions to narrow delivery. */
+export type Interest = {
+  readonly location?: LocationInterest
+  readonly sessions?: ReadonlyArray<string>
+}
+
 export interface Interface {
-  readonly subscribe: Effect.Effect<Stream.Stream<string, Error>, never, Scope.Scope>
+  readonly subscribe: (interest?: Interest) => Effect.Effect<Stream.Stream<string, Error>, never, Scope.Scope>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/server/EventFeed") {}
@@ -31,24 +54,107 @@ export function frame(event: OpenCodeEvent) {
   return `data: ${JSON.stringify(encode(event))}\n\n`
 }
 
+export function matchesInterest(event: EventV2.Payload, interest?: Interest): boolean {
+  if (interest === undefined) return true
+  if (GlobalEventTypes.has(event.type)) return true
+
+  const location = interest.location
+  if (location !== undefined) {
+    const ref = event.location
+    if (ref === undefined) return false
+    if (ref.directory !== location.directory) return false
+    if (location.workspace !== undefined && ref.workspaceID !== location.workspace) return false
+  }
+
+  const sessions = interest.sessions
+  if (sessions !== undefined && sessions.length > 0) {
+    const sessionID = sessionIDOf(event)
+    if (sessionID !== undefined) return sessions.includes(sessionID)
+  }
+
+  return true
+}
+
+const decodeSubscribeQuery = Schema.decodeUnknownOption(EventSubscribeQuery)
+
+/** Map an already-decoded subscribe query into feed interest (HttpApi `ctx.query`). */
+export function interestFromSubscribeQuery(query: typeof EventSubscribeQuery.Type): Interest | undefined {
+  const sessions = normalizeSessions(query.session)
+  if (query.location === undefined && sessions.length === 0) return undefined
+  return {
+    ...(query.location
+      ? {
+          location: {
+            directory: query.location.directory,
+            ...(query.location.workspace !== undefined ? { workspace: query.location.workspace } : {}),
+          },
+        }
+      : {}),
+    ...(sessions.length > 0 ? { sessions } : {}),
+  }
+}
+
+/** Parse deepObject / repeated `session` search params through `EventSubscribeQuery`. */
+export function interestFromQuery(params: URLSearchParams): Interest | undefined {
+  const decoded = Option.getOrUndefined(decodeSubscribeQuery(subscribeQueryCandidate(params)))
+  if (decoded === undefined) return undefined
+  return interestFromSubscribeQuery(decoded)
+}
+
+function subscribeQueryCandidate(params: URLSearchParams) {
+  const directory = params.get("location[directory]") ?? undefined
+  const workspace = params.get("location[workspace]") ?? undefined
+  const sessions = params
+    .getAll("session")
+    .flatMap((value) => value.split(","))
+    .map((value) => value.trim())
+    .filter((value) => value.length > 0)
+  return {
+    ...(directory !== undefined || workspace !== undefined
+      ? {
+          location: {
+            ...(directory !== undefined ? { directory } : {}),
+            ...(workspace !== undefined ? { workspace } : {}),
+          },
+        }
+      : {}),
+    ...(sessions.length === 0 ? {} : { session: sessions.length === 1 ? sessions[0] : sessions }),
+  }
+}
+
+function normalizeSessions(session: typeof EventSubscribeQuery.Type.session) {
+  if (session === undefined) return [] as string[]
+  if (typeof session === "string") return session.trim() === "" ? [] : [session]
+  return session.map((value) => value.trim()).filter((value) => value.length > 0)
+}
+
+type Subscriber = {
+  readonly queue: Queue.Queue<string, Error>
+  readonly interest?: Interest
+}
+
 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 render = options?.encode ?? frame
-  const subscribers = new Set<Queue.Queue<string, Error>>()
+  const subscribers = new Map<Queue.Queue<string, Error>, Subscriber>()
 
   const fail = (error: Error) =>
     Effect.sync(() => {
-      const current = Array.from(subscribers)
+      const current = Array.from(subscribers.values())
       subscribers.clear()
-      for (const subscriber of current) Queue.failCauseUnsafe(subscriber, Cause.fail(error))
+      for (const subscriber of current) Queue.failCauseUnsafe(subscriber.queue, Cause.fail(error))
     })
 
   const publish = Effect.fnUntraced(function* (event: EventV2.Payload) {
     if (!isOpenCodeEvent(event)) return
     if (subscribers.size === 0) return
+    const targets = Array.from(subscribers.values()).filter((subscriber) =>
+      matchesInterest(event, subscriber.interest),
+    )
+    if (targets.length === 0) return
     const encoded = yield* Effect.try({
       try: () => render(event),
       catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }),
@@ -62,22 +168,29 @@ export const make = Effect.fn("EventFeed.make")(function* (
       ),
     )
     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 })))
+    for (const subscriber of targets) {
+      if (Queue.offerUnsafe(subscriber.queue, encoded)) continue
+      subscribers.delete(subscriber.queue)
+      Queue.failCauseUnsafe(subscriber.queue, 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)),
+    subscribe: (interest) =>
+      Effect.acquireRelease(
+        Queue.dropping<string, Error>(capacity).pipe(
+          Effect.tap((queue) =>
+            Effect.sync(() => {
+              subscribers.set(queue, { queue, interest })
+            }),
+          ),
+        ),
+        (queue) =>
+          Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid),
+      ).pipe(Effect.map(Stream.fromQueue)),
   })
 })