From 4b5bdd2aecae61598dcabf60b590249f9307f5b4 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Thu, 1 Oct 2026 11:31:36 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20review=20round=20=E2=80=94=20incremental?= =?UTF-8?q?=20crossing=20search=20and=20cached=20routes,=20retry=20the=20l?= =?UTF-8?q?eftover=20draft=20save,=20drop=20the=20pipeline=20row=20when=20?= =?UTF-8?q?its=20last=20draft=20goes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 (1M context) --- .../assets/AssetGraph/AssetGraphCanvas.svelte | 42 +++++++++------ .../AssetGraph/assetGraphEdgeRouting.ts | 36 +++++++------ .../assets/AssetGraph/pipelineAiHelpers.ts | 3 ++ .../copilot/chat/AIChatManager.test.ts | 1 + .../copilot/chat/pipeline/core.test.ts | 34 ++++++++----- .../components/copilot/chat/pipeline/core.ts | 12 ++++- .../sessions/PipelineEditorView.svelte | 51 ++++++++++++------- .../(logged)/pipeline/[folder]/+page.svelte | 1 + 8 files changed, 115 insertions(+), 65 deletions(-) diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 31fd10030a..a16f6c8ec1 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -30,7 +30,7 @@ PIPELINE_NODE_HEADER, PIPELINE_NODE_HEADER_CARD_PAD } from './assetGraphLayout' - import { detourLane, routeAssetGraphEdges } from './assetGraphEdgeRouting' + import { detourLane, routeAssetGraphEdges, type EdgeRoute } from './assetGraphEdgeRouting' import { assetsOnlyView, upstreamDeletion, @@ -1877,23 +1877,33 @@ return s && t ? detourLane(s, t, routeNodes, ROUTE_PAD) : undefined } - let edgeRoutes = $derived( - routeAssetGraphEdges( - routeNodes, - view.edges - .filter((e) => e.kind !== 'add-anchor') - .map((e) => ({ - id: e.id, - source: e.source, - target: e.target, - laneX: detourForEdge(e.source, e.target), - // What sets how an edge is drawn (see the styling below): only - // edges that look the same may share a segment. - style: `${e.kind}:${e.reactive ? 1 : 0}:${e.missing ? 1 : 0}` - })), + // Routing depends on geometry and edges only. The nodes are rebuilt on every + // selection or run update, so the routes are kept until either really changes. + let routeInput = $derived.by(() => { + const edges = view.edges + .filter((e) => e.kind !== 'add-anchor') + .map((e) => ({ + id: e.id, + source: e.source, + target: e.target, + // What sets how an edge is drawn (see the styling below): only + // edges that look the same may share a segment. + style: `${e.kind}:${e.reactive ? 1 : 0}:${e.missing ? 1 : 0}` + })) + return { nodes: routeNodes, edges, key: JSON.stringify([routeNodes, edges]) } + }) + let routedFor: { key: string; routes: Map } | undefined + let edgeRoutes = $derived.by(() => { + const { nodes, edges, key } = routeInput + if (routedFor?.key === key) return routedFor.routes + const routes = routeAssetGraphEdges( + nodes, + edges.map((e) => ({ ...e, laneX: detourForEdge(e.source, e.target) })), LAYER_GAP ) - ) + routedFor = { key, routes } + return routes + }) let flowEdges = $derived.by(() => view.edges diff --git a/frontend/src/lib/components/assets/AssetGraph/assetGraphEdgeRouting.ts b/frontend/src/lib/components/assets/AssetGraph/assetGraphEdgeRouting.ts index 0e7794e291..c0e1c76f0c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/assetGraphEdgeRouting.ts +++ b/frontend/src/lib/components/assets/AssetGraph/assetGraphEdgeRouting.ts @@ -518,33 +518,37 @@ function assignTracks(list: Net[]): number[] { b.topPins.filter((x) => spans(a, x)).length ) ) - const total = (order: number[]) => { - let sum = 0 - for (let i = 0; i < order.length; i++) - for (let j = i + 1; j < order.length; j++) sum += cost[order[i]][order[j]] - return sum - } // Start from the order each pair prefers on its own, then move one line at a - // time to wherever it removes crossings, until no move helps. + // time to wherever it removes the most crossings. Moving a line only changes + // the pairs it is part of, so every position is scored in one sweep: linear + // per line, quadratic per pass — a graph of hundreds of lines routes at once. let order = list .map((_, i) => i) .sort((i, j) => cost[i][j] - cost[j][i] || list[i].from - list[j].from) - for (let improved = true; improved; ) { + const MAX_PASSES = 8 + for (let pass = 0, improved = true; improved && pass < MAX_PASSES; pass++) { improved = false for (let k = 0; k < order.length; k++) { + const v = order[k] const rest = order.filter((_, i) => i !== k) - let best = order - let bestCost = total(order) - for (let pos = 0; pos <= rest.length; pos++) { - const tried = [...rest.slice(0, pos), order[k], ...rest.slice(pos)] - const c = total(tried) + // Crossings involving v with v first: v above every other line. + let c = 0 + for (const j of rest) c += cost[v][j] + let best = 0 + let bestCost = c + let current = c + for (let pos = 0; pos < rest.length; pos++) { + // Moving v below rest[pos] swaps that one pair. + c += cost[rest[pos]][v] - cost[v][rest[pos]] + if (pos + 1 === k) current = c if (c < bestCost) { - best = tried + best = pos + 1 bestCost = c } } - if (best !== order) { - order = best + if (k === 0) current = rest.reduce((a, j) => a + cost[v][j], 0) + if (bestCost < current) { + order = [...rest.slice(0, best), v, ...rest.slice(best)] improved = true } } diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts index fef664602c..bde960b06d 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts @@ -62,6 +62,8 @@ export type PipelineAiHelperDeps = { onForgetPath?: (path: string) => void /** Notify the caller a test run started so it can light up its run UI. */ onRunStarted?: (jobId: string, path: string) => void + /** Whether the folder has trigger drafts, which live beside the script drafts. */ + hasTriggerDrafts?: () => boolean } export function makePipelineScript( @@ -215,6 +217,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC const helpers: PipelineAIChatHelpers = { getFolder: folderName, + hasDrafts: () => deps.getDrafts().size > 0 || (deps.hasTriggerDrafts?.() ?? false), getPipelineContext: buildContext, getNodeBody: async (path) => { const draft = deps.getDrafts().get(path) diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts index d83211b1e5..215858bd4e 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts @@ -4330,6 +4330,7 @@ describe('AIChatManager.waitForPipelineHelpers', () => { function fakePipelineHelpers(): PipelineAIChatHelpers { return { getFolder: () => 'f', + hasDrafts: () => false, getPipelineContext: () => ({ folder: 'f', mode: 'edit', nodes: [], assets: [] }), getNodeBody: async () => undefined, proposeNode: async () => ({ path: '', detectedReads: [], detectedWrites: [] }), diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts index 1c80b4aea8..c545fab85a 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts @@ -51,6 +51,7 @@ function makeHelpers(overrides: Partial = {}): { } const pipeline: PipelineAIChatHelpers = { getFolder: () => sampleContext.folder, + hasDrafts: () => false, getPipelineContext: () => sampleContext, getNodeBody: async (path: string) => { calls.getNodeBody = [...(calls.getNodeBody ?? []), [path]] @@ -198,20 +199,25 @@ describe('pipeline tools', () => { ).rejects.toThrow(/No pipeline editor is open/) }) - it('remove_pipeline_node records no change: undoing a draft leaves nothing to deploy', async () => { - const { helpers } = makeHelpers() - const modified: string[] = [] - await toolByName('remove_pipeline_node').fn({ - args: { path: 'f/analytics/clean' }, - workspace: 'w', - helpers, - toolCallbacks: { - ...noopCallbacks(), - onItemModified: (kind, path) => modified.push(`${kind}:${path}`) - }, - toolId: 't' - }) - expect(modified).toEqual([]) + it('remove_pipeline_node drops the pipeline from the changes once its last draft goes', async () => { + const run = async (hasDrafts: boolean) => { + const { helpers } = makeHelpers({ hasDrafts: () => hasDrafts }) + const log: string[] = [] + await toolByName('remove_pipeline_node').fn({ + args: { path: 'f/analytics/clean' }, + workspace: 'w', + helpers, + toolCallbacks: { + ...noopCallbacks(), + onItemModified: (kind, path) => log.push(`modified ${kind}:${path}`), + onItemDiscarded: (kind, path) => log.push(`discarded ${kind}:${path}`) + }, + toolId: 't' + }) + return log + } + expect(await run(false)).toEqual(['discarded data_pipeline:f/analytics/data_pipeline']) + expect(await run(true)).toEqual(['modified data_pipeline:f/analytics/data_pipeline']) }) it('test_pipeline_node requires confirmation', () => { diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.ts index 8c29fbcbeb..c48517335e 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.ts @@ -61,6 +61,8 @@ export type PipelineContext = { export interface PipelineAIChatHelpers { /** Folder name (no `f/` prefix) of the pipeline this editor shows. */ getFolder: () => string + /** Whether the folder has any script or trigger draft left. */ + hasDrafts: () => boolean getPipelineContext: () => PipelineContext /** Read a node's source (the in-flight draft body if one exists, else deployed). */ getNodeBody: (path: string) => Promise<{ language: ScriptLang; content: string } | undefined> @@ -366,9 +368,15 @@ export const pipelineTools: SessionTool[] = [ const { path } = removePipelineNodeSchema.parse(args) const pipeline = pipelineForPath(helpers, path) toolCallbacks.setToolStatus(toolId, { content: `Discarding draft '${path}'...` }) - // Not recorded: undoing a draft changes nothing to deploy, and the last - // one gone would leave the pipeline listed as this chat's change. await pipeline.removeProposedNode(path) + // With the last draft gone there is nothing left to deploy: the pipeline + // is no longer this chat's change, and listing it would show it deployed. + if (pipeline.hasDrafts()) recordPipelineModified(toolCallbacks, pipeline) + else + toolCallbacks.onItemDiscarded?.( + PIPELINE_DRAFT_KIND, + pipelineBundlePath(pipeline.getFolder()) + ) toolCallbacks.setToolStatus(toolId, { content: `Discarded draft '${path}'`, result: 'Success' diff --git a/frontend/src/lib/components/sessions/PipelineEditorView.svelte b/frontend/src/lib/components/sessions/PipelineEditorView.svelte index 20f0e91675..aabb211b67 100644 --- a/frontend/src/lib/components/sessions/PipelineEditorView.svelte +++ b/frontend/src/lib/components/sessions/PipelineEditorView.svelte @@ -120,6 +120,7 @@ getDrafts: () => pe.drafts, setDrafts: (next) => (pe.drafts = next), newDraftLocalId: pe.newDraftLocalId, + hasTriggerDrafts: () => pe.triggerDrafts.size > 0, onForgetPath: (p) => { pe.forgetPath(p) pe.discardTriggerDraftsFor(p) @@ -328,7 +329,12 @@ deploying = true try { if (!(await hydrated())) return { success: false, error: 'The pipeline drafts did not load.' } - if (pendingCount === 0) return { success: true } + if (pendingCount === 0) { + // Nothing left to deploy, but an earlier deploy may have failed to save + // that: the server would still hold the deployed drafts. + const unsaved = await saveRemaining() + return unsaved ? { success: false, error: unsaved } : { success: true } + } if (pe.triggerDrafts.size > 0 && !(await confirmTriggers())) { return { success: false, error: 'Deploy cancelled.' } } @@ -353,23 +359,9 @@ } // The session's changes list re-reads the draft once this resolves, so // what is left of it has to be saved by then. - await tick() - const draft = { - workspace: workspaceId, - itemKind: PIPELINE_DRAFT_KIND, - path: pipelineBundlePath(path) - } - await UserDraftDbSyncer.flush(draft) + const unsaved = await saveRemaining() deployErrors = errors - // A flush resolves even when the save failed: the server would keep the - // deployed drafts, and deploying them again would recreate their triggers. - const sync = UserDraftDbSyncer.getState(draft) - if (UserDraftDbSyncer.getConflict(draft).conflict || sync.state === 'failed') { - return { - success: false, - error: `Deployed, but the remaining drafts could not be saved: ${sync.failureMessage ?? 'they conflict with a newer version'}.` - } - } + if (unsaved) return { success: false, error: `Deployed, but ${unsaved}` } if (errors.size > 0) { deployErrorsOpen = true return { @@ -383,6 +375,31 @@ } } + // Saves what is left of the folder's draft and says why if that failed. A + // flush resolves even when the save failed, and the server would then keep + // drafts already deployed — deploying them again would recreate their + // triggers. After a failed save there may be nothing queued to flush, so an + // emptied draft is sent again. + async function saveRemaining(): Promise { + await tick() + const draft = { + workspace: workspaceId, + itemKind: PIPELINE_DRAFT_KIND, + path: pipelineBundlePath(path) + } + if (pendingCount === 0 && UserDraftDbSyncer.getState(draft).state === 'failed') { + await UserDraftDbSyncer.save({ ...draft, value: null, immediate: true }) + } else { + await UserDraftDbSyncer.flush(draft) + } + if (UserDraftDbSyncer.getConflict(draft).conflict) + return 'the remaining drafts conflict with a newer version.' + const sync = UserDraftDbSyncer.getState(draft) + if (sync.state === 'failed') + return `the remaining drafts could not be saved: ${sync.failureMessage ?? 'unknown error'}.` + return undefined + } + async function deployAllFromButton() { const count = pendingCount const res = await deployAll() diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index b546f0412f..88b311c7f7 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -497,6 +497,7 @@ getFolder: () => folder, getWorkspace: () => $workspaceStore, getResolvedGraph: () => graphWithDraft, + hasTriggerDrafts: () => pe.triggerDrafts.size > 0, getDrafts: () => pe.drafts, setDrafts: (next) => (pe.drafts = next), newDraftLocalId: pe.newDraftLocalId,