fix(route): advance free candidates after explicit 429

.github/workflows/nim-benchmark.yml

@@ -49,10 +49,15 @@ jobs:
         with:
           python-version: "3.12"
 
+      - name: Set up uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          version: "0.12.5"
+
       - name: Install pinned runtime
         run: |
-          python -m pip install --require-hashes -r requirements.lock
-          python -m pip install --no-deps -e .
+          uv sync --locked --extra api --extra db --extra queue
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
 
       - name: Run dry benchmark
         env:
@@ -104,10 +109,15 @@ jobs:
         with:
           python-version: "3.12"
 
+      - name: Set up uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          version: "0.12.5"
+
       - name: Install pinned runtime
         run: |
-          python -m pip install --require-hashes -r requirements.lock
-          python -m pip install --no-deps -e .
+          uv sync --locked --extra api --extra db --extra queue
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
 
       - name: Resolve live parameters
         id: live_params

.github/workflows/opencode-hourly-loop.yml

@@ -105,9 +105,16 @@ jobs:
         with:
           node-version: "22"
 
+      - name: Set up uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          version: "0.12.5"
+
       - name: Install pinned runtime dependencies
         timeout-minutes: 10
-        run: python -m pip install --require-hashes -r requirements.lock
+        run: |
+          uv sync --locked --extra api --extra db --extra queue
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
 
       - name: Install OpenCode CLI
         timeout-minutes: 10

.github/workflows/provider-catalog-sync.yml

@@ -44,8 +44,15 @@ jobs:
         with:
           python-version: "3.12"
 
-      - name: Install hash-pinned runtime dependencies
-        run: python -m pip install --disable-pip-version-check --no-input --require-hashes -r requirements.lock
+      - name: Set up uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          version: "0.12.5"
+
+      - name: Install pinned runtime dependencies
+        run: |
+          uv sync --locked --extra api --extra db --extra queue
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
 
       - name: Select durable or run-scoped PostgreSQL KV
         shell: bash

.github/workflows/security.yml

@@ -71,8 +71,8 @@ jobs:
 
       - name: Install hash-locked quality tools
         run: |
-          python -m pip install --require-hashes -r requirements.lock
-          python -m pip install --require-hashes -r requirements-opencode-review-ci.txt
+          uv pip install --python .venv/bin/python --require-hashes -r requirements-opencode-review-ci.txt
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
 
       - name: Prove complete benchmark coverage and public docstrings
         run: |
@@ -95,9 +95,9 @@ jobs:
         run: |
           set -euo pipefail
           rm -rf dist "$RUNNER_TEMP/nim-wheel-site"
