feat(console): reactive traces + session-events live streams

console/web/src/lib/backend/real.ts

@@ -7,11 +7,8 @@ import { parseCatalogModelKey } from '@/lib/catalog-model-key'
 import { getIiiClient } from '@/lib/iii-client'
 import { newMessageId } from '@/lib/session-id'
 import type { Mode, ModelId } from '@/types/chat'
-import type {
-  AgentEvent,
-  AgentMessage,
-  SessionEventEnvelope,
-} from '@/types/iii-agent-event'
+import type { AgentEvent, AgentMessage } from '@/types/iii-agent-event'
+import { startSessionEventsSubscription } from './session-events-live'
 import { createAgentEventTranslator } from './translate'
 import type {
   ChatBackend,
@@ -57,24 +54,25 @@ async function* realStream(
     r?.()
   }
 
-  const off = client.on<SessionEventEnvelope>('ui::session::event', (env) => {
-    if (!env || env.session_id !== sessionId || !env.event) return
-    queue.push(env.event)
-    wake()
-  })
+  // Subscribe directly to this session's `agent::events` stream via a scoped
+  // engine stream trigger (`group_id = sessionId`), replacing the harness
+  // fanout hop (`ui::subscribe` → per-browser `ui::session::event` push).
+  // Registered before the `harness::trigger` kickoff below — both travel the
+  // same ordered WS connection, so the trigger is in place before the turn's
+  // first event is written.
+  const stopSubscription = startSessionEventsSubscription(
+    client,
+    sessionId,
+    (event) => {
+      queue.push(event)
+      wake()
+    },
+  )
 
   const onAbort = () => wake()
   signal?.addEventListener('abort', onAbort, { once: true })
 
-  let subscribed = false
-
   try {
-    await client.call('ui::subscribe', {
-      browser_id: client.browserId,
-      session_id: sessionId,
-    })
-    subscribed = true
-
     const { translate } = createAgentEventTranslator()
 
     client
@@ -155,15 +153,7 @@ async function* realStream(
     }
   } finally {
     signal?.removeEventListener('abort', onAbort)
-    off()
-    if (subscribed) {
-      await client
-        .call('ui::unsubscribe', {
-          browser_id: client.browserId,
-          session_id: sessionId,
-        })
-        .catch(() => {})
-    }
+    stopSubscription()
   }
 }
 

console/web/src/lib/backend/session-events-live.test.ts

