diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index 79f5039036..e3af60c5ff 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -53,12 +53,18 @@ request count. `--runtime-root` selects an existing owner; default is `~/.loopx` The doctor distinguishes installation readiness from message delivery, and never uses a model or checks an API key against an upstream service. An unstarted relay is a valid installation state; a healthy legacy or mismatched running daemon is -not a qualified installation. Keep original-message acceptance as a separate -end-to-end check. +not a qualified installation. Since 0.15.4, `consumer_readiness` reports worker +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. +Keep original-message acceptance as a separate end-to-end check. For upgrades, install the reviewed target into a new immutable environment and -invoke its provider with `operation: "reconcile", execute: true`. The model -transport's ordinary `ensure` also performs this reconciliation. An authenticated +invoke its provider with `operation: "reconcile", execute: true`. This checks the +configured steward route as well as the worker, without creating credentials. +The model transport's ordinary `ensure` also performs reconciliation for its +own route. An authenticated new-protocol daemon with a different loaded build stops new admissions, drains already accepted requests and releases its existing owner lock before one new daemon starts. The model, route credentials, key policy and message queue stay diff --git a/packages/loopx-codex-provider-routing/pyproject.toml b/packages/loopx-codex-provider-routing/pyproject.toml index 43727653c7..945ac14367 100644 --- a/packages/loopx-codex-provider-routing/pyproject.toml +++ b/packages/loopx-codex-provider-routing/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "loopx-codex-provider-routing" -version = "0.15.3" +version = "0.15.4" description = "Native Codex account slots and exhausted-account handoff for LoopX" readme = "README.md" requires-python = ">=3.11" diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml index b6706285c9..67d6d014c9 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native-extension.toml @@ -1,6 +1,6 @@ schema_version = "loopx_extension_manifest_v0" id = "loopx-native-account-provider" -version = "0.15.3" +version = "0.15.4" requires_loopx_api = ">=1,<2" permissions = ["model-provider.local-read", "model-provider.transport-manage"] diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/openai_relay.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/openai_relay.py index 35237c4285..0ff0c41e98 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/openai_relay.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/openai_relay.py @@ -81,7 +81,11 @@ def route(owner: NativeAccounts) -> dict: def consumer_token(value: dict, consumer: str) -> str: token = value.get("steward_token" if consumer == "steward" else "token") - if not isinstance(token, str) or not token or consumer == "steward" and token == value.get("token"): + if not isinstance(token, str) or not token or ( + consumer == "steward" and ( + token == value.get("token") or any(not 33 <= ord(char) <= 126 for char in token) + ) + ): raise HostError("api_relay_upgrade_required") return token diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/provider.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/provider.py index f4ed67aafe..a0eaf01bce 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/provider.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/provider.py @@ -14,15 +14,19 @@ def doctor(owner): - from .relay_lifecycle import status + from .relay_lifecycle import consumer_statuses, same_instance # Exercises the shipped TS bridge with a synthetic disabled policy; no # credential discovery, account switching or upstream inference is invoked. policy({"operation": "api-role", "consumer": "worker", "policy": {}}) - relay = status(owner) - ready = relay["state"] in ("not_started", "ready") + consumers = consumer_statuses(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 ( + 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": relay, - "message_delivery_verified": False} + "installation_ready": ready, "relay": worker, "consumer_readiness": consumers, + "same_relay_instance": same, "message_delivery_verified": False} def main(): @@ -44,7 +48,9 @@ def main(): from .openai_relay import route from .relay_lifecycle import reconcile with owner._lock(name="api-relay-state.lock", wait_seconds=5): - reconcile(owner, route(owner), legacy_pid=request.get("legacy_pid")) + value = route(owner) + consumer = "steward" if "steward_token" in value else "worker" + reconcile(owner, value, consumer=consumer, legacy_pid=request.get("legacy_pid")) elif operation not in ("status", "reconcile"): raise HostError("invalid_request") result = doctor(owner) diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/relay_lifecycle.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/relay_lifecycle.py index 64d6eb8d12..d369bbde6f 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/relay_lifecycle.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/relay_lifecycle.py @@ -22,9 +22,9 @@ def runtime(value: dict, *, consumer="worker") -> dict | None: from .openai_relay import consumer_token, RUNTIME_SCHEMA prefix = "/steward" if consumer == "steward" else "" - request = urllib.request.Request(f"http://127.0.0.1:{value['port']}{prefix}/runtime", - headers={"Authorization": "Bearer " + consumer_token(value, consumer)}) try: + request = urllib.request.Request(f"http://127.0.0.1:{value['port']}{prefix}/runtime", + headers={"Authorization": "Bearer " + consumer_token(value, consumer)}) with urllib.request.build_opener(NoRedirect()).open(request, timeout=1) as response: data = response.read(4097) row = json.loads(data) if len(data) <= 4096 else None @@ -40,13 +40,19 @@ def runtime(value: dict, *, consumer="worker") -> dict | None: return None -def status(owner: NativeAccounts, *, consumer="worker") -> dict: - from .openai_relay import route, healthy, BUILD - if consumer not in ("worker", "steward"): - raise HostError("invalid_policy") +def _read_route(owner: NativeAccounts) -> dict | None: + from .openai_relay import route try: - value = route(owner) + return route(owner) except FileNotFoundError: + return None + + +def _consumer_status(value: dict | None, consumer: str) -> dict: + from .openai_relay import healthy, BUILD + if consumer == "steward" and (value is None or "steward_token" not in value): + return {"state": "not_configured", "ready": False, "expected_build": BUILD} + if value is None: return {"state": "not_started", "ready": False, "expected_build": BUILD} live = runtime(value, consumer=consumer) state = ("ready" if live["build"] == BUILD and live["accepting"] else @@ -56,6 +62,22 @@ def status(owner: NativeAccounts, *, consumer="worker") -> dict: "runtime": live} +def status(owner: NativeAccounts, *, consumer="worker") -> dict: + if consumer not in ("worker", "steward"): + raise HostError("invalid_policy") + return _consumer_status(_read_route(owner), consumer) + + +def consumer_statuses(owner: NativeAccounts) -> dict[str, dict]: + """Read both independent routes from one persisted credential snapshot.""" + value = _read_route(owner) + return {consumer: _consumer_status(value, consumer) for consumer in ("worker", "steward")} + + +def same_instance(left: dict, right: dict) -> bool: + return all(left[key] == right[key] for key in ("instance_id", "pid", "build")) + + def _owner_released(owner: NativeAccounts) -> bool: try: with owner._lock(name="api-relay.lock"): @@ -97,11 +119,16 @@ def _drain(value: dict, live: dict): def reconcile(owner: NativeAccounts, value: dict, *, consumer="worker", legacy_pid=None, timeout=10): """Called under the existing state lock; retain the server lock through drain.""" - from .openai_relay import healthy, BUILD + from .openai_relay import consumer_token, healthy, BUILD + if consumer == "steward": + consumer_token(value, consumer) # Invalid configuration cannot be repaired by a drain. live = runtime(value) - if (live and live["build"] == BUILD and live["accepting"] - and (consumer == "worker" or healthy(value, consumer))): - return + if live and live["build"] == BUILD and live["accepting"]: + if consumer == "worker": + return + scoped = runtime(value, consumer=consumer) + if scoped and scoped["accepting"] and same_instance(live, scoped): + return deadline = time.monotonic() + timeout if live and live["accepting"]: _drain(value, live) diff --git a/packages/loopx-codex-provider-routing/tests/test_provider_lifecycle.py b/packages/loopx-codex-provider-routing/tests/test_provider_lifecycle.py index 92dcf42be6..f88b5989b8 100644 --- a/packages/loopx-codex-provider-routing/tests/test_provider_lifecycle.py +++ b/packages/loopx-codex-provider-routing/tests/test_provider_lifecycle.py @@ -128,10 +128,19 @@ def test_real_detached_old_build_replaced_once_with_same_routes_and_private_stat value["steward_token"] = "replacement-steward-" + "q" * 32 native.atomic(owner.root / "api-relay.json", value) assert not life.status(owner, consumer="steward")["ready"] - with owner._lock(name="api-relay-state.lock"): - life.reconcile(owner, value, consumer="steward") - assert life.status(owner, consumer="steward")["ready"] + staged_route = (owner.root / "api-relay.json").read_bytes() + result = subprocess.run( + [sys.executable, "-m", "loopx_codex_provider_routing.provider", + "--runtime-root", str(owner.root.parents[1])], + input=json.dumps({"schema_version": "loopx_native_account_provider_v1", + "operation": "reconcile", "execute": True}), + capture_output=True, text=True, timeout=20) + assert result.returncode == 0, result.stdout + outcome = json.loads(result.stdout) + assert outcome["effect_performed"] and outcome["same_relay_instance"] + assert outcome["consumer_readiness"]["steward"]["ready"] assert life.status(owner)["runtime"]["pid"] != current["runtime"]["pid"] + assert (owner.root / "api-relay.json").read_bytes() == staged_route finally: live = life.runtime(value) if live: @@ -178,12 +187,91 @@ 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["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 monkeypatch.setenv("PATH", "") with pytest.raises(Exception, match="node_unavailable"): provider.doctor(owner) +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") + worker_token, steward_token = "worker-" + "w" * 32, "steward-" + "s" * 32 + monkeypatch.setenv("PYTHONPATH", str(SRC)) + server = relay.RelayServer(owner, {"port": 0, "token": worker_token, "steward_token": steward_token}) + thread = threading.Thread(target=server.serve_forever) + thread.start() + route_file = owner.root / "api-relay.json" + base = {"schema_version": relay.SCHEMA, "port": server.server_port, "token": worker_token} + def doctor(credentials): + native.atomic(route_file, {**base, **credentials}) + before = route_file.read_bytes() + result = subprocess.run( + [sys.executable, "-m", "loopx_codex_provider_routing.provider", + "--runtime-root", str(owner.root.parents[1]), "--doctor"], + capture_output=True, text=True, timeout=8) + assert route_file.read_bytes() == before + assert worker_token not in result.stdout and steward_token not in result.stdout + return result.returncode, json.loads(result.stdout) + try: + code, both = doctor({"steward_token": steward_token}) + assert code == 0 + code, stale = doctor({"steward_token": "stale-steward-" + "x" * 32}) + # Extension installation uses the process status, not the JSON output. + assert code == 1 + assert stale["installation_ready"] is False + assert stale["consumer_readiness"]["worker"]["ready"] is True + assert stale["consumer_readiness"]["steward"]["state"] == "unavailable" + assert stale["same_relay_instance"] is None + assert request(server, "/health", worker_token)[0] == 200 + assert both["same_relay_instance"] is True + assert all(row["ready"] for row in both["consumer_readiness"].values()) + assert both["relay"] == both["consumer_readiness"]["worker"] + assert both["message_delivery_verified"] is False + for invalid in (worker_token, "invalid\r\nheader"): + code, unusable = doctor({"steward_token": invalid}) + assert code == 1 + assert unusable["consumer_readiness"]["steward"]["state"] == "unavailable" + assert unusable["consumer_readiness"]["worker"]["ready"] + code, optional = doctor({}) + assert code == 0 and optional["consumer_readiness"]["steward"]["state"] == "not_configured" + with server._admission_lock: + server.draining = True + code, draining = doctor({"steward_token": steward_token}) + assert code == 1 + assert {v["state"] for v in draining["consumer_readiness"].values()} == {"draining"} + assert draining["same_relay_instance"] is True + 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") + monkeypatch.setattr(life, "runtime", lambda *_: pytest.fail("No relay may be drained")) + with pytest.raises(life.HostError, match="api_relay_upgrade_required"): + life.reconcile(owner, {"token": "shared", "steward_token": invalid}, consumer="steward") + + +def test_installed_doctor_does_not_combine_different_relay_instances(owner, monkeypatch): + provider = importlib.import_module("loopx_codex_provider_routing.provider") + relay = importlib.import_module("loopx_codex_provider_routing.openai_relay") + life = importlib.import_module("loopx_codex_provider_routing.relay_lifecycle") + monkeypatch.setattr(relay, "route", lambda _: {"steward_token": "configured"}) + monkeypatch.setattr(life, "runtime", lambda _value, *, consumer: { + "build": relay.BUILD, "accepting": True, "pid": 17, + "instance_id": ("a" if consumer == "worker" else "b") * 32, + }) + result = provider.doctor(owner) + assert all(row["ready"] for row in result["consumer_readiness"].values()) + assert result["same_relay_instance"] is False + assert result["installation_ready"] is False + + def test_existing_healthy_legacy_is_not_silently_adopted(owner): relay = importlib.import_module("loopx_codex_provider_routing.openai_relay") native = importlib.import_module("loopx_codex_provider_routing.native")