refactor: simplify pipeline code per review (dedup, single-parse, constant)

- ParseAssetsOutput::new() collapses the 6-line annotation copy-paste
  across the 4 asset-parser crates to one call site.
- asset_dispatch: parse the cascade trigger object once and pass it to
  the depth/partition readers instead of deserializing it twice; add a
  TRIGGER_ARG constant for the previously stringly-typed key (3 sites).
- scripts deploy: drop a redundant debounce_default clone.
No behavior change; 29 parser + 6 dispatch integration tests green.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

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

@@ -29,16 +29,11 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
     }
 
     let pipeline = parse_pipeline_annotations(input);
-    Ok(ParseAssetsOutput {
-        assets: merge_assets(assets_finder.assets),
-        in_pipeline: pipeline.in_pipeline,
-        triggers: pipeline.triggers,
-        partition: pipeline.partition,
-        freshness: pipeline.freshness,
-        join_mode: pipeline.join_mode,
-        debounce_default: pipeline.debounce_default,
-        ..Default::default()
-    })
+    Ok(ParseAssetsOutput::new(
+        merge_assets(assets_finder.assets),
+        Vec::new(),
+        pipeline,
+    ))
 }
 
 type VarAssetName = String;

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

@@ -34,16 +34,11 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
     }
 
     let pipeline = parse_pipeline_annotations(input);
-    Ok(ParseAssetsOutput {
-        assets: merge_assets(collector.assets),
-        in_pipeline: pipeline.in_pipeline,
-        triggers: pipeline.triggers,
-        partition: pipeline.partition,
-        freshness: pipeline.freshness,
-        join_mode: pipeline.join_mode,
-        debounce_default: pipeline.debounce_default,
-        ..Default::default()
-    })
+    Ok(ParseAssetsOutput::new(
+        merge_assets(collector.assets),
+        Vec::new(),
+        pipeline,
+    ))
 }
 
 /// Visitor that collects S3 asset literals from SQL statements

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

@@ -39,16 +39,11 @@ pub fn parse_assets(code: &str) -> anyhow::Result<ParseAssetsOutput> {
         AssetsFinder { assets: vec![], sql_queries: vec![], var_identifiers: HashMap::new() };
     assets_finder.visit_module_items(&ast);
     let pipeline = parse_pipeline_annotations(code);
-    Ok(ParseAssetsOutput {
-        assets: merge_assets(assets_finder.assets),
-        sql_queries: assets_finder.sql_queries,
-        in_pipeline: pipeline.in_pipeline,
-        triggers: pipeline.triggers,
-        partition: pipeline.partition,
-        freshness: pipeline.freshness,
-        join_mode: pipeline.join_mode,
-        debounce_default: pipeline.debounce_default,
-    })
+    Ok(ParseAssetsOutput::new(
+        merge_assets(assets_finder.assets),
+        assets_finder.sql_queries,
+        pipeline,
+    ))
 }
 
 type VarAssetName = String;

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

@@ -41,14 +41,9 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
     }
 
     let pipeline = parse_pipeline_annotations(input);
-    Ok(ParseAssetsOutput {
-        assets: merge_assets(assets),
-        in_pipeline: pipeline.in_pipeline,
-        triggers: pipeline.triggers,
-        partition: pipeline.partition,
-        freshness: pipeline.freshness,
-        join_mode: pipeline.join_mode,
-        debounce_default: pipeline.debounce_default,
-        ..Default::default()
-    })
+    Ok(ParseAssetsOutput::new(
+        merge_assets(assets),
+        Vec::new(),
+        pipeline,
+    ))
 }

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

@@ -225,6 +225,29 @@ pub struct PipelineAnnotations {
     pub debounce_default: Option<String>,
 }
 
