feat(server): opt-in location interest for event subscriptions

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,10 @@ 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"] }
 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,11 @@ 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"] }
+const Endpoint19_0 = (raw: RawClient["server.event"]) => (input?: Endpoint19_0Input) =>
   Stream.unwrap(
-    raw["event.subscribe"]({}).pipe(
+    raw["event.subscribe"]({ query: { location: input?.["location"] } }).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"] },
+            successStatus: 200,
+            declaredStatuses: [401, 400],
+            empty: false,
+          },
           requestOptions,
         ),
     },

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

@@ -4542,6 +4542,12 @@ export type SkillListOutput = {
   data: Array<SkillInfo>
 }
 
+export type EventSubscribeInput = {
+  readonly location?: {
+    readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
+  }["location"]
+}
+
 export type EventSubscribeOutput = V2Event
 
 export type PtyListInput = {

packages/protocol/src/groups/event.ts

@@ -4,6 +4,7 @@ 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 { LocationQuery, locationQueryOpenApi } from "./location.js"
 
 const fields = {
   id: Event.ID,
@@ -32,15 +33,18 @@ const make = <const Definitions extends ReadonlyArray<Definition>>(definitions:
     group: HttpApiGroup.make("server.event")
       .add(
         HttpApiEndpoint.get("event.subscribe", "/api/event", {
+          query: LocationQuery,
           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 location to receive the global public feed; pass location to narrow delivery to that directory (and optional workspace). 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/server/src/event-feed.ts

@@ -6,6 +6,13 @@ import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "eff
 
 export const SubscriberCapacity = 4_096
 
+/** Core-published types that always pass location interest filters. */
+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 +26,18 @@ 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 to narrow delivery. */
+export type Interest = {
+  readonly location?: LocationInterest
+}
+
 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 +48,52 @@ export function frame(event: OpenCodeEvent) {
   return `data: ${JSON.stringify(encode(event))}\n\n`
 }
 
+export function matchesInterest(event: EventV2.Payload, interest?: Interest): boolean {
+  const location = interest?.location
+  if (location === undefined) return true
+  if (GlobalEventTypes.has(event.type)) return true
+  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
+  return true
+}
+
+export function interestFromQuery(query: URLSearchParams): Interest | undefined {
+  const directory = query.get("location[directory]") ?? undefined
+  const workspace = query.get("location[workspace]") ?? undefined
+  if (directory === undefined && workspace === undefined) return undefined
+  if (directory === undefined) return undefined
+  return { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } }
+}
+
+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 +107,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)),
   })
 })
 

packages/server/src/handlers/event.ts

@@ -8,15 +8,18 @@ import { EventFeed } from "../event-feed"
 export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) =>
   Effect.gen(function* () {
     const feed = yield* EventFeed.Service
-    return handlers.handleRaw("event.subscribe", () =>
+    return handlers.handleRaw("event.subscribe", (ctx) =>
       Effect.gen(function* () {
+        const interest = EventFeed.interestFromQuery(new URL(ctx.request.url, "http://localhost").searchParams)
         const connected = {
           id: EventV2.ID.create(),
           type: "server.connected",
           data: {},
         } as const
         const output = Stream.unwrap(
-          feed.subscribe.pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))),
+          feed
+            .subscribe(interest)
+            .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(