From a48ae656ae59d600311f81ef357d08df5226515a Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 16 Sep 2026 09:57:43 +0200 Subject: [PATCH] feat: delete a browser's copy of an AI session past its workspace retention (#11156) * feat: delete a browser's copy of an AI session past its workspace retention Co-Authored-By: Claude Opus 5 (1M context) * fix: tell the AI session retention only to a member who can reach the workspace Co-Authored-By: Claude Opus 5 (1M context) * docs: keep the retention sweep's design narrative in the docs, not the code * fix: give the session retention its own route, leaving the status contract alone Co-Authored-By: Claude Opus 5 (1M context) * docs: shorten the retention route comment to its constraints * docs: name the two clocks in the retention setting, and the deploy window --------- Co-authored-by: Claude Opus 5 (1M context) --- ...c8a287bf1ff4b9d4301e3ea0efd8077936aff.json | 30 ++ backend/tests/session_workspace_status.rs | 68 ++++- .../windmill-api-workspaces/src/workspaces.rs | 40 +++ backend/windmill-api/openapi.yaml | 31 ++ docs/ai-session-backups.md | 104 +++++-- .../copilot/chat/HistoryManager.svelte.ts | 17 +- .../chat/artifacts/artifactsDB.test.ts | 2 +- .../copilot/chat/artifacts/artifactsDB.ts | 11 +- .../chat/files/attachedFilesDB.test.ts | 2 +- .../copilot/chat/files/attachedFilesDB.ts | 7 +- .../sessions/sessionMirror.svelte.ts | 12 +- .../components/sessions/sessionMirror.test.ts | 15 +- .../components/sessions/sessionMirrorPlan.ts | 6 +- .../sessions/sessionMirrorSignal.ts | 24 ++ .../sessions/sessionState.svelte.ts | 272 +++++++++++++++++- .../sessions/sessionStateIndexedDb.test.ts | 130 ++++++++- .../workspaceSettings/AISettings.svelte | 2 +- 17 files changed, 718 insertions(+), 55 deletions(-) create mode 100644 backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json diff --git a/backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json b/backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json new file mode 100644 index 0000000000..552a38a0d4 --- /dev/null +++ b/backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_settings.workspace_id AS \"id!\",\n workspace_settings.ai_config->'sessions_retention_days' AS retention\n FROM workspace_settings\n LEFT JOIN usr ON usr.workspace_id = workspace_settings.workspace_id AND usr.email = $2\n WHERE workspace_settings.workspace_id = ANY($1)\n AND ($3 OR (usr.email IS NOT NULL AND NOT usr.disabled))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "retention", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Text", + "Bool" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff" +} diff --git a/backend/tests/session_workspace_status.rs b/backend/tests/session_workspace_status.rs index b4fc98da5e..2ed4ac503a 100644 --- a/backend/tests/session_workspace_status.rs +++ b/backend/tests/session_workspace_status.rs @@ -3,18 +3,23 @@ //! extractor actually grants. Membership is not the only path: a superadmin is authed into //! any existing workspace without a `usr` row, and `admins` has no `usr` rows at all, so //! answering from `usr` alone reports live workspaces as unresolvable and the client deletes -//! sessions that still work. +//! sessions that still work. `POST /workspaces/session_workspace_retention`, the AI session +//! retention the same client deletes its own copies by, is a workspace setting and answers to +//! the stricter bar, which is why the two are separate routes and tested together. use serde_json::json; use sqlx::{Pool, Postgres}; use std::collections::HashMap; use windmill_test_utils::*; -async fn status(port: u16, token: &str, ids: &[&str]) -> anyhow::Result> { +async fn post( + port: u16, + route: &str, + token: &str, + ids: &[&str], +) -> anyhow::Result { let resp = reqwest::Client::new() - .post(format!( - "http://localhost:{port}/api/workspaces/session_workspace_status" - )) + .post(format!("http://localhost:{port}/api/workspaces/{route}")) .header("Authorization", format!("Bearer {token}")) .json(&json!({ "workspace_ids": ids })) .send() @@ -23,6 +28,14 @@ async fn status(port: u16, token: &str, ids: &[&str]) -> anyhow::Result anyhow::Result> { + post(port, "session_workspace_status", token, ids).await +} + +async fn retention(port: u16, token: &str, ids: &[&str]) -> anyhow::Result> { + post(port, "session_workspace_retention", token, ids).await +} + #[sqlx::test(fixtures("base", "session_workspace_status"))] async fn test_superadmin_reaches_workspaces_without_a_usr_row( db: Pool, @@ -60,3 +73,48 @@ async fn test_superadmin_reaches_workspaces_without_a_usr_row( Ok(()) } + +/// The retention a browser deletes its own copies by is a workspace setting, so unlike the +/// status it is told only to a caller the authed extractor would let in. +#[sqlx::test(fixtures("base", "session_workspace_status"))] +async fn test_session_retention_is_told_only_to_members_who_can_be_authed( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let ids = ["foreign-workspace", "test-workspace", "no-such-workspace"]; + sqlx::query( + "UPDATE workspace_settings SET ai_config = '{\"sessions_retention_days\": 7}' \ + WHERE workspace_id IN ('test-workspace', 'foreign-workspace')", + ) + .execute(&db) + .await?; + + // test@windmill.dev is a superadmin: authed into every workspace that exists. + let sa = retention(port, "SECRET_TOKEN", &ids).await?; + assert_eq!(sa["test-workspace"], 7); + assert_eq!(sa["foreign-workspace"], 7); + assert!(!sa.contains_key("no-such-workspace")); + + // test2@windmill.dev is a member of test-workspace only. + let usr = retention(port, "SECRET_TOKEN_2", &ids).await?; + assert_eq!(usr["test-workspace"], 7); + assert!(!usr.contains_key("foreign-workspace")); + + // A disabled membership still reconciles its sessions — the status stays `active` — but + // cannot be authed into the workspace, so it is told no setting. + sqlx::query("UPDATE usr SET disabled = true WHERE workspace_id = 'test-workspace'") + .execute(&db) + .await?; + assert_eq!( + status(port, "SECRET_TOKEN_2", &ids).await?["test-workspace"], + "active" + ); + assert!(!retention(port, "SECRET_TOKEN_2", &ids) + .await? + .contains_key("test-workspace")); + + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index c29d99d05b..3330ed697f 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -227,6 +227,10 @@ pub fn global_service() -> Router { .route("/list", get(list_workspaces)) .route("/users", get(user_workspaces)) .route("/session_workspace_status", post(session_workspace_status)) + .route( + "/session_workspace_retention", + post(session_workspace_retention), + ) .route("/create", post(create_workspace)) .route("/create_fork", post(deprecated_create_workspace_fork)) .route("/exists", post(exists_workspace)) @@ -5686,6 +5690,42 @@ async fn session_workspace_status( Ok(Json(statuses)) } +/// The AI session retention a browser deletes its local copies by (docs/ai-session-backups.md). +/// Its own route, not a field on the status above, whose shape an older tab still reads. Unlike +/// a status, it answers only for a workspace this caller can be authed into: a setting is the +/// workspace's to tell, so a disabled membership gets none though its sessions still reconcile. +async fn session_workspace_retention( + Extension(db): Extension, + authed: ApiAuthed, + Json(req): Json, +) -> JsonResult> { + if req.workspace_ids.len() > 1000 { + return Err(Error::BadRequest( + "Too many workspace ids (max 1000)".to_string(), + )); + } + let email = &authed.email; + let is_superadmin = windmill_api_auth::is_super_admin_authed(&db, &authed).await?; + let rows = sqlx::query!( + "SELECT workspace_settings.workspace_id AS \"id!\", + workspace_settings.ai_config->'sessions_retention_days' AS retention + FROM workspace_settings + LEFT JOIN usr ON usr.workspace_id = workspace_settings.workspace_id AND usr.email = $2 + WHERE workspace_settings.workspace_id = ANY($1) + AND ($3 OR (usr.email IS NOT NULL AND NOT usr.disabled))", + &req.workspace_ids[..], + email, + is_superadmin, + ) + .fetch_all(&db) + .await?; + let days = rows + .into_iter() + .filter_map(|r| sessions_retention_days(r.retention.as_ref()).map(|days| (r.id, days))) + .collect(); + Ok(Json(days)) +} + /// The instance critical alert channels belong to the instance operator, who on cloud is /// not the workspace owner and never opted into a tenant's job failures. Fork workspaces run /// throwaway copies of their parent's runnables, so instance-wide operational alerting must diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9ce1aef272..c88a298646 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1277,6 +1277,37 @@ paths: - archived - deleted + /workspaces/session_workspace_retention: + post: + summary: get the AI session retention of workspaces referenced by client-side sessions + operationId: getSessionWorkspaceRetention + tags: + - workspace + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + workspace_ids: + type: array + items: + type: string + required: + - workspace_ids + responses: + "200": + description: >- + map of workspace id to its `ai_config.sessions_retention_days`; a workspace + without a retention, or one the caller cannot be authenticated into, is absent + content: + application/json: + schema: + type: object + additionalProperties: + type: integer + /w/{workspace}/workspaces/get_as_superadmin: get: summary: get workspace as super admin (require to be super admin) diff --git a/docs/ai-session-backups.md b/docs/ai-session-backups.md index 9af2bd3a84..ffec41e182 100644 --- a/docs/ai-session-backups.md +++ b/docs/ai-session-backups.md @@ -240,32 +240,86 @@ the first push after it or on the next page load, whichever comes first. `ai_config.sessions_retention_days` (per workspace, in the AI settings; unset by default; the `sessions_storage_disabled` pattern: no migration, carried by settings export and the -CLI; 1 to 3650) puts an age on backups, counted from the last push of the session that -completed. It applies to the backup only: a browser keeps its copy whatever the retention, -and a backup swept while a browser still has the session comes back once that browser writes -to it again (its incremental push is refused and goes whole). +CLI; 1 to 3650) puts an age on sessions, counted from their last activity. Each side applies +it with its own clock against its own timestamps, so no clock is compared with another +machine's, and the two do not time the same event: the server counts the last push that +completed, a browser its last local activity, which includes reading new messages and is not +pushed. A backup swept while a browser still reads its copy comes back once that browser +writes to the session again (its incremental push is refused and goes whole): -The server sweeps the object store (`sweep_expired_ai_session_backups`, from the monitor about -every 40 minutes on each server, one pass at a time under a session-level advisory lock). For -every workspace with a retention it takes the store its backups live in, its own storage or -the instance store standing in, decided from the row it reads the generation from as the -routes do, names the users under the generation prefix (`list_with_delimiter`) and lists each -user's `index/` once: one object per session, nothing of what the sessions hold. A session -whose marker is older than the retention is removed under its lock (`lock_session`), once its -markers, listed again there, are still all older: a push that renewed the session between the -walk and the lock keeps it, and one split over parts either holds the lock or has the session -unlisted with its token next to the markers (`index/{sid}/push`), which the sweep leaves -alone while the token is younger than the retention: an older one is a push a browser -abandoned, whose landed parts nothing lists, and it goes the same way. Before deleting -anything the sweep writes a record next to the markers (`index/{sid}/sweep`, not an epoch, so -neither `list` nor `pull` counts it), and `remove_session` deletes it last: a removal cut -short, its markers already gone, is found by the next pass and finished, unless a push listed -the session again first. At most 1000 sessions per workspace and pass; the rest wait for the -next. `list` leaves an expired marker out of its answer meanwhile, so a browser never restores -a session the sweep has not reached. The marker's modification time is the storage's clock and -the cutoff the server's. The sweep reaches only the backups the routes would: a deleted -workspace's stay in its storage, and so do those a workspace keeps in the instance store once -`ai_sessions_instance_storage_fallback` is set to false. +- The server sweeps the object store (`sweep_expired_ai_session_backups`, from the monitor + about every 40 minutes on each server, one pass at a time under a session-level advisory + lock). For every workspace with a retention it takes the store its backups live in, its + own storage or the instance store standing in, decided from the row it reads the + generation from as the routes do, names the users under the generation prefix + (`list_with_delimiter`) and lists each user's `index/` once: one object + per session, nothing of what the sessions hold. A session whose marker is older than the + retention is removed under its lock (`lock_session`), once its markers, listed again + there, are still all older: a push that renewed the session between the walk and the lock + keeps it, and one split over parts either holds the lock or has the session unlisted with + its token next to the markers (`index/{sid}/push`), which the sweep leaves alone while the + token is younger than the retention: an older one is a push a browser abandoned, whose + landed parts nothing lists, and it goes the same way. Before deleting anything the sweep + writes a record next to + the markers (`index/{sid}/sweep`, not an epoch, so neither `list` nor `pull` counts it), + and `remove_session` deletes it last: a removal cut short, its markers already gone, is + found by the next pass and finished, unless a push listed the session again first. At + most 1000 sessions per workspace and pass; the rest wait for the next. `list` leaves an + expired marker out of its answer meanwhile, so a browser never restores a session the + sweep has not reached. The marker's modification time is the storage's clock and the + cutoff the server's. The sweep reaches only the backups the routes would: a deleted + workspace's stay in its storage, and so do those a workspace keeps in the instance store + once `ai_sessions_instance_storage_fallback` is set to false. +- The browser sweeps its own stores when a tab resolves the logged-in user + (`sweepExpiredSessions`, from the one `onUserChange` in `sessionState.svelte.ts`), before + that tab reads a single session. A session whose last activity is older than the retention + by the browser's clock is deleted locally, record, chats, images, attached files and + artifacts. A restored session carries the backup's time as its last activity, the storage's + clock, so it counts from the later of that and the moment it was restored here + (`restoredAt`): a browser clock ahead of the storage's never deletes a session it just + brought back. Archived sessions count like any other, and persisted unsent drafts by their + pending workspace. + + The stores are shared by the user's tabs, and each keeps copies of the sessions in memory, + so every tab holds a shared Web Lock from before it reads them until it stops using them, + and the sweep deletes only while holding that lock exclusively, requested if available: + granted exactly when no tab of the user has the sessions loaded, which is why the sweep + runs where it does and nowhere else. Nothing holds a copy of what it deletes and nothing + writes the stores meanwhile, so it deletes one record at a time and without re-reading. It + also takes the tab lock the flush and the restore take, again only if available, so neither + plans nor stages a session half deleted; like the restore, it does not run where Web Locks + do not exist. With several tabs open nothing is swept, until one of them reloads alone. + + The hold is only as good as the tabs that take it, so a tab still running a build from before + it has the sessions loaded and holds nothing. A tab loaded after that one, across a deploy, + can sweep a session the older tab has in memory, and a write there afterwards brings the + record back without its chats, which the next flush pushes. It needs a tab left open across a + deploy, a session untouched for the whole retention, and the user going back to that session + in the older tab; the next sweep deletes it again. The same window is open to the + workspace-lifecycle delete in `reconcileSessionsLifecycle`, which no lock guards at all. + + What deletes is the retention the server gives as the sweep runs, asked for under both locks + (`POST /workspaces/session_workspace_retention`, its own route rather than a field on the + lifecycle status, whose answer a tab loaded before this version still reads). Never a + remembered one: a retention raised or cleared since would otherwise delete a session that is + now within it, and a persisted unsent draft has no backup to come back from. What the sweep + keeps in localStorage decides only whether to ask again — it asks when it has asked nothing + yet, when the answer it has is a day old, or when that answer marks a session expired — so + an ordinary load costs no request at all. An answer that does not arrive within five seconds + leaves the sessions for the next load rather than delete on what this browser guessed. That + route answers for a workspace the caller can be authed into, unlike the status: a status is + what to do with the caller's own sessions, a setting is the workspace's to tell, so a + disabled membership is told nothing though its sessions still reconcile. + + Each session's record goes before its pieces, so nothing plans a push for it afterwards, + and a localStorage key written before the record and removed once every piece is gone makes + a later sweep finish a deletion that failed, unless a restore brought the session back + since. The record is deleted without the tombstone a user delete leaves, which is what lets + a restore bring it back. The session's dirty mark and sync row go with it (`sessionSwept`), + unless the row still carries a removal or a restore's staging. Nothing is sent to the storage: the local + copy's age says nothing about another device's, which may have pushed the session since, + and the server applies the rule to the backup on its own. A session swept here that the + storage still lists comes back on the next restore. ## Limits diff --git a/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts b/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts index 4edac6dc0a..d956b4c0c0 100644 --- a/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts @@ -267,15 +267,22 @@ export async function importStoredChats( return true } -/** Deletes these chats of the session (with their images) and these images: what an earlier - * restore staged for it and the backup no longer has. False when nothing could be deleted. */ +/** Every chat tagged with the session, and their images: for a session past its workspace's + * retention, which no runtime has mounted. */ +export function deleteSessionChats(sessionId: string, email: string): Promise { + return pruneSessionChats(sessionId, undefined, new Set(), email) +} + +/** Deletes chats of the session (with their images) and these images: what an earlier restore + * staged for it and the backup no longer has. `chats` names the ones to go; undefined is every + * chat of the session. False when nothing could be deleted. */ export async function pruneSessionChats( sessionId: string, - chats: Set, + chats: Set | undefined, images: Set, email: string ): Promise { - if (chats.size === 0 && images.size === 0) return true + if (chats?.size === 0 && images.size === 0) return true const db = await backupDb(email) if (!db) return false try { @@ -283,7 +290,7 @@ export async function pruneSessionChats( const chatStore = tx.objectStore('chats') const imageStore = tx.objectStore('images') for (const chatId of await chatStore.index('by-session').getAllKeys(sessionId)) { - if (!chats.has(String(chatId))) continue + if (chats && !chats.has(String(chatId))) continue await chatStore.delete(chatId) const keys = await imageStore .index('by-chat') diff --git a/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.test.ts b/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.test.ts index d70566ad18..a0fb5a4a4b 100644 --- a/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.test.ts +++ b/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.test.ts @@ -220,7 +220,7 @@ describe('artifactsDB', () => { expect(await noDb.getArtifact('a1')).toBeUndefined() expect(await noDb.listArtifactsForSession('s1')).toEqual([]) await expect(noDb.deleteArtifact('a1')).resolves.toBeUndefined() - await expect(noDb.deleteArtifactsForSession('s1')).resolves.toBeUndefined() + await expect(noDb.deleteArtifactsForSession('s1')).resolves.toBe(false) }) it('rejects a version read it could not make, instead of reading as absent', async () => { diff --git a/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.ts b/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.ts index e7ff4b1f60..f0cb32d71f 100644 --- a/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.ts +++ b/frontend/src/lib/components/copilot/chat/artifacts/artifactsDB.ts @@ -449,9 +449,14 @@ export async function pruneSessionArtifacts( } } -export async function deleteArtifactsForSession(sessionId: string): Promise { +/** False when the store could not be reached or the deletion failed. With `email`, only that + * user's store is touched: a caller that captured its user must not follow an account switch. */ +export async function deleteArtifactsForSession( + sessionId: string, + email?: string +): Promise { const db = await getDB() - if (!db) return + if (!db || (email !== undefined && db.name !== scopedKeyFor(ARTIFACTS_DB, email))) return false try { const tx = db.transaction(['items', 'versions'], 'readwrite') const items = tx.objectStore('items') @@ -464,8 +469,10 @@ export async function deleteArtifactsForSession(sessionId: string): Promise { putItem({ id: 'a', sessionId: 's1', kind: 'snapshot', name: 'x.txt', addedAt: 0 }) ).resolves.toBeUndefined() await expect(deleteItem('a')).resolves.toBeUndefined() - await expect(deleteItemsForSession('s1')).resolves.toBeUndefined() + await deleteItemsForSession('s1') }) it('does not throw when requesting persistent storage', async () => { diff --git a/frontend/src/lib/components/copilot/chat/files/attachedFilesDB.ts b/frontend/src/lib/components/copilot/chat/files/attachedFilesDB.ts index fe051ef8cc..dc872845cd 100644 --- a/frontend/src/lib/components/copilot/chat/files/attachedFilesDB.ts +++ b/frontend/src/lib/components/copilot/chat/files/attachedFilesDB.ts @@ -91,9 +91,10 @@ export async function deleteItem(id: string): Promise { await db?.delete('items', id) } -export async function deleteItemsForSession(sessionId: string): Promise { +/** False when the store could not be reached or the deletion failed. */ +export async function deleteItemsForSession(sessionId: string): Promise { const db = await getDB() - if (!db) return + if (!db) return false try { const tx = db.transaction('items', 'readwrite') const index = tx.store.index('by-session') @@ -103,8 +104,10 @@ export async function deleteItemsForSession(sessionId: string): Promise { cursor = await cursor.continue() } await tx.done + return true } catch (err) { console.error('Could not delete attached files for session', err) + return false } } diff --git a/frontend/src/lib/components/sessions/sessionMirror.svelte.ts b/frontend/src/lib/components/sessions/sessionMirror.svelte.ts index a18e2febd6..bfa1621822 100644 --- a/frontend/src/lib/components/sessions/sessionMirror.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionMirror.svelte.ts @@ -27,7 +27,7 @@ import { getCurrentUserEmail, onUserChange, scopedKey, scopedKeyFor } from '$lib import { logFeatureUsage } from '$lib/utils/featureUsage' import { randomUUID } from '$lib/utils/uuid' import { workspaceRootId } from './sessionScope.svelte' -import { onMirrorSignal } from './sessionMirrorSignal' +import { onMirrorSignal, onSessionSwept, sessionsLockName } from './sessionMirrorSignal' import { importSessions, isSessionTombstoned, @@ -385,7 +385,7 @@ function hasWebLocks(): boolean { async function withUserLock(email: string, fn: () => Promise, wait = false): Promise { const locks = webLocks() if (!locks) return fn() - await locks.request(`wm-ai-sessions-mirror::${email}`, { ifAvailable: !wait }, async (lock) => { + await locks.request(sessionsLockName(email), { ifAvailable: !wait }, async (lock) => { if (lock) await fn() // The other tab's flush read the marks before this one's were written: try again // once it is done, rather than wait for the next write or load. @@ -1572,6 +1572,14 @@ export function backupSettingsChanged(ws: string): void { // --- Wiring --- if (BROWSER) { + // Nothing pushes a swept session again, so its mark and sync row are dead weight; a row + // still carrying a removal or a restore's staging is left to those. + onSessionSwept(async (id, email) => { + if (email !== getCurrentUserEmail()) return + dropDirty(id) + const row = await readSync(id, email) + if (row && !row.removed && !row.staging) await deleteSync([id], email) + }) onMirrorSignal((signal) => { // A mark for another user waits for that user's next load. const mine = !signal.email || signal.email === getCurrentUserEmail() diff --git a/frontend/src/lib/components/sessions/sessionMirror.test.ts b/frontend/src/lib/components/sessions/sessionMirror.test.ts index ff2c8373da..7676662d31 100644 --- a/frontend/src/lib/components/sessions/sessionMirror.test.ts +++ b/frontend/src/lib/components/sessions/sessionMirror.test.ts @@ -93,7 +93,7 @@ import { sessionState, type Session } from './sessionState.svelte' -import { markSessionDirty } from './sessionMirrorSignal' +import { markSessionDirty, sessionSwept } from './sessionMirrorSignal' import { __flushForTesting, __resetMirrorForTesting, @@ -268,6 +268,19 @@ describe('sessionMirror flush', () => { await __settleForTesting() }) + it('forgets the sync row of a session the retention swept, unless it carries a removal', async () => { + await __writeSyncForTesting( + [ + { id: 'swept', ws: 'admins', head: '', chats: {}, images: {} }, + { id: 'swept-removed', ws: 'admins', head: '', chats: {}, images: {}, removed: true } + ], + EMAIL + ) + await sessionSwept('swept', EMAIL) + await sessionSwept('swept-removed', EMAIL) + expect((await __syncRowsForTesting(EMAIL)).map((r) => r.id)).toEqual(['swept-removed']) + }) + it('keeps a delete filed on the sync row while the first push is still in flight', async () => { const s: Session = { id: 'sr', name: 'session-1', createdAt: 1, workspace_id: 'ws' } sessionState.sessions = [s] diff --git a/frontend/src/lib/components/sessions/sessionMirrorPlan.ts b/frontend/src/lib/components/sessions/sessionMirrorPlan.ts index a664e61f5e..caea48c792 100644 --- a/frontend/src/lib/components/sessions/sessionMirrorPlan.ts +++ b/frontend/src/lib/components/sessions/sessionMirrorPlan.ts @@ -79,9 +79,9 @@ export function isFallbackStorage(name: string): boolean { /** * The part of a session record the backup keeps. Left out on purpose: `name` (a * per-browser counter), the unsent-draft fields (`pending_*`, `draftPrompt`, - * `autoSendDraftAt`), `workspace_root_id` (derived on import), `transient`, and the two - * fields reading a session bumps (`lastSeenCount`, `lastActivityAt`) — so opening a - * session and reading its new messages never costs a push. + * `autoSendDraftAt`), `workspace_root_id` (derived on import), `transient`, `restoredAt` + * (this browser's clock), and the two fields reading a session bumps (`lastSeenCount`, + * `lastActivityAt`) — so opening a session and reading its new messages never costs a push. */ export type SessionHead = Pick< Session, diff --git a/frontend/src/lib/components/sessions/sessionMirrorSignal.ts b/frontend/src/lib/components/sessions/sessionMirrorSignal.ts index eb936d2e22..abe600b4a6 100644 --- a/frontend/src/lib/components/sessions/sessionMirrorSignal.ts +++ b/frontend/src/lib/components/sessions/sessionMirrorSignal.ts @@ -29,6 +29,13 @@ export function markSessionRemoved(sessionId: string, workspaceId?: string, emai emit({ kind: 'removed', sessionId, workspaceId, email }) } +/** The Web Lock one tab of the user holds while it reads or writes the stores wholesale: the + * backup's flush and restore, and the retention sweep, which must not interleave with either + * (a flush planning a session half deleted would push the deletions to the backup). */ +export function sessionsLockName(email: string): string { + return `wm-ai-sessions-mirror::${email}` +} + export function onMirrorSignal(fn: (signal: MirrorSignal) => void): void { handler = fn const replay = buffered @@ -36,7 +43,24 @@ export function onMirrorSignal(fn: (signal: MirrorSignal) => void): void { for (const signal of replay) fn(signal) } +let sweptHandler: ((sessionId: string, email: string) => Promise) | undefined + +/** The retention sweep deleted this session's local copy in the store of `email`: what the + * backup keeps of it in this browser goes too. Awaited under the sweep's tab lock. */ +export async function sessionSwept(sessionId: string, email: string): Promise { + try { + await sweptHandler?.(sessionId, email) + } catch (e) { + console.error('Could not forget the backup state of a swept session', e) + } +} + +export function onSessionSwept(fn: (sessionId: string, email: string) => Promise): void { + sweptHandler = fn +} + export function __resetMirrorSignalForTesting(): void { handler = undefined + sweptHandler = undefined buffered = [] } diff --git a/frontend/src/lib/components/sessions/sessionState.svelte.ts b/frontend/src/lib/components/sessions/sessionState.svelte.ts index 2670b3fa4d..3014e6d5e7 100644 --- a/frontend/src/lib/components/sessions/sessionState.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionState.svelte.ts @@ -27,7 +27,13 @@ import { userScopedDb } from '$lib/userScopedDb' import { emailOfScopedKey, scopedKeyFor } from '$lib/userScopedStorage' import { deleteItemsForSession } from '../copilot/chat/files/attachedFilesDB' import { deleteArtifactsForSession } from '../copilot/chat/artifacts/artifactsDB' -import { markSessionDirty, markSessionRemoved } from './sessionMirrorSignal' +import { deleteSessionChats } from '../copilot/chat/HistoryManager.svelte' +import { + markSessionDirty, + markSessionRemoved, + sessionSwept, + sessionsLockName +} from './sessionMirrorSignal' // Switch the global workspace iff the target differs from the active one // and is non-empty. Centralises the "session needs its workspace in focus" @@ -124,6 +130,11 @@ export type Session = { // Absent on records last written before the field existed; readers fall back // to createdAt via sessionLastActivityAt. lastActivityAt?: number + // When this browser restored the session from its backup, by this browser's clock. + // The restore sets `lastActivityAt` to the backup's time, the storage's clock; the + // retention counts from whichever is later, so a browser clock ahead of the storage's + // never deletes a session it just brought back. Not backed up. + restoredAt?: number // Per-session unread watermark: the displayMessages count the last time // the user was on this session's page. Compared against the runtime's // current message count to derive the unread badge (see sessionUnread). @@ -438,8 +449,15 @@ export function __resetDeletedSessionIdsForTesting(): void { // The one way to remove a session's record. Tombstones BEFORE awaiting the delete so a // putSession racing this transaction cannot commit its write behind it — a direct // db.delete elsewhere would silently reopen that window. -async function deleteSessionRow(db: IDBPDatabase, id: string): Promise { - deletedSessionIds.add(id) +async function deleteSessionRow( + db: IDBPDatabase, + id: string, + // The retention sweep passes false. It holds the in-use lock exclusively, so no write can + // race its delete, and the backup may still hold the session: a tombstone would refuse the + // restore that is meant to bring it back. + tombstone = true +): Promise { + if (tombstone) deletedSessionIds.add(id) await db.delete('sessions', id) } @@ -634,6 +652,242 @@ export async function reconcileSessionsLifecycle(): Promise { } } +// --- Retention --- + +const DAY_MS = 24 * 60 * 60 * 1000 + +// Past the retention by this browser's clock, counted from the later of the session's last +// activity and its restore here: a restored session carries the backup's time, the storage's +// clock, so without `restoredAt` a browser running ahead would delete what it just brought +// back. Archived sessions count like any other. +function isSessionExpired( + session: Session, + retentionDays: number | undefined, + now: number +): boolean { + if (retentionDays === undefined || !(retentionDays >= 1)) return false + const since = Math.max(sessionLastActivityAt(session), session.restoredAt ?? 0) + return since < now - retentionDays * DAY_MS +} + +// What the server last told this browser, and when. It decides whether the sweep asks again, +// and nothing else: a retention raised or cleared since must not delete a session, and a +// persisted unsent draft has no backup to come back from. +const RETENTION_DAYS = 'windmill_sessions_retention_days' + +// Nothing remembered for longer than this is trusted even to say there is nothing to ask +// about, so a retention lowered while this browser saw nothing expiring still takes effect. +const RETENTION_STALE_MS = 24 * 60 * 60 * 1000 + +interface RememberedRetention { + at: number + days: Record +} + +function rememberRetention(email: string, days: Record): void { + try { + const remembered: RememberedRetention = { at: Date.now(), days } + localStorage.setItem(scopedKeyFor(RETENTION_DAYS, email), JSON.stringify(remembered)) + } catch {} +} + +function rememberedRetention(email: string): RememberedRetention | undefined { + try { + const stored = localStorage.getItem(scopedKeyFor(RETENTION_DAYS, email)) + const remembered = stored ? JSON.parse(stored) : undefined + if (remembered?.days && typeof remembered.days === 'object') { + return remembered as RememberedRetention + } + } catch {} + return undefined +} + +// How long the sweep waits for the retention of the workspaces it is about to sweep in. The +// tab reads its sessions after the sweep, so a request nothing answers costs the list this +// much and no more, and only in a tab that had something to delete. +const RETENTION_ASK_MS = 5000 + +// The retention the server gives now, or undefined when this browser could not be told: a +// session is deleted only on an answer of the moment. +async function askRetention(workspaceIds: string[]): Promise | undefined> { + try { + return await Promise.race([ + WorkspaceService.getSessionWorkspaceRetention({ + requestBody: { workspace_ids: workspaceIds } + }), + new Promise((resolve) => setTimeout(() => resolve(undefined), RETENTION_ASK_MS)) + ]) + } catch (e) { + console.error('Failed to read the AI session retention of the workspaces', e) + return undefined + } +} + +// One key per session this browser swept whose pieces are not all deleted yet. +const RETENTION_PENDING = 'windmill_sessions_retention_pending' + +function retentionPendingPrefix(email: string): string { + return `${scopedKeyFor(RETENTION_PENDING, email)}::` +} + +function forgetRetentionPending(email: string, id: string): void { + try { + localStorage.removeItem(retentionPendingPrefix(email) + id) + } catch {} +} + +function retentionPending(email: string): string[] { + const prefix = retentionPendingPrefix(email) + const ids: string[] = [] + try { + for (let i = 0; i < localStorage.length; i++) { + const key = localStorage.key(i) + if (key?.startsWith(prefix)) ids.push(key.slice(prefix.length)) + } + } catch {} + return ids +} + +function webLocks(): LockManager | undefined { + return typeof navigator === 'undefined' ? undefined : (navigator as { locks?: LockManager }).locks +} + +// Held, shared, by every tab from before it reads the user's sessions until it stops using +// them: the stores are shared, and each tab keeps copies of the sessions in memory, so the +// sweep deletes only while holding this exclusively. +function sessionsInUseLockName(email: string): string { + return `${sessionsLockName(email)}::in-use` +} + +interface InUseHold { + email: string + released: boolean + release?: () => void + done?: Promise +} + +let inUse: InUseHold | undefined + +// Resolves once the hold is granted, which waits for the sweep another tab is running. A +// request the browser refuses (a document that is not fully active) resolves it too, without +// a hold: the tab reads its sessions unguarded, as it does where Web Locks do not exist, and +// never sits waiting for a grant that is not coming. +async function holdSessionsInUse(email: string): Promise { + const locks = webLocks() + if (!locks || inUse?.email === email) return + await releaseSessionsInUse() + const hold: InUseHold = { email, released: false } + inUse = hold + await new Promise((granted) => { + hold.done = locks + .request(sessionsInUseLockName(email), { mode: 'shared' }, () => { + granted() + return hold.released ? undefined : new Promise((resolve) => (hold.release = resolve)) + }) + .catch((e) => { + console.error('Could not hold the AI sessions this tab is reading', e) + if (inUse === hold) inUse = undefined + granted() + }) + }) +} + +// Resolves once the hold is let go of, so an exclusive request made next can be granted. +async function releaseSessionsInUse(): Promise { + const hold = inUse + if (!hold) return + inUse = undefined + hold.released = true + hold.release?.() + await hold.done?.catch(() => {}) +} + +// Deletes one expired session: its record first, so nothing plans a push for it afterwards, +// then its pieces. The pending key, written before the record and removed once every piece +// is gone, is what a later sweep finishes a failed deletion from. +async function sweepSession( + db: IDBPDatabase, + id: string, + email: string +): Promise { + try { + localStorage.setItem(retentionPendingPrefix(email) + id, '1') + } catch { + return + } + await deleteSessionRow(db, id, false) + await sessionSwept(id, email) + if (await deleteSessionPieces(id, email)) forgetRetentionPending(email, id) +} + +// Chats with their images, artifacts and attached files. False when any of them could not +// be deleted. +async function deleteSessionPieces(id: string, email: string): Promise { + const chats = await deleteSessionChats(id, email) + const artifacts = await deleteArtifactsForSession(id, email) + const files = await deleteItemsForSession(id) + return chats && artifacts && files +} + +// The workspace a session's retention comes from: persisted unsent drafts count by the one +// they are waiting on. +function retentionWorkspaceOf(session: Session): string | undefined { + return session.workspace_id ?? session.pending_workspace_id +} + +// Deletes this browser's copies of the sessions past their workspace's retention, and the +// pieces of the ones an earlier sweep could not finish (docs/ai-session-backups.md). Deleting +// one record at a time, without re-reading it, is safe only under the in-use lock held +// exclusively, granted exactly when no tab has the sessions loaded — hence the call site. +async function sweepExpiredSessions(email: string): Promise { + const locks = webLocks() + if (!locks || inUse) return + try { + await locks.request(sessionsInUseLockName(email), { ifAvailable: true }, async (idle) => { + if (!idle) return + // The flush and the restore run under this one: neither must see a session half + // deleted, or plan a push from it. + await locks.request(sessionsLockName(email), { ifAvailable: true }, async (mirror) => { + if (!mirror) return + const db = await sessionsDb.whenReady() + if (!db || db.name !== scopedKeyFor(SESSIONS_DB, email)) return + for (const id of retentionPending(email)) { + // A restore brought the session back: its pieces are that copy's now. + const back = (await db.getKey('sessions', id)) !== undefined + if (back || (await deleteSessionPieces(id, email))) forgetRetentionPending(email, id) + } + const stored = await db.getAll('sessions') + const remembered = rememberedRetention(email) + const now = Date.now() + const workspaces = new Set() + let expired = false + for (const s of stored) { + const ws = retentionWorkspaceOf(s) + if (ws === undefined) continue + workspaces.add(ws) + expired ||= isSessionExpired(s, remembered?.days[ws], now) + } + // Nothing to sweep in, or nothing old enough by an answer recent enough to be + // believed about that: this load costs no request. + const fresh = remembered !== undefined && now - remembered.at < RETENTION_STALE_MS + if (workspaces.size === 0 || (!expired && fresh)) return + const retention = await askRetention([...workspaces]) + // Asked and not told: the sessions wait for the next load rather than go on an + // answer this browser does not have. + if (!retention) return + rememberRetention(email, retention) + for (const s of stored) { + const ws = retentionWorkspaceOf(s) + if (ws === undefined || !isSessionExpired(s, retention[ws], Date.now())) continue + await sweepSession(db, s.id, email) + } + }) + }) + } catch (e) { + console.error('Failed to sweep the sessions past their retention', e) + } +} + // The single seam for "a workspace just changed — bring sessions back in sync." // Refresh the workspace list FIRST — both reconcile and the putSession guard // read it, so it must reflect the change before reconcile runs — then reconcile. @@ -715,6 +969,15 @@ export async function deleteSessionsForWorkspace(workspaceId: string): Promise { if (!BROWSER) return + // The retention sweep runs here and nowhere else: this tab holds none of the new user's + // sessions yet, and letting go of the hold it had leaves it holding none of anyone's. The + // new hold is taken before the sessions are read, so another tab's sweep never deletes + // what this tab is about to load, and one already running is waited for. + await releaseSessionsInUse() + if (email) { + await sweepExpiredSessions(email) + await holdSessionsInUse(email) + } await hydrateSessions({ dropTransients: prevEmail !== email }) // onUserChange also fires at registration time, before the email resolves — // that hydration is an empty no-op and must not clear the loading state. @@ -1197,9 +1460,10 @@ export async function importSessions(records: Session[], email: string): Promise const tx = db.transaction('sessions', 'readwrite') const existing = new Set((await tx.store.getAllKeys()).map(String)) let next = nextSessionNumber([...(await tx.store.getAll()), ...sessionState.sessions]) + const restoredAt = Date.now() for (const r of records) { if (existing.has(r.id) || deletedSessionIds.has(r.id)) continue - const record: Session = { ...r, name: `session-${next++}` } + const record: Session = { ...r, name: `session-${next++}`, restoredAt } delete record.transient delete record.workspace_root_id ensureSessionRootId(record) diff --git a/frontend/src/lib/components/sessions/sessionStateIndexedDb.test.ts b/frontend/src/lib/components/sessions/sessionStateIndexedDb.test.ts index b9882d045e..dddfe795f7 100644 --- a/frontend/src/lib/components/sessions/sessionStateIndexedDb.test.ts +++ b/frontend/src/lib/components/sessions/sessionStateIndexedDb.test.ts @@ -9,20 +9,30 @@ vi.mock('esm-env', async (importOriginal) => ({ })) // Spy on the attached-file GC so we can assert lifecycle deletes clean it up. -const { deleteItemsForSessionMock } = vi.hoisted(() => ({ deleteItemsForSessionMock: vi.fn() })) +const { deleteItemsForSessionMock } = vi.hoisted(() => ({ + deleteItemsForSessionMock: vi.fn().mockResolvedValue(true) +})) vi.mock('../copilot/chat/files/attachedFilesDB', async (orig) => ({ ...(await orig()), deleteItemsForSession: deleteItemsForSessionMock })) const { deleteArtifactsForSessionMock } = vi.hoisted(() => ({ - deleteArtifactsForSessionMock: vi.fn() + deleteArtifactsForSessionMock: vi.fn().mockResolvedValue(true) })) vi.mock('../copilot/chat/artifacts/artifactsDB', async (orig) => ({ ...(await orig()), deleteArtifactsForSession: deleteArtifactsForSessionMock })) +const { deleteSessionChatsMock } = vi.hoisted(() => ({ + deleteSessionChatsMock: vi.fn().mockResolvedValue(true) +})) +vi.mock('../copilot/chat/HistoryManager.svelte', async (orig) => ({ + ...(await orig()), + deleteSessionChats: deleteSessionChatsMock +})) + // sessionState imports WorkspaceService; these tests don't touch the network. vi.mock('$lib/gen', async (orig) => { const actual = await orig() @@ -31,7 +41,8 @@ vi.mock('$lib/gen', async (orig) => { WorkspaceService: { ...actual.WorkspaceService, listUserWorkspaces: vi.fn().mockResolvedValue([]), - getSessionWorkspaceStatus: vi.fn().mockResolvedValue({}) + getSessionWorkspaceStatus: vi.fn().mockResolvedValue({}), + getSessionWorkspaceRetention: vi.fn().mockResolvedValue({}) } } }) @@ -75,6 +86,32 @@ function freshUser() { return asUser(`u${n++}@x.com`) } +// The Web Locks API, which the node test environment lacks: `holders` counts the shared holds +// on each name across tabs, against which an exclusive request made if available is not granted. +function installLocks(holders: Map): void { + if (typeof navigator === 'undefined') { + Object.defineProperty(globalThis, 'navigator', { value: {}, configurable: true }) + } + Object.defineProperty(navigator, 'locks', { + value: { + request: async (name: string, ...rest: unknown[]) => { + const run = rest[rest.length - 1] as (lock: unknown) => Promise + const options = (rest.length > 1 ? rest[0] : {}) as LockOptions + if (options.mode === 'shared') { + holders.set(name, (holders.get(name) ?? 0) + 1) + try { + return await run({}) + } finally { + holders.set(name, (holders.get(name) ?? 1) - 1) + } + } + return run(options.ifAvailable && (holders.get(name) ?? 0) > 0 ? null : {}) + } + }, + configurable: true + }) +} + // Hydration is fire-and-forget off the user store, so it can land after the test body // has populated sessionState.sessions and overwrite it with what the DB held at read // time; `hydrated` flips once the read has been applied. The logout is load-bearing: @@ -679,6 +716,93 @@ describe('sessionState IndexedDB persistence', () => { deleteSession('draftRec') }) + it('sweeps sessions past their workspace retention when a tab loads alone', async () => { + const user = freshUser() + usersWorkspaceStore.set({ + email: user.email, + workspaces: [ + { id: 'kept-ws', name: 'kept', disabled: false }, + { id: 'other-ws', name: 'other', disabled: false } + ] as never + }) + // The sweep runs only where Web Locks exist, and only as a tab loads: `login` is one. + const holders = new Map() + installLocks(holders) + const inUse = `wm-ai-sessions-mirror::${user.email}::in-use` + const otherTab = (n: number) => holders.set(inUse, (holders.get(inUse) ?? 0) + n) + await login(user) + const day = 24 * 60 * 60 * 1000 + const old = Date.now() - 31 * day + const stale = (id: string, over: Partial = {}) => + session({ id, createdAt: old, lastActivityAt: old, workspace_id: 'kept-ws', ...over }) + // Archived or not, a session is judged by its own last activity; one read a day ago + // stays, as do one restored here a day ago whatever the backup's time and one in a + // workspace without retention. + await putSession(stale('stale')) + await putSession(stale('stale-archived', { archived: true })) + await putSession(stale('read-lately', { lastActivityAt: Date.now() - day })) + await putSession(stale('restored-lately', { restoredAt: Date.now() - day })) + await putSession(stale('elsewhere', { workspace_id: 'other-ws' })) + + const retentionMock = vi.mocked(WorkspaceService.getSessionWorkspaceRetention) + let told: Record = { 'kept-ws': 30 } + retentionMock.mockImplementation(async () => told as never) + // The sweep believes a remembered answer for a day, so ageing it is how a later load + // is made to ask again. + const forgetWhenAsked = () => { + const key = `windmill_sessions_retention_days::${user.email}` + const remembered = JSON.parse(localStorage.getItem(key) ?? '{}') + localStorage.setItem(key, JSON.stringify({ ...remembered, at: Date.now() - 2 * day })) + } + const stored = async () => { + const db = await openDB(`windmill-sessions::${user.email}`, 1) + const ids = ((await db.getAll('sessions' as never)) as Session[]).map((s) => s.id) + db.close() + return ids.sort() + } + const chatDeletions = (id: string) => + deleteSessionChatsMock.mock.calls.filter(([sid, email]) => sid === id && email === user.email) + + // While another tab has the sessions loaded, nothing is swept. + otherTab(1) + await rehydrate(user) + expect(await stored()).toContain('stale') + expect(chatDeletions('stale')).toHaveLength(0) + otherTab(-1) + + // The retention is cleared when the sweep asks: what the server says then is what + // deletes, and a browser that remembered one deletes nothing on it. + told = {} + await rehydrate(user) + expect(await stored()).toContain('stale') + expect(chatDeletions('stale')).toHaveLength(0) + told = { 'kept-ws': 30 } + forgetWhenAsked() + + // The chats of the first expired session the sweep reaches, `stale` by key order, + // cannot be deleted this time. + deleteSessionChatsMock.mockResolvedValueOnce(false) + await rehydrate(user) + expect(await stored()).toEqual(['elsewhere', 'read-lately', 'restored-lately']) + const pending = (id: string) => + localStorage.getItem(`windmill_sessions_retention_pending::${user.email}::${id}`) + expect(chatDeletions('stale-archived')).toHaveLength(1) + expect(pending('stale')).toBe('1') + expect(pending('stale-archived')).toBeNull() + + // The next load finishes what that deletion left, with nothing else to sweep. + await rehydrate(user) + expect(pending('stale')).toBeNull() + expect(chatDeletions('stale')).toHaveLength(2) + + // A swept session is not tombstoned: the backup another device pushed to brings it back. + await importSessions([stale('stale')], user.email) + expect(await stored()).toContain('stale') + // Both are shared with the tests that follow, which expect neither. + retentionMock.mockResolvedValue({} as never) + Object.defineProperty(navigator, 'locks', { value: undefined, configurable: true }) + }) + it('clears the in-memory list on logout', async () => { const user = freshUser() await login(user) diff --git a/frontend/src/lib/components/workspaceSettings/AISettings.svelte b/frontend/src/lib/components/workspaceSettings/AISettings.svelte index a1dcc76758..14d27cab18 100644 --- a/frontend/src/lib/components/workspaceSettings/AISettings.svelte +++ b/frontend/src/lib/components/workspaceSettings/AISettings.svelte @@ -693,7 +693,7 @@