Own the live socket outside React so leaving Home cannot orphan it

The socket belonged to whichever hook instance ran its effect first. Passive
effects run children before parents, so on Home that was the brain graph rather
than the shell: navigating to a sibling route unmounted the graph, which closed
the socket, while the shell kept the reference count above zero. From there the
page was deaf for the rest of its life, with nothing left to reconnect it.

A module-level connection with a real refcount replaces it — connect on the
first subscriber, disconnect on the last — and the hook is a thin subscription
with the same signature, correct under StrictMode's mount/unmount/mount.

A 1008 now reconnects instead of returning silently: the token is read afresh
per attempt, and three consecutive rejections fall through to the caller's auth
handler so a revoked session surfaces rather than spins.

The snapshot's emit counts are read into a store of their own, apart from the
live count, so a graph that connects into a busy engine is drawn as busy without
every neuron claiming it just fired. The neuron and edge pulses now key off a
change seen while they were mounted, so returning to Home no longer replays
every emission of the session.

Home gets a live indicator for the case none of this can fix: quiet while the
socket is up, and named in words when it is down, since HTTP polling keeps the
rest of the page looking current.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HTsT1isxUjw5gtkJk8WhuA

frontend/src/components/Flow/BrainEdge.tsx

@@ -47,7 +47,10 @@ function BrainEdgeComponent({ id, source, target, data }: EdgeProps) {
   const targetNode = useInternalNode(target)
   const ts = useLatestTs(messages)
   const [pulsing, setPulsing] = useState(false)
-  const lastTs = useRef(0)
+  // Seeded with whatever the store already holds rather than with 0: the store
+  // outlives this component, so a value that arrived before it mounted is
+  // history and must not be replayed as a pulse.
+  const lastTs = useRef(ts)
 
   // The two centres, taken from the layout rather than from the `sourceX`/
   // `targetX` React Flow hands an edge. Those come off the handles, which it

frontend/src/components/Flow/liveStore.ts

@@ -58,9 +58,14 @@ const values = new Map<string, LiveValue>()
 const statuses = new Map<string, LiveStatus>()
 const health = new Map<string, NodeHealth>()
 let engineEvents: EngineEvent[] = []
-// How many times a node has emitted. The number itself means nothing; a change
-// is what restarts the pulse.
+// How many times this page has seen a node emit. The number itself means
+// nothing; a change is what restarts the pulse.
 const emits = new Map<string, number>()
+// What the engine had already counted when we connected, from the snapshot.
+// Kept apart from the live count on purpose: it says the node has been busy,
+// which is worth drawing, but it happened before anyone was watching, and a
+// pulse claims something is happening now.
+const priorEmits = new Map<string, number>()
 let logLines: LogLine[] = []
 const paused = new Set<string>()
 const listeners = new Map<string, Set<Listener>>()
@@ -133,6 +138,21 @@ export const liveStore = {
     emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1)
     notify(`emit:${nodeId}`)
   },
