Compare commits

...

11 Commits

Author SHA1 Message Date
Markus Hartung
da94bbb7a4 refactor(dashboard): split orchestration model into smaller helpers for new-code complexity ratchets 2026-08-30 20:47:10 -03:00
Markus Hartung
71e85daea3 fix(dashboard): orchestration review fixes — offline source node, orphaned-runner tasks, true overview counts
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.
2026-08-30 20:20:07 -03:00
Markus Hartung
88632788ec feat(dashboard): useOrchestrationSnapshot — allSettled polling + WS requests trigger 2026-08-30 18:59:32 -03:00
Markus Hartung
c5763c3ed8 feat(dashboard): overviewProjection — counters + kanban columns 2026-08-30 18:34:51 -03:00
Markus Hartung
aa651a9ef8 feat(dashboard): orchestrationToFlow — deterministic 3-layer layout + stable fitKey 2026-08-30 18:26:38 -03:00
Markus Hartung
82cf2458c6 fix(dashboard): mergeSnapshot input purity + conductor endedAt for staleness (review r1) 2026-08-30 18:16:10 -03:00
Markus Hartung
3719f97592 feat(dashboard): mergeSnapshot — root, conductor/a2a dedupe, staleness filter, overflow cap 2026-08-30 17:58:34 -03:00
Markus Hartung
aafcc0b462 feat(dashboard): fromConductor orchestration mapper 2026-08-30 17:47:26 -03:00
Markus Hartung
b3537be551 feat(dashboard): fromA2A orchestration mapper 2026-08-30 17:08:26 -03:00
Markus Hartung
f4a5c4e7ff feat(dashboard): fromCloudAgent orchestration mapper 2026-08-30 16:57:27 -03:00
Markus Hartung
3d429b0c59 feat(dashboard): orchestration canvas domain types + approval status color 2026-08-30 16:50:53 -03:00
13 changed files with 1455 additions and 0 deletions

View File

@@ -0,0 +1,157 @@
"use client";
/** Polls the 3 agent sources (allSettled), listens to the `requests` WS channel as a refetch trigger. */
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { useLiveDashboard } from "@/hooks/useLiveDashboard";
import type { CloudAgentTask } from "@/lib/cloudAgent/types";
import type { A2ATask } from "@/lib/a2a/taskManager";
import type { FleetSnapshot } from "@/lib/conductor/hubProxy";
import { fromCloudAgent } from "../model/fromCloudAgent";
import { fromA2A } from "../model/fromA2A";
import { fromConductor } from "../model/fromConductor";
import { mergeSnapshot } from "../model/mergeSnapshot";
import type { OrchSnapshot, SourceStatus } from "../model/orchestrationTypes";
export const POLL_MS = 5_000;
export const WS_REFETCH_DEBOUNCE_MS = 1_000;
interface Raw {
cloudAgent: CloudAgentTask[];
a2a: A2ATask[];
conductor: FleetSnapshot;
}
const EMPTY_RAW: Raw = {
cloudAgent: [],
a2a: [],
conductor: { offline: true, runners: [], tasks: [] },
};
async function fetchJson<T>(url: string, signal: AbortSignal): Promise<T> {
const res = await fetch(url, { signal, cache: "no-store" });
if (!res.ok) throw new Error(`HTTP ${res.status}`);
return res.json() as Promise<T>;
}
/** Builds the 3-source status list from a `Promise.allSettled` triple. */
function buildSourceStatuses(
ca: PromiseSettledResult<{ data: CloudAgentTask[] }>,
a2a: PromiseSettledResult<{ tasks: A2ATask[] }>,
cond: PromiseSettledResult<FleetSnapshot>,
nowIso: string
): SourceStatus[] {
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,
});
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,
});
return next;
}
export function useOrchestrationSnapshot() {
// `raw` and `polledAt` are React state (not refs) so the merge below reads them
// during render like any other state — a ref read during render trips the
// `react-hooks/refs` lint rule, and computing `Date.now()` inline in the memo
// factory trips `react-hooks/purity`. Sampling `Date.now()` once per poll (inside
// the effect, not during render) keeps `mergeSnapshot`'s staleness math correct
// without either violation.
const [raw, setRaw] = useState<Raw>(EMPTY_RAW);
const [statuses, setStatuses] = useState<SourceStatus[]>([]);
// 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<number>(() => Date.now());
const [isLoading, setIsLoading] = useState(true);
const [showCompleted, setShowCompleted] = useState(false);
const debounceRef = useRef<ReturnType<typeof setTimeout> | null>(null);
// Populated by the mount effect below; lets `refetch()` (and the WS debounce
// trigger) reach the same poll loop without hoisting it out of the effect —
// hoisting to a top-level `useCallback` invoked from the effect body trips
// `react-hooks/set-state-in-effect`.
const pollRef = useRef<() => void>(() => {});
useEffect(() => {
const controller = new AbortController();
const poll = async () => {
const [ca, a2a, cond] = await Promise.allSettled([
fetchJson<{ data: CloudAgentTask[] }>("/api/v1/agents/tasks?limit=100", controller.signal),
fetchJson<{ tasks: A2ATask[] }>("/api/a2a/tasks?limit=200", controller.signal),
fetchJson<FleetSnapshot>("/api/conductor/fleet", controller.signal),
]);
if (controller.signal.aborted) return;
const nowMs = Date.now();
const nowIso = new Date(nowMs).toISOString();
const next = buildSourceStatuses(ca, a2a, cond, nowIso);
// Failed sources keep the previously stored slice — only overwrite what
// actually resolved this round ("last good data" contract from the brief).
setRaw((prev) => ({
cloudAgent: ca.status === "fulfilled" ? ca.value.data : prev.cloudAgent,
a2a: a2a.status === "fulfilled" ? a2a.value.tasks : prev.a2a,
conductor: cond.status === "fulfilled" ? cond.value : prev.conductor,
}));
setStatuses(next);
setPolledAt(nowMs);
setIsLoading(false);
};
pollRef.current = () => void poll();
void poll();
const id = setInterval(() => void poll(), POLL_MS);
return () => {
clearInterval(id);
controller.abort();
if (debounceRef.current) {
clearTimeout(debounceRef.current);
debounceRef.current = null;
}
};
}, []);
const refetch = useCallback(() => {
pollRef.current();
}, []);
useLiveDashboard({
channels: ["requests"],
onEvent: (payload) => {
if (payload.channel !== "requests") return;
if (debounceRef.current) return; // debounce burst → one refetch
debounceRef.current = setTimeout(() => {
debounceRef.current = null;
refetch();
}, WS_REFETCH_DEBOUNCE_MS);
},
});
const snapshot: OrchSnapshot = useMemo(
() =>
mergeSnapshot(
{
cloudAgent: fromCloudAgent(raw.cloudAgent),
a2a: fromA2A(raw.a2a),
conductor: fromConductor(raw.conductor),
},
statuses,
{ now: polledAt, showCompleted }
),
[raw, statuses, showCompleted, polledAt]
);
return { snapshot, isLoading, showCompleted, setShowCompleted, refetch };
}

