fix(privacy): close the remaining Codex P1 gaps after #522 merged

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,

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",

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",

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);

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?;

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::{

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],

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;

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<ProjectMemoryPrivacyPurgeCursorV1>,
+        limit: usize,
+        write_control: &FactWriteControl,
+    ) -> FactStoreResult<ProjectMemoryPrivacyPurgeReceiptV1> {
+        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<ProjectMemoryPrivacyPurgeCursorV1>,
+            limit: usize,
+            write_control: &FactWriteControl,
+        ) -> FactStoreResult<ProjectMemoryPrivacyPurgeReceiptV1>;
         fn list_project_memory_facts(
             query: ProjectMemoryFactListQueryV1,
             read_control: &FactReadControl,