Merge remote-tracking branch 'origin/main' into datatable-external-instance

# Conflicts: # backend/ee-repo-ref.txt
2026-10-19 08:03:16 +00:00 · 2026-09-21 17:30:08 +02:00
parent 39ea61c1c9 9f2685d788
commit 2fc1f89746
28 changed files with 688 additions and 158 deletions
@@ -1,38 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "WITH all_audit AS (SELECT username, operation, timestamp FROM audit_partitioned UNION ALL SELECT username, operation, timestamp FROM audit),\n        active_users as (SELECT distinct username as email FROM all_audit WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n        active_authors as (SELECT distinct email FROM usr WHERE usr.operator IS false AND email IN (SELECT email FROM active_users)),\n        active_authors_agg as (SELECT array_agg(email) as authors FROM active_authors),\n        active_ops_agg as (SELECT array_agg(email) as operators from active_users WHERE email NOT IN (SELECT email FROM active_authors))\n        SELECT active_authors_agg.authors, active_ops_agg.operators, array_length(active_authors_agg.authors, 1) as author_count, array_length(active_ops_agg.operators, 1) as operator_count FROM active_authors_agg, active_ops_agg",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "authors",
-        "type_info": "VarcharArray"
-      },
-      {
-        "ordinal": 1,
-        "name": "operators",
-        "type_info": "VarcharArray"
-      },
-      {
-        "ordinal": 2,
-        "name": "author_count",
-        "type_info": "Int4"
-      },
-      {
-        "ordinal": 3,
-        "name": "operator_count",
-        "type_info": "Int4"
-      }
-    ],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": [
-      null,
-      null,
-      null,
-      null
-    ]
-  },
-  "hash": "359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b"
-}
@@ -1,38 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n        WITH capped AS (\n            SELECT timestamp, operation, resource, parameters\n            FROM audit_partitioned\n            WHERE workspace_id = 'admins'\n              AND operation = 'workspace_fairness.capped'\n            UNION ALL\n            SELECT timestamp, operation, resource, parameters\n            FROM audit\n            WHERE workspace_id = 'admins'\n              AND operation = 'workspace_fairness.capped'\n            ORDER BY timestamp DESC\n            LIMIT 200\n        ), uncapped AS (\n            SELECT timestamp, operation, resource, parameters\n            FROM audit_partitioned\n            WHERE workspace_id = 'admins'\n              AND operation = 'workspace_fairness.uncapped'\n            UNION ALL\n            SELECT timestamp, operation, resource, parameters\n            FROM audit\n            WHERE workspace_id = 'admins'\n              AND operation = 'workspace_fairness.uncapped'\n            ORDER BY timestamp DESC\n            LIMIT 200\n        )\n        SELECT timestamp AS \"timestamp!\",\n               operation::text AS \"operation!\",\n               resource AS workspace_id,\n               parameters\n        FROM (SELECT * FROM capped UNION ALL SELECT * FROM uncapped) e\n        ORDER BY timestamp DESC\n        ",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "timestamp!",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 1,
-        "name": "operation!",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 2,
-        "name": "workspace_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 3,
-        "name": "parameters",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": [
-      null,
-      null,
-      null,
-      null
-    ]
-  },
-  "hash": "6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a"
-}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "WITH active_users AS (SELECT distinct username as email FROM (SELECT username, timestamp, operation FROM audit_partitioned UNION ALL SELECT username, timestamp, operation FROM audit) AS a WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n            authors as (SELECT distinct email FROM usr WHERE usr.operator IS false)\n            SELECT email as \"email!\", (email NOT IN (SELECT email FROM authors)) as operator_only, NULL::bool as is_workspace_admin, login_type::text, verified as \"verified!\", super_admin as \"super_admin!\", devops as \"devops!\", name, company, username, first_time_user as \"first_time_user!\", role_source as \"role_source!\", disabled as \"disabled!\", NULL::text as workspace_id\n            FROM password\n            WHERE email IN (SELECT email FROM active_users)\n            UNION ALL\n            SELECT email as \"email!\", operator as operator_only, is_admin as is_workspace_admin, 'service_account'::text as login_type, true as \"verified!\", false as \"super_admin!\", false as \"devops!\", NULL::text as name, NULL::text as company, username, false as \"first_time_user!\", 'service_account'::text as \"role_source!\", disabled as \"disabled!\", workspace_id\n            FROM usr\n            WHERE is_service_account IS true\n            ORDER BY \"super_admin!\" DESC, \"devops!\" DESC\n            LIMIT $1 OFFSET $2",
+  "query": "WITH active_users AS (SELECT distinct username as email FROM audit_partitioned WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n            authors as (SELECT distinct email FROM usr WHERE usr.operator IS false)\n            SELECT email as \"email!\", (email NOT IN (SELECT email FROM authors)) as operator_only, NULL::bool as is_workspace_admin, login_type::text, verified as \"verified!\", super_admin as \"super_admin!\", devops as \"devops!\", name, company, username, first_time_user as \"first_time_user!\", role_source as \"role_source!\", disabled as \"disabled!\", NULL::text as workspace_id\n            FROM password\n            WHERE email IN (SELECT email FROM active_users)\n            UNION ALL\n            SELECT email as \"email!\", operator as operator_only, is_admin as is_workspace_admin, 'service_account'::text as login_type, true as \"verified!\", false as \"super_admin!\", false as \"devops!\", NULL::text as name, NULL::text as company, username, false as \"first_time_user!\", 'service_account'::text as \"role_source!\", disabled as \"disabled!\", workspace_id\n            FROM usr\n            WHERE is_service_account IS true\n            ORDER BY \"super_admin!\" DESC, \"devops!\" DESC\n            LIMIT $1 OFFSET $2",
  "describe": {
    "columns": [
      {
@@ -97,5 +97,5 @@
      null
    ]
  },
-  "hash": "8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000"
+  "hash": "d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e"
 }
