diff --git a/crates/tracedecay-runtime-core/src/db/memory_v2/schema/baseline.rs b/crates/tracedecay-runtime-core/src/db/memory_v2/schema/baseline.rs index 6f6842049..abcc5034b 100644 --- a/crates/tracedecay-runtime-core/src/db/memory_v2/schema/baseline.rs +++ b/crates/tracedecay-runtime-core/src/db/memory_v2/schema/baseline.rs @@ -90,6 +90,27 @@ pub(super) const BASELINE_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS memory_v2_f SELECT RAISE(ABORT, 'memory_v2 assertion payloads are immutable'); END; + CREATE TABLE IF NOT EXISTS memory_v2_assertion_payload_purges ( + assertion_id TEXT NOT NULL, + fact_id TEXT NOT NULL, + owner_kind TEXT NOT NULL, + project_id TEXT NOT NULL, + payload_reference_json TEXT NOT NULL CHECK(json_valid(payload_reference_json)), + detector_revision TEXT NOT NULL CHECK(length(detector_revision) > 0), + purge_reason TEXT NOT NULL CHECK(purge_reason = 'detector_flagged'), + PRIMARY KEY(assertion_id, fact_id, owner_kind, project_id), + FOREIGN KEY(assertion_id, fact_id, owner_kind, project_id) + REFERENCES memory_v2_assertions(assertion_id, fact_id, owner_kind, project_id) + ); + CREATE TRIGGER IF NOT EXISTS memory_v2_assertion_payload_purges_no_update + BEFORE UPDATE ON memory_v2_assertion_payload_purges BEGIN + SELECT RAISE(ABORT, 'memory_v2 assertion payload purge receipts are immutable'); + END; + CREATE TRIGGER IF NOT EXISTS memory_v2_assertion_payload_purges_no_delete + BEFORE DELETE ON memory_v2_assertion_payload_purges BEGIN + SELECT RAISE(ABORT, 'memory_v2 assertion payload purge receipts are immutable'); + END; + CREATE TABLE IF NOT EXISTS memory_v2_evidence ( evidence_id TEXT NOT NULL, fact_id TEXT NOT NULL, diff --git a/crates/tracedecay-runtime-core/src/db/memory_v2/tests.rs b/crates/tracedecay-runtime-core/src/db/memory_v2/tests.rs index 213e3681c..d5061f712 100644 --- a/crates/tracedecay-runtime-core/src/db/memory_v2/tests.rs +++ b/crates/tracedecay-runtime-core/src/db/memory_v2/tests.rs @@ -56,6 +56,7 @@ async fn fresh_store_carries_only_the_final_memory_shape() { .await, [ "memory_v2_assertion_evidence", + "memory_v2_assertion_payload_purges", "memory_v2_assertion_payloads", "memory_v2_assertion_payloads_fts", "memory_v2_assertion_payloads_fts_config", diff --git a/crates/tracedecay-runtime-core/src/db/migrations/tests.rs b/crates/tracedecay-runtime-core/src/db/migrations/tests.rs index 302d75b26..a086a9aed 100644 --- a/crates/tracedecay-runtime-core/src/db/migrations/tests.rs +++ b/crates/tracedecay-runtime-core/src/db/migrations/tests.rs @@ -394,6 +394,7 @@ async fn fresh_creation_installs_every_stage_of_the_final_shape() { .await, [ "memory_v2_assertion_evidence", + "memory_v2_assertion_payload_purges", "memory_v2_assertion_payloads", "memory_v2_assertion_payloads_fts", "memory_v2_assertion_payloads_fts_config", diff --git a/crates/tracedecay-runtime-core/src/store/memory/crud/commit.rs b/crates/tracedecay-runtime-core/src/store/memory/crud/commit.rs index db0858590..57e108f15 100644 --- a/crates/tracedecay-runtime-core/src/store/memory/crud/commit.rs +++ b/crates/tracedecay-runtime-core/src/store/memory/crud/commit.rs @@ -4,6 +4,9 @@ use super::super::primitives::{ COMMIT_OPERATION, OwnerKey, QUERY_OPERATION, row_exists, row_i64, row_optional_string, row_string, storage_error, storage_message, to_json, }; +use super::super::privacy_purge::{ + assertion_payload_is_explicitly_purged_tx, purge_superseded_payloads_for_fact_tx, +}; use super::{ CommitAttempt, ensure_event_references, ensure_fact_identity, event_exists, event_matches, insert_event, payload_is_purged_projection, publish_current_projection, receipt_outcome, @@ -94,6 +97,15 @@ pub(super) async fn commit_fact_tx( insert_event(transaction, &owner, event).await?; } publish_current_projection(transaction, &owner, batch).await?; + if let Some(assertion) = batch.assertion() { + purge_superseded_payloads_for_fact_tx( + transaction, + &owner, + batch.fact_id(), + assertion.assertion_id(), + ) + .await?; + } Ok(CommitAttempt { outcome: receipt_outcome(transaction, &owner, batch, false).await?, @@ -711,7 +723,12 @@ async fn assertion_matches( == to_json(assertion.payload(), "serialize assertion payload")? && row_string(&row, 1, QUERY_OPERATION)? == assertion.payload().content() } - None => payload_is_purged_projection(transaction, owner, assertion.fact_id()).await?, + // A missing payload row is consistent only with a terminal fact-wide + // purge or an immutable assertion-specific detector purge receipt. + None => { + payload_is_purged_projection(transaction, owner, assertion.fact_id()).await? + || assertion_payload_is_explicitly_purged_tx(transaction, owner, assertion).await? + } }; if !payload_matches { return Ok(false); diff --git a/crates/tracedecay-runtime-core/src/store/memory/crud/lineage.rs b/crates/tracedecay-runtime-core/src/store/memory/crud/lineage.rs index 1e2929a76..292414669 100644 --- a/crates/tracedecay-runtime-core/src/store/memory/crud/lineage.rs +++ b/crates/tracedecay-runtime-core/src/store/memory/crud/lineage.rs @@ -5,6 +5,7 @@ use super::super::primitives::{ payload_access_label, requires_payload_purge, row_exists, row_f64, row_i64, row_optional_string, row_string, storage_error, storage_message, to_json, }; +use super::super::privacy_purge::assertion_payload_exists_tx; use super::DEFAULT_TRUST; use crate::db::DatabaseMemoryTransaction as Transaction; use crate::db::engine::params; @@ -67,6 +68,14 @@ pub(super) async fn ensure_event_references( "lineage assertion reference is missing", )); } + if !assertion_payload_exists_tx(transaction, owner, event.fact_id(), assertion_id) + .await? + { + return Err(storage_message( + COMMIT_OPERATION, + "assertion without an available payload cannot be activated", + )); + } } FactLineageEventKindV1::TrustChanged { evidence_ids, .. } => { ensure_event_evidence(transaction, owner, event.fact_id(), evidence_ids).await?; diff --git a/crates/tracedecay-runtime-core/src/store/memory/crud/mod.rs b/crates/tracedecay-runtime-core/src/store/memory/crud/mod.rs index 71880edad..e9d43ccaf 100644 --- a/crates/tracedecay-runtime-core/src/store/memory/crud/mod.rs +++ b/crates/tracedecay-runtime-core/src/store/memory/crud/mod.rs @@ -41,7 +41,7 @@ use self::project::active_fact_count_tx; pub(super) use self::project::{ commit_batch_tx, find_project_memory_fact_by_content_digest_controlled_tx, find_project_memory_fact_by_content_digest_tx, get_project_memory_fact_controlled_tx, - initial_batch, list_project_memory_facts_controlled_tx, payload_metadata, + initial_batch, list_project_memory_facts_controlled_tx, payload_material, payload_metadata, project_memory_fact_history_controlled_tx, sanitize_payload, verified_payload, }; pub(super) use self::queries::{ diff --git a/crates/tracedecay-runtime-core/src/store/memory/crud/project.rs b/crates/tracedecay-runtime-core/src/store/memory/crud/project.rs index 852b8d244..7d23235ef 100644 --- a/crates/tracedecay-runtime-core/src/store/memory/crud/project.rs +++ b/crates/tracedecay-runtime-core/src/store/memory/crud/project.rs @@ -415,7 +415,7 @@ pub(in crate::store::memory) fn verified_payload( payload_from_parts(payload, category, receipt) } -fn payload_material( +pub(in crate::store::memory) fn payload_material( content: &str, category: FactCategoryV1, tags: &[String], diff --git a/crates/tracedecay-runtime-core/src/store/memory/crud/tests.rs b/crates/tracedecay-runtime-core/src/store/memory/crud/tests.rs index 17be2da58..7d35b4725 100644 --- a/crates/tracedecay-runtime-core/src/store/memory/crud/tests.rs +++ b/crates/tracedecay-runtime-core/src/store/memory/crud/tests.rs @@ -7,10 +7,11 @@ use crate::store::memory::{DatabaseFactStore, FactWriteControl}; use serde_json::{Value, json}; use tempfile::{TempDir, tempdir}; use tracedecay_domain::{ - Confidence, DomainError, FactCategoryV1, FactEventId, FactOwnerV1, LocatorDigest, ProvenanceId, + Confidence, DomainError, FactCategoryV1, FactEventId, FactLineageEventKindV1, + FactLineageEventV1, FactOwnerV1, LocatorDigest, ProvenanceId, UtcMicros, }; use tracedecay_store::{ - FactReadControl, FactStoreError, ProjectMemoryFactAddCommandV1, + FactReadControl, FactStore, FactStoreError, FactWriteBatch, ProjectMemoryFactAddCommandV1, ProjectMemoryFactAddDispositionV1, ProjectMemoryFactAddMaterialV1, ProjectMemoryFactAddOutcomeV1, ProjectMemoryFactContentDigestQueryV1, ProjectMemoryFactFeedbackActionV1, ProjectMemoryFactFeedbackCommandV1, @@ -186,6 +187,141 @@ async fn exact_update_replay_returns_the_same_fact_delta_and_commit_receipt() { assert_eq!(committed.commit_receipt(), replayed.commit_receipt()); } +/// The correction-time payload purge is scoped to detector findings: a clean +/// superseded payload row must survive an ordinary update so as-of reads keep +/// serving the fact's history. +#[tokio::test] +async fn update_retains_clean_superseded_payload_rows_for_as_of_reads() { + let (_directory, database) = database().await; + let store = DatabaseFactStore::new(&database); + let control = write_control(); + let target = added_target(&store, &control).await; + + store + .update_project_memory_fact( + update_command( + target.clone(), + "operation.crud-clean-supersede", + "The canonical update keeps clean history readable.", + 0.8, + ), + &control, + ) + .await + .expect("commit clean canonical update"); + + let mut rows = database + .read_connection() + .query( + "SELECT COUNT(*) FROM memory_v2_assertion_payloads WHERE fact_id = ?1", + [target.fact_id().as_str()], + ) + .await + .expect("count at-rest payload rows"); + let retained: i64 = rows + .next() + .await + .expect("read payload row count") + .expect("payload row count is present") + .get(0) + .expect("payload row count is an integer"); + assert_eq!( + retained, 2, + "a clean superseded payload row must stay readable for as-of history" + ); +} + +#[tokio::test] +async fn missing_superseded_payload_without_purge_receipt_cannot_be_reactivated() { + let (_directory, database) = database().await; + let store = DatabaseFactStore::new(&database); + let control = write_control(); + let target = added_target(&store, &control).await; + let ProjectMemoryFactProjectionV1::Available(original) = store + .get_project_memory_fact(target.clone(), &read_control()) + .await + .expect("load original fact") + .expect("original fact exists") + else { + panic!("original fact must be available"); + }; + let original_assertion_id = original.active_assertion_id().clone(); + + let updated = store + .update_project_memory_fact( + update_command( + target.clone(), + "operation.crud-missing-payload-supersede", + "The active assertion remains clean and available.", + 0.8, + ), + &control, + ) + .await + .expect("commit clean successor"); + let ProjectMemoryFactProjectionV1::Available(current) = updated.fact() else { + panic!("clean successor must be available"); + }; + + database + .writer_connection("simulate missing superseded payload") + .await + .expect("database writer") + .execute( + "DELETE FROM memory_v2_assertion_payloads + WHERE assertion_id = ?1 AND fact_id = ?2", + params![original_assertion_id.as_str(), target.fact_id().as_str()], + ) + .await + .expect("simulate a corrupt missing clean historical payload"); + + let reactivation_time = UtcMicros( + current + .projected_as_of() + .0 + .checked_add(1) + .expect("reactivation time"), + ); + let event = FactLineageEventV1::new( + target.fact_id().clone(), + target.owner().clone(), + FactLineageEventKindV1::AssertionRecorded { + assertion_id: original_assertion_id, + }, + reactivation_time, + None, + ) + .expect("reactivation event"); + let batch = FactWriteBatch::new( + target.fact_id().clone(), + target.owner().clone(), + None, + vec![event], + Vec::new(), + Vec::new(), + Some(current.last_event_id().clone()), + ) + .expect("reactivation batch"); + let error = store + .commit_fact(batch, &control) + .await + .expect_err("an assertion with a missing payload must not reactivate"); + assert!( + matches!(error, FactStoreError::Storage { .. }), + "unexpected refusal: {error}" + ); + + let ProjectMemoryFactProjectionV1::Available(after) = store + .get_project_memory_fact(target, &read_control()) + .await + .expect("reload current fact") + .expect("current fact exists") + else { + panic!("clean successor must remain available"); + }; + assert_eq!(after.active_assertion_id(), current.active_assertion_id()); +} + #[tokio::test] async fn reused_update_operation_with_changed_patch_conflicts_without_mutation() { let (_directory, database) = database().await; diff --git a/crates/tracedecay-runtime-core/src/store/memory/mod.rs b/crates/tracedecay-runtime-core/src/store/memory/mod.rs index a7e9b783a..9a5893c79 100644 --- a/crates/tracedecay-runtime-core/src/store/memory/mod.rs +++ b/crates/tracedecay-runtime-core/src/store/memory/mod.rs @@ -29,7 +29,8 @@ use tracedecay_store::{ ProjectMemoryFactRetrievalOutcomeV1, ProjectMemoryFactSearchPageV1, ProjectMemoryFactSearchQuery, ProjectMemoryFactStore, ProjectMemoryFactUpdateCommandV1, ProjectMemoryFactUpdateOutcomeV1, ProjectMemoryGraphPageV1, ProjectMemoryGraphQueryV1, - ProjectMemoryGraphStore, ProjectMemoryMemoryStatusV1, RetrievalAnchorQuery, StoredFactV1, + ProjectMemoryGraphStore, ProjectMemoryMemoryStatusV1, ProjectMemoryPrivacyPurgeCursorV1, + ProjectMemoryPrivacyPurgeReceiptV1, RetrievalAnchorQuery, StoredFactV1, }; use automatic_facts::{ @@ -56,6 +57,7 @@ use envelope::finish_read_snapshot; use primitives::{ COMMIT_OPERATION, QUERY_OPERATION, ensure_project_memory_read_active, storage_error, }; +use privacy_purge::purge_superseded_payloads_for_owner_tx; use search::{ find_project_memory_contradictions_tx, probe_project_memory_facts_tx, reason_project_memory_facts_tx, record_project_memory_fact_retrieval_tx, @@ -81,6 +83,7 @@ mod graph_reconciliation_tests; #[cfg(test)] mod graph_tests; mod primitives; +mod privacy_purge; mod projection; mod runtime; mod scoring; @@ -309,6 +312,31 @@ impl FactStore for DatabaseFactStore<'_> { } impl ProjectMemoryFactStore for DatabaseFactStore<'_> { + async fn purge_project_memory_superseded_payloads( + &self, + owner: FactOwnerV1, + after: Option, + limit: usize, + write_control: &FactWriteControl, + ) -> FactStoreResult { + self.project_memory_write( + write_control, + |_| false, + move |transaction| { + Box::pin(async move { + purge_superseded_payloads_for_owner_tx( + transaction, + &owner, + after.as_ref(), + limit, + ) + .await + }) + }, + ) + .await + } + async fn list_project_memory_facts( &self, query: ProjectMemoryFactListQueryV1, @@ -863,6 +891,12 @@ impl FactStore for ProjectFactStore<'_> { impl ProjectMemoryFactStore for ProjectFactStore<'_> { delegate_fact_store_methods! { + fn purge_project_memory_superseded_payloads( + owner: FactOwnerV1, + after: Option, + limit: usize, + write_control: &FactWriteControl, + ) -> FactStoreResult; fn list_project_memory_facts( query: ProjectMemoryFactListQueryV1, read_control: &FactReadControl, diff --git a/crates/tracedecay-runtime-core/src/store/memory/privacy_purge.rs b/crates/tracedecay-runtime-core/src/store/memory/privacy_purge.rs new file mode 100644 index 000000000..69a705a11 --- /dev/null +++ b/crates/tracedecay-runtime-core/src/store/memory/privacy_purge.rs @@ -0,0 +1,362 @@ +//! Canonical at-rest purge authority for superseded assertion payloads. + +use crate::db::DatabaseMemoryTransaction as Transaction; +use crate::db::engine::params; +use crate::privacy::{ + MEMORY_FACT_SANITIZER_VERSION_V1, MemoryFactSanitizationV1, sanitize_memory_fact_payload, +}; +use tracedecay_domain::{FactAssertionId, FactAssertionV1, FactId, FactOwnerV1, FactPayloadV1}; +use tracedecay_store::{ + FactStoreError, FactStoreResult, MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS, + ProjectMemoryPrivacyPurgeCursorV1, ProjectMemoryPrivacyPurgeReceiptV1, +}; + +use super::crud::{payload_material, payload_metadata}; +use super::primitives::{ + OwnerKey, PROJECT_MEMORY_WRITE_OPERATION, QUERY_OPERATION, from_json, row_string, + storage_error, storage_message, to_json, +}; + +struct SupersededPayload { + assertion_id: FactAssertionId, + fact_id: FactId, + payload_reference_json: String, + payload: FactPayloadV1, +} + +pub(super) async fn purge_superseded_payloads_for_owner_tx( + transaction: &Transaction<'_>, + owner: &FactOwnerV1, + after: Option<&ProjectMemoryPrivacyPurgeCursorV1>, + limit: usize, +) -> FactStoreResult { + if limit == 0 || limit > MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS { + return Err(FactStoreError::InvalidQueryLimit { + limit, + max: MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS, + }); + } + if after.is_some_and(|cursor| cursor.owner() != owner) { + return Err(FactStoreError::OwnerMismatch); + } + let owner_key = OwnerKey::new(owner)?; + let mut candidates = + load_superseded_payloads(transaction, &owner_key, None, None, after, limit + 1).await?; + let has_more = candidates.len() > limit; + if has_more { + candidates.pop(); + } + let next_after = if has_more { + candidates + .last() + .map(|candidate| { + ProjectMemoryPrivacyPurgeCursorV1::new( + owner.clone(), + candidate.fact_id.clone(), + candidate.assertion_id.clone(), + ) + }) + .transpose()? + } else { + None + }; + let scanned = u64::try_from(candidates.len()).map_err(|_| { + storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "superseded payload count exceeds the receipt range", + ) + })?; + let purged = purge_candidates(transaction, &owner_key, candidates).await?; + ProjectMemoryPrivacyPurgeReceiptV1::new( + owner.clone(), + MEMORY_FACT_SANITIZER_VERSION_V1.to_owned(), + scanned, + purged, + next_after, + ) +} + +pub(super) async fn purge_superseded_payloads_for_fact_tx( + transaction: &Transaction<'_>, + owner: &OwnerKey, + fact_id: &FactId, + superseding_assertion_id: &FactAssertionId, +) -> FactStoreResult<()> { + let candidates = load_superseded_payloads( + transaction, + owner, + Some(fact_id), + Some(superseding_assertion_id), + None, + MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS + 1, + ) + .await?; + if candidates.len() > MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS { + return Err(storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "one assertion supersedes more payloads than the bounded purge contract permits", + )); + } + purge_candidates(transaction, owner, candidates).await?; + Ok(()) +} + +async fn load_superseded_payloads( + transaction: &Transaction<'_>, + owner: &OwnerKey, + fact_id: Option<&FactId>, + superseding_assertion_id: Option<&FactAssertionId>, + after: Option<&ProjectMemoryPrivacyPurgeCursorV1>, + limit: usize, +) -> FactStoreResult> { + let mut rows = transaction + .query( + "SELECT payloads.assertion_id, payloads.fact_id, + assertions.payload_reference_json, payloads.payload_json + FROM memory_v2_assertion_payloads AS payloads + JOIN memory_v2_assertions AS assertions + ON assertions.assertion_id = payloads.assertion_id + AND assertions.fact_id = payloads.fact_id + AND assertions.owner_kind = payloads.owner_kind + AND assertions.project_id = payloads.project_id + WHERE payloads.owner_kind = ?1 AND payloads.project_id = ?2 + AND (?3 IS NULL OR payloads.fact_id = ?3) + AND ( + ?5 IS NULL OR payloads.fact_id > ?5 OR + (payloads.fact_id = ?5 AND payloads.assertion_id > ?6) + ) + AND EXISTS ( + SELECT 1 FROM memory_v2_assertion_supersession AS supersession + WHERE supersession.superseded_assertion_id = payloads.assertion_id + AND supersession.fact_id = payloads.fact_id + AND supersession.owner_kind = payloads.owner_kind + AND supersession.project_id = payloads.project_id + AND (?4 IS NULL OR supersession.assertion_id = ?4) + ) + ORDER BY payloads.fact_id, payloads.assertion_id + LIMIT ?7", + params![ + owner.kind, + owner.project_id.as_str(), + fact_id.map(FactId::as_str), + superseding_assertion_id.map(FactAssertionId::as_str), + after.map(|cursor| cursor.fact_id().as_str()), + after.map(|cursor| cursor.assertion_id().as_str()), + i64::try_from(limit).map_err(|_| storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "privacy purge query limit exceeds SQLite range", + ))?, + ], + ) + .await + .map_err(|error| storage_error(PROJECT_MEMORY_WRITE_OPERATION, error))?; + let mut candidates = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| storage_error(PROJECT_MEMORY_WRITE_OPERATION, error))? + { + candidates.push(SupersededPayload { + assertion_id: FactAssertionId::new(row_string( + &row, + 0, + PROJECT_MEMORY_WRITE_OPERATION, + )?)?, + fact_id: FactId::new(row_string(&row, 1, PROJECT_MEMORY_WRITE_OPERATION)?)?, + payload_reference_json: row_string(&row, 2, PROJECT_MEMORY_WRITE_OPERATION)?, + payload: from_json( + &row_string(&row, 3, PROJECT_MEMORY_WRITE_OPERATION)?, + PROJECT_MEMORY_WRITE_OPERATION, + )?, + }); + } + Ok(candidates) +} + +async fn purge_candidates( + transaction: &Transaction<'_>, + owner: &OwnerKey, + candidates: Vec, +) -> FactStoreResult { + let mut flagged = Vec::new(); + for candidate in candidates { + let payload = &candidate.payload; + let metadata = payload_metadata(payload.metadata()); + let wire = payload_material( + payload.content(), + payload.category(), + payload.tags(), + payload.entities(), + &metadata, + payload.source_label(), + ); + let sanitized = sanitize_memory_fact_payload(wire.clone()) + .map_err(|error| storage_error(PROJECT_MEMORY_WRITE_OPERATION, error))?; + let clean = matches!( + sanitized, + MemoryFactSanitizationV1::Durable { payload, .. } if payload == wire + ); + if !clean { + flagged.push(candidate); + } + } + if flagged.is_empty() { + return Ok(0); + } + + transaction + .execute_batch("PRAGMA secure_delete = ON;") + .await + .map_err(|error| storage_error(PROJECT_MEMORY_WRITE_OPERATION, error))?; + for candidate in &flagged { + record_purge_receipt(transaction, owner, candidate).await?; + let changed = transaction + .execute( + "DELETE FROM memory_v2_assertion_payloads + WHERE assertion_id = ?1 AND fact_id = ?2 + AND owner_kind = ?3 AND project_id = ?4", + params![ + candidate.assertion_id.as_str(), + candidate.fact_id.as_str(), + owner.kind, + owner.project_id.as_str(), + ], + ) + .await + .map_err(|error| storage_error(PROJECT_MEMORY_WRITE_OPERATION, error))?; + if changed != 1 { + return Err(storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "detector-flagged assertion payload disappeared before purge", + )); + } + } + u64::try_from(flagged.len()).map_err(|_| { + storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "purged payload count exceeds the receipt range", + ) + }) +} + +async fn record_purge_receipt( + transaction: &Transaction<'_>, + owner: &OwnerKey, + candidate: &SupersededPayload, +) -> FactStoreResult<()> { + transaction + .execute( + "INSERT OR IGNORE INTO memory_v2_assertion_payload_purges( + assertion_id, fact_id, owner_kind, project_id, + payload_reference_json, detector_revision, purge_reason + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, 'detector_flagged')", + params![ + candidate.assertion_id.as_str(), + candidate.fact_id.as_str(), + owner.kind, + owner.project_id.as_str(), + candidate.payload_reference_json.as_str(), + MEMORY_FACT_SANITIZER_VERSION_V1, + ], + ) + .await + .map_err(|error| storage_error(PROJECT_MEMORY_WRITE_OPERATION, error))?; + let mut rows = transaction + .query( + "SELECT payload_reference_json, detector_revision, purge_reason + FROM memory_v2_assertion_payload_purges + WHERE assertion_id = ?1 AND fact_id = ?2 + AND owner_kind = ?3 AND project_id = ?4", + params![ + candidate.assertion_id.as_str(), + candidate.fact_id.as_str(), + owner.kind, + owner.project_id.as_str(), + ], + ) + .await + .map_err(|error| storage_error(QUERY_OPERATION, error))?; + let Some(row) = rows + .next() + .await + .map_err(|error| storage_error(QUERY_OPERATION, error))? + else { + return Err(storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "assertion payload purge receipt insert disappeared", + )); + }; + if row_string(&row, 0, QUERY_OPERATION)? != candidate.payload_reference_json + || row_string(&row, 1, QUERY_OPERATION)? != MEMORY_FACT_SANITIZER_VERSION_V1 + || row_string(&row, 2, QUERY_OPERATION)? != "detector_flagged" + { + return Err(storage_message( + PROJECT_MEMORY_WRITE_OPERATION, + "assertion payload purge receipt identity collision", + )); + } + Ok(()) +} + +pub(super) async fn assertion_payload_is_explicitly_purged_tx( + transaction: &Transaction<'_>, + owner: &OwnerKey, + assertion: &FactAssertionV1, +) -> FactStoreResult { + let mut rows = transaction + .query( + "SELECT payload_reference_json, detector_revision, purge_reason + FROM memory_v2_assertion_payload_purges + WHERE assertion_id = ?1 AND fact_id = ?2 + AND owner_kind = ?3 AND project_id = ?4", + params![ + assertion.assertion_id().as_str(), + assertion.fact_id().as_str(), + owner.kind, + owner.project_id.as_str(), + ], + ) + .await + .map_err(|error| storage_error(QUERY_OPERATION, error))?; + let Some(row) = rows + .next() + .await + .map_err(|error| storage_error(QUERY_OPERATION, error))? + else { + return Ok(false); + }; + Ok(row_string(&row, 0, QUERY_OPERATION)? + == to_json( + &assertion.payload().payload_reference()?, + "serialize assertion payload reference", + )? + && !row_string(&row, 1, QUERY_OPERATION)?.is_empty() + && row_string(&row, 2, QUERY_OPERATION)? == "detector_flagged") +} + +pub(super) async fn assertion_payload_exists_tx( + transaction: &Transaction<'_>, + owner: &OwnerKey, + fact_id: &FactId, + assertion_id: &FactAssertionId, +) -> FactStoreResult { + let mut rows = transaction + .query( + "SELECT 1 FROM memory_v2_assertion_payloads + WHERE assertion_id = ?1 AND fact_id = ?2 + AND owner_kind = ?3 AND project_id = ?4", + params![ + assertion_id.as_str(), + fact_id.as_str(), + owner.kind, + owner.project_id.as_str(), + ], + ) + .await + .map_err(|error| storage_error(QUERY_OPERATION, error))?; + Ok(rows + .next() + .await + .map_err(|error| storage_error(QUERY_OPERATION, error))? + .is_some()) +} diff --git a/crates/tracedecay-store/src/lib.rs b/crates/tracedecay-store/src/lib.rs index c15e1ac9a..8bf263578 100644 --- a/crates/tracedecay-store/src/lib.rs +++ b/crates/tracedecay-store/src/lib.rs @@ -93,18 +93,19 @@ pub use memory::{ FactLineageCursor, FactLineageQuery, FactLineageResponseV1, FactQueryCoverageV1, FactReadControl, FactStore, FactStoreError, FactStoreResult, FactWriteBatch, FactWriteControl, MAX_FACT_QUERY_CONTRADICTIONS, MAX_PROJECT_MEMORY_AUTOMATIC_FACT_RECEIPTS, - MAX_PROJECT_MEMORY_GRAPH_RELATIONS, MAX_PROJECT_MEMORY_SEARCH_SCORE_MILLIONTHS, - ProjectMemoryAutomaticFactApplyDispositionV1, ProjectMemoryAutomaticFactApplyResultV1, - ProjectMemoryAutomaticFactEffectV1, ProjectMemoryAutomaticFactEvidenceV1, - ProjectMemoryAutomaticFactReceiptPageV1, ProjectMemoryAutomaticFactReceiptV1, - ProjectMemoryAutomaticFactStateV1, ProjectMemoryDashboardEntityV1, - ProjectMemoryDashboardFactDetailQueryV1, ProjectMemoryDashboardFactDetailV1, - ProjectMemoryDashboardFactEntityLinkV1, ProjectMemoryDashboardFactSummaryV1, - ProjectMemoryDashboardGrowthPointV1, ProjectMemoryDashboardMemoryOverviewQueryV1, - ProjectMemoryDashboardMemoryOverviewV1, ProjectMemoryDashboardNamedCountV1, - ProjectMemoryDashboardOplogEntryV1, ProjectMemoryDashboardOplogQueryV1, - ProjectMemoryDashboardVectorPointV1, ProjectMemoryDashboardVectorPointsQueryV1, - ProjectMemoryEntityIdV1, ProjectMemoryFactAddCommandV1, ProjectMemoryFactAddDispositionV1, + MAX_PROJECT_MEMORY_GRAPH_RELATIONS, MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS, + MAX_PROJECT_MEMORY_SEARCH_SCORE_MILLIONTHS, ProjectMemoryAutomaticFactApplyDispositionV1, + ProjectMemoryAutomaticFactApplyResultV1, ProjectMemoryAutomaticFactEffectV1, + ProjectMemoryAutomaticFactEvidenceV1, ProjectMemoryAutomaticFactReceiptPageV1, + ProjectMemoryAutomaticFactReceiptV1, ProjectMemoryAutomaticFactStateV1, + ProjectMemoryDashboardEntityV1, ProjectMemoryDashboardFactDetailQueryV1, + ProjectMemoryDashboardFactDetailV1, ProjectMemoryDashboardFactEntityLinkV1, + ProjectMemoryDashboardFactSummaryV1, ProjectMemoryDashboardGrowthPointV1, + ProjectMemoryDashboardMemoryOverviewQueryV1, ProjectMemoryDashboardMemoryOverviewV1, + ProjectMemoryDashboardNamedCountV1, ProjectMemoryDashboardOplogEntryV1, + ProjectMemoryDashboardOplogQueryV1, ProjectMemoryDashboardVectorPointV1, + ProjectMemoryDashboardVectorPointsQueryV1, ProjectMemoryEntityIdV1, + ProjectMemoryFactAddCommandV1, ProjectMemoryFactAddDispositionV1, ProjectMemoryFactAddMaterialV1, ProjectMemoryFactAddOutcomeV1, ProjectMemoryFactContentDigestQueryV1, ProjectMemoryFactContradictionPageV1, ProjectMemoryFactContradictionQueryV1, ProjectMemoryFactContradictionV1, @@ -135,7 +136,8 @@ pub use memory::{ ProjectMemoryFactUpdatePatchV1, ProjectMemoryFactV1, ProjectMemoryGraphPageV1, ProjectMemoryGraphQueryV1, ProjectMemoryGraphRelationV1, ProjectMemoryGraphStore, ProjectMemoryGraphTargetV1, ProjectMemoryMemoryAlgebraV1, ProjectMemoryMemoryFeedbackFunnelV1, - ProjectMemoryMemoryStatusV1, RetrievalAnchorQuery, StoredFactV1, + ProjectMemoryMemoryStatusV1, ProjectMemoryPrivacyPurgeCursorV1, + ProjectMemoryPrivacyPurgeReceiptV1, RetrievalAnchorQuery, StoredFactV1, derive_project_memory_fact_curation_child_operation_id, }; pub use native_integration::{ diff --git a/crates/tracedecay-store/src/memory/mod.rs b/crates/tracedecay-store/src/memory/mod.rs index 9faee5107..480ef64a6 100644 --- a/crates/tracedecay-store/src/memory/mod.rs +++ b/crates/tracedecay-store/src/memory/mod.rs @@ -20,18 +20,19 @@ pub use graph::{ }; pub use project_memory::ProjectMemoryAutomationRunReceiptsV1; pub use project_memory::{ - MAX_PROJECT_MEMORY_AUTOMATIC_FACT_RECEIPTS, MAX_PROJECT_MEMORY_SEARCH_SCORE_MILLIONTHS, - ProjectMemoryAutomaticFactApplyDispositionV1, ProjectMemoryAutomaticFactApplyResultV1, - ProjectMemoryAutomaticFactEffectV1, ProjectMemoryAutomaticFactEvidenceV1, - ProjectMemoryAutomaticFactReceiptPageV1, ProjectMemoryAutomaticFactReceiptV1, - ProjectMemoryAutomaticFactStateV1, ProjectMemoryDashboardEntityV1, - ProjectMemoryDashboardFactDetailQueryV1, ProjectMemoryDashboardFactDetailV1, - ProjectMemoryDashboardFactEntityLinkV1, ProjectMemoryDashboardFactSummaryV1, - ProjectMemoryDashboardGrowthPointV1, ProjectMemoryDashboardMemoryOverviewQueryV1, - ProjectMemoryDashboardMemoryOverviewV1, ProjectMemoryDashboardNamedCountV1, - ProjectMemoryDashboardOplogEntryV1, ProjectMemoryDashboardOplogQueryV1, - ProjectMemoryDashboardVectorPointV1, ProjectMemoryDashboardVectorPointsQueryV1, - ProjectMemoryEntityIdV1, ProjectMemoryFactAddCommandV1, ProjectMemoryFactAddDispositionV1, + MAX_PROJECT_MEMORY_AUTOMATIC_FACT_RECEIPTS, MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS, + MAX_PROJECT_MEMORY_SEARCH_SCORE_MILLIONTHS, ProjectMemoryAutomaticFactApplyDispositionV1, + ProjectMemoryAutomaticFactApplyResultV1, ProjectMemoryAutomaticFactEffectV1, + ProjectMemoryAutomaticFactEvidenceV1, ProjectMemoryAutomaticFactReceiptPageV1, + ProjectMemoryAutomaticFactReceiptV1, ProjectMemoryAutomaticFactStateV1, + ProjectMemoryDashboardEntityV1, ProjectMemoryDashboardFactDetailQueryV1, + ProjectMemoryDashboardFactDetailV1, ProjectMemoryDashboardFactEntityLinkV1, + ProjectMemoryDashboardFactSummaryV1, ProjectMemoryDashboardGrowthPointV1, + ProjectMemoryDashboardMemoryOverviewQueryV1, ProjectMemoryDashboardMemoryOverviewV1, + ProjectMemoryDashboardNamedCountV1, ProjectMemoryDashboardOplogEntryV1, + ProjectMemoryDashboardOplogQueryV1, ProjectMemoryDashboardVectorPointV1, + ProjectMemoryDashboardVectorPointsQueryV1, ProjectMemoryEntityIdV1, + ProjectMemoryFactAddCommandV1, ProjectMemoryFactAddDispositionV1, ProjectMemoryFactAddMaterialV1, ProjectMemoryFactAddOutcomeV1, ProjectMemoryFactContradictionPageV1, ProjectMemoryFactContradictionQueryV1, ProjectMemoryFactContradictionV1, ProjectMemoryFactCurationAddV1, @@ -54,6 +55,7 @@ pub use project_memory::{ ProjectMemoryFactSearchKindV1, ProjectMemoryFactSearchPageV1, ProjectMemoryFactSearchScoresV1, ProjectMemoryFactSnapshotV1, ProjectMemoryFactUnavailableV1, ProjectMemoryFactUpdateCommandV1, ProjectMemoryFactUpdateOutcomeV1, ProjectMemoryFactUpdatePatchV1, ProjectMemoryFactV1, + ProjectMemoryPrivacyPurgeCursorV1, ProjectMemoryPrivacyPurgeReceiptV1, derive_project_memory_fact_curation_child_operation_id, }; pub use queries::{ diff --git a/crates/tracedecay-store/src/memory/project_memory/mod.rs b/crates/tracedecay-store/src/memory/project_memory/mod.rs index 8ef08ef83..def86da1e 100644 --- a/crates/tracedecay-store/src/memory/project_memory/mod.rs +++ b/crates/tracedecay-store/src/memory/project_memory/mod.rs @@ -310,6 +310,106 @@ pub struct ProjectMemoryFactHistoryV1 { next_after: Option, } +/// Durable result of one owner-scoped sweep over superseded assertion +/// payloads under the active privacy detector. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProjectMemoryPrivacyPurgeReceiptV1 { + owner: FactOwnerV1, + detector_revision: String, + scanned_payloads: u64, + purged_payloads: u64, + next_after: Option, +} + +pub const MAX_PROJECT_MEMORY_PRIVACY_PURGE_PAYLOADS: usize = 256; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProjectMemoryPrivacyPurgeCursorV1 { + owner: FactOwnerV1, + fact_id: FactId, + assertion_id: FactAssertionId, +} + +impl ProjectMemoryPrivacyPurgeCursorV1 { + pub fn new( + owner: FactOwnerV1, + fact_id: FactId, + assertion_id: FactAssertionId, + ) -> FactStoreResult { + owner.validate()?; + validate_owned_fact_id(&fact_id, &owner)?; + assertion_id.validate()?; + Ok(Self { + owner, + fact_id, + assertion_id, + }) + } + + pub fn owner(&self) -> &FactOwnerV1 { + &self.owner + } + + pub fn fact_id(&self) -> &FactId { + &self.fact_id + } + + pub fn assertion_id(&self) -> &FactAssertionId { + &self.assertion_id + } +} + +impl ProjectMemoryPrivacyPurgeReceiptV1 { + pub fn new( + owner: FactOwnerV1, + detector_revision: String, + scanned_payloads: u64, + purged_payloads: u64, + next_after: Option, + ) -> FactStoreResult { + owner.validate()?; + validate_project_memory_text(&detector_revision, "privacy detector revision")?; + if purged_payloads > scanned_payloads { + return Err(FactStoreError::Contract(DomainError::NonCanonical { + field: "privacy purge counts", + })); + } + if next_after + .as_ref() + .is_some_and(|cursor| cursor.owner() != &owner) + { + return Err(FactStoreError::OwnerMismatch); + } + Ok(Self { + owner, + detector_revision, + scanned_payloads, + purged_payloads, + next_after, + }) + } + + pub fn owner(&self) -> &FactOwnerV1 { + &self.owner + } + + pub fn detector_revision(&self) -> &str { + &self.detector_revision + } + + pub fn scanned_payloads(&self) -> u64 { + self.scanned_payloads + } + + pub fn purged_payloads(&self) -> u64 { + self.purged_payloads + } + + pub fn next_after(&self) -> Option<&ProjectMemoryPrivacyPurgeCursorV1> { + self.next_after.as_ref() + } +} + impl ProjectMemoryFactHistoryV1 { pub fn new( owner: FactOwnerV1, diff --git a/crates/tracedecay-store/src/memory/traits.rs b/crates/tracedecay-store/src/memory/traits.rs index c88e58bba..c79fd7b5b 100644 --- a/crates/tracedecay-store/src/memory/traits.rs +++ b/crates/tracedecay-store/src/memory/traits.rs @@ -25,7 +25,8 @@ use super::{ ProjectMemoryFactRemoveOutcomeV1, ProjectMemoryFactRetrievalCommandV1, ProjectMemoryFactRetrievalOutcomeV1, ProjectMemoryFactSearchPageV1, ProjectMemoryFactSearchQuery, ProjectMemoryFactUpdateCommandV1, - ProjectMemoryFactUpdateOutcomeV1, ProjectMemoryMemoryStatusV1, RetrievalAnchorQuery, + ProjectMemoryFactUpdateOutcomeV1, ProjectMemoryMemoryStatusV1, + ProjectMemoryPrivacyPurgeCursorV1, ProjectMemoryPrivacyPurgeReceiptV1, RetrievalAnchorQuery, StoredFactV1, }; @@ -85,6 +86,17 @@ pub trait FactStore: Send + Sync { /// Single typed authority boundary for canonical project memory. pub trait ProjectMemoryFactStore: FactStore { + /// Re-evaluates every persisted superseded payload for one owner and + /// atomically records an immutable purge receipt before deleting each + /// detector-flagged payload and its FTS copy. + fn purge_project_memory_superseded_payloads( + &self, + owner: FactOwnerV1, + after: Option, + limit: usize, + write_control: &FactWriteControl, + ) -> impl Future> + Send; + fn list_project_memory_facts( &self, query: ProjectMemoryFactListQueryV1, diff --git a/crates/tracedecay-usecases/src/memory/mod.rs b/crates/tracedecay-usecases/src/memory/mod.rs index b1e72dab2..fe910283f 100644 --- a/crates/tracedecay-usecases/src/memory/mod.rs +++ b/crates/tracedecay-usecases/src/memory/mod.rs @@ -61,7 +61,8 @@ use tracedecay_store::{ ProjectMemoryFactRetrievalOutcomeV1, ProjectMemoryFactSearchGraphCoverageV1, ProjectMemoryFactSearchPageV1, ProjectMemoryFactSearchQuery, ProjectMemoryFactStore, ProjectMemoryFactUpdateCommandV1, ProjectMemoryFactUpdateOutcomeV1, - ProjectMemoryMemoryStatusV1, RetrievalAnchorQuery, StoredFactV1, + ProjectMemoryMemoryStatusV1, ProjectMemoryPrivacyPurgeCursorV1, + ProjectMemoryPrivacyPurgeReceiptV1, RetrievalAnchorQuery, StoredFactV1, }; /// Maps a [`MemoryApplicationError`] onto the root/dashboard-facing diff --git a/crates/tracedecay-usecases/src/memory/privacy_remediation.rs b/crates/tracedecay-usecases/src/memory/privacy_remediation.rs index d9ebdc9f8..334a672f1 100644 --- a/crates/tracedecay-usecases/src/memory/privacy_remediation.rs +++ b/crates/tracedecay-usecases/src/memory/privacy_remediation.rs @@ -42,6 +42,8 @@ pub enum PrivacyRemediationTriggerV1 { pub struct ProjectMemoryPrivacyRemediationReceiptV1 { pub detector_revision: String, pub trigger: PrivacyRemediationTriggerV1, + pub superseded_payloads_scanned: u64, + pub superseded_payloads_purged: u64, pub scanned_facts: u64, pub clean_facts: u64, pub quarantined_facts: u64, @@ -69,6 +71,50 @@ impl MemoryApplication { write_control: &FactWriteControl, ) -> Result { let confidence = remediation_confidence()?; + let mut superseded_payloads_scanned = 0_u64; + let mut superseded_payloads_purged = 0_u64; + let mut purge_after = None; + loop { + let requested_after = purge_after.take(); + let purge = self + .authority + .purge_project_memory_superseded_payloads( + self.owner.clone(), + requested_after.clone(), + RESCAN_PAGE_LIMIT, + write_control, + ) + .await?; + if purge.owner() != &self.owner + || purge.detector_revision() != MEMORY_FACT_SANITIZER_VERSION_V1 + { + return Err(MemoryApplicationError::InvalidAuthorityResult { + invariant: "privacy purge receipt binding", + }); + } + superseded_payloads_scanned = superseded_payloads_scanned + .checked_add(purge.scanned_payloads()) + .ok_or(MemoryApplicationError::InvalidAuthorityResult { + invariant: "privacy purge scanned count range", + })?; + superseded_payloads_purged = superseded_payloads_purged + .checked_add(purge.purged_payloads()) + .ok_or(MemoryApplicationError::InvalidAuthorityResult { + invariant: "privacy purge purged count range", + })?; + let Some(next) = purge.next_after() else { + break; + }; + if requested_after.as_ref().is_some_and(|previous| { + (next.fact_id(), next.assertion_id()) + <= (previous.fact_id(), previous.assertion_id()) + }) { + return Err(MemoryApplicationError::InvalidAuthorityResult { + invariant: "privacy purge cursor advancement", + }); + } + purge_after = Some(next.clone()); + } let mut scanned_facts = 0_u64; let mut clean_facts = 0_u64; let mut quarantined_facts = 0_u64; @@ -139,6 +185,8 @@ impl MemoryApplication { Ok(ProjectMemoryPrivacyRemediationReceiptV1 { detector_revision: MEMORY_FACT_SANITIZER_VERSION_V1.to_owned(), trigger, + superseded_payloads_scanned, + superseded_payloads_purged, scanned_facts, clean_facts, quarantined_facts, diff --git a/crates/tracedecay-usecases/src/memory/tests.rs b/crates/tracedecay-usecases/src/memory/tests.rs index 99a2d0484..7aea60677 100644 --- a/crates/tracedecay-usecases/src/memory/tests.rs +++ b/crates/tracedecay-usecases/src/memory/tests.rs @@ -181,6 +181,23 @@ impl FactStore for FakeAuthority { } impl ProjectMemoryFactStore for FakeAuthority { + async fn purge_project_memory_superseded_payloads( + &self, + owner: FactOwnerV1, + _after: Option, + _limit: usize, + _write_control: &FactWriteControl, + ) -> FactStoreResult { + self.authority_calls.lock().unwrap().push("privacy-purge"); + ProjectMemoryPrivacyPurgeReceiptV1::new( + owner, + "test-detector-revision".to_owned(), + 0, + 0, + None, + ) + } + async fn list_project_memory_facts( &self, query: ProjectMemoryFactListQueryV1, diff --git a/src/daemon/privacy_remediation.rs b/src/daemon/privacy_remediation.rs index 5f7cf9371..48550514e 100644 --- a/src/daemon/privacy_remediation.rs +++ b/src/daemon/privacy_remediation.rs @@ -32,6 +32,8 @@ pub(crate) fn spawn_at_rest_privacy_remediation( event = "project_memory_privacy_remediation", project = %project, detector_revision = %receipt.detector_revision, + superseded_payloads_scanned = receipt.superseded_payloads_scanned, + superseded_payloads_purged = receipt.superseded_payloads_purged, scanned_facts = receipt.scanned_facts, clean_facts = receipt.clean_facts, quarantined_facts = receipt.quarantined_facts, @@ -107,8 +109,9 @@ mod tests { SanitizerDispositionV1, SensitivityV1, }; use tracedecay_store::{ - ProjectMemoryFactAddMaterialV1, ProjectMemoryFactListQueryV1, - ProjectMemoryFactProjectionV1, ProjectMemoryFactStore, + ProjectMemoryFactAddMaterialV1, ProjectMemoryFactIdV1, ProjectMemoryFactListQueryV1, + ProjectMemoryFactProjectionV1, ProjectMemoryFactStore, ProjectMemoryFactUpdateCommandV1, + ProjectMemoryFactUpdatePatchV1, }; use tracedecay_usecases::memory::{MemoryApplication, PrivacyRemediationTriggerV1}; @@ -160,19 +163,18 @@ mod tests { .expect("legacy receipt id") } - /// Persists one fact exactly as an ingest path running an older vendored - /// ruleset could have: the receipt binds the raw payload without the - /// current detector rules ever evaluating it. The store's write firewall - /// pins the sanitizer revision string, so the legacy condition being - /// simulated is a ruleset refresh within the pinned revision. - async fn seed_legacy_fact( - database: &crate::db::Database, + /// Receipt-bound raw payload material exactly as an ingest path running + /// an older vendored ruleset could have persisted it: the receipt binds + /// the payload without the current detector rules ever evaluating it. The + /// store's write firewall pins the sanitizer revision string, so the + /// legacy condition being simulated is a ruleset refresh within the + /// pinned revision. + fn legacy_fact_material( owner: &FactOwnerV1, - label: &str, content: &str, source_label: Option<&str>, metadata: Value, - ) { + ) -> ProjectMemoryFactAddMaterialV1 { let mut tags = Vec::new(); let mut entities = Vec::new(); let payload_reference = FactPayloadV1::canonicalize_material( @@ -204,7 +206,7 @@ mod tests { Some(payload_reference), ) .expect("legacy sanitization receipt"); - let command = ProjectMemoryFactAddMaterialV1::new( + ProjectMemoryFactAddMaterialV1::new( owner.clone(), content.to_owned(), FactCategoryV1::Project, @@ -218,15 +220,26 @@ mod tests { None, ) .expect("legacy fact material") - .into_command( - ProvenanceId::new(format!("operation.privacy-legacy.{label}")) - .expect("legacy operation id"), - ) - .expect("legacy fact command"); + } + + async fn seed_legacy_fact( + database: &crate::db::Database, + owner: &FactOwnerV1, + label: &str, + content: &str, + source_label: Option<&str>, + metadata: Value, + ) -> tracedecay_store::ProjectMemoryFactAddOutcomeV1 { + let command = legacy_fact_material(owner, content, source_label, metadata) + .into_command( + ProvenanceId::new(format!("operation.privacy-legacy.{label}")) + .expect("legacy operation id"), + ) + .expect("legacy fact command"); DatabaseFactStore::new(database) .add_project_memory_fact(command, &remediation_write_control()) .await - .expect("persist legacy fact"); + .expect("persist legacy fact") } async fn served_contents( @@ -266,6 +279,110 @@ mod tests { .expect("inspect persisted memory payloads") } + async fn assertion_payload_purge_receipts(database: &crate::db::Database) -> i64 { + database + .query_scalar_i64( + "inspect at-rest privacy purge receipts", + "SELECT COUNT(*) FROM memory_v2_assertion_payload_purges", + ) + .await + .expect("inspect persisted privacy purge receipts") + } + + async fn orphaned_payload_fts_rows(database: &crate::db::Database) -> i64 { + database + .query_scalar_i64( + "inspect at-rest privacy FTS cleanup", + "SELECT COUNT(*) + FROM memory_v2_assertion_payloads_fts AS fts + LEFT JOIN memory_v2_assertion_payloads AS payloads ON payloads.rowid = fts.rowid + WHERE payloads.rowid IS NULL", + ) + .await + .expect("inspect payload FTS cleanup") + } + + struct LegacyPayloadRow { + assertion_id: String, + fact_id: String, + owner_kind: String, + project_id: String, + payload_json: String, + content: String, + } + + async fn capture_payload_row( + database: &crate::db::Database, + fact_id: &tracedecay_domain::FactId, + ) -> LegacyPayloadRow { + let mut rows = database + .read_connection() + .query( + "SELECT assertion_id, fact_id, owner_kind, project_id, payload_json, content + FROM memory_v2_assertion_payloads WHERE fact_id = ?1", + [fact_id.as_str()], + ) + .await + .expect("read legacy payload row"); + let row = rows + .next() + .await + .expect("read legacy payload result") + .expect("legacy payload row exists"); + LegacyPayloadRow { + assertion_id: row.get(0).expect("assertion id"), + fact_id: row.get(1).expect("fact id"), + owner_kind: row.get(2).expect("owner kind"), + project_id: row.get(3).expect("project id"), + payload_json: row.get(4).expect("payload json"), + content: row.get(5).expect("payload content"), + } + } + + /// Reconstructs the exact persisted shape an older binary left after it + /// superseded a secret-bearing assertion without an explicit purge + /// receipt. The final immutable trigger is restored before remediation. + async fn restore_pre_purge_superseded_payload( + database: &crate::db::Database, + payload: &LegacyPayloadRow, + ) { + let transaction = database + .begin_write_transaction("restore pre-purge superseded payload fixture") + .await + .expect("database transaction"); + transaction + .execute_batch( + "DROP TRIGGER memory_v2_assertion_payload_purges_no_delete; + DELETE FROM memory_v2_assertion_payload_purges; + CREATE TRIGGER memory_v2_assertion_payload_purges_no_delete + BEFORE DELETE ON memory_v2_assertion_payload_purges BEGIN + SELECT RAISE(ABORT, 'memory_v2 assertion payload purge receipts are immutable'); + END;", + ) + .await + .expect("restore pre-purge receipt shape"); + transaction + .execute( + "INSERT INTO memory_v2_assertion_payloads( + assertion_id, fact_id, owner_kind, project_id, payload_json, content + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6)", + crate::db::engine::params![ + payload.assertion_id.as_str(), + payload.fact_id.as_str(), + payload.owner_kind.as_str(), + payload.project_id.as_str(), + payload.payload_json.as_str(), + payload.content.as_str(), + ], + ) + .await + .expect("restore superseded payload from pre-purge binary"); + transaction + .commit() + .await + .expect("commit pre-purge superseded payload fixture"); + } + #[tokio::test] async fn at_rest_rescan_quarantines_and_erases_legacy_detector_hits() { let temp = TempDir::new().expect("privacy remediation fixture root"); @@ -383,6 +500,106 @@ mod tests { assert!(second.curation_receipts.is_empty()); } + #[tokio::test] + async fn at_rest_rescan_purges_detector_flagged_history_already_superseded_by_clean_content() { + let temp = TempDir::new().expect("privacy remediation fixture root"); + let profile_root = temp.path().join("profile"); + let project_id = + ProjectId::new("project.privacy-remediation-superseded").expect("project id"); + let project_root = enrolled_root(temp.path(), &project_id); + let _database_scope = crate::db::enter_daemon_database_scope( + &profile_root, + 43, + "superseded privacy remediation test", + ) + .expect("daemon database scope"); + let identity = profile_identity::load_or_create(&profile_root).expect("profile identity"); + let registry = DaemonSessionRuntimeRegistryV1::open(identity) + .await + .expect("daemon registry"); + let database = registry + .project_memory(project_id.clone(), [project_root]) + .await + .expect("project memory authority"); + let owner = FactOwnerV1::Project { project_id }; + + let added = seed_legacy_fact( + &database, + &owner, + "superseded-dirty", + &format!("deployment credential is {}", secret()), + None, + json!({"fixture": "superseded-dirty"}), + ) + .await; + let legacy_payload = capture_payload_row(&database, added.fact().fact_id()).await; + let target = ProjectMemoryFactIdV1::new(owner.clone(), added.fact().fact_id().clone()) + .expect("owner-bound legacy fact"); + DatabaseFactStore::new(&database) + .update_project_memory_fact( + ProjectMemoryFactUpdateCommandV1::new( + target, + ProvenanceId::new("operation.privacy-clean-correction") + .expect("correction operation id"), + None, + ProjectMemoryFactUpdatePatchV1::new( + Some("deployment authentication uses the managed vault".to_owned()), + None, + None, + None, + None, + None, + None, + ) + .expect("clean correction patch"), + None, + ) + .expect("clean correction command"), + &remediation_write_control(), + ) + .await + .expect("supersede legacy secret with clean content"); + + assert_eq!( + persisted_payload_rows_containing(&database, &secret()).await, + 0, + "the canonical correction boundary must purge the detector-flagged predecessor" + ); + assert_eq!(assertion_payload_purge_receipts(&database).await, 1); + assert_eq!(orphaned_payload_fts_rows(&database).await, 0); + + restore_pre_purge_superseded_payload(&database, &legacy_payload).await; + assert_eq!( + persisted_payload_rows_containing(&database, &secret()).await, + 1, + "the rollout fixture must contain one pre-existing superseded secret" + ); + assert_eq!(assertion_payload_purge_receipts(&database).await, 0); + + let memory = MemoryApplication::new(owner.clone(), DatabaseFactStore::new(&database)) + .expect("owner-bound memory application"); + let receipt = memory + .privacy_remediation_rescan( + PrivacyRemediationTriggerV1::DetectorRevisionAdoption, + &remediation_read_control(), + &remediation_write_control(), + ) + .await + .expect("at-rest privacy rescan"); + assert_eq!(receipt.superseded_payloads_scanned, 1); + assert_eq!(receipt.superseded_payloads_purged, 1); + assert_eq!(receipt.scanned_facts, 1); + assert_eq!(receipt.clean_facts, 1); + assert_eq!(receipt.quarantined_facts, 0); + assert_eq!(served_contents(&memory, &owner).await.len(), 1); + assert_eq!( + persisted_payload_rows_containing(&database, &secret()).await, + 0 + ); + assert_eq!(assertion_payload_purge_receipts(&database).await, 1); + assert_eq!(orphaned_payload_fts_rows(&database).await, 0); + } + #[tokio::test] async fn remediation_commits_more_than_one_curation_batch_without_leaving_secret_bytes() { let home = TempDir::new().expect("isolated home"); @@ -406,17 +623,86 @@ mod tests { let owner = FactOwnerV1::Project { project_id: project_id.clone(), }; - for index in 0..257_u16 { - seed_legacy_fact( - &database, - &owner, - &format!("dirty-{index}"), - &format!("credential {index} is {}", secret()), - None, - json!({"fixture": "many-dirty", "index": index}), - ) - .await; - } + + // Per-write graph publication makes 257 sequential adds quadratic in + // store size (and past the CI slow-timeout ceiling), so the bulk is + // seeded through one store-level curation batch: one commit for 256 + // dirty facts, one ordinary add for the 257th. The clean anchor fact + // supplies the reviewed evidence reference every curation add + // requires. + let anchor = seed_legacy_fact( + &database, + &owner, + "anchor", + "the retry budget is three attempts", + None, + json!({"fixture": "anchor"}), + ) + .await; + let ProjectMemoryFactProjectionV1::Available(anchor) = anchor.fact() else { + panic!("the anchor fact must be served"); + }; + let seed_confidence = Confidence::new(0.9).expect("seed confidence"); + let outer_operation_id = ProvenanceId::new("operation.privacy-legacy.batch-seed") + .expect("seed batch operation id"); + let operations = (0..256_usize) + .map(|index| { + let child_operation_id = + tracedecay_store::derive_project_memory_fact_curation_child_operation_id( + &outer_operation_id, + index, + tracedecay_store::ProjectMemoryFactCurationMutationKindV1::Add, + ) + .expect("seed child operation id"); + let command = legacy_fact_material( + &owner, + &format!("credential {index} is {}", secret()), + None, + json!({"fixture": "many-dirty", "index": index}), + ) + .into_command(child_operation_id) + .expect("seed add command"); + let evidence = tracedecay_store::ProjectMemoryFactCurationEvidenceV1::new( + &owner, + vec![tracedecay_store::ProjectMemoryFactCurationReviewRefV1::new( + tracedecay_store::ProjectMemoryFactIdV1::new( + owner.clone(), + anchor.fact_id().clone(), + ) + .expect("anchor fact identity"), + anchor.last_event_id().clone(), + )], + seed_confidence, + "legacy privacy fixture seed".to_owned(), + ) + .expect("seed evidence"); + tracedecay_store::ProjectMemoryFactCurationOperationV1::Add( + tracedecay_store::ProjectMemoryFactCurationAddV1::new(command, evidence) + .expect("seed curation add"), + ) + }) + .collect::>(); + let seed_batch = tracedecay_store::ProjectMemoryFactCurationBatchV1::new( + owner.clone(), + outer_operation_id, + None, + seed_confidence, + operations, + ) + .expect("seed curation batch"); + DatabaseFactStore::new(&database) + .apply_project_memory_fact_curation(seed_batch, &remediation_write_control()) + .await + .expect("persist bulk legacy facts"); + seed_legacy_fact( + &database, + &owner, + "dirty-tail", + &format!("credential tail is {}", secret()), + None, + json!({"fixture": "many-dirty", "index": "tail"}), + ) + .await; let memory = MemoryApplication::new(owner, DatabaseFactStore::new(&database)) .expect("owner-bound memory application"); @@ -429,10 +715,18 @@ mod tests { .await .expect("every bounded remediation batch commits"); - assert_eq!(receipt.scanned_facts, 257); - assert_eq!(receipt.clean_facts, 0); + assert_eq!(receipt.scanned_facts, 258); + assert_eq!(receipt.clean_facts, 1, "only the anchor fact is clean"); assert_eq!(receipt.quarantined_facts, 257); assert_eq!(receipt.curation_receipts.len(), 5); + assert_eq!( + receipt + .curation_receipts + .iter() + .map(tracedecay_store::ProjectMemoryFactCurationReceiptV1::facts_removed) + .sum::(), + 257 + ); assert_eq!( persisted_payload_rows_containing(&database, &secret()).await, 0,