(MOT-4412) fix(llm-router): harden provider lifecycle

.github/scripts/discover_changed_workers.py

@@ -6,6 +6,7 @@
     source_changed                : workers whose change wasn't only metadata
     rust / node / python          : language buckets (subset of changed_workers)
     integration_changed          : bool, did an integration-stack input change
+    llm_router_integration        : bool, must the live llm-router suite run
     provider_contract            : providers whose hermetic contract must run
     crates                        : shared crates/<name> dirs with source changes
     any                           : bool, any worker or crate change
@@ -65,6 +66,15 @@
     "harness/tests/quickstart/",
 )
 
+# The llm-router owns a separate real-engine lifecycle suite. Keep this gate
+# independent from Harness Integration: that stack intentionally substitutes a
+# ScriptedRouter and therefore cannot validate router transport/lifecycle bugs.
+LLM_ROUTER_INTEGRATION_WORKERS = {"llm-router"}
+LLM_ROUTER_INTEGRATION_INFRA_PATHS = {
+    ".github/scripts/discover_changed_workers.py",
+    ".github/workflows/ci.yml",
+}
+
 # Hermetic provider contracts run the real engine, llm-router, and selected
 # provider against a loopback HTTP/SSE upstream. Direct provider changes stay
 # narrow; shared router/testkit/CI changes fan out to every supported provider.
@@ -275,6 +285,13 @@ def main(argv: list[str] | None = None) -> int:
         INTEGRATION_INFRA_PATHS,
         INTEGRATION_EXCLUDED_PREFIXES,
     )
+    llm_router_integration = suite_changed(
+        files,
+        forced,
+        LLM_ROUTER_INTEGRATION_WORKERS,
+        LLM_ROUTER_INTEGRATION_INFRA_PATHS,
+        (),
+    )
     provider_contract = provider_contract_selection(files, workers)
     by_language: dict[str, list[str]] = {"rust": [], "node": [], "python": []}
     for w in changed:
@@ -289,6 +306,7 @@ def main(argv: list[str] | None = None) -> int:
         "source_changed": source_changed,
         "by_language": by_language,
         "integration_changed": integration_changed,
+        "llm_router_integration": llm_router_integration,
         "provider_contract": provider_contract,
         "crates": changed_crates,
     }
@@ -305,6 +323,10 @@ def main(argv: list[str] | None = None) -> int:
             f.write(
                 f"integration_changed={'true' if integration_changed else 'false'}\n"
             )
+            f.write(
+                "llm_router_integration="
+                f"{'true' if llm_router_integration else 'false'}\n"
+            )
             f.write(f"provider_contract={json.dumps(provider_contract)}\n")
             f.write(f"crates={json.dumps(changed_crates)}\n")
             f.write(f"any={'true' if any_change else 'false'}\n")

.github/scripts/tests/test_check_links.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import os
+import subprocess
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+CHECK_LINKS = ROOT / "scripts" / "check-links.sh"
+
+
+def fake_curl(tmp_path: Path, mitigation: str) -> Path:
+    curl = tmp_path / "curl"
+    curl.write_text(
+        """#!/usr/bin/env bash
+set -euo pipefail
+headers_file=
+while (( $# )); do
+  if [[ "$1" == "-D" ]]; then
+    headers_file=$2
+    shift 2
+  else
+    shift
+  fi
+done
+printf 'HTTP/2 403\\r\\nx-vercel-mitigated: %s\\r\\n\\r\\n' "$FAKE_VERCEL_MITIGATION" > "$headers_file"
+printf '403'
+""",
+        encoding="utf-8",
+    )
+    curl.chmod(0o755)
+    return curl
+
+
+def run_check(tmp_path: Path, mitigation: str) -> subprocess.CompletedProcess[str]:
+    fake_curl(tmp_path, mitigation)
+    env = os.environ.copy()
+    env["FAKE_VERCEL_MITIGATION"] = mitigation
+    env["PATH"] = f"{tmp_path}:{env['PATH']}"
+    return subprocess.run(
+        [str(CHECK_LINKS)],
+        cwd=ROOT,
+        env=env,
+        text=True,
+        capture_output=True,
+        timeout=30,
+        check=False,
+    )
+
+
+def test_vercel_security_challenge_is_reachable(tmp_path: Path) -> None:
+    result = run_check(tmp_path, "challenge")
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert "Vercel security challenge" in result.stdout
+
+
+def test_vercel_deny_remains_a_failure(tmp_path: Path) -> None:
+    result = run_check(tmp_path, "deny")
+
+    assert result.returncode == 1
+    assert "FAIL 403" in result.stdout

.github/scripts/tests/test_discover_changed_workers.py

@@ -283,8 +283,56 @@ def test_provider_change_stays_out_of_integration(self, tmp_path):
         data = json.loads(r.stdout)
         assert data["changed_workers"] == ["provider-anthropic"]
         assert data["integration_changed"] is False