@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM audit WHERE timestamp <= now() - ($1::bigint::text || ' s')::interval",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Int8"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1"
-}
@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH active_users as (SELECT distinct username as email FROM audit_partitioned WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n        active_authors as (SELECT distinct email FROM usr WHERE usr.operator IS false AND email IN (SELECT email FROM active_users)),\n        active_authors_agg as (SELECT array_agg(email) as authors FROM active_authors),\n        active_ops_agg as (SELECT array_agg(email) as operators from active_users WHERE email NOT IN (SELECT email FROM active_authors))\n        SELECT active_authors_agg.authors, active_ops_agg.operators, array_length(active_authors_agg.authors, 1) as author_count, array_length(active_ops_agg.operators, 1) as operator_count FROM active_authors_agg, active_ops_agg",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "authors",
+        "type_info": "VarcharArray"
+      },
+      {
+        "ordinal": 1,
+        "name": "operators",
+        "type_info": "VarcharArray"
+      },
+      {
+        "ordinal": 2,
+        "name": "author_count",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 3,
+        "name": "operator_count",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null,
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c"
+}
@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        WITH capped AS (\n            SELECT timestamp, operation, resource, parameters\n            FROM audit_partitioned\n            WHERE workspace_id = 'admins'\n              AND operation = 'workspace_fairness.capped'\n            ORDER BY timestamp DESC\n            LIMIT 200\n        ), uncapped AS (\n            SELECT timestamp, operation, resource, parameters\n            FROM audit_partitioned\n            WHERE workspace_id = 'admins'\n              AND operation = 'workspace_fairness.uncapped'\n            ORDER BY timestamp DESC\n            LIMIT 200\n        )\n        SELECT timestamp AS \"timestamp!\",\n               operation::text AS \"operation!\",\n               resource AS workspace_id,\n               parameters\n        FROM (SELECT * FROM capped UNION ALL SELECT * FROM uncapped) e\n        ORDER BY timestamp DESC\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "timestamp!",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 1,
+        "name": "operation!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "parameters",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null,
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290"
+}
@@ -1 +1 @@
-a4783137d4ba1121cb5c09d4f5fe3b628071ad78
+466fc69233459ea477e062cd3e03e4442f1ea5fd
@@ -1953,20 +1953,6 @@ pub async fn delete_expired_items(db: &DB) -> () {
        tracing::error!("Error deleting expired shared AI artifacts: {:?}", e);
    }

