diff --git a/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts b/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts new file mode 100644 index 0000000000..74a05b40e4 --- /dev/null +++ b/src/app/(dashboard)/dashboard/orchestration/model/fromConductor.ts @@ -0,0 +1,106 @@ +/** Conductor fleet snapshot → unified orchestration nodes. Pure. */ +import type { FleetSnapshot, FleetTask } from "@/lib/conductor/hubProxy"; +import type { OrchEdge, OrchNode, OrchState } from "./orchestrationTypes"; + +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); + return { + id: `conductor:task:${t.id}`, + kind, + source: "conductor", + state: mapped ?? "failed", + label: t.summary ?? t.id, + sublabel: mapped ? (t.repo ?? t.mode) : `unknown status: ${t.status}`, + updatedAt: t.updated_at ?? undefined, + raw: t, + }; +} + +export function fromConductor(snap: FleetSnapshot): { nodes: OrchNode[]; edges: OrchEdge[] } { + if (snap.offline || (snap.runners.length === 0 && snap.tasks.length === 0)) { + return { nodes: [], edges: [] }; + } + const nodes: OrchNode[] = []; + const edges: OrchEdge[] = []; + const counts: Partial> = {}; + const bump = (s: OrchState) => { + counts[s] = (counts[s] ?? 0) + 1; + }; + + const activeByRunner = new Map(); + for (const t of snap.tasks) { + if (t.runner && mapHubStatus(t.status) === "running") activeByRunner.set(t.runner, t); + } + + for (const r of snap.runners) { + const id = `conductor:runner:${r.id}`; + const activeTask = activeByRunner.get(r.id); + const state: OrchState = !r.online + ? "failed" + : r.draining + ? "cancelled" + : activeTask + ? "running" + : "queued"; + 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, + }); + } + } + + 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", + }); + } + + nodes.unshift({ + id: "source:conductor", + kind: "source", + source: "conductor", + label: "Conductor", + counts, + }); + return { nodes, edges }; +} diff --git a/tests/unit/ui/orchestrationModel.test.ts b/tests/unit/ui/orchestrationModel.test.ts index 37c8f26462..8dbaab4249 100644 --- a/tests/unit/ui/orchestrationModel.test.ts +++ b/tests/unit/ui/orchestrationModel.test.ts @@ -13,6 +13,8 @@ import { fromCloudAgent } from "../../../src/app/(dashboard)/dashboard/orchestra 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"; describe("orchestrationTypes", () => { it("covers all six states with a color each", () => { @@ -144,3 +146,58 @@ describe("fromA2A", () => { 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/); + }); +});