diff --git a/crates/tracedecay-code-extraction/src/rust_extractor.rs b/crates/tracedecay-code-extraction/src/rust_extractor.rs
index c00f740eda..973e7b59c4 100644
--- a/crates/tracedecay-code-extraction/src/rust_extractor.rs
+++ b/crates/tracedecay-code-extraction/src/rust_extractor.rs
@@ -61,6 +61,70 @@ impl ReceiverTypes {
     }
 }
 
+/// One type parameter's trait bounds, as a method owner.
+///
+/// `T: Processor` makes `value.process()` the callee `Processor::process`.
+/// Two bounds, or a bound the syntax does not name, stay unresolved so the
+/// call is not attached to both traits. This does not rename `self`: #1814
+/// still records that binding from the enclosing type.
+enum ParamBound {
+    Unbound,
+    Unique(String),
+    Ambiguous,
+}
+
+#[derive(Default)]
+struct TraitBounds {
+    parameters: BTreeMap<String, ParamBound>,
+}
+
+struct BoundClause {
+    paths: Vec<String>,
+    ambiguous: bool,
+}
+
+impl TraitBounds {
+    fn declare(&mut self, name: String) {
+        self.parameters.insert(name, ParamBound::Unbound);
+    }
+
+    fn knows(&self, name: &str) -> bool {
+        self.parameters.contains_key(name)
+    }
+
+    fn constrain(&mut self, name: &str, clause: BoundClause) {
+        if clause.paths.is_empty() && !clause.ambiguous {
+            return;
+        }
+        let Some(slot) = self.parameters.get_mut(name) else {
+            return;
+        };
+        if clause.ambiguous || clause.paths.len() != 1 {
+            *slot = ParamBound::Ambiguous;
+            return;
+        }
+        let Some(path) = clause.paths.into_iter().next() else {
+            *slot = ParamBound::Ambiguous;
+            return;
+        };
+        match slot {
+            ParamBound::Unbound => *slot = ParamBound::Unique(path),
+            ParamBound::Unique(existing) if existing == &path => {}
+            ParamBound::Unique(_) | ParamBound::Ambiguous => *slot = ParamBound::Ambiguous,
+        }
+    }
+
+    /// Replace a written type-parameter name with its unique trait. Any other
+    /// path, including the enclosing type recorded for `self`, is unchanged.
+    fn resolve(&self, path: String) -> Option<String> {
+        match self.parameters.get(path.as_str()) {
+            Some(ParamBound::Unique(bound)) => Some(bound.clone()),
+            Some(ParamBound::Ambiguous) => None,
+            Some(ParamBound::Unbound) | None => Some(path),
+        }
+    }
+}
+
 /// Internal state used during AST traversal.
 ///
 /// Borrows the caller's source for the lifetime of the walk: copying the
