smooth local pipeline dogfooding (#9888)

AGENTS.md

@@ -24,6 +24,7 @@ Open-source platform for internal tools, workflows, API integrations, background
 ## Dev Environment
 
 - **Backend**: `cargo run` from `backend/` (API at http://localhost:8000)
+- **DuckDB local jobs**: before running DuckDB scripts locally, build the FFI shared library with `cd backend/windmill-duckdb-ffi-internal && ./build_dev.sh`. Re-run it after clean builds or when `backend/target/debug/libwindmill_duckdb_ffi_internal.*` is missing.
 - **Frontend**: `REMOTE=http://localhost:8000 npm run dev` from `frontend/` (port 3000+)
 - **DB**: `psql postgres://postgres:changeme@localhost:5432/windmill`
 - **Login**: `admin@windmill.dev` / `changeme`

backend/CLAUDE.md

@@ -6,3 +6,8 @@
 - **DB schema**: `backend/summarized_schema.txt`
 - **API routes entry point**: `windmill-api/src/lib.rs`
 - **OpenAPI spec**: `windmill-api/openapi.yaml`
+- **DuckDB local jobs**: build the dynamic FFI library before running DuckDB scripts locally:
+  ```bash
+  cd backend/windmill-duckdb-ffi-internal && ./build_dev.sh
+  ```
+  Re-run after clean builds or when `target/debug/libwindmill_duckdb_ffi_internal.*` is missing.

cli/src/commands/pipeline/docs.ts

@@ -157,8 +157,6 @@ export async function generatePipelineDocs(
   opts: GlobalOptions & { local?: boolean; defaultTs?: "bun" | "deno" },
   folder: string,
 ) {
-  const workspace = await resolveWorkspace(opts);
-  await requireLogin(opts);
   const f = folder.replace(/^f\//, "").replace(/\/$/, "");
   // `docs` WRITES PIPELINE.md / AGENTS.md / CLAUDE.md under `f/<folder>`; a `..`
   // segment would escape the folder and clobber files elsewhere in the tree.
@@ -170,9 +168,12 @@ export async function generatePipelineDocs(
   // defaultTs (from wmill.yaml) drives .ts → bun/deno inference for the local graph.
   const merged = await mergeConfigWithConfigFile(opts);
 
+  const workspace = opts.local ? undefined : await resolveWorkspace(opts);
+  if (!opts.local) await requireLogin(opts);
+
   const graph = opts.local
     ? (await buildLocalPipelineGraph({ root, folder: f, defaultTs: merged.defaultTs })).graph
-    : await fetchDeployedGraph(workspace.workspaceId, f);
+    : await fetchDeployedGraph(workspace!.workspaceId, f);
 
   if (graph.runnables.length === 0) {
     // The deployed graph is empty — but a user/agent in a working tree may have
@@ -198,10 +199,20 @@ export async function generatePipelineDocs(
   }
 
   let datatableSchemas: any[] = [];
-  try {
-    datatableSchemas = await wmill.listDataTableSchemas({ workspace: workspace.workspaceId });
-  } catch (err: any) {
-    log.warn(colors.yellow(`Could not fetch datatable schemas: ${err.message}`));
+  const hasExplicitWorkspace =
+    !!opts.workspace ||
+    (!!opts.baseUrl && !!opts.token) ||
+    (!!process.env["WM_WORKSPACE"] &&
+      !!process.env["WM_TOKEN"] &&
+      !!(process.env["BASE_INTERNAL_URL"] ?? process.env["BASE_URL"]));
+  if (!opts.local || hasExplicitWorkspace) {
+    try {
+      const schemaWorkspace = workspace ?? await resolveWorkspace(opts);
+      if (opts.local) await requireLogin(opts);
+      datatableSchemas = await wmill.listDataTableSchemas({ workspace: schemaWorkspace.workspaceId });
+    } catch (err: any) {
+      log.warn(colors.yellow(`Could not fetch datatable schemas: ${err.message}`));
+    }
   }
 
   const md = generatePipelineMarkdown(f, graph, datatableSchemas, !!opts.local);

cli/src/commands/pipeline/localGraph.ts

@@ -241,6 +241,30 @@ function fallbackParse(content: string, language: string): ParseAssetsRaw {
   return out;
 }
 
+function recoverHeaderNativeTriggers(content: string, language: string): string[] {
+  const raw = commentPrefix(language);
+  const p = raw.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+  const out: string[] = [];
+  const seen = new Set<string>();
+  for (const line of content.split("\n")) {
+    const trimmed = line.trim();
+    if (trimmed === "") continue;
+    if (!trimmed.startsWith(raw)) break;
+    const marker = line.match(new RegExp(`^\\s*${p}\\s*on\\s+(\\S+)\\s*$`));
+    if (!marker) continue;
+    const kind = marker[1];
+    if (!NATIVE_KINDS.has(kind) || seen.has(kind)) continue;
+    seen.add(kind);
+    out.push(kind);
+  }
+  return out;
+}
+
+function normalizeRetry(retry: ParseAssetsRaw["retry"]): ParseAssetsRaw["retry"] {
+  if (!retry?.delay) return retry;
+  return { ...retry, delay: retry.delay.replace(/^delay=/, "") };
+}
+
 // Comment prefix for `volume:` annotations. Deliberately NOT `commentPrefix`
 // above (which returns `--` for SQL): volume annotations are only recognized for
 // the languages the backend/frontend recognize them for — mirrors
@@ -398,6 +422,8 @@ export async function buildLocalPipelineGraph(args: {
   for (const s of all) {
     const out = await inferScriptAssets(s.content, s.language);
     if (!out.in_pipeline) continue; // not a pipeline member
+    const retry = normalizeRetry(out.retry);
+    const nativeTriggers = recoverHeaderNativeTriggers(s.content, s.language);
     // Carry the parsed `// tag` so previews route to the same worker the
     // deployed pipeline would (both `pipeline run --local` and `/pipeline_dev`).
     pipelineScripts.push(out.tag ? { ...s, tag: out.tag } : s);
@@ -420,7 +446,7 @@ export async function buildLocalPipelineGraph(args: {
       ...(out.partition ? { partition_kind: out.partition.kind } : {}),
       ...(out.freshness ? { freshness: out.freshness.duration } : {}),
       ...(out.tag ? { tag: out.tag } : {}),
-      ...(out.retry ? { retry: out.retry } : {}),
+      ...(retry ? { retry } : {}),
       ...(out.data_tests && out.data_tests.length > 0 ? { data_tests: out.data_tests } : {}),
       ...(out.column_lineage && out.column_lineage.length > 0
         ? { column_lineage: out.column_lineage }
@@ -466,6 +492,7 @@ export async function buildLocalPipelineGraph(args: {
         });
       }
     }
+    const existingNativeTriggers = new Set<string>();
     for (const t of out.triggers ?? []) {
       if (t.kind === "asset") {
         const at = t as { kind: "asset"; asset_kind: string; path: string };
@@ -481,6 +508,7 @@ export async function buildLocalPipelineGraph(args: {
           runnable_path: s.path,
         });
       } else {
+        existingNativeTriggers.add(t.kind);
         triggers.push({
           trigger_kind: t.kind,
           runnable_kind: "script",
@@ -488,6 +516,14 @@ export async function buildLocalPipelineGraph(args: {
         });
       }
     }
+    for (const kind of nativeTriggers) {
+      if (existingNativeTriggers.has(kind)) continue;
+      triggers.push({
+        trigger_kind: kind,
+        runnable_kind: "script",
+        runnable_path: s.path,
+      });
+    }
   }
 
   return {

cli/test/pipeline_local_graph_unit.test.ts

@@ -67,11 +67,32 @@ test("native triggers surface as trigger rows (no deploy needed)", async () => {
     {
       "ingest.bun.ts":
         `// pipeline\n// on data_upload\nimport * as wmill from "windmill-client"\nexport async function main() {}\n`,
+      "upload_trigger.duckdb.sql": `-- pipeline\n-- on data_upload\nSELECT 1;\n`,
     },
     async (root, folder) => {
       const { graph } = await buildLocalPipelineGraph({ root, folder, defaultTs: "bun" });
       const native = graph.triggers.filter((t) => t.trigger_kind !== "asset");
-      expect(native.map((t) => t.trigger_kind)).toContain("data_upload");
+      expect(
+        native
+          .filter((t) => t.trigger_kind === "data_upload")
+          .map((t) => t.runnable_path)
+          .sort(),
+      ).toEqual(["f/mypipe/ingest", "f/mypipe/upload_trigger"]);
+    },
+  );
+});
+
+test("retry delay metadata strips the optional `delay=` prefix", async () => {
+  await withFolder(
+    {
+      "retry.duckdb.sql": `-- pipeline\n-- retry 2 delay=10s\nSELECT 1;\n`,
+    },
+    async (root, folder) => {
+      const { graph } = await buildLocalPipelineGraph({ root, folder, defaultTs: "bun" });
+      expect(graph.runnables.find((r) => r.path === "f/mypipe/retry")?.retry).toEqual({
+        count: 2,
+        delay: "10s",
+      });
     },
   );
 });