fix(sessions): block cover-past on observation commit failures

crates/tracedecay-sessions/src/observation.rs

@@ -438,9 +438,21 @@ where
                     if cancellation.is_cancelled() {
                         return Err(ObservationApplicationError::Cancelled);
                     }
-                    let projection_status = stored
-                        .ok_or(ObservationApplicationError::PersistedObservationUnavailable)?
-                        .projection_status();
+                    // A newly committed row is queued by this persist even
+                    // when an immediate reader snapshot trails the write. An
+                    // exact or covered duplicate wrote no projection state,
+                    // so a read-back miss cannot truthfully invent one.
+                    let projection_status = match stored {
+                        Some(stored) => stored.projection_status(),
+                        None if matches!(&outcome, ObservationPersistOutcome::Committed(_)) => {
+                            ObservationProjectionStatus::Queued
+                        }
+                        None => {
+                            return Err(
+                                ObservationApplicationError::PersistedObservationUnavailable,
+                            );
+                        }
+                    };
                     Ok(CaptureObservationOutcome::Persisted {
                         outcome: Box::new(outcome),
                         projection_status,

crates/tracedecay-sessions/src/observation_test.rs

@@ -26,6 +26,9 @@ struct FakeStore {
     cancel_on_replay: Mutex<Option<ObservationCancellation>>,
     cancel_on_advance: Mutex<Option<ObservationCancellation>>,
     cursor_advances: Mutex<Vec<ObservationCursorAdvance>>,
+    /// One read-your-writes miss: the next point read reports the committed
+    /// row as absent, the way a trailing reader snapshot does under load.
+    read_none_once: Mutex<bool>,
 }
 
 impl ObservationStore for FakeStore {
@@ -119,6 +122,9 @@ impl ObservationStore for FakeStore {
         &self,
         observation_id: &CanonicalObservationIdV1,
     ) -> ObservationStoreResult<Option<StoredObservation>> {
+        if std::mem::take(&mut *self.read_none_once.lock().unwrap()) {
+            return Ok(None);
+        }
         let observation = self
             .observations
             .lock()
@@ -446,6 +452,33 @@ fn request_accepts_only_bounded_parser_evidence_for_the_identity_range() {
     ));
 }
 
+#[tokio::test]
+async fn committed_capture_with_missed_read_back_stays_persisted_as_queued() {
+    let application = application();
+    *application.store.read_none_once.lock().unwrap() = true;
+
+    let outcome = application
+        .capture_claude_observation(request(&json!({
+            "type": "user",
+            "message": { "role": "user", "content": "read-your-writes miss" }
+        })))
+        .await
+        .expect("a committed persist must not fail on a missed read-back");
+
+    match outcome {
+        CaptureObservationOutcome::Persisted {
+            outcome,
+            projection_status,
+            ..
+        } => {
+            assert!(matches!(*outcome, ObservationPersistOutcome::Committed(_)));
+            assert_eq!(projection_status, ObservationProjectionStatus::Queued);
+        }
+        other => panic!("capture must stay persisted, got {other:?}"),
+    }
+    assert_eq!(application.store.observations.lock().unwrap().len(), 1);
+}
+
 #[tokio::test]
 async fn exact_duplicate_reports_authoritative_projection_status() {
     let application = application();
@@ -500,6 +533,31 @@ async fn exact_duplicate_reports_authoritative_projection_status() {
     assert_eq!(application.store.observations.lock().unwrap().len(), 1);
 }
 
+#[tokio::test]
+async fn exact_duplicate_with_missed_read_back_stays_typed_unavailable() {
+    let application = application();
+    let record = json!({
+        "type": "user",
+        "message": { "role": "user", "content": "duplicate read miss" }
+    });
+    application
+        .capture_claude_observation(request(&record))
+        .await
+        .expect("first capture persists");
+    *application.store.read_none_once.lock().unwrap() = true;
+
+    let error = application
+        .capture_claude_observation(request(&record))
+        .await
+        .expect_err("a duplicate read miss cannot fabricate projection status");
+
+    assert!(matches!(
+        error,
+        ObservationApplicationError::PersistedObservationUnavailable
+    ));
+    assert_eq!(application.store.observations.lock().unwrap().len(), 1);
+}
+
 #[tokio::test]
 async fn replay_reports_partial_coverage_and_a_truthful_continuation() {
     let application = application();

crates/tracedecay-sessions/src/runtime/jsonl_observation_admission.rs

@@ -8,7 +8,7 @@ use tracedecay_domain::{
 };
 use tracedecay_store::observation::{ObservationCoverageReason, ObservationCursorAdvance};
 
-use crate::admission::{HostAdmission, is_admission_cancellation};
+use crate::admission::{HostAdmission, HostAdmissionOutcome, is_admission_cancellation};
 use crate::observation::{
     CaptureObservationOutcome, CaptureObservationRequest, ObservationCancellation,
 };
@@ -309,19 +309,13 @@ impl ActiveAdmission<'_> {
                 )
                 .await
             }
-            // Deterministic refusals (content-derived identity conflicts and
-            // other non-retryable dispositions) re-fail identically forever;
+            // Deterministic content refusals re-fail identically forever;
             // advance coverage with a durable typed reason so the stream
             // converges instead of re-reporting the same records every sweep.
-            Err(outcome) if !outcome.retryable => {
-                if self.cancellation.is_cancelled() {
-                    return Err(TranscriptIngestError::NonDurableRecord {
-                        provider: self.provider,
-                        offset: frame.checkpoint.offset,
-                        end_offset: frame.checkpoint.end_offset,
-                        reason: outcome.reason_code.unwrap_or("host_admission_incomplete"),
-                    });
-                }
+            Err(outcome)
+                if is_deterministic_content_refusal(&outcome)
+                    && !self.cancellation.is_cancelled() =>
+            {
                 tracing::warn!(
                     provider = self.provider,
                     offset = frame.checkpoint.offset,
@@ -336,17 +330,24 @@ impl ActiveAdmission<'_> {
                 )
                 .await
             }
-            // Only retryable outcomes reach here (the arm above matched the
-            // rest). A retryable admission failure says nothing about the
-            // record itself, so the admission authority's own verdict must
-            // survive to classification — wrapping it as NonDurable laundered
-            // a converging cursor conflict into a terminal Degraded record.
             Err(outcome) => {
                 if is_admission_cancellation(&outcome, &self.cancellation) {
                     Err(TranscriptIngestError::Cancelled {
                         provider: self.provider,
                     })
                 } else {
+                    // Everything else says nothing about the record's
+                    // content: commit/read-back failures
+                    // (`observation_commit_failed`,
+                    // `authority_write_failed`,
+                    // `observation_persisted_value_unavailable`), unbound
+                    // authorities, and retryable races keep the admission
+                    // authority's own verdict as a typed block. The frontier
+                    // must not advance over a record whose durable fate is
+                    // unknown — the persist may already have committed and
+                    // advanced the source cursor, so a cover-past write here
+                    // would stack a second, conflicting cursor advance on
+                    // every frame.
                     Err(host_admission_error(self.provider, outcome))
                 }
             }
@@ -565,6 +566,21 @@ pub(super) async fn admit_jsonl_observations<State>(
     Ok(progress)
 }
 
+/// Non-retryable admission failures that are verdicts about the record's
+/// content. Only these may be covered past: they re-fail identically on every
+/// sweep, so a durable `AdmissionRefused` coverage row is what lets the
+/// stream converge. Every other failure — store commit/read-back failures,
+/// unbound authorities, retryable races — says nothing about the record and
+/// must surface as a typed block instead of writing coverage over a commit
+/// that never landed (or one that already landed and advanced the cursor).
+fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool {
+    !outcome.retryable
+        && matches!(
+            outcome.reason_code,
+            Some("invalid_observation_contract" | "privacy_boundary_failed")
+        )
+}
+
 /// Log identity for a transcript file. Transcript paths sit under the
 /// operator's home directory and name real sessions, so ingest logs carry the
 /// basename only rather than persisting an absolute path into the daemon log.
@@ -599,3 +615,6 @@ pub(super) fn preflight_and_parse_new(
     preflight_strict_jsonl(provider, path, prev, max_new_bytes)?;
     Ok(parse_new())
 }
+
+#[cfg(test)]
+mod tests;