test(global-db): prove refusal fast path with corruption tripwire

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]

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;

crates/tracedecay-domain/src/observation.rs

@@ -2307,8 +2307,11 @@ fn domain_digest(
     domain: &[u8],
     value: &impl Serialize,
 ) -> Result<String, ObservationContractError> {
-    #[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))
 }
 

crates/tracedecay-domain/src/research/canonical.rs

@@ -33,8 +33,6 @@ pub fn canonical_json_value(value: &Value) -> Result<String, DomainError> {
 /// 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<T: Serialize>(value: &T) -> Result<ManifestDigest, DomainError> {
-    #[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();

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

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,
 };

crates/tracedecay-global-db/src/observation/retention.rs

@@ -277,6 +277,23 @@ pub struct ObservationRetentionPhaseReport {
     pub oldest_eligible_at: Option<i64>,
 }
 
+/// 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<String>,
 }
 
@@ -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<String>,
+) -> 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<ObservationAdmissionWorkRollupV1, String> = 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::<i64>(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;

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