feat: retention for AI sessions on the object store and in the browser (#11152)

* feat: retention for AI sessions, swept on the object store and in the browser

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: make the retention sweeps retryable and safe against pushes

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

* fix: spare other tabs' sessions, reclaim abandoned split pushes

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

* fix: sweep under an exclusive session lock, keep the captured user

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

* docs: say a tab selecting a session mid-sweep is not held back

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

* fix: sweep local sessions only while no other tab has them loaded

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

* fix: one tab sweeps at a time, and keeps the switched user's hold

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

* test: push the fallback session again before the rotation assertions

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

* refactor: keep retention server-side here, move the browser sweep out

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

* docs: the retention setting no longer touches browser-local sessions

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

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

backend/src/monitor.rs

@@ -4380,6 +4380,23 @@ pub async fn monitor_db(
         }
     };
 
+    // Delete the AI session backups older than their workspace's retention. Every ~40 min
+    // (240 iterations at the default 10 s, the most a u8 `should_run` counts): the retention
+    // counts in days. Spawned for the same reason as the credential maintenance above, a
+    // sweep of many sessions outlasting the join's deadline; the sweep's own advisory lock
+    // keeps one server at a time at it.
+    let ai_session_retention_f = async {
+        #[cfg(feature = "parquet")]
+        if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(240) {
+            if let Some(db) = conn.as_sql() {
+                let db = db.clone();
+                tokio::spawn(
+                    async move { windmill_api::sweep_expired_ai_session_backups(&db).await },
+                );
+            }
+        }
+    };
+
     // run every 2 iterations (~20s at the default LISTEN_NEW_EVENTS_INTERVAL_SEC).
     // Enterprise feature: the active `// freshness` backstop lives in
     // windmill-queue's `freshness_watchdog` (`private`); OSS gets a no-op stub.
@@ -4434,6 +4451,7 @@ pub async fn monitor_db(
         cleanup_scheduled_job_deletions_f,
         git_auto_pull_f,
         git_credential_maintenance_f,
+        ai_session_retention_f,
         pipeline_freshness_watchdog_f,
         reconcile_unarmed_schedules_f,
     );

backend/tests/ai_sessions.rs

@@ -1235,6 +1235,161 @@ async fn test_backup_writes_are_refused_for_the_wrong_owner_token_or_id(
     Ok(())
 }
 