+  /**
+   * The counts a snapshot carries, so a page that connects into an engine
+   * already at work draws what has been busy rather than a graph of idle
+   * neurons.
+   *
+   * Absent keys are left alone rather than cleared: a restarted engine counts
+   * from zero again, which is no reason to forget what this page saw.
+   */
+  setEmits(entries: Record<string, number>) {
+    for (const [nodeId, count] of Object.entries(entries)) {
+      if (priorEmits.get(nodeId) === count) continue
+      priorEmits.set(nodeId, count)
+      notify(`emit:${nodeId}`)
+    }
+  },
   appendLog(line: LogLine) {
     // A new array each time, so the hook's snapshot comparison sees the change.
     logLines = [...logLines, line].slice(-LOG_LIMIT)
@@ -173,8 +193,10 @@ export const liveStore = {
     values.clear()
     for (const key of statuses.keys()) notify(`status:${key}`)
     statuses.clear()
-    for (const key of emits.keys()) notify(`emit:${key}`)
+    for (const key of new Set([...emits.keys(), ...priorEmits.keys()]))
+      notify(`emit:${key}`)
     emits.clear()
+    priorEmits.clear()
     for (const key of health.keys()) notify(`health:${key}`)
     health.clear()
     engineEvents = []
@@ -251,6 +273,9 @@ export function useFlowPaused(flow: string): boolean {
 
 const SEP = "\u0000"
 
+/** Back to the array. `"".split(SEP)` is `[""]`, which is one empty name. */
+const parts = (joined: string) => (joined ? joined.split(SEP) : [])
+
 function subscribeAll(keys: string[], listener: Listener) {
   const unsubscribes = keys.map((key) => subscribeKey(key, listener))
   return () => {
@@ -265,13 +290,37 @@ export function useGroupEmits(ids: string[]): number {
     useCallback(
       (listener: Listener) =>
         subscribeAll(
-          joined.split(SEP).map((id) => `emit:${id}`),
+          parts(joined).map((id) => `emit:${id}`),
+          listener,
+        ),
+      [joined],
+    ),
+    () => parts(joined).reduce((total, id) => total + (emits.get(id) ?? 0), 0),
+  )
+}
+
+/**
+ * Whether any of these nodes has ever published, this page or before it.
+ *
+ * Separate from the count above because it answers a different question: the
+ * count is what pulses, and this is what tells a neuron nobody has ever seen
+ * run from one that is merely quiet just now.
+ */
+export function useGroupActive(ids: string[]): boolean {
+  const joined = ids.join(SEP)
+  return useSyncExternalStore(
+    useCallback(
+      (listener: Listener) =>
+        subscribeAll(
+          parts(joined).map((id) => `emit:${id}`),
           listener,
         ),
       [joined],
     ),
     () =>
-      joined.split(SEP).reduce((total, id) => total + (emits.get(id) ?? 0), 0),
+      parts(joined).some(
+        (id) => (emits.get(id) ?? 0) + (priorEmits.get(id) ?? 0) > 0,
+      ),
   )
 }
 
@@ -282,12 +331,12 @@ export function useGroupError(ids: string[]): boolean {
     useCallback(
       (listener: Listener) =>
         subscribeAll(
-          joined.split(SEP).map((id) => `status:${id}`),
+          parts(joined).map((id) => `status:${id}`),
           listener,
         ),
       [joined],
     ),
-    () => joined.split(SEP).some((id) => statuses.get(id)?.status === "error"),
+    () => parts(joined).some((id) => statuses.get(id)?.status === "error"),
   )
 }
 
@@ -298,18 +347,16 @@ export function useLatestTs(names: string[]): number {
     useCallback(
       (listener: Listener) =>
         subscribeAll(
-          joined.split(SEP).map((name) => `value:${name}`),
+          parts(joined).map((name) => `value:${name}`),
           listener,
         ),
       [joined],
     ),
     () =>
-      joined
-        .split(SEP)
-        .reduce(
-          (latest, name) => Math.max(latest, values.get(name)?.ts ?? 0),
-          0,
-        ),
+      parts(joined).reduce(
+        (latest, name) => Math.max(latest, values.get(name)?.ts ?? 0),
+        0,
+      ),
   )
 }
 

frontend/src/components/Flow/useFlowSocket.ts

@@ -1,5 +1,6 @@
+import type { QueryClient } from "@tanstack/react-query"
 import { useQueryClient } from "@tanstack/react-query"
-import { useEffect, useRef } from "react"
+import { useEffect } from "react"
 
 import { OpenAPI } from "@/client"
 import { connectionStore } from "@/lib/connectionStore"
@@ -8,10 +9,18 @@ import { type LogLine, liveStore, type ValueSource } from "./liveStore"
 import { flowKeys } from "./queries"
 
 const RECONNECT_MIN = 1000
-
-/** How many components want the socket open. */
-let mounted = 0
 const RECONNECT_MAX = 30000
+/**
+ * Consecutive token rejections before the session is given up on.
+ *
+ * A 1008 means the server would not take the token. Under a portal that is
+ * routine — the handoff token lasts twelve hours against a local session's
+ * eight days — so it has to be recoverable rather than fatal: the token is read
+ * again for every attempt, so a refreshed one is picked up. A session that has
+ * genuinely been revoked fails all three and falls through to the caller's auth
+ * handler instead of retrying forever.
+ */
+const AUTH_ATTEMPTS = 3
 
 type FlowEvent =
   | {
@@ -20,6 +29,8 @@ type FlowEvent =
       nodes: { id: string; status: string; error?: string | null }[]
       paused?: string[]
       logs?: LogLine[]
+      /** Emission counts per qualified node. Absent on older installations. */
+      emits?: Record<string, number>
     }
   | {
       type: "message_value"
@@ -85,163 +96,202 @@ function socketUrl(): string {
   const url = new URL(`${base.replace(/\/$/, "")}/api/v1/flows/ws`)
   url.protocol = url.protocol === "https:" ? "wss:" : "ws:"
   // Browsers cannot set headers on a websocket handshake, so the token rides
-  // in the query string.
+  // in the query string. Read per attempt, so a reconnect after the old one
+  // expired carries whatever is current rather than what we started with.
   url.searchParams.set("token", apiToken())
   return url.toString()
 }
 
-/**
- * Keeps one socket open for the editor, feeding the live store.
+/*
+ * One socket for the page, owned here rather than by a component.
  *
- * @param onAuthFailure called when the server rejects the token, so the caller
- *   can send the user back to the login screen.
+ * Several places want live events at once — the app shell, the brain graph on
+ * Home, the editor — and React mounts and unmounts them in an order none of
+ * them controls: passive effects run children before parents, and StrictMode
+ * runs mount → unmount → mount. Keeping the socket inside whichever hook
+ * instance ran first made that one its owner, so leaving Home closed the socket
+ * while the shell kept the count above zero and nothing ever reopened it. The
+ * connection lives out here now with a plain refcount; a component only says
+ * whether it is still watching.
+ */
+let socket: WebSocket | null = null
+let timer: ReturnType<typeof setTimeout> | null = null
+let retry = RECONNECT_MIN
+/** How many mounted components want the socket open. */
+let watchers = 0
+/** Consecutive 1008s, cleared by a handshake the server accepts. */
+let rejected = 0
+let client: QueryClient | null = null
+const authHandlers = new Set<() => void>()
+
+function schedule() {
+  if (timer) return
+  timer = setTimeout(() => {
+    timer = null
+    connect()
+  }, retry)
+  retry = Math.min(retry * 2, RECONNECT_MAX)
+}
+
+function connect() {
+  if (watchers === 0 || socket || timer) return
+  const ws = new WebSocket(socketUrl())
+  socket = ws
+
+  ws.onopen = () => {
+    retry = RECONNECT_MIN
+    rejected = 0
+    liveStore.setConnected(true)
+    connectionStore.setOnline()
+    // Whatever happened while the socket was down was missed, so nothing
+    // held in cache can be trusted to still be current.
+    client?.invalidateQueries()
+  }
+
+  ws.onmessage = (event) => {
+    const message: FlowEvent = JSON.parse(event.data)
+    switch (message.type) {
+      case "snapshot":
+        liveStore.setValues(message.values)
+        liveStore.setStatuses(message.nodes)
+        liveStore.setPausedFlows(message.paused ?? [])
+        liveStore.setLogs(message.logs ?? [])
+        // Missing against an installation older than this bundle; the graph
+        // then starts from zero the way it always did.
+        liveStore.setEmits(message.emits ?? {})
+        break
+      case "message_value":
+        liveStore.setValue(message.name, {
+          value: message.value,
+          ts: message.ts,
+          source: message.source,
+        })
+        break
+      case "node_started":
+        liveStore.setStatus(message.node, { status: "running" })
+        break
+      case "node_executed":
+        liveStore.setStatus(message.node, { status: "success" })
+        if (message.outputs > 0) liveStore.recordEmit(message.node)
+        break
+      case "node_error":
+        liveStore.setStatus(message.node, {
+          status: "error",
+          error: message.error,
+        })
+        liveStore.recordEngineEvent({
+          type: message.type,
+          flow: message.flow,
+          node: message.node,
+          detail: message.error,
+          ts: message.ts ?? Date.now() / 1000,
+        })
+        break
+      case "node_health":
+        liveStore.setHealth(message.node, {
+          health: message.health,
+          detail: message.detail,
+        })
+        if (message.health === "down") {
+          liveStore.recordEngineEvent({
+            type: message.type,
+            flow: message.flow,
+            node: message.node,
+            detail: message.detail ?? "Reported itself down.",
+            ts: message.ts ?? Date.now() / 1000,
+          })
+        }
+        break
+      case "flow_quarantined":
+      case "engine_degraded":
+      case "engine_fatal":
+      case "cascade_dropped":
+      case "queue_unavailable":
+        liveStore.recordEngineEvent({
+          type: message.type,
+          flow: "flow" in message ? message.flow : undefined,
+          node: "node" in message ? message.node : undefined,
+          detail:
+            ("error" in message ? message.error : undefined) ??
+            ("reason" in message ? message.reason : undefined) ??
+            "",
+          ts: message.ts ?? Date.now() / 1000,
+        })
+        break
+      case "node_status":
+        liveStore.setStatus(message.node, {
+          status: message.status as "active" | "error",
+          error: message.error,
+        })
+        break
+      case "node_log":
+        liveStore.appendLog(message)
+        break
+      case "flow_paused":
+        liveStore.setPaused(message.flow, message.paused)
+        break
+      case "pipeline_rebuilt":
+        liveStore.setStatuses(message.nodes)
+        liveStore.setPausedFlows(message.paused ?? [])
+        // Someone published or started a flow, here or in another tab: the
+        // markers on the flow chips are stale until the list is refetched.
+        client?.invalidateQueries({ queryKey: flowKeys.all })
+        break
+    }
+  }
+
+  ws.onclose = (event) => {
+    // A socket we already dropped: its close says nothing about the connection
+    // we want now.
+    if (socket !== ws) return
+    socket = null
+    liveStore.setConnected(false)
+    if (watchers === 0) return
+    if (event.code === 1008) {
+      rejected += 1
+      if (rejected >= AUTH_ATTEMPTS) {
+        for (const handler of authHandlers) handler()
+        return
+      }
+    }
+    // 1013 is the portal saying the installation is not attached — the one
+    // close code that means "offline" rather than "the socket dropped".
+    if (event.code === 1013) {
+      connectionStore.setOffline(null)
+    }
+    schedule()
+  }
+}
+
+function release(onAuthFailure?: () => void) {
+  watchers -= 1
+  if (onAuthFailure) authHandlers.delete(onAuthFailure)
+  if (watchers > 0) return
+  if (timer) {
+    clearTimeout(timer)
+    timer = null
+  }
+  const ws = socket
+  socket = null
+  ws?.close()
+  liveStore.setConnected(false)
+}
+
+/**
+ * Keeps one socket open while anything on the page wants live events.
+ *
+ * @param onAuthFailure called once the server has refused the token often
+ *   enough to mean the session is gone, so the caller can send the user back to
+ *   the login screen.
  */
 export function useFlowSocket(onAuthFailure?: () => void): void {
-  const socket = useRef<WebSocket | null>(null)
-  const retry = useRef(RECONNECT_MIN)
-  const timer = useRef<ReturnType<typeof setTimeout> | null>(null)
-  const closed = useRef(false)
   const queryClient = useQueryClient()
 
   useEffect(() => {
-    // The editor and a dashboard can both be mounted; one socket serves both,
-    // and the second caller just rides along.
-    mounted += 1
-    if (mounted > 1) {
-      return () => {
-        mounted -= 1
-      }
-    }
-    closed.current = false
-
-    const connect = () => {
-      if (closed.current) return
-      const ws = new WebSocket(socketUrl())
-      socket.current = ws
-
-      ws.onopen = () => {
-        retry.current = RECONNECT_MIN
-        liveStore.setConnected(true)
-        connectionStore.setOnline()
-        // Whatever happened while the socket was down was missed, so nothing
-        // held in cache can be trusted to still be current.
-        queryClient.invalidateQueries()
-      }
-
-      ws.onmessage = (event) => {
-        const message: FlowEvent = JSON.parse(event.data)
-        switch (message.type) {
-          case "snapshot":
-            liveStore.setValues(message.values)
-            liveStore.setStatuses(message.nodes)
-            liveStore.setPausedFlows(message.paused ?? [])
-            liveStore.setLogs(message.logs ?? [])
-            break
-          case "message_value":
-            liveStore.setValue(message.name, {
-              value: message.value,
-              ts: message.ts,
-              source: message.source,
-            })
-            break
-          case "node_started":
-            liveStore.setStatus(message.node, { status: "running" })
-            break
-          case "node_executed":
-            liveStore.setStatus(message.node, { status: "success" })
-            if (message.outputs > 0) liveStore.recordEmit(message.node)
-            break
-          case "node_error":
-            liveStore.setStatus(message.node, {
-              status: "error",
-              error: message.error,
-            })
-            liveStore.recordEngineEvent({
-              type: message.type,
-              flow: message.flow,
-              node: message.node,
-              detail: message.error,
-              ts: message.ts ?? Date.now() / 1000,
-            })
-            break
-          case "node_health":
-            liveStore.setHealth(message.node, {
-              health: message.health,
-              detail: message.detail,
-            })
-            if (message.health === "down") {
-              liveStore.recordEngineEvent({
-                type: message.type,
-                flow: message.flow,
-                node: message.node,
-                detail: message.detail ?? "Reported itself down.",
-                ts: message.ts ?? Date.now() / 1000,
-              })
-            }
-            break
-          case "flow_quarantined":
-          case "engine_degraded":
-          case "engine_fatal":
-          case "cascade_dropped":
-          case "queue_unavailable":
-            liveStore.recordEngineEvent({
-              type: message.type,
-              flow: "flow" in message ? message.flow : undefined,
-              node: "node" in message ? message.node : undefined,
-              detail:
-                ("error" in message ? message.error : undefined) ??
-                ("reason" in message ? message.reason : undefined) ??
-                "",
-              ts: message.ts ?? Date.now() / 1000,
-            })
-            break
-          case "node_status":
-            liveStore.setStatus(message.node, {
-              status: message.status as "active" | "error",
-              error: message.error,
-            })
-            break
-          case "node_log":
-            liveStore.appendLog(message)
-            break
-          case "flow_paused":
-            liveStore.setPaused(message.flow, message.paused)
-            break
-          case "pipeline_rebuilt":
-            liveStore.setStatuses(message.nodes)
-            liveStore.setPausedFlows(message.paused ?? [])
-            // Someone published or started a flow, here or in another tab: the
-            // markers on the flow chips are stale until the list is refetched.
-            queryClient.invalidateQueries({ queryKey: flowKeys.all })
-            break
-        }
-      }
-
-      ws.onclose = (event) => {
-        liveStore.setConnected(false)
-        if (closed.current) return
-        if (event.code === 1008) {
-          onAuthFailure?.()
-          return
-        }
-        // 1013 is the portal saying the installation is not attached — the one
-        // close code that means "offline" rather than "the socket dropped".
-        if (event.code === 1013) {
-          connectionStore.setOffline(null)
-        }
-        timer.current = setTimeout(connect, retry.current)
-        retry.current = Math.min(retry.current * 2, RECONNECT_MAX)
-      }
-    }
-
+    client = queryClient
+    if (onAuthFailure) authHandlers.add(onAuthFailure)
+    watchers += 1
     connect()
-
-    return () => {
-      mounted -= 1
-      closed.current = true
-      if (timer.current) clearTimeout(timer.current)
-      socket.current?.close()
-      liveStore.setConnected(false)
-    }
+    return () => release(onAuthFailure)
   }, [onAuthFailure, queryClient])
 }

frontend/src/components/Health/LiveIndicator.tsx

@@ -0,0 +1,73 @@
+import { Activity } from "lucide-react"
+import { AnimatePresence, motion } from "motion/react"
+import { useEffect, useState, useSyncExternalStore } from "react"
+
+import { useLiveConnection } from "@/components/Flow/liveStore"
+import { connectionStore } from "@/lib/connectionStore"
+import { fadeIn } from "@/lib/motion"
+
+/**
+ * How long the socket may be down before it is worth saying so.
+ *
+ * The first reconnect attempt is a second away, so a dropped frame is back
+ * before this elapses and nothing appears. Long enough to cover the opening
+ * handshake too: a page that has just loaded is connecting, not disconnected.
+ */
+const GRACE = 3000
+
+/**
+ * Says when live values have stopped arriving.
+ *
+ * Everything else on Home is polled over HTTP, so a dead socket leaves the page
+ * looking perfectly current while nothing on it moves any more — the neurons
+ * stop pulsing and the values stop changing, with nothing to say why. Nothing
+ * at all while the socket is up: a page that works needs no chip saying so.
+ *
+ * Quiet too while `ConnectionBanner` is up, since an installation that cannot
+ * be reached has no socket either and one explanation of that is enough. Named
+ * in words rather than coloured, and deliberately not terracotta: the brain
+ * graph above it already owns that accent for a flow that cannot run.
+ */
+export function LiveIndicator() {
+  const connected = useLiveConnection()
+  const { offline } = useSyncExternalStore(
+    connectionStore.subscribe,
+    connectionStore.snapshot,
+    connectionStore.snapshot,
+  )
+  const [down, setDown] = useState(false)
+
+  useEffect(() => {
+    if (connected) {
+      setDown(false)
+      return
+    }
+    const timer = setTimeout(() => setDown(true), GRACE)
+    return () => clearTimeout(timer)
+  }, [connected])
+
+  return (
+    <AnimatePresence>
+      {down && !offline ? (
+        <motion.div
+          variants={fadeIn}
+          initial="hidden"
+          animate="visible"
+          exit="hidden"
+          role="status"
+          aria-live="polite"
+          data-testid="live-indicator"
+          // Wraps rather than widens: three phrases in a row do not fit a
+          // phone. See DESIGN-GUIDELINES.md → Responsive.
+          className="flex flex-wrap items-center gap-x-2 gap-y-1 rounded-lg border border-border px-4 py-2 text-sm shadow-e1"
+        >
+          <Activity className="size-4 shrink-0 text-muted-foreground" />
+          <span className="font-medium">Live updates disconnected</span>
+          <span className="text-muted-foreground">
+            values and activity are frozen — reconnecting…
+          </span>
+        </motion.div>
+      ) : null}
+    </AnimatePresence>
+  )
+}

frontend/src/routes/_layout/index.tsx

@@ -9,6 +9,7 @@ import { BrainView } from "@/components/Flow/BrainView"
 import { flowKeys, flowsQueryOptions } from "@/components/Flow/queries"
 import { HealthActivity } from "@/components/Health/HealthActivity"
 import { HealthOverview } from "@/components/Health/HealthOverview"
+import { LiveIndicator } from "@/components/Health/LiveIndicator"
 import { Badge } from "@/components/ui/badge"
 import { Card } from "@/components/ui/card"
 import { Skeleton } from "@/components/ui/skeleton"
@@ -116,6 +117,10 @@ function Dashboard() {
     // Every section here is free to shrink instead. See DESIGN-GUIDELINES.md
     // → Responsive.
     <div className="grid gap-6 [&>*]:min-w-0">
+      {/* Nothing at all while the socket is up. Everything below is polled, so
+          a dead socket would otherwise look like an engine that went quiet. */}
+      <LiveIndicator />
+
       {/* Nothing wired up yet means nothing to draw, and the band would still
           hold a screenful of empty space above the flows card. Node count
           rather than flow count: a flow made a minute ago has none. */}