diff --git a/backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json b/backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json deleted file mode 100644 index 23cf739f90..0000000000 --- a/backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json b/backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json deleted file mode 100644 index 17fbb0d5bb..0000000000 --- a/backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json b/backend/.sqlx/query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json similarity index 54% rename from backend/.sqlx/query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json rename to backend/.sqlx/query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json index d05c5b11af..edc0efd9ca 100644 --- a/backend/.sqlx/query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json +++ b/backend/.sqlx/query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json @@ -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" } diff --git a/backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json b/backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json deleted file mode 100644 index 70872a1328..0000000000 --- a/backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json b/backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json new file mode 100644 index 0000000000..02d889c38b --- /dev/null +++ b/backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json @@ -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" +} diff --git a/backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json b/backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json new file mode 100644 index 0000000000..7acdcf8801 --- /dev/null +++ b/backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json @@ -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" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index a59318e7e7..2a5fea1e7e 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -a4783137d4ba1121cb5c09d4f5fe3b628071ad78 +466fc69233459ea477e062cd3e03e4442f1ea5fd diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 750bc40560..7e4dbb820a 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -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 diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 47f9faca68..ea756fc23c 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -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) diff --git a/backend/windmill-api-users/src/users.rs b/backend/windmill-api-users/src/users.rs index da81c7e228..e46afede8b 100644 --- a/backend/windmill-api-users/src/users.rs +++ b/backend/windmill-api-users/src/users.rs @@ -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 diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index 07bef364c8..0102f1b033 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -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 diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index e8a6858db7..9619e96ebc 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -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( diff --git a/backend/windmill-api/src/live_migrations.rs b/backend/windmill-api/src/live_migrations.rs index 4c6f5e245c..334ee340a8 100644 --- a/backend/windmill-api/src/live_migrations.rs +++ b/backend/windmill-api/src/live_migrations.rs @@ -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 { + 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 = 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 = 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(()) + } +} diff --git a/frontend/src/lib/components/DropdownV2.svelte b/frontend/src/lib/components/DropdownV2.svelte index 6738e7ace8..5061636835 100644 --- a/frontend/src/lib/components/DropdownV2.svelte +++ b/frontend/src/lib/components/DropdownV2.svelte @@ -145,6 +145,9 @@ } } async function getMenuElements(): Promise { + // 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[] } diff --git a/frontend/src/lib/components/copilot/chat/AIChatInput.svelte b/frontend/src/lib/components/copilot/chat/AIChatInput.svelte index d3cc2c8bd9..6f15937cf2 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatInput.svelte +++ b/frontend/src/lib/components/copilot/chat/AIChatInput.svelte @@ -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 diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index 97854a166e..922f64ad93 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -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() + + 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 | undefined + setReadyGate(ready: Promise | 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[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. diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts index 35ac65253c..ef15df4e8d 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.test.ts @@ -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((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', () => { diff --git a/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts b/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts index d956b4c0c0..7843169dd9 100644 --- a/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/HistoryManager.svelte.ts @@ -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 diff --git a/frontend/src/lib/components/copilot/chat/HistoryManager.test.ts b/frontend/src/lib/components/copilot/chat/HistoryManager.test.ts index cdbf6f2020..3ee9edad39 100644 --- a/frontend/src/lib/components/copilot/chat/HistoryManager.test.ts +++ b/frontend/src/lib/components/copilot/chat/HistoryManager.test.ts @@ -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']) + }) +}) diff --git a/frontend/src/lib/components/copilot/chat/chatViewHost.ts b/frontend/src/lib/components/copilot/chat/chatViewHost.ts index 4af23337e2..5c4dceb7f1 100644 --- a/frontend/src/lib/components/copilot/chat/chatViewHost.ts +++ b/frontend/src/lib/components/copilot/chat/chatViewHost.ts @@ -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 diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 37bb868982..6ef18b7212 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts @@ -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 diff --git a/frontend/src/lib/components/meltComponents/Menu.svelte b/frontend/src/lib/components/meltComponents/Menu.svelte index 4a821ceacc..ab6de64123 100644 --- a/frontend/src/lib/components/meltComponents/Menu.svelte +++ b/frontend/src/lib/components/meltComponents/Menu.svelte @@ -176,6 +176,9 @@ } async function getMenuElements(): Promise { + // 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( diff --git a/frontend/src/lib/components/meltComponents/Popover.svelte b/frontend/src/lib/components/meltComponents/Popover.svelte index 6a6b7d681b..dd69d16917 100644 --- a/frontend/src/lib/components/meltComponents/Popover.svelte +++ b/frontend/src/lib/components/meltComponents/Popover.svelte @@ -194,6 +194,9 @@ } async function getMenuElements(): Promise { + // 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[] } diff --git a/frontend/src/lib/components/sessions/RawAppEditorView.svelte b/frontend/src/lib/components/sessions/RawAppEditorView.svelte index 4e2fb2bbd4..e95bc142e5 100644 --- a/frontend/src/lib/components/sessions/RawAppEditorView.svelte +++ b/frontend/src/lib/components/sessions/RawAppEditorView.svelte @@ -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 }) diff --git a/frontend/src/lib/components/sessions/SessionPicker.svelte b/frontend/src/lib/components/sessions/SessionPicker.svelte index e5b62da0eb..81ce3ff9af 100644 --- a/frontend/src/lib/components/sessions/SessionPicker.svelte +++ b/frontend/src/lib/components/sessions/SessionPicker.svelte @@ -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)} diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts index 640b484478..fbb84e1431 100644 --- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts @@ -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 { 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() +const peeksInFlight = new Set() +const peeksPendingRefresh = new Set() + +// 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 { + 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) + } + } +} diff --git a/frontend/src/lib/components/sessions/sessionUnread.svelte.ts b/frontend/src/lib/components/sessions/sessionUnread.svelte.ts index 281c6571bc..26ce3f205b 100644 --- a/frontend/src/lib/components/sessions/sessionUnread.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionUnread.svelte.ts @@ -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) } diff --git a/frontend/src/routes/(root)/(logged)/sessions/+page.svelte b/frontend/src/routes/(root)/(logged)/sessions/+page.svelte index a3712f3f24..e773e7603f 100644 --- a/frontend/src/routes/(root)/(logged)/sessions/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/sessions/+page.svelte @@ -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 => 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() + 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} - +
- {#each warmSessions as s (s.id)} + {#each mountedChatSessions as s (s.id)}