ci: restore green master (gates, clippy, bench, all Linux partitions)

benchmark_data/runtime/tests/test_lifecycle.py

@@ -160,7 +160,9 @@ def test_dashboard_http_variants_remain_typed_failures(self) -> None:
                             url,
                             request_timeout=0.05,
                         ),
-                        readiness_timeout=0.2,
+                        # The stub takes a moment to bind on a loaded runner;
+                        # 0.2s left the probe stuck in dashboard_connect.
+                        readiness_timeout=2.0,
                         poll_interval=0.01,
                         termination_grace=0.05,
                     )

crates/tracedecay-agent-hosts/src/agents/codex/tests.rs

@@ -979,9 +979,34 @@ fn codex_preflight_reports_inactive_cache_without_interactive_guidance() {
     assert!(CodexIntegration.interactive_removal_guidance().is_none());
 }
 
+/// Install an executable `codex` on the host-program search path only.
+///
+/// Preparation is `Ready` exactly when Codex's own plugin CLI is present, so
+/// the outcome under test is a property of the environment, not of the host
+/// integration. CI runners carry no `codex` binary while a developer box
+/// usually does; pin it here instead of reading whichever the machine has.
+/// Only host program resolution sees this directory, the process `PATH` is
+/// untouched.
+fn install_fake_codex_cli(
+    dir: &Path,
+) -> tracedecay_runtime_core::config::HostProgramSearchPathGuard {
+    let binary = dir.join(format!("codex{}", std::env::consts::EXE_SUFFIX));
+    std::fs::write(&binary, "#!/bin/sh\nexit 0\n").unwrap();
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        let mut permissions = std::fs::metadata(&binary).unwrap().permissions();
+        permissions.set_mode(0o755);
+        std::fs::set_permissions(&binary, permissions).unwrap();
+    }
+    tracedecay_runtime_core::config::HostProgramSearchPathGuard::set(dir)
+}
+
 #[test]
 fn prepare_stages_the_source_and_returns_ready_for_cli_activation() {
     let home = tempfile::tempdir().unwrap();
+    let cli_dir = tempfile::tempdir().unwrap();
+    let _codex_cli = install_fake_codex_cli(cli_dir.path());
     // Pre-existing user config: preparation runs before the component
     // transaction stages `config.toml`, so it must not write there, hook
     // trust is recorded by activation, inside the rollback boundary.

crates/tracedecay-application/src/work/work_evidence_retrieval.rs

@@ -22,9 +22,9 @@ use tracedecay_contracts::{
 };
 use tracedecay_domain::{
     AuthorizationRevision, ComponentRevision, EphemeralSanitizedQueryViewV1, FreshnessVectorDigest,
-    HydrationStateV1, PrincipalId, QueryNormalizationRevision, RetrievalCursor, RetrievalGrainV1,
-    RetrievalRequest, RetrievalScope, SanitizerRevision, ScoreDomainId, SingleRootScopeV1,
-    VectorWatermark,
+    HydrationStateV1, PrincipalId, QueryNormalizationRevision, RetrievalBudget, RetrievalCursor,
+    RetrievalGrainV1, RetrievalRequest, RetrievalScope, SanitizerRevision, ScoreDomainId,
+    SingleRootScopeV1, VectorWatermark,
 };
 use tracedecay_query::retrieval::QueryAuthorityV1;
 use tracedecay_query::retrieval::evidence_lanes::{
@@ -144,9 +144,10 @@ impl WorkTaskSessionEvidenceRetrievalV1 {
     fn temporal_query(
         &self,
         request: &WorkTaskSessionRequestV1,
+        page_size: u32,
     ) -> Result<SessionTemporalQuery, WorkEvidenceHydrationErrorV1> {
-        let page_size = usize::try_from(request.page_size)
-            .map_err(|_| WorkEvidenceHydrationErrorV1::Unavailable)?;
+        let page_size =
+            usize::try_from(page_size).map_err(|_| WorkEvidenceHydrationErrorV1::Unavailable)?;
         let context_bytes = WORK_EVIDENCE_CONTEXT_BYTES;
         let execution_limits = ExecutionLimits {
             candidate_total_bytes: context_bytes as usize,
@@ -227,7 +228,11 @@ impl WorkTaskSessionPortV1 for WorkTaskSessionEvidenceRetrievalV1 {
                     request.source.clone(),
                 )
                 .map_err(|_| WorkEvidenceHydrationErrorV1::NotFoundOrNotAuthorized)?;
-                let temporal_query = self.temporal_query(&request)?;
+                let page_size = task_session_page_size(
+                    request.page_size,
+                    authority.profile().retrieval_budget,
+                )?;
+                let temporal_query = self.temporal_query(&request, page_size)?;
                 let retrieval_request = retrieval_request(context, &request, authority.as_ref())?;
                 let query = EphemeralSanitizedQueryViewV1::sanitize(
                     task_session_query_text(&request),
@@ -255,7 +260,7 @@ impl WorkTaskSessionPortV1 for WorkTaskSessionEvidenceRetrievalV1 {
                     context,
                     request: &request,
                     reauthorization,
-                    page_size: usize::try_from(request.page_size)
+                    page_size: usize::try_from(page_size)
                         .map_err(|_| WorkEvidenceHydrationErrorV1::Unavailable)?,
                     ranking_cursor,
                 };
@@ -364,17 +369,35 @@ fn map_reauthorization_error(
     }
 }
 
+/// The per-attempt TaskSession page size the mounted authority can actually
+/// serve.
+///
+/// `WorkEvidenceRetrieveRequestV1::page_size` bounds evidence *sources* in the
+/// Work page (validated up to `MAX_WORK_ROOTED_EVIDENCE_SOURCES_V1`), which is
+/// a different quantity from how many ranked session anchors one attempt may
+/// hydrate. Passing it through unclamped made every legal Work request above
+/// the mounted profile's hydration budget permanently `Unavailable` instead of
+/// a served page plus a continuation, so clamp to the budget here. Zero stays a
+/// refusal: no budget can serve it.
+fn task_session_page_size(
+    requested: u32,
+    budget: RetrievalBudget,
+) -> Result<u32, WorkEvidenceHydrationErrorV1> {
+    let page_size = requested
+        .min(budget.max_hydrated_results)
+        .min(budget.max_candidates_per_lane);
+    if page_size == 0 {
+        return Err(WorkEvidenceHydrationErrorV1::Unavailable);
+    }
+    Ok(page_size)
+}
+
 fn retrieval_request(
     context: &RequestContext,
     request: &WorkTaskSessionRequestV1,
     authority: &QueryAuthorityV1,
 ) -> Result<RetrievalRequest, WorkEvidenceHydrationErrorV1> {
-    if request.page_size == 0
-        || request.page_size > authority.profile().retrieval_budget.max_hydrated_results
-        || request.page_size > authority.profile().retrieval_budget.max_candidates_per_lane
-    {
-        return Err(WorkEvidenceHydrationErrorV1::Unavailable);
-    }
+    task_session_page_size(request.page_size, authority.profile().retrieval_budget)?;
     Ok(RetrievalRequest {
         principal: PrincipalId::new(context.actor().as_str())
             .map_err(|_| WorkEvidenceHydrationErrorV1::Unavailable)?,
@@ -890,7 +913,39 @@ mod unit_tests {
     use tracedecay_contracts::WorkEvidenceHydrationErrorV1;
     use tracedecay_contracts::retrieval::SessionRetrievalStructuralRefusalV1;
 
-    use super::{budget_hydration_refusal, cursor_manifest_hydration_refusal};
+    use tracedecay_domain::RetrievalBudget;
+
+    use super::{
+        budget_hydration_refusal, cursor_manifest_hydration_refusal, task_session_page_size,
+    };
+
+    const fn budget(max_candidates_per_lane: u32, max_hydrated_results: u32) -> RetrievalBudget {
+        RetrievalBudget {
+            max_candidates_per_lane,
+            max_fused_candidates: 32,
+            max_hydrated_results,
+            max_hydration_bytes: 65_536,
+            deadline_micros: None,
+        }
+    }
+
+    #[test]
+    fn task_session_page_size_clamps_to_the_mounted_budget() {
+        // The checked-in core query fallback policy. A legal Work evidence
+        // request (up to MAX_WORK_ROOTED_EVIDENCE_SOURCES_V1) must be served,
+        // not refused, when it asks for more than one attempt can hydrate.
+        assert_eq!(task_session_page_size(100, budget(32, 16)), Ok(16));
+        assert_eq!(task_session_page_size(8, budget(32, 16)), Ok(8));
+        assert_eq!(task_session_page_size(100, budget(4, 16)), Ok(4));
+        assert_eq!(
+            task_session_page_size(100, budget(0, 16)),
+            Err(WorkEvidenceHydrationErrorV1::Unavailable)
+        );
+        assert_eq!(
+            task_session_page_size(0, budget(32, 16)),
+            Err(WorkEvidenceHydrationErrorV1::Unavailable)
+        );
+    }
 
     #[test]
     fn task_session_structural_refusals_retain_exact_hydration_causes() {

crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs

@@ -47,10 +47,7 @@ pub fn try_acquire_code_generation_store_read_lock(
 ) -> Result<Option<CodeGenerationStoreLockV1>, CodeGenerationRetentionErrorV1> {
     let store_root = canonical_store_root(store_root)?;
     let lock = open_lock_file(&store_root.join(STORE_LOCK_FILE))?;
-    match lock
-        .try_lock_shared()
-        .map_err(std::io::Error::from)
-    {
+    match lock.try_lock_shared().map_err(std::io::Error::from) {
         Ok(()) => Ok(Some(CodeGenerationStoreLockV1 {
             file: lock,
             store_root,

crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs

@@ -143,6 +143,15 @@ impl SharedCodeIndexBytePoolV1 {
 /// every unpinned query and must not be evictable by cursor traffic over
 /// superseded generations.
 pub(super) const DECODED_GENERATION_CACHE_CAPACITY: usize = 4;
+/// The exact detail a `try_acquire_code_generation_store_lock` refusal carries.
+///
+/// The store lock is a bounded shared resource: a concurrent publication in
+/// the same store root holds it and releases it on its own. Both the producer
+/// below and
+/// [`CodeIndexSchedulerErrorV1::is_transient_capacity_failure`] read this one
+/// token, so the retry classification cannot drift from the refusal it names.
+pub(super) const CODE_GENERATION_STORE_ACTIVE_OWNER_DETAIL_V1: &str =
+    "code-generation store has an active owner";
 
 /// Whether one generation resolution may enter the single-flight sealed-decode.
 ///
@@ -2150,7 +2159,7 @@ impl CodeIndexAtomicPublicationPort for DaemonCodeIndexPublicationStoreV1 {
         };
         let _store_lock = try_acquire_code_generation_store_lock(store_root)
             .map_err(Self::unavailable)?
-            .ok_or_else(|| Self::unavailable("code-generation store has an active owner"))?;
+            .ok_or_else(|| Self::unavailable(CODE_GENERATION_STORE_ACTIVE_OWNER_DETAIL_V1))?;
         let prior_pointer = if let Some(expected) = undecoded_expectation.as_ref() {
             if expected_active_generation.is_some() {
                 return Err(CodeIndexPublicationStoreErrorV1::CompareAndSwap);

crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs

@@ -398,6 +398,14 @@ impl CodeIndexSchedulerErrorV1 {
             }
             Self::SnapshotMemoryCapacityUnavailable => true,
             Self::GraphProjection(CodeGraphProjectionError::BudgetExhausted { .. }) => true,
+            // The code-generation store lock is bounded shared capacity: a
+            // concurrent publication in the same store root already holds it,
+            // and it releases on its own without waking this worktree. Every
+            // other `Unavailable` detail names a fault in this store, so only
+            // this one refusal is retried.
+            Self::Production(CodeIndexProductionErrorV1::Publication(
+                CodeIndexPublicationStoreErrorV1::Unavailable(detail),
+            )) => detail == super::publication_store::CODE_GENERATION_STORE_ACTIVE_OWNER_DETAIL_V1,
             _ => false,
         }
     }

crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs

@@ -521,6 +521,25 @@ impl CodeIndexSchedulerRegistryV1 {
             })
     }
 
+    /// The pending-wake slot for one exact mounted root, in unix micros; `0`
+    /// means no wake is outstanding. A pass that ends while a wake is already
+    /// pending re-arms a busy follow-up whose receipt lands later, so a test
+    /// pinning wake or receipt accounting needs this as well as
+    /// `reconcile_in_progress_for_test`.
+    #[cfg(test)]
+    pub(crate) async fn pending_wake_micros_for_root(&self, project_root: &Path) -> Option<u64> {
+        let project_root = project_root.canonicalize().ok()?;
+        let mounted = self.mounted.lock().await;
+        mounted.get(&project_root).map(|worktree| {
+            worktree
+                .pending_wake
+                .state
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .micros
+        })
+    }
+
     /// The exact-source currency witness for one mounted root, so tests can
     /// stage the unproven-seat state a restart restore leaves behind.
     #[cfg(test)]

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs

@@ -1203,6 +1203,30 @@ async fn wait_for_quiescent_owner_pass(
     }
 }
 
+/// Wait until the mounted worker for `path` is idle with nothing queued.
+///
+/// [`wait_for_quiescent_owner_pass`] only reports that no pass is *running*.
+/// A pass that ends while a wake is already pending re-arms a busy follow-up
+/// whose receipt lands later, so a test pinning receipt accounting has to wait
+/// for the pending-wake slot as well.
+async fn wait_for_settled_owner(registry: &CodeIndexSchedulerRegistryV1, path: &Path) {
+    let deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING;
+    loop {
+        wait_for_quiescent_owner_pass(registry, path).await;
+        if registry.pending_wake_micros_for_root(path).await == Some(0)
+            && !registry.reconcile_in_progress_for_test(path).await
+        {
+            return;
+        }
+        assert!(
+            Instant::now() <= deadline,
+            "the owner for {} never settled",
+            path.display()
+        );
+        tokio::time::sleep(Duration::from_millis(2)).await;
+    }
+}
+
 /// Drive the seated owner's clone-fingerprint backfill to completion.
 ///
 /// The seat no longer waits for that successor: exact and lexical serve as

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/noop_reconcile_tests.rs

@@ -13,8 +13,21 @@ async fn unchanged_reconcile_does_not_reactivate_the_serving_generation() {
         )
         .await
         .expect("mount");
-    let serving_generation = wait_for_initial_generation(&registry, fixture.path()).await;
+    wait_for_initial_generation(&registry, fixture.path()).await;
+    // The seat is published mid-pass and the seat no longer waits for the
+    // clone successor, so the mount's own receipt lands after the in-progress
+    // guard drops and the leftover backfill drains on later wakes that post
+    // receipts of their own. Settle that whole chain first: a wake still
+    // pending when the overflow arrives keeps its earlier arrival instant, and
+    // the pass would then answer for both.
+    drain_clone_backfill(&registry, fixture.path()).await;
+    wait_for_settled_owner(&registry, fixture.path()).await;
+    wait_for_event_to_ready(&registry).await;
     let admission = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
+    let serving_generation = registry
+        .latest_generation_id(fixture.path())
+        .await
+        .expect("serving generation");
     let scheduler = registry
         .scheduler_handle(fixture.path())
         .await
@@ -29,7 +42,11 @@ async fn unchanged_reconcile_does_not_reactivate_the_serving_generation() {
         .worktree
         .clone()
         .expect("worktree identity");
-    let before_receipts = registry.event_to_ready_receipts().len();
+    // Receipts are attributed by the arrival the pass claimed, not by list
+    // position: a mount-era receipt that lands after this instant still
+    // belongs to the mount. Only a wake accepted from here on is this
+    // reconcile's.
+    let overflow_at = tracedecay_contracts::now_micros().0;
 
     // Any redundant graph activation now fails. An unchanged reconcile must
     // still reach its Noop receipt by retaining the already-serving graph.
@@ -45,8 +62,16 @@ async fn unchanged_reconcile_does_not_reactivate_the_serving_generation() {
     let deadline = std::time::Instant::now() + Duration::from_secs(3);
     loop {
         let receipts = registry.event_to_ready_receipts();
-        if let Some(receipt) = receipts.get(before_receipts) {
-            assert!(receipt.is_noop(), "unchanged reconcile must be a no-op");
+        if let Some(receipt) = receipts.iter().find(|receipt| {
+            receipt
+                .arrival
+                .wake_micros()
+                .is_some_and(|wake_micros| wake_micros >= overflow_at)
+        }) {
+            assert!(
+                receipt.is_noop(),
+                "unchanged reconcile must be a no-op: {receipts:#?}"
+            );
             break;
         }
         assert!(