feat(codex): ship subscription-first OpenAI API fallback with one native owner

AGENTS.md

@@ -232,8 +232,9 @@ metadata, test, example or narrow-cleanup PRs when all these conditions hold:
 - none of the protected behavior surfaces above is changed;
 - the diff is easy to review and required checks/focused smokes passed;
 - its exact head has the published review, and
-  `uv run --extra test loopx pr-review --goal-id GOAL --check-merge-readiness NUMBER@HEAD_OID`
-  returned ready for that unchanged head;
+  `uv run --extra test loopx pr-review --check-merge-readiness NUMBER@HEAD_OID`
+  returned ready for that unchanged head (add `--goal-id GOAL` for already
+  Goal-managed work; standalone delivery requires no Goal);
 - no private state, raw benchmark evidence, credentials, local paths or
   generated logs are included;
 - no runtime behavior, benchmark adapter, permission, destructive Git or public

apps/presentation/dashboard/smoke/codex-account-clients-fixture.py

@@ -3,10 +3,12 @@
 
 import importlib
 import json
+import os
 from pathlib import Path
 import sys
 import tempfile
 import threading
+import time
 
 repo = Path(__file__).resolve().parents[4]
 sys.path.insert(0, str(repo))
@@ -18,6 +20,9 @@
 
 def main() -> None:
     native = importlib.import_module("loopx_codex_provider_routing.native")
+    keys_module = importlib.import_module("loopx_codex_provider_routing.openai_keys")
+    key_check = "models_readable"
+    native.NativeAccounts.api_keys = lambda self: keys_module.OpenAIKeys(self, checker=lambda _: key_check)
     native.discover = lambda: {"synthetic": True}
     clients = [
         {"pid": 303, "kind": "unmanaged", "process": "virtual-machine", "restartable": False,
@@ -31,12 +36,37 @@ def main() -> None:
            "restartable": True} for i in range(18)],
     ]
     mode, calls = "many", []
+    api_fixture = os.environ.get("LOOPX_API_FALLBACK_FIXTURE") == "1"
+    active, selection, api_slot, recovered, now = "a", "synthetic", None, False, time.time()
+    slots, registration_busy = ["a", "b", "c"], False
+    if api_fixture:
+        relay_module = importlib.import_module("loopx_codex_provider_routing.openai_relay")
+        relay_module.ensure = lambda _: {"slot": "openai-api", "api_key": "synthetic-relay-token", "base_url": "http://127.0.0.1:32100/v1"}
 
     def transport(adapter, operation, **parameters):
+        nonlocal active, selection, api_slot, slots
         calls.append(operation)
         if operation == "status":
-            return {"slots": ["a", "b", "c"], "active_slot": "a", "selection_revision": "synthetic",
-                    "current_login_matches": True, "recovery_required": False}
+            return {"slots": slots, "active_slot": active, "selection_revision": selection,
+                    "current_login_matches": True, "recovery_required": False,
+                    "api_fallback_supported": api_fixture, "api_slot": api_slot}
+        if operation == "enroll_current":
+            assert parameters["expected_revision"] == selection and slots == []
+            if registration_busy:
+                raise native.HostError("waiting_for_idle")
+            slots, active, selection = ["primary"], "primary", selection + "-registered"
+            return transport(adapter, "status")
+        if api_fixture and operation == "enroll_api":
+            api_slot = "openai-api"
+            return transport(adapter, "status")
+        if api_fixture and operation == "observe":
+            return {"observations": [{"slot": slot, "status": "ok", "observed_at": now,
+                     "usage": {"rate_limit": {"primary_window": {"used_percent": 10 if recovered and slot == "b" else 100,
+                                "reset_at": now + 3600}}, "credits": {"has_credits": False}}} for slot in parameters["slots"]]}
+        if api_fixture and operation == "switch":
+            assert parameters["expected_revision"] == selection
+            active, selection = parameters["slot"], selection + "-switched"
+            return transport(adapter, "status") | {"launch_failed": False}
         assert operation == "clients", "Reading settings cannot switch, signal, observe quota or recover"
         if mode == "unavailable":
             raise native.HostError("ownership_unverified")
@@ -45,30 +75,74 @@ def transport(adapter, operation, **parameters):
 
     with tempfile.TemporaryDirectory(prefix="loopx-account-clients-") as directory:
         runtime = Path(directory).resolve()
-        provider = native.NativeAccounts(runtime, transport=transport)
+        provider = native.NativeAccounts(runtime, transport=transport, **({"clock": lambda: now} if api_fixture else {}))
         provider.configure(expected_revision=0, enabled=False, slots=["a", "b", "c"], restart_clients=True)
         before = provider.path.read_bytes()
         calls.clear()
         service = CodexAccountService(runtime)