@@ -0,0 +1,155 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { IiiClient } from '@/lib/iii-client'
+import type { AgentEvent } from '@/types/iii-agent-event'
+import {
+  extractSessionEvent,
+  startSessionEventsSubscription,
+} from './session-events-live'
+
+/**
+ * Build the raw `agent::events` stream frame the engine delivers to a stream
+ * trigger handler: `serde_json::to_value(StreamWrapperMessage)` →
+ * `{ type, timestamp, streamName, groupId, id, event: { data } }`, where
+ * `event.data` is the AgentEvent the harness wrote via `stream::set`.
+ */
+function frame(groupId: string, event: AgentEvent, opts?: { flat?: boolean }) {
+  if (opts?.flat) {
+    return { groupId, streamName: 'agent::events', data: event }
+  }
+  return {
+    type: 'set',
+    timestamp: 1,
+    streamName: 'agent::events',
+    groupId,
+    id: `${groupId}-epoch-00000000`,
+    event: { data: event },
+  }
+}
+
+const UPDATE = { type: 'message_update' } as unknown as AgentEvent
+const END = { type: 'agent_end', messages: [] } as unknown as AgentEvent
+
+describe('extractSessionEvent', () => {
+  it('extracts the inner AgentEvent from a real engine frame (event.data)', () => {
+    expect(extractSessionEvent(frame('sess-1', UPDATE), 'sess-1')).toEqual(
+      UPDATE,
+    )
+  })
+
+  it('falls back to a flat `data` field when there is no event wrapper', () => {
+    expect(
+      extractSessionEvent(frame('sess-1', END, { flat: true }), 'sess-1'),
+    ).toEqual(END)
+  })
+
+  it('accepts the snake_case group_id key as well as camelCase groupId', () => {
+    const snake = { group_id: 'sess-1', data: UPDATE }
+    expect(extractSessionEvent(snake, 'sess-1')).toEqual(UPDATE)
+  })
+
+  it('drops a frame whose group_id is a different session', () => {
+    expect(extractSessionEvent(frame('sess-2', UPDATE), 'sess-1')).toBeNull()
+  })
+
+  it('drops a frame with no group_id (cannot confirm ownership)', () => {
+    expect(extractSessionEvent({ data: UPDATE }, 'sess-1')).toBeNull()
+  })
+
+  it('returns null for null / non-object / payload-less frames', () => {
+    expect(extractSessionEvent(null, 'sess-1')).toBeNull()
+    expect(extractSessionEvent('nope', 'sess-1')).toBeNull()
+    expect(extractSessionEvent({ groupId: 'sess-1' }, 'sess-1')).toBeNull()
+  })
+})
+
+function fakeClient() {
+  const triggers: Array<{
+    type: string
+    function_id: string
+    config?: unknown
+  }> = []
+  let handler: ((p: unknown) => void) | null = null
+  const offHandler = vi.fn()
+  const triggerUnregister = vi.fn()
+
+  const on = vi.fn((_fn: string, h: (p: unknown) => void) => {
+    handler = h
+    return offHandler
+  })
+  const registerTrigger = vi.fn(
+    (input: { type: string; function_id: string; config?: unknown }) => {
+      triggers.push(input)
+      return triggerUnregister
+    },
+  )
+
+  const client = {
+    browserId: 'console-test',
+    on,
+    registerTrigger,
+    call: vi.fn(),
+    addConnectionStateListener: vi.fn(),
+    dispose: vi.fn(async () => {}),
+  } as unknown as IiiClient
+
+  return {
+    client,
+    on,
+    registerTrigger,
+    triggers,
+    offHandler,
+    triggerUnregister,
+    fire: (f: unknown) => handler?.(f),
+  }
+}
+
+describe('startSessionEventsSubscription', () => {
+  it('registers an iii::-prefixed handler and a stream trigger scoped to the session', () => {
+    const { client, on, triggers } = fakeClient()
+
+    startSessionEventsSubscription(client, 'sess-1', () => {})
+
+    expect(on).toHaveBeenCalledWith(
+      'iii::console::session_event',
+      expect.any(Function),
+    )
+    expect(triggers).toEqual([
+      {
+        type: 'stream',
+        function_id: 'iii::console::session_event::console-test',
+        config: { stream_name: 'agent::events', group_id: 'sess-1' },
+      },
+    ])
+  })
+
+  it('delivers each extracted AgentEvent for this session to onEvent', () => {
+    const { client, fire } = fakeClient()
+    const onEvent = vi.fn()
+
+    startSessionEventsSubscription(client, 'sess-1', onEvent)
+    fire(frame('sess-1', UPDATE))
+
+    expect(onEvent).toHaveBeenCalledTimes(1)
+    expect(onEvent).toHaveBeenCalledWith(UPDATE)
+  })
+
+  it('does not deliver a frame addressed to another session', () => {
+    const { client, fire } = fakeClient()
+    const onEvent = vi.fn()
+
+    startSessionEventsSubscription(client, 'sess-1', onEvent)
+    fire(frame('sess-2', UPDATE))
+
+    expect(onEvent).not.toHaveBeenCalled()
+  })
+
+  it('unregisters the handler and the trigger on cleanup', () => {
+    const { client, offHandler, triggerUnregister } = fakeClient()
+
+    const stop = startSessionEventsSubscription(client, 'sess-1', () => {})
+    stop()
+
+    expect(offHandler).toHaveBeenCalledTimes(1)
+    expect(triggerUnregister).toHaveBeenCalledTimes(1)
+  })
+})

console/web/src/lib/backend/session-events-live.ts

