refactor(observability): trigger-driven trace updates — remove live streams and console polling

console/packages/console-frontend/src/api/config.ts

@@ -38,6 +38,14 @@ export function getStreamsWs(): string {
   return `${wsProtocol}//${host}/ws/streams`
 }
 
+export function getConsoleEventsWs(): string {
+  const wsProtocol =
+    typeof window !== 'undefined' && window.location.protocol === 'https:' ? 'wss:' : 'ws:'
+  const c = getConfig()
+  const host = typeof window !== 'undefined' ? window.location.host : `localhost:${c.consolePort}`
+  return `${wsProtocol}//${host}/ws/console-events`
+}
+
 export function getEngineBaseUrl(): string {
   return '/api/engine'
 }

console/packages/console-frontend/src/api/consoleEvents.ts

@@ -0,0 +1,65 @@
+import { getConsoleEventsWs } from './config'
+
+export interface ConsoleEventsHandlers {
+  /** A coalesced "these traces changed" tick. Carries ids only, never span
+   * data — consumers re-run their own filtered queries (notify-then-query),
+   * so the engine stays the single source of filter semantics. */
+  onTracesChanged: (traceIds: string[]) => void
+  /** Fired on every (re)connect; consumers resync anything missed while
+   * disconnected with one refetch. */
+  onConnect?: () => void
+}
+
+/**
+ * Live console-event feed (`/ws/console-events`): the console worker owns a
+ * `trace` trigger on the engine and forwards its coalesced ticks to every
+ * connected browser. Replaces the old 1s blind polling of the trace list —
+ * an idle engine produces no traffic at all.
+ */
+export function createConsoleEventsSubscription(handlers: ConsoleEventsHandlers): () => void {
+  let socket: WebSocket | null = null
+  let reconnectTimer: ReturnType<typeof setTimeout> | null = null
+  let disposed = false
+
+  const connect = () => {
+    if (disposed) return
+    socket = new WebSocket(getConsoleEventsWs())
+
+    socket.onopen = () => {
+      handlers.onConnect?.()
+    }
+
+    socket.onmessage = (event) => {
+      try {
+        const message = JSON.parse(event.data)
+        if (message?.type === 'traces_changed') {
+          const ids = Array.isArray(message.trace_ids)
+            ? message.trace_ids.filter((id: unknown): id is string => typeof id === 'string')
+            : []
+          handlers.onTracesChanged(ids)
+        }
+      } catch {
+        // Malformed frame — the next tick self-heals.
+      }
+    }
+
+    socket.onclose = () => {
+      socket = null
+      if (!disposed) {
+        reconnectTimer = setTimeout(connect, 3000)
+      }
+    }
+
+    socket.onerror = () => {
+      socket?.close()
+    }
+  }
+
+  connect()
+
+  return () => {
+    disposed = true
+    if (reconnectTimer) clearTimeout(reconnectTimer)
+    socket?.close()
+  }
+}

console/packages/console-frontend/src/hooks/useTraceData.ts

@@ -1,6 +1,7 @@
 import { useQuery } from '@tanstack/react-query'
 import { useEffect, useRef, useState } from 'react'
 import { fetchTraces } from '@/api'
+import { createConsoleEventsSubscription } from '@/api/consoleEvents'
 import type { TracesFilterParams } from '@/api/observability/traces'
 import { buildTraceGroups, type TraceGroup, traceGroupsFingerprint } from '@/lib/traceGroups'
 
@@ -13,6 +14,9 @@ export interface UseTraceDataOptions {
   showSystem: boolean
   debouncedSearch: string
   isPaused: boolean
+  /** Called on every trace-change tick (even while paused) with the touched
+   * trace ids, so the detail view can refresh an open trace. */
+  onTracesChanged?: (traceIds: string[]) => void
 }
 
 export interface UseTraceDataReturn {
@@ -31,6 +35,7 @@ export function useTraceData({
   showSystem,
   debouncedSearch,
   isPaused,
+  onTracesChanged,
 }: UseTraceDataOptions): UseTraceDataReturn {
   const [traceGroups, setTraceGroups] = useState<TraceGroup[]>([])
   const [hasOtelConfigured, setHasOtelConfigured] = useState(false)
@@ -58,13 +63,30 @@ export function useTraceData({
         limit: DEFAULT_TRACE_LIMIT,
         include_internal: showSystem,
       }),
-    // Interim: poll every 1s (was 3s) so freshly emitted spans surface sooner.
-    // The real fix is to subscribe to the engine's reactive trace-rows feed
-    // over the existing streams WebSocket instead of polling at all.
-    refetchInterval: isPaused ? false : 1000,
-    staleTime: 1000,
   })
 