+        service.watcher.close()
         service.provider = provider
+        service.watcher = service._new_observer(native)
         server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
         server.verbose = False
         server.codex_account_service = service
         thread = threading.Thread(target=server.serve_forever, daemon=True)
         thread.start()
+        observer_hold = None
         print(json.dumps({"origin": f"http://127.0.0.1:{server.server_port}"}), flush=True)
         try:
             for line in sys.stdin:
                 command = line.strip()
                 if command == "inspect":
                     print(json.dumps({"state_unchanged": provider.path.read_bytes() == before,
-                                      "calls": calls, "watcher_started": service.watcher.thread is not None}), flush=True)
+                                      "calls": calls, "policy": provider._load()["policy"],
+                                      "watcher_started": service.watcher.thread is not None}), flush=True)
+                elif command == "authorize-a" and api_fixture:
+                    result = service.call("configure", {"expected_revision": provider.status()["revision"],
+                                          "enabled": True, "slots": ["a"], "api_fallback": False})
+                    assert result["ok"]
+                    print(json.dumps({"policy": result["policy"]}), flush=True)
+                elif command == "observer-stale" and api_fixture:
+                    service.watcher.close()
+                    observer_hold = provider._lock(name="observer.lock")
+                    observer_hold.__enter__()
+                    native.atomic(provider.root / "observer.json", {"schema_version": native.OBSERVER_SCHEMA,
+                                  "build": native.OBSERVER_BUILD, "checked_at": time.time() - 100,
+                                  "deadline_at": time.time() - 1, "phase": "waiting"})
+                    print(json.dumps({"observer_stale": True, "revision": provider._load()["revision"]}), flush=True)
+                elif command == "observer-resume" and api_fixture:
+                    observer_hold.__exit__(None, None, None)
+                    observer_hold = None
+                    print(json.dumps({"observer_resumed": True}), flush=True)
                 elif command in {"many", "sparse", "empty", "partial", "unavailable"}:
                     mode = command
                     print(json.dumps({"mode": mode}), flush=True)
+                elif command in {"key-models_readable", "key-authentication_failed", "key-permission_denied", "key-rate_limited", "key-unavailable"}:
+                    key_check = command.removeprefix("key-")
+                    print(json.dumps({"key_check": key_check}), flush=True)
+                elif command == "subscription-recovered" and api_fixture:
+                    recovered, now = True, now + 301
+                    print(json.dumps({"subscription_recovered": True}), flush=True)
+                elif command == "api-observation" and api_fixture:
+                    keys = provider.api_keys()
+                    keys.record(keys.status()["keys"][0]["id"], status=200, code=None,
+                                model="example-selected-model", headers={
+                                    "x-ratelimit-remaining-requests": "42", "x-ratelimit-remaining-tokens": "12000"})
+                    print(json.dumps({"api_observation": True}), flush=True)
+                elif command == "unregistered":
+                    slots, active, selection, registration_busy = [], None, "unregistered", True
+                    result = service.call("configure", {"expected_revision": provider.status()["revision"],
+                                          "enabled": False, "slots": [], "api_fallback": False})
+                    assert result["ok"]
+                    print(json.dumps({"unregistered": True}), flush=True)
+                elif command == "registration-ready":
+                    registration_busy = False
+                    print(json.dumps({"registration_ready": True}), flush=True)
                 else:
                     raise ValueError("Unknown fixture command")
         finally:
+            if observer_hold:
+                observer_hold.__exit__(None, None, None)
             server.shutdown()
             server.server_close()
             thread.join(timeout=5)

apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs

@@ -232,7 +232,10 @@ try {
     // the parent message refresh must not replace it with the old page count.
     await fixture('report-window');
     const nextPage = english.getByRole('button', {name: 'Load next page', exact: true});
-    await nextPage.focus(); await page.keyboard.press('Enter'); await rows(13);
+    // A parent refresh can temporarily aria-disable paging after the fixture
+    // returns. Click waits for actionability; raw Enter can be ignored while
+    // busy. Keyboard activation and result focus are exercised above.
+    await nextPage.click(); await rows(13);
     assert.equal(await english.getByRole('button', {name: 'Load next page', exact: true}).count(), 0);
     assert.equal(await page.evaluate(() => document.activeElement?.textContent), '13 / 13 shown');
     await english.getByRole('button', {name: 'Awaiting follow-up', exact: true}).click(); await rows(11);