Merge remote-tracking branch 'origin/main' into change-68b704f7

# Conflicts:
#	backend/ee-repo-ref.txt
#	cli/src/commands/datatable/datatable.ts
#	cli/src/guidance/skills.gen.ts
#	system_prompts/auto-generated/cli/cli-commands.md
#	system_prompts/auto-generated/prompts.ts
#	system_prompts/auto-generated/skills/cli-commands/SKILL.md

cli/src/commands/app/bundle.ts

@@ -6,6 +6,7 @@ import * as log from "../../core/log.ts";
 import { colors } from "@cliffy/ansi/colors";
 import * as windmillUtils from "@windmill-labs/shared-utils";
 import { readTextFile, readTextFileSync } from "../../utils/utils.ts";
+import { getEsbuild, stopEsbuild } from "../../utils/esbuild_loader.ts";
 export interface BundleOptions {
   entryPoint?: string;
   outDir?: string;
@@ -170,8 +171,9 @@ export async function ensureNodeModules(appDir?: string): Promise<void> {
 export async function createBundle(
   options: BundleOptions = {}
 ): Promise<BundleResult> {
-  // Dynamically import esbuild
-  const esbuild = await import("esbuild");
+  // Native esbuild with a transparent esbuild-wasm fallback on host/binary
+  // version mismatch (see esbuild_loader.ts).
+  const esbuild = await getEsbuild();
 
   // Detect frameworks to determine default entry point.
   // Use the entryPoint's directory if provided, otherwise fall back to cwd.
@@ -286,6 +288,10 @@ export async function createBundle(
     outfile,
     sourcemap,
     minify,
+    // Keep outputs in memory: esbuild-wasm cannot write to the filesystem
+    // ("write" option unavailable), and the dist files were discarded after the
+    // read anyway. Native esbuild supports write:false + outputFiles too.
+    write: false as const,
     define: {
       "process.env.NODE_ENV": production ? '"production"' : '"development"',
     },
@@ -307,29 +313,24 @@ export async function createBundle(
 
     log.info(colors.green("✅ Bundle created successfully"));
 
-    // Read the generated files
-    const jsPath = path.join(process.cwd(), outfile);
-    const cssPath = path.join(process.cwd(), outDir, "bundle.css");
+    const outputFiles = result.outputFiles ?? [];
+    const jsFile = outputFiles.find((f) => f.path.endsWith(".js"));
+    const cssFile = outputFiles.find((f) => f.path.endsWith(".css"));
 
-    if (!fs.existsSync(jsPath)) {
-      throw new Error(`Expected JS bundle at ${jsPath} but file not found`);
+    if (!jsFile) {
+      throw new Error("Expected a JS bundle in esbuild output but none found");
     }
 
-    const jsContent = readTextFileSync(jsPath);
-    const cssContent = fs.existsSync(cssPath)
-      ? readTextFileSync(cssPath)
-      : "";
-
     try {
       fs.rmSync(distDir, { recursive: true });
     } catch {
       //ignore
     }
-    return { js: jsContent, css: cssContent };
-    
+    return { js: jsFile.text, css: cssFile?.text ?? "" };
+
   } finally {
-    // Stop esbuild
-    await esbuild.stop();
+    // Stop the native esbuild service so the process can exit (no-op for wasm).
+    await stopEsbuild();
   }
 }
 

cli/src/commands/app/dev.ts

@@ -437,7 +437,11 @@ async function dev(opts: DevOptions, appFolder?: string) {
   const rawApp = (await yamlParseFile(rawAppPath)) as any;
   const appPath = rawApp?.custom_path ?? "u/unknown/newapp";
 
-  // Dynamically import esbuild only when the dev command is called
+  // Dynamically import esbuild only when the dev command is called.
+  // Native-only here (no esbuild-wasm fallback via getEsbuild): dev is a local
+  // interactive command that relies on context()/watch, whose semantics under
+  // wasm are untested. The host/binary-mismatch fallback covers the bundling
+  // paths that run on workers/CI via `wmill sync push`.
   const esbuild = await import("esbuild");
 
   const host = opts.host ?? DEFAULT_HOST;

cli/src/commands/datatable/datatable.ts

@@ -118,6 +118,61 @@ const migrateCommand = new Command()
   )
   .action(migrateDown as any);
 
+async function create(
+  opts: GlobalOptions & { resource?: string; force?: boolean },
+  name?: string,
+) {
+  const workspace = await resolveWorkspace(opts);
+  await requireLogin(opts);
+  const dtName = name ?? DEFAULT_DATATABLE_NAME;
+
+  const existing = await wmill.listDataTables({
+    workspace: workspace.workspaceId,
+  });
+  if (existing.some((d) => d.name === dtName)) {
+    throw new Error(`Datatable '${dtName}' already exists in this workspace`);
+  }
+  // edit_datatable_config replaces the whole settings object, and fork
+  // metadata on existing datatables can't be read back through the API —
+  // so only touch a non-empty config when explicitly asked to.
+  if (existing.length > 0 && !opts.force) {
+    throw new Error(
+      `Workspace already has datatable(s): ${existing
+        .map((d) => d.name)
+        .join(", ")}. Re-run with --force to add '${dtName}' ` +
+        "(note: fork metadata on existing datatables is not preserved)",
+    );
+  }
+
+  const datatables: Record<
+    string,
+    { database: { resource_type: "postgresql" | "instance"; resource_path?: string } }
+  > = {};
+  for (const d of existing) {
+    datatables[d.name] = {
+      database: {
+        resource_type: d.resource_type as "postgresql" | "instance",
+        resource_path: d.resource_path ?? undefined,
+      },
+    };
+  }
+  datatables[dtName] = opts.resource
+    ? { database: { resource_type: "postgresql", resource_path: opts.resource } }
+    : { database: { resource_type: "instance", resource_path: "datatable_db" } };
+
+  await wmill.editDataTableConfig({
+    workspace: workspace.workspaceId,
+    requestBody: { settings: { datatables } },
+  });
+  log.info(
+    `Datatable '${dtName}' created (${
+      opts.resource
+        ? `postgresql resource ${opts.resource}`
+        : "instance-backed"
+    }). Scripts can now use datatable://${dtName}.`,
+  );
+}
+
 async function serve(
   opts: GlobalOptions & { port?: number; host?: string; password?: string },
 ) {
@@ -147,6 +202,20 @@ const command = new Command()
   )
   .action(run as any)
   .command("migrate", migrateCommand)
+  .command(
+    "create",
+    "register a datatable database in the workspace (default: instance-backed 'main') so scripts can use datatable://<name>",
+  )
+  .arguments("[name:string]")
+  .option(
+    "--resource <resource:string>",
+    "Back the datatable with an existing postgresql resource path instead of the instance database",
+  )
+  .option(
+    "--force",
+    "Allow adding to a workspace that already has datatables (fork metadata on existing ones is not preserved)",
+  )
+  .action(create as any)
   .command(
     "serve",
     "Serve all datatables as a Postgres-wire endpoint (psql, DBeaver, pgAdmin); the client picks the datatable via the database name in its connection string",

cli/src/commands/generate-metadata/generate-metadata.ts

@@ -803,7 +803,7 @@ async function rehashCommand(
 }
 
 const command = new Command()
-  .description("Generate metadata (locks, schemas) for all scripts, flows, and apps")
+  .description("Regenerate stale local locks and script schemas and refresh wmill-lock.yaml content hashes (scripts, flows, apps). Writes local files only, not a deploy. Run it after edits that add or remove imports or change a script's arguments, so the lock, the auto-generated UI schema, and wmill-lock.yaml stay in sync.")
   .arguments("[folder:string]")
   .option("--yes", "Skip confirmation prompt")
   .option("--dry-run", "Show what would be updated without making changes")
@@ -827,9 +827,7 @@ const command = new Command()
     "rehash",
     new Command()
       .description(
-        "Trust on-disk content; rewrite wmill-lock.yaml hashes without backend " +
-        "trips or yaml/lock rewrites. Useful for bootstrapping missing lockfile " +
-        "entries or recovering from older-CLI hash drift."
+        "Refresh wmill-lock.yaml content hashes from the on-disk .lock and .script.yaml without re-resolving dependencies or hitting the backend. Use when those files are already correct and only the hashes need updating: bootstrapping missing entries or recovering from hash drift."
       )
       .arguments("[folder:string]")
       .option("--skip-scripts", "Skip processing scripts")

cli/src/commands/pipeline/pipeline.ts

@@ -0,0 +1,287 @@
+import { Command } from "@cliffy/command";
+import { Table } from "@cliffy/table";
+import { colors } from "@cliffy/ansi/colors";
+
+import { OpenAPI } from "../../../gen/index.ts";
+import * as wmill from "../../../gen/services.gen.ts";
+import { requireLogin } from "../../core/auth.ts";
+import { resolveWorkspace } from "../../core/context.ts";
+import * as log from "../../core/log.ts";
+import { GlobalOptions } from "../../types.ts";
+
+// Mirrors the asset-graph endpoint payload (backend/windmill-api-assets).
+// TODO: the checked-in generated client (cli/gen, last regenerated 2025-04)
+// predates these routes, so we raw-fetch and hand-roll the types. Once
+// `cli/gen` is regenerated (run `cli/gen_wm_client.sh`, which is currently
+// >700 openapi.yaml commits stale and would churn the whole client), replace
+// `apiGet` + these types with the generated `wmill.getAssetsGraph(...)`
+// (operationId getAssetsGraph) and `wmill.listPipelineFolders(...)`
+// (operationId listPipelineFolders).
+type GraphRunnable = {
+  path: string;
+  usage_kind: "script" | "flow" | "job";
+  in_pipeline?: boolean;
+};
+type GraphEdge = {
+  runnable_kind: string;
+  runnable_path: string;
+  asset_kind: string;
+  asset_path: string;
+  access_type?: "r" | "w" | "rw";
+};
+type GraphTrigger =
+  | {
+      trigger_kind: "asset";
+      asset_kind: string;
+      asset_path: string;
+      runnable_kind: string;
+      runnable_path: string;
+    }
+  | {
+      trigger_kind: string;
+      path?: string;
+      runnable_kind: string;
+      runnable_path: string;
+      missing?: boolean;
+    };
+type AssetGraph = {
+  runnables: GraphRunnable[];
+  assets: { kind: string; path: string }[];
+  edges: GraphEdge[];
+  triggers: GraphTrigger[];
+};
+
+async function apiGet<T>(path: string): Promise<T> {
+  const response = await fetch(`${OpenAPI.BASE}${path}`, {
+    headers: { Authorization: `Bearer ${OpenAPI.TOKEN}` },
+  });
+  if (!response.ok) {
+    const body = await response.text();
+    throw new Error(`GET ${path} -> ${response.status}: ${body}`);
+  }
+  return (await response.json()) as T;
+}
+
+async function list(opts: GlobalOptions & { json?: boolean }) {
+  if (opts.json) log.setSilent(true);
+  const workspace = await resolveWorkspace(opts);
+  await requireLogin(opts);
+
+  const items = await apiGet<{ folder: string; script_count: number }[]>(
+    `/w/${workspace.workspaceId}/assets/pipelines`,
+  );
+  if (opts.json) {
+    console.log(JSON.stringify(items));
+  } else if (items.length === 0) {
+    log.info(
+      "No pipelines in this workspace. Mark scripts with a `// pipeline` comment (plus `// on <spec>` triggers) and push them into a folder.",
+    );
+  } else {
+    new Table()
+      .header(["Folder", "Scripts"])
+      .padding(2)
+      .border(true)
+      .body(items.map((p) => [`f/${p.folder}`, String(p.script_count)]))
+      .render();
+  }
+}
+
+const ASSET_KINDS = "s3object,ducklake,datatable,volume";
+
+function assetUri(kind: string, path: string): string {
+  const prefix = kind === "s3object" ? "s3" : kind;
+  return `${prefix}://${path}`;
+}
+
+function shortName(scriptPath: string): string {
+  return scriptPath.split("/").pop() ?? scriptPath;
+}
+
+// Append to a multimap value, creating the bucket on first use. Avoids the
+// O(n^2) spread-rebuild pattern (`map.set(k, [...(map.get(k) ?? []), v])`).
+function pushTo<K, V>(map: Map<K, V[]>, key: K, val: V): void {
+  (map.get(key) ?? map.set(key, []).get(key)!).push(val);
+}
+
+async function show(
+  opts: GlobalOptions & { json?: boolean },
+  folder: string,
+) {
+  if (opts.json) log.setSilent(true);
+  const workspace = await resolveWorkspace(opts);
+  await requireLogin(opts);
+
+  const f = folder.replace(/^f\//, "").replace(/\/$/, "");
+  const graph = await apiGet<AssetGraph>(
+    `/w/${workspace.workspaceId}/assets/graph?folder=${encodeURIComponent(f)}&asset_kinds=${ASSET_KINDS}`,
+  );
+  if (opts.json) {
+    console.log(JSON.stringify(graph));
+    return;
+  }
+  if (graph.runnables.length === 0) {
+    log.info(
+      `No pipeline scripts in f/${f}. Mark scripts with a \`// pipeline\` comment and push them.`,
+    );
+    return;
+  }
+
+  // Index the graph: writes per script, subscribers per asset, native
+  // trigger markers per script, asset subscriptions per script.
+  const writesByScript = new Map<string, string[]>();
+  for (const e of graph.edges) {
+    if (e.access_type === "w" || e.access_type === "rw") {
+      const uri = assetUri(e.asset_kind, e.asset_path);
+      pushTo(writesByScript, e.runnable_path, uri);
+    }
+  }
+  const subsByAsset = new Map<string, string[]>();
+  const subsByScript = new Map<string, string[]>();
+  const nativeByScript = new Map<
+    string,
+    { kind: string; path?: string; missing?: boolean }[]
+  >();
+  for (const t of graph.triggers) {
+    if (t.trigger_kind === "asset") {
+      const at = t as Extract<GraphTrigger, { trigger_kind: "asset" }>;
+      const uri = assetUri(at.asset_kind, at.asset_path);
+      pushTo(subsByAsset, uri, t.runnable_path);
+      pushTo(subsByScript, t.runnable_path, uri);
+    } else {
+      const nt = t as Exclude<GraphTrigger, { trigger_kind: "asset" }>;
+      pushTo(nativeByScript, t.runnable_path, {
+        kind: nt.trigger_kind,
+        path: nt.path,
+        missing: nt.missing,
+      });
+    }
+  }
+
+  function triggerBadges(script: string): string {
+    const out: string[] = [];
+    for (const t of nativeByScript.get(script) ?? []) {
+      if (t.kind === "data_upload") {
+        out.push(colors.magenta("[data upload]"));
+      } else if (t.missing) {
+        out.push(colors.red(`[${t.kind} ✗ missing]`));
+      } else {
+        out.push(colors.yellow(`[${t.kind}${t.path ? ` ${t.path}` : ""}]`));
+      }
+    }
+    return out.length > 0 ? " " + out.join(" ") : "";
+  }
+
+  const printed = new Set<string>();
+  const lines: string[] = [];
+
+  function printScript(script: string, prefix: string, extraOn?: string[]) {
+    const alsoOn =
+      extraOn && extraOn.length > 0
+        ? colors.dim(`  (also on: ${extraOn.join(", ")})`)
+        : "";
+    if (printed.has(script)) {
+      lines.push(
+        `${prefix}${colors.bold(shortName(script))}${colors.dim(" ↻ shown above")}`,
+      );
+      return;
+    }
+    printed.add(script);
+    lines.push(`${prefix}${colors.bold(shortName(script))}${triggerBadges(script)}${alsoOn}`);
+    const childPrefix = prefix.replace(/├─ $/, "│  ").replace(/└─ $/, "   ");
+    const writes = [...(writesByScript.get(script) ?? [])].sort();
+    writes.forEach((uri, i) => {
+      const lastAsset = i === writes.length - 1;
+      const assetBranch = lastAsset ? "└─▶ " : "├─▶ ";
+      lines.push(`${childPrefix}${assetBranch}${colors.cyan(uri)}`);
+      const assetChildPrefix = childPrefix + (lastAsset ? "    " : "│   ");
+      const subs = [...(subsByAsset.get(uri) ?? [])].sort();
+      subs.forEach((sub, j) => {
+        const branch = j === subs.length - 1 ? "└─ " : "├─ ";
+        const otherOn = (subsByScript.get(sub) ?? []).filter((u) => u !== uri);
+        printScript(sub, assetChildPrefix + branch, otherOn);
+      });
+    });
+  }
+
+  // Roots: pipeline scripts that aren't subscribed to any asset — sources
+  // (data upload, schedule, webhook) and manual entries.
+  const roots = graph.runnables
+    .map((r) => r.path)
+    .filter((p) => !(subsByScript.get(p)?.length))
+    .sort();
+
+  // UI-first markers (data_upload, webhook) have no trigger row — the
+  // graph endpoint's trigger enum (schedule/email/kafka/mqtt/nats/postgres/
+  // sqs/gcp) can't surface them, so they only exist as `// on <kind>`
+  // annotations in the script body. Roots are where sources matter, so fetch
+  // just those bodies and lift the marker kinds the canvas would show.
+  //
+  // DRIFT RISK: this regex + MARKER_KINDS is a divergent, partial copy of the
+  // canonical annotation parser. The proper fix is to have the graph endpoint
+  // emit these UI-only markers as trigger rows (a backend change), after which
+  // this whole Promise.all body-fetch can be deleted and read straight from
+  // the response. Until then, keep this list in sync with the canonical parser.
+  const MARKER_KINDS = ["data_upload", "webhook", "email"];
+  await Promise.all(
+    roots.map(async (p) => {
+      const r = graph.runnables.find((x) => x.path === p);
+      if (r?.usage_kind !== "script") return;
+      try {
+        const script = await wmill.getScriptByPath({
+          workspace: workspace.workspaceId,
+          path: p,
+        });
+        const existing = nativeByScript.get(p) ?? [];
+        for (const line of (script.content ?? "").split("\n")) {
+          const m = line.match(/^\s*(?:\/\/|--|#)\s*on\s+(\w+)\s*$/);
+          if (!m) continue;
+          const kind = m[1];
+          if (!MARKER_KINDS.includes(kind)) continue;
+          if (!existing.some((t) => t.kind === kind)) {
+            existing.push({ kind });
+          }
+        }
+        if (existing.length > 0) nativeByScript.set(p, existing);
+      } catch {
+        // body fetch is best-effort enrichment only
+      }
+    }),
+  );
+
+  const scriptCount = graph.runnables.length;
+  const assetCount = graph.assets.length;
+  log.info(
+    colors.bold(`Pipeline f/${f}`) +
+      colors.dim(` — ${scriptCount} script${scriptCount === 1 ? "" : "s"} · ${assetCount} asset${assetCount === 1 ? "" : "s"}`),
+  );
+  lines.push("");
+  for (const root of roots) {
+    printScript(root, "");
+    lines.push("");
+  }
+  // Anything unreachable from the roots (e.g. cycles) still gets listed.
+  for (const r of graph.runnables) {
+    if (!printed.has(r.path)) {
+      printScript(r.path, "");
+      lines.push("");
+    }
+  }
+  console.log(lines.join("\n"));
+}
+
+const command = new Command()
+  .description(
+    "inspect asset-driven pipelines (scripts marked `// pipeline`, wired by `// on <spec>` annotations)",
+  )
+  .command("list", "list pipeline folders in the workspace")
+  .option("--json", "Output as JSON (for piping to jq)")
+  .action(list as any)
+  .command(
+    "show",
+    "render a pipeline folder's DAG (sources, lineage, subscriptions) in the terminal",
+  )
+  .arguments("<folder:string>")
+  .option("--json", "Output the raw asset graph as JSON")
+  .action(show as any);
+
+export default command;

cli/src/commands/script/script.ts

@@ -58,6 +58,7 @@ import { SyncCodebase, listSyncCodebases } from "../../utils/codebase.ts";
 import { pollJobWithQueueLogging } from "../../utils/job_polling.ts";
 import fs from "node:fs";
 import { createTarBlob, type TarEntry } from "../../utils/tar.ts";
+import { getEsbuild } from "../../utils/esbuild_loader.ts";
 
 import { execSync } from "node:child_process";
 import { NewScript, Script, ScriptModule } from "../../../gen/types.gen.ts";
@@ -328,7 +329,7 @@ export async function handleFile(
         }).toString();
         log.info("Custom bundler executed for " + path);
       } else {
-        const esbuild = await import("esbuild");
+        const esbuild = await getEsbuild();
 
         log.info(`Started bundling ${path} ...`);
         const startTime = performance.now();
@@ -1565,7 +1566,7 @@ async function preview(
         maxBuffer: 1024 * 1024 * 50,
       }).toString();
     } else {
-      const esbuild = await import("esbuild");
+      const esbuild = await getEsbuild();
 
       if (!opts.silent) {
         log.info(`Bundling ${filePath} for preview...`);

cli/src/core/constants.ts

@@ -10,4 +10,4 @@ export const WM_FORK_PREFIX = "wm-fork";
 // (e.g. utils.ts) can read it without importing main.ts and creating a circular
 // dependency (main → workspace → utils → main) that triggers a TDZ.
 // Re-exported from main.ts for backwards compatibility.
-export const VERSION = "1.727.0";
+export const VERSION = "1.734.0";

cli/src/guidance/core.ts

@@ -117,6 +117,20 @@ Local previews exist for every entity type and don't deploy:
 
 Argument shapes and per-language details live in the \`write-script-<lang>\`, \`write-flow\`, and \`raw-app\` skills.
 
+## Keeping metadata in sync
+
+After editing a script, flow inline script, or app runnable, its generated metadata can go stale. \`wmill-lock.yaml\` stores a content hash per item, so a change that **adds or removes an import** or **changes a script's arguments** invalidates that hash and leaves the \`.lock\` (resolved dependencies) and \`.script.yaml\` (the input schema that drives the auto-generated args UI) out of date. \`wmill generate-metadata\` regenerates them and refreshes the hashes. Leaving them stale produces spurious diffs in git-sync and CI.
+
+This only writes local files — it is **not** a deploy — but it re-resolves dependencies, so it can bump unpinned versions (the same as deploying from the UI; expected, not a bug). So by default **offer it and run it once the user agrees**, rather than running it silently after every edit. YOU run the command (never tell the user to run it); the choice is only whether to confirm first.
+
+After running it, diff the regenerated lockfiles (e.g. \`git diff\` the \`.lock\` / \`.script.lock\` files): if any dependency versions changed, tell the user what bumped (e.g. \`requests 2.31.0 → 2.32.0\`) so they can catch an unwanted change before deploying. Do this even under \`Metadata: auto\` — it is information, not a confirmation gate. Pin a version in code to keep it fixed.
+
+With no path argument it regenerates only the items whose metadata is actually stale (content hash drifted), workspace-wide — not everything. The set can be larger than the file you edited for two reasons: imports propagate (editing a script that others import marks every importer stale too, so their locks regenerate against the new code — by design, since a lock must reflect the imported code), and any pre-existing drift is swept in. If it touches items you didn't expect, run \`wmill generate-metadata --dry-run\` first — it lists each stale item with a reason (\`content changed\` or \`depends on <path>\`) and changes nothing, so you can see why each is in scope. To narrow it, pass a folder or file path (\`wmill generate-metadata f/foo\`); add \`--strict-folder-boundaries\` to touch only items literally inside that folder (it warns about stale importers outside the folder that it skipped — they resurface as stale on the next unscoped run).
+
+**Save the preference so you don't ask every session.** If the user wants metadata regenerated automatically after edits (or always confirmed first), record it in the **project-specific instructions** section of \`AGENTS.md\` (user-owned — never overwritten by \`wmill refresh prompts\`), e.g. a line like \`Metadata: auto (run wmill generate-metadata after edits)\` or \`Metadata: ask first\`. Read that line first on later sessions and follow it.
+
+If the on-disk \`.lock\` and \`.script.yaml\` are already correct and only \`wmill-lock.yaml\` needs its hashes refreshed (hash drift, or bootstrapping missing entries), use \`wmill generate-metadata rehash\` — it re-records hashes from disk with no backend round-trip and no dependency changes.
+
 ## Deploying
 
 There are two ways local changes reach the workspace. Pick based on how the repo is wired, not habit.

cli/src/main.ts

@@ -51,6 +51,7 @@ import generateMetadata from "./commands/generate-metadata/generate-metadata.ts"
 import docs from "./commands/docs/docs.ts";
 import config from "./commands/config/config.ts";
 import datatable from "./commands/datatable/datatable.ts";
+import pipeline from "./commands/pipeline/pipeline.ts";
 import ducklake from "./commands/ducklake/ducklake.ts";
 import objectStorage from "./commands/object-storage/object-storage.ts";
 import { fetchVersion } from "./core/context.ts";
@@ -77,6 +78,7 @@ export {
   docs,
   config,
   datatable,
+  pipeline,
   ducklake,
   objectStorage,
   hubPull,
@@ -215,6 +217,7 @@ const command = new Command()
   .command("docs", docs)
   .command("config", config)
   .command("datatable", datatable)
+  .command("pipeline", pipeline)
   .command("ducklake", ducklake)
   .command("object-storage", objectStorage)
   .command("version --version", "Show version information")

cli/src/utils/esbuild_loader.ts

@@ -0,0 +1,228 @@
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import process from "node:process";
+import { createGunzip } from "node:zlib";
+import { Readable } from "node:stream";
+import { pathToFileURL } from "node:url";
+import * as tar from "tar-stream";
+import * as log from "../core/log.ts";
+
+// esbuild splits into a JS host package and a per-platform native binary
+// (@esbuild/<platform>). They must be the same version. A broken or incremental
+// install can leave the on-disk binary at a different version than the pinned
+// host, which crashes service start with
+//   Cannot start service: Host version "X" does not match binary version "Y"
+// The running code can't fix what npm/bun put on disk, so when that happens we
+// fall back to esbuild-wasm, whose binary is a single version-pinned .wasm. To
+// keep that 14MB out of every CLI install, the esbuild-wasm package is not a
+// dependency: it is downloaded once and cached on disk, on the fallback path
+// only. We download the whole package (not just the .wasm) because esbuild-wasm
+// reads the app's files from disk by spawning `node bin/esbuild`, which needs
+// bin/esbuild + esbuild.wasm + wasm_exec*.js co-located on disk.
+
+type Esbuild = typeof import("esbuild");
+
+// Version to fall back to if the native host's version can't be read. Keep in
+// sync with the "esbuild" pin in cli/package.json.
+const FALLBACK_VERSION = "0.28.0";
+
+let cached: Esbuild | undefined;
+let inFlight: Promise<Esbuild> | undefined;
+// Distinguishes concurrent extraction temp dirs within a process.
+let extractCounter = 0;
+
+/**
+ * Returns a working esbuild module, preferring the native binary and falling
+ * back to esbuild-wasm only when the native host/binary versions don't match.
+ * Memoized for the process: concurrent first callers (e.g. a parallel
+ * `wmill sync push`) share one probe/download instead of each running their own.
+ */
+export function getEsbuild(): Promise<Esbuild> {
+  if (cached) return Promise.resolve(cached);
+  if (inFlight) return inFlight;
+  inFlight = acquireEsbuild()
+    .then((esbuild) => {
+      cached = esbuild;
+      return esbuild;
+    })
+    .finally(() => {
+      inFlight = undefined;
+    });
+  return inFlight;
+}
+
+async function acquireEsbuild(): Promise<Esbuild> {
+  // Escape hatch: skip native entirely (e.g. a host known to have a broken
+  // install, or to exercise the fallback path).
+  if (process.env.WINDMILL_FORCE_ESBUILD_WASM) {
+    return loadWasmEsbuild(await nativeHostVersion());
+  }
+
+  try {
+    const esbuild = await import("esbuild");
+    // The native service only starts on the first call; force it with the most
+    // trivial op so any breakage (host/binary version mismatch, a dead service)
+    // surfaces now rather than mid-build. The mismatch detail is printed to the
+    // child's stderr while the thrown error is generic ("service was stopped"),
+    // so we fall back on ANY smoke-test failure rather than matching a string.
+    await esbuild.transform("");
+    return esbuild;
+  } catch (e) {
+    const msg = e instanceof Error ? e.message : String(e);
+    log.warn(
+      `native esbuild is not usable; falling back to esbuild-wasm (${msg.trim()})`
+    );
+  }
+
+  return loadWasmEsbuild(await nativeHostVersion());
+}
+
+/**
+ * Stops the esbuild service (native or wasm — both spawn a child process) so the
+ * process can exit. Safe to call repeatedly; the service restarts lazily on the
+ * next build.
+ */
+export async function stopEsbuild(): Promise<void> {
+  await cached?.stop();
+}
+
+async function nativeHostVersion(): Promise<string> {
+  try {
+    return (await import("esbuild")).version ?? FALLBACK_VERSION;
+  } catch {
+    return FALLBACK_VERSION;
+  }
+}
+
+async function loadWasmEsbuild(version: string): Promise<Esbuild> {
+  const pkgDir = await ensureWasmPackage(version);
+  const mainJs = path.join(pkgDir, "lib", "main.js");
+  // The Node build (lib/main.js) reads app files from disk by spawning
+  // `node bin/esbuild`, so it works with on-disk entry points and node_modules,
+  // unlike the browser build.
+  return (await import(pathToFileURL(mainJs).href)) as unknown as Esbuild;
+}
+
+/**
+ * Returns a directory containing an extracted esbuild-wasm package (with
+ * lib/main.js). Uses an explicit override, then an on-disk cache, then downloads
+ * and extracts the npm tarball.
+ */
+async function ensureWasmPackage(version: string): Promise<string> {
+  // Explicit local override wins (air-gapped / self-hosted workers): a path to
+  // an already-extracted esbuild-wasm package directory.
+  const override = process.env.WINDMILL_ESBUILD_WASM_PATH;
+  if (override) return override;
+
+  const destDir = path.join(cacheDir(), `esbuild-wasm-${version}`);
+  if (fs.existsSync(path.join(destDir, "lib", "main.js"))) {
+    return destDir;
+  }
+
+  const url = process.env.WINDMILL_ESBUILD_WASM_URL ??
+    `https://registry.npmjs.org/esbuild-wasm/-/esbuild-wasm-${version}.tgz`;
+  log.info(`Downloading esbuild-wasm@${version} from ${url} ...`);
+  const res = await fetch(url);
+  if (!res.ok || !res.body) {
+    throw new Error(
+      `Failed to download esbuild-wasm@${version} (${res.status} ${res.statusText}). ` +
+        `Set WINDMILL_ESBUILD_WASM_PATH to an extracted esbuild-wasm package dir, ` +
+        `point WINDMILL_ESBUILD_WASM_URL at a reachable tarball, or repair the native esbuild install.`
+    );
+  }
+
+  // Extract to a unique temp dir and rename into place so a crash or a
+  // concurrent writer can't leave a half-extracted package behind, and so two
+  // extractions never share an in-progress directory.
+  const tmpDir = `${destDir}.${process.pid}.${extractCounter++}.tmp`;
+  fs.rmSync(tmpDir, { recursive: true, force: true });
+  await extractTarball(res.body, tmpDir);
+  if (!fs.existsSync(path.join(tmpDir, "lib", "main.js"))) {
+    fs.rmSync(tmpDir, { recursive: true, force: true });
+    throw new Error(`esbuild-wasm@${version} tarball did not contain lib/main.js`);
+  }
+  try {
+    fs.renameSync(tmpDir, destDir);
+  } catch {
+    // Another process won the race, or rename across devices failed; clean up
+    // and let the existsSync check below decide whether the cache is usable.
+    fs.rmSync(tmpDir, { recursive: true, force: true });
+  }
+  if (!fs.existsSync(path.join(destDir, "lib", "main.js"))) {
+    throw new Error(`Failed to cache esbuild-wasm@${version} at ${destDir}`);
+  }
+  return destDir;
+}
+
+/**
+ * Resolves a tar entry to an absolute path inside destDir, stripping the leading
+ * "package/" component that npm tarballs use. Returns null if the entry would
+ * escape destDir (tar-slip), since WINDMILL_ESBUILD_WASM_URL allows untrusted
+ * tarball sources.
+ */
+export function resolveTarEntryPath(
+  destDir: string,
+  entryName: string
+): string | null {
+  const rel = entryName.replace(/^[^/]+\//, "");
+  const root = path.resolve(destDir);
+  const outPath = path.resolve(root, rel);
+  if (outPath !== root && !outPath.startsWith(root + path.sep)) {
+    return null;
+  }
+  return outPath;
+}
+
+// Extracts an npm tarball (gzipped tar) into destDir, stripping the leading
+// "package/" path component that npm tarballs use.
+async function extractTarball(
+  body: ReadableStream<Uint8Array>,
+  destDir: string
+): Promise<void> {
+  const extract = tar.extract();
+  extract.on("entry", (header, stream, next) => {
+    if (header.type !== "file") {
+      stream.resume();
+      stream.on("end", next);
+      return;
+    }
+    const outPath = resolveTarEntryPath(destDir, header.name);
+    if (!outPath) {
+      // Reject tar-slip entries that would write outside the cache dir.
+      stream.resume();
+      stream.on("end", () =>
+        next(new Error(`unsafe path in esbuild-wasm tarball: ${header.name}`))
+      );
+      return;
+    }
+    fs.mkdirSync(path.dirname(outPath), { recursive: true });
+    const ws = fs.createWriteStream(outPath, { mode: header.mode ?? 0o644 });
+    stream.pipe(ws);
+    ws.on("finish", next);
+    ws.on("error", next);
+    stream.on("error", next);
+  });
+
+  await new Promise<void>((resolve, reject) => {
+    extract.on("finish", resolve);
+    extract.on("error", reject);
+    Readable.fromWeb(body as unknown as Parameters<typeof Readable.fromWeb>[0])
+      .pipe(createGunzip())
+      .on("error", reject)
+      .pipe(extract)
+      .on("error", reject);
+  });
+}
+
+function cacheDir(): string {
+  const explicit = process.env.WINDMILL_CACHE_DIR;
+  if (explicit) return explicit;
+  const xdg = process.env.XDG_CACHE_HOME;
+  if (xdg) return path.join(xdg, "windmill");
+  try {
+    return path.join(os.homedir(), ".cache", "windmill");
+  } catch {
+    return path.join(os.tmpdir(), "windmill");
+  }
+}

cli/src/utils/local_path_scripts.ts

@@ -1,6 +1,7 @@
 import { execFileSync } from "node:child_process";
 import { stat } from "node:fs/promises";
 import { readTextFile } from "./utils.ts";
+import { getEsbuild } from "./esbuild_loader.ts";
 import type { SyncCodebase } from "./codebase.ts";
 import { parseMetadataFileIfExists } from "./metadata.ts";
 import { inferContentTypeFromFilePath } from "./script_common.ts";
@@ -43,7 +44,7 @@ async function bundleSingleFileCodebaseScript(
     ).toString();
   }
 
-  const esbuild = await import("esbuild");
+  const esbuild = await getEsbuild();
   const out = await esbuild.build({
     entryPoints: [filePath],
     // Inline rawscripts are executed through the standard module wrapper,

cli/src/utils/script_common.ts

@@ -110,8 +110,16 @@ export function inferContentTypeFromFilePath(
     return "rlang";
 	// for related places search: ADD_NEW_LANG
   } else {
+    const ext = contentPath.substring(contentPath.lastIndexOf("."));
+    let hint = "";
+    if (ext === ".sql") {
+      hint =
+        "\nBare .sql is ambiguous — use a dialect extension: .pg.sql (postgresql), .my.sql (mysql), .bq.sql (bigquery), .sf.sql (snowflake), .ms.sql (mssql), .odb.sql (oracledb), .duckdb.sql (duckdb)";
+    }
     throw new Error(
-      "Invalid language: " + contentPath.substring(contentPath.lastIndexOf("."))
+      `Cannot infer script language from extension '${ext}' (file ${contentPath}).` +
+        hint +
+        "\nSupported extensions: .ts (bun/deno), .py, .go, .sh, .ps1, .php, .rs, .cs, .nu, .java, .rb, .r, .gql, .playbook.yml, .pg.sql, .my.sql, .bq.sql, .sf.sql, .ms.sql, .odb.sql, .duckdb.sql"
     );
   }
 }