@@ -0,0 +1,99 @@
+/**
+ * Direct subscription to a session's `agent::events` stream, replacing the
+ * harness fanout hop (`ui::subscribe` → per-browser `ui::session::event` push).
+ *
+ * The browser registers a local handler and binds a SCOPED engine stream
+ * trigger (`config.group_id = session_id`) to it. The engine matches stream
+ * triggers by `(stream_name, group_id, item_id)` and delivers only matching
+ * frames straight to this browser's WS connection (engine
+ * `stream.rs::invoke_triggers`), so a browser receives exactly its own
+ * session's events — without the harness re-pushing them, and with no
+ * `harness::fanout::*` / per-browser `ui::session::event` spans.
+ *
+ * The handler is named with the `iii::` prefix (`is_iii_builtin_function_id`),
+ * so the spans produced by DELIVERING this trigger are tagged
+ * `iii.function.kind=internal` — hidden from the Traces view by the default
+ * `include_internal:false` query and skipped by the engine's trigger
+ * loop-break, matching the `traces-live.ts` approach. Without this, every
+ * delivery would flood the trace list with `session_event` spans.
+ *
+ * The iii-browser-sdk replays both registered functions and triggers on
+ * reconnect (see `onSocketOpen`), so the trigger re-binds automatically; the
+ * chat backend re-seeds turn state on start via `turn::get_state`, mirroring
+ * the pre-existing fanout behavior (no per-event replay on reconnect).
+ */
+
+import type { IiiClient } from '@/lib/iii-client'
+import type { AgentEvent } from '@/types/iii-agent-event'
+
+/** iii:: prefix → engine-internal → delivery spans hidden + trigger loop-break skip. */
+const SESSION_EVENT_FN = 'iii::console::session_event'
+/** The firehose stream the harness writes every agent event onto. */
+const EVENTS_STREAM = 'agent::events'
+
+/**
+ * Pull the AgentEvent out of a raw `agent::events` stream frame, scoped to one
+ * session. The engine serializes `StreamWrapperMessage` as
+ * `{ groupId, event: { data }, … }`; some shapes carry a flat `data`. This
+ * mirrors the extraction the (now-removed) harness `agent::events` fanout
+ * pump performed before browsers subscribed to the stream directly.
+ *
+ * Returns null when the frame is malformed, carries no extractable event, or
+ * is addressed to a different session. The stream trigger is already
+ * group-scoped, so the session check is defense-in-depth against mis-delivery
+ * and preserves the strict `session_id` guard the fanout path enforced.
+ */
+export function extractSessionEvent(
+  frame: unknown,
+  sessionId: string,
+): AgentEvent | null {
+  if (!frame || typeof frame !== 'object') return null
+  const obj = frame as Record<string, unknown>
+
+  const groupId =
+    (typeof obj.groupId === 'string' && obj.groupId) ||
+    (typeof obj.group_id === 'string' && obj.group_id) ||
+    null
+  if (groupId !== sessionId) return null
+
+  const wrapper =
+    obj.event && typeof obj.event === 'object'
+      ? (obj.event as Record<string, unknown>)
+      : null
+  const inner = wrapper && 'data' in wrapper ? wrapper.data : (obj.data ?? null)
+  if (!inner || typeof inner !== 'object') return null
+  return inner as AgentEvent
+}
+
+/**
+ * Register the handler + a `agent::events` stream trigger scoped to
+ * `sessionId`, delivering each extracted AgentEvent to `onEvent`. Returns a
+ * cleanup that unregisters both (replacing the old `ui::unsubscribe`).
+ */
+export function startSessionEventsSubscription(
+  client: Pick<IiiClient, 'browserId' | 'on' | 'registerTrigger'>,
+  sessionId: string,
+  onEvent: (event: AgentEvent) => void,
+): () => void {
+  const off = client.on(SESSION_EVENT_FN, (frame: unknown) => {
+    const event = extractSessionEvent(frame, sessionId)
+    if (event) onEvent(event)
+  })
+
+  // `on()` registers under `<fn>::<browserId>`; the trigger must target that id.
+  const functionId = `${SESSION_EVENT_FN}::${client.browserId}`
+  const offTrigger = client.registerTrigger({
+    type: 'stream',
+    function_id: functionId,
+    config: { stream_name: EVENTS_STREAM, group_id: sessionId },
+  })
+
+  return () => {
+    off()
+    try {
+      offTrigger()
+    } catch {
+      // SDK already disposed; nothing to do.
+    }
+  }
+}

console/web/src/lib/backend/types.ts

@@ -70,8 +70,9 @@ export interface ChatStreamOptions {
    * for the same conversation must pass the same value so the engine
    * groups every turn under one session in the traces UI.
    *
-   * The real backend uses it as `session_id` in `ui::subscribe` and
-   * `run::start`; the mock backend ignores it. When omitted, the real
+   * The real backend uses it as the `group_id` of its scoped `agent::events`
+   * stream trigger and as `session_id` in `harness::trigger`; the mock backend
+   * ignores it. When omitted, the real
    * backend falls back to a fresh `console-<uuid>` so callers that
    * haven't been updated yet still work (with the pre-fix behavior of
    * one session per send).

console/web/src/lib/iii-client.ts

@@ -10,8 +10,9 @@
  *      which proxies `/ws` to the engine).
  *   2. Open the WebSocket via `iii-browser-sdk::registerWorker(url)`.
  *   3. Mint a stable `browser_id` for this page; per-browser handlers are
- *      registered under `<functionId>::<browserId>` so the harness fanout
- *      can target this specific browser when it pushes events.
+ *      registered under `<functionId>::<browserId>` so engine triggers — the
+ *      browser's own scoped stream subscriptions and the harness sessions
+ *      fan-out — deliver to this specific browser's connection.
  *
  * Once `_clientPromise` is resolved, every other call (`call`, `on`,
  * `dispose`) goes over the single WS connection.