fix(hooks): keep ingest commit status off the drain residual

crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs

@@ -672,30 +672,31 @@ pub async fn ingest_transcript_with_cancellation(
         route_admission,
         observations_committed: route_observations_committed,
         exact_duplicate: route_exact_duplicate,
+        admission_owns_commit,
     } = capture;
-    // Admission is the durable commit; projection is downstream materialization
-    // off a queue this scope shares with the project catch-up sweep. Counting
-    // only the projections this pass drained itself reports a pass whose rows a
-    // peer drainer took as though it had captured nothing.
-    let authority_changed = messages_upserted > 0
-        || route_observations_committed > 0
-        || snapshot_capture
+    let verdict = ingest_commit_verdict(&IngestCommitAccount {
+        admission_owns_commit,
+        observations_committed: route_observations_committed,
+        route_exact_duplicate,
+        messages_upserted,
+        snapshot_messages_upserted: snapshot_capture
+            .as_ref()
+            .map_or(0, |capture| capture.stats.messages_upserted),
+        claude_observations_committed: claude_observation_stats
+            .as_ref()
+            .map_or(0, |stats| stats.observations_committed),
+        claude_cursor_advances: claude_observation_stats
             .as_ref()
-            .is_some_and(|capture| capture.stats.messages_upserted > 0)
-        || claude_observation_stats
+            .map_or(0, |stats| stats.cursor_advances),
+        claude_observation_duplicates: claude_observation_stats
             .as_ref()
-            .is_some_and(|stats| stats.observations_committed > 0 || stats.cursor_advances > 0);
-    // A pass that changed nothing is only `accepted_for_replay` when it cannot
-    // prove the data is already there. Routes that can prove it say so: Claude
-    // through its duplicate counters, every other route through
-    // `exact_duplicate`. Without this a replay whose observations a peer
-    // drainer already projected reports a terminal, non-retryable status that
-    // neither proves a commit nor invites a retry.
-    let exact_duplicate = !authority_changed
-        && (route_exact_duplicate
-            || claude_observation_stats.as_ref().is_some_and(|stats| {
-                stats.observation_duplicates > 0 || stats.cursor_duplicates > 0
-            }));
+            .map_or(0, |stats| stats.observation_duplicates),
+        claude_cursor_duplicates: claude_observation_stats
+            .as_ref()
+            .map_or(0, |stats| stats.cursor_duplicates),
+    });
+    let authority_changed = verdict.authority_changed;
+    let exact_duplicate = verdict.exact_duplicate;
     let deferred_by_byte_cap = source_deferred
         || snapshot_capture
             .as_ref()
@@ -779,6 +780,57 @@ pub async fn ingest_transcript_with_cancellation(
     Ok(output)
 }
 