View File

@@ -0,0 +1,53 @@
/** A2A tasks → unified orchestration nodes. Pure. */
import type { A2ATask } from "@/lib/a2a/taskManager";
import type { OrchEdge, OrchNode, OrchState } from "./orchestrationTypes";
const STATE_MAP: Record<string, OrchState> = {
submitted: "queued",
working: "running",
completed: "succeeded",
failed: "failed",
cancelled: "cancelled",
};
const TERMINAL: ReadonlySet<OrchState> = new Set(["succeeded", "failed", "cancelled"]);
function truncate(s: string, n = 60): string {
return s.length > n ? `${s.slice(0, n - 1)}` : s;
}
export function fromA2A(tasks: A2ATask[]): { nodes: OrchNode[]; edges: OrchEdge[] } {
if (tasks.length === 0) return { nodes: [], edges: [] };
const nodes: OrchNode[] = [];
const edges: OrchEdge[] = [];
const counts: Partial<Record<OrchState, number>> = {};
for (const t of tasks) {
const mapped = STATE_MAP[t.state];
const state: OrchState = mapped ?? "failed";
counts[state] = (counts[state] ?? 0) + 1;
const id = `a2a:${t.id}`;
const firstUser = t.input.messages.find((m) => m.role === "user")?.content ?? "";
nodes.push({
id,
kind: "work",
source: "a2a",
state,
label: t.skill,
sublabel: mapped ? truncate(firstUser) : `unknown state: ${String(t.state)}`,
startedAt: t.createdAt,
updatedAt: t.updatedAt,
endedAt: TERMINAL.has(state) ? t.updatedAt : undefined,
raw: t,
});
edges.push({
id: `e:source:a2a→${id}`,
from: "source:a2a",
to: id,
kind: "owns",
active: state === "running",
});
}
nodes.unshift({ id: "source:a2a", kind: "source", source: "a2a", label: "A2A", counts });
return { nodes, edges };
}

View File

@@ -0,0 +1,75 @@
/** Cloud Agent tasks → unified orchestration nodes. Pure. */
import type { CloudAgentTask } from "@/lib/cloudAgent/types";
import type { OrchEdge, OrchNode, OrchState } from "./orchestrationTypes";
const STATE_MAP: Record<string, OrchState> = {
queued: "queued",
running: "running",
awaiting_approval: "waiting_approval",
completed: "succeeded",
failed: "failed",
cancelled: "cancelled",
};
function truncate(s: string, n = 60): string {
return s.length > n ? `${s.slice(0, n - 1)}` : s;
}
export function fromCloudAgent(tasks: CloudAgentTask[]): { nodes: OrchNode[]; edges: OrchEdge[] } {
if (tasks.length === 0) return { nodes: [], edges: [] };
const nodes: OrchNode[] = [];
const edges: OrchEdge[] = [];
const counts: Partial<Record<OrchState, number>> = {};
for (const t of tasks) {
const mapped = STATE_MAP[t.status];
const state: OrchState = mapped ?? "failed";
counts[state] = (counts[state] ?? 0) + 1;
const id = `cloud-agent:${t.id}`;
const active = state === "running";
nodes.push({
id,
kind: "work",
source: "cloud-agent",
state,
label: truncate(t.prompt),
sublabel: mapped ? t.providerId : `${t.providerId} — unknown status: ${String(t.status)}`,
startedAt: t.createdAt,
updatedAt: t.updatedAt,
endedAt: t.completedAt,
cost: t.result?.cost,
raw: t,
});
edges.push({
id: `e:source:cloud-agent→${id}`,
from: "source:cloud-agent",
to: id,
kind: "owns",
active,
});
const last = t.activities[t.activities.length - 1];
if (active && last) {
const actId = `${id}:activity`;
nodes.push({
id: actId,
kind: "activity",
source: "cloud-agent",
state,
label: truncate(last.content),
sublabel: last.type,
updatedAt: last.timestamp,
});
edges.push({ id: `e:${id}${actId}`, from: id, to: actId, kind: "owns", active: true });
}
}
nodes.unshift({
id: "source:cloud-agent",
kind: "source",
source: "cloud-agent",
label: "Cloud Agent",
counts,
});
return { nodes, edges };
}

View File

