diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs
index e39fd6ea43..6c34db8169 100644
--- a/backend/windmill-common/src/cache.rs
+++ b/backend/windmill-common/src/cache.rs
@@ -1189,11 +1189,25 @@ const _: () = {
             use std::fs::OpenOptions;
             use std::io::Write;
 
+            // Atomic write: truncate+write a uniquely-named temp file (UUID, not pid — pids
+            // collide across container PID namespaces on a shared cache volume), fsync, then
+            // rename(2) over the target. Without this a shorter overwrite leaves a stale tail
+            // and concurrent writers tear the file — a corrupt entry a reader would import.
+            let final_path = item.path(self);
+            let tmp_path = final_path.with_extension(format!("tmp.{}", Uuid::new_v4()));
             OpenOptions::new()
                 .write(true)
                 .create(true)
-                .open(item.path(self))
-                .and_then(|mut file| file.write_all(data.as_ref()))
+                .truncate(true)
+                .open(&tmp_path)
+                .and_then(|mut file| {
+                    file.write_all(data.as_ref())?;
+                    file.sync_all()
+                })
+                .and_then(|()| std::fs::rename(&tmp_path, &final_path))
+                .inspect_err(|_| {
+                    let _ = std::fs::remove_file(&tmp_path);
+                })
         }
     }
 
