feat: path-less native trigger markers + missing-trigger placeholder

2026-08-20 08:01:35 +00:00 · 2026-05-21 20:55:28 +00:00
parent bdb97aba19
commit aeaa75ccf1
36 changed files with 780 additions and 251 deletions
@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        DELETE FROM asset\n        WHERE (workspace_id, path, kind) IN (\n            SELECT workspace_id, path, kind FROM (\n                SELECT a.workspace_id, a.path, a.kind, a.usage_kind, ROW_NUMBER() OVER (\n                    PARTITION BY a.workspace_id, a.path, a.kind\n                    ORDER BY a.created_at DESC\n                ) as rn,\n                limits.max_n\n                FROM asset a\n                INNER JOIN (\n                    SELECT * FROM UNNEST(\n                        $1::varchar[], \n                        $2::varchar[], \n                        $3::asset_kind[],\n                        $4::int[]\n                    ) AS t(workspace_id, path, kind, max_n)\n                ) limits\n                ON a.workspace_id = limits.workspace_id \n                AND a.path = limits.path \n                AND a.kind = limits.kind\n                WHERE a.usage_kind = 'job'\n            ) ranked\n            WHERE rn > max_n\n        )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "VarcharArray",
+        "VarcharArray",
+        {
+          "Custom": {
+            "name": "asset_kind[]",
+            "kind": {
+              "Array": {
+                "Custom": {
+                  "name": "asset_kind",
+                  "kind": {
+                    "Enum": [
+                      "s3object",
+                      "resource",
+                      "variable",
+                      "ducklake",
+                      "datatable",
+                      "volume"
+                    ]
+                  }
+                }
+              }
+            }
+          }
+        },
+        "Int4Array"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "02e526146f3584cd599dec708e1be48db3b0cd1c74adbfa2e4039377daa016f0"
+}
@@ -0,0 +1,78 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                                SELECT\n                                    flow_version.id AS version,\n                                    flow_version.value->>'early_return' as early_return,\n                                    flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor,\n                                    flow_version.value->>'failure_module' IS NOT NULL as has_failure_module,\n                                    (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled,\n                                    flow.tag,\n                                    flow.dedicated_worker,\n                                    flow.on_behalf_of_email,\n                                    flow.edited_by,\n                                    flow.labels\n                                FROM\n                                    flow_version\n                                INNER JOIN flow\n                                    ON flow.path = flow_version.path AND\n                                       flow.workspace_id = flow_version.workspace_id\n                                WHERE\n                                    flow_version.workspace_id = $1 AND\n                                    flow_version.path = $2 AND\n                                    flow_version.id = $3\n                            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "version",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "early_return",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "has_preprocessor",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 3,
+        "name": "has_failure_module",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 4,
+        "name": "chat_input_enabled",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 5,
+        "name": "tag",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 6,
+        "name": "dedicated_worker",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 7,
+        "name": "on_behalf_of_email",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 8,
+        "name": "edited_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 9,
+        "name": "labels",
+        "type_info": "TextArray"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      null,
+      null,
+      null,
+      null,
+      true,
+      true,
+      true,
+      false,
+      true
+    ]
+  },
+  "hash": "04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835"
+}
@@ -34,7 +34,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "DELETE FROM schedule\n                   WHERE workspace_id = $1\n                     AND managed_by_runnable_path = $2",
+  "query": "DELETE FROM schedule\n                   WHERE workspace_id = $1\n                     AND script_path = $2\n                     AND managed",
  "describe": {
    "columns": [],
    "parameters": {
@@ -11,5 +11,5 @@
    },
    "nullable": []
  },
-  "hash": "2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c"
+  "hash": "3194753fa26b1395fdbe3a7a434fe061db12e3bc4cb810e05110cd8115e1ca07"
 }
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                INSERT INTO schedule (\n                    workspace_id, path, schedule, timezone, edited_by, script_path,\n                    is_flow, enabled, email, permissioned_as,\n                    ws_error_handler_muted, no_flow_overlap, cron_version,\n                    managed\n                )\n                VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', true)\n                ON CONFLICT (workspace_id, path) DO UPDATE\n                SET schedule = EXCLUDED.schedule,\n                    edited_at = now(),\n                    edited_by = EXCLUDED.edited_by,\n                    managed = true\n                WHERE schedule.managed\n                   OR schedule.script_path = EXCLUDED.script_path\n                ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9"
+}
@@ -79,7 +79,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -15,7 +15,7 @@
      ]
    },
    "nullable": [
-      true
+      null
    ]
  },
  "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
@@ -160,7 +160,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -127,7 +127,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -34,7 +34,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -190,7 +190,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "\n        SELECT\n            runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n            runnable_path AS \"runnable_path!\",\n            trigger_kind::text AS \"trigger_kind!\",\n            trigger_ref   AS \"trigger_ref!\"\n        FROM script_trigger\n        WHERE workspace_id = $1\n          AND ($2::text IS NULL OR runnable_path LIKE $2)\n        ",
+  "query": "\n        SELECT\n            runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n            runnable_path AS \"runnable_path!\",\n            trigger_kind::text AS \"trigger_kind!\",\n            trigger_ref   AS \"trigger_ref!\"\n        FROM script_trigger\n        WHERE workspace_id = $1\n          AND trigger_kind IN ('asset', 'schedule')\n          AND ($2::text IS NULL OR runnable_path LIKE $2)\n        ",
  "describe": {
    "columns": [
      {
@@ -48,5 +48,5 @@
      false
    ]
  },
-  "hash": "c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08"
+  "hash": "ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce"
 }
@@ -165,7 +165,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -79,7 +79,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -110,7 +110,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -1,20 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n                INSERT INTO schedule (\n                    workspace_id, path, schedule, timezone, edited_by, script_path,\n                    is_flow, enabled, email, permissioned_as,\n                    ws_error_handler_muted, no_flow_overlap, cron_version,\n                    managed_by_runnable_path\n                )\n                VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', $2)\n                ON CONFLICT (workspace_id, path) DO UPDATE\n                SET schedule = EXCLUDED.schedule,\n                    edited_at = now(),\n                    edited_by = EXCLUDED.edited_by,\n                    managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n                WHERE schedule.managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n                   OR schedule.managed_by_runnable_path IS NULL AND schedule.script_path = EXCLUDED.script_path\n                ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Bool",
-        "Varchar",
-        "Varchar"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df"
-}
@@ -110,7 +110,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -250,7 +250,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job SET args = $1, preprocessed = TRUE WHERE id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Jsonb",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e3ee812acd5bb9d5af39ca7dc61481ddd56a641e8e6f6c7a145f2cd5f3dc4602"
+}
@@ -190,7 +190,8 @@
                "google",
                "ci_test",
                "github",
-                "azure"
+                "azure",
+                "asset"
              ]
            }
          }
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "DELETE FROM schedule\n           WHERE workspace_id = $1\n             AND managed_by_runnable_path = $2",
+  "query": "DELETE FROM schedule\n           WHERE workspace_id = $1\n             AND script_path = $2\n             AND managed",
  "describe": {
    "columns": [],
    "parameters": {
@@ -11,5 +11,5 @@
    },
    "nullable": []
  },