+/// The counters a capture route hands the terminal-status assembly.
+///
+/// `admission_owns_commit` routes (Cursor, Codex project) already know whether
+/// they persisted frames. Their projection drain reads a queue the project
+/// catch-up also empties, so `messages_upserted` on those routes is a residual
+/// of that queue, not a second copy of the commit.
+pub(super) struct IngestCommitAccount {
+    pub(super) admission_owns_commit: bool,
+    pub(super) observations_committed: u64,
+    pub(super) route_exact_duplicate: bool,
+    pub(super) messages_upserted: u64,
+    pub(super) snapshot_messages_upserted: u64,
+    pub(super) claude_observations_committed: u64,
+    pub(super) claude_cursor_advances: u64,
+    pub(super) claude_observation_duplicates: u64,
+    pub(super) claude_cursor_duplicates: u64,
+}
+
+pub(super) struct IngestCommitVerdict {
+    pub(super) authority_changed: bool,
+    pub(super) exact_duplicate: bool,
+}
+
+/// Commit status from the route that owns it.
+///
+/// When admission owns the commit, a non-zero drain residual cannot promote a
+/// pass that persisted nothing into `committed`, and a zero drain cannot hide
+/// frames this pass did persist. Routes without an admission tally still read
+/// their own message and duplicate counters.
+pub(super) fn ingest_commit_verdict(account: &IngestCommitAccount) -> IngestCommitVerdict {
+    if account.admission_owns_commit {
+        let authority_changed = account.observations_committed > 0;
+        return IngestCommitVerdict {
+            authority_changed,
+            exact_duplicate: !authority_changed && account.route_exact_duplicate,
+        };
+    }
+    let authority_changed = account.messages_upserted > 0
+        || account.observations_committed > 0
+        || account.snapshot_messages_upserted > 0
+        || account.claude_observations_committed > 0
+        || account.claude_cursor_advances > 0;
+    IngestCommitVerdict {
+        authority_changed,
+        exact_duplicate: !authority_changed
+            && (account.route_exact_duplicate
+                || account.claude_observation_duplicates > 0
+                || account.claude_cursor_duplicates > 0),
+    }
+}
+
 pub(super) fn complete_ingest_admission(
     admission: HostAdmissionOutcome,
     authority_changed: bool,

crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs

@@ -125,6 +125,11 @@ pub(super) struct TranscriptCaptureOutcome {
     /// `messages_upserted` counts only the projections this pass drained
     /// itself, which a peer drainer can legitimately take first.
     pub(super) observations_committed: u64,
+    /// This route's admission tally is the commit. The projection drain is a
+    /// shared per-scope queue, so its residual must not enter the terminal
+    /// status. Routes that have no admission tally leave this false and keep
+    /// using their own message counts.
+    pub(super) admission_owns_commit: bool,
     /// The route committed nothing because its observations were already
     /// durable. Kept apart from `messages_upserted == 0`, which cannot tell an
     /// already-committed replay from a pass that captured nothing.
@@ -429,6 +434,7 @@ async fn capture_codex_project(
         source_deferred: admitted.deferred,
         observations_committed: admitted.observations_committed,
         exact_duplicate: admitted.exact_duplicate,
+        admission_owns_commit: true,
         ..TranscriptCaptureOutcome::default()
     })
 }
@@ -457,6 +463,7 @@ fn cursor_capture_outcome(
         source_deferred: stats.source_deferred,
         observations_committed: stats.observations_committed,
         exact_duplicate: stats.exact_duplicate,
+        admission_owns_commit: true,
         ..TranscriptCaptureOutcome::default()
     }
 }

crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/tests.rs

@@ -1,9 +1,99 @@
 use super::super::*;
 use crate::structured_hook_error_data;
 use tracedecay_project::test_support::host_admission::HostAdmissionTestRuntimeV1;
+use tracedecay_sessions::admission::{HostAdmissionOutcome, HostAdmissionStatus};
 
 use super::*;
 