@@ -0,0 +1,148 @@
/** Conductor fleet snapshot → unified orchestration nodes. Pure. */
import type { FleetRunner, FleetSnapshot, FleetTask } from "@/lib/conductor/hubProxy";
import type { OrchEdge, OrchNode, OrchState } from "./orchestrationTypes";
const TERMINAL: ReadonlySet<OrchState> = new Set(["succeeded", "failed", "cancelled"]);
function mapHubStatus(status: string): OrchState | null {
const s = status.toLowerCase();
if (s === "queued" || s === "pending") return "queued";
if (s === "running" || s === "working" || s === "scheduled") return "running";
if (s === "done" || s === "completed" || s === "succeeded") return "succeeded";
if (s === "failed" || s === "error") return "failed";
if (s === "cancelled" || s === "canceled") return "cancelled";
return null;
}
function taskNode(t: FleetTask, kind: "work" | "activity"): OrchNode {
const mapped = mapHubStatus(t.status);
const state: OrchState = mapped ?? "failed";
return {
id: `conductor:task:${t.id}`,
kind,
source: "conductor",
state,
label: t.summary ?? t.id,
sublabel: mapped ? (t.repo ?? t.mode) : `unknown status: ${t.status}`,
updatedAt: t.updated_at ?? undefined,
// FleetTask has no dedicated completion timestamp — updated_at is the closest
// proxy, same pattern as fromA2A.ts (A2ATask has no completedAt either).
endedAt: TERMINAL.has(state) ? (t.updated_at ?? undefined) : undefined,
raw: t,
};
}
/**
* Tasks whose runner actually exists in `snap.runners` AND is currently "running" — those
* get "absorbed" into that runner's ActivityNode instead of getting their own work node.
* A running task pointing at a runner id that has since deregistered falls through to the
* normal work-node loop instead of being silently skipped as "already an activity".
*/
function computeActiveByRunner(snap: FleetSnapshot): Map<string, FleetTask> {
const runnerIds = new Set(snap.runners.map((r) => r.id));
const activeByRunner = new Map<string, FleetTask>();
for (const t of snap.tasks) {
if (t.runner && runnerIds.has(t.runner) && mapHubStatus(t.status) === "running") {
activeByRunner.set(t.runner, t);
}
}
return activeByRunner;
}
function runnerState(r: FleetRunner, activeTask: FleetTask | undefined): OrchState {
if (!r.online) return "failed";
if (r.draining) return "cancelled";
return activeTask ? "running" : "queued";
}
/** One work node per runner, plus an activity node for its currently-active task. */
function runnerWorkNodes(
snap: FleetSnapshot,
activeByRunner: Map<string, FleetTask>,
bump: (s: OrchState) => void
): { nodes: OrchNode[]; edges: OrchEdge[] } {
const nodes: OrchNode[] = [];
const edges: OrchEdge[] = [];
for (const r of snap.runners) {
const id = `conductor:runner:${r.id}`;
const activeTask = activeByRunner.get(r.id);
const state = runnerState(r, activeTask);
bump(state);
nodes.push({
id,
kind: "work",
source: "conductor",
state,
label: r.name,
sublabel: r.clis.join(", "),
raw: r,
});
edges.push({
id: `e:source:conductor→${id}`,
from: "source:conductor",
to: id,
kind: "owns",
active: state === "running",
});
if (activeTask) {
nodes.push(taskNode(activeTask, "activity"));
edges.push({
id: `e:${id}→conductor:task:${activeTask.id}`,
from: id,
to: `conductor:task:${activeTask.id}`,
kind: "owns",
active: true,
});
}
}
return { nodes, edges };
}
/** Work nodes for tasks not already absorbed as a runner's activity node. */
function remainingTaskWorkNodes(
snap: FleetSnapshot,
activeByRunner: Map<string, FleetTask>,
bump: (s: OrchState) => void
): { nodes: OrchNode[]; edges: OrchEdge[] } {
const nodes: OrchNode[] = [];
const edges: OrchEdge[] = [];
for (const t of snap.tasks) {
if (t.runner && activeByRunner.get(t.runner)?.id === t.id) continue; // already an activity
const node = taskNode(t, "work");
bump(node.state as OrchState);
nodes.push(node);
edges.push({
id: `e:source:conductor→${node.id}`,
from: "source:conductor",
to: node.id,
kind: "owns",
active: node.state === "running",
});
}
return { nodes, edges };
}
export function fromConductor(snap: FleetSnapshot): { nodes: OrchNode[]; edges: OrchEdge[] } {
if (snap.offline || (snap.runners.length === 0 && snap.tasks.length === 0)) {
return { nodes: [], edges: [] };
}
const counts: Partial<Record<OrchState, number>> = {};
const bump = (s: OrchState) => {
counts[s] = (counts[s] ?? 0) + 1;
};
const activeByRunner = computeActiveByRunner(snap);
const runners = runnerWorkNodes(snap, activeByRunner, bump);
const tasks = remainingTaskWorkNodes(snap, activeByRunner, bump);
const nodes = [...runners.nodes, ...tasks.nodes];
const edges = [...runners.edges, ...tasks.edges];
nodes.unshift({
id: "source:conductor",
kind: "source",
source: "conductor",
label: "Conductor",
counts,
});
return { nodes, edges };
}

View File