-          python -m pip wheel --no-deps --no-build-isolation . --wheel-dir dist
+          uv build --wheel --no-build-isolation --out-dir dist
           python scripts/verify_decision_wheel_manifest.py dist/contextual_orchestrator-*.whl "$RUNNER_TEMP"/decision-wheels/*.whl
-          python -m pip install --no-deps \
+          uv pip install --python .venv/bin/python --no-deps \
             --target "$RUNNER_TEMP/nim-wheel-site" \
             dist/contextual_orchestrator-*.whl "$RUNNER_TEMP"/decision-wheels/*.whl
           cd "$RUNNER_TEMP"
@@ -123,8 +123,16 @@ jobs:
         with:
           python-version: "3.12"
 
+      - name: Set up uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          version: "0.12.5"
+
       - name: Install fuzz dependencies
-        run: python -m pip install --require-hashes -r requirements.lock -r fuzz/requirements-property.txt -r fuzz/requirements-atheris.txt
+        run: |
+          uv sync --locked --extra api --extra db --extra queue --group dev
+          uv pip install --python .venv/bin/python --require-hashes -r fuzz/requirements-property.txt -r fuzz/requirements-atheris.txt
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
 
       - name: Run property-based fuzz tests
         run: python -m pytest tests/fuzz -q
@@ -266,17 +274,30 @@ jobs:
         with:
           python-version: "3.12"
 
+      - name: Set up uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          version: "0.12.5"
+
       - name: Install audit and project dependencies
         run: |
-          python -m pip install --require-hashes -r requirements-security-ci.txt
-          python -m pip install --require-hashes -r requirements.lock
-          python -m pip install --no-deps -e .
+          uv sync --locked --extra api --extra db --extra queue --group dev --no-install-project
+          uv sync --locked --extra api --extra db --extra queue --group dev --no-build-isolation
+          python -m venv "$RUNNER_TEMP/security-tools"
+          "$RUNNER_TEMP/security-tools/bin/python" -m pip install --require-hashes -r requirements-security-ci.txt
 
       - name: Audit pinned dependencies and generate SBOM
         run: |
           set -euo pipefail
-          python -m pip_audit -r requirements.lock
-          cyclonedx-py environment --output-format json --output-file cyclonedx-sbom.json
+          "$RUNNER_TEMP/security-tools/bin/pip-audit" --require-hashes -r requirements-security-ci.txt
+          audit_lock="${RUNNER_TEMP}/requirements-audit.txt"
+          python scripts/ci/prepare_runtime_lock_audit.py requirements.lock uv.lock "$audit_lock"
+          "$RUNNER_TEMP/security-tools/bin/pip-audit" --require-hashes --disable-pip -r "$audit_lock"
+          "$RUNNER_TEMP/security-tools/bin/pip-audit" --path "$PWD/.venv/lib/python3.12/site-packages" --format json --output "$RUNNER_TEMP/dependency-audit.json"
+          jq -e '([.dependencies[] | select(.skip_reason) | .name] | sort) == ["contextual-orchestrator", "fast-mlsirm"]' "$RUNNER_TEMP/dependency-audit.json"
+          "$RUNNER_TEMP/security-tools/bin/cyclonedx-py" environment "$PWD/.venv/bin/python" --output-format json --output-file cyclonedx-sbom.json
+          jq -e 'any(.components[]; .name == "contextual-orchestrator")' cyclonedx-sbom.json
+          jq -e '[.components[] | select(.name == "fast-mlsirm") | (.purl // "") | contains("09f762ded35786dd1078222a4577ff09d649816f")] == [true]' cyclonedx-sbom.json
 
       - name: Upload CycloneDX SBOM
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # actions/upload-artifact@v7.0.1

CHANGELOG.d/rate-limit-aware-admission.md

@@ -52,3 +52,5 @@ candidate, while an explicit concrete model id keeps failing fast
 unconditionally, regardless of how many failover candidates exist --
 preserving the `tests/test_provider_error_taxonomy.py` single-candidate,
 no-header 429 contract that must never wait.
+
+An explicit quota rejection advances to the next eligible candidate before a same-agent tool retry. Unknown-duration 429 recovery retains the existing administrator-owned assumed cooldown and labels it as assumed; it does not become a provider-declared timing claim.

CHANGELOG.d/restore-test-signal-main.md

@@ -0,0 +1,3 @@
+Restore the OpenRouter availability collector's separation from the answer-quality ledger. The restack merge for #1074 (`3078949c`) had silently reverted `openrouter_uptime.py` to its pre-`b3be48e3` form, so uptime polls once again rewrote the quality router's prior and blended in the benchmark quality prior. The collector now updates only the transport ledger again. The fixed `_UPTIME_FETCH_TIMEOUT_SECONDS` bound from #971 is kept. Also repair test drift that stopped the hosted `Tests and package quality` job at collection or kept suites red: the renamed provider-embedding claim-lease constant, a fake HTTP connection without `connect()`/`sock`, fingerprinted discovered-agent ids, a fake binary response without `headers.get`, and a stale batch-retention assertion that contradicted the unbounded-execution contract. Provider embedding waits now accept the application default `None` as an unbounded deadline, matching the synchronous endpoint contract instead of raising `TypeError` before provider completion.
+Bind `rustfmt` and `clippy` to the repository's pinned Rust 1.97.1 toolchain. The minimal rustup profile does not include either component, so the hosted Rust gate previously selected the directory override and failed before formatting or linting could run.
+Audit the hashed runtime index pins alongside the installed environment, and require the non-PyPI Git source to agree across locks before the security gate can pass.

contextual_orchestrator/batch_job_registry.py

@@ -214,6 +214,7 @@ def lock(
         *,
         lease_seconds: float | None = None,
         renew_until_epoch: float | None = None,
+        renew_while: Callable[[], bool] | None = None,
     ):
         """Return an atomic shard claim with bounded lease and acquisition wait."""
         lock_name = f"batch_job_registry:{name}:claim:{key}"
@@ -245,6 +246,9 @@ def acquired_claim():
                     def renew_claim() -> None:
                         interval = _claim_renewal_interval_seconds(lease_seconds)
                         while not stop_renewal.wait(interval):
+                            if renew_while is not None and not renew_while():
+                                lease.mark_lost()
+                                return
                             remaining = renew_until_epoch - time.time()
                             if remaining <= 0:
                                 lease.mark_lost()

contextual_orchestrator/batch_routing.py

@@ -1088,9 +1088,17 @@ def __init__(
 
     def close(self) -> None:
         """Release the bounded worker pool owned by this backend."""
-        self._closed.set()
         with self._executor_lock:
+            self._closed.set()
             executor, self._executor = self._executor, None
+        for job_id, event in list(self._terminal_events.items()):
+            if self._registry.durable:
+                # The next backend can reclaim persisted work after this worker exits.
+                event.set()
+            else:
+                self.cancel(
+                    BatchJob(job_id=job_id, backend=self.name), reason="backend closed"
+                )
         if executor is not None:
             executor.shutdown(wait=False, cancel_futures=True)
 
@@ -1164,6 +1172,7 @@ def _run_job(self, job_id: str) -> None:
                     job_id,
                     lease_seconds=self._claim_lease_seconds,
                     renew_until_epoch=deadline_epoch,
+                    renew_while=lambda: not self._closed.is_set(),
                 ) as execution_claim:
                     self._run_claimed_job(job_id, execution_claim)
             except ClaimNotAcquired:
@@ -1255,6 +1264,8 @@ def _run_claimed_job(self, job_id: str, execution_claim: Any) -> None:
         requests = list(self._requests[job_id])
         try:
             vectors, prompt_tokens = self._runner(requests)
+            if self._closed.is_set():
+                execution_claim.mark_lost()
             execution_claim.ensure_owned()
             if time.time() >= self._execution_deadline(job_id):
                 self._publish_terminal(
@@ -1295,6 +1306,9 @@ def _run_claimed_job(self, job_id: str, execution_claim: Any) -> None:
         except ClaimNotAcquired:
             raise
         except Exception as exc:  # noqa: BLE001 - polling exposes bounded failure metadata
+            if self._closed.is_set():
+                execution_claim.mark_lost()
+            execution_claim.ensure_owned()
             error = {
                 "error_type": type(exc).__name__,
                 "http_status": getattr(
@@ -1320,32 +1334,36 @@ def _publish_terminal(
         usage: Any = None,
         error: Any = None,
     ) -> None:
-        """Publish terminal state atomically for durable registries."""
-        if self._registry.durable:
-            self._registry.publish_provider_embedding_terminal(
-                execution_claim,
-                job_id,
-                status=status,
-                results=results,
-                usage=usage,
-                error=error,
-            )
-            return
-        with self._registry.lock(
-            "provider_embedding_job_states",
-            job_id,
-            lease_seconds=self._claim_lease_seconds,
-        ):
+        """Fence shutdown and publish one terminal outcome under the lifecycle lock."""
+        with self._executor_lock:
+            if self._closed.is_set():
+                execution_claim.mark_lost()
             execution_claim.ensure_owned()
-            if self._states.get(job_id) not in {"queued", "running"}:
-                raise ClaimNotAcquired("provider embedding job is already terminal")
-            if results is not None:
-                self._results[job_id] = results
-            if usage is not None:
-                self._usage[job_id] = usage
-            if error is not None:
-                self._errors[job_id] = error
-            self._states[job_id] = status
+            if self._registry.durable:
+                self._registry.publish_provider_embedding_terminal(
+                    execution_claim,
+                    job_id,
+                    status=status,
+                    results=results,
+                    usage=usage,
+                    error=error,
+                )
+                return
+            with self._registry.lock(
+                "provider_embedding_job_states",
+                job_id,
+                lease_seconds=self._claim_lease_seconds,
+            ):
+                execution_claim.ensure_owned()
+                if self._states.get(job_id) not in {"queued", "running"}:
+                    raise ClaimNotAcquired("provider embedding job is already terminal")
+                if results is not None:
+                    self._results[job_id] = results
+                if usage is not None:
+                    self._usage[job_id] = usage
+                if error is not None:
+                    self._errors[job_id] = error
+                self._states[job_id] = status
 
     def wait(self, job: BatchJob, *, timeout: float | None) -> Dict[str, Any]:
         """Wait within the caller's explicit deadline for a terminal state.
@@ -1354,10 +1372,18 @@ def wait(self, job: BatchJob, *, timeout: float | None) -> Dict[str, Any]:
         wall-clock deadline (contextual-orchestrator's no-implicit-deadline
         default). ``threading.Event.wait`` raises ``OverflowError`` for a
         non-finite timeout on CPython, so both forms are translated to ``None``
-        (block indefinitely) rather than passed through.
+        (block indefinitely) rather than passed through. Durable unbounded waits
+        observe shared terminal state at the backend's existing claim cadence;
+        another process cannot signal this process's local event.
         """
         event = self._terminal_events.get(job.job_id)
-        if event is not None:
+        if event is not None and self._registry.durable and (
+            timeout is None or not math.isfinite(timeout)
+        ):
+            while not self._closed.is_set() and not self.poll(job)["is_complete"]:
+                if event.wait(self.poll_after_ms / 1000):
+                    break
+        elif event is not None:
             event.wait(
                 timeout=(
                     timeout