diff --git a/Cargo.lock b/Cargo.lock index 6c0918b22b..52f6940d31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5826,6 +5826,7 @@ dependencies = [ "serde_json", "sha2 0.11.0", "thiserror", + "tracing", "url", ] diff --git a/crates/tracedecay-domain/Cargo.toml b/crates/tracedecay-domain/Cargo.toml index cf844b1b39..fdb6f6b137 100644 --- a/crates/tracedecay-domain/Cargo.toml +++ b/crates/tracedecay-domain/Cargo.toml @@ -7,20 +7,13 @@ license = "MIT" description = "Pure domain contracts for TraceDecay V2" repository = "https://github.com/ScriptedAlchemy/tracedecay" -[features] -# The one sanctioned test-only observability hook: thread-local counters over -# the canonicalize-then-SHA256 boundaries (observation identity derivation, -# payload-content hashing, canonical command digests) so store-level tests can -# prove a terminally refused record is never re-decoded, re-derived, or -# re-hashed. Never enable in production builds. -identity-digest-probe = [] - [dependencies] schemars = "1.2.1" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.11" thiserror = "2" +tracing = "0.1" url = "2" [dev-dependencies] diff --git a/crates/tracedecay-domain/src/identity_digest_probe.rs b/crates/tracedecay-domain/src/identity_digest_probe.rs deleted file mode 100644 index fb3afdb589..0000000000 --- a/crates/tracedecay-domain/src/identity_digest_probe.rs +++ /dev/null @@ -1,68 +0,0 @@ -//! The one sanctioned test-only observability hook: thread-local counters -//! over the domain's canonicalize-then-SHA256 boundaries. -//! -//! Store-level tests use these counters to prove a terminally refused -//! observation is never re-decoded, re-derived, or re-hashed. Counting here — -//! at the functions that perform the work — rather than at a store dispatch -//! seam means the proof cannot regress silently when digest work moves -//! earlier than the dispatch (e.g. idempotency-identity hashing computed -//! before a submit that then early-exits), and it requires no test-only port -//! in any store crate. -//! -//! Three boundaries are counted: -//! -//! * **identity** — [`crate::observation`]'s `domain_digest`: every fresh -//! canonical observation-identity derivation and every stored-row -//! decode-time verification (`accepted_identity_digests`) funnels through -//! it, so a zero delta proves no identity material was re-canonicalized or -//! re-hashed on the observed thread. -//! * **payload** — `sha256_digest` under -//! [`crate::observation::PayloadReferenceV1::for_payload`]: the only -//! payload-content hash, so a zero delta proves no payload was -//! re-canonicalized or re-hashed. -//! * **canonical** — [`crate::research::canonical_sha256`]: every runtime -//! read and write command digest is computed through it on the dispatching -//! thread *before* the request crosses into the store runtime, so the exact -//! delta bounds the record work a call dispatched — a stored-row read that -//! would be decoded off-thread still costs its command digest here first. -//! -//! Counters are thread-local so parallel tests cannot bleed counts into each -//! other. Never enable the `identity-digest-probe` feature in production -//! builds. - -use std::cell::Cell; - -thread_local! { - static IDENTITY_DIGESTS: Cell = const { Cell::new(0) }; - static PAYLOAD_DIGESTS: Cell = const { Cell::new(0) }; - static CANONICAL_DIGESTS: Cell = const { Cell::new(0) }; -} - -pub(crate) fn record_identity() { - IDENTITY_DIGESTS.with(|digests| digests.set(digests.get() + 1)); -} - -pub(crate) fn record_payload() { - PAYLOAD_DIGESTS.with(|digests| digests.set(digests.get() + 1)); -} - -pub(crate) fn record_canonical() { - CANONICAL_DIGESTS.with(|digests| digests.set(digests.get() + 1)); -} - -/// Canonical observation-identity digests computed on the current thread so -/// far (fresh derivations and decode-time verifications alike). -pub fn identity_digests() -> u64 { - IDENTITY_DIGESTS.with(Cell::get) -} - -/// Canonical payload-content digests computed on the current thread so far. -pub fn payload_digests() -> u64 { - PAYLOAD_DIGESTS.with(Cell::get) -} - -/// Canonical manifest/command digests ([`crate::research::canonical_sha256`]) -/// computed on the current thread so far. -pub fn canonical_digests() -> u64 { - CANONICAL_DIGESTS.with(Cell::get) -} diff --git a/crates/tracedecay-domain/src/lib.rs b/crates/tracedecay-domain/src/lib.rs index bc63e02206..56d390f1cc 100644 --- a/crates/tracedecay-domain/src/lib.rs +++ b/crates/tracedecay-domain/src/lib.rs @@ -12,8 +12,6 @@ pub mod external_source; pub mod feedback; pub mod framed_log; pub mod git; -#[cfg(feature = "identity-digest-probe")] -pub mod identity_digest_probe; pub mod integration; pub mod memory; pub mod multi_root; diff --git a/crates/tracedecay-domain/src/observation.rs b/crates/tracedecay-domain/src/observation.rs index 7c68b2e97e..5710034257 100644 --- a/crates/tracedecay-domain/src/observation.rs +++ b/crates/tracedecay-domain/src/observation.rs @@ -2307,8 +2307,11 @@ fn domain_digest( domain: &[u8], value: &impl Serialize, ) -> Result { - #[cfg(feature = "identity-digest-probe")] - crate::identity_digest_probe::record_identity(); + tracing::trace!( + target: "tracedecay::observation_admission_work", + work = "identity_derivation", + "derive canonical observation identity" + ); let bytes = canonical_json_bytes(value).map_err(|_| ObservationContractError::CanonicalEncoding)?; let mut hasher = Sha256::new(); @@ -2351,8 +2354,11 @@ fn accepted_identity_digests( } fn sha256_digest(bytes: &[u8]) -> String { - #[cfg(feature = "identity-digest-probe")] - crate::identity_digest_probe::record_payload(); + tracing::trace!( + target: "tracedecay::observation_admission_work", + work = "payload_digest", + "digest canonical observation payload" + ); format_sha256(&Sha256::digest(bytes)) } diff --git a/crates/tracedecay-domain/src/research/canonical.rs b/crates/tracedecay-domain/src/research/canonical.rs index 6ed11a51a2..fc0e0500d3 100644 --- a/crates/tracedecay-domain/src/research/canonical.rs +++ b/crates/tracedecay-domain/src/research/canonical.rs @@ -33,8 +33,6 @@ pub fn canonical_json_value(value: &Value) -> Result { /// intermediate `serde_json::Value` tree is materialized, which matters for /// the six-figure element sets the code index digests on every publish. pub fn canonical_sha256(value: &T) -> Result { - #[cfg(feature = "identity-digest-probe")] - crate::identity_digest_probe::record_canonical(); let mut sink = BufferedSink::new(Sha256::new()); canonical_serializer::serialize_canonical(value, &mut sink)?; let digest = sink.finish().finalize(); diff --git a/crates/tracedecay-global-db/Cargo.toml b/crates/tracedecay-global-db/Cargo.toml index 5b999eaee0..a8a3948f96 100644 --- a/crates/tracedecay-global-db/Cargo.toml +++ b/crates/tracedecay-global-db/Cargo.toml @@ -55,11 +55,6 @@ tracedecay-graph-db = { path = "../tracedecay-graph-db", version = "0.1.0", feat # crate's WAL reclaim tests need to prove exclusive-maintenance truncation. tracedecay-runtime-core = { path = "../tracedecay-runtime-core", version = "0.1.0", features = ["test-helpers", "test-transport"] } tracedecay-sessions = { path = "../tracedecay-sessions", version = "0.1.0", features = ["test-helpers"] } -# The one sanctioned test-only probe: thread-local counters at the domain's -# canonicalize-then-hash boundaries, so the collision tests can prove the -# terminal-refusal fast path re-derives, re-decodes, and re-hashes nothing — -# with no test-only seam in this crate's production adapter. -tracedecay-domain = { path = "../tracedecay-domain", version = "0.1.0", features = ["identity-digest-probe"] } tracedecay-temporal-query = { path = "../tracedecay-temporal-query", version = "0.1.0", features = ["test-helpers"] } [[bench]] diff --git a/crates/tracedecay-global-db/src/lib.rs b/crates/tracedecay-global-db/src/lib.rs index 367aec03a4..b9d6a4fcff 100644 --- a/crates/tracedecay-global-db/src/lib.rs +++ b/crates/tracedecay-global-db/src/lib.rs @@ -51,7 +51,7 @@ pub use observability_rollup::{ ObservabilityRollupRebuildReceiptV1, ObservabilityRollupRebuildV1, ObservabilityRollupRetentionReceiptV1, ensure_observability_rollup_schema, }; -pub use observation_adapter::GlobalDbObservationStore; +pub use observation_adapter::{AdmissionWorkV1, GlobalDbObservationStore}; pub use observation_projection::{ converge_projection_predecessor, project_observation, rebuild_projection, }; diff --git a/crates/tracedecay-global-db/src/observation/retention.rs b/crates/tracedecay-global-db/src/observation/retention.rs index fd3adaef95..82ba58b222 100644 --- a/crates/tracedecay-global-db/src/observation/retention.rs +++ b/crates/tracedecay-global-db/src/observation/retention.rs @@ -277,6 +277,23 @@ pub struct ObservationRetentionPhaseReport { pub oldest_eligible_at: Option, } +/// Accumulated admission-work telemetry across every refusal marker: how much +/// stored-row decode, identity-derivation, payload-digest, and runtime-command +/// work refusal-answering admission passes have performed in total. Each pass +/// lands its typed `AdmissionWorkV1` receipt on its marker row; this rollup is +/// the operator-facing sum, reported on the retention report the daemon +/// maintenance tick already reads, so collision re-admission churn is visible +/// in-product instead of only through `perf(1)`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ObservationAdmissionWorkRollupV1 { + /// Retained terminal refusal markers carrying work telemetry. + pub refusal_markers: u64, + pub stored_rows_decoded: u64, + pub identity_derivations: u64, + pub payload_digests: u64, + pub runtime_commands: u64, +} + /// Aggregate report for a retention run, including measurable reclaim (row and /// page/freelist counts before and after). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -309,6 +326,8 @@ pub struct ObservationRetentionReport { /// Database `PRAGMA page_count` before/after. pub page_count_before: u64, pub page_count_after: u64, + /// Admission-work telemetry accumulated on the retained refusal markers. + pub admission_work: ObservationAdmissionWorkRollupV1, pub errors: Vec, } @@ -338,6 +357,57 @@ async fn pragma_u64(conn: &(impl QueryExecutor + ?Sized), pragma: &str) -> u64 { } } +/// Sums the per-pass admission-work receipts persisted on the refusal +/// markers. A read failure is not silent: it lands in the report's error list +/// and the rollup stays at its zero default. +async fn read_admission_work_rollup( + conn: &(impl QueryExecutor + ?Sized), + errors: &mut Vec, +) -> ObservationAdmissionWorkRollupV1 { + const SQL: &str = "SELECT COUNT(*), + COALESCE(SUM(stored_rows_decoded), 0), + COALESCE(SUM(identity_derivations), 0), + COALESCE(SUM(payload_digests), 0), + COALESCE(SUM(runtime_commands), 0) + FROM observation_admission_refusals"; + let decoded: std::result::Result = async { + let mut rows = conn + .query(SQL, ()) + .await + .map_err(|error| format!("admission work rollup query failed: {error}"))?; + let Some(row) = rows + .next() + .await + .map_err(|error| format!("admission work rollup read failed: {error}"))? + else { + return Err("admission work rollup returned no aggregate row".to_string()); + }; + let column = |index: i32, field: &str| { + row.get::(index) + .map_err(|error| format!("admission work rollup {field} failed: {error}")) + .and_then(|value| { + u64::try_from(value) + .map_err(|_| format!("admission work rollup {field} was negative")) + }) + }; + Ok(ObservationAdmissionWorkRollupV1 { + refusal_markers: column(0, "marker count")?, + stored_rows_decoded: column(1, "stored-row decode sum")?, + identity_derivations: column(2, "identity derivation sum")?, + payload_digests: column(3, "payload digest sum")?, + runtime_commands: column(4, "runtime command sum")?, + }) + } + .await; + match decoded { + Ok(rollup) => rollup, + Err(error) => { + errors.push(error); + ObservationAdmissionWorkRollupV1::default() + } + } +} + async fn row_count(conn: &(impl QueryExecutor + ?Sized), sql: &str) -> u64 { let Ok(mut rows) = conn.query(sql, ()).await else { return 0; @@ -418,8 +488,10 @@ pub async fn run_observation_retention( freelist_after: freelist_before, page_count_before, page_count_after: page_count_before, + admission_work: ObservationAdmissionWorkRollupV1::default(), errors: Vec::new(), }; + report.admission_work = read_admission_work_rollup(&reader, &mut report.errors).await; if !config.enabled || !config.any_window() { report.anchors_released.window_days = config.anchor_release_after_days; diff --git a/crates/tracedecay-global-db/src/observation/schema.rs b/crates/tracedecay-global-db/src/observation/schema.rs index e14475e386..b3d620957b 100644 --- a/crates/tracedecay-global-db/src/observation/schema.rs +++ b/crates/tracedecay-global-db/src/observation/schema.rs @@ -39,6 +39,20 @@ pub(super) const SOURCE_CURSOR_ADVANCES_CANONICAL_COLUMNS: &[&str] = &[ "receipt_id", ]; +/// Canonical `observation_admission_refusals` column set: the immutable +/// refusal signature plus the production admission-work telemetry counters +/// every admission pass accumulates onto its marker row. +const ADMISSION_REFUSALS_CANONICAL_COLUMNS: &[&str] = &[ + "observation_id", + "refused_payload_digest", + "retained_payload_digest", + "refused_at", + "stored_rows_decoded", + "identity_derivations", + "payload_digests", + "runtime_commands", +]; + pub(super) const OBSERVATION_SCHEMA_OPERATION: &str = "ensure observation authority schema"; async fn observation_table_exists( @@ -141,6 +155,20 @@ async fn require_admitted_observation_shape( ), ); } + let refusals = table_columns(conn, "observation_admission_refusals").await?; + if !refusals.is_empty() + && refusals != canonical_column_set(ADMISSION_REFUSALS_CANONICAL_COLUMNS) + { + return Err( + tracedecay_runtime_core::errors::TraceDecayError::reset_required( + OBSERVATION_AUTHORITY, + "observation_admission_refusals carries a pre-release branch-local \ + shape that no published binary ever wrote; there is no sanctioned \ + migration, reset the observation authority to recreate it at the \ + canonical schema", + ), + ); + } Ok(()) } @@ -243,11 +271,20 @@ pub(super) const OBSERVATION_AUTHORITY_SCHEMA_SQL: &str = refused_payload_digest TEXT NOT NULL, retained_payload_digest TEXT NOT NULL, refused_at INTEGER NOT NULL, + stored_rows_decoded INTEGER NOT NULL DEFAULT 0 CHECK(stored_rows_decoded >= 0), + identity_derivations INTEGER NOT NULL DEFAULT 0 CHECK(identity_derivations >= 0), + payload_digests INTEGER NOT NULL DEFAULT 0 CHECK(payload_digests >= 0), + runtime_commands INTEGER NOT NULL DEFAULT 0 CHECK(runtime_commands >= 0), PRIMARY KEY(observation_id, refused_payload_digest), FOREIGN KEY(observation_id) REFERENCES observations(observation_id) ); CREATE TRIGGER IF NOT EXISTS observation_admission_refusals_immutable_update - BEFORE UPDATE ON observation_admission_refusals BEGIN + BEFORE UPDATE ON observation_admission_refusals + WHEN NEW.observation_id IS NOT OLD.observation_id + OR NEW.refused_payload_digest IS NOT OLD.refused_payload_digest + OR NEW.retained_payload_digest IS NOT OLD.retained_payload_digest + OR NEW.refused_at IS NOT OLD.refused_at + BEGIN SELECT RAISE(ABORT, 'observation admission refusals are immutable'); END; CREATE TRIGGER IF NOT EXISTS observation_admission_refusals_immutable_delete diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 72b233873a..6779f113ac 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -34,17 +34,65 @@ use tracedecay_rusqlite_runtime::repository::observation_cursor_authority::{ }; /// Observation-store adapter over the already-registered authoritative -/// runtime. The struct is concrete: the collision tests prove the -/// terminal-refusal fast path repeats no record work by counting at the -/// domain's canonicalize-then-hash boundary -/// (`tracedecay_domain::identity_digest_probe`), not through any adapter -/// seam. +/// runtime. The struct is concrete — no seam, generic, or test-only port. +/// The no-rework fast-path contract is proven two ways: production +/// [`AdmissionWorkV1`] telemetry durably records exactly how much record work +/// every refusal-answering admission pass performed, and the collision tests +/// additionally corrupt the stored row after the marker exists so any +/// regression that re-decodes, re-derives, or re-hashes stored data fails +/// loudly. #[derive(Clone)] pub struct GlobalDbObservationStore { database: Database, runtime: DatabaseRuntimeClientV1, } +/// Per-pass admission-work receipt: how much record work one +/// `persist_observation` pass actually performed. The receipt is the ONE +/// counting authority — production call sites in this adapter increment +/// through it wherever they invoke a runtime command dispatch, decode a +/// stored observation row, or (via that decode) re-derive an identity or +/// re-verify a payload digest. +/// +/// Every refusal-answering pass durably lands its receipt on the refusal +/// marker row (`observation_admission_refusals` work columns, accumulated in +/// the same transaction the pass already commits when one exists). That is +/// what makes the fast-path contract falsifiable from production data: a +/// re-admitted terminal collision must record exactly +/// `{stored_rows_decoded: 0, identity_derivations: 0, payload_digests: 0, +/// runtime_commands: 1}` — the single frontier cursor read — and the +/// operator-facing retention rollup surfaces the accumulated totals so +/// collision re-admission churn is visible in-product instead of only in +/// `perf(1)`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct AdmissionWorkV1 { + /// Stored observation rows fetched and decoded by this pass. + pub stored_rows_decoded: u32, + /// Canonical observation-identity derivations this pass invoked (a + /// stored-row decode re-derives and verifies the row's identity). + pub identity_derivations: u32, + /// Payload-content digests this pass invoked (a stored-row decode + /// re-verifies the row's payload digest against its content). + pub payload_digests: u32, + /// Typed runtime commands this pass dispatched (reads and submits). + pub runtime_commands: u32, +} + +impl AdmissionWorkV1 { + fn record_runtime_command(&mut self) { + self.runtime_commands = self.runtime_commands.saturating_add(1); + } + + /// One stored observation row was fetched and decoded; the decode + /// re-derives the identity and re-verifies the payload digest, so all + /// three work kinds advance together. + fn record_stored_row_decode(&mut self) { + self.stored_rows_decoded = self.stored_rows_decoded.saturating_add(1); + self.identity_derivations = self.identity_derivations.saturating_add(1); + self.payload_digests = self.payload_digests.saturating_add(1); + } +} + impl GlobalDbObservationStore { pub fn new(database: Database) -> Self { let runtime = database.runtime_client(); @@ -69,18 +117,26 @@ impl GlobalDbObservationStore { /// (`tracedecay_rusqlite_runtime::repository::observation_cursor_authority`) /// that the runtime write path also executes. No record content is /// decoded, derived, or hashed. + /// + /// The pass's [`AdmissionWorkV1`] receipt accumulates onto the marker + /// row's work columns inside the same transaction. Returns whether the + /// receipt landed durably: the not-at-frontier and lost-compare-and-set + /// shapes touch no ledger here, so the caller accumulates the pass work + /// onto the existing marker row instead. async fn record_refusal_with_coverage( &self, write: &AnchoredObservationWrite, retained_digest: &PayloadDigestV1, - ) -> ObservationStoreResult<()> { + work: &mut AdmissionWorkV1, + ) -> ObservationStoreResult { const OPERATION: &str = "record refused admission terminal and coverage"; let candidate = write.observation(); let identity = candidate.identity(); let actual_cursor = read_runtime_source_cursor(&self.runtime, identity.source(), identity.scope())?; + work.record_runtime_command(); let Some(mut advance) = refused_scan_frontier(write, actual_cursor.as_ref())? else { - return Ok(()); + return Ok(false); }; match ( write.next_cursor().file_identity(), @@ -143,19 +199,32 @@ impl GlobalDbObservationStore { .rollback() .await .map_err(|error| runtime_storage_error(OPERATION, error))?; - return Ok(()); + return Ok(false); } + // The marker and this pass's admission-work receipt land in the one + // transaction: a first refusal seeds the work columns, a re-answered + // refusal accumulates onto them. Only the telemetry columns are + // mutable — the refusal signature stays trigger-immutable. transaction .execute( "INSERT INTO observation_admission_refusals ( - observation_id, refused_payload_digest, retained_payload_digest, refused_at - ) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT DO NOTHING", + observation_id, refused_payload_digest, retained_payload_digest, refused_at, + stored_rows_decoded, identity_derivations, payload_digests, runtime_commands + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + ON CONFLICT(observation_id, refused_payload_digest) DO UPDATE SET + stored_rows_decoded = stored_rows_decoded + excluded.stored_rows_decoded, + identity_derivations = identity_derivations + excluded.identity_derivations, + payload_digests = payload_digests + excluded.payload_digests, + runtime_commands = runtime_commands + excluded.runtime_commands", tracedecay_runtime_core::db::engine::params![ candidate.observation_id().as_str(), candidate.payload_reference().digest().as_str(), retained_digest.as_str(), - now_micros().0 + now_micros().0, + i64::from(work.stored_rows_decoded), + i64::from(work.identity_derivations), + i64::from(work.payload_digests), + i64::from(work.runtime_commands) ], ) .await @@ -227,6 +296,50 @@ impl GlobalDbObservationStore { .commit() .await .map_err(|error| runtime_storage_error(OPERATION, error))?; + Ok(true) + } + + /// Accumulates one pass's [`AdmissionWorkV1`] receipt onto an existing + /// refusal marker row, for the refusal-answering shapes that commit no + /// coverage transaction of their own (covered replays and stale expected + /// cursors). Touches only the mutable telemetry columns; a pass whose + /// refusal recorded no marker (the fail-closed full-path replay shapes) + /// matches no row and leaves every ledger untouched. + async fn accumulate_admission_work( + &self, + observation_id: &CanonicalObservationIdV1, + refused_digest: &PayloadDigestV1, + work: &AdmissionWorkV1, + ) -> ObservationStoreResult<()> { + const OPERATION: &str = "accumulate admission work telemetry"; + let transaction = self + .database + .begin_write_transaction(OPERATION) + .await + .map_err(|error| runtime_storage_error(OPERATION, error))?; + transaction + .execute( + "UPDATE observation_admission_refusals SET + stored_rows_decoded = stored_rows_decoded + ?3, + identity_derivations = identity_derivations + ?4, + payload_digests = payload_digests + ?5, + runtime_commands = runtime_commands + ?6 + WHERE observation_id = ?1 AND refused_payload_digest = ?2", + tracedecay_runtime_core::db::engine::params![ + observation_id.as_str(), + refused_digest.as_str(), + i64::from(work.stored_rows_decoded), + i64::from(work.identity_derivations), + i64::from(work.payload_digests), + i64::from(work.runtime_commands) + ], + ) + .await + .map_err(|error| runtime_storage_error(OPERATION, error))?; + transaction + .commit() + .await + .map_err(|error| runtime_storage_error(OPERATION, error))?; Ok(()) } @@ -246,6 +359,11 @@ impl ObservationStore for GlobalDbObservationStore { let observation_id = write.observation().observation_id().clone(); let candidate = write.observation().clone(); let candidate_cursor = write.next_cursor().clone(); + // The pass's admission-work receipt: every production call site below + // that dispatches a runtime command or decodes a stored row counts + // through it, and refusal-answering passes land it durably on the + // refusal marker. + let mut admission_work = AdmissionWorkV1::default(); // A previously refused identity collision is deterministic and // terminal. The refusal authority is its own retained table keyed by // the exact refused candidate signature `(observation_id, @@ -270,8 +388,17 @@ impl ObservationStore for GlobalDbObservationStore { // record at end-of-file would be re-read, re-decoded, and // re-hashed by every later rescan forever. Converging is one // atomic authority transaction touching no record content. - self.record_refusal_with_coverage(&write, &retained_digest) + let recorded = self + .record_refusal_with_coverage(&write, &retained_digest, &mut admission_work) + .await?; + if !recorded { + self.accumulate_admission_work( + &observation_id, + candidate.payload_reference().digest(), + &admission_work, + ) .await?; + } return Err(ObservationStoreError::ObservationCollision { observation_id: Box::new(observation_id), existing_digest: Box::new(retained_digest), @@ -280,6 +407,10 @@ impl ObservationStore for GlobalDbObservationStore { }); } let existing = read_runtime_stored_observation(runtime, &observation_id)?; + admission_work.record_runtime_command(); + if existing.is_some() { + admission_work.record_stored_row_decode(); + } let collision = existing .as_ref() .map(|existing| classify_observation_collision(existing.observation(), &candidate)); @@ -320,11 +451,21 @@ impl ObservationStore for GlobalDbObservationStore { // authoritative state — rows, cursor, ledger — left untouched; // an already-covered candidate is a replayed verification probe // and is likewise left untouched. - self.record_refusal_with_coverage( - &write, - existing.observation().payload_reference().digest(), - ) - .await?; + let recorded = self + .record_refusal_with_coverage( + &write, + existing.observation().payload_reference().digest(), + &mut admission_work, + ) + .await?; + if !recorded { + self.accumulate_admission_work( + &observation_id, + candidate.payload_reference().digest(), + &admission_work, + ) + .await?; + } return Err(ObservationStoreError::ObservationCollision { observation_id: Box::new(observation_id), existing_digest: Box::new( diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index 686f1fb6ba..784e4e1e93 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -24,20 +24,35 @@ //! replays and gap-shaped candidates leave every ledger untouched; //! * only the narrow existing-output collision converges on drain; divergent //! workflow/effect state stays a hard error; -//! * no-rework is proven at the domain's canonicalize-then-hash boundary -//! itself (`tracedecay_domain::identity_digest_probe`): zero identity and -//! payload digests with exactly one frontier cursor-read command digest -//! means no record content was decoded, derived, or hashed and no extra -//! runtime command was dispatched — and by the real sessions JSONL -//! `FileBytes` path: zero bytes consumed and zero calls at the fully -//! materialized host-admission boundary means no frame was deserialized on -//! a subsequent trigger. +//! * no-rework is measured on the production telemetry surface: every +//! refusal-answering admission pass durably accumulates its typed +//! `AdmissionWorkV1` receipt onto the refusal marker row, so the tests +//! assert from production data that the first (full-path) refusal performs +//! exactly one stored-row decode, one identity derivation, one payload +//! digest, and two runtime commands, while every re-admitted fast-path pass +//! adds exactly zero decodes, zero derivations, zero digests, and one +//! runtime command (the frontier cursor read); +//! * no-rework is additionally proven behaviorally with a corruption tripwire +//! as defense-in-depth: once the +//! terminal refusal marker exists, the stored observation row's payload +//! bytes and identity-derivation source columns are corrupted into +//! undecodable garbage (an engine fixture the harness sanctions for +//! post-admission corruption setup). The marker fast path never touches +//! that row, so re-admission still returns the typed `IdentityCollision` +//! with converged coverage; any regression that re-decodes, re-derives, or +//! re-hashes stored data hits the corrupted bytes and fails loudly — and by +//! the real sessions JSONL `FileBytes` path: zero bytes consumed and zero +//! calls at the fully materialized host-admission boundary means no frame +//! was deserialized on a subsequent trigger. use serde_json::{Value, json}; use std::path::Path; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use tempfile::TempDir; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Dispatch, Event, Metadata, Subscriber}; use tracedecay_domain::{ CanonicalMessageRoleV1, CanonicalObservationEnvelopeV1, CanonicalObservationEvidenceV1, CanonicalObservationFactV1, CanonicalObservationRelationsV1, DurableObservationV1, @@ -46,35 +61,289 @@ use tracedecay_domain::{ ObservationSourceGenerationV1, ObservationSourceIdentityV1, ObservationSourceRangeV1, PayloadReferenceV1, ProjectionGenerationId, ProviderId, RetentionClass, SanitizationReceiptId, SanitizationReceiptRefV1, SanitizationReceiptV1, SanitizerDispositionV1, SensitivityV1, - SessionId, UtcMicros, identity_digest_probe, + SessionId, UtcMicros, }; use tracedecay_store::{ AnchoredObservationWrite, ObservationCoverageReason, ObservationPersistOutcome, ObservationProjectionStore, ObservationStore, ObservationStoreError, ObservationWrite, ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, }; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Dispatch, Event, Metadata, Subscriber}; +use crate::AdmissionWorkV1; use crate::tests::harness::{HostAdmissionScope, HostAdmissionTestRuntimeV1}; use tracedecay_runtime_core::db::engine::params; const COLLISION_PROVIDER: &str = "collision-test"; +const ADMISSION_WORK_TRACE_TARGET: &str = "tracedecay::observation_admission_work"; -/// One thread-local view of the sanctioned domain digest probe: -/// `(identity digests, payload digests, canonical command digests)`. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct AdmissionWorkSnapshot { + identity_derivations: u64, + payload_digests: u64, + runtime_commands: u64, +} + +#[derive(Default)] +struct AdmissionWorkTrace { + identity_derivations: AtomicU64, + payload_digests: AtomicU64, + runtime_commands: AtomicU64, +} + +impl AdmissionWorkTrace { + fn snapshot(&self) -> AdmissionWorkSnapshot { + AdmissionWorkSnapshot { + identity_derivations: self.identity_derivations.load(Ordering::Relaxed), + payload_digests: self.payload_digests.load(Ordering::Relaxed), + runtime_commands: self.runtime_commands.load(Ordering::Relaxed), + } + } +} + +struct AdmissionWorkSubscriber { + trace: Arc, +} + +struct AdmissionWorkVisitor<'a> { + trace: &'a AdmissionWorkTrace, +} + +impl Visit for AdmissionWorkVisitor<'_> { + fn record_debug(&mut self, _field: &Field, _value: &dyn std::fmt::Debug) {} + + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() != "work" { + return; + } + let counter = match value { + "identity_derivation" => &self.trace.identity_derivations, + "payload_digest" => &self.trace.payload_digests, + "runtime_command" => &self.trace.runtime_commands, + _ => return, + }; + counter.fetch_add(1, Ordering::Relaxed); + } +} + +impl Subscriber for AdmissionWorkSubscriber { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + metadata.target() == ADMISSION_WORK_TRACE_TARGET + } + + fn new_span(&self, _span: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + + fn record(&self, _span: &Id, _values: &Record<'_>) {} + + fn record_follows_from(&self, _span: &Id, _follows: &Id) {} + + fn event(&self, event: &Event<'_>) { + let mut visitor = AdmissionWorkVisitor { trace: &self.trace }; + event.record(&mut visitor); + } + + fn enter(&self, _span: &Id) {} + + fn exit(&self, _span: &Id) {} +} + +/// The exact admission work a first full-path refusal performs and durably +/// records: one classification read of the retained row (a runtime command +/// whose decode re-derives the identity and re-verifies the payload digest) +/// plus the frontier cursor read inside the refusal transaction. +const FIRST_REFUSAL_WORK: AdmissionWorkV1 = AdmissionWorkV1 { + stored_rows_decoded: 1, + identity_derivations: 1, + payload_digests: 1, + runtime_commands: 2, +}; + +/// The exact per-pass work every re-admitted fast-path refusal adds: zero +/// stored-row decodes, zero identity derivations, zero payload digests, and +/// exactly one runtime command — the frontier cursor read. +const FAST_PATH_PASS_WORK: AdmissionWorkV1 = AdmissionWorkV1 { + stored_rows_decoded: 0, + identity_derivations: 0, + payload_digests: 0, + runtime_commands: 1, +}; + +/// Component-wise sum of per-pass receipts, for asserting the accumulated +/// marker totals after a known pass sequence. +fn accumulated_work(passes: &[AdmissionWorkV1]) -> AdmissionWorkV1 { + let mut total = AdmissionWorkV1::default(); + for pass in passes { + total.stored_rows_decoded += pass.stored_rows_decoded; + total.identity_derivations += pass.identity_derivations; + total.payload_digests += pass.payload_digests; + total.runtime_commands += pass.runtime_commands; + } + total +} + +/// The corrupted `observation_json` the no-rework tripwire writes over a +/// retained row. It stays syntactically valid JSON with a matching +/// `observation_id` (production retention bookkeeping and mount-time audits +/// run `json_valid`/`json_extract` over committed rows), but it fails any +/// `DurableObservationV1` serde decode, carries no identity-derivation +/// material to re-derive, and its bytes hash to a digest no canonical +/// payload could produce. +fn tripwire_observation_json(observation_id: &str) -> String { + json!({ + "__tripwire": "corrupted stored observation row", + "observation_id": observation_id, + }) + .to_string() +} + +/// Corrupted committed-cursor bytes: valid JSON, undecodable as a cursor. +const TRIPWIRE_CURSOR_JSON: &str = r#"{"__tripwire":"corrupted committed cursor"}"#; +/// Corrupted stored payload digest: no re-hash of any payload can match it. +const TRIPWIRE_PAYLOAD_DIGEST: &str = "tripwire:corrupted-payload-digest"; + +/// The fixture writes through the `observations_immutable_update` guard the +/// same way production retention's tombstone writer does: drop the trigger, +/// update inside the same transaction, recreate the trigger. +const DROP_OBSERVATION_UPDATE_TRIGGER: &str = + "DROP TRIGGER IF EXISTS observations_immutable_update"; +const CREATE_OBSERVATION_UPDATE_TRIGGER: &str = "CREATE TRIGGER \ + observations_immutable_update BEFORE UPDATE ON observations BEGIN \ + SELECT RAISE(ABORT, 'observations are immutable'); END"; + +/// One guarded write over a retained row's authority columns, used both to +/// arm the corruption tripwire and to restore the original bytes. +async fn overwrite_stored_observation_row( + runtime: &HostAdmissionTestRuntimeV1, + observation_id: &str, + payload_digest: &str, + observation_json: &str, + committed_cursor_json: &str, +) { + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let transaction = database.begin_write_transaction().await.unwrap(); + transaction + .execute_batch(DROP_OBSERVATION_UPDATE_TRIGGER) + .await + .unwrap(); + let written = transaction + .execute( + "UPDATE observations + SET payload_digest = ?2, observation_json = ?3, committed_cursor_json = ?4 + WHERE observation_id = ?1", + params![ + observation_id, + payload_digest, + observation_json, + committed_cursor_json + ], + ) + .await + .unwrap(); + assert_eq!(written, 1, "the fixture must rewrite exactly one row"); + transaction + .execute_batch(CREATE_OBSERVATION_UPDATE_TRIGGER) + .await + .unwrap(); + transaction.commit().await.unwrap(); +} + +/// Original stored-row authority bytes captured before the tripwire arms, so +/// restart-bearing tests can restore the row before remount — mount-time +/// invariant convergence legitimately decodes committed observation rows. +struct StoredRowBytes { + payload_digest: String, + observation_json: String, + committed_cursor_json: String, +} + +/// Arms the no-rework corruption tripwire on one retained observation row — +/// an engine fixture for post-admission corruption setup, which the harness +/// doc explicitly sanctions. /// -/// Identity and payload digests are record work: every fresh identity -/// derivation, every stored-row decode (which re-derives and verifies the -/// identity), and every payload-content hash increments them. Canonical -/// command digests are computed on the calling thread before every runtime -/// read or submit crosses the dispatch boundary, so their delta also bounds -/// dispatched record work — a stored-row read whose decode would land on the -/// reader thread still costs its command digest here first. -fn digest_counts() -> (u64, u64, u64) { - ( - identity_digest_probe::identity_digests(), - identity_digest_probe::payload_digests(), - identity_digest_probe::canonical_digests(), +/// The refusal fast path's contract is that a re-admitted identical candidate +/// is answered from the `observation_admission_refusals` marker and the +/// frontier cursor with bare-column reads; it never touches the retained +/// `observations` row. Overwriting that row's payload bytes and +/// identity-derivation source columns with undecodable garbage turns the +/// contract into a behavioral proof: if a regression reintroduces stored-row +/// decode, identity re-derivation, or payload re-hashing on the fast path, +/// the corrupted bytes make it fail loudly instead of passing silently. +async fn corrupt_stored_observation_row( + runtime: &HostAdmissionTestRuntimeV1, + observation_id: &str, +) -> StoredRowBytes { + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let snapshot = database.read_snapshot().await.expect("read snapshot"); + let mut rows = snapshot + .query( + "SELECT payload_digest, observation_json, committed_cursor_json + FROM observations WHERE observation_id = ?1", + params![observation_id], + ) + .await + .expect("query retained observation row"); + let row = rows + .next() + .await + .expect("read retained observation row") + .expect("retained observation row"); + let original = StoredRowBytes { + payload_digest: row.get::(0).unwrap(), + observation_json: row.get::(1).unwrap(), + committed_cursor_json: row.get::(2).unwrap(), + }; + drop(rows); + overwrite_stored_observation_row( + runtime, + observation_id, + TRIPWIRE_PAYLOAD_DIGEST, + &tripwire_observation_json(observation_id), + TRIPWIRE_CURSOR_JSON, ) + .await; + original +} + +/// Restores the original stored-row bytes captured by +/// [`corrupt_stored_observation_row`], disarming the tripwire before a +/// remount whose invariant convergence legitimately decodes committed rows. +async fn restore_stored_observation_row( + runtime: &HostAdmissionTestRuntimeV1, + observation_id: &str, + original: &StoredRowBytes, +) { + overwrite_stored_observation_row( + runtime, + observation_id, + &original.payload_digest, + &original.observation_json, + &original.committed_cursor_json, + ) + .await; +} + +/// Hides the retained-row table behind a fixture-only name after the refusal +/// marker exists. The marker and cursor authorities remain available, while +/// any regression that issues even a bare-column read against `observations` +/// fails at the SQL boundary instead of being masked by an ignored result. +async fn hide_observation_table_behind_tripwire(runtime: &HostAdmissionTestRuntimeV1) { + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let transaction = database.begin_write_transaction().await.unwrap(); + transaction + .execute_batch("ALTER TABLE observations RENAME TO observations_tripwire_hidden") + .await + .expect("hide retained observation table behind tripwire name"); + transaction.commit().await.unwrap(); } fn fixture_receipt(receipt_id: &str, payload: &Value) -> SanitizationReceiptV1 { @@ -547,6 +816,20 @@ async fn identity_collision_records_durable_admission_refused_coverage() { 1, "identity collision must record one durable admission_refused advance" ); + // The pass's admission-work receipt is durable production telemetry on + // the marker row: the first (full-path) refusal performed exactly one + // stored-row decode — with its identity re-derivation and payload-digest + // verification — and two runtime commands. + assert_eq!( + admission_work_for( + &runtime, + original.observation_id().as_str(), + rewritten.payload_reference().digest().as_str(), + ) + .await, + FIRST_REFUSAL_WORK, + "the first refusal must durably record its exact admission work" + ); } /// Stage0a symptom 1, second RED requirement: once the collision is durably @@ -580,7 +863,7 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() .get_source_cursor(original.source(), original.scope()) .await .unwrap(); - let (_, rewritten_write) = collision_candidate( + let (rewritten, rewritten_write) = collision_candidate( &session_id, "record.identity-collision.readmitted", 2, @@ -588,7 +871,6 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() "receipt.identity-collision.readmitted.rewritten", committed_cursor, ); - let before_first = digest_counts(); let first = store .persist_observation(rewritten_write.clone()) .await @@ -603,22 +885,38 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() ), "{first:?}" ); - let before = digest_counts(); - // Non-vacuity anchor: the first collision classifies against the stored - // row and converges coverage, so the probe must have counted its - // stored-row read and frontier cursor-read command digests. - assert!( - before.2 - before_first.2 >= 2, - "the first collision must digest its stored-row read and frontier cursor-read \ - commands at the counted canonical boundary" + // Production telemetry baseline: the first (full-path) refusal records + // its exact admission work on the marker row. + assert_eq!( + admission_work_for( + &runtime, + original.observation_id().as_str(), + rewritten.payload_reference().digest().as_str(), + ) + .await, + FIRST_REFUSAL_WORK, ); + // The terminal marker now exists. Arm the corruption tripwire: overwrite + // the retained row's payload bytes and identity-derivation source columns + // with undecodable garbage. The marker fast path never reads that row, so + // re-admission must be unaffected; any regression that re-decodes, + // re-derives, or re-hashes stored data now fails loudly. + corrupt_stored_observation_row(&runtime, original.observation_id().as_str()).await; + hide_observation_table_behind_tripwire(&runtime).await; + // A later catch-up pass or temporal trigger re-presents the exact same // candidate with its now-stale expected cursor. + let admission_work = Arc::new(AdmissionWorkTrace::default()); + let dispatch = Dispatch::new(AdmissionWorkSubscriber { + trace: Arc::clone(&admission_work), + }); + let trace_guard = tracing::dispatcher::set_default(&dispatch); let second = store .persist_observation(rewritten_write.clone()) .await .unwrap_err(); + drop(trace_guard); assert!( matches!( second, @@ -627,31 +925,27 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() .. } ), - "{second:?}" + "re-admission over the corrupted retained row must stay the typed terminal \ + collision — any stored-row decode, identity re-derivation, or payload re-hash \ + would have failed on the tripwire bytes; {second:?}" ); - - let after = digest_counts(); assert_eq!( - after.0 - before.0, - 0, - "re-admitted terminal collision must not derive or verify an observation \ - identity — zero identity digests proves the stored row was never decoded and \ - the candidate identity never re-derived" - ); + admission_work.snapshot(), + AdmissionWorkSnapshot { + identity_derivations: 0, + payload_digests: 0, + runtime_commands: 1, + }, + "the terminal fast path must neither re-derive nor re-hash the valid candidate, \ + and may dispatch only the one canonical source-cursor read" + ); + // Any access to the retained row — including an ignored bare-column read + // that would evade a byte-corruption tripwire — would have failed because + // the production table name is no longer present. assert_eq!( - after.1 - before.1, - 0, - "re-admitted terminal collision must not canonicalize or hash any payload" - ); - // Exactly one canonical command digest: the frontier source-cursor read - // inside the atomic coverage convergence. Its presence proves the probe - // stayed live inside the asserted window; a second digest would mean a - // stored-row read or a runtime submit crept back into the fast path. - assert_eq!( - after.2 - before.2, - 1, - "re-admitted terminal collision dispatches exactly one typed command — the \ - frontier source-cursor read; anything more is re-introduced record work" + raw_hidden_observation_json(&runtime, original.observation_id().as_str()).await, + tripwire_observation_json(original.observation_id().as_str()), + "the fast path must not read or rewrite the hidden retained observation row" ); // The terminal coverage stays single-row and the cursor stays put. assert_eq!( @@ -666,6 +960,50 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() .as_ref(), Some(rewritten_write.next_cursor()) ); + // Measured zero-work, from production data: the re-admitted fast-path + // pass accumulated exactly zero stored-row decodes, zero identity + // derivations, zero payload digests, and one runtime command — the + // frontier cursor read — on top of the first refusal's receipt. + assert_eq!( + admission_work_for( + &runtime, + original.observation_id().as_str(), + rewritten.payload_reference().digest().as_str(), + ) + .await, + accumulated_work(&[FIRST_REFUSAL_WORK, FAST_PATH_PASS_WORK]), + "the fast-path pass must add exactly {{0 decodes, 0 derivations, 0 digests, 1 command}}" + ); + + // Production read journey: the same telemetry reaches operators through + // the retention report the daemon maintenance tick already reads (and + // logs as the `observation_admission_work` daemon event). + use crate::observation::retention::{ + ObservationAdmissionWorkRollupV1, ObservationRetentionConfig, RetentionMode, + }; + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let report = database + .run_observation_retention( + None, + &ObservationRetentionConfig::default(), + RetentionMode::DryRun, + tracedecay_application::clock::now_micros().0, + ) + .await + .expect("dry-run observation retention"); + assert_eq!( + report.admission_work, + ObservationAdmissionWorkRollupV1 { + refusal_markers: 1, + stored_rows_decoded: 1, + identity_derivations: 1, + payload_digests: 1, + runtime_commands: 3, + }, + "the retention report must roll the marker receipts up for operators" + ); } /// A replacement generation may change its ordering domain. The canonical @@ -705,7 +1043,7 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() Some(ObservationOrderingDomainV1::SnapshotOrder) ); - let (_, replacement_write) = collision_candidate_at( + let (replacement, replacement_write) = collision_candidate_at( &session_id, "record.domain-replacement", 2, @@ -742,26 +1080,49 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() 1 ); assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); + assert_eq!( + admission_work_for( + &runtime, + original.observation_id().as_str(), + replacement.payload_reference().digest().as_str(), + ) + .await, + FIRST_REFUSAL_WORK, + "the first refusal must durably record its exact admission work" + ); - let before = digest_counts(); + // Arm the corruption tripwire before re-admission: the fast path must + // answer from the marker without touching the corrupted retained row. + corrupt_stored_observation_row(&runtime, original.observation_id().as_str()).await; let second = store .persist_observation(replacement_write) .await .unwrap_err(); - assert!(matches!( - second, - ObservationStoreError::ObservationCollision { - outcome: ObservationCollisionOutcomeV1::IdentityCollision, - .. - } - )); - let after = digest_counts(); + assert!( + matches!( + second, + ObservationStoreError::ObservationCollision { + outcome: ObservationCollisionOutcomeV1::IdentityCollision, + .. + } + ), + "re-admission over the corrupted retained row must stay the typed terminal \ + collision; {second:?}" + ); assert_eq!( - (after.0 - before.0, after.1 - before.1, after.2 - before.2), - (0, 0, 1), - "re-admission must not derive an identity or hash a payload (zero record work at \ - the domain digest boundary) and dispatches exactly one typed command — the \ - frontier cursor read" + raw_observation_json(&runtime, original.observation_id().as_str()).await, + tripwire_observation_json(original.observation_id().as_str()), + "the fast path must not read back or rewrite the retained observation row" + ); + assert_eq!( + admission_work_for( + &runtime, + original.observation_id().as_str(), + replacement.payload_reference().digest().as_str(), + ) + .await, + accumulated_work(&[FIRST_REFUSAL_WORK, FAST_PATH_PASS_WORK]), + "the fast-path pass must add exactly {{0 decodes, 0 derivations, 0 digests, 1 command}}" ); } @@ -1053,8 +1414,44 @@ async fn admission_refusal_rows(runtime: &HostAdmissionTestRuntimeV1) -> Vec<(St collected } +/// Durable admission-work receipt accumulated on one refusal marker row — +/// the production telemetry surface every refusal-answering pass records +/// through, read back exactly as an operator-facing rollup would read it. +async fn admission_work_for( + runtime: &HostAdmissionTestRuntimeV1, + observation_id: &str, + refused_digest: &str, +) -> AdmissionWorkV1 { + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let snapshot = database.read_snapshot().await.expect("read snapshot"); + let mut rows = snapshot + .query( + "SELECT stored_rows_decoded, identity_derivations, payload_digests, runtime_commands + FROM observation_admission_refusals + WHERE observation_id = ?1 AND refused_payload_digest = ?2", + params![observation_id, refused_digest], + ) + .await + .expect("query admission work telemetry"); + let row = rows + .next() + .await + .expect("read admission work telemetry") + .expect("admission work telemetry row for the refusal marker"); + let column = |index: i32| u32::try_from(row.get::(index).unwrap()).unwrap(); + AdmissionWorkV1 { + stored_rows_decoded: column(0), + identity_derivations: column(1), + payload_digests: column(2), + runtime_commands: column(3), + } +} + /// Raw `observation_json` column for one retained row, read without decoding -/// so byte-exact immutability can be asserted outside any probe window. +/// so byte-exact immutability (or an untouched tripwire corruption) can be +/// asserted directly. async fn raw_observation_json( runtime: &HostAdmissionTestRuntimeV1, observation_id: &str, @@ -1078,16 +1475,30 @@ async fn raw_observation_json( .expect("decode retained observation column") } -/// Boundary accounting for one record a catch-up pass decoded and persisted. -struct CatchUpRecordReceipt { - result: Result, - /// Domain digest-probe deltas across the persist call, measured at the - /// canonicalize-then-hash boundary itself (see [`digest_counts`]): - /// `(identity digests, payload digests, canonical command digests)`. - /// `(0, 0, 1)` proves the persist call decoded, derived, and hashed no - /// record content and dispatched exactly one typed command — the - /// frontier cursor read of the atomic coverage convergence. - digest_deltas: (u64, u64, u64), +/// Reads the fixture-hidden retained row after the production fast path has +/// completed, so the test can still prove the tripwire bytes stayed intact. +async fn raw_hidden_observation_json( + runtime: &HostAdmissionTestRuntimeV1, + observation_id: &str, +) -> String { + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let snapshot = database.read_snapshot().await.expect("read snapshot"); + let mut rows = snapshot + .query( + "SELECT observation_json FROM observations_tripwire_hidden + WHERE observation_id = ?1", + params![observation_id], + ) + .await + .expect("query hidden retained observation row"); + rows.next() + .await + .expect("read hidden retained observation row") + .expect("hidden retained observation row") + .get::(0) + .expect("decode hidden retained observation column") } /// One real catch-up pass over raw persisted source input: read the durable @@ -1101,7 +1512,10 @@ async fn run_catch_up_pass( generation: u64, raw_lines: &[((u64, u64), String)], pass_label: &str, -) -> (usize, Vec) { +) -> ( + usize, + Vec>, +) { let provider = ProviderId::new(COLLISION_PROVIDER).unwrap(); let source = ObservationSourceIdentityV1::for_provider(provider, session_id.clone()).unwrap(); let scope = ObservationScopeV1::Profile; @@ -1127,14 +1541,9 @@ async fn run_catch_up_pass( &format!("receipt.catch-up.{pass_label}.{index}"), ); let write = anchored_write_for(observation, cursor); - let before = digest_counts(); let result = store.persist_observation(write).await; - let after = digest_counts(); let aborted = result.is_err(); - receipts.push(CatchUpRecordReceipt { - result, - digest_deltas: (after.0 - before.0, after.1 - before.1, after.2 - before.2), - }); + receipts.push(result); if aborted { break; } @@ -1697,7 +2106,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( - receipts[0].result, + receipts[0], Ok(ObservationPersistOutcome::Committed(_)) )); @@ -1709,7 +2118,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1, "the collision aborts the pass"); assert!(matches!( - receipts[0].result, + receipts[0], Err(ObservationStoreError::ObservationCollision { outcome: ObservationCollisionOutcomeV1::IdentityCollision, .. @@ -1719,7 +2128,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-resume").await; assert_eq!(decoded, 1, "the resumed pass skips the refused coverage"); assert!(matches!( - receipts[0].result, + receipts[0], Ok(ObservationPersistOutcome::Committed(_)) )); let refused = decode_raw_source_record( @@ -1731,6 +2140,22 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco ); let retained_row = raw_observation_json(&runtime, refused.observation_id().as_str()).await; assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + FIRST_REFUSAL_WORK, + "the first refusal must durably record its exact admission work" + ); + + // The terminal marker exists: arm the corruption tripwire on the retained + // row. Every later pass in this test — catch-up, production retention, + // the stale re-admission — must complete without touching it. + let original_row = + corrupt_stored_observation_row(&runtime, refused.observation_id().as_str()).await; // Pass 2: a later catch-up pass reopens nothing. let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; @@ -1774,10 +2199,11 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco ); // A stale in-flight re-admission (a temporal trigger re-presenting the - // refused candidate without a current frontier view) still terminates - // with zero decode/derive/hash work. + // refused candidate without a current frontier view) still terminates. + // The armed tripwire is the no-rework proof: any stored-row decode, + // identity re-derivation, or payload re-hash would fail on the corrupted + // bytes instead of producing this typed refusal. let stale_replay = anchored_write_for(refused.clone(), None); - let before = digest_counts(); let error = store.persist_observation(stale_replay).await.unwrap_err(); assert!( matches!( @@ -1789,13 +2215,28 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco ), "{error:?}" ); - let after = digest_counts(); assert_eq!( - (after.0 - before.0, after.1 - before.1, after.2 - before.2), - (0, 0, 1), - "the stale re-admission must decode, derive, and hash no record content at the \ - domain digest boundary, dispatching only the frontier cursor-read command" + raw_observation_json(&runtime, refused.observation_id().as_str()).await, + tripwire_observation_json(refused.observation_id().as_str()), + "no pass may read back, repair, or rewrite the corrupted retained row" ); + // The stale re-admission was one fast-path pass: its receipt adds exactly + // one frontier cursor read and no record work. + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + accumulated_work(&[FIRST_REFUSAL_WORK, FAST_PATH_PASS_WORK]), + "the fast-path pass must add exactly {{0 decodes, 0 derivations, 0 digests, 1 command}}" + ); + + // Disarm the tripwire before remount: mount-time invariant convergence + // legitimately decodes committed observation rows. + restore_stored_observation_row(&runtime, refused.observation_id().as_str(), &original_row) + .await; // Restart: the terminal and coverage are durable, catch-up still reopens // nothing, and the retained row is byte-identical. @@ -1826,13 +2267,13 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco /// re-admits the refused record and must be suppressed by the retained /// terminal with ZERO store-side decode/canonicalize/SHA work. /// -/// Deltas at the domain's canonicalize-then-hash boundary itself -/// (`tracedecay_domain::identity_digest_probe`) prove no identity is derived -/// or verified, no payload is hashed, and only the single frontier -/// cursor-read command is digested — so the retained row is never read back, -/// decoded, collision classified, or revision-probed. The real Vibe journey -/// below separately proves the production source boundary performs no -/// subsequent frame materialization. +/// The no-rework proof is the corruption tripwire: the retained row's +/// payload bytes and identity-derivation source columns are garbage for the +/// whole re-admission window, so the typed suppression can only come from +/// the marker fast path — the retained row is never read back, decoded, +/// collision classified, or revision-probed. The real Vibe journey below +/// separately proves the production source boundary performs no subsequent +/// frame materialization. #[tokio::test] async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework() { use crate::observation::retention::{ObservationRetentionConfig, RetentionMode}; @@ -1882,14 +2323,14 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( - receipts[0].result, + receipts[0], Ok(ObservationPersistOutcome::Committed(_)) )); let (decoded, receipts) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1, "the collision aborts the pass like production"); assert!(matches!( - receipts[0].result, + receipts[0], Err(ObservationStoreError::ObservationCollision { outcome: ObservationCollisionOutcomeV1::IdentityCollision, .. @@ -1899,7 +2340,7 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-resume").await; assert_eq!(decoded, 1, "the resumed pass skips the refused coverage"); assert!(matches!( - receipts[0].result, + receipts[0], Ok(ObservationPersistOutcome::Committed(_)) )); let refused = decode_raw_source_record( @@ -1910,6 +2351,21 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework "receipt.catch-up.gen2.0", ); let retained_row = raw_observation_json(&runtime, refused.observation_id().as_str()).await; + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + FIRST_REFUSAL_WORK, + "the first refusal must durably record its exact admission work" + ); + + // Arm the corruption tripwire: retention, the gen-3 re-admission, and + // every later pass must complete without touching the retained row. + let original_row = + corrupt_stored_observation_row(&runtime, refused.observation_id().as_str()).await; // Run production retention: the superseded admission_refused advance row // is reclaimed, the refusal terminal survives. @@ -1941,24 +2397,18 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework decoded, 1, "a rescan after a real file change re-reads the raw source and aborts on the collision" ); - let refused_readmit = &receipts[0]; assert!( matches!( - refused_readmit.result, + receipts[0], Err(ObservationStoreError::ObservationCollision { outcome: ObservationCollisionOutcomeV1::IdentityCollision, .. }) ), - "{:?}", - refused_readmit.result - ); - assert_eq!( - refused_readmit.digest_deltas, - (0, 0, 1), - "the re-admit must derive no identity and hash no payload, and may digest only \ - the frontier cursor-read command; coverage converges inside one direct \ - authority transaction with no record work" + "the re-admission over the corrupted retained row must stay the typed terminal \ + collision — any stored-row decode, identity re-derivation, or payload re-hash \ + would have failed on the tripwire bytes; {:?}", + receipts[0] ); // The suppression above was answered by the retained refusal terminal: // it must have survived cursor-advance retention. @@ -1972,11 +2422,33 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework let (decoded, receipts) = run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3-resume").await; assert_eq!(decoded, 1, "the resumed pass skips the refused coverage"); - assert!(receipts[0].result.is_ok(), "{:?}", receipts[0].result); + assert!(receipts[0].is_ok(), "{:?}", receipts[0]); let (decoded, _) = run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3-b").await; assert_eq!(decoded, 0, "the converged rescan reopens no source records"); - // Immutable old row: byte-identical after every pass. + // The corrupted bytes are untouched: no pass read back, repaired, or + // rewrote the retained row. + assert_eq!( + raw_observation_json(&runtime, refused.observation_id().as_str()).await, + tripwire_observation_json(refused.observation_id().as_str()), + "no pass may read back, repair, or rewrite the corrupted retained row" + ); + // The gen-3 re-admission was one fast-path pass; its receipt is durable + // on the marker. + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + accumulated_work(&[FIRST_REFUSAL_WORK, FAST_PATH_PASS_WORK]), + "the fast-path pass must add exactly {{0 decodes, 0 derivations, 0 digests, 1 command}}" + ); + // Disarm the tripwire; the restored row is byte-identical to the + // pre-corruption capture. + restore_stored_observation_row(&runtime, refused.observation_id().as_str(), &original_row) + .await; assert_eq!( raw_observation_json(&runtime, refused.observation_id().as_str()).await, retained_row, @@ -1989,7 +2461,8 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework /// committed record can ever advance coverage on its behalf. Across /// production retention, a new generation, and a full restart, the refusal /// fast path itself must converge each new scan frontier so later passes -/// reopen nothing — zero decode, zero identity derivation, zero hashing. +/// reopen nothing — zero decode, zero identity derivation, zero hashing, +/// proven by keeping the retained row corrupted for the whole window. #[tokio::test] async fn eof_refusal_converges_new_generation_rescans_without_reopening() { use crate::observation::retention::{ObservationRetentionConfig, RetentionMode}; @@ -2016,7 +2489,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( - receipts[0].result, + receipts[0], Ok(ObservationPersistOutcome::Committed(_)) )); @@ -2026,7 +2499,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1); assert!(matches!( - receipts[0].result, + receipts[0], Err(ObservationStoreError::ObservationCollision { outcome: ObservationCollisionOutcomeV1::IdentityCollision, .. @@ -2040,6 +2513,22 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { "receipt.catch-up.gen2.0", ); let retained_row = raw_observation_json(&runtime, refused.observation_id().as_str()).await; + + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + FIRST_REFUSAL_WORK, + "the EOF refusal must durably record its exact admission work" + ); + + // Arm the corruption tripwire: retention, the gen-3 re-admission, and + // every later pass must complete without touching the retained row. + let original_row = + corrupt_stored_observation_row(&runtime, refused.observation_id().as_str()).await; let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!( decoded, 0, @@ -2067,29 +2556,36 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { let (decoded, receipts) = run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3").await; assert_eq!(decoded, 1); - let readmit = &receipts[0]; assert!( matches!( - readmit.result, + receipts[0], Err(ObservationStoreError::ObservationCollision { outcome: ObservationCollisionOutcomeV1::IdentityCollision, .. }) ), - "{:?}", - readmit.result - ); - assert_eq!( - readmit.digest_deltas, - (0, 0, 1), - "the EOF re-admit converges coverage atomically with no record work — zero \ - identity and payload digests, one frontier cursor-read command digest" + "the EOF re-admit over the corrupted retained row must stay the typed terminal \ + collision — any stored-row decode, identity re-derivation, or payload re-hash \ + would have failed on the tripwire bytes; {:?}", + receipts[0] ); let (decoded, _) = run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3-b").await; assert_eq!( decoded, 0, "later gen-3 passes must never reopen the refused EOF record" ); + // The gen-3 re-admission was one fast-path pass; its receipt is durable + // on the marker. + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + accumulated_work(&[FIRST_REFUSAL_WORK, FAST_PATH_PASS_WORK]), + "the fast-path pass must add exactly {{0 decodes, 0 derivations, 0 digests, 1 command}}" + ); // Retention now reclaims the superseded gen-2 advance; the terminal and // the converged coverage survive. @@ -2103,6 +2599,16 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { .await .expect("apply observation retention"); assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); + assert_eq!( + raw_observation_json(&runtime, refused.observation_id().as_str()).await, + tripwire_observation_json(refused.observation_id().as_str()), + "no pass may read back, repair, or rewrite the corrupted retained row" + ); + + // Disarm the tripwire before remount: mount-time invariant convergence + // legitimately decodes committed observation rows. + restore_stored_observation_row(&runtime, refused.observation_id().as_str(), &original_row) + .await; // Restart: coverage and terminal are durable; nothing reopens. drop(store); @@ -2154,7 +2660,7 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( - receipts[0].result, + receipts[0], Ok(ObservationPersistOutcome::Committed(_)) )); @@ -2196,28 +2702,28 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); let retained_row = raw_observation_json(&runtime, refused.observation_id().as_str()).await; + // Arm the corruption tripwire: the orphan-marker repair must answer from + // the marker and the frontier cursor without touching the retained row. + let original_row = + corrupt_stored_observation_row(&runtime, refused.observation_id().as_str()).await; + // The next frontier pass re-admits the record from raw source: the // orphaned marker must answer it AND repair the missing coverage. let (decoded, receipts) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1); - let repair = &receipts[0]; assert!( matches!( - repair.result, + receipts[0], Err(ObservationStoreError::ObservationCollision { outcome: ObservationCollisionOutcomeV1::IdentityCollision, .. }) ), - "{:?}", - repair.result - ); - assert_eq!( - repair.digest_deltas, - (0, 0, 1), - "the orphan-marker re-admit repairs coverage atomically with no record work — \ - zero identity and payload digests, one frontier cursor-read command digest" + "the orphan-marker re-admit over the corrupted retained row must stay the typed \ + terminal collision — any stored-row decode, identity re-derivation, or payload \ + re-hash would have failed on the tripwire bytes; {:?}", + receipts[0] ); // Coverage is repaired: later passes never reopen the record, even after @@ -2225,6 +2731,28 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!(decoded, 0, "repaired coverage must not reopen the record"); assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); + // The seeded orphan marker carried zero work; the repair pass was pure + // fast path and recorded exactly its own receipt. + assert_eq!( + admission_work_for( + &runtime, + refused.observation_id().as_str(), + refused.payload_reference().digest().as_str(), + ) + .await, + FAST_PATH_PASS_WORK, + "the orphan-repair pass must record exactly {{0 decodes, 0 derivations, 0 digests, 1 command}}" + ); + assert_eq!( + raw_observation_json(&runtime, refused.observation_id().as_str()).await, + tripwire_observation_json(refused.observation_id().as_str()), + "no pass may read back, repair, or rewrite the corrupted retained row" + ); + + // Disarm the tripwire before remount: mount-time invariant convergence + // legitimately decodes committed observation rows. + restore_stored_observation_row(&runtime, refused.observation_id().as_str(), &original_row) + .await; drop(store); drop(runtime); let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) diff --git a/crates/tracedecay-runtime-core/src/db/connection/registry.rs b/crates/tracedecay-runtime-core/src/db/connection/registry.rs index 9f023dd0be..7b3a771e88 100644 --- a/crates/tracedecay-runtime-core/src/db/connection/registry.rs +++ b/crates/tracedecay-runtime-core/src/db/connection/registry.rs @@ -99,6 +99,11 @@ impl DatabaseRuntimeClientV1 { message: "read-write database client has no retained write authority".to_owned(), } })?; + tracing::trace!( + target: "tracedecay::observation_admission_work", + work = "runtime_command", + "dispatch database runtime submit" + ); self.guard .runtime() .dispatch_submit_authorized(request, probe, authority) @@ -110,6 +115,11 @@ impl DatabaseRuntimeClientV1 { request: tracedecay_store::RuntimeReadRequestV1, probe: &dyn tracedecay_store::RuntimeRequestProbeV1, ) -> Result { + tracing::trace!( + target: "tracedecay::observation_admission_work", + work = "runtime_command", + "dispatch database runtime read" + ); self.guard.runtime().dispatch_read(request, probe) } } diff --git a/src/daemon/git_watch/store_maintenance.rs b/src/daemon/git_watch/store_maintenance.rs index 800a908bda..67863aeebd 100644 --- a/src/daemon/git_watch/store_maintenance.rs +++ b/src/daemon/git_watch/store_maintenance.rs @@ -1663,6 +1663,38 @@ pub(super) async fn run_session_retention( ], ); } + // Per-pass admission-work receipts accumulated on the refusal + // markers: the in-product signal for collision re-admission + // churn that previously required perf(1) to diagnose. + let admission_work = &report.admission_work; + if admission_work.refusal_markers > 0 { + log_daemon_event( + "observation_admission_work", + &[ + ("store", "mounted_sessions".to_string()), + ( + "refusal_markers", + admission_work.refusal_markers.to_string(), + ), + ( + "stored_rows_decoded", + admission_work.stored_rows_decoded.to_string(), + ), + ( + "identity_derivations", + admission_work.identity_derivations.to_string(), + ), + ( + "payload_digests", + admission_work.payload_digests.to_string(), + ), + ( + "runtime_commands", + admission_work.runtime_commands.to_string(), + ), + ], + ); + } } Err(_) => { succeeded = false;