@@ -0,0 +1,217 @@
/** Merge the three source mappers into one snapshot: root, dedupe, staleness filter, cap. Pure. */
import {
MAX_WORK_NODES,
STALE_COMPLETED_MS,
type OrchEdge,
type OrchNode,
type OrchSnapshot,
type OrchSource,
type OrchState,
type SourceStatus,
} from "./orchestrationTypes";
const TERMINAL: ReadonlySet<OrchState> = new Set(["succeeded", "failed", "cancelled"]);
export interface MergeOptions {
now: number;
showCompleted?: boolean;
}
interface Part {
nodes: OrchNode[];
edges: OrchEdge[];
}
interface NodesAndEdges {
nodes: OrchNode[];
edges: OrchEdge[];
}
function conductorMirrorId(node: OrchNode): string | null {
const raw = node.raw as { metadata?: { conductor?: { task_id?: unknown } } } | undefined;
const id = raw?.metadata?.conductor?.task_id;
return typeof id === "string" ? id : null;
}
/**
* (2) Conductor↔A2A dedupe — key verified in src/lib/conductor/bridge.ts::ensureMirrored.
* Mutates `dropped` in place; returns the (possibly patched) nodes/edges.
*/
function dedupeConductorMirrors(
nodes: OrchNode[],
edges: OrchEdge[],
dropped: Set<string>
): NodesAndEdges {
const conductorTaskIds = new Set(
nodes
.filter((n) => n.source === "conductor" && n.id.startsWith("conductor:task:"))
.map((n) => n.id.slice("conductor:task:".length))
);
const nextNodes = [...nodes];
const nextEdges = [...edges];
for (const n of nextNodes) {
if (n.source !== "a2a" || n.kind !== "work") continue;
const mirror = conductorMirrorId(n);
if (mirror && conductorTaskIds.has(mirror)) {
dropped.add(n.id);
const cIndex = nextNodes.findIndex((c) => c.id === `conductor:task:${mirror}`);
if (cIndex !== -1) {
// Copy rather than mutate — the original object is still referenced by
// parts.conductor.nodes, and this function's contract is Pure.
const cNode: OrchNode = { ...nextNodes[cIndex], mirrorOf: n.id };
nextNodes[cIndex] = cNode;
nextEdges.push({
id: `e:mirror:${cNode.id}`,
from: cNode.id,
to: "source:a2a",
kind: "mirror",
active: false,
});
}
}
}
return { nodes: nextNodes, edges: nextEdges };
}
/** (3) staleness filter — adds stale terminal work/activity node ids to `dropped`. */
function markStaleCompleted(nodes: OrchNode[], now: number, dropped: Set<string>): void {
for (const n of nodes) {
if (n.kind !== "work" && n.kind !== "activity") continue;
if (
n.state &&
TERMINAL.has(n.state) &&
n.endedAt &&
now - Date.parse(n.endedAt) > STALE_COMPLETED_MS
) {
dropped.add(n.id);
}
}
}
/** One source's overflow placeholder node, or null when it fits under `budgetPer`. */
function overflowNodeForSource(
source: OrchSource,
list: OrchNode[],
budgetPer: number,
dropped: Set<string>
): OrchNode | null {
if (list.length <= budgetPer) return null;
list.sort((a, b) => Date.parse(b.updatedAt ?? "0") - Date.parse(a.updatedAt ?? "0"));
const excess = list.slice(budgetPer);
const counts: Partial<Record<OrchState, number>> = {};
for (const n of excess) {
dropped.add(n.id);
if (n.state) counts[n.state] = (counts[n.state] ?? 0) + 1;
}
return {
id: `overflow:${source}`,
kind: "overflow",
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,
};
}
/** (4) cap with per-source overflow, newest kept. */
function capWorkNodesWithOverflow(
nodes: OrchNode[],
edges: OrchEdge[],
dropped: Set<string>
): NodesAndEdges {
const works = nodes.filter((n) => n.kind === "work");
if (works.length <= MAX_WORK_NODES) return { nodes, edges };
const bySource = new Map<OrchSource, OrchNode[]>();
for (const w of works) {
const list = bySource.get(w.source as OrchSource) ?? [];
list.push(w);
bySource.set(w.source as OrchSource, list);
}
const budgetPer = Math.max(1, Math.floor(MAX_WORK_NODES / bySource.size));
const overflowNodes: OrchNode[] = [];
for (const [source, list] of bySource) {
const overflow = overflowNodeForSource(source, list, budgetPer, dropped);
if (overflow) overflowNodes.push(overflow);
}
const nextNodes = nodes.filter((n) => !dropped.has(n.id)).concat(overflowNodes);
const nextEdges = edges.filter((e) => !dropped.has(e.from) && !dropped.has(e.to));
for (const o of overflowNodes) {
nextEdges.push({
id: `e:source:${o.source}${o.id}`,
from: `source:${o.source}`,
to: o.id,
kind: "owns",
active: false,
});
}
return { nodes: nextNodes, edges: nextEdges };
}
/**
* (1) root — link every present SourceNode, plus failed sources so the UI can show them
* stale. Returns nodes with the root prepended.
*/
function buildRootAndSourceEdges(
nodes: OrchNode[],
edges: OrchEdge[],
sources: SourceStatus[]
): NodesAndEdges {
const root: OrchNode = { id: "orchestrator", kind: "orchestrator", label: "OmniRoute" };
const nextNodes = [...nodes];
const nextEdges = [...edges];
const sourceIds = new Set(nextNodes.filter((n) => n.kind === "source").map((n) => n.id));
for (const s of sources) {
// `!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") {
nextNodes.push({
id: `source:${s.source}`,
kind: "source",
source: s.source,
label: s.source,
sublabel: s.offline ? "offline" : "error",
});
sourceIds.add(`source:${s.source}`);
}
}
for (const id of sourceIds) {
nextEdges.push({
id: `e:orchestrator→${id}`,
from: "orchestrator",
to: id,
kind: "owns",
active: false,
});
}
return { nodes: [root, ...nextNodes], edges: nextEdges };
}
export function mergeSnapshot(
parts: { cloudAgent: Part; a2a: Part; conductor: Part },
sources: SourceStatus[],
opts: MergeOptions
): OrchSnapshot {
let nodes: OrchNode[] = [...parts.cloudAgent.nodes, ...parts.a2a.nodes, ...parts.conductor.nodes];
let edges: OrchEdge[] = [...parts.cloudAgent.edges, ...parts.a2a.edges, ...parts.conductor.edges];
const dropped = new Set<string>();
({ nodes, edges } = dedupeConductorMirrors(nodes, edges, dropped));
if (!opts.showCompleted) {
markStaleCompleted(nodes, opts.now, dropped);
}
nodes = nodes.filter((n) => !dropped.has(n.id));
edges = edges.filter((e) => !dropped.has(e.from) && !dropped.has(e.to));
({ nodes, edges } = capWorkNodesWithOverflow(nodes, edges, dropped));
({ nodes, edges } = buildRootAndSourceEdges(nodes, edges, sources));
return { nodes, edges, sources, generatedAt: new Date(opts.now).toISOString() };
}

