fix: refuse to move a pipeline draft based on a version the target lacks

A pipeline node deploys with its script hash as parent_hash, so a move to a
workspace without that version left a draft whose Deploy all fails there.

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

backend/.sqlx/query-65148bf40198716608e6b78e51ea606f1a13a5c1889537800710f72619cfb323.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT DISTINCT d.path FROM draft d,\n                   jsonb_array_elements(CASE WHEN jsonb_typeof(d.value::jsonb->'drafts') = 'array'\n                                             THEN d.value::jsonb->'drafts' ELSE '[]' END) node\n               WHERE d.workspace_id = $1 AND d.email = $2 AND d.typ = 'data_pipeline'\n                 AND d.path = ANY($3::text[])\n                 AND CASE WHEN (node->1->'script'->>'hash' ~ '^[0-9a-f]{16}$') IS NOT TRUE THEN false\n                          ELSE NOT EXISTS (SELECT 1 FROM script sc WHERE sc.workspace_id = $4::text\n                                   AND sc.hash = ('x' || (node->1->'script'->>'hash'))::bit(64)::bigint)\n                     END\n               ORDER BY d.path",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "TextArray",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "65148bf40198716608e6b78e51ea606f1a13a5c1889537800710f72619cfb323"
+}

backend/windmill-api-integration-tests/tests/drafts.rs

@@ -554,5 +554,39 @@ async fn test_transfer_drafts_to_fork(db: Pool<Postgres>) -> anyhow::Result<()>
         .await?;
     assert_eq!(r.status(), 400);
 
+    // A pipeline node based on a version the target lacks would fail Deploy all there: refused,
+    // and the source keeps it. One based on a version the target has moves.
+    let pipeline = |hash: &str| json!({ "drafts": [["f/p/n", { "localId": "n", "script": { "path": "f/p/n", "hash": hash } }]] });
+    for (path, hash) in [
+        ("f/p_stale", "00000000000abcde"),
+        ("f/p_ok", "ffffffffffffffff"),
+    ] {
+        sqlx::query(
+            "INSERT INTO draft (workspace_id, path, typ, value, email) VALUES ($1, $2, 'data_pipeline', $3, $4)",
+        )
+        .bind(WS)
+        .bind(path)
+        .bind(pipeline(hash))
+        .bind(me)
+        .execute(&db)
+        .await?;
+    }
+    let move_pipeline = |path: &'static str| {
+        c.post(&transfer_url)
+            .json(&json!({
+                "target_workspace": fork,
+                "items": [{ "kind": "data_pipeline", "path": path }],
+                "remove_from_source": true
+            }))
+            .send()
+    };
+    let r = move_pipeline("f/p_stale").await?;
+    assert_eq!(r.status(), 400, "{}", r.text().await?);
+    assert_eq!(draft_count(&db, "f/p_stale", "data_pipeline", me).await, 1);
+    assert_eq!(fork_count("f/p_stale").await?, 0);
+    let r = move_pipeline("f/p_ok").await?;
+    assert_eq!(r.status(), 200, "{}", r.text().await?);
+    assert_eq!(fork_count("f/p_ok").await?, 1);
+
     Ok(())
 }

backend/windmill-api/src/drafts.rs

@@ -1176,6 +1176,36 @@ async fn transfer_drafts(
             )
             .await?;
         }
+        // A pipeline node deploys with its `script.hash` as `parent_hash`, refused where that
+        // version does not exist, and repointing it would deploy over the target's newer one.
+        let stale_pipelines = sqlx::query_scalar!(
+            r#"SELECT DISTINCT d.path FROM draft d,
+                   jsonb_array_elements(CASE WHEN jsonb_typeof(d.value::jsonb->'drafts') = 'array'
+                                             THEN d.value::jsonb->'drafts' ELSE '[]' END) node
+               WHERE d.workspace_id = $1 AND d.email = $2 AND d.typ = 'data_pipeline'
+                 AND d.path = ANY($3::text[])
+                 AND CASE WHEN (node->1->'script'->>'hash' ~ '^[0-9a-f]{16}$') IS NOT TRUE THEN false
+                          ELSE NOT EXISTS (SELECT 1 FROM script sc WHERE sc.workspace_id = $4::text
+                                   AND sc.hash = ('x' || (node->1->'script'->>'hash'))::bit(64)::bigint)
+                     END
+               ORDER BY d.path"#,
+            &w_id,
+            &authed.email,
+            &locked
+                .iter()
+                .filter(|(k, _)| *k == UserDraftItemKind::DataPipeline)
+                .map(|(_, p)| p.clone())
+                .collect::<Vec<_>>(),
+            target,
+        )
+        .fetch_all(&mut *tx)
+        .await?;
+        if !stale_pipelines.is_empty() {
+            return Err(Error::BadRequest(format!(
+                "{target} lacks a script version these pipeline drafts are based on, so Deploy all would fail there: {}. Deploy them before moving, or pick a workspace that has those versions.",
+                stale_pipelines.join(", ")
+            )));
+        }
         if req.replace_existing {
             sqlx::query!(
                 "DELETE FROM draft