fix: review round — incremental crossing search and cached routes, retry the leftover draft save, drop the pipeline row when its last draft goes

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-10-01 11:31:36 +02:00
co-authored by Claude Opus 5.5
parent 8167cadbdf
commit 4b5bdd2aec
8 changed files with 115 additions and 65 deletions
@@ -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<string, EdgeRoute> } | 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
@@ -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
}
}
@@ -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)
@@ -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: [] }),
@@ -51,6 +51,7 @@ function makeHelpers(overrides: Partial<PipelineAIChatHelpers> = {}): {
}
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', () => {
@@ -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<PipelineToolHelpers>[] = [
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'
@@ -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<string | undefined> {
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()
@@ -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,