merge: bring EventSubscribeQuery interest decode into ownership stack

Also derive EventFeed.Diagnostics from the protocol Schema and lock the
generation boundary so prior-generation event IDs are not delivered.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: armancharan

packages/server/src/event-feed.ts

@@ -1,8 +1,14 @@
 export * as EventFeed from "./event-feed"
 
 import { EventV2 } from "@opencode-ai/core/event"
-import { isOpenCodeEvent, OpenCodeEvent, sessionIDOf } from "@opencode-ai/protocol/groups/event"
-import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "effect"
+import { EventFeedDiagnostics } from "@opencode-ai/protocol/groups/debug"
+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
 
@@ -38,14 +44,7 @@ export type Interest = {
 }
 
 /** Content-free counters for leak detection — never includes event payloads. */
-export type Diagnostics = {
-  readonly active: number
-  readonly opens: number
-  readonly closes: number
-  readonly serializedEvents: number
-  readonly serializedBytes: number
-  readonly overflows: number
-}
+export type Diagnostics = typeof EventFeedDiagnostics.Type
 
 export interface Interface {
   readonly subscribe: (interest?: Interest) => Effect.Effect<Stream.Stream<string, Error>, never, Scope.Scope>
@@ -81,25 +80,59 @@ export function matchesInterest(event: EventV2.Payload, interest?: Interest): bo
   return true
 }
 
-export function interestFromQuery(query: URLSearchParams): Interest | undefined {
-  const directory = query.get("location[directory]") ?? undefined
-  const workspace = query.get("location[workspace]") ?? undefined
-  const sessions = query
+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)
-  // Workspace interest requires a directory; alone it is an invalid location scope.
-  if (directory === undefined && workspace !== undefined) return undefined
-  if (directory === undefined && sessions.length === 0) return undefined
   return {
-    ...(directory !== undefined
-      ? { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } }
+    ...(directory !== undefined || workspace !== undefined
+      ? {
+          location: {
+            ...(directory !== undefined ? { directory } : {}),
+            ...(workspace !== undefined ? { workspace } : {}),
+          },
+        }
       : {}),
-    ...(sessions.length > 0 ? { sessions } : {}),
+    ...(sessions.length === 0 ? {} : { session: sessions.length === 1 ? sessions[0] : sessions }),
   }
 }
 
+function normalizeSessions(session: typeof EventSubscribeQuery.Type.session): string[] {
+  if (session === undefined) return []
+  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

packages/server/src/handlers/event.ts

@@ -33,7 +33,8 @@ export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers)
     return handlers.handleRaw("event.subscribe", (ctx) =>
       Effect.gen(function* () {
         yield* bridgeClientDisconnect(ctx.request)
-        const interest = EventFeed.interestFromQuery(new URL(ctx.request.url, "http://localhost").searchParams)
+        // handleRaw still Schema-decodes query; map into feed interest without re-parsing the URL.
+        const interest = EventFeed.interestFromSubscribeQuery(ctx.query)
         const connected = {
           id: EventV2.ID.create(),
           type: "server.connected",

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

@@ -265,6 +265,20 @@ describe("EventFeed", () => {
     expect(EventFeed.interestFromQuery(params)).toBeUndefined()
   })
 
+  test("maps a decoded EventSubscribeQuery into feed interest", () => {
+    expect(EventFeed.interestFromSubscribeQuery({})).toBeUndefined()
+    expect(
+      EventFeed.interestFromSubscribeQuery({
+        location: { directory: "/tmp/project", workspace: "ws_1" },
+        session: ["ses_a", "ses_b"],
+      }),
+    ).toEqual({
+      location: { directory: "/tmp/project", workspace: "ws_1" },
+      sessions: ["ses_a", "ses_b"],
+    })
+    expect(EventFeed.interestFromSubscribeQuery({ session: "ses_a" })).toEqual({ sessions: ["ses_a"] })
+  })
+
   test("matchesInterest delivers global types to location-scoped subscribers", () => {
     expect(
       EventFeed.matchesInterest(

packages/tui/test/cli/tui/client-ownership.test.tsx

@@ -1,5 +1,6 @@
 /** @jsxImportSource @opentui/solid */
 import { describe, expect, test } from "bun:test"
+import type { OpenCodeEvent } from "@opencode-ai/client"
 import { testRender } from "@opentui/solid"
 import { onMount } from "solid-js"
 import { ClientProvider, useClient } from "../../../src/context/client"
@@ -71,6 +72,59 @@ describe("ClientProvider stream ownership", () => {
     }
   })
 
+  test("event IDs from a prior generation are not delivered after interest reconnect", async () => {
+    const events = createEventStream()
+    const calls = createFetch((url, request) => {
+      if (url.pathname !== "/api/event") return undefined
+      return events.v2(request.signal)
+    }, events)
+
+    const seen: Array<{ id: string; generation: number }> = []
+    let client!: ReturnType<typeof useClient>
+    let done!: () => void
+    const ready = new Promise<void>((resolve) => {
+      done = resolve
+    })
+
+    const app = await testRender(() => (
+      <TestTuiContexts>
+        <ClientProvider api={createApi(calls.fetch)} interest={{ location: { directory } }}>
+          <Probe
+            onReady={(ctx) => {
+              client = ctx
+              client.event.on("vcs.branch.updated", (event) => {
+                seen.push({ id: event.id, generation: client.connection.internal.generation() })
+              })
+              done()
+            }}
+          />
+        </ClientProvider>
+      </TestTuiContexts>
+    ))
+
+    try {
+      await ready
+      await wait(() => client.connection.status() === "connected")
+      const generation = client.connection.internal.generation()
+      events.emit(vcs("before"))
+      await wait(() => seen.some((entry) => entry.id === "evt_vcs_before"))
+
+      client.event.scope({ location: { directory }, sessions: ["ses_next"] })
+      await wait(
+        () =>
+          client.connection.status() === "connected" && client.connection.internal.generation() > generation,
+      )
+      const next = client.connection.internal.generation()
+      events.emit(vcs("after"))
+      await wait(() => seen.some((entry) => entry.id === "evt_vcs_after"))
+
+      expect(seen.filter((entry) => entry.id === "evt_vcs_before")).toEqual([{ id: "evt_vcs_before", generation }])
+      expect(seen.filter((entry) => entry.id === "evt_vcs_after")).toEqual([{ id: "evt_vcs_after", generation: next }])
+    } finally {
+      app.renderer.destroy()
+    }
+  })
+
   test("teardown leaves zero active event streams", async () => {
     const events = createEventStream()
     let concurrent = 0
@@ -120,3 +174,12 @@ function Probe(props: { onReady: (client: ReturnType<typeof useClient>) => void
   onMount(() => props.onReady(client))
   return <box />
 }
+
+function vcs(branch: string): OpenCodeEvent {
+  return {
+    id: `evt_vcs_${branch}`,
+    created: 0,
+    type: "vcs.branch.updated",
+    data: { branch },
+  }
+}