diff --git a/benchmarks/aft-search/slice-descriptors/land-rich-ui-s03-presets.json b/benchmarks/aft-search/slice-descriptors/land-rich-ui-s03-presets.json new file mode 100644 index 0000000000..ebd574711e --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-rich-ui-s03-presets.json @@ -0,0 +1,8 @@ +{ + "targeted_mechanism": "none", + "kind": "harness", + "slice_class": "non_ranking", + "fixtures": [ + "harness-goldens" + ] +} diff --git a/crates/aft/src/db/compression_events.rs b/crates/aft/src/db/compression_events.rs index 03ac3c83de..a2f38b61fd 100644 --- a/crates/aft/src/db/compression_events.rs +++ b/crates/aft/src/db/compression_events.rs @@ -613,6 +613,7 @@ fn apply_prepared_retention_tick_timed( let task_delete_micros = task_delete_started.elapsed().as_micros(); let event_prune_started = Instant::now(); let compression_events_removed = prune_compression_events_in_transaction(&tx, now_ms)?; + crate::db::rich_ui::prune_presets(&tx, now_ms)?; let event_prune_micros = event_prune_started.elapsed().as_micros(); let commit_started = Instant::now(); tx.commit()?; diff --git a/crates/aft/src/db/mod.rs b/crates/aft/src/db/mod.rs index 7493493b01..00ee03c048 100644 --- a/crates/aft/src/db/mod.rs +++ b/crates/aft/src/db/mod.rs @@ -24,6 +24,7 @@ pub(crate) mod contention_probe; pub mod github_read_cache; pub mod remote_exec; pub mod removal; +pub mod rich_ui; pub mod standing_roots; pub mod state; @@ -33,10 +34,29 @@ pub mod state; #[cfg(test)] pub(crate) mod wal_credit_probe; -pub const CURRENT_SCHEMA_VERSION: u32 = 19; +pub const CURRENT_SCHEMA_VERSION: u32 = rich_ui::SCHEMA_VERSION; const MIGRATION_VERSIONS: &[u32] = &[ - 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, + 1, + 2, + 3, + 4, + 5, + 6, + 7, + 8, + 9, + 10, + 11, + 12, + 13, + 14, + 15, + 16, + 17, + 18, + 19, + rich_ui::SCHEMA_VERSION, ]; // Performance-only indexes do not change the table format or reader floor. @@ -772,12 +792,16 @@ fn apply_pragmas_with_timeout( /// Returns the post-migration schema version. Refuses to open databases created /// by newer AFT versions. pub fn run_migrations(conn: &mut Connection) -> Result { + run_migrations_supported(conn, CURRENT_SCHEMA_VERSION) +} + +fn run_migrations_supported(conn: &mut Connection, supported: u32) -> Result { // Also fence callers that supply their own connection. Read the version // on that connection, never a second descriptor while it is live. if let Some(path) = conn.path().filter(|path| !path.is_empty()) { if crate::production_storage::protected(Path::new(path)) { - if current_schema_version(conn).ok() == Some(CURRENT_SCHEMA_VERSION) { - return Ok(CURRENT_SCHEMA_VERSION); + if current_schema_version(conn).ok() == Some(supported) { + return Ok(supported); } crate::production_storage::refuse_write(Path::new(path))?; } @@ -787,24 +811,27 @@ pub fn run_migrations(conn: &mut Connection) -> Result { )?; let db_version = current_schema_version(conn)?; - if db_version == CURRENT_SCHEMA_VERSION { + if db_version == supported { return Ok(db_version); } - if db_version > CURRENT_SCHEMA_VERSION { + if db_version > supported { return Err(OpenError::DowngradeRefused { db_version, - supported: CURRENT_SCHEMA_VERSION, + supported, }); } // The bare read above keeps current-schema opens read-only. A lagging opener // may still have observed a stale version, so every planned step re-reads it // after acquiring SQLite's write lock and skips work another opener committed. - for &version in MIGRATION_VERSIONS.iter().filter(|&&v| v > db_version) { + for &version in MIGRATION_VERSIONS + .iter() + .filter(|&&v| v > db_version && v <= supported) + { apply_migration(conn, version)?; } - Ok(CURRENT_SCHEMA_VERSION) + Ok(supported) } fn ensure_additive_indexes(conn: &Connection) -> rusqlite::Result<()> { @@ -989,6 +1016,7 @@ fn apply_migration_statements(conn: &Connection, version: u32) -> rusqlite::Resu 17 => conn.execute_batch(access_asks::MIGRATION), 18 => conn.execute_batch(access_grants::MODES_MIGRATION), 19 => call_ledger::migrate_result_source(conn), + rich_ui::SCHEMA_VERSION => conn.execute_batch(rich_ui::MIGRATION), _ => Ok(()), } } @@ -1026,6 +1054,10 @@ mod tests { "write_ledger_process_minutes", "write_ledger_meta", "write_ledger_unmeasurable_minutes", + "bash_view_records", + "rich_ui_templates", + "rich_ui_cursor_key", + "rich_ui_session_presets", ]; const EXPECTED_INDEXES: &[&str] = &[ @@ -1051,6 +1083,7 @@ mod tests { "idx_standing_root_freshness_needs_verify", "idx_write_ledger_minutes_window", "idx_write_ledger_unmeasurable_window", + "idx_rich_ui_session_presets_recorded", ]; /// Opens `aft.db` twice: the first connection plays this process's shared diff --git a/crates/aft/src/db/rich_ui.rs b/crates/aft/src/db/rich_ui.rs new file mode 100644 index 0000000000..1e3f81aa45 --- /dev/null +++ b/crates/aft/src/db/rich_ui.rs @@ -0,0 +1,312 @@ +//! Durable rich-UI assets and catalog attestations in the process's shared aft.db. + +use rusqlite::{params, Connection}; + +pub use super::remote_exec::PolicyKey as PresetKey; + +pub const SCHEMA_VERSION: u32 = 20; + +pub(crate) const MIGRATION: &str = r#" +CREATE TABLE IF NOT EXISTS bash_view_records ( + harness TEXT NOT NULL, session_id TEXT NOT NULL, task_id TEXT NOT NULL, + record TEXT NOT NULL, + row_template_digest TEXT NOT NULL, page_template_digest TEXT NOT NULL, + PRIMARY KEY (harness, session_id, task_id), + FOREIGN KEY (harness, session_id, task_id) + REFERENCES bash_tasks(harness, session_id, task_id) ON DELETE CASCADE +); +CREATE TABLE IF NOT EXISTS rich_ui_templates ( + digest TEXT NOT NULL PRIMARY KEY, bytes BLOB NOT NULL +); +CREATE TABLE IF NOT EXISTS rich_ui_cursor_key ( + id INTEGER NOT NULL PRIMARY KEY CHECK (id = 1), + key BLOB NOT NULL CHECK (length(key) = 32) +); +INSERT OR IGNORE INTO rich_ui_cursor_key(id, key) VALUES (1, randomblob(32)); +CREATE TABLE IF NOT EXISTS rich_ui_session_presets ( + project_root TEXT NOT NULL, harness TEXT NOT NULL, session TEXT NOT NULL, + principal TEXT NOT NULL, owner TEXT NOT NULL, scope_ref TEXT NOT NULL, + epoch TEXT NOT NULL, preset TEXT NOT NULL CHECK (preset IN ('worker', 'reader')), + recorded_at INTEGER NOT NULL, + PRIMARY KEY (project_root, harness, session, principal, owner, scope_ref, epoch, preset) +); +CREATE INDEX IF NOT EXISTS idx_rich_ui_session_presets_recorded + ON rich_ui_session_presets(recorded_at); +"#; + +/// Each successful fetch refreshes only its exact identity, epoch and preset. +/// Unlike routing policies, attestations never carry forward from an older epoch. +pub fn record_preset(conn: &Connection, key: &PresetKey, now_ms: i64) -> rusqlite::Result<()> { + conn.execute( + "INSERT INTO rich_ui_session_presets + (project_root,harness,session,principal,owner,scope_ref,epoch,preset,recorded_at) + VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9) + ON CONFLICT(project_root,harness,session,principal,owner,scope_ref,epoch,preset) + DO UPDATE SET recorded_at=excluded.recorded_at", + params![ + key.project_root, + key.harness, + key.session, + key.principal, + key.owner, + key.scope_ref, + key.epoch.to_string(), + key.preset, + now_ms + ], + )?; + Ok(()) +} + +/// Read all presets for the exact route and scope without refreshing retention. +/// The key's preset is ignored so a worker/reader disagreement remains visible. +pub fn session_presets(conn: &Connection, key: &PresetKey) -> rusqlite::Result> { + conn.prepare( + "SELECT preset FROM rich_ui_session_presets + WHERE project_root=?1 AND harness=?2 AND session=?3 AND principal=?4 + AND owner=?5 AND scope_ref=?6 AND epoch=?7 ORDER BY preset", + )? + .query_map( + params![ + key.project_root, + key.harness, + key.session, + key.principal, + key.owner, + key.scope_ref, + key.epoch.to_string() + ], + |row| row.get(0), + )? + .collect() +} + +pub(crate) fn prune_presets(conn: &Connection, now_ms: i64) -> rusqlite::Result { + const RETENTION_MS: i64 = 30 * 24 * 60 * 60 * 1000; + conn.execute( + "DELETE FROM rich_ui_session_presets WHERE recorded_at < ?1", + [now_ms.saturating_sub(RETENTION_MS)], + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use rusqlite::Connection; + + fn assert_tables(conn: &Connection) { + for table in [ + "bash_view_records", + "rich_ui_templates", + "rich_ui_cursor_key", + "rich_ui_session_presets", + ] { + let exists: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1)", + [table], + |row| row.get(0), + ) + .unwrap(); + assert!(exists, "missing {table}"); + } + } + + #[test] + fn rich_ui_migration_fresh_database_has_four_tables() { + let mut conn = Connection::open_in_memory().unwrap(); + super::super::run_migrations(&mut conn).unwrap(); + assert_tables(&conn); + assert_eq!(cursor_key(&conn).len(), 32); + } + + fn cursor_key(conn: &Connection) -> Vec { + conn.query_row("SELECT key FROM rich_ui_cursor_key WHERE id=1", [], |row| { + row.get(0) + }) + .unwrap() + } + + #[test] + fn rich_ui_migration_reapplication_keeps_cursor_key() { + let mut conn = Connection::open_in_memory().unwrap(); + super::super::run_migrations(&mut conn).unwrap(); + let key = cursor_key(&conn); + conn.execute_batch(MIGRATION).unwrap(); + assert_eq!(cursor_key(&conn), key); + let count: i64 = conn + .query_row("SELECT count(*) FROM rich_ui_cursor_key", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(count, 1); + } + + #[test] + fn rich_ui_migration_upgrade_reopen_and_downgrade_fence() { + use super::super::{run_migrations, run_migrations_supported, CURRENT_SCHEMA_VERSION}; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("aft.db"); + let previous = CURRENT_SCHEMA_VERSION - 1; + let mut conn = Connection::open(&path).unwrap(); + run_migrations_supported(&mut conn, previous).unwrap(); + conn.execute( + "INSERT INTO host_state(key,value,updated_at) VALUES ('upgrade-canary','preserved',1)", + [], + ) + .unwrap(); + assert_eq!(run_migrations(&mut conn).unwrap(), CURRENT_SCHEMA_VERSION); + assert_tables(&conn); + let key = cursor_key(&conn); + assert_eq!(key.len(), 32); + assert_ne!(key, vec![0; 32]); + drop(conn); + + let mut conn = super::super::open(&path).unwrap(); + let changes = conn.total_changes(); + assert_eq!(run_migrations(&mut conn).unwrap(), CURRENT_SCHEMA_VERSION); + assert_eq!( + conn.total_changes(), + changes, + "current-version open must not write" + ); + assert_eq!(cursor_key(&conn), key); + let canary: String = conn + .query_row( + "SELECT value FROM host_state WHERE key='upgrade-canary'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(canary, "preserved"); + assert!(matches!( + run_migrations_supported(&mut conn, previous), + Err(super::super::OpenError::DowngradeRefused { db_version, supported }) + if db_version == CURRENT_SCHEMA_VERSION && supported == previous + )); + let mut fresh = Connection::open_in_memory().unwrap(); + run_migrations(&mut fresh).unwrap(); + assert_ne!( + cursor_key(&fresh), + key, + "databases must not share a fixed key" + ); + } + + #[test] + fn rich_ui_migration_upgrades_schema_18_without_losing_ledger_migration() { + let mut conn = Connection::open_in_memory().unwrap(); + super::super::run_migrations_supported(&mut conn, 18).unwrap(); + super::super::run_migrations(&mut conn).unwrap(); + assert_tables(&conn); + conn.prepare("SELECT tool_result_source FROM call_ledger") + .unwrap(); + } + + fn key(epoch: u64) -> PresetKey { + PresetKey { + project_root: "/project".into(), + harness: "Raw Harness".into(), + session: "session".into(), + principal: "reserved:broca".into(), + owner: r#"{"kind":"reserved","module_id":"prefrontal-core"}"#.into(), + scope_ref: "worker".into(), + epoch, + preset: "worker".into(), + } + } + + #[test] + fn rich_ui_presets_epochs_are_exact_text_across_reopen() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("aft.db"); + let conn = super::super::open(&path).unwrap(); + let epochs = [i64::MAX as u64, i64::MAX as u64 + 1, u64::MAX]; + for epoch in epochs { + record_preset(&conn, &key(epoch), 1).unwrap(); + } + drop(conn); + let conn = super::super::open(&path).unwrap(); + for epoch in epochs { + let k = key(epoch); + let stored: (String, String) = conn + .query_row( + "SELECT epoch,typeof(epoch) FROM rich_ui_session_presets WHERE epoch=?1", + [epoch.to_string()], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!(stored, (epoch.to_string(), "text".into())); + assert_eq!(session_presets(&conn, &k).unwrap(), ["worker"]); + // Other requested boundary epochs are also present. Remove them + // temporarily to prove each boundary misses both of its neighbours. + for other in epochs.into_iter().filter(|&other| other != epoch) { + conn.execute( + "DELETE FROM rich_ui_session_presets WHERE epoch=?1", + [other.to_string()], + ) + .unwrap(); + } + for neighbour in [epoch.checked_sub(1), epoch.checked_add(1)] + .into_iter() + .flatten() + { + assert!(session_presets(&conn, &key(neighbour)).unwrap().is_empty()); + } + for other in epochs.into_iter().filter(|&other| other != epoch) { + record_preset(&conn, &key(other), 1).unwrap(); + } + } + } + + #[test] + fn rich_ui_presets_exact_identity_and_distinct_presets_are_read_only() { + let mut conn = Connection::open_in_memory().unwrap(); + super::super::run_migrations(&mut conn).unwrap(); + let k = key(7); + record_preset(&conn, &k, 10).unwrap(); + record_preset( + &conn, + &PresetKey { + preset: "reader".into(), + ..k.clone() + }, + 20, + ) + .unwrap(); + let changes = conn.total_changes(); + assert_eq!(session_presets(&conn, &k).unwrap(), ["reader", "worker"]); + for field in 0..7 { + let mut wrong = k.clone(); + match field { + 0 => wrong.project_root.push('x'), + 1 => wrong.harness.push('x'), + 2 => wrong.session.push('x'), + 3 => wrong.principal.push('x'), + 4 => wrong.owner.push('x'), + 5 => wrong.scope_ref.push('x'), + _ => wrong.epoch += 1, + } + assert!( + session_presets(&conn, &wrong).unwrap().is_empty(), + "field {field}" + ); + } + assert_eq!(conn.total_changes(), changes); + } + + #[test] + fn rich_ui_presets_retention_tick_removes_only_older_than_thirty_days() { + let mut conn = Connection::open_in_memory().unwrap(); + super::super::run_migrations(&mut conn).unwrap(); + let now = 40 * 24 * 60 * 60 * 1000; + let cutoff = now - 30 * 24 * 60 * 60 * 1000; + record_preset(&conn, &key(1), cutoff - 1).unwrap(); + record_preset(&conn, &key(2), cutoff).unwrap(); + record_preset(&conn, &key(3), now).unwrap(); + super::super::compression_events::prune_retention_tick(&mut conn, now).unwrap(); + assert!(session_presets(&conn, &key(1)).unwrap().is_empty()); + assert_eq!(session_presets(&conn, &key(2)).unwrap(), ["worker"]); + assert_eq!(session_presets(&conn, &key(3)).unwrap(), ["worker"]); + } +} diff --git a/crates/aft/src/subc/persistence.rs b/crates/aft/src/subc/persistence.rs index 5887709a7a..36dcd55c53 100644 --- a/crates/aft/src/subc/persistence.rs +++ b/crates/aft/src/subc/persistence.rs @@ -230,8 +230,8 @@ impl DatabaseWaits { if let Some(operation) = operation.filter(|operation| tool_provider::recognized_operation(operation)) { - // Access settings (notably a head's approval mode) also use aft.db. - // Their readiness must not depend on having a saved worker policy. + // Access settings and scoped catalog preset attestations also use + // aft.db, even when no saved worker policy needs to be read. let catalog_wait = operation == "tool.catalog" && ctx.database_runtime_initializing() && remote_policy::catalog_needs_database(body, identity); @@ -743,6 +743,11 @@ mod tests { scope.owner = Principal::Reserved { module_id: "prefrontal-core".into(), }; + if let AuthenticatedPrincipal::RouteBind { principal_id, .. } = + &mut Arc::get_mut(&mut head.0).unwrap().spawn_principal + { + *principal_id = Some("reserved:prefrontal-core".into()); + } let decoded = DecodedFrame { parsed_body: Default::default(), confinement: None, @@ -845,6 +850,72 @@ mod tests { ); } + #[tokio::test] + async fn rich_ui_preset_writer_catalog_waits_for_open_then_records_without_refetch() { + use std::future::Future; + for preset in ["worker", "reader"] { + let (dir, root) = test_support::test_root("preset-catalog-database-wait"); + let ctx = test_support::test_ctx(); + ctx.mark_database_runtime_initializing_for_test(); + let executor = Arc::new(Executor::new()); + assert!(executor.register_actor(root.clone(), ctx.clone())); + let (mut identity, mut decoded) = head_catalog(&root, None); + let data = Arc::get_mut(&mut identity.0).unwrap(); + data.role = tool_provider::RouteRole::ToolProviderV1; + if preset == "worker" { + data.scope.as_mut().unwrap().kind = subc_protocol::scope::ScopeKind::Worker; + } + if let AuthenticatedPrincipal::RouteBind { principal_id, .. } = + &mut data.spawn_principal + { + *principal_id = Some("reserved:broca".into()); + } + decoded.frame = Frame::build( + FrameType::Request, + control_flags(), + 43, + 1, + 7, + serde_json::to_vec(&json!({"op":"tool.catalog","preset":preset})).unwrap(), + ) + .unwrap(); + let routes = HashMap::from([(route_key(43, 1), identity.clone())]); + let mut waits = DatabaseWaits::default(); + assert!( + waits.defer(decoded, &routes, &executor).is_ok(), + "{preset} catalog must wait" + ); + let resumed = waits.next(); + tokio::pin!(resumed); + tokio::task::yield_now().await; + std::future::poll_fn(|cx| { + assert!( + resumed.as_mut().poll(cx).is_pending(), + "catalog resumed before open" + ); + std::task::Poll::Ready(()) + }) + .await; + ctx.set_db(Arc::new(std::sync::Mutex::new( + crate::db::open(&dir.path().join("aft.db")).unwrap(), + ))); + ctx.finish_database_runtime(Ok(())); + let reply = head_catalog_reply(resumed.await, identity.clone(), &executor).await; + assert_eq!( + reply.header.ty, + FrameType::Response, + "{}", + String::from_utf8_lossy(&reply.body) + ); + let key = remote_policy::session_preset_key(&identity, preset).unwrap(); + assert_eq!( + crate::db::rich_ui::session_presets(&ctx.db().unwrap().lock().unwrap(), &key) + .unwrap(), + [preset] + ); + } + } + #[tokio::test] async fn head_catalog_database_deadline_and_failure_are_retryable() { let (_dir, root) = test_support::test_root("head-catalog-database-refusal"); diff --git a/crates/aft/src/subc/remote_policy.rs b/crates/aft/src/subc/remote_policy.rs index 350516cc18..729a7af38b 100644 --- a/crates/aft/src/subc/remote_policy.rs +++ b/crates/aft/src/subc/remote_policy.rs @@ -57,8 +57,13 @@ fn policy_key(identity: &RouteIdentity, preset: Option<&str>) -> Result bool { + let preset = body.get("preset").and_then(Value::as_str).unwrap_or("head"); + catalog_preset_key(identity, preset).is_some() || catalog_needs_policy_database(body, identity) +} + +fn catalog_needs_policy_database(body: &Value, identity: &RouteIdentity) -> bool { let preset = body.get("preset").and_then(Value::as_str).unwrap_or("head"); key(identity, Some("worker")).is_some() || body @@ -72,11 +77,41 @@ pub(super) fn catalog_needs_database(body: &Value, identity: &RouteIdentity) -> } pub(super) fn catalog( + body: Value, + identity: &RouteIdentity, + ctx: &AppContext, +) -> Result { + let preset = body.get("preset").and_then(Value::as_str).unwrap_or("head"); + let key = catalog_preset_key(identity, preset); + let answer = catalog_inner(body, identity, ctx)?; + if let Some(key) = key { + // An absent pool allows discovery without an attestation. Once a pool + // exists, do not acknowledge a fetch whose attestation could not commit. + if let Some(db) = ctx.db() { + let unavailable = || { + subc_protocol::ErrorBody::new( + "database_unavailable", + "catalog preset persistence unavailable", + ) + }; + let conn = db.lock().map_err(|_| unavailable())?; + crate::db::rich_ui::record_preset( + &conn, + &key, + crate::bash_background::persistence::unix_millis() as i64, + ) + .map_err(|_| unavailable())?; + } + } + Ok(answer) +} + +fn catalog_inner( mut body: Value, identity: &RouteIdentity, ctx: &AppContext, ) -> Result { - let needs_database = catalog_needs_database(&body, identity); + let needs_database = catalog_needs_policy_database(&body, identity); let preset = body .get("preset") .and_then(Value::as_str) @@ -335,6 +370,52 @@ pub(super) fn catalog( Ok(answer) } +/// Preserve the bind's harness label to identify the route that fetched the +/// catalog. Do not apply `HarnessId::storage_segment`, which identifies bash +/// task storage rather than catalog attestations. Carrier policy-route lookup +/// must construct the same key as catalog recording. +pub(super) fn session_preset_key( + identity: &RouteIdentity, + preset: &str, +) -> Option { + let scope = identity.scope.as_ref()?; + let AuthenticatedPrincipal::RouteBind { + principal_id: Some(principal), + .. + } = &identity.spawn_principal + else { + return None; + }; + Some(crate::db::rich_ui::PresetKey { + project_root: identity.project_root.display().to_string(), + harness: identity.harness.clone(), + session: identity.session.clone(), + principal: principal.clone(), + owner: serde_json::to_string(&scope.owner).expect("scope owner serializes"), + scope_ref: scope.scope_ref.clone(), + epoch: scope.scope_epoch, + preset: preset.into(), + }) +} + +fn catalog_preset_key( + identity: &RouteIdentity, + preset: &str, +) -> Option { + if !matches!(preset, "worker" | "reader") { + return None; + } + let scope = identity.scope.as_ref()?; + let key = session_preset_key(identity, preset)?; + if !scope.owner_authorized + || key.principal == "reserved:occipital" + || principal_id(&Some(scope.owner.clone())).as_deref() == Some(key.principal.as_str()) + { + return None; + } + Some(key) +} + // Keep the policy gates in their original short-circuit order. Both schema // selection and the diagnostic consume this result; request params alone do // not describe a saved worker policy or a head's user-config policy. @@ -694,3 +775,218 @@ pub(super) fn saved_policy( #[cfg(test)] #[path = "remote_policy_tests.rs"] mod tests; + +#[cfg(test)] +mod preset_writer_tests { + use super::*; + + fn identity(root: &Path, principal: &str) -> RouteIdentity { + RouteIdentity(Arc::new(RouteIdentityData { + root: ProjectRootId::from_path(root).unwrap(), + project_root: root.into(), + harness: "Raw Harness".into(), + session: "worker-session".into(), + role: tool_provider::RouteRole::ToolProviderV1, + trust: BindTrust::FirstParty, + spawn_principal: AuthenticatedPrincipal::RouteBind { + trust: crate::sandbox_spawn::PrincipalTrust::FirstParty, + route_channel: 41, + route_epoch: 1, + project_root: root.into(), + harness: "Raw Harness".into(), + session_id: "worker-session".into(), + principal_id: Some(principal.into()), + }, + consumer_elicitation_capable: false, + tool_result_body: false, + disabled_tools: Arc::new(vec![]), + run_scope: OnceLock::new(), + call_mode: subc_protocol::call_mode::CallMode::Classic, + scope: Some( + subc_protocol::scope::ScopeStamp::new( + subc_protocol::Principal::Reserved { + module_id: "prefrontal-core".into(), + }, + "worker", + 7, + subc_protocol::scope::ScopeKind::Worker, + ) + .with_owner_authorized(true), + ), + made_tool_call: AtomicBool::new(false), + run_owner: None, + })) + } + + fn context(root: &Path, storage: &Path, with_db: bool) -> AppContext { + let mut config = crate::config::Config::default(); + config.project_root = Some(root.into()); + config.storage_dir = Some(storage.into()); + let ctx = AppContext::new(Box::new(crate::parser::TreeSitterProvider::new()), config); + if with_db { + ctx.set_db(Arc::new(StdMutex::new( + crate::db::open(&storage.join("aft.db")).unwrap(), + ))); + } + ctx + } + + fn rows(ctx: &AppContext) -> i64 { + ctx.db() + .unwrap() + .lock() + .unwrap() + .query_row("SELECT count(*) FROM rich_ui_session_presets", [], |row| { + row.get(0) + }) + .unwrap() + } + + #[test] + fn rich_ui_preset_writer_records_worker_and_reader_once_and_refreshes() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let ctx = context(root.path(), storage.path(), true); + let route = identity(root.path(), "reserved:broca"); + for preset in ["worker", "reader"] { + let request = json!({"preset":preset}); + catalog(request.clone(), &route, &ctx).unwrap(); + let key = session_preset_key(&route, preset).unwrap(); + assert_eq!(key.harness, "Raw Harness"); + assert_eq!(key.project_root, root.path().display().to_string()); + assert_eq!(key.session, "worker-session"); + assert_eq!(key.scope_ref, "worker"); + assert_eq!(key.epoch, 7); + ctx.db() + .unwrap() + .lock() + .unwrap() + .execute( + "UPDATE rich_ui_session_presets SET recorded_at=1 WHERE preset=?1", + [preset], + ) + .unwrap(); + catalog(request, &route, &ctx).unwrap(); + let db = ctx.db().unwrap(); + let conn = db.lock().unwrap(); + let recorded: i64 = conn + .query_row( + "SELECT recorded_at FROM rich_ui_session_presets WHERE preset=?1", + [preset], + |row| row.get(0), + ) + .unwrap(); + assert!(recorded > 1); + let stored: (String, String, String) = conn + .query_row( + "SELECT harness,principal,owner FROM rich_ui_session_presets WHERE preset=?1", + [preset], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .unwrap(); + assert_eq!(stored, (key.harness, key.principal, key.owner)); + } + assert_eq!(rows(&ctx), 2); + } + + #[test] + fn rich_ui_preset_writer_occipital_fetch_writes_nothing() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let ctx = context(root.path(), storage.path(), true); + let route = identity(root.path(), "reserved:occipital"); + for preset in ["worker", "reader"] { + catalog(json!({"preset":preset}), &route, &ctx).unwrap(); + } + assert_eq!(rows(&ctx), 0); + let ordinary = identity(root.path(), "reserved:broca"); + catalog(json!({"preset":"worker"}), &ordinary, &ctx).unwrap(); + assert_eq!(rows(&ctx), 1); + } + + #[test] + fn rich_ui_preset_writer_skips_heads_unauthorized_unscoped_and_owner_principal() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let ctx = context(root.path(), storage.path(), true); + let route = identity(root.path(), "reserved:broca"); + catalog(json!({}), &route, &ctx).unwrap(); + catalog(json!({"preset":"head"}), &route, &ctx).unwrap(); + for preset in ["worker", "reader"] { + let mut unauthorized = identity(root.path(), "reserved:broca"); + Arc::get_mut(&mut unauthorized.0) + .unwrap() + .scope + .as_mut() + .unwrap() + .owner_authorized = false; + catalog(json!({"preset":preset}), &unauthorized, &ctx).unwrap(); + let mut unscoped = identity(root.path(), "reserved:broca"); + Arc::get_mut(&mut unscoped.0).unwrap().scope = None; + catalog(json!({"preset":preset}), &unscoped, &ctx).unwrap(); + let owner = identity(root.path(), "reserved:prefrontal-core"); + catalog(json!({"preset":preset}), &owner, &ctx).unwrap(); + } + assert_eq!(rows(&ctx), 0); + catalog(json!({"preset":"reader"}), &route, &ctx).unwrap(); + assert_eq!(rows(&ctx), 1); + } + + #[test] + fn rich_ui_preset_writer_failed_fetch_is_not_recorded() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let ctx = context(root.path(), storage.path(), true); + let route = identity(root.path(), "reserved:broca"); + assert!(catalog( + json!({"preset":"worker","params":{"invalid":true}}), + &route, + &ctx + ) + .is_err()); + assert_eq!(rows(&ctx), 0); + } + + #[test] + fn rich_ui_preset_writer_existing_row_write_failure_fails_fetch() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let ctx = context(root.path(), storage.path(), true); + let route = identity(root.path(), "reserved:broca"); + let request = json!({"preset":"worker"}); + catalog(request.clone(), &route, &ctx).unwrap(); + ctx.db() + .unwrap() + .lock() + .unwrap() + .execute_batch( + "UPDATE rich_ui_session_presets SET recorded_at=1; + CREATE TRIGGER fail_preset BEFORE UPDATE ON rich_ui_session_presets + BEGIN SELECT RAISE(FAIL, 'forced preset write failure'); END;", + ) + .unwrap(); + let error = catalog(request.clone(), &route, &ctx).unwrap_err(); + assert_eq!(error.code, "database_unavailable"); + assert_eq!(rows(&ctx), 1); + ctx.db() + .unwrap() + .lock() + .unwrap() + .execute_batch("DROP TRIGGER fail_preset") + .unwrap(); + catalog(request, &route, &ctx).unwrap(); + assert_eq!(rows(&ctx), 1); + } + + #[test] + fn rich_ui_preset_writer_missing_database_allows_unrecorded_fetch() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let ctx = context(root.path(), storage.path(), false); + let route = identity(root.path(), "reserved:broca"); + for preset in ["worker", "reader"] { + catalog(json!({"preset":preset}), &route, &ctx).unwrap(); + } + assert!(ctx.db().is_none()); + } +}