-    let audit_retention_days = audit_log_retention_days().await;
-    let audit_retention_secs: i64 = audit_retention_days * 60 * 60 * 24;
-
-    // Clean up old (non-partitioned) audit table — will eventually be empty and dropped
-    if let Err(e) = sqlx::query_scalar!(
-        "DELETE FROM audit WHERE timestamp <= now() - ($1::bigint::text || ' s')::interval",
-        audit_retention_secs,
-    )
-    .fetch_all(db)
-    .await
-    {
-        tracing::error!("Error deleting audit log: {:?}", e);
-    }
-
    if let Err(e) = sqlx::query_scalar!(
        "DELETE FROM autoscaling_event WHERE applied_at <= now() - ($1::bigint::text || ' s')::interval",
        30 * 24 * 60 * 60, // 30 days
@@ -56,7 +56,7 @@ app_version_lite: id(bigint), value(jsonb)
  FK: (id) -> app_version(id)
 asset: workspace_id(char), path(char), kind(asset_kind), usage_access_type(asset_access_type), usage_path(char), usage_kind(asset_usage_kind), created_at(ts), id(bigint), columns(jsonb)
  FK: (workspace_id) -> workspace(id)
-audit: workspace_id(char), id(bigint), timestamp(ts), username(char), operation(char), action_kind(action_kind), resource(char), parameters(jsonb), email(char), span(char)
+audit_partitioned: workspace_id(char), id(bigint), timestamp(ts), username(char), operation(char), action_kind(action_kind), resource(char), parameters(jsonb), email(char), span(char)
 autoscaling_event: id(int), worker_group(text), event_type(autoscaling_event_type), desired_workers(int), applied_at(ts), reason(text)
 capture: workspace_id(char), path(char), created_at(ts), created_by(char), main_args(jsonb), is_flow(bool), trigger_kind(trigger_kind), preprocessor_args(jsonb), id(bigint)
  FK: (workspace_id) -> workspace(id)
@@ -530,7 +530,7 @@ async fn list_users_as_super_admin(
    let rows = if active_only.is_some_and(|x| x) {
        sqlx::query_as!(
            GlobalUserInfo,
-            r#"WITH active_users AS (SELECT distinct username as email FROM (SELECT username, timestamp, operation FROM audit_partitioned UNION ALL SELECT username, timestamp, operation FROM audit) AS a WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),
+            r#"WITH active_users AS (SELECT distinct username as email FROM audit_partitioned WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),
            authors as (SELECT distinct email FROM usr WHERE usr.operator IS false)
            SELECT email as "email!", (email NOT IN (SELECT email FROM authors)) as operator_only, NULL::bool as is_workspace_admin, login_type::text, verified as "verified!", super_admin as "super_admin!", devops as "devops!", name, company, username, first_time_user as "first_time_user!", role_source as "role_source!", disabled as "disabled!", NULL::text as workspace_id
            FROM password
@@ -436,11 +436,6 @@ async fn get_workspace_fairness_events(
        WITH capped AS (
            SELECT timestamp, operation, resource, parameters
            FROM audit_partitioned
-            WHERE workspace_id = 'admins'
-              AND operation = 'workspace_fairness.capped'
-            UNION ALL
-            SELECT timestamp, operation, resource, parameters
-            FROM audit
            WHERE workspace_id = 'admins'
              AND operation = 'workspace_fairness.capped'
            ORDER BY timestamp DESC
@@ -448,11 +443,6 @@ async fn get_workspace_fairness_events(
        ), uncapped AS (
            SELECT timestamp, operation, resource, parameters
            FROM audit_partitioned
-            WHERE workspace_id = 'admins'
-              AND operation = 'workspace_fairness.uncapped'
-            UNION ALL
-            SELECT timestamp, operation, resource, parameters
-            FROM audit
            WHERE workspace_id = 'admins'
              AND operation = 'workspace_fairness.uncapped'
            ORDER BY timestamp DESC
@@ -372,7 +372,10 @@ pub async fn migrate(
    }

    crate::live_migrations::custom_migrations(&mut custom_migrator).await?;
-    Ok(None)
+    Ok(Some(crate::live_migrations::spawn_background_migrations(
+        db.clone(),
+        killpill_rx,
+    )))
 }

 pub async fn wait_for_migrations(
@@ -6,8 +6,9 @@
 * LICENSE-AGPL for a copy of the license.
 */

-use sqlx::Postgres;
-use windmill_common::error::Error;
+use sqlx::{PgConnection, Postgres};
+use tokio::task::JoinHandle;
+use windmill_common::{db::DB, error::Error};

 use crate::db::CustomMigrator;
 use sqlx::migrate::Migrate;
@@ -121,3 +122,294 @@ async fn fix_flow_versioning_migration(migrator: &mut CustomMigrator) -> Result<
    }
    Ok(())
 }
+
+// Held for the whole background run so only one server does it at a time.
+const BACKGROUND_MIGRATIONS_LOCK_ID: i64 = 4_931_072_518_336_401;
+
+/// Schema changes too slow to hold server startup for, run by one server at a time after the
+/// sqlx migrations. Each step is recorded in `windmill_migrations` once done; a step interrupted
+/// by a restart or an error starts over on the next start and must resume safely.
+pub fn spawn_background_migrations(
+    db: DB,
+    mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
+) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        tokio::select! {
+            r = run_background_migrations(&db) => {
+                if let Err(err) = r {
+                    tracing::error!("Background migrations stopped, retrying on the next start: {err:#}");
+                }
+            }
+            _ = killpill_rx.recv() => {
+                tracing::info!("Killpill received, stopping background migrations");
+            }
+        }
+    })
+}
+
+async fn run_background_migrations(db: &DB) -> Result<(), Error> {
+    // Detached so the pool's 5min statement_timeout, lifted here for the index builds, never
+    // comes back with this connection; closing it also releases the advisory lock.
+    let mut conn = db.acquire().await?.detach();
+    conn.execute("SET statement_timeout = 0").await?;
+    let locked = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)")
+        .bind(BACKGROUND_MIGRATIONS_LOCK_ID)
+        .fetch_one(&mut conn)
+        .await?;
+    if !locked {
+        return Ok(());
+    }
+
+    for step in [AUDIT_OPERATION_INDEX, RETIRE_LEGACY_AUDIT] {
+        if background_migration_done(&mut conn, step).await? {
+            continue;
+        }
+        // Steps bound the lock waits that would queue audit inserts behind them with
+        // lock_timeout, and resume where they stopped, so a lock timeout is retried here.
+        let mut attempt = 1;
+        loop {
+            let run = match step {
+                AUDIT_OPERATION_INDEX => create_audit_operation_index(&mut conn).await,
+                RETIRE_LEGACY_AUDIT => retire_legacy_audit_table(&mut conn).await,
+                _ => unreachable!("background migration {step} has no step function"),
+            };
+            match run {
+                Ok(()) => break,
+                Err(err) if attempt < 10 && is_lock_timeout(&err) => {
+                    tracing::warn!(
+                        "Background migration {step} timed out on a lock, retrying in 30s: {err:#}"
+                    );
+                    attempt += 1;
+                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
+                }
+                Err(err) => return Err(err),
+            }
+        }
+        mark_background_migration_done(&mut conn, step).await?;
+    }
+    Ok(())
+}
+
+const AUDIT_OPERATION_INDEX: &str = "audit_partitioned_workspace_operation_index";
+const RETIRE_LEGACY_AUDIT: &str = "retire_legacy_audit_table";
+
+// Short on purpose: a statement waiting for its lock is also a wait for every audit insert
+// queued behind it.
+const STEP_LOCK_TIMEOUT: &str = "SET lock_timeout = '1s'";
+
+fn is_lock_timeout(err: &Error) -> bool {
+    matches!(err, Error::SqlErr { error: sqlx::Error::Database(db_err), .. }
+        if db_err.code().as_deref() == Some("55P03"))
+}
+
+async fn background_migration_done(conn: &mut PgConnection, name: &str) -> Result<bool, Error> {
+    Ok(sqlx::query_scalar::<_, bool>(
+        "SELECT EXISTS(SELECT 1 FROM windmill_migrations WHERE name = $1)",
+    )
+    .bind(name)
+    .fetch_one(conn)
+    .await?)
+}
+
+async fn mark_background_migration_done(conn: &mut PgConnection, name: &str) -> Result<(), Error> {
+    sqlx::query("INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING")
+        .bind(name)
+        .execute(conn)
+        .await?;
+    tracing::info!("Background migration {name} done");
+    Ok(())
+}
+
+const AUDIT_OPERATION_INDEX_KEY: &str = r#"(workspace_id, operation, id DESC, "timestamp")"#;
+
+/// A plain `CREATE INDEX` on the partitioned table holds a SHARE lock on every partition until the
+/// whole build ends, blocking the audit insert each job push makes in its own transaction. Each
+/// partition is built CONCURRENTLY instead and attached to a parent created `ON ONLY`, which turns
+/// valid once all partitions are attached. Partitions created later get the index from the parent.
+async fn create_audit_operation_index(conn: &mut PgConnection) -> Result<(), Error> {
+    // Metadata only, but it waits for every open transaction that wrote an audit row, and every
+    // audit insert (so every job push) queues behind it meanwhile. Give up early instead.
+    conn.execute(STEP_LOCK_TIMEOUT).await?;
+    let created = conn
+        .execute(
+            format!(
+                "CREATE INDEX IF NOT EXISTS ix_audit_partitioned_workspace_operation \
+                 ON ONLY audit_partitioned {AUDIT_OPERATION_INDEX_KEY}"
+            )
+            .as_str(),
+        )
+        .await;
+    conn.execute("RESET lock_timeout").await?;
+    created?;
+    let partitions: Vec<String> = sqlx::query_scalar(
+        "SELECT c.relname::text FROM pg_inherits p JOIN pg_class c ON c.oid = p.inhrelid
+         WHERE p.inhparent = 'audit_partitioned'::regclass
+           AND NOT EXISTS (
+               SELECT 1 FROM pg_inherits ip JOIN pg_index i ON i.indexrelid = ip.inhrelid
+               WHERE ip.inhparent = 'ix_audit_partitioned_workspace_operation'::regclass
+                 AND i.indrelid = c.oid)
+         ORDER BY c.relname DESC",
+    )
+    .fetch_all(&mut *conn)
+    .await?;
+    let quote = |name: &str| format!("\"{}\"", name.replace('"', "\"\""));
+    for partition in partitions {
+        let index = quote(&format!(
+            "{partition}_workspace_id_operation_id_timestamp_idx"
+        ));
+        tracing::info!("Building ix_audit_partitioned_workspace_operation on {partition}");
+        // An interrupted CONCURRENTLY build leaves an invalid index under this name.
+        conn.execute(format!("DROP INDEX CONCURRENTLY IF EXISTS {index}").as_str())
+            .await?;
+        conn.execute(
+            format!(
+                "CREATE INDEX CONCURRENTLY {index} ON {} {AUDIT_OPERATION_INDEX_KEY}",
+                quote(&partition)
+            )
+            .as_str(),
+        )
+        .await?;
+        conn.execute(
+            format!(
+                "ALTER INDEX ix_audit_partitioned_workspace_operation ATTACH PARTITION {index}"
+            )
+            .as_str(),
+        )
+        .await?;
+    }
+    Ok(())
+}
+
+const AUDIT_COLUMNS: &str =
+    "workspace_id, id, timestamp, username, operation, action_kind, resource, parameters, email, span";
+
+/// Moves the last 30 days of the pre-partitioning `audit` table into daily partitions and drops
+/// it with anything older. An empty `audit` view takes its place for servers still running an
+/// older version, which read it in a `UNION ALL` with `audit_partitioned`.
+async fn retire_legacy_audit_table(conn: &mut PgConnection) -> Result<(), Error> {
+    let is_table = sqlx::query_scalar::<_, bool>(
+        "SELECT EXISTS(SELECT 1 FROM pg_class WHERE oid = to_regclass('audit') AND relkind = 'r')",
+    )
+    .fetch_one(&mut *conn)
+    .await?;
+    if !is_table {
+        return Ok(());
+    }
+    // Creating a partition and re-owning the sequence queue every audit insert (so every job
+    // push) behind them while they wait for their own lock, e.g. on a long-running reader, and
+    // the DROP queues older servers' reads the same way. Give up early and retry instead.
+    conn.execute(STEP_LOCK_TIMEOUT).await?;
+
+    let days: Vec<chrono::NaiveDate> = sqlx::query_scalar(
+        "SELECT DISTINCT timestamp::date FROM audit
+         WHERE timestamp > now() - interval '30 days' ORDER BY 1",
+    )
+    .fetch_all(&mut *conn)
+    .await?;
+    for day in days {
+        let next = day + chrono::Duration::days(1);
+        // Created outside any transaction: a partition created inside one keeps a lock on
+        // audit_partitioned that blocks every job push's audit insert until it commits.
+        conn.execute(
+            format!(
+                "CREATE TABLE IF NOT EXISTS \"audit_{}\" PARTITION OF audit_partitioned \
+                 FOR VALUES FROM ('{day}') TO ('{next}')",
+                day.format("%Y%m%d")
+            )
+            .as_str(),
+        )
+        .await?;
+        // One statement per day, so a row is always in exactly one of the two tables.
+        let moved = sqlx::query(&format!(
+            "WITH moved AS (
+                 DELETE FROM audit WHERE timestamp >= $1::date AND timestamp < $1::date + 1
+                 RETURNING {AUDIT_COLUMNS}
+             )
+             INSERT INTO audit_partitioned ({AUDIT_COLUMNS}) SELECT {AUDIT_COLUMNS} FROM moved"
+        ))
+        .bind(day)
+        .execute(&mut *conn)
+        .await?;
+        tracing::info!(
+            "Moved {} legacy audit rows of {day} into audit_partitioned",
+            moved.rows_affected()
+        );
+    }
+
+    // audit_partitioned draws its ids from this sequence, which dropping its owner would drop.
+    // Not in the transaction below: its lock stops every insert's nextval until commit, and the
+    // DROP can wait there on readers of the old table.
+    conn.execute("ALTER SEQUENCE audit_id_seq OWNED BY audit_partitioned.id")
+        .await?;
+    let mut tx = conn.begin().await?;
+    // Servers older than audit partitioning still write to the table during a rolling upgrade:
+    // rows they added after their day was moved above are swept up under the lock.
+    tx.execute("LOCK TABLE audit IN ACCESS EXCLUSIVE MODE")
+        .await?;
+    tx.execute(
+        format!(
+            "WITH moved AS (
+                 DELETE FROM audit WHERE timestamp > now() - interval '30 days'
+                 RETURNING {AUDIT_COLUMNS}
+             )
+             INSERT INTO audit_partitioned ({AUDIT_COLUMNS}) SELECT {AUDIT_COLUMNS} FROM moved"
+        )
+        .as_str(),
+    )
+    .await?;
+    tx.execute("DROP TABLE audit").await?;
+    tx.execute(
+        format!("CREATE VIEW audit AS SELECT {AUDIT_COLUMNS} FROM audit_partitioned WHERE false")
+            .as_str(),
+    )
+    .await?;
+    tx.execute("GRANT ALL ON audit TO windmill_user, windmill_admin")
+        .await?;
+    tx.commit().await?;
+    conn.execute("RESET lock_timeout").await?;
+    tracing::info!("Retired the legacy audit table");
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[sqlx::test(migrations = "../migrations")]
+    async fn retire_legacy_audit_table_keeps_the_last_30_days(db: DB) -> anyhow::Result<()> {
+        sqlx::query(
+            "INSERT INTO audit (workspace_id, id, timestamp, username, operation, action_kind)
+             VALUES ('w', -2, now() - interval '3 days', 'u', 'recent', 'execute'),
+                    ('w', -1, now() - interval '60 days', 'u', 'old', 'execute')",
+        )
+        .execute(&db)
+        .await?;
+
+        retire_legacy_audit_table(&mut *db.acquire().await?).await?;
+
+        let moved: Vec<(i64, String)> = sqlx::query_as(
+            "SELECT id, operation::text FROM audit_partitioned WHERE workspace_id = 'w'",
+        )
+        .fetch_all(&db)
+        .await?;
+        assert_eq!(moved, vec![(-2, "recent".to_string())]);
+
+        // The id sequence outlives the table that owned it.
+        sqlx::query(
+            "INSERT INTO audit_partitioned (workspace_id, username, operation, action_kind)
+             VALUES ('w', 'u', 'after', 'execute')",
+        )
+        .execute(&db)
+        .await?;
+
+        // Servers on an older version still read `audit` in a UNION with audit_partitioned.
+        let seen: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM (SELECT * FROM audit_partitioned UNION ALL SELECT * FROM audit) a
+             WHERE workspace_id = 'w'",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert_eq!(seen, 2);
+        Ok(())
+    }
+}
@@ -145,6 +145,9 @@
 		}
 	}
 	async function getMenuElements(): Promise<HTMLElement[]> {
+		// Runs on every pointerdown anywhere, for every mounted dropdown. Skip the
+		// whole-document query when the outside handler below cannot act anyway.
+		if (!usePointerDownOutside || !open) return []
 		return Array.from(document.querySelectorAll('[data-menu]')) as HTMLElement[]
 	}
 </script>
