refactor: dedup asset-graph code, squash migrations, drop artifacts

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-6aaddd80f8c07cfafea2021c1879c3d7b2fb156a43299e9a1209d05293c4f50f.json

@@ -0,0 +1,64 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO dispatch_event (\n             workspace_id, producer_job_id, subscriber_path,\n             asset_kind, asset_path, outcome,\n             child_job_id, partition,\n             received_inputs, required_inputs,\n             debounce_s, reason\n           )\n           SELECT $1, $2, sp, ak, ap, oc, cj, pt, ri, rq, db, rs\n           FROM unnest(\n             $3::text[], $4::ASSET_KIND[], $5::text[], $6::DISPATCH_OUTCOME[],\n             $7::uuid[], $8::text[], $9::int[], $10::int[], $11::int[], $12::text[]\n           ) AS t(sp, ak, ap, oc, cj, pt, ri, rq, db, rs)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Uuid",
+        "TextArray",
+        {
+          "Custom": {
+            "name": "asset_kind[]",
+            "kind": {
+              "Array": {
+                "Custom": {
+                  "name": "asset_kind",
+                  "kind": {
+                    "Enum": [
+                      "s3object",
+                      "resource",
+                      "variable",
+                      "ducklake",
+                      "datatable",
+                      "volume"
+                    ]
+                  }
+                }
+              }
+            }
+          }
+        },
+        "TextArray",
+        {
+          "Custom": {
+            "name": "dispatch_outcome[]",
+            "kind": {
+              "Array": {
+                "Custom": {
+                  "name": "dispatch_outcome",
+                  "kind": {
+                    "Enum": [
+                      "dispatched",
+                      "join_pending",
+                      "skipped"
+                    ]
+                  }
+                }
+              }
+            }
+          }
+        },
+        "UuidArray",
+        "TextArray",
+        "Int4Array",
+        "Int4Array",
+        "Int4Array",
+        "TextArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6aaddd80f8c07cfafea2021c1879c3d7b2fb156a43299e9a1209d05293c4f50f"
+}

backend/.sqlx/query-d3d87b9a4d62977dea5af95dd457bb759229805cff771cfa4275a46bfc80e1ab.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job\n         SET args = CASE\n                      WHEN args ? 'partition'\n                      THEN $1 || jsonb_build_object('partition', args -> 'partition')\n                      ELSE $1\n                    END,\n             preprocessed = TRUE\n         WHERE id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Jsonb",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d3d87b9a4d62977dea5af95dd457bb759229805cff771cfa4275a46bfc80e1ab"
+}

backend/.sqlx/query-f7790aaa332e81046ececc71b728364a3489a3c9151ac153c419ddce51d0d891.json

@@ -1,50 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO dispatch_event (\n             workspace_id, producer_job_id, subscriber_path,\n             asset_kind, asset_path, outcome,\n             child_job_id, partition,\n             received_inputs, required_inputs,\n             debounce_s, reason\n           ) VALUES ($1, $2, $3, $4::ASSET_KIND, $5, $6::DISPATCH_OUTCOME,\n                     $7, $8, $9, $10, $11, $12)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Uuid",
-        "Varchar",
-        {
-          "Custom": {
-            "name": "asset_kind",
-            "kind": {
-              "Enum": [
-                "s3object",
-                "resource",
-                "variable",
-                "ducklake",
-                "datatable",
-                "volume"
-              ]
-            }
-          }
-        },
-        "Text",
-        {
-          "Custom": {
-            "name": "dispatch_outcome",
-            "kind": {
-              "Enum": [
-                "dispatched",
-                "join_pending",
-                "skipped"
-              ]
-            }
-          }
-        },
-        "Uuid",
-        "Text",
-        "Int4",
-        "Int4",
-        "Int4",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "f7790aaa332e81046ececc71b728364a3489a3c9151ac153c419ddce51d0d891"
-}

backend/migrations/20260423050000_script_trigger.up.sql

@@ -6,9 +6,18 @@
 -- `// on <kind> <ref>` lines (every non-integration trigger kind; their
 -- trigger_ref is the trigger row path, or empty for marker-only forms).
 --
+-- Per-edge columns that are in fact script-level properties (every row for
+-- a given runnable carries the same value, set once at deploy) but live on
+-- the edge so the dispatcher reads everything from a single query:
+--   join_all      `// trigger all` AND-join barrier (else OR, the default).
+--   debounce_s    `// on debounce=` (else script-level `// debounce`); only
+--                 asset-cascade edges carry one. NULL = no debounce.
+--   retry_*       `// retry <count> [<delay>]` cascade retry. NULL = none.
+--
 -- The idempotency guards (IF NOT EXISTS / duplicate_object) are load-bearing:
 -- this migration squashes several pre-release ones, so databases migrated