+        assert data["llm_router_integration"] is False
         assert data["provider_contract"] == ["provider-anthropic"]
 
+    @pytest.mark.parametrize(
+        "changed_path",
+        [
+            "llm-router/src/lib.rs",
+            "llm-router/tests/integration.rs",
+            ".github/workflows/ci.yml",
+            ".github/scripts/discover_changed_workers.py",
+        ],
+    )
+    def test_llm_router_runtime_inputs_run_live_router_integration(
+        self, tmp_path, changed_path
+    ):
+        repo = make_repo_with_harness(tmp_path)
+        path = repo / changed_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("changed\n")
+        subprocess.run(
+            ["git", "add", "."], cwd=repo, check=True, env=GIT_HERMETIC_ENV
+        )
+        subprocess.run(
+            ["git", "commit", "-q", "-m", "router integration input"],
+            cwd=repo,
+            check=True,
+            env=GIT_HERMETIC_ENV,
+        )
+        r = run_script(repo, "main~1")
+        assert r.returncode == 0, r.stderr
+        assert json.loads(r.stdout)["llm_router_integration"] is True
+
+    def test_llm_router_docs_do_not_run_live_router_integration(self, tmp_path):
+        repo = make_repo_with_harness(tmp_path)
+        path = repo / "llm-router" / "README.md"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("# docs\n")
+        subprocess.run(
+            ["git", "add", "."], cwd=repo, check=True, env=GIT_HERMETIC_ENV
+        )
+        subprocess.run(
+            ["git", "commit", "-q", "-m", "router docs"],
+            cwd=repo,
+            check=True,
+            env=GIT_HERMETIC_ENV,
+        )
+        r = run_script(repo, "main~1")
+        assert r.returncode == 0, r.stderr
+        assert json.loads(r.stdout)["llm_router_integration"] is False
+
     def test_database_change_stays_out_of_integration(self, tmp_path):
         repo = make_repo_with_harness(tmp_path)
         (repo / "database" / "lib.rs").write_text("// change\n")

.github/scripts/tests/test_rust_ci_workflows.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+import re
 import tomllib
 
 import yaml
@@ -26,8 +27,9 @@ def test_rust_toolchain_is_pinned_to_the_last_verified_stable() -> None:
     assert toolchain["toolchain"]["channel"] == "1.97.1"
 
     bodies = "\n".join(path.read_text() for path in WORKFLOWS.glob("*.yml"))
-    assert "dtolnay/rust-toolchain@stable" not in bodies
-    assert bodies.count("dtolnay/rust-toolchain@1.97.1") == 13
+    workflow_toolchains = re.findall(r"dtolnay/rust-toolchain@([^\s]+)", bodies)
+    assert workflow_toolchains
+    assert set(workflow_toolchains) == {"1.97.1"}
 
 
 def test_prs_restore_rust_caches_and_main_pushes_publish_them() -> None:
@@ -108,7 +110,7 @@ def test_rust_security_audit_is_narrow_on_prs_and_complete_on_schedule() -> None
     steps = audit["jobs"]["audit"]["steps"]
     install = named_step(steps, "Install cargo-audit")
     run = named_step(steps, "Audit Rust lockfiles")["run"]
-    assert install["uses"] == "taiki-e/install-action@v2.79.11"
+    assert install["uses"] == "taiki-e/install-action@v2.85.13"
     assert install["with"]["tool"] == "cargo-audit@0.22.2"
     assert "git diff --name-only -z" in run
     assert "find . -name Cargo.lock" in run

.github/workflows/ci.yml

@@ -53,6 +53,7 @@ jobs:
       changed_workers: ${{ steps.bucket.outputs.changed_workers }}
       source_changed: ${{ steps.bucket.outputs.source_changed }}
       integration_changed: ${{ steps.bucket.outputs.integration_changed }}
+      llm_router_integration: ${{ steps.bucket.outputs.llm_router_integration }}
       provider_contract: ${{ steps.bucket.outputs.provider_contract }}
       any: ${{ steps.bucket.outputs.any }}
     steps:
@@ -267,6 +268,60 @@ jobs:
       - name: Run tests
         run: cargo test --locked --all-features
 