View File

@@ -0,0 +1,58 @@
/** OrchSnapshot → @xyflow nodes/edges with a deterministic shallow 3-layer layout. Pure. */
import type { Edge, Node } from "@xyflow/react";
import { edgeStyle } from "@/shared/components/flow/edgeStyles";
import type { OrchNodeKind, OrchSnapshot } from "./orchestrationTypes";
const LAYER_Y: Record<OrchNodeKind, number> = {
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;
} {
const byLayer = new Map<number, string[]>();
for (const n of [...snap.nodes].sort((a, b) => a.id.localeCompare(b.id))) {
const y = LAYER_Y[n.kind];
const ids = byLayer.get(y) ?? [];
ids.push(n.id);
byLayer.set(y, ids);
}
const pos = new Map<string, { x: number; y: number }>();
for (const [y, ids] of byLayer) {
const width = (ids.length - 1) * X_GAP;
ids.forEach((id, i) => pos.set(id, { x: i * X_GAP - width / 2, y }));
}
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<string, unknown>,
}));
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,
style: e.kind === "mirror" ? { ...style, strokeDasharray: "6 4" } : style,
};
});
const fitKey = snap.nodes
.filter((n) => n.kind === "work")
.map((n) => n.id)
.sort()
.join("|");
return { nodes, edges, fitKey };
}

View File

@@ -0,0 +1,79 @@
/**
* Pure domain vocabulary for the Orchestration Canvas — no React, no side effects.
* Spec: _tasks/superpowers/specs/2026-08-30-orchestration-canvas-design.md
*/
import { STATUS_HEX } from "@/shared/constants/statusColors";
export type OrchState =
"queued" | "running" | "waiting_approval" | "succeeded" | "failed" | "cancelled";
export type OrchSource = "cloud-agent" | "a2a" | "conductor" | "routing";
export type OrchNodeKind = "orchestrator" | "source" | "work" | "activity" | "overflow";
export interface OrchNode {
id: string; // `${source}:${sourceId}` for work nodes
kind: OrchNodeKind;
source?: OrchSource;
state?: OrchState;
label: string;
sublabel?: string;
startedAt?: string;
updatedAt?: string;
endedAt?: string;
cost?: number;
counts?: Partial<Record<OrchState, number>>;
// 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<Record<OrchState, number>>;
mirrorOf?: string;
raw?: unknown;
}
export interface OrchEdge {
id: string;
from: string;
to: string;
kind: "owns" | "mirror";
active: boolean; // true while the target work is `running`
}
export interface SourceStatus {
source: OrchSource;
ok: boolean;
offline?: boolean;
error?: string;
staleSince?: string;
}
export interface OrchSnapshot {
nodes: OrchNode[];
edges: OrchEdge[];
sources: SourceStatus[];
generatedAt: string;
}
export const ORCH_STATES = [
"queued",
"running",
"waiting_approval",
"succeeded",
"failed",
"cancelled",
] as const satisfies readonly OrchState[];
const STATE_HEX: Record<OrchState, string> = {
queued: STATUS_HEX.muted,
running: STATUS_HEX.warning,
waiting_approval: STATUS_HEX.approval,
succeeded: STATUS_HEX.success,
failed: STATUS_HEX.error,
cancelled: STATUS_HEX.muted,
};
export function orchStateColor(state: OrchState): string {
return STATE_HEX[state];
}
export const STALE_COMPLETED_MS = 600_000; // completed >10 min ago drop out of the live view
export const MAX_WORK_NODES = 40; // beyond this, per-source overflow nodes take over

View File

@@ -0,0 +1,50 @@
/** OrchSnapshot → overview counters + kanban columns. Pure. */
import {
ORCH_STATES,
type OrchNode,
type OrchSnapshot,
type OrchState,
} from "./orchestrationTypes";
export interface OverviewData {
counts: Record<OrchState, number>;
columns: {
queued: OrchNode[];
running: OrchNode[];
waiting_approval: OrchNode[];
done: OrchNode[];
};
}
export function overviewProjection(snap: OrchSnapshot, comboActive: number): OverviewData {
const counts = Object.fromEntries(ORCH_STATES.map((s) => [s, 0])) as Record<OrchState, number>;
const columns: OverviewData["columns"] = {
queued: [],
running: [],
waiting_approval: [],
done: [],
};
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") {
columns[n.state].push(n);
} else {
columns.done.push(n);
}
}
columns.done.sort((a, b) => Date.parse(b.updatedAt ?? "0") - Date.parse(a.updatedAt ?? "0"));
counts.running += comboActive;
return { counts, columns };
}

View File

@@ -14,4 +14,6 @@ export const STATUS_HEX = {
warning: "#f59e0b",
error: "#ef4444",
muted: "#6b7280",
/** Human-approval gate (waiting_approval) — violet, matching the industry de-facto palette. */
approval: "#8b5cf6",
} as const;

View File

