From 0e979f952a904906a00cc65a52855348dd8a62d3 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:20:49 -0800 Subject: [PATCH 1/7] fix(projection): validate rebuild outputs without TEMP cache Refs #4152 --- .../src/observation_projection/rebuild.rs | 197 ++++++++++++------ 1 file changed, 138 insertions(+), 59 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..19058685906 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,7 +2578,8 @@ mod activation_tests { use super::{ REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION, activate_projection_rebuild_transaction, activate_rebuild_sessions, - reconcile_overlapping_rebuild_sessions, write_staged_message, + prepare_rebuild_output_activation, reconcile_overlapping_rebuild_sessions, + write_staged_message, }; use crate::tests::harness::RegisteredGlobalDbHarness; use tracedecay_runtime_core::db::engine::{Executor, params}; @@ -2796,6 +2766,115 @@ 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 transaction = harness.registered.begin_write_transaction().await.unwrap(); + stage(&transaction, GENERATION, &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, + }; + tracedecay_lcm::raw::upsert_projection_raw_message( + &transaction, + &foreign, + Some("/foreign"), + ) + .await + .unwrap(); + let changed = SessionMessageRecord { + text: "overwritten body".to_owned(), + ..foreign.clone() + }; + write_staged_message(&transaction, GENERATION, &changed) + .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.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. + write_staged_message(&transaction, GENERATION, &foreign) + .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() + }; + write_staged_message(&transaction, GENERATION, &missing) + .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); + + let mut rows = transaction + .query( + "SELECT content FROM lcm_raw_messages + WHERE provider = ?1 AND message_id = ?2", + params![foreign.provider.as_str(), foreign.message_id.as_str()], + ) + .await + .unwrap(); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + foreign.text, + "a denied activation must not change a protected durable body" + ); + drop(rows); + + let mut integrity = transaction.query("PRAGMA main.integrity_check", ()).await.unwrap(); + assert_eq!( + integrity.next().await.unwrap().unwrap().get::(0).unwrap(), + "ok", + "the real registered SQLite store must remain structurally sound" + ); + } + #[asupersync::test] async fn activation_names_the_session_field_instead_of_a_message_collision() { let harness = RegisteredGlobalDbHarness::open("session-collision-field").await; From 61eefab3f0f039b7653c501167c424648d1215b9 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:21:27 -0800 Subject: [PATCH 2/7] test(projection): check durable SQLite integrity on preflight Refs #4152 --- .../src/observation_projection/rebuild.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 19058685906..64c04bc996d 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2867,7 +2867,10 @@ mod activation_tests { ); drop(rows); - let mut integrity = transaction.query("PRAGMA main.integrity_check", ()).await.unwrap(); + let mut integrity = transaction + .query("PRAGMA integrity_check", ()) + .await + .unwrap(); assert_eq!( integrity.next().await.unwrap().unwrap().get::(0).unwrap(), "ok", From 82b126a6057e37959356f56883fe7d8cef029cb2 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:33:33 -0800 Subject: [PATCH 3/7] style(projection): format output preflight regression --- .../src/observation_projection/rebuild.rs | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 64c04bc996d..223e635a151 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2861,7 +2861,12 @@ mod activation_tests { .await .unwrap(); assert_eq!( - rows.next().await.unwrap().unwrap().get::(0).unwrap(), + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), foreign.text, "a denied activation must not change a protected durable body" ); @@ -2872,7 +2877,13 @@ mod activation_tests { .await .unwrap(); assert_eq!( - integrity.next().await.unwrap().unwrap().get::(0).unwrap(), + integrity + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), "ok", "the real registered SQLite store must remain structurally sound" ); From c178ff556d225be94ba07825342d51aba389d449 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 18:09:04 -0800 Subject: [PATCH 4/7] test(projection): prove persisted output and TEMP pages survive rollback Refs #4152 --- .../src/observation_projection/rebuild.rs | 58 ++++++++++++++----- 1 file changed, 44 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 223e635a151..68668301f71 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2772,8 +2772,6 @@ mod activation_tests { let harness = RegisteredGlobalDbHarness::open("output-preflight-no-temp").await; let active = session(None, None); assert!(harness.registered.upsert_session(&active).await); - let transaction = harness.registered.begin_write_transaction().await.unwrap(); - stage(&transaction, GENERATION, &active).await; let foreign = SessionMessageRecord { provider: active.provider.clone(), @@ -2790,6 +2788,7 @@ mod activation_tests { source_offset: None, metadata_json: None, }; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); tracedecay_lcm::raw::upsert_projection_raw_message( &transaction, &foreign, @@ -2797,6 +2796,12 @@ mod activation_tests { ) .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() @@ -2852,30 +2857,55 @@ mod activation_tests { ); drop(rows); - let mut rows = transaction + // A dropped TEMP b-tree can disappear from sqlite_temp_master while + // its pages remain allocated on this writer. Measure actual SQLite + // TEMP database pages too, not just the surviving schema. + let mut pages = transaction + .query("PRAGMA temp.page_count", ()) + .await + .unwrap(); + assert_eq!( + pages.next().await.unwrap().unwrap().get::(0).unwrap(), + 0, + "preflight must not allocate TEMP b-tree pages on the long-lived writer" + ); + drop(pages); + + transaction.rollback().await.unwrap(); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot .query( - "SELECT content FROM lcm_raw_messages + "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!( - rows.next() - .await - .unwrap() - .unwrap() - .get::(0) - .unwrap(), + row.get::(0).unwrap(), foreign.text, - "a denied activation must not change a protected durable body" + "a denied rebuild must not replace a committed foreign body" ); + assert_eq!(row.get::(1).unwrap(), "/foreign"); drop(rows); - let mut integrity = transaction - .query("PRAGMA integrity_check", ()) + 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() @@ -2885,7 +2915,7 @@ mod activation_tests { .get::(0) .unwrap(), "ok", - "the real registered SQLite store must remain structurally sound" + "the committed SQLite store must remain structurally sound after rollback" ); } From b14ced87a74cdd78b3b3dd17d35da63681cb393c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:16:19 -0800 Subject: [PATCH 5/7] test(projection): compare TEMP page allocation to preflight baseline Refs #4152 --- .../src/observation_projection/rebuild.rs | 25 ++++++++++++++++--- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 68668301f71..d5f88c5ded4 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2809,6 +2809,22 @@ mod activation_tests { write_staged_message(&transaction, GENERATION, &changed) .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(); @@ -2858,16 +2874,17 @@ mod activation_tests { drop(rows); // A dropped TEMP b-tree can disappear from sqlite_temp_master while - // its pages remain allocated on this writer. Measure actual SQLite - // TEMP database pages too, not just the surviving schema. + // 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(), - 0, - "preflight must not allocate TEMP b-tree pages on the long-lived writer" + baseline_page_count, + "repeated preflight checks must not grow SQLite TEMP database pages" ); drop(pages); From 4025e7763c6a7ed3a9973487abea1580e1ef7436 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:29:26 -0800 Subject: [PATCH 6/7] test(projection): stage normalized stored messages in preflight regression Refs #4152 --- .../src/observation_projection/rebuild.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index d5f88c5ded4..b391d21be53 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2806,7 +2806,8 @@ mod activation_tests { text: "overwritten body".to_owned(), ..foreign.clone() }; - write_staged_message(&transaction, GENERATION, &changed) + 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 @@ -2836,7 +2837,8 @@ mod activation_tests { // Matching foreign bodies are compatible; rejecting every unowned row // would also avoid spills, but would break legitimate rebuilds. - write_staged_message(&transaction, GENERATION, &foreign) + let compatible_stored = projected_stored_message(&foreign).unwrap(); + write_staged_message(&transaction, GENERATION, &compatible_stored) .await .unwrap(); prepare_rebuild_output_activation(&transaction, GENERATION) @@ -2847,7 +2849,8 @@ mod activation_tests { message_id: "message.not-created".to_owned(), ..foreign.clone() }; - write_staged_message(&transaction, GENERATION, &missing) + 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) From 976e8586f8d441122716c8c8725089e2ecebdb71 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:47:02 -0800 Subject: [PATCH 7/7] fix(projection): import normalized message helper in regression Refs #4152 --- .../src/observation_projection/rebuild.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index b391d21be53..2eee7419d61 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -2578,8 +2578,8 @@ mod activation_tests { use super::{ REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION, activate_projection_rebuild_transaction, activate_rebuild_sessions, - prepare_rebuild_output_activation, reconcile_overlapping_rebuild_sessions, - write_staged_message, + prepare_rebuild_output_activation, projected_stored_message, + reconcile_overlapping_rebuild_sessions, write_staged_message, }; use crate::tests::harness::RegisteredGlobalDbHarness; use tracedecay_runtime_core::db::engine::{Executor, params};