diff --git a/frontend/src/components/Flow/EndpointNode.tsx b/frontend/src/components/Flow/EndpointNode.tsx index 44d8bdb..e76dbb2 100644 --- a/frontend/src/components/Flow/EndpointNode.tsx +++ b/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" diff --git a/frontend/src/components/Flow/LiveEdge.tsx b/frontend/src/components/Flow/LiveEdge.tsx index 2ebd5b7..3a4940f 100644 --- a/frontend/src/components/Flow/LiveEdge.tsx +++ b/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(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>(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 ( <> diff --git a/frontend/src/components/Flow/endpoints.ts b/frontend/src/components/Flow/endpoints.ts index 4ed68fb..d1abb3c 100644 --- a/frontend/src/components/Flow/endpoints.ts +++ b/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, }) }