@@ -0,0 +1,372 @@
/**
* tests/unit/ui/orchestrationModel.test.ts
* Run: node --import tsx/esm --test tests/unit/ui/orchestrationModel.test.ts
*/
import { describe, it } from "node:test";
import assert from "node:assert/strict";
import {
ORCH_STATES,
orchStateColor,
} from "../../../src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts";
import { STATUS_HEX } from "../../../src/shared/constants/statusColors.ts";
import { fromCloudAgent } from "../../../src/app/(dashboard)/dashboard/orchestration/model/fromCloudAgent.ts";
import type { CloudAgentTask } from "../../../src/lib/cloudAgent/types.ts";
import { fromA2A } from "../../../src/app/(dashboard)/dashboard/orchestration/model/fromA2A.ts";
import type { A2ATask } from "../../../src/lib/a2a/taskManager.ts";
import { fromConductor } from "../../../src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts";
import type { FleetSnapshot } from "../../../src/lib/conductor/hubProxy.ts";
import { mergeSnapshot } from "../../../src/app/(dashboard)/dashboard/orchestration/model/mergeSnapshot.ts";
import {
STALE_COMPLETED_MS,
MAX_WORK_NODES,
} from "../../../src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts";
describe("orchestrationTypes", () => {
it("covers all six states with a color each", () => {
assert.equal(ORCH_STATES.length, 6);
for (const s of ORCH_STATES) {
assert.match(orchStateColor(s), /^#[0-9a-f]{6}$/i, s);
}
});
it("waiting_approval maps to the new STATUS_HEX.approval violet", () => {
assert.equal(orchStateColor("waiting_approval"), STATUS_HEX.approval);
assert.equal(STATUS_HEX.approval, "#8b5cf6");
});
it("running maps to warning, succeeded to success, failed to error", () => {
assert.equal(orchStateColor("running"), STATUS_HEX.warning);
assert.equal(orchStateColor("succeeded"), STATUS_HEX.success);
assert.equal(orchStateColor("failed"), STATUS_HEX.error);
});
});
function caTask(over: Partial<CloudAgentTask>): CloudAgentTask {
return {
id: "t1",
providerId: "devin",
status: "running",
prompt: "Fix the flaky test in CI",
source: { repoName: "acme/app", repoUrl: "https://github.com/acme/app" },
options: {},
activities: [],
createdAt: "2026-08-30T10:00:00Z",
updatedAt: "2026-08-30T10:05:00Z",
...over,
} as CloudAgentTask;
}
describe("fromCloudAgent", () => {
it("maps every status to the unified OrchState", () => {
const cases: Array<[CloudAgentTask["status"], string]> = [
["queued", "queued"],
["running", "running"],
["awaiting_approval", "waiting_approval"],
["completed", "succeeded"],
["failed", "failed"],
["cancelled", "cancelled"],
];
for (const [input, expected] of cases) {
const { nodes } = fromCloudAgent([caTask({ status: input })]);
const work = nodes.find((n) => n.kind === "work");
assert.equal(work?.state, expected, input);
}
});
it("unknown status becomes failed with the raw value in sublabel", () => {
const { nodes } = fromCloudAgent([caTask({ status: "exploded" as CloudAgentTask["status"] })]);
const work = nodes.find((n) => n.kind === "work");
assert.equal(work?.state, "failed");
assert.match(work?.sublabel ?? "", /exploded/);
});
it("running task with activities gets one ActivityNode; completed does not", () => {
const running = caTask({
activities: [
{ id: "a1", type: "command", content: "npm test", timestamp: "2026-08-30T10:04:00Z" },
],
});
const done = caTask({ id: "t2", status: "completed", activities: running.activities });
const { nodes, edges } = fromCloudAgent([running, done]);
const acts = nodes.filter((n) => n.kind === "activity");
assert.equal(acts.length, 1);
assert.equal(acts[0].id, "cloud-agent:t1:activity");
assert.ok(
edges.some(
(e) => e.from === "cloud-agent:t1" && e.to === "cloud-agent:t1:activity" && e.active
)
);
});
it("emits a SourceNode with per-state counts and owns-edges from it", () => {
const { nodes, edges } = fromCloudAgent([caTask({}), caTask({ id: "t2", status: "failed" })]);
const src = nodes.find((n) => n.id === "source:cloud-agent");
assert.equal(src?.counts?.running, 1);
assert.equal(src?.counts?.failed, 1);
assert.ok(
edges.some(
(e) => e.from === "source:cloud-agent" && e.to === "cloud-agent:t1" && e.kind === "owns"
)
);
});
it("empty input emits nothing", () => {
const out = fromCloudAgent([]);
assert.equal(out.nodes.length, 0);
assert.equal(out.edges.length, 0);
});
});
function a2aTask(over: Partial<A2ATask>): A2ATask {
return {
id: "a1",
skill: "smart-routing",
state: "working",
input: { skill: "smart-routing", messages: [{ role: "user", content: "route this well" }] },
artifacts: [],
events: [{ timestamp: "2026-08-30T10:00:00Z", state: "submitted" }],
metadata: {},
createdAt: "2026-08-30T10:00:00Z",
updatedAt: "2026-08-30T10:01:00Z",
expiresAt: "2026-08-30T10:05:00Z",
...over,
} as A2ATask;
}
describe("fromA2A", () => {
it("maps the five A2A states", () => {
const cases: Array<[A2ATask["state"], string]> = [
["submitted", "queued"],
["working", "running"],
["completed", "succeeded"],
["failed", "failed"],
["cancelled", "cancelled"],
];
for (const [input, expected] of cases) {
const { nodes } = fromA2A([a2aTask({ state: input })]);
assert.equal(nodes.find((n) => n.kind === "work")?.state, expected, input);
}
});
it("work node id is a2a:<id>, label is the skill", () => {
const { nodes } = fromA2A([a2aTask({})]);
const w = nodes.find((n) => n.kind === "work");
assert.equal(w?.id, "a2a:a1");
assert.equal(w?.label, "smart-routing");
});
it("empty input emits nothing", () => {
assert.equal(fromA2A([]).nodes.length, 0);
});
});
const baseSnap: FleetSnapshot = {
offline: false,
runners: [{ id: "r1", name: "runner-one", clis: ["claude"], online: true, draining: false }],
tasks: [
{
id: "ct1",
status: "running",
mode: "auto",
repo: "acme/app",
runner: "r1",
summary: "Refactor auth",
branch: null,
error: null,
updated_at: "2026-08-30T10:00:00Z",
},
],
};
describe("fromConductor", () => {
it("online runner with a running task → running WorkNode + ActivityNode for the task", () => {
const { nodes, edges } = fromConductor(baseSnap);
const runner = nodes.find((n) => n.id === "conductor:runner:r1");
assert.equal(runner?.state, "running");
const act = nodes.find((n) => n.id === "conductor:task:ct1");
assert.equal(act?.kind, "activity");
assert.ok(edges.some((e) => e.from === "conductor:runner:r1" && e.to === "conductor:task:ct1"));
});
it("queued task without runner hangs directly under the source as a work node", () => {
const snap: FleetSnapshot = {
...baseSnap,
runners: [],
tasks: [{ ...baseSnap.tasks[0], id: "ct2", status: "queued", runner: null }],
};
const { nodes, edges } = fromConductor(snap);
const w = nodes.find((n) => n.id === "conductor:task:ct2");
assert.equal(w?.kind, "work");
assert.equal(w?.state, "queued");
assert.ok(edges.some((e) => e.from === "source:conductor" && e.to === "conductor:task:ct2"));
});
it("offline snapshot emits only nothing (hook marks the source offline separately)", () => {
const out = fromConductor({ offline: true, runners: [], tasks: [] });
assert.equal(out.nodes.length, 0);
});
it("unknown hub status maps to failed with the raw value in sublabel", () => {
const snap: FleetSnapshot = {
...baseSnap,
runners: [],
tasks: [{ ...baseSnap.tasks[0], id: "ct3", status: "vaporized", runner: null }],
};
const w = fromConductor(snap).nodes.find((n) => n.id === "conductor:task:ct3");
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 = [
{ source: "cloud-agent" as const, ok: true },
{ source: "a2a" as const, ok: true },
{ source: "conductor" as const, ok: true },
];
const NOW = Date.parse("2026-08-30T12:00:00Z");
const empty = { nodes: [], edges: [] };
describe("mergeSnapshot", () => {
it("adds the orchestrator root and root→source edges", () => {
const snap = mergeSnapshot(
{ cloudAgent: fromCloudAgent([caTask({})]), a2a: empty, conductor: empty },
OK_SOURCES,
{ now: NOW }
);
assert.ok(snap.nodes.some((n) => n.id === "orchestrator"));
assert.ok(snap.edges.some((e) => e.from === "orchestrator" && e.to === "source:cloud-agent"));
});
it("dedupes a Conductor-mirrored A2A task into one node with a mirror edge", () => {
const a2a = fromA2A([
a2aTask({ id: "am1", skill: "conductor", metadata: { conductor: { task_id: "ct1" } } }),
]);
const conductor = fromConductor(baseSnap); // contains conductor:task:ct1 as activity of r1
const snap = mergeSnapshot({ cloudAgent: empty, a2a, conductor }, OK_SOURCES, { now: NOW });
assert.ok(
!snap.nodes.some((n) => n.id === "a2a:am1"),
"mirrored A2A work node must be dropped"
);
const mirrorEdge = snap.edges.find((e) => e.kind === "mirror");
assert.equal(mirrorEdge?.to, "source:a2a");
});
it("drops terminal work older than STALE_COMPLETED_MS unless showCompleted", () => {
const old = caTask({
id: "old",
status: "completed",
completedAt: new Date(NOW - STALE_COMPLETED_MS - 1000).toISOString(),
});
const parts = { cloudAgent: fromCloudAgent([old]), a2a: empty, conductor: empty };
assert.ok(
!mergeSnapshot(parts, OK_SOURCES, { now: NOW }).nodes.some((n) => n.id === "cloud-agent:old")
);
assert.ok(
mergeSnapshot(parts, OK_SOURCES, { now: NOW, showCompleted: true }).nodes.some(
(n) => n.id === "cloud-agent:old"
)
);
});
it("caps work nodes at MAX_WORK_NODES with a per-source overflow node", () => {
const many = Array.from({ length: MAX_WORK_NODES + 10 }, (_, i) =>
caTask({ id: `m${i}`, updatedAt: new Date(NOW - i * 1000).toISOString() })
);
const snap = mergeSnapshot(
{ cloudAgent: fromCloudAgent(many), a2a: empty, conductor: empty },
OK_SOURCES,
{ now: NOW }
);
const works = snap.nodes.filter((n) => n.kind === "work");
assert.ok(works.length <= MAX_WORK_NODES, `got ${works.length}`);
const overflow = snap.nodes.find((n) => n.id === "overflow:cloud-agent");
assert.ok(overflow, "overflow node expected");
});
it("keeps SourceStatus[] verbatim on the snapshot", () => {
const src = [{ source: "conductor" as const, ok: false, offline: true }];
const snap = mergeSnapshot({ cloudAgent: empty, a2a: empty, conductor: empty }, src, {
now: NOW,
});
assert.deepEqual(snap.sources, src);
});
it("does not mutate input node objects when deduping (honors the Pure contract)", () => {
const a2a = fromA2A([
a2aTask({ id: "am2", skill: "conductor", metadata: { conductor: { task_id: "ct1" } } }),
]);
const conductor = fromConductor(baseSnap); // contains conductor:task:ct1 as activity of r1
const originalNode = conductor.nodes.find((n) => n.id === "conductor:task:ct1");
assert.ok(originalNode, "conductor:task:ct1 must exist in the source part");
const snap = mergeSnapshot({ cloudAgent: empty, a2a, conductor }, OK_SOURCES, { now: NOW });
assert.equal(
"mirrorOf" in (originalNode as object),
false,
"the original conductor input node must not be mutated"
);
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,
runners: [],
tasks: [
{
...baseSnap.tasks[0],
id: "ctOld",
status: "completed",
runner: null,
updated_at: new Date(NOW - STALE_COMPLETED_MS - 1000).toISOString(),
},
],
};
const parts = { cloudAgent: empty, a2a: empty, conductor: fromConductor(staleSnap) };
assert.ok(
!mergeSnapshot(parts, OK_SOURCES, { now: NOW }).nodes.some(
(n) => n.id === "conductor:task:ctOld"
)
);
assert.ok(
mergeSnapshot(parts, OK_SOURCES, { now: NOW, showCompleted: true }).nodes.some(
(n) => n.id === "conductor:task:ctOld"
)
);
});
});

View File

@@ -0,0 +1,56 @@
/** Run: node --import tsx/esm --test tests/unit/ui/orchestrationToFlow.test.ts */
import { describe, it } from "node:test";
import assert from "node:assert/strict";
import { orchestrationToFlow } from "../../../src/app/(dashboard)/dashboard/orchestration/model/orchestrationToFlow.ts";
import type { OrchSnapshot } from "../../../src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts";
const snap: OrchSnapshot = {
nodes: [
{ id: "orchestrator", kind: "orchestrator", label: "OmniRoute" },
{ id: "source:a2a", kind: "source", source: "a2a", label: "A2A" },
{ id: "a2a:t1", kind: "work", source: "a2a", state: "running", label: "smart-routing" },
{ id: "a2a:t2", kind: "work", source: "a2a", state: "failed", label: "cost-analysis" },
],
edges: [
{ id: "e1", from: "orchestrator", to: "source:a2a", kind: "owns", active: false },
{ 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",
};
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)
);
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);
assert.equal(ys.get("a2a:t1"), 320);
});
it("active edge is animated; edge to failed work is red", () => {
const { edges } = orchestrationToFlow(snap);
assert.equal(edges.find((e) => e.id === "e2")?.animated, true);
const failedEdge = edges.find((e) => e.id === "e3");
assert.equal((failedEdge?.style as { stroke?: string })?.stroke, "#ef4444");
});
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)),
};
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"),
};
assert.notEqual(orchestrationToFlow(nodeRemoved).fitKey, k1);
});
});

