diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs index 846387e8ca..00925df9c0 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs +++ b/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, diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs index fbf97ef025..245e6f64ba 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs +++ b/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() } } diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/tests.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/tests.rs index c62091c5d4..9963dd2df8 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/tests.rs +++ b/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"); diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs index 4bc895f9b2..16a37a8027 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs +++ b/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< diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs index b548c3bc95..e80482175c 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs +++ b/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] diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs index f397d6df9d..ba9e35e8d4 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs +++ b/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!({