From 1c3344ecace1cca520ff4d245225b680f5a45b34 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 22:28:18 -0800 Subject: [PATCH 1/5] fix(projection): remove SQLite TEMP output preflight Refs #4152 --- .../src/observation_projection/rebuild.rs | 259 ++++++++++++++---- 1 file changed, 201 insertions(+), 58 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index e32499a11b5..2eee7419d61 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2328,57 +2328,6 @@ async fn prepare_rebuild_output_activation( conn: &impl Executor, generation: &str, ) -> ProjectionStoreResult<()> { - conn.execute_batch( - "DROP TABLE IF EXISTS temp.observation_projection_rebuild_preexisting_outputs; - CREATE TEMP TABLE observation_projection_rebuild_preexisting_outputs ( - output_provider TEXT NOT NULL, - output_message_id TEXT NOT NULL, - active_exists INTEGER NOT NULL CHECK(active_exists IN (0, 1)), - current_owned INTEGER NOT NULL CHECK(current_owned IN (0, 1)), - cross_owned INTEGER NOT NULL CHECK(cross_owned IN (0, 1)), - staged_created INTEGER NOT NULL CHECK(staged_created IN (0, 1)), - PRIMARY KEY(output_provider, output_message_id) - ) WITHOUT ROWID;", - ) - .await - .map_err(|error| storage("prepare staged projection output activation", error))?; - conn.execute( - "INSERT INTO temp.observation_projection_rebuild_preexisting_outputs ( - output_provider, output_message_id, active_exists, - current_owned, cross_owned, staged_created - ) - SELECT staged.output_provider, staged.output_message_id, - EXISTS ( - SELECT 1 FROM lcm_raw_messages AS active - WHERE active.provider = staged.output_provider - AND active.message_id = staged.output_message_id - ), - EXISTS ( - SELECT 1 FROM observation_projection_provenance AS owner - WHERE owner.projector_version = ?1 AND owner.message_created = 1 - AND owner.output_provider = staged.output_provider - AND owner.output_message_id = staged.output_message_id - ), - EXISTS ( - SELECT 1 FROM observation_projection_provenance AS owner - WHERE owner.projector_version <> ?1 - AND owner.output_provider = staged.output_provider - AND owner.output_message_id = staged.output_message_id - ), - EXISTS ( - SELECT 1 FROM observation_projection_rebuild_provenance AS owner - WHERE owner.projector_version = ?1 AND owner.generation = ?2 - AND owner.message_created = 1 - AND owner.output_provider = staged.output_provider - AND owner.output_message_id = staged.output_message_id - ) - FROM observation_projection_rebuild_messages AS staged - WHERE staged.projector_version = ?1 AND staged.generation = ?2", - params![SESSION_MESSAGE_PROJECTOR_VERSION, generation], - ) - .await - .map_err(|error| storage("materialize preexisting projection outputs", error))?; - let message_conflicts = format!( "{} OR (active.provider <> 'hermes' AND ( @@ -2393,26 +2342,46 @@ async fn prepare_rebuild_output_activation( json_extract_expr(STAGED_MESSAGE_JSON_COLUMN, "text"), json_extract_expr(STAGED_MESSAGE_JSON_COLUMN, "metadata_json"), ); + // Validate against live indexed ownership while the old projector still owns + // its outputs. A long-lived writer must not materialize connection-local + // TEMP tables: they can keep unlinked SQLite files allocated between rebuilds. let mut conflicts = conn .query( &format!( "SELECT staged.output_provider, staged.output_message_id FROM observation_projection_rebuild_messages AS staged - JOIN temp.observation_projection_rebuild_preexisting_outputs AS ownership - ON ownership.output_provider = staged.output_provider - AND ownership.output_message_id = staged.output_message_id LEFT JOIN lcm_raw_messages AS active ON active.provider = staged.output_provider AND active.message_id = staged.output_message_id WHERE staged.projector_version = ?1 AND staged.generation = ?2 AND ( - (ownership.active_exists = 0 AND ownership.staged_created = 0) + (active.message_id IS NULL + AND NOT EXISTS ( + SELECT 1 FROM observation_projection_rebuild_provenance AS staged_owner + WHERE staged_owner.projector_version = ?1 + AND staged_owner.generation = ?2 + AND staged_owner.message_created = 1 + AND staged_owner.output_provider = staged.output_provider + AND staged_owner.output_message_id = staged.output_message_id + )) OR ( - ownership.active_exists = 1 - AND NOT (ownership.current_owned = 1 AND ownership.cross_owned = 0) + active.message_id IS NOT NULL AND ( - {message_conflicts} + NOT EXISTS ( + SELECT 1 FROM observation_projection_provenance AS current_owner + WHERE current_owner.projector_version = ?1 + AND current_owner.message_created = 1 + AND current_owner.output_provider = staged.output_provider + AND current_owner.output_message_id = staged.output_message_id + ) + OR EXISTS ( + SELECT 1 FROM observation_projection_provenance AS foreign_owner + WHERE foreign_owner.projector_version <> ?1 + AND foreign_owner.output_provider = staged.output_provider + AND foreign_owner.output_message_id = staged.output_message_id + ) ) + AND ({message_conflicts}) ) ) LIMIT 1" @@ -2609,6 +2578,7 @@ mod activation_tests { use super::{ REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION, activate_projection_rebuild_transaction, activate_rebuild_sessions, + prepare_rebuild_output_activation, projected_stored_message, reconcile_overlapping_rebuild_sessions, write_staged_message, }; use crate::tests::harness::RegisteredGlobalDbHarness; @@ -2796,6 +2766,179 @@ mod activation_tests { ); } + #[asupersync::test] + async fn output_preflight_rejects_unowned_changes_without_retaining_temp_storage() { + const GENERATION: &str = "generation.output-preflight"; + let harness = RegisteredGlobalDbHarness::open("output-preflight-no-temp").await; + let active = session(None, None); + assert!(harness.registered.upsert_session(&active).await); + + let foreign = SessionMessageRecord { + provider: active.provider.clone(), + session_id: active.session_id.clone(), + message_id: "message.unowned".to_owned(), + role: "assistant".to_owned(), + timestamp: Some(1), + ordinal: 1, + text: "foreign durable body".to_owned(), + kind: None, + model: None, + tool_names: None, + source_path: None, + source_offset: None, + metadata_json: None, + }; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + tracedecay_lcm::raw::upsert_projection_raw_message( + &transaction, + &foreign, + Some("/foreign"), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + // The foreign message must exist in the committed store before a + // separate rebuild transaction attempts to replace it. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + stage(&transaction, GENERATION, &active).await; + let changed = SessionMessageRecord { + text: "overwritten body".to_owned(), + ..foreign.clone() + }; + let changed_stored = projected_stored_message(&changed).unwrap(); + write_staged_message(&transaction, GENERATION, &changed_stored) + .await + .unwrap(); + // Schema initialization and other writer operations can already + // allocate TEMP pages. The preflight must not grow that allocation, + // regardless of unrelated TEMP storage already held by this connection. + let mut baseline_pages = transaction + .query("PRAGMA temp.page_count", ()) + .await + .unwrap(); + let baseline_page_count = baseline_pages + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(); + drop(baseline_pages); + + let error = prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .unwrap_err(); + assert!( + matches!(error, ProjectionStoreError::OutputCollision { provider, message_id } + if provider == "cursor" && message_id == "message.unowned"), + "rebuild must refuse a foreign output with different content" + ); + + // Matching foreign bodies are compatible; rejecting every unowned row + // would also avoid spills, but would break legitimate rebuilds. + let compatible_stored = projected_stored_message(&foreign).unwrap(); + write_staged_message(&transaction, GENERATION, &compatible_stored) + .await + .unwrap(); + prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .expect("unchanged foreign output must remain eligible"); + + let missing = SessionMessageRecord { + message_id: "message.not-created".to_owned(), + ..foreign.clone() + }; + let missing_stored = projected_stored_message(&missing).unwrap(); + write_staged_message(&transaction, GENERATION, &missing_stored) + .await + .unwrap(); + let error = prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .unwrap_err(); + assert!( + matches!(error, ProjectionStoreError::OutputCollision { provider, message_id } + if provider == "cursor" && message_id == "message.not-created"), + "an absent output without staged creation evidence must be rejected" + ); + + let mut rows = transaction + .query( + "SELECT name FROM sqlite_temp_master + WHERE name = 'observation_projection_rebuild_preexisting_outputs'", + (), + ) + .await + .unwrap(); + assert!( + rows.next().await.unwrap().is_none(), + "repeated preflight checks must not retain a connection-local TEMP table" + ); + drop(rows); + + // A dropped TEMP b-tree can disappear from sqlite_temp_master while + // its pages remain allocated on this writer. Compare page counts on + // this same long-lived connection before and after repeated preflight + // calls; requiring an absolute zero would conflate earlier storage. + let mut pages = transaction + .query("PRAGMA temp.page_count", ()) + .await + .unwrap(); + assert_eq!( + pages.next().await.unwrap().unwrap().get::(0).unwrap(), + baseline_page_count, + "repeated preflight checks must not grow SQLite TEMP database pages" + ); + drop(pages); + + transaction.rollback().await.unwrap(); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT content, location_path FROM lcm_raw_messages + WHERE provider = ?1 AND message_id = ?2", + params![foreign.provider.as_str(), foreign.message_id.as_str()], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert_eq!( + row.get::(0).unwrap(), + foreign.text, + "a denied rebuild must not replace a committed foreign body" + ); + assert_eq!(row.get::(1).unwrap(), "/foreign"); + drop(rows); + + let mut rows = snapshot + .query( + "SELECT COUNT(*) FROM observation_projection_rebuilds + WHERE generation = ?1", + params![GENERATION], + ) + .await + .unwrap(); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + 0, + "a failed rebuild must leave no durable staging generation" + ); + drop(rows); + + let mut integrity = snapshot.query("PRAGMA integrity_check", ()).await.unwrap(); + assert_eq!( + integrity + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + "ok", + "the committed SQLite store must remain structurally sound after rollback" + ); + } + #[asupersync::test] async fn activation_names_the_session_field_instead_of_a_message_collision() { let harness = RegisteredGlobalDbHarness::open("session-collision-field").await; From ad4c9b77d86989ac504f81464ad7bc3d460ab02c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:25:04 -0800 Subject: [PATCH 2/5] test(storage): exercise SQLite TEMP integrity and spill release Refs #4152 Exercise MEMORY and FILE TEMP databases under the shipped allocator. Check live main/TEMP pages and exact committed bodies across rollback, cross-thread frees, allocator collection, sorting and connection close. Verify real Linux unlinked files against the production spill census. Keep the glibc routing control and run integrity checks on other Unix hosts. The local SQLite 3.46.1 SQL/FD probe passed both modes; repository Bazel, mimalloc and native macOS execution remain pending. --- .../tests/c_library_allocator.rs | 271 +++++++++++++++++- 1 file changed, 261 insertions(+), 10 deletions(-) diff --git a/crates/tracedecay-cli/tests/c_library_allocator.rs b/crates/tracedecay-cli/tests/c_library_allocator.rs index df4f9ef0adc..d0d7d506d68 100644 --- a/crates/tracedecay-cli/tests/c_library_allocator.rs +++ b/crates/tracedecay-cli/tests/c_library_allocator.rs @@ -1,28 +1,30 @@ -//! The shipped binary's C libraries allocate from mimalloc, not glibc. +//! The shipped binary's routed C allocations survive collection and TEMP I/O. //! -//! One test in its own binary: startup routing has to run before SQLite or -//! tree-sitter allocates anything, and nothing else may touch glibc while it -//! measures. Rust allocates through mimalloc here as it does in the binary. +//! One test owns this process: startup routing must precede every SQLite and +//! tree-sitter allocation. The glibc counter control remains Linux-specific; +//! live-page integrity and committed-data checks also run on other Unix hosts. -#![cfg(all( - target_os = "linux", - target_env = "gnu", - not(feature = "alloc-jemalloc") -))] +#![cfg(all(unix, not(feature = "alloc-jemalloc")))] #[path = "../src/process_allocator.rs"] mod process_allocator; +use std::collections::BTreeSet; + +use tracedecay_domain::process_heap::collect_calling_thread_allocator_v1; + #[global_allocator] static MIMALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; const MIB: usize = 1024 * 1024; +const ROW_COUNT: i64 = 1024; /// Bytes glibc's arenas and its mmapped chunks hold for live allocations. /// /// The hermetic toolchain links against glibc 2.31, which predates /// `mallinfo2`. `mallinfo` reports the same counters as `int`, which this /// test's few-MiB working set stays far below. +#[cfg(all(target_os = "linux", target_env = "gnu"))] fn glibc_in_use() -> usize { // SAFETY: `mallinfo` only reads allocator statistics. let info = unsafe { libc::mallinfo() }; @@ -31,9 +33,20 @@ fn glibc_in_use() -> usize { } #[test] -fn sqlite_and_tree_sitter_allocate_from_mimalloc_not_glibc() { +fn routed_c_libraries_preserve_live_pages_and_release_temp_files() { process_allocator::configure_process_allocator(); + #[cfg(all(target_os = "linux", target_env = "gnu"))] + assert_c_libraries_avoid_glibc(); + // Both modes matter: an integrity check of a copied TEMP file cannot + // inspect dirty pages that exist only in the live connection's cache. + for temp_store in [2, 1] { + assert_temp_integrity_and_reclamation(temp_store); + } +} + +#[cfg(all(target_os = "linux", target_env = "gnu"))] +fn assert_c_libraries_avoid_glibc() { let before = glibc_in_use(); // SAFETY: a plain allocation, written in full and freed below. let direct = std::hint::black_box(unsafe { libc::malloc(8 * MIB) }); @@ -86,3 +99,241 @@ fn sqlite_and_tree_sitter_allocate_from_mimalloc_not_glibc() { ); drop((tree, parser, connection)); } + +fn payload(id: i64) -> Vec { + id.to_le_bytes().repeat(512) +} + +fn assert_rows(connection: &rusqlite::Connection, table: &str) { + let mut statement = connection + .prepare(&format!("SELECT id, body FROM {table} ORDER BY id")) + .expect("read the fixture's complete row set"); + let mut rows = statement.query([]).expect("query fixture rows"); + for id in 0..ROW_COUNT { + let row = rows.next().expect("read row").expect("no missing row"); + assert_eq!(row.get::<_, i64>(0).expect("id"), id); + assert_eq!(row.get::<_, Vec>(1).expect("body"), payload(id)); + } + assert!(rows.next().expect("end of rows").is_none()); +} + +fn assert_integrity(connection: &rusqlite::Connection, schema: &str) { + let mut statement = connection + .prepare(&format!("PRAGMA {schema}.integrity_check")) + .expect("prepare live integrity check"); + let results = statement + .query_map([], |row| row.get::<_, String>(0)) + .expect("run live integrity check") + .collect::>>() + .expect("read every integrity result"); + assert_eq!( + results, + vec!["ok".to_owned()], + "{schema} must remain structurally sound" + ); +} + +fn collect_with_cross_thread_frees() { + let handed_off: Vec<_> = (0..64).map(payload).collect(); + let live = payload(97); + std::thread::spawn(move || { + drop(handed_off); + collect_calling_thread_allocator_v1(); + }) + .join() + .expect("cross-thread frees complete"); + collect_calling_thread_allocator_v1(); + assert_eq!(live, payload(97), "collection must preserve live blocks"); +} + +fn assert_temp_integrity_and_reclamation(temp_store: i64) { + let root = tempfile::tempdir().expect("isolated storage root"); + let path = root.path().join("durable.db"); + #[cfg(target_os = "linux")] + let before = sqlite_spills::snapshot(); + let mut connection = rusqlite::Connection::open(&path).expect("open durable fixture"); + connection + .execute_batch(&format!( + "PRAGMA journal_mode=WAL; + PRAGMA temp_store={temp_store}; + PRAGMA cache_size=8; + PRAGMA temp.cache_size=8; + CREATE TABLE durable(id INTEGER PRIMARY KEY, body BLOB NOT NULL);" + )) + .expect("configure fixture"); + assert_eq!( + connection + .query_row("PRAGMA temp_store", [], |row| row.get::<_, i64>(0)) + .expect("read TEMP mode"), + temp_store + ); + { + let transaction = connection.transaction().expect("seed durable data"); + { + let mut insert = transaction + .prepare("INSERT INTO durable VALUES (?1, ?2)") + .expect("prepare fixture insert"); + for id in 0..ROW_COUNT { + insert + .execute(rusqlite::params![id, payload(id)]) + .expect("insert fixture row"); + } + } + transaction.commit().expect("commit durable data"); + } + connection + .execute_batch( + "CREATE TEMP TABLE scratch(id INTEGER PRIMARY KEY, body BLOB NOT NULL); + INSERT INTO scratch SELECT id, body FROM main.durable;", + ) + .expect("populate live TEMP pages"); + let temp_pages: i64 = connection + .query_row("PRAGMA temp.page_count", [], |row| row.get(0)) + .expect("TEMP page count"); + let page_size: i64 = connection + .query_row("PRAGMA temp.page_size", [], |row| row.get(0)) + .expect("TEMP page size"); + assert!(temp_pages * page_size > i64::try_from(MIB).expect("MiB fits")); + + #[cfg(target_os = "linux")] + let temp_files = sqlite_spills::new_since(&before); + #[cfg(target_os = "linux")] + if temp_store == 1 { + assert!( + temp_files.values().sum::() > 1024 * 1024, + "FILE mode must really spill; a heap-only run is not disk evidence" + ); + } else { + assert!(temp_files.is_empty(), "MEMORY mode must keep TEMP in heap"); + } + + for _ in 0..8 { + connection + .execute_batch( + "BEGIN IMMEDIATE; + UPDATE main.durable SET body=zeroblob(4096); + DELETE FROM temp.scratch WHERE id % 3 = 0;", + ) + .expect("start mutations in both databases"); + collect_with_cross_thread_frees(); + connection + .execute_batch("ROLLBACK") + .expect("roll back both databases"); + assert_rows(&connection, "main.durable"); + assert_rows(&connection, "temp.scratch"); + assert_integrity(&connection, "main"); + assert_integrity(&connection, "temp"); + } + assert_sorter_reclamation(&connection, temp_store); + connection + .execute_batch("DROP TABLE temp.scratch; PRAGMA shrink_memory;") + .expect("release unused scratch without deleting durable data"); + collect_with_cross_thread_frees(); + assert_rows(&connection, "main.durable"); + connection.close().expect("close the owning SQLite connection"); + #[cfg(target_os = "linux")] + sqlite_spills::assert_reclaimed(&temp_files); + + let reopened = rusqlite::Connection::open(&path).expect("reopen the same durable store"); + assert_rows(&reopened, "main.durable"); + assert_integrity(&reopened, "main"); + reopened.close().expect("close reopened fixture"); +} + +fn assert_sorter_reclamation(connection: &rusqlite::Connection, temp_store: i64) { + #[cfg(target_os = "linux")] + let before = sqlite_spills::snapshot(); + // The expression defeats the rowid order; wide keys force a real sorter + // spill with the eight-page cache, distinct from the TEMP database file. + let mut statement = connection + .prepare("SELECT id, body FROM temp.scratch ORDER BY hex(body)") + .expect("prepare an unindexed sort"); + let mut rows = statement.query([]).expect("start sort"); + let first = rows.next().expect("sort first row").expect("nonempty sort"); + let mut seen = BTreeSet::new(); + let id = first.get::<_, i64>(0).expect("first sorted id"); + assert_eq!(first.get::<_, Vec>(1).expect("body"), payload(id)); + assert!(seen.insert(id)); + #[cfg(target_os = "linux")] + let sorter_files = sqlite_spills::new_since(&before); + #[cfg(target_os = "linux")] + if temp_store == 1 { + assert!( + sorter_files.values().sum::() > 1024 * 1024, + "the active statement must own a real unlinked sorter file" + ); + } + #[cfg(not(target_os = "linux"))] + let _ = temp_store; + collect_with_cross_thread_frees(); + while let Some(row) = rows.next().expect("read sorted rows") { + let id = row.get::<_, i64>(0).expect("sorted id"); + assert_eq!(row.get::<_, Vec>(1).expect("body"), payload(id)); + assert!(seen.insert(id), "sort must not duplicate rows"); + } + assert_eq!(seen, (0..ROW_COUNT).collect::>()); + drop(rows); + drop(statement); + #[cfg(target_os = "linux")] + sqlite_spills::assert_reclaimed(&sorter_files); +} + +#[cfg(target_os = "linux")] +mod sqlite_spills { + use std::collections::BTreeMap; + use std::io::ErrorKind; + use std::os::unix::fs::MetadataExt; + use std::path::Path; + + use tracedecay_maintenance::retention::footprint::census_sqlite_temp_spill; + + type Files = BTreeMap<(u64, u64), u64>; + + pub(super) fn snapshot() -> Files { + let mut files = Files::new(); + for entry in std::fs::read_dir("/proc/self/fd").expect("read process descriptors") { + let path = entry.expect("enumerate descriptor").path(); + let target = match std::fs::read_link(&path) { + Ok(target) => target, + Err(error) if error.kind() == ErrorKind::NotFound => continue, + Err(error) => panic!("inspect descriptor {path:?}: {error}"), + }; + if !target + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| { + name.starts_with("etilqs_") && name.ends_with(" (deleted)") + }) + { + continue; + } + let metadata = std::fs::metadata(&path).expect("stat live SQLite spill"); + assert!(metadata.is_file()); + assert_eq!(metadata.nlink(), 0); + // Stat, never open or duplicate the descriptor: observing lifetime + // must not itself keep the file alive or disturb SQLite locks. + files.insert((metadata.dev(), metadata.ino()), metadata.len()); + } + let census = census_sqlite_temp_spill(Path::new("/proc/self/fd"), &BTreeMap::new()) + .expect("production census can inspect this process"); + assert_eq!(census.unavailable_entry_count, 0); + assert_eq!(census.bytes, files.values().sum::()); + files + } + + pub(super) fn new_since(before: &Files) -> Files { + let mut files = snapshot(); + files.retain(|identity, _| !before.contains_key(identity)); + files + } + + pub(super) fn assert_reclaimed(owned: &Files) { + let remaining = snapshot(); + for identity in owned.keys() { + assert!( + !remaining.contains_key(identity), + "spill {identity:?} is still open" + ); + } + } +} From d5a9ef5e8c741eef529c5518851a88b5100fc182 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:53:55 -0800 Subject: [PATCH 3/5] fix(projection): verify canonical bodies before rebuild activation Refs #4152 Unchanged compacted foreign outputs were compared against their shortened placeholders instead of their full staged bodies. Hydrate canonical body candidates through the existing verified reader in bounded keyset pages, without restoring connection-local TEMP materialization or trusting only a stored hash. Preserve existing ownership and Hermes collision rules. Add a registered-SQLite regression for 129 matching canonical bodies, a later-page conflict, corrupt and missing persisted bodies, and rollback preserving committed messages and database integrity. The independent SQLite SQL/hydration probe passes these scenarios. Repository Bazel and native macOS execution remain pending. --- .../src/observation_projection/rebuild.rs | 339 ++++++++++++++---- 1 file changed, 275 insertions(+), 64 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 2eee7419d61..9dba419171a 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2324,10 +2324,7 @@ async fn activate_rebuild_sessions( .map_err(|error| storage("bind rebuilt child session spawn calls", error)) } -async fn prepare_rebuild_output_activation( - conn: &impl Executor, - generation: &str, -) -> ProjectionStoreResult<()> { +fn rebuild_output_preflight_query(has_cursor: bool) -> String { let message_conflicts = format!( "{} OR (active.provider <> 'hermes' AND ( @@ -2342,69 +2339,123 @@ async fn prepare_rebuild_output_activation( json_extract_expr(STAGED_MESSAGE_JSON_COLUMN, "text"), json_extract_expr(STAGED_MESSAGE_JSON_COLUMN, "metadata_json"), ); - // Validate against live indexed ownership while the old projector still owns - // its outputs. A long-lived writer must not materialize connection-local - // TEMP tables: they can keep unlinked SQLite files allocated between rebuilds. - let mut conflicts = conn - .query( - &format!( - "SELECT staged.output_provider, staged.output_message_id - FROM observation_projection_rebuild_messages AS staged - LEFT JOIN lcm_raw_messages AS active - ON active.provider = staged.output_provider - AND active.message_id = staged.output_message_id - WHERE staged.projector_version = ?1 AND staged.generation = ?2 + let canonical_body = "active.provider <> 'hermes' + AND active.storage_kind = 'inline' AND active.content IS NULL"; + let cursor_clause = if has_cursor { + "AND (staged.output_provider, staged.output_message_id) > (?3, ?4)" + } else { + "" + }; + format!( + "SELECT staged.output_provider, staged.output_message_id, + CASE WHEN {canonical_body} THEN 1 ELSE 0 END + FROM observation_projection_rebuild_messages AS staged + LEFT JOIN lcm_raw_messages AS active + ON active.provider = staged.output_provider + AND active.message_id = staged.output_message_id + WHERE staged.projector_version = ?1 AND staged.generation = ?2 + {cursor_clause} + AND ( + (active.message_id IS NULL + AND NOT EXISTS ( + SELECT 1 FROM observation_projection_rebuild_provenance AS staged_owner + WHERE staged_owner.projector_version = ?1 + AND staged_owner.generation = ?2 + AND staged_owner.message_created = 1 + AND staged_owner.output_provider = staged.output_provider + AND staged_owner.output_message_id = staged.output_message_id + )) + OR ( + active.message_id IS NOT NULL AND ( - (active.message_id IS NULL - AND NOT EXISTS ( - SELECT 1 FROM observation_projection_rebuild_provenance AS staged_owner - WHERE staged_owner.projector_version = ?1 - AND staged_owner.generation = ?2 - AND staged_owner.message_created = 1 - AND staged_owner.output_provider = staged.output_provider - AND staged_owner.output_message_id = staged.output_message_id - )) - OR ( - active.message_id IS NOT NULL - AND ( - NOT EXISTS ( - SELECT 1 FROM observation_projection_provenance AS current_owner - WHERE current_owner.projector_version = ?1 - AND current_owner.message_created = 1 - AND current_owner.output_provider = staged.output_provider - AND current_owner.output_message_id = staged.output_message_id - ) - OR EXISTS ( - SELECT 1 FROM observation_projection_provenance AS foreign_owner - WHERE foreign_owner.projector_version <> ?1 - AND foreign_owner.output_provider = staged.output_provider - AND foreign_owner.output_message_id = staged.output_message_id - ) - ) - AND ({message_conflicts}) + NOT EXISTS ( + SELECT 1 FROM observation_projection_provenance AS current_owner + WHERE current_owner.projector_version = ?1 + AND current_owner.message_created = 1 + AND current_owner.output_provider = staged.output_provider + AND current_owner.output_message_id = staged.output_message_id + ) + OR EXISTS ( + SELECT 1 FROM observation_projection_provenance AS foreign_owner + WHERE foreign_owner.projector_version <> ?1 + AND foreign_owner.output_provider = staged.output_provider + AND foreign_owner.output_message_id = staged.output_message_id ) ) - LIMIT 1" - ), - params![SESSION_MESSAGE_PROJECTOR_VERSION, generation], - ) - .await - .map_err(|error| storage("validate staged projection outputs", error))?; - if let Some(row) = conflicts - .next() - .await - .map_err(|error| storage("validate staged projection outputs", error))? - { - return Err(ProjectionStoreError::OutputCollision { - provider: row - .get(0) - .map_err(|error| storage("validate staged projection outputs", error))?, - message_id: row - .get(1) - .map_err(|error| storage("validate staged projection outputs", error))?, - }); + AND (({message_conflicts}) OR ({canonical_body})) + ) + ) + ORDER BY staged.output_provider, staged.output_message_id + LIMIT ?5" + ) +} + +async fn prepare_rebuild_output_activation( + conn: &impl Executor, + generation: &str, +) -> ProjectionStoreResult<()> { + // Keep ownership on the indexed durable tables, not connection-local TEMP + // storage. Only compacted bodies need hydration: their placeholders are not + // the text staged by the rebuild, and a stored hash alone cannot prove that + // the referenced bytes still exist and are intact. + let mut cursor: Option<(String, String)> = None; + loop { + let (after_provider, after_message) = match cursor.as_ref() { + Some((provider, message_id)) => (Some(provider.as_str()), Some(message_id.as_str())), + None => (None, None), + }; + let mut rows = conn + .query( + &rebuild_output_preflight_query(cursor.is_some()), + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + generation, + after_provider, + after_message, + REBUILD_PAGE_SIZE + ], + ) + .await + .map_err(|error| storage("validate staged projection outputs", error))?; + let mut candidates = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| storage("validate staged projection outputs", error))? + { + candidates.push(( + row.get::(0) + .map_err(|error| storage("validate staged projection outputs", error))?, + row.get::(1) + .map_err(|error| storage("validate staged projection outputs", error))?, + row.get::(2) + .map_err(|error| storage("validate staged projection outputs", error))? + != 0, + )); + } + // Hydrate at most one body at a time through the existing verified + // reader, after dropping the key-only page's cursor. Large bodies must + // not multiply into a transport-sized page or an unbounded Rust cache. + drop(rows); + let scanned = candidates.len(); + for (provider, message_id, canonical_body) in candidates { + if canonical_body { + let actual = read_message(conn, &provider, &message_id).await?; + let staged = read_staged_message(conn, generation, &provider, &message_id).await?; + if matches!((actual, staged), (Some(actual), Some(staged)) if actual == staged) { + cursor = Some((provider, message_id)); + continue; + } + } + return Err(ProjectionStoreError::OutputCollision { + provider, + message_id, + }); + } + if scanned < REBUILD_PAGE_SIZE as usize { + return Ok(()); + } } - Ok(()) } async fn activate_rebuild_messages( @@ -2578,7 +2629,7 @@ mod activation_tests { use super::{ REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION, activate_projection_rebuild_transaction, activate_rebuild_sessions, - prepare_rebuild_output_activation, projected_stored_message, + prepare_rebuild_output_activation, projected_stored_message, read_message, reconcile_overlapping_rebuild_sessions, write_staged_message, }; use crate::tests::harness::RegisteredGlobalDbHarness; @@ -2939,6 +2990,166 @@ mod activation_tests { ); } + #[asupersync::test] + async fn output_preflight_verifies_canonical_bodies_past_one_page() { + const GENERATION: &str = "generation.canonical-output-preflight"; + let harness = RegisteredGlobalDbHarness::open("canonical-output-preflight").await; + let active = session(None, None); + assert!(harness.registered.upsert_session(&active).await); + let body = "Canonical retained message body.\n".repeat(256); + assert!(body.len() > tracedecay_store::INLINE_BODY_BYTES); + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + let mut expected = Vec::new(); + for index in 0..=REBUILD_PAGE_SIZE { + let message = SessionMessageRecord { + provider: active.provider.clone(), + session_id: active.session_id.clone(), + message_id: format!("message.canonical.{index:03}"), + role: "assistant".to_owned(), + timestamp: Some(1), + ordinal: index, + text: body.clone(), + kind: None, + model: None, + tool_names: None, + source_path: None, + source_offset: None, + metadata_json: None, + }; + tracedecay_lcm::raw::upsert_projection_raw_message( + &transaction, + &message, + Some("/foreign"), + ) + .await + .unwrap(); + expected.push(projected_stored_message(&message).unwrap()); + } + transaction.commit().await.unwrap(); + + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + stage(&transaction, GENERATION, &active).await; + for message in &expected { + write_staged_message(&transaction, GENERATION, message) + .await + .unwrap(); + } + let mut rows = transaction + .query( + "SELECT COUNT(*) FROM lcm_raw_messages + WHERE provider = 'cursor' AND message_id LIKE 'message.canonical.%' + AND content IS NULL AND storage_kind = 'inline'", + (), + ) + .await + .unwrap(); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + REBUILD_PAGE_SIZE + 1, + "the fixture must exercise canonical bodies, not short inline text" + ); + drop(rows); + prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .expect("matching canonical bodies must not collide with their placeholders"); + + // Keep metadata identical and change only a tail beyond the snippet. + // A placeholder-only or first-page-only check must not accept this last row. + let last = expected.last().unwrap(); + let mut changed = last.clone(); + changed.text.push_str("different retained tail"); + assert_eq!( + tracedecay_lcm::retrieval_content::derived_text_for_snippet(&changed.text), + tracedecay_lcm::retrieval_content::derived_text_for_snippet(&last.text) + ); + write_staged_message(&transaction, GENERATION, &changed) + .await + .unwrap(); + assert!(matches!( + prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .unwrap_err(), + ProjectionStoreError::OutputCollision { provider, message_id } + if provider == last.provider && message_id == last.message_id + )); + write_staged_message(&transaction, GENERATION, last) + .await + .unwrap(); + + // Damage the actual persisted body behind an unchanged hash, then + // remove it entirely. Neither state is permission to overwrite or + // repair foreign data; each must fail through the verified reader. + let hash = tracedecay_lcm::retrieval_content::projected_content_hash(&expected[0].text); + // Defer (do not disable) the fixture's foreign keys so a missing + // body can be inspected before rollback. A commit would still refuse + // the missing parent; no ownership guards or schema are removed. + transaction + .execute_batch("PRAGMA defer_foreign_keys = ON") + .await + .unwrap(); + for mutation in [ + "UPDATE session_canonical_bodies SET body = X'00' WHERE content_hash = ?1", + "DELETE FROM session_canonical_bodies WHERE content_hash = ?1", + ] { + transaction + .execute_batch("SAVEPOINT canonical_body_fault") + .await + .unwrap(); + assert_eq!( + transaction + .execute(mutation, params![hash.as_str()]) + .await + .unwrap(), + 1, + "the corruption fixture must alter a real stored body" + ); + assert!(matches!( + prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .unwrap_err(), + ProjectionStoreError::Storage { .. } + )); + transaction + .execute_batch("ROLLBACK TO canonical_body_fault; RELEASE canonical_body_fault;") + .await + .unwrap(); + } + prepare_rebuild_output_activation(&transaction, GENERATION) + .await + .expect("restoring the body must restore compatibility"); + transaction.rollback().await.unwrap(); + + let snapshot = harness.registered.read_snapshot().await.unwrap(); + for message in &expected { + assert_eq!( + read_message(&snapshot, &message.provider, &message.message_id) + .await + .unwrap() + .as_ref(), + Some(message), + "refused rebuilds must preserve every committed body and its metadata" + ); + } + let mut rows = snapshot + .query( + "SELECT COUNT(*) FROM observation_projection_rebuilds WHERE generation = ?1", + params![GENERATION], + ) + .await + .unwrap(); + assert_eq!(rows.next().await.unwrap().unwrap().get::(0).unwrap(), 0); + drop(rows); + let mut rows = snapshot.query("PRAGMA integrity_check", ()).await.unwrap(); + let mut integrity = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + integrity.push(row.get::(0).unwrap()); + } + assert_eq!(integrity, vec!["ok".to_owned()]); + drop(rows); + let mut foreign_keys = snapshot.query("PRAGMA foreign_key_check", ()).await.unwrap(); + assert!(foreign_keys.next().await.unwrap().is_none()); + } + #[asupersync::test] async fn activation_names_the_session_field_instead_of_a_message_collision() { let harness = RegisteredGlobalDbHarness::open("session-collision-field").await; From a9c68c2ba2b203e2a1e7d005a409f89b81c41274 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:47:27 -0800 Subject: [PATCH 4/5] fix(ci): format SQLite TEMP regressions Refs #4152 Match the Bazel rustfmt check on both issue-specific files without changing the production preflight or altering test assertions. --- crates/tracedecay-cli/tests/c_library_allocator.rs | 8 ++++---- .../src/observation_projection/rebuild.rs | 10 ++++++++-- 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/crates/tracedecay-cli/tests/c_library_allocator.rs b/crates/tracedecay-cli/tests/c_library_allocator.rs index d0d7d506d68..9d155302575 100644 --- a/crates/tracedecay-cli/tests/c_library_allocator.rs +++ b/crates/tracedecay-cli/tests/c_library_allocator.rs @@ -230,7 +230,9 @@ fn assert_temp_integrity_and_reclamation(temp_store: i64) { .expect("release unused scratch without deleting durable data"); collect_with_cross_thread_frees(); assert_rows(&connection, "main.durable"); - connection.close().expect("close the owning SQLite connection"); + connection + .close() + .expect("close the owning SQLite connection"); #[cfg(target_os = "linux")] sqlite_spills::assert_reclaimed(&temp_files); @@ -301,9 +303,7 @@ mod sqlite_spills { if !target .file_name() .and_then(|name| name.to_str()) - .is_some_and(|name| { - name.starts_with("etilqs_") && name.ends_with(" (deleted)") - }) + .is_some_and(|name| name.starts_with("etilqs_") && name.ends_with(" (deleted)")) { continue; } diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 9dba419171a..beb861a9d9a 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -3137,7 +3137,10 @@ mod activation_tests { ) .await .unwrap(); - assert_eq!(rows.next().await.unwrap().unwrap().get::(0).unwrap(), 0); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + 0 + ); drop(rows); let mut rows = snapshot.query("PRAGMA integrity_check", ()).await.unwrap(); let mut integrity = Vec::new(); @@ -3146,7 +3149,10 @@ mod activation_tests { } assert_eq!(integrity, vec!["ok".to_owned()]); drop(rows); - let mut foreign_keys = snapshot.query("PRAGMA foreign_key_check", ()).await.unwrap(); + let mut foreign_keys = snapshot + .query("PRAGMA foreign_key_check", ()) + .await + .unwrap(); assert!(foreign_keys.next().await.unwrap().is_none()); } From 1569c5b78981d0595c93df2d983ec5ac360fc1a5 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:52:34 -0800 Subject: [PATCH 5/5] test(automation): bind scheduler fixtures to backend identities Repair three hosted automation_runner_test failures by supplying the configured executable identity required by the production prefilter. Use the same identity for the prefilter and retained runner, preserving manual-schedule, exact-ledger, live-lock and lock-race assertions. Add a production-prefilter regression for unconfigured refusal, deduplication and same-occurrence recovery after binding, with zero backend calls and no effectful terminal. Do not relax backend admission, change timeouts, or alter the graph publication changes. Refs #4532 --- .../tests/automation_runner_test/jobs.rs | 124 ++++++++++++++++-- 1 file changed, 110 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 740283d56e8..7342786381b 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -86,6 +86,7 @@ fn enabled_job_config() -> AutomationSettingsV1 { struct ContentBackend { calls: AtomicUsize, content: &'static str, + executable: Option, } impl ContentBackend { @@ -93,9 +94,15 @@ impl ContentBackend { Self { calls: AtomicUsize::new(0), content, + executable: None, } } + fn with_executable(mut self, executable: &Path) -> Self { + self.executable = Some(executable.to_path_buf()); + self + } + fn calls(&self) -> usize { self.calls.load(Ordering::SeqCst) } @@ -126,7 +133,7 @@ impl AgentTaskBackend for ContentBackend { } fn executable(&self) -> Option<&std::path::Path> { - None + self.executable.as_deref() } } @@ -988,6 +995,10 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let mut job = sample_job("sched-job"); job.schedule = Some("manual".to_string()); let config = enabled_job_config(); + // Reach the schedule gate with a bound identity, not the earlier + // backend-unconfigured refusal. This gate never invokes the backend. + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); // Manual schedule: every scheduler tick of the same occurrence reaches // the prefilter, which records the skip once. @@ -996,7 +1007,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let run = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1019,7 +1030,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let repeated = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1031,6 +1042,76 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { repeated.ledger_record, records[0], "an immediate repeat skip must return the exact durable prior row" ); + assert_eq!(backend.calls(), 0); +} + +#[tokio::test] +async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() { + let temp = tempdir().unwrap(); + let dashboard_root = temp.path().join("dashboard"); + let job = sample_job("backend-recovery"); + let config = enabled_job_config(); + let occurrence = "backend-recovery-occurrence"; + let backend = ContentBackend::new("unused"); + + let first = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("a due job still requires a configured backend"); + assert_eq!(first.report["reason"], json!("backend_unconfigured")); + assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped); + assert_ne!(first.run_id, occurrence); + let repeated = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(repeated.ledger_record, first.ledger_record); + + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = backend.with_executable(executable.path()); + assert!( + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), + "binding the backend must admit the same due occurrence without cooldown" + ); + assert_eq!(backend.calls(), 0); + assert_eq!( + load_run_records(&dashboard_root, 10).await.unwrap(), + vec![first.ledger_record], + "advisory admission must not append an effectful terminal" + ); + assert!( + tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded( + &dashboard_root, + occurrence, + ) + .await + .unwrap() + .is_none() + ); } #[tokio::test] @@ -1206,15 +1287,22 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() { let config = enabled_job_config(); let occurrence = "locked-prefilter-occurrence"; - let prefilter = - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .expect("the live job lock must produce a diagnostic"); + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); + let prefilter = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("the live job lock must produce a diagnostic"); assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active")); assert_ne!(prefilter.run_id, occurrence); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, @@ -1262,12 +1350,21 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { job.schedule = Some("every 1h".to_owned()); let config = enabled_job_config(); let occurrence = "prefilter-gap-occurrence"; + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); assert!( - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .is_none(), + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), "a new interval job is advisory-due before effect admission" ); let _raced_in = @@ -1276,7 +1373,6 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .unwrap() .expect("the advisory prefilter must release its reservation lock"); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root,