View File

@@ -0,0 +1,67 @@
/** Run: node --import tsx/esm --test tests/unit/ui/overviewProjection.test.ts */
import { describe, it } from "node:test";
import assert from "node:assert/strict";
import { overviewProjection } from "../../../src/app/(dashboard)/dashboard/orchestration/model/overviewProjection.ts";
import type { OrchSnapshot } from "../../../src/app/(dashboard)/dashboard/orchestration/model/orchestrationTypes.ts";
const snap: OrchSnapshot = {
nodes: [
{ id: "orchestrator", kind: "orchestrator", label: "OmniRoute" },
{ id: "cloud-agent:1", kind: "work", source: "cloud-agent", state: "running", label: "a" },
{
id: "cloud-agent:2",
kind: "work",
source: "cloud-agent",
state: "waiting_approval",
label: "b",
},
{
id: "a2a:3",
kind: "work",
source: "a2a",
state: "failed",
label: "c",
updatedAt: "2026-08-30T11:00:00Z",
},
{ id: "a2a:3:activity", kind: "activity", source: "a2a", state: "running", label: "noise" },
],
edges: [],
sources: [],
generatedAt: "2026-08-30T12:00:00Z",
};
describe("overviewProjection", () => {
it("counts only work nodes and adds comboActive to running", () => {
const { counts } = overviewProjection(snap, 4);
assert.equal(counts.running, 1 + 4);
assert.equal(counts.waiting_approval, 1);
assert.equal(counts.failed, 1);
assert.equal(counts.queued, 0);
});
it("folds terminals into the done column", () => {
const { columns } = overviewProjection(snap, 0);
assert.equal(columns.done.length, 1);
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);
});
});

