From 1c58d39caf5311ff30bb121cdc1e71ec68764632 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 19 Aug 2026 17:01:16 +0000 Subject: [PATCH 1/3] feat(privacy): rescan at-rest LCM store bytes under current detector Co-authored-by: Zack Jackson --- crates/tracedecay-global-db/src/lib.rs | 2 + .../src/registered_lcm.rs | 4 +- .../src/registered_lcm_privacy.rs | 478 ++++++++++++++++++ crates/tracedecay-global-db/src/tests.rs | 2 + .../src/tests/lcm_privacy_rescan.rs | 397 +++++++++++++++ .../src/privacy/mod.rs | 4 +- .../src/privacy/rules.rs | 13 + .../src/privacy/structured_text.rs | 17 + .../src/runtime/lcm/raw.rs | 24 + src/daemon/privacy_remediation.rs | 37 +- src/daemon/project_open_owners.rs | 7 +- 11 files changed, 973 insertions(+), 12 deletions(-) create mode 100644 crates/tracedecay-global-db/src/registered_lcm_privacy.rs create mode 100644 crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs 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..9729a247b --- /dev/null +++ b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs @@ -0,0 +1,478 @@ +//! 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-runs the current in-process detector over +//! every at-rest raw-message body — inline `content` and whole-message +//! external payload bytes — and re-ingests each hit through the exact staging +//! and commit path new ingest uses, so redaction, externalization, +//! quarantine, receipts, and FTS maintenance all follow the one canonical +//! sanitizer. The `session_messages` projection twin is resynchronized from +//! the re-ingest's projection output, and 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 the effective +//! detector revision (sanitizer contract + compiled rule-document digest), 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, so the rerun settles nothing it already fixed. +//! Rows the sanitizer cannot re-evaluate fail the run with a typed error — +//! never a silent skip — and the watermark stays unset until a pass covers +//! every row. +//! +//! Boundary: media-span payload files are byte ranges the ingest scan already +//! evaluated inside their owning message text before externalizing them; the +//! rescan re-evaluates every message body (where those placeholders live), +//! not the extracted media bytes themselves. Unreceipted rows are first +//! protected through the existing [`RegisteredGlobalDb::lcm_protect_session_raw_messages`] +//! pass rather than a duplicate path. + +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 tracedecay_runtime_core::db::engine::params; + +use super::RegisteredGlobalDb; + +/// Watermark row in `lcm_gc_meta`: the effective detector revision whose +/// rescan last completed over this store. +const LCM_PRIVACY_RESCAN_META_KEY: &str = "privacy_rescan_completed_revision"; + +/// One page of raw rows per authority read. +const RESCAN_PAGE_LIMIT: usize = 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 effective + /// detector revision; nothing was scanned. + AlreadyCurrent { detector_revision: String }, + /// 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); they serve only a placeholder, 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, + has_projection: bool, + 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 + /// effective 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 { + detector_revision: detector_revision.to_owned(), + }); + } + } + + 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 = unavailable_payload_rows.saturating_add(1); + continue; + } + }; + scanned_rows = scanned_rows.saturating_add(1); + if !row_requires_remediation(&row, &text)? { + clean_rows = clean_rows.saturating_add(1); + continue; + } + self.remediate_row(&storage_root, &row, text, replaced_payload_ref) + .await?; + remediated_rows = remediated_rows.saturating_add(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 = protected_rows.saturating_add( + 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 limit = i64::try_from(RESCAN_PAGE_LIMIT) + .map_err(|error| LcmError::Db(format!("invalid rescan page limit: {error}")))?; + 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.message_id IS NOT NULL, + 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, 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)?, + has_projection: row.get::(11)? != 0, + projection_kind: row.get(12)?, + projection_model: row.get(13)?, + projection_tool_names: row.get(14)?, + projection_source_path: row.get(15)?, + projection_source_offset: row.get(16)?, + }); + } + 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: &std::path::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: &std::path::Path, + row: &RescanRow, + text: String, + 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: stored_provider_metadata(row)?, + }; + 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?; + if row.has_projection { + 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?; + } + let replaced = match replaced_payload_ref { + Some(old_ref) => { + let mut rows = transaction + .query( + "SELECT payload_ref FROM lcm_raw_messages + WHERE provider = ?1 AND message_id = ?2", + params![record.provider.as_str(), record.message_id.as_str()], + ) + .await?; + let current: Option = rows + .next() + .await? + .ok_or_else(|| { + LcmError::Db("remediated raw message disappeared mid-commit".to_owned()) + })? + .get(0)?; + drop(rows); + if current.as_deref() == Some(old_ref.as_str()) { + None + } else { + payload::delete_external_payload_in_transaction( + &transaction, + storage_root, + &old_ref, + &DeleteOpts { + rewrite_placeholders: false, + remove_file: true, + verify_hash: false, + }, + ) + .await?; + Some(old_ref) + } + } + None => None, + }; + transaction.commit().await?; + payload_rollback.disarm(); + if let Some(old_ref) = replaced { + 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 body or metadata the detector refuses to +/// re-evaluate fails the rescan with a typed error instead of passing as +/// clean. +fn row_requires_remediation(row: &RescanRow, text: &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.findings().is_empty() || sanitization.sanitized_text() != text { + return Ok(true); + } + match stored_provider_metadata(row)? { + 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..dccf23676 --- /dev/null +++ b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs @@ -0,0 +1,397 @@ +//! 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::tests::harness::RegisteredGlobalDbHarness; + +/// The rescan watermark row this suite clears to force a repeat pass. +const RESCAN_META_KEY: &str = "privacy_rescan_completed_revision"; + +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 { + detector_revision: lcm_payload_detector_revision().to_owned(), + } + ); + + // 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, 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..ecc34c407 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/raw.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/raw.rs @@ -802,6 +802,30 @@ 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..f4af3125c 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 From 3dcaa684182f503ce80937a66bda68209960e180 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 19 Aug 2026 17:44:51 +0000 Subject: [PATCH 2/3] style(privacy): format at-rest LCM rescan module and test Co-authored-by: Zack Jackson --- .../src/registered_lcm_privacy.rs | 17 +++++----- .../src/tests/lcm_privacy_rescan.rs | 31 ++++++++++++------- .../src/runtime/lcm/raw.rs | 11 +++---- 3 files changed, 34 insertions(+), 25 deletions(-) diff --git a/crates/tracedecay-global-db/src/registered_lcm_privacy.rs b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs index 9729a247b..886122374 100644 --- a/crates/tracedecay-global-db/src/registered_lcm_privacy.rs +++ b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs @@ -28,6 +28,7 @@ //! protected through the existing [`RegisteredGlobalDb::lcm_protect_session_raw_messages`] //! pass rather than a duplicate 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, @@ -37,7 +38,6 @@ use tracedecay_sessions::runtime::{ raw, schema, }, }; -use tracedecay_runtime_core::db::engine::params; use super::RegisteredGlobalDb; @@ -354,8 +354,11 @@ impl RegisteredGlobalDb { }; 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 staged = raw::stage_raw_message_with_payload_tracked( + storage_root, + &record, + &mut payload_rollback, + )?; let transaction = self .begin_write_transaction() .await @@ -463,11 +466,11 @@ fn stored_provider_metadata(row: &RescanRow) -> Result, LcmError> reason: format!("stored LCM metadata is not valid JSON: {error}"), } })?; - let object = metadata.as_object_mut().ok_or_else(|| { - LcmError::SanitizationRefused { + 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); diff --git a/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs index dccf23676..1e8e5c688 100644 --- a/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs +++ b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs @@ -30,8 +30,8 @@ fn secret() -> String { 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 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() @@ -244,9 +244,7 @@ fn payload_dir_holds(storage_root: &std::path::Path, needle: &str) -> bool { }; entries.filter_map(Result::ok).any(|entry| { std::fs::read(entry.path()) - .map(|bytes| { - String::from_utf8_lossy(&bytes).contains(needle) - }) + .map(|bytes| String::from_utf8_lossy(&bytes).contains(needle)) .unwrap_or(false) }) } @@ -287,7 +285,10 @@ async fn at_rest_rescan_remediates_legacy_rows_and_settles_watermark() { .await .expect("ingest clean message"); - let inline_content = format!("the deploy pipeline authenticates with api_key={}", secret()); + 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(), @@ -333,7 +334,10 @@ async fn at_rest_rescan_remediates_legacy_rows_and_settles_watermark() { // 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"); + 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. @@ -342,12 +346,15 @@ async fn at_rest_rescan_remediates_legacy_rows_and_settles_watermark() { .await .expect("verified load of remediated row") .expect("remediated row still serves"); - assert!(remediated.content.contains("the deploy pipeline authenticates")); + 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"); + 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)); diff --git a/crates/tracedecay-sessions/src/runtime/lcm/raw.rs b/crates/tracedecay-sessions/src/runtime/lcm/raw.rs index ecc34c407..eb759ee6e 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/raw.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/raw.rs @@ -812,12 +812,11 @@ const MAX_PROVIDER_METADATA_BYTES: u64 = 1_048_576; 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 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 { From d22241a8850d844f1652aa65a2c347ce10d50e91 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 19 Aug 2026 18:07:32 +0000 Subject: [PATCH 3/3] refactor(privacy): trim at-rest LCM rescan to its load-bearing shape The remediation commit no longer re-reads the row it just upserted to guard payload deletion: an external row remediates only when its body changed and payload refs are content-addressed, so the replacement ref can never equal the replaced one. The dirty probe compares at-rest bytes directly (a finding that changes no byte changes no at-rest state), parses stored provider metadata once per row instead of twice, and the projection-twin resync runs unconditionally since an absent twin matches no rows. AlreadyCurrent no longer carries the detector revision no caller reads, and the watermark key is shared with the test instead of duplicated. Co-authored-by: Zack Jackson --- .../src/registered_lcm_privacy.rs | 210 ++++++++---------- .../src/tests/lcm_privacy_rescan.rs | 10 +- src/daemon/privacy_remediation.rs | 2 +- 3 files changed, 96 insertions(+), 126 deletions(-) diff --git a/crates/tracedecay-global-db/src/registered_lcm_privacy.rs b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs index 886122374..ad6a3007c 100644 --- a/crates/tracedecay-global-db/src/registered_lcm_privacy.rs +++ b/crates/tracedecay-global-db/src/registered_lcm_privacy.rs @@ -2,31 +2,28 @@ //! //! 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-runs the current in-process detector over -//! every at-rest raw-message body — inline `content` and whole-message -//! external payload bytes — and re-ingests each hit through the exact staging -//! and commit path new ingest uses, so redaction, externalization, -//! quarantine, receipts, and FTS maintenance all follow the one canonical -//! sanitizer. The `session_messages` projection twin is resynchronized from -//! the re-ingest's projection output, and a replaced external payload file is +//! 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 the effective -//! detector revision (sanitizer contract + compiled rule-document digest), 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, so the rerun settles nothing it already fixed. -//! Rows the sanitizer cannot re-evaluate fail the run with a typed error — -//! never a silent skip — and the watermark stays unset until a pass covers -//! every row. +//! 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. //! -//! Boundary: media-span payload files are byte ranges the ingest scan already -//! evaluated inside their owning message text before externalizing them; the -//! rescan re-evaluates every message body (where those placeholders live), -//! not the extracted media bytes themselves. Unreceipted rows are first -//! protected through the existing [`RegisteredGlobalDb::lcm_protect_session_raw_messages`] -//! pass rather than a duplicate path. +//! 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}; @@ -41,19 +38,19 @@ use tracedecay_sessions::runtime::{ use super::RegisteredGlobalDb; -/// Watermark row in `lcm_gc_meta`: the effective detector revision whose -/// rescan last completed over this store. -const LCM_PRIVACY_RESCAN_META_KEY: &str = "privacy_rescan_completed_revision"; +/// 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: usize = 64; +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 effective - /// detector revision; nothing was scanned. - AlreadyCurrent { detector_revision: String }, + /// 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), } @@ -73,8 +70,8 @@ pub struct LcmPrivacyRescanReceiptV1 { /// before the scan. pub protected_rows: u64, /// External rows whose payload bytes are no longer at rest (offloaded or - /// collected); they serve only a placeholder, so there is nothing left to - /// rescan or disclose. + /// collected); only their placeholder remains, so there is nothing left + /// to rescan or disclose. pub unavailable_payload_rows: u64, } @@ -91,7 +88,6 @@ struct RescanRow { storage_kind: LcmStorageKind, payload_ref: Option, metadata_json: Option, - has_projection: bool, projection_kind: Option, projection_model: Option, projection_tool_names: Option, @@ -112,8 +108,8 @@ enum RescanInput { impl RegisteredGlobalDb { /// Rescans every persisted LCM raw-message body under the current - /// effective detector revision, remediating hits through the canonical - /// ingest path. Runs at most once per store per detector revision. + /// 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 { @@ -125,9 +121,7 @@ impl RegisteredGlobalDb { .as_deref() == Some(detector_revision) { - return Ok(LcmPrivacyRescanOutcomeV1::AlreadyCurrent { - detector_revision: detector_revision.to_owned(), - }); + return Ok(LcmPrivacyRescanOutcomeV1::AlreadyCurrent); } } @@ -153,18 +147,25 @@ impl RegisteredGlobalDb { replaced_payload_ref, } => (text, replaced_payload_ref), RescanInput::PayloadUnavailable => { - unavailable_payload_rows = unavailable_payload_rows.saturating_add(1); + unavailable_payload_rows += 1; continue; } }; - scanned_rows = scanned_rows.saturating_add(1); - if !row_requires_remediation(&row, &text)? { - clean_rows = clean_rows.saturating_add(1); + 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, replaced_payload_ref) - .await?; - remediated_rows = remediated_rows.saturating_add(1); + self.remediate_row( + &storage_root, + &row, + text, + provider_metadata, + replaced_payload_ref, + ) + .await?; + remediated_rows += 1; } } @@ -212,24 +213,20 @@ impl RegisteredGlobalDb { }; let mut protected_rows = 0_u64; for (provider, session_id) in sessions { - protected_rows = protected_rows.saturating_add( - self.lcm_protect_session_raw_messages(&provider, &session_id) - .await?, - ); + 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 limit = i64::try_from(RESCAN_PAGE_LIMIT) - .map_err(|error| LcmError::Db(format!("invalid rescan page limit: {error}")))?; 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.message_id IS NOT NULL, message.kind, message.model, message.tool_names, message.source_path, message.source_offset FROM lcm_raw_messages AS raw @@ -239,7 +236,7 @@ impl RegisteredGlobalDb { WHERE raw.store_id > ?1 ORDER BY raw.store_id LIMIT ?2", - params![after_store_id, limit], + params![after_store_id, RESCAN_PAGE_LIMIT], ) .await?; let mut page = Vec::new(); @@ -260,12 +257,11 @@ impl RegisteredGlobalDb { storage_kind, payload_ref: row.get(9)?, metadata_json: row.get(10)?, - has_projection: row.get::(11)? != 0, - projection_kind: row.get(12)?, - projection_model: row.get(13)?, - projection_tool_names: row.get(14)?, - projection_source_path: row.get(15)?, - projection_source_offset: row.get(16)?, + 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) @@ -275,7 +271,7 @@ impl RegisteredGlobalDb { /// the verified external payload bytes. async fn rescan_input( &self, - storage_root: &std::path::Path, + storage_root: &Path, row: &RescanRow, ) -> Result { match row.storage_kind { @@ -332,9 +328,10 @@ impl RegisteredGlobalDb { /// external payload so the superseded bytes leave the disk. async fn remediate_row( &self, - storage_root: &std::path::Path, + storage_root: &Path, row: &RescanRow, text: String, + provider_metadata_json: Option, replaced_payload_ref: Option, ) -> Result<(), LcmError> { let record = SessionMessageRecord { @@ -350,7 +347,7 @@ impl RegisteredGlobalDb { tool_names: row.projection_tool_names.clone(), source_path: row.projection_source_path.clone(), source_offset: row.projection_source_offset, - metadata_json: stored_provider_metadata(row)?, + metadata_json: provider_metadata_json, }; let mut payload_rollback = payload::PayloadFileRollback::begin_cancellation_safe(storage_root); @@ -364,64 +361,42 @@ impl RegisteredGlobalDb { .await .map_err(|error| LcmError::Db(error.to_string()))?; let upsert = raw::commit_staged_raw_message(&transaction, &record, staged).await?; - if row.has_projection { - 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?; + 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?; } - let replaced = match replaced_payload_ref { - Some(old_ref) => { - let mut rows = transaction - .query( - "SELECT payload_ref FROM lcm_raw_messages - WHERE provider = ?1 AND message_id = ?2", - params![record.provider.as_str(), record.message_id.as_str()], - ) - .await?; - let current: Option = rows - .next() - .await? - .ok_or_else(|| { - LcmError::Db("remediated raw message disappeared mid-commit".to_owned()) - })? - .get(0)?; - drop(rows); - if current.as_deref() == Some(old_ref.as_str()) { - None - } else { - payload::delete_external_payload_in_transaction( - &transaction, - storage_root, - &old_ref, - &DeleteOpts { - rewrite_placeholders: false, - remove_file: true, - verify_hash: false, - }, - ) - .await?; - Some(old_ref) - } - } - None => None, - }; transaction.commit().await?; payload_rollback.disarm(); - if let Some(old_ref) = replaced { + 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) + gc::drain_pending_payload_delete_in_transaction(&transaction, storage_root, old_ref) .await?; transaction.commit().await?; } @@ -430,19 +405,18 @@ impl RegisteredGlobalDb { } /// Returns whether the current detector would change this row's served body -/// or provider metadata. A body or metadata the detector refuses to -/// re-evaluate fails the rescan with a typed error instead of passing as -/// clean. -fn row_requires_remediation(row: &RescanRow, text: &str) -> Result { +/// 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.findings().is_empty() || sanitization.sanitized_text() != text { + if sanitization.sanitized_text() != text { return Ok(true); } - match stored_provider_metadata(row)? { - Some(metadata) => raw::provider_metadata_requires_resanitization(&metadata), + match provider_metadata { + Some(metadata) => raw::provider_metadata_requires_resanitization(metadata), None => Ok(false), } } diff --git a/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs index 1e8e5c688..c0d4bc1e0 100644 --- a/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs +++ b/crates/tracedecay-global-db/src/tests/lcm_privacy_rescan.rs @@ -15,11 +15,9 @@ 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; -/// The rescan watermark row this suite clears to force a repeat pass. -const RESCAN_META_KEY: &str = "privacy_rescan_completed_revision"; - fn secret() -> String { ["sk-at-rest-rescan-secret-", "1234567890abcdef"].concat() } @@ -373,9 +371,7 @@ async fn at_rest_rescan_remediates_legacy_rows_and_settles_watermark() { .lcm_privacy_rescan_raw_messages() .await .expect("watermarked rescan"), - LcmPrivacyRescanOutcomeV1::AlreadyCurrent { - detector_revision: lcm_payload_detector_revision().to_owned(), - } + LcmPrivacyRescanOutcomeV1::AlreadyCurrent ); // A forced repeat pass (rule-refresh simulation: watermark cleared) finds @@ -385,7 +381,7 @@ async fn at_rest_rescan_remediates_legacy_rows_and_settles_watermark() { .begin_write_transaction() .await .expect("watermark transaction"); - schema::clear_gc_meta(&transaction, RESCAN_META_KEY) + schema::clear_gc_meta(&transaction, LCM_PRIVACY_RESCAN_META_KEY) .await .expect("clear watermark"); transaction.commit().await.expect("commit watermark clear"); diff --git a/src/daemon/privacy_remediation.rs b/src/daemon/privacy_remediation.rs index f4af3125c..5f7cf9371 100644 --- a/src/daemon/privacy_remediation.rs +++ b/src/daemon/privacy_remediation.rs @@ -47,7 +47,7 @@ pub(crate) fn spawn_at_rest_privacy_remediation( } } match session_db.lcm_privacy_rescan_raw_messages().await { - Ok(LcmPrivacyRescanOutcomeV1::AlreadyCurrent { .. }) => {} + Ok(LcmPrivacyRescanOutcomeV1::AlreadyCurrent) => {} Ok(LcmPrivacyRescanOutcomeV1::Completed(receipt)) => { tracing::info!( event = "lcm_privacy_remediation",