+fn status_for(account: &IngestCommitAccount) -> HostAdmissionStatus {
+    let verdict = ingest_commit_verdict(account);
+    complete_ingest_admission(
+        HostAdmissionOutcome::accepted_for_replay(),
+        verdict.authority_changed,
+        verdict.exact_duplicate,
+        false,
+    )
+    .status
+}
+
+/// A shared-queue residual is not this pass's commit. Nine projected rows
+/// left by a peer, or by another provider on the same scope queue, must stay
+/// `accepted_for_replay` when admission persisted nothing and cannot prove a
+/// duplicate.
+#[test]
+fn drain_residual_does_not_commit_an_admission_owned_pass() {
+    let status = status_for(&IngestCommitAccount {
+        admission_owns_commit: true,
+        observations_committed: 0,
+        route_exact_duplicate: false,
+        messages_upserted: 9,
+        snapshot_messages_upserted: 0,
+        claude_observations_committed: 0,
+        claude_cursor_advances: 0,
+        claude_observation_duplicates: 0,
+        claude_cursor_duplicates: 0,
+    });
+
+    assert_eq!(status, HostAdmissionStatus::AcceptedForReplay);
+}
+
+/// The pass persisted two observations and the drain found nothing. The
+/// commit still stands.
+#[test]
+fn admission_commit_stands_when_the_drain_is_empty() {
+    let status = status_for(&IngestCommitAccount {
+        admission_owns_commit: true,
+        observations_committed: 2,
+        route_exact_duplicate: false,
+        messages_upserted: 0,
+        snapshot_messages_upserted: 4,
+        claude_observations_committed: 1,
+        claude_cursor_advances: 1,
+        claude_observation_duplicates: 0,
+        claude_cursor_duplicates: 0,
+    });
+
+    assert_eq!(status, HostAdmissionStatus::Committed);
+}
+
+/// A peer already admitted the source. Residual projected rows must not
+/// rewrite that duplicate into a fresh commit.
+#[test]
+fn drain_residual_does_not_promote_an_exact_duplicate() {
+    let status = status_for(&IngestCommitAccount {
+        admission_owns_commit: true,
+        observations_committed: 0,
+        route_exact_duplicate: true,
+        messages_upserted: 3,
+        snapshot_messages_upserted: 0,
+        claude_observations_committed: 0,
+        claude_cursor_advances: 0,
+        claude_observation_duplicates: 0,
+        claude_cursor_duplicates: 0,
+    });
+
+    assert_eq!(status, HostAdmissionStatus::ExactDuplicate);
+}
+
+/// Hermes and the other routes that have no admission tally still commit
+/// from the messages they themselves upserted.
+#[test]
+fn message_counted_route_still_commits_from_its_own_upserts() {
+    let status = status_for(&IngestCommitAccount {
+        admission_owns_commit: false,
+        observations_committed: 0,
+        route_exact_duplicate: false,
+        messages_upserted: 1,
+        snapshot_messages_upserted: 0,
+        claude_observations_committed: 0,
+        claude_cursor_advances: 0,
+        claude_observation_duplicates: 0,
+        claude_cursor_duplicates: 0,
+    });
+
+    assert_eq!(status, HostAdmissionStatus::Committed);
+}
+
 #[test]
 fn cursor_compaction_response_matches_hook_contract() {
     let value = cursor_compact_skipped("no messages to compact");

crates/tracedecay-sessions/src/runtime/hosts/cursor.rs

@@ -218,6 +218,27 @@ impl CursorSourceAdmissionTally {
     }
 }
 
+/// Fold one hook pass's admission over the projection drain it happened to run.
+///
+/// The projection queue is per scope. The project catch-up drains it too, so
+/// `drain.source_deferred`, `drain.exact_duplicate`, and `drain.messages_upserted`
+/// describe whoever last touched that queue, not this pass. A residual there
+/// must not hide a commit, invent a duplicate, or turn a byte-finished pass
+/// into backpressure. Admission is the commit.
+fn account_hook_admission(
+    mut drain: projection::CursorTranscriptIngestStats,
+    observations_committed: u64,
+    fully_replayed: bool,
+    admission_deferred: bool,
+    bytes_consumed: u64,
+) -> projection::CursorTranscriptIngestStats {
+    drain.bytes_consumed = bytes_consumed;
+    drain.source_deferred = admission_deferred;
+    drain.observations_committed = observations_committed;
+    drain.exact_duplicate = observations_committed == 0 && fully_replayed;
+    drain
+}
+
 // Cursor JSONL admission chokepoint: the whole per-file admission future is
 // boxed here so the per-file sweep loop no longer pins each call, keeping the
 // debug poll frame bounded through the deep ingest recursion chain.
@@ -631,19 +652,19 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission(
         admitted.record(&progress);
         budget.record_progress(progress.bytes_consumed, progress.source_deferred);
     }
