diff --git a/crates/tracedecay-global-db/src/lib.rs b/crates/tracedecay-global-db/src/lib.rs index aa200ddbb..3d3e4cf14 100644 --- a/crates/tracedecay-global-db/src/lib.rs +++ b/crates/tracedecay-global-db/src/lib.rs @@ -60,6 +60,7 @@ mod registered_accounting; mod registered_analytics; mod registered_dashboard; mod registered_lcm; +mod registered_lcm_privacy; mod registered_legacy_relations; mod registered_session_sync; mod registered_sessions; @@ -123,6 +124,7 @@ pub use registered::{ WorkAttemptDeliveryCensusReadV1, }; pub use registered_analytics::ObservabilityRetentionReceiptV1; +pub use registered_lcm_privacy::{LcmPrivacyRescanOutcomeV1, LcmPrivacyRescanReceiptV1}; pub use remote_deletion::{ RemoteDeletionCleanupState, RemoteDeletionFailureCode, RemoteDeletionPhase, RemoteDeletionTarget, RemoteDeletionTombstone, RemoteDeletionTombstoneRecordOutcome, diff --git a/crates/tracedecay-global-db/src/registered_lcm.rs b/crates/tracedecay-global-db/src/registered_lcm.rs index b8f84f2d0..f11b0f048 100644 --- a/crates/tracedecay-global-db/src/registered_lcm.rs +++ b/crates/tracedecay-global-db/src/registered_lcm.rs @@ -70,7 +70,7 @@ impl Executor for RegisteredGlobalDbWriterConnection<'_> { } impl RegisteredGlobalDb { - async fn lcm_read_snapshot( + pub(super) async fn lcm_read_snapshot( &self, ) -> Result { self.read_snapshot() @@ -78,7 +78,7 @@ impl RegisteredGlobalDb { .map_err(|error| LcmError::Db(error.to_string())) } - fn lcm_storage_root(&self) -> Result<&Path, LcmError> { + pub(super) fn lcm_storage_root(&self) -> Result<&Path, LcmError> { self.db_path() .parent() .ok_or_else(|| LcmError::Db("registered session database has no parent".to_string())) diff --git a/crates/tracedecay-global-db/src/registered_lcm_privacy.rs b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs new file mode 100644 index 000000000..ad6a3007c --- /dev/null +++ b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs @@ -0,0 +1,455 @@ +//! At-rest privacy rescan over the persisted LCM raw-message store. +//! +//! Ingest sanitizes every raw message before persistence, but rows written +//! under older detector rules can hold values the current detector would +//! redact or refuse. This owner re-evaluates every at-rest raw-message body — +//! inline `content` and whole-message external payload bytes — and re-ingests +//! each hit through the same staging and commit path new ingest uses, so +//! redaction, externalization, quarantine, receipts, and FTS maintenance all +//! follow the one canonical sanitizer. A replaced external payload file is +//! deleted through the payload tombstone machinery so the superseded bytes do +//! not survive on disk. +//! +//! One completed pass settles a per-store watermark keyed by +//! [`lcm_payload_detector_revision`], so the sweep runs once per rule refresh +//! instead of on every project open. An interrupted pass leaves the watermark +//! unset and reruns from the start; sanitization is idempotent. A row the +//! sanitizer cannot re-evaluate fails the run with a typed error — never a +//! silent skip — and the watermark stays unset until a pass covers every row. +//! +//! Media-span payload files are byte ranges the ingest scan already evaluated +//! inside their owning message text before externalizing them; the rescan +//! re-evaluates message bodies (where those placeholders live), not the +//! extracted media bytes. Unreceipted rows are first protected through the +//! existing [`RegisteredGlobalDb::lcm_protect_session_raw_messages`] pass. + +use std::path::Path; + +use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::privacy::{lcm_payload_detector_revision, sanitize_lcm_payload_text}; +use tracedecay_sessions::runtime::{ + SessionMessageRecord, + lcm::{ + LcmError, LcmStorageKind, gc, + payload::{self, DeleteOpts}, + raw, schema, + }, +}; + +use super::RegisteredGlobalDb; + +/// Watermark row in `lcm_gc_meta`: the detector revision whose rescan last +/// completed over this store. +pub(crate) const LCM_PRIVACY_RESCAN_META_KEY: &str = "privacy_rescan_completed_revision"; + +/// One page of raw rows per authority read. +const RESCAN_PAGE_LIMIT: i64 = 64; + +/// Truthful outcome of one at-rest LCM privacy rescan request. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum LcmPrivacyRescanOutcomeV1 { + /// The store already completed a rescan under the current detector + /// revision; nothing was scanned. + AlreadyCurrent, + /// A full pass ran to completion and settled the watermark. + Completed(LcmPrivacyRescanReceiptV1), +} + +/// Counts of one completed at-rest rescan pass. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LcmPrivacyRescanReceiptV1 { + pub detector_revision: String, + /// Rows whose at-rest body was re-evaluated under the current detector. + pub scanned_rows: u64, + /// Rows the current detector left byte-identical. + pub clean_rows: u64, + /// Rows re-ingested because the current detector changed their body or + /// provider metadata. + pub remediated_rows: u64, + /// Unreceipted rows re-protected through the canonical protect pass + /// before the scan. + pub protected_rows: u64, + /// External rows whose payload bytes are no longer at rest (offloaded or + /// collected); only their placeholder remains, so there is nothing left + /// to rescan or disclose. + pub unavailable_payload_rows: u64, +} + +/// One at-rest raw row joined with its optional `session_messages` twin. +struct RescanRow { + store_id: i64, + provider: String, + message_id: String, + session_id: String, + role: String, + ordinal: i64, + timestamp: Option, + content: Option, + storage_kind: LcmStorageKind, + payload_ref: Option, + metadata_json: Option, + projection_kind: Option, + projection_model: Option, + projection_tool_names: Option, + projection_source_path: Option, + projection_source_offset: Option, +} + +/// The rescan input recovered from one row's at-rest bytes. +enum RescanInput { + Body { + text: String, + replaced_payload_ref: Option, + }, + /// The external payload bytes are gone (offloaded or collected); the row + /// serves only its placeholder. + PayloadUnavailable, +} + +impl RegisteredGlobalDb { + /// Rescans every persisted LCM raw-message body under the current + /// detector revision, remediating hits through the canonical ingest path. + /// Runs at most once per store per detector revision. + pub async fn lcm_privacy_rescan_raw_messages( + &self, + ) -> Result { + let detector_revision = lcm_payload_detector_revision(); + { + let snapshot = self.lcm_read_snapshot().await?; + if schema::get_gc_meta(&snapshot, LCM_PRIVACY_RESCAN_META_KEY) + .await? + .as_deref() + == Some(detector_revision) + { + return Ok(LcmPrivacyRescanOutcomeV1::AlreadyCurrent); + } + } + + let protected_rows = self.protect_unreceipted_sessions().await?; + + let storage_root = self.lcm_storage_root()?.to_path_buf(); + let mut scanned_rows = 0_u64; + let mut clean_rows = 0_u64; + let mut remediated_rows = 0_u64; + let mut unavailable_payload_rows = 0_u64; + let mut after_store_id = 0_i64; + loop { + let page = self.load_rescan_page(after_store_id).await?; + let Some(last) = page.last() else { + break; + }; + after_store_id = last.store_id; + for row in page { + let (text, replaced_payload_ref) = + match self.rescan_input(&storage_root, &row).await? { + RescanInput::Body { + text, + replaced_payload_ref, + } => (text, replaced_payload_ref), + RescanInput::PayloadUnavailable => { + unavailable_payload_rows += 1; + continue; + } + }; + scanned_rows += 1; + let provider_metadata = stored_provider_metadata(&row)?; + if !requires_remediation(&text, provider_metadata.as_deref())? { + clean_rows += 1; + continue; + } + self.remediate_row( + &storage_root, + &row, + text, + provider_metadata, + replaced_payload_ref, + ) + .await?; + remediated_rows += 1; + } + } + + let transaction = self + .begin_write_transaction() + .await + .map_err(|error| LcmError::Db(error.to_string()))?; + schema::set_gc_meta(&transaction, LCM_PRIVACY_RESCAN_META_KEY, detector_revision).await?; + transaction.commit().await?; + + Ok(LcmPrivacyRescanOutcomeV1::Completed( + LcmPrivacyRescanReceiptV1 { + detector_revision: detector_revision.to_owned(), + scanned_rows, + clean_rows, + remediated_rows, + protected_rows, + unavailable_payload_rows, + }, + )) + } + + /// Binds receipts to every unreceipted row through the one existing + /// protect pass, per owning session. + async fn protect_unreceipted_sessions(&self) -> Result { + let sessions = { + let snapshot = self.lcm_read_snapshot().await?; + let mut rows = snapshot + .query( + "SELECT DISTINCT provider, session_id + FROM lcm_raw_messages + WHERE json_extract( + metadata_json, + '$.ingest_protection.sanitization_receipt' + ) IS NULL + ORDER BY provider, session_id", + (), + ) + .await?; + let mut sessions: Vec<(String, String)> = Vec::new(); + while let Some(row) = rows.next().await? { + sessions.push((row.get(0)?, row.get(1)?)); + } + sessions + }; + let mut protected_rows = 0_u64; + for (provider, session_id) in sessions { + protected_rows += self + .lcm_protect_session_raw_messages(&provider, &session_id) + .await?; + } + Ok(protected_rows) + } + + async fn load_rescan_page(&self, after_store_id: i64) -> Result, LcmError> { + let snapshot = self.lcm_read_snapshot().await?; + let mut rows = snapshot + .query( + "SELECT raw.store_id, raw.provider, raw.message_id, raw.session_id, + raw.role, raw.ordinal, raw.timestamp, raw.content, + raw.storage_kind, raw.payload_ref, raw.metadata_json, + message.kind, message.model, message.tool_names, + message.source_path, message.source_offset + FROM lcm_raw_messages AS raw + LEFT JOIN session_messages AS message + ON message.provider = raw.provider + AND message.message_id = raw.message_id + WHERE raw.store_id > ?1 + ORDER BY raw.store_id + LIMIT ?2", + params![after_store_id, RESCAN_PAGE_LIMIT], + ) + .await?; + let mut page = Vec::new(); + while let Some(row) = rows.next().await? { + let storage_kind_text: String = row.get(8)?; + let storage_kind = LcmStorageKind::from_db(&storage_kind_text).ok_or_else(|| { + LcmError::Db(format!("invalid storage_kind: {storage_kind_text}")) + })?; + page.push(RescanRow { + store_id: row.get(0)?, + provider: row.get(1)?, + message_id: row.get(2)?, + session_id: row.get(3)?, + role: row.get(4)?, + ordinal: row.get(5)?, + timestamp: row.get(6)?, + content: row.get(7)?, + storage_kind, + payload_ref: row.get(9)?, + metadata_json: row.get(10)?, + projection_kind: row.get(11)?, + projection_model: row.get(12)?, + projection_tool_names: row.get(13)?, + projection_source_path: row.get(14)?, + projection_source_offset: row.get(15)?, + }); + } + Ok(page) + } + + /// Recovers the at-rest body this row actually serves: inline content, or + /// the verified external payload bytes. + async fn rescan_input( + &self, + storage_root: &Path, + row: &RescanRow, + ) -> Result { + match row.storage_kind { + LcmStorageKind::Inline => { + let text = row + .content + .clone() + .ok_or(LcmError::PayloadIntegrityMismatch)?; + Ok(RescanInput::Body { + text, + replaced_payload_ref: None, + }) + } + LcmStorageKind::External => { + let payload_ref = row.payload_ref.as_deref().ok_or_else(|| { + LcmError::Db("external raw message carries no payload_ref".to_owned()) + })?; + let metadata = { + let snapshot = self.lcm_read_snapshot().await?; + match payload::load_payload_metadata(&snapshot, payload_ref).await { + Ok(metadata) => metadata, + Err(LcmError::PayloadNotFound | LcmError::PayloadGcd) => { + return Ok(RescanInput::PayloadUnavailable); + } + Err(error) => return Err(error), + } + }; + let byte_count = usize::try_from(metadata.byte_count) + .map_err(|_| LcmError::PayloadIntegrityMismatch)?; + let char_count = usize::try_from(metadata.char_count) + .map_err(|_| LcmError::PayloadIntegrityMismatch)?; + match payload::read_verified_payload_content( + storage_root, + payload_ref, + &metadata.content_hash, + byte_count, + char_count, + ) { + Ok(text) => Ok(RescanInput::Body { + text, + replaced_payload_ref: Some(payload_ref.to_owned()), + }), + Err(LcmError::PayloadMissing | LcmError::PayloadGcd) => { + Ok(RescanInput::PayloadUnavailable) + } + Err(error) => Err(error), + } + } + } + } + + /// Re-ingests one dirty row through the canonical staging and commit + /// path, resynchronizes its projection twin, and tombstones a replaced + /// external payload so the superseded bytes leave the disk. + async fn remediate_row( + &self, + storage_root: &Path, + row: &RescanRow, + text: String, + provider_metadata_json: Option, + replaced_payload_ref: Option, + ) -> Result<(), LcmError> { + let record = SessionMessageRecord { + provider: row.provider.clone(), + message_id: row.message_id.clone(), + session_id: row.session_id.clone(), + role: row.role.clone(), + timestamp: row.timestamp, + ordinal: row.ordinal, + text, + kind: row.projection_kind.clone(), + model: row.projection_model.clone(), + tool_names: row.projection_tool_names.clone(), + source_path: row.projection_source_path.clone(), + source_offset: row.projection_source_offset, + metadata_json: provider_metadata_json, + }; + let mut payload_rollback = + payload::PayloadFileRollback::begin_cancellation_safe(storage_root); + let staged = raw::stage_raw_message_with_payload_tracked( + storage_root, + &record, + &mut payload_rollback, + )?; + let transaction = self + .begin_write_transaction() + .await + .map_err(|error| LcmError::Db(error.to_string()))?; + let upsert = raw::commit_staged_raw_message(&transaction, &record, staged).await?; + transaction + .execute( + "UPDATE session_messages SET text = ?3, metadata_json = ?4 + WHERE provider = ?1 AND message_id = ?2", + params![ + record.provider.as_str(), + record.message_id.as_str(), + upsert.projection_text.as_str(), + upsert.projection_metadata_json.as_deref(), + ], + ) + .await?; + // An external row remediates only when its body changed, and payload + // refs are content-addressed, so the re-ingest can never reuse the + // replaced ref: the superseded payload is always safe to delete. + if let Some(old_ref) = replaced_payload_ref.as_deref() { + payload::delete_external_payload_in_transaction( + &transaction, + storage_root, + old_ref, + &DeleteOpts { + rewrite_placeholders: false, + remove_file: true, + verify_hash: false, + }, + ) + .await?; + } + transaction.commit().await?; + payload_rollback.disarm(); + if let Some(old_ref) = replaced_payload_ref.as_deref() { + let transaction = self + .begin_write_transaction() + .await + .map_err(|error| LcmError::Db(error.to_string()))?; + gc::drain_pending_payload_delete_in_transaction(&transaction, storage_root, old_ref) + .await?; + transaction.commit().await?; + } + Ok(()) + } +} + +/// Returns whether the current detector would change this row's served body +/// or provider metadata. A payload the detector refuses to re-evaluate fails +/// the rescan with a typed error instead of passing as clean. +fn requires_remediation(text: &str, provider_metadata: Option<&str>) -> Result { + let sanitization = + sanitize_lcm_payload_text(text).map_err(|error| LcmError::SanitizationRefused { + reason: format!("at-rest LCM privacy rescan refused a stored payload: {error}"), + })?; + if sanitization.sanitized_text() != text { + return Ok(true); + } + match provider_metadata { + Some(metadata) => raw::provider_metadata_requires_resanitization(metadata), + None => Ok(false), + } +} + +/// Recovers the provider metadata a re-ingest must carry: the stored metadata +/// without the `ingest_protection` envelope the ingest path re-derives. +/// +/// Whole-message external rows never persisted provider metadata (their +/// stored metadata is the payload envelope ingest builds fresh), so they +/// re-ingest with none — exactly what ingest produced the first time. +fn stored_provider_metadata(row: &RescanRow) -> Result, LcmError> { + if row.storage_kind == LcmStorageKind::External { + return Ok(None); + } + let Some(metadata_json) = row.metadata_json.as_deref() else { + return Ok(None); + }; + let mut metadata = + serde_json::from_str::(metadata_json).map_err(|error| { + LcmError::SanitizationRefused { + reason: format!("stored LCM metadata is not valid JSON: {error}"), + } + })?; + let object = metadata + .as_object_mut() + .ok_or_else(|| LcmError::SanitizationRefused { + reason: "stored LCM metadata must be a JSON object".to_owned(), + })?; + object.remove("ingest_protection"); + if object.is_empty() { + return Ok(None); + } + serde_json::to_string(&metadata) + .map(Some) + .map_err(|error| LcmError::Db(format!("LCM metadata encoding failed: {error}"))) +} diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs index 09e181379..54798c80f 100644 --- a/crates/tracedecay-global-db/src/tests.rs +++ b/crates/tracedecay-global-db/src/tests.rs @@ -7,6 +7,8 @@ use super::{ pub mod harness; #[cfg(test)] +mod lcm_privacy_rescan; +#[cfg(test)] mod lcm_schema; #[cfg(test)] mod session_sync; diff --git a/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs new file mode 100644 index 000000000..c0d4bc1e0 --- /dev/null +++ b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs @@ -0,0 +1,400 @@ +//! At-rest LCM privacy rescan: legacy rows written under older detector +//! rules are re-scanned and remediated through the canonical ingest path. + +use serde_json::{Value, json}; +use tracedecay_domain::{ + ComponentVersion, PayloadReferenceV1, SanitizationReceiptId, SanitizationReceiptRefV1, + SanitizationReceiptV1, SanitizerDispositionV1, SensitivityV1, +}; +use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::privacy::{ + LCM_PAYLOAD_SANITIZER_VERSION_V1, lcm_payload_detector_revision, sanitize_lcm_payload_text, +}; +use tracedecay_sessions::retrieval_content::projected_content_hash; +use tracedecay_sessions::runtime::SessionMessageRecord; +use tracedecay_sessions::runtime::lcm::{payload, schema}; + +use crate::LcmPrivacyRescanOutcomeV1; +use crate::registered_lcm_privacy::LCM_PRIVACY_RESCAN_META_KEY; +use crate::tests::harness::RegisteredGlobalDbHarness; + +fn secret() -> String { + ["sk-at-rest-rescan-secret-", "1234567890abcdef"].concat() +} + +/// A receipt exactly as an older binary bound it: the stored bytes are the +/// receipt's payload, accepted as non-sensitive, under the pinned sanitizer +/// contract — but the current detector rules never evaluated them. +fn legacy_receipt(content: &str) -> SanitizationReceiptV1 { + let payload_reference = PayloadReferenceV1::for_payload(&Value::String(content.to_owned())) + .expect("legacy payload reference"); + let sanitizer_version = + ComponentVersion::new(LCM_PAYLOAD_SANITIZER_VERSION_V1).expect("pinned sanitizer contract"); + let receipt_id = SanitizationReceiptId::new(format!( + "privacy.lcm-payload.v1.{}", + payload_reference.digest().as_str() + )) + .expect("legacy receipt id"); + SanitizationReceiptV1::new( + SanitizationReceiptRefV1::new(receipt_id, sanitizer_version).expect("legacy receipt ref"), + SanitizerDispositionV1::Accepted, + SensitivityV1::NonSensitive, + Some(payload_reference), + ) + .expect("legacy receipt") +} + +async fn seed_session(harness: &RegisteredGlobalDbHarness, session_id: &str) { + harness + .registered + .writer_connection() + .expect("writer") + .execute( + "INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES ('cursor', ?1, '/tmp/project', '/tmp/project')", + params![session_id], + ) + .await + .expect("seed session"); +} + +/// Persists one inline raw row plus its projection twin exactly as an older +/// ingest could have: receipt-bound bytes the current rules never evaluated. +async fn seed_legacy_inline_row( + harness: &RegisteredGlobalDbHarness, + session_id: &str, + message_id: &str, + content: &str, +) { + let metadata = json!({ + "fixture": "legacy-inline", + "ingest_protection": { "sanitization_receipt": legacy_receipt(content) } + }) + .to_string(); + let writer = harness.registered.writer_connection().expect("writer"); + writer + .execute( + "INSERT INTO session_messages(provider, message_id, session_id, role, ordinal, text) + VALUES ('cursor', ?1, ?2, 'user', 1, ?3)", + params![message_id, session_id, content], + ) + .await + .expect("seed legacy projection twin"); + writer + .execute( + "INSERT INTO lcm_raw_messages( + provider, message_id, session_id, role, ordinal, timestamp, + content, content_hash, storage_kind, payload_ref, snippet_text, + index_text, legacy_source, legacy_truncated, metadata_json + ) + VALUES ('cursor', ?1, ?2, 'user', 1, 10, ?3, ?4, 'inline', NULL, ?3, ?3, 0, 0, ?5)", + params![ + message_id, + session_id, + content, + projected_content_hash(content), + metadata + ], + ) + .await + .expect("seed legacy inline raw row"); +} + +/// Persists one external raw row whose at-rest payload file holds bytes the +/// current detector would redact. +async fn seed_legacy_external_row( + harness: &RegisteredGlobalDbHarness, + session_id: &str, + message_id: &str, + content: &str, +) -> String { + let storage_root = harness + .registered + .db_path() + .parent() + .expect("registered database storage root") + .to_path_buf(); + let mut rollback = payload::PayloadFileRollback::begin_cancellation_safe(&storage_root); + let payload_ref = payload::write_external_payload_tracked( + &storage_root, + payload::ExternalPayloadWrite { + provider: "cursor", + session_id, + message_id, + kind: "tool_output", + content, + metadata_json: None, + }, + &mut rollback, + ) + .expect("write legacy external payload"); + let placeholder = format!( + "[Externalized LCM ingest payload: kind=tool_output; field=content; chars={}; bytes={}; ref={}]", + payload_ref.char_count, payload_ref.byte_count, payload_ref.payload_ref + ); + let metadata = json!({ + "external_payload": true, + "payload_ref": payload_ref.payload_ref, + "kind": "tool_output", + "byte_count": payload_ref.byte_count, + "char_count": payload_ref.char_count, + "sha256": payload_ref.content_hash, + "ingest_protection": { "sanitization_receipt": legacy_receipt(content) } + }) + .to_string(); + let transaction = harness + .registered + .begin_write_transaction() + .await + .expect("seed transaction"); + payload::upsert_payload_metadata(&transaction, &payload_ref) + .await + .expect("seed external payload metadata"); + transaction + .execute( + "INSERT INTO lcm_raw_messages( + provider, message_id, session_id, role, ordinal, timestamp, + content, content_hash, storage_kind, payload_ref, snippet_text, + index_text, legacy_source, legacy_truncated, metadata_json + ) + VALUES ('cursor', ?1, ?2, 'user', 2, 20, NULL, ?3, 'external', ?4, ?5, ?5, 0, 0, ?6)", + params![ + message_id, + session_id, + payload_ref.content_hash.as_str(), + payload_ref.payload_ref.as_str(), + placeholder.as_str(), + metadata + ], + ) + .await + .expect("seed legacy external raw row"); + transaction.commit().await.expect("commit seed"); + rollback.disarm(); + payload_ref.payload_ref +} + +/// Persists one projection-landed row without a sanitization receipt: the +/// shape the bulk observation rebuild writes and the protect pass owns. +async fn seed_unreceipted_row( + harness: &RegisteredGlobalDbHarness, + session_id: &str, + message_id: &str, + content: &str, +) { + let writer = harness.registered.writer_connection().expect("writer"); + writer + .execute( + "INSERT INTO session_messages(provider, message_id, session_id, role, ordinal, text) + VALUES ('cursor', ?1, ?2, 'assistant', 3, ?3)", + params![message_id, session_id, content], + ) + .await + .expect("seed unreceipted projection twin"); + writer + .execute( + "INSERT INTO lcm_raw_messages( + provider, message_id, session_id, role, ordinal, timestamp, + content, content_hash, storage_kind, payload_ref, snippet_text, + index_text, legacy_source, legacy_truncated, metadata_json + ) + VALUES ('cursor', ?1, ?2, 'assistant', 3, 30, ?3, ?4, 'inline', NULL, ?3, ?3, 0, 0, NULL)", + params![ + message_id, + session_id, + content, + projected_content_hash(content) + ], + ) + .await + .expect("seed unreceipted raw row"); +} + +async fn count_rows_holding(harness: &RegisteredGlobalDbHarness, needle: &str) -> i64 { + let snapshot = harness.registered.read_snapshot().await.expect("snapshot"); + let pattern = format!("%{needle}%"); + let mut rows = snapshot + .query( + "SELECT + (SELECT COUNT(*) FROM lcm_raw_messages + WHERE COALESCE(content, '') LIKE ?1 + OR snippet_text LIKE ?1 + OR index_text LIKE ?1 + OR COALESCE(metadata_json, '') LIKE ?1) + + (SELECT COUNT(*) FROM session_messages + WHERE text LIKE ?1 OR COALESCE(metadata_json, '') LIKE ?1)", + params![pattern], + ) + .await + .expect("count query"); + rows.next() + .await + .expect("count row") + .expect("count present") + .get(0) + .expect("count value") +} + +fn payload_dir_holds(storage_root: &std::path::Path, needle: &str) -> bool { + let dir = payload::payload_dir(storage_root); + let Ok(entries) = std::fs::read_dir(&dir) else { + return false; + }; + entries.filter_map(Result::ok).any(|entry| { + std::fs::read(entry.path()) + .map(|bytes| String::from_utf8_lossy(&bytes).contains(needle)) + .unwrap_or(false) + }) +} + +#[tokio::test] +async fn at_rest_rescan_remediates_legacy_rows_and_settles_watermark() { + let harness = RegisteredGlobalDbHarness::open("lcm-privacy-rescan").await; + let storage_root = harness + .registered + .db_path() + .parent() + .expect("registered database storage root") + .to_path_buf(); + let session_id = "privacy-rescan-session"; + seed_session(&harness, session_id).await; + + // A message ingested through the current path stays untouched. + harness + .registered + .lcm_ingest_raw_message( + &storage_root, + &SessionMessageRecord { + provider: "cursor".to_owned(), + message_id: "clean-message".to_owned(), + session_id: session_id.to_owned(), + role: "user".to_owned(), + timestamp: Some(5), + ordinal: 0, + text: "the retry budget is three attempts".to_owned(), + kind: None, + model: None, + tool_names: None, + source_path: None, + source_offset: None, + metadata_json: None, + }, + ) + .await + .expect("ingest clean message"); + + let inline_content = format!( + "the deploy pipeline authenticates with api_key={}", + secret() + ); + let inline_sanitized = sanitize_lcm_payload_text(&inline_content).expect("evaluate fixture"); + assert_ne!( + inline_sanitized.sanitized_text(), + inline_content, + "the fixture must be a value the current detector redacts" + ); + seed_legacy_inline_row(&harness, session_id, "legacy-inline", &inline_content).await; + + let external_content = format!("captured tool output leaked api_key={}", secret()); + let old_payload_ref = + seed_legacy_external_row(&harness, session_id, "legacy-external", &external_content).await; + let old_payload_path = payload::payload_dir(&storage_root).join(&old_payload_ref); + assert!(old_payload_path.is_file(), "seeded payload must be at rest"); + + seed_unreceipted_row( + &harness, + session_id, + "unreceipted-message", + "projection-landed row without a receipt", + ) + .await; + + assert!(count_rows_holding(&harness, &secret()).await > 0); + assert!(payload_dir_holds(&storage_root, &secret())); + + let outcome = harness + .registered + .lcm_privacy_rescan_raw_messages() + .await + .expect("at-rest rescan"); + let LcmPrivacyRescanOutcomeV1::Completed(receipt) = outcome else { + panic!("first rescan must complete a full pass: {outcome:?}"); + }; + assert_eq!(receipt.detector_revision, lcm_payload_detector_revision()); + assert_eq!(receipt.protected_rows, 1); + assert_eq!(receipt.scanned_rows, 4); + assert_eq!(receipt.clean_rows, 2); + assert_eq!(receipt.remediated_rows, 2); + assert_eq!(receipt.unavailable_payload_rows, 0); + + // The detector hit is gone from every at-rest surface: raw rows, the + // projection twin, and the payload directory (the replaced payload file + // is deleted, not merely superseded). + assert_eq!(count_rows_holding(&harness, &secret()).await, 0); + assert!(!payload_dir_holds(&storage_root, &secret())); + assert!( + !old_payload_path.exists(), + "replaced payload must be deleted" + ); + + // The remediated inline row still serves through the verified raw-read + // authority, with a fresh receipt and its provider metadata retained. + let snapshot = harness.registered.read_snapshot().await.expect("snapshot"); + let remediated = schema::load_raw_message(&snapshot, "cursor", "legacy-inline") + .await + .expect("verified load of remediated row") + .expect("remediated row still serves"); + assert!( + remediated + .content + .contains("the deploy pipeline authenticates") + ); + assert!(!remediated.content.contains(&secret())); + let metadata: Value = + serde_json::from_str(remediated.metadata_json.as_deref().expect("metadata")) + .expect("metadata JSON"); + assert_eq!(metadata["fixture"], json!("legacy-inline")); + assert_eq!(metadata["ingest_protection"]["redacted"], json!(true)); + + // The unreceipted row now carries a receipt bound by the protect pass. + let protected = schema::load_raw_message(&snapshot, "cursor", "unreceipted-message") + .await + .expect("verified load of protected row") + .expect("protected row serves"); + assert_eq!(protected.content, "projection-landed row without a receipt"); + drop(snapshot); + + // A second request is answered by the watermark without scanning. + assert_eq!( + harness + .registered + .lcm_privacy_rescan_raw_messages() + .await + .expect("watermarked rescan"), + LcmPrivacyRescanOutcomeV1::AlreadyCurrent + ); + + // A forced repeat pass (rule-refresh simulation: watermark cleared) finds + // the remediated store clean and settles nothing further. + let transaction = harness + .registered + .begin_write_transaction() + .await + .expect("watermark transaction"); + schema::clear_gc_meta(&transaction, LCM_PRIVACY_RESCAN_META_KEY) + .await + .expect("clear watermark"); + transaction.commit().await.expect("commit watermark clear"); + let outcome = harness + .registered + .lcm_privacy_rescan_raw_messages() + .await + .expect("repeat rescan"); + let LcmPrivacyRescanOutcomeV1::Completed(repeat) = outcome else { + panic!("cleared watermark must force a full pass: {outcome:?}"); + }; + assert_eq!(repeat.scanned_rows, 4); + assert_eq!(repeat.clean_rows, 4); + assert_eq!(repeat.remediated_rows, 0); + assert_eq!(repeat.protected_rows, 0); +} diff --git a/crates/tracedecay-runtime-core/src/privacy/mod.rs b/crates/tracedecay-runtime-core/src/privacy/mod.rs index 5cc3eeb47..e37024cdc 100644 --- a/crates/tracedecay-runtime-core/src/privacy/mod.rs +++ b/crates/tracedecay-runtime-core/src/privacy/mod.rs @@ -59,8 +59,8 @@ pub use structured::{StructuredTextFormatV1, sanitize_provider_metadata_json}; pub use structured_text::{ CODE_SOURCE_SANITIZER_VERSION_V1, CodeSourceSanitizationV1, CodeSourceShapeV1, LCM_PAYLOAD_SANITIZER_VERSION_V1, LcmPayloadSanitizationV1, bind_sanitized_lcm_payload_text, - quarantine_lcm_payload_text, sanitize_code_source_bytes, sanitize_lcm_payload_text, - sanitize_provider_metadata_text, + lcm_payload_detector_revision, quarantine_lcm_payload_text, sanitize_code_source_bytes, + sanitize_lcm_payload_text, sanitize_provider_metadata_text, }; pub use tracedecay_capture::{ ClaudeRecordParseErrorV1, MAX_OBSERVATION_RECORD_BYTES, ObservationRecordParseErrorV1, diff --git a/crates/tracedecay-runtime-core/src/privacy/rules.rs b/crates/tracedecay-runtime-core/src/privacy/rules.rs index 1e4bda160..b2811ceeb 100644 --- a/crates/tracedecay-runtime-core/src/privacy/rules.rs +++ b/crates/tracedecay-runtime-core/src/privacy/rules.rs @@ -42,6 +42,19 @@ const VENDORED_SOURCE: &str = "vendor/gitleaks/gitleaks.toml"; const SUPPLEMENT_RULES_TOML: &str = include_str!("rules/supplement.toml"); const SUPPLEMENT_SOURCE: &str = "supplement.toml"; +/// The exact rule-document bytes the detector compiles, in evaluation order. +/// +/// Revision-sensitive consumers (the at-rest privacy rescan watermark) bind to +/// this data rather than to the pinned sanitizer contract string, because a +/// vendored-catalogue or supplement refresh changes what the detector finds +/// without changing the receipt contract. +pub(crate) const fn rule_document_bytes() -> [&'static [u8]; 2] { + [ + VENDORED_RULES_TOML.as_bytes(), + SUPPLEMENT_RULES_TOML.as_bytes(), + ] +} + /// Upstream's generated "context" rules all open with this preamble: an /// unanchored run of identifier bytes ahead of the provider keyword. Its /// presence is what distinguishes a rule that matches `provider_key = ` diff --git a/crates/tracedecay-runtime-core/src/privacy/structured_text.rs b/crates/tracedecay-runtime-core/src/privacy/structured_text.rs index fedd0d0d6..844eaf2b9 100644 --- a/crates/tracedecay-runtime-core/src/privacy/structured_text.rs +++ b/crates/tracedecay-runtime-core/src/privacy/structured_text.rs @@ -15,6 +15,7 @@ use std::collections::BTreeSet; use std::ops::Range; +use std::sync::OnceLock; use serde_json::Value; use sha2::{Digest, Sha256}; @@ -659,6 +660,22 @@ pub fn sanitize_code_source_bytes( }) } +/// Effective LCM-payload detector revision: the pinned sanitizer contract +/// bound to a digest of the compiled credential rule documents. +/// +/// The contract string names the receipt shape and never changes with a rule +/// refresh, so it cannot tell an at-rest rescan whether previously accepted +/// bytes were evaluated under the current rules. This revision changes exactly +/// when the vendored catalogue or the local supplement changes, which is +/// exactly when a completed-rescan watermark must invalidate. +pub fn lcm_payload_detector_revision() -> &'static str { + static REVISION: OnceLock = OnceLock::new(); + REVISION.get_or_init(|| { + let digest = super::length_prefixed_sha256_hex(&super::rules::rule_document_bytes()); + format!("{LCM_PAYLOAD_SANITIZER_VERSION_V1}+rules.{}", &digest[..16]) + }) +} + pub fn sanitize_lcm_payload_text(raw: &str) -> Result { let (sanitized_text, findings) = detect_lcm_payload(raw)?; bind_lcm_payload(raw, sanitized_text, findings) diff --git a/crates/tracedecay-sessions/src/runtime/lcm/raw.rs b/crates/tracedecay-sessions/src/runtime/lcm/raw.rs index 31a6550c3..eb759ee6e 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/raw.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/raw.rs @@ -802,6 +802,29 @@ fn safe_placeholder_metadata(value: &str) -> String { const MAX_PROVIDER_METADATA_BYTES: u64 = 1_048_576; +/// Returns whether the current detector would change this provider metadata. +/// +/// This is the at-rest rescan's change probe for the exact transformation +/// ingest applies through [`protected_metadata_json`]: metadata persisted +/// under older detector rules is dirty when re-sanitizing it under the +/// current rules yields a different document. A document the sanitizer +/// refuses to evaluate is a typed refusal, never implicitly clean. +pub fn provider_metadata_requires_resanitization( + provider_metadata_json: &str, +) -> Result { + let original = serde_json::from_str::(provider_metadata_json).map_err(|error| { + LcmError::SanitizationRefused { + reason: format!("stored LCM provider metadata is not valid JSON: {error}"), + } + })?; + let sanitized = + sanitize_provider_metadata_json(provider_metadata_json, MAX_PROVIDER_METADATA_BYTES) + .ok_or_else(|| LcmError::SanitizationRefused { + reason: "LCM metadata sanitization failed".to_owned(), + })?; + Ok(sanitized != original) +} + /// Pure function of the provider metadata bytes: every failure is a /// deterministic content refusal, never an environmental fault. fn protected_metadata_json( diff --git a/src/daemon/privacy_remediation.rs b/src/daemon/privacy_remediation.rs index 7125de3c4..5f7cf9371 100644 --- a/src/daemon/privacy_remediation.rs +++ b/src/daemon/privacy_remediation.rs @@ -2,10 +2,11 @@ //! //! Project-open spawns one bounded background rescan per adopted project //! store after fail-closed admission has finished; it never blocks admission -//! or retrieval. The rescan re-runs the current in-process detector over -//! persisted project-memory facts, quarantines detector hits, and settles -//! every mutation through the canonical curation authority so durable -//! curation receipts record what changed. No scanner binary runs. +//! or retrieval. The rescan re-runs the current in-process detector over the +//! persisted stores it owns. Project-memory detector hits are terminally +//! quarantined so historical payloads are erased; LCM raw messages settle +//! through their canonical remediation authority. Durable receipts record +//! every mutation, and no scanner binary runs. use std::sync::Arc; @@ -15,10 +16,14 @@ use tracedecay_usecases::memory::{ }; use crate::errors::Result; +use crate::global_db::{LcmPrivacyRescanOutcomeV1, RegisteredGlobalDbLeaseV1}; use crate::tracedecay::TraceDecay; /// Spawns the bounded background rescan for one adopted project store. -pub(crate) fn spawn_project_memory_privacy_remediation(graph: Arc) { +pub(crate) fn spawn_at_rest_privacy_remediation( + graph: Arc, + session_db: RegisteredGlobalDbLeaseV1, +) { tokio::spawn(async move { let project = graph.project_root().display().to_string(); match run_project_memory_privacy_remediation(&graph).await { @@ -41,6 +46,28 @@ pub(crate) fn spawn_project_memory_privacy_remediation(graph: Arc) { ); } } + match session_db.lcm_privacy_rescan_raw_messages().await { + Ok(LcmPrivacyRescanOutcomeV1::AlreadyCurrent) => {} + Ok(LcmPrivacyRescanOutcomeV1::Completed(receipt)) => { + tracing::info!( + event = "lcm_privacy_remediation", + project = %project, + detector_revision = %receipt.detector_revision, + scanned_rows = receipt.scanned_rows, + clean_rows = receipt.clean_rows, + remediated_rows = receipt.remediated_rows, + protected_rows = receipt.protected_rows, + unavailable_payload_rows = receipt.unavailable_payload_rows, + ); + } + Err(error) => { + tracing::warn!( + event = "lcm_privacy_remediation_failed", + project = %project, + %error, + ); + } + } }); } diff --git a/src/daemon/project_open_owners.rs b/src/daemon/project_open_owners.rs index 0814c06c4..2b800b9a4 100644 --- a/src/daemon/project_open_owners.rs +++ b/src/daemon/project_open_owners.rs @@ -1088,9 +1088,10 @@ pub(super) async fn register_project_open_production_owners( // At-rest privacy remediation is bounded background work after fail-closed // admission; it never blocks admission or retrieval. - crate::daemon::privacy_remediation::spawn_project_memory_privacy_remediation(Arc::clone( - &graph, - )); + crate::daemon::privacy_remediation::spawn_at_rest_privacy_remediation( + Arc::clone(&graph), + session_db.clone(), + ); // Once-per-project-open adoption-eligibility census over the composed // capability catalog, recorded through the project-bound session