+impl ParseAssetsOutput {
+    /// Build from detected assets/queries plus the script's parsed
+    /// pipeline annotations, so each language asset-parser does not
+    /// re-list the per-annotation fields (one call site instead of six
+    /// lines kept in lockstep across the parser crates).
+    pub fn new(
+        assets: Vec<ParseAssetsResult>,
+        sql_queries: Vec<SqlQueryDetails>,
+        pipeline: PipelineAnnotations,
+    ) -> Self {
+        ParseAssetsOutput {
+            assets,
+            sql_queries,
+            in_pipeline: pipeline.in_pipeline,
+            triggers: pipeline.triggers,
+            partition: pipeline.partition,
+            freshness: pipeline.freshness,
+            join_mode: pipeline.join_mode,
+            debounce_default: pipeline.debounce_default,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 pub struct DelegateToGitRepoDetails {
     pub resource: String,

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

@@ -1244,7 +1244,7 @@ async fn create_script_internal<'c>(
     let pipeline_join_all = !pipeline_annotations.join_mode.is_any();
     // Script-level `// debounce <dur>` default; a per-`// on debounce=`
     // overrides it (precedence resolved per edge below).
-    let pipeline_debounce_default = pipeline_annotations.debounce_default.clone();
+    let pipeline_debounce_default = pipeline_annotations.debounce_default;
     let pipeline_triggers = pipeline_annotations.triggers;
     let auto_kind = if in_pipeline {
         Some("pipeline".to_string())

backend/windmill-queue/src/asset_dispatch.rs

@@ -64,6 +64,10 @@ use windmill_common::DB;
 /// Set by the test panel when the user opts out of the cascade.
 pub const SKIP_ASSET_DISPATCH_ARG: &str = "_wmill_skip_asset_dispatch";
 
+/// Arg key holding the cascade trigger object (carries `depth`, `partition`,
+/// producer metadata) injected into every dispatched subscriber.
+const TRIGGER_ARG: &str = "trigger";
+
 /// Reserved arg key (under `trigger.depth`) that carries cascade depth.
 const CHAIN_DEPTH_KEY: &str = "depth";
 
@@ -105,8 +109,14 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
     if read_skip_arg(args.as_ref()) {
         return Ok(DispatchResult::default());
     }
-    let depth = read_chain_depth(args.as_ref());
-    let partition = read_partition(args.as_ref());
+    // Parse the cascade `trigger` object once; both depth and the
+    // propagated partition are read from it.
+    let trigger_map = args
+        .as_ref()
+        .and_then(|a| a.get(TRIGGER_ARG))
+        .and_then(|t| serde_json::from_str::<HashMap<String, Box<RawValue>>>(t.get()).ok());
+    let depth = read_chain_depth(trigger_map.as_ref());
+    let partition = read_partition(args.as_ref(), trigger_map.as_ref());
     if depth >= MAX_CHAIN_DEPTH {
         tracing::warn!(
             "asset-trigger dispatch skipped: chain depth {} >= cap {} (job {}, path {})",
@@ -237,17 +247,9 @@ fn read_skip_arg(args: Option<&HashMap<String, Box<RawValue>>>) -> bool {
         .unwrap_or(false)
 }
 
-fn read_chain_depth(args: Option<&HashMap<String, Box<RawValue>>>) -> i64 {
-    let Some(args) = args else {
-        return 0;
-    };
-    let Some(trigger) = args.get("trigger") else {
-        return 0;
-    };
-    let Ok(map) = serde_json::from_str::<HashMap<String, Box<RawValue>>>(trigger.get()) else {
-        return 0;
-    };
-    map.get(CHAIN_DEPTH_KEY)
+fn read_chain_depth(trigger_map: Option<&HashMap<String, Box<RawValue>>>) -> i64 {
+    trigger_map
+        .and_then(|m| m.get(CHAIN_DEPTH_KEY))
         .and_then(|v| serde_json::from_str::<i64>(v.get()).ok())
         .unwrap_or(0)
 }
@@ -257,16 +259,16 @@ fn read_chain_depth(args: Option<&HashMap<String, Box<RawValue>>>) -> i64 {
 /// materializes the same partition without re-resolving. Top-level
 /// `partition` arg (run-start injection) takes precedence over the
 /// `trigger.partition` carried from an upstream cascade hop.
-fn read_partition(args: Option<&HashMap<String, Box<RawValue>>>) -> Option<String> {
-    let args = args?;
-    if let Some(v) = args.get(PARTITION_ARG) {
+fn read_partition(
+    args: Option<&HashMap<String, Box<RawValue>>>,
+    trigger_map: Option<&HashMap<String, Box<RawValue>>>,
+) -> Option<String> {
+    if let Some(v) = args.and_then(|a| a.get(PARTITION_ARG)) {
         if let Ok(s) = serde_json::from_str::<String>(v.get()) {
             return Some(s);
         }
     }
-    let trigger = args.get("trigger")?;
-    let map = serde_json::from_str::<HashMap<String, Box<RawValue>>>(trigger.get()).ok()?;
-    serde_json::from_str::<String>(map.get(PARTITION_ARG)?.get()).ok()
+    serde_json::from_str::<String>(trigger_map?.get(PARTITION_ARG)?.get()).ok()
 }
 
 fn prefix_for(kind: AssetKind) -> Option<&'static str> {
@@ -517,7 +519,7 @@ async fn push_subscriber(
         CHAIN_DEPTH_KEY: depth,
         PARTITION_ARG: partition,
     });
-    args.insert("trigger".to_string(), to_raw_value(&trigger_payload));
+    args.insert(TRIGGER_ARG.to_string(), to_raw_value(&trigger_payload));
     // Carry the producer's resolved partition forward as a top-level arg so
     // the subscriber's body can read it and the next cascade hop's
     // `read_partition` picks it up — keeps the whole chain on one partition,