From a5b451cc2b3baf448c15f30d752e57f9df92bf37 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Mon, 26 Oct 2026 08:23:20 -0700 Subject: [PATCH] fix: re-copy drafts into a new fork and keep other tabs in step with a move - A new fork now gets the same transfer as an existing one, with replace_existing set only when this move created the fork, so a draft saved after the clone reaches the fork and an existing fork's own drafts still go through the conflict check. - The transfer runs the operator composition check on each draft in the target workspace, the same check as a direct save there. - Editors already open on a moved item in the fork reload it from the server. - Other tabs on the session hold their composer during a move and adopt the session's new workspace when it ends. Co-Authored-By: Claude Opus 5.5 (1M context) --- ...24603e7390ef36aebaf53afffa7e3c2a7be04.json | 25 --- ...7035261318b5c8401059c1c2c73a1b7e62e8b.json | 17 ++ ...59f8bb5227c70c31a3162595a7f07b2298e4f.json | 65 +++++++ .../tests/drafts.rs | 25 ++- backend/windmill-api/openapi.yaml | 5 +- backend/windmill-api/src/drafts.rs | 158 +++++++++++------- backend/windmill-common/src/user_drafts.rs | 3 +- .../components/sessions/SessionWrapper.svelte | 19 ++- .../sessions/moveSessionToFork.test.ts | 42 ++++- .../components/sessions/moveSessionToFork.ts | 132 +++++++++------ .../sessions/sessionRuntime.svelte.ts | 4 + .../sessions/sessionState.svelte.ts | 44 ++++- frontend/src/lib/userDraft.svelte.ts | 5 + 13 files changed, 390 insertions(+), 154 deletions(-) delete mode 100644 backend/.sqlx/query-2e4e783de828f8e33a98a1b082e24603e7390ef36aebaf53afffa7e3c2a7be04.json create mode 100644 backend/.sqlx/query-ab6c20641004ad0a93e379cd2287035261318b5c8401059c1c2c73a1b7e62e8b.json create mode 100644 backend/.sqlx/query-f20626f8db303b11f7bd258274559f8bb5227c70c31a3162595a7f07b2298e4f.json diff --git a/backend/.sqlx/query-2e4e783de828f8e33a98a1b082e24603e7390ef36aebaf53afffa7e3c2a7be04.json b/backend/.sqlx/query-2e4e783de828f8e33a98a1b082e24603e7390ef36aebaf53afffa7e3c2a7be04.json deleted file mode 100644 index e608f4e7e0..0000000000 --- a/backend/.sqlx/query-2e4e783de828f8e33a98a1b082e24603e7390ef36aebaf53afffa7e3c2a7be04.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT 1 AS \"one\" 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": [ - { - "ordinal": 0, - "name": "one", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "TextArray", - "TextArray" - ] - }, - "nullable": [ - null - ] - }, - "hash": "2e4e783de828f8e33a98a1b082e24603e7390ef36aebaf53afffa7e3c2a7be04" -} diff --git a/backend/.sqlx/query-ab6c20641004ad0a93e379cd2287035261318b5c8401059c1c2c73a1b7e62e8b.json b/backend/.sqlx/query-ab6c20641004ad0a93e379cd2287035261318b5c8401059c1c2c73a1b7e62e8b.json new file mode 100644 index 0000000000..bd4d03780c --- /dev/null +++ b/backend/.sqlx/query-ab6c20641004ad0a93e379cd2287035261318b5c8401059c1c2c73a1b7e62e8b.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM draft\n WHERE workspace_id = $1 AND email = $2\n AND (typ::text, path) IN (SELECT * FROM unnest($3::text[], $4::text[]))", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "TextArray", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "ab6c20641004ad0a93e379cd2287035261318b5c8401059c1c2c73a1b7e62e8b" +} diff --git a/backend/.sqlx/query-f20626f8db303b11f7bd258274559f8bb5227c70c31a3162595a7f07b2298e4f.json b/backend/.sqlx/query-f20626f8db303b11f7bd258274559f8bb5227c70c31a3162595a7f07b2298e4f.json new file mode 100644 index 0000000000..2579d6fbb7 --- /dev/null +++ b/backend/.sqlx/query-f20626f8db303b11f7bd258274559f8bb5227c70c31a3162595a7f07b2298e4f.json @@ -0,0 +1,65 @@ +{ + "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", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "typ: UserDraftItemKind", + "type_info": { + "Custom": { + "name": "draft_kind", + "kind": { + "Enum": [ + "script", + "flow", + "app", + "raw_app", + "resource", + "variable", + "trigger_schedule", + "trigger_webhook", + "trigger_default_email", + "trigger_email", + "trigger_http", + "trigger_websocket", + "trigger_postgres", + "trigger_kafka", + "trigger_nats", + "trigger_mqtt", + "trigger_sqs", + "trigger_gcp", + "trigger_azure", + "trigger_poll", + "trigger_cli", + "trigger_nextcloud", + "trigger_google", + "trigger_github", + "data_pipeline", + "trigger_amqp" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "value!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "TextArray", + "TextArray" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "f20626f8db303b11f7bd258274559f8bb5227c70c31a3162595a7f07b2298e4f" +} diff --git a/backend/windmill-api-integration-tests/tests/drafts.rs b/backend/windmill-api-integration-tests/tests/drafts.rs index 2c0ab54c42..1f7c62db6b 100644 --- a/backend/windmill-api-integration-tests/tests/drafts.rs +++ b/backend/windmill-api-integration-tests/tests/drafts.rs @@ -313,8 +313,8 @@ async fn test_cross_user_draft_privacy(db: Pool) -> anyhow::Result<()> } /// Moving a session's drafts into a fork is all-or-nothing: a draft the caller already has in -/// the fork blocks the whole transfer rather than one side silently winning, and the source -/// copy is removed only together with a successful copy. +/// the fork blocks the whole transfer rather than one side silently winning, unless the caller +/// asks to replace it, and the source copy is removed only together with a successful copy. #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn test_transfer_drafts_to_fork(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; @@ -394,13 +394,19 @@ async fn test_transfer_drafts_to_fork(db: Pool) -> anyhow::Result<()> assert_eq!(draft_count(&db, "u/test-user/a", "script", me).await, 1); assert_eq!(draft_count(&db, "u/test-user/b", "script", me).await, 1); - sqlx::query("DELETE FROM draft WHERE workspace_id = $1 AND path = 'u/test-user/a'") - .bind(fork) - .execute(&db) + // A save after the clone: replacing the clone's copy must carry it into the fork. + let r = c + .post(save_url(port, "script", "u/test-user/a")) + .json(&json!({ "value": { "content": "a2" } })) + .send() .await?; + assert_eq!(r.status(), 200); let body: serde_json::Value = c .post(&transfer_url) - .json(&json!({ "target_workspace": fork, "items": items, "remove_from_source": true })) + .json(&json!({ + "target_workspace": fork, "items": items, + "remove_from_source": true, "replace_existing": true + })) .send() .await? .json() @@ -410,6 +416,13 @@ async fn test_transfer_drafts_to_fork(db: Pool) -> anyhow::Result<()> assert_eq!(body["removed"].as_array().unwrap().len(), 2); assert_eq!(fork_count("u/test-user/a").await?, 1); assert_eq!(fork_count("u/test-user/b").await?, 1); + let a_in_fork: String = sqlx::query_scalar( + "SELECT value->>'content' FROM draft WHERE workspace_id = $1 AND path = 'u/test-user/a'", + ) + .bind(fork) + .fetch_one(&db) + .await?; + assert_eq!(a_in_fork, "a2"); assert_eq!(draft_count(&db, "u/test-user/a", "script", me).await, 0); assert_eq!(draft_count(&db, "u/test-user/b", "script", me).await, 0); diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 43fc4b1af0..e4d9c709a5 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -11283,7 +11283,7 @@ paths: /w/{workspace}/drafts/transfer: post: summary: copy the current user's drafts into a fork and/or remove them here - description: In one transaction, copies the authed user's own drafts of `items` into `target_workspace` (a direct fork of this workspace) and, with `remove_from_source`, deletes them here. Without `target_workspace` it only removes. If the user already has a draft of any item in the target, nothing is written and the items are returned in `conflicts`. + description: In one transaction, copies the authed user's own drafts of `items` into `target_workspace` (a direct fork of this workspace) and, with `remove_from_source`, deletes them here. Without `target_workspace` it only removes. If the user already has a draft of any item in the target, nothing is written and the items are returned in `conflicts`, unless `replace_existing` is set. A user who can only compose flows or apps in the target is held to the same rules as a direct draft save there. operationId: transferDrafts tags: - draft @@ -11304,6 +11304,9 @@ paths: $ref: "#/components/schemas/DraftItemRef" remove_from_source: type: boolean + replace_existing: + type: boolean + description: Overwrite the user's own drafts of `items` already in the target instead of returning them as conflicts. required: [items] responses: "200": diff --git a/backend/windmill-api/src/drafts.rs b/backend/windmill-api/src/drafts.rs index 8d3d27eb4a..9d753a2b32 100644 --- a/backend/windmill-api/src/drafts.rs +++ b/backend/windmill-api/src/drafts.rs @@ -385,6 +385,69 @@ fn draft_lineage(kind: UserDraftItemKind, value: &str) -> Option Result<()> { + let caps = windmill_common::roles::caller_capabilities(db, w_id, authed).await?; + if caps.flows == AuthorLevel::Compose && kind == UserDraftItemKind::Flow { + #[derive(Deserialize)] + struct FlowDraft { + #[serde(default)] + value: FlowValue, + schema: Option>, + tag: Option, + } + // The text as stored, NULs stripped: a stripped NUL can rename a key into one this + // check reads. + let draft: FlowDraft = serde_json::from_str(&strip_json_nul(value)) + .map_err(|e| Error::BadRequest(format!("Invalid flow draft: {e}")))?; + validate_operator_flow( + &draft.value, + &draft.tag, + draft.schema.as_deref().map(|s| s.get()), + authed, + db, + user_db, + w_id, + ) + .await?; + } + + // A raw-app draft carries the app value and its policy side by side. + if caps.apps == AuthorLevel::Compose && kind == UserDraftItemKind::RawApp { + let stripped = strip_json_nul(value); + let draft: serde_json::Value = serde_json::from_str(&stripped) + .map_err(|e| Error::BadRequest(format!("Invalid app draft: {e}")))?; + let mut policy = draft + .get("policy") + .filter(|p| !p.is_null()) + .map(|p| serde_json::from_value::(p.clone())) + .transpose() + .map_err(|e| Error::BadRequest(format!("Invalid app draft policy: {e}")))?; + let raw = serde_json::value::RawValue::from_string(stripped.into_owned()) + .map_err(|e| Error::BadRequest(format!("Invalid app draft: {e}")))?; + crate::apps::validate_operator_composed_app( + authed, + db, + user_db, + w_id, + crate::apps::BuilderAppWrite::Draft, + true, + Some(&raw), + policy.as_mut(), + ) + .await?; + } + Ok(()) +} + /// Apply the current user's draft: non-null `value` upserts, `null` (or /// omitted) deletes. Either way, when the existing row is newer than /// `last_sync` (and `force` is false) the op is skipped and the response is @@ -471,58 +534,8 @@ async fn update_draft( } } - // A role that composes but does not author in full is held to the composition-only rules on - // its drafts as on its deploys. - let caps = windmill_common::roles::caller_capabilities(&db, &w_id, &authed).await?; - if caps.flows == AuthorLevel::Compose && kind == UserDraftItemKind::Flow { - if let Some(value) = &req.value { - #[derive(Deserialize)] - struct FlowDraft { - #[serde(default)] - value: FlowValue, - schema: Option>, - tag: Option, - } - // The text stored below, NULs stripped: a stripped NUL can rename a key into one - // this check reads. - let draft: FlowDraft = serde_json::from_str(&strip_json_nul(value.0.get())) - .map_err(|e| Error::BadRequest(format!("Invalid flow draft: {e}")))?; - validate_operator_flow( - &draft.value, - &draft.tag, - draft.schema.as_deref().map(|s| s.get()), - &authed, - &db, - &user_db, - &w_id, - ) - .await?; - } - } - - // A raw-app draft carries the app value and its policy side by side. - if caps.apps == AuthorLevel::Compose && kind == UserDraftItemKind::RawApp { - if let Some(value) = &req.value { - let draft: serde_json::Value = serde_json::from_str(&strip_json_nul(value.0.get())) - .map_err(|e| Error::BadRequest(format!("Invalid app draft: {e}")))?; - let mut policy = draft - .get("policy") - .filter(|p| !p.is_null()) - .map(|p| serde_json::from_value::(p.clone())) - .transpose() - .map_err(|e| Error::BadRequest(format!("Invalid app draft policy: {e}")))?; - crate::apps::validate_operator_composed_app( - &authed, - &db, - &user_db, - &w_id, - crate::apps::BuilderAppWrite::Draft, - true, - Some(&value.0), - policy.as_mut(), - ) - .await?; - } + if let Some(value) = &req.value { + validate_composed_draft(&authed, &db, &user_db, &w_id, kind, value.0.get()).await?; } let applied = if let Some(value) = &req.value { @@ -969,6 +982,12 @@ pub struct TransferDraftsRequest { pub items: Vec, #[serde(default)] pub remove_from_source: bool, + /// Overwrite the caller's own drafts of `items` already in the target instead of + /// reporting them as conflicts. For a fork created moments ago, whose only such rows are + /// the fork clone's copies of the very drafts being moved: a save made after the clone + /// must reach the fork before its parent's copy is removed. + #[serde(default)] + pub replace_existing: bool, } #[derive(Serialize)] @@ -1005,6 +1024,7 @@ async fn transfer_drafts( .collect() }; + let mut target_authed = None; if let Some(target) = req.target_workspace.as_deref() { let is_fork = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM workspace WHERE id = $1 AND parent_workspace_id = $2 AND deleted = false)", @@ -1047,7 +1067,7 @@ async fn transfer_drafts( } // `authed` speaks for this workspace only; the copy writes into the fork, so its // rights there are loaded separately. A non-member gets an empty authed, not an error. - let target_authed = + let authed_there = windmill_api_workspaces::workspaces::load_workspace_authed(&db, &authed, target) .await?; let is_member = sqlx::query_scalar!( @@ -1058,18 +1078,19 @@ async fn transfer_drafts( .fetch_one(&db) .await? .unwrap_or(false); - if !is_member && !target_authed.is_admin { + if !is_member && !authed_there.is_admin { return Err(Error::PermissionDenied(format!("not a member of {target}"))); } for (kind, path) in &items { - require_can_write_path(&target_authed, &db, &user_db, target, *kind, path).await?; + require_can_write_path(&authed_there, &db, &user_db, target, *kind, path).await?; } + target_authed = Some(authed_there); } let mut tx = db.begin().await?; // Holds off a concurrent save of these rows, so the version removed below is the one copied. - sqlx::query!( - r#"SELECT 1 AS "one" FROM draft + let source_rows = sqlx::query!( + r#"SELECT typ AS "typ: UserDraftItemKind", 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"#, @@ -1081,7 +1102,28 @@ async fn transfer_drafts( .fetch_all(&mut *tx) .await?; let mut copied = vec![]; - if let Some(target) = req.target_workspace.as_deref() { + if let (Some(target), Some(target_authed)) = + (req.target_workspace.as_deref(), target_authed.as_ref()) + { + // The copy is a save in the target, so a role that only composes there is held to the + // rules a direct save would apply. Checked on the locked rows, the ones copied. + for row in &source_rows { + validate_composed_draft(target_authed, &db, &user_db, target, row.typ, &row.value) + .await?; + } + if req.replace_existing { + sqlx::query!( + "DELETE FROM draft + WHERE workspace_id = $1 AND email = $2 + AND (typ::text, path) IN (SELECT * FROM unnest($3::text[], $4::text[]))", + target, + &authed.email, + &kinds as &[&str], + &paths as &[&str], + ) + .execute(&mut *tx) + .await?; + } let conflicts = sqlx::query!( r#"SELECT typ AS "typ: UserDraftItemKind", path FROM draft WHERE workspace_id = $1 AND email = $2 diff --git a/backend/windmill-common/src/user_drafts.rs b/backend/windmill-common/src/user_drafts.rs index 8bee972a56..1a4316415e 100644 --- a/backend/windmill-common/src/user_drafts.rs +++ b/backend/windmill-common/src/user_drafts.rs @@ -670,7 +670,8 @@ pub async fn delete_own_draft_for_path( /// there. Secret values are copied as encrypted, so the two workspaces must share a key. /// /// A plain INSERT: the caller guarantees the target holds none of the copied rows (an empty -/// fork, or a conflict check under the same transaction), else `draft_pkey_with_user` aborts it. +/// fork, or a conflict check or delete under the same transaction), else `draft_pkey_with_user` +/// aborts it. /// `created_at` is preserved so the per-tab `last_sync` baseline the editor reads lines up with /// the source's timeline — otherwise the first save from an open editor would trip the conflict /// modal on every copied draft. diff --git a/frontend/src/lib/components/sessions/SessionWrapper.svelte b/frontend/src/lib/components/sessions/SessionWrapper.svelte index 68afde0c5f..df530d8a4e 100644 --- a/frontend/src/lib/components/sessions/SessionWrapper.svelte +++ b/frontend/src/lib/components/sessions/SessionWrapper.svelte @@ -34,6 +34,7 @@ import SessionMoveModal from './SessionMoveModal.svelte' import { flushPendingSaves, moveSessionToFork, type MoveTarget } from './moveSessionToFork' import { maskHasDraftRow } from './modifiedItemsMask' + import { localRunEnded, localRunStarted } from './sessionSync.svelte' import { sendUserToast } from '$lib/toast' import Toggle from '$lib/components/Toggle.svelte' import { copilotInfo, copilotWorkspace } from '$lib/aiStore' @@ -480,6 +481,10 @@ if (!session || !req || moving) return moving = true const releaseAutoResume = runtime?.manager.holdAutoResume() + // Bracketed like a turn, so another tab on this session cannot send while its drafts + // move, then re-reads the session (its new workspace, the model note) once it ends. + const chatId = runtime?.manager.historyManager.getCurrentChatId() + if (chatId) localRunStarted(session.id, chatId) try { const res = await moveSessionToFork(session.id, req.parent, req.target, { items: req.items.map(({ kind, path }) => ({ kind, path })), @@ -491,12 +496,17 @@ sendUserToast(res.error, true) moveRequest = undefined } else { - moveRequest = { ...req, error: res.error, conflicts: res.conflicts ?? [] } + moveRequest = { + ...req, + target: res.retry ?? req.target, + error: res.error, + conflicts: res.conflicts ?? [] + } } return } moveRequest = undefined - void runtime?.manager + await runtime?.manager .queueModelNote( `This session moved from workspace "${req.parent}" to its fork "${res.forkId}": every further tool call acts on "${res.forkId}". ` + (res.copied > 0 @@ -508,10 +518,11 @@ .catch((e) => console.error('Failed to persist the move note', e)) runtime?.manager.rebuildGlobalSystemMessage() sendUserToast(`Session moved to ${req.targetName}`) - if (res.warning) sendUserToast(res.warning, true) } finally { moving = false releaseAutoResume?.() + if (chatId) + localRunEnded(session.id, runtime?.manager.historyManager.getCurrentChatId() ?? chatId) } } @@ -833,7 +844,7 @@ open={moveRequest.open} parentName={moveRequest.parentName} targetName={moveRequest.targetName} - newFork={moveRequest.target.kind === 'new'} + newFork={moveRequest.target.kind === 'new' || !!moveRequest.target.replace} items={moveRequest.items} conflicts={moveRequest.conflicts} error={moveRequest.error} diff --git a/frontend/src/lib/components/sessions/moveSessionToFork.test.ts b/frontend/src/lib/components/sessions/moveSessionToFork.test.ts index 6deef29a0d..f044802946 100644 --- a/frontend/src/lib/components/sessions/moveSessionToFork.test.ts +++ b/frontend/src/lib/components/sessions/moveSessionToFork.test.ts @@ -2,7 +2,7 @@ import { describe, it, expect, vi, afterEach } from 'vitest' import { get } from 'svelte/store' import { DraftService } from '$lib/gen' import { workspaceStore } from '$lib/stores' -import { sessionState, type Session } from './sessionState.svelte' +import { materializeForkOutcome, sessionState, type Session } from './sessionState.svelte' import { moveSessionToFork } from './moveSessionToFork' vi.mock('$lib/gen', async (orig) => { @@ -13,6 +13,11 @@ vi.mock('$lib/gen', async (orig) => { } }) +vi.mock('./sessionState.svelte', async (orig) => { + const actual = await orig() + return { ...actual, materializeForkOutcome: vi.fn() } +}) + const items = [{ kind: 'script' as const, path: 'u/a/s' }] function addSession(): Session { @@ -77,4 +82,39 @@ describe('moveSessionToFork', () => { workspaceStore.set(prev) } }) + + it.each([ + [true, true], + [false, false] + ])( + 're-copies into a new fork, overwriting the clone only when this move created it (created %s)', + async (created, replace) => { + vi.mocked(materializeForkOutcome).mockResolvedValue({ id: 'wm-fork-new', created }) + vi.mocked(DraftService.transferDrafts).mockResolvedValue({ + copied: items, + removed: items, + conflicts: [] + }) + const s = addSession() + const res = await moveSessionToFork( + s.id, + 'root_ws', + { + kind: 'new', + fork: { id: 'wm-fork-new', name: 'new', parent_workspace_id: 'root_ws' } + }, + { items, removeFromParent: true } + ) + expect(res).toMatchObject({ ok: true, forkId: 'wm-fork-new', copied: 1, removed: 1 }) + expect(DraftService.transferDrafts).toHaveBeenLastCalledWith({ + workspace: 'root_ws', + requestBody: { + target_workspace: 'wm-fork-new', + items, + remove_from_source: true, + replace_existing: replace + } + }) + } + ) }) diff --git a/frontend/src/lib/components/sessions/moveSessionToFork.ts b/frontend/src/lib/components/sessions/moveSessionToFork.ts index 4e50959142..ccaa8550d1 100644 --- a/frontend/src/lib/components/sessions/moveSessionToFork.ts +++ b/frontend/src/lib/components/sessions/moveSessionToFork.ts @@ -7,13 +7,24 @@ import { invalidateWorkspaceDiffCache } from '$lib/components/copilot/chat/globa import { logFeatureUsage } from '$lib/utils/featureUsage' import { flushPendingKeystrokes } from '$lib/components/editorKeystrokeFlush' import { flushEditorSaves } from './useUserDraftSync.svelte' -import { materializeFork, moveSessionToWorkspace, type PendingFork } from './sessionState.svelte' +import { + materializeForkOutcome, + moveSessionToWorkspace, + type PendingFork +} from './sessionState.svelte' -export type MoveTarget = { kind: 'existing'; id: string } | { kind: 'new'; fork: PendingFork } +/** + * `replace` marks a fork this move created itself, whose own drafts at the moved paths are the + * fork clone's copies of them: they are overwritten rather than reported as conflicts. + */ +export type MoveTarget = + | { kind: 'existing'; id: string; replace?: boolean } + | { kind: 'new'; fork: PendingFork } export type MoveSessionResult = - | { ok: true; forkId: string; copied: number; removed: number; warning?: string } - | { ok: false; error: string; conflicts?: DraftItemRef[] } + | { ok: true; forkId: string; copied: number; removed: number } + // `retry` is the target to retry with once the fork exists, so a retry never creates it again. + | { ok: false; error: string; conflicts?: DraftItemRef[]; retry?: MoveTarget } /** * Send every debounced draft save for `workspace` to the server. Returns an error message @@ -35,15 +46,39 @@ export async function flushPendingSaves(workspace: string): Promise { + await Promise.all( + items + .filter((it) => UserDraft.isLive(it.kind, it.path, { workspace })) + .map(async (it) => { + const draft = await DraftService.getOwnDraft({ + workspace, + kind: it.kind, + path: it.path + }).catch(() => undefined) + if (draft?.value == null) return + UserDraftDbSyncer.recordRemoteSync( + { workspace, itemKind: it.kind, path: it.path }, + draft.created_at + ) + UserDraft.seed(it.kind, it.path, draft.value, { workspace }) + }) + ) +} + /** * Move a session into a fork of its workspace, bringing its drafts along. The session's * workspace changes only once its drafts are where they must be, so a failure at any step * leaves it acting on the parent with its drafts intact. * - * A new fork already receives every draft of the user through the fork clone, so only an - * existing fork needs the copy. Once a new fork exists the move always completes: a retry - * would adopt that fork without cloning again, and then remove parent drafts saved after the - * clone. Removing the parent's copies is therefore best-effort there, reported as `warning`. + * A new fork already holds the user's drafts through the fork clone, but a save made after the + * clone is only in the parent, so the drafts are copied again, under the same lock as their + * removal from the parent. */ export async function moveSessionToFork( sessionId: string, @@ -56,71 +91,60 @@ export async function moveSessionToFork( isBusy?: () => boolean } ): Promise { - const busy = { ok: false as const, error: 'A chat turn is running. Retry once it ends.' } 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.' } - if (opts.isBusy?.()) return busy + let fork: Extract + 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 } + } else { + fork = target + } - let forkId: string let copied = 0 let removed = 0 - let warning: string | undefined - try { - if (target.kind === 'new') { - const created = await materializeFork(target.fork) - if (!created) return { ok: false, error: 'Could not create the fork' } - forkId = created - copied = opts.items.length - if (opts.removeFromParent && opts.items.length > 0) { - try { - const res = await DraftService.transferDrafts({ - workspace: parent, - requestBody: { items: opts.items, remove_from_source: true } - }) - removed = res.removed.length - } catch (e: any) { - warning = `The drafts are in the fork but could not be removed from ${parent}: ${e?.body ?? e?.message ?? e}` + if (opts.items.length > 0) { + try { + const res = await DraftService.transferDrafts({ + workspace: parent, + requestBody: { + target_workspace: fork.id, + items: opts.items, + remove_from_source: opts.removeFromParent, + replace_existing: fork.replace + } + }) + if (res.conflicts.length > 0) { + return { + ok: false, + error: 'You already have drafts of these items in the fork', + conflicts: res.conflicts, + retry: fork } } - } else { - forkId = target.id - if (opts.items.length > 0) { - const res = await DraftService.transferDrafts({ - workspace: parent, - requestBody: { - target_workspace: forkId, - items: opts.items, - remove_from_source: opts.removeFromParent - } - }) - if (res.conflicts.length > 0) { - return { - ok: false, - error: 'You already have drafts of these items in the fork', - conflicts: res.conflicts - } - } - copied = res.copied.length - removed = res.removed.length - } + copied = res.copied.length + removed = res.removed.length + } catch (e: any) { + return { ok: false, error: String(e?.body ?? e?.message ?? e), retry: fork } } - } catch (e: any) { - return { ok: false, error: String(e?.body ?? e?.message ?? e) } } if (removed > 0) { for (const it of opts.items) UserDraft.forgetLocal(it.kind, it.path, { workspace: parent }) } - await moveSessionToWorkspace(sessionId, forkId) - for (const ws of [parent, forkId]) { + await reloadOpenEditors(fork.id, opts.items) + await moveSessionToWorkspace(sessionId, fork.id) + for (const ws of [parent, fork.id]) { invalidateWorkspaceDrafts(ws) invalidateWorkspaceDiffCache(ws) } logFeatureUsage('ai_session', 'moved_to_fork', { key: opts.removeFromParent ? 'copy_and_clean_parent' : 'copy', entityId: sessionId, - workspace: forkId + workspace: fork.id }) - return { ok: true, forkId, copied, removed, warning } + return { ok: true, forkId: fork.id, copied, removed } } diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts index a79d98d713..773269c33a 100644 --- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts @@ -48,6 +48,7 @@ import { runHeldElsewhere } from './sessionSync.svelte' import { + adoptStoredSessionWorkspace, commitSessionWorkspace, deleteSession as deleteSessionState, getEffectiveWorkspaceId, @@ -1409,7 +1410,10 @@ onRemoteTurnEnd((sessionId, chatId) => { }) async function applyRemoteTurnEnd(sessionId: string, chatId: string): Promise { + // The other tab may have moved the session to a fork (it brackets the move like a turn). + const moved = await adoptStoredSessionWorkspace(sessionId) const runtime = runtimes.get(sessionId) + if (moved) runtime?.manager.rebuildGlobalSystemMessage() if (!runtime) { // The driving tab may have rotated to a new chat; follow it, or the next // read and a later runtime would open the old one. diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index 4a3abebb4f..19288c6ca7 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -1381,7 +1381,18 @@ export async function materializeFork( fork: PendingFork, onCreationStarted?: (creationId: string) => void ): Promise { - if (get(userWorkspaces).some((w) => w.id === fork.id)) return fork.id + return (await materializeForkOutcome(fork, onCreationStarted))?.id +} + +/** + * `materializeFork`, also telling whether this call created the fork. `created: false` is a + * fork that already existed under that id, whose drafts the clone never wrote. + */ +export async function materializeForkOutcome( + fork: PendingFork, + onCreationStarted?: (creationId: string) => void +): Promise<{ id: string; created: boolean } | undefined> { + if (get(userWorkspaces).some((w) => w.id === fork.id)) return { id: fork.id, created: false } try { let resumed = false if (fork.creation_id) { @@ -1405,12 +1416,12 @@ export async function materializeFork( } usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces()) sendUserToast(`Created fork ${fork.name}`) - return fork.id + return { id: fork.id, created: true } } catch (e: any) { const msg = String(e?.body ?? e?.message ?? e) if (msg.includes('is already being created') && (await forkJoinedWorkspaces(fork.id))) { sendUserToast(`Created fork ${fork.name}`) - return fork.id + return { id: fork.id, created: false } } if (/workspace_pkey|duplicate key|already exists/i.test(msg)) { // Self-heal: the create likely already succeeded. Refresh + adopt the @@ -1420,7 +1431,8 @@ export async function materializeFork( // it and propagate up through commitSessionWorkspace/beforeSend). try { usersWorkspaceStore.set(await WorkspaceService.listUserWorkspaces()) - if (get(userWorkspaces).some((w) => w.id === fork.id)) return fork.id + if (get(userWorkspaces).some((w) => w.id === fork.id)) + return { id: fork.id, created: false } } catch (refreshErr) { console.error('Failed to refresh workspaces during fork self-heal', refreshErr) } @@ -1461,6 +1473,30 @@ export async function moveSessionToWorkspace(id: string, newWorkspaceId: string) await putSession(s) } +/** + * Adopt the workspace another tab moved this session to. A tab acting on the session's + * previous workspace would otherwise keep writing drafts and running jobs there. + */ +export async function adoptStoredSessionWorkspace(id: string): Promise { + const s = sessionState.sessions.find((x) => x.id === id) + if (!BROWSER || !s || s.transient) return false + const db = await sessionsDb.whenReady() + const stored = await db?.get('sessions', id).catch(() => undefined) + if (!stored?.workspace_id || stored.workspace_id === s.workspace_id) return false + // A fork created by that tab is missing from this tab's list, which would read the + // session's workspace as gone. + if (!get(userWorkspaces).some((w) => w.id === stored.workspace_id)) { + const listed = await WorkspaceService.listUserWorkspaces().catch(() => undefined) + if (listed) usersWorkspaceStore.set(listed) + } + s.workspace_id = stored.workspace_id + s.workspace_root_id = stored.workspace_root_id + s.moves = stored.moves + delete s.pending_workspace_id + delete s.pending_fork + return true +} + // Create a brand-new fork and re-assign a committed session to it. Used // by the unavailable-session banner's "Create new fork" path in the // move dropdown. On success the global workspace is switched to the diff --git a/frontend/src/lib/userDraft.svelte.ts b/frontend/src/lib/userDraft.svelte.ts index 4648201a3c..e28caeb895 100644 --- a/frontend/src/lib/userDraft.svelte.ts +++ b/frontend/src/lib/userDraft.svelte.ts @@ -341,6 +341,11 @@ export const UserDraft = { return writtenCache.get(mk)?.val !== undefined }, + /** Whether an editor in this tab holds a live handle on this key. */ + isLive(itemKind: UserDraftItemKind, path: string, opts?: UserDraftOptions): boolean { + return entries.has(mapKey(resolveWorkspace(opts), itemKind, path)) + }, + remove(itemKind: UserDraftItemKind, path: string, opts?: UserDraftOptions): void { const ws = resolveWorkspace(opts) const mk = mapKey(ws, itemKind, path)