feat(pipelines): auto-derive cascade edges from ducklake/s3 reads (+ muted-read badge) (#9963)

* feat(pipelines): auto-derive cascade trigger edges from ducklake/s3 reads

Within a `// pipeline`, a read of a ducklake table or s3 object now
auto-wires its cascade trigger edge straight from the FROM clause, so
`// on <asset>` is only needed for edges inference can't see (dynamic SQL)
or to carry per-edge opts. Two opt-outs: `// mute <asset>` suppresses a
single derived edge (a lookup / SCD input read every run but not cascaded
on), and `// mute all` opts the script out of derivation entirely (back to
explicit-`// on`-only). Explicit `// on` still wins the dedup.

Scoped to ducklake + s3 reads; resource/datatable/volume stay explicit.
Read-write (RW) and write inputs are excluded so a self-referential
merge can't loop-trigger itself; ambiguous (None) access is skipped.

- parser: `mute` / `mute_all` in PipelineAnnotations (Rust + TS mirror)
- deploy: derive_pipeline_asset_trigger_refs → script_trigger rows
- frontend: resolveGraph mirrors derivation for the live edit-mode canvas
- tests: shared parity corpus + derive-helper units + resolveGraph overlays

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

* feat(pipelines): mark auto-derived cascade edges with a persisted derived flag + "auto" badge

Persist script_trigger.derived (deploy: true for ducklake/s3-read derivation,
false for explicit // on) and return it from the asset-graph endpoint so the
canvas renders a Sparkles "auto" badge on auto-wired edges — the inference is
now visible on both the deployed graph and the live edit canvas, not just
implied. Dispatch (fetch_subscribers) ignores the flag, so a derived edge fires
identically to an explicit // on. Also copy derived in the workspace-clone
trigger copy, and backfill muteAssets/muteAll into two empty PipelineAnnotations
literals the base commit left stale (check:fast).

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

* fix(pipelines): derive cascade edge from effective (alt-fallback) asset access

derive_pipeline_asset_trigger_refs gated on the raw parser access_type, but the
persisted asset.usage_access_type and the frontend canvas both use
access_type.or(alt_access_type). An ambiguous parse with a manual read override
was persisted/drawn as a read yet derived no edge, so the auto edge silently
vanished on deploy. Gate on the effective access type for parity.

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

* feat(pipelines): badge muted reads instead of auto-derived edges

Auto-derivation is the default now, so badging every derived cascade edge is
noise. Drop the "auto" badge and the persisted `script_trigger.derived` flag
(migration + insert param + graph field + clone copy) that only powered it, and
instead badge the exception: a ducklake/s3 asset a script reads but does NOT
cascade — `// mute <asset>` / `// mute all`. `computeMutedReadKeys` marks a
read-only ('r') supported read with no cascade trigger and no self-write; the
canvas renders a bell-off "muted" badge on that read edge.

Also fixes two review parity nits:
- TS `// on` parser now strips trailing `key=value` opts (e.g. `debounce=60s`)
  like the Rust `split_trailing_kv_opts`, so the ref dedups against inference.
- A `// materialize` producer reading its own target is upgraded to `rw`
  (deploy) / excluded via the materialize write refs (canvas), so it neither
  self-cascades nor shows as a muted read.

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

* fix(pipelines): drop redundant // on for auto-derived reads; gate muted badge to pipeline scripts

- Templates no longer scaffold `// on <asset>` for a ducklake/s3 input the body
  reads — the read auto-wires the cascade now that derivation is the default.
  Kept for datatable/resource (not auto-derived) and native triggers. The
  discoverability hint now mentions `// mute` (the newly relevant annotation).
- computeMutedReadKeys only badges reads by `// pipeline` scripts. A plain
  script or flow reading a ducklake/s3 asset never had an auto trigger to
  suppress, so it must render as ordinary lineage, not "muted" (Codex review).

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

* fix(pipelines): only drop template // on when the body actually reads the input

The redundant-`// on` removal assumed the generated body reads the ducklake/s3
input, but postgres/bash/generic bodies (and `data_upload`, which reads the
picker file) ignore `input` — dropping `// on` there left the asset-created
script with no cascade at all. Gate the drop on READS_INPUT_LANGS
(bun/deno/python/duckdb) so non-reading templates keep the explicit trigger.

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

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-07-06 18:12:12 +00:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 91e1b087a2
commit 3dcd3949a1
16 changed files with 950 additions and 19 deletions
@@ -528,6 +528,16 @@ pub struct PipelineAnnotations {
pub column_lineage: Vec<ColumnLineage>,
pub macros: bool,
pub use_libs: Vec<String>,
// `// mute <asset>` — suppress the auto-derived cascade edge for a read
// that would otherwise trigger this script (a lookup / slowly-changing
// dimension you read every run but don't want to re-run on). Only Asset
// specs are stored; native trigger kinds are never auto-derived, so
// muting them is meaningless.
pub mute: Vec<TriggerSpec>,
// `// mute all` — opt out of auto-derivation entirely for this script.
// Falls back to explicit-`// on`-only semantics. Explicit `// on` edges
// are unaffected.
pub mute_all: bool,
}
impl ParseAssetsOutput {
@@ -956,6 +966,23 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations {
continue;
}
// `// mute all` opts out of auto-derived cascade edges entirely;
// `// mute <asset>` suppresses the one edge. Only asset refs are
// muteable — native trigger kinds are never auto-derived. Checked
// before the generic `on`/asset shorthand (a complete word, so
// prose like `// muted for now` never matches).
if let Some(after_kw) = consume_keyword(rest, "mute") {
let arg = after_kw.trim();
if arg == "all" {
out.mute_all = true;
} else if let Some(spec @ TriggerSpec::Asset { .. }) = parse_trigger_spec(arg) {
if !out.mute.contains(&spec) {
out.mute.push(spec);
}
}
continue;
}
// `data_test` is checked before `on`/asset shorthands and is a complete
// word (so it never collides with the `// test:` CI annotation, which
// has no whitespace after `test`). Accumulates — every well-formed line
@@ -769,5 +769,109 @@
"tag": null,
"retry": null
}
},
{
"name": "mute suppresses a single ducklake read edge",
"code": "// pipeline\n// mute ducklake://main.orders\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null,
"mute": [
"ducklake:main.orders"
]
}
},
{
"name": "mute all opts out of all auto-derivation",
"code": "// pipeline\n// mute all\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null,
"mute_all": true
}
},
{
"name": "mute accumulates in order and dedups",
"code": "// pipeline\n// mute ducklake://main.a\n// mute s3://raw/b\n// mute ducklake://main.a\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null,
"mute": [
"ducklake:main.a",
"s3object:raw/b"
]
}
},
{
"name": "mute of a native trigger kind is dropped (only assets are muteable)",
"code": "// pipeline\n// mute kafka\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null
}
},
{
"name": "mute prose without an asset ref never false-positives",
"code": "// pipeline\n// muted for now\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null
}
},
{
"name": "mute all coexists with explicit on edges",
"code": "// pipeline\n// mute all\n// on ducklake://main.orders\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [
"ducklake:main.orders"
],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null,
"mute_all": true
}
},
{
"name": "on asset ref strips trailing key=value opts",
"code": "// pipeline\n// on ducklake://main.orders debounce=60s\nselect 1",
"expected": {
"in_pipeline": true,
"asset_triggers": [
"ducklake:main.orders"
],
"native_triggers": [],
"partition": null,
"freshness": null,
"tag": null,
"retry": null
}
}
]
@@ -53,6 +53,13 @@ struct Expected {
// `// use <lib_path>` accumulation, declaration order, deduped. Absent === [].
#[serde(default)]
use_libs: Vec<String>,
// `// mute <asset>` accumulation as `kind:path`, declaration order, deduped.
// Absent === [].
#[serde(default)]
mute: Vec<String>,
// `// mute all` marker. Absent === false.
#[serde(default)]
mute_all: bool,
}
#[derive(Deserialize)]
@@ -251,5 +258,18 @@ fn pipeline_annotation_fixtures_match() {
assert_eq!(got.macros, f.expected.macros, "{ctx}: macros");
assert_eq!(got.use_libs, f.expected.use_libs, "{ctx}: use_libs");
let mute: Vec<String> = got
.mute
.iter()
.filter_map(|t| match t {
TriggerSpec::Asset { asset_kind, path, .. } => {
Some(format!("{}:{}", kind_str(*asset_kind), path))
}
_ => None,
})
.collect();
assert_eq!(mute, f.expected.mute, "{ctx}: mute");
assert_eq!(got.mute_all, f.expected.mute_all, "{ctx}: mute_all");
}
}
+73 -4
View File
@@ -45,8 +45,9 @@ use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap;
use windmill_common::{
assets::{
clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash,
insert_script_trigger, parse_duration_secs, parse_pipeline_annotations,
replace_static_asset_usage, trigger_spec_to_row, AssetUsageKind, TriggerSpec,
derive_pipeline_asset_trigger_refs, insert_script_trigger, parse_duration_secs,
parse_pipeline_annotations, replace_static_asset_usage, trigger_spec_to_row,
AssetUsageKind, ScriptTriggerKind, TriggerSpec,
},
error::{self, to_anyhow},
min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2},
@@ -1457,11 +1458,22 @@ async fn create_script_internal<'c>(
// fire and the view would be an orphan node in the lineage graph.
for (target_kind, path) in m.write_targets() {
let kind = windmill_common::assets::asset_kind_from_parser(target_kind);
if !a.iter().any(|x| x.kind == kind && x.path == path) {
use windmill_common::assets::AssetUsageAccessType;
if let Some(existing) = a.iter_mut().find(|x| x.kind == kind && x.path == path) {
// The body reads its own managed target (an incremental/merge
// model `SELECT`ing from the table it materializes). The runtime
// still generates the write, so the effective access is RW — mark
// it so, otherwise it stays a plain read and (a) auto-derives a
// self-cascade edge back to this producer and (b) shows as a muted
// read on the canvas. Both are wrong: it's the script's own output.
if existing.access_type != Some(AssetUsageAccessType::W) {
existing.access_type = Some(AssetUsageAccessType::RW);
}
} else {
a.push(windmill_common::assets::AssetWithAltAccessType {
path,
kind,
access_type: Some(windmill_common::assets::AssetUsageAccessType::W),
access_type: Some(AssetUsageAccessType::W),
alt_access_type: None,
columns: None,
});
@@ -2035,6 +2047,63 @@ async fn create_script_internal<'c>(
.await?;
}
// Auto-derived cascade edges: within a `// pipeline`, a read of a ducklake
// table or s3 object wires the cascade edge straight from the FROM clause,
// so `// on <asset>` is only needed for edges inference can't see (dynamic
// SQL) or to carry per-edge opts. `// mute <asset>` / `// mute all` opt out.
// Explicit `// on` asset edges (inserted just above) win the dedup — they
// carry the per-edge debounce, so a derived row must not shadow them.
if in_pipeline && !pipeline_annotations.mute_all {
let asset_ref = |spec: &TriggerSpec| {
trigger_spec_to_row(spec)
.filter(|(k, _)| *k == ScriptTriggerKind::Asset)
.map(|(_, r)| r)
};
let explicit_refs: std::collections::HashSet<String> =
pipeline_triggers.iter().filter_map(asset_ref).collect();
let muted_refs: std::collections::HashSet<String> = pipeline_annotations
.mute
.iter()
.filter_map(asset_ref)
.collect();
let derived = derive_pipeline_asset_trigger_refs(
effective_assets.as_deref().unwrap_or(&[]),
&explicit_refs,
&muted_refs,
pipeline_annotations.mute_all,
);
// Derived edges have no per-`// on` opts, so they take the script-level
// `// debounce` default and `// retry` policy — same as writing a bare
// `// on <asset>` would.
let derived_debounce_s = pipeline_debounce_default
.as_deref()
.and_then(parse_duration_secs);
let derived_retry_count = pipeline_annotations
.retry
.as_ref()
.map(|r| r.count.min(i16::MAX as u32) as i16);
let derived_retry_delay_s = pipeline_annotations
.retry
.as_ref()
.and_then(|r| r.delay.as_deref())
.and_then(parse_duration_secs);
for trigger_ref in derived {
insert_script_trigger(
&mut *tx,
&w_id,
AssetUsageKind::Script,
&ns.path,
ScriptTriggerKind::Asset,
&trigger_ref,
pipeline_join_all,
derived_debounce_s,
derived_retry_count,
derived_retry_delay_s,
)
.await?;
}
}
// Schedule annotations (`// on schedule`) are marker-only — the binding
// lives on the schedule row's own `script_path` field, which the user
// creates separately via the schedule editor. No script-create-time
+190
View File
@@ -169,6 +169,68 @@ fn is_write_access(access: Option<AssetUsageAccessType>) -> bool {
)
}
/// Kinds whose *read* usage auto-derives a cascade trigger edge inside a
/// `// pipeline`. Scoped to the two intra-pipeline data kinds — a ducklake
/// table read (the core case) and an s3 object read (file-ingestion
/// producers). Resource / datatable / volume reads stay explicit-`// on`:
/// a config/lookup read cascading is more often surprising than wanted.
fn is_auto_trigger_kind(kind: AssetKind) -> bool {
matches!(kind, AssetKind::Ducklake | AssetKind::S3Object)
}
/// Trigger refs auto-derived from a pipeline script's inferred reads, so the
/// FROM clause alone wires the cascade edge (no redundant `// on <asset>`).
///
/// Included: an input read read-*only* (`R`) of a supported kind
/// ([`is_auto_trigger_kind`]). The effective access type is
/// `access_type.or(alt_access_type)` — same precedence as the persisted
/// `asset.usage_access_type` and the frontend mirror's `access_type ??
/// alt_access_type`, so a manual read override on an ambiguous parse still
/// derives an edge (and the live canvas and the deployed graph agree).
/// Excluded, each for a reason:
/// - `RW` / `W` — the script also writes the asset; an edge would be a
/// self-triggering loop.
/// - `None` access — usage is ambiguous (poisoned merge) with no override;
/// can't confirm a read, so fail safe and don't cascade.
/// - already in `explicit_refs` — the author wrote `// on <asset>`, which
/// wins (it carries the per-edge debounce/opts).
/// - in `muted_refs` — a `// mute <asset>` opt-out (lookup / SCD input).
///
/// `mute_all` (from `// mute all`) short-circuits to no derivation, leaving
/// only the explicit `// on` edges. Returns canonical refs (e.g.
/// `ducklake://main.orders`), deduped, in input order.
pub fn derive_pipeline_asset_trigger_refs(
assets: &[AssetWithAltAccessType],
explicit_refs: &HashSet<String>,
muted_refs: &HashSet<String>,
mute_all: bool,
) -> Vec<String> {
if mute_all {
return vec![];
}
let mut out = vec![];
let mut seen = HashSet::new();
for a in assets {
// Effective access mirrors the persisted `usage_access_type` and the
// frontend derivation: an explicit parse wins, else the manual override.
let access = a.access_type.or(a.alt_access_type);
if access != Some(AssetUsageAccessType::R) || !is_auto_trigger_kind(a.kind) {
continue;
}
let Some(prefix) = a.kind.canonical_prefix() else {
continue;
};
let r = format!("{}{}", prefix, a.path);
if explicit_refs.contains(&r) || muted_refs.contains(&r) {
continue;
}
if seen.insert(r.clone()) {
out.push(r);
}
}
out
}
/// Clear and reinsert the full static-asset usage set of a script in one tx,
/// invalidating the producer-writes cache at most once and only on a real
/// change. The cache (asset_dispatch::ASSET_PRODUCER_WRITES_CACHE) keys a
@@ -369,6 +431,134 @@ mod debounce_duration_tests {
}
}
#[cfg(test)]
mod derive_trigger_tests {
use super::{derive_pipeline_asset_trigger_refs, AssetKind, AssetUsageAccessType};
use std::collections::HashSet;
use windmill_types::assets::AssetWithAltAccessType;
fn asset(
kind: AssetKind,
path: &str,
at: Option<AssetUsageAccessType>,
) -> AssetWithAltAccessType {
AssetWithAltAccessType {
path: path.to_string(),
kind,
access_type: at,
alt_access_type: None,
columns: None,
}
}
fn derive(assets: &[AssetWithAltAccessType]) -> Vec<String> {
derive_pipeline_asset_trigger_refs(assets, &HashSet::new(), &HashSet::new(), false)
}
#[test]
fn read_only_ducklake_and_s3_derive_an_edge() {
use AssetUsageAccessType::R;
let a = [
asset(AssetKind::Ducklake, "main.orders", Some(R)),
asset(AssetKind::S3Object, "raw/events", Some(R)),
];
assert_eq!(
derive(&a),
vec![
"ducklake://main.orders".to_string(),
"s3://raw/events".to_string()
]
);
}
#[test]
fn writes_and_rw_are_skipped_to_avoid_self_edges() {
use AssetUsageAccessType::*;
// W (pure producer) and RW (reads *and* writes the same table — a
// self-cascade if edged) both derive nothing.
let a = [
asset(AssetKind::Ducklake, "main.out", Some(W)),
asset(AssetKind::Ducklake, "main.self", Some(RW)),
];
assert!(derive(&a).is_empty());
}
#[test]
fn ambiguous_access_and_unsupported_kinds_are_skipped() {
use AssetUsageAccessType::R;
let a = [
asset(AssetKind::Ducklake, "main.ambiguous", None), // poisoned merge
asset(AssetKind::Resource, "f/db", Some(R)), // out of scope
asset(AssetKind::DataTable, "main.dt", Some(R)), // out of scope
];
assert!(derive(&a).is_empty());
}
#[test]
fn manual_read_override_on_ambiguous_parse_derives_an_edge() {
use AssetUsageAccessType::{R, W};
// Parser can't confirm access (`access_type: None`) but the user manually
// overrode it. Effective access = `access_type.or(alt_access_type)`, the
// same value persisted to `asset.usage_access_type` and used by the
// frontend canvas — so a read override derives an edge (parity, no
// silently-vanishing edge on deploy) and a write override does not.
let read_override = AssetWithAltAccessType {
path: "main.override_r".to_string(),
kind: AssetKind::Ducklake,
access_type: None,
alt_access_type: Some(R),
columns: None,
};
let write_override = AssetWithAltAccessType {
path: "main.override_w".to_string(),
kind: AssetKind::Ducklake,
access_type: None,
alt_access_type: Some(W),
columns: None,
};
assert_eq!(
derive(&[read_override, write_override]),
vec!["ducklake://main.override_r".to_string()]
);
}
#[test]
fn explicit_and_muted_refs_are_excluded() {
use AssetUsageAccessType::R;
let a = [
asset(AssetKind::Ducklake, "main.explicit", Some(R)),
asset(AssetKind::Ducklake, "main.muted", Some(R)),
asset(AssetKind::Ducklake, "main.keep", Some(R)),
];
let explicit: HashSet<String> = ["ducklake://main.explicit".to_string()].into();
let muted: HashSet<String> = ["ducklake://main.muted".to_string()].into();
assert_eq!(
derive_pipeline_asset_trigger_refs(&a, &explicit, &muted, false),
vec!["ducklake://main.keep".to_string()]
);
}
#[test]
fn mute_all_derives_nothing() {
use AssetUsageAccessType::R;
let a = [asset(AssetKind::Ducklake, "main.orders", Some(R))];
assert!(
derive_pipeline_asset_trigger_refs(&a, &HashSet::new(), &HashSet::new(), true)
.is_empty()
);
}
#[test]
fn duplicate_reads_dedup() {
use AssetUsageAccessType::R;
let a = [
asset(AssetKind::Ducklake, "main.orders", Some(R)),
asset(AssetKind::Ducklake, "main.orders", Some(R)),
];
assert_eq!(derive(&a), vec!["ducklake://main.orders".to_string()]);
}
}
#[cfg(test)]
mod trigger_ref_roundtrip_tests {
use super::{parse_asset_trigger_ref, trigger_spec_to_row, AssetKind, ScriptTriggerKind};
@@ -18,6 +18,7 @@
import PanToNode from './PanToNode.svelte'
import InitialFitView from './InitialFitView.svelte'
import { layoutAssetGraph } from './assetGraphLayout'
import { computeMutedReadKeys } from './resolveGraph'
import { buildDownstreamMap } from './graphTraversal'
import { buildLineageDownstreamMap } from './boundedCascade'
import type { AssetGraphResponse, AssetGraphSelection, NativeTriggerKind } from './types'
@@ -228,6 +229,11 @@
| 'macro'
| 'test-dependency'
unsaved?: boolean
// Muted read edge: a ducklake/s3 input read every run whose (default)
// auto cascade trigger is suppressed by `// mute` / `// mute all`.
// Rendered with a bell-off badge — auto-wiring is the norm, so we mark
// the read that deliberately does NOT cascade, not every derived edge.
muted?: 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".
@@ -490,6 +496,10 @@
})
}
// Read edges of a ducklake/s3 asset with no cascade trigger = muted
// (`// mute` / `// mute all` opted the default auto trigger out). Gated
// on pipeline scripts inside the helper (non-pipeline reads never derive).
const mutedReadKeys = computeMutedReadKeys(g.edges, g.triggers, g.runnables)
for (const e of g.edges) {
const runnableId = `${e.runnable_kind}:${e.runnable_path}`
const assetId = `asset:${e.asset_kind}:${e.asset_path}`
@@ -542,7 +552,13 @@
source: assetId,
target: runnableId,
kind: 'lineage-read',
unsaved: e.unsaved
unsaved: e.unsaved,
// Only a pure `'r'` read can be muted; `'rw'` is a self-read.
muted:
access === 'r' &&
mutedReadKeys.has(
`${e.asset_kind}:${e.asset_path}->${e.runnable_kind}:${e.runnable_path}`
)
})
}
}
@@ -1079,7 +1095,9 @@
// Macro-edge badge: which of the library's macros the consumer
// calls (all of them when pulled in via `// use`).
macro_names: e.macro_names,
via_use: e.via_use
via_use: e.via_use,
// Muted read edge — bell-off badge on the read link.
muted: e.muted
},
animated,
label,
@@ -636,7 +636,9 @@
dataTests: [],
columnLineage: [],
macros: false,
useLibs: []
useLibs: [],
muteAssets: [],
muteAll: false
}
)
// `// macros` library: the defined signatures for the strip above the
@@ -1,7 +1,7 @@
<script lang="ts">
import { BaseEdge, getBezierPath, type EdgeProps } from '@xyflow/svelte'
import { NODE } from '$lib/components/graph/util'
import { FlaskConical, Columns3, SquareFunction } from 'lucide-svelte'
import { FlaskConical, Columns3, SquareFunction, BellOff } from 'lucide-svelte'
import type { ColumnLineage, DataTest } from './parsePipelineAnnotations'
let {
@@ -76,6 +76,14 @@
// they don't overlap on the link (each badge is 18px tall; +18 clears it).
let columnsBadgeY = $derived(tests && tests.length > 0 ? labelY + 18 : labelY)
// Muted read edge: a ducklake/s3 input the script reads every run but that
// does NOT cascade — `// mute <asset>` or `// mute all` opted it out of the
// (default) auto-derived trigger. Auto-wiring is the norm, so we badge the
// exception (a read with no trigger) rather than every derived edge.
let isMuted = $derived((data as { muted?: boolean } | undefined)?.muted ?? false)
const mutedBadgeTitle =
'Read but not cascaded — `// mute` (or `// mute all`) suppresses the auto trigger, so changes to this asset do not re-run this script.'
// Macro-edge badge: which of the library's macros the consumer calls (all
// of them when the whole lib is pulled in via `// use`).
let macroNames = $derived((data as { macro_names?: string[] } | undefined)?.macro_names)
@@ -222,6 +230,27 @@
</foreignObject>
{/if}
{#if isMuted}
<!-- Muted-read badge, centered on the read link. Bell-off = this input is
read but its auto cascade trigger is suppressed (`// mute`). -->
<foreignObject x={labelX - 30} y={labelY - 9} width="60" height="18" class="overflow-visible">
<div
xmlns="http://www.w3.org/1999/xhtml"
class="w-full h-full flex items-center justify-center"
style="pointer-events: none;"
>
<div
class="flex items-center gap-0.5 px-1 py-0.5 rounded-sm border shadow-sm text-3xs leading-none font-mono cursor-default bg-surface border-amber-300 dark:border-amber-900/60 text-amber-700 dark:text-amber-300"
style="pointer-events: all;"
title={mutedBadgeTitle}
>
<BellOff size={10} />
<span>muted</span>
</div>
</div>
</foreignObject>
{/if}
{#if columnLineage && columnLineage.length > 0}
<!-- Column-lineage badge, stacked below the data-test badge when both exist. -->
<foreignObject
@@ -22,7 +22,9 @@ const ASSERTED_TS_FIELDS: Record<keyof PipelineAnnotations, true> = {
dataTests: true,
columnLineage: true,
macros: true,
useLibs: true
useLibs: true,
muteAssets: true,
muteAll: true
}
// Parser-parity guard: this TS parser (drives the live graph preview) and
@@ -86,6 +88,11 @@ type Fixture = {
macros?: boolean
// `// use <lib_path>` accumulation, declaration order, deduped. Absent === [].
use_libs?: string[]
// `// mute <asset>` accumulation as `kind:path`, declaration order, deduped.
// Absent === [].
mute?: string[]
// `// mute all` marker. Absent === false.
mute_all?: boolean
}
}
@@ -194,6 +201,13 @@ describe('parsePipelineAnnotations matches the shared Rust fixture corpus', () =
expect(got.macros, 'macros').toBe(f.expected.macros ?? false)
expect(got.useLibs, 'use_libs').toEqual(f.expected.use_libs ?? [])
expect(
got.muteAssets.map((a) => `${a.kind}:${a.path}`),
'mute'
).toEqual(f.expected.mute ?? [])
expect(got.muteAll, 'mute_all').toBe(f.expected.mute_all ?? false)
})
}
})
@@ -62,6 +62,13 @@ describe('parsePipelineAnnotations: header scan', () => {
const out = parsePipelineAnnotations(code)
expect(out.tag).toBe('heavy')
})
it('strips trailing key=value opts from a // on asset ref', () => {
// The path must be the bare asset, not `main.orders debounce=60s`, so it
// dedups against body inference and matches the deploy path.
const out = parsePipelineAnnotations('// on ducklake://main.orders debounce=60s')
expect(out.triggerAssets).toEqual([{ kind: 'ducklake', path: 'main.orders' }])
})
})
describe('parsePipelineAnnotations: retry', () => {
@@ -215,6 +215,14 @@ export type PipelineAnnotations = {
// `// use <lib_script_path>` — force-inject the named macro library into
// this script's jobs. Accumulating, declaration order, deduped.
useLibs: string[]
// `// mute <asset>` — suppress the auto-derived cascade edge for a read
// that would otherwise trigger this script (a lookup / SCD input). Only
// asset refs; native trigger kinds are never auto-derived. Accumulating,
// deduped.
muteAssets: PipelineTriggerAsset[]
// `// mute all` — opt out of auto-derivation entirely (explicit `// on`
// edges are unaffected).
muteAll: boolean
}
// Tokenize a `key=value [key="quoted value"] ...` option string. Bare
@@ -254,6 +262,17 @@ function parseKvOpts(s: string): Map<string, string> {
return out
}
// Drop a `// on` right-hand side's trailing `key=value` opts (e.g.
// `debounce=60s`), returning just the trigger ref. The opts start at the first
// whitespace-delimited token shaped like `<ident>=…`; everything before is the
// ref. Mirrors Rust `split_trailing_kv_opts` — asset refs never contain a space
// followed by an `ident=` token, so without this `// on ducklake://t debounce=1s`
// would keep the path as `t debounce=1s` and desync the live canvas from deploy.
function stripTrailingKvOpts(s: string): string {
const m = s.match(/\s([A-Za-z_][A-Za-z0-9_]*)=/)
return (m?.index !== undefined ? s.slice(0, m.index) : s).trimEnd()
}
function parseAssetSyntax(s: string): PipelineTriggerAsset | undefined {
for (const [prefix, kind] of ASSET_PREFIXES) {
if (s.startsWith(prefix)) {
@@ -570,7 +589,9 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
dataTests: [],
columnLineage: [],
macros: false,
useLibs: []
useLibs: [],
muteAssets: [],
muteAll: false
}
for (const rawLine of code.split('\n')) {
@@ -658,6 +679,27 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
continue
}
// `// mute all` opts out of auto-derived cascade edges entirely;
// `// mute <asset>` suppresses the one edge. Only asset refs are
// muteable — native trigger kinds are never auto-derived. A complete
// word, so prose like `// muted for now` never matches.
const afterMute = consumeKeyword(inner, 'mute')
if (afterMute !== undefined) {
const arg = afterMute.trim()
if (arg === 'all') {
out.muteAll = true
} else {
const spec = parseTriggerSpec(arg)
if (
spec?.kind === 'asset' &&
!out.muteAssets.some((a) => a.kind === spec.value.kind && a.path === spec.value.path)
) {
out.muteAssets.push(spec.value)
}
}
continue
}
// `data_test` is a complete word (so it never collides with the `// test:`
// CI annotation, which has no whitespace after `test`). Accumulates.
const afterDataTest = consumeKeyword(inner, 'data_test')
@@ -678,7 +720,10 @@ export function parsePipelineAnnotations(code: string): PipelineAnnotations {
const afterOn = consumeKeyword(inner, 'on')
if (afterOn !== undefined) {
const specText = afterOn.trim()
// Strip trailing `key=value` opts (e.g. `debounce=60s`) so the ref
// matches body inference and the deploy path — otherwise the opts
// leak into the asset path and the explicit edge dedups wrong.
const specText = stripTrailingKvOpts(afterOn.trim())
if (!specText) continue
const spec = parseTriggerSpec(specText)
if (!spec) continue
@@ -91,3 +91,70 @@ describe('pipelineTemplates materialize partition filter', () => {
expect(body).not.toContain('strftime')
})
})
describe('pipelineTemplates: auto-derived reads drop the redundant // on', () => {
const draft = (
input: { kind: 'ducklake' | 's3object' | 'datatable'; path: string } | undefined,
triggers: Parameters<typeof generatePipelineDraft>[0]['triggers'],
language: ScriptLang = 'duckdb',
outputKind: PipelineOutputKind = 'materialize'
) =>
generatePipelineDraft({
language,
outputKind,
output: autoOutputAsset(outputKind, 'demo', language),
input,
triggers
})
it('omits // on for a ducklake input the body reads (auto-derived)', () => {
const body = draft({ kind: 'ducklake', path: 'main/orders' }, [
{ kind: 'asset', ref: 'ducklake://main/orders' }
])
expect(body).not.toContain('on ducklake://main/orders')
// The body still attaches and reads the table (`ducklake://main` →
// `lake.orders`), which is what wires the cascade now that the explicit
// `// on` is gone.
expect(body).toContain(`ATTACH 'ducklake://main'`)
expect(body).toContain('lake.orders')
})
it('omits // on for an s3 input the body reads', () => {
const body = draft({ kind: 's3object', path: 'raw/events.parquet' }, [
{ kind: 'asset', ref: 's3://raw/events.parquet' }
])
expect(body).not.toContain('on s3://raw/events.parquet')
})
it('keeps // on for a datatable input (not auto-derived)', () => {
const body = draft({ kind: 'datatable', path: 'main/dt' }, [
{ kind: 'asset', ref: 'datatable://main/dt' }
])
expect(body).toContain('on datatable://main/dt')
})
it('keeps // on for a native trigger', () => {
const body = draft(undefined, [{ kind: 'schedule', path: undefined }])
expect(body).toContain('on schedule')
})
it('keeps // on when the body does not read the input (postgres, bash)', () => {
// postgres/bash bodies ignore `input`, so dropping `// on` would leave the
// script with neither an explicit trigger nor an inferred read → no cascade.
const pg = draft(
{ kind: 'ducklake', path: 'main/orders' },
[{ kind: 'asset', ref: 'ducklake://main/orders' }],
'postgresql',
'datatable'
)
expect(pg).toContain('on ducklake://main/orders')
const bash = draft(
{ kind: 's3object', path: 'raw/events.parquet' },
[{ kind: 'asset', ref: 's3://raw/events.parquet' }],
'bash',
's3_object'
)
expect(bash).toContain('on s3://raw/events.parquet')
})
})
@@ -332,7 +332,7 @@ export type TemplateContext = {
// Output asset is NOT declared here — it's reconstructed from the body's
// SDK calls / SQL by the asset parser, same as production scripts.
function header(ctx: TemplateContext): string {
const { language, triggers, output, outputKind } = ctx
const { language, triggers, output, outputKind, input } = ctx
const p = commentPrefix(language)
// A custom data test is a standalone script, not a graph node that produces
// an asset — so it gets no `// pipeline` / output annotation. Instead, tell
@@ -345,14 +345,26 @@ function header(ctx: TemplateContext): string {
''
].join('\n')
}
const lines = triggers.map((t) => {
// A ducklake/s3 read the body performs auto-wires its cascade edge from the
// FROM clause, so an explicit `// on` for that same asset would be redundant
// now that auto-derivation is the default. Only drop it when the generated
// body actually READS the input: bun/deno/python/duckdb emit an input load,
// but postgres/bash/generic bodies (and `data_upload`, which reads the picker
// `file` instead) do not — dropping there would leave the script with no
// cascade at all. Also keep `// on` for kinds inference can't derive
// (datatable/resource/…) and native triggers.
const bodyReadsInput = READS_INPUT_LANGS.has(language) && !isDataUpload(triggers)
const inputRef = input ? assetUri(input) : undefined
const inputAutoDerives = input?.kind === 'ducklake' || input?.kind === 's3object'
const lines = triggers.flatMap((t) => {
switch (t.kind) {
case 'asset':
return `${p} on ${t.ref}`
if (bodyReadsInput && inputAutoDerives && t.ref === inputRef) return []
return [`${p} on ${t.ref}`]
default:
// Native triggers (incl. schedule): marker-only — the
// binding lives on the trigger row's own `script_path`.
return `${p} on ${t.kind}`
return [`${p} on ${t.kind}`]
}
})
// Managed materialization is the one kind that declares its output
@@ -384,7 +396,10 @@ function header(ctx: TemplateContext): string {
// link is the canonical reference once they want the details. A blank
// line separates it from the parsed annotations above (`// pipeline`,
// `// on …`) so the editor reads as "real annotations, then a hint".
const more = `${p} More: partitioned daily, freshness 1h, retry 3, tag heavy — https://www.windmill.dev/docs/core_concepts/pipelines`
// Reads of a ducklake/s3 table auto-trigger this script when that table
// changes — `// mute <asset>` opts a read out of the cascade. The rest are
// the annotations users most often miss.
const more = `${p} More: mute <asset> to not cascade on a read, partitioned daily, freshness 1h, retry 3, tag heavy — https://www.windmill.dev/docs/core_concepts/pipelines`
return [`${p} pipeline`, ...lines, ...matLine, ...macrosLine, '', more, ''].join('\n')
}
@@ -402,6 +417,11 @@ function isDataUpload(triggers: DraftTriggerSource[]): boolean {
return triggers.some((t) => t.kind === 'data_upload')
}
// Languages whose generated body reads the `input` asset (an SDK load / SQL
// FROM), so a ducklake/s3 input auto-derives its cascade and the explicit
// `// on` can be dropped. postgres/bash/generic bodies ignore `input`.
const READS_INPUT_LANGS: ReadonlySet<ScriptLang> = new Set(['bun', 'deno', 'python3', 'duckdb'])
function bodyTs(ctx: TemplateContext): string {
const { input, output, outputKind } = ctx
const dataUpload = isDataUpload(ctx.triggers)
@@ -1,5 +1,5 @@
import { describe, expect, it } from 'vitest'
import { resolveGraph, type ResolveGraphInput } from './resolveGraph'
import { resolveGraph, computeMutedReadKeys, type ResolveGraphInput } from './resolveGraph'
import type { PipelineAnnotations } from './parsePipelineAnnotations'
import type { AssetGraphResponse } from './types'
import type { AssetWithAltAccessType } from '$lib/components/assets/lib'
@@ -12,6 +12,8 @@ const ann = (over: Partial<PipelineAnnotations> = {}): PipelineAnnotations => ({
columnLineage: [],
macros: false,
useLibs: [],
muteAssets: [],
muteAll: false,
...over
})
@@ -37,6 +39,15 @@ const input = (over: Partial<ResolveGraphInput> = {}): ResolveGraphInput => ({
const s3 = (path: string, access_type: 'r' | 'w' | 'rw'): AssetWithAltAccessType =>
({ kind: 's3object', path, access_type }) as AssetWithAltAccessType
const duck = (path: string, access_type: 'r' | 'w' | 'rw'): AssetWithAltAccessType =>
({ kind: 'ducklake', path, access_type }) as AssetWithAltAccessType
/** kind:path of the unsaved asset-trigger overlays for a runnable. */
const assetTrigKeys = (r: AssetGraphResponse, path: string): string[] =>
r.triggers
.filter((t) => t.trigger_kind === 'asset' && t.runnable_path === path && (t as any).unsaved)
.map((t) => `${(t as any).asset_kind}:${(t as any).asset_path}`)
describe('resolveGraph', () => {
it('passes an empty graph through unchanged', () => {
const r = resolveGraph(input())
@@ -224,6 +235,85 @@ describe('resolveGraph', () => {
expect(r.edges.find((e) => e.asset_path === '/should-not.json')).toBeUndefined()
})
it('open buffer: a ducklake/s3 read auto-derives an unsaved cascade trigger', () => {
const liveBodyAssets = {
scriptPath: 'f/x/open',
assets: [duck('main.orders', 'r'), s3('raw/events', 'r'), duck('main.out', 'w')]
}
const liveAnnotations = { scriptPath: 'f/x/open', annotations: ann({ inPipeline: true }) }
const r = resolveGraph(input({ liveBodyAssets, liveAnnotations }))
// The two reads derive edges; the write (main.out) does not (self-edge).
expect(assetTrigKeys(r, 'f/x/open').sort()).toEqual([
'ducklake:main.orders',
's3object:raw/events'
])
})
it('open buffer: a materialize producer reading its own target does not self-cascade', () => {
// The body `SELECT`s from the table it materializes (incremental model).
// The write is annotation-declared, not in the body, so without excluding
// the materialize target the read would auto-derive a self-cascade edge.
const liveBodyAssets = { scriptPath: 'f/x/p', assets: [duck('main.orders', 'r')] }
const liveAnnotations = {
scriptPath: 'f/x/p',
annotations: ann({
inPipeline: true,
materialize: { targetKind: 'ducklake' as const, targetPath: 'main.orders' }
})
}
const r = resolveGraph(input({ liveBodyAssets, liveAnnotations }))
expect(assetTrigKeys(r, 'f/x/p')).toEqual([])
})
it('open buffer: a read that is also written (rw) does not self-trigger', () => {
const liveBodyAssets = { scriptPath: 'f/x/open', assets: [duck('main.self', 'rw')] }
const liveAnnotations = { scriptPath: 'f/x/open', annotations: ann({ inPipeline: true }) }
const r = resolveGraph(input({ liveBodyAssets, liveAnnotations }))
expect(assetTrigKeys(r, 'f/x/open')).toEqual([])
})
it('open buffer: // mute suppresses one derived edge, // mute all suppresses all', () => {
const liveBodyAssets = {
scriptPath: 'f/x/open',
assets: [duck('main.a', 'r'), duck('main.b', 'r')]
}
const muted = resolveGraph(
input({
liveBodyAssets,
liveAnnotations: {
scriptPath: 'f/x/open',
annotations: ann({ inPipeline: true, muteAssets: [{ kind: 'ducklake', path: 'main.a' }] })
}
})
)
expect(assetTrigKeys(muted, 'f/x/open')).toEqual(['ducklake:main.b'])
const all = resolveGraph(
input({
liveBodyAssets,
liveAnnotations: {
scriptPath: 'f/x/open',
annotations: ann({ inPipeline: true, muteAll: true })
}
})
)
expect(assetTrigKeys(all, 'f/x/open')).toEqual([])
})
it('open buffer: an explicit // on is not double-emitted as a derived edge', () => {
const liveBodyAssets = { scriptPath: 'f/x/open', assets: [duck('main.orders', 'r')] }
const liveAnnotations = {
scriptPath: 'f/x/open',
annotations: ann({
inPipeline: true,
triggerAssets: [{ kind: 'ducklake', path: 'main.orders' }]
})
}
const r = resolveGraph(input({ liveBodyAssets, liveAnnotations }))
// Exactly one overlay for the table, not one explicit + one derived.
expect(assetTrigKeys(r, 'f/x/open')).toEqual(['ducklake:main.orders'])
})
it('open-script live annotations add unsaved triggers, deduped vs persisted', () => {
const base = baseGraph({
triggers: [
@@ -630,3 +720,79 @@ describe('resolveGraph', () => {
expect(lib?.macros).toEqual([{ name: 'dbl', params: 'a', is_table: false }])
})
})
describe('computeMutedReadKeys', () => {
const readEdge = (asset_kind: any, asset_path: string, access: any = 'r') => ({
runnable_path: 'f/x/c',
runnable_kind: 'script' as const,
asset_kind,
asset_path,
access_type: access
})
const assetTrigger = (asset_kind: any, asset_path: string) => ({
trigger_kind: 'asset' as const,
asset_kind,
asset_path,
runnable_kind: 'script' as const,
runnable_path: 'f/x/c'
})
// `f/x/c` (the read consumer) is an in-pipeline script.
const pipelineRunnable = [{ path: 'f/x/c', usage_kind: 'script' as const, in_pipeline: true }]
it('flags a ducklake/s3 read with no cascade trigger as muted', () => {
const muted = computeMutedReadKeys(
[readEdge('ducklake', 'main.orders'), readEdge('s3object', 'raw/events')],
[],
pipelineRunnable
)
expect([...muted].sort()).toEqual([
'ducklake:main.orders->script:f/x/c',
's3object:raw/events->script:f/x/c'
])
})
it('does not flag a read that has a cascade trigger', () => {
const muted = computeMutedReadKeys(
[readEdge('ducklake', 'main.orders')],
[assetTrigger('ducklake', 'main.orders')],
pipelineRunnable
)
expect(muted.size).toBe(0)
})
it('ignores rw self-reads and unsupported kinds', () => {
const muted = computeMutedReadKeys(
[
readEdge('ducklake', 'main.self', 'rw'), // self-read, not muted
readEdge('resource', 'f/db'), // out of scope
readEdge('datatable', 'main.dt') // out of scope
],
[],
pipelineRunnable
)
expect(muted.size).toBe(0)
})
it('does not flag a read whose script also writes the asset', () => {
// Live-overlay shape of a `// materialize` producer reading its own target:
// a separate `'r'` read edge and `'w'` write edge for the same asset. It's
// the script's own output, not a suppressed input, so it is not muted.
const muted = computeMutedReadKeys(
[readEdge('ducklake', 'main.orders', 'r'), readEdge('ducklake', 'main.orders', 'w')],
[],
pipelineRunnable
)
expect(muted.size).toBe(0)
})
it('does not flag a read by a non-pipeline script (auto-derivation never applied)', () => {
// Same read, but the consumer is a plain script (or a flow), not a
// `// pipeline` member — no auto trigger was ever derived to suppress.
const plainScript = [{ path: 'f/x/c', usage_kind: 'script' as const, in_pipeline: false }]
expect(computeMutedReadKeys([readEdge('ducklake', 'main.orders')], [], plainScript).size).toBe(
0
)
const flow = [{ path: 'f/x/c', usage_kind: 'flow' as const, in_pipeline: true }]
expect(computeMutedReadKeys([readEdge('ducklake', 'main.orders')], [], flow).size).toBe(0)
})
})
@@ -70,6 +70,108 @@ function persistedNativeKinds(base: AssetGraphResponse, path: string): Set<strin
)
}
// Read-asset kinds that auto-derive a cascade trigger edge inside a
// `// pipeline`. Mirror of the backend `is_auto_trigger_kind`
// (windmill-common assets.rs) — ducklake tables and s3 objects only.
const AUTO_TRIGGER_KINDS: ReadonlySet<AssetKind> = new Set(['ducklake', 's3object'])
/** `kind:path` refs of a script's `// materialize` write target(s) (base +
* the scd2 `<dim>_current` companion), which the body `SELECT` doesn't express. */
function materializeWriteRefs(parsed: PipelineAnnotations): string[] {
const m = parsed.materialize
if (!m) return []
const refs = [`${m.targetKind}:${m.targetPath}`]
const current = scd2CurrentTargetPath(m)
if (current) refs.push(`${m.targetKind}:${current}`)
return refs
}
/**
* Backend-mirror of `derive_pipeline_asset_trigger_refs` (windmill-common
* assets.rs): a pipeline script's ducklake/s3 read auto-wires a cascade
* trigger edge from the FROM clause, so `// on <asset>` is only needed for
* edges inference can't see. Excluded: assets the script also writes (`writes`
* covers `w`/`rw`, so an `rw` self-read can't loop-trigger), muted assets,
* `// mute all`, and any explicit `// on` (which already emits its own edge).
* Returns the `{kind, path}` refs to overlay as unsaved asset triggers, deduped.
*/
function deriveAutoAssetTriggers(
reads: Array<{ kind: AssetKind; path: string }>,
writes: Array<{ kind: AssetKind; path: string }>,
parsed: PipelineAnnotations
): Array<{ kind: AssetKind; path: string }> {
// Auto-derivation is scoped to `// pipeline` scripts (backend gates on
// `in_pipeline`); `// mute all` opts a pipeline script back out.
if (!parsed.inPipeline || parsed.muteAll) return []
const skip = new Set<string>([
...writes.map((w) => `${w.kind}:${w.path}`),
// The `// materialize` target(s) are this script's own output — the body
// `SELECT` doesn't express the write, so an incremental model that reads
// its own target would otherwise auto-derive a self-cascade edge (backend
// parity: the deploy path upgrades the same read to `rw`).
...materializeWriteRefs(parsed),
...parsed.muteAssets.map((a) => `${a.kind}:${a.path}`),
...parsed.triggerAssets.map((a) => `${a.kind}:${a.path}`)
])
const out: Array<{ kind: AssetKind; path: string }> = []
for (const r of reads) {
const key = `${r.kind}:${r.path}`
if (!AUTO_TRIGGER_KINDS.has(r.kind) || skip.has(key)) continue
skip.add(key) // dedup within reads too
out.push({ kind: r.kind, path: r.path })
}
return out
}
/** Edge key matching a lineage/trigger edge's `(asset, runnable)` pair. */
function edgeKey(a: {
asset_kind: string
asset_path: string
runnable_kind: string
runnable_path: string
}): string {
return `${a.asset_kind}:${a.asset_path}->${a.runnable_kind}:${a.runnable_path}`
}
/**
* Keys of "muted read" edges: a ducklake/s3 asset a script reads read-*only*
* (`access_type === 'r'`) yet has NO cascade trigger for. Auto-derivation is the
* default, so a supported read with no trigger means the author opted it out
* with `// mute <asset>` / `// mute all` (or it's an explicit non-triggering
* read). The canvas badges these — the exception — instead of every derived
* edge. Self-reads are excluded: a script that also writes the asset carries
* `'rw'` on the deployed graph (one edge) or a separate `'w'` edge in the live
* overlay (a `// materialize` producer reading its own target), and neither
* should badge as muted — it's the script's own output, not a suppressed input.
*
* Only `// pipeline` scripts are considered: auto-derivation never applies to a
* plain script or a flow, so a non-pipeline runnable reading a ducklake/s3
* asset has no auto trigger to suppress and must render as ordinary lineage.
*/
export function computeMutedReadKeys(
edges: AssetGraphResponse['edges'],
triggers: AssetGraphResponse['triggers'],
runnables: AssetGraphResponse['runnables']
): Set<string> {
const pipelineScripts = new Set(
runnables.filter((r) => r.usage_kind === 'script' && r.in_pipeline).map((r) => r.path)
)
const cascaded = new Set(
triggers.filter((t) => t.trigger_kind === 'asset').map((t) => edgeKey(t))
)
const written = new Set(
edges.filter((e) => e.access_type === 'w' || e.access_type === 'rw').map((e) => edgeKey(e))
)
const muted = new Set<string>()
for (const e of edges) {
if (e.access_type !== 'r' || !AUTO_TRIGGER_KINDS.has(e.asset_kind)) continue
if (e.runnable_kind !== 'script' || !pipelineScripts.has(e.runnable_path)) continue
const key = edgeKey(e)
if (!cascaded.has(key) && !written.has(key)) muted.add(key)
}
return muted
}
/** `kind:path` keys of persisted asset (`// on <asset>`) triggers for `path`. */
function persistedAssetKeys(base: AssetGraphResponse, path: string): Set<string> {
return new Set(
@@ -313,7 +415,7 @@ function seedAccumulator(input: ResolveGraphInput, ctx: ResolveContext): Accumul
* `drafts` map so multiple concurrent drafts all render at once.
*/
function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
const { base, drafts, liveBodyAssets } = input
const { base, drafts, liveBodyAssets, inferredReadsByPath, inferredWritesByPath } = input
const { runnables, assets, edges, extraTriggers } = acc
for (const [path, d] of drafts) {
@@ -489,6 +591,31 @@ function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
const hasTriggerAsset = assets.some((x) => x.kind === a.kind && x.path === a.path)
if (!hasTriggerAsset) assets.push({ kind: a.kind, path: a.path })
}
// Auto-derived cascade edges (backend parity): a ducklake/s3 read wires
// the edge from the body alone. Reads/writes come from the active draft's
// live inference, else the sticky session cache. The open buffer's derived
// edges are re-computed authoritatively in applyLiveBufferOverlay (which
// strips this path's seeded triggers first), same as the explicit `// on`
// triggers above.
const draftReads = liveForThisDraft
? extractReads(liveBodyAssets.assets)
: (inferredReadsByPath.get(path) ?? [])
const draftWrites = liveForThisDraft
? extractWrites(liveBodyAssets.assets)
: (inferredWritesByPath.get(path) ?? [])
for (const a of deriveAutoAssetTriggers(draftReads, draftWrites, parsed)) {
extraTriggers.push({
trigger_kind: 'asset',
asset_kind: a.kind,
asset_path: a.path,
runnable_kind: 'script',
runnable_path: path,
unsaved: true
})
if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
assets.push({ kind: a.kind, path: a.path })
}
}
// Native trigger annotations on a draft are "missing" until a
// matching trigger row exists. A brand-new draft never has one (the
// script isn't deployed yet), but a draft promoted from unsaved
@@ -549,6 +676,30 @@ function applyLiveBufferOverlay(acc: Accumulator, input: ResolveGraphInput, ctx:
assets.push({ kind: a.kind, path: a.path })
}
}
// Auto-derived cascade edges (backend parity): a ducklake/s3 read wires
// the edge from the FROM clause alone, keystroke-live. The open buffer's
// live body inference is authoritative for its reads/writes. `// mute` /
// `// mute all` and explicit `// on` (above) suppress a derived edge; a
// derived edge already persisted for a non-draft open script is deduped
// via `assetKeys`.
if (input.liveBodyAssets.scriptPath === livePath) {
const reads = extractReads(input.liveBodyAssets.assets)
const writes = extractWrites(input.liveBodyAssets.assets)
for (const a of deriveAutoAssetTriggers(reads, writes, liveAnnotations.annotations)) {
if (assetKeys.has(`${a.kind}:${a.path}`)) continue
extraTriggers.push({
trigger_kind: 'asset',
asset_kind: a.kind,
asset_path: a.path,
runnable_kind: 'script',
runnable_path: livePath,
unsaved: true
})
if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
assets.push({ kind: a.kind, path: a.path })
}
}
}
// Native trigger annotations: kinds for which a matching trigger
// row was found in the backend response. If the live buffer
// declares `// on kafka` and at least one kafka_trigger row points
@@ -259,7 +259,9 @@
dataTests: [],
columnLineage: [],
macros: false,
useLibs: []
useLibs: [],
muteAssets: [],
muteAll: false
}
}