From 0b4fafc02e37357f6b587572a8be78da7c06aafc Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Thu, 1 Oct 2026 14:13:33 +0200 Subject: [PATCH] feat: AI gives every pipeline node a succinct summary Co-Authored-By: Claude Opus 5.5 (1M context) --- .../assets/AssetGraph/pipelineAiHelpers.ts | 16 ++- .../copilot/chat/pipeline/core.test.ts | 85 ++++++++++-- .../components/copilot/chat/pipeline/core.ts | 121 +++++++++++++----- 3 files changed, 176 insertions(+), 46 deletions(-) diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts index 9f2cb4359f..e360a6b577 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineAiHelpers.ts @@ -78,14 +78,15 @@ export function makePipelineScript( language: ScriptLang, scriptPath: string, content: string, - createdAt: string + createdAt: string, + summary = '' ): Script { // Cast through unknown: a local draft only needs path/language/content/schema; // the many readonly deployment fields on Script don't matter until createScript. return { hash: '', path: scriptPath, - summary: '', + summary, description: '', content, schema: emptySchema(), @@ -209,7 +210,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC path: r.path, language: draft?.script.language, unsaved: r.unsaved ?? false, - summary: draft?.script.summary || undefined, + summary: (draft ? draft.script.summary : r.summary) || undefined, writes: [...new Set(writes)], reads: [...new Set(reads)], triggers: [...new Set(triggers)] @@ -262,7 +263,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC return undefined } }, - proposeNode: async ({ path, language, content, outputKind }) => { + proposeNode: async ({ path, language, content, outputKind, summary }) => { deps.ensureEditable?.() // build_pipeline_node creates a NEW node in the OPEN folder. Reject a path // outside the folder (it would silently stage into this folder's bundle) @@ -318,7 +319,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC const next = new Map(drafts) next.set(path, { localId: deps.newDraftLocalId(), - script: makePipelineScript(language, path, content, new Date().toISOString()), + script: makePipelineScript(language, path, content, new Date().toISOString(), summary), outputAssets, inputAssets: inferred.reads }) @@ -336,7 +337,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC ) } }, - editNode: async (path, content) => { + editNode: async (path, content, summary) => { deps.ensureEditable?.() assertInFolder(path) assertPipelineAnnotation(content) @@ -369,7 +370,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC const next = new Map(drafts) next.set(path, { localId: existing?.localId ?? deps.newDraftLocalId(), - script: { ...baseScript, content }, + script: { ...baseScript, content, ...(summary !== undefined ? { summary } : {}) }, outputAssets, inputAssets: inferred.reads }) @@ -378,6 +379,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC deps.onProposeNode?.(path) // Deployable lineage only (see proposeNode): body writes + materialize target. return { + summary: summary ?? baseScript.summary ?? '', detectedReads: inferred.reads.map(assetUri), detectedWrites: dedupeAssets([...inferred.writes, ...materializeWrites(content)]).map( assetUri 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 f8918d792a..57f68df10b 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.test.ts @@ -61,9 +61,9 @@ function makeHelpers(overrides: Partial = {}): { calls.proposeNode = [...(calls.proposeNode ?? []), [input]] return { path: input.path, detectedReads: [], detectedWrites: [] } }, - editNode: async (path, content) => { - calls.editNode = [...(calls.editNode ?? []), [path, content]] - return { detectedReads: [], detectedWrites: [] } + editNode: async (path, content, summary) => { + calls.editNode = [...(calls.editNode ?? []), [path, content, summary]] + return { summary: summary ?? 'Existing summary', detectedReads: [], detectedWrites: [] } }, removeProposedNode: record('removeProposedNode'), unconfiguredTriggers: async () => [], @@ -109,6 +109,7 @@ describe('pipeline tools', () => { args: { path: 'f/analytics/clean', language: 'bun', + summary: 'Test node', content: '// pipeline\nexport async function main() {}', output_kind: 'ducklake' }, @@ -136,6 +137,7 @@ describe('pipeline tools', () => { args: { path: 'f/analytics/ingest', language: 'python3', + summary: 'Test node', content: '# pipeline\n# on schedule\n# on kafka\ndef main():\n return 1', triggers: [{ kind: 'schedule', config: { schedule: '0 0 6 * * *', timezone: 'UTC' } }] }, @@ -161,6 +163,7 @@ describe('pipeline tools', () => { args: { path: 'f/analytics/ingest', language: 'python3', + summary: 'Test node', content: '# pipeline\n# on schedule\ndef main():\n return 1', triggers: [{ kind: 'schedule', config: { timezone: 'UTC' } }] }, @@ -182,7 +185,12 @@ describe('pipeline tools', () => { }) }) const out = await toolByName('build_pipeline_node').fn({ - args: { path: 'f/analytics/clean', language: 'duckdb', content: '-- pipeline' }, + args: { + path: 'f/analytics/clean', + language: 'duckdb', + content: '-- pipeline', + summary: 'Test node' + }, workspace: 'w', helpers, toolCallbacks: noopCallbacks(), @@ -197,7 +205,12 @@ describe('pipeline tools', () => { proposeNode: async (input) => ({ path: input.path, detectedReads: [], detectedWrites: [] }) }) const out = await toolByName('build_pipeline_node').fn({ - args: { path: 'f/analytics/clean', language: 'duckdb', content: '-- pipeline' }, + args: { + path: 'f/analytics/clean', + language: 'duckdb', + content: '-- pipeline', + summary: 'Test node' + }, workspace: 'w', helpers, toolCallbacks: noopCallbacks(), @@ -221,6 +234,34 @@ describe('pipeline tools', () => { expect(calls.editNode?.[0]?.[1]).toContain('const x = 42') }) + it('edit_pipeline_node renames a node by summary alone, and asks for one when missing', async () => { + const body = { language: 'bun' as const, content: 'const x = 1\n' } + const { helpers, calls } = makeHelpers({ getNodeBody: async () => body }) + const run = (args: Record) => + toolByName('edit_pipeline_node').fn({ + args: { path: 'f/analytics/orders', ...args }, + workspace: 'w', + helpers, + toolCallbacks: noopCallbacks(), + toolId: 't' + }) + await run({ summary: 'Clean orders' }) + expect(calls.editNode?.[0]).toEqual(['f/analytics/orders', body.content, 'Clean orders']) + + const bare = makeHelpers({ + getNodeBody: async () => body, + editNode: async () => ({ summary: '', detectedReads: [], detectedWrites: [] }) + }) + const out = await toolByName('edit_pipeline_node').fn({ + args: { path: 'f/analytics/orders', old_string: 'x = 1', new_string: 'x = 2' }, + workspace: 'w', + helpers: bare.helpers, + toolCallbacks: noopCallbacks(), + toolId: 't' + }) + expect(out).toContain('has no summary') + }) + it('edit_pipeline_node surfaces a clear error when old_string is absent', async () => { const { helpers } = makeHelpers({ getNodeBody: async () => ({ language: 'bun', content: 'const x = 1\n' }) @@ -239,7 +280,7 @@ describe('pipeline tools', () => { it('mutation tools fail clearly when no pipeline editor is registered', async () => { await expect( toolByName('build_pipeline_node').fn({ - args: { path: 'f/a/b', language: 'bun', content: 'x' }, + args: { path: 'f/a/b', language: 'bun', content: 'x', summary: 'Test node' }, workspace: 'w', helpers: {}, toolCallbacks: noopCallbacks(), @@ -295,19 +336,47 @@ describe('pipeline tools with several editors open', () => { toolId: 't' }) + it('remounts an opened folder whose editor another tab replaced', async () => { + const built: string[] = [] + const mounted: PipelineAIChatHelpers[] = [] + const ensured: string[] = [] + await toolByName('build_pipeline_node').fn({ + args: { + path: 'f/sales/clean', + language: 'bun', + content: '// pipeline', + summary: 'Test node' + }, + workspace: 'w', + helpers: { + pipelines: () => mounted, + pipelineFolders: () => ['sales'], + ensurePipeline: async (folder: string) => { + ensured.push(folder) + mounted.push(editorFor(folder, built)) + return mounted[0] + } + }, + toolCallbacks: noopCallbacks(), + toolId: 't' + }) + expect(ensured).toEqual(['sales']) + expect(built).toEqual(['sales:f/sales/clean']) + }) + it('routes a node to the editor of its folder', async () => { const built: string[] = [] const editors = [editorFor('crm', built), editorFor('sales', built)] await call( 'build_pipeline_node', - { path: 'f/sales/clean', language: 'bun', content: '// pipeline' }, + { path: 'f/sales/clean', language: 'bun', content: '// pipeline', summary: 'Test node' }, editors ) expect(built).toEqual(['sales:f/sales/clean']) await expect( call( 'build_pipeline_node', - { path: 'f/other/clean', language: 'bun', content: '// pipeline' }, + { path: 'f/other/clean', language: 'bun', content: '// pipeline', summary: 'Test node' }, editors ) ).rejects.toThrow(/open_preview\(kind="pipeline", path="other"\)/) diff --git a/frontend/src/lib/components/copilot/chat/pipeline/core.ts b/frontend/src/lib/components/copilot/chat/pipeline/core.ts index cdd7e79eac..f47decce08 100644 --- a/frontend/src/lib/components/copilot/chat/pipeline/core.ts +++ b/frontend/src/lib/components/copilot/chat/pipeline/core.ts @@ -77,13 +77,15 @@ export interface PipelineAIChatHelpers { language: ScriptLang content: string outputKind?: PipelineOutputKind + summary?: string }) => Promise<{ path: string; detectedReads: string[]; detectedWrites: string[] }> /** Replace an existing node's body, applied as an unsaved draft. Returns the * re-inferred asset lineage (URIs) so the caller sees the effect of the edit. */ editNode: ( path: string, - content: string - ) => Promise<{ detectedReads: string[]; detectedWrites: string[] }> + content: string, + summary?: string + ) => Promise<{ summary: string; detectedReads: string[]; detectedWrites: string[] }> /** Discard the unsaved draft at a path (undo a build_pipeline_node). */ removeProposedNode: (path: string) => Promise /** Trigger kinds the node declares (`on `) that no trigger row or draft fills. */ @@ -101,7 +103,25 @@ export interface PipelineAIChatHelpers { /** Helper bag the pipeline tools receive from the manager in global mode: every * pipeline editor currently open, one per folder. */ -export type PipelineToolHelpers = { pipelines?: () => readonly PipelineAIChatHelpers[] } +export type PipelineToolHelpers = { + pipelines?: () => readonly PipelineAIChatHelpers[] + /** Folders the chat may work on, including ones whose editor is not mounted now. */ + pipelineFolders?: () => readonly string[] + /** Remounts the folder's editor (a session reopens its preview tab) and resolves + * to it, or `undefined` when it cannot be brought back. */ + ensurePipeline?: (folder: string) => Promise +} + +/** The folder's editor, remounted when the chat opened it before and something else + * has taken its tab since. */ +async function remount( + helpers: PipelineToolHelpers, + folder: string | undefined +): Promise { + if (!folder || !helpers?.ensurePipeline) return undefined + if (!(helpers.pipelineFolders?.() ?? []).includes(folder)) return undefined + return helpers.ensurePipeline(folder) +} function openPipelines(helpers: PipelineToolHelpers): readonly PipelineAIChatHelpers[] { const open = helpers?.pipelines?.() ?? [] @@ -123,11 +143,15 @@ function notOpenError(folder: string, open: readonly PipelineAIChatHelpers[]): E /** The editor of the pipeline a node path belongs to (`f//`). With a * single editor open, a path outside it still goes to that editor, whose own check * names the corrected path. */ -function pipelineForPath(helpers: PipelineToolHelpers, path: string): PipelineAIChatHelpers { - const open = openPipelines(helpers) +async function pipelineForPath( + helpers: PipelineToolHelpers, + path: string +): Promise { const folder = path.match(/^f\/([^/]+)\//)?.[1] - const match = folder ? open.find((p) => p.getFolder() === folder) : undefined + const live = folder ? helpers?.pipelines?.().find((p) => p.getFolder() === folder) : undefined + const match = live ?? (await remount(helpers, folder)) if (match) return match + const open = openPipelines(helpers) if (open.length === 1) return open[0] throw folder ? notOpenError(folder, open) @@ -136,15 +160,19 @@ function pipelineForPath(helpers: PipelineToolHelpers, path: string): PipelineAI ) } -function pipelineForFolder( +async function pipelineForFolder( helpers: PipelineToolHelpers, folder: string | undefined -): PipelineAIChatHelpers { +): Promise { + const known = helpers?.pipelineFolders?.() ?? [] + const name = folder ? normalizePipelineFolder(folder) : known.length === 1 ? known[0] : undefined + const live = name ? helpers?.pipelines?.().find((p) => p.getFolder() === name) : undefined + const remounted = live ?? (await remount(helpers, name)) + if (remounted) return remounted const open = openPipelines(helpers) if (folder) { - const name = normalizePipelineFolder(folder) const match = open.find((p) => p.getFolder() === name) - if (!match) throw notOpenError(name, open) + if (!match) throw notOpenError(normalizePipelineFolder(folder), open) return match } if (open.length === 1) return open[0] @@ -250,6 +278,15 @@ function unconfiguredTriggersNote(path: string, kinds: readonly string[]): strin return ` ACTION REQUIRED: '${path}' declares \`on ${kinds.join('`, `on ')}\` but has no ${kinds.join('/')} trigger yet, so nothing would start it once deployed. Call set_pipeline_trigger for it now (a schedule needs its cron and timezone; ask the user only if the cadence cannot be inferred).` } +const summaryField = z + .string() + .trim() + .min(1) + .max(80) + .describe( + 'Succinct summary of what the node does, a few words in sentence case with no trailing period (e.g. "Ingest Stripe charges", "Daily revenue by region"). It is the title the node shows in the workspace.' + ) + const buildPipelineNodeSchema = z.object({ path: z .string() @@ -264,6 +301,7 @@ const buildPipelineNodeSchema = z.object({ .describe( "Full script source. Start it with the `pipeline` annotation as a top-of-file comment in the LANGUAGE'S comment syntax — `-- pipeline` for SQL (duckdb/postgresql), `# pipeline` for python3/bash, `// pipeline` for bun/TS — to mark it a pipeline member; declare inputs the same way (e.g. `-- on `), and write outputs via the wmill SDK / SQL so the lineage edges are inferred. A `// pipeline` line in a SQL node is a syntax error. Read existing node bodies first to match conventions." ), + summary: summaryField, output_kind: outputKindSchema.optional(), triggers: z .array(nodeTriggerSchema) @@ -282,8 +320,17 @@ const buildPipelineNodeToolDef = createToolDef( const editPipelineNodeSchema = z.object({ path: z.string().describe('Workspace path of the node to edit.'), - old_string: z.string().min(1).describe("Exact text to find in the node's current source."), - new_string: z.string().describe('Replacement text.'), + old_string: z + .string() + .min(1) + .optional() + .describe("Exact text to find in the node's current source. Omit to change only the summary."), + new_string: z.string().optional().describe('Replacement text.'), + summary: summaryField + .optional() + .describe( + 'New succinct summary (a few words, no trailing period). Pass it whenever the node has none or the edit changes what the node does.' + ), replace_all: z .boolean() .optional() @@ -354,7 +401,7 @@ export const pipelineTools: SessionTool[] = [ planModeSafe: true, fn: async ({ args, helpers, toolId, toolCallbacks }) => { const { folder } = getPipelineGraphSchema.parse(args ?? {}) - const pipeline = pipelineForFolder(helpers, folder) + const pipeline = await pipelineForFolder(helpers, folder) toolCallbacks.setToolStatus(toolId, { content: 'Reading pipeline graph...' }) const ctx = pipeline.getPipelineContext() toolCallbacks.setToolStatus(toolId, { @@ -370,7 +417,7 @@ export const pipelineTools: SessionTool[] = [ planModeSafe: true, fn: async ({ args, helpers, toolId, toolCallbacks }) => { const { path } = readPipelineNodeSchema.parse(args) - const pipeline = pipelineForPath(helpers, path) + const pipeline = await pipelineForPath(helpers, path) toolCallbacks.setToolStatus(toolId, { content: `Reading node '${path}'...` }) const node = await pipeline.getNodeBody(path) if (!node) { @@ -387,9 +434,9 @@ export const pipelineTools: SessionTool[] = [ showDetails: true, showFade: true, fn: async ({ args, helpers, toolId, toolCallbacks }) => { - const { path, language, content, output_kind, triggers } = + const { path, language, content, summary, output_kind, triggers } = buildPipelineNodeSchema.parse(args) - const pipeline = pipelineForPath(helpers, path) + const pipeline = await pipelineForPath(helpers, path) // Validated before the node is staged, so a bad config leaves nothing half-built. const configs = (triggers ?? []).map((t) => ({ kind: t.kind, @@ -400,6 +447,7 @@ export const pipelineTools: SessionTool[] = [ path, language: language as ScriptLang, content, + summary, outputKind: output_kind as PipelineOutputKind | undefined }) const attached: string[] = [] @@ -425,27 +473,38 @@ export const pipelineTools: SessionTool[] = [ showDetails: true, showFade: true, fn: async ({ args, helpers, toolId, toolCallbacks }) => { - const { path, old_string, new_string, replace_all } = editPipelineNodeSchema.parse(args) - const pipeline = pipelineForPath(helpers, path) + const { path, old_string, new_string, replace_all, summary } = + editPipelineNodeSchema.parse(args) + if (old_string === undefined && summary === undefined) { + throw new Error('Pass old_string/new_string to change the source, summary to rename it, or both.') + } + const pipeline = await pipelineForPath(helpers, path) const node = await pipeline.getNodeBody(path) if (!node) { return `No pipeline node found at '${path}'. Call get_pipeline_graph to list the available nodes.` } toolCallbacks.setToolStatus(toolId, { content: `Editing node '${path}'...` }) - const updated = findAndReplace( - node.content, - old_string, - new_string, - replace_all ?? false, - 'node source' - ) - const { detectedReads, detectedWrites } = await pipeline.editNode(path, updated) + const updated = + old_string === undefined + ? node.content + : findAndReplace( + node.content, + old_string, + new_string ?? '', + replace_all ?? false, + 'node source' + ) + const edited = await pipeline.editNode(path, updated, summary) + const { detectedReads, detectedWrites } = edited + const summaryNote = edited.summary.trim() + ? '' + : ` '${path}' has no summary: call edit_pipeline_node with just \`summary\` (a few words saying what it does).` recordPipelineModified(toolCallbacks, pipeline) toolCallbacks.setToolStatus(toolId, { content: `Edited draft '${path}'`, result: 'Success' }) - return `Pipeline node '${path}' updated as an unsaved draft on the canvas (not deployed).${inferredLineageNote(detectedReads, detectedWrites)}${unconfiguredTriggersNote(path, await pipeline.unconfiguredTriggers(path))}` + return `Pipeline node '${path}' updated as an unsaved draft on the canvas (not deployed).${inferredLineageNote(detectedReads, detectedWrites)}${summaryNote}${unconfiguredTriggersNote(path, await pipeline.unconfiguredTriggers(path))}` } }, { @@ -454,7 +513,7 @@ export const pipelineTools: SessionTool[] = [ showDetails: true, fn: async ({ args, helpers, toolId, toolCallbacks }) => { const { path, kind, config } = setPipelineTriggerSchema.parse(args) - const pipeline = pipelineForPath(helpers, path) + const pipeline = await pipelineForPath(helpers, path) toolCallbacks.setToolStatus(toolId, { content: `Setting the ${kind} trigger of '${path}'...` }) const valid = validateTriggerConfig(kind, config, path) const { path: triggerPath, replaced } = await pipeline.setNodeTrigger(path, kind, valid) @@ -471,7 +530,7 @@ export const pipelineTools: SessionTool[] = [ def: removePipelineNodeToolDef, fn: async ({ args, helpers, toolId, toolCallbacks }) => { const { path } = removePipelineNodeSchema.parse(args) - const pipeline = pipelineForPath(helpers, path) + const pipeline = await pipelineForPath(helpers, path) toolCallbacks.setToolStatus(toolId, { content: `Discarding draft '${path}'...` }) await pipeline.removeProposedNode(path) // With the last draft gone there is nothing left to deploy: the pipeline @@ -498,7 +557,7 @@ export const pipelineTools: SessionTool[] = [ autoCollapseDetails: false, fn: async ({ args, workspace, helpers, toolId, toolCallbacks }) => { const { path, args: runArgs } = testPipelineNodeSchema.parse(args) - const pipeline = pipelineForPath(helpers, path) + const pipeline = await pipelineForPath(helpers, path) return executeTestRun({ jobStarter: async () => { const jobId = await pipeline.testNode(path, runArgs ?? undefined) @@ -554,7 +613,7 @@ Data Pipeline editor (ACTIVE): - ${pathLine}${ canWriteDraft ? ` -- Build new nodes with build_pipeline_node and edit existing ones with edit_pipeline_node. These apply directly as unsaved drafts on the canvas (like the flow/script editor applies AI edits) — they DO NOT deploy. There is no separate Accept/Reject step. Prefer these over the generic write_script/edit_script draft tools while a pipeline is open. +- Build new nodes with build_pipeline_node and edit existing ones with edit_pipeline_node. Give every node a succinct \`summary\` (a few words saying what it does, e.g. "Ingest Stripe charges"), and update it when an edit changes what the node does. These apply directly as unsaved drafts on the canvas (like the flow/script editor applies AI edits) — they DO NOT deploy. There is no separate Accept/Reject step. Prefer these over the generic write_script/edit_script draft tools while a pipeline is open. - Reuse existing asset paths from the graph when wiring a downstream node to an upstream one (read the upstream's write asset, then \`// on\` that same URI). - TRIGGERS: an ingestion node (one with no upstream asset to react to) must declare how it starts, e.g. \`// on schedule\`, and every \`on \` line needs its trigger: pass it in build_pipeline_node's \`triggers\` (or call set_pipeline_trigger) in the same step you build the node. A schedule needs a 6-field cron and an IANA timezone — pick a sensible cadence from the request (e.g. daily at 06:00 UTC for a daily sync) rather than leaving it empty; for the other kinds call get_trigger_schema(kind) for the config fields and ask the user for any connection/resource you cannot find. Never leave a declared trigger without one: the node would deploy with nothing to start it. \`webhook\` and \`data_upload\` need no trigger. - Only deploy when the user explicitly asks; the user deploys drafts from the canvas.`