-    let mut stats = drain_cursor_observation_projections(
+    let drain = drain_cursor_observation_projections(
         admission,
         &scope,
         &ObservationCancellation::default(),
     )
     .await?;
-    stats.bytes_consumed = budget.consumed();
-    stats.source_deferred |= budget.deferred();
-    stats.observations_committed = admitted.observations_committed;
-    stats.exact_duplicate |= stats.messages_upserted == 0
-        && stats.observations_committed == 0
-        && admitted.fully_replayed();
-    Ok(stats)
+    Ok(account_hook_admission(
+        drain,
+        admitted.observations_committed,
+        admitted.fully_replayed(),
+        budget.deferred(),
+        budget.consumed(),
+    ))
 }
 
 pub async fn ingest_cursor_user_transcript_event_capped(
@@ -781,19 +802,19 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission(
         admitted.record(&progress);
         budget.record_progress(progress.bytes_consumed, progress.source_deferred);
     }
-    let mut stats = drain_cursor_observation_projections(
+    let drain = drain_cursor_observation_projections(
         admission,
         &scope,
         &ObservationCancellation::default(),
     )
     .await?;
-    stats.bytes_consumed = budget.consumed();
-    stats.source_deferred |= budget.deferred();
-    stats.observations_committed = admitted.observations_committed;
-    stats.exact_duplicate |= stats.messages_upserted == 0
-        && stats.observations_committed == 0
-        && admitted.fully_replayed();
-    Ok(stats)
+    Ok(account_hook_admission(
+        drain,
+        admitted.observations_committed,
+        admitted.fully_replayed(),
+        budget.deferred(),
+        budget.consumed(),
+    ))
 }
 
 pub(in crate::runtime) fn try_ingest_cursor_project_sweep_capped_with_session_ids<

crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs

@@ -509,6 +509,52 @@ async fn replayed_cursor_ingest_reports_an_exact_duplicate_not_a_bare_replay() {
     );
 }
 
+/// The shared projection queue's residual is not this pass. A deferred drain
+/// with leftover rows does not defer the pass, invent a duplicate, or hide
+/// the frames admission persisted.
+#[test]
+fn hook_admission_ignores_a_shared_drain_residual() {
+    let committed = account_hook_admission(
+        CursorTranscriptIngestStats {
+            messages_upserted: 9,
+            source_deferred: true,
+            exact_duplicate: true,
+            ..CursorTranscriptIngestStats::default()
+        },
+        2,
+        false,
+        false,
+        40,
+    );
+    assert_eq!(committed.observations_committed, 2);
+    assert_eq!(committed.bytes_consumed, 40);
+    assert_eq!(committed.messages_upserted, 9);
+    assert!(!committed.source_deferred);
+    assert!(!committed.exact_duplicate);
+
+    let replayed = account_hook_admission(
+        CursorTranscriptIngestStats {
+            messages_upserted: 4,
+            source_deferred: true,
+            exact_duplicate: false,
+            ..CursorTranscriptIngestStats::default()
+        },
+        0,
+        true,
+        false,
+        0,
+    );
+    assert_eq!(replayed.observations_committed, 0);
+    assert!(replayed.exact_duplicate);
+    assert!(!replayed.source_deferred);
+
+    let deferred =
+        account_hook_admission(CursorTranscriptIngestStats::default(), 0, false, true, 8);
+    assert!(deferred.source_deferred);
+    assert!(!deferred.exact_duplicate);
+    assert_eq!(deferred.observations_committed, 0);
+}
+
 /// The duplicate verdict is evidence, not a default: a source this pass has
 /// never opened carries no proof that anything was committed before.
 #[tokio::test]

crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs

@@ -1115,9 +1115,18 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() {
             .expect("registered daemon stop response text"),
     )
     .expect("registered daemon stop payload");
-    assert_eq!(
-        stop_payload["status"], "committed",
-        "registered daemon stop ingest did not commit: {stop_response}"
+    // Same durable-terminal contract as the Cursor ingest above. The project
+    // catch-up can admit the rollout first; the hook then reports
+    // `exact_duplicate`. A drain residual on the shared projection queue must
+    // not be what flips that into `committed`, and `accepted_for_replay`
+    // still proves neither a commit nor a duplicate.
+    assert!(
+        matches!(
+            stop_payload["status"].as_str(),
+            Some("committed" | "exact_duplicate")
+        ),
+        "registered daemon stop ingest did not commit: {stop_response}\ndaemon log:\n{}",
+        std::fs::read_to_string(&daemon_log).expect("read isolated advisory daemon log"),
     );
 
     let advisory_args = json!({