From ca98738eb8c75c600dd5e618fe65606ed31f7dc7 Mon Sep 17 00:00:00 2001 From: Camiel van Schoonhoven Date: Thu, 30 Jul 2026 15:03:05 -0700 Subject: [PATCH] feat: run comparison - merged graph & change summary logic --- .../utils/buildMergedGraph.test.ts | 165 ++++++++++++++++++ .../CompareView/utils/buildMergedGraph.ts | 145 +++++++++++++++ .../CompareView/utils/summarizeChange.test.ts | 159 +++++++++++++++++ .../CompareView/utils/summarizeChange.ts | 72 ++++++++ 4 files changed, 541 insertions(+) create mode 100644 src/routes/v2/pages/CompareView/utils/buildMergedGraph.test.ts create mode 100644 src/routes/v2/pages/CompareView/utils/buildMergedGraph.ts create mode 100644 src/routes/v2/pages/CompareView/utils/summarizeChange.test.ts create mode 100644 src/routes/v2/pages/CompareView/utils/summarizeChange.ts diff --git a/src/routes/v2/pages/CompareView/utils/buildMergedGraph.test.ts b/src/routes/v2/pages/CompareView/utils/buildMergedGraph.test.ts new file mode 100644 index 000000000..85e8bb7cc --- /dev/null +++ b/src/routes/v2/pages/CompareView/utils/buildMergedGraph.test.ts @@ -0,0 +1,165 @@ +import { describe, expect, test } from "vitest"; + +import type { ComponentSpec, TaskSpec } from "@/utils/componentSpec"; + +import { buildMergedGraph } from "./buildMergedGraph"; +import { buildPipelineComparison } from "./comparePipelines"; + +const task = (digest: string, overrides: Partial = {}): TaskSpec => ({ + componentRef: { name: "comp", digest }, + ...overrides, +}); + +const fromOutput = (taskId: string) => ({ + taskOutput: { taskId, outputName: "out" }, +}); + +const fromInput = (inputName: string) => ({ graphInput: { inputName } }); + +const ioSpec = (): ComponentSpec => ({ + inputs: [{ name: "data" }], + outputs: [{ name: "model" }], + implementation: { + graph: { + tasks: { + train: task("d1", { arguments: { x: fromInput("data") } }), + }, + outputValues: { + model: { taskOutput: { taskId: "train", outputName: "out" } }, + }, + }, + }, +}); + +const graphSpec = (tasks: Record): ComponentSpec => ({ + implementation: { graph: { tasks } }, +}); + +const containerSpec = (): ComponentSpec => ({ + implementation: { container: { image: "python:3.11" } }, +}); + +const noStatus = new Map(); + +const merge = (specA: ComponentSpec, specB: ComponentSpec) => + buildMergedGraph( + buildPipelineComparison( + { spec: specA, taskStatusMap: noStatus }, + { spec: specB, taskStatusMap: noStatus }, + ), + ); + +describe("buildMergedGraph()", () => { + test("unions task ids from both runs into one node each", () => { + const specA = graphSpec({ train: task("d1"), evaluate: task("d2") }); + const specB = graphSpec({ train: task("d1"), deploy: task("d3") }); + + const { nodes } = merge(specA, specB); + + expect(nodes.map((n) => n.id).sort()).toEqual([ + "deploy", + "evaluate", + "train", + ]); + expect(nodes.every((n) => n.type === "mergedTask")).toBe(true); + }); + + test("marks an edge present in both runs as unchanged", () => { + const tasks = { + gen: task("d0"), + sink: task("d1", { arguments: { x: fromOutput("gen") } }), + }; + const { edges } = merge(graphSpec(tasks), graphSpec(tasks)); + + expect(edges).toHaveLength(1); + expect(edges[0]).toMatchObject({ + source: "gen", + target: "sink", + data: { membership: "unchanged" }, + }); + }); + + test("marks run-exclusive edges as lost (A-only) and new (B-only)", () => { + const specA = graphSpec({ + gen: task("d0"), + sink: task("d1", { arguments: { x: fromOutput("gen") } }), + }); + const specB = graphSpec({ + gen: task("d0"), + sink: task("d1"), + extra: task("d2", { arguments: { y: fromOutput("gen") } }), + }); + + const { edges } = merge(specA, specB); + const byId = Object.fromEntries( + edges.map((e) => [e.id, e.data?.membership]), + ); + + expect(byId["gen->sink"]).toBe("lost"); + expect(byId["gen->extra"]).toBe("new"); + }); + + test("drops edges that reference an unknown task", () => { + const specA = graphSpec({ + only: task("d0", { arguments: { x: fromOutput("ghost") } }), + }); + + const { nodes, edges } = merge(specA, specA); + + expect(nodes.map((n) => n.id)).toEqual(["only"]); + expect(edges).toEqual([]); + }); + + test("produces no nodes for container-implementation specs", () => { + const { nodes, edges } = merge(containerSpec(), containerSpec()); + + expect(nodes).toEqual([]); + expect(edges).toEqual([]); + }); + + test("adds prefixed nodes for pipeline inputs and outputs", () => { + const { nodes } = merge(ioSpec(), ioSpec()); + + const byId = Object.fromEntries(nodes.map((n) => [n.id, n.type])); + expect(byId["input:data"]).toBe("mergedIo"); + expect(byId["output:model"]).toBe("mergedIo"); + expect(byId["train"]).toBe("mergedTask"); + }); + + test("wires input→task and task→output edges", () => { + const { edges } = merge(ioSpec(), ioSpec()); + const byId = Object.fromEntries( + edges.map((e) => [e.id, e.data?.membership]), + ); + + expect(byId["input:data->train"]).toBe("unchanged"); + expect(byId["train->output:model"]).toBe("unchanged"); + }); + + test("marks a rewired output edge as run-exclusive", () => { + const specA = ioSpec(); + const specB: ComponentSpec = { + inputs: [{ name: "data" }], + outputs: [{ name: "model" }], + implementation: { + graph: { + tasks: { + train: task("d1", { arguments: { x: fromInput("data") } }), + tune: task("d2"), + }, + outputValues: { + model: { taskOutput: { taskId: "tune", outputName: "out" } }, + }, + }, + }, + }; + + const { edges } = merge(specA, specB); + const byId = Object.fromEntries( + edges.map((e) => [e.id, e.data?.membership]), + ); + + expect(byId["train->output:model"]).toBe("lost"); + expect(byId["tune->output:model"]).toBe("new"); + }); +}); diff --git a/src/routes/v2/pages/CompareView/utils/buildMergedGraph.ts b/src/routes/v2/pages/CompareView/utils/buildMergedGraph.ts new file mode 100644 index 000000000..e35365173 --- /dev/null +++ b/src/routes/v2/pages/CompareView/utils/buildMergedGraph.ts @@ -0,0 +1,145 @@ +import type { Edge, Node } from "@xyflow/react"; + +import type { TaskSpec } from "@/utils/componentSpec"; +import { + isGraphInputArgument, + isTaskOutputArgument, +} from "@/utils/componentSpec"; + +import type { + DiffStatus, + IoDiff, + PipelineComparison, + TaskDiff, +} from "./comparePipelines"; + +export type SpotlightMode = "both" | "a" | "b"; + +export interface MergedTaskNodeData extends Record { + diff: TaskDiff; + spotlight: SpotlightMode; + singleRun?: boolean; +} + +export interface MergedIoNodeData extends Record { + diff: IoDiff; + spotlight: SpotlightMode; +} + +type MergedTaskNode = Node; +type MergedIoNode = Node; +export type MergedNode = MergedTaskNode | MergedIoNode; + +interface MergedEdgeData extends Record { + membership: DiffStatus; +} + +type MergedEdge = Edge; + +export interface MergedGraphModel { + nodes: MergedNode[]; + edges: MergedEdge[]; +} + +const inputNodeId = (name: string) => `input:${name}`; +const outputNodeId = (name: string) => `output:${name}`; + +interface EdgeSides { + source: string; + target: string; + inA: boolean; + inB: boolean; +} + +export function buildMergedGraph( + comparison: PipelineComparison, +): MergedGraphModel { + const { taskDiffs, inputDiffs, outputDiffs } = comparison; + + const taskIds = new Set(taskDiffs.map((diff) => diff.taskId)); + const inputIds = new Set(inputDiffs.map((diff) => inputNodeId(diff.name))); + + const inputNodes: MergedNode[] = inputDiffs.map((diff) => ({ + id: inputNodeId(diff.name), + type: "mergedIo", + position: { x: 0, y: 0 }, + data: { diff, spotlight: "both" }, + })); + + const taskNodes: MergedNode[] = taskDiffs.map((diff) => ({ + id: diff.taskId, + type: "mergedTask", + position: { x: 0, y: 0 }, + data: { diff, spotlight: "both" }, + })); + + const outputNodes: MergedNode[] = outputDiffs.map((diff) => ({ + id: outputNodeId(diff.name), + type: "mergedIo", + position: { x: 0, y: 0 }, + data: { diff, spotlight: "both" }, + })); + + const edgeSides = new Map(); + + const addEdge = (source: string, target: string, side: "a" | "b") => { + const key = `${source}->${target}`; + const existing = edgeSides.get(key) ?? { + source, + target, + inA: false, + inB: false, + }; + if (side === "a") existing.inA = true; + else existing.inB = true; + edgeSides.set(key, existing); + }; + + const collectTaskInputs = ( + targetId: string, + taskSpec: TaskSpec | undefined, + side: "a" | "b", + ) => { + if (!taskSpec?.arguments) return; + + for (const argument of Object.values(taskSpec.arguments)) { + if (isTaskOutputArgument(argument)) { + const source = argument.taskOutput.taskId; + if (taskIds.has(source)) addEdge(source, targetId, side); + } else if (isGraphInputArgument(argument)) { + const source = inputNodeId(argument.graphInput.inputName); + if (inputIds.has(source)) addEdge(source, targetId, side); + } + } + }; + + for (const diff of taskDiffs) { + collectTaskInputs(diff.taskId, diff.a, "a"); + collectTaskInputs(diff.taskId, diff.b, "b"); + } + + for (const diff of outputDiffs) { + const target = outputNodeId(diff.name); + if (diff.sourceTaskIdA && taskIds.has(diff.sourceTaskIdA)) { + addEdge(diff.sourceTaskIdA, target, "a"); + } + if (diff.sourceTaskIdB && taskIds.has(diff.sourceTaskIdB)) { + addEdge(diff.sourceTaskIdB, target, "b"); + } + } + + const edges: MergedEdge[] = Array.from(edgeSides.values()).map( + ({ source, target, inA, inB }) => { + const membership: DiffStatus = + inA && inB ? "unchanged" : inA ? "lost" : "new"; + return { + id: `${source}->${target}`, + source, + target, + data: { membership }, + }; + }, + ); + + return { nodes: [...inputNodes, ...taskNodes, ...outputNodes], edges }; +} diff --git a/src/routes/v2/pages/CompareView/utils/summarizeChange.test.ts b/src/routes/v2/pages/CompareView/utils/summarizeChange.test.ts new file mode 100644 index 000000000..a5bbe7bd2 --- /dev/null +++ b/src/routes/v2/pages/CompareView/utils/summarizeChange.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, test } from "vitest"; + +import type { IoDiff, TaskDiff } from "./comparePipelines"; +import { summarizeIoChange, summarizeTaskChange } from "./summarizeChange"; + +const taskDiff = (overrides: Partial = {}): TaskDiff => ({ + taskId: "train", + status: "changed", + a: { componentRef: { name: "comp", digest: "d1" } }, + b: { componentRef: { name: "comp", digest: "d1" } }, + digestA: "d1", + digestB: "d1", + componentChanged: false, + cacheDisabledA: false, + cacheDisabledB: false, + cacheChanged: false, + outcomeChanged: false, + argumentDiffs: [], + annotationDiffs: [], + settingDiffs: [], + ...overrides, +}); + +const ioDiff = (overrides: Partial = {}): IoDiff => ({ + name: "model", + kind: "output", + status: "changed", + fieldDiffs: [], + ...overrides, +}); + +describe("summarizeTaskChange()", () => { + test("reports a component version change", () => { + const summary = summarizeTaskChange( + taskDiff({ digestA: "d1", digestB: "d2", componentChanged: true }), + ); + expect(summary).toBe("component"); + }); + + test("counts and pluralizes changed arguments", () => { + const summary = summarizeTaskChange( + taskDiff({ + argumentDiffs: [ + { key: "epochs", status: "changed" }, + { key: "lr", status: "new" }, + { key: "region", status: "unchanged" }, + ], + }), + ); + expect(summary).toBe("2 arguments"); + }); + + test("combines component, arguments, and cache into one caption", () => { + const summary = summarizeTaskChange( + taskDiff({ + digestA: "d1", + digestB: "d2", + componentChanged: true, + argumentDiffs: [{ key: "epochs", status: "changed" }], + cacheChanged: true, + cacheDisabledB: true, + }), + ); + expect(summary).toBe("component · 1 argument · cache disabled"); + }); + + test("describes a cache re-enable", () => { + const summary = summarizeTaskChange( + taskDiff({ + cacheChanged: true, + cacheDisabledA: true, + cacheDisabledB: false, + }), + ); + expect(summary).toBe("cache enabled"); + }); + + test("counts changed execution settings", () => { + const summary = summarizeTaskChange( + taskDiff({ + settingDiffs: [ + { key: "retryStrategy", status: "changed" }, + { key: "isEnabled", status: "new" }, + ], + }), + ); + expect(summary).toBe("2 settings"); + }); + + test("does not repeat a cache flip as a setting change", () => { + const summary = summarizeTaskChange( + taskDiff({ + cacheChanged: true, + cacheDisabledB: true, + settingDiffs: [{ key: "cachingStrategy", status: "changed" }], + }), + ); + expect(summary).toBe("cache disabled"); + }); + + test("returns an empty string when nothing structural changed", () => { + expect(summarizeTaskChange(taskDiff())).toBe(""); + }); +}); + +describe("summarizeIoChange()", () => { + test("calls out a rewired producing task alongside other changed fields", () => { + const summary = summarizeIoChange( + ioDiff({ + fieldDiffs: [ + { key: "source", status: "changed" }, + { key: "type", status: "changed" }, + ], + }), + ); + expect(summary).toBe("source rewired · 1 field changed"); + }); + + test("reports a rewire on its own when nothing else changed", () => { + const summary = summarizeIoChange( + ioDiff({ fieldDiffs: [{ key: "source", status: "changed" }] }), + ); + expect(summary).toBe("source rewired"); + }); + + test("counts and pluralizes changed fields", () => { + const summary = summarizeIoChange( + ioDiff({ + fieldDiffs: [ + { key: "type", status: "changed" }, + { key: "default", status: "changed" }, + { key: "description", status: "unchanged" }, + ], + }), + ); + expect(summary).toBe("2 fields changed"); + }); + + test("returns an empty string with no changed fields", () => { + expect(summarizeIoChange(ioDiff({ fieldDiffs: [] }))).toBe(""); + }); + + test("names an artifact difference the spec cannot show", () => { + expect( + summarizeIoChange(ioDiff({ fieldDiffs: [], artifactStatus: "changed" })), + ).toBe("artifact differs"); + expect( + summarizeIoChange(ioDiff({ fieldDiffs: [], artifactStatus: "lost" })), + ).toBe("artifact only in A"); + expect( + summarizeIoChange(ioDiff({ fieldDiffs: [], artifactStatus: "new" })), + ).toBe("artifact only in B"); + expect( + summarizeIoChange( + ioDiff({ fieldDiffs: [], artifactStatus: "unchanged" }), + ), + ).toBe(""); + }); +}); diff --git a/src/routes/v2/pages/CompareView/utils/summarizeChange.ts b/src/routes/v2/pages/CompareView/utils/summarizeChange.ts new file mode 100644 index 000000000..7d7652f88 --- /dev/null +++ b/src/routes/v2/pages/CompareView/utils/summarizeChange.ts @@ -0,0 +1,72 @@ +import type { DiffStatus } from "@/utils/diffStatus"; +import { pluralize } from "@/utils/string"; + +import { countChanged, type IoDiff, type TaskDiff } from "./comparePipelines"; + +function counted(count: number, noun: string): string { + return `${count} ${pluralize(count, noun)}`; +} + +/** + * Short human summary of how a changed task differs between the two runs, used + * as a subdued caption on graph nodes. Returns an empty string when there is + * nothing structural to report (e.g. an outcome-only difference), letting the + * caller decide whether to render anything. + */ +export function summarizeTaskChange(diff: TaskDiff): string { + const parts: string[] = []; + + if (diff.componentChanged) parts.push("component"); + + const changedArguments = countChanged(diff.argumentDiffs); + if (changedArguments > 0) parts.push(counted(changedArguments, "argument")); + + const changedAnnotations = countChanged(diff.annotationDiffs); + if (changedAnnotations > 0) { + parts.push(counted(changedAnnotations, "annotation")); + } + + if (diff.cacheChanged) { + parts.push(diff.cacheDisabledB ? "cache disabled" : "cache enabled"); + } + + const changedSettings = diff.settingDiffs.filter( + (entry) => + entry.status !== "unchanged" && + !(diff.cacheChanged && entry.key === "cachingStrategy"), + ).length; + if (changedSettings > 0) parts.push(counted(changedSettings, "setting")); + + return parts.join(" · "); +} + +const ARTIFACT_CHANGE_LABEL: Partial> = { + changed: "artifact differs", + lost: "artifact only in A", + new: "artifact only in B", +}; + +/** + * Short human summary of how a changed pipeline input/output differs. A rewired + * producing task is called out explicitly, alongside the count of any other + * fields that also changed, and a difference in the artifact the run actually + * produced — which is invisible in the spec — is named on its own. + */ +export function summarizeIoChange(diff: IoDiff): string { + const changedFields = diff.fieldDiffs.filter( + (entry) => entry.status !== "unchanged", + ); + const rewired = changedFields.some((entry) => entry.key === "source"); + const otherFields = rewired ? changedFields.length - 1 : changedFields.length; + + const parts: string[] = []; + if (rewired) parts.push("source rewired"); + if (otherFields > 0) parts.push(`${counted(otherFields, "field")} changed`); + + const artifactLabel = diff.artifactStatus + ? ARTIFACT_CHANGE_LABEL[diff.artifactStatus] + : undefined; + if (artifactLabel) parts.push(artifactLabel); + + return parts.join(" · "); +}