loopx-agent · Nov 11, 2026 · Nov 11, 2026 · Nov 11, 2026
@@ -72,6 +72,17 @@ and steward independently. An unconfigured steward route does not block the
 worker. If a configured route is unavailable, or the two health responses
 identify different relay instances, installation verification fails and each
 route keeps its own status. The legacy `relay` field still reports the worker.
+
+Version 0.15.8 reports `adapter_readiness` alongside relay readiness. With an
+enabled account policy and a saved adapter, the local probe verifies the same
+owner-file pins and native route as Chat model preparation. A stale pin reports
+`adapter_changed` and fails installation readiness even when both relays answer.
+A malformed saved policy reports `state_untrusted` and retains the two relay
+observations.
+After adoption of the reviewed package, use the existing **Reconnect** action
+with the current policy and check the live status again. For package-only
+verification before adoption, pass a disposable `--runtime-root` to `--doctor`.
+
 Keep original-message acceptance as a separate end-to-end check.
 
 For upgrades, install the reviewed target into a new immutable environment and

@@ -1,6 +1,6 @@
 schema_version = "loopx_extension_manifest_v0"
 id = "loopx-codex-provider-routing"
-version = "0.15.7"
+version = "0.15.8"
 requires_loopx_api = ">=1,<2"
 permissions = []
 

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "loopx-codex-provider-routing"
-version = "0.15.7"
+version = "0.15.8"
 description = "Native Codex account slots and exhausted-account handoff for LoopX"
 readme = "README.md"
 requires-python = ">=3.11"

@@ -1,6 +1,6 @@
 schema_version = "loopx_extension_manifest_v0"
 id = "loopx-native-account-provider"
-version = "0.15.7"
+version = "0.15.8"
 requires_loopx_api = ">=1,<2"
 permissions = ["model-provider.local-read", "model-provider.transport-manage"]
 

@@ -129,21 +129,26 @@ def discover(*, user_home: Path | None = None) -> dict:
             "profile": {key: str(value) for key, value in profile.items()}}
 
 
+def checked_adapter(adapter: dict, home: Path) -> tuple[Path, dict]:
+    if adapter.get("owner") != OWNER:
+        raise HostError("adapter_changed")
+    source = Path(__file__).parent
+    for name in OWNER_FILES:
+        if hashlib.sha256(trusted_file(source / name).read_bytes()).hexdigest() != adapter["pins"][name]:
+            raise HostError("adapter_changed")
+    codex = trusted_file(Path(adapter["codex"]))
+    spec = native_app.route(home, "gpt")
+    if {key: str(value) for key, value in spec.items()} != adapter.get("profile"):
+        raise HostError("adapter_changed")
+    return codex, spec
+
+
 class NativeHost:
     def __init__(self, adapter: dict, *, user_home: Path | None = None, reconcile_services: bool = True):
-        if adapter.get("owner") != OWNER:
-            raise HostError("adapter_changed")
-        source = Path(__file__).parent
-        for name in OWNER_FILES:
-            if hashlib.sha256(trusted_file(source / name).read_bytes()).hexdigest() != adapter["pins"][name]:
-                raise HostError("adapter_changed")
-        self.codex = trusted_file(Path(adapter["codex"]))
+        self.home = user_home or Path.home()
+        self.codex, self.spec = checked_adapter(adapter, self.home)
         self.apps = native_app
         self.accounts = native_account_store
-        self.home = user_home or Path.home()
-        self.spec = self.apps.route(self.home, "gpt")
-        if {key: str(value) for key, value in self.spec.items()} != adapter.get("profile"):
-            raise HostError("adapter_changed")
         self.expected = None
         self.not_after = None
 

@@ -8,11 +8,28 @@
 import sys
 
 from .native import NativeAccounts, policy
-from .native_host import HostError
+from .native_host import HostError, checked_adapter
 
 SCHEMA = "loopx_native_account_provider_v1"
 
 
+def adapter_readiness(owner: NativeAccounts) -> dict:
+    try:
+        state = owner._load()
+        saved_policy = state.get("policy")
+        if not isinstance(saved_policy, dict) or type(saved_policy.get("enabled")) is not bool:
+            return {"state": "state_untrusted", "ready": False}
+        adapter = state.get("adapter")
+        if adapter is None or not saved_policy["enabled"]:
+            return {"state": "not_required", "ready": True}
+        checked_adapter(adapter, Path.home())
+    except HostError as error:
+        return {"state": error.code, "ready": False}
+    except (AttributeError, KeyError, OSError, TypeError, ValueError):
+        return {"state": "native_unavailable", "ready": False}
+    return {"state": "ready", "ready": True}
+
+
 def doctor(owner):
     from .relay_lifecycle import consumer_statuses, same_instance
     # Exercises the shipped TS bridge with a synthetic disabled policy; no
@@ -22,10 +39,12 @@ def doctor(owner):
     worker, steward = consumers["worker"], consumers["steward"]
     worker_runtime, steward_runtime = worker.get("runtime"), steward.get("runtime")
     same = same_instance(worker_runtime, steward_runtime) if worker_runtime and steward_runtime else None
