diff --git a/crates/tracedecay-domain/Cargo.toml b/crates/tracedecay-domain/Cargo.toml index 7cfac3754f..cf844b1b39 100644 --- a/crates/tracedecay-domain/Cargo.toml +++ b/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"] } diff --git a/crates/tracedecay-domain/src/identity_digest_probe.rs b/crates/tracedecay-domain/src/identity_digest_probe.rs new file mode 100644 index 0000000000..fb3afdb589 --- /dev/null +++ b/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 = 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 56d390f1cc..bc63e02206 100644 --- a/crates/tracedecay-domain/src/lib.rs +++ b/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; diff --git a/crates/tracedecay-domain/src/observation.rs b/crates/tracedecay-domain/src/observation.rs index d295e27546..7c68b2e97e 100644 --- a/crates/tracedecay-domain/src/observation.rs +++ b/crates/tracedecay-domain/src/observation.rs @@ -2307,6 +2307,8 @@ fn domain_digest( domain: &[u8], value: &impl Serialize, ) -> Result { + #[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)) } diff --git a/crates/tracedecay-domain/src/research/canonical.rs b/crates/tracedecay-domain/src/research/canonical.rs index fc0e0500d3..6ed11a51a2 100644 --- a/crates/tracedecay-domain/src/research/canonical.rs +++ b/crates/tracedecay-domain/src/research/canonical.rs @@ -33,6 +33,8 @@ 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 a8a3948f96..5b999eaee0 100644 --- a/crates/tracedecay-global-db/Cargo.toml +++ b/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]] diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 1e88baae1a..72b233873a 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/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,75 +21,28 @@ 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; - - fn dispatch_submit( - &self, - request: RuntimeSubmitRequestV1, - probe: Arc, - ) -> impl Future> + Send; -} - -impl ObservationRuntimeDispatch for DatabaseRuntimeClientV1 { - fn binding(&self) -> &StoreRuntimeBindingV1 { - DatabaseRuntimeClientV1::binding(self) - } - - fn dispatch_read( - &self, - request: RuntimeReadRequestV1, - probe: &dyn RuntimeRequestProbeV1, - ) -> Result { - DatabaseRuntimeClientV1::dispatch_read(self, request, probe) - } - - async fn dispatch_submit( - &self, - request: RuntimeSubmitRequestV1, - probe: Arc, - ) -> Result { - 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 { +pub struct GlobalDbObservationStore { database: Database, - runtime: R, + runtime: DatabaseRuntimeClientV1, } impl GlobalDbObservationStore { @@ -98,15 +50,6 @@ impl GlobalDbObservationStore { let runtime = database.runtime_client(); Self { database, runtime } } -} - -impl GlobalDbObservationStore { - /// 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 GlobalDbObservationStore { &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 GlobalDbObservationStore { } } -impl ObservationStore for GlobalDbObservationStore { +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 { 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> { @@ -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> { @@ -945,7 +885,7 @@ async fn read_admission_refusal( } fn read_runtime_stored_observation( - runtime: &impl ObservationRuntimeDispatch, + runtime: &DatabaseRuntimeClientV1, observation_id: &CanonicalObservationIdV1, ) -> ObservationStoreResult> { 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 ObservationProjectionStore for GlobalDbObservationStore { +impl ObservationProjectionStore for GlobalDbObservationStore { async fn next_queued_observation( &self, ) -> ProjectionStoreResult> { diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index 4d47258d3e..686f1fb6ba 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -24,10 +24,14 @@ //! 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 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 and no native/canonical identity -//! or payload digest was constructed on a subsequent trigger. +//! * 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. use serde_json::{Value, json}; use std::path::Path; @@ -42,7 +46,7 @@ use tracedecay_domain::{ ObservationSourceGenerationV1, ObservationSourceIdentityV1, ObservationSourceRangeV1, PayloadReferenceV1, ProjectionGenerationId, ProviderId, RetentionClass, SanitizationReceiptId, SanitizationReceiptRefV1, SanitizationReceiptV1, SanitizerDispositionV1, SensitivityV1, - SessionId, UtcMicros, + SessionId, UtcMicros, identity_digest_probe, }; use tracedecay_store::{ AnchoredObservationWrite, ObservationCoverageReason, ObservationPersistOutcome, @@ -50,14 +54,29 @@ use tracedecay_store::{ ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, }; -use crate::tests::harness::{ - CountingObservationRuntime, HostAdmissionScope, HostAdmissionTestRuntimeV1, - ObservationDispatchCounts, -}; +use crate::tests::harness::{HostAdmissionScope, HostAdmissionTestRuntimeV1}; use tracedecay_runtime_core::db::engine::params; const COLLISION_PROVIDER: &str = "collision-test"; +/// One thread-local view of the sanctioned domain digest probe: +/// `(identity digests, payload digests, canonical command digests)`. +/// +/// 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(), + ) +} + fn fixture_receipt(receipt_id: &str, payload: &Value) -> SanitizationReceiptV1 { SanitizationReceiptV1::new( SanitizationReceiptRefV1::new( @@ -541,8 +560,8 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (store, counts) = runtime - .counting_observation_store(HostAdmissionScope::Profile) + let store = runtime + .observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.identity-collision.readmitted").unwrap(); let (original, original_write) = collision_candidate( @@ -569,7 +588,7 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() "receipt.identity-collision.readmitted.rewritten", committed_cursor, ); - let before_first = counts.snapshot(); + let before_first = digest_counts(); let first = store .persist_observation(rewritten_write.clone()) .await @@ -584,12 +603,14 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() ), "{first:?}" ); - let before = counts.snapshot(); + let before = digest_counts(); // Non-vacuity anchor: the first collision classifies against the stored - // row, so the counting seam must have observed its stored-row read. + // row and converges coverage, so the probe must have counted its + // stored-row read and frontier cursor-read command digests. assert!( - before.stored_observation_reads > before_first.stored_observation_reads, - "the first collision must read the stored row through the counted dispatch seam" + 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" ); // A later catch-up pass or temporal trigger re-presents the exact same @@ -609,25 +630,28 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() "{second:?}" ); - let after = counts.snapshot(); + let after = digest_counts(); assert_eq!( - after.stored_observation_reads - before.stored_observation_reads, + after.0 - before.0, 0, - "re-admitted terminal collision must not read the stored observation row again — \ - without that read it cannot decode, re-classify the collision, or re-probe the \ - payload revision" + "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" ); assert_eq!( - after.submits - before.submits, + after.1 - before.1, 0, - "re-admitted terminal collision must not submit a runtime write, so it never \ - canonicalizes or hashes a command again" + "re-admitted terminal collision must not canonicalize or hash any payload" ); - // The seam stayed live inside the asserted window: the re-admission's own - // frontier check dispatches exactly one typed source-cursor read. - assert!( - after.source_cursor_reads > before.source_cursor_reads, - "the counting seam must observe the re-admission's frontier cursor read" + // 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" ); // The terminal coverage stays single-row and the cursor stays put. assert_eq!( @@ -654,8 +678,8 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (store, counts) = runtime - .counting_observation_store(HostAdmissionScope::Profile) + let store = runtime + .observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.identity-collision.domain-replacement").unwrap(); let (original, original_write) = collision_candidate( @@ -719,7 +743,7 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() ); assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); - let before = counts.snapshot(); + let before = digest_counts(); let second = store .persist_observation(replacement_write) .await @@ -731,16 +755,13 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() .. } )); - let after = counts.snapshot(); + let after = digest_counts(); assert_eq!( - ( - after.stored_observation_reads - before.stored_observation_reads, - after.submits - before.submits, - ), - (0, 0), - "re-admission must not read the stored row (the only route to decoding, \ - classifying, or revision-probing it) and must not submit (so nothing is \ - canonicalized or hashed)" + (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" ); } @@ -1060,13 +1081,13 @@ async fn raw_observation_json( /// Boundary accounting for one record a catch-up pass decoded and persisted. struct CatchUpRecordReceipt { result: Result, - /// Runtime-dispatch deltas across the persist call, observed at the - /// adapter's real dispatch seam: `(stored-observation reads, runtime - /// submits)`. A stored-observation read is the only route to decoding, - /// collision-classifying, or revision-probing the retained row, and - /// every submit is keyed by exactly one canonical command digest — so - /// `(0, 0)` proves the persist call repeated none of that record work. - dispatch_deltas: (u64, u64), + /// 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), } /// One real catch-up pass over raw persisted source input: read the durable @@ -1075,8 +1096,7 @@ struct CatchUpRecordReceipt { /// ABORTING the pass on a persist error — an identity collision ends the /// pass, it does not skip to the next record. async fn run_catch_up_pass( - store: &crate::GlobalDbObservationStore, - counts: &ObservationDispatchCounts, + store: &crate::GlobalDbObservationStore, session_id: &SessionId, generation: u64, raw_lines: &[((u64, u64), String)], @@ -1107,16 +1127,13 @@ async fn run_catch_up_pass( &format!("receipt.catch-up.{pass_label}.{index}"), ); let write = anchored_write_for(observation, cursor); - let before = counts.snapshot(); + let before = digest_counts(); let result = store.persist_observation(write).await; - let after = counts.snapshot(); + let after = digest_counts(); let aborted = result.is_err(); receipts.push(CatchUpRecordReceipt { result, - dispatch_deltas: ( - after.stored_observation_reads - before.stored_observation_reads, - after.submits - before.submits, - ), + digest_deltas: (after.0 - before.0, after.1 - before.1, after.2 - before.2), }); if aborted { break; @@ -1639,8 +1656,8 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (store, counts) = runtime - .counting_observation_store(HostAdmissionScope::Profile) + let store = runtime + .observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.retention").unwrap(); @@ -1677,7 +1694,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco // Pass 0: gen-1 ingest of the original file. let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -1689,7 +1706,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco // refusal's own coverage advance lets the follow-up pass move on to // record one and converge. let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; + 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, @@ -1698,15 +1715,8 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco .. }) )); - let (decoded, receipts) = run_catch_up_pass( - &store, - &counts, - &session_id, - 2, - &rewritten_lines, - "gen2-resume", - ) - .await; + let (decoded, receipts) = + 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, @@ -1723,8 +1733,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); // Pass 2: a later catch-up pass reopens nothing. - let (decoded, _) = - run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2-b").await; + let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!( decoded, 0, "catch-up must not reopen covered source records" @@ -1768,7 +1777,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco // refused candidate without a current frontier view) still terminates // with zero decode/derive/hash work. let stale_replay = anchored_write_for(refused.clone(), None); - let before = counts.snapshot(); + let before = digest_counts(); let error = store.persist_observation(stale_replay).await.unwrap_err(); assert!( matches!( @@ -1780,17 +1789,12 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco ), "{error:?}" ); - let after = counts.snapshot(); + let after = digest_counts(); assert_eq!( - after.stored_observation_reads - before.stored_observation_reads, - 0, - "the stale re-admission must not read the stored row, so it cannot decode, \ - classify, or revision-probe it" - ); - assert_eq!( - after.submits - before.submits, - 0, - "the stale re-admission must not submit a runtime write, so nothing is hashed" + (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" ); // Restart: the terminal and coverage are durable, catch-up still reopens @@ -1800,18 +1804,11 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (reopened_store, reopened_counts) = reopened - .counting_observation_store(HostAdmissionScope::Profile) + let reopened_store = reopened + .observation_store(HostAdmissionScope::Profile) .unwrap(); - let (decoded, _) = run_catch_up_pass( - &reopened_store, - &reopened_counts, - &session_id, - 2, - &rewritten_lines, - "gen2-c", - ) - .await; + let (decoded, _) = + run_catch_up_pass(&reopened_store, &session_id, 2, &rewritten_lines, "gen2-c").await; assert_eq!(decoded, 0); assert_eq!(admission_refusal_rows(&reopened).await.len(), 1); assert_eq!( @@ -1829,11 +1826,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. /// -/// Runtime-dispatch counts at the adapter's dispatch seam prove the retained -/// row is not read again (so never decoded, collision classified, or -/// revision-probed) and nothing is submitted (so never command-digested). The -/// real Vibe journey below separately proves the production source boundary -/// performs no subsequent frame materialization. +/// 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. #[tokio::test] async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework() { use crate::observation::retention::{ObservationRetentionConfig, RetentionMode}; @@ -1842,8 +1841,8 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (store, counts) = runtime - .counting_observation_store(HostAdmissionScope::Profile) + let store = runtime + .observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.rescan").unwrap(); let original_lines = vec![( @@ -1880,14 +1879,14 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework // record terminally and commits the appended record, advancing the cursor // strictly past the refused coverage. let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, Ok(ObservationPersistOutcome::Committed(_)) )); let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; + 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, @@ -1896,15 +1895,8 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework .. }) )); - let (decoded, receipts) = run_catch_up_pass( - &store, - &counts, - &session_id, - 2, - &rewritten_lines, - "gen2-resume", - ) - .await; + let (decoded, receipts) = + 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, @@ -1944,7 +1936,7 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework // fast path answers from the terminal, converges coverage with one typed // cursor-advance write, and aborts the pass like production. let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3").await; + run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3").await; assert_eq!( decoded, 1, "a rescan after a real file change re-reads the raw source and aborts on the collision" @@ -1962,11 +1954,11 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework refused_readmit.result ); assert_eq!( - refused_readmit.dispatch_deltas, - (0, 0), - "the re-admit must not read the stored row (so it cannot classify or probe \ - revisions) and must not submit (so it digests nothing); coverage converges \ - inside one direct authority transaction with no record work" + 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 suppression above was answered by the retained refusal terminal: // it must have survived cursor-advance retention. @@ -1977,19 +1969,11 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework ); // The resumed pass commits the appended record past the converged // coverage, and the NEXT pass reopens zero source records. - let (decoded, receipts) = run_catch_up_pass( - &store, - &counts, - &session_id, - 3, - &rewritten_lines, - "gen3-resume", - ) - .await; + 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); - let (decoded, _) = - run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3-b").await; + 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. @@ -2014,8 +1998,8 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (store, counts) = runtime - .counting_observation_store(HostAdmissionScope::Profile) + let store = runtime + .observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.eof").unwrap(); // The refused record is the ONLY record: nothing follows it, ever. @@ -2029,7 +2013,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { )]; let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -2039,7 +2023,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { // Gen-2 rescan: the EOF record collides and the pass aborts. The refusal // records terminal + coverage, so the SAME generation never reopens it. let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; + run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -2056,8 +2040,7 @@ 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; - let (decoded, _) = - run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2-b").await; + let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!( decoded, 0, "the refused EOF coverage holds within its generation" @@ -2082,7 +2065,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { // terminal AND converges the new generation's coverage, so this exact // decode happens once per real file change — never again for gen 3. let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3").await; + run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3").await; assert_eq!(decoded, 1); let readmit = &receipts[0]; assert!( @@ -2097,12 +2080,12 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { readmit.result ); assert_eq!( - readmit.dispatch_deltas, - (0, 0), - "the EOF re-admit converges coverage atomically with no record work" + 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" ); - let (decoded, _) = - run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3-b").await; + 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" @@ -2127,18 +2110,11 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (reopened_store, reopened_counts) = reopened - .counting_observation_store(HostAdmissionScope::Profile) + let reopened_store = reopened + .observation_store(HostAdmissionScope::Profile) .unwrap(); - let (decoded, _) = run_catch_up_pass( - &reopened_store, - &reopened_counts, - &session_id, - 3, - &rewritten_lines, - "gen3-c", - ) - .await; + let (decoded, _) = + run_catch_up_pass(&reopened_store, &session_id, 3, &rewritten_lines, "gen3-c").await; assert_eq!( decoded, 0, "restarted rescans must never reopen the refused EOF record" @@ -2162,8 +2138,8 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (store, counts) = runtime - .counting_observation_store(HostAdmissionScope::Profile) + let store = runtime + .observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.orphan").unwrap(); let original_lines = vec![( @@ -2175,7 +2151,7 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { raw_source_line(&session_id, "record.orphan.0", (0, 1), "rewritten record"), )]; let (decoded, receipts) = - run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -2223,7 +2199,7 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { // 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, &counts, &session_id, 2, &rewritten_lines, "gen2").await; + run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1); let repair = &receipts[0]; assert!( @@ -2238,15 +2214,15 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { repair.result ); assert_eq!( - repair.dispatch_deltas, - (0, 0), - "the orphan-marker re-admit repairs coverage atomically with no record work" + 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" ); // Coverage is repaired: later passes never reopen the record, even after // a restart, and no duplicate marker rows appear. - let (decoded, _) = - run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2-b").await; + 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); drop(store); @@ -2254,18 +2230,11 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let (reopened_store, reopened_counts) = reopened - .counting_observation_store(HostAdmissionScope::Profile) + let reopened_store = reopened + .observation_store(HostAdmissionScope::Profile) .unwrap(); - let (decoded, _) = run_catch_up_pass( - &reopened_store, - &reopened_counts, - &session_id, - 2, - &rewritten_lines, - "gen2-c", - ) - .await; + let (decoded, _) = + run_catch_up_pass(&reopened_store, &session_id, 2, &rewritten_lines, "gen2-c").await; assert_eq!(decoded, 0); assert_eq!( raw_observation_json(&reopened, refused.observation_id().as_str()).await, diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index 28bc9772aa..b9ad7352f3 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/crates/tracedecay-global-db/src/registered.rs @@ -581,22 +581,6 @@ impl RegisteredGlobalDb { crate::GlobalDbObservationStore::new(self.database.clone()) } - /// Test-only [`Self::observation_store`] variant that binds the adapter - /// to an explicit runtime dispatch seam (e.g. a counting wrapper over - /// this client's runtime client), so tests can observe the record work a - /// persist call dispatches without changing the adapter's production - /// shape. - #[cfg(test)] - pub(crate) fn observation_store_with_runtime_dispatch( - &self, - runtime: R, - ) -> crate::GlobalDbObservationStore - where - R: crate::observation_adapter::ObservationRuntimeDispatch, - { - crate::GlobalDbObservationStore::with_runtime_dispatch(self.database.clone(), runtime) - } - /// Retains this exact client for closed runtime read/submit requests. /// /// The returned capability has no raw Store runtime, connection, or diff --git a/crates/tracedecay-global-db/src/tests/harness.rs b/crates/tracedecay-global-db/src/tests/harness.rs index 68bc86e033..504862e60a 100644 --- a/crates/tracedecay-global-db/src/tests/harness.rs +++ b/crates/tracedecay-global-db/src/tests/harness.rs @@ -476,107 +476,6 @@ pub(crate) enum SessionTemporalFixtureCountV1 { RefreshProgress, } -/// Runtime-dispatch counts observed by [`CountingObservationRuntime`]. -/// -/// `stored_observation_reads` counts stored-record runtime reads — the only -/// way a persist path can obtain, decode, collision-classify, or -/// revision-probe a retained observation row. `submits` counts runtime -/// writes — every submit is keyed by exactly one canonical command digest -/// computed immediately before dispatch, and the digest call sites are -/// unreachable without either a stored-record read or a submit. A window -/// with zero deltas on both therefore proves the persist call repeated none -/// of the expensive record work, while `source_cursor_reads` shows the seam -/// stayed live inside that same window. -#[cfg(test)] -#[derive(Debug, Default)] -pub(crate) struct ObservationDispatchCounts { - stored_observation_reads: AtomicU64, - source_cursor_reads: AtomicU64, - submits: AtomicU64, -} - -/// One consistent view of [`ObservationDispatchCounts`]. -#[cfg(test)] -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(crate) struct ObservationDispatchSnapshot { - pub(crate) stored_observation_reads: u64, - pub(crate) source_cursor_reads: u64, - pub(crate) submits: u64, -} - -#[cfg(test)] -impl ObservationDispatchCounts { - pub(crate) fn snapshot(&self) -> ObservationDispatchSnapshot { - ObservationDispatchSnapshot { - stored_observation_reads: self.stored_observation_reads.load(Ordering::Relaxed), - source_cursor_reads: self.source_cursor_reads.load(Ordering::Relaxed), - submits: self.submits.load(Ordering::Relaxed), - } - } -} - -/// Counting test double over the production runtime client at the adapter's -/// real dispatch seam: every read and submit is forwarded unchanged to the -/// wrapped [`DatabaseRuntimeClientV1`] after classifying it into -/// [`ObservationDispatchCounts`]. -#[cfg(test)] -#[derive(Clone)] -pub(crate) struct CountingObservationRuntime { - inner: tracedecay_runtime_core::db::DatabaseRuntimeClientV1, - counts: Arc, -} - -#[cfg(test)] -impl crate::observation_adapter::ObservationRuntimeDispatch for CountingObservationRuntime { - fn binding(&self) -> &tracedecay_store::StoreRuntimeBindingV1 { - self.inner.binding() - } - - fn dispatch_read( - &self, - request: tracedecay_store::RuntimeReadRequestV1, - probe: &dyn tracedecay_store::RuntimeRequestProbeV1, - ) -> Result< - tracedecay_store::RuntimeReadOutcomeV1, - tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure, - > { - if let tracedecay_store::RuntimeReadOperationV1::Repository { - op: - tracedecay_store::RepositoryReadOperationV1::Project( - tracedecay_store::ProjectReadOperationV1::Observation(operation), - ), - } = request.operation() - { - match operation { - tracedecay_store::ObservationReadOperationV1::Observation { .. } => { - self.counts - .stored_observation_reads - .fetch_add(1, Ordering::Relaxed); - } - tracedecay_store::ObservationReadOperationV1::SourceCursor { .. } => { - self.counts - .source_cursor_reads - .fetch_add(1, Ordering::Relaxed); - } - _ => {} - } - } - self.inner.dispatch_read(request, probe) - } - - async fn dispatch_submit( - &self, - request: tracedecay_store::RuntimeSubmitRequestV1, - probe: Arc, - ) -> Result< - tracedecay_store::RuntimeSubmitOutcomeV1, - tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure, - > { - self.counts.submits.fetch_add(1, Ordering::Relaxed); - self.inner.dispatch_submit(request, probe).await - } -} - /// Test-only registered database fixture retained below the use-case layer. #[doc(hidden)] pub struct HostAdmissionTestRuntimeV1 { @@ -736,29 +635,6 @@ impl HostAdmissionTestRuntimeV1 { Ok(database.observation_store()) } - /// An observation store whose runtime dispatch seam is wrapped in - /// [`CountingObservationRuntime`], plus the shared counters, so a test - /// can prove exactly which record work a persist call dispatched. - #[cfg(test)] - pub(crate) fn counting_observation_store( - &self, - scope: HostAdmissionScope, - ) -> tracedecay_runtime_core::errors::Result<( - crate::GlobalDbObservationStore, - Arc, - )> { - let database = self.session_database_for_test(scope)?; - let counts = Arc::new(ObservationDispatchCounts::default()); - let runtime = CountingObservationRuntime { - inner: database.runtime_client(), - counts: Arc::clone(&counts), - }; - Ok(( - database.observation_store_with_runtime_dispatch(runtime), - counts, - )) - } - pub async fn upsert_session_for_test( &self, scope: HostAdmissionScope,