--- from the unsquashed history already contain the final objects.
+-- from the unsquashed history already contain the final objects and
+-- re-applying must be a no-op.
 DO $$ BEGIN
     CREATE TYPE SCRIPT_TRIGGER_KIND AS ENUM (
         'asset', 'schedule', 'webhook', 'email', 'kafka', 'mqtt', 'nats',
@@ -22,7 +31,11 @@ CREATE TABLE IF NOT EXISTS script_trigger (
   runnable_kind  ASSET_USAGE_KIND NOT NULL,
   runnable_path  VARCHAR(255) NOT NULL,
   trigger_kind   SCRIPT_TRIGGER_KIND NOT NULL,
-  trigger_ref    TEXT NOT NULL
+  trigger_ref    TEXT NOT NULL,
+  join_all       BOOLEAN NOT NULL DEFAULT FALSE,
+  debounce_s     INTEGER,
+  retry_count    SMALLINT,
+  retry_delay_s  INTEGER
 );
 
 -- Per-runnable lookup (wipe-on-deploy, list-triggers-for-script).

backend/migrations/20260516190545_add_script_trigger_join_all.down.sql

@@ -1,2 +0,0 @@
-ALTER TABLE script_trigger
-  DROP COLUMN join_all;

backend/migrations/20260516190545_add_script_trigger_join_all.up.sql

@@ -1,9 +0,0 @@
--- AND-join barrier (`// trigger all`). When true, the subscriber runs only
--- once every partition-bearing input has materialized at the same partition
--- (plus every reference input exists), rather than firing on any input (OR,
--- the default). Stored per script_trigger row — it is a script-level
--- property so every row for a given (workspace, runnable) carries the same
--- value; this matches the wipe-and-reinsert-on-deploy pattern and keeps the
--- subscriber lookup a single query.
-ALTER TABLE script_trigger
-  ADD COLUMN join_all BOOLEAN NOT NULL DEFAULT FALSE;

backend/migrations/20260516201317_script_trigger_debounce_s.down.sql

@@ -1,2 +0,0 @@
-ALTER TABLE script_trigger
-  DROP COLUMN debounce_s;

backend/migrations/20260516201317_script_trigger_debounce_s.up.sql

@@ -1,8 +0,0 @@
--- Opt-in debounce window (seconds) for an asset-cascade subscriber edge.
--- NULL = no debounce (fan-out, the default/current behaviour). Resolved
--- per `// on` edge at deploy as: edge `debounce=<dur>` ?? script-level
--- `// debounce <dur>` ?? none. Only asset subscriber rows carry it; the
--- dispatcher builds DebouncingSettings keyed (subscriber, partition) so
--- distinct partitions never collapse and "latest in window" falls out.
-ALTER TABLE script_trigger
-  ADD COLUMN debounce_s INTEGER;

backend/migrations/20260523054912_script_trigger_retry.down.sql

@@ -1,3 +0,0 @@
-ALTER TABLE script_trigger
-  DROP COLUMN retry_count,
-  DROP COLUMN retry_delay_s;

backend/migrations/20260523054912_script_trigger_retry.up.sql

@@ -1,10 +0,0 @@
--- Cascade retry policy (`// retry <count> [<delay>]`). Stored per
--- `script_trigger` row — like `join_all`, it is a script-level property
--- (every row for a given runnable carries the same value) but lives
--- alongside the edge so the dispatcher reads everything it needs from a
--- single query. NULL = no retry (current behaviour); set = re-run on
--- failure up to `retry_count` times, waiting `retry_delay_s` seconds
--- between attempts (0 = back-to-back).
-ALTER TABLE script_trigger
-  ADD COLUMN retry_count   SMALLINT,
-  ADD COLUMN retry_delay_s INTEGER;

backend/migrations/20260523055641_dispatch_event.up.sql

@@ -9,13 +9,19 @@
 -- retention sweep (monitor.rs delete_expired_jobs_batch -> DELETE FROM
 -- v2_job WHERE id = ANY(...)) reaps these rows along with their producer.
 -- No separate cleanup path needed.
-CREATE TYPE DISPATCH_OUTCOME AS ENUM (
-  'dispatched',
-  'join_pending',
-  'skipped'
-);
+--
+-- Idempotency guards (duplicate_object / IF NOT EXISTS) are load-bearing:
+-- re-applying this migration after a squash must be a no-op.
+DO $$ BEGIN
+  CREATE TYPE DISPATCH_OUTCOME AS ENUM (
+    'dispatched',
+    'join_pending',
+    'skipped'
+  );
+EXCEPTION WHEN duplicate_object THEN NULL;
+END $$;
 
-CREATE TABLE dispatch_event (
+CREATE TABLE IF NOT EXISTS dispatch_event (
   id              BIGSERIAL PRIMARY KEY,
   workspace_id    VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
   producer_job_id UUID NOT NULL REFERENCES v2_job(id) ON DELETE CASCADE,
@@ -41,5 +47,17 @@ CREATE TABLE dispatch_event (
 
 -- Primary access pattern: list events for one producer (the job detail
 -- panel). Ordered scans by id give chronological order for free.
-CREATE INDEX idx_dispatch_event_producer
+CREATE INDEX IF NOT EXISTS idx_dispatch_event_producer
   ON dispatch_event (producer_job_id, id);
+
+-- Backs the asset-graph edge listing (jobs.rs list_asset_dispatch_edges):
+--   WHERE workspace_id = $1 AND subscriber_path LIKE 'prefix%'
+--         AND created_at >= $3
+--   ORDER BY created_at DESC, id DESC
+-- The (producer_job_id, id) index above doesn't help this access path, so
+-- without this one a high-volume dispatch_event seq-scans + sorts.
+-- text_pattern_ops makes the anchored LIKE prefix (built as `path_start || '%'`)
+-- index-usable regardless of the column collation; created_at DESC matches
+-- the ORDER BY so Postgres can satisfy ordering from the index.
+CREATE INDEX IF NOT EXISTS idx_dispatch_event_subscriber
+  ON dispatch_event (workspace_id, subscriber_path text_pattern_ops, created_at DESC);

backend/migrations/20260523091727_backfill_asset_from_script_assets.down.sql

@@ -1,6 +0,0 @@
--- No-op: the backfill is a one-shot data migration. We can't safely
--- distinguish backfilled rows from those a subsequent deploy inserted via
--- the normal `insert_static_asset_usage` path (same tuple, no marker), so
--- rolling back would risk deleting live data. Leaving the rows in place on
--- downgrade is harmless — they'll get rewritten on the next deploy of
--- each affected script.

backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql

@@ -1,39 +0,0 @@
--- Backfill the `asset` table from `script.assets` JSONB for scripts that
--- were deployed before the deploy path started populating asset rows from
--- the parsed `ns.assets`. The JSONB column carried the parser's findings
--- the whole time; the corresponding asset rows are what `fetch_producer_writes`
--- (asset-trigger cascade) and the asset-graph lineage view actually query.
--- Without this, a pre-feature script "succeeds" but the dispatcher sees
--- no writes, no subscribers are matched, and the dispatch_event panel
--- stays empty.
---
--- Idempotent: scoped to (workspace, path) pairs that have ZERO asset rows
--- under `usage_kind = 'script'`, so re-running can't duplicate. The
--- (workspace_id, path, kind, usage_path, usage_kind) primary key catches
--- any residual overlap via `ON CONFLICT DO NOTHING` as a belt-and-braces.
--- Skips archived/deleted script versions so we backfill from the *current*
--- snapshot, matching what the live deploy path would write.
-INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind)
-SELECT
-  s.workspace_id,
-  a->>'path',
-  (a->>'kind')::asset_kind,
-  (a->>'access_type')::asset_access_type,
-  s.path,
-  'script'::asset_usage_kind
-FROM script s
-CROSS JOIN LATERAL jsonb_array_elements(s.assets) AS a
-WHERE s.archived = false
-  AND s.deleted = false
-  AND s.assets IS NOT NULL
-  AND jsonb_typeof(s.assets) = 'array'
-  AND a->>'path' IS NOT NULL
-  AND a->>'kind' IS NOT NULL
-  AND NOT EXISTS (
-    SELECT 1
-    FROM asset existing
-    WHERE existing.workspace_id = s.workspace_id
-      AND existing.usage_kind = 'script'
-      AND existing.usage_path = s.path
-  )
-ON CONFLICT DO NOTHING;

backend/parsers/windmill-parser/src/asset_parser.rs

@@ -281,13 +281,12 @@ pub fn merge_assets(assets: Vec<ParseAssetsResult>) -> Vec<ParseAssetsResult> {
             .iter_mut()
             .find(|x| x.path == asset.path && x.kind == asset.kind)
         {
-            // merge access types
+            // merge access types — a None on either side means ambiguous
+            // usage (unknown access), which poisons the merge to None;
+            // otherwise delegate to the shared truth table.
             existing.access_type = match (asset.access_type, existing.access_type) {
                 (None, _) | (_, None) => None,
-                (Some(R), Some(W)) | (Some(W), Some(R)) => Some(RW),
-                (Some(RW), _) | (_, Some(RW)) => Some(RW),
-                (Some(R), Some(R)) => Some(R),
-                (Some(W), Some(W)) => Some(W),
+                (Some(a), Some(b)) => Some(merge_access_types(a, b)),
             };
             // merge columns: union the column sets and merge access types per column
             existing.columns = merge_column_maps(existing.columns.take(), asset.columns);
@@ -391,10 +390,11 @@ fn unquote(s: &str) -> Option<&str> {
 // `ident=` token — the same assumption `// partitioned` already makes.
 fn split_trailing_kv_opts(s: &str) -> (&str, BTreeMap<String, String>) {
     let mut split_at: Option<usize> = None;
-    let mut pos = 0usize;
     for tok in s.split_whitespace() {
-        let tok_start = s[pos..].find(tok).map(|o| pos + o).unwrap_or(pos);
-        pos = tok_start + tok.len();
+        // `split_whitespace` yields slices borrowed from `s`, so the exact
+        // byte offset is the pointer delta — substring search (`find`) would
+        // misfire when an earlier token also appears inside a later one.
+        let tok_start = tok.as_ptr() as usize - s.as_ptr() as usize;
         if let Some(eq) = tok.find('=') {
             let key = &tok[..eq];
             if !key.is_empty()
@@ -489,6 +489,20 @@ fn parse_kv_opts(s: &str) -> BTreeMap<String, String> {
 // multiple lines declare them, the first one is kept (last would be
 // reasonable too, but first matches the file-top convention developers
 // follow).
+// Try to consume `<kw>` as a complete word from `rest`. Returns the trailing
+// text after the keyword if it matched (empty or whitespace-bounded),
+// `None` otherwise. Prevents `partitioned` matching `partition`, `pipelines`
+// matching `pipeline`, etc. Mirrors `consumeKeyword` in
+// parsePipelineAnnotations.ts.
+fn consume_keyword<'a>(rest: &'a str, kw: &str) -> Option<&'a str> {
+    let after = rest.strip_prefix(kw)?;
+    if after.is_empty() || after.starts_with(|c: char| c.is_whitespace()) {
+        Some(after)
+    } else {
+        None
+    }
+}
+
 pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
     let mut out = PipelineAnnotations::default();
 
@@ -505,7 +519,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
         };
         let rest = rest.trim_start();
 
-        if let Some(after_kw) = rest.strip_prefix("pipeline") {
+        if let Some(after_kw) = consume_keyword(rest, "pipeline") {
             // Strict: keyword must be the only content on the line. Rejects
             // `pipeline broken`, `pipelines`, `pipeline-related`, etc.
             if after_kw.trim().is_empty() {
@@ -514,10 +528,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("partitioned") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "partitioned") {
             if out.partition.is_none() {
                 if let Some(spec) = parse_partitioned_spec(after_kw.trim()) {
                     out.partition = Some(spec);
@@ -526,10 +537,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("freshness") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "freshness") {
             let dur = after_kw.trim();
             if !dur.is_empty() && out.freshness.is_none() {
                 out.freshness = Some(FreshnessSpec { duration: dur.to_string() });
@@ -537,10 +545,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("trigger") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "trigger") {
             match after_kw.trim() {
                 "all" => out.join_mode = JoinMode::All,
                 "any" => out.join_mode = JoinMode::Any,
@@ -550,10 +555,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("debounce") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "debounce") {
             let dur = after_kw.trim();
             if !dur.is_empty() && out.debounce_default.is_none() {
                 out.debounce_default = Some(dur.to_string());
@@ -561,10 +563,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("tag") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "tag") {
             let name = after_kw.trim();
             if !name.is_empty() && out.tag.is_none() {
                 out.tag = Some(name.to_string());
@@ -572,10 +571,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("retry") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "retry") {
             if out.retry.is_none() {
                 if let Some(spec) = parse_retry_spec(after_kw.trim()) {
                     out.retry = Some(spec);
@@ -584,10 +580,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
             continue;
         }
 
-        if let Some(after_kw) = rest.strip_prefix("on") {
-            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
-                continue;
-            }
+        if let Some(after_kw) = consume_keyword(rest, "on") {
             let spec_text = after_kw.trim();
             if spec_text.is_empty() {
                 continue;
@@ -1082,6 +1075,33 @@ mod pipeline_annotation_tests {
         assert_eq!(r.delay.as_deref(), Some("5s"));
     }
 
+    #[test]
+    fn split_trailing_kv_opts_uses_exact_token_offset() {
+        // Regression for the token-offset computation in
+        // `split_trailing_kv_opts`. The asset ref's path token contains the
+        // exact text of the trailing `debounce=60s` opt as a substring, and
+        // the same `debounce=60s` text also appears a second time before the
+        // real opt. Offsetting by the `&str` slice's pointer is exact; a
+        // substring scan (`find`) is what this guards against regressing to.
+        let (ref_part, opts) = split_trailing_kv_opts("s3://lake/debounce=60s/raw debounce=60s");
+        assert_eq!(ref_part, "s3://lake/debounce=60s/raw");
+        assert_eq!(opts.get("debounce").map(String::as_str), Some("60s"));
+
+        // End-to-end through the annotation parser: the ref must parse to the
+        // full S3 path (not truncated at the embedded `debounce=`), and the
+        // per-edge debounce override must be picked up from the trailing opt.
+        let out = parse_pipeline_annotations("// on s3://lake/debounce=60s/raw debounce=90s");
+        assert_eq!(out.triggers.len(), 1);
+        match &out.triggers[0] {
+            TriggerSpec::Asset { asset_kind, path, debounce } => {
+                assert_eq!(*asset_kind, AssetKind::S3Object);
+                assert_eq!(path, "lake/debounce=60s/raw");
+                assert_eq!(debounce.as_deref(), Some("90s"));
+            }
+            other => panic!("expected asset trigger, got {other:?}"),
+        }
+    }
+
     #[test]
     fn kv_opts_quoted_with_spaces() {
         let m = parse_kv_opts("a=\"hello world\" b=plain c='single quoted'");

backend/windmill-api/src/db.rs

@@ -282,6 +282,12 @@ pub async fn migrate(
         20260207000002,
         20260207000003,
         20260207000004,
+        // Squashed pre-release pipeline migrations: the per-column ALTERs on
+        // script_trigger and the dispatch_event subscriber index were folded
+        // back into these two CREATEs, changing their checksum. Both are
+        // idempotent, so re-applying on an already-migrated DB is a no-op.
+        20260423050000,
+        20260523055641,
     ];
     for m in migrator.migrations.iter() {
         if m.migration_type.is_down_migration() {

backend/windmill-common/src/assets.rs

@@ -211,13 +211,10 @@ pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, String)> {
 pub fn trigger_spec_to_row(spec: &TriggerSpec) -> Option<(ScriptTriggerKind, String)> {
     match spec {
         TriggerSpec::Asset { asset_kind, path, .. } => {
-            let prefix = match asset_kind {
-                windmill_parser::asset_parser::AssetKind::S3Object => "s3://",
-                windmill_parser::asset_parser::AssetKind::Resource => "$res:",
-                windmill_parser::asset_parser::AssetKind::Ducklake => "ducklake://",
-                windmill_parser::asset_parser::AssetKind::DataTable => "datatable://",
-                windmill_parser::asset_parser::AssetKind::Volume => "volume://",
-            };
+            // Single source of truth for the canonical prefix lives on the
+            // common AssetKind; map the parser kind across first. The parser
+            // enum has no Variable variant, so canonical_prefix is always Some.
+            let prefix = asset_kind_from_parser(*asset_kind).canonical_prefix()?;
             Some((ScriptTriggerKind::Asset, format!("{}{}", prefix, path)))
         }
         // Schedule joins the native-trigger family — no script_trigger row

backend/windmill-common/src/partition.rs

@@ -14,8 +14,14 @@
 //! - `dynamic key=<path>`: extracted from the triggering payload via a
 //!   minimal `$.a.b` JSON path (the realistic per-tenant/shard/event case).
 
+use std::collections::HashMap;
+
 use chrono::{DateTime, NaiveDate, Utc};
 use chrono_tz::Tz;
+use serde_json::value::RawValue;
+use sqlx::types::Json;
+use sqlx::PgExecutor;
+use uuid::Uuid;
 use windmill_parser::asset_parser::{PartitionKind, PartitionSpec};
 
 use crate::error::{Error, Result};
@@ -24,6 +30,60 @@ use crate::error::{Error, Result};
 /// mirrored at `trigger.partition` for cascade propagation).
 pub const PARTITION_ARG: &str = "partition";
 
+/// Persist a freshly resolved `partition` into a job's `v2_job.args`,
+/// merging it on top of whatever args already exist.
+///
+/// INVARIANT: partition, once resolved, is immutable for the job's
+/// lifetime. This is the single place that *establishes* it (run-start
+/// resolution); every later args rewrite must preserve it — see
+/// [`merge_args_preserving_partition`]. Keeping both the set and the
+/// preserve here keeps the JSONB shape and the invariant in one module.
+pub async fn set_resolved_partition<'e>(
+    executor: impl PgExecutor<'e>,
+    job_id: Uuid,
+    value: &str,
+) -> Result<()> {
+    sqlx::query!(
+        "UPDATE v2_job SET args = coalesce(args, '{}'::jsonb) \
+         || jsonb_build_object('partition', $1::text) WHERE id = $2",
+        value,
+        job_id,
+    )
+    .execute(executor)
+    .await?;
+    Ok(())
+}
+
+/// Replace a job's `v2_job.args` with `new_args`, but carry forward any
+/// existing top-level `partition` key so a wholesale rewrite (e.g. the
+/// preprocessor swapping in preprocessed args) cannot clobber a resolved
+/// partition.
+///
+/// INVARIANT: partition, once resolved, is immutable for the job's
+/// lifetime (set by [`set_resolved_partition`]); this guards every later
+/// args rewrite against dropping it.
+pub async fn merge_args_preserving_partition<'e>(
+    executor: impl PgExecutor<'e>,
+    job_id: Uuid,
+    new_args: HashMap<String, Box<RawValue>>,
+) -> Result<()> {
+    sqlx::query!(
+        "UPDATE v2_job
+         SET args = CASE
+                      WHEN args ? 'partition'
+                      THEN $1 || jsonb_build_object('partition', args -> 'partition')
+                      ELSE $1
+                    END,
+             preprocessed = TRUE
+         WHERE id = $2",
+        Json(new_args) as Json<HashMap<String, Box<RawValue>>>,
+        job_id,
+    )
+    .execute(executor)
+    .await?;
+    Ok(())
+}
+
 fn default_format(kind: &PartitionKind) -> &'static str {
     match kind {
         PartitionKind::Hourly => "%Y-%m-%dT%H",
@@ -40,7 +100,7 @@ fn default_format(kind: &PartitionKind) -> &'static str {
 /// `spec.start` (anchor: older partitions are not backfilled). Errors on an
 /// invalid `tz` or `start` rather than silently materializing the wrong
 /// partition. `Dynamic` is rejected — use [`extract_dynamic_partition`].
-pub fn resolve_time_partition(spec: &PartitionSpec, at: DateTime<Utc>) -> Result<Option<String>> {
+fn resolve_time_partition(spec: &PartitionSpec, at: DateTime<Utc>) -> Result<Option<String>> {
     if matches!(spec.kind, PartitionKind::Dynamic { .. }) {
         return Err(Error::BadRequest(
             "resolve_time_partition called on a dynamic partition".to_string(),
@@ -74,7 +134,7 @@ pub fn resolve_time_partition(spec: &PartitionSpec, at: DateTime<Utc>) -> Result
 /// `$.a.b.c` dotted path (no brackets/wildcards/filters — that covers the
 /// realistic per-tenant/shard/event keys). The leaf must be a string,
 /// number or bool.
-pub fn extract_dynamic_partition(key: &str, payload: &serde_json::Value) -> Result<String> {
+fn extract_dynamic_partition(key: &str, payload: &serde_json::Value) -> Result<String> {
     let path = key.strip_prefix("$").unwrap_or(key);
     let path = path.strip_prefix('.').unwrap_or(path);
     let mut cur = payload;

backend/windmill-queue/src/asset_dispatch.rs

@@ -94,8 +94,8 @@ enum DispatchOutcome {
     Skipped,
 }
 
-/// Outcome-specific fields. `record_event` takes the four "always-present"
-/// columns positionally and bundles the rest here so each call site only
+/// Outcome-specific fields. Event constructors take the four "always-present"
+/// columns positionally and bundle the rest here so each call site only
 /// names what it actually carries.
 #[derive(Debug, Default)]
 struct EventOptions<'a> {
@@ -107,6 +107,46 @@ struct EventOptions<'a> {
     reason: Option<&'a str>,
 }
 
+/// One accumulated `dispatch_event` row. Owned (not borrowed) so the whole
+/// dispatch pass can collect rows and flush them in a single batched INSERT
+/// at the end, avoiding an N+1 (one INSERT per subscriber × asset write).
+#[derive(Debug)]
+struct EventRow {
+    subscriber_path: String,
+    asset_kind: AssetKind,
+    asset_path: String,
+    outcome: DispatchOutcome,
+    child_job_id: Option<Uuid>,
+    partition: Option<String>,
+    received_inputs: Option<i32>,
+    required_inputs: Option<i32>,
+    debounce_s: Option<i32>,
+    reason: Option<String>,
+}
+
+impl EventRow {
+    fn new(
+        subscriber_path: &str,
+        asset_kind: AssetKind,
+        asset_path: &str,
+        outcome: DispatchOutcome,
+        opts: EventOptions<'_>,
+    ) -> Self {
+        EventRow {
+            subscriber_path: subscriber_path.to_string(),
+            asset_kind,
+            asset_path: asset_path.to_string(),
+            outcome,
+            child_job_id: opts.child_job_id,
+            partition: opts.partition.map(str::to_string),
+            received_inputs: opts.received_inputs,
+            required_inputs: opts.required_inputs,
+            debounce_s: opts.debounce_s,
+            reason: opts.reason.map(str::to_string),
+        }
+    }
+}
+
 /// AND-join slot progress at the moment a partition-bearing input
 /// arrived. `fired` = all required inputs are now present (the slot has
 /// been cleared and the subscriber will be dispatched). `received` /
@@ -120,19 +160,75 @@ struct JoinSlotStatus {
     required: i32,
 }
 
-/// Best-effort insert into `dispatch_event`. Never propagates — the
-/// dispatch contract is "logging failures must not retroactively fail
-/// the producer's job."
-async fn record_event(
+/// Outcome of evaluating an AND-join barrier for one (subscriber, input).
+/// The caller turns this into a `dispatch_event` row and decides whether to
+/// push, all in one place.
+enum JoinDecision {
+    /// Input does not advance the join (recorded as Skipped with `reason`).
+    Skip(&'static str),
+    /// Join advanced but is not yet complete (recorded as JoinPending).
+    Pending { received: i32, required: i32 },
+    /// All required inputs are present — push the subscriber.
+    Fire,
+}
+
+/// Evaluate the AND-join barrier for a partition-bearing subscriber input.
+/// Only a partition-bearing input carrying a concrete partition advances the
+/// join — a reference input or an unpartitioned producer must never fire a
+/// partitioned join (the case-3 silent-wrong guard). On `Err` the caller
+/// should log and skip without recording an event.
+async fn handle_join(
     db: &DB,
     workspace_id: &str,
-    producer_job_id: Uuid,
-    subscriber_path: &str,
-    asset_kind: AssetKind,
-    asset_path: &str,
-    outcome: DispatchOutcome,
-    opts: EventOptions<'_>,
-) {
+    sub_path: &str,
+    trigger_ref: &str,
+    partition: Option<&str>,
+) -> Result<JoinDecision> {
+    if !is_partition_bearing_ref(trigger_ref) {
+        tracing::debug!(
+            "AND subscriber {}: non-partition-bearing input {} does not fire the join",
+            sub_path,
+            trigger_ref
+        );
+        return Ok(JoinDecision::Skip("case3_non_partition_bearing"));
+    }
+    let Some(pv) = partition else {
+        tracing::warn!(
+            "AND subscriber {}: partition-bearing input {} arrived with no resolved \
+             partition; not dispatching (case-3 guard)",
+            sub_path,
+            trigger_ref
+        );
+        return Ok(JoinDecision::Skip("case3_missing_partition"));
+    };
+    match record_and_check_join_slot(db, workspace_id, sub_path, pv, trigger_ref).await? {
+        JoinSlotStatus { fired: false, received, required } => {
+            Ok(JoinDecision::Pending { received, required })
+        }
+        JoinSlotStatus { fired: true, .. } => Ok(JoinDecision::Fire),
+    }
+}
+
+/// Best-effort batched insert into `dispatch_event`. Never propagates — the
+/// dispatch contract is "logging failures must not retroactively fail the
+/// producer's job." All rows accumulated over a dispatch pass go in one
+/// INSERT (UNNEST) to avoid an N+1 across (subscriber × asset write).
+async fn flush_events(db: &DB, workspace_id: &str, producer_job_id: Uuid, events: &[EventRow]) {
+    if events.is_empty() {
+        return;
+    }
+    // Column-oriented arrays for UNNEST. Each Vec is one column across all rows.
+    let subscriber_paths: Vec<String> = events.iter().map(|e| e.subscriber_path.clone()).collect();
+    let asset_kinds: Vec<AssetKind> = events.iter().map(|e| e.asset_kind).collect();
+    let asset_paths: Vec<String> = events.iter().map(|e| e.asset_path.clone()).collect();
+    let outcomes: Vec<DispatchOutcome> = events.iter().map(|e| e.outcome).collect();
+    let child_job_ids: Vec<Option<Uuid>> = events.iter().map(|e| e.child_job_id).collect();
+    let partitions: Vec<Option<String>> = events.iter().map(|e| e.partition.clone()).collect();
+    let received_inputs: Vec<Option<i32>> = events.iter().map(|e| e.received_inputs).collect();
+    let required_inputs: Vec<Option<i32>> = events.iter().map(|e| e.required_inputs).collect();
+    let debounce_s: Vec<Option<i32>> = events.iter().map(|e| e.debounce_s).collect();
+    let reasons: Vec<Option<String>> = events.iter().map(|e| e.reason.clone()).collect();
+
     let res = sqlx::query!(
         r#"INSERT INTO dispatch_event (
              workspace_id, producer_job_id, subscriber_path,
@@ -140,26 +236,31 @@ async fn record_event(
              child_job_id, partition,
              received_inputs, required_inputs,
              debounce_s, reason
-           ) VALUES ($1, $2, $3, $4::ASSET_KIND, $5, $6::DISPATCH_OUTCOME,
-                     $7, $8, $9, $10, $11, $12)"#,
+           )
+           SELECT $1, $2, sp, ak, ap, oc, cj, pt, ri, rq, db, rs
+           FROM unnest(
+             $3::text[], $4::ASSET_KIND[], $5::text[], $6::DISPATCH_OUTCOME[],
+             $7::uuid[], $8::text[], $9::int[], $10::int[], $11::int[], $12::text[]
+           ) AS t(sp, ak, ap, oc, cj, pt, ri, rq, db, rs)"#,
         workspace_id,
         producer_job_id,
-        subscriber_path,
-        asset_kind as AssetKind,
-        asset_path,
-        outcome as DispatchOutcome,
-        opts.child_job_id,
-        opts.partition,
-        opts.received_inputs,
-        opts.required_inputs,
-        opts.debounce_s,
-        opts.reason,
+        &subscriber_paths,
+        asset_kinds as Vec<AssetKind>,
+        &asset_paths,
+        outcomes as Vec<DispatchOutcome>,
+        &child_job_ids as &[Option<Uuid>],
+        &partitions as &[Option<String>],
+        &received_inputs as &[Option<i32>],
+        &required_inputs as &[Option<i32>],
+        &debounce_s as &[Option<i32>],
+        &reasons as &[Option<String>],
     )
     .execute(db)
     .await;
     if let Err(e) = res {
         tracing::error!(
-            "failed to record dispatch_event for producer {}: {e:#}",
+            "failed to record {} dispatch_event row(s) for producer {}: {e:#}",
+            events.len(),
             producer_job_id
         );
     }
@@ -220,8 +321,13 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
     }
 
     let mut dispatched = Vec::new();
+    // Best-effort dispatch_event rows accumulated over the whole pass and
+    // flushed in one batched INSERT at the end (avoids an N+1 over
+    // subscriber × asset write). The mid-pass join-slot writes
+    // (record_and_check_join_slot) are a separate table and unaffected.
+    let mut events: Vec<EventRow> = Vec::new();
     for (asset_kind, asset_path) in writes {
-        let Some(prefix) = prefix_for(asset_kind) else {
+        let Some(prefix) = asset_kind.canonical_prefix() else {
             continue;
         };
         let trigger_ref = format!("{}{}", prefix, asset_path);
@@ -230,93 +336,51 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
             let Subscriber { path: sub_path, join_all, debounce_s, retry_count, retry_delay_s } =
                 sub;
             if sub_path == runnable_path {
-                record_event(
-                    db,
-                    &job.workspace_id,
-                    job.id,
+                events.push(EventRow::new(
                     &sub_path,
                     asset_kind,
                     &asset_path,
                     DispatchOutcome::Skipped,
                     EventOptions { reason: Some("self_loop"), ..Default::default() },
-                )
-                .await;
+                ));
                 continue;
             }
             if join_all {
-                // AND join barrier. Only a partition-bearing input
-                // (`// on …/{partition}/…`) carrying a concrete partition
-                // advances the join — a reference input or an
-                // unpartitioned producer must never fire a partitioned
-                // join (the case-3 silent-wrong guard).
-                if !is_partition_bearing_ref(&trigger_ref) {
-                    tracing::debug!(
-                        "AND subscriber {}: non-partition-bearing input {} does not fire the join",
-                        sub_path,
-                        trigger_ref
-                    );
-                    record_event(
-                        db,
-                        &job.workspace_id,
-                        job.id,
-                        &sub_path,
-                        asset_kind,
-                        &asset_path,
-                        DispatchOutcome::Skipped,
-                        EventOptions {
-                            reason: Some("case3_non_partition_bearing"),
-                            ..Default::default()
-                        },
-                    )
-                    .await;
-                    continue;
-                }
-                let Some(pv) = partition.as_deref() else {
-                    tracing::warn!(
-                        "AND subscriber {}: partition-bearing input {} arrived with no resolved \
-                         partition; not dispatching (case-3 guard)",
-                        sub_path,
-                        trigger_ref
-                    );
-                    record_event(
-                        db,
-                        &job.workspace_id,
-                        job.id,
-                        &sub_path,
-                        asset_kind,
-                        &asset_path,
-                        DispatchOutcome::Skipped,
-                        EventOptions {
-                            reason: Some("case3_missing_partition"),
-                            ..Default::default()
-                        },
-                    )
-                    .await;
-                    continue;
-                };
-                match record_and_check_join_slot(db, &job.workspace_id, &sub_path, pv, &trigger_ref)
-                    .await
+                match handle_join(
+                    db,
+                    &job.workspace_id,
+                    &sub_path,
+                    &trigger_ref,
+                    partition.as_deref(),
+                )
+                .await
                 {
-                    Ok(JoinSlotStatus { fired: false, received, required }) => {
-                        record_event(
-                            db,
-                            &job.workspace_id,
-                            job.id,
+                    Ok(JoinDecision::Skip(reason)) => {
+                        events.push(EventRow::new(
+                            &sub_path,
+                            asset_kind,
+                            &asset_path,
+                            DispatchOutcome::Skipped,
+                            EventOptions { reason: Some(reason), ..Default::default() },
+                        ));
+                        continue;
+                    }
+                    Ok(JoinDecision::Pending { received, required }) => {
+                        events.push(EventRow::new(
                             &sub_path,
                             asset_kind,
                             &asset_path,
                             DispatchOutcome::JoinPending,
                             EventOptions {
-                                partition: Some(pv),
+                                partition: partition.as_deref(),
                                 received_inputs: Some(received),
                                 required_inputs: Some(required),
                                 ..Default::default()
                             },
-                        )
-                        .await;
+                        ));
                         continue; // slot incomplete — wait for the rest
                     }
-                    Ok(JoinSlotStatus { fired: true, .. }) => {} // fall through to push
+                    Ok(JoinDecision::Fire) => {} // fall through to push
                     Err(e) => {
                         tracing::error!("join-slot check failed for {}: {e:#}", sub_path);
                         continue;
@@ -339,10 +403,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
             .await
             {
                 Ok(id) => {
-                    record_event(
-                        db,
-                        &job.workspace_id,
-                        job.id,
+                    events.push(EventRow::new(
                         &sub_path,
                         asset_kind,
                         &asset_path,
@@ -353,8 +414,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
                             debounce_s,
                             ..Default::default()
                         },
-                    )
-                    .await;
+                    ));
                     dispatched.push(id);
                 }
                 Err(e) => {
@@ -364,6 +424,8 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
         }
     }
 
+    flush_events(db, &job.workspace_id, job.id, &events).await;
+
     if !dispatched.is_empty() {
         tracing::info!(
             "asset-trigger dispatch from job {} ({}): pushed {} downstream jobs",
@@ -436,19 +498,6 @@ fn read_partition(
     serde_json::from_str::<String>(trigger_map?.get(PARTITION_ARG)?.get()).ok()
 }
 
-fn prefix_for(kind: AssetKind) -> Option<&'static str> {
-    match kind {
-        AssetKind::S3Object => Some("s3://"),
-        AssetKind::Resource => Some("$res:"),
-        AssetKind::Ducklake => Some("ducklake://"),
-        AssetKind::DataTable => Some("datatable://"),
-        AssetKind::Volume => Some("volume://"),
-        // Deprecated kind from before the parser was unified — has no
-        // canonical trigger ref and never produced trigger rows.
-        AssetKind::Variable => None,
-    }
-}
-
 async fn fetch_producer_writes(
     db: &Pool<Postgres>,
     workspace_id: &str,
@@ -763,7 +812,7 @@ async fn push_subscriber(
     let mut args: HashMap<String, Box<RawValue>> = HashMap::new();
     let trigger_payload = serde_json::json!({
         "kind": "asset",
-        "asset_kind": serde_json::to_value(&asset_kind).unwrap_or(serde_json::Value::Null),
+        "asset_kind": serde_json::to_value(&asset_kind).expect("AssetKind serializes"),
         "asset_path": asset_path,
         "producer_path": producer_path,
         "producer_job_id": producer.id.to_string(),

backend/windmill-types/src/assets.rs

@@ -16,6 +16,23 @@ pub enum AssetKind {
     Volume,
 }
 
+impl AssetKind {
+    /// The canonical URI prefix used in asset trigger refs (e.g. `s3://`,
+    /// `$res:`). Single source of truth for both trigger-ref construction
+    /// and runtime cascade dispatch. `Variable` is deprecated and has no
+    /// canonical ref, so it returns `None`.
+    pub fn canonical_prefix(&self) -> Option<&'static str> {
+        match self {
+            AssetKind::S3Object => Some("s3://"),
+            AssetKind::Resource => Some("$res:"),
+            AssetKind::Ducklake => Some("ducklake://"),
+            AssetKind::DataTable => Some("datatable://"),
+            AssetKind::Volume => Some("volume://"),
+            AssetKind::Variable => None,
+        }
+    }
+}
+
 #[derive(
     Serialize, Deserialize, Debug, PartialEq, Copy, Clone, Hash, Eq, sqlx::Type, PartialOrd, Ord,
 )]

backend/windmill-worker/src/result_processor.rs

@@ -784,19 +784,11 @@ pub async fn process_completed_job(
             // by resolve_partition_for_job). Run identity is immutable —
             // the preprocessor must not change or drop it, or the asset
             // cascade would read no partition for this producer.
-            sqlx::query!(
-                "UPDATE v2_job
-                 SET args = CASE
-                              WHEN args ? 'partition'
-                              THEN $1 || jsonb_build_object('partition', args -> 'partition')
-                              ELSE $1
-                            END,
-                     preprocessed = TRUE
-                 WHERE id = $2",
-                Json(preprocessed_args) as Json<HashMap<String, Box<RawValue>>>,
-                job.id
+            windmill_common::partition::merge_args_preserving_partition(
+                db,
+                job.id,
+                preprocessed_args,
             )
-            .execute(db)
             .await?;
         }
 

backend/windmill-worker/src/worker.rs

@@ -4385,14 +4385,7 @@ async fn resolve_partition_for_job(
     // Persist back so dispatch_asset_triggers (which reads the producer's
     // completed v2_job.args) propagates the same value down the cascade.
     if let Some(db) = conn.as_sql() {
-        sqlx::query!(
-            "UPDATE v2_job SET args = coalesce(args, '{}'::jsonb) \
-             || jsonb_build_object('partition', $1::text) WHERE id = $2",
-            value,
-            job.id,
-        )
-        .execute(db)
-        .await?;
+        windmill_common::partition::set_resolved_partition(db, job.id, &value).await?;
     } else {
         tracing::warn!(
             job_id = %job.id,

cli/src/commands/pipeline/pipeline.ts

@@ -10,8 +10,13 @@ import * as log from "../../core/log.ts";
 import { GlobalOptions } from "../../types.ts";
 
 // Mirrors the asset-graph endpoint payload (backend/windmill-api-assets).
-// Raw-fetched because these routes are newer than the checked-in generated
-// client; swap to the generated functions on the next client regen.
+// 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";
@@ -92,6 +97,12 @@ 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,
@@ -121,10 +132,7 @@ async function show(
   for (const e of graph.edges) {
     if (e.access_type === "w" || e.access_type === "rw") {
       const uri = assetUri(e.asset_kind, e.asset_path);
-      writesByScript.set(e.runnable_path, [
-        ...(writesByScript.get(e.runnable_path) ?? []),
-        uri,
-      ]);
+      pushTo(writesByScript, e.runnable_path, uri);
     }
   }
   const subsByAsset = new Map<string, string[]>();
@@ -137,17 +145,15 @@ async function show(
     if (t.trigger_kind === "asset") {
       const at = t as Extract<GraphTrigger, { trigger_kind: "asset" }>;
       const uri = assetUri(at.asset_kind, at.asset_path);
-      subsByAsset.set(uri, [...(subsByAsset.get(uri) ?? []), t.runnable_path]);
-      subsByScript.set(t.runnable_path, [
-        ...(subsByScript.get(t.runnable_path) ?? []),
-        uri,
-      ]);
+      pushTo(subsByAsset, uri, t.runnable_path);
+      pushTo(subsByScript, t.runnable_path, uri);
     } else {
       const nt = t as Exclude<GraphTrigger, { trigger_kind: "asset" }>;
-      nativeByScript.set(t.runnable_path, [
-        ...(nativeByScript.get(t.runnable_path) ?? []),
-        { kind: nt.trigger_kind, path: nt.path, missing: nt.missing },
-      ]);
+      pushTo(nativeByScript, t.runnable_path, {
+        kind: nt.trigger_kind,
+        path: nt.path,
+        missing: nt.missing,
+      });
     }
   }
 
@@ -205,9 +211,16 @@ async function show(
     .sort();
 
   // UI-first markers (data_upload, webhook) have no trigger row — the
-  // graph endpoint can't surface them, they live as `// on <kind>`
-  // annotations in the body. Roots are where sources matter, so fetch just
-  // those bodies and lift the marker kinds the canvas would show.
+  // 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) => {

frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte

@@ -530,7 +530,6 @@
 			: {
 					inPipeline: false,
 					triggerAssets: [],
-					schedules: [],
 					nativeTriggers: []
 				}
 	)

frontend/src/lib/components/assets/AssetGraph/AssetRunsPanel.svelte

@@ -404,6 +404,7 @@
 					<div class="text-3xs uppercase tracking-wide text-tertiary">Logs</div>
 					<LogViewer
 						jobId={selectedJob.id}
+						tag={selectedJob.tag}
 						duration={durationMs(selectedJob)}
 						mem={(selectedJob as any).mem_peak}
 						isLoading={jobLoaderLoading && !(selectedJob as any).logs}

frontend/src/lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte

@@ -0,0 +1,202 @@
+<script lang="ts">
+	import { workspaceStore } from '$lib/stores'
+	import { sendUserToast } from '$lib/utils'
+	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
+	import type { NativeTriggerKind } from './types'
+	import {
+		EmailTriggerService,
+		GcpTriggerService,
+		KafkaTriggerService,
+		MqttTriggerService,
+		NatsTriggerService,
+		PostgresTriggerService,
+		ScheduleService,
+		SqsTriggerService
+	} from '$lib/gen'
+	import KafkaTriggerEditor from '$lib/components/triggers/kafka/KafkaTriggerEditor.svelte'
+	import MqttTriggerEditor from '$lib/components/triggers/mqtt/MqttTriggerEditor.svelte'
+	import NatsTriggerEditor from '$lib/components/triggers/nats/NatsTriggerEditor.svelte'
+	import PostgresTriggerEditor from '$lib/components/triggers/postgres/PostgresTriggerEditor.svelte'
+	import SqsTriggerEditor from '$lib/components/triggers/sqs/SqsTriggerEditor.svelte'
+	import GcpTriggerEditor from '$lib/components/triggers/gcp/GcpTriggerEditor.svelte'
+	import EmailTriggerEditor from '$lib/components/triggers/email/EmailTriggerEditor.svelte'
+	import ScheduleEditor from '$lib/components/triggers/schedules/ScheduleEditor.svelte'
+	import WebhookEditor from '$lib/components/triggers/webhook/WebhookEditor.svelte'
+
+	// Owns the native-trigger drawer wiring for the pipeline canvas: the nine
+	// editor instances, the create/edit dispatch by kind, and the delete
+	// confirmation flow. The page drives it imperatively via `bind:this` so
+	// the draft-guard toasts (which depend on the page's drafts map) stay in
+	// the page while the editor plumbing lives here.
+	//
+	// `mountTriggerEditors` gates the eight editor instances: they're
+	// edit-mode-only on the canvas, so the page unmounts them off edit mode
+	// (matching the previous `{#if mode === 'edit'}` wrapper). The webhook
+	// editor stays mounted in every mode — its node is clickable in view mode
+	// too (informational endpoint URLs/token).
+	type Props = { onUpdate: () => void; mountTriggerEditors: boolean }
+	let { onUpdate, mountTriggerEditors }: Props = $props()
+
+	let kafkaEditor: KafkaTriggerEditor | undefined = $state()
+	let mqttEditor: MqttTriggerEditor | undefined = $state()
+	let natsEditor: NatsTriggerEditor | undefined = $state()
+	let postgresEditor: PostgresTriggerEditor | undefined = $state()
+	let sqsEditor: SqsTriggerEditor | undefined = $state()
+	let gcpEditor: GcpTriggerEditor | undefined = $state()
+	let emailEditor: EmailTriggerEditor | undefined = $state()
+	let scheduleEditor: ScheduleEditor | undefined = $state()
+	let webhookEditor: WebhookEditor | undefined = $state()
+
+	// Open the create-trigger drawer for `kind`, pre-filling `script_path`.
+	// Caller is responsible for the draft guard (a trigger row can only point
+	// at a deployed script).
+	export function openNew(kind: NativeTriggerKind, scriptPath: string) {
+		switch (kind) {
+			case 'schedule':
+				return scheduleEditor?.openNew(false, scriptPath, undefined, scriptPath)
+			case 'kafka':
+				return kafkaEditor?.openNew(false, scriptPath)
+			case 'mqtt':
+				return mqttEditor?.openNew(false, scriptPath)
+			case 'nats':
+				return natsEditor?.openNew(false, scriptPath)
+			case 'postgres':
+				return postgresEditor?.openNew(false, scriptPath)
+			case 'sqs':
+				return sqsEditor?.openNew(false, scriptPath)
+			case 'gcp':
+				return gcpEditor?.openNew(false, scriptPath)
+			case 'email':
+				return emailEditor?.openNew(false, scriptPath)
+			// webhook has no dedicated editor.
+			default:
+				return
+		}
+	}
+
+	export function openEdit(kind: NativeTriggerKind, triggerPath: string, scriptPath: string) {
+		switch (kind) {
+			case 'schedule':
+				return scheduleEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'kafka':
+				return kafkaEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'mqtt':
+				return mqttEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'nats':
+				return natsEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'postgres':
+				return postgresEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'sqs':
+				return sqsEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'gcp':
+				return gcpEditor?.openEdit(triggerPath, false, scriptPath)
+			case 'email':
+				return emailEditor?.openEdit(triggerPath, false, scriptPath)
+			default:
+				return
+		}
+	}
+
+	// Open the webhook drawer (endpoint URLs + token flow). Caller owns the
+	// draft guard.
+	export function openWebhook(scriptPath: string) {
+		webhookEditor?.openDrawer(scriptPath, false)
+	}
+
+	// Delete-trigger confirmation state. Kebab → Delete opens the standard
+	// ConfirmationModal; the actual delete is dispatched from onConfirmed so
+	// the dialog stays consistent with the rest of the app.
+	let triggerDeleteTarget = $state<{ kind: NativeTriggerKind; path: string } | undefined>(undefined)
+	let triggerDeleteLoading = $state(false)
+	let triggerDeleteOpen = $derived(triggerDeleteTarget != undefined)
+
+	export function requestDelete(kind: NativeTriggerKind, triggerPath: string) {
+		triggerDeleteTarget = { kind, path: triggerPath }
+	}
+
+	async function confirmDeleteAttachedTrigger() {
+		if (!triggerDeleteTarget || !$workspaceStore) return
+		const { kind, path: triggerPath } = triggerDeleteTarget
+		const workspace = $workspaceStore
+		triggerDeleteLoading = true
+		try {
+			switch (kind) {
+				case 'schedule':
+					await ScheduleService.deleteSchedule({ workspace, path: triggerPath })
+					break
+				case 'kafka':
+					await KafkaTriggerService.deleteKafkaTrigger({ workspace, path: triggerPath })
+					break
+				case 'mqtt':
+					await MqttTriggerService.deleteMqttTrigger({ workspace, path: triggerPath })
+					break
+				case 'nats':
+					await NatsTriggerService.deleteNatsTrigger({ workspace, path: triggerPath })
+					break
+				case 'postgres':
+					await PostgresTriggerService.deletePostgresTrigger({ workspace, path: triggerPath })
+					break
+				case 'sqs':
+					await SqsTriggerService.deleteSqsTrigger({ workspace, path: triggerPath })
+					break
+				case 'gcp':
+					await GcpTriggerService.deleteGcpTrigger({ workspace, path: triggerPath })
+					break
+				case 'email':
+					await EmailTriggerService.deleteEmailTrigger({ workspace, path: triggerPath })
+					break
+				default:
+					return
+			}
+			sendUserToast(`Deleted ${kind} trigger "${triggerPath}"`)
+			triggerDeleteTarget = undefined
+			onUpdate()
+		} catch (e: any) {
+			sendUserToast(
+				`Could not delete ${kind} trigger "${triggerPath}": ${e?.body ?? e?.message ?? String(e)}`,
+				true
+			)
+		} finally {
+			triggerDeleteLoading = false
+		}
+	}
+</script>
+
+{#if mountTriggerEditors}
+	<ConfirmationModal
+		open={triggerDeleteOpen}
+		loading={triggerDeleteLoading}
+		title={triggerDeleteTarget ? `Delete ${triggerDeleteTarget.kind} trigger` : 'Delete trigger'}
+		confirmationText="Delete"
+		onConfirmed={confirmDeleteAttachedTrigger}
+		onCanceled={() => {
+			if (!triggerDeleteLoading) triggerDeleteTarget = undefined
+		}}
+	>
+		{#if triggerDeleteTarget}
+			<p>
+				Delete <code class="font-mono">{triggerDeleteTarget.path}</code>? The
+				<code class="font-mono">// on {triggerDeleteTarget.kind}</code> annotation on the script stays
+				— the trigger will read as missing on the canvas until you recreate it or remove the annotation.
+			</p>
+		{/if}
+	</ConfirmationModal>
+
+	<!-- Native trigger editors mounted off-screen. Each only renders its
+	     inner drawer when `open=true` (set by openNew/openEdit), so this
+	     adds ~zero render cost while idle. `onUpdate` refreshes the graph
+	     so the new trigger row replaces the red missing placeholder.
+	     Edit-mode only: every entry point (create/edit/delete trigger) is
+	     gated off the canvas outside edit mode. -->
+	<KafkaTriggerEditor bind:this={kafkaEditor} {onUpdate} />
+	<MqttTriggerEditor bind:this={mqttEditor} {onUpdate} />
+	<NatsTriggerEditor bind:this={natsEditor} {onUpdate} />
+	<PostgresTriggerEditor bind:this={postgresEditor} {onUpdate} />
+	<SqsTriggerEditor bind:this={sqsEditor} {onUpdate} />
+	<GcpTriggerEditor bind:this={gcpEditor} {onUpdate} />
+	<EmailTriggerEditor bind:this={emailEditor} {onUpdate} />
+	<ScheduleEditor bind:this={scheduleEditor} {onUpdate} />
+{/if}
+<!-- Webhook drawer stays mounted in every mode — the webhook trigger node
+     is clickable in view mode too (informational: endpoint URLs/token). -->
+<WebhookEditor bind:this={webhookEditor} />

frontend/src/lib/components/assets/AssetGraph/deployGraphDiff.ts

@@ -1,4 +1,5 @@
 import type { AssetGraphResponse } from './types'
+import { assetKey, isWriteEdge } from './lib'
 
 // Post-deploy graph verification: the canvas previews a draft's edges from
 // the *frontend* parsers (resolveGraph overlay); the deployed rows are
@@ -24,16 +25,14 @@ export type CascadeFacts = {
 export function extractCascadeFacts(g: AssetGraphResponse, scriptPath: string): CascadeFacts {
 	const writes = new Set<string>()
 	for (const e of g.edges ?? []) {
-		if (e.runnable_kind !== 'script' || e.runnable_path !== scriptPath) continue
-		const access = e.access_type ?? 'r'
-		if (access !== 'w' && access !== 'rw') continue
-		writes.add(`${e.asset_kind}:${e.asset_path}`)
+		if (e.runnable_path !== scriptPath || !isWriteEdge(e)) continue
+		writes.add(assetKey(e))
 	}
 	const subs = new Set<string>()
 	for (const t of g.triggers ?? []) {
 		if (t.trigger_kind !== 'asset') continue
 		if (t.runnable_kind !== 'script' || t.runnable_path !== scriptPath) continue
-		subs.add(`${t.asset_kind}:${t.asset_path}`)
+		subs.add(assetKey(t))
 	}
 	return { writes, subs }
 }

frontend/src/lib/components/assets/AssetGraph/graphTraversal.test.ts

@@ -1,6 +1,11 @@
 import { describe, expect, it } from 'vitest'
 import type { AssetGraphResponse } from './types'
-import { computeDownstreamClosure, getDownstreamSubscribers } from './graphTraversal'
+import { buildDownstreamMap, computeDownstreamClosure } from './graphTraversal'
+
+/** Direct (one-hop) subscriber script paths of `scriptPath`. */
+function downstreamSubscribers(g: AssetGraphResponse, scriptPath: string): string[] {
+	return Array.from(buildDownstreamMap(g).get(scriptPath) ?? [])
+}
 
 // Minimal graph builder: `writes` are producer→asset write edges, `subs` are
 // `// on <asset>` subscriptions. All datatable/script for brevity.
@@ -37,7 +42,7 @@ describe('buildDownstreamMap', () => {
 				['c', 'x']
 			]
 		)
-		expect(getDownstreamSubscribers(g, 'a').sort()).toEqual(['b', 'c'])
+		expect(downstreamSubscribers(g, 'a').sort()).toEqual(['b', 'c'])
 	})
 
 	it('ignores read edges, flow subscribers and self-loops', () => {
@@ -57,8 +62,8 @@ describe('buildDownstreamMap', () => {
 			runnable_kind: 'flow',
 			runnable_path: 'f'
 		})
-		expect(getDownstreamSubscribers(g, 'a')).toEqual([])
-		expect(getDownstreamSubscribers(g, 'r')).toEqual([])
+		expect(downstreamSubscribers(g, 'a')).toEqual([])
+		expect(downstreamSubscribers(g, 'r')).toEqual([])
 	})
 
 	it('matches on asset kind as well as path', () => {
@@ -70,7 +75,7 @@ describe('buildDownstreamMap', () => {
 			runnable_kind: 'script',
 			runnable_path: 'b'
 		})
-		expect(getDownstreamSubscribers(g, 'a')).toEqual([])
+		expect(downstreamSubscribers(g, 'a')).toEqual([])
 	})
 })
 

frontend/src/lib/components/assets/AssetGraph/graphTraversal.ts

@@ -1,4 +1,5 @@
 import type { AssetGraphResponse } from './types'
+import { assetKey, buildAssetSubscribers, isWriteEdge } from './lib'
 
 // Execution-DAG traversal over the resolved asset graph (drafts included).
 //
@@ -13,20 +14,11 @@ import type { AssetGraphResponse } from './types'
 
 /** `producer script path` → set of subscriber script paths (one hop). */
 export function buildDownstreamMap(g: AssetGraphResponse): Map<string, Set<string>> {
-	const subscribersByAsset = new Map<string, Set<string>>()
-	for (const t of g.triggers ?? []) {
-		if (t.trigger_kind !== 'asset' || t.runnable_kind !== 'script') continue
-		const key = `${t.asset_kind}:${t.asset_path}`
-		const set = subscribersByAsset.get(key) ?? new Set<string>()
-		set.add(t.runnable_path)
-		subscribersByAsset.set(key, set)
-	}
+	const subscribersByAsset = buildAssetSubscribers(g)
 	const downstream = new Map<string, Set<string>>()
 	for (const e of g.edges ?? []) {
-		if (e.runnable_kind !== 'script') continue
-		const access = e.access_type ?? 'r'
-		if (access !== 'w' && access !== 'rw') continue
-		const subs = subscribersByAsset.get(`${e.asset_kind}:${e.asset_path}`)
+		if (!isWriteEdge(e)) continue
+		const subs = subscribersByAsset.get(assetKey(e))
 		if (!subs) continue
 		const merged = downstream.get(e.runnable_path) ?? new Set<string>()
 		for (const s of subs) if (s !== e.runnable_path) merged.add(s)
@@ -35,11 +27,6 @@ export function buildDownstreamMap(g: AssetGraphResponse): Map<string, Set<strin
 	return downstream
 }
 
-/** Direct (one-hop) subscriber script paths of `scriptPath`. */
-export function getDownstreamSubscribers(g: AssetGraphResponse, scriptPath: string): string[] {
-	return Array.from(buildDownstreamMap(g).get(scriptPath) ?? [])
-}
-
 export type DownstreamClosure = {
 	/** Every script in the transitive downstream of root, root excluded. */
 	nodes: string[]

frontend/src/lib/components/assets/AssetGraph/lib.ts

@@ -0,0 +1,35 @@
+import type { AssetGraphEdge, AssetGraphResponse } from './types'
+
+// Shared graph predicates used by both the execution-DAG traversal
+// (graphTraversal.ts) and the post-deploy drift check (deployGraphDiff.ts).
+// A "write edge" is what makes a script a *producer* of an asset; the default
+// `?? 'r'` treats a null access_type as a read so a missing direction never
+// counts as a write.
+
+/** `kind:path` identity for an asset, the dedup key used across the graph. */
+export function assetKey(asset: { asset_kind: string; asset_path: string }): string {
+	return `${asset.asset_kind}:${asset.asset_path}`
+}
+
+/** True for a script write lineage edge (access 'w' or 'rw'). */
+export function isWriteEdge(edge: AssetGraphEdge): boolean {
+	if (edge.runnable_kind !== 'script') return false
+	const access = edge.access_type ?? 'r'
+	return access === 'w' || access === 'rw'
+}
+
+/**
+ * `asset key` → set of subscriber script paths (`// on <asset>` declarations).
+ * Flows are excluded, mirroring the backend dispatch policy.
+ */
+export function buildAssetSubscribers(g: AssetGraphResponse): Map<string, Set<string>> {
+	const subscribersByAsset = new Map<string, Set<string>>()
+	for (const t of g.triggers ?? []) {
+		if (t.trigger_kind !== 'asset' || t.runnable_kind !== 'script') continue
+		const key = assetKey(t)
+		const set = subscribersByAsset.get(key) ?? new Set<string>()
+		set.add(t.runnable_path)
+		subscribersByAsset.set(key, set)
+	}
+	return subscribersByAsset
+}

frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.parity.test.ts

@@ -2,7 +2,23 @@ import { readFileSync } from 'node:fs'
 import { resolve, dirname } from 'node:path'
 import { fileURLToPath } from 'node:url'
 import { describe, expect, it } from 'vitest'
-import { parsePipelineAnnotations } from './parsePipelineAnnotations'
+import { parsePipelineAnnotations, type PipelineAnnotations } from './parsePipelineAnnotations'
+
+// Every field the TS parser produces, each with an assertion in the per-fixture
+// test below. Typed as `Record<keyof PipelineAnnotations, true>` so adding a
+// field to `PipelineAnnotations` is a compile error here until it's listed —
+// forcing the author to also wire up its parity assertion. (Deploy-only Rust
+// fields like join_mode / debounce_default are not part of this type, so they
+// correctly never appear here.)
+const ASSERTED_TS_FIELDS: Record<keyof PipelineAnnotations, true> = {
+	inPipeline: true,
+	triggerAssets: true,
+	nativeTriggers: true,
+	partition: true,
+	freshness: true,
+	tag: true,
+	retry: true
+}
 
 // Parser-parity guard: this TS parser (drives the live graph preview) and
 // the Rust `parse_pipeline_annotations` (drives deploy) must stay
@@ -14,8 +30,16 @@ import { parsePipelineAnnotations } from './parsePipelineAnnotations'
 // Rust counterpart: backend/parsers/windmill-parser/tests/
 // pipeline_annotations_parity.rs. Extend the corpus when the grammar
 // changes; a fixture passing on one side and failing on the other is
-// exactly the drift this exists to catch. Only fields both parsers produce
-// are compared (join_mode / debounce_default are deploy-only, Rust-side).
+// exactly the drift this exists to catch.
+//
+// Intentional divergence — the Rust parser is a superset. It also parses
+// `join_mode` and `debounce_default`, which are DEPLOY-ONLY: they affect how
+// the backend schedules cascade runs, never the rendered graph, so the TS
+// preview parser deliberately doesn't produce them and they are not compared
+// here. Every field the TS parser DOES produce is compared, and the
+// `ASSERTED_TS_FIELDS` exhaustiveness check above fails the suite if a new TS
+// field is added without a matching assertion — so a field can't be parsed on
+// the TS side yet silently skipped by this guard.
 
 type Fixture = {
 	name: string
@@ -48,6 +72,18 @@ describe('parsePipelineAnnotations matches the shared Rust fixture corpus', () =
 		expect(fixtures.length).toBeGreaterThan(0)
 	})
 
+	it('every field the parser emits across the corpus has a parity assertion', () => {
+		const asserted = new Set(Object.keys(ASSERTED_TS_FIELDS))
+		const emitted = new Set<string>()
+		for (const f of fixtures) {
+			for (const k of Object.keys(parsePipelineAnnotations(f.code))) emitted.add(k)
+		}
+		expect(
+			[...emitted].filter((k) => !asserted.has(k)),
+			'unasserted parser fields'
+		).toEqual([])
+	})
+
 	for (const f of fixtures) {
 		it(f.name, () => {
 			const got = parsePipelineAnnotations(f.code)

frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts

@@ -46,6 +46,10 @@ export function usePipelineHistory(
 
 	async function load(ws: string, prefix: string, days: number) {
 		const myGen = ++gen
+		// A folder/days change (or unmount) bumps `gen`; once it diverges from
+		// this call's `myGen`, every write below this point belongs to a stale
+		// scope and must be dropped.
+		const isStale = () => gen !== myGen
 		loading = true
 		error = undefined
 		try {
@@ -64,7 +68,7 @@ export function usePipelineHistory(
 					perPage: PER_PAGE,
 					page: pages
 				})
-				if (gen !== myGen) return
+				if (isStale()) return
 				for (const j of rows) {
 					out.push({
 						id: j.id,
@@ -89,15 +93,15 @@ export function usePipelineHistory(
 			} catch (e) {
 				console.warn('failed to load pipeline dispatch edges', e)
 			}
-			if (gen !== myGen) return
+			if (isStale()) return
 			events = out
 			edges = edgeRows
 			truncated = sawFullPage
 		} catch (e: any) {
-			if (gen !== myGen) return
+			if (isStale()) return
 			error = e?.body ?? e?.message ?? String(e)
 		} finally {
-			if (gen === myGen) loading = false
+			if (!isStale()) loading = false
 		}
 	}
 

frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts

@@ -1,5 +1,6 @@
 import type { ScriptLang, AssetKind } from '$lib/gen'
 import { random_adj } from '$lib/components/random_positive_adjetive'
+import { parseDbInputFromAssetSyntax } from '$lib/utils'
 
 // What kind of asset the new script will produce. Drives the random output
 // path scheme and the body skeleton. The output asset is NOT declared in a
@@ -77,8 +78,10 @@ const LANG_COMPATIBILITY: Record<ScriptLang, PipelineOutputKind[]> = {
 	csharp: ['none'],
 	graphql: ['none'],
 	bunnative: ['none'],
-	nativets: ['none']
-} as any
+	nativets: ['none'],
+	ruby: ['none'],
+	rlang: ['none']
+}
 
 export function compatibleOutputKinds(lang: ScriptLang): PipelineOutputKind[] {
 	return LANG_COMPATIBILITY[lang] ?? ['none']
@@ -182,9 +185,12 @@ function s3Key(path: string): string {
 }
 
 // Splits a datatable asset path (`<db>/<table>` or `<db>/<schema>.<table>`)
-// into its constituent parts. Mirrors `parseDbInputFromAssetSyntax` in
-// $lib/utils.ts: schema is omitted when the table lives in the default
-// (`public`) schema, prefixed with `<schema>.` only when explicit.
+// into its constituent parts. The `<schema>.<table>` grammar is owned by
+// `parseDbInputFromAssetSyntax` in $lib/utils.ts (which consumes a full
+// `datatable://…` URI): we delegate to it for the schema/table split so the
+// two stay in lockstep, and only keep the leading `<db>/` extraction here —
+// the util folds the db into `resourcePath`, which is awkward to read back,
+// and it has no slashless fallback (the SQL emitters tolerate a bare db).
 function parseDatatablePath(p: string): {
 	db: string
 	schema: string | undefined
@@ -193,10 +199,11 @@ function parseDatatablePath(p: string): {
 	const slash = p.indexOf('/')
 	if (slash < 0) return { db: p, schema: undefined, table: '' }
 	const db = p.slice(0, slash)
-	const tail = p.slice(slash + 1)
-	const dot = tail.indexOf('.')
-	if (dot < 0) return { db, schema: undefined, table: tail }
-	return { db, schema: tail.slice(0, dot), table: tail.slice(dot + 1) }
+	// `datatable://` URIs always parse to the `database` variant, so the
+	// `specificSchema`/`specificTable` accessors below are always present.
+	const parsed = parseDbInputFromAssetSyntax(`datatable://${p}`)
+	const schema = parsed?.type === 'database' ? parsed.specificSchema : undefined
+	return { db, schema, table: parsed?.specificTable ?? '' }
 }
 
 // Schema-qualified table reference for SQL emission. We always emit the

frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts

@@ -35,6 +35,67 @@ export type ResolveGraphInput = {
 	annotatedNativeKindsByPath: Map<string, Set<NativeTriggerKind>>
 }
 
+/** Mutable bag the sequential passes accumulate the resolved graph into. */
+type Accumulator = {
+	runnables: AssetGraphResponse['runnables']
+	assets: AssetGraphResponse['assets']
+	edges: AssetGraphResponse['edges']
+	extraTriggers: AssetGraphResponse['triggers']
+}
+
+/**
+ * Native trigger kinds with a persisted (non-asset) trigger row pointing at
+ * `path`. These bind by script path and survive content edits, so they dedup
+ * against draft/live annotations to avoid showing a real source node next to a
+ * red "missing" placeholder for the same kind.
+ */
+function persistedNativeKinds(base: AssetGraphResponse, path: string): Set<string> {
+	return new Set(
+		base.triggers
+			.filter(
+				(t) =>
+					t.trigger_kind !== 'asset' && t.runnable_kind === 'script' && t.runnable_path === path
+			)
+			.map((t) => t.trigger_kind)
+	)
+}
+
+/** `kind:path` keys of persisted asset (`// on <asset>`) triggers for `path`. */
+function persistedAssetKeys(base: AssetGraphResponse, path: string): Set<string> {
+	return new Set(
+		base.triggers
+			.filter(
+				(t) =>
+					t.trigger_kind === 'asset' && t.runnable_kind === 'script' && t.runnable_path === path
+			)
+			.map((t) => (t.trigger_kind === 'asset' ? `${t.asset_kind}:${t.asset_path}` : ''))
+	)
+}
+
+/**
+ * Emit a red "missing" placeholder for each annotated native kind that has no
+ * matching attached trigger row. `opts.unsaved` marks editor-driven overlays
+ * (drafts / live buffer); deployed-but-unswept scripts pass `unsaved: false`.
+ */
+function pushMissingNativeTriggers(
+	extraTriggers: AssetGraphResponse['triggers'],
+	annotatedKinds: Iterable<NativeTriggerKind>,
+	attachedKinds: Set<string>,
+	path: string,
+	opts: { unsaved: boolean }
+) {
+	for (const kind of annotatedKinds) {
+		if (attachedKinds.has(kind)) continue
+		extraTriggers.push({
+			trigger_kind: kind,
+			runnable_kind: 'script',
+			runnable_path: path,
+			...(opts.unsaved ? { unsaved: true } : {}),
+			missing: true
+		})
+	}
+}
+
 /**
  * Merge the persisted base graph with the draft, session-inferred and
  * open-script live overlays into one `AssetGraphResponse`.
@@ -50,41 +111,61 @@ export type ResolveGraphInput = {
  * `graphWithDraft` `$derived`. See `resolveGraph.test.ts`.
  */
 export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
-	const {
-		base,
-		drafts,
-		liveBodyAssets,
-		liveAnnotations,
-		inferredWritesByPath,
-		inferredReadsByPath,
-		annotatedNativeKindsByPath
-	} = input
+	const { base } = input
+	const ctx = makeContext(input)
+
+	// Pass order mirrors the precedence comment, lowest → highest:
+	const acc = seedAccumulator(input, ctx)
+	seedDraftOverlays(acc, input)
+	applyLiveBufferOverlay(acc, input, ctx)
+	crossCheckSweptScripts(acc, input)
+	overlayInferredLineage(acc, input)
+
+	// Drop persisted ASSET triggers for drafted paths — those come from the
+	// deployed `// on <asset>` annotations, which the draft's live/parsed
+	// annotations now own. Native triggers (kafka/schedule/…) are kept: they
+	// bind by `script_path`, which the draft shares, so the attachment is still
+	// valid regardless of content edits.
+	const baseTriggers = base.triggers.filter((t) => {
+		if (t.trigger_kind !== 'asset') return true
+		if (ctx.isDrafted(t.runnable_kind, t.runnable_path)) return false
+		if (
+			t.runnable_kind === 'script' &&
+			t.runnable_path === ctx.openPath &&
+			ctx.staleForOpen(t.asset_kind, t.asset_path)
+		)
+			return false
+		return true
+	})
+	return {
+		...base,
+		assets: acc.assets,
+		runnables: acc.runnables,
+		edges: acc.edges,
+		triggers: [...baseTriggers, ...acc.extraTriggers]
+	}
+}
+
+type ResolveContext = {
+	draftedPaths: Set<string>
+	isDrafted: (kind: string, p: string) => boolean
+	openPath: string | undefined
+	staleForOpen: (kind: AssetKind, path: string) => boolean
+}
+
+/** Derive the shared "what's drafted / open / stale" predicates once. */
+function makeContext(input: ResolveGraphInput): ResolveContext {
+	const { drafts, liveBodyAssets, liveAnnotations } = input
 
-	// Every draft contributes: a runnable, an output asset, a write edge,
-	// plus its own seeded schedule trigger (template includes `// on
-	// schedule "0 * * * *"` by default, picked up through live parse).
-	// We iterate the whole `drafts` map so multiple concurrent drafts
-	// all render as their own subgraph at once.
-	const runnables = [...base.runnables]
-	const assets = [...base.assets]
 	// A drafted path's content-derived lineage is owned entirely by the draft
 	// overlay (live inference for the active draft, parsed annotations /
-	// snapshot for inactive ones). Drop the persisted base asset edges for those
+	// snapshot for inactive ones). Drop persisted base asset edges for those
 	// paths so it shows only the draft's I/O, not the union of the saved version
 	// and the in-flight edits. (Base asset triggers are dropped at the return;
 	// native triggers are kept — they bind by path.)
 	const draftedPaths = new Set(drafts.keys())
 	const isDrafted = (kind: string, p: string) => kind === 'script' && draftedPaths.has(p)
 
-	// Editing a *saved* script does NOT create a `drafts` entry — it stays an
-	// open selection with live overlays. So the draft filter above doesn't cover
-	// it: its persisted edges/triggers would linger next to the live edits (e.g.
-	// renaming an input leaves the old asset linked alongside the new one). For
-	// the open, live-inferred saved script, treat its current content as
-	// authoritative and drop persisted asset edges/triggers it no longer
-	// references. `liveRefKeys` unions the reliable `// on <asset>` annotations
-	// with the inferred body assets, so an unchanged selection drops nothing
-	// (every base asset is still referenced).
 	const openPath = liveBodyAssets.scriptPath
 	const openIsSavedEdit = openPath !== undefined && !draftedPaths.has(openPath)
 	const liveRefKeys = new Set<string>()
@@ -98,17 +179,39 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 	const staleForOpen = (kind: AssetKind, path: string) =>
 		openIsSavedEdit && !liveRefKeys.has(`${kind}:${path}`)
 
+	return { draftedPaths, isDrafted, openPath, staleForOpen }
+}
+
+/** Base graph plus the persisted edges that survive the draft/open filter. */
+function seedAccumulator(input: ResolveGraphInput, ctx: ResolveContext): Accumulator {
+	const { base } = input
 	const edges = base.edges.filter((e) => {
-		if (isDrafted(e.runnable_kind, e.runnable_path)) return false
+		if (ctx.isDrafted(e.runnable_kind, e.runnable_path)) return false
 		if (
 			e.runnable_kind === 'script' &&
-			e.runnable_path === openPath &&
-			staleForOpen(e.asset_kind, e.asset_path)
+			e.runnable_path === ctx.openPath &&
+			ctx.staleForOpen(e.asset_kind, e.asset_path)
 		)
 			return false
 		return true
 	})
-	const extraTriggers: AssetGraphResponse['triggers'] = []
+	return {
+		runnables: [...base.runnables],
+		assets: [...base.assets],
+		edges,
+		extraTriggers: []
+	}
+}
+
+/**
+ * Every draft contributes: a runnable, output asset(s), a write edge, live
+ * read lineage for the active draft, plus its seeded asset/native triggers
+ * (template includes `// on schedule …` by default). Iterates the whole
+ * `drafts` map so multiple concurrent drafts all render at once.
+ */
+function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
+	const { base, drafts, liveBodyAssets } = input
+	const { runnables, assets, edges, extraTriggers } = acc
 
 	for (const [path, d] of drafts) {
 		const parsed = parsePipelineAnnotations(d.script.content)
@@ -238,140 +341,110 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 		// they bind by path and are not dropped above. Dedup against those
 		// so the canvas doesn't show the real schedule node next to a red
 		// "missing" placeholder for the same kind.
-		const persistedNativeKinds = new Set(
-			base.triggers
-				.filter(
-					(t) =>
-						t.trigger_kind !== 'asset' &&
-						t.runnable_kind === 'script' &&
-						t.runnable_path === path
-				)
-				.map((t) => t.trigger_kind)
+		pushMissingNativeTriggers(
+			extraTriggers,
+			parsed.nativeTriggers.map((n) => n.kind),
+			persistedNativeKinds(base, path),
+			path,
+			{ unsaved: true }
 		)
-		for (const n of parsed.nativeTriggers) {
-			if (persistedNativeKinds.has(n.kind)) continue
-			extraTriggers.push({
-				trigger_kind: n.kind,
-				runnable_kind: 'script',
-				runnable_path: path,
-				unsaved: true,
-				missing: true
-			})
-		}
 	}
+}
 
-	// Live-parsed overlay for the currently-open script — takes precedence
-	// over the seeded-template triggers for the same path by swapping
-	// them out. Scoped to one path (only one pane is open at a time).
+/**
+ * Live-parsed overlay for the currently-open script — takes precedence over
+ * the seeded-template triggers for the same path by swapping them out. Scoped
+ * to one path (only one pane is open at a time).
+ */
+function applyLiveBufferOverlay(acc: Accumulator, input: ResolveGraphInput, ctx: ResolveContext) {
+	const { base, liveAnnotations } = input
+	const { assets, extraTriggers } = acc
 	const livePath = liveAnnotations.scriptPath
-	if (livePath) {
-		// When the open buffer is a draft, its base asset triggers were dropped
-		// above (draft-owned), so don't dedup live asset triggers against them —
-		// every live `// on <asset>` is emitted as unsaved. For a non-draft open
-		// script (viewing a saved one), keep deduping so live re-parse doesn't
-		// double the persisted triggers.
-		const persistedAssetKeys = draftedPaths.has(livePath)
-			? new Set<string>()
-			: new Set(
-					base.triggers
-						.filter(
-							(t) =>
-								t.trigger_kind === 'asset' &&
-								t.runnable_kind === 'script' &&
-								t.runnable_path === livePath
-						)
-						.map((t) => (t.trigger_kind === 'asset' ? `${t.asset_kind}:${t.asset_path}` : ''))
-				)
-		// Strip seeded triggers we computed above for the active draft;
-		// live annotations are authoritative for the open buffer.
-		for (let i = extraTriggers.length - 1; i >= 0; i--) {
-			if (extraTriggers[i].runnable_path === livePath) extraTriggers.splice(i, 1)
-		}
-		for (const a of liveAnnotations.annotations.triggerAssets) {
-			const key = `${a.kind}:${a.path}`
-			if (persistedAssetKeys.has(key)) continue
-			extraTriggers.push({
-				trigger_kind: 'asset',
-				asset_kind: a.kind,
-				asset_path: a.path,
-				runnable_kind: 'script',
-				runnable_path: livePath,
-				unsaved: true
-			})
-			// Synthesize the asset node so the new trigger edge has a
-			// target — without this, typing `// on s3:///...` adds an
-			// edge to a node that doesn't exist and the canvas silently
-			// drops it. Mirrors the draft branch above.
-			if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
-				assets.push({ kind: a.kind, path: a.path })
-			}
-		}
-		// Native trigger annotations: kinds for which a matching trigger
-		// row was found in the backend response. If the live buffer
-		// declares `// on kafka` and at least one kafka_trigger row points
-		// at this script, the source node is already on the canvas — no
-		// overlay needed. Otherwise emit a "missing" placeholder so the
-		// user can either create the trigger row or remove the annotation.
-		const persistedNativeKinds = new Set(
-			base.triggers
-				.filter(
-					(t) =>
-						t.trigger_kind !== 'asset' &&
-						t.runnable_kind === 'script' &&
-						t.runnable_path === livePath
-				)
-				.map((t) => t.trigger_kind)
-		)
-		for (const n of liveAnnotations.annotations.nativeTriggers) {
-			if (persistedNativeKinds.has(n.kind)) continue
-			extraTriggers.push({
-				trigger_kind: n.kind,
-				runnable_kind: 'script',
-				runnable_path: livePath,
-				unsaved: true,
-				missing: true
-			})
+	if (!livePath) return
+
+	// When the open buffer is a draft, its base asset triggers were dropped
+	// above (draft-owned), so don't dedup live asset triggers against them —
+	// every live `// on <asset>` is emitted as unsaved. For a non-draft open
+	// script (viewing a saved one), keep deduping so live re-parse doesn't
+	// double the persisted triggers.
+	const assetKeys = ctx.draftedPaths.has(livePath)
+		? new Set<string>()
+		: persistedAssetKeys(base, livePath)
+	// Strip seeded triggers we computed above for the active draft;
+	// live annotations are authoritative for the open buffer.
+	for (let i = extraTriggers.length - 1; i >= 0; i--) {
+		if (extraTriggers[i].runnable_path === livePath) extraTriggers.splice(i, 1)
+	}
+	for (const a of liveAnnotations.annotations.triggerAssets) {
+		const key = `${a.kind}:${a.path}`
+		if (assetKeys.has(key)) continue
+		extraTriggers.push({
+			trigger_kind: 'asset',
+			asset_kind: a.kind,
+			asset_path: a.path,
+			runnable_kind: 'script',
+			runnable_path: livePath,
+			unsaved: true
+		})
+		// Synthesize the asset node so the new trigger edge has a
+		// target — without this, typing `// on s3:///...` adds an
+		// edge to a node that doesn't exist and the canvas silently
+		// drops it. Mirrors the draft branch above.
+		if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
+			assets.push({ kind: a.kind, path: a.path })
 		}
 	}
+	// Native trigger annotations: kinds for which a matching trigger
+	// row was found in the backend response. If the live buffer
+	// declares `// on kafka` and at least one kafka_trigger row points
+	// at this script, the source node is already on the canvas — no
+	// overlay needed. Otherwise emit a "missing" placeholder so the
+	// user can either create the trigger row or remove the annotation.
+	pushMissingNativeTriggers(
+		extraTriggers,
+		liveAnnotations.annotations.nativeTriggers.map((n) => n.kind),
+		persistedNativeKinds(base, livePath),
+		livePath,
+		{ unsaved: true }
+	)
+}
 
-	// Cross-check for already-deployed scripts (not the open buffer): if a
-	// script's persisted body declares `// on kafka` but no matching
-	// kafka_trigger row points at it, surface a red placeholder. The
-	// annotated-kinds map is filled by the page-level prefetch sweep
-	// (one read per script in the folder); drafts and the active editor
-	// are handled by the loops above. Scripts that haven't been swept
-	// yet contribute nothing here — they'll surface on the next refetch.
-	const livePathExcl = livePath
+/**
+ * Cross-check for already-deployed scripts (not the open buffer): if a
+ * script's persisted body declares `// on kafka` but no matching kafka_trigger
+ * row points at it, surface a red placeholder. The annotated-kinds map is
+ * filled by the page-level prefetch sweep (one read per script in the folder);
+ * drafts and the active editor are handled by the earlier passes. Scripts not
+ * yet swept contribute nothing — they surface on the next refetch.
+ */
+function crossCheckSweptScripts(acc: Accumulator, input: ResolveGraphInput) {
+	const { base, drafts, liveAnnotations, annotatedNativeKindsByPath } = input
+	const livePath = liveAnnotations.scriptPath
 	for (const [scriptPath, kinds] of annotatedNativeKindsByPath) {
 		if (drafts.has(scriptPath)) continue
-		if (scriptPath === livePathExcl) continue
-		const attachedKinds = new Set(
-			base.triggers
-				.filter(
-					(t) =>
-						t.trigger_kind !== 'asset' &&
-						t.runnable_kind === 'script' &&
-						t.runnable_path === scriptPath
-				)
-				.map((t) => t.trigger_kind)
+		if (scriptPath === livePath) continue
+		pushMissingNativeTriggers(
+			acc.extraTriggers,
+			kinds,
+			persistedNativeKinds(base, scriptPath),
+			scriptPath,
+			{ unsaved: false }
 		)
-		for (const kind of kinds) {
-			if (attachedKinds.has(kind)) continue
-			extraTriggers.push({
-				trigger_kind: kind,
-				runnable_kind: 'script',
-				runnable_path: scriptPath,
-				missing: true
-			})
-		}
 	}
+}
+
+/**
+ * Live body-asset lineage for any persisted script inferred at least once this
+ * session (maps filled by `handleAssetsChange` + the load prefetch). Drafts
+ * are handled by `seedDraftOverlays`. For scripts whose deploy didn't persist
+ * their body assets (e.g. older WASM at save time, or object-form
+ * writeS3File), this keeps the lineage edge on the canvas across selection
+ * changes — not just while selected.
+ */
+function overlayInferredLineage(acc: Accumulator, input: ResolveGraphInput) {
+	const { base, drafts, inferredWritesByPath, inferredReadsByPath } = input
+	const { assets, edges } = acc
 
-	// Live body-asset lineage for any persisted script inferred at least
-	// once this session (maps filled by `handleAssetsChange` + the load
-	// prefetch). Drafts are handled by the loop above. For scripts whose
-	// deploy didn't persist their body assets (e.g. older WASM at save
-	// time, or object-form writeS3File), this keeps the lineage edge on
-	// the canvas across selection changes — not just while selected.
 	const overlayLineage = (
 		byPath: Map<string, Array<{ kind: AssetKind; path: string }>>,
 		access: 'w' | 'r'
@@ -407,22 +480,4 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 	}
 	overlayLineage(inferredWritesByPath, 'w')
 	overlayLineage(inferredReadsByPath, 'r')
-
-	// Drop persisted ASSET triggers for drafted paths — those come from the
-	// deployed `// on <asset>` annotations, which the draft's live/parsed
-	// annotations now own. Native triggers (kafka/schedule/…) are kept: they
-	// bind by `script_path`, which the draft shares, so the attachment is still
-	// valid regardless of content edits.
-	const baseTriggers = base.triggers.filter((t) => {
-		if (t.trigger_kind !== 'asset') return true
-		if (isDrafted(t.runnable_kind, t.runnable_path)) return false
-		if (
-			t.runnable_kind === 'script' &&
-			t.runnable_path === openPath &&
-			staleForOpen(t.asset_kind, t.asset_path)
-		)
-			return false
-		return true
-	})
-	return { ...base, assets, runnables, edges, triggers: [...baseTriggers, ...extraTriggers] }
 }

frontend/src/lib/components/runs/DispatchEventsButton.svelte

@@ -1,8 +1,7 @@
 <script lang="ts">
-	import { resource } from 'runed'
-	import { JobService } from '$lib/gen'
 	import Popover from '$lib/components/meltComponents/Popover.svelte'
 	import DispatchEventsTable from './DispatchEventsTable.svelte'
+	import { useDispatchEvents } from './useDispatchEvents.svelte'
 	import { GitFork } from 'lucide-svelte'
 	import { twMerge } from 'tailwind-merge'
 
@@ -12,12 +11,11 @@
 	type Props = { workspace: string; jobId: string; class?: string }
 	let { workspace, jobId, class: klass = '' }: Props = $props()
 
-	const events = resource(
-		() => ({ workspace, jobId }),
-		async ({ workspace, jobId }) =>
-			workspace && jobId ? await JobService.listDispatchEvents({ workspace, id: jobId }) : []
+	const events = useDispatchEvents(
+		() => workspace,
+		() => jobId
 	)
-	const list = $derived(events.current ?? [])
+	const list = $derived(events.list)
 </script>
 
 {#if list.length > 0}

frontend/src/lib/components/runs/DispatchEventsPanel.svelte

@@ -1,18 +1,16 @@
 <script lang="ts">
-	import { resource } from 'runed'
-	import { JobService } from '$lib/gen'
 	import DispatchEventsTable from './DispatchEventsTable.svelte'
+	import { useDispatchEvents } from './useDispatchEvents.svelte'
 
 	type Props = { workspace: string; jobId: string }
 	let { workspace, jobId }: Props = $props()
 
-	const events = resource(
-		() => ({ workspace, jobId }),
-		async ({ workspace, jobId }) =>
-			workspace && jobId ? await JobService.listDispatchEvents({ workspace, id: jobId }) : []
+	const events = useDispatchEvents(
+		() => workspace,
+		() => jobId
 	)
 
-	const list = $derived(events.current ?? [])
+	const list = $derived(events.list)
 </script>
 
 {#if list.length > 0}

frontend/src/lib/components/runs/DispatchEventsTable.svelte

@@ -7,6 +7,9 @@
 	type Props = { events: Event[]; workspace: string }
 	let { events, workspace }: Props = $props()
 
+	// These reason strings are produced verbatim by the backend asset-dispatch
+	// reason enum (backend asset_dispatch.rs) — keep them in sync if that enum
+	// changes; an unmatched reason falls through to the raw string.
 	function reasonLabel(reason: string | undefined): string {
 		switch (reason) {
 			case 'self_loop':

frontend/src/lib/components/runs/useDispatchEvents.svelte.ts

@@ -0,0 +1,24 @@
+import { resource } from 'runed'
+import { JobService, type ListDispatchEventsResponse } from '$lib/gen'
+
+export type DispatchEvent = ListDispatchEventsResponse[number]
+
+// Shared loader for "the jobs this run dispatched". Both the inline button and
+// the run-detail panel render the same list, so keep the fetch in one place.
+// Returns a getter for the (never-undefined) event list — empty while loading,
+// when there's nothing to load, or before workspace/jobId are known.
+export function useDispatchEvents(
+	workspace: () => string,
+	jobId: () => string
+): { readonly list: DispatchEvent[] } {
+	const events = resource(
+		() => ({ workspace: workspace(), jobId: jobId() }),
+		async ({ workspace, jobId }) =>
+			workspace && jobId ? await JobService.listDispatchEvents({ workspace, id: jobId }) : []
+	)
+	return {
+		get list() {
+			return events.current ?? []
+		}
+	}
+}

frontend/src/lib/components/triggers/TriggerRunnablePicker.svelte

@@ -0,0 +1,59 @@
+<script lang="ts">
+	import type { Snippet } from 'svelte'
+	import Required from '$lib/components/Required.svelte'
+	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
+	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
+
+	// Shared runnable section for trigger editors. When `fixedScriptPath` is
+	// non-empty the drawer was opened from the pipeline editor for an
+	// already-bound script, so the ScriptPicker is swapped for a read-only
+	// viewer (see PipelineLockedRunnableInfo) to keep the trigger from being
+	// silently reassigned off the pipeline.
+	interface Props {
+		fixedScriptPath: string
+		itemKind: 'script' | 'flow'
+		scriptPath: string
+		initialScriptPath: string
+		canWrite: boolean
+		isOperator: boolean
+		promptText?: string
+		promptClass?: string
+		// Per-trigger "Create from template" button (hub URL, variant and any
+		// extra guard/tooltip differ per trigger kind, so the caller owns it).
+		createButton?: Snippet
+	}
+
+	let {
+		fixedScriptPath,
+		itemKind = $bindable(),
+		scriptPath = $bindable(),
+		initialScriptPath,
+		canWrite,
+		isOperator,
+		promptText = 'Pick a script or flow to be triggered',
+		promptClass = 'text-xs mb-1 text-primary',
+		createButton
+	}: Props = $props()
+</script>
+
+{#if fixedScriptPath != ''}
+	<PipelineLockedRunnableInfo path={fixedScriptPath} />
+{:else}
+	<p class={promptClass}>
+		{promptText}<Required required={true} />
+	</p>
+	<div class="flex flex-row mb-2">
+		<ScriptPicker
+			disabled={!canWrite}
+			initialPath={initialScriptPath}
+			kinds={['script']}
+			allowFlow={true}
+			bind:itemKind
+			bind:scriptPath
+			allowRefresh={canWrite}
+			allowEdit={!isOperator}
+			clearable
+		/>
+		{@render createButton?.()}
+	</div>
+{/if}

frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte

@@ -3,9 +3,7 @@
 	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import Path from '$lib/components/Path.svelte'
-	import Required from '$lib/components/Required.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import {
 		EmailTriggerService,
 		type ErrorHandler,
@@ -240,8 +238,8 @@
 		return {
 			noDeployed: !!(s as any)?.no_deployed,
 			overlay: draftFromBackend
-			? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
-			: undefined
+				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
+				: undefined
 		}
 	}
 
@@ -390,38 +388,27 @@
 
 			{#if !hideTarget}
 				<Section label="Target">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs mt-3 mb-1 text-primary">
-							Pick a script or flow to be triggered<Required required={true} />
-						</p>
-						<div class="flex flex-col gap-2">
-							<div class="flex flex-row mb-2">
-								<ScriptPicker
-									disabled={!can_write}
-									initialPath={initialScriptPath}
-									kinds={['script']}
-									allowFlow={true}
-									bind:itemKind
-									bind:scriptPath={script_path}
-									allowRefresh={can_write}
-									allowEdit={!$userStore?.operator}
-									clearable
-								/>
-
-								{#if emptyString(script_path)}
-									<Button
-										btnClasses="ml-4"
-										variant="accent"
-										size="xs"
-										href={itemKind === 'flow' ? '/flows/add?hub=72' : '/scripts/add?hub=hub%2F19813'}
-										target="_blank">Create from template</Button
-									>
-								{/if}
-							</div>
-						</div>
-					{/if}
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+						promptClass="text-xs mt-3 mb-1 text-primary"
+					>
+						{#snippet createButton()}
+							{#if emptyString(script_path)}
+								<Button
+									btnClasses="ml-4"
+									variant="accent"
+									size="xs"
+									href={itemKind === 'flow' ? '/flows/add?hub=72' : '/scripts/add?hub=hub%2F19813'}
+									target="_blank">Create from template</Button
+								>
+							{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte

@@ -18,9 +18,7 @@
 		type TriggerMode
 	} from '$lib/gen'
 	import Section from '$lib/components/Section.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
-	import Required from '$lib/components/Required.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import GcpTriggerEditorConfigSection from './GcpTriggerEditorConfigSection.svelte'
 	import { untrack, type Snippet } from 'svelte'
 	import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte'
@@ -213,11 +211,11 @@
 			const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any
 			loadTriggerConfig(deployedTrigger)
 			return {
-			noDeployed: !!(s as any)?.no_deployed,
-			overlay: draftFromBackend
-				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
-				: undefined
-		}
+				noDeployed: !!(s as any)?.no_deployed,
+				overlay: draftFromBackend
+					? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
+					: undefined
+			}
 		} catch (error) {
 			sendUserToast(`Could not load GCP Pub/Sub trigger: ${error.body}`, true)
 			return { overlay: undefined, noDeployed: false }
@@ -477,24 +475,16 @@
 
 			{#if !hideTarget}
 				<Section label="Runnable">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs mb-1 text-primary">
-							Pick a script or flow to be triggered <Required required={true} />
-						</p>
-						<div class="flex flex-row mb-2">
-							<ScriptPicker
-								disabled={!can_write}
-								initialPath={initialScriptPath}
-								kinds={['script']}
-								allowFlow={true}
-								bind:itemKind
-								bind:scriptPath={script_path}
-								allowRefresh={can_write}
-								allowEdit={!$userStore?.operator}
-								clearable
-							/>
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+						promptText="Pick a script or flow to be triggered "
+					>
+						{#snippet createButton()}
 							{#if emptyString(script_path)}
 								<Button
 									btnClasses="ml-4"
@@ -505,8 +495,8 @@
 									target="_blank">Create from template</Button
 								>
 							{/if}
-						</div>
-					{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte

@@ -4,9 +4,7 @@
 	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import Path from '$lib/components/Path.svelte'
-	import Required from '$lib/components/Required.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import { KafkaTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
 	import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
 	import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
@@ -271,8 +269,8 @@
 		return {
 			noDeployed: !!(s as any)?.no_deployed,
 			overlay: draftFromBackend
-			? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
-			: undefined
+				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
+				: undefined
 		}
 	}
 
@@ -525,24 +523,15 @@
 
 			{#if !hideTarget}
 				<Section label="Runnable">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs mb-1 text-primary">
-							Pick a script or flow to be triggered<Required required={true} />
-						</p>
-						<div class="flex flex-row mb-2">
-							<ScriptPicker
-								disabled={!can_write}
-								initialPath={initialScriptPath}
-								kinds={['script']}
-								allowFlow={true}
-								bind:itemKind
-								bind:scriptPath={script_path}
-								allowRefresh={can_write}
-								allowEdit={!$userStore?.operator}
-								clearable
-							/>
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+					>
+						{#snippet createButton()}
 							{#if emptyString(script_path)}
 								<Button
 									btnClasses="ml-4"
@@ -553,8 +542,8 @@
 									target="_blank">Create from template</Button
 								>
 							{/if}
-						</div>
-					{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte

@@ -4,9 +4,7 @@
 	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import Path from '$lib/components/Path.svelte'
-	import Required from '$lib/components/Required.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
 	import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
 	import { withForkConflictRetry } from '$lib/utils/forkConflict'
@@ -269,11 +267,11 @@
 			const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any
 			loadTriggerConfig(deployedTrigger)
 			return {
-			noDeployed: !!(s as any)?.no_deployed,
-			overlay: draftFromBackend
-				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
-				: undefined
-		}
+				noDeployed: !!(s as any)?.no_deployed,
+				overlay: draftFromBackend
+					? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
+					: undefined
+			}
 		} catch (error) {
 			sendUserToast(`Could not load mqtt trigger: ${error.body}`, true)
 			return { overlay: undefined, noDeployed: false }
@@ -500,24 +498,15 @@
 
 			{#if !hideTarget}
 				<Section label="Runnable">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs mb-1 text-primary">
-							Pick a script or flow to be triggered<Required required={true} />
-						</p>
-						<div class="flex flex-row mb-2">
-							<ScriptPicker
-								disabled={!can_write}
-								initialPath={initialScriptPath}
-								kinds={['script']}
-								allowFlow={true}
-								bind:itemKind
-								bind:scriptPath={script_path}
-								allowRefresh={can_write}
-								allowEdit={!$userStore?.operator}
-								clearable
-							/>
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+					>
+						{#snippet createButton()}
 							{#if emptyString(script_path)}
 								<Button
 									btnClasses="ml-4"
@@ -530,8 +519,8 @@
 									Create from template
 								</Button>
 							{/if}
-						</div>
-					{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte

@@ -3,9 +3,7 @@
 	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import Path from '$lib/components/Path.svelte'
-	import Required from '$lib/components/Required.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import { NatsTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
 	import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
 	import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
@@ -267,8 +265,8 @@
 		return {
 			noDeployed: !!(s as any)?.no_deployed,
 			overlay: draftFromBackend
-			? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
-			: undefined
+				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
+				: undefined
 		}
 	}
 
@@ -495,24 +493,15 @@
 			</div>
 			{#if !hideTarget}
 				<Section label="Runnable">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs mb-1 text-primary">
-							Pick a script or flow to be triggered<Required required={true} />
-						</p>
-						<div class="flex flex-row mb-2">
-							<ScriptPicker
-								disabled={!can_write}
-								initialPath={initialScriptPath}
-								kinds={['script']}
-								allowFlow={true}
-								bind:itemKind
-								bind:scriptPath={script_path}
-								allowRefresh={can_write}
-								allowEdit={!$userStore?.operator}
-								clearable
-							/>
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+					>
+						{#snippet createButton()}
 							{#if emptyString(script_path)}
 								<Button
 									btnClasses="ml-4"
@@ -524,8 +513,8 @@
 									Create from template
 								</Button>
 							{/if}
-						</div>
-					{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte

@@ -4,8 +4,7 @@
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import Path from '$lib/components/Path.svelte'
 	import Required from '$lib/components/Required.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import {
 		PostgresTriggerService,
 		type ErrorHandler,
@@ -672,25 +671,17 @@
 			</Label>
 			{#if !hideTarget}
 				<Section label="Runnable">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs text-primary">
-							Pick a script or flow to be triggered <Required required={true} />
-						</p>
-						<div class="flex flex-row mb-2">
-							<ScriptPicker
-								disabled={!can_write}
-								initialPath={initialScriptPath}
-								kinds={['script']}
-								allowFlow={true}
-								bind:itemKind
-								bind:scriptPath={script_path}
-								allowRefresh={can_write}
-								allowEdit={!$userStore?.operator}
-								clearable
-							/>
-
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+						promptText="Pick a script or flow to be triggered "
+						promptClass="text-xs text-primary"
+					>
+						{#snippet createButton()}
 							{#if emptyString(script_path) && is_flow === false}
 								<div class="flex">
 									<Button
@@ -711,8 +702,8 @@
 									</Button>
 								</div>
 							{/if}
-						</div>
-					{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte

@@ -17,9 +17,7 @@
 	} from '$lib/gen'
 	import SqsTriggerEditorConfigSection from './SqsTriggerEditorConfigSection.svelte'
 	import Section from '$lib/components/Section.svelte'
-	import ScriptPicker from '$lib/components/ScriptPicker.svelte'
-	import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte'
-	import Required from '$lib/components/Required.svelte'
+	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
 	import { untrack, type Snippet } from 'svelte'
 	import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte'
 	import PermissionedAsLine from '../PermissionedAsLine.svelte'
@@ -247,11 +245,11 @@
 			const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any
 			loadTriggerConfig(deployedTrigger)
 			return {
-			noDeployed: !!(s as any)?.no_deployed,
-			overlay: draftFromBackend
-				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
-				: undefined
-		}
+				noDeployed: !!(s as any)?.no_deployed,
+				overlay: draftFromBackend
+					? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
+					: undefined
+			}
 		} catch (error) {
 			sendUserToast(`Could not load SQS trigger: ${error.body}`, true)
 			return { overlay: undefined, noDeployed: false }
@@ -479,24 +477,16 @@
 
 			{#if !hideTarget}
 				<Section label="Runnable">
-					{#if fixedScriptPath != ''}
-						<PipelineLockedRunnableInfo path={fixedScriptPath} />
-					{:else}
-						<p class="text-xs mb-1 text-primary">
-							Pick a script or flow to be triggered <Required required={true} />
-						</p>
-						<div class="flex flex-row mb-2">
-							<ScriptPicker
-								disabled={!can_write}
-								initialPath={initialScriptPath}
-								kinds={['script']}
-								allowFlow={true}
-								bind:itemKind
-								bind:scriptPath={script_path}
-								allowRefresh={can_write}
-								allowEdit={!$userStore?.operator}
-								clearable
-							/>
+					<TriggerRunnablePicker
+						{fixedScriptPath}
+						bind:itemKind
+						bind:scriptPath={script_path}
+						{initialScriptPath}
+						canWrite={can_write}
+						isOperator={!!$userStore?.operator}
+						promptText="Pick a script or flow to be triggered "
+					>
+						{#snippet createButton()}
 							{#if emptyString(script_path)}
 								<Button
 									btnClasses="ml-4"
@@ -509,8 +499,8 @@
 									Create from template
 								</Button>
 							{/if}
-						</div>
-					{/if}
+						{/snippet}
+					</TriggerRunnablePicker>
 				</Section>
 			{/if}
 

frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte

@@ -19,7 +19,6 @@
 	import PipelineActivityPanel from '$lib/components/assets/AssetGraph/PipelineActivityPanel.svelte'
 	import AssetGraphDetailsPane from '$lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte'
 	import PipelinePickerModal from '$lib/components/assets/AssetGraph/PipelinePickerModal.svelte'
-	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
 	import {
 		extractWrites,
 		extractReads,
@@ -71,17 +70,9 @@
 		Telescope
 	} from 'lucide-svelte'
 	import {
-		EmailTriggerService,
-		GcpTriggerService,
 		JobService,
-		KafkaTriggerService,
-		MqttTriggerService,
-		NatsTriggerService,
 		OpenAPI,
-		PostgresTriggerService,
-		ScheduleService,
 		ScriptService,
-		SqsTriggerService,
 		type AssetKind,
 		type Script,
 		type ScriptLang
@@ -95,15 +86,7 @@
 	import Popover from '$lib/components/meltComponents/Popover.svelte'
 	import HideButton from '$lib/components/apps/editor/settingsPanel/HideButton.svelte'
 	import { inferArgs, inferAssets } from '$lib/infer'
-	import KafkaTriggerEditor from '$lib/components/triggers/kafka/KafkaTriggerEditor.svelte'
-	import MqttTriggerEditor from '$lib/components/triggers/mqtt/MqttTriggerEditor.svelte'
-	import NatsTriggerEditor from '$lib/components/triggers/nats/NatsTriggerEditor.svelte'
-	import PostgresTriggerEditor from '$lib/components/triggers/postgres/PostgresTriggerEditor.svelte'
-	import SqsTriggerEditor from '$lib/components/triggers/sqs/SqsTriggerEditor.svelte'
-	import GcpTriggerEditor from '$lib/components/triggers/gcp/GcpTriggerEditor.svelte'
-	import EmailTriggerEditor from '$lib/components/triggers/email/EmailTriggerEditor.svelte'
-	import ScheduleEditor from '$lib/components/triggers/schedules/ScheduleEditor.svelte'
-	import WebhookEditor from '$lib/components/triggers/webhook/WebhookEditor.svelte'
+	import PipelineTriggerEditors from '$lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte'
 
 	// Variables and resources are declarative config, not pipeline assets —
 	// they're hub-shaped (referenced by most runnables) and would swamp the
@@ -128,10 +111,7 @@
 			selection = undefined
 			activeDraftPath = undefined
 			panelHidden = false
-			liveAnnotations = {
-				scriptPath: undefined,
-				annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
-			}
+			liveAnnotations = EMPTY_LIVE_ANNOTATIONS
 		}
 		const url = new URL(page.url)
 		if (m === 'view') url.searchParams.delete('mode')
@@ -242,10 +222,7 @@
 			panelHidden = false
 			// Same reset as the pane's close button — clears the live
 			// annotation overlay of whichever script was open.
-			liveAnnotations = {
-				scriptPath: undefined,
-				annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
-			}
+			liveAnnotations = EMPTY_LIVE_ANNOTATIONS
 		}
 	}
 
@@ -503,6 +480,24 @@
 		content: ''
 	})
 
+	// Canonical "empty" overlay literals, reused both as reset values for the
+	// live-* state above and as the no-overlay inputs to the deployed graph.
+	const EMPTY_LIVE_ASSETS = { scriptPath: undefined, assets: [] }
+	const EMPTY_LIVE_ANNOTATIONS = {
+		scriptPath: undefined,
+		annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
+	}
+
+	// Reset every live editor overlay (annotations / body assets / content)
+	// back to empty, unconditionally. Used by the leave-edit path so a stale
+	// buffer for the previously-open script can't leak into the view graphs.
+	// (forgetPath resets these per-path instead — see there.)
+	function clearLiveOverlays() {
+		liveAnnotations = EMPTY_LIVE_ANNOTATIONS
+		liveBodyAssets = EMPTY_LIVE_ASSETS
+		liveContent = { scriptPath: undefined, content: '' }
+	}
+
 	// Only-add cache of (script_path → body content) populated lazily by
 	// `bodyFetchEffect` for every script in the current folder. We never
 	// remove entries: stale keys (renamed-away, deleted) are simply ignored
@@ -524,10 +519,15 @@
 	// Iteration is gated on `graphRes.current.runnables`, so a path that
 	// gets renamed / deleted disappears from the derived map as soon as
 	// the refetch lands — no manual rekey, no phantom edges.
-	let inferredWritesByPath = $derived.by(() => {
-		const out = new Map<string, Array<{ kind: AssetKind; path: string }>>()
+	// Single pass over the graph's scripts producing both the write- and
+	// read-asset maps (they only differ by extractWrites vs extractReads over
+	// the same `liveBodyAssets`-vs-cache asset source). Split into two derives
+	// below so consumers can depend on one without invalidating on the other.
+	let inferredAssetEdges = $derived.by(() => {
+		const writes = new Map<string, Array<{ kind: AssetKind; path: string }>>()
+		const reads = new Map<string, Array<{ kind: AssetKind; path: string }>>()
 		const g = graphRes.current
-		if (!g) return out
+		if (!g) return { writes, reads }
 		const liveAssetsForPath = (path: string) =>
 			liveBodyAssets.scriptPath === path ? liveBodyAssets.assets : inferredAssetsByPath.get(path)
 		for (const r of g.runnables) {
@@ -535,25 +535,14 @@
 			const assets = liveAssetsForPath(r.path)
 			if (!assets) continue
 			const w = extractWrites(assets)
-			if (w.length > 0) out.set(r.path, w)
+			if (w.length > 0) writes.set(r.path, w)
+			const rd = extractReads(assets)
+			if (rd.length > 0) reads.set(r.path, rd)
 		}
-		return out
-	})
-	let inferredReadsByPath = $derived.by(() => {
-		const out = new Map<string, Array<{ kind: AssetKind; path: string }>>()
-		const g = graphRes.current
-		if (!g) return out
-		const liveAssetsForPath = (path: string) =>
-			liveBodyAssets.scriptPath === path ? liveBodyAssets.assets : inferredAssetsByPath.get(path)
-		for (const r of g.runnables) {
-			if (r.usage_kind !== 'script') continue
-			const assets = liveAssetsForPath(r.path)
-			if (!assets) continue
-			const reads = extractReads(assets)
-			if (reads.length > 0) out.set(r.path, reads)
-		}
-		return out
+		return { writes, reads }
 	})
+	let inferredWritesByPath = $derived(inferredAssetEdges.writes)
+	let inferredReadsByPath = $derived(inferredAssetEdges.reads)
 	// Same derived shape for `// on kafka` etc. annotations. Live buffer
 	// wins for the open script; everyone else is parsed from the
 	// prefetched body content.
@@ -947,13 +936,10 @@
 			selection = undefined
 		}
 		if (liveAnnotations.scriptPath === path) {
-			liveAnnotations = {
-				scriptPath: undefined,
-				annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
-			}
+			liveAnnotations = EMPTY_LIVE_ANNOTATIONS
 		}
 		if (liveBodyAssets.scriptPath === path) {
-			liveBodyAssets = { scriptPath: undefined, assets: [] }
+			liveBodyAssets = EMPTY_LIVE_ASSETS
 		}
 		if (liveContent.scriptPath === path) {
 			liveContent = { scriptPath: undefined, content: '' }
@@ -1135,6 +1121,16 @@
 	// Currently-open draft shape (if any) — fed into the details pane.
 	let activeDraft = $derived(activeDraftPath ? drafts.get(activeDraftPath) : undefined)
 
+	// Path of the script currently open in the details pane (draft or
+	// persisted selection), used wherever run-routing / overlay logic needs
+	// "the one script the user is editing right now".
+	let openScriptPath = $derived(
+		activeDraftPath ??
+			(selection?.kind === 'runnable' && selection.runnable_kind === 'script'
+				? selection.path
+				: undefined)
+	)
+
 	// Entering edit mode with a selection whose path has a draft (e.g. a
 	// deployed script whose unsaved edits were promoted on the switch to
 	// view) re-opens the draft, not the stale deployed version — same
@@ -1170,7 +1166,7 @@
 	// `handleDraftContentChange`, that creates a parent ↔ child feedback
 	// loop ("effect_update_depth_exceeded"). Named functions keep the
 	// prop reference stable so the $effects only re-fire on real
-	// content changes.
+	// content changes (e.g. handleContentChange below mutating drafts).
 	function handleAnnotationsChange(
 		scriptPath: string | undefined,
 		annotations: PipelineAnnotations
@@ -1365,12 +1361,7 @@
 		// pane, route through ScriptEditor's Test path — the test panel then
 		// shows logs/result and the user can cancel from there. Same UX as
 		// hitting the Test button directly.
-		const openPath =
-			activeDraftPath ??
-			(selection?.kind === 'runnable' && selection.runnable_kind === 'script'
-				? selection.path
-				: undefined)
-		if (openPath === producer.path) {
+		if (openScriptPath === producer.path) {
 			if (cascade) requestRunCascadeSignal++
 			else requestRunSignal++
 			return undefined
@@ -1441,11 +1432,7 @@
 	// drives the read-only pane's run form. Derived from the displayed
 	// graph's triggers so the drafts overlay picks up draft annotations too.
 	let openScriptHasDataUpload = $derived.by(() => {
-		const path =
-			activeDraftPath ??
-			(selection?.kind === 'runnable' && selection.runnable_kind === 'script'
-				? selection.path
-				: undefined)
+		const path = openScriptPath
 		if (!path) return false
 		return displayGraph.triggers.some(
 			(t) =>
@@ -1476,11 +1463,6 @@
 	// but with no drafts and no live editor buffer. resolveGraph is pure, so
 	// feeding it empty overlays is the cheapest way to share the logic.
 	const EMPTY_DRAFTS: Map<string, Draft> = new Map()
-	const EMPTY_LIVE_ASSETS = { scriptPath: undefined, assets: [] }
-	const EMPTY_LIVE_ANNOTATIONS = {
-		scriptPath: undefined,
-		annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
-	}
 	let deployedGraph = $derived.by<AssetGraphResponse>(() =>
 		resolveGraph({
 			base: graphRes.current ?? EMPTY_GRAPH,
@@ -1504,15 +1486,14 @@
 	$effect(() => {
 		if (mode === 'edit') return
 		untrack(() => {
-			if (liveAnnotations.scriptPath != undefined || liveBodyAssets.scriptPath != undefined) {
-				liveAnnotations = {
-					scriptPath: undefined,
-					annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] }
-				}
-				liveBodyAssets = { scriptPath: undefined, assets: [] }
-			}
-			if (liveContent.scriptPath != undefined) {
-				liveContent = { scriptPath: undefined, content: '' }
+			// Guard before reassigning so an already-empty overlay doesn't
+			// needlessly invalidate the graph derives every mode toggle.
+			if (
+				liveAnnotations.scriptPath != undefined ||
+				liveBodyAssets.scriptPath != undefined ||
+				liveContent.scriptPath != undefined
+			) {
+				clearLiveOverlays()
 			}
 		})
 	})
@@ -1685,9 +1666,15 @@
 	}
 	// Poll a launched cascade job to a terminal state. Modest fixed cadence —
 	// chains are short and the folder poll is already watching the same jobs
-	// for the canvas animation.
+	// for the canvas animation. Capped so a never-terminating job can't pin
+	// `cascadeRunningRoot` forever and wedge every future cascade: after the
+	// timeout we throw, which the orchestrator surfaces as a chain failure and
+	// `runDraftAwareCascade`'s finally clears the running-root guard.
+	const CASCADE_POLL_INTERVAL_MS = 1000
+	const CASCADE_JOB_TIMEOUT_MS = 30 * 60 * 1000
 	async function waitJobTerminal(jobId: string): Promise<'success' | 'failure'> {
-		while (true) {
+		const deadline = Date.now() + CASCADE_JOB_TIMEOUT_MS
+		while (Date.now() < deadline) {
 			try {
 				const r = await JobService.getCompletedJobResultMaybe({
 					workspace: $workspaceStore!,
@@ -1698,8 +1685,11 @@
 			} catch {
 				// transient — retry on the next tick
 			}
-			await new Promise((res) => setTimeout(res, 1000))
+			await new Promise((res) => setTimeout(res, CASCADE_POLL_INTERVAL_MS))
 		}
+		throw new Error(
+			`Timed out after ${Math.round(CASCADE_JOB_TIMEOUT_MS / 60000)}min waiting for job ${jobId} to finish`
+		)
 	}
 	// "Run + downstream" over a chain that includes drafts: the backend
 	// asset-trigger dispatcher only resolves deployed rows, so the page
@@ -1806,11 +1796,7 @@
 	// in `triggers`) other than self. Flows are excluded because V1 dispatch
 	// only fans out to scripts.
 	let editedScriptDownstreamCount = $derived.by(() => {
-		const editedPath =
-			activeDraftPath ??
-			(selection?.kind === 'runnable' && selection.runnable_kind === 'script'
-				? selection.path
-				: undefined)
+		const editedPath = openScriptPath
 		if (!editedPath) return 0
 		const writes = graphWithDraft.edges.filter(
 			(e) =>
@@ -1841,20 +1827,10 @@
 	// folder…" entry in the dropdown otherwise.
 	let pickerModalOpen = $state(false)
 
-	// Native trigger editors mounted inline so clicking a "missing"
-	// placeholder opens the matching drawer with `script_path` pre-filled
-	// — keeps pipeline drafts intact instead of navigating away. Each
-	// editor's wrapper lazily mounts its Inner only when `open=true`, so
-	// holding refs to all seven is cheap.
-	let kafkaEditor: KafkaTriggerEditor | undefined = $state()
-	let mqttEditor: MqttTriggerEditor | undefined = $state()
-	let natsEditor: NatsTriggerEditor | undefined = $state()
-	let postgresEditor: PostgresTriggerEditor | undefined = $state()
-	let sqsEditor: SqsTriggerEditor | undefined = $state()
-	let gcpEditor: GcpTriggerEditor | undefined = $state()
-	let emailEditor: EmailTriggerEditor | undefined = $state()
-	let scheduleEditor: ScheduleEditor | undefined = $state()
-	let webhookEditor: WebhookEditor | undefined = $state()
+	// Native trigger editor drawers live in <PipelineTriggerEditors>; the page
+	// drives them imperatively. The draft guards stay here because they depend
+	// on the drafts map.
+	let triggerEditors: PipelineTriggerEditors | undefined = $state()
 
 	// Webhooks have no trigger row to create — clicking the node opens a
 	// drawer with the endpoint URLs + the webhook-specific token creation
@@ -1868,7 +1844,7 @@
 			)
 			return
 		}
-		webhookEditor?.openDrawer(scriptPath, false)
+		triggerEditors?.openWebhook(scriptPath)
 	}
 
 	// Data upload is a UI-first entry point — no trigger row. Clicking the
@@ -1910,112 +1886,19 @@
 			)
 			return
 		}
-		switch (kind) {
-			case 'schedule':
-				return scheduleEditor?.openNew(false, scriptPath, undefined, scriptPath)
-			case 'kafka':
-				return kafkaEditor?.openNew(false, scriptPath)
-			case 'mqtt':
-				return mqttEditor?.openNew(false, scriptPath)
-			case 'nats':
-				return natsEditor?.openNew(false, scriptPath)
-			case 'postgres':
-				return postgresEditor?.openNew(false, scriptPath)
-			case 'sqs':
-				return sqsEditor?.openNew(false, scriptPath)
-			case 'gcp':
-				return gcpEditor?.openNew(false, scriptPath)
-			case 'email':
-				return emailEditor?.openNew(false, scriptPath)
-			// webhook has no dedicated editor.
-			default:
-				return
-		}
+		triggerEditors?.openNew(kind, scriptPath)
 	}
 
 	// Lock the script-picker to the related script so the user can't
 	// reassign the trigger off this pipeline from the canvas. The trigger
 	// can still be edited everywhere else (TriggersPanel, etc.) where the
 	// picker stays editable.
-	// Delete-trigger confirmation state. Kebab → Delete opens the standard
-	// ConfirmationModal; the actual delete is dispatched from onConfirmed
-	// so the dialog stays consistent with the rest of the app.
-	let triggerDeleteTarget = $state<{ kind: NativeTriggerKind; path: string } | undefined>(undefined)
-	let triggerDeleteLoading = $state(false)
-	let triggerDeleteOpen = $derived(triggerDeleteTarget != undefined)
-
 	function deleteAttachedTrigger(kind: NativeTriggerKind, triggerPath: string) {
-		triggerDeleteTarget = { kind, path: triggerPath }
-	}
-
-	async function confirmDeleteAttachedTrigger() {
-		if (!triggerDeleteTarget || !$workspaceStore) return
-		const { kind, path: triggerPath } = triggerDeleteTarget
-		const workspace = $workspaceStore
-		triggerDeleteLoading = true
-		try {
-			switch (kind) {
-				case 'schedule':
-					await ScheduleService.deleteSchedule({ workspace, path: triggerPath })
-					break
-				case 'kafka':
-					await KafkaTriggerService.deleteKafkaTrigger({ workspace, path: triggerPath })
-					break
-				case 'mqtt':
-					await MqttTriggerService.deleteMqttTrigger({ workspace, path: triggerPath })
-					break
-				case 'nats':
-					await NatsTriggerService.deleteNatsTrigger({ workspace, path: triggerPath })
-					break
-				case 'postgres':
-					await PostgresTriggerService.deletePostgresTrigger({ workspace, path: triggerPath })
-					break
-				case 'sqs':
-					await SqsTriggerService.deleteSqsTrigger({ workspace, path: triggerPath })
-					break
-				case 'gcp':
-					await GcpTriggerService.deleteGcpTrigger({ workspace, path: triggerPath })
-					break
-				case 'email':
-					await EmailTriggerService.deleteEmailTrigger({ workspace, path: triggerPath })
-					break
-				default:
-					return
-			}
-			sendUserToast(`Deleted ${kind} trigger "${triggerPath}"`)
-			triggerDeleteTarget = undefined
-			await graphRes.refetch()
-		} catch (e: any) {
-			sendUserToast(
-				`Could not delete ${kind} trigger "${triggerPath}": ${e?.body ?? e?.message ?? String(e)}`,
-				true
-			)
-		} finally {
-			triggerDeleteLoading = false
-		}
+		triggerEditors?.requestDelete(kind, triggerPath)
 	}
 
 	function openEditTriggerDrawer(kind: NativeTriggerKind, triggerPath: string, scriptPath: string) {
-		switch (kind) {
-			case 'schedule':
-				return scheduleEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'kafka':
-				return kafkaEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'mqtt':
-				return mqttEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'nats':
-				return natsEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'postgres':
-				return postgresEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'sqs':
-				return sqsEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'gcp':
-				return gcpEditor?.openEdit(triggerPath, false, scriptPath)
-			case 'email':
-				return emailEditor?.openEdit(triggerPath, false, scriptPath)
-			default:
-				return
-		}
+		triggerEditors?.openEdit(kind, triggerPath, scriptPath)
 	}
 
 	// Reuse the empty AssetGraphResponse so we can still render the canvas
@@ -2456,11 +2339,7 @@
 									// per-node Run button. The currently-edited script
 									// is whichever path is open in the pane (active
 									// draft, or the persisted-script selection).
-									const openPath =
-										activeDraftPath ??
-										(selection?.kind === 'runnable' && selection.runnable_kind === 'script'
-											? selection.path
-											: undefined)
+									const openPath = openScriptPath
 									if (running && openPath) {
 										activeRunnable = { kind: 'script', path: openPath }
 										// Mark the tested runnable as launched-from-here so the
@@ -2497,14 +2376,7 @@
 									// via the explicit "Discard" button in the pane.
 									selection = undefined
 									activeDraftPath = undefined
-									liveAnnotations = {
-										scriptPath: undefined,
-										annotations: {
-											inPipeline: false,
-											triggerAssets: [],
-											nativeTriggers: []
-										}
-									}
+									liveAnnotations = EMPTY_LIVE_ANNOTATIONS
 								}}
 								onHide={() => (panelHidden = true)}
 								onDiscard={() => {
@@ -2583,44 +2455,14 @@
 
 <PipelinePickerModal bind:open={pickerModalOpen} currentFolder={folder} {mode} />
 
-{#if mode === 'edit'}
-	<ConfirmationModal
-		open={triggerDeleteOpen}
-		loading={triggerDeleteLoading}
-		title={triggerDeleteTarget ? `Delete ${triggerDeleteTarget.kind} trigger` : 'Delete trigger'}
-		confirmationText="Delete"
-		onConfirmed={confirmDeleteAttachedTrigger}
-		onCanceled={() => {
-			if (!triggerDeleteLoading) triggerDeleteTarget = undefined
-		}}
-	>
-		{#if triggerDeleteTarget}
-			<p>
-				Delete <code class="font-mono">{triggerDeleteTarget.path}</code>? The
-				<code class="font-mono">// on {triggerDeleteTarget.kind}</code> annotation on the script stays
-				— the trigger will read as missing on the canvas until you recreate it or remove the annotation.
-			</p>
-		{/if}
-	</ConfirmationModal>
-
-	<!-- Native trigger editors mounted off-screen. Each only renders its
-	     inner drawer when `open=true` (set by openNew/openEdit), so this
-	     adds ~zero render cost while idle. `onUpdate` refreshes the graph
-	     so the new trigger row replaces the red missing placeholder.
-	     Edit-mode only: every entry point (create/edit/delete trigger) is
-	     gated off the canvas outside edit mode. -->
-	<KafkaTriggerEditor bind:this={kafkaEditor} onUpdate={() => graphRes.refetch()} />
-	<MqttTriggerEditor bind:this={mqttEditor} onUpdate={() => graphRes.refetch()} />
-	<NatsTriggerEditor bind:this={natsEditor} onUpdate={() => graphRes.refetch()} />
-	<PostgresTriggerEditor bind:this={postgresEditor} onUpdate={() => graphRes.refetch()} />
-	<SqsTriggerEditor bind:this={sqsEditor} onUpdate={() => graphRes.refetch()} />
-	<GcpTriggerEditor bind:this={gcpEditor} onUpdate={() => graphRes.refetch()} />
-	<EmailTriggerEditor bind:this={emailEditor} onUpdate={() => graphRes.refetch()} />
-	<ScheduleEditor bind:this={scheduleEditor} onUpdate={() => graphRes.refetch()} />
-{/if}
-<!-- Webhook drawer stays mounted in every mode — the webhook trigger node
-     is clickable in view mode too (informational: endpoint URLs/token). -->
-<WebhookEditor bind:this={webhookEditor} />
+<!-- Native trigger drawer wiring: create/edit/delete drawers (edit-mode
+     only) + the always-mounted webhook drawer. Driven imperatively from the
+     page via `triggerEditors`. -->
+<PipelineTriggerEditors
+	bind:this={triggerEditors}
+	mountTriggerEditors={mode === 'edit'}
+	onUpdate={() => graphRes.refetch()}
+/>
 
 {#if leaveModalOpen}
 	<!-- Three-button leave guard. Built inline rather than reusing