-  "hash": "818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05"
+  "hash": "edf2814409bd480ac7c61140ec8d91b461782169bad6d47127c83bc5992d30ec"
 }
@@ -0,0 +1,41 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT kind, path, script_path, is_flow FROM (\n            SELECT 'email' AS kind, path, script_path, is_flow FROM email_trigger\n                WHERE workspace_id = $1\n            UNION ALL\n            SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger\n                WHERE workspace_id = $1\n            UNION ALL\n            SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger\n                WHERE workspace_id = $1\n            UNION ALL\n            SELECT 'nats', path, script_path, is_flow FROM nats_trigger\n                WHERE workspace_id = $1\n            UNION ALL\n            SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger\n                WHERE workspace_id = $1\n            UNION ALL\n            SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger\n                WHERE workspace_id = $1\n            UNION ALL\n            SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger\n                WHERE workspace_id = $1\n        ) t\n        WHERE ($2::text IS NULL OR script_path LIKE $2)\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "kind",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "script_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "is_flow",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null,
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3"
+}
@@ -1,2 +1,2 @@
-DROP INDEX IF EXISTS idx_schedule_managed_by_runnable_path;
-ALTER TABLE schedule DROP COLUMN IF EXISTS managed_by_runnable_path;
+DROP INDEX IF EXISTS idx_schedule_managed;
+ALTER TABLE schedule DROP COLUMN IF EXISTS managed;
@@ -1,12 +1,14 @@
-- Track schedules auto-created from a materializer's `// schedule "<cron>"`
+-- Flag schedules auto-created from a pipeline script's `// schedule "<cron>"`
 -- annotation so reconciliation can update / drop them on subsequent deploys
-- without touching schedules a user created manually. Null for all
-- pre-existing rows.
-ALTER TABLE schedule ADD COLUMN IF NOT EXISTS managed_by_runnable_path VARCHAR(255) DEFAULT NULL;
+-- without touching schedules a user created manually. Defaults to false for
+-- pre-existing rows. `script_path` already tells us which script owns the
+-- row — this is just a boolean discriminator.
+ALTER TABLE schedule ADD COLUMN IF NOT EXISTS managed BOOLEAN NOT NULL DEFAULT false;

