mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-27 16:02:43 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
e88dd4ae28
commit
a5b451cc2b
-25
@@ -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"
|
||||
}
|
||||
+17
@@ -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"
|
||||
}
|
||||
+65
@@ -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"
|
||||
}
|
||||
@@ -313,8 +313,8 @@ async fn test_cross_user_draft_privacy(db: Pool<Postgres>) -> 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<Postgres>) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
@@ -394,13 +394,19 @@ async fn test_transfer_drafts_to_fork(db: Pool<Postgres>) -> 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<Postgres>) -> 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);
|
||||
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -385,6 +385,69 @@ fn draft_lineage(kind: UserDraftItemKind, value: &str) -> Option<DraftBaseVersio
|
||||
has_base.then_some(base)
|
||||
}
|
||||
|
||||
/// A role that composes but does not author in full is held to the composition-only rules on
|
||||
/// its drafts as on its deploys. `value` is the draft text as stored or about to be.
|
||||
async fn validate_composed_draft(
|
||||
authed: &ApiAuthed,
|
||||
db: &DB,
|
||||
user_db: &UserDB,
|
||||
w_id: &str,
|
||||
kind: UserDraftItemKind,
|
||||
value: &str,
|
||||
) -> 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<Box<serde_json::value::RawValue>>,
|
||||
tag: Option<String>,
|
||||
}
|
||||
// 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::<crate::apps::Policy>(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<Box<serde_json::value::RawValue>>,
|
||||
tag: Option<String>,
|
||||
}
|
||||
// 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::<crate::apps::Policy>(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<DraftItemRef>,
|
||||
#[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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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<typeof import('./sessionState.svelte')>()
|
||||
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
|
||||
}
|
||||
})
|
||||
}
|
||||
)
|
||||
})
|
||||
|
||||
@@ -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<string | und
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the moved drafts into editors already open on them in `workspace`. Such an editor
|
||||
* still holds what it loaded before the move, with no baseline to conflict on, so its next
|
||||
* save would overwrite the moved draft.
|
||||
*/
|
||||
async function reloadOpenEditors(workspace: string, items: DraftItemRef[]): Promise<void> {
|
||||
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<MoveSessionResult> {
|
||||
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<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 }
|
||||
} 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 }
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
// 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.
|
||||
|
||||
@@ -1381,7 +1381,18 @@ export async function materializeFork(
|
||||
fork: PendingFork,
|
||||
onCreationStarted?: (creationId: string) => void
|
||||
): Promise<string | undefined> {
|
||||
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<boolean> {
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user