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

@@ -57,8 +57,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: |
@@ -81,9 +81,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"
@@ -109,8 +109,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
@@ -224,17 +232,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

@@ -23,16 +23,15 @@ a `502` connection failure. `provider_readiness_report`
 instead of exiting.
 
 A 429 that states no cooldown at all (RFC 9110 permits omitting
-`Retry-After`/`x-ratelimit-reset*`, and NIM/OpenRouter routinely do) now
-records an assumed cooldown -- the new `rate_limit_unknown_cooldown_seconds`
-default (5s) -- instead of nothing, so an all-omitted-header storm can no
-longer look identical to "nothing is rate-limited" and fail as if this
-feature did not exist; every cooldown surface labels itself
-`cooldown_source: "provider"` or `"assumed"` accordingly, and a provider-stated
-cooldown is never shortened or relabeled by a later assumed one. This
-assumption applies to 429 only (a 503 with no header keeps requiring a real
-provider-stated duration) -- scoped narrowly after concrete pre-existing-test
-regression evidence, not by design intent alone.
+`Retry-After`) is recorded as unavailable without inventing a deadline.
+An all-omitted-header storm therefore returns the typed
+`provider_rate_limited` 429 with `cooldown_source: "unavailable"`, no
+`Retry-After`, and `retryable: false`; other eligible candidates are still
+tried. Provider-declared finite timing remains labeled `"provider"` and can
+authorize bounded waiting. Readiness renders unavailable timing as JSON
+`null`, not non-standard `Infinity`. The former 5-second assumed cooldown and
+its constructor/CLI option were removed because neither RFC 9110 nor provider
+evidence authorizes that retry decision.
 
 The wait admission decision no longer turns on candidate count. An earlier
 version of this guard returned immediately whenever fewer than two

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/__main__.py

@@ -1094,19 +1094,6 @@ def main(argv: list[str] | None = None) -> None:
             "model_timeout_seconds deadline (default: 30)."
         ),
     )
-    parser.add_argument(
-        "--rate-limit-unknown-cooldown-seconds",
-        type=float,
-        default=5.0,
-        help=(
-            "Assumed cooldown applied when a 429/503 provider response "
-            "states no Retry-After/x-ratelimit-reset* at all (RFC 9110 "
-            "permits omitting it, and some providers routinely do). This is "
-            "a caller-contract bound, not a discovered provider fact -- kept "
-            "short by default so an unknown cooldown is re-probed soon "
-            "rather than parked (default: 5)."
-        ),
-    )
     parser.add_argument("--eval", nargs="+", metavar="PROMPT",
                         help="Measure orchestration vs a single-worker baseline on these prompts and print the report.")
     parser.add_argument(
@@ -1153,7 +1140,6 @@ def main(argv: list[str] | None = None) -> None:
         budget_max_cost_usd=args.budget_max_cost_usd,
         cache_ttl=args.cache_ttl,
         rate_limit_wait_seconds=args.rate_limit_wait_seconds,
-        rate_limit_unknown_cooldown_seconds=args.rate_limit_unknown_cooldown_seconds,
         allow_empty_agents=args.auto_discover_model_agents,
         role_effort_catalog=(
             default_role_effort_catalog() if args.role_effort_catalog == "default" else None

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,45 +1334,63 @@ 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
-
-    def wait(self, job: BatchJob, *, timeout: float) -> Dict[str, Any]:
+            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.
 
-        ``timeout`` may be ``float("inf")`` when the caller has no wall-clock
-        deadline (contextual-orchestrator's no-implicit-deadline default);
-        ``threading.Event.wait`` raises ``OverflowError`` for a non-finite
-        timeout on CPython, so a non-finite value is translated to ``None``
-        (block indefinitely) rather than passed through.
+        ``timeout`` may be ``None`` or ``float("inf")`` when the caller has no
+        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. 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:
-            event.wait(timeout=timeout if math.isfinite(timeout) else 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
+                    if timeout is not None and math.isfinite(timeout)
+                    else None
+                )
+            )
         return self.poll(job)
 
     def poll(self, job: BatchJob) -> Dict[str, Any]:

contextual_orchestrator/nim_benchmark.py

@@ -2386,15 +2386,6 @@ def complete_cell() -> dict[str, Any]:
 
 
 
-def _require_declared_positive_int(value: object, field_name: str) -> int:
-    """Reject missing, boolean, or non-positive integer declarations."""
-    if type(value) is not int or value < 1:
-        raise BenchmarkContractError(
-            f"{field_name} must be a declared positive integer"
-        )
-    return value
-
-
 # --------------------------------------------------------------------------
 # Statistics: paired bootstrap + Pareto frontiers
 # --------------------------------------------------------------------------