fix(tests): restore main's test signal; re-apply dropped OpenRouter availability/quality split

.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 --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,22 @@ 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
 
       - 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
+          python -m pip_audit --require-hashes -r requirements.lock
+          python -m pip_audit --path .venv/lib/python3.12/site-packages
+          cyclonedx-py environment .venv/bin/python --output-format json --output-file cyclonedx-sbom.json
 
       - name: Upload CycloneDX SBOM
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # actions/upload-artifact@v7.0.1

CHANGELOG.d/provider-owned-quota-timing.md

@@ -0,0 +1 @@
+Honor provider-declared retry timing that is already carried on a classified provider error (chat, passthrough and structured requests), stop replaying the same agent after an explicit quota rejection, and let a fresh successful provider call clear that agent's cooldown. A 429 without `Retry-After`/`x-ratelimit-reset*` still records the finite, administrator-owned `rate_limit_unknown_cooldown_seconds` assumed cooldown (`cooldown_source: assumed`), exactly as on `main`: the agent is skipped only while that cooldown runs and becomes selectable again once it elapses. An intermediate revision had recorded such a 429 as an infinite cooldown, which excluded the agent permanently whenever an alternative existed; that regression is fixed and covered by a test.

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

@@ -0,0 +1,7 @@
+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.
+Bound the untrusted OpenRouter uptime response to the same 8 MiB provider-response ceiling and reject oversized JSON before parsing, while preserving valid endpoint telemetry.
+Preserve one deployment identity entry for every real same-agent retry in deterministic selection receipts, so replay evidence no longer understates provider calls.
+Keep provider-backed contextual embedding outside the psychometric persistence lock while revalidating the deployment on both sides of the call, so an unbounded provider wait cannot block observation persistence or Agent-pool edits.
+Repair the OpenRouter bounded-read test double and the finite embedding-wait fixture so both tests exercise their intended production boundary without a swallowed `TypeError` or a timer-driven race.
+Restore the required hash-locked `requirements.lock` supply-chain audit alongside the installed-environment audit. Replace the obsolete fast-mlsirm VCS pin with immutable released `fast-mlsirm==0.11.4`, declare an `anyio>=4.14.2` security floor and upgrade the locks from vulnerable `anyio==4.14.1` to fixed `anyio==4.14.2`, align the constrained fuzz locks, migrate the held-out DIF benchmark to the release's non-deprecated API, remove an inactive pytest-asyncio option, and keep the environment-derived CycloneDX SBOM as a separate artifact.

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

contextual_orchestrator/openrouter_uptime.py

@@ -47,6 +47,10 @@
 # uptime update -- so it keeps its own fixed, independent bound instead.
 _UPTIME_FETCH_TIMEOUT_SECONDS = 10.0
 
+# Match the inference response boundary: telemetry is untrusted provider input
+# and must not consume unbounded memory before JSON validation.
+_OPENROUTER_UPTIME_RESPONSE_MAX_BYTES = 8 * 1024 * 1024
+
 
 class OpenRouterUptimeCollector:
     """Periodically fold upstream availability into the transport prior."""
@@ -159,7 +163,12 @@ def _fetch_uptime(self, model_id: str) -> float | None:
             with urllib.request.urlopen(
                 request, timeout=_UPTIME_FETCH_TIMEOUT_SECONDS
             ) as response:
-                payload = json.loads(response.read().decode("utf-8"))
+                response_body = response.read(
+                    _OPENROUTER_UPTIME_RESPONSE_MAX_BYTES + 1
+                )
+                if len(response_body) > _OPENROUTER_UPTIME_RESPONSE_MAX_BYTES:
+                    raise ValueError("OpenRouter uptime response exceeds byte limit")
+                payload = json.loads(response_body.decode("utf-8"))
                 endpoints = payload.get("data", {}).get("endpoints", [])
                 uptimes = [
                     endpoint["uptime_last_30m"]