refactor(global-db): concrete adapter with domain digest-boundary proof

crates/tracedecay-domain/Cargo.toml

@@ -7,6 +7,14 @@ 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"] }

crates/tracedecay-domain/src/identity_digest_probe.rs

@@ -0,0 +1,68 @@
+//! 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<u64> = const { Cell::new(0) };
+    static PAYLOAD_DIGESTS: Cell<u64> = const { Cell::new(0) };
+    static CANONICAL_DIGESTS: Cell<u64> = 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)
+}

crates/tracedecay-domain/src/lib.rs

@@ -12,6 +12,8 @@ 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,6 +2307,8 @@ fn domain_digest(
     domain: &[u8],
     value: &impl Serialize,
 ) -> Result<String, ObservationContractError> {
+    #[cfg(feature = "identity-digest-probe")]
+    crate::identity_digest_probe::record_identity();
     let bytes =
         canonical_json_bytes(value).map_err(|_| ObservationContractError::CanonicalEncoding)?;
     let mut hasher = Sha256::new();
@@ -2349,6 +2351,8 @@ fn accepted_identity_digests(
 }
 
 fn sha256_digest(bytes: &[u8]) -> String {
+    #[cfg(feature = "identity-digest-probe")]
+    crate::identity_digest_probe::record_payload();
     format_sha256(&Sha256::digest(bytes))
 }
 

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

@@ -33,6 +33,8 @@ 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,6 +55,11 @@ 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/observation_adapter.rs

@@ -1,4 +1,3 @@
-use std::future::Future;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use tracedecay_application::clock::now_micros;
@@ -22,91 +21,35 @@ use tracedecay_store::{
     RepositoryReadOperationV1, RepositoryReadResultV1, RepositoryWritePayloadV1,
     RuntimeBatchCompatibilityV1, RuntimeCancellationIdV1, RuntimeCancellationIdentityV1,
     RuntimeDeadlineIdV1, RuntimeDeadlineV1, RuntimeInterruptionV1, RuntimeReadCoverageV1,
-    RuntimeReadOperationV1, RuntimeReadOutcomeV1, RuntimeReadRequestV1, RuntimeReadResultV1,
-    RuntimeRequestControlV1, RuntimeRequestProbeV1, RuntimeSubmitOutcomeV1, RuntimeSubmitRequestV1,
-    RuntimeTransactionIdV1, RuntimeTransactionScopeV1, StoreClientIdV1, StoreIdempotencyKeyV1,
-    StoreOperationIdV1, StoreOperationMetadataV1, StoreRuntimeBindingV1, StoredObservation,
-    StoredObservationRowV1,
+    RuntimeReadOperationV1, RuntimeReadRequestV1, RuntimeReadResultV1, RuntimeRequestControlV1,
+    RuntimeRequestProbeV1, RuntimeSubmitOutcomeV1, RuntimeSubmitRequestV1, RuntimeTransactionIdV1,
+    RuntimeTransactionScopeV1, StoreClientIdV1, StoreIdempotencyKeyV1, StoreOperationIdV1,
+    StoreOperationMetadataV1, StoredObservation, StoredObservationRowV1,
 };
 
 use tracedecay_runtime_core::db::{Database, DatabaseRuntimeClientV1};
-use tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure;
 use tracedecay_rusqlite_runtime::repository::observation_cursor_authority::{
     COMMIT_SOURCE_CURSOR_SQL, READ_CURSOR_ADVANCE_SQL, READ_SOURCE_CURSOR_SQL,
     RECORD_CURSOR_ADVANCE_SQL, cursor_advance_ledger_row_matches,
 };
 
-/// The closed dispatch boundary between this adapter and the authoritative
-/// store runtime. Every stored-record read and every runtime write the
-/// adapter performs crosses this seam, so an observer wrapped around it sees
-/// exactly the record work one persist call dispatches — the collision tests
-/// prove the terminal-refusal fast path repeats no stored-row read (and thus
-/// no decode, classification, or revision probing) and no submit (and thus no
-/// canonical command digest) by counting here.
-pub(crate) trait ObservationRuntimeDispatch: Clone + Send + Sync {
-    fn binding(&self) -> &StoreRuntimeBindingV1;
-
-    fn dispatch_read(
-        &self,
-        request: RuntimeReadRequestV1,
-        probe: &dyn RuntimeRequestProbeV1,
-    ) -> Result<RuntimeReadOutcomeV1, StoreRuntimeRegistryFailure>;
-
-    fn dispatch_submit(
-        &self,
-        request: RuntimeSubmitRequestV1,
-        probe: Arc<dyn RuntimeRequestProbeV1>,
-    ) -> impl Future<Output = Result<RuntimeSubmitOutcomeV1, StoreRuntimeRegistryFailure>> + Send;
-}
-
-impl ObservationRuntimeDispatch for DatabaseRuntimeClientV1 {
-    fn binding(&self) -> &StoreRuntimeBindingV1 {
-        DatabaseRuntimeClientV1::binding(self)
-    }
-
-    fn dispatch_read(
-        &self,
-        request: RuntimeReadRequestV1,
-        probe: &dyn RuntimeRequestProbeV1,
-    ) -> Result<RuntimeReadOutcomeV1, StoreRuntimeRegistryFailure> {
-        DatabaseRuntimeClientV1::dispatch_read(self, request, probe)
-    }
-
-    async fn dispatch_submit(
-        &self,
-        request: RuntimeSubmitRequestV1,
-        probe: Arc<dyn RuntimeRequestProbeV1>,
-    ) -> Result<RuntimeSubmitOutcomeV1, StoreRuntimeRegistryFailure> {
-        DatabaseRuntimeClientV1::dispatch_submit(self, request, probe).await
-    }
-}
-
-/// Observation-store adapter over the already-registered authoritative runtime.
-///
-/// The runtime parameter exists so tests can observe the dispatch seam;
-/// production only ever constructs the default [`DatabaseRuntimeClientV1`]
-/// via [`GlobalDbObservationStore::new`], and the struct shape is identical
-/// in every build.
+/// 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.
 #[derive(Clone)]
-pub struct GlobalDbObservationStore<R = DatabaseRuntimeClientV1> {
+pub struct GlobalDbObservationStore {
     database: Database,
-    runtime: R,
+    runtime: DatabaseRuntimeClientV1,
 }
 
 impl GlobalDbObservationStore {
     pub fn new(database: Database) -> Self {
         let runtime = database.runtime_client();
         Self { database, runtime }
     }
-}
-
-impl<R> GlobalDbObservationStore<R> {
-    /// Binds the adapter to an explicit runtime dispatch seam so a test can
-    /// count the record work a persist path performs.
-    #[cfg(test)]
-    pub(crate) fn with_runtime_dispatch(database: Database, runtime: R) -> Self {
-        Self { database, runtime }
-    }
 
     /// Records a terminal refusal — the marker in
     /// `observation_admission_refusals` AND the typed `admission_refused`
@@ -130,10 +73,7 @@ impl<R> GlobalDbObservationStore<R> {
         &self,
         write: &AnchoredObservationWrite,
         retained_digest: &PayloadDigestV1,
-    ) -> ObservationStoreResult<()>
-    where
-        R: ObservationRuntimeDispatch,
-    {
+    ) -> ObservationStoreResult<()> {
         const OPERATION: &str = "record refused admission terminal and coverage";
         let candidate = write.observation();
         let identity = candidate.identity();
@@ -297,7 +237,7 @@ impl<R> GlobalDbObservationStore<R> {
     }
 }
 
-impl<R: ObservationRuntimeDispatch> ObservationStore for GlobalDbObservationStore<R> {
+impl ObservationStore for GlobalDbObservationStore {
     async fn persist_observation(
         &self,
         write: AnchoredObservationWrite,
@@ -719,7 +659,7 @@ impl RuntimeRequestProbeV1 for RuntimeObservationProbe {
 }
 
 fn dispatch_runtime_observation_read(
-    runtime: &impl ObservationRuntimeDispatch,
+    runtime: &DatabaseRuntimeClientV1,
     operation: ObservationReadOperationV1,
 ) -> ObservationStoreResult<ObservationReadResultV1> {
     let command_digest = canonical_sha256(&operation)
@@ -820,7 +760,7 @@ fn stored_observation_from_runtime_row(
 }
 
 fn read_runtime_source_cursor(
-    runtime: &impl ObservationRuntimeDispatch,
+    runtime: &DatabaseRuntimeClientV1,
     source: &ClaudeSourceIdentityV1,
     scope: &ObservationScopeV1,
 ) -> ObservationStoreResult<Option<ClaudeSourceCursorV1>> {
@@ -840,7 +780,7 @@ fn read_runtime_source_cursor(
 }
 
 fn read_runtime_retrieval_anchor_by_alias(
-    runtime: &impl ObservationRuntimeDispatch,
+    runtime: &DatabaseRuntimeClientV1,
     scope: &ObservationScopeV1,
     alias: &tracedecay_domain::NativeAliasV2,
 ) -> ObservationStoreResult<Option<tracedecay_domain::RetrievalAnchorId>> {
@@ -945,7 +885,7 @@ async fn read_admission_refusal(
 }
 
 fn read_runtime_stored_observation(
-    runtime: &impl ObservationRuntimeDispatch,
+    runtime: &DatabaseRuntimeClientV1,
     observation_id: &CanonicalObservationIdV1,
 ) -> ObservationStoreResult<Option<StoredObservation>> {
     match dispatch_runtime_observation_read(
@@ -965,7 +905,7 @@ fn read_runtime_stored_observation(
 }
 
 async fn submit_runtime_write(
-    runtime: &impl ObservationRuntimeDispatch,
+    runtime: &DatabaseRuntimeClientV1,
     payload: RepositoryWritePayloadV1,
     idempotency_key: String,
     operation: &'static str,
@@ -1108,7 +1048,7 @@ fn runtime_storage_error(
     }
 }
 
-impl<R: ObservationRuntimeDispatch> ObservationProjectionStore for GlobalDbObservationStore<R> {
+impl ObservationProjectionStore for GlobalDbObservationStore {
     async fn next_queued_observation(
         &self,
     ) -> ProjectionStoreResult<Option<CanonicalObservationIdV1>> {