View File

@@ -0,0 +1,121 @@
// @vitest-environment jsdom
import React, { act } from "react";
import { createRoot } from "react-dom/client";
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
// Capture the onEvent handler the hook registers on the requests channel.
let capturedOnEvent: ((p: { channel: string }) => void) | null = null;
vi.mock("@/hooks/useLiveDashboard", () => ({
useLiveDashboard: (opts: { onEvent?: (p: { channel: string }) => void }) => {
capturedOnEvent = opts.onEvent ?? null;
return { connection: { isConnected: true }, events: [] };
},
}));
import { useOrchestrationSnapshot } from "@/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot";
function HookProbe({
onRender,
}: {
onRender: (v: ReturnType<typeof useOrchestrationSnapshot>) => void;
}) {
onRender(useOrchestrationSnapshot());
return null;
}
const okJson = (body: unknown) =>
Promise.resolve({ ok: true, json: () => Promise.resolve(body) } as Response);
describe("useOrchestrationSnapshot", () => {
let container: HTMLDivElement;
let root: ReturnType<typeof createRoot>;
beforeEach(() => {
vi.useFakeTimers();
container = document.createElement("div");
document.body.appendChild(container);
root = createRoot(container);
});
afterEach(() => {
act(() => root.unmount());
container.remove();
vi.useRealTimers();
vi.restoreAllMocks();
});
it("polls the three endpoints and builds a snapshot; a failed source keeps the last good data", async () => {
let latest: ReturnType<typeof useOrchestrationSnapshot> | 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 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 });
return okJson({ offline: false, runners: [], tasks: [] });
});
vi.stubGlobal("fetch", fetchMock);
await act(async () => {
root.render(
<HookProbe
onRender={(v) => {
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 });
return okJson({ offline: false, runners: [], tasks: [] });
});
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);
});
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 });
return okJson({ offline: false, runners: [], tasks: [] });
});
vi.stubGlobal("fetch", fetchMock);
await act(async () => {
root.render(<HookProbe onRender={() => {}} />);
});
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);
});
// Two burst events → exactly ONE extra round of 3 fetches (debounce), not two.
expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3);
});
});