+  // Notify-then-query: the console worker owns a `trace` trigger on the
+  // engine and forwards its coalesced `{trace_ids}` ticks over
+  // /ws/console-events; each tick re-runs the filtered query above. No
+  // polling — an idle engine produces zero traffic — and every (re)connect
+  // resyncs with one refetch, covering ticks missed while disconnected.
+  const isPausedRef = useRef(isPaused)
+  isPausedRef.current = isPaused
+  const refetchRef = useRef(refetch)
+  refetchRef.current = refetch
+  const onTracesChangedRef = useRef(onTracesChanged)
+  onTracesChangedRef.current = onTracesChanged
+
+  useEffect(() => {
+    return createConsoleEventsSubscription({
+      onConnect: () => refetchRef.current(),
+      onTracesChanged: (traceIds) => {
+        if (!isPausedRef.current) refetchRef.current()
+        onTracesChangedRef.current?.(traceIds)
+      },
+    })
+  }, [])
+
   useEffect(() => {
     if (!tracesData) return
 

console/packages/console-frontend/src/routes/traces.tsx

@@ -90,6 +90,35 @@ function TracesPage() {
   const [spansError, setSpansError] = useState<string | null>(null)
   const [isPaused, setIsPaused] = useState(false)
 
+  // Silent refresh of the OPEN trace on a trace-change tick: no loading
+  // spinner (ticks arrive continuously while the trace runs), one request in
+  // flight at a time, and a stale response for a since-deselected trace is
+  // dropped.
+  const selectedTraceIdRef = useRef<string | null>(null)
+  useEffect(() => {
+    selectedTraceIdRef.current = selectedTraceId
+  }, [selectedTraceId])
+  const refreshInFlightRef = useRef(false)
+  const handleTracesChanged = useCallback((traceIds: string[]) => {
+    const current = selectedTraceIdRef.current
+    if (!current || !traceIds.includes(current) || refreshInFlightRef.current) return
+    refreshInFlightRef.current = true
+    fetchTraceTree(current)
+      .then((data) => {
+        if (selectedTraceIdRef.current !== current) return
+        if (data.roots && data.roots.length > 0) {
+          const wfData = treeToWaterfallData(data.roots)
+          if (wfData) setWaterfallData(wfData)
+        }
+      })
+      .catch(() => {
+        // Keep the current waterfall; the next tick retries.
+      })
+      .finally(() => {
+        refreshInFlightRef.current = false
+      })
+  }, [])
+
   const {
     filters: filterState,
     updateFilter,
@@ -118,6 +147,7 @@ function TracesPage() {
     showSystem,
     debouncedSearch,
     isPaused,
+    onTracesChanged: handleTracesChanged,
   })
 
   const loadTraceSpans = useCallback(async (traceId: string) => {

console/packages/console-rust/src/bridge/events.rs

@@ -0,0 +1,46 @@
+use iii_sdk::protocol::RegisterTriggerInput;
+use iii_sdk::{Error, IIIClient, RegisterFunction};
+use serde_json::{json, Value};
+use tracing::info;
+
+/// Broadcast channel carrying serialized console events (currently only
+/// `traces_changed` ticks) from the engine bridge to every browser connected
+/// on `/ws/console-events`.
+pub type ConsoleEvents = tokio::sync::broadcast::Sender<String>;
+
+/// Register the trace-trigger handler and its `trace` trigger.
+///
+/// The engine coalesces span activity into one `{trace_ids}` tick per
+/// window and invokes the handler fire-and-forget; the handler fans the tick
+/// out to the browsers, which re-run their own filtered queries
+/// (notify-then-query — the engine stays the single source of filter
+/// semantics, and an idle engine produces no traffic at all). The trigger is
+/// owned by this bridge connection, so the engine unregisters it when the
+/// console disconnects.
+pub fn register_trace_events(bridge: &IIIClient, events: ConsoleEvents) -> Result<(), Error> {
+    let tick = events.clone();
+    bridge.register_function(
+        "engine::console::traces_changed",
+        RegisterFunction::new_async(move |input: Value| {
+            let events = tick.clone();
+            async move {
+                let trace_ids = input
+                    .get("trace_ids")
+                    .cloned()
+                    .unwrap_or_else(|| Value::Array(Vec::new()));
+                let frame = json!({ "type": "traces_changed", "trace_ids": trace_ids }).to_string();
+                // No receiver just means no console tab is open right now.
+                let _ = events.send(frame);
+                Ok(json!({ "delivered": true }))
+            }
+        }),
+    );
+
+    info!("Registering trace trigger: engine::console::traces_changed");
+    bridge.register_trigger(RegisterTriggerInput::new(
+        "trace",
+        "engine::console::traces_changed",
+        json!({}),
+    ))?;
+    Ok(())
+}

console/packages/console-rust/src/bridge/mod.rs

@@ -1,6 +1,8 @@
 mod error;
+mod events;
 mod functions;
 mod triggers;
 
+pub use events::{register_trace_events, ConsoleEvents};
 pub use functions::register_functions;
 pub use triggers::register_triggers;

console/packages/console-rust/src/main.rs

@@ -180,13 +180,21 @@ async fn main() -> Result<()> {
         tracing::warn!("Trigger registration failed: {}", e);
     }
 
+    // Trace-change ticks flow engine trigger -> bridge handler -> this
+    // channel -> every browser on /ws/console-events.
+    let (events, _) = tokio::sync::broadcast::channel::<String>(64);
+    if let Err(e) = bridge::register_trace_events(&bridge, events.clone()) {
+        tracing::warn!("Trace trigger registration failed: {}", e);
+    }
+
     let config = server::ServerConfig {
         port: args.port,
         host: args.host,
         engine_host: args.engine_host,
         engine_port: args.engine_port,
         ws_port: args.ws_port,
         enable_flow: args.enable_flow,
+        events,
     };
 
     // Run server with graceful shutdown

console/packages/console-rust/src/server.rs

@@ -29,6 +29,9 @@ pub struct ServerConfig {
     pub engine_port: u16,
     pub ws_port: u16,
     pub enable_flow: bool,
+    /// Console-event broadcast (trace-change ticks) fanned out on
+    /// `/ws/console-events`.
+    pub events: crate::bridge::ConsoleEvents,
 }
 
 pub struct AppState {
@@ -91,6 +94,40 @@ async fn serve_config(
     }))
 }
 
+/// Fan console events (trace-change ticks) out to a browser. Send-only from
+/// the browser's perspective; incoming frames are ignored except to detect
+/// close. A lagged receiver just skips ticks — the client refetches on the
+/// next one, and resyncs once per (re)connect anyway.
+async fn console_events_handler(
+    axum::extract::State(state): axum::extract::State<std::sync::Arc<AppState>>,
+    ws: axum::extract::ws::WebSocketUpgrade,
+) -> Response {
+    let mut events = state.config.events.subscribe();
+    ws.on_upgrade(move |mut socket| async move {
+        loop {
+            tokio::select! {
+                event = events.recv() => match event {
+                    Ok(frame) => {
+                        if socket
+                            .send(axum::extract::ws::Message::Text(frame.into()))
+                            .await
+                            .is_err()
+                        {
+                            break;
+                        }
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                },
+                incoming = socket.recv() => match incoming {
+                    Some(Ok(_)) => continue,
+                    _ => break,
+                },
+            }
+        }
+    })
+}
+
 /// Serve the index.html with runtime config
 async fn serve_index(
     axum::extract::State(state): axum::extract::State<std::sync::Arc<AppState>>,
@@ -214,7 +251,8 @@ pub async fn run_server(config: ServerConfig) -> Result<()> {
 
     let mut app = Router::new()
         .route("/", get(serve_index))
-        .route("/api/config", get(serve_config));
+        .route("/api/config", get(serve_config))
+        .route("/ws/console-events", any(console_events_handler));
 
     // IMPORTANT: Call .with_state(proxy) BEFORE merging. This converts
     // Router<Arc<ProxyState>> to Router<()>, which can be merged into

engine/src/workers/observability/README.md

@@ -230,7 +230,7 @@ final span **replaces the snapshot in place** — same storage position, one ent
 
 This is what lets live trace views show in-progress work: engine-side parent spans (`trigger <fn>`,
 `enqueue`, builtin `call <fn>`) become visible while the work under them is still running, traces
-appear in list views as soon as they start, and the trace trigger / devtools streams tick on span
+appear in list views as soon as they start, and the trace triggers tick on span
 start as well as close. Worker (SDK) spans still arrive only when they close — they are ingested
 over OTLP, which has no notion of an unfinished span.
 
@@ -239,8 +239,8 @@ Consumers of the Traces API should treat `pending: true` (or `end_time_unix_nano
 `engine::metrics::list` excludes them from latency statistics. The `pending` field is only
 serialized when true, so the wire shape of finished spans is unchanged.
 
-**OTEL compliance:** pending snapshots exist ONLY in the in-memory store and the `iii:devtools:*`
-streams fed from it. The OTLP export path (both the engine exporter and the SDK-span forwarder) is
+**OTEL compliance:** pending snapshots exist ONLY in the in-memory store (and the query views and
+trace-trigger ticks fed from it). The OTLP export path (both the engine exporter and the SDK-span forwarder) is
 untouched and only ever ships complete spans — `end_time_unix_nano` is semantically required on the
 OTLP wire, and nothing partial ever reaches it. Spans that never close (e.g. a leaked guard) remain
 pending until buffer eviction; they are a dev-store artifact, not exported data.