-- One managed schedule per (workspace, runnable). Index supports the
-- reconciliation lookup ("does this script already have a managed schedule?")
-- and the cleanup-on-delete query.
-CREATE INDEX IF NOT EXISTS idx_schedule_managed_by_runnable_path
-    ON schedule (workspace_id, managed_by_runnable_path)
-    WHERE managed_by_runnable_path IS NOT NULL;
+-- Partial index for the two hot reconciliation queries:
+--   * "does this script already have a managed schedule?"   (script_path lookup)
+--   * "drop any managed schedules for this deleted script"  (same lookup)
+-- The boolean predicate keeps the index narrow (only managed rows are stored).
+CREATE INDEX IF NOT EXISTS idx_schedule_managed
+    ON schedule (workspace_id, script_path)
+    WHERE managed;
@@ -115,35 +115,20 @@ pub enum TriggerSpec {
    Schedule {
        cron: String,
    },
-    // `// on <kind> <path>` style. The `path` is a workspace-relative
-    // reference to a trigger row already configured in the corresponding
-    // trigger table (http_trigger, email_trigger, kafka_trigger, …). Keeps
-    // the annotation terse; auth/broker/topic details live in the trigger's
-    // own UI.
-    Webhook {
-        path: String,
-    },
-    Email {
-        path: String,
-    },
-    Kafka {
-        path: String,
-    },
-    Mqtt {
-        path: String,
-    },
-    Nats {
-        path: String,
-    },
-    Postgres {
-        path: String,
-    },
-    Sqs {
-        path: String,
-    },
-    Gcp {
-        path: String,
-    },
+    // `// on <kind>` — marker-only declaration that this script wants to be
+    // triggered by a native trigger of the given kind. No path: the binding
+    // is the trigger row's own `script_path` field (set when the user creates
+    // the kafka/mqtt/… trigger in its dedicated UI). The graph endpoint
+    // discovers attached triggers by `WHERE script_path = <this script>` and
+    // surfaces a "missing" placeholder when an annotation has no matching row.
+    Webhook,
+    Email,
+    Kafka,
+    Mqtt,
+    Nats,
+    Postgres,
+    Sqs,
+    Gcp,
 }

 impl TriggerSpec {
@@ -620,33 +605,38 @@ fn parse_partitioned_spec(s: &str) -> Option<PartitionSpec> {
 //     webhook | email | kafka | mqtt | nats | postgres | sqs | gcp
 //   <asset-path-with-prefix>        (e.g. s3://bucket/key, $res:f/foo)
 //
+// Native trigger keywords are *marker-only* — no trailing path. The actual
+// binding lives on the native trigger row (`script_path` column). Anything
+// trailing the keyword is rejected so the form stays unambiguous.
+//
 // Note: `on schedule "..."` is no longer accepted — schedule moved to a
 // top-level `// schedule "..."` annotation. The `Schedule` TriggerSpec
 // variant is still produced, just from a different keyword.
 fn parse_trigger_spec(s: &str) -> Option<TriggerSpec> {
-    // `<kind> <path>` — delegate to a tiny table so the annotation set
-    // stays in lockstep with `TriggerSpec`.
-    type Ctor = fn(String) -> TriggerSpec;
-    const KINDS: &[(&str, Ctor)] = &[
-        ("webhook", |p| TriggerSpec::Webhook { path: p }),
-        ("email", |p| TriggerSpec::Email { path: p }),
-        ("kafka", |p| TriggerSpec::Kafka { path: p }),
-        ("mqtt", |p| TriggerSpec::Mqtt { path: p }),
-        ("nats", |p| TriggerSpec::Nats { path: p }),
-        ("postgres", |p| TriggerSpec::Postgres { path: p }),
-        ("sqs", |p| TriggerSpec::Sqs { path: p }),
-        ("gcp", |p| TriggerSpec::Gcp { path: p }),
+    // Marker-only native trigger keywords. The match table keeps the
+    // annotation set in lockstep with `TriggerSpec`.
+    const NATIVE_KINDS: &[(&str, TriggerSpec)] = &[
+        ("webhook", TriggerSpec::Webhook),
+        ("email", TriggerSpec::Email),
+        ("kafka", TriggerSpec::Kafka),
+        ("mqtt", TriggerSpec::Mqtt),
+        ("nats", TriggerSpec::Nats),
+        ("postgres", TriggerSpec::Postgres),
+        ("sqs", TriggerSpec::Sqs),
+        ("gcp", TriggerSpec::Gcp),
    ];
-    for (kw, ctor) in KINDS {
+    for (kw, spec) in NATIVE_KINDS {
        if let Some(rest) = s.strip_prefix(kw) {
-            if !rest.starts_with(|c: char| c.is_whitespace()) {
+            // Must be a complete word — `kafkalike` doesn't match `kafka`.
+            // Trailing whitespace alone is fine; any non-empty trailing
+            // content is treated as malformed (the annotation is marker-only).
+            if !rest.is_empty() && !rest.starts_with(|c: char| c.is_whitespace()) {
                continue;
            }
-            let path = rest.trim();
-            if path.is_empty() {
+            if !rest.trim().is_empty() {
                return None;
            }
-            return Some(ctor(path.to_string()));
+            return Some(spec.clone());
        }
    }

@@ -845,12 +835,43 @@ mod pipeline_annotation_tests {
    }

    #[test]
-    fn on_kv_split_preserves_non_asset_and_spaced_refs() {
-        // `<kind> <path>` ref with a trailing opt still parses; the opt is
-        // simply not carried for non-asset triggers.
-        let out = parse_pipeline_annotations("// on webhook f/foo debounce=30s");
+    fn native_trigger_keywords_are_marker_only() {
+        // Marker form: `// on kafka` parses to the unit variant.
+        let out = parse_pipeline_annotations("// on kafka");
        assert_eq!(out.triggers.len(), 1);
-        assert!(matches!(out.triggers[0], TriggerSpec::Webhook { .. }));
+        assert!(matches!(out.triggers[0], TriggerSpec::Kafka));
+
+        // Old path-bearing form is rejected (no path on native markers).
+        let out = parse_pipeline_annotations("// on webhook f/foo");
+        assert!(out.triggers.is_empty());
+
+        // Trailing key=value opts are silently dropped by the line-level
+        // KV splitter before parse_trigger_spec sees them — same behaviour
+        // for both asset and native kinds. The opts have no meaning for a
+        // marker, but the marker still parses.
+        let out = parse_pipeline_annotations("// on mqtt debounce=30s");
+        assert_eq!(out.triggers.len(), 1);
+        assert!(matches!(out.triggers[0], TriggerSpec::Mqtt));
+
+        // `kafkalike` mustn't match `kafka`.
+        let out = parse_pipeline_annotations("// on kafkalike");
+        assert!(out.triggers.is_empty());
+    }
+
+    #[test]
+    fn all_native_marker_keywords_parse() {
+        let code = "// on webhook\n// on email\n// on kafka\n// on mqtt\n\
+                    // on nats\n// on postgres\n// on sqs\n// on gcp";
+        let out = parse_pipeline_annotations(code);
+        assert_eq!(out.triggers.len(), 8);
+        assert!(matches!(out.triggers[0], TriggerSpec::Webhook));
+        assert!(matches!(out.triggers[1], TriggerSpec::Email));
+        assert!(matches!(out.triggers[2], TriggerSpec::Kafka));
+        assert!(matches!(out.triggers[3], TriggerSpec::Mqtt));
+        assert!(matches!(out.triggers[4], TriggerSpec::Nats));
+        assert!(matches!(out.triggers[5], TriggerSpec::Postgres));
+        assert!(matches!(out.triggers[6], TriggerSpec::Sqs));
+        assert!(matches!(out.triggers[7], TriggerSpec::Gcp));
    }

    #[test]
@@ -407,9 +407,14 @@ struct GraphEdge {
 }

 // Declared `// on <trigger>` trigger edge — the actual execution DAG.
-// For the eight non-native, non-schedule trigger kinds the variant carries
-// just the trigger's workspace path; the config (broker, topic, auth, …)
-// lives in its own trigger table and UI.
+// Asset / Schedule come from `script_trigger`; the seven native variants
+// (Email/Kafka/…/Gcp) come from the per-kind trigger tables joined on
+// `script_path`. Each native variant carries just the trigger row's path;
+// the config (broker, topic, auth, …) lives in its own UI.
+//
+// `webhook` is parsed as an annotation marker but has no dedicated trigger
+// table — every script gets an implicit webhook endpoint — so no variant
+// here. The frontend renders the marker from the source annotations alone.
 #[derive(Serialize, Debug)]
 #[serde(tag = "trigger_kind", rename_all = "lowercase")]
 enum TriggerEdge {
@@ -424,11 +429,6 @@ enum TriggerEdge {
        runnable_kind: AssetUsageKind,
        runnable_path: String,
    },
-    Webhook {
-        path: String,
-        runnable_kind: AssetUsageKind,
-        runnable_path: String,
-    },
    Email {
        path: String,
        runnable_kind: AssetUsageKind,
@@ -518,9 +518,11 @@ async fn asset_graph(
    .fetch_all(&mut *tx)
    .await?;

-    // Pipeline triggers attached to scripts in scope. Fetched separately so
-    // we can widen the runnable_set for trigger-only endpoints (e.g. an
-    // asset trigger whose asset has no usage in the pipeline yet).
+    // Pipeline asset / schedule trigger edges, fetched separately so we can
+    // widen the runnable_set for trigger-only endpoints (e.g. an asset
+    // trigger whose asset has no usage in the pipeline yet). Native trigger
+    // kinds (kafka, mqtt, …) are *not* in `script_trigger` anymore — they're
+    // discovered below by querying each native trigger table directly.
    let trigger_rows = sqlx::query!(
        r#"
        SELECT
@@ -530,6 +532,7 @@ async fn asset_graph(
            trigger_ref   AS "trigger_ref!"
        FROM script_trigger
        WHERE workspace_id = $1
+          AND trigger_kind IN ('asset', 'schedule')
          AND ($2::text IS NULL OR runnable_path LIKE $2)
        "#,
        &w_id,
@@ -538,6 +541,43 @@ async fn asset_graph(
    .fetch_all(&mut *tx)
    .await?;

+    // Native triggers in scope. Each native trigger table stores its
+    // single-destination `script_path` directly, so we resolve attachment by
+    // joining on that field rather than via `script_trigger`. UNION ALL keeps
+    // it a single round trip; the `kind` column drives the TriggerEdge ctor
+    // below.
+    let native_trigger_rows = sqlx::query!(
+        r#"
+        SELECT kind, path, script_path, is_flow FROM (
+            SELECT 'email' AS kind, path, script_path, is_flow FROM email_trigger
+                WHERE workspace_id = $1
+            UNION ALL
+            SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger
+                WHERE workspace_id = $1
+            UNION ALL
+            SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger
+                WHERE workspace_id = $1
+            UNION ALL
+            SELECT 'nats', path, script_path, is_flow FROM nats_trigger
+                WHERE workspace_id = $1
+            UNION ALL
+            SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger
+                WHERE workspace_id = $1
+            UNION ALL
+            SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger
+                WHERE workspace_id = $1
+            UNION ALL
+            SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger
+                WHERE workspace_id = $1
+        ) t
+        WHERE ($2::text IS NULL OR script_path LIKE $2)
+        "#,
+        &w_id,
+        folder_filter.as_deref(),
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
    // Which scripts in scope are pipeline members (have `// pipeline`).
    let pipeline_member_paths = sqlx::query!(
        r#"
@@ -584,7 +624,8 @@ async fn asset_graph(
        });
    }

-    let mut triggers: Vec<TriggerEdge> = Vec::with_capacity(trigger_rows.len());
+    let mut triggers: Vec<TriggerEdge> =
+        Vec::with_capacity(trigger_rows.len() + native_trigger_rows.len());
    for t in trigger_rows {
        runnable_set.insert((t.runnable_kind, t.runnable_path.clone()));
        match t.trigger_kind.as_str() {
@@ -610,53 +651,36 @@ async fn asset_graph(
                    runnable_path: t.runnable_path,
                });
            }
-            // One-liners for the `<kind> <path>` trigger variants. Kept as a
-            // flat match rather than a helper — each arm's variant ctor is
-            // different and we don't benefit from abstracting it.
-            "webhook" => triggers.push(TriggerEdge::Webhook {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "email" => triggers.push(TriggerEdge::Email {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "kafka" => triggers.push(TriggerEdge::Kafka {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "mqtt" => triggers.push(TriggerEdge::Mqtt {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "nats" => triggers.push(TriggerEdge::Nats {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "postgres" => triggers.push(TriggerEdge::Postgres {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "sqs" => triggers.push(TriggerEdge::Sqs {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            "gcp" => triggers.push(TriggerEdge::Gcp {
-                path: t.trigger_ref,
-                runnable_kind: t.runnable_kind,
-                runnable_path: t.runnable_path,
-            }),
-            _ => {} // Unknown trigger_kind — forward-compat.
+            _ => {} // Native kinds come from per-kind trigger tables below.
        }
    }

+    // Native trigger attachments — one TriggerEdge per row, the kind chosen
+    // from the discriminator. Add the runnable to the set so a script with
+    // no asset edges but a kafka attachment still renders on the canvas.
+    for t in native_trigger_rows {
+        let kind = t.kind.unwrap_or_default();
+        let path = t.path.unwrap_or_default();
+        let script_path = t.script_path.unwrap_or_default();
+        let runnable_kind = if t.is_flow.unwrap_or(false) {
+            AssetUsageKind::Flow
+        } else {
+            AssetUsageKind::Script
+        };
+        runnable_set.insert((runnable_kind, script_path.clone()));
+        let edge = match kind.as_str() {
+            "email" => TriggerEdge::Email { path, runnable_kind, runnable_path: script_path },
+            "kafka" => TriggerEdge::Kafka { path, runnable_kind, runnable_path: script_path },
+            "mqtt" => TriggerEdge::Mqtt { path, runnable_kind, runnable_path: script_path },
+            "nats" => TriggerEdge::Nats { path, runnable_kind, runnable_path: script_path },
+            "postgres" => TriggerEdge::Postgres { path, runnable_kind, runnable_path: script_path },
+            "sqs" => TriggerEdge::Sqs { path, runnable_kind, runnable_path: script_path },
+            "gcp" => TriggerEdge::Gcp { path, runnable_kind, runnable_path: script_path },
+            _ => continue,
+        };
+        triggers.push(edge);
+    }
+
    let mut assets: Vec<GraphAssetNode> = asset_set
        .into_iter()
        .map(|(kind, path)| GraphAssetNode { kind, path })
@@ -1566,10 +1566,15 @@ async fn create_script_internal<'c>(
    }

    // Pipeline trigger edges: wipe-and-reinsert per deploy so removing an
-    // `// on ...` annotation drops the edge.
+    // `// on ...` annotation drops the edge. Only Asset / Schedule produce
+    // a row — native trigger marker annotations (`// on kafka`, etc.) are
+    // discovered by the graph endpoint directly from the per-kind trigger
+    // tables, so `trigger_spec_to_row` returns None for those.
    clear_script_triggers(&mut *tx, &w_id, &ns.path, AssetUsageKind::Script).await?;
    for spec in &pipeline_triggers {
-        let (trigger_kind, trigger_ref) = trigger_spec_to_row(spec);
+        let Some((trigger_kind, trigger_ref)) = trigger_spec_to_row(spec) else {
+            continue;
+        };
        // Effective debounce for this edge: per-`// on debounce=` wins,
        // else the script-level `// debounce` default. Debounce only
        // applies to asset-cascade edges; other trigger kinds get none.
@@ -107,7 +107,7 @@ pub async fn clear_script_triggers<'e>(
 // `// schedule "<cron>"` annotation. Idempotent: each call brings the
 // `schedule` row in line with the annotation as of *this* deploy.
 //
-// The schedule lives at the same path as the runnable. `managed_by_runnable_path`
+// The schedule lives at the same path as the runnable. The `managed` flag
 // disambiguates auto-created rows from user-managed ones — only managed
 // rows are updated or removed by reconciliation; manually-created schedules
 // at the same path are left alone (the annotation is silently ignored).
@@ -135,16 +135,16 @@ pub async fn reconcile_pipeline_schedule<'e>(
                    workspace_id, path, schedule, timezone, edited_by, script_path,
                    is_flow, enabled, email, permissioned_as,
                    ws_error_handler_muted, no_flow_overlap, cron_version,
-                    managed_by_runnable_path
+                    managed
                )
-                VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', $2)
+                VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', true)
                ON CONFLICT (workspace_id, path) DO UPDATE
                SET schedule = EXCLUDED.schedule,
                    edited_at = now(),
                    edited_by = EXCLUDED.edited_by,
-                    managed_by_runnable_path = EXCLUDED.managed_by_runnable_path
-                WHERE schedule.managed_by_runnable_path = EXCLUDED.managed_by_runnable_path
-                   OR schedule.managed_by_runnable_path IS NULL AND schedule.script_path = EXCLUDED.script_path
+                    managed = true
+                WHERE schedule.managed
+                   OR schedule.script_path = EXCLUDED.script_path
                "#,
                workspace_id,
                runnable_path,
@@ -159,12 +159,13 @@ pub async fn reconcile_pipeline_schedule<'e>(
        }
        None => {
            // Drop any prior managed schedule for this runnable. Manual
-            // schedules at the same path keep `managed_by_runnable_path =
-            // NULL` and are unaffected.
+            // schedules at the same path keep `managed = false` and are
+            // unaffected.
            sqlx::query!(
                r#"DELETE FROM schedule
                   WHERE workspace_id = $1
-                     AND managed_by_runnable_path = $2"#,
+                     AND script_path = $2
+                     AND managed"#,
                workspace_id,
                runnable_path,
            )
@@ -186,7 +187,8 @@ pub async fn delete_managed_pipeline_schedule<'e>(
    sqlx::query!(
        r#"DELETE FROM schedule
           WHERE workspace_id = $1
-             AND managed_by_runnable_path = $2"#,
+             AND script_path = $2
+             AND managed"#,
        workspace_id,
        runnable_path,
    )
@@ -289,7 +291,12 @@ pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, String)> {
 // Convert a parser TriggerSpec into the `(kind, ref)` pair stored in
 // script_trigger. Asset refs get their canonical prefix back so the
 // trigger_ref matches what downstream lookups expect.
-pub fn trigger_spec_to_row(spec: &TriggerSpec) -> (ScriptTriggerKind, String) {
+//
+// Returns `None` for native trigger kinds (Kafka, Mqtt, Postgres, …) —
+// those annotations are marker-only and don't produce a `script_trigger`
+// row. The actual binding lives on the trigger row's own `script_path`
+// column; the graph endpoint looks it up directly per kind.
+pub fn trigger_spec_to_row(spec: &TriggerSpec) -> Option<(ScriptTriggerKind, String)> {
    match spec {
        TriggerSpec::Asset { asset_kind, path, .. } => {
            let prefix = match asset_kind {
@@ -299,17 +306,17 @@ pub fn trigger_spec_to_row(spec: &TriggerSpec) -> (ScriptTriggerKind, String) {
                windmill_parser::asset_parser::AssetKind::DataTable => "datatable://",
                windmill_parser::asset_parser::AssetKind::Volume => "volume://",
            };
-            (ScriptTriggerKind::Asset, format!("{}{}", prefix, path))
+            Some((ScriptTriggerKind::Asset, format!("{}{}", prefix, path)))
        }
-        TriggerSpec::Schedule { cron } => (ScriptTriggerKind::Schedule, cron.clone()),
-        TriggerSpec::Webhook { path } => (ScriptTriggerKind::Webhook, path.clone()),
-        TriggerSpec::Email { path } => (ScriptTriggerKind::Email, path.clone()),
-        TriggerSpec::Kafka { path } => (ScriptTriggerKind::Kafka, path.clone()),
-        TriggerSpec::Mqtt { path } => (ScriptTriggerKind::Mqtt, path.clone()),
-        TriggerSpec::Nats { path } => (ScriptTriggerKind::Nats, path.clone()),
-        TriggerSpec::Postgres { path } => (ScriptTriggerKind::Postgres, path.clone()),
-        TriggerSpec::Sqs { path } => (ScriptTriggerKind::Sqs, path.clone()),
-        TriggerSpec::Gcp { path } => (ScriptTriggerKind::Gcp, path.clone()),
+        TriggerSpec::Schedule { cron } => Some((ScriptTriggerKind::Schedule, cron.clone())),
+        TriggerSpec::Webhook
+        | TriggerSpec::Email
+        | TriggerSpec::Kafka
+        | TriggerSpec::Mqtt
+        | TriggerSpec::Nats
+        | TriggerSpec::Postgres
+        | TriggerSpec::Sqs
+        | TriggerSpec::Gcp => None,
    }
 }

@@ -15,7 +15,7 @@
 	import AddNode from './AddNode.svelte'
 	import AssetGraphEdge from './AssetGraphEdge.svelte'
 	import { layoutAssetGraph } from './assetGraphLayout'
-	import type { AssetGraphResponse, AssetGraphSelection } from './types'
+	import type { AssetGraphResponse, AssetGraphSelection, NativeTriggerKind } from './types'
 	import type { RunnableRunState } from './activeRunnables.svelte'
 	import type { AssetKind } from '$lib/gen'
 	import { NODE } from '$lib/components/graph/util'
@@ -51,10 +51,7 @@
 			path: string,
 			source:
 				| { kind: 'schedule'; cron: string }
-				| {
-						kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
-						path: string | undefined
-				  },
+				| { kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp' },
 			outputKind: import('./pipelineTemplates').PipelineOutputKind,
 			aiPrompt?: string
 		) => void
@@ -105,6 +102,11 @@
 		// a small badge on each runnable node. Same poll source as
 		// `activeRunnableIds`; persists the last status while idle.
 		runStates?: ReadonlyMap<string, RunnableRunState>
+		// Click handler for a "missing trigger" placeholder. The page wires
+		// this to its native trigger drawer set so clicking the red node
+		// opens the matching editor with `script_path` pre-filled — no
+		// navigation, drafts stay intact.
+		onCreateMissingTrigger?: (kind: NativeTriggerKind, scriptPath: string) => void
 	}
 	let {
 		graph,
@@ -119,7 +121,8 @@
 		onRunnableMenuRemove,
 		activeRunnable,
 		activeRunnableIds,
-		runStates
+		runStates,
+		onCreateMissingTrigger
 	}: Props = $props()

 	const ADD_NODE_ID = '__add__'
@@ -140,6 +143,10 @@
 			| 'trigger-native'
 			| 'add-anchor'
 		unsaved?: boolean
+		// Edge from a missing-trigger placeholder — styled red dashed to
+		// signal "this script declared `// on kafka` but no trigger row
+		// targets it; create one or remove the annotation".
+		missing?: boolean
 	}

 	// Graph-id of the script the user just launched (zero-latency hint),
@@ -315,18 +322,33 @@
 		}

 		// Non-asset triggers (schedule + native) are rendered as source nodes
-		// above the pipeline script. Nodes are deduplicated per (kind, ref)
-		// tuple so a single schedule/webhook shared across multiple scripts
-		// shows as one node with N outgoing edges. A trigger node is
-		// considered unsaved if every attachment referencing it is unsaved.
+		// above the pipeline script. Real (non-missing) nodes are
+		// deduplicated per (kind, ref) tuple so a single schedule shared
+		// across multiple scripts shows as one node with N outgoing edges.
+		// "missing" placeholders are scoped per-(kind, script) — each script
+		// gets its own placeholder so the prompt "create / delete" tells
+		// the user which script the annotation lives on.
 		const triggerSourceNodes = new Map<
 			string,
-			{ allUnsaved: boolean; kind: TriggerNodeKind; ref: string }
+			{
+				allUnsaved: boolean
+				kind: TriggerNodeKind
+				ref: string
+				missing: boolean
+				runnable_path?: string
+			}
 		>()
-		function recordSourceTrigger(id: string, kind: TriggerNodeKind, ref: string, unsaved: boolean) {
+		function recordSourceTrigger(
+			id: string,
+			kind: TriggerNodeKind,
+			ref: string,
+			unsaved: boolean,
+			missing: boolean,
+			runnable_path?: string
+		) {
 			const prev = triggerSourceNodes.get(id)
 			if (!prev) {
-				triggerSourceNodes.set(id, { allUnsaved: unsaved, kind, ref })
+				triggerSourceNodes.set(id, { allUnsaved: unsaved, kind, ref, missing, runnable_path })
 			} else {
 				prev.allUnsaved = prev.allUnsaved && unsaved
 			}
@@ -345,22 +367,45 @@
 				})
 				continue
 			}
-			const ref = t.trigger_kind === 'schedule' ? (t as any).cron : (t as any).path
+			const isMissing = t.trigger_kind !== 'schedule' && (t as any).missing === true
+			// Schedule: cron is the ref. Native (attached): trigger row path.
+			// Native (missing): synthesize a per-script ref so each placeholder
+			// is its own node ("missing kafka on f/foo/bar").
+			const ref = isMissing
+				? `missing:${t.runnable_path}`
+				: t.trigger_kind === 'schedule'
+					? (t as any).cron
+					: ((t as any).path ?? '')
 			const sourceId = `trigger:${t.trigger_kind}:${ref}`
-			recordSourceTrigger(sourceId, t.trigger_kind, ref, !!t.unsaved)
+			recordSourceTrigger(
+				sourceId,
+				t.trigger_kind,
+				ref,
+				!!t.unsaved,
+				isMissing,
+				isMissing ? t.runnable_path : undefined
+			)
 			edges.push({
 				id: `trig-${t.trigger_kind}:${sourceId}->${runnableId}`,
 				source: sourceId,
 				target: runnableId,
 				kind: t.trigger_kind === 'schedule' ? 'trigger-schedule' : 'trigger-native',
-				unsaved: t.unsaved
+				unsaved: t.unsaved,
+				missing: isMissing
 			})
 		}
 		for (const [id, info] of triggerSourceNodes) {
 			nodes.push({
 				id,
 				type: 'trigger',
-				data: { kind: info.kind, ref: info.ref, unsaved: info.allUnsaved }
+				data: {
+					kind: info.kind,
+					ref: info.ref,
+					unsaved: info.allUnsaved,
+					missing: info.missing,
+					runnable_path: info.runnable_path,
+					onCreateMissingTrigger
+				}
 			})
 		}

@@ -517,6 +562,18 @@
 					style = `${style} opacity: 0.7;`
 					if (label) label = `${label} (unsaved)`
 				}
+				// Missing-trigger edge: overrides the per-kind stroke colour
+				// with red so the entire "annotated but no row" branch reads
+				// as broken at a glance. Composes with `unsaved` if both
+				// (red dashed dimmed — fresh draft annotation that also has
+				// no matching row, which is the common case).
+				if (e.missing) {
+					style = 'stroke: rgb(239 68 68); stroke-width: 2px;'
+					strokeDasharray = '3 3'
+					markerColor = 'rgb(239 68 68)'
+					label = 'missing trigger'
+					labelStyle = 'fill: rgb(239 68 68); font-size: 10px; font-weight: 600;'
+				}
 				if (strokeDasharray) {
 					style = `${style} stroke-dasharray: ${strokeDasharray};`
 				}
@@ -110,36 +110,117 @@
 	import { Handle, Position } from '@xyflow/svelte'
 	import { NODE } from '$lib/components/graph/util'
 	import { twMerge } from 'tailwind-merge'
+	import { AlertTriangle } from 'lucide-svelte'

 	interface Props {
 		// `ref` is the cron expression for schedules, the trigger-path for
-		// every other kind. Rendered verbatim — no formatting per kind.
-		data: { kind: TriggerNodeKind; ref: string; unsaved?: boolean }
+		// attached native triggers, and a synthetic `missing:<script>` for
+		// placeholders. `missing: true` swaps the styling to a red broken
+		// state and surfaces "no trigger row" instead of a path; the
+		// owning script is in `runnable_path` (used by the title and the
+		// "+ Create trigger" drawer hook passed by the page).
+		data: {
+			kind: TriggerNodeKind
+			ref: string
+			unsaved?: boolean
+			missing?: boolean
+			runnable_path?: string
+			// Page-supplied dispatcher that opens the matching native
+			// trigger drawer with `script_path` pre-filled. When absent
+			// (e.g. webhook, schedule, or a kind without an editor) the
+			// placeholder is non-clickable.
+			onCreateMissingTrigger?: (kind: NativeTriggerKind, scriptPath: string) => void
+		}
 	}
 	let { data }: Props = $props()

 	let style = $derived(TRIGGER_NODE_STYLE[data.kind])
-	let Icon = $derived(style.icon)
+	let Icon = $derived(data.missing ? AlertTriangle : style.icon)
+	let missingTitle = $derived(
+		data.missing
+			? `Missing ${style.label} trigger: ${data.runnable_path ?? ''} declares \`// on ${style.label}\` but no ${style.label} trigger targets it. Click to create one, or remove the annotation.`
+			: undefined
+	)
+	// Schedule and webhook have no dedicated drawer (schedules are
+	// inline-managed; webhooks are implicit endpoints), so the placeholder
+	// is not clickable for those kinds.
+	let canCreate = $derived(
+		data.missing &&
+			data.kind !== 'schedule' &&
+			data.kind !== 'webhook' &&
+			!!data.runnable_path &&
+			!!data.onCreateMissingTrigger
+	)
+
+	function handleMissingClick() {
+		if (!canCreate || !data.runnable_path || !data.onCreateMissingTrigger) return
+		data.onCreateMissingTrigger(data.kind as NativeTriggerKind, data.runnable_path)
+	}
 </script>

 <div class="relative">
-	<div
-		class={twMerge(
-			'flex items-center rounded-md drop-shadow-sm overflow-hidden outline outline-1',
-			style.bg,
-			data.unsaved ? `opacity-80 ${style.borderUnsaved}` : style.border
-		)}
-		style="width: {NODE.width}px; min-height: {NODE.height}px;"
-		title={data.unsaved ? `Unsaved ${style.label}: ${data.ref}` : `${style.label}: ${data.ref}`}
-	>
-		<Icon size={14} class={`shrink-0 ml-2 mr-2 ${style.iconText}`} />
-		<div class="flex flex-col min-w-0 flex-1 pr-2 py-0.5 leading-tight">
-			<span class="text-3xs uppercase tracking-wide text-tertiary truncate">
-				{style.label}{data.unsaved ? ' · unsaved' : ''}
-			</span>
-			<span class="text-2xs font-mono text-emphasis truncate">{data.ref}</span>
+	{#if canCreate}
+		<!-- Whole missing placeholder is a button so the cursor + a11y
+		     affordance reads "clickable to fix". Sits in the same visual
+		     box as the non-clickable variants. -->
+		<button
+			type="button"
+			onclick={handleMissingClick}
+			class={twMerge(
+				'flex items-center rounded-md drop-shadow-sm overflow-hidden outline outline-1 w-full text-left',
+				'bg-red-50 dark:bg-red-900/30 outline-dashed outline-red-400 dark:outline-red-500',
+				'hover:bg-red-100 dark:hover:bg-red-900/40 transition-colors'
+			)}
+			style="width: {NODE.width}px; min-height: {NODE.height}px;"
+			title={missingTitle}
+		>
+			<Icon size={14} class="shrink-0 ml-2 mr-2 text-red-600 dark:text-red-400" />
+			<div class="flex flex-col min-w-0 flex-1 pr-2 py-0.5 leading-tight">
+				<span class="text-3xs uppercase tracking-wide truncate text-red-700 dark:text-red-400">
+					{style.label} · missing
+				</span>
+				<span class="text-2xs font-mono truncate text-red-700 dark:text-red-400">
+					Click to create
+				</span>
+			</div>
+		</button>
+	{:else}
+		<div
+			class={twMerge(
+				'flex items-center rounded-md drop-shadow-sm overflow-hidden outline outline-1',
+				data.missing
+					? 'bg-red-50 dark:bg-red-900/30 outline-dashed outline-red-400 dark:outline-red-500'
+					: style.bg,
+				data.missing ? '' : data.unsaved ? `opacity-80 ${style.borderUnsaved}` : style.border
+			)}
+			style="width: {NODE.width}px; min-height: {NODE.height}px;"
+			title={missingTitle ??
+				(data.unsaved ? `Unsaved ${style.label}: ${data.ref}` : `${style.label}: ${data.ref}`)}
+		>
+			<Icon
+				size={14}
+				class={`shrink-0 ml-2 mr-2 ${data.missing ? 'text-red-600 dark:text-red-400' : style.iconText}`}
+			/>
+			<div class="flex flex-col min-w-0 flex-1 pr-2 py-0.5 leading-tight">
+				<span
+					class={twMerge(
+						'text-3xs uppercase tracking-wide truncate',
+						data.missing ? 'text-red-700 dark:text-red-400' : 'text-tertiary'
+					)}
+				>
+					{style.label}{data.missing ? ' · missing' : data.unsaved ? ' · unsaved' : ''}
+				</span>
+				<span
+					class={twMerge(
+						'text-2xs font-mono truncate',
+						data.missing ? 'text-red-700 dark:text-red-400' : 'text-emphasis'
+					)}
+				>
+					{data.missing ? 'no trigger row' : data.ref}
+				</span>
+			</div>
 		</div>
-	</div>
+	{/if}
 </div>

 <Handle type="source" position={Position.Bottom} isConnectable={false} />
@@ -40,7 +40,10 @@ const NATIVE_TRIGGER_KEYWORDS: NativeTriggerKind[] = [
 export const PARTITION_TOKEN = '{partition}'

 export type PipelineTriggerAsset = { kind: AssetKind; path: string }
-export type PipelineNativeTrigger = { kind: NativeTriggerKind; path: string }
+// Marker-only — native trigger annotations carry no path. The binding lives
+// on the trigger row's own `script_path` field; the graph endpoint resolves
+// it by querying the per-kind trigger tables.
+export type PipelineNativeTrigger = { kind: NativeTriggerKind }

 export type PartitionKind =
 	| { kind: 'daily' }
@@ -130,15 +133,22 @@ type ParsedTriggerSpec =

 // Parse a single `on <spec>` right-hand side. The top-level `// schedule`
 // is handled separately at the line level (not via `on`).
+//
+// Native trigger keywords (kafka, mqtt, …) are marker-only — `// on kafka`
+// without a trailing path. Trailing content makes the line malformed and
+// is rejected. Asset triggers always carry an `<asset-prefix><path>` ref.
 function parseTriggerSpec(s: string): ParsedTriggerSpec | undefined {
 	for (const kw of NATIVE_TRIGGER_KEYWORDS) {
 		if (s.startsWith(kw)) {
 			const after = s.slice(kw.length)
-			// Require whitespace so `kafkalike` doesn't match `kafka`.
-			if (after.length === 0 || !/\s/.test(after[0])) continue
-			const path = after.trim()
-			if (!path) return undefined
-			return { kind: 'native', value: { kind: kw, path } }
+			// `kafka` must end the line (modulo whitespace) — `kafkalike`
+			// is not `kafka`. Anything trailing makes it malformed.
+			if (after.length === 0) {
+				return { kind: 'native', value: { kind: kw } }
+			}
+			if (!/\s/.test(after[0])) continue
+			if (after.trim().length > 0) return undefined
+			return { kind: 'native', value: { kind: kw } }
 		}
 	}
 	const asset = parseAssetSyntax(s)
@@ -262,9 +272,7 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
 					out.triggerAssets.push(spec.value)
 				}
 			} else {
-				if (
-					!out.nativeTriggers.some((n) => n.kind === spec.value.kind && n.path === spec.value.path)
-				) {
+				if (!out.nativeTriggers.some((n) => n.kind === spec.value.kind)) {
 					out.nativeTriggers.push(spec.value)
 				}
 			}
@@ -234,10 +234,11 @@ function commentPrefix(lang: ScriptLang): string {
 export type DraftTriggerSource =
 	| { kind: 'schedule'; cron: string }
 	| { kind: 'asset'; ref: string }
-	| {
-			kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
-			path: string | undefined
-	  }
+	// Native trigger marker — the annotation is path-less (`// on kafka`).
+	// The binding lives on the trigger row's own `script_path`; the user
+	// creates that row via the drawer the canvas opens on the missing-
+	// trigger placeholder.
+	| { kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp' }

 export type TemplateContext = {
 	language: ScriptLang
@@ -262,7 +263,8 @@ function header(language: ScriptLang, triggers: DraftTriggerSource[]): string {
 			case 'asset':
 				return `${p} on ${t.ref}`
 			default:
-				return `${p} on ${t.kind} ${t.path ?? '<trigger-path>'}`
+				// Native triggers: marker-only — no trailing path.
+				return `${p} on ${t.kind}`
 		}
 	})
 	// Discoverability hint — the three annotations users most often miss
@@ -27,6 +27,7 @@ const input = (over: Partial<ResolveGraphInput> = {}): ResolveGraphInput => ({
 	liveAnnotations: { scriptPath: undefined, annotations: ann() },
 	inferredWritesByPath: new Map(),
 	inferredReadsByPath: new Map(),
+	annotatedNativeKindsByPath: new Map(),
 	...over
 })

@@ -1,4 +1,4 @@
-import type { AssetGraphResponse } from './types'
+import type { AssetGraphResponse, NativeTriggerKind } from './types'
 import { parsePipelineAnnotations, type PipelineAnnotations } from './parsePipelineAnnotations'
 import {
 	extractWrites,
@@ -25,6 +25,13 @@ export type ResolveGraphInput = {
 	/** Sticky session caches of inferred body writes/reads per script path. */
 	inferredWritesByPath: Map<string, Array<{ kind: AssetKind; path: string }>>
 	inferredReadsByPath: Map<string, Array<{ kind: AssetKind; path: string }>>
+	/**
+	 * Sticky cache of native trigger kinds declared via `// on <kind>` in
+	 * each script's deployed source. Filled by the load-time prefetch
+	 * sweep. Used here to emit "missing" placeholders for scripts whose
+	 * annotation has no matching trigger row in `base.triggers`.
+	 */
+	annotatedNativeKindsByPath: Map<string, Set<NativeTriggerKind>>
 }

 /**
@@ -48,7 +55,8 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 		liveBodyAssets,
 		liveAnnotations,
 		inferredWritesByPath,
-		inferredReadsByPath
+		inferredReadsByPath,
+		annotatedNativeKindsByPath
 	} = input

 	// Every draft contributes: a runnable, an output asset, a write edge,
@@ -136,13 +144,17 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 			const hasTriggerAsset = assets.some((x) => x.kind === a.kind && x.path === a.path)
 			if (!hasTriggerAsset) assets.push({ kind: a.kind, path: a.path })
 		}
+		// Native trigger annotations on a draft are always "missing" until
+		// the user creates the matching trigger row — drafts can't carry a
+		// real trigger row since the script isn't deployed yet. Surface a
+		// red placeholder so the user knows to wire it up.
 		for (const n of parsed.nativeTriggers) {
 			extraTriggers.push({
 				trigger_kind: n.kind,
-				path: n.path,
 				runnable_kind: 'script',
 				runnable_path: path,
-				unsaved: true
+				unsaved: true,
+				missing: true
 			})
 		}
 	}
@@ -206,11 +218,13 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 				unsaved: true
 			})
 		}
-		// Persisted native triggers keyed by `<kind>:<path>`, used to
-		// suppress duplicate overlay for already-saved `// on <kind>`
-		// annotations. trigger_kind is narrower than the union so we
-		// cast through string.
-		const persistedNativeKeys = new Set(
+		// 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) =>
@@ -219,17 +233,49 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
 						t.runnable_kind === 'script' &&
 						t.runnable_path === livePath
 				)
-				.map((t) => `${t.trigger_kind}:${(t as { path: string }).path}`)
+				.map((t) => t.trigger_kind)
 		)
 		for (const n of liveAnnotations.annotations.nativeTriggers) {
-			const key = `${n.kind}:${n.path}`
-			if (persistedNativeKeys.has(key)) continue
+			if (persistedNativeKinds.has(n.kind)) continue
 			extraTriggers.push({
 				trigger_kind: n.kind,
-				path: n.path,
 				runnable_kind: 'script',
 				runnable_path: livePath,
-				unsaved: true
+				unsaved: true,
+				missing: 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
+	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.trigger_kind !== 'schedule' &&
+						t.runnable_kind === 'script' &&
+						t.runnable_path === scriptPath
+				)
+				.map((t) => t.trigger_kind)
+		)
+		for (const kind of kinds) {
+			if (attachedKinds.has(kind)) continue
+			extraTriggers.push({
+				trigger_kind: kind,
+				runnable_kind: 'script',
+				runnable_path: scriptPath,
+				missing: true
 			})
 		}
 	}
@@ -70,10 +70,17 @@ export type AssetGraphTrigger =
 	  }
 	| {
 			trigger_kind: NativeTriggerKind
-			path: string
+			// path of the matching trigger row (kafka_trigger.path, etc.).
+			// Undefined when `missing` is true — the script has the
+			// annotation marker but no trigger row points at it.
+			path?: string
 			runnable_kind: GraphUsageKind
 			runnable_path: string
 			unsaved?: boolean
+			// Annotation declared but no matching trigger row was found —
+			// the canvas renders a red placeholder with a "Create trigger"
+			// affordance instead of a fully-wired source.
+			missing?: boolean
 	  }

 export interface AssetGraphResponse {
@@ -16,9 +16,13 @@
 	} from '$lib/components/assets/lib'
 	import type {
 		AssetGraphResponse,
-		AssetGraphSelection
+		AssetGraphSelection,
+		NativeTriggerKind
 	} from '$lib/components/assets/AssetGraph/types'
-	import type { PipelineAnnotations } from '$lib/components/assets/AssetGraph/parsePipelineAnnotations'
+	import {
+		parsePipelineAnnotations,
+		type PipelineAnnotations
+	} from '$lib/components/assets/AssetGraph/parsePipelineAnnotations'
 	import { resolveGraph } from '$lib/components/assets/AssetGraph/resolveGraph'
 	import {
 		generatePipelineDraft,
@@ -56,6 +60,13 @@
 	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'

 	// Variables and resources are declarative config, not pipeline assets —
 	// they're hub-shaped (referenced by most runnables) and would swamp the
@@ -274,6 +285,12 @@
 	// persisted asset rows.
 	let inferredReadsByPath = $state<Map<string, Array<{ kind: AssetKind; path: string }>>>(new Map())

+	// Sticky cache of native trigger kinds declared via `// on kafka` etc.
+	// in each deployed script's source. Filled by the prefetch sweep below.
+	// resolveGraph uses this to flag scripts whose annotation has no
+	// matching trigger row — red placeholder on the canvas.
+	let annotatedNativeKindsByPath = $state<Map<string, Set<NativeTriggerKind>>>(new Map())
+
 	// Build a runnable Script from picked language / triggers / output.
 	// Delegates to the shared template generator (pipelineTemplates.ts) so
 	// the same logic is reachable from anywhere a draft is needed.
@@ -678,7 +695,8 @@
 			liveBodyAssets,
 			liveAnnotations,
 			inferredWritesByPath,
-			inferredReadsByPath
+			inferredReadsByPath,
+			annotatedNativeKindsByPath
 		})
 	)

@@ -799,6 +817,41 @@
 	// 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()
+
+	function openMissingTriggerDrawer(kind: NativeTriggerKind, scriptPath: string) {
+		switch (kind) {
+			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; schedule is inline-managed.
+			default:
+				return
+		}
+	}
+
 	// Reuse the empty AssetGraphResponse so we can still render the canvas
 	// (layout, controls, mini-map) on a fresh pipeline.
 	const EMPTY_GRAPH: AssetGraphResponse = {
@@ -888,7 +941,11 @@
 				.filter((r) => r.usage_kind === 'script')
 				.map((r) => r.path)
 				.filter(
-					(p) => !drafts.has(p) && !inferredWritesByPath.has(p) && !inferredReadsByPath.has(p)
+					(p) =>
+						!drafts.has(p) &&
+						!inferredWritesByPath.has(p) &&
+						!inferredReadsByPath.has(p) &&
+						!annotatedNativeKindsByPath.has(p)
 				)
 		)
 		if (targets.length === 0) return
@@ -900,11 +957,18 @@
 				try {
 					const s = await ScriptService.getScriptByPath({ workspace: ws, path })
 					if (gen !== assetPrefetchGen) return
-					const res = await inferAssets(s.language, s.content ?? '')
+					const content = s.content ?? ''
+					const res = await inferAssets(s.language, content)
 					if (gen !== assetPrefetchGen) return
 					const inferred = (res?.assets ?? []) as AssetWithAltAccessType[]
 					const writes = extractWrites(inferred)
 					const reads = extractReads(inferred)
+					// Parse `// on kafka` markers in parallel with the asset
+					// inference. Cheap pure-TS pass — runs on the same content
+					// we already loaded for inferAssets.
+					const annotated = new Set(
+						parsePipelineAnnotations(content).nativeTriggers.map((n) => n.kind)
+					)
 					untrack(() => {
 						// A live edit / prior sweep may have filled either meanwhile.
 						if (writes.length > 0 && !inferredWritesByPath.has(path)) {
@@ -917,6 +981,13 @@
 							next.set(path, reads)
 							inferredReadsByPath = next
 						}
+						// Always seed the annotation map (even if empty) so a
+						// later removal of all `// on kafka` lines retires the
+						// placeholder on the next deploy + refetch.
+						const nextAnnot = new Map(annotatedNativeKindsByPath)
+						if (annotated.size > 0) nextAnnot.set(path, annotated)
+						else nextAnnot.delete(path)
+						annotatedNativeKindsByPath = nextAnnot
 					})
 				} catch {
 					// Skip — that node just falls back to base-graph edges.
@@ -1085,6 +1156,7 @@
 								{pathPrefix}
 								defaultPathSuffix={DEFAULT_PATH_SUFFIX}
 								defaultScheduleCron={DEFAULT_SCHEDULE_CRON}
+								onCreateMissingTrigger={openMissingTriggerDrawer}
 								onselect={(s) => {
 									// Clicking a draft runnable node re-opens it in the pane;
 									// clicking anything else selects it normally and detaches
@@ -1346,6 +1418,18 @@

 	<PipelinePickerModal bind:open={pickerModalOpen} currentFolder={folder} />

+	<!-- 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. -->
+	<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()} />
+
 	{#if leaveModalOpen}
 		<!-- Three-button leave guard. Built inline rather than reusing
 		     ConfirmationModal because that one is binary (confirm/cancel) and