all

backend/migrations/20260421205023_add_asset_materializer.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS asset_materializer_history;
+DROP TABLE IF EXISTS asset_materializer;

backend/migrations/20260421205023_add_asset_materializer.up.sql

@@ -0,0 +1,33 @@
+-- Current materializer: exactly one runnable owns each (workspace, kind, path) asset.
+-- Updated via last-deploy-wins on `// materialize <path>` annotations.
+CREATE TABLE asset_materializer (
+  workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  asset_kind ASSET_KIND NOT NULL,
+  asset_path VARCHAR(255) NOT NULL,
+  runnable_kind ASSET_USAGE_KIND NOT NULL,
+  runnable_path VARCHAR(255) NOT NULL,
+  deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  deployed_by VARCHAR(50) NOT NULL,
+  PRIMARY KEY (workspace_id, asset_kind, asset_path)
+);
+
+CREATE INDEX idx_asset_materializer_runnable
+  ON asset_materializer (workspace_id, runnable_kind, runnable_path);
+
+-- Append-only history. Every claim (including takeovers) gets a row.
+-- Revocations (redeploy without annotation) are represented by absence in
+-- `asset_materializer` rather than a row here, so "who last materialized X"
+-- is the most recent history row for X regardless of current state.
+CREATE TABLE asset_materializer_history (
+  id BIGSERIAL PRIMARY KEY,
+  workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  asset_kind ASSET_KIND NOT NULL,
+  asset_path VARCHAR(255) NOT NULL,
+  runnable_kind ASSET_USAGE_KIND NOT NULL,
+  runnable_path VARCHAR(255) NOT NULL,
+  deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  deployed_by VARCHAR(50) NOT NULL
+);
+
+CREATE INDEX idx_asset_materializer_history_asset
+  ON asset_materializer_history (workspace_id, asset_kind, asset_path, deployed_at DESC);

backend/migrations/20260421212900_pipeline_folders_index.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS idx_script_materializer_path;

backend/migrations/20260421212900_pipeline_folders_index.up.sql

@@ -0,0 +1,8 @@
+-- Fast lookups for:
+--   1. "does folder F have a pipeline?"  (exists check on prefix)
+--   2. "list all folders with a pipeline" (distinct folder from path)
+-- The partial predicate keeps the index tiny on workspaces with few
+-- materializer scripts, and text_pattern_ops lets 'f/foo/%' LIKE scans use it.
+CREATE INDEX IF NOT EXISTS idx_script_materializer_path
+  ON script (workspace_id, path text_pattern_ops)
+  WHERE auto_kind = 'materializer' AND archived = false AND deleted = false;

backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql

@@ -0,0 +1,30 @@
+DROP TABLE IF EXISTS script_trigger;
+DROP TYPE IF EXISTS SCRIPT_TRIGGER_KIND;
+
+-- Recreate the materializer tables so rolling back to the pre-refactor
+-- backend code boots. Data is gone either way (irrecoverable).
+CREATE TABLE asset_materializer (
+  workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  asset_kind ASSET_KIND NOT NULL,
+  asset_path VARCHAR(255) NOT NULL,
+  runnable_kind ASSET_USAGE_KIND NOT NULL,
+  runnable_path VARCHAR(255) NOT NULL,
+  deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  deployed_by VARCHAR(50) NOT NULL,
+  PRIMARY KEY (workspace_id, asset_kind, asset_path)
+);
+CREATE INDEX idx_asset_materializer_runnable
+  ON asset_materializer (workspace_id, runnable_kind, runnable_path);
+
+CREATE TABLE asset_materializer_history (
+  id BIGSERIAL PRIMARY KEY,
+  workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  asset_kind ASSET_KIND NOT NULL,
+  asset_path VARCHAR(255) NOT NULL,
+  runnable_kind ASSET_USAGE_KIND NOT NULL,
+  runnable_path VARCHAR(255) NOT NULL,
+  deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  deployed_by VARCHAR(50) NOT NULL
+);
+CREATE INDEX idx_asset_materializer_history_asset
+  ON asset_materializer_history (workspace_id, asset_kind, asset_path, deployed_at DESC);

backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql

@@ -0,0 +1,31 @@
+-- Simplified pipeline model: `// materialize` is a bare opt-in marker (sets
+-- auto_kind='materializer') and the per-asset ownership tracking goes away.
+-- Writes are already tracked in the `asset` table via the parser, so the
+-- `asset_materializer*` tables no longer earn their keep.
+DROP TABLE IF EXISTS asset_materializer_history;
+DROP TABLE IF EXISTS asset_materializer;
+
+-- Execution DAG edges declared via `// on <asset | schedule>` annotations.
+-- For `trigger_kind='asset'`: trigger_ref is `<kind>://<path>` (kind from
+--   parse_asset_syntax, so downstream lookups match the `asset` table).
+-- For `trigger_kind='schedule'`: trigger_ref is the raw cron expression.
+CREATE TYPE SCRIPT_TRIGGER_KIND AS ENUM ('asset', 'schedule');
+
+CREATE TABLE script_trigger (
+  id             BIGSERIAL PRIMARY KEY,
+  workspace_id   VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  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
+);
+
+-- Per-runnable lookup (wipe-on-deploy, list-triggers-for-script).
+CREATE INDEX idx_script_trigger_runnable
+  ON script_trigger (workspace_id, runnable_kind, runnable_path);
+
+-- Reverse lookup: "which scripts are triggered by asset X?" (the asset → script
+-- edges in the graph). trigger_ref is unbounded text so can't share the
+-- asset_kind btree, but this covers the common prefix-scan use case.
+CREATE INDEX idx_script_trigger_ref
+  ON script_trigger (workspace_id, trigger_kind, trigger_ref);

backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql

@@ -0,0 +1,12 @@
+-- Postgres doesn't support removing enum values in-place. The only safe
+-- rollback is to recreate the type with the original set and rewrite the
+-- column, deleting any rows using values introduced in the up migration.
+DELETE FROM script_trigger
+ WHERE trigger_kind NOT IN ('asset', 'schedule');
+
+CREATE TYPE SCRIPT_TRIGGER_KIND_OLD AS ENUM ('asset', 'schedule');
+ALTER TABLE script_trigger
+  ALTER COLUMN trigger_kind TYPE SCRIPT_TRIGGER_KIND_OLD
+  USING trigger_kind::text::SCRIPT_TRIGGER_KIND_OLD;
+DROP TYPE SCRIPT_TRIGGER_KIND;
+ALTER TYPE SCRIPT_TRIGGER_KIND_OLD RENAME TO SCRIPT_TRIGGER_KIND;

backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql

@@ -0,0 +1,11 @@
+-- Expand the pipeline trigger enum to cover every non-integration
+-- (i.e. non-native) trigger kind Windmill supports. Each value mirrors a
+-- keyword the annotation parser recognises in `// on <kind> <ref>` lines.
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'webhook';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'email';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'kafka';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'mqtt';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'nats';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'postgres';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'sqs';
+ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'gcp';

backend/parsers/windmill-parser-py-asset/src/lib.rs

@@ -2,8 +2,8 @@ use rustpython_ast::{Constant, Expr, ExprConstant, Visitor};
 use rustpython_parser::{ast::Suite, Parse};
 use std::collections::HashMap;
 use windmill_parser::asset_parser::{
-    asset_was_used, merge_assets, parse_asset_syntax, AssetKind, AssetUsageAccessType,
-    ParseAssetsOutput, ParseAssetsResult,
+    asset_was_used, merge_assets, parse_asset_syntax, parse_pipeline_annotations, AssetKind,
+    AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult,
 };
 use AssetUsageAccessType::*;
 
@@ -28,7 +28,13 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         }
     }
 
-    Ok(ParseAssetsOutput { assets: merge_assets(assets_finder.assets), ..Default::default() })
+    let (is_materializer, triggers) = parse_pipeline_annotations(input);
+    Ok(ParseAssetsOutput {
+        assets: merge_assets(assets_finder.assets),
+        is_materializer,
+        triggers,
+        ..Default::default()
+    })
 }
 
 type VarAssetName = String;

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

@@ -9,8 +9,8 @@ use sqlparser::{
     parser::Parser,
 };
 use windmill_parser::asset_parser::{
-    asset_was_used, merge_assets, parse_asset_syntax, AssetKind, AssetUsageAccessType,
-    ParseAssetsOutput, ParseAssetsResult,
+    asset_was_used, merge_assets, parse_asset_syntax, parse_pipeline_annotations, AssetKind,
+    AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult,
 };
 use AssetUsageAccessType::*;
 
@@ -33,7 +33,13 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         }
     }
 
-    Ok(ParseAssetsOutput { assets: merge_assets(collector.assets), ..Default::default() })
+    let (is_materializer, triggers) = parse_pipeline_annotations(input);
+    Ok(ParseAssetsOutput {
+        assets: merge_assets(collector.assets),
+        is_materializer,
+        triggers,
+        ..Default::default()
+    })
 }
 
 /// Visitor that collects S3 asset literals from SQL statements

backend/parsers/windmill-parser-ts-asset/src/lib.rs

@@ -5,8 +5,8 @@ use swc_ecma_ast::{CallExpr, Expr, Lit, MemberExpr, MemberProp, Str};
 use swc_ecma_parser::{lexer::Lexer, Parser, StringInput, Syntax, TsSyntax};
 use swc_ecma_visit::{Visit, VisitWith};
 use windmill_parser::asset_parser::{
-    asset_was_used, merge_assets, parse_asset_syntax, AssetKind, AssetUsageAccessType,
-    ParseAssetsOutput, ParseAssetsResult, SqlQueryDetails,
+    asset_was_used, merge_assets, parse_asset_syntax, parse_pipeline_annotations, AssetKind,
+    AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult, SqlQueryDetails,
 };
 use AssetUsageAccessType::*;
 
@@ -38,9 +38,12 @@ pub fn parse_assets(code: &str) -> anyhow::Result<ParseAssetsOutput> {
     let mut assets_finder =
         AssetsFinder { assets: vec![], sql_queries: vec![], var_identifiers: HashMap::new() };
     assets_finder.visit_module_items(&ast);
+    let (is_materializer, triggers) = parse_pipeline_annotations(code);
     Ok(ParseAssetsOutput {
         assets: merge_assets(assets_finder.assets),
         sql_queries: assets_finder.sql_queries,
+        is_materializer,
+        triggers,
     })
 }
 

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

@@ -1,5 +1,6 @@
 use windmill_parser::asset_parser::{
-    merge_assets, AssetKind, AssetUsageAccessType, ParseAssetsOutput, ParseAssetsResult,
+    merge_assets, parse_pipeline_annotations, AssetKind, AssetUsageAccessType, ParseAssetsOutput,
+    ParseAssetsResult,
 };
 
 use crate::{parse_ansible_reqs, ResourceOrVariablePath};
