feat: pipelines in AI sessions with several folders open at once

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-09-30 20:24:40 +02:00
co-authored by Claude Opus 5.5
parent 53066d7239
commit a25f9cf199
22 changed files with 345 additions and 177 deletions
@@ -214,6 +214,7 @@ export function createPipelineAiHelpers(deps: PipelineAiHelperDeps): PipelineAIC
}
const helpers: PipelineAIChatHelpers = {
getFolder: folderName,
getPipelineContext: buildContext,
getNodeBody: async (path) => {
const draft = deps.getDrafts().get(path)
@@ -498,11 +498,12 @@ export class AIChatManager implements ChatViewHost {
skipResponsesApi = $state(false)
mode = $state<AIMode>(AIMode.NAVIGATOR)
pipelineAiChatHelpers = $state<PipelineAIChatHelpers | undefined>(undefined)
// One per mounted pipeline editor; a session can have several folders open.
#pipelineEditors = new Set<PipelineAIChatHelpers>()
// Resolved when a pipeline editor registers its tools. open_preview(pipeline)
// awaits this so the model's next build_pipeline_node call can't race ahead of
// the async canvas mount and hit "Unknown tool call".
#pipelineHelpersWaiters = new Set<() => void>()
#pipelineHelpersWaiters = new Set<(helpers: PipelineAIChatHelpers) => void>()
readonly isOpen = $derived(chatState.size > 0)
savedSize = $state<number>(0)
instructions = $state<string>('')
@@ -2485,7 +2486,7 @@ export class AIChatManager implements ChatViewHost {
// Public because it is purely local, unlike `changeMode(GLOBAL)`, which also
// fires the three network refreshes.
configureGlobalMode = () => {
const pipeline = this.pipelineAiChatHelpers
const hasPipelines = this.#pipelineEditors.size > 0
const opts = this.globalAssemblyOpts()
const baseHelpers: GlobalToolHelpers = {
// A session targets its own fixed (possibly forked) workspace, so capture it for
@@ -2517,7 +2518,9 @@ export class AIChatManager implements ChatViewHost {
}
}
this.#assembledTools = assembleGlobalTools(opts)
this.helpers = pipeline ? { ...baseHelpers, pipeline } : baseHelpers
this.helpers = hasPipelines
? { ...baseHelpers, pipelines: () => [...this.#pipelineEditors] }
: baseHelpers
this.systemMessage = assembleGlobalSystemMessage(getCustomPromptParts(AIMode.GLOBAL), opts)
this.syncArtifactsSession()
}
@@ -2530,7 +2533,7 @@ export class AIChatManager implements ChatViewHost {
mcpServers: this.mcpServers,
access: this.sessionAccess,
sessionContext: this.sessionContextResolver?.(),
pipelineContext: this.pipelineAiChatHelpers?.getPipelineContext()
pipelineFolders: [...new Set([...this.#pipelineEditors].map((p) => p.getFolder()))]
})
// Fetch the workspace's AI skills and, if GLOBAL mode is still active, rebuild
@@ -4862,19 +4865,17 @@ export class AIChatManager implements ChatViewHost {
// AI edits apply directly as drafts, so there is nothing to auto-accept.
// Returns a cleanup that tears the registration back down.
setPipelineHelpers = (pipelineHelpers: PipelineAIChatHelpers) => {
this.pipelineAiChatHelpers = pipelineHelpers
this.#pipelineEditors.add(pipelineHelpers)
untrack(() => {
if (this.mode === AIMode.GLOBAL) {
this.configureGlobalMode()
}
})
// The pipeline tools are now registered — release anything awaiting them.
const waiters = [...this.#pipelineHelpersWaiters]
this.#pipelineHelpersWaiters.clear()
waiters.forEach((resolve) => resolve())
;[...this.#pipelineHelpersWaiters].forEach((notify) => notify(pipelineHelpers))
return () => {
this.pipelineAiChatHelpers = undefined
this.#pipelineEditors.delete(pipelineHelpers)
untrack(() => {
if (this.mode === AIMode.GLOBAL) {
this.configureGlobalMode()
@@ -4884,15 +4885,16 @@ export class AIChatManager implements ChatViewHost {
}
/**
* Await the pipeline editor's tool registration. Resolves `true` immediately
* when a pipeline editor is already mounted, or when the next one registers;
* resolves `false` after `timeoutMs` if none registers (e.g. a backgrounded
* session whose preview tab is not mounted, or a closed tab). Callers must
* treat `false` as "tools NOT available" rather than silently reporting
* success — the open_preview handler surfaces that to the model.
* Await the tool registration of the pipeline editor for `folder` (any folder
* when unset). Resolves `true` immediately when one is already mounted, or when
* it registers; resolves `false` after `timeoutMs` if none registers (e.g. a
* session whose runtime is not mounted, or a closed tab). Callers must treat
* `false` as "tools NOT available" rather than silently reporting success —
* the open_preview handler surfaces that to the model.
*/
waitForPipelineHelpers = (timeoutMs = 8000): Promise<boolean> => {
if (this.pipelineAiChatHelpers) return Promise.resolve(true)
waitForPipelineHelpers = (folder?: string, timeoutMs = 8000): Promise<boolean> => {
const matches = (p: PipelineAIChatHelpers) => folder === undefined || p.getFolder() === folder
if ([...this.#pipelineEditors].some(matches)) return Promise.resolve(true)
return new Promise<boolean>((resolve) => {
let settled = false
const finish = (registered: boolean) => {
@@ -4901,7 +4903,9 @@ export class AIChatManager implements ChatViewHost {
this.#pipelineHelpersWaiters.delete(onRegister)
resolve(registered)
}
const onRegister = () => finish(true)
const onRegister = (p: PipelineAIChatHelpers) => {
if (matches(p)) finish(true)
}
this.#pipelineHelpersWaiters.add(onRegister)
setTimeout(() => finish(false), timeoutMs)
})
@@ -4329,6 +4329,7 @@ describe('DOM selector chips scoped by app path', () => {
describe('AIChatManager.waitForPipelineHelpers', () => {
function fakePipelineHelpers(): PipelineAIChatHelpers {
return {
getFolder: () => 'f',
getPipelineContext: () => ({ folder: 'f', mode: 'edit', nodes: [], assets: [] }),
getNodeBody: async () => undefined,
proposeNode: async () => ({ path: '', detectedReads: [], detectedWrites: [] }),
@@ -4341,13 +4342,13 @@ describe('AIChatManager.waitForPipelineHelpers', () => {
it('resolves true immediately when a pipeline editor is already registered', async () => {
const manager = new AIChatManager()
manager.setPipelineHelpers(fakePipelineHelpers())
await expect(manager.waitForPipelineHelpers(1000)).resolves.toBe(true)
await expect(manager.waitForPipelineHelpers(undefined, 1000)).resolves.toBe(true)
})
it('resolves true once a pipeline editor registers', async () => {
const manager = new AIChatManager()
let outcome: boolean | undefined
const wait = manager.waitForPipelineHelpers(1000).then((v) => (outcome = v))
const wait = manager.waitForPipelineHelpers(undefined, 1000).then((v) => (outcome = v))
await Promise.resolve()
expect(outcome).toBeUndefined()
manager.setPipelineHelpers(fakePipelineHelpers())
@@ -4355,12 +4356,24 @@ describe('AIChatManager.waitForPipelineHelpers', () => {
expect(outcome).toBe(true)
})
it('waits for the editor of the requested folder, not any editor', async () => {
const manager = new AIChatManager()
manager.setPipelineHelpers(fakePipelineHelpers())
let outcome: boolean | undefined
const wait = manager.waitForPipelineHelpers('sales', 1000).then((v) => (outcome = v))
await Promise.resolve()
expect(outcome).toBeUndefined()
manager.setPipelineHelpers({ ...fakePipelineHelpers(), getFolder: () => 'sales' })
await wait
expect(outcome).toBe(true)
})
// The false result is the signal the open_preview handler needs: a backgrounded
// session's editor never mounts, so it must report "tools unavailable" rather
// than silently claim success.
it('resolves false after the timeout when no editor ever registers', async () => {
const manager = new AIChatManager()
await expect(manager.waitForPipelineHelpers(10)).resolves.toBe(false)
await expect(manager.waitForPipelineHelpers(undefined, 10)).resolves.toBe(false)
})
})
@@ -15,22 +15,17 @@ vi.mock('@codingame/monaco-vscode-languages-service-override', () => ({ default:
vi.mock('$lib/components/vscode', () => ({}))
import { getSessionContextPromptSection, type SessionPromptContext } from './core'
import { getPipelinePromptSection, type PipelineContext } from '../pipeline/core'
import { getPipelinePromptSection } from '../pipeline/core'
import { assembleGlobalSystemMessage, assembleGlobalTools } from './globalAssembly'
const sessionContext: SessionPromptContext = { workspaceId: 'ws', parentWorkspaceId: 'parent' }
const pipelineContext: PipelineContext = {
folder: 'myfolder',
mode: 'edit',
nodes: [],
assets: []
}
const pipelineFolders = ['myfolder']
describe('assembleGlobalTools', () => {
it('includes the pipeline tools only when a pipeline editor is active', () => {
const names = (opts: Parameters<typeof assembleGlobalTools>[0]) =>
assembleGlobalTools(opts).map((t) => t.def.function.name)
expect(names({ pipelineContext })).toContain('get_pipeline_graph')
expect(names({ pipelineFolders })).toContain('get_pipeline_graph')
expect(names({})).not.toContain('get_pipeline_graph')
})
})
@@ -41,13 +36,13 @@ describe('assembleGlobalSystemMessage', () => {
it('carries both optional sections when both contexts are given', () => {
const content = assembleGlobalSystemMessage(undefined, {
sessionContext,
pipelineContext
pipelineFolders
}).content as string
expect(content).toContain(getSessionContextPromptSection(sessionContext))
expect(content).toContain(getPipelinePromptSection(pipelineContext))
expect(content).toContain(getPipelinePromptSection(pipelineFolders))
// Order is part of the assembled output, so it is pinned too.
expect(content.indexOf(getSessionContextPromptSection(sessionContext))).toBeLessThan(
content.indexOf(getPipelinePromptSection(pipelineContext))
content.indexOf(getPipelinePromptSection(pipelineFolders))
)
})
@@ -58,6 +53,6 @@ describe('assembleGlobalSystemMessage', () => {
// shares, so a stray section built from any other context still trips them.
const heading = (s: string) => s.trim().split('\n')[0]
expect(content).not.toContain(heading(getSessionContextPromptSection(sessionContext)))
expect(content).not.toContain(heading(getPipelinePromptSection(pipelineContext)))
expect(content).not.toContain(heading(getPipelinePromptSection(pipelineFolders)))
})
})
@@ -12,7 +12,7 @@ import {
type SessionPromptContext
} from './core'
import { createMcpTools } from './mcpTools'
import { getPipelinePromptSection, pipelineTools, type PipelineContext } from '../pipeline/core'
import { getPipelinePromptSection, pipelineTools } from '../pipeline/core'
import { getFolderInstructionsPromptSection, type FolderInstruction } from '../folderInstructions'
// Derived, not retyped: an option added to prepareGlobalSystemMessage is reachable
@@ -21,7 +21,8 @@ import { getFolderInstructionsPromptSection, type FolderInstruction } from '../f
// directly — would get it.
export type GlobalAssemblyOpts = NonNullable<Parameters<typeof prepareGlobalSystemMessage>[1]> & {
sessionContext?: SessionPromptContext
pipelineContext?: PipelineContext
/** Folders of the open pipeline editors; the pipeline tools and prompt come with any. */
pipelineFolders?: readonly string[]
folderInstructions?: readonly FolderInstruction[]
}
@@ -34,8 +35,8 @@ export function assembleGlobalSystemMessage(
if (opts.sessionContext) {
systemMessage.content += getSessionContextPromptSection(opts.sessionContext, opts.access)
}
if (opts.pipelineContext) {
systemMessage.content += getPipelinePromptSection(opts.pipelineContext, opts.access)
if (opts.pipelineFolders?.length) {
systemMessage.content += getPipelinePromptSection(opts.pipelineFolders, opts.access)
}
return systemMessage
}
@@ -43,7 +44,7 @@ export function assembleGlobalSystemMessage(
export function assembleGlobalTools(opts: GlobalAssemblyOpts): SessionTool<any>[] {
return [
...globalToolsFor({ sessionPreview: opts.previewTools ?? false }),
...(opts.pipelineContext ? pipelineTools : []),
...(opts.pipelineFolders?.length ? pipelineTools : []),
...createMcpTools(opts.mcpServers ?? [])
]
}
@@ -32,7 +32,7 @@ import type { SessionAccess, SessionCapability, SessionTool } from '../sessionCa
const ASSEMBLY_OPTS = {
previewTools: true,
pipelineContext: { folder: 'my_pipeline', mode: 'edit', nodes: [], assets: [] },
pipelineFolders: ['my_pipeline'],
mcpServers: [{ path: 'f/test/server' } as any]
}
@@ -168,7 +168,7 @@ describe('session tool policies', () => {
user: { username: 'alex', folders: ['shared'], folders_read: ['shared'] },
access,
sessionContext: ctx,
pipelineContext: { folder: 'my_pipeline', mode: 'edit', nodes: [], assets: [] }
pipelineFolders: ['my_pipeline']
})
// Both decoration variants: the escalation one adds its own tool mentions.
for (const blocks of [0, 9]) {
@@ -39,7 +39,7 @@ const sampleContext: PipelineContext = {
}
function makeHelpers(overrides: Partial<PipelineAIChatHelpers> = {}): {
helpers: { pipeline: PipelineAIChatHelpers }
helpers: { pipelines: () => PipelineAIChatHelpers[] }
calls: Record<string, any[]>
} {
const calls: Record<string, any[]> = {}
@@ -50,6 +50,7 @@ function makeHelpers(overrides: Partial<PipelineAIChatHelpers> = {}): {
return ret
}
const pipeline: PipelineAIChatHelpers = {
getFolder: () => sampleContext.folder,
getPipelineContext: () => sampleContext,
getNodeBody: async (path: string) => {
calls.getNodeBody = [...(calls.getNodeBody ?? []), [path]]
@@ -67,7 +68,7 @@ function makeHelpers(overrides: Partial<PipelineAIChatHelpers> = {}): {
testNode: async () => 'job-123',
...overrides
}
return { helpers: { pipeline }, calls }
return { helpers: { pipelines: () => [pipeline] }, calls }
}
describe('pipeline tools', () => {
@@ -196,9 +197,60 @@ describe('pipeline tools', () => {
})
})
describe('pipeline tools with several editors open', () => {
function editorFor(folder: string, built: string[]): PipelineAIChatHelpers {
return {
...makeHelpers().helpers.pipelines()[0],
getFolder: () => folder,
getPipelineContext: () => ({ ...sampleContext, folder }),
proposeNode: async (input) => {
built.push(`${folder}:${input.path}`)
return { path: input.path, detectedReads: [], detectedWrites: [] }
}
}
}
const call = (name: string, args: any, editors: PipelineAIChatHelpers[]) =>
toolByName(name).fn({
args,
workspace: 'w',
helpers: { pipelines: () => editors },
toolCallbacks: noopCallbacks(),
toolId: 't'
})
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' },
editors
)
expect(built).toEqual(['sales:f/sales/clean'])
await expect(
call(
'build_pipeline_node',
{ path: 'f/other/clean', language: 'bun', content: '// pipeline' },
editors
)
).rejects.toThrow(/open_preview\(kind="pipeline", path="other"\)/)
})
it('get_pipeline_graph needs the folder only when it is ambiguous', async () => {
const editors = [editorFor('crm', []), editorFor('sales', [])]
await expect(call('get_pipeline_graph', {}, editors)).rejects.toThrow(/pass the folder/)
expect(
JSON.parse(await call('get_pipeline_graph', { folder: 'f/sales' }, editors))
).toMatchObject({ folder: 'sales' })
expect(JSON.parse(await call('get_pipeline_graph', {}, [editors[0]]))).toMatchObject({
folder: 'crm'
})
})
})
describe('getPipelinePromptSection', () => {
it('names the active folder and the direct-draft workflow', () => {
const section = getPipelinePromptSection(sampleContext)
const section = getPipelinePromptSection([sampleContext.folder])
expect(section).toContain('/pipeline/analytics')
expect(section).toContain('build_pipeline_node')
expect(section).toContain('directly as unsaved drafts')
@@ -10,6 +10,7 @@ import {
type SessionTool
} from '../sessionCapabilities'
import type { PipelineOutputKind } from '$lib/components/assets/AssetGraph/pipelineTemplates'
import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
// ============================================================================
// Pipeline AI chat tools.
@@ -57,6 +58,8 @@ export type PipelineContext = {
* small — the page owns the draft Map and canvas rendering.
*/
export interface PipelineAIChatHelpers {
/** Folder name (no `f/` prefix) of the pipeline this editor shows. */
getFolder: () => string
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>
@@ -81,16 +84,58 @@ export interface PipelineAIChatHelpers {
testNode: (path: string, args?: Record<string, any>) => Promise<string | undefined>
}
/** Helper bag the pipeline tools receive from the manager in global mode. */
export type PipelineToolHelpers = { pipeline?: 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[] }
function requirePipeline(helpers: PipelineToolHelpers): PipelineAIChatHelpers {
if (!helpers?.pipeline) {
function openPipelines(helpers: PipelineToolHelpers): readonly PipelineAIChatHelpers[] {
const open = helpers?.pipelines?.() ?? []
if (open.length === 0) {
throw new Error(
'No pipeline editor is open. Pipeline tools only work on a /pipeline/<folder> page in edit mode.'
)
}
return helpers.pipeline
return open
}
function notOpenError(folder: string, open: readonly PipelineAIChatHelpers[]): Error {
const names = open.map((p) => `'${p.getFolder()}'`).join(', ')
return new Error(
`No pipeline editor is open for folder '${folder}' (open: ${names}). Open it with open_preview(kind="pipeline", path="${folder}") first.`
)
}
/** The editor of the pipeline a node path belongs to (`f/<folder>/<node>`). 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)
const folder = path.match(/^f\/([^/]+)\//)?.[1]
const match = folder ? open.find((p) => p.getFolder() === folder) : undefined
if (match) return match
if (open.length === 1) return open[0]
throw folder
? notOpenError(folder, open)
: new Error(
`'${path}' is not a pipeline node path — nodes live at f/<folder>/<node_name>, with <folder> one of: ${open.map((p) => `'${p.getFolder()}'`).join(', ')}.`
)
}
function pipelineForFolder(
helpers: PipelineToolHelpers,
folder: string | undefined
): PipelineAIChatHelpers {
const open = openPipelines(helpers)
if (folder) {
const name = normalizePipelineFolder(folder)
const match = open.find((p) => p.getFolder() === name)
if (!match) throw notOpenError(name, open)
return match
}
if (open.length === 1) return open[0]
throw new Error(
`Several pipeline editors are open (${open.map((p) => `'${p.getFolder()}'`).join(', ')}) — pass the folder to read.`
)
}
const scriptLangSchema = z.enum($ScriptLang.enum)
@@ -105,12 +150,20 @@ const outputKindSchema = z
// Read tools
// ----------------------------------------------------------------------------
const getPipelineGraphSchema = z.object({})
const getPipelineGraphSchema = z.object({
folder: z
.string()
.optional()
.describe(
'Folder of the pipeline to read (e.g. `analytics`). Required when several pipeline editors are open; may be omitted when only one is.'
)
})
const getPipelineGraphToolDef = createToolDef(
getPipelineGraphSchema,
'get_pipeline_graph',
"Read the live pipeline graph for the open /pipeline/<folder> editor: its nodes (scripts), each node's language, asset reads/writes, declared triggers, and whether it has an unsaved draft edit. Call this before building or editing nodes so you reuse existing assets/paths and understand the current DAG."
"Read the live pipeline graph of an open /pipeline/<folder> editor: its nodes (scripts), each node's language, asset reads/writes, declared triggers, and whether it has an unsaved draft edit. Call this before building or editing nodes so you reuse existing assets/paths and understand the current DAG.",
{ strict: false }
)
const readPipelineNodeSchema = z.object({
@@ -131,7 +184,7 @@ const buildPipelineNodeSchema = z.object({
path: z
.string()
.describe(
"Workspace path for the new node, e.g. f/<folder>/<name>. Use the open pipeline's folder. Must not collide with an existing node."
'Workspace path for the new node, e.g. f/<folder>/<name>, in the folder of an open pipeline editor. Must not collide with an existing node.'
),
language: scriptLangSchema.describe(
'Script language. SQL-shaped data work uses duckdb (DuckLake/S3) or postgresql (data tables); bun/python3 for general transforms.'
@@ -217,8 +270,9 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
requires: NONE,
def: getPipelineGraphToolDef,
planModeSafe: true,
fn: async ({ helpers, toolId, toolCallbacks }) => {
const pipeline = requirePipeline(helpers)
fn: async ({ args, helpers, toolId, toolCallbacks }) => {
const { folder } = getPipelineGraphSchema.parse(args ?? {})
const pipeline = pipelineForFolder(helpers, folder)
toolCallbacks.setToolStatus(toolId, { content: 'Reading pipeline graph...' })
const ctx = pipeline.getPipelineContext()
toolCallbacks.setToolStatus(toolId, {
@@ -233,8 +287,8 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
def: readPipelineNodeToolDef,
planModeSafe: true,
fn: async ({ args, helpers, toolId, toolCallbacks }) => {
const pipeline = requirePipeline(helpers)
const { path } = readPipelineNodeSchema.parse(args)
const pipeline = pipelineForPath(helpers, path)
toolCallbacks.setToolStatus(toolId, { content: `Reading node '${path}'...` })
const node = await pipeline.getNodeBody(path)
if (!node) {
@@ -251,8 +305,8 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
showDetails: true,
showFade: true,
fn: async ({ args, helpers, toolId, toolCallbacks }) => {
const pipeline = requirePipeline(helpers)
const { path, language, content, output_kind } = buildPipelineNodeSchema.parse(args)
const pipeline = pipelineForPath(helpers, path)
toolCallbacks.setToolStatus(toolId, { content: `Building node '${path}'...` })
const { detectedReads, detectedWrites } = await pipeline.proposeNode({
path,
@@ -274,8 +328,8 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
showDetails: true,
showFade: true,
fn: async ({ args, helpers, toolId, toolCallbacks }) => {
const pipeline = requirePipeline(helpers)
const { path, old_string, new_string, replace_all } = editPipelineNodeSchema.parse(args)
const pipeline = 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.`
@@ -300,8 +354,8 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
requires: WRITE_DRAFT,
def: removePipelineNodeToolDef,
fn: async ({ args, helpers, toolId, toolCallbacks }) => {
const pipeline = requirePipeline(helpers)
const { path } = removePipelineNodeSchema.parse(args)
const pipeline = pipelineForPath(helpers, path)
toolCallbacks.setToolStatus(toolId, { content: `Discarding draft '${path}'...` })
await pipeline.removeProposedNode(path)
toolCallbacks.setToolStatus(toolId, {
@@ -319,8 +373,8 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
showDetails: true,
autoCollapseDetails: false,
fn: async ({ args, workspace, helpers, toolId, toolCallbacks }) => {
const pipeline = requirePipeline(helpers)
const { path, args: runArgs } = testPipelineNodeSchema.parse(args)
const pipeline = pipelineForPath(helpers, path)
return executeTestRun({
jobStarter: async () => {
const jobId = await pipeline.testNode(path, runArgs ?? undefined)
@@ -350,18 +404,30 @@ export const pipelineTools: SessionTool<PipelineToolHelpers>[] = [
* since the tools it names are withheld from that session; the annotation model stays,
* as it is what lets the model read and explain an existing pipeline.
*/
export function getPipelinePromptSection(ctx: PipelineContext, access?: SessionAccess): string {
export function getPipelinePromptSection(
folders: readonly string[],
access?: SessionAccess
): string {
const canWriteDraft = !access || access.has('write_draft')
const [first] = folders
const openLine =
folders.length === 1
? `The user has the /pipeline/${first} editor open.`
: `The user has ${folders.length} pipeline editors open: ${folders.map((f) => `/pipeline/${f}`).join(', ')}. Each tool acts on the pipeline its node path's folder names; pass \`folder\` to get_pipeline_graph.`
const pathLine =
folders.length === 1
? `Every node of this pipeline lives at \`f/${first}/<node_name>\` — \`${first}\` is the folder name and \`f/\` is the owner prefix every workspace path carries, so write it exactly once (never \`f/f/…\`, and never a bare \`<node_name>\`).`
: `Every node of a pipeline lives at \`f/<folder>/<node_name>\` (e.g. \`f/${first}/<node_name>\`) — \`f/\` is the owner prefix every workspace path carries, so write it exactly once (never \`f/f/…\`, and never a bare \`<node_name>\`). A node can only be built in the folder of an open pipeline editor.`
return `
Data Pipeline editor (ACTIVE):
- The user has the /pipeline/${ctx.folder} editor open. A pipeline is a DAG of scripts (nodes) connected by storage assets (DuckLake tables, data tables, S3 objects, volumes, resources) and execution triggers.
- ${openLine} A pipeline is a DAG of scripts (nodes) connected by storage assets (DuckLake tables, data tables, S3 objects, volumes, resources) and execution triggers.
- Annotations are top-of-file comments in the NODE'S OWN comment syntax: \`--\` for SQL (duckdb/postgresql), \`#\` for python3/bash, \`//\` for bun/TS. The \`//\` shown below is the TS form — translate it (a \`// pipeline\` line in a SQL node is a syntax error that won't deploy).
- A script becomes a pipeline node when its source starts with the \`// pipeline\` annotation. Declare execution-DAG inputs with \`// on <asset-uri | schedule | webhook | email | kafka | mqtt | amqp | nats | postgres | sqs | gcp | data_upload>\` (e.g. \`// on ducklake://main/orders\`). Outputs are inferred from what the body writes (wmill SDK calls / SQL CREATE TABLE / writeS3File); declare a managed output with \`// materialize <asset-uri>\`. Optional badges: \`// partitioned <daily|hourly|weekly|monthly|dynamic>\`, \`// freshness <duration>\`, \`// tag <name>\`, \`// retry <count> [delay]\`, \`// data_test <kind> ...\` (managed DuckLake targets only — deploy rejects it beside a \`dbt://\` target), \`// measure <name> = <agg> [where <pred>]\`, \`// dimension <name> = <expr>\`.
- \`materialize\` (the managed output): a managed \`// materialize ducklake://<name>/<table>\` means the runtime writes the node's output table FOR you — write the body as a single SELECT and the runtime wraps it in the create/replace, so do NOT also write your own CREATE TABLE / INSERT. The \`dbt://\` target below is the opposite: the node writes its own DDL and none of the write strategies apply to it. IMPORTANT: a MANAGED \`// materialize\` is **DuckDB-only** and its target MUST be a DuckLake table (\`ducklake://<name>/<table>\`) — deploy rejects a \`ducklake://\` target on any other language. For a \`python3\`/\`bun\`/\`postgresql\` node writing the lake, do NOT use \`// materialize\`; write the output via the SDK instead (e.g. \`wmill.writeS3File(...)\`, a \`CREATE TABLE\` in postgresql, or \`wmill.databaseUrlFromResource\`/ducklake helpers) and let the output be inferred. Reach for \`duckdb\` when a node should materialize a DuckLake table. The one target any language BUT DBT'S OWN may declare (a dbt project's writes come from its manifest, so \`// materialize\` on a dbt script is rejected at deploy) is a WAREHOUSE RELATION: \`// materialize manual dbt://<warehouse>/<schema>/<name>\`, with \`<warehouse>\` a warehouse the workspace configures under Settings → dbt. \`manual\` is its only mode — nothing generates warehouse DDL, so the node issues its own write and the annotation records the outcome. Use it on an ingestion node a dbt project reads as a \`source\`: the declared relation and the dbt model become ONE graph node, and a downstream \`// on dbt://<warehouse>/<schema>/<name>\` fires when that node completes. Write strategy: with no option it REPLACES the whole table each run (full refresh; the only mode whose output columns may change); \`// materialize <uri> append\` INSERT-appends rows (incremental); \`// materialize <uri> key=<col>\` merges/upserts on \`<col>\`. \`// materialize manual <uri>\` opts OUT of managed writes — the script writes its own DDL and the annotation only records the output asset for lineage. \`materialize\` is paired with partitioning for incremental pipelines: a \`// partitioned <daily|hourly|weekly|monthly|dynamic>\` node runs once per partition (append/merge into a fixed-schema table), and the \`{partition}\` token — usable in any asset URI AND in the body SQL — is substituted with the current partition's IDENTITY string at run time. To filter the source to the active slice on a time grain, use the runtime-injected macro: \`WHERE wm_partition(<ts_col>) = {partition}\`. \`wm_partition(ts)\` buckets a timestamp with the exact identity format the runtime used (daily/hourly/weekly/monthly), so it always matches and you never hand-write a \`strftime\` format. Do NOT write \`= TIMESTAMP {partition}\`: the identity string is not a valid timestamp literal for hourly/weekly/monthly and errors at runtime. For \`dynamic\` partitioning the identity is your caller-supplied key (not a timestamp, no macro), so filter on it directly: \`WHERE <your_key_col> = {partition}\`. \`materialize\` is an output DECLARATION on the node — it is not a command; there is no "materialize run".
- \`measure\` / \`dimension\` (declared metrics): on a node that materializes a DuckLake table, \`// measure <name> = <aggregate> [where <predicate>]\` names the canonical way to aggregate that table (e.g. \`// measure revenue = sum(amount) where not is_refund\`), and \`// dimension <name> = <expr>\` names a way to slice it (e.g. \`// dimension region = region\`, \`// dimension month = date_trunc('month', ordered_at)\`). They execute nothing: they are catalogued at deploy so the editor and other agents can reuse the definition instead of re-deriving it and silently disagreeing. Keep the predicate in the \`where\` clause rather than folding it into the aggregate: it is rendered as \`<agg> FILTER (WHERE <pred>)\`, which is what lets two measures with different predicates sit under one GROUP BY. DuckLake-only, and only meaningful next to \`// materialize\`. Declare one when a number carries a judgement call someone else would get wrong (refunds excluded, test rows dropped, which column is the amount); do NOT blanket every table with measures, an obvious \`count(*)\` earns nothing. To USE a metric another node declares, read that node with read_pipeline_node and reuse its exact expression rather than guessing it.
- Use get_pipeline_graph to see the current nodes/assets/triggers, and read_pipeline_node before editing one.
- Every node of this pipeline lives at \`f/${ctx.folder}/<node_name>\` — \`${ctx.folder}\` is the folder name and \`f/\` is the owner prefix every workspace path carries, so write it exactly once (never \`f/f/…\`, and never a bare \`<node_name>\`).${
- ${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.
@@ -29,7 +29,7 @@
/** Folder name the pipeline graph is scoped to (not a workspace item path). */
path: string
workspaceId: string
/** Only the visible session registers the pipeline tools on its manager. */
/** Only the visible session polls the live run badges. */
isActiveSession?: boolean
/** Whether this is the foreground preview tab. */
active?: boolean
@@ -43,39 +43,13 @@
// wrong chat — leaving this session's chat unable to build canvas nodes.
const aiChatManager = runtime.manager
// Only the foreground tab of the foreground session owns the chat's pipeline
// tools and runs the live-badge poll — a background tab (another preview tab is
// showing, or another session is active) must not shadow that context or poll.
// Only the foreground tab of the foreground session runs the live-badge poll.
const engaged = $derived(isActiveSession && active)
// Externalized editor state — lives on the runtime so the drafts persist across
// hide/show of the preview pane (the pane unmounts on hide).
const pe = runtime.pipelineEditorState
// The reused `pe` is scoped to a folder. A same-folder remount (hide→show)
// keeps the drafts; a retarget to a different folder resets so stale drafts
// don't bleed across folders. untrack the writes so this can't self-loop.
$effect(() => {
const folder = path
untrack(() => {
if (pe.folder !== folder) {
if (pe.folder !== undefined) {
pe.reset()
// A retarget without remount re-scopes the poll, so the release
// effect could never match the old folder's job — drop the hint.
activeRunnable = undefined
activeRunnableJobId = undefined
// Re-scope the Global pipeline prompt to the new folder (the helper
// methods already read the reactive path, but the system message
// string was built for the old one). Only when this tab is engaged —
// its helpers are the registered set; a background tab reconfigures
// when it next becomes the foreground one.
if (engaged) aiChatManager.rebuildGlobalSystemMessage()
}
pe.folder = folder
}
})
})
// This folder's editor state lives on the runtime, so its drafts survive a
// remount. The host remounts this view when the tab moves to another folder, so
// `path` is fixed for its lifetime.
const pe = untrack(() => runtime.pipelineEditor(path))
const EMPTY_GRAPH: AssetGraphResponse = { assets: [], runnables: [], edges: [], triggers: [] }
const EMPTY_PATH_MAP = new Map<string, Array<{ kind: AssetKind; path: string }>>()
@@ -285,14 +259,10 @@
}
}
// Register the pipeline tools on this session's manager while this tab is the
// engaged (foreground) one. setPipelineHelpers rebuilds the global tool set to
// include the pipeline tools and tears them down on cleanup — so switching to
// another preview tab or session releases them.
$effect(() => {
if (!engaged) return
return aiChatManager.setPipelineHelpers(helpers)
})
// Register this folder's pipeline tools on the session's own manager for as long
// as the view is mounted — background tabs included, since each folder's tools
// act on its own state — and release them on unmount.
$effect(() => aiChatManager.setPipelineHelpers(helpers))
</script>
<div class="flex flex-col h-full w-full bg-surface">
@@ -306,7 +306,10 @@
{#await import('./PipelineEditorView.svelte')}
{@render editorLoading()}
{:then Module}
<Module.default {runtime} path={slot.path} {workspaceId} {isActiveSession} {active} />
<!-- Keyed: the view binds one folder's state for its lifetime. -->
{#key slot.path}
<Module.default {runtime} path={slot.path} {workspaceId} {isActiveSession} {active} />
{/key}
{/await}
{:else}
{#await import('./RawAppEditorView.svelte')}
@@ -1335,8 +1335,8 @@
bind:open={seedOfferOpen}
kind="X"
enterConfirms={false}
title="Keep this {seedOffer?.route.kind ?? 'item'} in the new session?"
description="You came here from {seedOffer?.route.itemPath ?? ''}."
title="Keep this {seedOffer?.kind ?? 'item'} in the new session?"
description="You came here from {seedOffer?.path ?? ''}."
>
<p class="text-sm text-secondary">
Keeping it opens it in the preview, so the chat starts with it as context.
@@ -1,6 +1,6 @@
import { getContext, onDestroy, setContext } from 'svelte'
import type { OpenInSessionSource } from './OpenInSessionButton.svelte'
import type { PreviewItemRoute } from './previewPaths'
import type { SessionTarget } from './sessionState.svelte'
// The "Open in AI session" hand-off, published by the component that owns the
// item being edited (FlowBuilder, RawAppEditor) for AI entry points too deep in
@@ -38,20 +38,17 @@ export function getOpenInSessionHandoff(): OpenInSessionHandoff | undefined {
return getContext<OpenInSessionHandoff | undefined>(KEY)
}
/** The mounted editor's hand-off for the item `route` names, or undefined when
* no editor on screen publishes one for it (a legacy app, a detail page).
* Matched on the item rather than taken as "the latest registered": a script
* editor mounted in a flow's drawer registers too, and the rail wants the
* page's own item. */
/** The mounted editor's hand-off for `target`, or undefined when no editor on
* screen publishes one for it (a detail page). Matched on the item rather than
* taken as "the latest registered": a script editor mounted in a flow's drawer
* registers too, and the rail wants the page's own item. */
export function findMountedOpenInSessionSource(
route: PreviewItemRoute
target: SessionTarget
): OpenInSessionSource | undefined {
const kind = route.kind === 'app' ? (route.raw_app ? 'raw_app' : undefined) : route.kind
if (!kind) return undefined
for (const handoff of mounted) {
const source = handoff.source()
const target = source?.target
if (target && target.kind === kind && target.path === route.itemPath) return source
const t = source?.target
if (t && t.kind === target.kind && t.path === target.path) return source
}
return undefined
}
@@ -1,5 +1,6 @@
import { base } from '$lib/base'
import type { WorkspaceItemKind } from '$lib/components/workspacePicker'
import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
// The paths a preview location can point at, and the base handling around them. Kept apart
// from `previewRouter`, which reads a location's *view* from each page's filter schema and
@@ -166,3 +167,11 @@ export function parsePreviewItemRoute(fullPath: string): PreviewItemRoute | null
if (m[1] === 'apps_raw') return { kind: 'app', raw_app: true, itemPath }
return { kind: 'app', raw_app: false, itemPath }
}
// A `/pipeline/<folder>` route is the data-pipeline graph editor for that folder
// (the folder is a single path segment, not a workspace item path). The bare
// `/pipeline` list page is not an editor. Returns the folder name, or null.
export function parsePipelineRoute(fullPath: string): string | null {
const m = stripBase(fullPath).match(/^\/pipeline\/([^/?#]+)/)
return m ? normalizePipelineFolder(decodeURIComponent(m[1])) : null
}
@@ -11,6 +11,7 @@ import {
pageKey,
pageHref,
parsePageItemRoute,
parsePipelineRoute,
parsePreviewItemRoute,
RESOURCES_PATH,
RUNS_PATH,
@@ -32,6 +33,7 @@ export {
pageKey,
pageHref,
parsePageItemRoute,
parsePipelineRoute,
parsePreviewItemRoute,
stripBase,
TRIGGER_PAGES,
@@ -59,7 +61,6 @@ import { buildVariablesFilterSchema } from '$lib/components/variables/variablesF
import { buildResourcesFilterSchema } from '$lib/components/resources/resourcesFilter'
import { buildAssetsFilterSchema } from '$lib/components/assets/assetsFilter'
import { COMPARE_ITEMS_PARAM } from './modifiedItemsMask'
import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
import type { WorkspaceItem } from '$lib/components/workspacePicker'
import type { SessionTargetKind } from './sessionRuntime.svelte'
@@ -470,13 +471,6 @@ export function parsePreviewSelectedId(url: string): string | undefined {
}
}
// A `/pipeline/<folder>` route is the data-pipeline graph editor for that folder
// (the folder is a single path segment, not a workspace item path). The bare
// `/pipeline` list page is not an editor. Returns the folder name, or null.
export function parsePipelineRoute(fullPath: string): string | null {
const m = stripBase(fullPath).match(/^\/pipeline\/([^/?#]+)/)
return m ? normalizePipelineFolder(decodeURIComponent(m[1])) : null
}
// The id (before the query) is the artifact's stable routing identity; the name rides in the
// hash so the tab strip labels it without a store lookup, and the version in the query so the
@@ -530,8 +524,7 @@ export const isArtifactKey = (key: string) => key.startsWith('artifact:')
// How a preview tab should render: as an in-process live editor or an iframe
// fallback. Any editable item of a wrappable kind (script, flow, raw app) mounts
// its per-(kind,path) cell editor; a `/pipeline/<folder>` route mounts the
// data-pipeline graph editor (single, shared runtime.pipelineEditorState — `path`
// is the folder); the list page of a page item kind mounts that list; everything
// data-pipeline graph editor of that folder (`path` is the folder); the list page of a page item kind mounts that list; everything
// else (static pages, regular drag-and-drop apps, any other route) stays an iframe.
export type PreviewSlot =
| { kind: 'editor'; editorKind: SessionTargetKind | 'pipeline'; path: string }
@@ -1,9 +1,13 @@
import { base } from '$app/paths'
import type { SessionTarget } from './sessionState.svelte'
import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
// Maps a session's editor target to the canonical full-page Windmill route.
export function sessionTargetHref(target: SessionTarget | undefined): string | undefined {
if (!target) return undefined
if (target.kind === 'pipeline') {
return `${base}/pipeline/${encodeURIComponent(normalizePipelineFolder(target.path))}`
}
const seg =
target.kind === 'script'
? 'scripts/edit'
@@ -13,9 +13,13 @@ describe('sessionTargetHref', () => {
expect(sessionTargetHref({ kind: 'raw_app', path: 'u/me/app' })).toBe('/apps_raw/edit/u/me/app')
})
it('returns undefined for no target or a kind without a full-page route', () => {
it('maps a pipeline to its folder page, whichever form the folder is given in', () => {
expect(sessionTargetHref({ kind: 'pipeline', path: 'crm' })).toBe('/pipeline/crm')
expect(sessionTargetHref({ kind: 'pipeline', path: 'f/crm' })).toBe('/pipeline/crm')
})
it('returns undefined for no target', () => {
expect(sessionTargetHref(undefined)).toBeUndefined()
expect(sessionTargetHref({ kind: 'pipeline', path: 'u/me/pipe' })).toBeUndefined()
})
})
@@ -421,20 +421,15 @@ export class SessionPreviewTabs {
}
}
const url = targetUrl(target)
// Pipeline previews all share one runtime.pipelineEditorState, so keep at
// most one pipeline tab: re-point the existing one to the requested folder
// rather than opening a second pipeline editor that would fight over the
// shared state (`focused` when it already showed this folder, else `opened`
// since the view now shows a different pipeline).
// One tab per pipeline folder: two would mount two editors over one folder's
// state and draft.
const pipelineFolder = parsePipelineRoute(url)
if (pipelineFolder) {
const existing = this.#tabs.find((t) => parsePipelineRoute(t.url) !== null)
const existing = this.#tabs.find((t) => parsePipelineRoute(t.url) === pipelineFolder)
if (existing) {
const same = existing.url === url
this.#retarget(existing, url)
this.#activeId = existing.id
this.#flush()
return { status: same ? 'focused' : 'opened' }
return { status: 'focused' }
}
}
// Dedupe artifacts by id, not full url: an update may have changed the name the url carries.
@@ -506,15 +501,11 @@ export class SessionPreviewTabs {
return
}
}
// Keep at most one pipeline tab (all share runtime.pipelineEditorState): if a
// *different* tab already hosts a pipeline, retarget and focus it rather than
// turning the active tab into a second pipeline editor racing the shared
// state. Same invariant as open(); a no-op when the active tab is that tab.
// Same one-tab-per-folder rule as open(): focus the tab already on this folder.
const pipelineFolder = parsePipelineRoute(targetUrl(target))
if (pipelineFolder) {
const existing = this.#tabs.find((x) => parsePipelineRoute(x.url) !== null)
const existing = this.#tabs.find((x) => parsePipelineRoute(x.url) === pipelineFolder)
if (existing && existing.id !== t.id) {
this.#retarget(existing, targetUrl(target, existing))
this.#activeId = existing.id
this.#flush()
return
@@ -552,19 +552,22 @@ describe('SessionPreviewTabs.navigate', () => {
expect(o.tabs.find((t) => t.id === pageTabId)?.url).toBe('/runs')
})
it('retargets the one pipeline tab instead of turning the active tab into a second', () => {
it('keeps one tab per pipeline folder, several folders side by side', () => {
const o = owner()
o.open(pipelineTarget)
const pipelineTabId = o.activeId
o.open(scriptTarget) // a second, non-pipeline tab is now active
const scriptTabId = o.activeId
o.navigate(pipelineTarget2)
// No second pipeline editor: the existing one is retargeted and focused.
const crmTabId = o.activeId
o.open(pipelineTarget2)
const salesTabId = o.activeId
expect(o.tabs).toHaveLength(2)
expect(o.activeId).toBe(pipelineTabId)
expect(o.tabs.find((t) => t.id === pipelineTabId)?.url).toBe(`${base}/pipeline/sales`)
// The script tab is untouched.
expect(o.tabs.find((t) => t.id === scriptTabId)?.url).toBe('/scripts/edit/u/me/foo')
expect(salesTabId).not.toBe(crmTabId)
o.open(scriptTarget)
// A folder already open is focused, from open() and navigate() alike.
o.open(pipelineTarget)
expect(o.activeId).toBe(crmTabId)
o.navigate(pipelineTarget2)
expect(o.activeId).toBe(salesTabId)
expect(o.tabs).toHaveLength(3)
expect(o.tabs.find((t) => t.id === crmTabId)?.url).toBe(`${base}/pipeline/crm`)
})
it('retargets the active pipeline tab in place to a new folder', () => {
@@ -191,9 +191,9 @@ export interface SessionRuntime {
// (renderer) and the open_preview/get_preview_status tools cross it, so the
// tab model has exactly one live copy.
readonly previewTabs: SessionPreviewTabs
// Pipeline target state — persists across editor hide/show (the pane unmounts
// on hide, so this can't be component-local) and across session switches.
readonly pipelineEditorState: PipelineEditorState
// Pipeline editor state per folder — persists across editor remounts and
// session switches, so it can't be component-local. Created on first use.
pipelineEditor(folder: string): PipelineEditorState
// Per-(kind, path) editor cells (content/baseline stores + load slot), created
// on demand. Each editable preview tab resolves its own cell, so several items
// stay live at once.
@@ -577,10 +577,17 @@ function createRuntime(session: Session): SessionRuntime {
// picker lists, so changeMode's network refreshes would fire once per listing.
manager.configureGlobalMode()
// Pipeline target state lives on the runtime (not the PipelineEditorView
// component) so the in-session drafts survive hide/show of the editor pane —
// the pane unmounts on hide, and a component-local store would be discarded.
const pipelineEditorState = new PipelineEditorState()
const pipelineEditors = new Map<string, PipelineEditorState>()
function pipelineEditor(folder: string): PipelineEditorState {
const key = normalizePipelineFolder(folder)
let editor = pipelineEditors.get(key)
if (!editor) {
editor = new PipelineEditorState()
editor.folder = key
pipelineEditors.set(key, editor)
}
return editor
}
const runtimeLogRequesters = new Map<string, RawAppRuntimeLogRequester>()
// appPath → requester, one entry per mounted raw-app preview tab.
@@ -594,7 +601,7 @@ function createRuntime(session: Session): SessionRuntime {
sessionId: session.id,
manager,
previewTabs,
pipelineEditorState,
pipelineEditor,
flowCell,
loadedEditorPath,
@@ -1211,8 +1218,8 @@ setOpenPreviewHandler(async ({ sessionId: callerSessionId, kind, path }) => {
// "Unknown tool call" error on the first node it tries to build.
if (kind === 'pipeline') {
const folder = normalizePipelineFolder(path)
const ready = await runtime.manager.waitForPipelineHelpers()
// A backgrounded session's preview tab does not mount, so its editor never
const ready = await runtime.manager.waitForPipelineHelpers(folder)
// A session whose page is not mounted never mounts the editor, so it never
// registers — don't claim success, or the model calls build_pipeline_node
// into the void. Tell it the tools aren't available and how to recover.
if (!ready) {
@@ -15,7 +15,7 @@ import {
type SessionTarget
} from './sessionState.svelte'
import { sessionTargetHref, withPreviewParams } from './sessionMode.svelte'
import { parsePreviewItemRoute, type PreviewItemRoute } from './previewPaths'
import { parsePipelineRoute, parsePreviewItemRoute } from './previewPaths'
import { findMountedOpenInSessionSource } from './openInSessionContext'
// Type-only: erased at compile time, so the component graph stays out of this
// navigation seam (see the dynamic import in openEditorInSession).
@@ -42,10 +42,28 @@ export function rememberNavRoute(pathnameWithSearch: string): void {
}
/** An item editor the user reached session mode from, as the in-app href a new
* session's preview can open, plus the item it edits (for naming it). */
export type NewSessionSeed = { url: string; route: PreviewItemRoute }
* session's preview can open, plus what it edits (for naming it). */
export type NewSessionSeed = { url: string; kind: string; path: string }
// The item (flow, script, app) the user came to session mode from, or undefined
/** What an editor route edits: a session target, or an unset target for an item
* with no in-session editor (a legacy app), which is opened by its page instead. */
function editedAt(
url: string
): { target: SessionTarget | undefined; kind: string; path: string } | undefined {
const folder = parsePipelineRoute(url)
if (folder)
return { target: { kind: 'pipeline', path: folder }, kind: 'pipeline', path: `f/${folder}` }
const route = parsePreviewItemRoute(url)
if (!route) return undefined
const kind = route.kind === 'app' ? (route.raw_app ? 'raw_app' : undefined) : route.kind
return {
target: kind ? { kind, path: route.itemPath } : undefined,
kind: route.kind,
path: route.itemPath
}
}
// The item (flow, script, app, pipeline) the user came to session mode from, or undefined
// when they came from anywhere else. A "New session" asked for after arriving
// from an item is usually a session about that item, but the arrival route
// (`enterSessionMode`) resumes whatever session was open, so the picker offers
@@ -56,8 +74,8 @@ export type NewSessionSeed = { url: string; route: PreviewItemRoute }
export function takeNewSessionSeed(): NewSessionSeed | undefined {
if (navRouteOffered) return undefined
navRouteOffered = true
const route = parsePreviewItemRoute(lastNavRoute)
return route ? { url: lastNavRoute, route } : undefined
const edited = editedAt(lastNavRoute)
return edited ? { url: lastNavRoute, kind: edited.kind, path: edited.path } : undefined
}
// The session entering session mode resumes: the active one if selected, else
@@ -102,8 +120,8 @@ const RESUME_IDLE_LIMIT_MS = 60 * 60 * 1000
// enterSessionMode does. Rejects when the editor could not persist its draft,
// so the caller can stay on the page and say so.
export async function enterSessionModeFromNav(): Promise<void> {
const route = parsePreviewItemRoute(lastNavRoute)
if (route) {
const edited = editedAt(lastNavRoute)
if (edited) {
const resumable = resumableSession()
if (!resumable || Date.now() - sessionLastActivityAt(resumable) > RESUME_IDLE_LIMIT_MS) {
// The editor's own hand-off is what its "Open in AI session" button uses:
@@ -112,7 +130,7 @@ export async function enterSessionModeFromNav(): Promise<void> {
// workspace the item lives in. Only an item with no such editor on
// screen (a legacy app, a detail page) is opened by route, as last
// persisted, in the workspace the route was scoped to.
const source = findMountedOpenInSessionSource(route)
const source = edited.target && findMountedOpenInSessionSource(edited.target)
if (source) await openSourceInSession(source)
else await openPageInSession(lastNavRoute, workspaceParamOf(lastNavRoute))
return
@@ -259,15 +259,21 @@ describe('takeNewSessionSeed', () => {
rememberNavRoute('/flows/edit/u/me/my_flow?workspace=ws')
expect(takeNewSessionSeed()).toEqual({
url: '/flows/edit/u/me/my_flow?workspace=ws',
route: { kind: 'flow', raw_app: false, itemPath: 'u/me/my_flow' }
kind: 'flow',
path: 'u/me/my_flow'
})
// A dismissed or taken offer is not repeated until the user leaves again.
expect(takeNewSessionSeed()).toBeUndefined()
rememberNavRoute('/apps_raw/edit/f/team/dashboard?workspace=ws')
expect(takeNewSessionSeed()?.route).toEqual({
kind: 'app',
raw_app: true,
itemPath: 'f/team/dashboard'
expect(takeNewSessionSeed()).toMatchObject({ kind: 'app', path: 'f/team/dashboard' })
})
it('offers the pipeline the user came from, named by its folder', () => {
rememberNavRoute('/pipeline/crm?workspace=ws')
expect(takeNewSessionSeed()).toEqual({
url: '/pipeline/crm?workspace=ws',
kind: 'pipeline',
path: 'f/crm'
})
})
@@ -118,6 +118,11 @@
import type { PipelineRecording } from '$lib/components/recording/types'
import AutosaveIndicator from '$lib/components/AutosaveIndicator.svelte'
import { onMount, tick, untrack } from 'svelte'
import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte'
import OpenInSessionButton, {
type OpenInSessionSource
} from '$lib/components/sessions/OpenInSessionButton.svelte'
import { setOpenInSessionHandoff } from '$lib/components/sessions/openInSessionContext'
import { aiChatManager } from '$lib/components/copilot/chat/AIChatManager.svelte'
import {
AlertTriangle,
@@ -589,6 +594,31 @@
onMount(() => aiChatManager.setPipelineHelpers(pipelineAiHelpers))
// The session's preview hydrates this folder's drafts from its DB bundle, so the
// autosave still inside its debounce has to land first.
async function persistDraftsForSession(): Promise<void> {
const workspace = $workspaceStore
if (!workspace) return
await tick()
await UserDraftDbSyncer.flush({
workspace,
itemKind: PIPELINE_DRAFT_KIND,
path: pipelineDraftPath
})
}
const sessionOpen: OpenInSessionSource | undefined = $derived(
isOperator
? undefined
: {
target: { kind: 'pipeline', path: folder },
workspaceId: $workspaceStore ?? undefined,
beforeOpen: persistDraftsForSession
}
)
// Lets the navigation rail's session switch open this pipeline, with its drafts.
setOpenInSessionHandoff({ source: () => sessionOpen })
// Navigation guard state. `pendingNavigationUrl` holds the URL the user
// tried to leave to so we can complete the navigation after they pick
// "Save all" or "Discard all"; `bypassNavigationGuard` is the standard
@@ -3265,6 +3295,7 @@
Activity
</Button>
{/if}
<OpenInSessionButton source={sessionOpen} />
<Button
variant="subtle"
unifiedSize="sm"