#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import json import sys import tempfile from datetime import UTC, datetime, timedelta from pathlib import Path from types import SimpleNamespace HERE = Path(__file__).resolve().parents[1] sys.path.insert(0, str(HERE)) from account_registry import Account, AccountRegistry, AccountRegistryKind from authority_migration import AuthorityMigrationCoordinator, AuthorityProvision, MigrationAcceptance from health_client import AccountSnapshot, HealthStatus from provider_services import AuthorityLifecycleAdapter, LifecycleEvent class FixtureAdmin: def __init__(self, grant: Path) -> None: self.grant = grant self.generation = 1 self.calls: list[str] = [] def login_and_provision(self, account: Account, authority_name: str) -> AuthorityProvision: self.calls.append("login") self.grant.write_text("synthetic-fixture-proxy-key", encoding="utf-8") self.grant.chmod(0o600) return AuthorityProvision(authority_name, f"route-{account.tool}-fixture-20260822", self.grant, self.generation) def status_generation(self, provider: str, account_id: str) -> int: del provider, account_id self.calls.append("status") return self.generation def disable_route(self, route_id: str) -> None: del route_id self.calls.append("disable-route") def remove_route_grants(self, route_id: str) -> None: del route_id self.calls.append("remove-route-grants") def remove_route(self, route_id: str) -> None: del route_id self.calls.append("remove-route") def remove_account(self, provider: str, account_id: str) -> None: del provider, account_id self.calls.append("remove-account") class NativeLifecycle: def reauthenticate(self, account: Account): if account.authority_binding is not None: raise RuntimeError("native fixture received authority binding") yield LifecycleEvent(kind="success", account=account) def add(self, alias: str, *, login_hint=None, slug=None): del alias, login_hint, slug raise RuntimeError("unexpected add") def submit_code(self, session, code: str) -> None: del session, code class NativeHealth: def fetch(self, account: Account, timeout_secs: float = 10.0) -> AccountSnapshot: del timeout_secs if account.authority_binding is not None: raise RuntimeError("native fixture health received authority binding") return AccountSnapshot(HealthStatus.OK, 0, 0, detail="synthetic native ready") class NativeRepair: def add(self, alias: str, *, login_hint=None, slug=None): del alias, login_hint, slug yield LifecycleEvent(kind="success") def reauthenticate(self, account: Account): raise RuntimeError(f"native repair unexpectedly used for {account.tool}") def submit_code(self, session, code: str) -> None: del session, code def acceptance(provider: str) -> MigrationAcceptance: common = dict(cli=True, resume=True, streaming=True, quota_health=True, repair=True, restart=True, exact_route=True, no_provider_credentials=True) if provider == "codex": return MigrationAcceptance(**common, responses=True, realtime=True, app_server=True) return MigrationAcceptance(**common, http_sse=True) def fixture(provider: str, root: Path) -> dict[str, object]: registry = AccountRegistry( base_dir=root / "registry", kind=AccountRegistryKind(provider), legacy_codex_home=root / "native-codex", legacy_claude_home=root / "native-claude", active_claude_json=root / "native-claude.json", ) slug = f"fixture-{provider}" account_home = registry.add_dir(slug, f"Fixture {provider.title()}") if provider == "codex": claims = { "email": "fixture-codex@example.invalid", "https://api.openai.com/auth": { "chatgpt_plan_type": "fixture", "chatgpt_account_id": "acct_fixture_codex", }, } encoded = base64.urlsafe_b64encode(json.dumps(claims).encode("utf-8")).decode("ascii").rstrip("=") account_home.joinpath("auth.json").write_text( json.dumps({"tokens": {"id_token": f"header.{encoded}.signature"}}), encoding="utf-8", ) account = registry.list()[0] grant = root / "fixture-authority-grant" admin = FixtureAdmin(grant) ticks = iter(datetime(2026, 8, 22, 4, 0, tzinfo=UTC) + timedelta(minutes=i) for i in range(20)) coordinator = AuthorityMigrationCoordinator(registry, admin, clock=lambda: next(ticks)) if coordinator.stage_dark(account, authority_name="fixture-authority").state != "dark": raise RuntimeError("dark staging failed") repair_calls: list[list[str]] = [] adapter = AuthorityLifecycleAdapter( NativeRepair(), binary=Path("/fixture/subrouter"), state_dir=Path("/fixture/state"), command_runner=lambda argv, check=False: repair_calls.append(list(argv)) or SimpleNamespace(returncode=0), ) if [event.kind for event in adapter.reauthenticate(registry.list()[0])] != ["started", "success"]: raise RuntimeError("authority repair fixture failed") if not repair_calls or "authority-account" not in repair_calls[0] or "repair" not in repair_calls[0]: raise RuntimeError("repair did not use authority admin") admin.generation = 2 restarted_registry = AccountRegistry( base_dir=registry.base_dir, kind=AccountRegistryKind(provider), legacy_codex_home=root / "native-codex", legacy_claude_home=root / "native-claude", active_claude_json=root / "native-claude.json", ) restarted = AuthorityMigrationCoordinator(restarted_registry, admin, clock=lambda: next(ticks)) checks = acceptance(provider) if checks.missing(provider): raise RuntimeError("fixture acceptance incomplete") promoted = restarted.promote(restarted_registry.list()[0], checks) if promoted.state != "authority" or promoted.generation != 2: raise RuntimeError("authority promotion failed") rolled = restarted.rollback( restarted_registry.list()[0], native_lifecycle=NativeLifecycle(), native_health=NativeHealth() ) if rolled.state != "native" or not rolled.cleanup_complete or restarted_registry.list()[0].authority_binding is not None: raise RuntimeError("fixture rollback failed") if grant.exists(): raise RuntimeError("fixture rollback retained grant material") required_order = ["disable-route", "remove-route-grants", "remove-route", "remove-account"] positions = [admin.calls.index(name) for name in required_order] if positions != sorted(positions): raise RuntimeError("fixture rollback cleanup order invalid") payload = rolled.as_dict() serialized = json.dumps(payload, sort_keys=True) for forbidden in (slug, "synthetic-fixture-proxy-key", f"route-{provider}-fixture-20260822"): if forbidden in serialized: raise RuntimeError("fixture receipt leaked protected/raw identifiers") if payload["provider_credential_copied"] is not False or payload["native_fallback_used"] is not False: raise RuntimeError("fixture receipt violated credential boundary") return payload def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() args.output_dir.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="overdeck-s5-fixtures-") as temporary: root = Path(temporary) for provider in ("codex", "claude"): payload = fixture(provider, root / provider) destination = args.output_dir / f"2026-08-22-subrouter-authority-s5-{provider}.json" destination.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") print(json.dumps({"status": "pass", "providers": ["codex", "claude"]}, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())