-    ready = worker["state"] in ("not_started", "ready") and (
+    adapter = adapter_readiness(owner)
+    ready = adapter["ready"] and worker["state"] in ("not_started", "ready") and (
         steward["state"] == "not_configured" or (steward["ready"] and same is True))
     return {"ok": ready, "schema_version": SCHEMA, "provider_id": "loopx-native-account-provider",
-            "installation_ready": ready, "relay": worker, "consumer_readiness": consumers,
+            "installation_ready": ready, "adapter_readiness": adapter,
+            "relay": worker, "consumer_readiness": consumers,
             "same_relay_instance": same, "message_delivery_verified": False}
 
 

@@ -187,6 +187,7 @@ def test_provider_doctor_checks_real_dependencies_and_does_not_start_unconfigure
     before = list(owner.root.iterdir())
     result = provider.doctor(owner)
     assert result["installation_ready"] and result["relay"]["state"] == "not_started"
+    assert result["adapter_readiness"] == {"state": "not_required", "ready": True}
     assert result["consumer_readiness"]["steward"]["state"] == "not_configured"
     assert result["same_relay_instance"] is None
     assert result["message_delivery_verified"] is False and list(owner.root.iterdir()) == before
@@ -195,6 +196,41 @@ def test_provider_doctor_checks_real_dependencies_and_does_not_start_unconfigure
         provider.doctor(owner)
 
 
+def test_provider_doctor_rejects_stale_native_adapter_with_healthy_relay(owner, tmp_path, monkeypatch):
+    native = importlib.import_module("loopx_codex_provider_routing.native")
+    host = importlib.import_module("loopx_codex_provider_routing.native_host")
+    life = importlib.import_module("loopx_codex_provider_routing.relay_lifecycle")
+    provider = importlib.import_module("loopx_codex_provider_routing.provider")
+    fake_codex = tmp_path / "codex"
+    fake_codex.write_text("")
+    fake_codex.chmod(0o700)
+    route = {"home": tmp_path / "codex-home", "frontend": tmp_path / "frontend"}
+    monkeypatch.setattr(host.native_app, "route", lambda *_: route)
+    original_which = host.shutil.which
+    monkeypatch.setattr(host.shutil, "which", lambda name: str(fake_codex) if name == "codex" else original_which(name))
+    current = host.discover()
+    stale = {**current, "pins": {**current["pins"], "native_account_store.py": "0" * 64}}
+    state = owner._load()
+    native.atomic(owner.path, {**state, "adapter": stale})
+    before = owner.path.read_bytes()
+    runtime = {"instance_id": "relay", "pid": 1, "build": "current"}
+    monkeypatch.setattr(life, "consumer_statuses", lambda _: {
+        name: {"state": "ready", "ready": True, "runtime": runtime} for name in ("worker", "steward")
+    })
+    monkeypatch.setattr(host.NativeHost, "status", lambda *_: pytest.fail("Probe must use adapter checks only"))
+
+    result = provider.doctor(owner)
+    assert result["same_relay_instance"] is True
+    assert result["installation_ready"] is False and result["ok"] is False
+    assert result["adapter_readiness"] == {"state": "adapter_changed", "ready": False}
+    assert owner.path.read_bytes() == before
+
+    native.atomic(owner.path, {**state, "adapter": current})
+    result = provider.doctor(owner)
+    assert result["adapter_readiness"] == {"state": "ready", "ready": True}
+    assert result["installation_ready"] and result["ok"]
+
+
 def test_installed_doctor_separates_steward_readiness_from_worker_health(owner, monkeypatch):
     relay = importlib.import_module("loopx_codex_provider_routing.openai_relay")
     native = importlib.import_module("loopx_codex_provider_routing.native")
@@ -249,6 +285,52 @@ def doctor(credentials):
         thread.join(5)
 
 
+@pytest.mark.parametrize("saved_policy", [
+    {},
+    {"policy": None},
+    {"policy": []},
+    {"policy": {"enabled": "true"}},
+], ids=["missing", "null", "list", "invalid-enabled"])
+@pytest.mark.parametrize("operation", ["doctor", "status"])
+def test_provider_readiness_reports_invalid_saved_policy(owner, tmp_path, monkeypatch, saved_policy, operation):
+    native = importlib.import_module("loopx_codex_provider_routing.native")
+    relay = importlib.import_module("loopx_codex_provider_routing.openai_relay")
+    provider = importlib.import_module("loopx_codex_provider_routing.provider")
+    probe = native.NativeAccounts(tmp_path / "probe")
+    probe.root.mkdir(mode=0o700, parents=True)
+    native.atomic(probe.path, {"schema_version": native.SCHEMA, "adapter": {}, **saved_policy})
+    before = probe.path.read_bytes()
+    credentials = {"token": "worker-" + "w" * 32, "steward_token": "steward-" + "s" * 32}
+    server = relay.RelayServer(owner, {"port": 0, **credentials})
+    native.atomic(probe.root / "api-relay.json", {
+        "schema_version": relay.SCHEMA, "port": server.server_port, **credentials,
+    })
+    thread = threading.Thread(target=server.serve_forever)
+    thread.start()
+    monkeypatch.setenv("PYTHONPATH", str(SRC))
+    command = [sys.executable, "-m", "loopx_codex_provider_routing.provider",
+               "--runtime-root", str(tmp_path / "probe")]
+    payload = None
+    if operation == "doctor":
+        command.append("--doctor")
+    else:
+        payload = json.dumps({"schema_version": provider.SCHEMA, "operation": "status"})
+    try:
+        result = subprocess.run(command, input=payload, capture_output=True, text=True, timeout=8)
+        assert result.returncode == 1 and not result.stderr, result.stderr
+        status = json.loads(result.stdout)
+        assert status["ok"] is False and status["installation_ready"] is False
+        assert status["adapter_readiness"] == {"state": "state_untrusted", "ready": False}
+        assert status["same_relay_instance"] is True
+        assert all(observation["ready"] for observation in status["consumer_readiness"].values())
+        assert probe.path.read_bytes() == before
+        assert all(token not in result.stdout for token in credentials.values())
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(5)
+
+
 @pytest.mark.parametrize("invalid", ["shared", "", "invalid\r\nheader"])
 def test_reconcile_refuses_invalid_steward_credential_before_drain(owner, monkeypatch, invalid):
     life = importlib.import_module("loopx_codex_provider_routing.relay_lifecycle")