@@ -1738,6 +1802,17 @@ impl RustExtractor {
         node: TsNode<'_>,
         function: TsNode<'_>,
         receivers: &mut ReceiverTypes,
+    ) {
+        let bounds = Self::trait_bounds_for(state, function);
+        Self::collect_receiver_bindings(state, node, function, receivers, &bounds);
+    }
+
+    fn collect_receiver_bindings(
+        state: &ExtractionState<'_>,
+        node: TsNode<'_>,
+        function: TsNode<'_>,
+        receivers: &mut ReceiverTypes,
+        bounds: &TraitBounds,
     ) {
         match node.kind() {
             "self_parameter" => {
@@ -1749,14 +1824,14 @@ impl RustExtractor {
                 if let Some(pattern) = node.child_by_field_name("pattern") {
                     let type_path = node
                         .child_by_field_name("type")
-                        .and_then(|ty| Self::stated_type_path(state, ty));
+                        .and_then(|ty| Self::receiver_type_path(state, ty, bounds));
                     Self::record_receiver_pattern(state, pattern, type_path, receivers);
                 }
             }
             "let_declaration" => {
                 if let Some(pattern) = node.child_by_field_name("pattern") {
                     let type_path = match node.child_by_field_name("type") {
-                        Some(ty) => Self::stated_type_path(state, ty),
+                        Some(ty) => Self::receiver_type_path(state, ty, bounds),
                         None => node
                             .child_by_field_name("value")
                             .and_then(|value| Self::stated_initializer_type_path(state, value)),
@@ -1793,7 +1868,7 @@ impl RustExtractor {
         let mut cursor = node.walk();
         if cursor.goto_first_child() {
             loop {
-                Self::collect_receiver_types(state, cursor.node(), function, receivers);
+                Self::collect_receiver_bindings(state, cursor.node(), function, receivers, bounds);
                 if !cursor.goto_next_sibling() {
                     break;
                 }
@@ -1846,10 +1921,185 @@ impl RustExtractor {
             "dynamic_type" | "abstract_type" => ty
                 .child_by_field_name("trait")
                 .and_then(|inner| Self::stated_type_path(state, inner)),
+            "higher_ranked_trait_bound" => ty
+                .child_by_field_name("type")
+                .and_then(|inner| Self::stated_type_path(state, inner)),
+            // `impl Trait + 'a` still names that trait. Two nominals do not.
+            "bounded_type" => Self::unique_sum_type_path(state, ty),
             _ => None,
         }
     }
 
+    /// A parameter type, with a type parameter replaced by its unique trait
+    /// bound. `Self` is left as #1814 mapped it: the enclosing type, not the
+    /// trait the parameter happens to implement.
+    fn receiver_type_path(
+        state: &ExtractionState<'_>,
+        ty: TsNode<'_>,
+        bounds: &TraitBounds,
+    ) -> Option<String> {
+        if Self::annotation_is_self(state, ty) {
+            return Self::enclosing_receiver_type(state);
+        }
+        bounds.resolve(Self::stated_type_path(state, ty)?)
+    }
+
+    fn annotation_is_self(state: &ExtractionState<'_>, ty: TsNode<'_>) -> bool {
+        match ty.kind() {
+            "type_identifier" => state.node_text(ty) == "Self",
+            "reference_type" => ty
+                .child_by_field_name("type")
+                .is_some_and(|inner| Self::annotation_is_self(state, inner)),
+            _ => false,
+        }
+    }
+
+    fn unique_sum_type_path(state: &ExtractionState<'_>, ty: TsNode<'_>) -> Option<String> {
+        let mut found = None;
+        let mut cursor = ty.walk();
+        if !cursor.goto_first_child() {
+            return None;
+        }
+        loop {
+            let child = cursor.node();
+            if child.is_named() {
+                let path = match child.kind() {
+                    "lifetime" | "use_bounds" => None,
+                    "bounded_type" => Self::unique_sum_type_path(state, child),
+                    _ => Self::stated_type_path(state, child),
+                };
+                match path {
+                    None if matches!(child.kind(), "lifetime" | "use_bounds") => {}
+                    None => return None,
+                    Some(path) => {
+                        if found.replace(path).is_some() {
+                            return None;
+                        }
+                    }
+                }
+            }
+            if !cursor.goto_next_sibling() {
+                break;
+            }
+        }
+        found
+    }
+
+    fn trait_bounds_for(state: &ExtractionState<'_>, function: TsNode<'_>) -> TraitBounds {
+        let mut ancestors = Vec::new();
+        let mut current = function.parent();
+        while let Some(node) = current {
+            if matches!(node.kind(), "function_item" | "function_signature_item") {
+                break;
+            }
+            if matches!(node.kind(), "impl_item" | "trait_item") {
+                ancestors.push(node);
+            }
+            current = node.parent();
+        }
+        ancestors.reverse();
+        let mut bounds = TraitBounds::default();
+        for item in ancestors {
+            Self::absorb_generic_bounds(state, item, &mut bounds);
+        }
+        Self::absorb_generic_bounds(state, function, &mut bounds);
+        bounds
+    }
+
+    fn absorb_generic_bounds(
+        state: &ExtractionState<'_>,
+        item: TsNode<'_>,
+        bounds: &mut TraitBounds,
+    ) {
+        if let Some(parameters) = item.child_by_field_name("type_parameters") {
+            let mut cursor = parameters.walk();
+            if cursor.goto_first_child() {
+                loop {
+                    let child = cursor.node();
+                    if child.kind() == "type_parameter"
+                        && let Some(name_node) = child.child_by_field_name("name")
+                    {
+                        let name = state.node_text(name_node).to_owned();
+                        bounds.declare(name.clone());
+                        if let Some(clause) = child.child_by_field_name("bounds") {
+                            bounds.constrain(&name, Self::trait_bound_clause(state, clause));
+                        }
+                    }
+                    if !cursor.goto_next_sibling() {
+                        break;
+                    }
+                }
+            }
+        }
+        let Some(where_clause) = Self::child_of_kind(item, "where_clause") else {
+            return;
+        };
+        let mut cursor = where_clause.walk();
+        if !cursor.goto_first_child() {
+            return;
+        }
+        loop {
+            let child = cursor.node();
+            if child.kind() == "where_predicate"
+                && let Some(left) = child.child_by_field_name("left")
+                && left.kind() == "type_identifier"
+            {
+                let name = state.node_text(left);
+                if bounds.knows(name)
+                    && let Some(clause) = child.child_by_field_name("bounds")
+                {
+                    bounds.constrain(name, Self::trait_bound_clause(state, clause));
+                }
+            }
+            if !cursor.goto_next_sibling() {
+                break;
+            }
+        }
+    }
+
+    fn trait_bound_clause(state: &ExtractionState<'_>, bounds: TsNode<'_>) -> BoundClause {
+        let mut clause = BoundClause {
+            paths: Vec::new(),
+            ambiguous: false,
+        };
+        let mut cursor = bounds.walk();
+        if !cursor.goto_first_child() {
+            return clause;
+        }
+        loop {
+            let child = cursor.node();
+            if child.is_named() {
+                match child.kind() {
+                    "lifetime" | "use_bounds" | "removed_trait_bound" => {}
+                    _ => match Self::stated_type_path(state, child) {
+                        Some(path) => clause.paths.push(path),
+                        None => clause.ambiguous = true,
+                    },
+                }
+            }
+            if !cursor.goto_next_sibling() {
+                break;
+            }
+        }
+        clause
+    }
+
+    fn child_of_kind<'t>(node: TsNode<'t>, kind: &str) -> Option<TsNode<'t>> {
+        let mut cursor = node.walk();
+        if !cursor.goto_first_child() {
+            return None;
+        }
+        loop {
+            let child = cursor.node();
+            if child.kind() == kind {
+                return Some(child);
+            }
+            if !cursor.goto_next_sibling() {
+                return None;
+            }
+        }
+    }
+
     /// The type a `let` initialiser states in syntax: a `T { .. }` literal,
     /// optionally behind `?`. Method names are never return-type evidence.
     /// Abstain rather than fabricate a receiver type.
diff --git a/crates/tracedecay-code-extraction/tests/main/rust.rs b/crates/tracedecay-code-extraction/tests/main/rust.rs
index 664d131ec8..bd9fc4cf10 100644
--- a/crates/tracedecay-code-extraction/tests/main/rust.rs
+++ b/crates/tracedecay-code-extraction/tests/main/rust.rs
@@ -1237,6 +1237,93 @@ fn make() -> Vec<i32> { Vec::new() }
     );
 }
 
+#[test]
+fn trait_bound_calls_name_the_trait_without_a_bare_method() {
+    let source = r#"
+trait Processor {
+    fn process(&self, input: u32) -> u32;
+    fn via_self(&self, input: u32) -> u32 {
+        self.process(input)
+    }
+}
+trait Other {
+    fn process(&self, input: u32) -> u32;
+}
+struct Doubler;
+impl Doubler {
+    fn kick(&self, input: u32) -> u32 {
+        self.process(input)
+    }
+}
+impl Processor for Doubler {
+    fn process(&self, input: u32) -> u32 {
+        input * 2
+    }
+}
+fn via_dyn(processor: &dyn Processor, input: u32) -> u32 {
+    processor.process(input)
+}
+fn via_impl(processor: impl Processor + 'static, input: u32) -> u32 {
+    processor.process(input)
+}
+fn via_bound<T: Processor>(processor: &T, input: u32) -> u32 {
+    processor.process(input)
+}
+fn via_where<T>(processor: &T, input: u32) -> u32
+where
+    T: Processor,
+{
+    processor.process(input)
+}
+fn ambiguous<T: Processor + Other>(processor: &T, input: u32) -> u32 {
+    processor.process(input)
+}
+"#;
+    let result = RustExtractor.extract("src/lib.rs", source);
+    assert!(result.errors.is_empty(), "{:?}", result.errors);
+    let from = |name: &str| {
+        let function = result
+            .nodes
+            .iter()
+            .find(|node| {
+                matches!(node.kind, NodeKind::Function | NodeKind::Method) && node.name == name
+            })
+            .unwrap_or_else(|| panic!("{name} is extracted"));
+        result
+            .unresolved_refs
+            .iter()
+            .filter(|reference| {
+                reference.reference_kind == EdgeKind::Calls && reference.from_node_id == function.id
+            })
+            .map(|reference| reference.reference_name.as_str())
+            .collect::<Vec<_>>()
+    };
+
+    for owner in ["via_self", "via_dyn", "via_impl", "via_bound", "via_where"] {
+        let names = from(owner);
+        assert!(
+            names.contains(&"Processor::process"),
+            "{owner} must name the trait callee: {names:?}"
+        );
+        assert!(
+            !names.contains(&"process"),
+            "{owner} must not reintroduce the bare method name: {names:?}"
+        );
+    }
+    let kick = from("kick");
+    assert!(
+        kick.contains(&"Doubler::process") && !kick.contains(&"Processor::process"),
+        "self in an inherent impl stays the type, not the trait: {kick:?}"
+    );
+    assert!(!kick.contains(&"process"), "{kick:?}");
+    let ambiguous = from("ambiguous");
+    assert!(
+        !ambiguous.iter().any(|name| name.contains("::process")),
+        "two trait bounds must not pick a callee: {ambiguous:?}"
+    );
+    assert!(!ambiguous.contains(&"process"), "{ambiguous:?}");
+}
+
 #[test]
 fn self_receiver_names_carry_module_scope_and_the_outer_as_delimiter() {
     let source = r#"
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
index 44e4d9f5b1..5f19b178a6 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
+++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
@@ -851,7 +851,7 @@ fn retained_stale_rust_extractor_generation_is_refused_and_rebuilt() {
             .iter()
             .find(|(language, _)| language.as_str() == "rust")
             .map(|(_, revision)| revision.as_str()),
-        Some("extractor.rust.v11")
+        Some("extractor.rust.v12")
     );
 }
 
diff --git a/crates/tracedecay-code-index/src/chunks.rs b/crates/tracedecay-code-index/src/chunks.rs
index 08e9be3684..15fd1f44b0 100644
--- a/crates/tracedecay-code-index/src/chunks.rs
+++ b/crates/tracedecay-code-index/src/chunks.rs
@@ -4458,6 +4458,71 @@ pub fn real_symbol() {}
         );
     }
 
+    #[test]
+    fn trait_bound_method_call_binds_the_trait_callee() {
+        let source = concat!(
+            "pub trait Processor {\n",
+            "    fn process(&self, input: u32) -> u32;\n",
+            "    fn via_self(&self, input: u32) -> u32 { self.process(input) }\n",
+            "}\n",
+            "pub trait Other { fn process(&self, input: u32) -> u32; }\n",
+            "pub struct Doubler;\n",
+            "impl Doubler { fn kick(&self, input: u32) -> u32 { self.process(input) } }\n",
+            "impl Processor for Doubler { fn process(&self, input: u32) -> u32 { input * 2 } }\n",
+            "pub fn via_dyn(processor: &dyn Processor, input: u32) -> u32 { processor.process(input) }\n",
+            "pub fn via_impl(processor: impl Processor + 'static, input: u32) -> u32 { processor.process(input) }\n",
+            "pub fn via_bound<T: Processor>(processor: &T, input: u32) -> u32 { processor.process(input) }\n",
+            "pub fn via_where<T>(processor: &T, input: u32) -> u32 where T: Processor { processor.process(input) }\n",
+            "pub fn ambiguous<T: Processor + Other>(processor: &T, input: u32) -> u32 { processor.process(input) }\n",
+        );
+        let file = validated_file("src/lib.rs", source.as_bytes());
+        let batch = batch_for(&file, ParseOutcomeV1::Complete);
+        let artifacts = chunker()
+            .index_file(&file, &batch, &rust_descriptor(), &NeverCancelled)
+            .expect("indexing succeeds");
+        let qualified = |occurrence: &SymbolOccurrenceId| {
+            artifacts
+                .symbols
+                .iter()
+                .find(|symbol| &symbol.occurrence == occurrence)
+                .map(|symbol| symbol.qualified_name.as_str())
+                .unwrap_or("<missing>")
+        };
+        let mut calls = artifacts
+            .edges
+            .iter()
+            .filter(|edge| edge.kind == RelationEdgeKindV1::Calls)
+            .map(|edge| {
+                (
+                    qualified(&edge.from_occurrence).to_owned(),
+                    qualified(&edge.to_occurrence).to_owned(),
+                )
+            })
+            .collect::<Vec<_>>();
+        calls.sort();
+
+        let trait_method = "src/lib.rs::Processor::process";
+        let impl_method = "src/lib.rs::<Doubler as Processor>::process";
+        assert_eq!(
+            calls,
+            vec![
+                (
+                    "src/lib.rs::Doubler::kick".to_owned(),
+                    impl_method.to_owned()
+                ),
+                (
+                    "src/lib.rs::Processor::via_self".to_owned(),
+                    trait_method.to_owned()
+                ),
+                ("src/lib.rs::via_bound".to_owned(), trait_method.to_owned()),
+                ("src/lib.rs::via_dyn".to_owned(), trait_method.to_owned()),
+                ("src/lib.rs::via_impl".to_owned(), trait_method.to_owned()),
+                ("src/lib.rs::via_where".to_owned(), trait_method.to_owned()),
+            ],
+            "a unique trait bound is the callee; two bounds and a bare method name are not: {calls:?}"
+        );
+    }
+
     #[test]
     fn self_call_inside_an_inline_module_binds_the_module_scoped_method() {
         let source = concat!(
diff --git a/crates/tracedecay-code-index/src/extract.rs b/crates/tracedecay-code-index/src/extract.rs
index cd4f39eb1a..3aedf93cb1 100644
--- a/crates/tracedecay-code-index/src/extract.rs
+++ b/crates/tracedecay-code-index/src/extract.rs
@@ -777,12 +777,14 @@ mod tests {
         // typed value and no longer fabricates receiver types for method
         // initializers; v9 adds the clone-body token bound and v10 the byte
         // bound; v11 drops the bare method name of a dotted call and types
-        // `self` from the enclosing impl or trait. The revision is part of the
-        // batch identity, so the pinned digest moves with it.
-        assert_eq!(descriptor.extractor_revision.as_str(), "extractor.rust.v11");
+        // `self` from the enclosing impl or trait; v12 binds a receiver typed
+        // by a type parameter with one trait bound to `Trait::method`. The
+        // revision is part of the batch identity, so the pinned digest moves
+        // with it.
+        assert_eq!(descriptor.extractor_revision.as_str(), "extractor.rust.v12");
         assert_eq!(
             extraction.batch().rows_digest.as_str(),
-            "sha256:e92b7ad8f93e3576c70adafb0690d064bd996c207a4ecd7b855c96e3ad3959ad"
+            "sha256:4e483806dfce308dfc97ec460c65f674b28f2dc8bad0e57c28a3b1519c8e495c"
         );
     }
 
diff --git a/crates/tracedecay-code-index/src/languages.rs b/crates/tracedecay-code-index/src/languages.rs
index 5d7ea779ed..fdcf1392e6 100644
--- a/crates/tracedecay-code-index/src/languages.rs
+++ b/crates/tracedecay-code-index/src/languages.rs
@@ -214,9 +214,11 @@ impl StaticLanguageRegistry {
             // stopped emitting the bare method name of a dotted call, so an
             // unrelated same-file callable sharing that name is no longer a
             // caller, and types `self` through the enclosing impl or trait.
-            // Only re-extraction removes the poisoned record.
+            // Rust v12 resolves a receiver whose type is a type parameter with
+            // one trait bound to `Trait::method`, so that call now has a
+            // callee. Only re-extraction removes the poisoned record.
             let extractor_revision = if language == "rust" {
-                11
+                12
             } else if matches!(language.as_str(), "typescript" | "protobuf" | "sql") {
                 6
             } else {
@@ -421,7 +423,7 @@ mod tests {
         assert!(rust.stable_member_spans);
         assert!(rust.capabilities.extraction);
         assert_eq!(rust.root_markers, vec!["Cargo.toml".to_owned()]);
-        assert_eq!(rust.extractor_revision.as_str(), "extractor.rust.v11");
+        assert_eq!(rust.extractor_revision.as_str(), "extractor.rust.v12");
 
         assert_eq!(
             registry
diff --git a/crates/tracedecay-code-index/src/production/worker_tests.rs b/crates/tracedecay-code-index/src/production/worker_tests.rs
index 7531f94588..ba58d2d4d2 100644
--- a/crates/tracedecay-code-index/src/production/worker_tests.rs
+++ b/crates/tracedecay-code-index/src/production/worker_tests.rs
@@ -357,7 +357,7 @@ fn extractor_revision_change_reextracts_before_validating_retained_import_rows()
 
     assert_eq!(
         rebuilt.files[0].extraction.extractor_revision.as_str(),
-        "extractor.rust.v11"
+        "extractor.rust.v12"
     );
     assert_ne!(
         rebuilt.files[0].extraction.parser_import_rows_digest,
@@ -425,7 +425,7 @@ fn physical_artifact_reuse_rejects_a_stale_extractor_revision() {
 
     assert_eq!(
         rebuilt.files[0].extraction.extractor_revision.as_str(),
-        "extractor.rust.v11"
+        "extractor.rust.v12"
     );
     assert!(
         rebuilt.files[0]
diff --git a/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs b/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs
index 6e7d5f7894..f4f4d73f23 100644
--- a/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs
+++ b/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs
@@ -3244,22 +3244,22 @@ fn partitioned_codec_fixture() -> (
 }
 
 const PARTITIONED_FORMAT_STATE_DIGEST: &str =
-    "sha256:4f78e1a1b0a4ea366f748e4699d4c28d9913782912809bbe2afafba4eac25268";
+    "sha256:89fa34dded4c696b44119ee9b658cea4d627fcef4b37b98360ca2298e36d1df3";
 const PARTITIONED_FORMAT_SEGMENTS: &[(&str, u64)] = &[
     (
-        "sha256:7cd13a44df02cc2dc2e13d5a867aaab5410594eaa398fe535ed59450ddc63b36",
+        "sha256:8caabc1a8e5b468ba6ccd6826e81efa2fa2627763e7217df64ea0f7313a383a6",
         11_071,
     ),
     (
-        "sha256:64b5c4d5c08f363d66c1dc3922fb61df7c8df0b022dab6b72507e61e6e1bf403",
+        "sha256:c44d7072ba9f13b9e5c734950f94bec9bb646ba4f5b9d6c691d43f822fd657c8",
         5_171,
     ),
     (
-        "sha256:782d1321bdc39aac9efcff85a44f8d48da66e24212d28150020cb04309d23880",
+        "sha256:7709e7304333ac9c4a94e0e02508029f09712072e1606a707854f42c2c1f9ce3",
         6_279,
     ),
     (
-        "sha256:3a307f49e46059b54a86dac6921287afbe0b25cadc581840b5143ce2bd5a04d1",
+        "sha256:2da2f3ca63181ae45a17db21cfeb68b8a42ae6e104820cd91a0b5c0673c5d249",
         6_837,
     ),
 ];
diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs
index 8148f6b9e1..571f4c6fbc 100644
--- a/crates/tracedecay-global-db/src/observation_adapter.rs
+++ b/crates/tracedecay-global-db/src/observation_adapter.rs
@@ -1526,6 +1526,17 @@ impl ObservationStore for GlobalDbObservationStore {
                 Ok(CursorAdvanceOutcome::Committed)
             }
             RuntimeSubmitOutcomeV1::ExactReplay { .. } => Ok(CursorAdvanceOutcome::ExactDuplicate),
+            // The idempotency key covers the advanced coverage, not the whole
+            // command, so a re-scan of already-admitted history reuses the key
+            // with different bytes (a fresh `expected_cursor` or resume
+            // checkpoint) and the writer reports a conflict against the earlier
+            // committed receipt. When the durable cursor is already exactly
+            // `next_cursor`, that earlier commit is this advance: the coverage
+            // is applied and the replay is a duplicate. Only a conflict that
+            // left the cursor somewhere else is an unresolved collision.
+            RuntimeSubmitOutcomeV1::IdempotencyConflict { .. } if existed_at_next => {
+                Ok(CursorAdvanceOutcome::ExactDuplicate)
+            }
             RuntimeSubmitOutcomeV1::IdempotencyConflict { .. } => {
                 Err(ObservationStoreError::CursorAdvanceCollision)
             }
diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs
index 31cd26d4db..4fd33d97cc 100644
--- a/crates/tracedecay-global-db/src/observation_collision_tests.rs
+++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs
@@ -55,9 +55,10 @@ use tracedecay_domain::{
 use tracedecay_store::observation::ObservationIdentityCollisionDispositionV1;
 use tracedecay_store::{
     AnchoredObservationWrite, CursorAdvanceLedgerReasonV1, CursorAdvanceLedgerReceiptIdV1,
-    ObservationCoverageReason, ObservationCursorAdvance, ObservationPersistOutcome,
-    ObservationProjectionStore, ObservationStore, ObservationStoreError, ObservationWrite,
-    ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION,
+    CursorAdvanceOutcome, ObservationCoverageReason, ObservationCursorAdvance,
+    ObservationPersistOutcome, ObservationProjectionStore, ObservationStore, ObservationStoreError,
+    ObservationWrite, ProjectionPersistOutcome, ProjectionSkipReason,
+    SESSION_MESSAGE_PROJECTOR_VERSION,
 };
 use tracing::field::{Field, Visit};
 use tracing::span::{Attributes, Id, Record};
@@ -3073,6 +3074,130 @@ async fn runtime_cursor_replay_without_a_ledger_row_keeps_generic_collision_sema
     ));
 }
 
+/// Overwrites the durable cursor for one source, the shape a retained-history
+/// rescan sees when it resumes behind the admitted frontier.
+async fn rewind_source_cursor(
+    runtime: &HostAdmissionTestRuntimeV1,
+    cursor: &ObservationSourceCursorV1,
+) {
+    let database = runtime
+        .registered_database(HostAdmissionScope::Profile)
+        .expect("registered profile database");
+    let transaction = database.begin_write_transaction().await.unwrap();
+    transaction
+        .execute(
+            COMMIT_SOURCE_CURSOR_SQL,
+            params![
+                serde_json::to_string(cursor.source()).unwrap().as_str(),
+                serde_json::to_string(cursor.scope()).unwrap().as_str(),
+                serde_json::to_string(cursor).unwrap().as_str()
+            ],
+        )
+        .await
+        .unwrap();
+    transaction.commit().await.unwrap();
+}
+
+/// A cursor advance is idempotency-keyed by its coverage, not by the whole
+/// command, so a retained-history rescan replays an already-admitted coverage
+/// with different command bytes: the `expected_cursor` it resumed from carries
+/// a freshly computed resume checkpoint. The writer answers with a conflict
+/// against the earlier receipt, and treating that as a permanent collision is
+/// what wedges retained ingest. The durable cursor already sits exactly at
+/// `next_cursor`, so the coverage is applied and the replay is a duplicate.
+/// A conflict that leaves the cursor somewhere else is still a collision.
+#[tokio::test]
+async fn already_positioned_cursor_replay_with_new_command_bytes_is_a_duplicate() {
+    const FILE_IDENTITY: u64 = 41;
+
+    let tmp = TempDir::new().unwrap();
+    let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path())
+        .await
+        .unwrap();
+    let store = runtime
+        .observation_store(HostAdmissionScope::Profile)
+        .unwrap();
+    let source = ObservationSourceIdentityV1::for_provider(
+        ProviderId::new(COLLISION_PROVIDER).unwrap(),
+        SessionId::new("session.cursor-replay-new-command-bytes").unwrap(),
+    )
+    .unwrap();
+    let generation = ObservationSourceGenerationV1::new(7).unwrap();
+    let cursor_at = |offset: u64, resume_fingerprint: u64| {
+        ObservationSourceCursorV1::new(
+            source.clone(),
+            ObservationScopeV1::Profile,
+            generation,
+            offset,
+        )
+        .unwrap()
+        .with_resume_checkpoint(FILE_IDENTITY, resume_fingerprint)
+    };
+    let advance_over = |expected: Option<ObservationSourceCursorV1>,
+                        covered: (u64, u64),
+                        resume_fingerprint: u64| {
+        ObservationCursorAdvance::new(
+            source.clone(),
+            ObservationScopeV1::Profile,
+            generation,
+            expected,
+            ObservationSourceRangeV1::new(covered.0, covered.1).unwrap(),
+            ObservationCoverageReason::BlankFrame,
+        )
+        .unwrap()
+        .with_resume_checkpoint(FILE_IDENTITY, resume_fingerprint)
+    };
+
+    assert_eq!(
+        store
+            .advance_source_cursor(advance_over(None, (0, 5), 11))
+            .await
+            .unwrap(),
+        CursorAdvanceOutcome::Committed
+    );
+    let admitted = advance_over(Some(cursor_at(5, 11)), (5, 10), 22);
+    assert_eq!(
+        store.advance_source_cursor(admitted.clone()).await.unwrap(),
+        CursorAdvanceOutcome::Committed
+    );
+
+    let replay = advance_over(Some(cursor_at(5, 33)), (5, 10), 22);
+    assert_eq!(
+        replay.coverage(),
+        admitted.coverage(),
+        "the replay must reuse the admitted coverage idempotency key"
+    );
+    assert_ne!(
+        replay, admitted,
+        "the replay must carry different command bytes"
+    );
+    assert_eq!(
+        store.advance_source_cursor(replay).await.unwrap(),
+        CursorAdvanceOutcome::ExactDuplicate,
+        "a coverage replay whose cursor is already at next must not wedge history"
+    );
+    assert_eq!(
+        store
+            .get_source_cursor(&source, &ObservationScopeV1::Profile)
+            .await
+            .unwrap(),
+        Some(cursor_at(10, 22)),
+        "a duplicate advance must leave the admitted frontier untouched"
+    );
+
+    rewind_source_cursor(&runtime, &cursor_at(5, 11)).await;
+    assert!(
+        matches!(
+            store
+                .advance_source_cursor(advance_over(Some(cursor_at(5, 11)), (5, 10), 44))
+                .await
+                .unwrap_err(),
+            ObservationStoreError::CursorAdvanceCollision
+        ),
+        "a conflicting replay that does not leave the cursor at next stays a collision"
+    );
+}
+
 #[tokio::test]
 async fn runtime_cursor_replay_preserves_storage_failure() {
     let tmp = TempDir::new().unwrap();
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs
index e988993e7c..632898ca24 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs
@@ -119,9 +119,11 @@ impl SessionTemporalRefreshProjector for CanonicalSessionTemporalProjector {
                 // Empty remaining range is a durable no-op: terminalize with an
                 // empty complete progress batch instead of deferring forever.
                 Ok(None) => canonical_noop_complete_effect(&recovery),
-                Err(error) if error.is_storage() => Err(
-                    SessionTemporalRefreshProjectorError::retryable("source_busy"),
-                ),
+                Err(error) if error.is_storage() => {
+                    Err(SessionTemporalRefreshProjectorError::retryable(format!(
+                        "source_busy: {error}"
+                    )))
+                }
                 Err(_) => Err(SessionTemporalRefreshProjectorError::terminal(
                     "projector_failed",
                 )),
diff --git a/crates/tracedecay-session-temporal-store/src/doctor_health.rs b/crates/tracedecay-session-temporal-store/src/doctor_health.rs
index 9dbe978c52..9bdf989170 100644
--- a/crates/tracedecay-session-temporal-store/src/doctor_health.rs
+++ b/crates/tracedecay-session-temporal-store/src/doctor_health.rs
@@ -22,7 +22,8 @@ const MAX_FINDING_COUNT: u64 = 1_000_000;
 const SQLITE_CORRUPT_VTAB: i32 = 267;
 const SESSION_TEMPORAL_HEALTH_CACHE_TTL: Duration = Duration::from_secs(2);
 const MAX_CACHED_SESSION_TEMPORAL_STORES: usize = 64;
-const MAX_SYNCHRONOUS_SESSION_TEMPORAL_HEALTH_BYTES: u64 = 64 * 1024 * 1024;
+const HEALTH_PROBE_PAGE_SIZE: i64 = 512;
+const HEALTH_PROBE_QUERY_LIMIT: i64 = HEALTH_PROBE_PAGE_SIZE + 1;
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 struct SessionTemporalStoreFileFingerprint {
@@ -119,77 +120,6 @@ fn session_temporal_store_fingerprint(
     })
 }
 
-fn session_temporal_store_family_bytes(database_path: &Path) -> std::io::Result<u64> {
-    hotpath::measure_block!("session_temporal.doctor.stat", {
-        let database = std::fs::metadata(database_path)?.len();
-        let mut wal_path = database_path.as_os_str().to_os_string();
-        wal_path.push("-wal");
-        match std::fs::metadata(PathBuf::from(wal_path)) {
-            Ok(wal) => database.checked_add(wal.len()).ok_or_else(|| {
-                std::io::Error::new(
-                    std::io::ErrorKind::InvalidData,
-                    "session temporal store size overflowed",
-                )
-            }),
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(database),
-            Err(error) => Err(error),
-        }
-    })
-}
-
-fn permits_synchronous_session_temporal_health(database_path: &Path) -> bool {
-    session_temporal_store_family_bytes(database_path)
-        .is_ok_and(|bytes| bytes <= MAX_SYNCHRONOUS_SESSION_TEMPORAL_HEALTH_BYTES)
-}
-
-// Occurrence and summary FTS integrity share the same content/docsize
-// EXCEPT + probe-token shape; only the table names differ.
-macro_rules! fts_integrity_check_sql {
-    ($content:literal, $fts:literal, $docsize:literal) => {
-        concat!(
-            "SELECT
-    (SELECT COUNT(*) FROM (
-        SELECT rowid AS id FROM ",
-            $content,
-            "
-        EXCEPT SELECT id FROM ",
-            $docsize,
-            "
-        LIMIT 1000001
-    ))
-    + (SELECT COUNT(*) FROM (
-        SELECT id FROM ",
-            $docsize,
-            "
-        EXCEPT SELECT rowid AS id FROM ",
-            $content,
-            "
-        LIMIT 1000001
-    ))
-    + COALESCE((
-        SELECT 0 FROM ",
-            $fts,
-            "
-        WHERE ",
-            $fts,
-            " MATCH 'tracedecay_health_probe_token'
-        LIMIT 1
-    ), 0)"
-        )
-    };
-}
-
-const OCCURRENCE_FTS_CHECK_SQL: &str = fts_integrity_check_sql!(
-    "session_occurrences",
-    "session_occurrences_fts",
-    "session_occurrences_fts_docsize"
-);
-const SUMMARY_FTS_CHECK_SQL: &str = fts_integrity_check_sql!(
-    "session_summary_nodes",
-    "session_summary_nodes_fts",
-    "session_summary_nodes_fts_docsize"
-);
-
 const REQUIRED_BASE_TABLES: &[&str] = &[
     "lcm_summary_nodes",
     "lcm_summary_sources",
@@ -305,226 +235,428 @@ const REQUIRED_TRIGGERS: &[(&str, &str)] = &[
     ),
 ];
 
+const INVALID_GENERATION_TAIL: &str = "WHERE candidate.generation <= 0
+    OR json_valid(candidate.frozen_watermarks_json) = 0
+    OR CASE WHEN json_valid(candidate.frozen_watermarks_json) = 1 THEN (
+         json_type(candidate.frozen_watermarks_json, '$.active_generation') IS NOT 'integer'
+         OR CAST(json_extract(
+             candidate.frozen_watermarks_json, '$.active_generation'
+         ) AS INTEGER) <= 0
+         OR CAST(json_extract(
+             candidate.frozen_watermarks_json, '$.active_generation'
+         ) AS INTEGER) > candidate.generation
+         OR json_type(candidate.frozen_watermarks_json, '$.source_frontier') IS NOT 'integer'
+         OR CAST(json_extract(
+             candidate.frozen_watermarks_json, '$.source_frontier'
+         ) AS INTEGER) < 0
+         OR json_type(candidate.frozen_watermarks_json, '$.projection_frontier') IS NOT 'integer'
+         OR CAST(json_extract(
+             candidate.frozen_watermarks_json, '$.projection_frontier'
+         ) AS INTEGER) < 0
+         OR json_type(candidate.frozen_watermarks_json, '$.summary_frontier') IS NOT 'integer'
+         OR CAST(json_extract(
+             candidate.frozen_watermarks_json, '$.summary_frontier'
+         ) AS INTEGER) < 0
+         OR NOT (
+              (candidate.state = 'building' AND candidate.ready_at IS NULL
+                   AND candidate.activated_at IS NULL AND candidate.completed_at IS NULL)
+           OR (candidate.state = 'ready' AND candidate.ready_at IS NOT NULL
+                   AND candidate.activated_at IS NULL AND candidate.completed_at IS NULL)
+           OR (candidate.state = 'active' AND candidate.ready_at IS NOT NULL
+                   AND candidate.activated_at IS NOT NULL AND candidate.completed_at IS NULL)
+           OR (candidate.state = 'superseded' AND candidate.ready_at IS NOT NULL
+                   AND candidate.activated_at IS NOT NULL
+                   AND candidate.completed_at IS NOT NULL)
+           OR (candidate.state IN ('failed', 'cancelled')
+                   AND candidate.completed_at IS NOT NULL)
+         )
+    ) ELSE 0 END";
+
+const MULTI_ACTIVE_GENERATION_TAIL: &str = "WHERE candidate.state = 'active'
+    AND NOT EXISTS (
+        SELECT 1
+        FROM session_temporal_generations AS earlier
+        WHERE earlier.session_id = candidate.session_id
+          AND earlier.state = 'active'
+          AND earlier.rowid < candidate.source_rowid
+    )
+    AND EXISTS (
+        SELECT 1
+        FROM session_temporal_generations AS later
+        WHERE later.session_id = candidate.session_id
+          AND later.state = 'active'
+          AND later.rowid > candidate.source_rowid
+    )";
+
+const CURSOR_KEY_ABSENT_TAIL: &str = "LEFT JOIN session_query_cursor_keys AS key
+    ON key.key_id = json_extract(candidate.frozen_watermarks_json, '$.cursor_key.key_id')
+   AND key.key_version = CAST(json_extract(
+          candidate.frozen_watermarks_json, '$.cursor_key.version'
+       ) AS INTEGER)
+   AND key.retired_at IS NULL
+  WHERE candidate.state = 'active'
+    AND (
+        json_type(candidate.frozen_watermarks_json, '$.cursor_key') IS NOT 'object'
+        OR key.key_id IS NULL
+    )";
+
+const STUCK_BINDING_TAIL: &str = "LEFT JOIN session_refresh_bindings AS binding
+    ON binding.session_id = candidate.session_id
+   AND binding.operation_id = candidate.operation_id
+  LEFT JOIN session_temporal_generations AS generation
+    ON generation.session_id = binding.session_id
+   AND generation.generation = binding.generation
+  WHERE candidate.state = 'running'
+    AND (
+        binding.operation_id IS NULL
+        OR generation.session_id IS NULL
+        OR generation.state <> 'building'
+    )";
+
+const STUCK_PROGRESS_SQL: &str = "WITH operation_source AS MATERIALIZED (
+        SELECT rowid AS source_rowid, session_id, operation_id, state, updated_at
+        FROM session_refresh_operations
+        ORDER BY rowid
+        LIMIT ?1
+    ),
+    operation_page AS MATERIALIZED (
+        SELECT * FROM operation_source ORDER BY source_rowid LIMIT ?2
+    ),
+    operation_progress AS MATERIALIZED (
+        SELECT operation.*,
+               (
+                   SELECT MAX(progress.recorded_at)
+                   FROM session_refresh_progress AS progress
+                   WHERE progress.session_id = operation.session_id
+                     AND progress.operation_id = operation.operation_id
+                     AND progress.progress_ordinal < ?2
+               ) AS latest_progress
+        FROM operation_page AS operation
+    )
+    SELECT
+      (SELECT COUNT(*)
+         FROM operation_progress AS operation
+         JOIN session_refresh_bindings AS binding
+           ON binding.session_id = operation.session_id
+          AND binding.operation_id = operation.operation_id
+         WHERE operation.state = 'running'
+           AND NOT EXISTS(
+               SELECT 1
+               FROM session_refresh_progress AS progress
+               WHERE progress.session_id = operation.session_id
+                 AND progress.operation_id = operation.operation_id
+                 AND progress.progress_ordinal >= ?2
+           )
+           AND (
+               (operation.latest_progress IS NULL
+                AND operation.updated_at
+                    < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000)
+               OR operation.latest_progress
+                    < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000
+           )),
+      EXISTS(SELECT 1 FROM operation_source LIMIT 1 OFFSET ?2)
+      OR EXISTS(
+          SELECT 1
+          FROM operation_page AS operation
+          WHERE EXISTS(
+              SELECT 1
+              FROM session_refresh_progress AS progress
+              WHERE progress.session_id = operation.session_id
+                AND progress.operation_id = operation.operation_id
+                AND progress.progress_ordinal >= ?2
+          )
+      )";
+
+const STUCK_RECEIPT_TAIL: &str = "LEFT JOIN session_refresh_receipts AS receipt
+    ON receipt.session_id = candidate.session_id
+   AND receipt.operation_id = candidate.operation_id
+  WHERE (candidate.state = 'running' AND receipt.operation_id IS NOT NULL)
+     OR (candidate.state <> 'running' AND receipt.operation_id IS NULL)
+     OR (receipt.operation_id IS NOT NULL
+         AND (
+             receipt.terminal_state <> candidate.state
+             OR receipt.terminal_at IS NOT candidate.terminal_at
+             OR receipt.failure_code IS NOT candidate.failure_code
+         ))";
+
+const COMPATIBILITY_DRIFT_TAIL: &str = "LEFT JOIN lcm_summary_nodes AS compatibility
+    ON compatibility.node_id = candidate.summary_id
+  WHERE compatibility.node_id IS NULL
+     OR candidate.publication_json IS NULL
+     OR json_extract(candidate.publication_json, '$.summary_hash') IS NULL
+     OR compatibility.session_id <> candidate.session_id
+     OR compatibility.summary_text <> candidate.summary_text
+     OR compatibility.summary_hash
+          <> json_extract(candidate.publication_json, '$.summary_hash')";
+
+macro_rules! row_health_check {
+    (
+        $kind:ident,
+        $tables:expr,
+        $source_table:literal,
+        $source_columns:literal,
+        $count:literal,
+        $tail:expr
+    ) => {
+        HealthCheck {
+            kind: SessionTemporalHealthFindingKind::$kind,
+            tables: $tables,
+            probe: HealthProbe::Rows {
+                source_table: $source_table,
+                source_columns: $source_columns,
+                count: $count,
+                tail: $tail,
+            },
+        }
+    };
+}
+
 const CHECKS: &[HealthCheck] = &[
+    row_health_check!(
+        OccurrenceFtsCorruption,
+        &["session_occurrences", "session_occurrences_fts_docsize"],
+        "session_occurrences",
+        "",
+        "COUNT(*)",
+        "LEFT JOIN session_occurrences_fts_docsize AS docsize
+           ON docsize.id = candidate.source_rowid
+         WHERE docsize.id IS NULL"
+    ),
+    row_health_check!(
+        OccurrenceFtsCorruption,
+        &["session_occurrences", "session_occurrences_fts_docsize"],
+        "session_occurrences_fts_docsize",
+        ", id",
+        "COUNT(*)",
+        "LEFT JOIN session_occurrences AS occurrence
+           ON occurrence.rowid = candidate.id
+         WHERE occurrence.rowid IS NULL"
+    ),
     HealthCheck {
         kind: SessionTemporalHealthFindingKind::OccurrenceFtsCorruption,
-        tables: &[
-            "session_occurrences",
-            "session_occurrences_fts",
-            "session_occurrences_fts_docsize",
-        ],
-        sql: OCCURRENCE_FTS_CHECK_SQL,
+        tables: &["session_occurrences_fts"],
+        probe: HealthProbe::Sql(
+            "SELECT COALESCE((
+                 SELECT 0 FROM session_occurrences_fts
+                 WHERE session_occurrences_fts MATCH 'tracedecay_health_probe_token'
+                 LIMIT 1
+             ), 0), (?1 - ?1) + (?2 - ?2)",
+        ),
     },
+    row_health_check!(
+        SummaryFtsCorruption,
+        &["session_summary_nodes", "session_summary_nodes_fts_docsize"],
+        "session_summary_nodes",
+        "",
+        "COUNT(*)",
+        "LEFT JOIN session_summary_nodes_fts_docsize AS docsize
+           ON docsize.id = candidate.source_rowid
+         WHERE docsize.id IS NULL"
+    ),
+    row_health_check!(
+        SummaryFtsCorruption,
+        &["session_summary_nodes", "session_summary_nodes_fts_docsize"],
+        "session_summary_nodes_fts_docsize",
+        ", id",
+        "COUNT(*)",
+        "LEFT JOIN session_summary_nodes AS summary
+           ON summary.rowid = candidate.id
+         WHERE summary.rowid IS NULL"
+    ),
     HealthCheck {
         kind: SessionTemporalHealthFindingKind::SummaryFtsCorruption,
-        tables: &[
-            "session_summary_nodes",
-            "session_summary_nodes_fts",
-            "session_summary_nodes_fts_docsize",
-        ],
-        sql: SUMMARY_FTS_CHECK_SQL,
+        tables: &["session_summary_nodes_fts"],
+        probe: HealthProbe::Sql(
+            "SELECT COALESCE((
+                 SELECT 0 FROM session_summary_nodes_fts
+                 WHERE session_summary_nodes_fts MATCH 'tracedecay_health_probe_token'
+                 LIMIT 1
+             ), 0), (?1 - ?1) + (?2 - ?2)",
+        ),
     },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::MissingAnchor,
-        tables: &[
-            "retrieval_anchors",
-            "session_assertions",
-            "session_occurrences",
-            "session_summary_nodes",
+    row_health_check!(
+        MissingAnchor,
+        &["retrieval_anchors", "session_summary_nodes"],
+        "session_summary_nodes",
+        ", summary_anchor_id",
+        "COUNT(*)",
+        "LEFT JOIN retrieval_anchors AS anchor
+           ON anchor.anchor_id = candidate.summary_anchor_id
+         WHERE anchor.anchor_id IS NULL"
+    ),
+    row_health_check!(
+        MissingAnchor,
+        &["retrieval_anchors", "session_occurrences"],
+        "session_occurrences",
+        ", retrieval_anchor_id",
+        "COUNT(*)",
+        "LEFT JOIN retrieval_anchors AS anchor
+           ON anchor.anchor_id = candidate.retrieval_anchor_id
+         WHERE anchor.anchor_id IS NULL"
+    ),
+    row_health_check!(
+        MissingAnchor,
+        &["retrieval_anchors", "session_assertions"],
+        "session_assertions",
+        ", subject_anchor_id, object_anchor_id",
+        "COUNT(*)",
+        "LEFT JOIN retrieval_anchors AS subject
+           ON subject.anchor_id = candidate.subject_anchor_id
+         LEFT JOIN retrieval_anchors AS object
+           ON object.anchor_id = candidate.object_anchor_id
+         WHERE subject.anchor_id IS NULL OR object.anchor_id IS NULL"
+    ),
+    row_health_check!(
+        MissingReceipt,
+        &[
+            "sanitization_receipts",
+            "session_external_payload_manifests"
         ],
-        sql: "SELECT
-            (SELECT COUNT(*) FROM session_summary_nodes AS node
-             LEFT JOIN retrieval_anchors AS anchor
-               ON anchor.anchor_id = node.summary_anchor_id
-             WHERE anchor.anchor_id IS NULL)
-            + (SELECT COUNT(*) FROM session_occurrences AS occurrence
-               LEFT JOIN retrieval_anchors AS anchor
-                 ON anchor.anchor_id = occurrence.retrieval_anchor_id
-               WHERE anchor.anchor_id IS NULL)
-            + (SELECT COUNT(*) FROM session_assertions AS assertion
-               LEFT JOIN retrieval_anchors AS subject
-                 ON subject.anchor_id = assertion.subject_anchor_id
-               LEFT JOIN retrieval_anchors AS object
-                 ON object.anchor_id = assertion.object_anchor_id
-               WHERE subject.anchor_id IS NULL OR object.anchor_id IS NULL)",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::MissingReceipt,
-        tables: &[
+        "session_external_payload_manifests",
+        ", receipt_id",
+        "COUNT(*)",
+        "LEFT JOIN sanitization_receipts AS receipt
+           ON receipt.receipt_id = candidate.receipt_id
+         WHERE receipt.receipt_id IS NULL"
+    ),
+    row_health_check!(
+        MissingReceipt,
+        &[
             "sanitization_receipts",
-            "session_external_payload_manifests",
-            "session_refresh_batch_bindings",
-            "session_summary_nodes",
-            "session_temporal_observation_effects",
-            "session_temporal_projection_receipts",
+            "session_temporal_observation_effects"
         ],
-        sql: "SELECT
-            (SELECT COUNT(*) FROM session_external_payload_manifests AS manifest
-             LEFT JOIN sanitization_receipts AS receipt
-               ON receipt.receipt_id = manifest.receipt_id
-             WHERE receipt.receipt_id IS NULL)
-            + (SELECT COUNT(*) FROM session_temporal_observation_effects AS effect
-               LEFT JOIN sanitization_receipts AS receipt
-                 ON receipt.receipt_id = effect.receipt_id
-               WHERE receipt.receipt_id IS NULL)
-            + (SELECT COUNT(*) FROM session_summary_nodes AS summary
-               LEFT JOIN sanitization_receipts AS receipt
-                 ON receipt.receipt_id = json_extract(summary.publication_json, '$.receipt_id')
-               WHERE summary.publication_json IS NULL OR receipt.receipt_id IS NULL)
-            + (SELECT COUNT(*) FROM session_refresh_batch_bindings AS binding
-               LEFT JOIN session_temporal_projection_receipts AS receipt
-                 ON receipt.session_id = binding.session_id
-                AND receipt.generation = binding.generation
-                AND receipt.batch_ordinal = binding.batch_ordinal
-               WHERE receipt.session_id IS NULL)",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::InvalidGeneration,
-        tables: &["session_temporal_generations"],
-        sql: "SELECT COUNT(*) FROM session_temporal_generations
-            WHERE generation <= 0
-               OR json_valid(frozen_watermarks_json) = 0
-               OR CASE WHEN json_valid(frozen_watermarks_json) = 1 THEN (
-                    json_type(frozen_watermarks_json, '$.active_generation') IS NOT 'integer'
-                    OR CAST(json_extract(
-                        frozen_watermarks_json, '$.active_generation'
-                    ) AS INTEGER) <= 0
-                    OR CAST(json_extract(
-                        frozen_watermarks_json, '$.active_generation'
-                    ) AS INTEGER) > generation
-                    OR json_type(
-                        frozen_watermarks_json, '$.source_frontier'
-                    ) IS NOT 'integer'
-                    OR CAST(json_extract(
-                        frozen_watermarks_json, '$.source_frontier'
-                    ) AS INTEGER) < 0
-                    OR json_type(
-                        frozen_watermarks_json, '$.projection_frontier'
-                    ) IS NOT 'integer'
-                    OR CAST(json_extract(
-                        frozen_watermarks_json, '$.projection_frontier'
-                    ) AS INTEGER) < 0
-                    OR json_type(
-                        frozen_watermarks_json, '$.summary_frontier'
-                    ) IS NOT 'integer'
-                    OR CAST(json_extract(
-                        frozen_watermarks_json, '$.summary_frontier'
-                    ) AS INTEGER) < 0
-                    OR NOT (
-                         (state = 'building' AND ready_at IS NULL
-                              AND activated_at IS NULL AND completed_at IS NULL)
-                      OR (state = 'ready' AND ready_at IS NOT NULL
-                              AND activated_at IS NULL AND completed_at IS NULL)
-                      OR (state = 'active' AND ready_at IS NOT NULL
-                              AND activated_at IS NOT NULL AND completed_at IS NULL)
-                      OR (state = 'superseded' AND ready_at IS NOT NULL
-                              AND activated_at IS NOT NULL AND completed_at IS NOT NULL)
-                      OR (state IN ('failed', 'cancelled') AND completed_at IS NOT NULL)
-                    )
-               ) ELSE 0 END",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::MultiActiveGeneration,
-        tables: &["session_temporal_generations"],
-        sql: "SELECT COUNT(*) FROM (
-                SELECT session_id
-                FROM session_temporal_generations
-                WHERE state = 'active'
-                GROUP BY session_id
-                HAVING COUNT(*) > 1
-            )",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::CursorChainAbsent,
-        tables: &["session_query_cursor_keys", "session_temporal_generations"],
-        sql: "SELECT
-            (SELECT COUNT(*) FROM session_query_cursor_keys AS key
-             WHERE key.key_version > 1
-               AND NOT EXISTS (
-                   SELECT 1 FROM session_query_cursor_keys AS predecessor
-                   WHERE predecessor.key_version = key.key_version - 1
-               ))
-            + (SELECT CASE
-                 WHEN EXISTS(
-                     SELECT 1 FROM session_temporal_generations
-                     WHERE state = 'active'
-                 ) AND (
-                     SELECT COUNT(*) FROM session_query_cursor_keys
-                     WHERE retired_at IS NULL
-                 ) <> 1
-                 THEN 1 ELSE 0 END)",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::CursorKeyAbsent,
-        tables: &["session_query_cursor_keys", "session_temporal_generations"],
-        sql: "SELECT COUNT(*)
-            FROM session_temporal_generations AS generation
-            LEFT JOIN session_query_cursor_keys AS key
-              ON key.key_id = json_extract(
-                    generation.frozen_watermarks_json, '$.cursor_key.key_id'
-                 )
-             AND key.key_version = CAST(json_extract(
-                    generation.frozen_watermarks_json, '$.cursor_key.version'
-                 ) AS INTEGER)
-             AND key.retired_at IS NULL
-            WHERE generation.state = 'active'
-              AND (
-                  json_type(generation.frozen_watermarks_json, '$.cursor_key') IS NOT 'object'
-                  OR key.key_id IS NULL
-              )",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::OwnershipDrift,
-        tables: &[
+        "session_temporal_observation_effects",
+        ", receipt_id",
+        "COUNT(*)",
+        "LEFT JOIN sanitization_receipts AS receipt
+           ON receipt.receipt_id = candidate.receipt_id
+         WHERE receipt.receipt_id IS NULL"
+    ),
+    row_health_check!(
+        MissingReceipt,
+        &["sanitization_receipts", "session_summary_nodes"],
+        "session_summary_nodes",
+        ", publication_json",
+        "COUNT(*)",
+        "LEFT JOIN sanitization_receipts AS receipt
+           ON receipt.receipt_id =
+              json_extract(candidate.publication_json, '$.receipt_id')
+         WHERE candidate.publication_json IS NULL OR receipt.receipt_id IS NULL"
+    ),
+    row_health_check!(
+        MissingReceipt,
+        &[
             "session_refresh_batch_bindings",
-            "session_refresh_bindings",
-            "session_summary_availability",
-            "session_summary_nodes",
+            "session_temporal_projection_receipts"
         ],
-        sql: "SELECT
-            (SELECT COUNT(*)
-               FROM session_summary_availability AS availability
-               LEFT JOIN session_summary_nodes AS summary
-                 ON summary.summary_id = availability.summary_id
-               WHERE summary.summary_id IS NULL
-                  OR availability.session_id IS NOT summary.session_id)
-            + (SELECT COUNT(*)
-               FROM session_refresh_batch_bindings AS batch
-               LEFT JOIN session_refresh_bindings AS binding
-                 ON binding.session_id = batch.session_id
-                AND binding.operation_id = batch.operation_id
-               WHERE binding.operation_id IS NULL
-                  OR batch.generation IS NOT binding.generation)",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::StuckRefresh,
-        tables: &["session_refresh_operations"],
-        sql: "SELECT COUNT(*) FROM session_refresh_operations
-            WHERE state = 'running'
-              AND updated_at < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::StuckBinding,
-        tables: &[
+        "session_refresh_batch_bindings",
+        ", session_id, generation, batch_ordinal",
+        "COUNT(*)",
+        "LEFT JOIN session_temporal_projection_receipts AS receipt
+           ON receipt.session_id = candidate.session_id
+          AND receipt.generation = candidate.generation
+          AND receipt.batch_ordinal = candidate.batch_ordinal
+         WHERE receipt.session_id IS NULL"
+    ),
+    row_health_check!(
+        InvalidGeneration,
+        &["session_temporal_generations"],
+        "session_temporal_generations",
+        ", generation, state, frozen_watermarks_json, ready_at, activated_at, completed_at",
+        "COUNT(*)",
+        INVALID_GENERATION_TAIL
+    ),
+    row_health_check!(
+        MultiActiveGeneration,
+        &["session_temporal_generations"],
+        "session_temporal_generations",
+        ", session_id, state",
+        "COUNT(*)",
+        MULTI_ACTIVE_GENERATION_TAIL
+    ),
+    row_health_check!(
+        CursorChainAbsent,
+        &["session_query_cursor_keys"],
+        "session_query_cursor_keys",
+        ", key_version",
+        "COUNT(*)",
+        "WHERE candidate.key_version > 1
+           AND NOT EXISTS (
+               SELECT 1
+               FROM session_query_cursor_keys AS predecessor
+               WHERE predecessor.key_version = candidate.key_version - 1
+           )"
+    ),
+    row_health_check!(
+        CursorChainAbsent,
+        &["session_query_cursor_keys", "session_temporal_generations"],
+        "session_temporal_generations",
+        ", state",
+        "CASE WHEN COUNT(*) > 0 THEN 1 ELSE 0 END",
+        "WHERE candidate.state = 'active'
+           AND (
+               SELECT COUNT(*)
+               FROM (
+                   SELECT 1
+                   FROM session_query_cursor_keys
+                   WHERE retired_at IS NULL
+                   LIMIT 2
+               )
+           ) <> 1"
+    ),
+    row_health_check!(
+        CursorKeyAbsent,
+        &["session_query_cursor_keys", "session_temporal_generations"],
+        "session_temporal_generations",
+        ", state, frozen_watermarks_json",
+        "COUNT(*)",
+        CURSOR_KEY_ABSENT_TAIL
+    ),
+    row_health_check!(
+        OwnershipDrift,
+        &["session_summary_availability", "session_summary_nodes"],
+        "session_summary_availability",
+        ", session_id, summary_id",
+        "COUNT(*)",
+        "LEFT JOIN session_summary_nodes AS summary
+           ON summary.summary_id = candidate.summary_id
+         WHERE summary.summary_id IS NULL
+            OR candidate.session_id IS NOT summary.session_id"
+    ),
+    row_health_check!(
+        OwnershipDrift,
+        &["session_refresh_batch_bindings", "session_refresh_bindings"],
+        "session_refresh_batch_bindings",
+        ", session_id, operation_id, generation",
+        "COUNT(*)",
+        "LEFT JOIN session_refresh_bindings AS binding
+           ON binding.session_id = candidate.session_id
+          AND binding.operation_id = candidate.operation_id
+         WHERE binding.operation_id IS NULL
+            OR candidate.generation IS NOT binding.generation"
+    ),
+    row_health_check!(
+        StuckRefresh,
+        &["session_refresh_operations"],
+        "session_refresh_operations",
+        ", state, updated_at",
+        "COUNT(*)",
+        "WHERE candidate.state = 'running'
+           AND candidate.updated_at
+               < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000"
+    ),
+    row_health_check!(
+        StuckBinding,
+        &[
             "session_refresh_bindings",
             "session_refresh_operations",
-            "session_temporal_generations",
+            "session_temporal_generations"
         ],
-        sql: "SELECT COUNT(*)
-            FROM session_refresh_operations AS operation
-            LEFT JOIN session_refresh_bindings AS binding
-              ON binding.session_id = operation.session_id
-             AND binding.operation_id = operation.operation_id
-            LEFT JOIN session_temporal_generations AS generation
-              ON generation.session_id = binding.session_id
-             AND generation.generation = binding.generation
-            WHERE operation.state = 'running'
-              AND (
-                  binding.operation_id IS NULL
-                  OR generation.session_id IS NULL
-                  OR generation.state <> 'building'
-              )",
-    },
+        "session_refresh_operations",
+        ", session_id, operation_id, state",
+        "COUNT(*)",
+        STUCK_BINDING_TAIL
+    ),
     HealthCheck {
         kind: SessionTemporalHealthFindingKind::StuckProgress,
         tables: &[
@@ -532,56 +664,24 @@ const CHECKS: &[HealthCheck] = &[
             "session_refresh_operations",
             "session_refresh_progress",
         ],
-        sql: "SELECT COUNT(*) FROM (
-                SELECT operation.session_id, operation.operation_id
-                FROM session_refresh_operations AS operation
-                JOIN session_refresh_bindings AS binding
-                  ON binding.session_id = operation.session_id
-                 AND binding.operation_id = operation.operation_id
-                LEFT JOIN session_refresh_progress AS progress
-                  ON progress.session_id = operation.session_id
-                 AND progress.operation_id = operation.operation_id
-                WHERE operation.state = 'running'
-                GROUP BY operation.session_id, operation.operation_id
-                HAVING (MAX(progress.recorded_at) IS NULL
-                        AND MAX(operation.updated_at)
-                            < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000)
-                    OR MAX(progress.recorded_at)
-                         < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000
-            )",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::StuckReceipt,
-        tables: &["session_refresh_operations", "session_refresh_receipts"],
-        sql: "SELECT COUNT(*)
-            FROM session_refresh_operations AS operation
-            LEFT JOIN session_refresh_receipts AS receipt
-              ON receipt.session_id = operation.session_id
-             AND receipt.operation_id = operation.operation_id
-            WHERE (operation.state = 'running' AND receipt.operation_id IS NOT NULL)
-               OR (operation.state <> 'running' AND receipt.operation_id IS NULL)
-               OR (receipt.operation_id IS NOT NULL
-                   AND (
-                       receipt.terminal_state <> operation.state
-                       OR receipt.terminal_at IS NOT operation.terminal_at
-                       OR receipt.failure_code IS NOT operation.failure_code
-                   ))",
-    },
-    HealthCheck {
-        kind: SessionTemporalHealthFindingKind::CompatibilityDrift,
-        tables: &["lcm_summary_nodes", "session_summary_nodes"],
-        sql: "SELECT COUNT(*)
-            FROM session_summary_nodes AS canonical
-            LEFT JOIN lcm_summary_nodes AS compatibility
-              ON compatibility.node_id = canonical.summary_id
-            WHERE compatibility.node_id IS NULL
-               OR canonical.publication_json IS NULL
-               OR json_extract(canonical.publication_json, '$.summary_hash') IS NULL
-               OR compatibility.session_id <> canonical.session_id
-               OR compatibility.summary_text <> canonical.summary_text
-               OR compatibility.summary_hash
-                    <> json_extract(canonical.publication_json, '$.summary_hash')",
+        probe: HealthProbe::Sql(STUCK_PROGRESS_SQL),
     },
+    row_health_check!(
+        StuckReceipt,
+        &["session_refresh_operations", "session_refresh_receipts"],
+        "session_refresh_operations",
+        ", session_id, operation_id, state, terminal_at, failure_code",
+        "COUNT(*)",
+        STUCK_RECEIPT_TAIL
+    ),
+    row_health_check!(
+        CompatibilityDrift,
+        &["lcm_summary_nodes", "session_summary_nodes"],
+        "session_summary_nodes",
+        ", summary_id, session_id, summary_text, publication_json",
+        "COUNT(*)",
+        COMPATIBILITY_DRIFT_TAIL
+    ),
 ];
 
 #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
@@ -640,9 +740,8 @@ impl SessionTemporalHealthFinding {
 pub struct SessionTemporalHealthReport {
     status: SessionTemporalHealthStatus,
     findings: Vec<SessionTemporalHealthFinding>,
-    /// Why diagnosis could not complete: a fixed machine reason for path-API
-    /// unavailability (for example `synchronous_diagnosis_size_budget_exceeded`)
-    /// or a `<probe>: <storage error>` detail naming the read that failed.
+    /// Why diagnosis could not complete: a fixed bounded-probe reason or a
+    /// `<probe>: <storage error>` detail naming the read that failed.
     /// Omitted when diagnosis ran to completion against an immutable snapshot.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     reason: Option<String>,
@@ -666,7 +765,17 @@ impl SessionTemporalHealthReport {
 struct HealthCheck {
     kind: SessionTemporalHealthFindingKind,
     tables: &'static [&'static str],
-    sql: &'static str,
+    probe: HealthProbe,
+}
+
+enum HealthProbe {
+    Rows {
+        source_table: &'static str,
+        source_columns: &'static str,
+        count: &'static str,
+        tail: &'static str,
+    },
+    Sql(&'static str),
 }
 
 impl<D: SessionTemporalRegisteredDb + Sync> SessionTemporalAccess<'_, D> {
@@ -677,12 +786,6 @@ impl<D: SessionTemporalRegisteredDb + Sync> SessionTemporalAccess<'_, D> {
     #[hotpath::measure(future = true, label = "session_temporal.doctor.query")]
     pub async fn session_temporal_doctor_health(&self) -> SessionTemporalHealthReport {
         let database_path = self.db_path();
-        if !permits_synchronous_session_temporal_health(database_path) {
-            return unavailable_report_with_reason(
-                SessionTemporalHealthStatus::Unavailable,
-                Some("synchronous_diagnosis_size_budget_exceeded"),
-            );
-        }
         let cache = session_temporal_health_cache_cell(database_path);
         let mut cached = cache.lock().await;
         let before = session_temporal_store_fingerprint(database_path).ok();
@@ -753,6 +856,7 @@ async fn diagnose_snapshot(
 
     let mut status = SessionTemporalHealthStatus::Complete;
     let mut findings = Vec::new();
+    let mut partial_reasons = BTreeSet::new();
     let missing_tables = required_table_names()
         .filter(|table| !inventory.tables.contains(*table))
         .count() as u64;
@@ -835,33 +939,61 @@ async fn diagnose_snapshot(
             continue;
         }
         record_session_doctor_check();
-        match snapshot_count(conn, check.sql).await {
-            Ok(0) => {}
-            Ok(value) => merge_finding(&mut findings, check.kind, value),
-            Err(error) if is_fts_finding(check.kind) && is_fts_virtual_table_corruption(&error) => {
-                merge_finding(&mut findings, check.kind, 1);
-            }
-            Err(error) if is_engine_locked(&error) => {
-                return SessionTemporalHealthReport {
-                    status: SessionTemporalHealthStatus::Locked,
-                    findings,
-                    reason: None,
-                };
-            }
-            Err(error) => {
-                return unavailable_report_with_detail(
-                    SessionTemporalHealthStatus::Unavailable,
-                    check_probe_name(check.kind),
-                    &error,
-                );
-            }
+        if diagnose_health_check(
+            conn,
+            check,
+            &mut status,
+            &mut findings,
+            &mut partial_reasons,
+        )
+        .await
+        {
+            return SessionTemporalHealthReport {
+                status: SessionTemporalHealthStatus::Locked,
+                findings,
+                reason: None,
+            };
         }
     }
     findings.sort_by_key(SessionTemporalHealthFinding::kind);
     SessionTemporalHealthReport {
         status,
         findings,
-        reason: None,
+        reason: (!partial_reasons.is_empty())
+            .then(|| partial_reasons.into_iter().collect::<Vec<_>>().join("; ")),
+    }
+}
+
+async fn diagnose_health_check(
+    conn: &impl crate::handle::SessionTemporalQuery,
+    check: &HealthCheck,
+    status: &mut SessionTemporalHealthStatus,
+    findings: &mut Vec<SessionTemporalHealthFinding>,
+    partial_reasons: &mut BTreeSet<String>,
+) -> bool {
+    let probe_name = check_probe_name(check.kind);
+    match snapshot_count(conn, check).await {
+        Ok(outcome) => {
+            if outcome.count > 0 || !outcome.complete {
+                merge_finding(findings, check.kind, outcome.count);
+            }
+            if !outcome.complete {
+                *status = SessionTemporalHealthStatus::Partial;
+                partial_reasons.insert(format!("{probe_name}: bounded_row_probe_incomplete"));
+            }
+            false
+        }
+        Err(error) if is_fts_finding(check.kind) && is_fts_virtual_table_corruption(&error) => {
+            merge_finding(findings, check.kind, 1);
+            false
+        }
+        Err(error) if is_engine_locked(&error) => true,
+        Err(error) => {
+            *status = SessionTemporalHealthStatus::Partial;
+            merge_finding(findings, check.kind, 0);
+            partial_reasons.insert(format!("{probe_name}: {error}"));
+            false
+        }
     }
 }
 
@@ -959,19 +1091,54 @@ async fn snapshot_schema_version(
 #[hotpath::measure(future = true, label = "session_temporal.doctor.query.count")]
 async fn snapshot_count(
     conn: &impl crate::handle::SessionTemporalQuery,
-    sql: &str,
-) -> tracedecay_runtime_core::db::engine::Result<u64> {
-    let mut rows = conn.query(sql, ()).await?;
-    let value = rows
-        .next()
-        .await?
-        .map(|row| row.get::<Option<i64>>(0))
-        .transpose()?
-        .flatten();
-    Ok(value
-        .and_then(|value| u64::try_from(value).ok())
-        .unwrap_or(0)
-        .min(MAX_FINDING_COUNT))
+    check: &HealthCheck,
+) -> tracedecay_runtime_core::db::engine::Result<HealthProbeOutcome> {
+    let sql = match &check.probe {
+        HealthProbe::Rows {
+            source_table,
+            source_columns,
+            count,
+            tail,
+        } => format!(
+            "WITH source AS MATERIALIZED (
+                 SELECT rowid AS source_rowid{source_columns}
+                 FROM {source_table}
+                 ORDER BY rowid
+                 LIMIT ?1
+             ),
+             page AS MATERIALIZED (
+                 SELECT * FROM source ORDER BY source_rowid LIMIT ?2
+             )
+             SELECT {count},
+                    EXISTS(SELECT 1 FROM source LIMIT 1 OFFSET ?2)
+             FROM page AS candidate
+             {tail}"
+        ),
+        HealthProbe::Sql(sql) => (*sql).to_owned(),
+    };
+    let mut rows = conn
+        .query(&sql, [HEALTH_PROBE_QUERY_LIMIT, HEALTH_PROBE_PAGE_SIZE])
+        .await?;
+    let Some(row) = rows.next().await? else {
+        return Ok(HealthProbeOutcome {
+            count: 0,
+            complete: true,
+        });
+    };
+    let value = row.get::<Option<i64>>(0)?;
+    let incomplete = row.get::<i64>(1)? != 0;
+    Ok(HealthProbeOutcome {
+        count: value
+            .and_then(|value| u64::try_from(value).ok())
+            .unwrap_or(0)
+            .min(MAX_FINDING_COUNT),
+        complete: !incomplete,
+    })
+}
+
+struct HealthProbeOutcome {
+    count: u64,
+    complete: bool,
 }
 
 fn finding(kind: SessionTemporalHealthFindingKind, count: u64) -> SessionTemporalHealthFinding {
@@ -1070,17 +1237,6 @@ fn unavailable_report_with_detail(
     }
 }
 
-fn unavailable_report_with_reason(
-    status: SessionTemporalHealthStatus,
-    reason: Option<&'static str>,
-) -> SessionTemporalHealthReport {
-    SessionTemporalHealthReport {
-        status,
-        findings: Vec::new(),
-        reason: reason.map(str::to_string),
-    }
-}
-
 #[inline(always)]
 fn record_session_doctor_cache_hit() {
     #[cfg(feature = "hotpath")]
@@ -1102,8 +1258,6 @@ fn record_session_doctor_check() {
 #[cfg(test)]
 mod cache_tests {
     use super::*;
-    use crate::handle::SessionTemporalAccess;
-    use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness;
 
     #[test]
     fn session_temporal_fingerprint_tracks_database_and_wal_changes() {
@@ -1121,20 +1275,114 @@ mod cache_tests {
         let expanded = session_temporal_store_fingerprint(&database).expect("expanded fingerprint");
         assert_ne!(with_wal, expanded);
     }
+}
 
-    #[test]
-    fn session_temporal_size_budget_includes_wal_bytes() {
+#[cfg(test)]
+mod probe_tests {
+    use super::*;
+    use crate::handle::SessionTemporalExec;
+    use tracedecay_runtime_core::db::engine::TestConnection;
+
+    #[tokio::test]
+    async fn row_probe_reports_observed_findings_when_its_page_is_incomplete() {
         let tmp = tempfile::TempDir::new().expect("tempdir");
-        let database = tmp.path().join("sessions.db");
-        std::fs::write(&database, b"database").expect("database");
-        assert!(permits_synchronous_session_temporal_health(&database));
+        let connection = TestConnection::open(&tmp.path().join("doctor-page.db"));
+        SessionTemporalExec::execute_batch(
+            &connection,
+            &format!(
+                "CREATE TABLE retrieval_anchors (anchor_id TEXT PRIMARY KEY);
+                 CREATE TABLE session_summary_nodes (summary_anchor_id TEXT NOT NULL);
+                 CREATE TABLE session_occurrences (retrieval_anchor_id TEXT NOT NULL);
+                 CREATE TABLE session_assertions (
+                     subject_anchor_id TEXT NOT NULL,
+                     object_anchor_id TEXT NOT NULL
+                 );
+                 WITH RECURSIVE sequence(value) AS (
+                     VALUES(0)
+                     UNION ALL
+                     SELECT value + 1 FROM sequence WHERE value < {}
+                 )
+                 INSERT INTO session_summary_nodes (summary_anchor_id)
+                 SELECT printf('missing-%d', value) FROM sequence;",
+                HEALTH_PROBE_PAGE_SIZE
+            ),
+        )
+        .await
+        .expect("seed oversized health probe");
+        let check = CHECKS
+            .iter()
+            .find(|check| check.kind == SessionTemporalHealthFindingKind::MissingAnchor)
+            .expect("missing-anchor check");
+
+        let mut status = SessionTemporalHealthStatus::Complete;
+        let mut findings = Vec::new();
+        let mut partial_reasons = BTreeSet::new();
+
+        assert!(
+            !diagnose_health_check(
+                &connection,
+                check,
+                &mut status,
+                &mut findings,
+                &mut partial_reasons,
+            )
+            .await
+        );
+        assert_eq!(status, SessionTemporalHealthStatus::Partial);
+        assert_eq!(
+            findings,
+            vec![finding(
+                SessionTemporalHealthFindingKind::MissingAnchor,
+                HEALTH_PROBE_PAGE_SIZE as u64,
+            )]
+        );
+        assert_eq!(
+            partial_reasons,
+            BTreeSet::from(["missing_anchor: bounded_row_probe_incomplete".to_owned()])
+        );
+    }
+}
 
-        let wal = tmp.path().join("sessions.db-wal");
-        std::fs::File::create(wal)
-            .expect("wal")
-            .set_len(MAX_SYNCHRONOUS_SESSION_TEMPORAL_HEALTH_BYTES)
-            .expect("wal size");
-        assert!(!permits_synchronous_session_temporal_health(&database));
+#[cfg(test)]
+mod registered_tests {
+    use super::*;
+    use crate::handle::{SessionTemporalAccess, SessionTemporalExec, SessionTemporalRegisteredDb};
+    use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness;
+
+    #[tokio::test]
+    async fn oversized_session_temporal_store_still_reports_schema_findings() {
+        const OLD_SYNCHRONOUS_HEALTH_BUDGET_BYTES: u64 = 64 * 1024 * 1024;
+        let harness =
+            RegisteredGlobalDbHarness::open_without_relation_graph("doctor-oversized-store").await;
+        let writer = harness.registered.writer_connection().expect("writer");
+        SessionTemporalExec::execute(
+            &writer,
+            "DROP INDEX idx_session_occurrences_generation_order",
+            (),
+        )
+        .await
+        .expect("drop required index");
+
+        let database = SessionTemporalRegisteredDb::db_path(&harness.registered);
+        std::fs::OpenOptions::new()
+            .write(true)
+            .open(database)
+            .expect("open session database")
+            .set_len(OLD_SYNCHRONOUS_HEALTH_BUDGET_BYTES + 4096)
+            .expect("grow session database past the former admission budget");
+
+        let report = SessionTemporalAccess::new(&harness.registered)
+            .session_temporal_doctor_health()
+            .await;
+
+        assert_eq!(report.status(), SessionTemporalHealthStatus::Partial);
+        assert_ne!(
+            report.reason(),
+            Some("synchronous_diagnosis_size_budget_exceeded")
+        );
+        assert!(report.findings().iter().any(|finding| {
+            finding.kind() == SessionTemporalHealthFindingKind::MigrationGap && finding.count() >= 1
+        }));
     }
 
     #[tokio::test]
diff --git a/crates/tracedecay-session-temporal-store/src/projection.rs b/crates/tracedecay-session-temporal-store/src/projection.rs
index c5ebf7ecc4..037e5cbed7 100644
--- a/crates/tracedecay-session-temporal-store/src/projection.rs
+++ b/crates/tracedecay-session-temporal-store/src/projection.rs
@@ -9,7 +9,7 @@ use tracedecay_store::{
 };
 use tracedecay_temporal_query::ports::ExecutionControl;
 
-use super::query::{PERSIST_OPERATION, storage, storage_message};
+use super::query::{PERSIST_OPERATION, storage};
 use super::refresh::{SessionRefreshRecoveryV1, SessionRefreshRestartStateV1};
 use super::relations::SessionRelationError;
 use crate::handle::{SessionTemporalAccess, SessionTemporalRegisteredDb, SessionTemporalWriteTxn};
@@ -36,8 +36,6 @@ pub(super) use receipts::validate_final_projection_receipt;
 
 const DISCOVER_REFRESH: &str = "discover session temporal refresh";
 const MATERIALIZE_REFRESH: &str = "materialize session temporal refresh";
-const MAX_BASELINE_RELATION_ITEMS: usize = 100_000;
-
 pub struct SessionTemporalRefreshDiscoveryPage {
     requests: Vec<SessionRefreshBeginOrJoinRequestV1>,
     active_scanned_through: Option<SessionId>,
@@ -279,52 +277,23 @@ impl<D: SessionTemporalRegisteredDb + Sync> SessionTemporalAccess<'_, D> {
                 let (scope, relation_store) = self
                     .session_relation_store()
                     .map_err(|error| storage(MATERIALIZE_REFRESH, error))?;
-                match relation_store.load_projection(
+                match relation_store.logical_copy_count(
                     &scope,
                     recovery.session_id(),
                     recovery.frozen_watermarks().active_generation().value(),
-                    MAX_BASELINE_RELATION_ITEMS,
-                    MAX_BASELINE_RELATION_ITEMS,
                     Arc::new(NeverCancelled),
                 ) {
-                    Ok(projection) => u64::try_from(projection.logical_copies.len())
-                        .map_err(|error| storage(MATERIALIZE_REFRESH, error))?,
+                    Ok(copies) => copies,
                     Err(SessionRelationError::NotFound) => {
-                        let mut rows = snapshot
-                            .query(
-                                "SELECT COUNT(*)
-                             FROM session_occurrences
-                             WHERE session_id = ?1 AND generation = ?2",
-                                params![
-                                    recovery.session_id().as_str(),
-                                    i64::try_from(
-                                        recovery.frozen_watermarks().active_generation().value()
-                                    )
-                                    .map_err(|error| storage(MATERIALIZE_REFRESH, error))?,
-                                ],
-                            )
-                            .await
-                            .map_err(|error| storage(MATERIALIZE_REFRESH, error))?;
-                        let retained: i64 = rows
-                            .next()
-                            .await
-                            .map_err(|error| storage(MATERIALIZE_REFRESH, error))?
-                            .ok_or_else(|| {
-                                storage_message(
-                                    MATERIALIZE_REFRESH,
-                                    "active projection count returned no row",
-                                )
-                            })?
-                            .get(0)
-                            .map_err(|error| storage(MATERIALIZE_REFRESH, error))?;
-                        if retained == 0 {
-                            0
-                        } else {
-                            return Err(storage_message(
-                                MATERIALIZE_REFRESH,
-                                "active native relation projection is unavailable",
-                            ));
-                        }
+                        // No native graph was applied for this generation. Reconstruct
+                        // the copy count from the sealed rows instead of retrying the
+                        // absence as a busy source.
+                        crate::relation_projection::count_canonical_logical_copies(
+                            &snapshot,
+                            recovery.session_id(),
+                            recovery.frozen_watermarks().active_generation(),
+                        )
+                        .await?
                     }
                     Err(error) => return Err(storage(MATERIALIZE_REFRESH, error)),
                 }
diff --git a/crates/tracedecay-session-temporal-store/src/projection/derived.rs b/crates/tracedecay-session-temporal-store/src/projection/derived.rs
index fbc84ea164..df18db3bf5 100644
--- a/crates/tracedecay-session-temporal-store/src/projection/derived.rs
+++ b/crates/tracedecay-session-temporal-store/src/projection/derived.rs
@@ -11,6 +11,10 @@ use tracedecay_temporal_query::ports::ExecutionControl;
 use super::super::query::{PERSIST_OPERATION, generation_i64, storage, storage_message};
 use super::super::rebuild::checkpoint_relation_rebuild_control;
 
+/// Keep one occurrence-ref read under the exact-SQL materialization ceiling
+/// (10_000 rows / 64 MiB). A terminal generation is larger than that ceiling.
+const OCCURRENCE_REF_PAGE_ROWS: i64 = 512;
+
 #[hotpath::measure(future = true, label = "session_temporal.projection.rebuild_derived")]
 pub(super) async fn rebuild_derived_evidence(
     conn: &impl crate::handle::SessionTemporalExec,
@@ -63,79 +67,138 @@ async fn load_occurrence_refs(
     generation: i64,
     control: &ExecutionControl,
 ) -> SessionStoreResult<Vec<DerivedEvidenceOccurrenceRefV1>> {
-    checkpoint_relation_rebuild_control(control)?;
-    let mut rows = conn
-        .query(
-            "SELECT occurrence.occurrence_id,
-                    occurrence.retrieval_anchor_id,
-                    occurrence.thread_id,
-                    occurrence.message_id,
-                    occurrence.knowledge_at,
-                    effect.observation_sequence,
-                    occurrence.projection_output_ordinal
-             FROM session_occurrences AS occurrence
-             JOIN session_temporal_observation_effects AS effect
-               ON effect.observation_id = occurrence.source_observation_id
-              AND effect.session_id = occurrence.session_id
-             WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2
-             ORDER BY effect.observation_sequence ASC,
-                      occurrence.projection_output_ordinal ASC,
-                      occurrence.occurrence_id ASC",
-            params![session_id.as_str(), generation],
-        )
-        .await
-        .map_err(|error| storage(PERSIST_OPERATION, error))?;
     let mut occurrences = Vec::new();
-    while let Some(row) = rows
-        .next()
-        .await
-        .map_err(|error| storage(PERSIST_OPERATION, error))?
-    {
+    let mut cursor: Option<(i64, i64, String)> = None;
+    loop {
         checkpoint_relation_rebuild_control(control)?;
-        let occurrence_id = row
-            .get::<String>(0)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        let retrieval_anchor_id = row
-            .get::<String>(1)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        let thread_id = row
-            .get::<Option<String>>(2)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        let message_id = row
-            .get::<Option<String>>(3)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        let knowledge_at = row
-            .get::<i64>(4)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        let observation_sequence = row
-            .get::<i64>(5)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        let projection_output_ordinal = row
-            .get::<i64>(6)
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        occurrences.push(DerivedEvidenceOccurrenceRefV1 {
-            occurrence_id: MessageOccurrenceIdV1::new(occurrence_id)
-                .map_err(|error| storage(PERSIST_OPERATION, error))?,
-            retrieval_anchor_id: RetrievalAnchorId::new(retrieval_anchor_id)
-                .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))?,
-            thread_id: thread_id
-                .map(|value| {
-                    ThreadId::new(value)
-                        .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))
-                })
-                .transpose()?,
-            message_id: message_id
-                .map(|value| {
-                    MessageId::new(value)
-                        .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))
-                })
-                .transpose()?,
-            knowledge_at: UtcMicros(knowledge_at),
-            observation_sequence: u64::try_from(observation_sequence)
+        let mut rows = match &cursor {
+            None => conn
+                .query(
+                    "SELECT occurrence.occurrence_id,
+                            occurrence.retrieval_anchor_id,
+                            occurrence.thread_id,
+                            occurrence.message_id,
+                            occurrence.knowledge_at,
+                            effect.observation_sequence,
+                            occurrence.projection_output_ordinal
+                     FROM session_occurrences AS occurrence
+                     JOIN session_temporal_observation_effects AS effect
+                       ON effect.observation_id = occurrence.source_observation_id
+                      AND effect.session_id = occurrence.session_id
+                     WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2
+                     ORDER BY effect.observation_sequence ASC,
+                              occurrence.projection_output_ordinal ASC,
+                              occurrence.occurrence_id ASC
+                     LIMIT ?3",
+                    params![session_id.as_str(), generation, OCCURRENCE_REF_PAGE_ROWS],
+                )
+                .await
                 .map_err(|error| storage(PERSIST_OPERATION, error))?,
-            projection_output_ordinal: u32::try_from(projection_output_ordinal)
+            Some((sequence, ordinal, occurrence_id)) => conn
+                .query(
+                    "SELECT occurrence.occurrence_id,
+                            occurrence.retrieval_anchor_id,
+                            occurrence.thread_id,
+                            occurrence.message_id,
+                            occurrence.knowledge_at,
+                            effect.observation_sequence,
+                            occurrence.projection_output_ordinal
+                     FROM session_occurrences AS occurrence
+                     JOIN session_temporal_observation_effects AS effect
+                       ON effect.observation_id = occurrence.source_observation_id
+                      AND effect.session_id = occurrence.session_id
+                     WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2
+                       AND (
+                            effect.observation_sequence > ?3
+                            OR (
+                                effect.observation_sequence = ?3
+                                AND occurrence.projection_output_ordinal > ?4
+                            )
+                            OR (
+                                effect.observation_sequence = ?3
+                                AND occurrence.projection_output_ordinal = ?4
+                                AND occurrence.occurrence_id > ?5
+                            )
+                       )
+                     ORDER BY effect.observation_sequence ASC,
+                              occurrence.projection_output_ordinal ASC,
+                              occurrence.occurrence_id ASC
+                     LIMIT ?6",
+                    params![
+                        session_id.as_str(),
+                        generation,
+                        *sequence,
+                        *ordinal,
+                        occurrence_id.as_str(),
+                        OCCURRENCE_REF_PAGE_ROWS
+                    ],
+                )
+                .await
                 .map_err(|error| storage(PERSIST_OPERATION, error))?,
-        });
+        };
+        let mut page_rows = 0_i64;
+        let mut page_cursor = None;
+        while let Some(row) = rows
+            .next()
+            .await
+            .map_err(|error| storage(PERSIST_OPERATION, error))?
+        {
+            checkpoint_relation_rebuild_control(control)?;
+            let occurrence_id = row
+                .get::<String>(0)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            let retrieval_anchor_id = row
+                .get::<String>(1)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            let thread_id = row
+                .get::<Option<String>>(2)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            let message_id = row
+                .get::<Option<String>>(3)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            let knowledge_at = row
+                .get::<i64>(4)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            let observation_sequence = row
+                .get::<i64>(5)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            let projection_output_ordinal = row
+                .get::<i64>(6)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            page_cursor = Some((
+                observation_sequence,
+                projection_output_ordinal,
+                occurrence_id.clone(),
+            ));
+            page_rows += 1;
+            occurrences.push(DerivedEvidenceOccurrenceRefV1 {
+                occurrence_id: MessageOccurrenceIdV1::new(occurrence_id)
+                    .map_err(|error| storage(PERSIST_OPERATION, error))?,
+                retrieval_anchor_id: RetrievalAnchorId::new(retrieval_anchor_id)
+                    .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))?,
+                thread_id: thread_id
+                    .map(|value| {
+                        ThreadId::new(value)
+                            .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))
+                    })
+                    .transpose()?,
+                message_id: message_id
+                    .map(|value| {
+                        MessageId::new(value)
+                            .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))
+                    })
+                    .transpose()?,
+                knowledge_at: UtcMicros(knowledge_at),
+                observation_sequence: u64::try_from(observation_sequence)
+                    .map_err(|error| storage(PERSIST_OPERATION, error))?,
+                projection_output_ordinal: u32::try_from(projection_output_ordinal)
+                    .map_err(|error| storage(PERSIST_OPERATION, error))?,
+            });
+        }
+        if page_rows < OCCURRENCE_REF_PAGE_ROWS {
+            break;
+        }
+        cursor = page_cursor;
     }
     Ok(occurrences)
 }
@@ -260,3 +323,67 @@ async fn ensure_derived_anchor(
     .map_err(|error| storage(PERSIST_OPERATION, error))?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tracedecay_runtime_core::db::engine::Executor;
+
+    #[tokio::test]
+    async fn occurrence_refs_span_more_than_one_exact_sql_page() {
+        let dir = tempfile::TempDir::new().expect("occurrence dir");
+        let conn = tracedecay_runtime_core::db::engine::TestConnection::open(
+            &dir.path().join("occurrences.db"),
+        );
+        Executor::execute_batch(
+            &conn,
+            "CREATE TABLE session_occurrences (
+                session_id TEXT, generation INTEGER, occurrence_id TEXT,
+                retrieval_anchor_id TEXT, thread_id TEXT, message_id TEXT,
+                knowledge_at INTEGER, projection_output_ordinal INTEGER,
+                source_observation_id TEXT
+             );
+             CREATE TABLE session_temporal_observation_effects (
+                observation_id TEXT, session_id TEXT, observation_sequence INTEGER
+             );",
+        )
+        .await
+        .expect("schema");
+        let total = OCCURRENCE_REF_PAGE_ROWS + 3;
+        for index in 0..total {
+            let occurrence_id = format!("sha256:{index:064x}");
+            let observation_id = format!("obs-{index}");
+            Executor::execute(
+                &conn,
+                "INSERT INTO session_occurrences (
+                    session_id, generation, occurrence_id, retrieval_anchor_id,
+                    knowledge_at, projection_output_ordinal, source_observation_id
+                 ) VALUES ('page-session', 1, ?1, 'anchor.page', ?2, 0, ?3)",
+                params![occurrence_id, index, observation_id.clone()],
+            )
+            .await
+            .expect("occurrence");
+            Executor::execute(
+                &conn,
+                "INSERT INTO session_temporal_observation_effects (
+                    observation_id, session_id, observation_sequence
+                 ) VALUES (?1, 'page-session', ?2)",
+                params![observation_id, index],
+            )
+            .await
+            .expect("effect");
+        }
+
+        let session_id = SessionId::new("page-session").expect("session");
+        let refs = load_occurrence_refs(&conn, &session_id, 1, &ExecutionControl::default())
+            .await
+            .expect("paged refs");
+
+        assert_eq!(refs.len(), usize::try_from(total).expect("total"));
+        assert_eq!(refs[0].observation_sequence, 0);
+        assert_eq!(
+            refs.last().expect("last").observation_sequence,
+            u64::try_from(total - 1).expect("last sequence")
+        );
+    }
+}
diff --git a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs
index d73162675e..07f34b3533 100644
--- a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs
+++ b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs
@@ -22,6 +22,11 @@ use super::super::relations::{LogicalCopyRelation, SessionRelationProjection};
 use super::persist::*;
 
 const MAX_RECEIPT_COPY_ENTITIES: usize = 100_000;
+/// One coverage digest page stays under the exact-SQL materialization ceiling.
+/// `snippet_text` and `index_text` are the row bulk, so this is row-count-small
+/// on purpose. OFFSET is O(n²) row visits on a terminal batch; keyset on the
+/// ORDER BY column is the upgrade if a generation makes this the hot path.
+const COVERAGE_DIGEST_PAGE_ROWS: i64 = 32;
 
 #[hotpath::measure(future = true, label = "session_temporal.projection.validate_receipt")]
 pub async fn validate_final_projection_receipt(
@@ -642,40 +647,69 @@ pub(super) async fn digest_query_rows(
     batch: &SessionTemporalProjectionBatchV1,
     control: Option<&ExecutionControl>,
 ) -> SessionStoreResult<(usize, String)> {
-    if let Some(control) = control {
-        checkpoint_relation_rebuild_control(control)?;
-    }
-    record_coverage_query_probe();
-    let mut rows = conn
-        .query(
-            sql,
-            params![
-                batch.session_id().as_str(),
-                generation_i64(batch.generation(), PERSIST_OPERATION)?,
-            ],
-        )
-        .await
-        .map_err(|error| storage(PERSIST_OPERATION, error))?;
+    digest_paged_rows(
+        conn,
+        sql,
+        batch.session_id().as_str(),
+        generation_i64(batch.generation(), PERSIST_OPERATION)?,
+        control,
+    )
+    .await
+}
+
+async fn digest_paged_rows(
+    conn: &impl crate::handle::SessionTemporalExec,
+    sql: &str,
+    session_id: &str,
+    generation: i64,
+    control: Option<&ExecutionControl>,
+) -> SessionStoreResult<(usize, String)> {
+    // SQLite drops ORDER BY inside a subquery unless that subquery has LIMIT.
+    // The inner LIMIT -1 keeps digest order; the outer page is what exact SQL
+    // materializes.
+    let paged = format!("SELECT * FROM ({sql} LIMIT -1) LIMIT ?3 OFFSET ?4");
+    let mut offset = 0_i64;
     let mut digest = Sha256::new();
     let mut count = 0usize;
-    while let Some(row) = rows
-        .next()
-        .await
-        .map_err(|error| storage(PERSIST_OPERATION, error))?
-    {
+    loop {
         if let Some(control) = control {
             checkpoint_relation_rebuild_control(control)?;
         }
-        let value = row
-            .get::<String>(0)
+        record_coverage_query_probe();
+        let mut rows = conn
+            .query(
+                &paged,
+                params![session_id, generation, COVERAGE_DIGEST_PAGE_ROWS, offset],
+            )
+            .await
             .map_err(|error| storage(PERSIST_OPERATION, error))?;
-        record_coverage_row(
-            u64::try_from(value.len()).map_err(|error| storage(PERSIST_OPERATION, error))?,
-        );
-        update_ordered_row_digest(&mut digest, count, value.as_bytes());
-        count = count
-            .checked_add(1)
-            .ok_or_else(|| storage_message(PERSIST_OPERATION, "coverage row count overflow"))?;
+        let mut page_rows = 0_i64;
+        while let Some(row) = rows
+            .next()
+            .await
+            .map_err(|error| storage(PERSIST_OPERATION, error))?
+        {
+            if let Some(control) = control {
+                checkpoint_relation_rebuild_control(control)?;
+            }
+            let value = row
+                .get::<String>(0)
+                .map_err(|error| storage(PERSIST_OPERATION, error))?;
+            record_coverage_row(
+                u64::try_from(value.len()).map_err(|error| storage(PERSIST_OPERATION, error))?,
+            );
+            update_ordered_row_digest(&mut digest, count, value.as_bytes());
+            count = count
+                .checked_add(1)
+                .ok_or_else(|| storage_message(PERSIST_OPERATION, "coverage row count overflow"))?;
+            page_rows += 1;
+        }
+        if page_rows < COVERAGE_DIGEST_PAGE_ROWS {
+            break;
+        }
+        offset = offset.checked_add(page_rows).ok_or_else(|| {
+            storage_message(PERSIST_OPERATION, "coverage digest page offset overflow")
+        })?;
     }
     Ok((
         count,
@@ -1029,4 +1063,51 @@ mod digest_tests {
             digest_bytes(b"alpha\n\nomega")
         );
     }
+
+    #[tokio::test]
+    async fn coverage_digest_reads_every_ordered_row_across_pages() {
+        let dir = tempfile::TempDir::new().expect("digest dir");
+        let conn = tracedecay_runtime_core::db::engine::TestConnection::open(
+            &dir.path().join("digest.db"),
+        );
+        tracedecay_runtime_core::db::engine::Executor::execute_batch(
+            &conn,
+            "CREATE TABLE coverage_rows(session_id TEXT, generation INTEGER, encoded TEXT)",
+        )
+        .await
+        .expect("create coverage rows");
+        let total = COVERAGE_DIGEST_PAGE_ROWS + 8;
+        let mut expected = Sha256::new();
+        for index in 0..total {
+            let encoded = format!("row-{index:04}");
+            update_ordered_row_digest(
+                &mut expected,
+                usize::try_from(index).expect("index"),
+                encoded.as_bytes(),
+            );
+            tracedecay_runtime_core::db::engine::Executor::execute(
+                &conn,
+                "INSERT INTO coverage_rows(session_id, generation, encoded) VALUES ('session', 7, ?1)",
+                params![encoded],
+            )
+            .await
+            .expect("insert coverage row");
+        }
+
+        let (count, digest) = digest_paged_rows(
+            &conn,
+            "SELECT encoded FROM coverage_rows WHERE session_id = ?1 AND generation = ?2 ORDER BY encoded",
+            "session",
+            7,
+            None,
+        )
+        .await
+        .expect("paged digest");
+
+        assert_eq!(count, usize::try_from(total).expect("total"));
+        assert_eq!(
+            digest,
+            encode_tagged_lowercase_hex("sha256:", &expected.finalize())
+        );
+    }
 }
diff --git a/crates/tracedecay-session-temporal-store/src/projection/tests.rs b/crates/tracedecay-session-temporal-store/src/projection/tests.rs
index 8ea0fa39c2..595a63953d 100644
--- a/crates/tracedecay-session-temporal-store/src/projection/tests.rs
+++ b/crates/tracedecay-session-temporal-store/src/projection/tests.rs
@@ -1582,6 +1582,17 @@ async fn explicit_copy_survives_reconstruction_in_the_native_relation_graph() {
             Arc::new(NeverCancelled),
         )
         .unwrap();
+    assert_eq!(
+        relation_store
+            .logical_copy_count(
+                &scope,
+                &session_id,
+                batch.generation().value(),
+                Arc::new(NeverCancelled),
+            )
+            .expect("paged logical copy count"),
+        loaded.logical_copies.len() as u64
+    );
     assert_eq!(
         loaded.logical_copies,
         vec![crate::relations::LogicalCopyRelation {
diff --git a/crates/tracedecay-session-temporal-store/src/query.rs b/crates/tracedecay-session-temporal-store/src/query.rs
index 8dd5a56501..ae0ab6b4de 100644
--- a/crates/tracedecay-session-temporal-store/src/query.rs
+++ b/crates/tracedecay-session-temporal-store/src/query.rs
@@ -185,9 +185,10 @@ pub(super) async fn read_observation(
     Ok((sequence, observation))
 }
 
-/// The largest `observation_id IN (...)` batch one observation prefetch binds,
-/// kept clear of `SQLite`'s default variable ceiling.
-const OBSERVATION_READ_BATCH: usize = 500;
+/// First prefetch width. A batch of full `observation_json` bodies is bounded
+/// by the exact-SQL materialization ceiling (64 MiB), not by SQLite's variable
+/// limit, so this starts small and splits when a page still does not fit.
+const OBSERVATION_READ_BATCH: usize = 32;
 
 /// Prefetches the observations a projection pass is about to decode.
 ///
@@ -211,7 +212,20 @@ pub(super) async fn read_observations(
     if unique.is_empty() {
         return Ok(observations);
     }
-    for chunk in unique.chunks(OBSERVATION_READ_BATCH) {
+    let mut pending = Vec::new();
+    let mut offset = 0usize;
+    while offset < unique.len() {
+        let end = offset
+            .saturating_add(OBSERVATION_READ_BATCH)
+            .min(unique.len());
+        pending.push((offset, end));
+        offset = end;
+    }
+    while let Some((start, end)) = pending.pop() {
+        if start >= end {
+            continue;
+        }
+        let chunk = &unique[start..end];
         let placeholders = (1..=chunk.len())
             .map(|index| format!("?{index}"))
             .collect::<Vec<_>>()
@@ -221,10 +235,26 @@ pub(super) async fn read_observations(
              FROM observations
              WHERE observation_id IN ({placeholders})"
         );
-        let mut rows = conn
+        let mut rows = match conn
             .query(&sql, params_from_iter(chunk.iter().copied()))
             .await
-            .map_err(|error| storage(PERSIST_OPERATION, error))?;
+        {
+            Ok(rows) => rows,
+            Err(error) => {
+                let error = storage(PERSIST_OPERATION, error);
+                // One materialized page of bodies exceeded the exact-SQL
+                // ceiling. Split until a single observation remains; that
+                // observation is then a typed storage failure, not a retry
+                // that looks like a busy source.
+                if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 {
+                    let mid = start + chunk.len() / 2;
+                    pending.push((mid, end));
+                    pending.push((start, mid));
+                    continue;
+                }
+                return Err(error);
+            }
+        };
         while let Some(row) = rows
             .next()
             .await
@@ -249,6 +279,15 @@ pub(super) async fn read_observations(
     Ok(observations)
 }
 
+fn observation_prefetch_exceeded_materialization_limit(error: &SessionStoreError) -> bool {
+    match error {
+        SessionStoreError::Storage { source, .. } => source
+            .to_string()
+            .contains("materialization exceeded its limit"),
+        _ => false,
+    }
+}
+
 /// The error `read_observation` raises for an id the store does not hold, reused
 /// by callers that resolve prefetched observations out of a batch map.
 pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError {
@@ -257,3 +296,26 @@ pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) ->
         format!("source observation {} is missing", observation_id.as_str()),
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage};
+
+    #[test]
+    fn materialization_limit_is_the_prefetch_split_signal() {
+        let exceeded = storage(
+            PERSIST_OPERATION,
+            std::io::Error::other("exact SQL query materialization exceeded its limit"),
+        );
+        let locked = storage(
+            PERSIST_OPERATION,
+            std::io::Error::other("database is locked"),
+        );
+        assert!(observation_prefetch_exceeded_materialization_limit(
+            &exceeded
+        ));
+        assert!(!observation_prefetch_exceeded_materialization_limit(
+            &locked
+        ));
+    }
+}
diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs
index d881616136..520304afb8 100644
--- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs
+++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs
@@ -533,6 +533,139 @@ pub(crate) async fn reconstruct_session_relation_projection(
     Ok(projection)
 }
 
+/// Counts logical copies for a generation whose native graph was never applied.
+///
+/// The refresh baseline used to refuse that state. The graph comment says the
+/// caller must reconstruct instead. This walks occurrences in precedence order
+/// and keeps only the latest predecessor per message, so a large generation is
+/// not one exact-SQL page of `observation_json`.
+pub(crate) async fn count_canonical_logical_copies(
+    conn: &impl crate::handle::SessionTemporalQuery,
+    session_id: &SessionId,
+    generation: SessionProjectionGenerationV1,
+) -> SessionStoreResult<u64> {
+    const PAGE: i64 = 8;
+    let generation = generation_i64(generation, RECONSTRUCT_OPERATION)?;
+    let mut cursor: Option<(i64, i64, String)> = None;
+    let mut predecessors: BTreeMap<String, (UtcMicros, u32, String)> = BTreeMap::new();
+    let mut copies = 0_u64;
+    loop {
+        let mut rows = match &cursor {
+            None => conn
+                .query(
+                    "SELECT occurrence.occurrence_id, occurrence.message_id,
+                            occurrence.projection_output_ordinal, occurrence.knowledge_at,
+                            observation.observation_json
+                     FROM session_occurrences AS occurrence
+                     JOIN observations AS observation
+                       ON observation.observation_id = occurrence.source_observation_id
+                     WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2
+                     ORDER BY occurrence.knowledge_at, occurrence.projection_output_ordinal,
+                              occurrence.occurrence_id
+                     LIMIT ?3",
+                    params![session_id.as_str(), generation, PAGE],
+                )
+                .await
+                .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?,
+            Some((knowledge_at, ordinal, occurrence_id)) => conn
+                .query(
+                    "SELECT occurrence.occurrence_id, occurrence.message_id,
+                            occurrence.projection_output_ordinal, occurrence.knowledge_at,
+                            observation.observation_json
+                     FROM session_occurrences AS occurrence
+                     JOIN observations AS observation
+                       ON observation.observation_id = occurrence.source_observation_id
+                     WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2
+                       AND (
+                            occurrence.knowledge_at > ?3
+                            OR (
+                                occurrence.knowledge_at = ?3
+                                AND occurrence.projection_output_ordinal > ?4
+                            )
+                            OR (
+                                occurrence.knowledge_at = ?3
+                                AND occurrence.projection_output_ordinal = ?4
+                                AND occurrence.occurrence_id > ?5
+                            )
+                       )
+                     ORDER BY occurrence.knowledge_at, occurrence.projection_output_ordinal,
+                              occurrence.occurrence_id
+                     LIMIT ?6",
+                    params![
+                        session_id.as_str(),
+                        generation,
+                        *knowledge_at,
+                        *ordinal,
+                        occurrence_id.as_str(),
+                        PAGE
+                    ],
+                )
+                .await
+                .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?,
+        };
+        let mut page_rows = 0_i64;
+        let mut page_cursor = None;
+        while let Some(row) = rows
+            .next()
+            .await
+            .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?
+        {
+            let occurrence_id: String = row
+                .get(0)
+                .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?;
+            let message_id: Option<String> = row
+                .get(1)
+                .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?;
+            let ordinal = u32::try_from(
+                row.get::<i64>(2)
+                    .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?,
+            )
+            .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?;
+            let knowledge_at = UtcMicros(
+                row.get(3)
+                    .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?,
+            );
+            let observation: DurableObservationV1 = serde_json::from_str(
+                &row.get::<String>(4)
+                    .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?,
+            )
+            .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?;
+            let parent_message_id = observation_envelope_from_payload(observation.payload())
+                .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?
+                .relations()
+                .parent_message_id()
+                .map(|id| id.as_str().to_owned());
+            page_cursor = Some((knowledge_at.0, i64::from(ordinal), occurrence_id.clone()));
+            page_rows += 1;
+            if let (Some(message_id), Some(parent_message_id)) = (&message_id, &parent_message_id)
+                && message_id == parent_message_id
+                && predecessors.get(parent_message_id).is_some_and(|source| {
+                    (source.0, source.1, source.2.as_str())
+                        < (knowledge_at, ordinal, occurrence_id.as_str())
+                })
+            {
+                copies = copies.saturating_add(1);
+            }
+            if let Some(message_id) = message_id {
+                let candidate = (knowledge_at, ordinal, occurrence_id);
+                match predecessors.get(&message_id) {
+                    Some(existing)
+                        if (existing.0, existing.1, existing.2.as_str())
+                            >= (candidate.0, candidate.1, candidate.2.as_str()) => {}
+                    _ => {
+                        predecessors.insert(message_id, candidate);
+                    }
+                }
+            }
+        }
+        if page_rows < PAGE {
+            break;
+        }
+        cursor = page_cursor;
+    }
+    Ok(copies)
+}
+
 pub(crate) async fn reconstruct_logical_copy_relations(
     conn: &impl crate::handle::SessionTemporalQuery,
     session_id: &SessionId,
diff --git a/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs b/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs
index 0cd64ebb53..d9f435fa6a 100644
--- a/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs
+++ b/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs
@@ -70,6 +70,63 @@ impl SessionRelationGraphStore {
         }
         decode_projection(scope, session_id, generation, page.entities, page.relations)
     }
+
+    /// Counts logical-copy relations without loading the generation.
+    ///
+    /// A single `load_projection` asks for up to 100_000 relations. That read
+    /// is one exact-SQL page, whose ceiling is 10_000 rows, so a large parent
+    /// generation fails and refresh retries it as storage busy. The successor
+    /// baseline only needs the count.
+    pub fn logical_copy_count(
+        &self,
+        scope: &super::SessionRelationScope,
+        session_id: &SessionId,
+        generation: u64,
+        cancellation: Arc<dyn GraphCancellation>,
+    ) -> Result<u64, SessionRelationError> {
+        // Stay under the exact-SQL materialization ceiling (10_000 rows).
+        const RELATION_PAGE: usize = 4_096;
+        let namespace = namespace(scope)?;
+        let projection_id = projection(session_id, generation)?;
+        if self
+            .database
+            .projection_telemetry(GraphProjectionTelemetryRequest {
+                namespace: namespace.clone(),
+                projection: projection_id.clone(),
+                cancellation: Arc::clone(&cancellation),
+            })
+            .map_err(map_graph_error)?
+            .is_none()
+        {
+            return Err(SessionRelationError::NotFound);
+        }
+        let mut after_relation = None;
+        let mut copies = 0_u64;
+        loop {
+            let page = self
+                .database
+                .read_projection(GraphProjectionReadRequest {
+                    namespace: namespace.clone(),
+                    projection: projection_id.clone(),
+                    after_entity: None,
+                    after_relation,
+                    max_entities: 1,
+                    max_relations: RELATION_PAGE,
+                    cancellation: Arc::clone(&cancellation),
+                })
+                .map_err(map_graph_error)?;
+            for relation in &page.relations {
+                if relation.kind.as_str() == LOGICAL_COPY_KIND {
+                    copies = copies.saturating_add(1);
+                }
+            }
+            match page.next_relation {
+                Some(next) => after_relation = Some(next),
+                None => break,
+            }
+        }
+        Ok(copies)
+    }
 }
 
 fn decode_projection(
diff --git a/crates/tracedecay-sessions/src/admission/mod.rs b/crates/tracedecay-sessions/src/admission/mod.rs
index dc51b835a1..092d05457c 100644
--- a/crates/tracedecay-sessions/src/admission/mod.rs
+++ b/crates/tracedecay-sessions/src/admission/mod.rs
@@ -924,6 +924,7 @@ pub(crate) mod test_support {
         projection_failure: Arc<Mutex<Option<(HostAdmissionOutcome, ObservationCancellation)>>>,
         cancel_on_discovery_queue_read: Arc<Mutex<Option<ObservationCancellation>>>,
         session_backfill_page_pause: Arc<Mutex<Option<SessionBackfillPagePause>>>,
+        deterministic_capture_refusal: Arc<Mutex<Option<&'static str>>>,
     }
 
     impl MemoryHostAdmission {
@@ -940,6 +941,16 @@ pub(crate) mod test_support {
             self.store.state().capture_failures_remaining = 1;
         }
 
+        /// Refuse every capture the way a deterministic content refusal does:
+        /// the same record fails identically on every retry, so callers must
+        /// converge past it rather than re-attempt the source forever.
+        pub(crate) fn refuse_captures_deterministically(&self, reason: &'static str) {
+            *self
+                .deterministic_capture_refusal
+                .lock()
+                .unwrap_or_else(|error| error.into_inner()) = Some(reason);
+        }
+
         /// Make the next `count` session-message lookups report the store as
         /// unavailable, the way reader-pool saturation does.
         pub(crate) fn fail_next_session_message_lookups(&self, count: usize) {
@@ -1042,6 +1053,13 @@ pub(crate) mod test_support {
             request: CaptureObservationRequest,
         ) -> AdmissionFuture<'a, CaptureObservationOutcome> {
             Box::pin(async move {
+                if let Some(reason) = *self
+                    .deterministic_capture_refusal
+                    .lock()
+                    .unwrap_or_else(|error| error.into_inner())
+                {
+                    return Err(HostAdmissionOutcome::deterministic_content_refusal(reason));
+                }
                 {
                     let mut state = self.store.state();
                     state.scalar_capture_calls = state.scalar_capture_calls.saturating_add(1);
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs
index 67439f9621..b663217469 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs
@@ -269,17 +269,23 @@ impl Default for CodexReplayIndex {
 }
 
 #[cfg(test)]
-static CODEX_REPLAY_INDEX_ENTRIES_VISITED: std::sync::atomic::AtomicU64 =
-    std::sync::atomic::AtomicU64::new(0);
+thread_local! {
+    /// Per-thread, because `indexed_replay_pass` runs entirely on its caller's
+    /// thread and the one test that measures B-tree traversal shares the
+    /// process with every other test replaying an index in parallel. A global
+    /// counter measures the whole suite's traversal, not this pass's.
+    static CODEX_REPLAY_INDEX_ENTRIES_VISITED: std::cell::Cell<u64> =
+        const { std::cell::Cell::new(0) };
+}
 
 #[cfg(test)]
 fn reset_replay_index_entries_visited_for_test() {
-    CODEX_REPLAY_INDEX_ENTRIES_VISITED.store(0, std::sync::atomic::Ordering::Release);
+    CODEX_REPLAY_INDEX_ENTRIES_VISITED.with(|visited| visited.set(0));
 }
 
 #[cfg(test)]
 fn replay_index_entries_visited_for_test() -> u64 {
-    CODEX_REPLAY_INDEX_ENTRIES_VISITED.load(std::sync::atomic::Ordering::Acquire)
+    CODEX_REPLAY_INDEX_ENTRIES_VISITED.with(std::cell::Cell::get)
 }
 
 fn indexed_replay_pass(
@@ -295,7 +301,8 @@ fn indexed_replay_pass(
     let lower = position.map_or(Bound::Unbounded, Bound::Excluded);
     for indexed in index.paths.range((lower, Bound::Unbounded)) {
         #[cfg(test)]
-        CODEX_REPLAY_INDEX_ENTRIES_VISITED.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+        CODEX_REPLAY_INDEX_ENTRIES_VISITED
+            .with(|visited| visited.set(visited.get().saturating_add(1)));
         let path_bytes =
             u64::try_from(crate::runtime::source::path_byte_len(&indexed.path)).unwrap_or(u64::MAX);
         if paths.len() >= bounds.max_files.max(1)
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs
index 117e0d70ca..8982f1a1e2 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs
@@ -15,6 +15,7 @@ use tracedecay_store::observation::{ObservationCoverageReason, ObservationCursor
 
 use crate::admission::{HostAdmission, HostAdmissionOutcome};
 use crate::observation::{CaptureObservationOutcome, ObservationCancellation};
+use crate::runtime::jsonl_observation_admission::is_deterministic_content_refusal;
 use crate::runtime::shared::TranscriptIngestStats;
 use tracedecay_runtime_core::db::{SqliteFileIdentityOperation, sqlite_generation_identity};
 
@@ -89,8 +90,29 @@ async fn advance_coverage(
         .map_err(host_admission_error)
 }
 
+/// The admission's own verdict, verbatim.
+///
+/// The status alone names a family ("degraded"), not a cause: every
+/// deterministic refusal, cursor mismatch and contract violation collapsed
+/// into one indistinguishable sentence, so a sweep that skipped the same
+/// `state.db` every five seconds forever gave an operator nothing to act on.
+/// Carry the reason code, retryability and storage cause the outcome already
+/// holds.
 fn host_admission_error(outcome: HostAdmissionOutcome) -> String {
-    crate::runtime::snapshot_observation::host_admission_status_message("Hermes", outcome.status)
+    let mut message = crate::runtime::snapshot_observation::host_admission_status_message(
+        "Hermes",
+        outcome.status,
+    );
+    if let Some(reason) = outcome.reason_code {
+        message.push_str(&format!(
+            " (reason_code={reason}, retryable={})",
+            outcome.retryable
+        ));
+    }
+    if let Some(cause) = outcome.storage_cause {
+        message.push_str(&format!(": {cause}"));
+    }
+    message
 }
 
 pub(super) async fn drain_hermes_projections_with_admission(
@@ -221,11 +243,44 @@ pub(super) async fn admit_rows_with_admission_and_cancellation(
                 .await?;
             }
             HermesAdmissionAction::Capture(request) => {
-                match facade
-                    .capture_observation(*request)
-                    .await
-                    .map_err(host_admission_error)?
-                {
+                let captured = match facade.capture_observation(*request).await {
+                    Ok(captured) => captured,
+                    // A deterministic content refusal re-fails identically on
+                    // every pass. Without a durable skip the source's cursor
+                    // never clears the offending row, so the whole `state.db`
+                    // is abandoned every sweep, forever, with one WARN each
+                    // time. Cover past it with a typed reason exactly as the
+                    // shared JSONL path does so the stream converges.
+                    Err(outcome) if is_deterministic_content_refusal(&outcome) => {
+                        tracing::warn!(
+                            provider = PROVIDER,
+                            row = row.id,
+                            reason = outcome.reason_code.unwrap_or("host_admission_refused"),
+                            "admission refused a Hermes row; covering past it"
+                        );
+                        advance_coverage(
+                            facade,
+                            source,
+                            range,
+                            expected_cursor,
+                            scope.clone(),
+                            generation,
+                            if outcome.reason_code == Some("observation_identity_collision") {
+                                ObservationCoverageReason::ObservationIdentityCollision
+                            } else {
+                                ObservationCoverageReason::AdmissionRefused
+                            },
+                            None,
+                            file_identity,
+                            resume_fingerprint,
+                            cancellation,
+                        )
+                        .await?;
+                        continue;
+                    }
+                    Err(outcome) => return Err(host_admission_error(outcome)),
+                };
+                match captured {
                     CaptureObservationOutcome::Persisted { outcome, .. }
                     | CaptureObservationOutcome::AcceptedForReplay { outcome, .. } => {
                         if matches!(*outcome, ObservationPersistOutcome::Committed(_)) {
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs
index 88476f6bf4..3d2ca7642f 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs
@@ -26,6 +26,38 @@ fn new_sweep_budget(max_new_bytes: Option<u64>) -> IngestByteBudget {
     IngestByteBudget::bounded(max_new_bytes.unwrap_or(DEFAULT_HERMES_SWEEP_BYTES))
 }
 
+/// Whether this sweep pass should report a source outcome, given what the last
+/// pass reported for the same `state.db`.
+///
+/// A source that cannot be admitted stays unadmittable until something about
+/// the store or the file changes, and the sweep runs every few seconds. Logging
+/// the identical line each pass buried every other daemon warning without
+/// telling an operator anything the first line did not. Report a failure when
+/// it is new or its reason changed; `None` records a recovered source so its
+/// next failure is reported again. The map is keyed by discovered Hermes
+/// profile, so it is bounded by the number of profiles on disk.
+fn hermes_source_outcome_is_new(state_db: &Path, error: Option<&str>) -> bool {
+    use std::collections::BTreeMap;
+    use std::sync::{LazyLock, Mutex, PoisonError};
+
+    static REPORTED: LazyLock<Mutex<BTreeMap<PathBuf, String>>> =
+        LazyLock::new(|| Mutex::new(BTreeMap::new()));
+    let mut reported = REPORTED.lock().unwrap_or_else(PoisonError::into_inner);
+    match error {
+        Some(error) => {
+            if reported.get(state_db).is_some_and(|last| last == error) {
+                return false;
+            }
+            reported.insert(state_db.to_path_buf(), error.to_owned());
+            true
+        }
+        None => {
+            reported.remove(state_db);
+            false
+        }
+    }
+}
+
 /// Default Hermes profile homes under the resolved user home.
 ///
 /// Missing home is a typed absence (`None`), never an empty successful sweep.
@@ -283,14 +315,19 @@ pub(super) async fn ingest_homes_capped_with_admission_and_cancellation(
         )
         .await
         {
-            Ok(source_stats) => outcome.stats = outcome.stats.merge(source_stats),
+            Ok(source_stats) => {
+                hermes_source_outcome_is_new(&source.state_db, None);
+                outcome.stats = outcome.stats.merge(source_stats);
+            }
             Err(error) => {
                 outcome.source_failures = outcome.source_failures.saturating_add(1);
-                tracing::warn!(
-                    state_db = %source.state_db.display(),
-                    error,
-                    "skipping Hermes transcript source"
-                );
+                if hermes_source_outcome_is_new(&source.state_db, Some(&error)) {
+                    tracing::warn!(
+                        state_db = %source.state_db.display(),
+                        error,
+                        "skipping Hermes transcript source"
+                    );
+                }
             }
         }
     }
@@ -403,14 +440,19 @@ async fn ingest_user_homes_capped_with_admission(
         )
         .await
         {
-            Ok(source_stats) => outcome.stats = outcome.stats.merge(source_stats),
+            Ok(source_stats) => {
+                hermes_source_outcome_is_new(&source.state_db, None);
+                outcome.stats = outcome.stats.merge(source_stats);
+            }
             Err(error) => {
                 outcome.source_failures = outcome.source_failures.saturating_add(1);
-                tracing::warn!(
-                    state_db = %source.state_db.display(),
-                    error,
-                    "skipping projectless Hermes transcript source"
-                );
+                if hermes_source_outcome_is_new(&source.state_db, Some(&error)) {
+                    tracing::warn!(
+                        state_db = %source.state_db.display(),
+                        error,
+                        "skipping projectless Hermes transcript source"
+                    );
+                }
             }
         }
     }
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs
index 7d5f30d72c..d69d457c28 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs
@@ -1688,3 +1688,59 @@ async fn unreadable_state_db_is_a_counted_source_failure_not_a_clean_sweep() {
     );
     assert!(admission.observations().is_empty());
 }
+
+/// A deterministic admission refusal is permanent: the same row fails the same
+/// way on every sweep. Without a durable skip the source cursor never clears
+/// it, so the whole profile `state.db` is abandoned every pass forever, which
+/// is what produced an endless "skipping projectless Hermes transcript source"
+/// WARN on a live daemon. Cover past it, exactly as the shared JSONL path
+/// does, so the source converges.
+mod deterministic_refusal_recovery {
+    use super::*;
+
+    async fn admit_one_refused_row(reason: &'static str) -> MemoryHostAdmission {
+        let admission = MemoryHostAdmission::default();
+        admission.refuse_captures_deterministically(reason);
+        let stats = admit_rows_with_admission_and_cancellation(
+            &admission,
+            &[fixture(1)],
+            ObservationScopeV1::Profile,
+            ObservationSourceGenerationV1::new(1).unwrap(),
+            1,
+            1,
+            |_| Some(fixture_projection()),
+            &ObservationCancellation::default(),
+        )
+        .await
+        .expect("a permanently refused row must not abandon the whole source");
+        assert_eq!(stats.messages_upserted, 0);
+        admission
+    }
+
+    #[tokio::test]
+    async fn refused_row_is_covered_past_instead_of_skipping_the_source() {
+        let admission = admit_one_refused_row("privacy_boundary_failed").await;
+
+        let advances = admission.non_durable_advances();
+        assert_eq!(
+            advances.len(),
+            1,
+            "the refused row must be covered exactly once"
+        );
+        assert_eq!(
+            advances[0].reason(),
+            ObservationCoverageReason::AdmissionRefused
+        );
+        assert_eq!(advances[0].next_cursor().position(), 1);
+    }
+
+    #[tokio::test]
+    async fn identity_collision_keeps_its_own_coverage_reason() {
+        let admission = admit_one_refused_row("observation_identity_collision").await;
+
+        assert_eq!(
+            admission.non_durable_advances()[0].reason(),
+            ObservationCoverageReason::ObservationIdentityCollision
+        );
+    }
+}
diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs
index 468022fd84..2f40a44a2f 100644
--- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs
+++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs
@@ -483,10 +483,18 @@ pub(in crate::runtime) fn install_test_shared_jsonl_preparation_authority() {
     use std::num::NonZeroUsize;
     use tracedecay_runtime_core::resident_memory::ProcessResidentMemoryV1;
 
+    // One process-wide budget serves the whole suite, so every test thread
+    // holding a `SHARED_JSONL_WORKER_RESERVATION_BYTES` page charges it at
+    // once. At 32 GiB a wide harness drove the derived preparation capacity
+    // down to two entries, which is the shared metadata cache's degraded mode,
+    // not the product's: a production process meters one ingest workload
+    // against the machine. Size the budget past what the harness's own
+    // parallelism can reserve so capacity stays CPU-bound, the way the
+    // composition root installs it.
     static MEMORY: OnceLock<Arc<ProcessResidentMemoryV1>> = OnceLock::new();
     let memory = Arc::clone(MEMORY.get_or_init(|| {
         Arc::new(ProcessResidentMemoryV1::new(
-            NonZeroU64::new(32 * 1024 * 1024 * 1024).unwrap(),
+            NonZeroU64::new(1024 * 1024 * 1024 * 1024).unwrap(),
         ))
     }));
     let background_cpu = Arc::new(ProcessBackgroundCpuV1::new(NonZeroUsize::new(48).unwrap()));
@@ -1849,10 +1857,19 @@ impl ActiveAdmission<'_> {
         .with_resume_checkpoint(self.file_identity, checkpoint.resume_fingerprint);
         hotpath::gauge!("jsonl_admission_coverage_frames").inc(1.0);
         hotpath::gauge!("jsonl_admission_writer_submits").inc(1.0);
-        self.admission
+        if let Err(outcome) = self
+            .admission
             .advance_non_durable_source_cursor(advance, self.cancellation.clone())
             .await
-            .map_err(|outcome| {
+        {
+            if is_lost_cursor_cas(&outcome)
+                && self
+                    .peer_already_covered(expected_cursor, checkpoint.end_offset)
+                    .await
+            {
+                return Ok(());
+            }
+            return Err({
                 if is_admission_cancellation(&outcome, &self.cancellation) {
                     TranscriptIngestError::Cancelled {
                         provider: self.provider,
@@ -1874,12 +1891,47 @@ impl ActiveAdmission<'_> {
                             .unwrap_or("non_durable_cursor_advance_failed"),
                     }
                 }
-            })?;
+            });
+        }
         *expected_cursor =
             Some(self.cursor_at(checkpoint.end_offset, checkpoint.resume_fingerprint)?);
         Ok(())
     }
 
+    /// Whether the peer that won a cursor CAS already covered this range.
+    ///
+    /// Live hook ingest and the catch-up sweep own the same `(source, scope)`
+    /// cursor and routinely read the same transcript at once; the store's
+    /// compare-and-swap is what keeps them honest, so one of them loses. The
+    /// loser's frames are almost always already durable behind the winner's
+    /// cursor, and re-reading that cursor is enough to prove it. Adopt the
+    /// winner's cursor and let the pass continue instead of failing the whole
+    /// source over work that is already committed.
+    ///
+    /// A read failure, a different generation, or a cursor short of this frame
+    /// all answer "not covered", which keeps the caller's typed block.
+    #[hotpath::skip]
+    async fn peer_already_covered(
+        &self,
+        expected_cursor: &mut Option<ObservationSourceCursorV1>,
+        end_offset: u64,
+    ) -> bool {
+        let Ok(actual) = self
+            .admission
+            .get_source_cursor(&self.source, &self.scope)
+            .await
+        else {
+            return false;
+        };
+        let covered = actual.as_ref().is_some_and(|cursor| {
+            cursor.generation() == self.generation && cursor.position() >= end_offset
+        });
+        if covered {
+            *expected_cursor = actual;
+        }
+        covered
+    }
+
     fn capture_request(
         &self,
         expected_cursor: Option<ObservationSourceCursorV1>,
@@ -1987,6 +2039,13 @@ impl ActiveAdmission<'_> {
                 if outcome.status == HostAdmissionStatus::Backpressured {
                     hotpath::gauge!("jsonl_admission_backpressure_writer").inc(1.0);
                 }
+                if is_lost_cursor_cas(&outcome)
+                    && self
+                        .peer_already_covered(expected_cursor, checkpoint.end_offset)
+                        .await
+                {
+                    return Ok(DurableFrameDisposition::AlreadyDurable);
+                }
                 if is_admission_cancellation(&outcome, &self.cancellation) {
                     Err(TranscriptIngestError::Cancelled {
                         provider: self.provider,
@@ -2172,6 +2231,21 @@ impl ActiveAdmission<'_> {
                         }
                     }
                 }
+                // The batch is atomic: nothing in this window committed. When
+                // the peer that won the CAS is already past the window's last
+                // frame, every frame in it is durable behind the winner's
+                // cursor, so this is a no-op rather than a failed source pass.
+                if is_lost_cursor_cas(&outcome)
+                    && let Some(last) = checkpoints.last()
+                    && self
+                        .peer_already_covered(expected_cursor, last.end_offset)
+                        .await
+                {
+                    progress.frames_skipped = progress
+                        .frames_skipped
+                        .saturating_add(checkpoints.len() as u64);
+                    return Ok(());
+                }
                 if is_admission_cancellation(&outcome, &self.cancellation) {
                     Err(CaptureWindowError::Ingest(
                         TranscriptIngestError::Cancelled {
@@ -2852,7 +2926,14 @@ pub(in crate::runtime) async fn admit_jsonl_observations<State: Clone>(
 /// 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 {
+/// A cursor compare-and-swap lost to a peer that owns the same
+/// `(source, scope)` cursor. Retryable by construction; whether it is a
+/// failure at all depends on what the winner already covered.
+fn is_lost_cursor_cas(outcome: &HostAdmissionOutcome) -> bool {
+    outcome.reason_code == Some("cursor_conflict")
+}
+
+pub(in crate::runtime) fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool {
     matches!(
         outcome.recovery,
         Some(HostAdmissionRecovery::DeterministicContentRefusal)
diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs
index cb9b053e7d..3915ac089b 100644
--- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs
+++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs
@@ -59,6 +59,10 @@ struct SeamSpyAdmission {
     capture_calls: AtomicU64,
     capture_collision_dispositions: Mutex<Vec<ObservationIdentityCollisionDispositionV1>>,
     cover_past_advances: Mutex<Vec<ObservationCursorAdvance>>,
+    /// Commit the next capture through the shared store and then report the
+    /// cursor CAS as lost, the way a live hook ingest wins the race a sweep
+    /// was still trying to write.
+    peer_wins_next_cursor_cas: AtomicBool,
 }
 
 #[tokio::test]
@@ -651,6 +655,14 @@ impl SeamSpyAdmission {
         *self.scripted_capture_error.lock().unwrap() = Some(outcome);
     }
 
+    fn script_peer_wins_next_cursor_cas(&self) {
+        self.peer_wins_next_cursor_cas.store(true, Ordering::SeqCst);
+    }
+
+    fn peer_won_cursor_cas(&self) -> bool {
+        self.peer_wins_next_cursor_cas.swap(false, Ordering::SeqCst)
+    }
+
     fn script_batch_error(&self, outcome: HostAdmissionOutcome) {
         *self.scripted_batch_error.lock().unwrap() = Some(outcome);
     }
@@ -683,6 +695,12 @@ impl HostAdmission for SeamSpyAdmission {
                 .lock()
                 .unwrap()
                 .push(request.identity_collision_disposition());
+            if self.peer_won_cursor_cas() {
+                let _ = self.inner.capture_observation(request).await;
+                return Err(HostAdmissionOutcome::retained_backpressured(
+                    "cursor_conflict",
+                ));
+            }
             if let Some(outcome) = self.scripted_capture_error_once.lock().unwrap().take() {
                 return Err(outcome);
             }
@@ -703,6 +721,12 @@ impl HostAdmission for SeamSpyAdmission {
                     .iter()
                     .map(CaptureObservationRequest::identity_collision_disposition),
             );
+            if self.peer_won_cursor_cas() {
+                let _ = self.inner.capture_observations(requests).await;
+                return Err(HostAdmissionOutcome::retained_backpressured(
+                    "cursor_conflict",
+                ));
+            }
             if let Some(outcome) = self.scripted_batch_error.lock().unwrap().take() {
                 return Err(outcome);
             }
@@ -898,6 +922,68 @@ async fn retryable_admission_failures_keep_their_own_verdict() {
     assert!(stored_cursor(&spy).await.is_none());
 }
 
+/// Live hook ingest and the catch-up sweep own the same `(source, scope)`
+/// cursor, so one of them loses the store's compare-and-swap. When the winner
+/// already covered the range the loser was writing, the loser's work is
+/// durable and the pass is a no-op, not a failed source: reporting it as a
+/// failure produced a "Cursor transcript catch-up failed" WARN roughly every
+/// ten seconds on a live daemon for work that was already committed.
+#[tokio::test]
+async fn cursor_cas_lost_to_a_peer_that_covered_the_range_is_a_no_op() {
+    let (_temp, path, len) = rollout_fixture();
+    let spy = SeamSpyAdmission::default();
+    spy.script_peer_wins_next_cursor_cas();
+
+    let stats =
+        try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None)
+            .await
+            .expect("a CAS the peer already covered must not fail the source pass");
+
+    assert_eq!(
+        stored_cursor(&spy).await.map(|cursor| cursor.position()),
+        Some(len),
+        "the pass must adopt the winner's frontier"
+    );
+    assert!(
+        !spy.inner.observations().is_empty(),
+        "the peer's commit is the durable record this pass stopped duplicating"
+    );
+    assert_eq!(
+        stats.frames_accepted, 0,
+        "the loser accepts nothing of its own"
+    );
+    assert!(
+        stats.frames_skipped > 0,
+        "the covered frames are counted as skipped, not lost"
+    );
+}
+
+/// The same lost CAS with nothing behind it stays a typed retryable block:
+/// adopting a frontier the winner never reached would skip real records.
+#[tokio::test]
+async fn cursor_cas_lost_without_peer_coverage_stays_a_typed_block() {
+    let (_temp, path, _len) = rollout_fixture();
+    let spy = SeamSpyAdmission::default();
+    spy.script_capture_error(HostAdmissionOutcome::retained_backpressured(
+        "cursor_conflict",
+    ));
+
+    let error =
+        try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None)
+            .await
+            .expect_err("an uncovered race must surface for another pass");
+
+    assert!(matches!(
+        error,
+        TranscriptIngestError::HostAdmission {
+            reason: "cursor_conflict",
+            retryable: true,
+            ..
+        }
+    ));
+    assert!(stored_cursor(&spy).await.is_none());
+}
+
 #[tokio::test]
 async fn eligible_identity_collision_retries_once_with_normalizer_fallback() {
     super::install_test_shared_jsonl_preparation_authority();
@@ -1175,6 +1261,11 @@ async fn content_refusals_cover_past_so_the_stream_converges() {
 
 #[tokio::test]
 async fn codex_session_meta_prefix_is_decoded_once_across_consumers() {
+    // The shared metadata cache retains entries up to
+    // `shared_jsonl_preparation_capacity()`, so this test only observes the
+    // shared decode once the preparation authority is installed: without it the
+    // capacity is the degraded fallback of one entry.
+    super::install_test_shared_jsonl_preparation_authority();
     let (_temp, path, _) = rollout_fixture();
     let first = SeamSpyAdmission::default();
     let second = SeamSpyAdmission::default();
diff --git a/crates/tracedecay/src/daemon/store_runtime_tests.rs b/crates/tracedecay/src/daemon/store_runtime_tests.rs
index 3b58371f8a..8799d13b61 100644
--- a/crates/tracedecay/src/daemon/store_runtime_tests.rs
+++ b/crates/tracedecay/src/daemon/store_runtime_tests.rs
@@ -30,9 +30,9 @@ use tracedecay_session_memory::memory::{
 };
 use tracedecay_store::{
     CursorAdvanceOutcome, FactReadControl, FactWriteControl, ObservationCoverageReason,
-    ObservationCursorAdvance, ObservationStore, ObservationStoreError, ProjectId,
-    ProjectMemoryFactHistoryQueryV1, ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1,
-    RetainedGraphStoreLeaseV1, StoreShardIdV1,
+    ObservationCursorAdvance, ObservationStore, ProjectId, ProjectMemoryFactHistoryQueryV1,
+    ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1, RetainedGraphStoreLeaseV1,
+    StoreShardIdV1,
 };
 use tracedecay_store_runtime::{
     DaemonSessionRuntimeRegistryV1, RegisteredSchemaConvergenceStatus, process_runtime_generation,
@@ -1193,19 +1193,22 @@ async fn retained_runtime_ledger_replays_during_bounded_background_convergence()
             .expect("replay retained cursor while convergence is pending"),
         CursorAdvanceOutcome::ExactDuplicate
     );
-    let conflicting_advance = runtime_cursor_advance(
+    // The same range under a different coverage reason finds the retained
+    // cursor already at its `next_cursor`, so it is a duplicate of the
+    // applied coverage rather than a collision (#1842).
+    let rereasoned_advance = runtime_cursor_advance(
         &project_id,
         "retired",
         ObservationCoverageReason::BlankFrame,
     );
-    assert!(matches!(
+    assert_eq!(
         database
             .observation_store()
-            .advance_source_cursor(conflicting_advance)
+            .advance_source_cursor(rereasoned_advance)
             .await
-            .expect_err("classify retained cursor collision while convergence is pending"),
-        ObservationStoreError::CursorAdvanceCollision
-    ));
+            .expect("classify a re-reasoned retained cursor while convergence is pending"),
+        CursorAdvanceOutcome::ExactDuplicate
+    );
 
     let fresh_advance =
         runtime_cursor_advance(&project_id, "fresh", ObservationCoverageReason::OutOfScope);
diff --git a/crates/tracedecay/tests/session_suite/observation_store/mod.rs b/crates/tracedecay/tests/session_suite/observation_store/mod.rs
index c4d27abd67..eeeb421313 100644
--- a/crates/tracedecay/tests/session_suite/observation_store/mod.rs
+++ b/crates/tracedecay/tests/session_suite/observation_store/mod.rs
@@ -1476,7 +1476,11 @@ async fn cursor_only_progress_persists_non_payload_receipt_and_retries_idempoten
 }
 
 #[tokio::test]
-async fn cursor_only_retry_rejects_same_cursor_with_different_reason() {
+/// A second owner replaying the same range with a different coverage reason
+/// finds the durable cursor already at its `next_cursor`: the coverage it
+/// wanted to record is applied, so the replay is a duplicate, not a
+/// collision that blocks ingest (#1842). The committed reason stays.
+async fn cursor_only_retry_with_same_cursor_and_different_reason_is_a_duplicate() {
     let tmp = TempDir::new().unwrap();
     let runtime = profile_runtime(&tmp).await;
     let store = runtime
@@ -1493,7 +1497,7 @@ async fn cursor_only_retry_rejects_same_cursor_with_different_reason() {
         .await
         .unwrap();
 
-    assert!(matches!(
+    assert_eq!(
         store
             .advance_source_cursor(cursor_advance(
                 None,
@@ -1501,9 +1505,10 @@ async fn cursor_only_retry_rejects_same_cursor_with_different_reason() {
                 10,
                 NonDurableFrameReason::OutOfScope,
             ))
-            .await,
-        Err(ObservationStoreError::CursorAdvanceCollision)
-    ));
+            .await
+            .unwrap(),
+        CursorAdvanceOutcome::ExactDuplicate
+    );
     assert_eq!(
         store.get_source_cursor(&source(), &scope()).await.unwrap(),
         Some(cursor(10))