@@ -39,5 +40,11 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         }
     }
 
-    Ok(ParseAssetsOutput { assets: merge_assets(assets), ..Default::default() })
+    let (is_materializer, triggers) = parse_pipeline_annotations(input);
+    Ok(ParseAssetsOutput {
+        assets: merge_assets(assets),
+        is_materializer,
+        triggers,
+        ..Default::default()
+    })
 }

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

@@ -47,6 +47,42 @@ pub struct SqlQueryDetails {
 pub struct ParseAssetsOutput {
     pub assets: Vec<ParseAssetsResult>,
     pub sql_queries: Vec<SqlQueryDetails>,
+    // Bare `// materialize` (or `#` / `--`) anywhere in the source — opt-in
+    // marker that sets auto_kind='materializer' and includes the script in
+    // its folder's pipeline. Does NOT declare what is materialized: the
+    // parser-detected `w`/`rw` usages in `assets` are the outputs.
+    #[serde(skip_serializing_if = "std::ops::Not::not", default)]
+    pub is_materializer: bool,
+    // `// on <trigger-spec>` annotations — execution DAG edges. Each is an
+    // independent OR (any fires the script). Empty = script has no
+    // automatic triggers (still runnable manually / via existing cron
+    // triggers).
+    #[serde(skip_serializing_if = "Vec::is_empty", default)]
+    pub triggers: Vec<TriggerSpec>,
+}
+
+#[derive(Serialize, Debug, PartialEq, Clone)]
+#[serde(tag = "kind", rename_all = "lowercase")]
+pub enum TriggerSpec {
+    // Refresh when `<asset>` changes. Kind comes from parse_asset_syntax so
+    // it matches the `asset` table.
+    Asset { asset_kind: AssetKind, path: String },
+    // Refresh on cron. The raw expression is passed through as-is so the
+    // existing schedule subsystem can validate it.
+    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 },
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -153,3 +189,181 @@ pub const ASSET_KINDS: &[(&str, AssetKind)] = &[
     ("datatable://", AssetKind::DataTable),
     ("volume://", AssetKind::Volume),
 ];
+
+// Scan raw source for pipeline annotations. Language-agnostic: any line
+// whose first non-whitespace tokens are a comment prefix (`//`, `#`, or
+// `--`) followed by either:
+//   - bare `materialize` (no arguments)  → is_materializer = true
+//   - `on <trigger-spec>`                 → one TriggerSpec entry
+//
+// Trailing content on a `// materialize` line (beyond whitespace) is ignored
+// so the old `// materialize s3://foo` style is forward-compatible: the
+// bare marker still fires even if users have stale per-asset arguments.
+pub fn parse_pipeline_annotations(code: &str) -> (bool, Vec<TriggerSpec>) {
+    let mut is_materializer = false;
+    let mut triggers: Vec<TriggerSpec> = vec![];
+
+    for raw_line in code.lines() {
+        let line = raw_line.trim_start();
+        let rest = if let Some(r) = line.strip_prefix("//") {
+            r
+        } else if let Some(r) = line.strip_prefix("--") {
+            r
+        } else if let Some(r) = line.strip_prefix('#') {
+            r
+        } else {
+            continue;
+        };
+        let rest = rest.trim_start();
+
+        if let Some(after_kw) = rest.strip_prefix("materialize") {
+            // Bare marker: must be end-of-line, whitespace, or nothing.
+            // Rejects `materializer`, `materialized`, etc.
+            if after_kw.is_empty() || after_kw.starts_with(|c: char| c.is_whitespace()) {
+                is_materializer = true;
+            }
+            continue;
+        }
+
+        if let Some(after_kw) = rest.strip_prefix("on") {
+            if !after_kw.starts_with(|c: char| c.is_whitespace()) {
+                continue;
+            }
+            let spec_text = after_kw.trim();
+            if spec_text.is_empty() {
+                continue;
+            }
+            if let Some(trig) = parse_trigger_spec(spec_text) {
+                if !triggers.contains(&trig) {
+                    triggers.push(trig);
+                }
+            }
+        }
+    }
+
+    (is_materializer, triggers)
+}
+
+// Parse a single `on <spec>` right-hand side. Accepted forms:
+//   schedule "<cron expr>"
+//   <kind> <trigger-path>           — where <kind> is one of
+//     webhook | email | kafka | mqtt | nats | postgres | sqs | gcp
+//   <asset-path-with-prefix>        (e.g. s3://bucket/key, $res:f/foo)
+fn parse_trigger_spec(s: &str) -> Option<TriggerSpec> {
+    if let Some(rest) = s.strip_prefix("schedule") {
+        let rest = rest.trim_start();
+        // Accept either double or single quotes around the cron expression.
+        let cron = rest
+            .strip_prefix('"')
+            .and_then(|r| r.strip_suffix('"'))
+            .or_else(|| rest.strip_prefix('\'').and_then(|r| r.strip_suffix('\'')))?;
+        if cron.trim().is_empty() {
+            return None;
+        }
+        return Some(TriggerSpec::Schedule { cron: cron.to_string() });
+    }
+
+    // `<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 }),
+    ];
+    for (kw, ctor) in KINDS {
+        if let Some(rest) = s.strip_prefix(kw) {
+            if !rest.starts_with(|c: char| c.is_whitespace()) {
+                continue;
+            }
+            let path = rest.trim();
+            if path.is_empty() {
+                return None;
+            }
+            return Some(ctor(path.to_string()));
+        }
+    }
+
+    let (kind, path) = parse_asset_syntax(s, false)?;
+    Some(TriggerSpec::Asset { asset_kind: kind, path: path.to_string() })
+}
+
+#[cfg(test)]
+mod pipeline_annotation_tests {
+    use super::*;
+
+    #[test]
+    fn bare_materialize_marker() {
+        let (is_m, triggers) = parse_pipeline_annotations("// materialize\nconsole.log('hi')");
+        assert!(is_m);
+        assert!(triggers.is_empty());
+    }
+
+    #[test]
+    fn bare_materialize_with_trailing_noise_still_fires() {
+        // Forward-compat with old `// materialize s3://foo` style.
+        let (is_m, _) = parse_pipeline_annotations("// materialize s3://legacy");
+        assert!(is_m);
+    }
+
+    #[test]
+    fn rejects_materializer_keyword_variants() {
+        let (is_m, _) = parse_pipeline_annotations("// materializer\n# materialized");
+        assert!(!is_m);
+    }
+
+    #[test]
+    fn on_schedule() {
+        let (_, triggers) = parse_pipeline_annotations("// on schedule \"0 */6 * * *\"");
+        assert_eq!(triggers.len(), 1);
+        assert_eq!(
+            triggers[0],
+            TriggerSpec::Schedule { cron: "0 */6 * * *".to_string() }
+        );
+    }
+
+    #[test]
+    fn on_asset_ts_py_sql() {
+        let code = "// on s3://a/b\n# on datatable://main\n-- on $res:f/foo";
+        let (_, triggers) = parse_pipeline_annotations(code);
+        assert_eq!(triggers.len(), 3);
+        assert!(matches!(
+            triggers[0],
+            TriggerSpec::Asset { asset_kind: AssetKind::S3Object, .. }
+        ));
+        assert!(matches!(
+            triggers[1],
+            TriggerSpec::Asset { asset_kind: AssetKind::DataTable, .. }
+        ));
+        assert!(matches!(
+            triggers[2],
+            TriggerSpec::Asset { asset_kind: AssetKind::Resource, .. }
+        ));
+    }
+
+    #[test]
+    fn on_deduplicates() {
+        let code = "// on s3://a/b\n# on s3://a/b";
+        let (_, triggers) = parse_pipeline_annotations(code);
+        assert_eq!(triggers.len(), 1);
+    }
+
+    #[test]
+    fn rejects_unknown_trigger_spec() {
+        let (_, triggers) = parse_pipeline_annotations("// on unknown://nope\n# on schedule");
+        assert!(triggers.is_empty());
+    }
+
+    #[test]
+    fn combined() {
+        let code = "// materialize\n// on s3://in.csv\n// on schedule \"0 0 * * *\"";
+        let (is_m, triggers) = parse_pipeline_annotations(code);
+        assert!(is_m);
+        assert_eq!(triggers.len(), 2);
+    }
+}

backend/windmill-api-assets/src/lib.rs

