fix: lock the composer during a move and copy only the locked source drafts

- The tab doing a move disables its composer until the move ends, and the
  move checks again for a running turn once a new fork exists.
- Saves still pending in editors open in an existing fork are flushed
  before the transfer, so they surface as conflicts instead of landing on
  the moved draft later.
- The transfer copies, conflict-checks and removes only the source rows it
  locked and validated; a draft saved at a requested key after the lock is
  left out.

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

backend/.sqlx/query-b951a9ebcac1ac5e4fe4f763bafcc95db906a50d063585be8c4e34c1a233f2b1.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT typ AS \"typ: UserDraftItemKind\", value::text AS \"value!\" FROM draft\n           WHERE workspace_id = $1 AND email = $2\n             AND (typ::text, path) IN (SELECT * FROM unnest($3::text[], $4::text[]))\n           FOR UPDATE",
+  "query": "SELECT typ AS \"typ: UserDraftItemKind\", path, value::text AS \"value!\" FROM draft\n           WHERE workspace_id = $1 AND email = $2\n             AND (typ::text, path) IN (SELECT * FROM unnest($3::text[], $4::text[]))\n           FOR UPDATE",
   "describe": {
     "columns": [
       {
@@ -44,6 +44,11 @@
       },
       {
         "ordinal": 1,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
         "name": "value!",
         "type_info": "Text"
       }
@@ -57,9 +62,10 @@
       ]
     },
     "nullable": [
+      false,
       false,
       null
     ]
   },
-  "hash": "f20626f8db303b11f7bd258274559f8bb5227c70c31a3162595a7f07b2298e4f"
+  "hash": "b951a9ebcac1ac5e4fe4f763bafcc95db906a50d063585be8c4e34c1a233f2b1"
 }

backend/windmill-api/src/drafts.rs

@@ -1089,8 +1089,10 @@ async fn transfer_drafts(
 
     let mut tx = db.begin().await?;
     // Holds off a concurrent save of these rows, so the version removed below is the one copied.
+    // A requested key with no row takes no lock and a draft saved there meanwhile skips the
+    // checks below, so everything after this works on the locked keys only.
     let source_rows = sqlx::query!(
-        r#"SELECT typ AS "typ: UserDraftItemKind", value::text AS "value!" FROM draft
+        r#"SELECT typ AS "typ: UserDraftItemKind", path, value::text AS "value!" FROM draft
            WHERE workspace_id = $1 AND email = $2
              AND (typ::text, path) IN (SELECT * FROM unnest($3::text[], $4::text[]))
            FOR UPDATE"#,
@@ -1101,6 +1103,12 @@ async fn transfer_drafts(
     )
     .fetch_all(&mut *tx)
     .await?;
+    let locked: Vec<(UserDraftItemKind, String)> = source_rows
+        .iter()
+        .map(|r| (r.typ, r.path.clone()))
+        .collect();
+    let locked_kinds: Vec<&str> = locked.iter().map(|(k, _)| k.as_str()).collect();
+    let locked_paths: Vec<&str> = locked.iter().map(|(_, p)| p.as_str()).collect();
     let mut copied = vec![];
     if let (Some(target), Some(target_authed)) =
         (req.target_workspace.as_deref(), target_authed.as_ref())
@@ -1131,8 +1139,8 @@ async fn transfer_drafts(
                FOR UPDATE"#,
             target,
             &authed.email,
-            &kinds as &[&str],
-            &paths as &[&str],
+            &locked_kinds as &[&str],
+            &locked_paths as &[&str],
         )
         .fetch_all(&mut *tx)
         .await?;
@@ -1148,7 +1156,7 @@ async fn transfer_drafts(
             &w_id,
             target,
             &authed.email,
-            Some(&items),
+            Some(&locked),
         )
         .await?;
     }
@@ -1162,8 +1170,8 @@ async fn transfer_drafts(
                RETURNING typ AS "typ: UserDraftItemKind", path"#,
             &w_id,
             &authed.email,
-            &kinds as &[&str],
-            &paths as &[&str],
+            &locked_kinds as &[&str],
+            &locked_paths as &[&str],
         )
         .fetch_all(&mut *tx)
         .await?

frontend/src/lib/components/sessions/SessionWrapper.svelte

@@ -738,12 +738,14 @@
 							initialAttachments={armedAtInit ? undefined : getSessionDraftAttachments(sessionId)}
 							onAttachmentsChange={(a) => setSessionDraftAttachments(sessionId, a)}
 							placeholder={composerPlaceholder}
-							forceDisabled={isUnavailable || !!session.archived}
+							forceDisabled={isUnavailable || !!session.archived || moving}
 							forceDisabledMessage={isUnavailable
 								? 'This session is linked to a workspace that no longer exists. Move it or discard it from the banner above to keep working.'
 								: session.archived
 									? 'This session is archived. Unarchive it from the banner above to keep working.'
-									: ''}
+									: moving
+										? `Moving this session to ${moveRequest?.targetName ?? 'the fork'}…`
+										: ''}
 							starter={opensSession
 								? { where: 'session', suggestions: starterPrompts.suggestions }
 								: undefined}

frontend/src/lib/components/sessions/moveSessionToFork.ts

@@ -93,15 +93,22 @@ export async function moveSessionToFork(
 ): Promise<MoveSessionResult> {
 	const unsaved = await flushPendingSaves(parent)
 	if (unsaved) return { ok: false, error: unsaved }
-	if (opts.isBusy?.()) return { ok: false, error: 'A chat turn is running. Retry once it ends.' }
+	const busy = 'A chat turn is running. Retry once it ends.'
+	if (opts.isBusy?.()) return { ok: false, error: busy }
 
 	let fork: Extract<MoveTarget, { kind: 'existing' }>
 	if (target.kind === 'new') {
 		const made = await materializeForkOutcome(target.fork)
 		if (!made) return { ok: false, error: 'Could not create the fork' }
 		fork = { kind: 'existing', id: made.id, replace: made.created }
+		// Creating the fork takes seconds, long enough for a job's auto-resume to start a turn.
+		if (opts.isBusy?.()) return { ok: false, error: busy, retry: fork }
 	} else {
 		fork = target
+		// An editor open in the fork may hold edits not yet saved. Saved now, they become the
+		// conflict they are; still parked, they would land on top of the moved draft later.
+		const unsavedThere = await flushPendingSaves(fork.id)
+		if (unsavedThere) return { ok: false, error: unsavedThere, retry: fork }
 	}
 
 	let copied = 0