@@ -378,6 +378,16 @@
 	)
 	$effect(() => {
 		chatHost.setComposerStaged(composerKey, editingMessageIndex, stagedBytes)
+		chatHost.setComposerHasDraft(
+			composerKey,
+			!draft.isEmpty ||
+				(chatHost.mode === AIMode.GLOBAL && selectedContext.length > 0) ||
+				pendingFiles > 0 ||
+				pendingFileBytes > 0 ||
+				pendingImages > 0 ||
+				pendingBlobs > 0 ||
+				ingestionHolds > 0
+		)
 	})
 	$effect(() => () => chatHost.clearComposerStaged(composerKey))

@@ -889,7 +899,7 @@
 			contextTextareaComponent?.clearForSend()
 			return
 		}
-		if (chatHost.loading) {
+		if (chatHost.loading || chatHost.sendInFlight || chatHost.sendPending) {
 			// Queue the message instead of silently discarding it — it is
 			// auto-sent when the streaming turn completes successfully.
 			// Editing-while-loading keeps the old discard behavior. Paste
@@ -2224,7 +2224,7 @@ export class AIChatManager implements ChatViewHost {
 	 * Gated on `sendInFlight` as well as `loading`: `loading` only rises after a
 	 * send's attachment upkeep, so between the two a click would slip past. */
 	sendOrQueue(text: string) {
-		if (this.loading || this.sendInFlight) {
+		if (this.loading || this.sendInFlight || this.sendPending) {
 			this.queueMessage(text)
 			return
 		}
@@ -2819,6 +2819,23 @@ export class AIChatManager implements ChatViewHost {

 	clearComposerStaged(key: string) {
 		this.#composerStaged.delete(key)
+		this.#composersWithDraft.delete(key)
+	}
+
+	// Composers whose draft is non-empty. The draft itself is component-local, so
+	// this is the only way to know an unmount would lose unsent input.
+	#composersWithDraft = new SvelteSet<string>()
+
+	setComposerHasDraft(key: string, hasDraft: boolean) {
+		if (hasDraft) this.#composersWithDraft.add(key)
+		else this.#composersWithDraft.delete(key)
+	}
+
+	/** Unsent input an unmount of this chat would lose: a composer draft, or a
+	 *  queued message (whose text may be empty when it carries only attachments
+	 *  or context). */
+	get hasUnsentInput(): boolean {
+		return this.#composersWithDraft.size > 0 || this.#hasQueuedMessage()
 	}

 	/** Release the outgoing-files reservation identified by `key` (a per-send token).
@@ -3174,7 +3191,39 @@ export class AIChatManager implements ChatViewHost {
 		return this.#sendsInFlight > 0
 	}

+	// Resolves once this chat's stored transcript has been restored. A session
+	// runtime hands its manager out before that finishes, and `loadPastChat`
+	// refuses to swap a transcript under a send, so a turn started first lands in
+	// a chat of its own and is lost when the session's real one loads.
+	//
+	// Cleared once awaited, and only awaited when set: every later send — and
+	// every send of the chats that have no gate — reaches the `sendInFlight`
+	// counter below synchronously, which is what makes a second Enter queue
+	// behind the first instead of starting a turn of its own.
+	#ready: Promise<unknown> | undefined
+	setReadyGate(ready: Promise<unknown> | undefined) {
+		this.#ready = ready
+	}
+
+	#sendsAwaitingReady = $state(0)
+	/** A send parked on the gate above. It has not reached `sendInFlight` yet —
+	 * and must not, since `loadPastChat` reads that flag and would then skip the
+	 * restore this send is waiting for — so every guard that decides between
+	 * sending and queueing tests this as well. */
+	get sendPending(): boolean {
+		return this.#sendsAwaitingReady > 0
+	}
+
 	sendRequest = async (options: Parameters<typeof this.sendRequestImpl>[0] = {}) => {
+		if (this.#ready) {
+			this.#sendsAwaitingReady++
+			try {
+				await this.#ready
+			} finally {
+				this.#sendsAwaitingReady--
+				this.#ready = undefined
+			}
+		}
 		// A turn with nowhere to render still streams, spends tokens and applies
 		// tool calls — entirely off-screen. Refuse instead. `sendInlineRequest` is
 		// exempt: the ⌘K widget renders its own composer inside Monaco.
@@ -514,6 +514,25 @@ describe('AIChatManager.sendOrQueue', () => {
 		releaseUpkeep?.()
 		await sending
 	})
+
+	// A session's chat is restored after its manager is handed out, so the first
+	// send waits on that restore — before `sendInFlight`, which the restore reads
+	// to decide whether to bail.
+	it('queues while a send waits on the chat restore gate', async () => {
+		const manager = new AIChatManager()
+		let openGate: (() => void) | undefined
+		manager.setReadyGate(new Promise<void>((resolve) => (openGate = resolve)))
+		manager.instructions = 'first turn'
+		const sending = manager.sendRequest()
+		await vi.waitFor(() => expect(manager.sendPending).toBe(true))
+		expect(manager.sendInFlight).toBe(false)
+
+		manager.sendOrQueue('fix the failing run')
+		expect(manager.queuedMessage).toBe('fix the failing run')
+
+		openGate?.()
+		await sending
+	})
 })

 describe('AIChatManager request errors', () => {
@@ -394,7 +394,10 @@ export default class HistoryManager {
 			.sort((a, b) => b.lastModified - a.lastModified)
 	)

-	async init() {
+	/** With `session`, loads only that session's chats plus `session.chatId`, which
+	 *  may predate session tagging. Loading every chat instead costs one full copy
+	 *  of the user's history per session runtime. */
+	async init(session?: { id: string; chatId?: string }) {
 		// (Re)initializing adopts a new identity's history: drop the previous
 		// identity's cached blob ids with it.
 		this.imageIdByUrl.clear()
@@ -404,7 +407,16 @@ export default class HistoryManager {
 		const db = await this.dbh.whenReady()
 		if (!db) return
 		try {
-			const chats = await db.getAll('chats')
+			let chats: StoredChat[]
+			if (session) {
+				chats = await db.getAllFromIndex('chats', 'by-session', session.id)
+				if (session.chatId && !chats.some((c) => c.id === session.chatId)) {
+					const current = await db.get('chats', session.chatId)
+					if (current) chats.push(current)
+				}
+			} else {
+				chats = await db.getAll('chats')
+			}
 			this.savedChats = chats.reduce(
 				(acc, chat) => {
 					acc[chat.id] = chat
@@ -816,3 +816,36 @@ describe('HistoryManager.reloadChat', () => {
 		expect(await hm.reloadChat(hm.getCurrentChatId())).toBe('unavailable')
 	})
 })
+
+describe('HistoryManager.init for a session', () => {
+	it("loads the session's chats and its current chat, not other sessions'", async () => {
+		const setup = new HistoryManager()
+		await setup.init()
+		setup.close()
+		const chat = (id: string, sessionId?: string) => ({
+			id,
+			actualMessages: [],
+			displayMessages: [],
+			title: id,
+			lastModified: 1,
+			...(sessionId ? { sessionId } : {})
+		})
+		const db = await openDB('copilot-chat-history::admin@test')
+		await db.put('chats' as never, chat('mine', 's1') as never)
+		await db.put('chats' as never, chat('other-session', 's2') as never)
+		await db.put('chats' as never, chat('untagged-current') as never)
+		await db.put('chats' as never, chat('untagged-other') as never)
+		db.close()
+
+		const hm = new HistoryManager()
+		await hm.init({ id: 's1', chatId: 'untagged-current' })
+
+		// The current chat can predate session tagging, so it is read by key.
+		expect(
+			hm
+				.getAllSavedChats()
+				.map((c) => c.id)
+				.sort()
+		).toEqual(['mine', 'untagged-current'])
+	})
+})
@@ -52,6 +52,8 @@ export interface ChatViewHost {
 	 * pins away from the navigated one. */
 	readonly operatingWorkspace: string | undefined
 	loading: boolean
+	/** A send waiting for the chat's transcript to be restored; queue behind it. */
+	sendPending: boolean
 	/** A turn this tab can neither follow nor stop, held by another tab on the same chat. */
 	readonly runHeldElsewhere: boolean
 	loadingLabel: string | undefined
@@ -85,6 +87,8 @@ export interface ChatViewHost {
 	) => void
 	dequeueMessage: () => void
 	setComposerStaged: (key: string, editingIndex: number | null, bytes: number) => void
+	/** Whether this composer holds anything unsent (text, pastes, attachments, reads in flight). */
+	setComposerHasDraft: (key: string, hasDraft: boolean) => void
 	clearComposerStaged: (key: string) => void
 	attachmentBytesExcluding: (selfKey: string) => number

@@ -398,6 +398,8 @@ export class FlowChatViewHost implements ChatViewHost {
 	get loading(): boolean {
 		return isBusy(this.#state.status)
 	}
+	// No transcript restore to wait for: a flow chat renders from its rows.
+	sendPending = false
 	runHeldElsewhere = false
 	loadingLabel = undefined
 	compacting = false
@@ -580,6 +582,7 @@ export class FlowChatViewHost implements ChatViewHost {
 		return taken
 	}
 	setComposerStaged = () => {}
+	setComposerHasDraft = () => {}
 	clearComposerStaged = () => {}
 	attachmentBytesExcluding = () => 0

@@ -176,6 +176,9 @@
 	}

 	async function getMenuElements(): Promise<HTMLElement[]> {
+		// Runs on every pointerdown anywhere, for every mounted menu. Skip the
+		// whole-document query when the outside handler below cannot act anyway.
+		if (!usePointerDownOutside || !open) return []
 		// Tooltip content counts as menu territory for the same reason as the
 		// onOutsideClick veto above.
 		return Array.from(
@@ -194,6 +194,9 @@
 	}

 	async function getMenuElements(): Promise<HTMLElement[]> {
+		// Runs on every pointerdown anywhere, for every mounted popover. Skip the
+		// whole-document query when the outside handler below cannot act anyway.
+		if (!usePointerDownOutside || (!isOpen && !fullScreen)) return []
 		const selector = excludeSelectors ? `[data-popover], ${excludeSelectors}` : '[data-popover]'
 		return Array.from(document.querySelectorAll(selector)) as HTMLElement[]
 	}
@@ -172,9 +172,10 @@

 	// The inline mini-composer over a selected element sends a chat turn; the
 	// element is already an app_dom_selector chip, so it rides along as context.
-	// Mirror the composer: while a turn is streaming, queue it (a second concurrent
-	// sendRequest would race the shared abortController / streaming buffers) — it
-	// auto-sends when the current turn completes.
+	// Mirror the composer: while another send holds the chat — streaming, in
+	// preflight, or waiting for the transcript to be restored — queue it (a second
+	// concurrent sendRequest would race the shared abortController / streaming
+	// buffers) — it auto-sends when the current turn completes.
 	function onInlinePrompt(selector: string, prompt: string) {
 		// Snapshot the selection synchronously at submit time so a re-selection
 		// during the async send preflight (immediate path) or before the queue
@@ -190,7 +191,7 @@
 			.filter(
 				(c) => c.type !== 'app_dom_selector' || (c.selector === selector && c.appPath === path)
 			)
-		if (runtime.manager.loading) {
+		if (runtime.manager.loading || runtime.manager.sendInFlight || runtime.manager.sendPending) {
 			runtime.manager.queueMessage(prompt, [], snapshot)
 		} else {
 			void runtime.manager.sendRequest({ instructions: prompt, contextOverride: snapshot })
@@ -44,8 +44,9 @@
 	import { unreadCountFor } from './sessionUnread.svelte'
 	import Toggle from '$lib/components/Toggle.svelte'
 	import {
-		getOrCreateRuntime,
+		ensureSessionChatPeek,
 		getRuntime,
+		getSessionChatPeek,
 		getSessionChatStatus,
 		removeSession,
 		resetSessionPreviewTabs
@@ -87,7 +88,7 @@
 	// displayMessages array vs. the localStorage-backed lastSeen map;
 	// both are reactive so the badge updates without polling.
 	function unreadFor(session: Session): number {
-		return unreadCountFor(session.id, getRuntime(session.id))
+		return unreadCountFor(session.id, getRuntime(session.id), getSessionChatPeek(session.id))
 	}

 	// Whether the composer for a session holds non-whitespace text. We
@@ -378,14 +379,12 @@
 		}
 	})

-	// Eagerly create a runtime per VISIBLE session so the status dot reflects
-	// the persisted chat (last message, pending confirmation, etc.) without
-	// requiring the user to open the session first. Sessions outside the
-	// current workspace scope are left cold to avoid opening IDB connections
-	// for unrelated work.
+	// Read each VISIBLE session's stored chat so the status dot and unread count
+	// reflect it without the user opening the session. A runtime per listed
+	// session would hold every listed chat in memory for as long as the page lives.
 	$effect(() => {
 		for (const session of visibleSessions) {
-			getOrCreateRuntime(session)
+			if (!getRuntime(session.id)) void ensureSessionChatPeek(session)
 		}
 	})

@@ -768,7 +767,9 @@
 									{/if}
 									{#each group.sessions as session (session.id)}
 										{@const runtime = getRuntime(session.id)}
-										{@const status = runtime ? getSessionChatStatus(runtime) : 'idle'}
+										{@const status = runtime
+											? getSessionChatStatus(runtime)
+											: (getSessionChatPeek(session.id)?.status ?? 'idle')}
 										{@const isSelected =
 											sessionActive && session.id === sessionState.currentSessionId}
 										{@const unread = unreadFor(session)}
@@ -986,7 +987,9 @@
 				{/snippet}
 				{#snippet sessionRow(session, indented, treeDepth)}
 					{@const runtime = getRuntime(session.id)}
-					{@const status = runtime ? getSessionChatStatus(runtime) : 'idle'}
+					{@const status = runtime
+						? getSessionChatStatus(runtime)
+						: (getSessionChatPeek(session.id)?.status ?? 'idle')}
 					{@const isSelected = sessionActive && session.id === sessionState.currentSessionId}
 					{@const isEditing = editingId === session.id}
 					{@const unread = unreadFor(session)}
@@ -99,7 +99,13 @@ import type {
 	RawAppDomResult
 } from '$lib/components/raw_apps/rawAppDom'
 import { getNonStreamingMetadataCompletion } from '$lib/components/copilot/lib'
-import { pendingUserAction, type DisplayMessage } from '$lib/components/copilot/chat/shared'
+import {
+	pendingUserAction,
+	type ChatJob,
+	type DisplayMessage
+} from '$lib/components/copilot/chat/shared'
+import { readStoredChat } from '$lib/components/copilot/chat/HistoryManager.svelte'
+import { getCurrentUserEmail } from '$lib/userScopedStorage'
 import type { ChatCompletionMessageParam } from 'openai/resources/index.mjs'

 // Per-kind load state for a session's editor target. Pure state container the
@@ -985,7 +991,7 @@ function createRuntime(session: Session): SessionRuntime {

 async function initRuntime(runtime: SessionRuntime, session: Session) {
 	const { manager } = runtime
-	await manager.historyManager.init()
+	await manager.historyManager.init(session)
 	manager.historyManager.setSessionId(session.id)
 	// Restore linked files persisted for this session (live handles re-grant on send;
 	// snapshots restore directly). Non-transient sessions persist immediately.
@@ -1000,7 +1006,9 @@ async function initRuntime(runtime: SessionRuntime, session: Session) {
 	if (session.chatId) {
 		manager.historyManager.setCurrentChatId(session.chatId)
 		await manager.historyManager.tagChatWithSession(session.chatId, session.id)
-		await manager.loadPastChat(session.chatId)
+		// preserveQueue: restoring a session's own chat is not a conversation switch.
+		// A message queued while the restore ran is unsent user input waiting for it.
+		await manager.loadPastChat(session.chatId, { preserveQueue: true })
 		// loadPastChat only seeds the mask when the chat exists in history; a chatId
 		// pointing at a chat not yet persisted (no turn saved) would leave it
 		// undefined, and the Edits surface would then show every workspace draft.
@@ -1020,7 +1028,11 @@ export function getOrCreateRuntime(session: Session): SessionRuntime {
 	if (!runtime) {
 		runtime = createRuntime(session)
 		runtimes.set(session.id, runtime)
-		initRuntime(runtime, session).catch((e) => console.error('Failed to init session runtime', e))
+		// The gate opens even when the restore fails: a session that cannot read its
+		// history is still usable, and a send that waits forever is not.
+		runtime.manager.setReadyGate(
+			initRuntime(runtime, session).catch((e) => console.error('Failed to init session runtime', e))
+		)
 	}
 	return runtime
 }
@@ -1031,6 +1043,9 @@ export function disposeRuntime(sessionId: string) {
 	runtime.manager.cancel('runtime disposed')
 	runtime.manager.historyManager.close()
 	runtimes.delete(sessionId)
+	// The sidebar reads the stored chat again rather than trust a read that
+	// predates everything this runtime saw.
+	peeks.delete(sessionId)
 }

 export function listRuntimes(): SessionRuntime[] {
@@ -1064,7 +1079,13 @@ onRemoteTurnEnd((sessionId, chatId) => {

 async function applyRemoteTurnEnd(sessionId: string, chatId: string): Promise<void> {
 	const runtime = runtimes.get(sessionId)
-	if (!runtime) return
+	if (!runtime) {
+		// The driving tab may have rotated to a new chat; follow it, or the next
+		// read and a later runtime would open the old one.
+		setSessionChatId(sessionId, chatId)
+		refreshSessionChatPeek(sessionId)
+		return
+	}
 	const m = runtime.manager
 	// Two transient states get a short retry rather than a skip, because the
 	// composer unlocks when this promise settles and a skip would unlock it on
@@ -1405,3 +1426,87 @@ export function getSessionChatStatus(runtime: SessionRuntime): SessionChatStatus
 	if (last && (last.role === 'assistant' || last.role === 'tool')) return 'awaiting-user'
 	return 'idle'
 }
+
+// ---------------------------------------------------------------------------
+// Sessions without a runtime
+// ---------------------------------------------------------------------------
+
+// A runtime costs a chat manager and its loaded transcript, and lives until its
+// session is deleted, so the sidebar does not create one per listed session. It
+// reads the session's one stored chat instead, once. Without a runtime nothing
+// runs in this tab, so only two facts of that chat can matter: its length (for
+// the unread count) and whether its last message failed. Every other status is
+// a live one.
+export interface SessionChatPeek {
+	status: 'error' | 'idle'
+	messageCount: number
+}
+
+const peeks = new SvelteMap<string, SessionChatPeek>()
+const peeksInFlight = new Set<string>()
+const peeksPendingRefresh = new Set<string>()
+
+// Not gated on `detached`: a chat saved mid-wait stores its job undetached, and
+// loadPastChat resumes polling every unfinished job either way.
+function isLiveJob(j: ChatJob): boolean {
+	return (
+		j.status === 'queued' ||
+		j.status === 'running' ||
+		j.status === 'suspended' ||
+		j.status === 'scheduled'
+	)
+}
+
+function sessionById(sessionId: string): Session | undefined {
+	return sessionState.sessions.find((s) => s.id === sessionId)
+}
+
+function refreshSessionChatPeek(sessionId: string): void {
+	peeks.delete(sessionId)
+	if (peeksInFlight.has(sessionId)) {
+		peeksPendingRefresh.add(sessionId)
+		return
+	}
+	const session = sessionById(sessionId)
+	if (session) void ensureSessionChatPeek(session)
+}
+
+export function getSessionChatPeek(sessionId: string): SessionChatPeek | undefined {
+	return peeks.get(sessionId)
+}
+
+/** Read the stored chat of a session that has no runtime. A chat with a detached
+ *  job still running gets its runtime instead: only a runtime polls the job and
+ *  resumes the conversation when it finishes. */
+export async function ensureSessionChatPeek(session: Session): Promise<void> {
+	const id = session.id
+	const chatId = session.chatId
+	if (runtimes.has(id) || peeks.has(id) || peeksInFlight.has(id) || !chatId) return
+	const email = getCurrentUserEmail()
+	if (!email) return
+	peeksInFlight.add(id)
+	try {
+		const chat = await readStoredChat(chatId, email)
+		if (sessionById(id)?.chatId !== chatId) {
+			refreshSessionChatPeek(id)
+			return
+		}
+		if (!chat || runtimes.has(id)) return
+		if (chat.backgroundJobs?.some(isLiveJob)) {
+			getOrCreateRuntime(session)
+			return
+		}
+		const last = chat.displayMessages[chat.displayMessages.length - 1]
+		peeks.set(id, {
+			status: last?.role === 'user' && last.error ? 'error' : 'idle',
+			messageCount: chat.displayMessages.length
+		})
+	} catch (e) {
+		console.error('Failed to read session chat', e)
+	} finally {
+		peeksInFlight.delete(id)
+		if (peeksPendingRefresh.delete(id)) {
+			refreshSessionChatPeek(id)
+		}
+	}
+}
@@ -1,4 +1,4 @@
-import type { SessionRuntime } from './sessionRuntime.svelte'
+import type { SessionChatPeek, SessionRuntime } from './sessionRuntime.svelte'
 import { sessionState, putSession } from './sessionState.svelte'

 // Per-session "last seen" watermark — the displayMessages count the last time
@@ -20,11 +20,16 @@ export function markSessionSeen(sessionId: string, count: number) {
 	void putSession(s)
 }

-// Number of unread messages for a session. Undefined / unloaded runtime
-// returns 0 — until messages are hydrated we don't know what's new.
-export function unreadCountFor(sessionId: string, runtime: SessionRuntime | undefined): number {
-	if (!runtime) return 0
+// Number of unread messages for a session, from its runtime when it has one, else
+// from the sidebar's read of its stored chat. Neither loaded returns 0 — until
+// messages are read we don't know what's new.
+export function unreadCountFor(
+	sessionId: string,
+	runtime: SessionRuntime | undefined,
+	peek: SessionChatPeek | undefined
+): number {
+	const total = runtime ? runtime.manager.displayMessages.length : peek?.messageCount
+	if (total === undefined) return 0
 	const seen = sessionState.sessions.find((x) => x.id === sessionId)?.lastSeenCount ?? 0
-	const total = runtime.manager.displayMessages.length
 	return Math.max(0, total - seen)
 }
@@ -220,19 +220,39 @@
 			// deep links the same way they react to picker clicks.
 			selectSession(session.id)
 			getOrCreateRuntime(session)
+			mountChat(session.id)
 		})
 	})

-	// Warm = sessions with a live runtime. The picker eagerly creates runtimes
-	// for its visible sessions, so this tracks whatever it shows. Keeping warm
-	// chats mounted (stacked, visibility-toggled) preserves their scroll/draft
-	// state across switches.
+	// Warm = sessions with a live runtime: the ones visited this page load plus
+	// any a chat tool or a running job started.
 	const warmSessions = $derived(
 		listRuntimes()
 			.map((r) => sessionState.sessions.find((s) => s.id === r.sessionId))
 			.filter((s): s is NonNullable<typeof s> => s != null)
 	)

+	// Chat columns stay mounted (stacked, visibility-toggled) so switching back
+	// keeps scroll position, MRU-capped like preview tabs: every mounted column
+	// is a full transcript in the DOM. Unmounting drops only the column — the
+	// runtime keeps the chat — so a column is kept while its composer holds
+	// unsent input (component-local) or its turn is running.
+	const MAX_MOUNTED_CHATS = 5
+	const mountedChatIds = new SvelteSet<string>()
+	function keepsChatMounted(id: string): boolean {
+		const m = getRuntime(id)?.manager
+		return !!m && (m.loading || m.sendInFlight || m.sendPending || m.hasUnsentInput)
+	}
+	function mountChat(id: string) {
+		mountedChatIds.delete(id)
+		mountedChatIds.add(id)
+		for (const oldest of [...mountedChatIds]) {
+			if (mountedChatIds.size <= MAX_MOUNTED_CHATS) break
+			if (oldest !== id && !keepsChatMounted(oldest)) mountedChatIds.delete(oldest)
+		}
+	}
+	const mountedChatSessions = $derived(warmSessions.filter((s) => mountedChatIds.has(s.id)))
+
 	// Mark the active session "seen" up to its current message count: arrive →
 	// clear unread; AI streams a new message while we're here → clear again. The
 	// effect depends only on the length, not the array contents, so token-by-token
@@ -915,11 +935,11 @@
 					class="flex-1 min-h-0 session-splitter {previewCollapsed ? 'splitter-off' : ''}"
 				>
 					{#if !fullscreen}
-						<!-- Chat column. Warm sessions stay mounted (stacked, visibility-toggled)
-					     so switching between them preserves chat scroll/draft state. -->
+						<!-- Chat column. Recently visited sessions stay mounted (stacked,
+					     visibility-toggled) — see mountChat. -->
 						<Pane bind:size={chatPaneSize} minSize={25} class="flex flex-col min-h-0">
 							<div class="relative flex-1 min-h-0">
-								{#each warmSessions as s (s.id)}
+								{#each mountedChatSessions as s (s.id)}
 									<div
 										class="absolute inset-0 flex flex-col {s.id === activeSession?.id
 											? 'z-10 opacity-100 pointer-events-auto'