fix: reject trailing tokens in defer classifier, keep cache_ttl steps eager

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/windmill-worker/src/transform_var_defer.rs

@@ -67,6 +67,13 @@ pub fn classify_deferable_variables(expr: &str) -> Option<DeferPlan> {
     if !parser.take_errors().is_empty() {
         return None;
     }
+    // `parse_expr` does not require EOF and records no error for unconsumed
+    // trailing tokens; eager evaluation of such an expression fails, so it
+    // must not classify as deferable.
+    let end = (ast.span().hi.0.saturating_sub(fm.start_pos.0)) as usize;
+    if !expr.get(end..).is_some_and(|rest| rest.trim().is_empty()) {
+        return None;
+    }
 
     let snippet = |e: &dyn Spanned| cm.span_to_snippet(e.span()).ok();
 
@@ -239,6 +246,23 @@ mod tests {
         }
     }
 
+    #[test]
+    fn trailing_tokens_stay_eager() {
+        for expr in [
+            r#"variable("f/pw") garbage"#,
+            r#"variable("f/pw"))"#,
+            r#"variable("f/pw");"#,
+            "`x=${variable(\"f/pw\")}` extra",
+        ] {
+            assert_eq!(classify_deferable_variables(expr), None, "expr: {expr}");
+        }
+        // trailing whitespace is fine
+        assert_eq!(
+            classify_deferable_variables("variable(\"f/pw\")  \n"),
+            Some(DeferPlan::WholeVar(PathSpec::Literal("f/pw".to_string())))
+        );
+    }
+
     #[test]
     fn template_without_var_slots_stays_eager() {
         // contains the word variable only in a literal part

backend/windmill-worker/src/worker_flow.rs

@@ -4062,12 +4062,16 @@ async fn push_next_flow_job(
                     }),
                     _ => None,
                 };
-                let defer_variables = matches!(
-                    &value,
-                    Ok(FlowModuleValue::Script { .. }
-                        | FlowModuleValue::RawScript { .. }
-                        | FlowModuleValue::FlowScript { .. })
-                );
+                // cache_ttl steps must keep eager values: the result-cache key
+                // hashes raw args, and a stable `$var:`/`$interpolate` reference
+                // would serve stale cached results across variable rotations.
+                let defer_variables = module.cache_ttl.is_none()
+                    && matches!(
+                        &value,
+                        Ok(FlowModuleValue::Script { .. }
+                            | FlowModuleValue::RawScript { .. }
+                            | FlowModuleValue::FlowScript { .. })
+                    );
                 transform_input(
                     arc_flow_job_args.clone(),
                     flow_env,
@@ -4304,7 +4308,8 @@ async fn push_next_flow_job(
                         resume.clone(),
                         approvers.clone(),
                         None,
-                        // simple loop modules are always leaf scripts
+                        // simple loop modules are always leaf scripts without
+                        // cache_ttl (see `is_simple_modules`)
                         true,
                         &ctx,
                         client,