From 71e85daea310de8ab6ce379dd3550d8b690dba6c Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Sun, 30 Aug 2026 20:20:07 -0300 Subject: [PATCH] =?UTF-8?q?fix(dashboard):=20orchestration=20review=20fixe?= =?UTF-8?q?s=20=E2=80=94=20offline=20source=20node,=20orphaned-runner=20ta?= =?UTF-8?q?sks,=20true=20overview=20counts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the final whole-branch review findings on the Orchestration Canvas model layer before merge: - mergeSnapshot: materialize the placeholder source node when a source status reports offline:true (not only !ok), so PR-2's SourceNode can actually reach the sublabel === "offline" render arm for an offline Conductor. - fromConductor: a running task can only be "absorbed" into a runner's ActivityNode when that runner id is present in snap.runners — a task whose runner deregistered mid-flight now emits as a normal work node instead of vanishing. - OrchNode gains an additive droppedByState field; mergeSnapshot stamps it on overflow nodes with the per-state counts of the work nodes the 40-node cap dropped, and overviewProjection folds it into counts (never columns) so the overview counters always show true totals per operator ruling. - Fixes the pre-first-poll 1970 generatedAt (lazy Date.now() seed for polledAt) and nulls debounceRef after clearTimeout in the poll effect's cleanup to avoid a StrictMode remount permanently swallowing WS triggers. - Prettier formatting pass on the touched model/hook/test files. --- .../hooks/useOrchestrationSnapshot.ts | 33 ++++++++-- .../orchestration/model/fromConductor.ts | 9 ++- .../orchestration/model/mergeSnapshot.ts | 10 ++- .../model/orchestrationToFlow.ts | 28 ++++++-- .../orchestration/model/orchestrationTypes.ts | 5 ++ .../orchestration/model/overviewProjection.ts | 10 +++ tests/unit/ui/orchestrationModel.test.ts | 52 +++++++++++++++ tests/unit/ui/orchestrationToFlow.test.ts | 19 ++++-- tests/unit/ui/overviewProjection.test.ts | 20 ++++++ .../unit/ui/useOrchestrationSnapshot.test.tsx | 65 +++++++++++++++---- 10 files changed, 222 insertions(+), 29 deletions(-) diff --git a/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts b/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts index 759d6ba831..51a106fa9e 100644 --- a/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts +++ b/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts @@ -19,7 +19,11 @@ interface Raw { a2a: A2ATask[]; conductor: FleetSnapshot; } -const EMPTY_RAW: Raw = { cloudAgent: [], a2a: [], conductor: { offline: true, runners: [], tasks: [] } }; +const EMPTY_RAW: Raw = { + cloudAgent: [], + a2a: [], + conductor: { offline: true, runners: [], tasks: [] }, +}; async function fetchJson(url: string, signal: AbortSignal): Promise { const res = await fetch(url, { signal, cache: "no-store" }); @@ -36,7 +40,11 @@ export function useOrchestrationSnapshot() { // without either violation. const [raw, setRaw] = useState(EMPTY_RAW); const [statuses, setStatuses] = useState([]); - const [polledAt, setPolledAt] = useState(0); + // Lazy initializer (not a literal 0) so the pre-first-poll render already has a + // real timestamp — with `0` the very first `mergeSnapshot` call stamped + // `generatedAt` as the 1970 epoch. Safe: with EMPTY_RAW there is nothing to + // staleness-filter at mount, so seeding `Date.now()` here changes no behavior. + const [polledAt, setPolledAt] = useState(() => Date.now()); const [isLoading, setIsLoading] = useState(true); const [showCompleted, setShowCompleted] = useState(false); const debounceRef = useRef | null>(null); @@ -60,12 +68,24 @@ export function useOrchestrationSnapshot() { const nowIso = new Date(nowMs).toISOString(); const next: SourceStatus[] = []; if (ca.status === "fulfilled") next.push({ source: "cloud-agent", ok: true }); - else next.push({ source: "cloud-agent", ok: false, error: String(ca.reason), staleSince: nowIso }); + else + next.push({ + source: "cloud-agent", + ok: false, + error: String(ca.reason), + staleSince: nowIso, + }); if (a2a.status === "fulfilled") next.push({ source: "a2a", ok: true }); else next.push({ source: "a2a", ok: false, error: String(a2a.reason), staleSince: nowIso }); if (cond.status === "fulfilled") { next.push({ source: "conductor", ok: true, offline: cond.value.offline }); - } else next.push({ source: "conductor", ok: false, error: String(cond.reason), staleSince: nowIso }); + } else + next.push({ + source: "conductor", + ok: false, + error: String(cond.reason), + staleSince: nowIso, + }); // Failed sources keep the previously stored slice — only overwrite what // actually resolved this round ("last good data" contract from the brief). @@ -85,7 +105,10 @@ export function useOrchestrationSnapshot() { return () => { clearInterval(id); controller.abort(); - if (debounceRef.current) clearTimeout(debounceRef.current); + if (debounceRef.current) { + clearTimeout(debounceRef.current); + debounceRef.current = null; + } }; }, []); diff --git a/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts b/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts index 0a21d291b6..b629e2621a 100644 --- a/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts +++ b/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts @@ -43,9 +43,16 @@ export function fromConductor(snap: FleetSnapshot): { nodes: OrchNode[]; edges: counts[s] = (counts[s] ?? 0) + 1; }; + // Only tasks whose runner actually exists in snap.runners can be "absorbed" + // into that runner's ActivityNode below — a running task pointing at a runner + // id that has since deregistered must fall through to the normal work-node + // loop instead of being silently skipped as "already an activity". + const runnerIds = new Set(snap.runners.map((r) => r.id)); const activeByRunner = new Map(); for (const t of snap.tasks) { - if (t.runner && mapHubStatus(t.status) === "running") activeByRunner.set(t.runner, t); + if (t.runner && runnerIds.has(t.runner) && mapHubStatus(t.status) === "running") { + activeByRunner.set(t.runner, t); + } } for (const r of snap.runners) { diff --git a/src/app/(dashboard)/dashboard/orchestration/model/mergeSnapshot.ts b/src/app/(dashboard)/dashboard/orchestration/model/mergeSnapshot.ts index 2b01c62ca8..993bced1ae 100644 --- a/src/app/(dashboard)/dashboard/orchestration/model/mergeSnapshot.ts +++ b/src/app/(dashboard)/dashboard/orchestration/model/mergeSnapshot.ts @@ -108,6 +108,10 @@ export function mergeSnapshot( source, label: `+${excess.length} more`, counts, + // Additive: lets overviewProjection fold true per-state totals into its + // counters even though these nodes no longer render on the canvas + // (operator ruling — spec governs, counters must show TRUE totals). + droppedByState: counts, }); } nodes = nodes.filter((n) => !dropped.has(n.id)).concat(overflowNodes); @@ -127,7 +131,11 @@ export function mergeSnapshot( const root: OrchNode = { id: "orchestrator", kind: "orchestrator", label: "OmniRoute" }; const sourceIds = new Set(nodes.filter((n) => n.kind === "source").map((n) => n.id)); for (const s of sources) { - if (!s.ok && !sourceIds.has(`source:${s.source}`) && s.source !== "routing") { + // `!s.ok` covers hard failures; `s.offline` also materializes a placeholder + // for a source that reported ok:true but offline:true (e.g. Conductor with + // no hub configured) — otherwise that source never gets a SourceNode at all + // and its "offline" sublabel can never render. + if ((!s.ok || s.offline) && !sourceIds.has(`source:${s.source}`) && s.source !== "routing") { nodes.push({ id: `source:${s.source}`, kind: "source", diff --git a/src/app/(dashboard)/dashboard/orchestration/model/orchestrationToFlow.ts b/src/app/(dashboard)/dashboard/orchestration/model/orchestrationToFlow.ts index 9c5ad1889d..1c3971cd87 100644 --- a/src/app/(dashboard)/dashboard/orchestration/model/orchestrationToFlow.ts +++ b/src/app/(dashboard)/dashboard/orchestration/model/orchestrationToFlow.ts @@ -4,11 +4,19 @@ import { edgeStyle } from "@/shared/components/flow/edgeStyles"; import type { OrchNodeKind, OrchSnapshot } from "./orchestrationTypes"; const LAYER_Y: Record = { - orchestrator: 0, source: 150, work: 320, overflow: 320, activity: 470, + orchestrator: 0, + source: 150, + work: 320, + overflow: 320, + activity: 470, }; const X_GAP = 260; -export function orchestrationToFlow(snap: OrchSnapshot): { nodes: Node[]; edges: Edge[]; fitKey: string } { +export function orchestrationToFlow(snap: OrchSnapshot): { + nodes: Node[]; + edges: Edge[]; + fitKey: string; +} { const byLayer = new Map(); for (const n of [...snap.nodes].sort((a, b) => a.id.localeCompare(b.id))) { const y = LAYER_Y[n.kind]; @@ -24,17 +32,27 @@ export function orchestrationToFlow(snap: OrchSnapshot): { nodes: Node[]; edges: const stateOf = new Map(snap.nodes.map((n) => [n.id, n.state])); const nodes: Node[] = snap.nodes.map((n) => ({ - id: n.id, type: n.kind, position: pos.get(n.id)!, data: n as unknown as Record, + id: n.id, + type: n.kind, + position: pos.get(n.id)!, + data: n as unknown as Record, })); const edges: Edge[] = snap.edges.map((e) => { const target = stateOf.get(e.to); const style = edgeStyle(e.active, false, target === "failed", target === "succeeded"); return { - id: e.id, source: e.from, target: e.to, animated: e.active, + id: e.id, + source: e.from, + target: e.to, + animated: e.active, style: e.kind === "mirror" ? { ...style, strokeDasharray: "6 4" } : style, }; }); - const fitKey = snap.nodes.filter((n) => n.kind === "work").map((n) => n.id).sort().join("|"); + const fitKey = snap.nodes + .filter((n) => n.kind === "work") + .map((n) => n.id) + .sort() + .join("|"); return { nodes, edges, fitKey }; } diff --git a/src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts b/src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts index 362ecb13d8..8c8426a572 100644 --- a/src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts +++ b/src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts @@ -21,6 +21,11 @@ export interface OrchNode { endedAt?: string; cost?: number; counts?: Partial>; + // Overflow nodes only: per-state counts of the work nodes folded into this + // overflow node when the MAX_WORK_NODES cap engages. overviewProjection folds + // this into its `counts` totals (never into `columns`) so operators still see + // TRUE totals even when the canvas caps the rendered node count. + droppedByState?: Partial>; mirrorOf?: string; raw?: unknown; } diff --git a/src/app/(dashboard)/dashboard/orchestration/model/overviewProjection.ts b/src/app/(dashboard)/dashboard/orchestration/model/overviewProjection.ts index abfb606b15..b5566b4cf2 100644 --- a/src/app/(dashboard)/dashboard/orchestration/model/overviewProjection.ts +++ b/src/app/(dashboard)/dashboard/orchestration/model/overviewProjection.ts @@ -26,6 +26,16 @@ export function overviewProjection(snap: OrchSnapshot, comboActive: number): Ove }; for (const n of snap.nodes) { + // Overflow nodes (MAX_WORK_NODES cap) fold their dropped work nodes' true + // per-state counts into `counts` only — never into `columns`, since those + // nodes are not rendered on the canvas. Counters must show TRUE totals + // even when the canvas caps the rendered node count (operator ruling). + if (n.kind === "overflow" && n.droppedByState) { + for (const s of ORCH_STATES) { + counts[s] += n.droppedByState[s] ?? 0; + } + continue; + } if (n.kind !== "work" || !n.state) continue; counts[n.state] += 1; if (n.state === "queued" || n.state === "running" || n.state === "waiting_approval") { diff --git a/tests/unit/ui/orchestrationModel.test.ts b/tests/unit/ui/orchestrationModel.test.ts index 7a0d847756..08a728e29a 100644 --- a/tests/unit/ui/orchestrationModel.test.ts +++ b/tests/unit/ui/orchestrationModel.test.ts @@ -205,6 +205,27 @@ describe("fromConductor", () => { assert.equal(w?.state, "failed"); assert.match(w?.sublabel ?? "", /vaporized/); }); + it("running task whose runner is not present in snap.runners is emitted as a work node, not swallowed", () => { + const snap: FleetSnapshot = { + offline: false, + runners: [], + tasks: [ + { + ...baseSnap.tasks[0], + id: "ct-orphan", + status: "running", + runner: "ghost-runner", + }, + ], + }; + const { nodes, edges } = fromConductor(snap); + const w = nodes.find((n) => n.id === "conductor:task:ct-orphan"); + assert.equal(w?.kind, "work"); + assert.equal(w?.state, "running"); + assert.ok( + edges.some((e) => e.from === "source:conductor" && e.to === "conductor:task:ct-orphan") + ); + }); }); const OK_SOURCES = [ @@ -291,6 +312,37 @@ describe("mergeSnapshot", () => { const mergedNode = snap.nodes.find((n) => n.id === "conductor:task:ct1"); assert.equal(mergedNode?.mirrorOf, "a2a:am2"); }); + it("materializes an offline placeholder source node for a source reporting offline:true even when ok:true", () => { + const src = [{ source: "conductor" as const, ok: true, offline: true }]; + const snap = mergeSnapshot({ cloudAgent: empty, a2a: empty, conductor: empty }, src, { + now: NOW, + }); + const node = snap.nodes.find((n) => n.id === "source:conductor"); + assert.ok(node, "offline placeholder source:conductor node expected"); + assert.equal(node?.sublabel, "offline"); + assert.ok( + snap.edges.some( + (e) => e.from === "orchestrator" && e.to === "source:conductor" && e.kind === "owns" + ) + ); + }); + it("overflow node carries droppedByState with the per-state counts of dropped work nodes", () => { + const many = Array.from({ length: MAX_WORK_NODES + 5 }, (_, i) => + caTask({ + id: `ov${i}`, + status: i < MAX_WORK_NODES ? "running" : "failed", + updatedAt: new Date(NOW - i * 1000).toISOString(), + }) + ); + const snap = mergeSnapshot( + { cloudAgent: fromCloudAgent(many), a2a: empty, conductor: empty }, + OK_SOURCES, + { now: NOW } + ); + const overflow = snap.nodes.find((n) => n.id === "overflow:cloud-agent"); + assert.ok(overflow, "overflow node expected"); + assert.deepEqual(overflow?.droppedByState, { failed: 5 }); + }); it("drops a stale terminal Conductor task older than STALE_COMPLETED_MS unless showCompleted", () => { const staleSnap: FleetSnapshot = { ...baseSnap, diff --git a/tests/unit/ui/orchestrationToFlow.test.ts b/tests/unit/ui/orchestrationToFlow.test.ts index f44bb054a0..e3796adcb5 100644 --- a/tests/unit/ui/orchestrationToFlow.test.ts +++ b/tests/unit/ui/orchestrationToFlow.test.ts @@ -16,14 +16,18 @@ const snap: OrchSnapshot = { { id: "e2", from: "source:a2a", to: "a2a:t1", kind: "owns", active: true }, { id: "e3", from: "source:a2a", to: "a2a:t2", kind: "owns", active: false }, ], - sources: [], generatedAt: "2026-08-30T12:00:00Z", + sources: [], + generatedAt: "2026-08-30T12:00:00Z", }; describe("orchestrationToFlow", () => { it("puts each kind on its own Y layer and is deterministic", () => { const a = orchestrationToFlow(snap); const b = orchestrationToFlow(snap); - assert.deepEqual(a.nodes.map((n) => n.position), b.nodes.map((n) => n.position)); + assert.deepEqual( + a.nodes.map((n) => n.position), + b.nodes.map((n) => n.position) + ); const ys = new Map(a.nodes.map((n) => [n.id, n.position.y])); assert.equal(ys.get("orchestrator"), 0); assert.equal(ys.get("source:a2a"), 150); @@ -37,9 +41,16 @@ describe("orchestrationToFlow", () => { }); it("fitKey only tracks the set of work ids", () => { const k1 = orchestrationToFlow(snap).fitKey; - const stateChanged = { ...snap, nodes: snap.nodes.map((n) => n.id === "a2a:t1" ? { ...n, state: "succeeded" as const } : n) }; + const stateChanged = { + ...snap, + nodes: snap.nodes.map((n) => (n.id === "a2a:t1" ? { ...n, state: "succeeded" as const } : n)), + }; assert.equal(orchestrationToFlow(stateChanged).fitKey, k1); - const nodeRemoved = { ...snap, nodes: snap.nodes.filter((n) => n.id !== "a2a:t2"), edges: snap.edges.filter((e) => e.to !== "a2a:t2") }; + const nodeRemoved = { + ...snap, + nodes: snap.nodes.filter((n) => n.id !== "a2a:t2"), + edges: snap.edges.filter((e) => e.to !== "a2a:t2"), + }; assert.notEqual(orchestrationToFlow(nodeRemoved).fitKey, k1); }); }); diff --git a/tests/unit/ui/overviewProjection.test.ts b/tests/unit/ui/overviewProjection.test.ts index bfbb3e5e57..56c133cc96 100644 --- a/tests/unit/ui/overviewProjection.test.ts +++ b/tests/unit/ui/overviewProjection.test.ts @@ -44,4 +44,24 @@ describe("overviewProjection", () => { assert.equal(columns.done[0].id, "a2a:3"); assert.equal(columns.running.length, 1); }); + it("folds an overflow node's droppedByState into counts but not into columns", () => { + const snapWithOverflow: OrchSnapshot = { + ...snap, + nodes: [ + ...snap.nodes, + { + id: "overflow:cloud-agent", + kind: "overflow", + source: "cloud-agent", + label: "+7 more", + droppedByState: { running: 5, failed: 2 }, + }, + ], + }; + const { counts, columns } = overviewProjection(snapWithOverflow, 0); + assert.equal(counts.running, 1 + 5); + assert.equal(counts.failed, 1 + 2); + assert.equal(columns.running.length, 1); + assert.equal(columns.done.length, 1); + }); }); diff --git a/tests/unit/ui/useOrchestrationSnapshot.test.tsx b/tests/unit/ui/useOrchestrationSnapshot.test.tsx index f79f95f624..d90e1d8a2a 100644 --- a/tests/unit/ui/useOrchestrationSnapshot.test.tsx +++ b/tests/unit/ui/useOrchestrationSnapshot.test.tsx @@ -14,12 +14,17 @@ vi.mock("@/hooks/useLiveDashboard", () => ({ import { useOrchestrationSnapshot } from "@/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot"; -function HookProbe({ onRender }: { onRender: (v: ReturnType) => void }) { +function HookProbe({ + onRender, +}: { + onRender: (v: ReturnType) => void; +}) { onRender(useOrchestrationSnapshot()); return null; } -const okJson = (body: unknown) => Promise.resolve({ ok: true, json: () => Promise.resolve(body) } as Response); +const okJson = (body: unknown) => + Promise.resolve({ ok: true, json: () => Promise.resolve(body) } as Response); describe("useOrchestrationSnapshot", () => { let container: HTMLDivElement; @@ -39,25 +44,49 @@ describe("useOrchestrationSnapshot", () => { it("polls the three endpoints and builds a snapshot; a failed source keeps the last good data", async () => { let latest: ReturnType | null = null; - const task = { id: "t1", providerId: "devin", status: "running", prompt: "p", source: { repoName: "r", repoUrl: "https://x" }, options: {}, activities: [], createdAt: "2026-08-30T10:00:00Z", updatedAt: "2026-08-30T10:00:00Z" }; + const task = { + id: "t1", + providerId: "devin", + status: "running", + prompt: "p", + source: { repoName: "r", repoUrl: "https://x" }, + options: {}, + activities: [], + createdAt: "2026-08-30T10:00:00Z", + updatedAt: "2026-08-30T10:00:00Z", + }; const fetchMock = vi.fn((url: string) => { if (url.startsWith("/api/v1/agents/tasks")) return okJson({ data: [task] }); - if (url.startsWith("/api/a2a/tasks")) return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + if (url.startsWith("/api/a2a/tasks")) + return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); return okJson({ offline: false, runners: [], tasks: [] }); }); vi.stubGlobal("fetch", fetchMock); - await act(async () => { root.render( { latest = v; }} />); }); - await act(async () => { await vi.advanceTimersByTimeAsync(10); }); + await act(async () => { + root.render( + { + latest = v; + }} + /> + ); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(10); + }); expect(latest!.snapshot.nodes.some((n) => n.id === "cloud-agent:t1")).toBe(true); // Second tick: cloud agent fails — node must survive from the last good photo, source marked stale. fetchMock.mockImplementation((url: string) => { if (url.startsWith("/api/v1/agents/tasks")) return Promise.reject(new Error("boom")); - if (url.startsWith("/api/a2a/tasks")) return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + if (url.startsWith("/api/a2a/tasks")) + return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); return okJson({ offline: false, runners: [], tasks: [] }); }); - await act(async () => { await vi.advanceTimersByTimeAsync(5_100); }); + await act(async () => { + await vi.advanceTimersByTimeAsync(5_100); + }); expect(latest!.snapshot.nodes.some((n) => n.id === "cloud-agent:t1")).toBe(true); const st = latest!.snapshot.sources.find((s) => s.source === "cloud-agent"); expect(st?.ok).toBe(false); @@ -66,16 +95,26 @@ describe("useOrchestrationSnapshot", () => { it("a requests-channel WS event triggers a debounced immediate refetch", async () => { const fetchMock = vi.fn((url: string) => { if (url.startsWith("/api/v1/agents/tasks")) return okJson({ data: [] }); - if (url.startsWith("/api/a2a/tasks")) return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + if (url.startsWith("/api/a2a/tasks")) + return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); return okJson({ offline: false, runners: [], tasks: [] }); }); vi.stubGlobal("fetch", fetchMock); - await act(async () => { root.render( {}} />); }); - await act(async () => { await vi.advanceTimersByTimeAsync(10); }); + await act(async () => { + root.render( {}} />); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(10); + }); const callsAfterMount = fetchMock.mock.calls.length; - act(() => { capturedOnEvent?.({ channel: "requests" }); capturedOnEvent?.({ channel: "requests" }); }); - await act(async () => { await vi.advanceTimersByTimeAsync(1_100); }); + act(() => { + capturedOnEvent?.({ channel: "requests" }); + capturedOnEvent?.({ channel: "requests" }); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(1_100); + }); // Two burst events → exactly ONE extra round of 3 fetches (debounce), not two. expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3); });