LiveEdge pulse gaps: reset timer survives other producers, snapshots do not flash, flow: endpoints pulse as their node

frontend/src/components/Flow/EndpointNode.tsx

@@ -4,7 +4,7 @@ import { memo, useEffect, useMemo, useRef, useState } from "react"
 
 import { useIsMobile } from "@/hooks/useMobile"
 import { cn } from "@/lib/utils"
-import type { EndpointNodeData } from "./endpoints"
+import { type EndpointNodeData, publisherOf } from "./endpoints"
 import { handleOffset, nodeHeight } from "./layout"
 import { useLiveValues } from "./liveStore"
 
@@ -57,7 +57,10 @@ function EndpointNodeComponent({ id, data, selected }: NodeProps) {
       seen.current.set(name, live.ts)
       if (first || live.prior) continue
       // Published by this label, by the rule its own edges pulse on (LiveEdge).
-      if (provides.includes(name) && (!live.source || live.source.id === id)) {
+      if (
+        provides.includes(name) &&
+        (!live.source || live.source.id === publisherOf(id))
+      ) {
         dir = "in"
       } else if (requires.includes(name)) {
         dir ??= "out"

frontend/src/components/Flow/LiveEdge.tsx

@@ -89,6 +89,11 @@ function LiveEdgeComponent({
   const zoom = useStore((state) => state.transform[2])
   const [pulsing, setPulsing] = useState(false)
   const lastTs = useRef<number | null>(null)
+  // Owned by the pulse rather than by the effect: a value from another
+  // producer re-runs the effect too, and must not cancel a reset it does not
+  // replace, or the edge stays lit and the next pulse cannot restart.
+  const timer = useRef<ReturnType<typeof setTimeout>>(undefined)
+  useEffect(() => () => clearTimeout(timer.current), [])
 
   // Every rank sits further along the flow than the one before it, so an edge
   // whose target has ended up behind its source is one closing a cycle.
@@ -113,13 +118,15 @@ function LiveEdgeComponent({
   useEffect(() => {
     if (!live?.ts || live.ts === lastTs.current) return
     lastTs.current = live.ts
+    // A snapshot catching the page up is not a value passing now.
+    if (live.prior) return
     const from = live.source
     // No source at all is an older engine; pulse rather than go silent.
     if (from && from.id !== producerId) return
     setPulsing(true)
-    const timer = setTimeout(() => setPulsing(false), duration.pulse * 1000)
-    return () => clearTimeout(timer)
-  }, [live?.ts, live?.source, producerId])
+    clearTimeout(timer.current)
+    timer.current = setTimeout(() => setPulsing(false), duration.pulse * 1000)
+  }, [live?.ts, live?.source, live?.prior, producerId])
 
   return (
     <>

frontend/src/components/Flow/endpoints.ts

@@ -37,6 +37,19 @@ export function isEndpointNode(node: { id: string }): boolean {
   return PREFIXES.some((prefix) => node.id.startsWith(prefix))
 }
 
+/**
+ * Whom a value published through this endpoint is credited to.
+ *
+ * Every other kind publishes under its endpoint id, but a node in another
+ * flow publishes as itself — `other.node`, while its label here is
+ * `flow:other.node`.
+ */
+export function publisherOf(endpointId: string): string {
+  return endpointId.startsWith("flow:")
+    ? endpointId.slice("flow:".length)
+    : endpointId
+}
+
 /**
  * The flow's own boundary, as endpoints.
  *
@@ -167,7 +180,7 @@ export function deriveEndpoints(
           data: {
             message,
             flow,
-            producerId: endpoint.id,
+            producerId: publisherOf(endpoint.id),
           } satisfies FlowEdgeData,
         })
       }