+  # ──────────────────────────────────────────────────────────────
+  # llm-router lifecycle contract: unlike the regular Rust job, this always
+  # supplies the pinned engine and therefore cannot silently self-skip the
+  # real-bus registration, streaming, cancellation, and restart scenarios.
+  # ──────────────────────────────────────────────────────────────
+  llm-router-integration:
+    name: "llm-router: live engine integration"
+    needs: discover
+    if: needs.discover.outputs.llm_router_integration == 'true'
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Rewrite SSH to HTTPS for public deps
+        run: git config --global url."https://github.com/".insteadOf "ssh://git@github.com/"
+
+      - uses: dtolnay/rust-toolchain@1.97.1
+
+      - uses: Swatinem/rust-cache@v2
+        with:
+          shared-key: llm-router-integration
+          save-if: false
+          workspaces: llm-router -> target
+
+      - name: Install pinned iii engine
+        env:
+          VERSION: '0.22.1'
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          set -euo pipefail
+          curl -fsSL https://install.iii.dev/iii/main/install.sh -o /tmp/install-iii.sh
+          sh /tmp/install-iii.sh
+          {
+            echo "$HOME/.local/bin"
+            echo "$HOME/.iii/bin"
+          } >> "$GITHUB_PATH"
+          export PATH="$HOME/.local/bin:$HOME/.iii/bin:$PATH"
+          engine_bin=$(command -v iii)
+          [[ -x "$engine_bin" ]] || { echo "::error::iii engine is not executable"; exit 3; }
+          echo "III_ENGINE_BIN=$engine_bin" >> "$GITHUB_ENV"
+          iii --version
+
+      - name: Run real-engine router lifecycle suite
+        run: |
+          set -euo pipefail
+          [[ -x "$III_ENGINE_BIN" ]] || { echo "::error::III_ENGINE_BIN is unavailable"; exit 3; }
+          cargo test \
+            --locked \
+            --manifest-path llm-router/Cargo.toml \
+            --no-default-features \
+            --test integration \
+            -- --nocapture --test-threads=1
+
   # ──────────────────────────────────────────────────────────────
   # Provider contracts: real engine + real router + selected provider,
   # with the vendor HTTP/SSE boundary replaced by a loopback stub. No live

.github/workflows/rust-security-audit.yml

@@ -24,7 +24,7 @@ jobs:
           fetch-depth: 0
 
       - name: Install cargo-audit
-        uses: taiki-e/install-action@v2.79.11
+        uses: taiki-e/install-action@v2.85.13
         with:
           tool: cargo-audit@0.22.2
           fallback: none

harness/tests/integration/README.md

@@ -28,6 +28,7 @@ No provider key or network access is required.
 | INT-018 | `spawn-reuse-guard` | direct | an in-turn spawn into an existing session owned by another parent is refused naming the owner (no hijack turn ever starts); re-spawning its own child appends the new task to the retained transcript and reports `reused: true` |
 | INT-019 | `condition-failure-notice` | direct | a binding whose condition ERRORS on a fire wakes its owner with an actionable `[notification]` (once per binding) instead of starving silently; the skip record still lands and the binding stays armed |
 | INT-020 | `child-discovery-granted` | direct | a child narrowed to its work functions can still dispatch the mandatory `engine::functions::list`/`::info` round (the discovery union); its native toolset stays the work functions only |
+| INT-021 | `router-midstream-terminal-error` | direct | partial content and keepalive noise followed by one permanent router error preserve the partial, fail exactly once, and leave no pending work |
 | UI-001 | `console-streamed-text` | playground | a message sent by the Console streams to durable completion |
 | UI-002 | `multi-turn-traces` | playground | a native function turn and a Console turn expose distinct traces and function-call events |
 

harness/tests/integration/src/fixtures/loading.rs

@@ -16,8 +16,8 @@ pub struct ScenarioFixture {
     /// first in the statuses list.
     pub expected_terminal_turns: usize,
     /// Each completion's lifecycle status, in completion order — parked
-    /// completions first, then terminal turns. The last must be `completed` —
-    /// the floor's durable-status check binds to it.
+    /// completions first, then terminal turns. The last status is also the
+    /// durable outcome that the floor requires from `harness::status`.
     pub expected_turn_statuses: Vec<String>,
     pub scenario: CompiledScenarioV1,
     pub script: RouterScriptV1,
@@ -155,12 +155,6 @@ impl ScenarioFixture {
                 );
             }
         }
-        if self.intervention.is_none() {
-            anyhow::ensure!(
-                self.expected_turn_statuses.last().map(String::as_str) == Some("completed"),
-                "the last terminal turn must be completed"
-            );
-        }
         if let Some(intervention) = &self.intervention {
             match intervention {
                 ScenarioIntervention::StopCancelCascade {

harness/tests/integration/src/fixtures/tests.rs

@@ -12,15 +12,15 @@ fn all_selection_returns_the_checked_in_fixtures() {
         std::collections::BTreeSet::from([
             "INT-001", "INT-002", "INT-003", "INT-005", "INT-006", "INT-010", "INT-011", "INT-012",
             "INT-013", "INT-014", "INT-015", "INT-016", "INT-017", "INT-018", "INT-019", "INT-020",
-            "UI-001", "UI-002"
+            "INT-021", "UI-001", "UI-002"
         ])
     );
     assert_eq!(
         fixtures
             .iter()
             .filter(|fixture| fixture.driver == crate::scenarios::ScenarioDriver::Direct)
             .count(),
-        16
+        17
     );
 }