From db2ee84366fd3ef92535eea45a79cf901e00fbd3 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 18 Sep 2026 23:22:32 +0000 Subject: [PATCH 01/18] fix(hooks): account an ingest commit apart from its own drain `packaged_host_ingest_delivers_a_registered_advisory_cycle` failed roughly every other run on the 4 vCPU arm runner with `admission: accepted_for_replay, completed: true, messages_upserted: 0`: a terminal, non-retryable status that neither proves a commit nor invites a retry. Reproduced locally at ~30% under `taskset -c 0-3`. Root cause: `authority_changed` for the Cursor and Codex hook routes was derived only from the projections the pass drained itself. The projection queue is per scope and consumed on projection, and the daemon's project catch-up sweep (`ingest::project_provider::run_cursor`) drains the whole scope's queue, not just the rows it admitted. On a slow runner the scheduler tick lands between the ingest's admit and its own drain, so the ingest finds an empty queue and reports zero, discarding the two observations it had just persisted. Daemon debug evidence from a failing run: `transcript_admission_batch phase=complete total_frames=2 bytes_consumed=175 source_deferred=false` followed by `messages_upserted: 0`. Admission is the durable commit; projection is downstream materialization. Carry `observations_committed` (frames persisted) from the Cursor and Codex admit loops through to the capture outcome, count it as an authority change, and report it in the ingest payload so a zero-projection pass is readable without guessing which drainer won. A second, distinct state was equally vacuous: a pass that re-scans a source already at its stored cursor, whose rows a peer drained earlier, persists nothing and finds nothing. That is `exact_duplicate`, but Cursor had no way to say so: `CursorProjectionDrainStats::into_transcript_stats` dropped `exact_duplicates`, and `exact_duplicate` was derived only from Claude observation stats, which are always `None` for Cursor. Both gaps are closed; `JsonlObservationAdmissionProgress::resumed` supplies the evidence that a source was already admitted rather than empty, so a first-ever scan is never reported as a duplicate. A genuinely deferred pass still reports that honestly: `source_deferred` and the backpressure mapping are untouched. The acceptance assertion now accepts `committed` or `exact_duplicate` and still rejects `accepted_for_replay`. Both prove the transcript is durable; the retry path (a first pass deferred by an incomplete projection rebuild, then a re-scan of the exhausted source) legitimately terminates in the latter. Verification: `taskset -c 0-3` acceptance run 12/12 pass (was 2/6 and 0/1 before the fix, same binary shape); three new deterministic tests in `runtime::hosts::cursor::tests` cover the peer-drain steal, the replay, and the empty-source negative; `tracedecay-sessions --lib` 569/569, `tracedecay-mcp --lib` 367/367, `tracedecay-host-admission` 1/1. Co-Authored-By: Claude Fable 5.1 --- .../src/handlers/hook_runtime/ingest.rs | 43 +++- .../handlers/hook_runtime/ingest/kernels.rs | 15 +- .../src/runtime/hosts/cursor.rs | 57 +++++- .../src/runtime/hosts/cursor/projection.rs | 16 ++ .../src/runtime/hosts/cursor/tests.rs | 186 ++++++++++++++++++ .../src/runtime/ingest/project_provider.rs | 2 + .../src/runtime/ingest/user.rs | 2 + .../jsonl_observation_admission.rs | 6 + .../advisory_runtime_acceptance.rs | 14 +- 9 files changed, 330 insertions(+), 11 deletions(-) diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs index c0a7668ffb..3cd2df6eee 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs +++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs @@ -119,9 +119,10 @@ async fn admit_codex_project_rollouts( project_id: ProjectId, max_new_bytes: Option, cancellation: &ObservationCancellation, -) -> Result { +) -> Result { let mut budget = max_new_bytes; let mut deferred = false; + let mut observations_committed = 0_u64; let mut paths = source.transcript_paths(project_root).into_iter().peekable(); while let Some(path) = paths.next() { let progress = @@ -136,6 +137,7 @@ async fn admit_codex_project_rollouts( .await .map_err(|error| map_transcript_ingest_error(&error))?; deferred |= progress.source_deferred; + observations_committed = observations_committed.saturating_add(progress.frames_persisted); if let Some(remaining) = budget.as_mut() { *remaining = remaining.saturating_sub(progress.bytes_consumed); if *remaining == 0 { @@ -144,7 +146,18 @@ async fn admit_codex_project_rollouts( } } } - Ok(deferred) + Ok(CodexRolloutAdmission { + deferred, + observations_committed, + }) +} + +/// What one Codex rollout admission pass committed, apart from what its own +/// projection drain later catches. The projection queue is shared per scope +/// with the project catch-up sweep, which can consume these rows first. +pub(super) struct CodexRolloutAdmission { + pub(super) deferred: bool, + pub(super) observations_committed: u64, } async fn drain_host_observation_projections( @@ -642,18 +655,32 @@ pub async fn ingest_transcript_with_cancellation( source_deferred, lcm_receipt, route_admission, + observations_committed: route_observations_committed, + exact_duplicate: route_exact_duplicate, } = 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 .as_ref() .is_some_and(|capture| capture.stats.messages_upserted > 0) || 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 - && claude_observation_stats - .as_ref() - .is_some_and(|stats| stats.observation_duplicates > 0 || stats.cursor_duplicates > 0); + && (route_exact_duplicate + || claude_observation_stats.as_ref().is_some_and(|stats| { + stats.observation_duplicates > 0 || stats.cursor_duplicates > 0 + })); let deferred_by_byte_cap = source_deferred || snapshot_capture .as_ref() @@ -710,6 +737,12 @@ pub async fn ingest_transcript_with_cancellation( .await; output["hint_outcomes"] = settlement.as_json(); } + // Routes that admit observations directly report what they committed, so a + // `messages_upserted: 0` pass is readable without guessing which drainer + // won. The snapshot and Claude blocks below own the key for their routes. + if route_observations_committed > 0 { + output["observations_committed"] = json!(route_observations_committed); + } if let Some(capture) = snapshot_capture { output["observations_committed"] = json!(capture.stats.messages_upserted); output["bytes_consumed"] = json!(capture.bytes_consumed); 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 dbfda00811..d8918e65af 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs +++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs @@ -121,6 +121,14 @@ pub(super) struct TranscriptCaptureOutcome { pub(super) snapshot: Option, pub(super) claude_observation: Option, pub(super) source_deferred: bool, + /// Observations the route durably admitted, whoever later projects them. + /// `messages_upserted` counts only the projections this pass drained + /// itself, which a peer drainer can legitimately take first. + pub(super) observations_committed: u64, + /// 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. + pub(super) exact_duplicate: bool, /// Set by routes that commit through the LCM authority instead of a source /// scan; rendered as `authority_outcome` and `committed_state`. pub(super) lcm_receipt: Option, @@ -405,7 +413,7 @@ async fn capture_codex_project( let scope = ObservationScopeV1::Project { project_id: project_id.clone(), }; - let source_deferred = admit_codex_project_rollouts( + let admitted = admit_codex_project_rollouts( ctx.facade, &source, cg.project_root(), @@ -418,7 +426,8 @@ async fn capture_codex_project( drain_host_observation_projections(ctx.facade, &scope, ctx.cancellation).await?; Ok(TranscriptCaptureOutcome { messages_upserted, - source_deferred, + source_deferred: admitted.deferred, + observations_committed: admitted.observations_committed, ..TranscriptCaptureOutcome::default() }) } @@ -445,6 +454,8 @@ fn cursor_capture_outcome( TranscriptCaptureOutcome { messages_upserted: stats.messages_upserted, source_deferred: stats.source_deferred, + observations_committed: stats.observations_committed, + exact_duplicate: stats.exact_duplicate, ..TranscriptCaptureOutcome::default() } } diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs index 4c36450be6..4bc895f9b2 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs @@ -178,6 +178,46 @@ fn cursor_admission_record_id( Ok((identity.into_primary(), retry_eligible)) } +/// What one Cursor ingest pass did to the sources it scanned, independently of +/// what its own projection drain happened to catch. +/// +/// The projection queue is shared per scope and consumed on projection, so the +/// project catch-up sweep in [`crate::runtime::ingest::project_provider`] can +/// drain the rows this pass just admitted before this pass drains them itself. +/// Projection-output counts alone therefore cannot answer whether the pass's +/// transcript is durable, and both of the states below report zero outputs: +/// +/// - `observations_committed > 0`: this pass persisted new observations. They +/// are durable at admission; which drainer materializes them is not the +/// host's question. +/// - `fully_replayed()`: every scanned source resumed at its stored cursor +/// with nothing new to persist, so its observations were already durable. +#[derive(Debug, Default, Clone, Copy)] +struct CursorSourceAdmissionTally { + scanned: u64, + replayed: u64, + observations_committed: u64, +} + +impl CursorSourceAdmissionTally { + fn record(&mut self, progress: &JsonlObservationAdmissionProgress) { + self.scanned = self.scanned.saturating_add(1); + self.observations_committed = self + .observations_committed + .saturating_add(progress.frames_persisted); + if progress.resumed && progress.frames_persisted == 0 { + self.replayed = self.replayed.saturating_add(1); + } + } + + /// True only when at least one source was scanned and every one of them + /// was a pure replay. One source with new frames makes the pass a commit, + /// not a duplicate. + const fn fully_replayed(self) -> bool { + self.scanned > 0 && self.scanned == self.replayed + } +} + // 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. @@ -575,6 +615,7 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission( "sessions.hosts.cursor.discover_blocking", run_blocking_transcript_section(|| source.transcript_paths(&project_root)) ); + let mut admitted = CursorSourceAdmissionTally::default(); for path in paths { let context = cursor_observation_context(&source.event, &path, false); let progress = admit_cursor_jsonl_observations( @@ -587,6 +628,7 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission( &ObservationCancellation::default(), ) .await?; + admitted.record(&progress); budget.record_progress(progress.bytes_consumed, progress.source_deferred); } let mut stats = drain_cursor_observation_projections( @@ -597,6 +639,10 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission( .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) } @@ -719,6 +765,7 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission( "sessions.hosts.cursor.discover_blocking", run_blocking_transcript_section(|| source.transcript_paths(&placeholder)) ); + let mut admitted = CursorSourceAdmissionTally::default(); for path in paths { let context = cursor_observation_context(&source.event, &path, true); let progress = admit_cursor_jsonl_observations( @@ -731,6 +778,7 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission( &ObservationCancellation::default(), ) .await?; + admitted.record(&progress); budget.record_progress(progress.bytes_consumed, progress.source_deferred); } let mut stats = drain_cursor_observation_projections( @@ -741,6 +789,10 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission( .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) } @@ -820,6 +872,7 @@ async fn admit_cursor_sweep_observations_with_session_ids( "sessions.hosts.cursor.discover_blocking", run_blocking_transcript_section(|| source.transcript_paths(project_root)) ); + let mut admitted = CursorSourceAdmissionTally::default(); for path in paths { if cancellation.is_cancelled() { return Err(TranscriptIngestError::Cancelled { provider: "cursor" }); @@ -847,18 +900,20 @@ async fn admit_cursor_sweep_observations_with_session_ids( cancellation, ) .await?; + admitted.record(&progress); budget.record_progress(progress.bytes_consumed, progress.source_deferred); } if cancellation.is_cancelled() { return Err(TranscriptIngestError::Cancelled { provider: "cursor" }); } - let outcome = projection::drain_cursor_observation_projections_with_sessions( + let mut outcome = projection::drain_cursor_observation_projections_with_sessions( admission, &scope, cancellation, ) .await .map(|stats| stats.into_sweep_outcome(budget.consumed(), budget.deferred()))?; + outcome.stats.observations_committed = admitted.observations_committed; persist_host_provider_coverage( admission, &scope, diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs index 79f1ae795f..099fa77bd8 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs @@ -19,6 +19,18 @@ pub struct CursorTranscriptIngestStats { pub messages_upserted: u64, pub bytes_consumed: u64, pub source_deferred: bool, + /// Observations this pass durably admitted. `messages_upserted` counts + /// only what this pass's own projection drain materialized, and the + /// projection queue is shared per scope, so a peer drainer can consume + /// these rows first. Admission is the commit; keep it accounted for + /// separately from whoever projects it. + pub observations_committed: u64, + /// This pass changed nothing because its observations were already + /// durable: every scanned source resumed at its stored cursor with no new + /// frame to persist, or the projection drain met only exact duplicates. + /// Distinguishes an already-committed replay from a pass that committed + /// nothing at all; both report `messages_upserted == 0`. + pub exact_duplicate: bool, } #[derive(Debug, Default)] @@ -102,6 +114,7 @@ pub async fn try_ingest_cursor_user_sweep_capped_with_admission( pub(in crate::runtime) struct CursorProjectionDrainStats { pub session_ids: Vec, pub messages_upserted: u64, + pub exact_duplicates: u64, pub source_deferred: bool, } @@ -141,6 +154,7 @@ pub(in crate::runtime) async fn drain_cursor_observation_projections_with_sessio Ok(CursorProjectionDrainStats { session_ids: outcome.session_ids, messages_upserted: outcome.projected_outputs, + exact_duplicates: outcome.exact_duplicates, source_deferred: outcome.deferred, }) } @@ -152,6 +166,8 @@ impl CursorProjectionDrainStats { messages_upserted: self.messages_upserted, bytes_consumed: 0, source_deferred: self.source_deferred, + observations_committed: 0, + exact_duplicate: self.messages_upserted == 0 && self.exact_duplicates > 0, } } diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs index 6d2c90c07f..b548c3bc95 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs @@ -354,3 +354,189 @@ fn user_scope_selects_one_physical_authority_for_a_mirrored_session() { Some("session-mirrored") ); } + +/// Fixture for the replayed-ingest journey: one project-scoped Cursor hook +/// event whose transcript already carries two records. +fn cursor_replay_fixture() -> (tempfile::TempDir, String, ProjectId) { + // Production installs the process-wide capture authorities during daemon + // bootstrap; capture refuses with a typed `BackgroundResourceUnavailable` + // without them. + crate::runtime::observation::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); + let project = tempfile::tempdir().unwrap(); + let transcript = project.path().join("cursor-replayed.jsonl"); + std::fs::write( + &transcript, + concat!( + "{\"role\":\"user\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"Edit the shared file.\"}]}}\n", + "{\"role\":\"assistant\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"Saved src/lib.rs.\"}]}}\n" + ), + ) + .unwrap(); + let event = json!({ + "session_id": "session-replayed", + "conversation_id": "conversation-replayed", + "generation_id": "generation-replayed", + "transcript_path": transcript, + "workspace_roots": [project.path()], + }) + .to_string(); + let project_id = ProjectId::new("project.cursor-replayed").unwrap(); + (project, event, project_id) +} + +/// A pass that admits observations and then loses the projection queue to a +/// peer drainer still committed those observations. +/// +/// This is the production interleaving on a slow runner: the explicit hook +/// ingest admits the transcript, the project catch-up sweep's scheduler tick +/// drains the scope-wide projection queue, and the ingest's own drain then +/// finds nothing left. Reporting only the projections this pass drained itself +/// turns a real commit into a terminal, non-retryable `accepted_for_replay`. +#[tokio::test] +async fn cursor_ingest_reports_its_commit_when_a_peer_drains_the_projection_queue() { + let (_project, event, project_id) = cursor_replay_fixture(); + let admission = crate::admission::test_support::MemoryHostAdmission::default(); + let scope = ObservationScopeV1::Project { + project_id: project_id.clone(), + }; + + // Admit exactly as the hook ingest does, then let a peer empty the queue + // before the ingest's own drain can run. + let transcript: PathBuf = serde_json::from_str::(&event).unwrap()["transcript_path"] + .as_str() + .map(PathBuf::from) + .unwrap(); + let source_event: Value = serde_json::from_str(&event).unwrap(); + let context = cursor_observation_context(&source_event, &transcript, false); + let progress = admit_cursor_jsonl_observations( + "session-replayed", + &transcript, + &context, + &admission, + &scope, + None, + &ObservationCancellation::default(), + ) + .await + .unwrap(); + assert!( + progress.frames_persisted > 0, + "the admit persists the transcript frames: {progress:?}" + ); + let peer = projection::drain_cursor_observation_projections( + &admission, + &scope, + &ObservationCancellation::default(), + ) + .await + .unwrap(); + assert!( + peer.messages_upserted > 0, + "the peer drainer takes the queued rows: {peer:?}" + ); + + // The ingest now re-scans an exhausted source against an empty queue. + let stats = try_ingest_cursor_transcript_event_capped_with_admission( + &event, project_id, &admission, None, + ) + .await + .unwrap(); + assert_eq!( + stats.messages_upserted, 0, + "the peer already projected these rows: {stats:?}" + ); + assert!( + stats.observations_committed > 0 || stats.exact_duplicate, + "an ingest whose observations are durable must not look like a pass that captured nothing: {stats:?}" + ); +} + +/// A pass whose observations a peer drainer already projected must not report +/// the same zero-change accounting as a pass that captured nothing. +/// +/// The daemon's project catch-up sweep drains the whole Cursor projection +/// queue for a scope, not just the rows it admitted itself, and projection +/// consumes the queue row. So an explicit hook ingest that admitted on a +/// deferred first call can find the queue empty on its next call even though +/// its own observations are durably committed. Reported as an unqualified +/// zero, the admission completes as `accepted_for_replay`: terminal, +/// non-retryable, and proving nothing. +#[tokio::test] +async fn replayed_cursor_ingest_reports_an_exact_duplicate_not_a_bare_replay() { + let (_project, event, project_id) = cursor_replay_fixture(); + let admission = crate::admission::test_support::MemoryHostAdmission::default(); + + let committed = try_ingest_cursor_transcript_event_capped_with_admission( + &event, + project_id.clone(), + &admission, + None, + ) + .await + .unwrap(); + assert!( + committed.messages_upserted > 0, + "the first pass admits and projects the transcript: {committed:?}" + ); + assert_eq!( + committed.observations_committed, 2, + "admission is the commit and is accounted for independently of whichever \ + drainer projects it: {committed:?}" + ); + assert!( + !committed.exact_duplicate, + "a pass that committed rows is not a duplicate: {committed:?}" + ); + + // Same event again: the source cursor is at end of file and the projection + // queue this scope shares with the catch-up sweep is already empty. + let replayed = try_ingest_cursor_transcript_event_capped_with_admission( + &event, project_id, &admission, None, + ) + .await + .unwrap(); + assert_eq!( + replayed.messages_upserted, 0, + "an already-projected replay upserts nothing: {replayed:?}" + ); + assert!( + !replayed.source_deferred, + "nothing is left to defer: {replayed:?}" + ); + assert!( + replayed.exact_duplicate, + "a replay of already-durable observations is an exact duplicate, not a bare accepted-for-replay: {replayed:?}" + ); +} + +/// 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] +async fn first_cursor_ingest_of_an_empty_source_is_never_an_exact_duplicate() { + crate::runtime::observation::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); + let project = tempfile::tempdir().unwrap(); + let transcript = project.path().join("cursor-empty.jsonl"); + std::fs::write(&transcript, "").unwrap(); + let event = json!({ + "session_id": "session-empty", + "transcript_path": transcript, + "workspace_roots": [project.path()], + }) + .to_string(); + let admission = crate::admission::test_support::MemoryHostAdmission::default(); + + let stats = try_ingest_cursor_transcript_event_capped_with_admission( + &event, + ProjectId::new("project.cursor-empty").unwrap(), + &admission, + None, + ) + .await + .unwrap(); + + assert_eq!(stats.messages_upserted, 0); + assert!( + !stats.exact_duplicate, + "a first-ever scan proves no prior commit: {stats:?}" + ); +} diff --git a/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs b/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs index 0d2029f659..e6b7f9e605 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs @@ -922,6 +922,8 @@ mod tests { messages_upserted: 3, bytes_consumed: 4, source_deferred: true, + observations_committed: 0, + exact_duplicate: false, }, session_ids: BTreeSet::from(["shared-session".to_string()]), }; diff --git a/crates/tracedecay-sessions/src/runtime/ingest/user.rs b/crates/tracedecay-sessions/src/runtime/ingest/user.rs index 43904b8849..f2e15c29b3 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/user.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/user.rs @@ -779,6 +779,8 @@ mod cursor_tests { messages_upserted: 3, bytes_consumed: 4, source_deferred: true, + observations_committed: 0, + exact_duplicate: false, }, session_ids: BTreeSet::from(["shared-session".to_string()]), }; 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 22ff24262b..468022fd84 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -276,6 +276,11 @@ pub(in crate::runtime) struct JsonlObservationAdmissionProgress { pub frames_rejected_before_decode: u64, pub frames_refused: u64, pub frames_persisted: u64, + /// This pass resumed from a durable source cursor rather than opening the + /// source for the first time. With `frames_persisted == 0` that is the + /// only evidence a caller has that the source was *already* admitted, + /// versus never carrying anything: both report zero new frames. + pub resumed: bool, pub io: crate::runtime::source::JsonlIoAccounting, } @@ -2280,6 +2285,7 @@ pub(in crate::runtime) async fn admit_jsonl_observations( let mut progress = JsonlObservationAdmissionProgress { bytes_consumed: raw.read_through.saturating_sub(raw.start_offset), source_deferred: raw.deferred.is_some(), + resumed: had_expected_cursor, io: if shared_page_hit { JsonlIoAccounting::default() } else { 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 a02f6328ed..f397d6df9d 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs @@ -1036,9 +1036,17 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { .expect("registered daemon ingest response text"), ) .expect("registered daemon ingest payload"); - assert_eq!( - payload["status"], "committed", - "registered daemon ingest did not commit: {response}" + // The daemon's project catch-up sweep drains the whole Cursor projection + // queue for this scope, so it can project the observations this ingest + // admitted on an earlier deferred pass. Both terminal states below prove + // the transcript is durable; only `accepted_for_replay` would not. + assert!( + matches!( + payload["status"].as_str(), + Some("committed" | "exact_duplicate") + ), + "registered daemon ingest did not commit: {response}\ndaemon log:\n{}", + std::fs::read_to_string(&daemon_log).expect("read isolated advisory daemon log"), ); // Codex records a turn in its rollout, not in the Stop event, so the From 13a36df9fea054c6566948da362267ec06ddd41f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:06:14 +0000 Subject: [PATCH 02/18] test(code-index): burn the banked wake permit before sampling dashboard_progress_does_not_wait_for_the_scheduler_mutex asserted Fresh and ignored_dependency_waits_for_global_admission_before_publication_gate asserted an idle global admission, and both sampled a background pass instead of the quiet worktree they set up. Master run 35402148043 failed both on try 1 (the dashboard read Some(Verifying), the admission read 0 permits); under `taskset -c 0,1` they reproduced 3/25 and 4/25. Both read the same state through different windows: the freshness ladder's `refresh_in_flight` is the pass counter *or* the pending-arrival slot, and the admission permit is held for a pass's whole source reconcile. Settling the mount-era chain the seat now leaves behind (6318c180d2, 1828d6e1af) is not enough, because an empty arrival slot is not an empty queue: `note_wake` posts a `Notify` permit for an arrival a running pass then claims, and `note_worker_continuation` replenishes that permit whenever it cannot claim the slot. Either leaves a banked permit behind a settled owner, and the worker's next `notified()` spends it on a no-op pass. A dump at the dashboard failure showed exactly that: two receipts (Mount/Published, BusyFollowUp/Noop), `pending=Some(0)`, and a third pass already in progress. That replenish is the liveness net for a pending arrival whose permit this pass consumed, so it stays. The tests drain the banked permit instead: release the admission, give the worker's claim its turn, and settle, until a release leaves the permit free. The dashboard test also drops to a single background permit, since the default bound is the host core count and one held permit parks nothing there; it then holds that permit across the sample. Both pass 40/40 under `taskset -c 0,1`, and noop_reconcile_tests, unchanged_background_freshness_probe_posts_no_overflow_wake, distinct_stores_reconcile_in_parallel_under_bounded_admission and registry_clone_freshness_reports_coverage_and_update_accounting pass 6/6. Co-Authored-By: Claude Fable 5.1 --- .../src/code_index_scheduler/tests/mod.rs | 46 +++++++++++++++++++ .../code_index_scheduler/tests/reconcile.rs | 30 +++++++++--- 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs index 3b6ba3ceb6..972169e649 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs @@ -1288,6 +1288,52 @@ async fn quiesced_background_reconcile_admission( admission } +/// Settle the owner *and* burn the coalesced wake permit a settled owner can +/// still be holding, so the global admission is idle and stays idle. +/// +/// [`wait_for_settled_owner`] proves the pending-arrival slot is empty now, but +/// emptiness is not the whole queue: `note_worker_continuation` replenishes the +/// `Notify` permit whenever it cannot claim the slot, and `note_wake` posts a +/// permit of its own for an arrival a running pass then claims. Either leaves a +/// banked permit behind a settled owner, and the no-op pass it starts owns the +/// single background admission while it runs. A test that reads +/// `available_permits`, or one that reads the freshness ladder (whose +/// `refresh_in_flight` is the pass counter *or* the pending slot), samples that +/// pass and not the quiet worktree it set up. +/// +/// Holding the permit parks such a pass at its dequeue point, before it claims +/// an arrival or enters its guard. Releasing it hands it straight over, so the +/// drain is done only once a release leaves the permit free. +/// +/// The registry must be single-permit +/// ([`CodeIndexSchedulerRegistryV1::with_background_reconcile_permits`]): with +/// the host's default bound, one held permit parks nothing. +async fn settled_owner_with_idle_admission( + registry: &CodeIndexSchedulerRegistryV1, + project_root: &Path, +) { + let admission = registry.background_reconcile_admission(); + let deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING; + loop { + drop(quiesced_background_reconcile_admission(registry, project_root).await); + // A banked permit is claimed by the worker's very next `notified()`, + // whose first act is to take this admission. Give that claim its turn, + // then settle: a pass that did start moves the guard or the slot this + // wait joins, and the free permit afterwards is the proof none is left. + tokio::time::sleep(Duration::from_millis(5)).await; + wait_for_settled_owner(registry, project_root).await; + if admission.available_permits() == 1 { + return; + } + assert!( + Instant::now() <= deadline, + "the admission for {} never went idle", + project_root.display() + ); + tokio::time::sleep(Duration::from_millis(2)).await; + } +} + const CALLER_STAR: usize = 2_000; const CALLER_STAR_FILES: usize = 8; const CALLER_PAGE: u32 = 10; 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 41254144b5..e79be3c75c 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 @@ -30,11 +30,12 @@ use super::{ quiesced_background_reconcile_admission, replace_scheduler_chunker_revision, replace_scheduler_policy_revision, rewrite_active_rust_extractor_revision, rewrite_preserving_stat, scheduler, scheduler_with_policy, served_lexical_texts, - test_project_id, wait_for_dashboard_ready, wait_for_event_to_ready, wait_for_generation_change, - wait_for_initial_generation, wait_for_live_complete_generation, - wait_for_live_complete_generation_by_polling, wait_for_queryable_text_generation, - wait_for_queryable_text_generation_change, wait_for_queryable_text_generation_id, - wait_for_quiescent_owner_pass, wait_for_settled_owner, wait_until_serving_seat, write, + settled_owner_with_idle_admission, test_project_id, wait_for_dashboard_ready, + wait_for_event_to_ready, wait_for_generation_change, wait_for_initial_generation, + wait_for_live_complete_generation, wait_for_live_complete_generation_by_polling, + wait_for_queryable_text_generation, wait_for_queryable_text_generation_change, + wait_for_queryable_text_generation_id, wait_for_quiescent_owner_pass, wait_for_settled_owner, + wait_until_serving_seat, write, }; use crate::{ code_index::{ @@ -2868,7 +2869,11 @@ async fn ignored_dependency_waits_for_global_admission_before_publication_gate() let store = TempDir::new().expect("store root"); let (registry, _) = mounted_core_query_worktree_with_one_permit(&fixture, &store).await; let latest = wait_for_live_complete_generation(®istry, fixture.path()).await; + // Draining leaves the busy follow-up wake armed, and every pass it starts + // owns the single global admission permit this test needs idle. Settle + // that chain and burn the banked permit behind it before sampling. drain_clone_backfill(®istry, fixture.path()).await; + settled_owner_with_idle_admission(®istry, fixture.path()).await; let generation = latest.generation(); let verified_import = generation .imports() @@ -3670,7 +3675,10 @@ async fn dashboard_progress_does_not_wait_for_the_scheduler_mutex() { .collect::>(); let fixture = GitFixture::new(&borrowed); let store = TempDir::new().expect("store root"); - let registry = CodeIndexSchedulerRegistryV1::new(1); + // One background permit, so holding it is what parks the owner: the + // default bound is the host core count and a single held permit would + // leave the other passes free to run under the sample below. + let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); registry .mount_worktree( test_project_id(), @@ -3681,6 +3689,16 @@ async fn dashboard_progress_does_not_wait_for_the_scheduler_mutex() { .expect("mount daemon-owned scheduler"); wait_for_initial_generation(®istry, fixture.path()).await; wait_for_dashboard_ready(®istry, fixture.path()).await; + // `refresh_in_flight` is the pass counter *or* the pending-wake slot, and + // `wait_for_dashboard_ready` only joins the running pass. The seat no + // longer waits for the clone successor, so the mount leaves backfill work + // behind, and the wakes that drain it leave a banked permit whose no-op + // pass projects Verifying instead of Fresh. Settle the whole mount-era + // chain, then hold the admission so no further pass can start under the + // sample below. + drain_clone_backfill(®istry, fixture.path()).await; + settled_owner_with_idle_admission(®istry, fixture.path()).await; + let _quiet_owner = quiesced_background_reconcile_admission(®istry, fixture.path()).await; let canonical_root = fixture .path() .canonicalize() From 3e821288bb10f4e67eb8c1734b9850e9a71feaae Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:22:12 +0000 Subject: [PATCH 03/18] test(code-index): prove the legacy restore by its read shape `legacy_generation_restore_does_not_materialize_its_evidence_segment` failed run 35407045249 at 1.288x of the evidence segment, inside the band 30c0a5d683 measured as clean on four earlier runs. The guard was measuring the wrong thing. Peak RSS cannot see this regression. VmHWM only moves for an allocation that does not fit in the heap the restored generation already made resident, and the restore itself grows that heap by 6.6 MiB before the first measured probe. Injecting a clone of every evidence page into a buffer held for the whole decode - the retention this test exists to refuse - moves the shipped guard by 0.302x and it passes. What VmHWM does register is additive noise: when the allocator trims between probes the next one faults fresh pages, which on a loaded 4 vCPU runner costs more than the 1.7 MiB segment under test. That, not a regression, is what failed the run. Assert instead on what the restore states about itself. Every segment read it issues is already visible to the caller through the read callback, and a restore that materializes the segment has to ask for it: the whole segment in one read, or a range that grows with it. The paged form of the identical generation is the reference, so the pre-paging form must ask for the same count of equally bounded ranges into a buffer bounded the same way. That is exact, catches the shipped regression (a whole-segment read) on the first probe in 0.47s, needs no memory reading, and holds on every platform rather than returning early off Linux. Peak RSS stays as a loose ceiling on the rest of the restore, and is made noise-tolerant without loosening the bound: each form is probed over four alternating rounds and the smallest growth is taken as its cost, because the noise is additive while a materializing restore pays on every round. The bound stays at one evidence segment. Verified on this lane: 10 consecutive runs pinned to 4 CPUs against a concurrent workspace `cargo check` on the same cores, all passing at 0.000-0.031x of the segment (was 0.12-1.288x); full code_index_suite 165/165 and the crate's 256 lib tests; clippy -p tracedecay-code-index --all-targets -D warnings. Co-Authored-By: Claude Fable 5.1 --- .../production_orchestration.rs | 308 ++++++++++++++---- 1 file changed, 237 insertions(+), 71 deletions(-) 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 c07cdca96a..89c3b65fb3 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 @@ -4548,19 +4548,42 @@ fn rss_scaled_request(file_count: usize) -> CodeIndexBuildRequestV1 { } } -/// Decode `manifest`, serving every segment read from `segments`, and return -/// the peak RSS growth over a freshly reset high water mark. -fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap>) -> u64 { +/// What one decode of a generation observed: the shape of the segment reads +/// the restore issued, and the peak RSS it grew over a freshly reset high +/// water mark. +struct RssDecodeProbeV1 { + hwm_delta_kib: u64, + evidence_reads: usize, + evidence_read_whole: bool, + largest_evidence_read: u64, + largest_evidence_buffer: usize, +} + +/// Decode `manifest`, serving every segment read from `segments`, and report +/// both the read shape and the peak RSS growth. +fn rss_measure_decode( + label: &str, + manifest: &[u8], + segments: &BTreeMap>, + evidence_digest: &str, +) -> RssDecodeProbeV1 { assert!(rss_reset_peak(), "reset VmHWM"); let hwm_before = rss_proc_kib("VmHWM").expect("VmHWM"); let mut whole_reads = 0_usize; let mut ranged_reads = 0_usize; + let mut probe = RssDecodeProbeV1 { + hwm_delta_kib: 0, + evidence_reads: 0, + evidence_read_whole: false, + largest_evidence_read: 0, + largest_evidence_buffer: 0, + }; let restored = CodeIndexPublishedGenerationV1::decode_partitioned_sealed(manifest, |request, buffer| { - let (digest, offset, length) = match request { + let (digest, offset, length, whole) = match request { SealedGenerationSegmentReadV1::Whole { digest, size_bytes } => { whole_reads += 1; - (digest, 0, size_bytes) + (digest, 0, size_bytes, true) } SealedGenerationSegmentReadV1::Range { digest, @@ -4569,9 +4592,15 @@ fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap { ranged_reads += 1; - (digest, offset, length) + (digest, offset, length, false) } }; + let evidence = digest.as_str() == evidence_digest; + if evidence { + probe.evidence_reads += 1; + probe.evidence_read_whole |= whole; + probe.largest_evidence_read = probe.largest_evidence_read.max(length); + } let bytes = segments.get(digest.as_str()).ok_or_else(|| { CodeIndexProductionErrorV1::Contract("measured segment is missing".to_owned()) })?; @@ -4579,6 +4608,12 @@ fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap, + legacy_manifest: Vec, + segments: BTreeMap>, + evidence_digest: String, + evidence_bytes: usize, + generation_bytes: usize, + file_count: usize, +} +/// Publish one generation, then rewrite its descriptor into the pre-paging +/// shape a historical writer emitted: one whole authenticated evidence +/// segment, no page table. +fn legacy_rss_fixture(file_count: usize) -> LegacyRssFixtureV1 { let store = SharedPublicationStore::default(); let mut owner = CodeIndexProductionOwnerV1::new(config(), store, ApplyingProjectionSink) .expect("rss fixture owner"); @@ -4669,8 +4685,6 @@ fn legacy_generation_restore_does_not_materialize_its_evidence_segment() { }) .expect("rss fixture encodes"); - // Rewrite the descriptor into the pre-paging shape a historical writer - // emitted: one whole authenticated evidence segment, no page table. let mut envelope: serde_json::Value = serde_json::from_slice(&paged_manifest).expect("manifest JSON"); envelope["generation"]["generation_evidence"] @@ -4696,13 +4710,160 @@ fn legacy_generation_restore_does_not_materialize_its_evidence_segment() { drop(generation); drop(owner); + LegacyRssFixtureV1 { + paged_manifest, + legacy_manifest, + segments, + evidence_digest, + evidence_bytes, + generation_bytes, + file_count, + } +} + +/// Restoring a pre-paging generation must not materialize its evidence +/// segment. +/// +/// The shipped restore read the whole segment, parsed a `serde_json::Value` +/// from it, rewrote identities in that tree and deserialized the tree again: +/// peak memory was 2.35x the on-disk generation and grew with the corpus. +/// +/// The proof is the read shape the restore asks the caller for, not its +/// memory footprint. A restore that materializes the segment has to hold it, +/// so it must ask for the whole segment in one read or for a range that grows +/// with the segment - the restore states its own peak segment residency in the +/// requests it issues. The paged form of the identical generation is the +/// reference: the pre-paging form must ask for the same bounded ranges into +/// the same bounded buffer. That is exact, needs no memory reading, and holds +/// on every platform. +/// +/// Peak RSS follows only as a loose ceiling on the whole restore, and it is +/// deliberately not the proof. VmHWM cannot see an allocation that fits inside +/// the heap the restored generation already made resident, so it does not +/// detect retention on its own: injecting a copy of every evidence page into a +/// buffer held for the decode moves it by a fifth of the segment or less, well +/// inside the clean band. What it does see is additive noise - a trimmed +/// allocator makes the next probe fault fresh pages, which on a loaded runner +/// cost more than the segment under test - so each form is measured over +/// several alternating rounds and the smallest growth is taken as its cost. +/// A restore that materializes pays that cost on every round, so the minimum +/// keeps whatever signal RSS carries and drops the noise that made a single +/// pair of probes flaky. +#[test] +fn legacy_generation_restore_does_not_materialize_its_evidence_segment() { + const RSS_CHILD: &str = "TD_LEGACY_RSS_CHILD"; + const RSS_TEST: &str = concat!( + "production_orchestration::", + "legacy_generation_restore_does_not_materialize_its_evidence_segment" + ); + /// Alternating paged/legacy rounds behind the discarded warm-up. + const RSS_ROUNDS: usize = 4; + + // VmHWM and `clear_refs` are Linux-only. The read-shape guard needs + // neither, so run it alone elsewhere rather than skipping the test. + let rss_readable = rss_proc_kib("VmHWM").is_some() && rss_reset_peak(); + + // VmHWM is process-wide, so the reading only means anything while nothing + // else is allocating: take it in a child that runs this test alone. + if rss_readable && std::env::var_os(RSS_CHILD).is_none() { + let status = std::process::Command::new(std::env::current_exe().expect("test binary")) + .args([RSS_TEST, "--exact", "--nocapture", "--test-threads=1"]) + .env(RSS_CHILD, "1") + .status() + .expect("run the peak-RSS measurement alone"); + assert!( + status.success(), + "the isolated restore measurement failed; its own failure is above" + ); + return; + } + + let file_count: usize = std::env::var("TD_LEGACY_RSS_FILES") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(300); + let fixture = legacy_rss_fixture(file_count); + + let paged = rss_measure_decode( + "paged", + &fixture.paged_manifest, + &fixture.segments, + &fixture.evidence_digest, + ); + let legacy = rss_measure_decode( + "legacy", + &fixture.legacy_manifest, + &fixture.segments, + &fixture.evidence_digest, + ); + + // --- Guard 1: the read shape, exact. ------------------------------------ + assert!( + !paged.evidence_read_whole && paged.evidence_reads > 1, + "the paged control must itself page the evidence segment: \ + evidence_reads={} evidence_read_whole={}", + paged.evidence_reads, + paged.evidence_read_whole + ); + assert!( + !legacy.evidence_read_whole, + "restoring a pre-paging generation asked for its whole \ + {}-byte evidence segment in one read: the segment is being materialized", + fixture.evidence_bytes + ); + assert!( + legacy.largest_evidence_read <= paged.largest_evidence_read, + "restoring a pre-paging generation read up to {} bytes of its \ + {}-byte evidence segment at once, beyond the {}-byte bound the paged \ + restore of the same generation holds to", + legacy.largest_evidence_read, + fixture.evidence_bytes, + paged.largest_evidence_read + ); + assert!( + legacy.largest_evidence_buffer <= paged.largest_evidence_buffer, + "restoring a pre-paging generation held a {}-byte evidence buffer, \ + beyond the {}-byte buffer the paged restore of the same generation \ + holds to: the segment is being materialized", + legacy.largest_evidence_buffer, + paged.largest_evidence_buffer + ); + assert_eq!( + legacy.evidence_reads, paged.evidence_reads, + "the pre-paging restore must read the evidence segment in the same \ + bounded chunks the page table would have named" + ); + + // --- Guard 2: peak RSS, noise-tolerant. --------------------------------- + if !rss_readable { + return; + } // The first restore in a process pays a cold-start cost (the arena a // restored generation needs) that has nothing to do with the form being - // restored. Spend it on a discarded probe so the two measured probes - // start from the same allocator state and stay comparable. - rss_measure_decode("warmup", &paged_manifest, &segments); - let paged_hwm = rss_measure_decode("paged", &paged_manifest, &segments); - let legacy_hwm = rss_measure_decode("legacy", &legacy_manifest, &segments); + // restored; the two probes above spent it. Alternate from here so neither + // form is systematically the one that inherits a trimmed allocator. + let mut paged_hwm = u64::MAX; + let mut legacy_hwm = u64::MAX; + for _ in 0..RSS_ROUNDS { + paged_hwm = paged_hwm.min( + rss_measure_decode( + "paged", + &fixture.paged_manifest, + &fixture.segments, + &fixture.evidence_digest, + ) + .hwm_delta_kib, + ); + legacy_hwm = legacy_hwm.min( + rss_measure_decode( + "legacy", + &fixture.legacy_manifest, + &fixture.segments, + &fixture.evidence_digest, + ) + .hwm_delta_kib, + ); + } // The paged decode of the same generation is the control, not a warm-up: // both forms restore the identical generation, so only the difference is // the pre-paging path's own cost. An absolute peak is not a usable @@ -4712,23 +4873,28 @@ fn legacy_generation_restore_does_not_materialize_its_evidence_segment() { let legacy_extra_bytes = legacy_hwm.saturating_sub(paged_hwm) * 1024; println!( - "rss_summary files={file_count} generation_on_disk_bytes={generation_bytes} \ -evidence_segment_bytes={evidence_bytes} paged_hwm_delta_kib={paged_hwm} \ + "rss_summary files={} generation_on_disk_bytes={} \ +evidence_segment_bytes={} rounds={RSS_ROUNDS} paged_hwm_delta_kib={paged_hwm} \ legacy_hwm_delta_kib={legacy_hwm} legacy_extra_bytes={legacy_extra_bytes} \ legacy_extra_over_evidence={:.3}", - legacy_extra_bytes as f64 / evidence_bytes as f64, - ); - - // A restore that materializes the segment holds all of it at once and - // parses it on top, so it costs at least the segment; the streaming - // restore costs one bounded page buffer plus allocator slack, measured - // at a fifth to a third of the segment. The bound sits between them, and - // is tighter than the half-the-whole-generation bound it replaces (which - // permitted two and a half times this segment). + fixture.file_count, + fixture.generation_bytes, + fixture.evidence_bytes, + legacy_extra_bytes as f64 / fixture.evidence_bytes as f64, + ); + + // The read-shape guard above already refused a restore that holds the + // segment. This is the ceiling on the rest of the restore: the pre-paging + // path must not cost a segment's worth of anything over the paged path. + // The bound is the segment because that is the size the guard is about, + // not a threshold tuned to the noise - the minimum over the rounds is + // what removes the noise. assert!( - legacy_extra_bytes < evidence_bytes as u64, + legacy_extra_bytes < fixture.evidence_bytes as u64, "restoring a pre-paging generation cost {legacy_extra_bytes} bytes of peak RSS beyond the \ - paged restore of the same {generation_bytes}-byte generation, which is not far below its \ - {evidence_bytes}-byte evidence segment: the segment is being materialized" + paged restore of the same {}-byte generation, which is not far below its \ + {}-byte evidence segment: the segment is being materialized", + fixture.generation_bytes, + fixture.evidence_bytes ); } From bf04c93ae708dc3dfa3a86a3e200323c9144f86d Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:20:23 +0000 Subject: [PATCH 04/18] test(transport): pay the refresh batch once, not at every reopen `background_refresh_and_reopen_report_only_servable_generations` was FLKY-FL 2/2 on the 4 vCPU Linux root-transport runner: TRY 1 failed after 63s inside `ProductionProjectCompositionHarnessV1::open`, TRY 2 passed in 24s as the last test on the box. The failure is the harness's own publish gate, not any assertion: production-composition code index did not publish ... after 20000 ms; composition gate capacity=1, admitted=1, waiting=0 with `progress.phase: Ready, completed_files: 98/98, committed_pages: 290, committed_payload_bytes: 31005972, completed_lexical_units: 56970602` and `clone_index: Stale { completed_source_pages: 0, total_source_pages: 290 }`. The reopen was still working through the refresh batch when its 20s budget expired. Root cause is fixture weight, not a regression. The batch exists only to keep one background refresh observable across a few status polls, but it stayed installed for the rest of the journey, so both reopens re-indexed 12,288 symbols inside a fixed 20s gate the test neither controls nor asserts on. dccdb15cdd already cut the batch 768 -> 96 files for exactly this reason; 96 is still ~8x more than the observation needs and the cost is paid three times. Two changes, both to the fixture and none to an assertion or a budget: retire the batch in the offline commit before the first reopen, so its weight lands on the refresh it exists for; and drop each file from 128 symbols to 16, which buys no fewer `partial_refresh_in_progress` samples at the 25ms poll interval. Attribution: PR #1792's `51402cdf8d` (the published branch worktree's query authority mount) is not the cause. Reverting it locally and re-running under the same contention failed 2/2 (79.8s, 88.5s), byte-identical gate message. Verification under `taskset -c 0-3`, contention emulated with four spinners pinned to the same cores (base: FAIL at 66.6s, matching CI's 63.3s): loaded 5/5 pass (34.0-47.1s), idle 5/5 pass (7.9-9.2s, was 36s). The `wait_for_background_refresh` window is still observed on every idle run, which is where it is tightest. Clippy clean on `tracedecay --features tracedecay/test-transport --test transport_acceptance_suite`. Co-Authored-By: Claude Fable 5.1 --- .../graph_rebuild_status_test.rs | 32 ++++++++++++++----- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs b/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs index 4b5eda1bee..6e741f8222 100644 --- a/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs +++ b/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs @@ -279,12 +279,15 @@ async fn wait_for_background_refresh( /// /// The batch only has to keep one refresh observable across a few status polls. /// At 768 files it instead indexed 98,304 symbols into 455 million lexical -/// units and 645 MB on disk, which on a four-core runner takes ~61s to commit -/// and then pushes the reopen past the composition harness's own 20s publish -/// gate: no `RECEIPT_TIMEOUT` can rescue that, the journey simply cannot finish. -/// 96 files still take seconds, so `partial_refresh_in_progress` is sampled -/// many times over at [`POLL_INTERVAL`], and every later open stays inside its -/// gate. +/// units and 645 MB on disk, which on a four-core runner takes ~61s to commit: +/// no `RECEIPT_TIMEOUT` can rescue that, the journey simply cannot finish. 96 +/// files of 16 symbols still take seconds, so `partial_refresh_in_progress` is +/// sampled many times over at [`POLL_INTERVAL`]; the 128 symbols a file used to +/// carry bought no extra samples and cost eight times the commit. +/// +/// The batch is also retired in the offline commit before the first reopen, so +/// its weight is paid by the refresh it exists for and not again by two reopens +/// bounded by a publish gate this journey cannot raise. const REFRESH_BATCH_FILES: u32 = 96; fn install_background_batch(isolation_root: &Path, project: &Path) { @@ -292,7 +295,7 @@ fn install_background_batch(isolation_root: &Path, project: &Path) { fs::create_dir_all(&staging).expect("background batch staging directory"); for file_index in 0..REFRESH_BATCH_FILES { let mut source = String::new(); - for symbol_index in 0..128_u32 { + for symbol_index in 0..16_u32 { writeln!( source, "pub fn refresh_probe_{file_index:04}_{symbol_index:03}(input: u32) -> u32 {{ input + {symbol_index} }}" @@ -360,12 +363,25 @@ async fn background_refresh_and_reopen_report_only_servable_generations_inner() ); harness.shutdown().await; + // The batch has done its only job: one background refresh stayed + // observable across many status polls. Leaving it installed makes every + // later reopen re-index the whole batch inside + // `ProductionProjectCompositionHarnessV1::open`'s fixed 20s publish gate, a + // budget this journey neither controls nor asserts on: on a contended + // four-core runner that reopen exhausts the gate and the open fails before + // any reopen assertion runs. Retiring the batch in the same offline commit + // keeps both reopen assertions exact -- a source change the closed daemon + // never saw, then a quiet checkout -- at the cost they actually need. + fs::remove_dir_all(project.join("src/refresh_batch")).expect("retire the background batch"); fs::write( project.join("src/after_reopen.rs"), "pub fn after_reopen() -> &'static str { \"current\" }\n", ) .expect("post-shutdown source"); - commit_all(&project, "change source while daemon is closed"); + commit_all( + &project, + "retire the batch and change source while daemon is closed", + ); let reopened_revision = head(&project); let reopened = ProductionProjectCompositionHarnessV1::open(isolation.path(), [project.clone()]) From 63f23d163d5f43f7243155b237ebee927f103672 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:48:14 +0000 Subject: [PATCH 05/18] test(daemon): honour the daemon's retry directive and the seat's cost `admitted_project_id` in the transport restart journey asserted success on its first `storage_status` right after `init`. The daemon can still be mounting the project's query authority then, and says so with a pre-admission `application.surface.unavailable` whose own retry directive is `after_delay` / `retry_after_millis`. Honour that directive the way a production client would instead of reading it as a verdict. `read_only_project_binding_refuses_before_scheduler_mutation` waited 5 s for the complete generation to seat; on a loaded 4 vCPU runner that background work has taken longer (CI: TRY 1 timed out at 5.2 s, TRY 2 seated in 0.6 s, and nextest's `flaky-result = "fail"` counts that as red). The bound is now 60 s; the assertion is unchanged. Co-Authored-By: Claude Fable 5.1 --- .../ignored_dependency_admission_tests.rs | 6 ++- .../transport_boundaries.rs | 47 +++++++++++++------ 2 files changed, 37 insertions(+), 16 deletions(-) diff --git a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs index 8c3a2499c6..d20e8879af 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs @@ -386,8 +386,10 @@ async fn latest( project_root: &Path, ) -> LatestCompleteCodeIndexV1 { // Lightweight publication precedes complete-generation seating. Demand - // that complete state before using its imports as admission evidence. - tokio::time::timeout(Duration::from_secs(5), async { + // that complete state before using its imports as admission evidence. The + // seat is background work behind the scheduler mutex; under a loaded CI + // runner it has taken over 5 s, so the bound is generous. + tokio::time::timeout(Duration::from_secs(60), async { loop { let _ = registry.latest_complete_fresh(project_root).await; if registry diff --git a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs index 34cc736911..0c79a835e3 100644 --- a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs +++ b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs @@ -110,20 +110,39 @@ fn http_mount(home: &Path) -> HttpMount { /// this identity locally: its route accepts only the daemon's public ID. fn admitted_project_id(home: &Path, project: &Path) -> String { let project_arg = project.to_string_lossy().into_owned(); - let output = tracedecay_command_with_home(home) - .current_dir(project) - .args([ - "tool", - "--project", - project_arg.as_str(), - "storage_status", - "--args", - r#"{"include_details":false}"#, - "--json", - ]) - .stdin(Stdio::null()) - .output() - .expect("read daemon-admitted project identity"); + let deadline = Instant::now() + AUTHORITY_TIMEOUT; + let output = loop { + let output = tracedecay_command_with_home(home) + .current_dir(project) + .args([ + "tool", + "--project", + project_arg.as_str(), + "storage_status", + "--args", + r#"{"include_details":false}"#, + "--json", + ]) + .stdin(Stdio::null()) + .output() + .expect("read daemon-admitted project identity"); + // Right after `init` the daemon can still be mounting the project's + // query authority; it says so with a pre-admission problem whose own + // retry directive is `after_delay`. Honour that directive, as a + // production client would, instead of treating it as a verdict. + let problem = serde_json::from_slice::(&output.stdout) + .ok() + .map(|envelope| envelope["problem"].clone()) + .filter(|problem| problem["terminality"] == "pre_admission") + .filter(|problem| problem["retry"] == "after_delay"); + match problem { + Some(problem) if !output.status.success() && Instant::now() < deadline => { + let millis = problem["retry_after_millis"].as_u64().unwrap_or(250); + std::thread::sleep(Duration::from_millis(millis)); + } + _ => break output, + } + }; assert!( output.status.success(), "storage_status failed while admitting the fixture project\nstdout:\n{}\nstderr:\n{}", From 48cc9d5b5951384e1f690dde9e2bd40e3ef2bd2f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:49:00 +0000 Subject: [PATCH 06/18] fix(sessions): bound a project ingest pass by bytes, not one source `completed_session_import_immediately_searches_canonical_message` timed out on its own 30 s deadline in the master root-transport partition (run 35402148043, both nextest tries). Reproduced under `taskset -c 0-3` with four competing CPU hogs: 42-45 s, deterministic. Unloaded it took 23 s against a 30 s wall, so the partition's own parallelism is all it takes to cross it. `ProjectProviderRun::run_codex` ended its pass at the first rollout that consumed bytes whenever another discovered path followed it, and marked the discovery frontier uncommittable on the way out. So a pass admitted exactly one rollout however much of its byte budget was left, the frontier never advanced, and the next pass rediscovered and re-read every rollout it had already exhausted. For the fixture's 33 rollouts that is 33 passes and 561 source admissions for 99 messages: the debug log shows 554 `transcript_admission_batch` events, n(n+1)/2. It is quadratic in the rollouts a project has, so a real project pays far worse than this test. The profile-scope loop in `ingest/user.rs` has always walked every discovered path until its byte budget runs out, over the same admission call for the same provider. The bound that owns a resumable cursor is `MAX_CAPTURE_WINDOW`: it stops inside one rollout and reports `source_deferred`, and the next pass resumes at the stored byte offset. Yielding per finished rollout has no cursor behind it, which is why it re-reads instead of resuming. Break only on `source_deferred`, exactly what `frontier_committable` already tests, and let `remaining` bound the pass as it does for the profile scope. The import now takes 5.3 s of ingest instead of 20 s, and the guarantees are unchanged: a partially read source still ends the pass and still blocks the frontier, cancellation and the byte budget still break at the loop head, and coverage is still Partial until a pass finishes every discovered path. `project_provider_yields_between_dated_rollouts_and_converges_without_loss` pinned the one-rollout-per-pass step with `max_new_bytes: u64::MAX`, so it asserted a bound the byte budget was never allowed to express. It now sizes its newest rollout past one capture window and asserts the yield that has a cursor: pass 0 stops inside that rollout, pass 1 resumes it and finishes the two behind it. Its convergence, no-loss and no-duplication assertions are unchanged. Co-Authored-By: Claude Fable 5.1 --- .../src/runtime/hosts/codex/tests.rs | 82 ++++++++++++++----- .../src/runtime/ingest/project_provider.rs | 15 ++-- 2 files changed, 71 insertions(+), 26 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index bdbd6126fa..195997fa0e 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -799,20 +799,32 @@ mod goal_event_tests { ); } + /// A pass yields on the source the capture window left mid-file, not on + /// every source it finishes. + /// + /// `MAX_CAPTURE_WINDOW` is the cooperative bound that owns a resumable + /// cursor: it stops inside one rollout and the next pass resumes at that + /// byte offset. Yielding once per *finished* rollout has no such cursor, so + /// the discovery frontier can never commit and every later pass re-reads + /// every rollout it already exhausted, which is quadratic in the rollouts a + /// project has. The profile-scope loop has always been bounded this way. #[tokio::test] - async fn project_provider_yields_between_dated_rollouts_and_converges_without_loss() { + async fn project_provider_yields_on_a_deferred_rollout_and_converges_without_loss() { crate::runtime::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); let temp = tempfile::tempdir().unwrap(); let home = temp.path().canonicalize().unwrap(); let project = home.join("project"); std::fs::create_dir_all(&project).unwrap(); + // The newest rollout alone exceeds one capture window, so pass 0 must + // stop inside it; the two behind it fit in the pass that finishes it. + let deferring_messages = + crate::runtime::jsonl_observation_admission::MAX_CAPTURE_WINDOW + 44; let rollouts = [ - (("2026", "09", "03"), "session-newest"), - (("2026", "09", "02"), "session-middle"), - (("2026", "09", "01"), "session-oldest"), + (("2026", "09", "03"), "session-newest", deferring_messages), + (("2026", "09", "02"), "session-middle", 3), + (("2026", "09", "01"), "session-oldest", 3), ]; - let messages_per_rollout = 3; - for (date, session_id) in rollouts { + for (date, session_id, messages_per_rollout) in rollouts { let directory = home .join(".codex/sessions") .join(date.0) @@ -826,7 +838,7 @@ mod goal_event_tests { })]; lines.extend((0..messages_per_rollout).map(|ordinal| { json!({ - "timestamp": format!("{}-{}-{}T12:00:0{}.000Z", date.0, date.1, date.2, ordinal + 1), + "timestamp": format!("{}-{}-{}T12:00:01.{:03}Z", date.0, date.1, date.2, ordinal), "type": "event_msg", "payload": { "type": "user_message", @@ -852,7 +864,22 @@ mod goal_event_tests { }; let admission = MemoryHostAdmission::default(); let cancellation = ObservationCancellation::default(); - let failure_ceiling = rollouts.len().saturating_add(1); + // Pass 0 stops inside the newest rollout at the capture window; pass 1 + // resumes it from the stored byte offset and, still inside its byte + // budget, finishes the two rollouts behind it. + let expected_passes: [(&[&str], usize); 2] = [ + ( + &["session-newest"], + crate::runtime::jsonl_observation_admission::MAX_CAPTURE_WINDOW, + ), + ( + &["session-newest", "session-middle", "session-oldest"], + (deferring_messages + 1) + - crate::runtime::jsonl_observation_admission::MAX_CAPTURE_WINDOW + + 8, + ), + ]; + let failure_ceiling = expected_passes.len().saturating_add(1); let mut completed_after = None; for pass_index in 0..failure_ceiling { @@ -877,20 +904,30 @@ mod goal_event_tests { let after = admission.observations(); let admitted_this_pass = &after[before..]; assert!( - pass_index < rollouts.len(), + pass_index < expected_passes.len(), "coverage did not complete within the fixture-derived ceiling" ); - let expected_session = rollouts[pass_index].1; + let (expected_sessions, expected_admitted) = expected_passes[pass_index]; assert_eq!( admitted_this_pass.len(), - messages_per_rollout + 1, - "pass {pass_index} must finish exactly {expected_session}" + expected_admitted, + "pass {pass_index} must admit one capture window of {expected_sessions:?}" + ); + assert_eq!( + admitted_this_pass + .iter() + .map(|stored| { + let envelope: CanonicalObservationEnvelopeV1 = + serde_json::from_value(stored.observation().payload().clone()).unwrap(); + envelope.relations().session_id().as_str().to_owned() + }) + .collect::>(), + expected_sessions + .iter() + .map(|session_id| (*session_id).to_owned()) + .collect::>(), + "pass {pass_index} admitted the wrong rollouts" ); - assert!(admitted_this_pass.iter().all(|stored| { - let envelope: CanonicalObservationEnvelopeV1 = - serde_json::from_value(stored.observation().payload().clone()).unwrap(); - envelope.relations().session_id().as_str() == expected_session - })); let coverage = read_host_provider_coverage(&admission, &scope, "codex") .await @@ -902,11 +939,14 @@ mod goal_event_tests { assert_eq!(coverage, Some(HostProviderCoverage::Partial)); } - assert_eq!(completed_after, Some(rollouts.len())); + assert_eq!(completed_after, Some(expected_passes.len())); let observations = admission.observations(); assert_eq!( observations.len(), - rollouts.len() * (messages_per_rollout + 1) + rollouts + .iter() + .map(|(_, _, messages)| messages + 1) + .sum::() ); let envelopes = observations .iter() @@ -925,10 +965,10 @@ mod goal_event_tests { admitted_sessions.iter().cloned().collect::>(), rollouts .iter() - .map(|(_, session_id)| (*session_id).to_owned()) + .map(|(_, session_id, _)| (*session_id).to_owned()) .collect::>() ); - for (_, session_id) in rollouts { + for (_, session_id, messages_per_rollout) in rollouts { assert_eq!( admitted_sessions .iter() diff --git a/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs b/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs index e6b7f9e605..24f02e9a60 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs @@ -260,7 +260,7 @@ impl<'a> ProjectProviderRun<'a> { let mut deferred = discovery.is_truncated(); let mut frontier_committable = true; let mut outcome = ProviderRunOutcome::bounded(TranscriptIngestStats::default(), 0, false); - for (path_index, path) in discovery.paths.iter().enumerate() { + for path in &discovery.paths { if remaining == 0 { deferred = true; frontier_committable = false; @@ -286,10 +286,15 @@ impl<'a> ProjectProviderRun<'a> { frontier_committable &= !progress.source_deferred && progress.bytes_consumed <= remaining; remaining = remaining.saturating_sub(progress.bytes_consumed); - if progress.bytes_consumed > 0 - && (progress.source_deferred - || path_index.saturating_add(1) < discovery.paths.len()) - { + // Only a source the admission left mid-window ends the + // pass: it owns the next one, and no discovery frontier may + // commit past it. An exhausted source must not, or a pass + // admits at most one source however much budget is left, + // never commits a frontier, and the next pass rediscovers + // and re-reads every source it already finished. The byte + // budget above is the pass bound here, exactly as it is in + // the profile-scope loop. + if progress.source_deferred { deferred = true; frontier_committable = false; break; From ba881a17a57f49b56d185d736d4e270bb683b703 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:49:17 +0000 Subject: [PATCH 07/18] fix(session-sync): gate the import's history stage on history alone With the ingest pass no longer quadratic, the same import test failed on a new terminal instead: `termination: "failed"`, `failure_codes: ["session_history_not_current"]`, every stat zero, in 5.3 s. `await_import_history` is stage one of a two-stage import gate. It waits for both historical ingest workers through `wake_history_and_wait_until_idle`, and then also required both projection serving states to be `Current` and both projection stores to hold no pending rows. It does not wait for any of that. Stage two, `await_import_projection`, does: it wakes the projection workers, waits for them, and re-checks those same four conditions plus backlog and availability. But it only runs when stage one reported coverage complete with no failure codes, so the moment stage one read a projection that had not drained yet, the stage that exists to wait for it was skipped and the import reported a history failure for a history that was current. A slow ingest hid this: the projection always drained inside the 20 s stage one spent waiting on history, so the extra conditions were incidentally true by the time they were read. Making the ingest fast is what exposed the ordering. Check what the stage's name and its `session_history_not_current` failure code claim, and leave projection currency to the stage that waits for it. Stage two's conditions are a strict superset, so nothing stops being verified; a projection that genuinely never converges still terminates the import, through `session_temporal_projection_not_current`. Verification: the previously failing `completed_session_import_immediately_searches_canonical_message` passes 6/6 under `taskset -c 0-3`, three of them against four competing CPU hogs, in 14-16 s against its 30 s deadline (was 42-45 s and failing). Its whole module 6/6, `tracedecay-sessions --lib` 569/569, `tracedecay-session-runtime --lib` 113/113, `session_suite` 448/448, `transcript_ingest_suite` 190/190. Co-Authored-By: Claude Fable 5.1 --- .../src/session_sync.rs | 25 ++++++------------- .../src/session_sync/work.rs | 5 +--- 2 files changed, 9 insertions(+), 21 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_sync.rs b/crates/tracedecay-session-runtime/src/session_sync.rs index adfb143bc0..57125baf99 100644 --- a/crates/tracedecay-session-runtime/src/session_sync.rs +++ b/crates/tracedecay-session-runtime/src/session_sync.rs @@ -638,7 +638,6 @@ impl DaemonSessionSyncService { async fn await_import_history( &self, context: &SessionSyncProjectContext, - project_sessions: &RegisteredGlobalDbLeaseV1, request: &SessionSyncRequestV1, ) -> Result< crate::session_temporal_refresh_scheduler::history::SessionHistoricalIngestProgress, @@ -673,22 +672,14 @@ impl DaemonSessionSyncService { return Err(Some(interruption)); } }; - if let (Some(project), Some(user)) = settled - && matches!( - context.project_refresh.serving_status().state, - SessionProjectionServingState::Current - ) - && matches!( - context.user_refresh.serving_status().state, - SessionProjectionServingState::Current - ) - && self - .projection_store_is_current(project_sessions, request) - .await? - && self - .projection_store_is_current(&context.user_sessions, request) - .await? - { + // Only the historical frontier is decided here. Projection currency is + // `await_import_projection`'s gate, which waits for the projection + // workers and then re-checks these same serving states and stores. This + // gate does not wait for them, so asserting them here reports + // `session_history_not_current` for a history that is current and whose + // projection has simply not drained yet, and skips the stage that would + // have waited for it. + if let (Some(project), Some(user)) = settled { Ok( crate::session_temporal_refresh_scheduler::history::SessionHistoricalIngestProgress { stats: project.stats.merge(user.stats), diff --git a/crates/tracedecay-session-runtime/src/session_sync/work.rs b/crates/tracedecay-session-runtime/src/session_sync/work.rs index 4b2ed150ce..80b21f6e6d 100644 --- a/crates/tracedecay-session-runtime/src/session_sync/work.rs +++ b/crates/tracedecay-session-runtime/src/session_sync/work.rs @@ -491,10 +491,7 @@ impl SessionSyncProjectContext { request: &SessionSyncRequestV1, project_sessions: RegisteredGlobalDbLeaseV1, ) -> SessionSyncWorkResult { - let history = match service - .await_import_history(self, &project_sessions, request) - .await - { + let history = match service.await_import_history(self, request).await { Ok(progress) => Some(progress), Err(Some(interruption)) => { return SessionSyncWorkResult::Interrupted(interruption); From 656b5328e75a9d73884900e90a1a2b506ab00265 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:55:31 +0000 Subject: [PATCH 08/18] test(code-index): fence query-claim tests against the pass tail `concurrent_query_admissions_claim_one_pending_wake_before_worker_coalescing` failed in CI with zero of eight admissions accepted. The owner's worker releases the background admission halfway through a pass and drops its `reconcile_pass` guard before the branches that call `note_worker_continuation`, so a settled-looking owner can still stamp `BusyFollowUp` into the coalesced pending-wake slot. The test's raw `acquire_owned` returned at that mid-pass release, and the tail then refilled the slot the test had just emptied, so every concurrent `PendingWakeClaimV1::claim` saw it occupied and declined. Instrumenting the slot under contention reproduced it and named the stamper: `trigger=5` (`BusyFollowUp`) with the pass still in flight and all eight verdicts `Unavailable`. Take the admission through `quiesced_background_reconcile_admission` so no new pass can start, then clear the slot until it survives a quiet window, which is the proof `wait_for_settled_owner` cannot give while the guard drops early. The two query-claim-gate tests get the same fence: there a tail's stamp declines the request before the gate and hangs `wait_for_query_claim`. Co-Authored-By: Claude Fable 5.1 --- .../src/code_index_scheduler/tests/mod.rs | 31 ++++++++++++ .../code_index_scheduler/tests/reconcile.rs | 50 +++++++++++-------- 2 files changed, 61 insertions(+), 20 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs index 972169e649..c82155ff9b 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs @@ -1334,6 +1334,37 @@ async fn settled_owner_with_idle_admission( } } +/// Empty the coalesced pending-wake slot and prove the owner's pass tail is +/// done disturbing it. +/// +/// The caller must already hold the single background admission, so no further +/// pass can start. One pass can still be finishing: the worker releases that +/// admission halfway through its body and drops its `reconcile_pass` guard +/// before the branches that call `note_worker_continuation`, so both +/// `reconcile_in_progress` and the slot read quiet while the tail is still +/// about to stamp `BusyFollowUp` into it. [`wait_for_settled_owner`] samples +/// exactly those two, so it cannot see that tail. With the admission held the +/// tail is finite and unrepeatable, so clearing until the slot survives a quiet +/// window is the proof the settle cannot give. +async fn clear_pending_wake_until_quiet( + registry: &CodeIndexSchedulerRegistryV1, + scope: &tracedecay_contracts::ResolvedScope, +) { + let deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING; + loop { + registry.clear_pending_wake_for_scope(scope).await; + tokio::time::sleep(Duration::from_millis(10)).await; + if registry.pending_wake_micros_for_scope(scope).await == Some(0) { + return; + } + assert!( + Instant::now() <= deadline, + "the pending-wake slot for {:?} never stayed empty", + scope.worktree_id + ); + } +} + const CALLER_STAR: usize = 2_000; const CALLER_STAR_FILES: usize = 8; const CALLER_PAGE: u32 = 10; 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 e79be3c75c..8d06ded020 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 @@ -24,18 +24,18 @@ use tracedecay_runtime_core::resident_memory::{ use super::{ ALPHA_LIB_V1, GitFixture, RETAINED_REVISION_0, SERVING_SEAT_FAILURE_CEILING, - advance_pointer_to_unseated_successor, application_context, committed_capture_corpus_files, - core_search_request, drain_clone_backfill, git, git_stdout, mounted_core_query_worktree, - mounted_core_query_worktree_with_one_permit, published, query_authority, query_meta, - quiesced_background_reconcile_admission, replace_scheduler_chunker_revision, - replace_scheduler_policy_revision, rewrite_active_rust_extractor_revision, - rewrite_preserving_stat, scheduler, scheduler_with_policy, served_lexical_texts, - settled_owner_with_idle_admission, test_project_id, wait_for_dashboard_ready, - wait_for_event_to_ready, wait_for_generation_change, wait_for_initial_generation, - wait_for_live_complete_generation, wait_for_live_complete_generation_by_polling, - wait_for_queryable_text_generation, wait_for_queryable_text_generation_change, - wait_for_queryable_text_generation_id, wait_for_quiescent_owner_pass, wait_for_settled_owner, - wait_until_serving_seat, write, + advance_pointer_to_unseated_successor, application_context, clear_pending_wake_until_quiet, + committed_capture_corpus_files, core_search_request, drain_clone_backfill, git, git_stdout, + mounted_core_query_worktree, mounted_core_query_worktree_with_one_permit, published, + query_authority, query_meta, quiesced_background_reconcile_admission, + replace_scheduler_chunker_revision, replace_scheduler_policy_revision, + rewrite_active_rust_extractor_revision, rewrite_preserving_stat, scheduler, + scheduler_with_policy, served_lexical_texts, settled_owner_with_idle_admission, + test_project_id, wait_for_dashboard_ready, wait_for_event_to_ready, wait_for_generation_change, + wait_for_initial_generation, wait_for_live_complete_generation, + wait_for_live_complete_generation_by_polling, wait_for_queryable_text_generation, + wait_for_queryable_text_generation_change, wait_for_queryable_text_generation_id, + wait_for_quiescent_owner_pass, wait_for_settled_owner, wait_until_serving_seat, write, }; use crate::{ code_index::{ @@ -5221,11 +5221,13 @@ async fn concurrent_query_admissions_claim_one_pending_wake_before_worker_coales // about simultaneous query admissions, so finish that independent // production journey before establishing the empty-slot precondition. drain_clone_backfill(®istry, fixture.path()).await; - let admission = registry - .background_reconcile_admission() - .acquire_owned() - .await - .expect("background reconcile admission"); + // Take the shared admission first, through the helper that also waits out + // an in-flight pass: from here no new pass can start, so the quiet window + // established below stays quiet. A raw `acquire_owned` returns the instant + // a running pass releases the admission mid-body, and that pass's tail then + // stamps `BusyFollowUp` over the empty slot this test set up, which makes + // every claim below decline. + let admission = quiesced_background_reconcile_admission(®istry, fixture.path()).await; let scheduler = { let mounted = registry.mounted.lock().await; Arc::clone( @@ -5235,8 +5237,11 @@ async fn concurrent_query_admissions_claim_one_pending_wake_before_worker_coales .scheduler, ) }; + // The tail of the pass the admission was taken from also publishes text + // owners, so empty the wake slot and prove it stays empty before clearing + // the generations this test needs absent. + clear_pending_wake_until_quiet(®istry, &scope).await; registry.clear_serving_generation_for_scope(&scope).await; - registry.clear_pending_wake_for_scope(&scope).await; let held = scheduler .lock() .expect("hold the scheduler as a rebuild would"); @@ -5461,7 +5466,10 @@ async fn foreign_wake_keeps_pending_arrival_when_query_claim_is_released() { let store = TempDir::new().expect("store root"); let (registry, scope) = mounted_core_query_worktree_with_one_permit(&fixture, &store).await; let admission = quiesced_background_reconcile_admission(®istry, fixture.path()).await; - registry.clear_pending_wake_for_scope(&scope).await; + // A `BusyFollowUp` stamp from the finishing pass's tail would make the + // request below decline before it ever reaches the claim gate, and + // `wait_for_query_claim` would then hang instead of failing. + clear_pending_wake_until_quiet(®istry, &scope).await; registry.install_query_claim_gate(&scope); let request = { @@ -5511,7 +5519,9 @@ async fn foreign_wake_arriving_during_query_claim_drop_is_retained() { // reaches the claim gate under test; settle the backfill first. drain_clone_backfill(®istry, fixture.path()).await; let admission = quiesced_background_reconcile_admission(®istry, fixture.path()).await; - registry.clear_pending_wake_for_scope(&scope).await; + // Same hang: a tail's `BusyFollowUp` stamp declines the request before the + // claim gate this test waits on. + clear_pending_wake_until_quiet(®istry, &scope).await; registry.install_query_claim_gate(&scope); registry.install_pending_wake_drop_gate(&scope).await; From b5a784cf32ea751b71f37e247e3d81354a29a3c2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 00:59:59 +0000 Subject: [PATCH 09/18] test(code-index): fence three more slot-emptiness preconditions Audit of the reconcile module for the same pattern: a single-permit registry that holds the background admission, clears the coalesced pending-wake slot, then asserts the slot stayed empty. Each is exposed to the same pass tail, which drops its `reconcile_pass` guard before its `note_worker_continuation` branches and can stamp `BusyFollowUp` after the clear. Co-Authored-By: Claude Fable 5.1 --- .../src/code_index_scheduler/tests/reconcile.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) 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 8d06ded020..445e165aea 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 @@ -2156,7 +2156,10 @@ async fn unchanged_git_watcher_probe_does_not_enqueue_authoritative_capture() { .acquire_owned() .await .expect("hold background reconcile admission"); - registry.clear_pending_wake_for_scope(&scope).await; + // The settle above cannot see a pass tail that has dropped its guard and + // not yet stamped its `BusyFollowUp` follow-up, so prove the slot stays + // empty before asserting that nothing queued a capture pass. + clear_pending_wake_until_quiet(®istry, &scope).await; assert_eq!( scheduler .lock() @@ -6949,7 +6952,9 @@ async fn text_freshness_query_during_owner_work_is_current_when_source_is_unchan .acquire_owned() .await .expect("hold background reconcile admission"); - registry.clear_pending_wake_for_scope(&scope).await; + // A pass tail that stamps `BusyFollowUp` after this clear would fail the + // "no wake" assertion below, so prove the empty slot holds. + clear_pending_wake_until_quiet(®istry, &scope).await; // Stand in for a worker pass re-observing an unchanged tree: in-progress, // scheduler mutex free, nothing moved on disk or in git. let owner_pass = registry @@ -9137,7 +9142,8 @@ async fn graph_off_remount_preserves_an_unhinted_source_reconcile() { .acquire_owned() .await .expect("hold worker after remount dequeue"); - registry.clear_pending_wake_for_scope(&scope).await; + // Same tail: its stamp would look like the unhinted edit's own wake below. + clear_pending_wake_until_quiet(®istry, &scope).await; fixture.edit("src/lib.rs", "pub fn beta() -> usize { 2 }\n"); git(fixture.path(), &["commit", "-qam", "unhinted remount edit"]); From 92713c1817022326c1a7e878f7d783f7ca889bf9 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 01:06:10 +0000 Subject: [PATCH 10/18] test(daemon): spell the seat wait in minutes for clippy Co-Authored-By: Claude Fable 5.1 --- .../code_index_reads/ignored_dependency_admission_tests.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs index d20e8879af..b29578464c 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs @@ -388,8 +388,8 @@ async fn latest( // Lightweight publication precedes complete-generation seating. Demand // that complete state before using its imports as admission evidence. The // seat is background work behind the scheduler mutex; under a loaded CI - // runner it has taken over 5 s, so the bound is generous. - tokio::time::timeout(Duration::from_secs(60), async { + // runner it has taken over 5 s, so the bound is a minute. + tokio::time::timeout(Duration::from_mins(1), async { loop { let _ = registry.latest_complete_fresh(project_root).await; if registry From e2c8b592cd6bcbae855ec20065f771206f601417 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 01:20:32 +0000 Subject: [PATCH 11/18] test(transport): assert reset on the first admitted observation After a physical restart the authority record is published before the daemon has opened the project, and the first HTTP observation in that window is a typed pre-admission `unavailable` with `retry: after_delay` (master run 35411306897, TRY 1 of `reset_required_survives_http_mcp_and_rust_sdk_across_restart`). Honour the directive as a production client would and assert on the first admitted reply. Co-Authored-By: Claude Fable 5.1 --- .../transport_boundaries.rs | 36 +++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs index 0c79a835e3..daca6a9303 100644 --- a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs +++ b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs @@ -105,6 +105,39 @@ fn http_mount(home: &Path) -> HttpMount { } } +/// Posts to the daemon's HTTP mount, repeating while the daemon answers with a +/// pre-admission problem whose own retry directive is `after_delay`. +/// +/// The authority record can be published before a restarted daemon has opened +/// the project, and the reply for that window is a typed, retryable +/// `unavailable`, not a verdict on the project. The first observation a +/// journey asserts on is the first one the daemon *admitted*, which is what a +/// production client that honours the directive sees. +fn post_application_once_admitted( + mount: &HttpMount, + project_id: &str, + route: &str, + body: &Value, +) -> (u16, Value) { + let deadline = Instant::now() + AUTHORITY_TIMEOUT; + loop { + let (status, payload) = post_application(mount, project_id, route, body, None); + let problem = [&payload, &payload["value"], &payload["data"]] + .into_iter() + .map(|candidate| &candidate["problem"]) + .find(|problem| problem.is_object()) + .filter(|problem| problem["terminality"] == "pre_admission") + .filter(|problem| problem["retry"] == "after_delay"); + match problem { + Some(problem) if Instant::now() < deadline => { + let millis = problem["retry_after_millis"].as_u64().unwrap_or(250); + std::thread::sleep(Duration::from_millis(millis)); + } + _ => return (status, payload), + } + } +} + /// Opens the exact project through the same daemon-owned route as a production /// CLI client and returns the identity the daemon admitted. HTTP cannot infer /// this identity locally: its route accepts only the daemon's public ID. @@ -617,12 +650,11 @@ fn reset_required_survives_http_mcp_and_rust_sdk_across_restart() { let storage_status_body = json!({ "include_details": false }); - let (http_status, http_body) = post_application( + let (http_status, http_body) = post_application_once_admitted( &mount, &identity, STORAGE_STATUS_ROUTE, &storage_status_body, - None, ); super::assert_reset_required( &problem_envelope(&http_body, "HTTP reset required"), From 36b0b780c929ecb6a0aecb9a80d52be2237e8f8f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 01:52:52 +0000 Subject: [PATCH 12/18] test(code-index): empty the wake slot under the held admission `dashboard_progress_does_not_wait_for_the_scheduler_mutex` still read `Verifying` on CI run 35412193695 after the settle: `refresh_in_flight` also reads the pending-wake slot, and the settled owner's pass tail can stamp `BusyFollowUp` into it after every settle check. With the single admission held that tail is finite, so clear the slot until it stays empty, the way the query-claim tests already do. Co-Authored-By: Claude Fable 5.1 --- .../src/code_index_scheduler/tests/reconcile.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) 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 445e165aea..138157469d 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 @@ -3706,14 +3706,26 @@ async fn dashboard_progress_does_not_wait_for_the_scheduler_mutex() { .path() .canonicalize() .expect("canonical fixture root"); - let (scheduler, progress_slot) = { + let (scheduler, progress_slot, scope) = { let mounted = registry.mounted.lock().await; let worktree = mounted.get(&canonical_root).expect("mounted worktree"); ( Arc::clone(&worktree.scheduler), Arc::clone(&worktree.build_progress), + tracedecay_contracts::ResolvedScope::new( + test_project_id(), + worktree.repository_id.clone(), + worktree.worktree_id.clone(), + None, + ) + .expect("resolved scope"), ) }; + // `refresh_in_flight` also reads the pending-wake slot, and the settled + // owner's pass tail can still stamp `BusyFollowUp` into it after every + // settle check above (CI run 35412193695). With the admission held that + // tail is finite: empty the slot until it stays empty. + clear_pending_wake_until_quiet(®istry, &scope).await; let expected = tokio::time::timeout(Duration::from_secs(5), async { loop { if let Some(progress) = progress_slot.read().expect("progress slot").snapshot() { From 0d328fa0a9db50497249d489acc63988197e0694 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 02:05:17 +0000 Subject: [PATCH 13/18] test(code-index): fence pointer corruption behind the store lock `coalesced_publication_failure_preserves_the_scheduler_error_family` induces an owner publication failure by writing `{` over `active-code-generation-v1.json`, then asserts the coalesced follower sees the owner's publication error family. Master run 35411306897 failed it on try 1 in 0.64 s: the owner published successfully and returned an outcome instead of failing closed. CI's nextest profile has `flaky-result = "fail"`, so the try-2 pass still reported red. Every production writer of that pointer reads it, edits it in memory and renames a temporary over it while holding the exclusive generation-store lock. The test wrote the file with no lock at all, so its corruption raced an in-flight read-modify-write whose rename restored a valid pointer; the owner's `validate_serving_is_active` then read an intact generation and published. An ordered trace of failing runs named the racer. The last event before the corruption is always a text-artifact mutation about to rename its edited pointer, and the pointer the owner reads afterwards is that same valid pointer - one run even observed the restore from the test thread, `on_disk_len=Ok(1753)` immediately after `hold.release()`. That mutation runs in the background pass tail, after the pass released the background admission permit this owner then took ("release the background admission permit before HeadOpening / graph work", registry/mount.rs), so neither the held admission nor the held scheduler mutex proves the store is quiet - the same pass-tail exposure fenced in 656b5328e7 and b5a784cf32. Corrupt under the store lock instead. Being granted it proves no writer is mid-transaction, and a writer that starts after it is released reads the corruption under the lock and refuses rather than overwriting it. The assertions are unchanged: the owner and the follower must both fail with `Production(Publication(_))`. Under `taskset -c 0,1` the target test reproduced 3/48, 5/48 and 7/60 before with 6 concurrent copies, 0/40 serially. After: 0/40 serially, 0/60 across 10 rounds of 6 concurrent copies, and the whole `code_index_ignored_dependencies_test::` module passes 21/21. Co-Authored-By: Claude Fable 5.1 --- .../flight_tests.rs | 35 +++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs index 2cec3b2c8c..bc1388df7f 100644 --- a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs +++ b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs @@ -283,8 +283,39 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() { &fixture.path().canonicalize().expect("canonical fixture"), ); let pointer_path = scoped_store.join("active-code-generation-v1.json"); - let pointer_bytes = std::fs::read(&pointer_path).expect("read active pointer"); - std::fs::write(&pointer_path, b"{").expect("corrupt active pointer"); + // Every production writer of the active pointer reads it, edits it in + // memory and renames a temporary over it while holding the exclusive + // generation-store lock. Corrupting the file without that lock races an + // in-flight read-modify-write whose rename then restores a valid pointer, + // and this owner publishes instead of failing closed. The racer is the + // background pass tail: it releases the background admission permit this + // owner then takes (registry/mount.rs, "release the background admission + // permit before HeadOpening / graph work") and keeps attaching the + // generation's text artifact afterwards, so neither the held admission + // nor the held scheduler mutex proves the store is quiet. Taking the + // store lock does: being granted it means no writer is mid-transaction, + // and any writer that starts after it is released reads the corruption + // under the lock and refuses instead of overwriting it. + let pointer_bytes = { + use tracedecay_code_index_retention::code_index_generations::try_acquire_code_generation_store_lock; + + let store_lock = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Some(lock) = try_acquire_code_generation_store_lock(&scoped_store) + .expect("generation store lock") + { + break lock; + } + tokio::time::sleep(Duration::from_millis(2)).await; + } + }) + .await + .expect("no generation-store writer is mid-transaction"); + let pointer_bytes = std::fs::read(&pointer_path).expect("read active pointer"); + std::fs::write(&pointer_path, b"{").expect("corrupt active pointer"); + drop(store_lock); + pointer_bytes + }; owner_control.release(); hold.release(); From d2b57e4ad70fc0dbd70a01cc953f8434e469b6ba Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 02:16:05 +0000 Subject: [PATCH 14/18] test(daemon): await a stopped predecessor before respawning `spawn_tracedecay_daemon_process` refused to start when the profile's endpoint was still connectable, asserting instantaneously. Stopping a daemon is asynchronous with respect to its endpoint: `TestChildProcess` kills and reaps the PID it spawned, but the kernel keeps the listening socket alive while any duplicate of that descriptor survives, including one a subprocess inherited across `fork` and still holds because it has not reached its own `exec` yet. A reaped PID is therefore not proof that the endpoint stopped accepting. `init_project_fixture` journeys walk straight into that tail: spawn a daemon, run `tracedecay init`, drop the daemon, spawn another one. On a loaded runner the teardown tail outlives the reap and the second spawn reported an ordinary teardown as a live daemon, which is how `branch_search_serves_a_committed_generation_behind_dirty_worktree_state` and its `branch_add_...` sibling went red. Wait a bounded time for the endpoint to stop accepting instead. A predecessor that keeps accepting past the timeout still fails with the same refusal, so a genuine daemon leak is still reported. Co-Authored-By: Claude Fable 5.1 --- crates/tracedecay/tests/common/mod.rs | 54 +++++++++++++++++++++------ 1 file changed, 43 insertions(+), 11 deletions(-) diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs index 8ea4bdd407..28bbc596c4 100644 --- a/crates/tracedecay/tests/common/mod.rs +++ b/crates/tracedecay/tests/common/mod.rs @@ -1089,6 +1089,13 @@ pub fn spawn_tracedecay_daemon_with( spawn_tracedecay_daemon_process(&home, &binary, configure) } +/// How long a replacement daemon waits for a stopped predecessor's endpoint to +/// stop accepting before reporting it as still live. +/// +/// Generous on purpose: the wait only costs time when a predecessor is +/// genuinely still reachable, and a real leak still fails rather than hangs. +const PREDECESSOR_DAEMON_VACATE_TIMEOUT: Duration = Duration::from_secs(10); + fn spawn_tracedecay_daemon_process( home: &Path, binary: &Path, @@ -1111,17 +1118,42 @@ fn spawn_tracedecay_daemon_process( }) .is_some_and(|address| TcpStream::connect(address).is_ok()) }; - #[cfg(unix)] - assert!( - std::os::unix::net::UnixStream::connect(&socket_path).is_err(), - "refusing to replace a live test daemon at {}", - socket_path.display() - ); - #[cfg(not(unix))] - assert!( - !portable_daemon_connectable(), - "refusing to replace a live test daemon recorded at {}", - authority_path.display() + // Stopping a predecessor daemon is asynchronous with respect to its + // endpoint: `kill` plus `wait` reaps the PID the harness spawned, but the + // kernel keeps the listening socket alive while *any* duplicate of that + // descriptor survives, including one a subprocess inherited across `fork` + // and still holds because it has not reached its own `exec` yet. Asserting + // instantaneously therefore reports an ordinary teardown tail as a live + // daemon, which is what `init_project_fixture` journeys (spawn, init, drop, + // spawn again) hit on a loaded runner. Wait a bounded time for the endpoint + // to stop accepting; a daemon that keeps accepting still fails with the + // same refusal. + poll_until( + Instant::now() + PREDECESSOR_DAEMON_VACATE_TIMEOUT, + Duration::from_millis(25), + || { + #[cfg(unix)] + let live = std::os::unix::net::UnixStream::connect(&socket_path).is_ok(); + #[cfg(not(unix))] + let live = portable_daemon_connectable(); + (!live).then_some(()) + }, + || { + #[cfg(unix)] + { + format!( + "refusing to replace a live test daemon at {}", + socket_path.display() + ) + } + #[cfg(not(unix))] + { + format!( + "refusing to replace a live test daemon recorded at {}", + authority_path.display() + ) + } + }, ); let mut command = Command::new(binary); From b725b7a0c6b24266c252e6a06c4a0c238ce3d78b Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 03:36:11 +0000 Subject: [PATCH 15/18] test(lcm): retry window reads through typed staleness `preserved_profile_lcm_discovery_converges_without_blocking_retrieval` failed both nextest tries of run 35414007809 at ~21 s, on `12-hour direct-user lcm_grep must return parsed session identities` with `status: "stale", count: 0`. Reproduced 4/8 under `taskset -c 0-3` against four CPU hogs; every local failure carried that same typed `stale`, never a wrong-window hit and never a thin corpus. The journey asserts that retrieval stays admitted while convergence runs, so a window the projection has not caught up to is answered with typed staleness rather than blocked. That makes any single read racy against background ingest: `wait_for_preserved_discovery` had already been served non-empty hits for this exact query and window, and the projection fell behind again while the test walked the assertions between that gate and the final grep. Reverting `48cc9d5b59` (bound a project ingest pass by bytes) locally takes the failure to 0/8 and the run from 21-46 s to 99-129 s: the quadratic re-read it removed was pacing the test, so discovery converged only once ingest was nearly done and no lag window was left to observe. It is not a regression, it is the same journey run against an ingest that is no longer quadratic. Reverting `ba881a17a5` (5/8) or `7f76433808` (5/8) changes nothing; this journey never drives the import gate. `wait_for_pre_window_search` already looped on exactly this typed staleness, for one of the three window reads. Generalize it into `converged_window_read` and route all three through it, bounded by the same `CONVERGENCE_WAIT` the discovery gate uses. The elapsed time it returns is the served call alone, so `SEARCH_BUDGET` still measures one answer rather than the wait in front of it, and every existing window, budget and payload assertion is unchanged. Verification: 20/20 under `taskset -c 0-3` with four CPU hogs (was 4/8 failing), the module once, `cargo clippy -p tracedecay --all-targets --locked -- -D warnings` and `cargo fmt --all -- --check` clean. Co-Authored-By: Claude Fable 5.1 --- .../lcm_preserved_profile_journey_test.rs | 75 +++++++++++-------- 1 file changed, 42 insertions(+), 33 deletions(-) diff --git a/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs b/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs index bb17b3c310..8ad630e212 100644 --- a/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs @@ -531,40 +531,37 @@ fn assert_window_side(label: &str, returned: &[String], in_window: bool, payload } } -/// The complement of the 12-hour window, once temporal convergence can serve -/// it. +/// One window read, retried until the projection can serve that window. /// -/// A `stale` outcome is the projection reporting it has not caught up to those -/// generations yet; reading absence out of it would let the window assertions -/// pass on lag instead of on a window decision. -async fn wait_for_pre_window_search( +/// Retrieval is never blocked on convergence, which is the admission this +/// journey asserts: a window the projection has not caught up to is answered +/// with typed staleness instead of waiting for it. So any single read here can +/// land in a lag window that background ingest opened after an earlier read of +/// the same window was served, and reading absence out of that would let the +/// window assertions pass on lag instead of on a window decision. Every window +/// read retries under the same convergence budget the discovery wait uses. +/// +/// The returned elapsed time is the served call alone, so the product search +/// budget still measures one answer and not the wait in front of it. +async fn converged_window_read( harness: &ProductionProjectCompositionHarnessV1, project: &Path, - origin: i64, - since: i64, -) -> Value { + label: &str, + tool: &str, + arguments: Value, +) -> (Duration, Value) { let deadline = Instant::now() + CONVERGENCE_WAIT; loop { - let payload = answered( - harness, - project, - "tracedecay_message_search", - json!({ - "query": DIRECT_USER_QUERY, - "message_type": "direct_user", - "since": origin, - "until": since - 1, - "limit": SESSION_REPLAYS, - "format": "json", - }), - ) - .await; - if payload["outcome"] != json!("stale") { - return payload; + let (elapsed, response) = timed_call(harness, project, tool, arguments.clone()).await; + let payload = retained_payload(&resolved(harness, project, tool, response).await); + // `lcm_grep` names typed staleness on `status`, `message_search` on + // `outcome`; a served page carries "stale" on neither. + if payload["status"] != json!("stale") && payload["outcome"] != json!("stale") { + return (elapsed, payload); } assert!( Instant::now() < deadline, - "the pre-window search never left typed staleness: {payload}" + "{label} never left typed staleness: {payload}" ); tokio::time::sleep(Duration::from_millis(200)).await; } @@ -765,9 +762,10 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval() "known TraceDecay worktree must return its correlated session: {sessions_for}" ); - let (search_elapsed, search) = timed_call( + let (search_elapsed, search_payload) = converged_window_read( &harness, &project, + "direct-user 12-hour message_search", "tracedecay_message_search", json!({ "query": DIRECT_USER_QUERY, @@ -783,8 +781,6 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval() search_elapsed, SEARCH_BUDGET, ); - let search_payload = - retained_payload(&resolved(&harness, &project, "tracedecay_message_search", search).await); assert_ne!( search_payload["status"], json!("error"), @@ -807,7 +803,21 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval() // The exclusion above must be a window decision, not an empty corpus: the // complementary query over the same span returns exactly the replays the // 12-hour window drops, every one of them. - let excluded_search = wait_for_pre_window_search(&harness, &project, origin, since).await; + let (_, excluded_search) = converged_window_read( + &harness, + &project, + "pre-window direct-user search", + "tracedecay_message_search", + json!({ + "query": DIRECT_USER_QUERY, + "message_type": "direct_user", + "since": origin, + "until": since - 1, + "limit": SESSION_REPLAYS, + "format": "json", + }), + ) + .await; let excluded_sessions = message_hit_session_ids(&excluded_search); assert_window_side( "pre-window direct-user search", @@ -823,9 +833,10 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval() ); } - let (grep_elapsed, grep) = timed_call( + let (grep_elapsed, grep_payload) = converged_window_read( &harness, &project, + "direct-user 12-hour lcm_grep", "tracedecay_lcm_grep", json!({ "query": DIRECT_USER_QUERY, @@ -837,8 +848,6 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval() ) .await; assert_under_budget("direct-user 12-hour lcm_grep", grep_elapsed, SEARCH_BUDGET); - let grep_payload = - retained_payload(&resolved(&harness, &project, "tracedecay_lcm_grep", grep).await); for hit in grep_hits(&grep_payload) { let snippet = hit["snippet"].as_str().unwrap_or_default(); assert!( From d8aaa0fa68363009a03803d509674be46e0ca7f1 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 03:47:56 +0000 Subject: [PATCH 16/18] fix(hooks): serialize Codex rollout admission per scope `production_codex_hook_ingest_survives_message_search_reopen` failed TRY 1 of run 35414007809 at 6 s with transcript ingest failed: authority_write_failed: submit observation batch: ... Invalid parameter name: observation repository provenance collision: retained provenance is not replayable `Invalid parameter name` is not a binding bug: `repository::support::invalid` builds `rusqlite::Error::InvalidParameterName` for every domain refusal this crate raises through a `rusqlite::Result`, so the text after it is the whole error. The refusal is `verify_observation_authority`'s, and it is a replay verification: a write whose observation id already exists must read back the rows the first apply wrote. Root cause: the MCP hook route (`admit_codex_project_rollouts`) and the daemon's project catch-up sweep (`ingest::project_provider::run_codex`) both reach `try_admit_codex_jsonl_observations` for the same rollout under the same scope, with no mutual exclusion, and both read the source cursor before they write. Interleaved, the loser reads a cursor the winner has not published yet, re-reads frames the winner has already committed, and re-submits the same observation ids with its own independently captured repository provenance. `repository_replay_anchor` normalizes only a later local capture clock, so a second capture that differs in any other way is refused, and the host reports a retryable infrastructure failure for what is a duplicate. Hold one gate across the cursor read and the admit for one (scope, rollout) pair. The loser then reads the advanced cursor, persists nothing, and returns the `resumed` replay. The gate is per process; cross-process writers still meet at the store transaction, which is what the replay verification is for. That replay had no way to say it was durable. db2ee84366 closed exactly this for Cursor ("`CursorProjectionDrainStats::into_transcript_stats` dropped `exact_duplicates`") but the Codex half was never landed: `CodexJsonlAdmissionProgress` dropped `resumed` from the inner `JsonlObservationAdmissionProgress`, so `capture_codex_project` could not set `exact_duplicate` and every already-admitted rollout reported `accepted_for_replay` -- db2ee84366's own "terminal, non-retryable status that neither proves a commit nor invites a retry". Carry `resumed` through and tally it the way Cursor does: a pass is a duplicate only when it scanned at least one rollout and every one of them resumed with nothing new to persist. Evidence: instrumented at 6 concurrent copies under `taskset -c 0,1`, 12/12 runs reported `admission: accepted_for_replay, messages_upserted: 0` -- the catch-up won every time and the hook never described the rollout it had just proved durable. The test now rejects `accepted_for_replay` and accepts `committed` or `exact_duplicate`, the same bar db2ee84366 set for the Cursor acceptance test; 72 post-fix runs, none reported the vacuous status. Not fixed here, and not this race: the first `tracedecay_message_search` after the ingest can still answer `outcome: complete_zero` with `freshness.state: partial, generation_lag: 1` from a store that already holds the rows. The project store's search generation is published only by the background session-temporal refresh tick; no hook ingest route wakes it, and `tracedecay_session_refresh_begin` converges the profile scope, not the project store the search reads. Measured: converges within 500 ms of the failing read in every instrumented case. Verification, 6-way concurrent under `taskset -c 0,1` on a host shared with other build lanes, so the rates drift with load and only their shape is meaningful: baseline 22/30 failures; reverting 48cc9d5b59 17/30, ba881a17a5 22/30, 7f76433808 13/30, so none of the three opens the race and all three widen a pre-existing window; with this change 15/30 and later 30/42 at a higher host load, which is the same rate, not an improvement. What did change is categorical: every remaining failure is the search-generation lag above, and across 114 post-fix runs none reported an admission status other than `committed`/`exact_duplicate` and none hit a provenance collision. `session_search_test::` module 6/6, `tracedecay-mcp --lib` 367/367, `tracedecay-sessions --lib` 568/569 (the one failure, `codex_session_meta_prefix_is_decoded_once_across_consumers`, fails identically on the pristine tree and passes in isolation). Co-Authored-By: Claude Fable 5.1 --- .../src/handlers/hook_runtime/ingest.rs | 15 ++++++ .../handlers/hook_runtime/ingest/kernels.rs | 1 + .../src/runtime/hosts/codex/observation.rs | 52 ++++++++++++++++++- .../mcp_handler_test/session_search_test.rs | 20 ++++--- 4 files changed, 79 insertions(+), 9 deletions(-) diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs index 3cd2df6eee..846387e8ca 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs +++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs @@ -123,6 +123,8 @@ async fn admit_codex_project_rollouts( let mut budget = max_new_bytes; let mut deferred = false; let mut observations_committed = 0_u64; + let mut scanned = 0_u64; + let mut replayed = 0_u64; let mut paths = source.transcript_paths(project_root).into_iter().peekable(); while let Some(path) = paths.next() { let progress = @@ -138,6 +140,10 @@ async fn admit_codex_project_rollouts( .map_err(|error| map_transcript_ingest_error(&error))?; deferred |= progress.source_deferred; observations_committed = observations_committed.saturating_add(progress.frames_persisted); + scanned = scanned.saturating_add(1); + if progress.resumed && progress.frames_persisted == 0 { + replayed = replayed.saturating_add(1); + } if let Some(remaining) = budget.as_mut() { *remaining = remaining.saturating_sub(progress.bytes_consumed); if *remaining == 0 { @@ -148,6 +154,10 @@ async fn admit_codex_project_rollouts( } Ok(CodexRolloutAdmission { deferred, + // Only when every scanned rollout was a pure replay. One rollout with + // new frames makes the pass a commit, not a duplicate, and a pass that + // scanned nothing has nothing to call durable. + exact_duplicate: scanned > 0 && scanned == replayed, observations_committed, }) } @@ -157,6 +167,11 @@ async fn admit_codex_project_rollouts( /// with the project catch-up sweep, which can consume these rows first. pub(super) struct CodexRolloutAdmission { pub(super) deferred: bool, + /// Every scanned rollout resumed at its stored cursor with nothing new to + /// persist, so the transcript was already durable when this pass ran. + /// `JsonlObservationAdmissionProgress::resumed` is what separates that from + /// an empty source, so a first-ever scan is never called a duplicate. + pub(super) exact_duplicate: bool, pub(super) observations_committed: u64, } 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 d8918e65af..fbf97ef025 100644 --- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs +++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs @@ -428,6 +428,7 @@ async fn capture_codex_project( messages_upserted, source_deferred: admitted.deferred, observations_committed: admitted.observations_committed, + exact_duplicate: admitted.exact_duplicate, ..TranscriptCaptureOutcome::default() }) } diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index 66a8dc6bee..b02eecaab6 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError}; +use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError, Weak}; use tokio::sync::Notify; use tokio::sync::futures::OwnedNotified; @@ -233,6 +233,12 @@ pub struct CodexJsonlAdmissionProgress { pub frames_rejected_before_decode: u64, pub frames_refused: u64, pub frames_persisted: u64, + /// This pass resumed from a durable source cursor instead of opening the + /// rollout for the first time. With `frames_persisted == 0` it is the only + /// evidence that separates an already-admitted rollout from an empty one, + /// so a caller can report the replay as a duplicate rather than as a pass + /// that captured nothing. + pub resumed: bool, } /// Admit a Codex rollout for one exact project identity. @@ -651,6 +657,44 @@ async fn shared_session_meta_with_provenance( } } +/// Serializes the read-cursor-then-admit window for one rollout in one scope. +/// +/// The MCP hook route (`admit_codex_project_rollouts`) and the daemon's project +/// catch-up sweep (`ingest::project_provider::run_codex`) both reach +/// [`try_admit_codex_jsonl_observations`] for the same rollout under the same +/// scope, and both read the source cursor before they write. Interleaved, the +/// loser reads a cursor the winner has not published yet, re-reads frames the +/// winner has already committed, and re-submits the same observation ids with +/// its own independently captured repository provenance. The store's replay +/// verification refuses that second provenance as `observation repository +/// provenance collision`, which reaches the host as a retryable +/// `authority_write_failed` infrastructure error rather than the duplicate it +/// is. Serialized, the loser reads the advanced cursor, persists nothing, and +/// reports the `resumed` replay its caller renders as an exact duplicate. +/// +/// The gate is per process. Cross-process writers still meet at the store's +/// own transaction, which is what the replay verification is there for. +type CodexAdmissionGate = Arc>; +type CodexAdmissionGates = + Mutex>>>; + +static CODEX_ADMISSION_GATES: OnceLock = OnceLock::new(); + +/// ponytail: linear sweep of live gates per acquisition; keyed eviction if a +/// scope ever admits enough rollouts at once for the sweep to show up. +fn codex_admission_gate(scope: &ObservationScopeV1, path: &Path) -> CodexAdmissionGate { + let gates = CODEX_ADMISSION_GATES.get_or_init(|| Mutex::new(HashMap::new())); + let mut gates = gates.lock().unwrap_or_else(PoisonError::into_inner); + gates.retain(|_, gate| gate.strong_count() > 0); + let key = (scope.clone(), path.to_path_buf()); + if let Some(gate) = gates.get(&key).and_then(Weak::upgrade) { + return gate; + } + let gate = CodexAdmissionGate::new(tokio::sync::Mutex::new(())); + gates.insert(key, Arc::downgrade(&gate)); + gate +} + async fn try_admit_codex_jsonl_observations( path: &Path, admission_scope: CodexObservationAdmission<'_>, @@ -686,6 +730,11 @@ async fn try_admit_codex_jsonl_observations( cancellation, }; let scope = admission_scope.scope(); + // Held across the cursor read and the admit below: both are one pass over + // this rollout, and a peer that interleaves between them re-submits what + // this pass is about to commit. + let gate = codex_admission_gate(&scope, path); + let _admitting = gate.lock().await; if let Some(target) = admission .get_source_cursor(&ordinary_source, &scope) .await @@ -895,6 +944,7 @@ async fn admit_codex_jsonl_page( frames_rejected_before_decode: progress.frames_rejected_before_decode, frames_refused: progress.frames_refused, frames_persisted: progress.frames_persisted, + resumed: progress.resumed, }) } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs index c394a8c651..09534aa0b3 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs @@ -534,15 +534,19 @@ async fn production_codex_hook_ingest_survives_message_search_reopen() { ) .expect("production Codex hook ingest JSON"); assert_eq!(ingest["completed"], true, "{ingest}"); - // The composition's background Codex catch-up may admit the rollout - // before the hook pass reaches it, in which case the hook truthfully - // reports zero new bytes. Either path must leave the rollout durable and - // searchable, which the retrieval assertions below verify directly. + // The composition's background Codex catch-up may admit the rollout before + // the hook pass reaches it, in which case the hook persists no new frames + // and reports the rollout as an exact duplicate. Both terminals prove the + // transcript is durable; `accepted_for_replay` proves neither a commit nor + // a duplicate and must not be reported for a rollout that is on disk and + // admitted. Either path must also leave the rollout searchable, which the + // retrieval assertions below verify directly. assert!( - ingest["admission"]["status"] - .as_str() - .is_some_and(|status| status != "unavailable" && status != "unknown"), - "real Codex hook ingest was refused: {ingest}" + matches!( + ingest["admission"]["status"].as_str(), + Some("committed" | "exact_duplicate") + ), + "real Codex hook ingest proved neither a commit nor a duplicate: {ingest}" ); let initial = production_codex_message_search(&harness, &project).await; From e8fe3561ed3b4bb7066e4a4a1eadee7745a38a39 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 03:47:57 +0000 Subject: [PATCH 17/18] style(mcp-tests): sort pr_context facts by key `cargo clippy -p tracedecay --all-targets --features tracedecay/test-transport -- -D warnings` fails on `unnecessary_sort_by` in `pr_context_facts`, which blocks clippy for the whole `mcp_suite` target under the feature lens the suite's own tests need. CI's workspace lens does not build that target, so the lint has been latent since c8fbc38aae. Co-Authored-By: Claude Fable 5.1 --- .../tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs index fdf85beb3d..8acf0b25e8 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs @@ -3108,7 +3108,7 @@ fn symbol_facts(symbols: &Value) -> Value { } }) .collect::>(); - facts.sort_by(|left, right| left.to_string().cmp(&right.to_string())); + facts.sort_by_key(ToString::to_string); Value::Array(facts) } From d0dd529802a77a62df28868ac4229373b2c637a3 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 04:17:58 +0000 Subject: [PATCH 18/18] test(daemon): poll the dashboard read while the seat settles `read_only_project_binding_refuses_before_scheduler_mutation` polled `latest_complete_fresh` every 25 ms while waiting for `Fresh`. That read leaves a coalesced wake behind whenever it finds the worker holding the scheduler with an expired proof, so the poll re-armed a no-op pass faster than the ladder could settle: CI run 35419627712 sat in `Verifying` for the full minute and passed in 0.3 s on the retry. Poll the side-effect-free dashboard projection and read the generation once the ladder is `Fresh`. 60/60 as six concurrent copies on two cores. Co-Authored-By: Claude Fable 5.1 --- .../ignored_dependency_admission_tests.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs index b29578464c..946f6bf383 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs @@ -389,9 +389,15 @@ async fn latest( // that complete state before using its imports as admission evidence. The // seat is background work behind the scheduler mutex; under a loaded CI // runner it has taken over 5 s, so the bound is a minute. + // Poll the dashboard projection alone while the owner is busy: the + // query-admission read (`latest_complete_fresh`) leaves a coalesced wake + // behind whenever it finds the worker holding the scheduler with an + // expired proof, and polling it every 25 ms re-armed a no-op pass faster + // than the ladder could settle to `Fresh` (CI run 35419627712: one + // minute of `Verifying`, then 0.3 s on the retry). Read the generation + // only once the ladder has settled. tokio::time::timeout(Duration::from_mins(1), async { loop { - let _ = registry.latest_complete_fresh(project_root).await; if registry .dashboard_freshness(project_root) .await