@@ -1237,6 +1251,33 @@ mod tests {
     use super::*;
     use serde_json::json;
 
+    #[test]
+    fn fs_cache_put_overwrites_without_stale_tail() {
+        // Regression for the non-truncating, non-atomic `put`: overwriting a value with a
+        // shorter one must not leave stale trailing bytes (which imported as corrupt/wrong
+        // content — the #9751 worker-cache hazard).
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+
+        root.put("k", b"a-long-cached-value-0123456789").unwrap();
+        assert_eq!(root.get("k").unwrap(), b"a-long-cached-value-0123456789");
+
+        root.put("k", b"short").unwrap();
+        assert_eq!(
+            root.get("k").unwrap(),
+            b"short",
+            "shorter overwrite must fully replace, no stale tail"
+        );
+
+        // No temp files left behind after a successful write.
+        let leftover: Vec<_> = std::fs::read_dir(root)
+            .unwrap()
+            .filter_map(|e| e.ok())
+            .filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
+            .collect();
+        assert!(leftover.is_empty(), "temp files must be renamed/cleaned up");
+    }
+
     #[test]
     fn flow_data_extras_preserves_notes_and_groups() {
         let raw = serde_json::value::to_raw_value(&json!({
diff --git a/backend/windmill-types/src/flows.rs b/backend/windmill-types/src/flows.rs
index 2d1a538124..006185dbd9 100644
--- a/backend/windmill-types/src/flows.rs
+++ b/backend/windmill-types/src/flows.rs
@@ -155,6 +155,19 @@ fn validate_retry(retry: &Retry, module_id: &str) -> anyhow::Result<()> {
     Ok(())
 }
 
+/// Script/sub-flow step references must be workspace paths (`u/`, `f/`, `g/`) or a hub
+/// reference (`hub/`). Empty is tolerated for intermediate/incomplete steps. This blocks
+/// absolute or local filesystem paths (e.g. `/tmp/.../ops/scripts/...` baked in by a
+/// `wmill sync push` from a feature-branch checkout) from being persisted into a flow,
+/// where they silently mis-resolve to an unrelated script at runtime (#9751).
+fn is_workspace_runnable_path(path: &str) -> bool {
+    path.is_empty()
+        || path.starts_with("u/")
+        || path.starts_with("f/")
+        || path.starts_with("g/")
+        || path.starts_with("hub/")
+}
+
 fn validate_flow_value<'de, D>(deserializer: D) -> Result<Box<RawValue>, D::Error>
 where
     D: Deserializer<'de>,
@@ -164,21 +177,38 @@ where
     let flow_value: FlowValue = serde_json::from_str(raw_value.get())
         .map_err(|e| serde::de::Error::custom(format!("Invalid flow value: {}", e)))?;
 
-    FlowModule::traverse_modules(&flow_value.modules, &mut |module| {
+    let mut validate_module = |module: &FlowModule| -> anyhow::Result<()> {
         if let Some(ref retry) = module.retry {
             validate_retry(retry, &module.id)?;
         }
-        return Ok(());
-    })
-    .map_err(|e| serde::de::Error::custom(e.to_string()))?;
+        if let Ok(FlowModuleValue::Script { path, .. } | FlowModuleValue::Flow { path, .. }) =
+            module.get_value()
+        {
+            if !is_workspace_runnable_path(&path) {
+                return Err(anyhow::anyhow!(
+                    "step '{}' references '{}', which is not a workspace path (expected u/, \
+                     f/, g/ or hub/). Absolute or local filesystem paths are not allowed in \
+                     flow steps.",
+                    module.id,
+                    path
+                ));
+            }
+        }
+        Ok(())
+    };
 
-    if let Some(ref _failure_module) = flow_value.failure_module {
-        //add validation logic here for failure module
-    }
-
-    if let Some(ref _preprocessor_module) = flow_value.preprocessor_module {
-        //add validation logic here for preprocessor module
-    }
+    // The API is the authoritative guard (it can be called directly, bypassing the CLI), so
+    // it must cover every step that resolves a path: the main modules AND the failure /
+    // preprocessor modules (which can themselves be sub-flows/loops/branches).
+    let extra_modules: Vec<FlowModule> = flow_value
+        .failure_module
+        .iter()
+        .chain(flow_value.preprocessor_module.iter())
+        .map(|m| (**m).clone())
+        .collect();
+    FlowModule::traverse_modules(&flow_value.modules, &mut validate_module)
+        .and_then(|()| FlowModule::traverse_modules(&extra_modules, &mut validate_module))
+        .map_err(|e| serde::de::Error::custom(e.to_string()))?;
 
     Ok(raw_value)
 }
@@ -1228,6 +1258,108 @@ mod tests {
         assert_eq!(val.modules.len(), 1);
     }
 
+    #[test]
+    fn flow_rejects_absolute_step_path() {
+        // #9751: an absolute local path baked into a step must be rejected on deploy.
+        let bad = json!({
+            "path": "f/test/flow",
+            "summary": "",
+            "value": { "modules": [{
+                "id": "validate_onboard_target",
+                "value": {
+                    "type": "script",
+                    "path": "/tmp/tmp.X/f/ops/scripts/clean_device/pre_clean",
+                    "input_transforms": {}
+                }
+            }]}
+        });
+        let err = serde_json::from_value::<NewFlow>(bad)
+            .unwrap_err()
+            .to_string();
+        assert!(
+            err.contains("not a workspace path"),
+            "unexpected error: {err}"
+        );
+        assert!(
+            err.contains("validate_onboard_target"),
+            "error should name the step: {err}"
+        );
+    }
+
+    #[test]
+    fn flow_rejects_absolute_step_path_in_nested_module() {
+        let bad = json!({
+            "path": "f/test/flow",
+            "summary": "",
+            "value": { "modules": [{
+                "id": "loop",
+                "value": {
+                    "type": "forloopflow",
+                    "iterator": {"type": "javascript", "expr": "[1]"},
+                    "modules": [{
+                        "id": "inner",
+                        "value": {"type": "script", "path": "/abs/path", "input_transforms": {}}
+                    }]
+                }
+            }]}
+        });
+        let err = serde_json::from_value::<NewFlow>(bad)
+            .unwrap_err()
+            .to_string();
+        assert!(
+            err.contains("not a workspace path"),
+            "unexpected error: {err}"
+        );
+    }
+
+    #[test]
+    fn flow_rejects_absolute_path_in_failure_and_preprocessor_modules() {
+        for slot in ["failure_module", "preprocessor_module"] {
+            // Build the value with the slot as an explicit (interpolated) key.
+            let mut value = serde_json::Map::new();
+            value.insert("modules".to_string(), json!([]));
+            value.insert(
+                slot.to_string(),
+                json!({
+                    "id": slot,
+                    "value": {"type": "script", "path": "/abs/path", "input_transforms": {}}
+                }),
+            );
+            let bad = json!({ "path": "f/test/flow", "summary": "", "value": value });
+            let err = serde_json::from_value::<NewFlow>(bad)
+                .unwrap_err()
+                .to_string();
+            assert!(
+                err.contains("not a workspace path"),
+                "{slot} should be validated, got: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn flow_accepts_workspace_step_paths() {
+        for p in [
+            "f/ops/scripts/x",
+            "u/me/y",
+            "g/grp/z",
+            "hub/123/foo",
+            "", // tolerated for incomplete steps
+        ] {
+            let ok = json!({
+                "path": "f/test/flow",
+                "summary": "",
+                "value": { "modules": [{
+                    "id": "a",
+                    "value": {"type": "script", "path": p, "input_transforms": {}}
+                }]}
+            });
+            assert!(
+                serde_json::from_value::<NewFlow>(ok).is_ok(),
+                "path {p:?} should be accepted"
+            );
+        }
+    }
+
     #[test]
     fn ai_agent_omit_output_from_conversation_defaults_to_false() {
         let input = json!({
diff --git a/cli/src/commands/flow/flow.ts b/cli/src/commands/flow/flow.ts
index c0e94c478d..71d90af6fe 100644
--- a/cli/src/commands/flow/flow.ts
+++ b/cli/src/commands/flow/flow.ts
@@ -135,6 +135,31 @@ function warnAboutLocalPathScriptDivergence(
 
 const alreadySynced: string[] = [];
 
+// Collect every script/sub-flow step path in a flow value — recursively through loops,
+// branches, and the failure/preprocessor modules — for workspace-path validation. Unlike
+// `collectPathScriptPaths` this also includes `type: "flow"` sub-flow steps.
+function collectStepPaths(flowValue: any): string[] {
+  const paths: string[] = [];
+  const walk = (modules: any[] | undefined) => {
+    for (const m of modules ?? []) {
+      const v = m?.value;
+      if (!v) continue;
+      if ((v.type === "script" || v.type === "flow") && typeof v.path === "string") {
+        paths.push(v.path);
+      }
+      walk(v.modules);
+      walk(v.default);
+      for (const b of v.branches ?? []) walk(b?.modules);
+      // AI-agent tools are step-like and can carry script paths too.
+      walk(v.tools);
+    }
+  };
+  walk(flowValue?.modules);
+  if (flowValue?.failure_module) walk([flowValue.failure_module]);
+  if (flowValue?.preprocessor_module) walk([flowValue.preprocessor_module]);
+  return paths;
+}
+
 export async function pushFlow(
   workspace: string,
   remotePath: string,
@@ -190,6 +215,21 @@ export async function pushFlow(
     );
   }
 
+  // Reject script/sub-flow steps whose path is not a workspace path (u/, f/, g/ or hub/).
+  // A flow.yaml generated from a feature-branch checkout can carry absolute local paths
+  // (e.g. /tmp/.../ops/scripts/...); pushed, they silently mis-resolve at runtime (#9751).
+  // The backend re-validates the same rule for every step type, so this is a fail-fast.
+  const badStepPaths = collectStepPaths(localFlow.value).filter(
+    (p) => p !== "" && !/^(u|f|g|hub)\//.test(p)
+  );
+  if (badStepPaths.length > 0) {
+    throw new Error(
+      `Cannot push flow ${remotePath}: step(s) reference non-workspace path(s): ${badStepPaths.join(", ")}. ` +
+      `Flow step paths must be workspace paths (u/, f/, g/ or hub/), not absolute or local filesystem paths. ` +
+      `This usually means flow.yaml was generated with paths from a checkout directory.`
+    );
+  }
+
   const hasOnBehalfOf = (localFlow as any).has_on_behalf_of ?? !!localFlow.on_behalf_of_email;
   delete (localFlow as any).has_on_behalf_of;