+/// Sets the object's modification time `days` back: the FilesystemStorage answers
+/// `last_modified` from it, so this is a session no push touched since.
+fn age_object(path: &std::path::Path, days: u64) -> std::io::Result<()> {
+    let at = std::time::SystemTime::now() - std::time::Duration::from_secs(days * 86_400);
+    std::fs::File::options()
+        .write(true)
+        .open(path)?
+        .set_modified(at)
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_expired_backups_are_swept_by_age_and_left_out_of_the_listing(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let base = format!(
+        "http://localhost:{}/api/w/test-workspace",
+        server.addr.port()
+    );
+    let storage_dir = tempfile::tempdir()?;
+    configure_primary_lfs(&db, &storage_dir.path().to_string_lossy()).await?;
+
+    let chat = |sid: &str, cid: &str| {
+        json!({ "id": cid, "record": { "id": cid, "sessionId": sid, "lastModified": 2,
+                                        "actualMessages": [], "displayMessages": [] } })
+    };
+    let whole = |sid: &str| {
+        json!({
+            "id": sid, "whole": true, "epoch": 0,
+            "head": { "id": sid, "workspace_id": "test-workspace", "createdAt": 1, "chatId": "c1" },
+            "chats": [chat(sid, "c1")],
+            "images": [{ "chat_id": "c1", "id": "img1", "data_url": "data:image/png;base64,AAAA" }],
+            "artifacts": { "items": [], "versions": [] }
+        })
+    };
+    // Two pushes split over parts of which only the first part landed: one a browser
+    // abandoned long ago (its token aged past the retention), one still in flight.
+    let opening = |sid: &str| {
+        json!({
+            "id": sid, "whole": true, "epoch": 0, "push": format!("t-{sid}"), "opens": true,
+            "partial": true, "chats": [chat(sid, "c1")],
+            "head": { "id": sid, "workspace_id": "test-workspace", "createdAt": 1, "chatId": "c1" }
+        })
+    };
+    let resp = push(
+        &base,
+        "SECRET_TOKEN",
+        json!({ "owner": "test@windmill.dev", "sessions": [
+            whole("old"), whole("live"), opening("abandoned"), opening("inflight")
+        ] }),
+    )
+    .await?;
+    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
+    let root = user_root(storage_dir.path(), "test@windmill.dev");
+    age_object(&root.join("index/old/0"), 40)?;
+    age_object(&root.join("index/abandoned/push"), 40)?;
+
+    let listed = |listing: Value| -> Vec<String> {
+        let mut ids: Vec<String> = listing["sessions"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|s| s["id"].as_str().unwrap().to_string())
+            .collect();
+        ids.sort();
+        ids
+    };
+    let objects = |root: &std::path::Path| -> Vec<String> {
+        files_under(root)
+            .into_iter()
+            .map(|(p, _)| p.strip_prefix(root).unwrap().to_string_lossy().into_owned())
+            .collect()
+    };
+
+    // Without a retention nothing is swept, however old.
+    windmill_api::sweep_expired_ai_session_backups(&db).await;
+    assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live", "old"]);
+
+    let set_retention = |days: Value| {
+        authed(
+            client().post(format!("{base}/workspaces/edit_copilot_config")),
+            "SECRET_TOKEN",
+        )
+        .json(&json!({ "sessions_retention_days": days }))
+        .send()
+    };
+    let resp = set_retention(json!(0)).await?;
+    assert_eq!(resp.status(), 400, "{}", resp.text().await?);
+    let resp = set_retention(json!(30)).await?;
+    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
+
+    // The listing leaves the expired session out before the sweep reaches it.
+    assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]);
+    assert!(root.join("sessions/old/head.json").exists());
+
+    // A removal cut short (a directory stands where the head is, so it cannot be unlinked)
+    // leaves the sweep's record with the markers gone; the next pass finds it and finishes.
+    let head = root.join("sessions/old/head.json");
+    std::fs::remove_file(&head)?;
+    std::fs::create_dir(&head)?;
+    std::fs::write(head.join("planted"), b"")?;
+    windmill_api::sweep_expired_ai_session_backups(&db).await;
+    assert!(root.join("index/old/sweep").exists());
+    assert!(!root.join("index/old/0").exists());
+    assert!(root.join("sessions/old/chats/c1.json").exists());
+    assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]);
+    std::fs::remove_dir_all(&head)?;
+
+    windmill_api::sweep_expired_ai_session_backups(&db).await;
+    let remaining = objects(&root);
+    assert!(
+        remaining
+            .iter()
+            .all(|p| !p.contains("/old/") && !p.contains("/abandoned/")),
+        "{remaining:?}"
+    );
+    for kept in [
+        "index/live/0",
+        "sessions/live/head.json",
+        "sessions/live/chats/c1.json",
+        "images/live/c1/img1",
+        "index/inflight/push",
+        "sessions/inflight/head.json",
+    ] {
+        assert!(
+            remaining.iter().any(|p| p == kept),
+            "{kept} in {remaining:?}"
+        );
+    }
+    assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]);
+
+    // A second pass has nothing to do; a session pushed again since its marker aged is
+    // renewed by the push, which rewrites the marker.
+    windmill_api::sweep_expired_ai_session_backups(&db).await;
+    age_object(&root.join("index/live/0"), 40)?;
+    let resp = push(
+        &base,
+        "SECRET_TOKEN",
+        json!({ "owner": "test@windmill.dev",
+                "sessions": [{ "id": "live", "epoch": 0, "chats": [chat("live", "c2")] }] }),
+    )
+    .await?;
+    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
+    windmill_api::sweep_expired_ai_session_backups(&db).await;
+    let mut after = objects(&root);
+    after.sort();
+    let mut expected = remaining.clone();
+    expected.push("sessions/live/chats/c2.json".to_string());
+    expected.sort();
+    assert_eq!(after, expected);
+    assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]);
+    Ok(())
+}
+
 /// Puts the process-wide instance store back to none, even when an assertion fails.
 struct ResetInstanceStore;
 impl Drop for ResetInstanceStore {
@@ -1333,6 +1488,23 @@ async fn test_backups_fall_back_to_the_instance_storage(db: Pool<Postgres>) -> a
         .unwrap_or_else(|| panic!("no fallback usage in {usage}"));
     assert!(fallback_usage["bytes"].as_i64().unwrap() > 0);
 
+    // The retention sweep reaches what the instance store keeps for the workspace, choosing
+    // that store from the row it reads the generation from.
+    let instance_user = user_root(instance_dir.path(), "test@windmill.dev");
+    age_object(&instance_user.join("index/s1/0"), 40)?;
+    sqlx::query(
+        "UPDATE workspace_settings SET ai_config = '{\"sessions_retention_days\": 30}' \
+         WHERE workspace_id = 'test-workspace'",
+    )
+    .execute(&db)
+    .await?;
+    windmill_api::sweep_expired_ai_session_backups(&db).await;
+    assert!(!instance_user.join("index/s1/0").exists());
+    assert!(!instance_user.join("sessions/s1/head.json").exists());
+    // Pushed again, so the rotation below has a backup to delete.
+    assert_eq!(push_whole().await?.status(), 200);
+    assert!(!files_under(&in_instance).is_empty());
+
     // A key rotation sweeps the older generation out of the instance store too.
     rotate(&base, &"c".repeat(64)).await?;
     wait_until_empty(&in_instance, "a rotation on the instance store").await;

backend/windmill-api-workspaces/src/ai_session_backups.rs

@@ -123,8 +123,9 @@ pub async fn fallback_store(db: &DB) -> Result<Option<BackupStore>> {
 }
 
 /// The workspace's primary storage, resolved without a caller: a rotation runs its deletion
-/// off its own request.
-async fn primary_store(db: &DB, w_id: &str) -> Result<Option<BackupStore>> {
+/// off its own request, and the retention sweep off any. The caller must be the server
+/// itself; nothing here checks who asks.
+pub async fn primary_store(db: &DB, w_id: &str) -> Result<Option<BackupStore>> {
     let Some(lfs_json) = sqlx::query_scalar!(
         "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1",
         w_id

backend/windmill-api-workspaces/src/workspaces.rs

@@ -5629,6 +5629,14 @@ struct SessionWorkspaceStatusRequest {
     workspace_ids: Vec<String>,
 }
 
+/// `ai_config.sessions_retention_days` as stored, `None` when unset or not a count of days.
+pub fn sessions_retention_days(value: Option<&serde_json::Value>) -> Option<u32> {
+    value
+        .and_then(|v| v.as_u64())
+        .filter(|days| *days >= 1)
+        .and_then(|days| u32::try_from(days).ok())
+}
+
 /// Reconciliation support for client-side AI sessions, which the backend cannot touch
 /// directly. The client posts the workspace ids its sessions reference and uses the
 /// per-id status to keep sessions in sync with workspace lifecycle: `deleted` (no row, or

backend/windmill-api/openapi.yaml

@@ -28448,6 +28448,14 @@ components:
           description: >-
             Stops browsers from backing their AI sessions up to the workspace's object
             storage. Read from the workspace's own settings like `copilot_disabled`.
+        sessions_retention_days:
+          type: integer
+          minimum: 1
+          maximum: 3650
+          description: >-
+            The server deletes the backup of a session no push has reached for this many
+            days. Unset keeps backups until the user deletes the session. Read from the
+            workspace's own settings like `copilot_disabled`.
 
     AISessionBackupListing:
       type: object

backend/windmill-api/src/ai.rs

@@ -455,6 +455,12 @@ pub struct AIConfig {
     /// (`ai_sessions.rs`). Read from the workspace's own row like `copilot_disabled`.
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub sessions_storage_disabled: bool,
+    /// The server's sweep (`ai_sessions.rs`) deletes the backup of a session no push has
+    /// reached for this many days. The copies in members' browsers are untouched. Unset
+    /// keeps backups until the user deletes the session. Read from the workspace's own row
+    /// like `copilot_disabled`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sessions_retention_days: Option<u32>,
 }
 
 /// Negotiated rates in USD per million tokens. An unset cache rate is read as the
@@ -491,6 +497,9 @@ impl ModelPriceOverride {
     }
 }
 
+/// Ten years: past any plausible retention, and well within what a day count is turned into.
+pub const MAX_SESSIONS_RETENTION_DAYS: u32 = 3650;
+
 impl AIConfig {
     pub fn validate_model_pricing(&self) -> Result<()> {
         for (key, price) in self.model_pricing.iter().flatten() {
@@ -499,6 +508,17 @@ impl AIConfig {
         Ok(())
     }
 
+    pub fn validate_sessions_retention(&self) -> Result<()> {
+        match self.sessions_retention_days {
+            Some(days) if !(1..=MAX_SESSIONS_RETENTION_DAYS).contains(&days) => {
+                Err(Error::BadRequest(format!(
+                    "AI session retention must be between 1 and {MAX_SESSIONS_RETENTION_DAYS} days (got {days})"
+                )))
+            }
+            _ => Ok(()),
+        }
+    }
+
     pub fn has_providers(&self) -> bool {
         self.providers
             .as_ref()

backend/windmill-api/src/ai_sessions.rs

@@ -30,8 +30,9 @@ use serde_json::value::RawValue;
 use std::sync::Arc;
 use windmill_api_auth::is_effectively_unscoped;
 use windmill_api_workspaces::ai_session_backups::{
-    fallback_store, generation_prefix, storage_id, MAX_OBJECT_BYTES,
+    fallback_store, generation_prefix, primary_store, storage_id, MAX_OBJECT_BYTES,
 };
+use windmill_api_workspaces::workspaces::sessions_retention_days;
 use windmill_common::error::{Error, JsonResult, Result};
 use windmill_common::utils::calculate_hash;
 use windmill_common::variables::{crypt_from_key_with_suffix, get_workspace_key};
@@ -65,6 +66,46 @@ const MAX_LISTED_OBJECTS: usize = 5000;
 const MAX_LIST_SCAN: usize = 50_000;
 const LIST_MAX: usize = 500;
 const IO_CONCURRENCY: usize = 8;
+/// Sessions the retention sweep deletes per workspace and pass at most; the rest wait for
+/// the next pass.
+const SWEEP_MAX_PER_WORKSPACE: usize = 1000;
+/// Session-level advisory lock of the retention sweep: one server at a time runs it.
+const SWEEP_LOCK_ID: i64 = 0x5745_4550_4149;
+/// The name of the sweep's record next to a session's markers (see `Backend::sweep_key`).
+const SWEEP_RECORD: &str = "sweep";
+/// The name of a split push's token next to a session's markers (see `Backend::push_key`).
+const PUSH_TOKEN: &str = "push";
+
+/// A marker modified before this is past a retention of `days`.
+fn retention_cutoff(days: u32) -> chrono::DateTime<chrono::Utc> {
+    chrono::Utc::now() - chrono::Duration::days(i64::from(days))
+}
+
+/// What a key under the `index/` prefix is.
+enum IndexEntry {
+    /// The marker that lists the session, named by its epoch.
+    Marker(u32),
+    /// The retention sweep's record (see `Backend::sweep_key`).
+    Sweep,
+    /// The token of a push split over parts (see `Backend::push_key`).
+    Push,
+}
+
+/// The session a key under the `index/` prefix belongs to, and what the key is.
+fn index_entry<'a>(index: &ObjectPath, key: &'a ObjectPath) -> Option<(&'a str, IndexEntry)> {
+    // `Path` drops the trailing delimiter, so the remainder starts with one.
+    let rel = key.as_ref().strip_prefix(index.as_ref())?;
+    let (sid, name) = rel.trim_start_matches('/').split_once('/')?;
+    if sid.is_empty() {
+        return None;
+    }
+    let entry = match name {
+        SWEEP_RECORD => IndexEntry::Sweep,
+        PUSH_TOKEN => IndexEntry::Push,
+        epoch => IndexEntry::Marker(epoch.parse().ok()?),
+    };
+    Some((sid, entry))
+}
 
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -101,6 +142,9 @@ struct Backend {
     /// owed to the storage alone (a rotation deleted the older generation's copy anyway).
     storage_id: String,
     generation: i64,
+    /// `ai_config.sessions_retention_days`: a session whose marker is older is not listed,
+    /// whether or not the sweep has deleted it yet.
+    retention_days: Option<u32>,
     /// The store is the instance object store standing in for a workspace without storage
     /// of its own (`ai_session_backups::fallback_store`).
     fallback: bool,
@@ -111,6 +155,12 @@ impl Backend {
         ObjectPath::from(format!("{}/index/", self.prefix))
     }
 
+    /// The moment a marker's modification time must reach to count as live, under the
+    /// workspace's retention; `None` without one.
+    fn retention_cutoff(&self) -> Option<chrono::DateTime<chrono::Utc>> {
+        self.retention_days.map(retention_cutoff)
+    }
+
     /// The marker that lists the session, named by the session's move count so that of a
     /// session two workspaces list, the copy moved last is told from the listing alone.
     fn index_key(&self, sid: &str, epoch: u32) -> ObjectPath {
@@ -121,10 +171,18 @@ impl Backend {
         ObjectPath::from(format!("{}/index/{sid}/", self.prefix))
     }
 
-    /// The token of the push split over parts in progress, under the session so a removal
-    /// or the next whole push clears it with the rest.
+    /// Written by the retention sweep before it deletes anything of a session, and deleted
+    /// last (`remove_session`): what finds a removal the sweep started and could not finish,
+    /// the markers being gone by then. Not an epoch, so nothing lists or pulls a session by it.
+    fn sweep_key(&self, sid: &str) -> ObjectPath {
+        ObjectPath::from(format!("{}/index/{sid}/{SWEEP_RECORD}", self.prefix))
+    }
+
+    /// The token of the push split over parts in progress, next to the markers so a removal
+    /// or the next whole push clears it with them, and the retention sweep, which walks the
+    /// markers, finds one a browser abandoned.
     fn push_key(&self, sid: &str) -> ObjectPath {
-        ObjectPath::from(format!("{}/sessions/{sid}/push", self.prefix))
+        ObjectPath::from(format!("{}/index/{sid}/{PUSH_TOKEN}", self.prefix))
     }
 
     fn session_prefix(&self, sid: &str) -> ObjectPath {
@@ -310,7 +368,12 @@ impl Backend {
             let mut stream = self.store.list(Some(&prefix));
             while let Some(meta) = stream.next().await {
                 let meta = meta.map_err(object_store_error_to_error)?;
-                listed |= marker;
+                // The sweep's record is not a marker: a session it started removing is absent.
+                listed |= marker
+                    && meta
+                        .location
+                        .filename()
+                        .is_some_and(|name| name.parse::<u32>().is_ok());
                 acc = fold(
                     acc,
                     meta.location.as_ref(),
@@ -415,17 +478,21 @@ fn require_json_object(kind: &str, raw: &RawValue, max_bytes: usize) -> Result<(
 /// no instance store to stand in, or the admin switched them off. Both read as
 /// `enabled: false` so the browser stops trying.
 async fn backend(authed: &ApiAuthed, db: &DB, w_id: &str) -> Result<Option<Backend>> {
-    let (disabled, generation, has_storage) = sqlx::query_as::<_, (Option<bool>, i64, bool)>(
-        "SELECT (ai_config->>'sessions_storage_disabled')::bool, ai_sessions_backup_generation, \
-         large_file_storage IS NOT NULL FROM workspace_settings WHERE workspace_id = $1",
-    )
-    .bind(w_id)
-    .fetch_optional(db)
-    .await?
-    .unwrap_or((None, 0, false));
+    let (disabled, retention, generation, has_storage) =
+        sqlx::query_as::<_, (Option<bool>, Option<serde_json::Value>, i64, bool)>(
+            "SELECT (ai_config->>'sessions_storage_disabled')::bool, \
+                    ai_config->'sessions_retention_days', ai_sessions_backup_generation, \
+                    large_file_storage IS NOT NULL \
+             FROM workspace_settings WHERE workspace_id = $1",
+        )
+        .bind(w_id)
+        .fetch_optional(db)
+        .await?
+        .unwrap_or((None, None, 0, false));
     if disabled.unwrap_or(false) {
         return Ok(None);
     }
+    let retention_days = sessions_retention_days(retention.as_ref());
     // Decided from the row the generation came from: the instance store is written only
     // under a generation read while the workspace had no storage of its own, which
     // configuring one moves past (`ai_session_backups`).
@@ -459,6 +526,7 @@ async fn backend(authed: &ApiAuthed, db: &DB, w_id: &str) -> Result<Option<Backe
         prefix,
         storage_id,
         generation,
+        retention_days,
         fallback,
     }))
 }
@@ -493,7 +561,9 @@ struct ListResponse {
 }
 
 /// A session is listed once a push entry of it landed whole (its marker is written last);
-/// a push that failed before that left objects the listing does not name.
+/// a push that failed before that left objects the listing does not name. One past the
+/// workspace's retention is not listed either, whether or not the sweep has reached it, so a
+/// browser never restores it.
 async fn list(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -511,6 +581,7 @@ async fn list(
         }));
     };
     let prefix = backend.index_prefix();
+    let cutoff = backend.retention_cutoff();
     let mut stream = backend.store.list(Some(&prefix));
     // One marker per session, whatever the session holds: the newest LIST_MAX are kept as
     // the scan goes (a min-heap drops the oldest), and the scan itself is bounded.
@@ -526,19 +597,12 @@ async fn list(
             truncated = true;
             break;
         }
-        // `Path` drops the trailing delimiter, so the remainder starts with one.
-        let Some(rel) = meta.location.as_ref().strip_prefix(prefix.as_ref()) else {
-            continue;
-        };
-        let Some((sid, epoch)) = rel.trim_start_matches('/').split_once('/') else {
-            continue;
-        };
-        let Ok(epoch) = epoch.parse::<u32>() else {
-            continue;
-        };
-        if sid.is_empty() || sid.contains('/') {
+        if cutoff.is_some_and(|cutoff| meta.last_modified < cutoff) {
             continue;
         }
+        let Some((sid, IndexEntry::Marker(epoch))) = index_entry(&prefix, &meta.location) else {
+            continue;
+        };
         newest.push(std::cmp::Reverse((
             meta.last_modified,
             epoch,
@@ -1206,15 +1270,29 @@ async fn push_session(backend: &Backend, s: &PushedSession) -> Result<(usize, bo
     Ok((written, false))
 }
 
-/// The marker goes first so a removal cut short leaves nothing listed, then the head so
+/// The markers go first so a removal cut short leaves nothing listed, then the head so
 /// nothing pulls either, and no push takes it for a session still there (see `push_session`).
+/// The retention sweep's record goes last (see `Backend::sweep_key`).
 async fn remove_session(backend: &Backend, sid: &str) -> Result<()> {
+    let sweep = backend.sweep_key(sid);
     backend
-        .delete_prefix(&backend.index_session_prefix(sid))
+        .store
+        .list(Some(&backend.index_session_prefix(sid)))
+        .map_err(object_store_error_to_error)
+        .try_for_each_concurrent(IO_CONCURRENCY, |meta| {
+            let sweep = &sweep;
+            async move {
+                if meta.location == *sweep {
+                    return Ok(());
+                }
+                backend.delete(&meta.location).await
+            }
+        })
         .await?;
     backend.delete(&backend.head_key(sid)).await?;
     backend.delete_prefix(&backend.session_prefix(sid)).await?;
-    backend.delete_prefix(&backend.images_prefix(sid)).await
+    backend.delete_prefix(&backend.images_prefix(sid)).await?;
+    backend.delete(&sweep).await
 }
 
 /// One writer per session at a time, across servers: a push and a removal of the same
@@ -1261,6 +1339,205 @@ async fn remove_session_locked(db: &DB, backend: &Backend, sid: &str) -> Result<
     result
 }
 
+/// Deletes, in every workspace with `ai_config.sessions_retention_days`, the backups of the
+/// sessions whose marker is older than that: the marker is rewritten by every push that
+/// completes, so its modification time is the session's last activity as the storage clocks
+/// it. For the monitor, on every server: a session-level advisory lock keeps one pass at a
+/// time across them. The walk reads markers only, one object per session and nothing of what
+/// the sessions hold, under each user's prefix in turn (`list_with_delimiter` names the
+/// users), and deletes at most `SWEEP_MAX_PER_WORKSPACE` sessions per workspace and pass. A
+/// session goes under its lock (`lock_session`), once its markers are listed again there and
+/// still all older (see `sweep_session`). A removal cut short leaves the sweep's record next
+/// to the markers, which the walk also collects, so the next pass finishes it.
+pub async fn sweep_expired_ai_session_backups(db: &DB) {
+    let mut lock_conn = match db.acquire().await {
+        Ok(conn) => conn,
+        Err(e) => {
+            tracing::error!("AI session retention: could not acquire a connection: {e:#}");
+            return;
+        }
+    };
+    let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
+        .bind(SWEEP_LOCK_ID)
+        .fetch_one(&mut *lock_conn)
+        .await
+    {
+        Ok(locked) => locked,
+        Err(e) => {
+            tracing::error!("AI session retention: advisory lock failed: {e:#}");
+            return;
+        }
+    };
+    if !locked {
+        return;
+    }
+    if let Err(e) = sweep_workspaces(db).await {
+        tracing::error!("AI session retention sweep failed: {e:#}");
+    }
+    if let Err(e) = sqlx::query("SELECT pg_advisory_unlock($1)")
+        .bind(SWEEP_LOCK_ID)
+        .execute(&mut *lock_conn)
+        .await
+    {
+        tracing::error!("AI session retention: advisory unlock failed: {e:#}");
+    }
+}
+
+async fn sweep_workspaces(db: &DB) -> Result<()> {
+    let workspaces = sqlx::query_as::<_, (String, Option<serde_json::Value>, i64, bool)>(
+        "SELECT workspace_id, ai_config->'sessions_retention_days', ai_sessions_backup_generation, \
+                large_file_storage IS NOT NULL \
+         FROM workspace_settings \
+         WHERE ai_config->'sessions_retention_days' IS NOT NULL",
+    )
+    .fetch_all(db)
+    .await?;
+    for (w_id, retention, generation, has_storage) in workspaces {
+        let Some(days) = sessions_retention_days(retention.as_ref()) else {
+            continue;
+        };
+        match sweep_workspace(db, &w_id, days, generation, has_storage).await {
+            Ok(0) => {}
+            Ok(deleted) => tracing::info!(
+                "AI session retention deleted {deleted} session backups of {w_id} older than {days} days"
+            ),
+            Err(e) => tracing::warn!("AI session retention sweep of {w_id}: {e:#}"),
+        }
+    }
+    Ok(())
+}
+
+/// `has_storage` comes from the row `generation` was read from, as in `backend`: the instance
+/// store is swept only under a generation read while the workspace had no storage of its own.
+async fn sweep_workspace(
+    db: &DB,
+    w_id: &str,
+    days: u32,
+    generation: i64,
+    has_storage: bool,
+) -> Result<usize> {
+    let resolved = if has_storage {
+        primary_store(db, w_id).await?
+    } else {
+        fallback_store(db).await?
+    };
+    let Some(resolved) = resolved else {
+        return Ok(0);
+    };
+    let key = get_workspace_key(w_id, db).await?;
+    let (store, storage_id) = (resolved.store, resolved.storage_id);
+    let cutoff = retention_cutoff(days);
+    let root = ObjectPath::from(generation_prefix(w_id, generation));
+    let users = store
+        .list_with_delimiter(Some(&root))
+        .await
+        .map_err(object_store_error_to_error)?
+        .common_prefixes;
+    let mut deleted = 0;
+    for user_prefix in users {
+        let Some(user) = user_prefix.filename() else {
+            continue;
+        };
+        // The sweep decrypts nothing; the cipher is only what a `Backend` is made of.
+        let backend = Backend {
+            store: store.clone(),
+            mc: crypt_from_key_with_suffix(&key, user),
+            prefix: user_prefix.to_string(),
+            storage_id: storage_id.clone(),
+            generation,
+            retention_days: Some(days),
+            fallback: resolved.fallback,
+        };
+        let index = backend.index_prefix();
+        let mut markers = backend.store.list(Some(&index));
+        let mut expired = std::collections::BTreeSet::new();
+        while let Some(meta) = markers.next().await {
+            let meta = meta.map_err(object_store_error_to_error)?;
+            let sid = match index_entry(&index, &meta.location) {
+                Some((sid, IndexEntry::Sweep)) => sid,
+                Some((sid, IndexEntry::Marker(_) | IndexEntry::Push))
+                    if meta.last_modified < cutoff =>
+                {
+                    sid
+                }
+                _ => continue,
+            };
+            expired.insert(sid.to_string());
+            if deleted + expired.len() >= SWEEP_MAX_PER_WORKSPACE {
+                break;
+            }
+        }
+        for sid in expired {
+            match sweep_session(db, &backend, &sid, cutoff).await {
+                Ok(true) => deleted += 1,
+                Ok(false) => {}
+                Err(e) => tracing::warn!(
+                    "AI session retention left the backup of {sid} in {w_id} for the next pass: {e:#}"
+                ),
+            }
+        }
+        if deleted >= SWEEP_MAX_PER_WORKSPACE {
+            break;
+        }
+    }
+    Ok(deleted)
+}
+
+/// True when the session was deleted. Under the session's lock its markers are listed again:
+/// one a push renewed since the walk keeps the session. A session with none is left alone
+/// while a push split over parts is between two of them (its token younger than the
+/// retention) or it is gone, unless the sweep's record says a removal was started; an older
+/// token is a split push a browser abandoned, whose landed parts nothing lists. The record is
+/// written before anything is deleted and removed last, so a removal cut short is found again
+/// by the next pass.
+async fn sweep_session(
+    db: &DB,
+    backend: &Backend,
+    sid: &str,
+    cutoff: chrono::DateTime<chrono::Utc>,
+) -> Result<bool> {
+    let tx = lock_session(db, backend, sid).await?;
+    let result = async {
+        let (sweep, push) = (backend.sweep_key(sid), backend.push_key(sid));
+        let mut entries = backend.store.list(Some(&backend.index_session_prefix(sid)));
+        let (mut listed, mut renewed, mut started, mut abandoned) = (false, false, false, false);
+        while let Some(meta) = entries.next().await {
+            let meta = meta.map_err(object_store_error_to_error)?;
+            if meta.location == sweep {
+                started = true;
+            } else if meta.location == push {
+                abandoned = meta.last_modified < cutoff;
+            } else {
+                listed = true;
+                renewed |= meta.last_modified >= cutoff;
+            }
+        }
+        if renewed {
+            // A push listed the session again over a removal cut short before its markers
+            // went, which had deleted nothing else.
+            if started {
+                backend.delete(&sweep).await?;
+            }
+            return Ok(false);
+        }
+        if !listed && !started && !abandoned {
+            return Ok(false);
+        }
+        if !started {
+            backend
+                .store
+                .put(&sweep, PutPayload::new())
+                .await
+                .map_err(object_store_error_to_error)?;
+        }
+        remove_session(backend, sid).await?;
+        Ok(true)
+    }
+    .await;
+    tx.commit().await?;
+    result
+}
+
 async fn push(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,

backend/windmill-api/src/lib.rs

@@ -71,6 +71,8 @@ mod ai_free_tier_ee;
 mod ai_free_tier_oss;
 #[cfg(feature = "parquet")]
 mod ai_sessions;
+#[cfg(feature = "parquet")]
+pub use ai_sessions::sweep_expired_ai_session_backups;
 mod ai_shared_artifacts;
 mod apps;
 mod apps_raw_bundle;

backend/windmill-api/src/workspaces.rs

@@ -108,6 +108,7 @@ async fn edit_copilot_config(
     }
 
     ai_config.validate_model_pricing()?;
+    ai_config.validate_sessions_retention()?;
 
     let mut tx = db.begin().await?;
 
@@ -148,6 +149,7 @@ async fn edit_copilot_config(
     let workspace_has_config = ai_config.has_providers();
     let copilot_disabled = ai_config.copilot_disabled;
     let sessions_storage_disabled = ai_config.sessions_storage_disabled;
+    let sessions_retention_days = ai_config.sessions_retention_days;
     let instance_ai_config =
         sqlx::query_scalar!("SELECT value FROM global_settings WHERE name = 'ai_config'")
             .fetch_optional(&db)
@@ -176,6 +178,7 @@ async fn edit_copilot_config(
     };
     effective_ai_config.copilot_disabled = copilot_disabled;
     effective_ai_config.sessions_storage_disabled = sessions_storage_disabled;
+    effective_ai_config.sessions_retention_days = sessions_retention_days;
 
     Ok(Json(EditCopilotConfigResponse {
         effective_ai_config,
@@ -217,6 +220,9 @@ async fn get_copilot_info(
     let sessions_storage_disabled = workspace_ai_config
         .as_ref()
         .is_some_and(|c| c.0.sessions_storage_disabled);
+    let sessions_retention_days = workspace_ai_config
+        .as_ref()
+        .and_then(|c| c.0.sessions_retention_days);
     let instance_config =
         sqlx::query_scalar!("SELECT value FROM global_settings WHERE name = 'ai_config'")
             .fetch_optional(&db)
@@ -242,6 +248,7 @@ async fn get_copilot_info(
         };
     effective.copilot_disabled = copilot_disabled;
     effective.sessions_storage_disabled = sessions_storage_disabled;
+    effective.sessions_retention_days = sessions_retention_days;
     Ok(Json(effective))
 }
 

docs/ai-session-backups.md

@@ -201,7 +201,8 @@ its pulls are done, and just before they do the whole family is listed again (me
 backups were off included, since a move from another device can land in a workspace between
 the first listings and the pulls; a family of one, with nowhere else for a copy to show up, is
 not): a session a later copy of which showed up elsewhere is left, with the family, for the
-next time. Only a user-initiated `deleteSession` removes the backup; the next push from
+next time. Only a user-initiated `deleteSession`, and the retention sweep below, remove the
+backup; the next push from
 another device that still has the session is refused with `needs_whole` (nothing of it is
 written), its row goes stale without a backoff, and that device's next flush sends the session
 whole; the workspace-lifecycle
@@ -235,6 +236,37 @@ old one; a session whose own parts were answered from different storages is not
 all. The listing a restore starts with runs the same check, so a storage switch is noticed at
 the first push after it or on the next page load, whichever comes first.
 
+## Retention
+
+`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).
+
+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.
+
 ## Limits
 
 Push bodies are packed to about 8 MB (UTF-8 bytes as sent), at most 100 entries, 200 removals and

frontend/src/lib/components/workspaceSettings/AISettings.svelte

@@ -29,6 +29,7 @@
 	import AiUsagePanel from './AiUsagePanel.svelte'
 	import { setCopilotInfo } from '$lib/aiStore'
 	import { backupSettingsChanged } from '$lib/components/sessions/sessionMirror.svelte'
+	import TextInput from '../text_input/TextInput.svelte'
 	import AIPromptsModal from '../settings/AIPromptsModal.svelte'
 	import { Settings } from 'lucide-svelte'
 	import { untrack } from 'svelte'
@@ -81,6 +82,7 @@
 	let workspaceOverrideEditorOpened = $state(false)
 	let copilotDisabled = $state(false)
 	let sessionsStorageDisabled = $state(false)
+	let sessionsRetentionDays: number | undefined = $state(undefined)
 
 	// --- Initial state for dirty tracking ---
 	let initialAiProviders: Exclude<AIConfig['providers'], undefined> = $state({})
@@ -93,6 +95,7 @@
 	let initialPrompts: Record<string, string> = $state({})
 	let initialCopilotDisabled = $state(false)
 	let initialSessionsStorageDisabled = $state(false)
+	let initialSessionsRetentionDays: number | undefined = $state(undefined)
 	let lastLoadedConfigKey = $state<string | undefined>(undefined)
 
 	function clone<T>(v: T): T {
@@ -122,6 +125,7 @@
 		modelPricing = clone(config?.model_pricing ?? {})
 		copilotDisabled = config?.copilot_disabled === true
 		sessionsStorageDisabled = config?.sessions_storage_disabled === true
+		sessionsRetentionDays = config?.sessions_retention_days
 		for (const mode of ['edit', 'fix', 'gen']) {
 			if (!(mode in customPrompts)) {
 				customPrompts[mode] = ''
@@ -140,6 +144,7 @@
 		initialPrompts = clone(customPrompts)
 		initialCopilotDisabled = copilotDisabled
 		initialSessionsStorageDisabled = sessionsStorageDisabled
+		initialSessionsRetentionDays = sessionsRetentionDays
 	}
 
 	export function loadFromConfig(config: AIConfig | undefined) {
@@ -157,6 +162,7 @@
 		modelPricing = clone(initialModelPricing)
 		copilotDisabled = initialCopilotDisabled
 		sessionsStorageDisabled = initialSessionsStorageDisabled
+		sessionsRetentionDays = initialSessionsRetentionDays
 	}
 
 	$effect(() => {
@@ -193,7 +199,8 @@
 			JSON.stringify(maxTokensPerModel) !== JSON.stringify(initialMaxTokensPerModel) ||
 			JSON.stringify(modelPricing) !== JSON.stringify(initialModelPricing) ||
 			copilotDisabled !== initialCopilotDisabled ||
-			sessionsStorageDisabled !== initialSessionsStorageDisabled
+			sessionsStorageDisabled !== initialSessionsStorageDisabled ||
+			sessionsRetentionDays !== initialSessionsRetentionDays
 	)
 
 	$effect(() => {
@@ -298,9 +305,11 @@
 			.filter(([_, prompt]) => prompt.trim().length > 0)
 			.reduce((acc, [mode, prompt]) => ({ ...acc, [mode]: prompt }), {})
 
-		// The flags are what a workspace on instance defaults still stores of its own.
+		// The flags and the retention are what a workspace on instance defaults still stores
+		// of its own.
 		const copilot_disabled = copilotDisabled ? true : undefined
 		const sessions_storage_disabled = sessionsStorageDisabled ? true : undefined
+		const sessions_retention_days = sessionsRetentionDays
 		return Object.keys(aiProviders ?? {}).length > 0
 			? {
 					providers: aiProviders,
@@ -312,16 +321,27 @@
 						Object.keys(maxTokensPerModel).length > 0 ? maxTokensPerModel : undefined,
 					model_pricing: Object.keys(modelPricing).length > 0 ? modelPricing : undefined,
 					copilot_disabled,
-					sessions_storage_disabled
+					sessions_storage_disabled,
+					sessions_retention_days
 				}
-			: { copilot_disabled, sessions_storage_disabled }
+			: { copilot_disabled, sessions_storage_disabled, sessions_retention_days }
 	}
 
+	// The server refuses anything outside this range; the input holds the same bounds.
+	const MAX_SESSIONS_RETENTION_DAYS = 3650
+	let retentionInvalid = $derived(
+		sessionsRetentionDays !== undefined &&
+			(!Number.isInteger(sessionsRetentionDays) ||
+				sessionsRetentionDays < 1 ||
+				sessionsRetentionDays > MAX_SESSIONS_RETENTION_DAYS)
+	)
+
 	function isSaveDisabled(): boolean {
 		return (
 			!Object.values(aiProviders).every((p) => p.resource_path) ||
 			(metadataModel != undefined && metadataModel.length === 0) ||
-			(codeCompletionModel != undefined && codeCompletionModel.length === 0)
+			(codeCompletionModel != undefined && codeCompletionModel.length === 0) ||
+			retentionInvalid
 		)
 	}
 
@@ -671,6 +691,33 @@
 				options={{ right: 'Do not back AI sessions up to the workspace storage' }}
 			/>
 		</SettingCard>
+		<SettingCard
+			label="AI session retention"
+			description="Deletes the backup of an AI session from the workspace's object storage once no browser has pushed to it for this many days, archived sessions included. Sessions in members' browsers are not affected. Leave empty to keep backups until their owner deletes the session."
+		>
+			<div class="flex items-center gap-2">
+				<div class="w-28">
+					<TextInput
+						inputProps={{
+							type: 'number',
+							min: 1,
+							max: MAX_SESSIONS_RETENTION_DAYS,
+							step: 1,
+							placeholder: 'Forever'
+						}}
+						error={retentionInvalid ? `1 to ${MAX_SESSIONS_RETENTION_DAYS}` : undefined}
+						bind:value={
+							() => sessionsRetentionDays ?? '',
+							(v) => {
+								const n = typeof v === 'number' ? v : parseInt(v ?? '')
+								sessionsRetentionDays = Number.isNaN(n) ? undefined : n
+							}
+						}
+					/>
+				</div>
+				<span class="text-xs text-secondary">days</span>
+			</div>
+		</SettingCard>
 	{/if}
 </div>