@@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use sqlx::Row;
 use windmill_common::{
-    assets::{AssetKind, AssetUsageKind},
+    assets::{parse_asset_trigger_ref, AssetKind, AssetUsageKind},
     db::UserDB,
     error::JsonResult,
     utils::escape_ilike_pattern,
@@ -21,6 +21,7 @@ pub fn workspaced_service() -> Router {
         .route("/list_by_usages", post(list_assets_by_usages))
         .route("/list_favorites", get(list_favorites))
         .route("/graph", get(asset_graph))
+        .route("/pipelines", get(list_pipeline_folders))
 }
 
 #[derive(Deserialize)]
@@ -388,8 +389,14 @@ struct GraphAssetNode {
 struct GraphRunnableNode {
     path: String,
     usage_kind: AssetUsageKind,
+    // True iff the script was deployed with `// materialize` — drives the
+    // pipeline-member visual state on the frontend.
+    #[serde(skip_serializing_if = "std::ops::Not::not", default)]
+    is_materializer: bool,
 }
 
+// Lineage edge from parsed r/w usages. One per (runnable, asset, access_type)
+// tuple. Informational — not the DAG execution edges.
 #[derive(Serialize, Debug)]
 struct GraphEdge {
     runnable_path: String,
@@ -399,11 +406,72 @@ struct GraphEdge {
     access_type: Option<String>,
 }
 
+// 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.
+#[derive(Serialize, Debug)]
+#[serde(tag = "trigger_kind", rename_all = "lowercase")]
+enum TriggerEdge {
+    Asset {
+        asset_kind: AssetKind,
+        asset_path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Schedule {
+        cron: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Webhook {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Email {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Kafka {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Mqtt {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Nats {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Postgres {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Sqs {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+    Gcp {
+        path: String,
+        runnable_kind: AssetUsageKind,
+        runnable_path: String,
+    },
+}
+
 #[derive(Serialize, Debug)]
 struct AssetGraphResponse {
     assets: Vec<GraphAssetNode>,
     runnables: Vec<GraphRunnableNode>,
     edges: Vec<GraphEdge>,
+    triggers: Vec<TriggerEdge>,
 }
 
 async fn asset_graph(
@@ -450,12 +518,60 @@ 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).
+    let trigger_rows = sqlx::query!(
+        r#"
+        SELECT
+            runnable_kind AS "runnable_kind!: AssetUsageKind",
+            runnable_path AS "runnable_path!",
+            trigger_kind::text AS "trigger_kind!",
+            trigger_ref   AS "trigger_ref!"
+        FROM script_trigger
+        WHERE workspace_id = $1
+          AND ($2::text IS NULL OR runnable_path LIKE $2)
+        "#,
+        &w_id,
+        folder_filter.as_deref(),
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    // Which scripts in scope are pipeline members (have `// materialize`).
+    let materializer_paths = sqlx::query!(
+        r#"
+        SELECT path AS "path!"
+        FROM script
+        WHERE workspace_id = $1
+          AND auto_kind = 'materializer'
+          AND archived = false
+          AND deleted = false
+          AND ($2::text IS NULL OR path LIKE $2)
+        "#,
+        &w_id,
+        folder_filter.as_deref(),
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
     tx.commit().await?;
 
+    let materializer_script_paths: std::collections::HashSet<String> =
+        materializer_paths.into_iter().map(|r| r.path).collect();
+
     let mut edges = Vec::with_capacity(rows.len());
     let mut asset_set: std::collections::HashSet<(AssetKind, String)> = Default::default();
     let mut runnable_set: std::collections::HashSet<(AssetUsageKind, String)> = Default::default();
 
+    // Every materializer in scope goes into the graph, even when the parser
+    // didn't detect any asset r/w and the script has no triggers yet. Without
+    // this, a freshly-saved materializer whose template body hasn't been
+    // filled in would vanish from the pipeline view on graph refetch.
+    for path in &materializer_script_paths {
+        runnable_set.insert((AssetUsageKind::Script, path.clone()));
+    }
+
     for r in rows {
         asset_set.insert((r.asset_kind, r.asset_path.clone()));
         runnable_set.insert((r.usage_kind, r.usage_path.clone()));
@@ -468,6 +584,79 @@ async fn asset_graph(
         });
     }
 
+    let mut triggers: Vec<TriggerEdge> = Vec::with_capacity(trigger_rows.len());
+    for t in trigger_rows {
+        runnable_set.insert((t.runnable_kind, t.runnable_path.clone()));
+        match t.trigger_kind.as_str() {
+            "asset" => {
+                // trigger_ref is `<prefix><path>` — parse back out so both
+                // endpoints match what the frontend uses for node ids.
+                if let Some((asset_kind, asset_path)) = parse_asset_trigger_ref(&t.trigger_ref) {
+                    // Make sure the source asset has a node even if nothing
+                    // reads/writes it in this folder.
+                    asset_set.insert((asset_kind, asset_path.clone()));
+                    triggers.push(TriggerEdge::Asset {
+                        asset_kind,
+                        asset_path,
+                        runnable_kind: t.runnable_kind,
+                        runnable_path: t.runnable_path,
+                    });
+                }
+            }
+            "schedule" => {
+                triggers.push(TriggerEdge::Schedule {
+                    cron: t.trigger_ref,
+                    runnable_kind: t.runnable_kind,
+                    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.
+        }
+    }
+
     let mut assets: Vec<GraphAssetNode> = asset_set
         .into_iter()
         .map(|(kind, path)| GraphAssetNode { kind, path })
@@ -476,9 +665,66 @@ async fn asset_graph(
 
     let mut runnables: Vec<GraphRunnableNode> = runnable_set
         .into_iter()
-        .map(|(usage_kind, path)| GraphRunnableNode { path, usage_kind })
+        .map(|(usage_kind, path)| {
+            let is_materializer =
+                usage_kind == AssetUsageKind::Script && materializer_script_paths.contains(&path);
+            GraphRunnableNode { path, usage_kind, is_materializer }
+        })
         .collect();
     runnables.sort_by(|a, b| a.path.cmp(&b.path));
 
-    Ok(Json(AssetGraphResponse { assets, runnables, edges }))
+    Ok(Json(AssetGraphResponse {
+        assets,
+        runnables,
+        edges,
+        triggers,
+    }))
+}
+
+// ------------------------------------------------------------------
+// GET /w/:workspace/assets/pipelines
+// ------------------------------------------------------------------
+// Distinct folder names that contain at least one materializer script
+// (auto_kind='materializer'). Used by the pipeline-editor folder picker
+// and the "Pipeline" entry in folder views. Keyed by the partial index
+// on `script (workspace_id, path) WHERE auto_kind='materializer' ...`
+// so this is effectively O(matches).
+
+#[derive(Serialize, Debug)]
+struct PipelineFolder {
+    folder: String,
+    script_count: i64,
+}
+
+async fn list_pipeline_folders(
+    authed: ApiAuthed,
+    Path(w_id): Path<String>,
+    Extension(user_db): Extension<UserDB>,
+) -> JsonResult<Vec<PipelineFolder>> {
+    let mut tx = user_db.begin(&authed).await?;
+    let rows = sqlx::query!(
+        r#"
+        SELECT
+            substring(path from '^f/([^/]+)/') AS "folder!",
+            COUNT(*) AS "script_count!"
+        FROM script
+        WHERE workspace_id = $1
+          AND auto_kind = 'materializer'
+          AND archived = false
+          AND deleted = false
+          AND path LIKE 'f/%'
+        GROUP BY substring(path from '^f/([^/]+)/')
+        ORDER BY substring(path from '^f/([^/]+)/')
+        "#,
+        &w_id,
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    tx.commit().await?;
+
+    Ok(Json(
+        rows.into_iter()
+            .map(|r| PipelineFolder { folder: r.folder, script_count: r.script_count })
+            .collect(),
+    ))
 }

backend/windmill-api-scripts/src/scripts.rs

@@ -43,8 +43,9 @@ use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap;
 
 use windmill_common::{
     assets::{
-        clear_static_asset_usage, clear_static_asset_usage_by_script_hash,
-        insert_static_asset_usage, AssetUsageKind, AssetWithAltAccessType,
+        clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash,
+        insert_script_trigger, insert_static_asset_usage, parse_pipeline_annotations,
+        trigger_spec_to_row, AssetUsageKind, AssetWithAltAccessType,
     },
     error::{self, to_anyhow},
     min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2},
@@ -966,7 +967,14 @@ async fn create_script_internal<'c>(
 
     let ci_test_refs =
         windmill_common::schema::parse_ci_test_annotation(&ns.content, &lang.as_comment_lit());
-    let auto_kind = if ci_test_refs.is_some() {
+    // `materializer` wins over `test` and any client-supplied auto_kind. The
+    // bare `// materialize` marker is the opt-in signal for pipeline
+    // membership; parsed writes tell us what is produced (we don't record
+    // them in auto_kind itself).
+    let (is_materializer, pipeline_triggers) = parse_pipeline_annotations(&ns.content);
+    let auto_kind = if is_materializer {
+        Some("materializer".to_string())
+    } else if ci_test_refs.is_some() {
         Some("test".to_string())
     } else {
         auto_kind
@@ -1277,6 +1285,22 @@ async fn create_script_internal<'c>(
             .await?;
     }
 
+    // Pipeline trigger edges: wipe-and-reinsert per deploy so removing an
+    // `// on ...` annotation drops the edge.
+    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);
+        insert_script_trigger(
+            &mut *tx,
+            &w_id,
+            AssetUsageKind::Script,
+            &ns.path,
+            trigger_kind,
+            &trigger_ref,
+        )
+        .await?;
+    }
+
     let permissioned_as = username_to_permissioned_as(&authed.username);
     if let Some(parent_hash) = ns.parent_hash {
         tracing::info!(

backend/windmill-common/src/assets.rs

@@ -2,8 +2,24 @@ use sqlx::PgExecutor;
 
 use crate::{error, scripts::ScriptHash};
 
+pub use windmill_parser::asset_parser::{parse_pipeline_annotations, TriggerSpec};
 pub use windmill_types::assets::*;
 
+#[derive(sqlx::Type, Debug, Clone, Copy, PartialEq)]
+#[sqlx(type_name = "SCRIPT_TRIGGER_KIND", rename_all = "lowercase")]
+pub enum ScriptTriggerKind {
+    Asset,
+    Schedule,
+    Webhook,
+    Email,
+    Kafka,
+    Mqtt,
+    Nats,
+    Postgres,
+    Sqs,
+    Gcp,
+}
+
 pub async fn insert_static_asset_usage<'e>(
     executor: impl PgExecutor<'e>,
     workspace_id: &str,
@@ -66,6 +82,87 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>(
     Ok(())
 }
 
+// Wipe all pipeline trigger declarations held by the given runnable. Used at
+// deploy time: redeploying a script wipes its prior `// on` annotations so
+// removing them implicitly un-declares those edges.
+pub async fn clear_script_triggers<'e>(
+    executor: impl PgExecutor<'e>,
+    workspace_id: &str,
+    runnable_path: &str,
+    runnable_kind: AssetUsageKind,
+) -> error::Result<()> {
+    sqlx::query!(
+        r#"DELETE FROM script_trigger
+           WHERE workspace_id = $1 AND runnable_kind = $2 AND runnable_path = $3"#,
+        workspace_id,
+        runnable_kind as AssetUsageKind,
+        runnable_path,
+    )
+    .execute(executor)
+    .await?;
+    Ok(())
+}
+
+// Insert a single trigger declaration. Caller is expected to wipe first.
+pub async fn insert_script_trigger<'e>(
+    executor: impl PgExecutor<'e>,
+    workspace_id: &str,
+    runnable_kind: AssetUsageKind,
+    runnable_path: &str,
+    trigger_kind: ScriptTriggerKind,
+    trigger_ref: &str,
+) -> error::Result<()> {
+    sqlx::query!(
+        r#"INSERT INTO script_trigger
+             (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref)
+           VALUES ($1, $2, $3, $4, $5)"#,
+        workspace_id,
+        runnable_kind as AssetUsageKind,
+        runnable_path,
+        trigger_kind as ScriptTriggerKind,
+        trigger_ref,
+    )
+    .execute(executor)
+    .await?;
+    Ok(())
+}
+
+// Inverse of trigger_spec_to_row for the Asset variant: parses a stored
+// trigger_ref (e.g. `s3://foo`, `$res:bar`) back into the (kind, path) pair
+// used as a graph node id. Returns None for refs that don't match any known
+// asset prefix — callers should skip those edges.
+pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, String)> {
+    let (kind, path) = windmill_parser::asset_parser::parse_asset_syntax(s, false)?;
+    Some((asset_kind_from_parser(kind), path.to_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) {
+    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://",
+            };
+            (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()),
+    }
+}
+
 pub fn asset_kind_from_parser(parser_kind: windmill_parser::asset_parser::AssetKind) -> AssetKind {
     match parser_kind {
         windmill_parser::asset_parser::AssetKind::S3Object => AssetKind::S3Object,

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

@@ -2,7 +2,6 @@
 	import '@xyflow/svelte/dist/base.css'
 	import {
 		SvelteFlow,
-		Background,
 		Controls,
 		MiniMap,
 		ConnectionLineType,
@@ -12,36 +11,132 @@
 	} from '@xyflow/svelte'
 	import AssetNode from './AssetNode.svelte'
 	import RunnableNode from './RunnableNode.svelte'
+	import TriggerNode, { type TriggerNodeKind } from './TriggerNode.svelte'
+	import AddNode from './AddNode.svelte'
 	import AssetGraphEdge from './AssetGraphEdge.svelte'
 	import { layoutAssetGraph } from './assetGraphLayout'
 	import type { AssetGraphResponse, AssetGraphSelection } from './types'
+	import type { AssetKind } from '$lib/gen'
+	import { NODE } from '$lib/components/graph/util'
+
+	// Width of the + node's rendered DOM element. Sugiyama allocates a full
+	// NODE.width slot for every node, so the small round button ends up
+	// left-aligned in its slot. We compensate by shifting the + node right
+	// by half the difference so its visual center matches the slot center.
+	const ADD_NODE_WIDTH = 40
 
 	interface Props {
 		graph: AssetGraphResponse
 		selection?: AssetGraphSelection | undefined
 		onselect?: (selection: AssetGraphSelection | undefined) => void
+		// Called when the user clicks the per-asset + button (consumer-script
+		// entry). Kept optional so the canvas stays usable outside the
+		// pipeline editor.
+		onAddScriptForAsset?: (
+			asset: { kind: AssetKind; path: string },
+			language: import('$lib/gen').ScriptLang,
+			scriptPath: string
+		) => void
+		// Pipeline-wide + node shown at the top of the graph. Picking any
+		// kind from the menu invokes this one callback with the chosen
+		// trigger source — the page uses it to seed the draft's annotation.
+		onAddMaterializer?: (
+			language: import('$lib/gen').ScriptLang,
+			path: string,
+			source:
+				| { kind: 'schedule'; cron: string }
+				| {
+						kind: 'webhook' | 'email' | 'kafka' | 'mqtt' | 'nats' | 'postgres' | 'sqs' | 'gcp'
+						path: string | undefined
+				  }
+		) => void
+		// Folder-scoped prefix shown as a read-only chip in the insert menu
+		// path input (e.g. `f/{folder}/`). Shared across top + and per-asset +.
+		pathPrefix?: string
+		// Seeded editable suffix (e.g. `new_materializer`).
+		defaultPathSuffix?: string
+		// Default cron expression seeded into the top + template.
+		defaultScheduleCron?: string
 	}
-	let { graph, selection, onselect }: Props = $props()
+	let {
+		graph,
+		selection,
+		onselect,
+		onAddScriptForAsset,
+		onAddMaterializer,
+		pathPrefix = '',
+		defaultPathSuffix = '',
+		defaultScheduleCron = ''
+	}: Props = $props()
 
-	// Producers (w/rw) point runnable → asset; consumers (r) point asset →
-	// runnable. The sugiyama layout (assetGraphLayout.ts) renders this as a
-	// top-down DAG: producers above, assets in the middle, consumers below.
+	const ADD_NODE_ID = '__add__'
+
+	type BuiltEdge = {
+		id: string
+		source: string
+		target: string
+		// 'add-anchor' edges connect the + node to every otherwise-root node in
+		// the DAG. They force sugiyama to put + at layer 0 (top) and center
+		// it horizontally over the roots — same mechanism the flow editor
+		// uses for its Trigger node. Filtered out of rendered edges.
+		kind:
+			| 'lineage-write'
+			| 'lineage-read'
+			| 'trigger-asset'
+			| 'trigger-schedule'
+			| 'trigger-native'
+			| 'add-anchor'
+		unsaved?: boolean
+	}
+
+	// Lineage edges (parsed r/w usages): writer → asset, asset → reader.
+	// Trigger edges (`// on <x>`): asset → script or schedule → script. The
+	// lineage subgraph is informational; the trigger subgraph is executable.
 	function build(g: AssetGraphResponse) {
-		const nodes: Array<{ id: string; type: 'asset' | 'runnable'; data: any }> = []
-		const edges: Array<{ id: string; source: string; target: string; access: string | null }> = []
+		const nodes: Array<{
+			id: string
+			type: 'asset' | 'runnable' | 'trigger' | 'add'
+			data: any
+		}> = []
+		const edges: BuiltEdge[] = []
+
+		const hasAddNode = onAddMaterializer != null
+		if (hasAddNode) {
+			nodes.push({
+				id: ADD_NODE_ID,
+				type: 'add',
+				data: {
+					onAddMaterializer: onAddMaterializer!,
+					pathPrefix,
+					defaultPathSuffix,
+					defaultScheduleCron
+				}
+			})
+		}
 
 		for (const a of g.assets) {
+			const assetId = `asset:${a.kind}:${a.path}`
 			nodes.push({
-				id: `asset:${a.kind}:${a.path}`,
+				id: assetId,
 				type: 'asset',
-				data: { asset_kind: a.kind, path: a.path }
+				data: {
+					asset_kind: a.kind,
+					path: a.path,
+					onAddScript: onAddScriptForAsset,
+					pathPrefix,
+					defaultPathSuffix
+				}
 			})
 		}
 		for (const r of g.runnables) {
 			nodes.push({
 				id: `${r.usage_kind}:${r.path}`,
 				type: 'runnable',
-				data: { runnable_kind: r.usage_kind, path: r.path }
+				data: {
+					runnable_kind: r.usage_kind,
+					path: r.path,
+					is_materializer: r.is_materializer ?? false
+				}
 			})
 		}
 
@@ -54,7 +149,8 @@
 					id: `prod:${runnableId}->${assetId}`,
 					source: runnableId,
 					target: assetId,
-					access
+					kind: 'lineage-write',
+					unsaved: e.unsaved
 				})
 			}
 			if (access === 'r' || access === 'rw') {
@@ -62,10 +158,83 @@
 					id: `cons:${assetId}->${runnableId}`,
 					source: assetId,
 					target: runnableId,
-					access
+					kind: 'lineage-read',
+					unsaved: e.unsaved
 				})
 			}
 		}
+
+		// Non-asset triggers (schedule + native) are rendered as source nodes
+		// above the materializer. 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.
+		const triggerSourceNodes = new Map<
+			string,
+			{ allUnsaved: boolean; kind: TriggerNodeKind; ref: string }
+		>()
+		function recordSourceTrigger(id: string, kind: TriggerNodeKind, ref: string, unsaved: boolean) {
+			const prev = triggerSourceNodes.get(id)
+			if (!prev) {
+				triggerSourceNodes.set(id, { allUnsaved: unsaved, kind, ref })
+			} else {
+				prev.allUnsaved = prev.allUnsaved && unsaved
+			}
+		}
+
+		for (const t of g.triggers ?? []) {
+			const runnableId = `${t.runnable_kind}:${t.runnable_path}`
+			if (t.trigger_kind === 'asset') {
+				const assetId = `asset:${t.asset_kind}:${t.asset_path}`
+				edges.push({
+					id: `trig-a:${assetId}->${runnableId}`,
+					source: assetId,
+					target: runnableId,
+					kind: 'trigger-asset',
+					unsaved: t.unsaved
+				})
+				continue
+			}
+			const ref = 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)
+			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
+			})
+		}
+		for (const [id, info] of triggerSourceNodes) {
+			nodes.push({
+				id,
+				type: 'trigger',
+				data: { kind: info.kind, ref: info.ref, unsaved: info.allUnsaved }
+			})
+		}
+
+		// Anchor the + to layer 0 by making it a parent of every node that
+		// would otherwise be a root (no incoming edge), *including* schedule
+		// nodes. Must run AFTER all real edges are added so we catch the
+		// current set of roots. Sugiyama then places + on layer 0 and
+		// centers it over whatever's below.
+		if (hasAddNode) {
+			const hasIncoming = new Set<string>()
+			for (const e of edges) hasIncoming.add(e.target)
+			for (const n of nodes) {
+				if (n.id === ADD_NODE_ID) continue
+				if (!hasIncoming.has(n.id)) {
+					edges.push({
+						id: `add-anchor:${n.id}`,
+						source: ADD_NODE_ID,
+						target: n.id,
+						kind: 'add-anchor'
+					})
+				}
+			}
+		}
+
 		return { nodes, edges }
 	}
 
@@ -78,36 +247,125 @@
 			: `${selection.runnable_kind}:${selection.path}`
 	})
 
+	// Pane width drives horizontal centering (same pattern FlowGraphV2 uses).
+	// Bound on the outer wrapper; updates on pane resize via $state.
+	let paneWidth = $state(800)
+
 	let positionedNodes = $derived.by(() => {
 		const positions = layoutAssetGraph({
 			nodes: model.nodes.map((n) => ({ id: n.id, data: n.data })),
 			edges: model.edges.map((e) => ({ source: e.source, target: e.target }))
 		})
+		// Compute bbox width from layout; shift every x so the graph is
+		// horizontally centered inside the pane. y is untouched so layer 0
+		// sits at the top of the viewport (matches flow editor's Trigger
+		// placement — no fitView reshuffling).
+		let minX = Infinity
+		let maxX = -Infinity
+		for (const p of positions.values()) {
+			if (p.x < minX) minX = p.x
+			if (p.x > maxX) maxX = p.x
+		}
+		const bboxWidth = isFinite(minX) ? maxX - minX : 0
+		const xCenter = paneWidth / 2 - bboxWidth / 2
 		return model.nodes.map<Node>((n) => {
 			const p = positions.get(n.id) ?? { x: 0, y: 0 }
+			// Compensate for the + node being narrower than its layout slot
+			// so it visually centers over the node(s) below.
+			const xShift = n.id === ADD_NODE_ID ? (NODE.width - ADD_NODE_WIDTH) / 2 : 0
 			return {
 				id: n.id,
 				type: n.type,
-				position: { x: p.x, y: p.y },
+				position: { x: p.x + xCenter + xShift, y: p.y + 40 },
 				data: n.data,
-				selected: n.id === selectedId
+				selected: n.id === selectedId,
+				// All nodes non-draggable: the layout is sugiyama-computed,
+				// dragging would fight the reactive re-layout. Selection is
+				// still allowed on asset/runnable (not on the + or schedules)
+				// for the details-pane click-through.
+				draggable: false,
+				selectable: n.id !== ADD_NODE_ID && n.type !== 'trigger'
 			}
 		})
 	})
 
 	let flowEdges = $derived.by(() =>
-		model.edges.map<Edge>((e) => ({
-			id: e.id,
-			source: e.source,
-			target: e.target,
-			type: 'asset',
-			animated: e.access === 'rw',
-			style:
-				e.access === 'w' || e.access === 'rw'
-					? 'stroke: rgb(59 130 246); stroke-width: 1.5px;'
-					: 'stroke: rgb(107 114 128); stroke-width: 1.25px;',
-			markerEnd: { type: MarkerType.ArrowClosed, width: 14, height: 14 }
-		}))
+		model.edges
+			// Anchor edges are layout-only.
+			.filter((e) => e.kind !== 'add-anchor')
+			.map<Edge>((e) => {
+				let style: string
+				let animated = false
+				let markerColor: string | undefined = undefined
+				let strokeDasharray: string | undefined = undefined
+				let label: string | undefined = undefined
+				let labelStyle: string | undefined = undefined
+				switch (e.kind) {
+					case 'lineage-write':
+						style = 'stroke: rgb(59 130 246); stroke-width: 1.5px;'
+						break
+					case 'lineage-read':
+						style = 'stroke: rgb(107 114 128); stroke-width: 1.25px;'
+						break
+					case 'trigger-asset':
+						style = 'stroke: rgb(16 185 129); stroke-width: 2px;'
+						animated = true
+						markerColor = 'rgb(16 185 129)'
+						label = 'triggers'
+						labelStyle = 'fill: rgb(16 185 129); font-size: 10px; font-weight: 600;'
+						break
+					case 'trigger-schedule':
+						style = 'stroke: rgb(245 158 11); stroke-width: 2px;'
+						strokeDasharray = '6 3'
+						markerColor = 'rgb(245 158 11)'
+						label = 'schedule'
+						labelStyle = 'fill: rgb(245 158 11); font-size: 10px; font-weight: 600;'
+						break
+					case 'trigger-native':
+						// Colour neutral here because the trigger source node
+						// already carries per-kind colour; edge just needs to
+						// say "this fires it" distinctly from lineage.
+						style = 'stroke: rgb(100 116 139); stroke-width: 2px;'
+						strokeDasharray = '6 3'
+						markerColor = 'rgb(100 116 139)'
+						label = 'triggers'
+						labelStyle = 'fill: rgb(100 116 139); font-size: 10px; font-weight: 600;'
+						break
+					default:
+						style = ''
+				}
+				// Unsaved edges (trigger or lineage) get a distinct dashed
+				// pattern + dimmed opacity so the user sees they're live-
+				// parsed, not persisted. Lineage edges from base graph data
+				// don't carry `unsaved`, only those synthesized by the draft
+				// overlay (e.g. the random output asset).
+				if (e.unsaved) {
+					strokeDasharray = '3 3'
+					style = `${style} opacity: 0.7;`
+					animated = false
+					if (label) label = `${label} (unsaved)`
+				}
+				if (strokeDasharray) {
+					style = `${style} stroke-dasharray: ${strokeDasharray};`
+				}
+				return {
+					id: e.id,
+					source: e.source,
+					target: e.target,
+					type: 'asset',
+					animated,
+					label,
+					labelStyle,
+					labelBgStyle: label ? 'fill: rgb(255 255 255 / 0.9);' : undefined,
+					style,
+					markerEnd: {
+						type: MarkerType.ArrowClosed,
+						width: 14,
+						height: 14,
+						color: markerColor
+					}
+				}
+			})
 	)
 
 	let nodes = $state.raw<Node[]>([])
@@ -121,7 +379,9 @@
 
 	const nodeTypes = {
 		asset: AssetNode as any,
-		runnable: RunnableNode as any
+		runnable: RunnableNode as any,
+		trigger: TriggerNode as any,
+		add: AddNode as any
 	}
 
 	const edgeTypes = {
@@ -133,19 +393,19 @@
 		const data = node.data as any
 		if (node.type === 'asset') {
 			onselect({ kind: 'asset', asset_kind: data.asset_kind, path: data.path })
-		} else {
+		} else if (node.type === 'runnable') {
 			onselect({ kind: 'runnable', runnable_kind: data.runnable_kind, path: data.path })
 		}
+		// 'schedule' doesn't produce a selection.
 	}
 </script>
 
-<div class="w-full h-full relative">
+<div class="w-full h-full relative" bind:clientWidth={paneWidth}>
 	<SvelteFlow
 		{nodes}
 		{edges}
 		{nodeTypes}
 		{edgeTypes}
-		fitView
 		minZoom={0.2}
 		maxZoom={1.6}
 		nodesDraggable={false}
@@ -159,14 +419,18 @@
 		onpaneclick={() => onselect?.(undefined)}
 		--background-color={false}
 	>
-		<div class="absolute inset-0 !bg-surface-secondary h-full -z-10"></div>
-		<Background />
+		<div class="absolute inset-0 !bg-surface-secondary h-full"></div>
 		<Controls />
 		<MiniMap
 			pannable
 			zoomable
 			class="!bg-surface"
-			nodeColor={(n) => (n.type === 'asset' ? 'rgb(96 165 250 / 0.5)' : 'rgb(52 211 153 / 0.5)')}
+			nodeColor={(n) =>
+				n.type === 'asset'
+					? 'rgb(96 165 250 / 0.5)'
+					: n.type === 'trigger'
+						? 'rgb(251 191 36 / 0.5)'
+						: 'rgb(52 211 153 / 0.5)'}
 			nodeStrokeColor="transparent"
 			maskColor="rgb(0 0 0 / 0.2)"
 		/>

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

@@ -7,39 +7,87 @@
 	import WorkerTagSelect from '$lib/components/WorkerTagSelect.svelte'
 	import AssetGenericIcon from '$lib/components/icons/AssetGenericIcon.svelte'
 	import { formatAssetKind } from '$lib/components/assets/lib'
-	import { Code2, ExternalLink, GitBranch, Loader2, Save, X } from 'lucide-svelte'
+	import { Code2, ExternalLink, GitBranch, Loader2, Save, Trash2, X } from 'lucide-svelte'
 	import { inferArgs } from '$lib/infer'
 	import { emptySchema, sendUserToast } from '$lib/utils'
 	import type { AssetGraphSelection } from './types'
+	import { parsePipelineAnnotations, type PipelineAnnotations } from './parsePipelineAnnotations'
 
 	interface Props {
-		selection: AssetGraphSelection
+		// Regular selection — loads the script by path for inline editing.
+		selection?: AssetGraphSelection | undefined
+		// Pre-built draft script not yet persisted. Takes precedence over
+		// `selection` when present. Used by the pipeline + menu so a new
+		// materializer opens inline instead of navigating to /scripts/add.
+		draftScript?: Script | undefined
 		workspace: string
 		onclose: () => void
+		// Called after a draft is saved for the first time so the page can
+		// refetch the graph and clear its local draft.
+		onDraftSaved?: (savedPath: string) => void
+		// Called when the user hits "Discard" on the draft header. Separate
+		// from `onclose` so the page can drop the draft from its map vs
+		// just dismissing the pane.
+		onDiscard?: () => void
+		// Emits live-parsed pipeline annotations from the open script so the
+		// canvas can overlay unsaved schedule / trigger nodes in real time.
+		// Fires whenever the script content changes.
+		onAnnotationsChange?: (scriptPath: string | undefined, annotations: PipelineAnnotations) => void
 	}
-	let { selection, workspace, onclose }: Props = $props()
+	let {
+		selection,
+		draftScript,
+		workspace,
+		onclose,
+		onDraftSaved,
+		onDiscard,
+		onAnnotationsChange
+	}: Props = $props()
 
-	// Mirror the workspace-script editing flow from ScriptEditorDrawer: load
-	// the script by path, hand a mutable copy to <ScriptEditor>, and on save
-	// re-infer args then POST a new version with parent_hash chained.
+	// When `draftScript` is provided we bypass the fetch entirely and edit
+	// it locally; saving calls ScriptService.createScript to deploy it.
 	let scriptRes = resource(
-		[() => workspace, () => selection],
-		async ([ws, sel], _prev, { signal }) => {
-			if (sel.kind !== 'runnable' || sel.runnable_kind !== 'script') return undefined
+		[() => workspace, () => selection, () => draftScript],
+		async ([ws, sel, draft], _prev, { signal }) => {
+			if (draft) return undefined
+			if (!sel || sel.kind !== 'runnable' || sel.runnable_kind !== 'script') return undefined
 			return await ScriptService.getScriptByPath({ workspace: ws, path: sel.path }, signal as any)
 		}
 	)
 
-	// Local mutable copy that ScriptEditor binds to. Reset whenever the
-	// underlying resource yields a new script (selection change or refetch).
+	// Local mutable copy. For drafts: seeded once from the incoming prop
+	// (subsequent typing stays in `script`, not `draftScript`). For fetched
+	// scripts: reset whenever the resource yields new data.
 	let script = $state<Script | undefined>(undefined)
 	$effect.pre(() => {
+		if (draftScript) {
+			script = structuredClone($state.snapshot(draftScript) as Script)
+			return
+		}
 		const fresh = scriptRes.current
 		script = fresh ? structuredClone($state.snapshot(fresh) as Script) : undefined
 	})
 
 	let args = $state<Record<string, any>>({})
 	let saving = $state(false)
+	let isDraft = $derived(draftScript != undefined)
+
+	// Live-parse the editor buffer so the page can render unsaved schedule /
+	// trigger nodes on the canvas. Parsed TS output mirrors the Rust
+	// `parse_pipeline_annotations` used at deploy time.
+	let liveAnnotations = $derived<PipelineAnnotations>(
+		script
+			? parsePipelineAnnotations(script.content ?? '')
+			: {
+					isMaterializer: false,
+					triggerAssets: [],
+					schedules: [],
+					nativeTriggers: []
+				}
+	)
+	$effect(() => {
+		onAnnotationsChange?.(script?.path, liveAnnotations)
+	})
 
 	async function save() {
 		if (!script) return
@@ -59,7 +107,8 @@
 					...script,
 					language: script.language,
 					description: script.description ?? '',
-					parent_hash: script.hash != undefined ? String(script.hash) : undefined,
+					// Drafts have no prior hash; workspace scripts chain off their last hash.
+					parent_hash: isDraft || script.hash == undefined ? undefined : String(script.hash),
 					is_template: false,
 					tag: script.tag,
 					kind: script.kind as Script['kind'] | undefined,
@@ -67,13 +116,21 @@
 				}
 			})
 			sendUserToast(`Saved ${script.path}`)
-			await scriptRes.refetch()
+			if (isDraft) {
+				onDraftSaved?.(script.path)
+			} else {
+				await scriptRes.refetch()
+			}
 		} catch (e: any) {
 			sendUserToast(`Save failed: ${e?.body ?? e?.message ?? e}`, true)
 		} finally {
 			saving = false
 		}
 	}
+
+	let isScriptView = $derived(
+		isDraft || (selection?.kind === 'runnable' && selection.runnable_kind === 'script')
+	)
 </script>
 
 <div class="flex flex-col h-full bg-surface">
@@ -81,7 +138,15 @@
 		class="flex items-center justify-between gap-2 px-3 py-2 border-b shrink-0 min-h-10 whitespace-nowrap"
 	>
 		<div class="flex items-center gap-2 min-w-0">
-			{#if selection.kind === 'asset'}
+			{#if isDraft && script}
+				<Code2 size={16} class="shrink-0 text-emerald-700 dark:text-emerald-400" />
+				<div class="flex flex-col min-w-0">
+					<span class="text-3xs uppercase tracking-wide text-emerald-600 dark:text-emerald-400">
+						Draft materializer
+					</span>
+					<span class="text-xs font-mono truncate" title={script.path}>{script.path}</span>
+				</div>
+			{:else if selection?.kind === 'asset'}
 				<AssetGenericIcon
 					assetKind={selection.asset_kind}
 					size="16px"
@@ -93,13 +158,13 @@
 					</span>
 					<span class="text-xs font-mono truncate" title={selection.path}>{selection.path}</span>
 				</div>
-			{:else if selection.runnable_kind === 'script'}
+			{:else if selection?.kind === 'runnable' && selection.runnable_kind === 'script'}
 				<Code2 size={16} class="shrink-0 text-emerald-700 dark:text-emerald-400" />
 				<div class="flex flex-col min-w-0">
 					<span class="text-3xs uppercase tracking-wide text-tertiary">Script</span>
 					<span class="text-xs font-mono truncate" title={selection.path}>{selection.path}</span>
 				</div>
-			{:else}
+			{:else if selection?.kind === 'runnable' && selection.runnable_kind === 'flow'}
 				<GitBranch size={16} class="shrink-0 text-emerald-700 dark:text-emerald-400" />
 				<div class="flex flex-col min-w-0">
 					<span class="text-3xs uppercase tracking-wide text-tertiary">Flow</span>
@@ -108,7 +173,17 @@
 			{/if}
 		</div>
 		<div class="flex items-center gap-1 shrink-0">
-			{#if selection.kind === 'runnable' && selection.runnable_kind === 'script' && script}
+			{#if isDraft && onDiscard}
+				<Button
+					variant="subtle"
+					unifiedSize="sm"
+					startIcon={{ icon: Trash2 }}
+					onclick={onDiscard}
+					iconOnly
+					title="Discard draft"
+				/>
+			{/if}
+			{#if isScriptView && script}
 				<Button
 					variant="accent"
 					unifiedSize="sm"
@@ -116,10 +191,10 @@
 					onclick={save}
 					disabled={saving}
 				>
-					{saving ? 'Saving…' : 'Save'}
+					{saving ? 'Saving…' : isDraft ? 'Create' : 'Save'}
 				</Button>
 			{/if}
-			{#if selection.kind === 'runnable'}
+			{#if !isDraft && selection?.kind === 'runnable'}
 				<Button
 					variant="subtle"
 					unifiedSize="sm"
@@ -144,25 +219,25 @@
 	</div>
 
 	<div class="flex-1 min-h-0 relative">
-		{#if selection.kind === 'asset'}
+		{#if selection?.kind === 'asset' && !isDraft}
 			<div class="p-3 text-xs text-secondary">
 				Asset details. Use the producer/consumer arrows in the graph to navigate.
 			</div>
-		{:else if selection.runnable_kind === 'flow'}
+		{:else if selection?.kind === 'runnable' && selection.runnable_kind === 'flow' && !isDraft}
 			<div class="p-3 text-xs text-secondary">
 				Flows are not editable inline. Use the open-in-editor button above.
 			</div>
-		{:else if scriptRes.loading && !script}
+		{:else if !isDraft && scriptRes.loading && !script}
 			<div class="absolute inset-0 flex items-center justify-center gap-2 text-tertiary">
 				<Loader2 size={16} class="animate-spin" />
 				<span class="text-xs">Loading script…</span>
 			</div>
-		{:else if scriptRes.error}
+		{:else if !isDraft && scriptRes.error}
 			<div class="p-3 text-xs text-red-500">
 				Failed to load: {scriptRes.error.message}
 			</div>
 		{:else if script}
-			{#key script.hash}
+			{#key (script.hash ?? 'draft') + script.language}
 				<ScriptEditor
 					showCaptures={false}
 					noSyncFromGithub

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

@@ -4,34 +4,106 @@
 	import AssetGenericIcon from '$lib/components/icons/AssetGenericIcon.svelte'
 	import { formatShortAssetPath, type AssetKind } from '$lib/components/assets/lib'
 	import { NODE } from '$lib/components/graph/util'
+	import PipelineInsertMenu, { type PipelineInsertPick } from './PipelineInsertMenu.svelte'
+	import { Code2, Plus } from 'lucide-svelte'
+	import type { ScriptLang } from '$lib/gen'
 
 	interface Props {
-		data: { asset_kind: AssetKind; path: string }
+		data: {
+			asset_kind: AssetKind
+			path: string
+			onAddScript?: (
+				asset: { kind: AssetKind; path: string },
+				language: ScriptLang,
+				scriptPath: string
+			) => void
+			pathPrefix?: string
+			defaultPathSuffix?: string
+		}
 	}
 	let { data }: Props = $props()
 
 	let asset = $derived({ kind: data.asset_kind, path: data.path })
+
+	const LANGUAGES: Array<{ label: string; lang: ScriptLang }> = [
+		{ label: 'TypeScript (Bun)', lang: 'bun' },
+		{ label: 'TypeScript (Deno)', lang: 'deno' },
+		{ label: 'Python', lang: 'python3' },
+		{ label: 'PostgreSQL', lang: 'postgresql' },
+		{ label: 'DuckDB', lang: 'duckdb' },
+		{ label: 'BigQuery', lang: 'bigquery' },
+		{ label: 'Snowflake', lang: 'snowflake' },
+		{ label: 'MySQL', lang: 'mysql' },
+		{ label: 'MS SQL', lang: 'mssql' },
+		{ label: 'Bash', lang: 'bash' },
+		{ label: 'Go', lang: 'go' }
+	]
+
+	function handlePick(pick: PipelineInsertPick) {
+		if (pick.kindId === 'materializer' && pick.language && pick.path) {
+			data.onAddScript?.(
+				{ kind: data.asset_kind, path: data.path },
+				pick.language as ScriptLang,
+				pick.path
+			)
+		}
+	}
+
+	let showAdd = $derived(data.onAddScript != undefined)
 </script>
 
 <div class="relative">
 	<div
 		class={twMerge(
-			'flex items-center rounded-md drop-shadow-sm overflow-hidden',
-			'bg-surface-secondary outline outline-1 outline-transparent hover:outline-blue-400 transition-colors'
+			'flex items-center rounded-md drop-shadow-sm overflow-hidden border',
+			'bg-surface-secondary border-gray-300 dark:border-gray-600 hover:border-blue-400 transition-colors'
 		)}
-		style="width: {NODE.width}px; min-height: {NODE.height + 30}px;"
+		style="width: {NODE.width}px; min-height: {NODE.height}px;"
 		title={data.path}
 	>
 		<AssetGenericIcon
 			assetKind={data.asset_kind}
 			class="shrink-0 ml-2 mr-2 text-blue-600 dark:text-blue-400"
-			size="16px"
+			size="14px"
 		/>
-		<div class="flex flex-col min-w-0 flex-1 pr-2 py-1.5">
+		<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">{data.asset_kind}</span>
 			<span class="text-2xs font-mono text-emphasis truncate">{formatShortAssetPath(asset)}</span>
 		</div>
 	</div>
+	{#if showAdd}
+		<!-- Always-visible + below the asset for downstream materializer
+		     creation. Half-overlapping the bottom edge so it visually attaches
+		     to the node like the flow editor's between-step inserter. -->
+		<div class="absolute left-1/2 -bottom-3 -translate-x-1/2 z-10">
+			<PipelineInsertMenu
+				kinds={[
+					{
+						id: 'materializer',
+						label: 'Add downstream materializer',
+						description: 'Triggered when this asset changes',
+						icon: Code2,
+						pickLanguage: true
+					}
+				]}
+				languages={LANGUAGES as any}
+				pathPrefix={data.pathPrefix ?? ''}
+				defaultPathSuffix={data.defaultPathSuffix ?? ''}
+				onPick={handlePick}
+			>
+				{#snippet trigger()}
+					<button
+						type="button"
+						onclick={(e) => e.stopPropagation()}
+						class="bg-emerald-500 hover:bg-emerald-600 text-white rounded-full w-5 h-5 flex items-center justify-center shadow border-2 border-surface-secondary"
+						title="Add downstream materializer"
+					>
+						<Plus size={12} />
+					</button>
+				{/snippet}
+			</PipelineInsertMenu>
+		</div>
+	{/if}
 </div>
 
 <Handle type="target" position={Position.Top} isConnectable={false} />

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

@@ -1,33 +1,44 @@
 <script lang="ts">
 	import { Handle, Position } from '@xyflow/svelte'
-	import { Code2, GitBranch } from 'lucide-svelte'
+	import { Code2, GitBranch, Sparkles } from 'lucide-svelte'
 	import { twMerge } from 'tailwind-merge'
 	import type { GraphUsageKind } from './types'
 	import { NODE } from '$lib/components/graph/util'
 
 	interface Props {
-		data: { runnable_kind: GraphUsageKind; path: string }
+		data: { runnable_kind: GraphUsageKind; path: string; is_materializer?: boolean }
 	}
 	let { data }: Props = $props()
 
 	let Icon = $derived(data.runnable_kind === 'flow' ? GitBranch : Code2)
-	let label = $derived(data.runnable_kind === 'flow' ? 'flow' : 'script')
+	let label = $derived(
+		data.is_materializer ? 'materializer' : data.runnable_kind === 'flow' ? 'flow' : 'script'
+	)
 </script>
 
 <div class="relative">
 	<div
 		class={twMerge(
-			'flex items-center rounded-md drop-shadow-sm overflow-hidden',
-			'bg-surface-tertiary outline outline-1 outline-transparent hover:outline-emerald-500 transition-colors'
+			'flex items-center rounded-md drop-shadow-sm overflow-hidden border',
+			'bg-surface-tertiary border-gray-300 dark:border-gray-600 hover:border-emerald-500 transition-colors',
+			data.is_materializer && 'border-emerald-400/60'
 		)}
-		style="width: {NODE.width}px; min-height: {NODE.height + 30}px;"
+		style="width: {NODE.width}px; min-height: {NODE.height}px;"
 		title={data.path}
 	>
-		<Icon size={16} class="shrink-0 ml-2 mr-2 text-emerald-700 dark:text-emerald-400" />
-		<div class="flex flex-col min-w-0 flex-1 pr-2 py-1.5">
+		<Icon size={14} class="shrink-0 ml-2 mr-2 text-emerald-700 dark:text-emerald-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 text-tertiary truncate">{label}</span>
 			<span class="text-2xs font-mono text-emphasis truncate">{data.path}</span>
 		</div>
+		{#if data.is_materializer}
+			<div
+				class="shrink-0 flex items-center gap-0.5 px-1.5 py-0.5 mr-1.5 rounded-sm bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300"
+				title="// materialize — pipeline member"
+			>
+				<Sparkles size={10} />
+			</div>
+		{/if}
 	</div>
 </div>
 

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

@@ -10,20 +10,66 @@ export interface AssetGraphAssetNode {
 export interface AssetGraphRunnableNode {
 	path: string
 	usage_kind: GraphUsageKind
+	// Script has `// materialize` annotation. Drives the pipeline-member
+	// visual state; unrelated to what the script actually writes (that's
+	// parsed separately into lineage edges).
+	is_materializer?: boolean
 }
 
+// Lineage edge from parsed r/w usages — informational only, not the
+// execution DAG. `unsaved: true` for edges synthesized by a draft overlay
+// (e.g. the random output asset attached at draft creation).
 export interface AssetGraphEdge {
 	runnable_path: string
 	runnable_kind: GraphUsageKind
 	asset_kind: AssetKind
 	asset_path: string
 	access_type: 'r' | 'w' | 'rw' | null
+	unsaved?: boolean
 }
 
+// Declared `// on <trigger>` — the actual execution DAG edges.
+// `unsaved: true` marks overlays computed live from editor buffer that
+// haven't been persisted to script_trigger yet.
+export type NativeTriggerKind =
+	| 'webhook'
+	| 'email'
+	| 'kafka'
+	| 'mqtt'
+	| 'nats'
+	| 'postgres'
+	| 'sqs'
+	| 'gcp'
+
+export type AssetGraphTrigger =
+	| {
+			trigger_kind: 'asset'
+			asset_kind: AssetKind
+			asset_path: string
+			runnable_kind: GraphUsageKind
+			runnable_path: string
+			unsaved?: boolean
+	  }
+	| {
+			trigger_kind: 'schedule'
+			cron: string
+			runnable_kind: GraphUsageKind
+			runnable_path: string
+			unsaved?: boolean
+	  }
+	| {
+			trigger_kind: NativeTriggerKind
+			path: string
+			runnable_kind: GraphUsageKind
+			runnable_path: string
+			unsaved?: boolean
+	  }
+
 export interface AssetGraphResponse {
 	assets: AssetGraphAssetNode[]
 	runnables: AssetGraphRunnableNode[]
 	edges: AssetGraphEdge[]
+	triggers: AssetGraphTrigger[]
 }
 
 export type AssetGraphNodeData =

frontend/src/lib/components/home/ItemsList.svelte

@@ -9,8 +9,10 @@
 		type Script,
 		ScriptService,
 		type Flow,
-		type ListableRawApp
+		type ListableRawApp,
+		OpenAPI
 	} from '$lib/gen'
+	import { resource } from 'runed'
 	import { userStore, workspaceStore } from '$lib/stores'
 	import type uFuzzy from '@leeoniya/ufuzzy'
 	import {
@@ -70,6 +72,25 @@
 	type TableApp = TableItem<ListableApp, 'app'>
 	type TableRawApp = TableItem<ListableRawApp, 'raw_app'>
 
+	// Folders with ≥1 materializer script (auto_kind='materializer'). Used by
+	// TreeView to surface a "Pipeline" entry inside those folders. Cheap
+	// thanks to the partial index on script.auto_kind.
+	let pipelineFoldersRes = resource(
+		() => $workspaceStore,
+		async (ws, _prev, { signal }) => {
+			if (!ws) return new Set<string>()
+			const base_url = OpenAPI.BASE ?? ''
+			const res = await fetch(`${base_url}/w/${ws}/assets/pipelines`, {
+				credentials: 'include',
+				signal
+			})
+			if (!res.ok) return new Set<string>()
+			const rows = (await res.json()) as Array<{ folder: string }>
+			return new Set(rows.map((r) => r.folder))
+		}
+	)
+	let pipelineFolders = $derived(pipelineFoldersRes.current ?? new Set<string>())
+
 	let scripts: TableScript[] | undefined = $state()
 	let flows: TableFlow[] | undefined = $state()
 	let apps: TableApp[] | undefined = $state()
@@ -557,6 +578,7 @@
 				{items}
 				{nbDisplayed}
 				{collapseAll}
+				{pipelineFolders}
 				isSearching={filter !== ''}
 				on:scriptChanged={() => loadScripts(includeWithoutMain)}
 				on:flowChanged={loadFlows}

frontend/src/lib/components/home/TreeView.svelte

@@ -1,11 +1,12 @@
 <script lang="ts">
 	import TreeView from './TreeView.svelte'
 
-	import { ChevronDown, ChevronUp, Folder, FolderTree, User } from 'lucide-svelte'
+	import { ChevronDown, ChevronUp, Folder, FolderTree, NetworkIcon, User } from 'lucide-svelte'
 	import Item from './Item.svelte'
 	import type { FolderItem, ItemType, UserItem } from './treeViewUtils'
 	import { twMerge } from 'tailwind-merge'
 	import { pluralize } from '$lib/utils'
+	import { base } from '$lib/base'
 
 	interface Props {
 		item: ItemType | FolderItem | UserItem
@@ -13,11 +14,24 @@
 		depth?: number
 		showCode: (path: string, summary: string) => void
 		isSearching?: boolean
+		pipelineFolders?: Set<string>
 	}
 
-	let { item, collapseAll, depth = 0, showCode, isSearching = false }: Props = $props()
+	let {
+		item,
+		collapseAll,
+		depth = 0,
+		showCode,
+		isSearching = false,
+		pipelineFolders
+	}: Props = $props()
 
-	const isFolder = (i: typeof item): i is FolderItem => i && 'folderName' in i
+	const isFolderItem = (i: typeof item): i is FolderItem => i && 'folderName' in i
+	let hasPipeline = $derived(
+		depth === 0 && isFolderItem(item) && (pipelineFolders?.has(item.folderName) ?? false)
+	)
+
+	const isFolder = isFolderItem
 	const isUser = (i: typeof item): i is UserItem => i && 'username' in i
 
 	let opened: boolean = $state(true)
@@ -67,11 +81,25 @@
 		</div>
 		{#if opened || isSearching}
 			<div>
+				{#if hasPipeline && isFolder(item)}
+					<a
+						href="{base}/pipeline/{encodeURIComponent(item.folderName)}"
+						class="flex items-center gap-4 px-4 py-2 border-b text-sm hover:bg-surface-hover transition-colors"
+						style="padding-left: {(depth + 1) * 16}px;"
+					>
+						<NetworkIcon size={16} class="text-emerald-600 dark:text-emerald-400" />
+						<div class="flex flex-col">
+							<span class="text-xs font-medium text-emphasis">Pipeline</span>
+							<span class="text-2xs text-secondary">Open pipeline editor</span>
+						</div>
+					</a>
+				{/if}
 				{#each item.items.slice(0, showMax) as subItem, index ((subItem['path'] ? subItem['type'] + '__' + subItem['path'] + '__' + index : undefined) ?? 'folder__' + subItem['folderName'] + '__' + index)}
 					<TreeView
 						{isSearching}
 						{collapseAll}
 						item={subItem}
+						{pipelineFolders}
 						on:scriptChanged
 						on:flowChanged
 						on:appChanged

frontend/src/lib/components/home/TreeViewRoot.svelte

@@ -9,6 +9,7 @@
 		nbDisplayed: number
 		items: ItemType[] | undefined
 		isSearching?: boolean
+		pipelineFolders?: Set<string>
 	}
 
 	let {
@@ -16,7 +17,8 @@
 		showCode,
 		nbDisplayed = $bindable(),
 		items,
-		isSearching = false
+		isSearching = false,
+		pipelineFolders
 	}: Props = $props()
 
 	let groupedItems: ReturnType<typeof groupItems> | 'loading' = $state('loading')
@@ -43,6 +45,7 @@
 					{isSearching}
 					{collapseAll}
 					{item}
+					{pipelineFolders}
 					on:scriptChanged
 					on:flowChanged
 					on:appChanged
@@ -56,7 +59,10 @@
 	{#if groupedItems.length > 15 && nbDisplayed < groupedItems.length}
 		<span class="text-xs font-normal text-secondary"
 			>{nbDisplayed} root nodes out of {groupedItems.length}
-			<button class="ml-4 text-xs font-normal text-primary hover:text-emphasis" onclick={() => (nbDisplayed += 30)}>load 30 more</button></span
+			<button
+				class="ml-4 text-xs font-normal text-primary hover:text-emphasis"
+				onclick={() => (nbDisplayed += 30)}>load 30 more</button
+			></span
 		>
 	{/if}
 {/if}

frontend/src/routes/(root)/(logged)/+page.svelte

@@ -20,7 +20,8 @@
 		Globe2,
 		Loader2,
 		Code,
-		LayoutDashboard
+		LayoutDashboard,
+		NetworkIcon
 	} from 'lucide-svelte'
 	import { hubBaseUrlStore } from '$lib/stores'
 	import { base } from '$lib/base'
@@ -117,7 +118,8 @@
 			setTimeout(() => {
 				workspaceTutorials?.runTutorialById('workspace-onboarding')
 			}, 500)
-		} else if (tutorialParam === 'workspace-onboarding-operator') { // Small delay to ensure page is fully loaded
+		} else if (tutorialParam === 'workspace-onboarding-operator') {
+			// Small delay to ensure page is fully loaded
 			setTimeout(() => {
 				workspaceTutorials?.runTutorialById('workspace-onboarding-operator')
 			}, 500)
@@ -284,16 +286,26 @@
 			childrenWrapperDivClasses="flex-1 flex flex-row gap-4 flex-wrap justify-end items-center"
 		>
 			{#if !$userStore?.operator && showCreateButtons}
-				<span class="text-xs font-normal text-primary">Create a</span>
 				<CreateActionsScript aiId="create-script-button" aiDescription="Creates a new script" />
 				{#if HOME_SHOW_CREATE_FLOW}<CreateActionsFlow />{/if}
+				<Button
+					id="pipeline-button"
+					aiId="pipeline-button"
+					aiDescription="Opens the pipeline editor"
+					unifiedSize="lg"
+					startIcon={{ icon: NetworkIcon }}
+					href="{base}/pipeline"
+					variant="accent"
+				>
+					Pipeline
+				</Button>
 				{#if HOME_SHOW_CREATE_APP}<CreateActionsApp />{/if}
 			{/if}
 		</PageHeader>
 
 		<TutorialBanner />
 
-		<NoDirectDeployAlert onUpdateCanEditStatus={(v) => showCreateButtons = v}/>
+		<NoDirectDeployAlert onUpdateCanEditStatus={(v) => (showCreateButtons = v)} />
 
 		{#if !$userStore?.operator}
 			<div class="w-full overflow-auto scrollbar-hidden pb-2">

frontend/src/routes/(root)/(logged)/assets/+page.svelte

@@ -161,10 +161,10 @@
 				<Button
 					variant="accent-secondary"
 					unifiedSize="sm"
-					href="{base}/assets/graph"
+					href="{base}/pipeline"
 					startIcon={{ icon: NetworkIcon }}
 				>
-					Graph view
+					Pipelines
 				</Button>
 			</div>
 		</PageHeader>

frontend/src/routes/(root)/(logged)/assets/graph/+page.js

@@ -1,5 +0,0 @@
-export function load() {
-	return {
-		stuff: { title: 'Asset Graph' }
-	}
-}

frontend/src/routes/(root)/(logged)/assets/graph/+page.svelte

@@ -1,141 +0,0 @@
-<script lang="ts">
-	import { workspaceStore, userStore } from '$lib/stores'
-	import { base } from '$lib/base'
-	import Button from '$lib/components/common/button/Button.svelte'
-	import Toggle from '$lib/components/Toggle.svelte'
-	import AssetGraphCanvas from '$lib/components/assets/AssetGraph/AssetGraphCanvas.svelte'
-	import AssetGraphDetailsPane from '$lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte'
-	import type {
-		AssetGraphResponse,
-		AssetGraphSelection
-	} from '$lib/components/assets/AssetGraph/types'
-	import { ArrowLeft, Loader2, NetworkIcon, RefreshCw } from 'lucide-svelte'
-	import { OpenAPI } from '$lib/gen'
-	import { resource } from 'runed'
-	import { Pane, Splitpanes } from 'svelte-splitpanes'
-
-	// Variables and resources tend to be hubs (DB creds, API keys) used by
-	// most runnables, so they swamp the layout. Hidden by default; the
-	// toggle in the header opts back in.
-	const DATA_KINDS = ['s3object', 'ducklake', 'datatable', 'volume']
-
-	let includeConfigKinds = $state(false)
-	let selection = $state<AssetGraphSelection | undefined>(undefined)
-
-	let graphRes = resource(
-		[() => $workspaceStore, () => includeConfigKinds],
-		async ([ws, includeAll], _prev, { signal }) => {
-			if (!ws) return undefined
-			const base_url = OpenAPI.BASE ?? ''
-			const qs = includeAll ? '' : `?asset_kinds=${DATA_KINDS.join(',')}`
-			const res = await fetch(`${base_url}/w/${ws}/assets/graph${qs}`, {
-				credentials: 'include',
-				signal
-			})
-			if (!res.ok) throw new Error(`GET /assets/graph → ${res.status}`)
-			return (await res.json()) as AssetGraphResponse
-		}
-	)
-
-	// Pluralize the kind label: 1 script, 2 scripts, 1 s3object, 2 s3objects.
-	function pluralize(n: number, singular: string): string {
-		return `${n} ${singular}${n === 1 ? '' : 's'}`
-	}
-
-	let summary = $derived.by<string[]>(() => {
-		const g = graphRes.current
-		if (!g) return []
-		const parts: string[] = []
-		const scripts = g.runnables.filter((r) => r.usage_kind === 'script').length
-		const flows = g.runnables.filter((r) => r.usage_kind === 'flow').length
-		if (scripts) parts.push(pluralize(scripts, 'script'))
-		if (flows) parts.push(pluralize(flows, 'flow'))
-		const byKind = new Map<string, number>()
-		for (const a of g.assets) byKind.set(a.kind, (byKind.get(a.kind) ?? 0) + 1)
-		for (const [kind, n] of byKind) parts.push(pluralize(n, kind))
-		return parts
-	})
-</script>
-
-<svelte:head>
-	<title>Asset graph — Windmill</title>
-</svelte:head>
-
-{#if $userStore?.operator}
-	<div class="p-8 text-tertiary">Page not available for operators.</div>
-{:else}
-	<div class="flex flex-col h-full">
-		<div
-			class="border-b flex flex-row justify-between gap-2 px-2 py-1 items-center overflow-y-visible overflow-x-auto min-h-10 shrink-0 whitespace-nowrap"
-		>
-			<div class="flex flex-row items-center gap-2">
-				<Button
-					variant="subtle"
-					unifiedSize="sm"
-					href="{base}/assets"
-					startIcon={{ icon: ArrowLeft }}
-					iconOnly
-					title="Back to assets"
-				/>
-				<NetworkIcon size={16} class="text-tertiary shrink-0" />
-				<h1 class="text-sm font-semibold">Asset graph</h1>
-				{#if summary.length > 0}
-					<span class="text-xs text-tertiary">· {summary.join(' · ')}</span>
-				{/if}
-			</div>
-			<div class="flex flex-row items-center gap-2">
-				<Toggle
-					bind:checked={includeConfigKinds}
-					size="xs"
-					options={{ right: 'Vars & resources' }}
-					disabled={graphRes.loading}
-				/>
-				<Button
-					variant="subtle"
-					unifiedSize="sm"
-					startIcon={{ icon: RefreshCw }}
-					onclick={() => graphRes.refetch()}
-					disabled={graphRes.loading}
-					iconOnly
-					title="Refresh"
-				/>
-			</div>
-		</div>
-
-		<div class="flex-1 min-h-0">
-			{#if graphRes.loading && !graphRes.current}
-				<div class="h-full flex items-center justify-center gap-2 text-tertiary">
-					<Loader2 size={18} class="animate-spin" />
-					<span>Loading graph…</span>
-				</div>
-			{:else if graphRes.error}
-				<div class="h-full flex items-center justify-center text-red-500 text-sm">
-					Failed to load graph: {graphRes.error.message}
-				</div>
-			{:else if graphRes.current && graphRes.current.assets.length === 0 && graphRes.current.runnables.length === 0}
-				<div class="h-full flex items-center justify-center text-tertiary text-sm">
-					No assets are referenced by scripts or flows in this workspace yet.
-				</div>
-			{:else if graphRes.current}
-				<Splitpanes class="!h-full">
-					<Pane size={selection ? 60 : 100}>
-						<AssetGraphCanvas
-							graph={graphRes.current}
-							{selection}
-							onselect={(s) => (selection = s)}
-						/>
-					</Pane>
-					{#if selection && $workspaceStore}
-						<Pane size={40} minSize={25}>
-							<AssetGraphDetailsPane
-								{selection}
-								workspace={$workspaceStore}
-								onclose={() => (selection = undefined)}
-							/>
-						</Pane>
-					{/if}
-				</Splitpanes>
-			{/if}
-		</div>
-	</div>
-{/if}

frontend/src/routes/(root)/(logged)/scripts/add/+page.svelte

@@ -33,6 +33,14 @@
 	const collabLang = page.url.searchParams.get('lang') as ScriptLang | null
 	const wacParam = page.url.searchParams.get('wac')
 	const importParam = page.url.searchParams.get('import')
+	// Pipeline opt-in marker. Any truthy value prefills `<comment-prefix>
+	// materialize` at the top of the new script so the bare marker fires on
+	// deploy. ScriptBuilder's initialCode path only runs when content is
+	// empty, so this survives.
+	const materializeParam = page.url.searchParams.get('materialize')
+	// Optional `// on <asset>` trigger. Value is the full asset-syntax string
+	// (e.g. s3://bucket/key) — written verbatim after the `materialize` line.
+	const onAssetParam = page.url.searchParams.get('on_asset')
 
 	let initialArgs = urlArgs ? decodeState(urlArgs) : (get(initialArgsStore) ?? {})
 	if (get(initialArgsStore)) $initialArgsStore = undefined
@@ -53,19 +61,61 @@
 		}
 	}
 
+	// Language → comment prefix recognized by the pipeline annotation
+	// parser. Any of //, #, -- is accepted.
+	function commentPrefix(lang: ScriptLang): string {
+		switch (lang) {
+			case 'python3':
+			case 'bash':
+			case 'powershell':
+			case 'nu':
+			case 'ansible':
+				return '#'
+			case 'postgresql':
+			case 'mysql':
+			case 'bigquery':
+			case 'snowflake':
+			case 'mssql':
+			case 'oracledb':
+			case 'duckdb':
+				return '--'
+			default:
+				return '//'
+		}
+	}
+
+	// Compose the prefix block for a new pipeline script: bare `// materialize`
+	// marker plus optional `// on <ref>` trigger. Placed at the very top so
+	// comments precede any language-specific preamble the builder injects.
+	function pipelinePreamble(lang: ScriptLang): string {
+		const p = commentPrefix(lang)
+		const lines: string[] = []
+		if (materializeParam) lines.push(`${p} materialize`)
+		if (onAssetParam) lines.push(`${p} on ${onAssetParam}`)
+		return lines.length ? lines.join('\n') + '\n' : ''
+	}
+
 	function defaultScript(): Script {
+		const language = ((wacParam === 'python'
+			? 'python3'
+			: wacParam === 'typescript'
+				? 'bun'
+				: null) ??
+			collabLang ??
+			$defaultScripts?.order?.filter(
+				(x) => $defaultScripts?.hidden == undefined || !$defaultScripts.hidden.includes(x)
+			)?.[0] ??
+			'bun') as ScriptLang
 		return {
 			hash: '',
 			path: path ?? '',
 			summary: '',
-			content: '',
+			content: pipelinePreamble(language),
 			description: '',
 			schema: schema,
 			is_template: false,
 			extra_perms: {},
-			language: (wacParam === 'python' ? 'python3' : wacParam === 'typescript' ? 'bun' : null) ?? collabLang ?? ($defaultScripts?.order?.filter(
-				(x) => $defaultScripts?.hidden == undefined || !$defaultScripts.hidden.includes(x)
-			)?.[0] ?? 'bun') as ScriptLang,
+			language,
 			kind: 'script'
 		}
 	}
@@ -139,8 +189,7 @@
 			extra_perms: {}
 		}
 		if (isWac) {
-			importedWacTemplate =
-				imported.language === 'python3' ? 'wac_python' : 'wac_typescript'
+			importedWacTemplate = imported.language === 'python3' ? 'wac_python' : 'wac_typescript'
 			sendUserToast('WAC script loaded from YAML/JSON')
 		} else {
 			sendUserToast('Script loaded from YAML/JSON')
@@ -159,7 +208,12 @@
 		{initialArgs}
 		bind:this={scriptBuilder}
 		lockedLanguage={templatePath != null || hubPath != null}
-		template={importedWacTemplate ?? (wacParam === 'python' ? 'wac_python' : wacParam === 'typescript' ? 'wac_typescript' : 'script')}
+		template={importedWacTemplate ??
+			(wacParam === 'python'
+				? 'wac_python'
+				: wacParam === 'typescript'
+					? 'wac_typescript'
+					: 'script')}
 		onDeploy={(e) => {
 			goto(`/scripts/get/${e.hash}?workspace=${$workspaceStore}`)
 		}}