from __future__ import annotations

import hashlib
from concurrent.futures import ThreadPoolExecutor
import json
import stat
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import authority_client
import provider_services
from account_registry import (
    Account,
    AccountRef,
    AccountRegistryKind,
    AuthorityBinding,
    AuthorityMode,
)
from authority_client import (
    AuthorityConfigurationError,
    GatewayStatus,
    GatewayUsage,
    GatewayUsageWindow,
    build_authority_launch,
    fetch_gateway_status,
    fetch_gateway_usage,
    gateway_health_snapshot,
    materialize_gateway_home,
    resolve_authority_endpoint,
)
from health_client import AccountSnapshot, HealthStatus
from provider_services import (
    AuthorityHealthAdapter,
    ProviderServiceMap,
    ProviderServices,
)


_READY_STATUS = {
    "state": "ready",
    "provider": "codex",
    "route_fingerprint": authority_client._fingerprint("route-fixture"),
    "grant_fingerprint": "0123456789ab",
    "grant_expires_at": "2099-01-01T00:00:00Z",
    "grant_revoked": False,
    "account_availability": "available",
}


class _StatusHandler(BaseHTTPRequestHandler):
    body = dict(_READY_STATUS)
    requests: list[tuple[str, str | None]] = []

    def do_GET(self) -> None:
        self.__class__.requests.append((self.path, self.headers.get("Authorization")))
        payload = json.dumps(self.__class__.body).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def do_POST(self) -> None:
        self.__class__.requests.append((self.path, self.headers.get("Authorization")))
        payload = json.dumps(
            {
                "checked_at": "2026-08-23T12:00:00Z",
                "fresh": True,
                "windows": [
                    {
                        "name": "primary",
                        "used_percent": 17.4,
                        "limit_window_seconds": 18000,
                        "reset_after_seconds": 600,
                    },
                    {
                        "name": "secondary",
                        "used_percent": 42,
                        "limit_window_seconds": 604800,
                        "reset_after_seconds": 1200,
                    },
                    {
                        "name": "GPT-Spark/primary",
                        "feature": "GPT-Spark",
                        "used_percent": 3,
                        "limit_window_seconds": 18000,
                        "reset_after_seconds": 300,
                    },
                ],
            }
        ).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, _format: str, *_args: object) -> None:
        return


@pytest.fixture
def authority(tmp_path: Path):
    _StatusHandler.requests = []
    _StatusHandler.body = dict(_READY_STATUS)
    server = ThreadingHTTPServer(("127.0.0.1", 0), _StatusHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    base_dir = tmp_path / "runtime"
    base_dir.mkdir(mode=0o700)
    base_dir.chmod(0o700)
    origin = f"http://127.0.0.1:{server.server_port}"
    endpoints = base_dir / "authority_endpoints.json"
    endpoints.write_text(
        json.dumps({"authorities": {"workstation": {"origin": origin}}}),
        encoding="utf-8",
    )
    endpoints.chmod(0o600)
    grant = tmp_path / "fixture.grant"
    grant.write_text("synthetic-route-key", encoding="utf-8")
    grant.chmod(0o600)
    binding = AuthorityBinding(
        mode=AuthorityMode.SUBROUTER_DARK,
        authority_name="workstation",
        route_id="route-fixture",
        provider="codex",
        proxy_grant_ref=grant.resolve(),
    )
    account = Account(
        ref=AccountRef("codex", "fixture"),
        alias="Fixture",
        account_home=tmp_path / "native" / "CODEX_HOME",
        email=None,
        plan=None,
        account_id=None,
        authority_binding=binding,
    )
    try:
        yield base_dir, account, binding, origin
    finally:
        server.shutdown()
        thread.join()
        server.server_close()


def test_build_authority_launch_uses_gateway_only_home_and_strips_native_auth(
    authority,
) -> None:
    base_dir, account, _binding, origin = authority

    launch = build_authority_launch(
        base_dir,
        account,
        "codex",
        ["exec", "task"],
        inherited_environment={
            "OPENAI_API_KEY": "must-not-survive",
            "OPENAI_BASE_URL": "https://wrong.invalid",
            "ANTHROPIC_AUTH_TOKEN": "must-not-survive",
            "ANTHROPIC_BASE_URL": "https://wrong.invalid",
            "CLAUDE_CONFIG_DIR": "/native",
            "CODEX_HOME": "/native",
            "SUBROUTER_PROXY_KEY": "wrong-grant",
            "HTTPS_PROXY": "http://wrong.invalid",
            "SAFE": "preserved",
        },
    )

    assert launch.environment["SAFE"] == "preserved"
    assert "OPENAI_API_KEY" not in launch.environment
    assert "OPENAI_BASE_URL" not in launch.environment
    assert "ANTHROPIC_AUTH_TOKEN" not in launch.environment
    assert "ANTHROPIC_BASE_URL" not in launch.environment
    assert "CLAUDE_CONFIG_DIR" not in launch.environment
    assert "HTTPS_PROXY" not in launch.environment
    assert launch.environment["SUBROUTER_PROXY_KEY"] == "synthetic-route-key"
    assert launch.environment["HOME"] == str(launch.gateway_home)
    assert launch.environment["CODEX_HOME"] == str(launch.gateway_home)
    assert launch.gateway_home != account.account_home
    assert stat.S_IMODE(launch.gateway_home.stat().st_mode) == 0o700
    assert sorted(path.name for path in launch.gateway_home.iterdir()) == [
        "config.toml",
        "gateway-manifest.json",
    ]
    assert not any((launch.gateway_home / name).exists() for name in ("auth.json", ".credentials.json", "claude.json"))
    config = (launch.gateway_home / "config.toml").read_text(encoding="utf-8")
    assert f'{origin}/r/route-fixture/v1' in config
    assert "synthetic-route-key" not in config
    manifest = json.loads(
        (launch.gateway_home / "gateway-manifest.json").read_text(encoding="utf-8")
    )
    assert manifest["provider_auth_files_present"] is False
    assert "route-fixture" not in json.dumps(manifest)
    assert _StatusHandler.requests == [
        ("/r/route-fixture/_subrouter/status", "Bearer synthetic-route-key")
    ]


def test_launch_uses_one_grant_snapshot_for_status_and_execution(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, _binding, _origin = authority
    reads: list[str] = []
    status_keys: list[str] = []

    def read_once(_path: Path) -> str:
        value = "first-key" if not reads else "second-key"
        reads.append(value)
        return value

    def status_with_key(
        _endpoint: object,
        _binding: object,
        key: str,
        *,
        timeout_secs: float,
    ) -> GatewayStatus:
        del timeout_secs
        status_keys.append(key)
        return GatewayStatus("ready", "Gateway: ready", 1.0)

    monkeypatch.setattr(authority_client, "_read_proxy_key", read_once)
    monkeypatch.setattr(authority_client, "_fetch_gateway_status", status_with_key)

    launch = build_authority_launch(base_dir, account, "codex", ["exec", "task"])

    assert reads == ["first-key"]
    assert status_keys == ["first-key"]
    assert launch.environment["SUBROUTER_PROXY_KEY"] == "first-key"


def test_claude_launch_uses_gateway_only_home_and_exact_route(authority) -> None:
    base_dir, _account, binding, origin = authority
    claude_binding = AuthorityBinding(
        mode=binding.mode,
        authority_name=binding.authority_name,
        route_id=binding.route_id,
        provider="claude",
        proxy_grant_ref=binding.proxy_grant_ref,
    )
    account = Account(
        ref=AccountRef("claude", "fixture"),
        alias="Fixture",
        account_home=base_dir / "native" / "CLAUDE_CONFIG_DIR",
        email=None,
        plan=None,
        account_id=None,
        authority_binding=claude_binding,
    )

    _StatusHandler.body = {**_READY_STATUS, "provider": "claude"}
    launch = build_authority_launch(
        base_dir,
        account,
        "claude",
        ["--print", "task"],
        inherited_environment={
            "CLAUDE_CONFIG_DIR": "/native",
            "CLAUDE_CODE_OAUTH_TOKEN": "remove",
            "SUBROUTER_PROXY_KEY": "wrong-grant",
            "OPENAI_BASE_URL": "https://wrong.invalid",
        },
    )

    assert launch.environment["CLAUDE_CONFIG_DIR"] == str(launch.gateway_home)
    assert launch.environment["ANTHROPIC_BASE_URL"] == f"{origin}/r/route-fixture"
    assert launch.environment["ANTHROPIC_AUTH_TOKEN"] == "synthetic-route-key"
    assert "CLAUDE_CODE_OAUTH_TOKEN" not in launch.environment
    assert "SUBROUTER_PROXY_KEY" not in launch.environment
    assert "OPENAI_BASE_URL" not in launch.environment
    assert launch.environment["HOME"] == str(launch.gateway_home)
    assert sorted(path.name for path in launch.gateway_home.iterdir()) == [
        "gateway-manifest.json"
    ]
    assert not any((launch.gateway_home / name).exists() for name in ("auth.json", ".credentials.json", "claude.json"))


def test_gateway_home_refuses_provider_auth_files(authority) -> None:
    base_dir, account, binding, _origin = authority
    endpoint = resolve_authority_endpoint(base_dir, binding)
    root = base_dir / "gateway-homes" / "codex"
    first = materialize_gateway_home(base_dir, account, endpoint, binding)
    (first / "auth.json").write_text("{}", encoding="utf-8")

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        materialize_gateway_home(base_dir, account, endpoint, binding)

    assert root.exists()


def test_gateway_home_interrupted_materialization_publishes_nothing(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, binding, _origin = authority
    rebound = Account(
        ref=account.ref,
        alias=account.alias,
        account_home=account.account_home,
        email=account.email,
        plan=account.plan,
        account_id=account.account_id,
        authority_binding=AuthorityBinding(
            mode=binding.mode,
            authority_name=binding.authority_name,
            route_id="route-interrupted",
            provider=binding.provider,
            proxy_grant_ref=binding.proxy_grant_ref,
        ),
    )
    endpoint = resolve_authority_endpoint(base_dir, rebound.authority_binding)
    original_write = authority_client._atomic_write
    writes = 0

    def interrupted_write(path: Path, body: str, mode: int) -> None:
        nonlocal writes
        writes += 1
        if writes == 2:
            raise OSError("interrupted")
        original_write(path, body, mode)

    monkeypatch.setattr(authority_client, "_atomic_write", interrupted_write)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        materialize_gateway_home(base_dir, rebound, endpoint, rebound.authority_binding)

    root = authority_client._gateway_root(base_dir, rebound, rebound.authority_binding)
    assert not root.exists()
    assert not list(root.parent.glob(f".{root.name}.*"))


def test_gateway_status_renews_rejected_local_grant_once(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, _account, binding, _origin = authority
    calls: list[str] = []
    renewed: list[tuple[str, str]] = []
    future_ms = int((time.time() + 7200) * 1000)

    def fetch(_endpoint, _binding, key: str, *, timeout_secs: float) -> GatewayStatus:
        del timeout_secs
        calls.append(key)
        if len(calls) <= 2:
            raise authority_client._GatewayGrantRejected("Gateway: unavailable")
        return GatewayStatus("ready", "Gateway: ready", time.time(), future_ms)

    def renew(endpoint, candidate, key: str, *, timeout_secs: float) -> int:
        del timeout_secs
        renewed.append((candidate.route_id, key))
        assert endpoint.name == binding.authority_name
        return future_ms

    monkeypatch.setattr(authority_client, "_can_renew_gateway_grant", lambda *_args: True)
    monkeypatch.setattr(authority_client, "_fetch_gateway_status", fetch)
    monkeypatch.setattr(authority_client, "_renew_gateway_grant", renew)

    status = fetch_gateway_status(base_dir, binding)

    assert status.state == "ready"
    assert calls == ["synthetic-route-key", "synthetic-route-key", "synthetic-route-key"]
    assert renewed == [("route-fixture", "synthetic-route-key")]


def test_gateway_status_recovers_fifty_concurrent_reconnects_after_grant_rejection(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, _account, binding, _origin = authority
    renewed = threading.Event()
    renewal_lock = threading.Lock()
    renewals = 0
    future_ms = int((time.time() + 24 * 60 * 60) * 1000)

    def fetch(_endpoint, _binding, _key: str, *, timeout_secs: float) -> GatewayStatus:
        del timeout_secs
        if not renewed.is_set():
            raise authority_client._GatewayGrantRejected("Gateway: unavailable")
        return GatewayStatus("ready", "Gateway: ready", time.time(), future_ms)

    def renew(_endpoint, _binding, _key: str, *, timeout_secs: float) -> int:
        nonlocal renewals
        del timeout_secs
        with renewal_lock:
            renewals += 1
            renewed.set()
        return future_ms

    monkeypatch.setattr(authority_client, "_can_renew_gateway_grant", lambda *_args: True)
    monkeypatch.setattr(authority_client, "_fetch_gateway_status", fetch)
    monkeypatch.setattr(authority_client, "_renew_gateway_grant", renew)

    with ThreadPoolExecutor(max_workers=50) as pool:
        statuses = list(pool.map(lambda _index: fetch_gateway_status(base_dir, binding), range(50)))

    assert all(status.state == "ready" for status in statuses)
    assert renewals == 1


def test_gateway_status_rejected_nonlocal_grant_never_uses_admin_renewal(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, _account, binding, _origin = authority
    renewals: list[str] = []

    monkeypatch.setattr(
        authority_client,
        "_fetch_gateway_status",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            authority_client._GatewayGrantRejected("Gateway: unavailable")
        ),
    )
    monkeypatch.setattr(
        authority_client,
        "_renew_gateway_grant",
        lambda _endpoint, candidate, _key, *, timeout_secs: renewals.append(
            candidate.route_id
        ),
    )

    with pytest.raises(AuthorityConfigurationError, match="unavailable"):
        fetch_gateway_status(base_dir, binding)

    assert renewals == []


def test_gateway_status_proactively_renews_near_expiry(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, _account, binding, _origin = authority
    now = time.time()
    statuses = [
        GatewayStatus("ready", "Gateway: ready", now, int((now + 30 * 60) * 1000)),
        GatewayStatus("ready", "Gateway: ready", now, int((now + 30 * 60) * 1000)),
        GatewayStatus("ready", "Gateway: ready", now, int((now + 24 * 60 * 60) * 1000)),
    ]
    renewals: list[str] = []

    monkeypatch.setattr(authority_client, "_can_renew_gateway_grant", lambda *_args: True)
    monkeypatch.setattr(
        authority_client,
        "_fetch_gateway_status",
        lambda *_args, **_kwargs: statuses.pop(0),
    )
    monkeypatch.setattr(
        authority_client,
        "_renew_gateway_grant",
        lambda _endpoint, candidate, _key, *, timeout_secs: renewals.append(candidate.route_id) or int((now + 24 * 60 * 60) * 1000),
    )

    status = fetch_gateway_status(base_dir, binding)

    assert status.grant_expires_at_ms is not None
    assert status.grant_expires_at_ms > int((now + 23 * 60 * 60) * 1000)
    assert renewals == ["route-fixture"]
    assert statuses == []


def test_gateway_grant_renewal_sends_hash_not_proxy_key(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    _base_dir, _account, binding, _origin = authority
    captured: list[tuple[str, str | None, bytes]] = []

    class _Admin(BaseHTTPRequestHandler):
        def do_POST(self) -> None:
            length = int(self.headers.get("Content-Length", "0"))
            body = self.rfile.read(length) if length else b""
            captured.append((self.path, self.headers.get("Authorization"), body))
            payload = json.dumps({"expires_at": "2099-01-01T00:00:00Z"}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

        def log_message(self, _format: str, *_args: object) -> None:
            return

    server = ThreadingHTTPServer(("127.0.0.1", 0), _Admin)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    origin = f"http://127.0.0.1:{server.server_port}"
    endpoint = authority_client.AuthorityEndpoint(binding.authority_name, origin)
    monkeypatch.setattr(authority_client, "LOCAL_AUTHORITY_NAME", binding.authority_name)
    monkeypatch.setattr(authority_client, "LOCAL_AUTHORITY_ORIGIN", origin)
    monkeypatch.setattr(authority_client, "_read_authority_admin_token", lambda: "a" * 64)
    try:
        authority_client._renew_gateway_grant(
            endpoint, binding, "synthetic-route-key", timeout_secs=5.0
        )
    finally:
        server.shutdown()
        thread.join()
        server.server_close()

    assert len(captured) == 1
    path, authorization, body = captured[0]
    assert authorization == "Bearer " + "a" * 64
    assert body == b""
    assert "synthetic-route-key" not in path
    assert hashlib.sha256(b"synthetic-route-key").hexdigest() in path
    assert "route-fixture" in path


def test_gateway_usage_reads_sanitized_local_admin_windows(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, binding, origin = authority
    active = Account(
        ref=account.ref,
        alias=account.alias,
        account_home=account.account_home,
        email=account.email,
        plan=account.plan,
        account_id=account.account_id,
        authority_binding=AuthorityBinding(
            mode=AuthorityMode.SUBROUTER,
            authority_name=binding.authority_name,
            route_id=binding.route_id,
            provider=binding.provider,
            proxy_grant_ref=binding.proxy_grant_ref,
        ),
    )
    monkeypatch.setattr(authority_client, "LOCAL_AUTHORITY_NAME", binding.authority_name)
    monkeypatch.setattr(authority_client, "LOCAL_AUTHORITY_ORIGIN", origin)
    monkeypatch.setattr(authority_client, "_read_authority_admin_token", lambda: "a" * 64)
    _StatusHandler.requests = []

    usage = fetch_gateway_usage(base_dir, active)

    assert usage.fresh is True
    assert [(w.name, w.used_percent, w.feature) for w in usage.windows] == [
        ("primary", 17.4, None),
        ("secondary", 42.0, None),
        ("GPT-Spark/primary", 3.0, "GPT-Spark"),
    ]
    assert len(_StatusHandler.requests) == 1
    path, authorization = _StatusHandler.requests[0]
    assert path.startswith("/_subrouter/authority-admin/v1/usage?")
    assert "provider=codex" in path
    assert "account_id=fixture" in path
    assert "synthetic-route-key" not in path
    assert authorization == "Bearer " + "a" * 64


def test_authority_health_adapter_merges_gateway_usage_limits(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, binding, _origin = authority
    active = Account(
        ref=account.ref,
        alias=account.alias,
        account_home=account.account_home,
        email=account.email,
        plan=account.plan,
        account_id=account.account_id,
        authority_binding=AuthorityBinding(
            mode=AuthorityMode.SUBROUTER,
            authority_name=binding.authority_name,
            route_id=binding.route_id,
            provider=binding.provider,
            proxy_grant_ref=binding.proxy_grant_ref,
        ),
    )

    class _Registry:
        def __init__(self, path: Path) -> None:
            self.base_dir = path

    monkeypatch.setattr(
        provider_services,
        "gateway_health_snapshot",
        lambda *_args, **_kwargs: AccountSnapshot(
            HealthStatus.OK, None, None, checked_at=1.0, detail="Gateway: ready"
        ),
    )
    monkeypatch.setattr(
        provider_services,
        "fetch_gateway_usage",
        lambda *_args, **_kwargs: GatewayUsage(
            checked_at=1000.0,
            fresh=True,
            windows=(
                GatewayUsageWindow("primary", 17.4, 18000, 600),
                GatewayUsageWindow("secondary", 42.0, 604800, 1200),
                GatewayUsageWindow("GPT-Spark/primary", 3.0, 18000, 300, "GPT-Spark"),
            ),
        ),
    )
    adapter = AuthorityHealthAdapter(_Registry(base_dir))  # type: ignore[arg-type]

    snapshot = adapter.fetch(active)

    assert snapshot.status == HealthStatus.OK
    assert snapshot.primary_used_pct == 17
    assert snapshot.secondary_used_pct == 42
    assert snapshot.primary_reset_at == 1600.0
    assert snapshot.secondary_reset_at == 2200.0
    assert snapshot.detail == "Gateway: ready"
    assert [(item.kind, item.group, item.percent, item.resets_at) for item in snapshot.named_limits] == [
        ("five_hour", "GPT-Spark", 3, 1300.0)
    ]


def test_authority_health_adapter_uses_duration_before_generic_window_name() -> None:
    snapshot = AuthorityHealthAdapter._quota_snapshot(
        AccountSnapshot(
            HealthStatus.OK, None, None, checked_at=1.0, detail="Gateway: ready"
        ),
        GatewayUsage(
            checked_at=1000.0,
            fresh=True,
            windows=(GatewayUsageWindow("primary", 1.0, 604800, 1200),),
        ),
    )

    assert snapshot.primary_used_pct is None
    assert snapshot.primary_reset_at is None
    assert snapshot.secondary_used_pct == 1
    assert snapshot.secondary_reset_at == 2200.0


def test_gateway_status_is_sanitized_and_missing_grant_fails_closed(authority) -> None:
    base_dir, account, binding, _origin = authority

    snapshot = gateway_health_snapshot(base_dir, account)

    assert snapshot.status == HealthStatus.OK
    assert snapshot.detail == "Gateway: ready"
    binding.proxy_grant_ref.unlink()

    unavailable = gateway_health_snapshot(base_dir, account)

    assert unavailable.status == HealthStatus.UNKNOWN
    assert unavailable.detail == "Gateway: migration required"
    assert "fixture" not in unavailable.detail


def test_gateway_status_maps_exact_exhausted_state(authority) -> None:
    base_dir, _account, binding, _origin = authority
    _StatusHandler.body = {
        **_READY_STATUS,
        "state": "exhausted",
        "account_availability": "exhausted",
    }

    status = fetch_gateway_status(base_dir, binding)

    assert status.state == "exhausted"
    assert status.detail == "Gateway: quota exhausted"


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("provider", "claude"),
        ("route_fingerprint", "ffffffffffff"),
        ("grant_fingerprint", "not-bounded"),
        ("grant_expires_at", "2000-01-01T00:00:00Z"),
        ("grant_revoked", True),
        ("account_availability", "unavailable"),
        ("credential", "must-be-rejected"),
    ],
)
def test_gateway_status_rejects_wrong_or_unbounded_schema(
    authority,
    field: str,
    value: object,
) -> None:
    base_dir, _account, binding, _origin = authority
    _StatusHandler.body = {**_READY_STATUS, field: value}

    with pytest.raises(AuthorityConfigurationError, match="unavailable"):
        fetch_gateway_status(base_dir, binding)


def test_gateway_status_refuses_redirect_without_forwarding_grant(authority) -> None:
    base_dir, _account, binding, _origin = authority
    captured: list[str | None] = []

    class _Sink(BaseHTTPRequestHandler):
        def do_GET(self) -> None:
            captured.append(self.headers.get("Authorization"))
            self.send_response(200)
            self.end_headers()

        def log_message(self, _format: str, *_args: object) -> None:
            return

    sink = ThreadingHTTPServer(("127.0.0.1", 0), _Sink)

    class _Redirect(BaseHTTPRequestHandler):
        def do_GET(self) -> None:
            self.send_response(302)
            self.send_header(
                "Location", f"http://127.0.0.1:{sink.server_port}/capture"
            )
            self.end_headers()

        def log_message(self, _format: str, *_args: object) -> None:
            return

    redirect = ThreadingHTTPServer(("127.0.0.1", 0), _Redirect)
    threads = [
        threading.Thread(target=sink.serve_forever),
        threading.Thread(target=redirect.serve_forever),
    ]
    for thread in threads:
        thread.start()
    try:
        endpoints = base_dir / "authority_endpoints.json"
        endpoints.write_text(
            json.dumps(
                {
                    "authorities": {
                        "workstation": {
                            "origin": f"http://127.0.0.1:{redirect.server_port}"
                        }
                    }
                }
            ),
            encoding="utf-8",
        )

        with pytest.raises(AuthorityConfigurationError, match="unavailable"):
            fetch_gateway_status(base_dir, binding)

        assert captured == []
    finally:
        redirect.shutdown()
        sink.shutdown()
        for thread in threads:
            thread.join()
        redirect.server_close()
        sink.server_close()


def test_binding_rejects_permissive_or_symlinked_grant(authority, tmp_path: Path) -> None:
    base_dir, account, binding, _origin = authority
    binding.proxy_grant_ref.chmod(0o644)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        build_authority_launch(base_dir, account, "codex", [])

    target = tmp_path / "target"
    target.write_text("synthetic", encoding="utf-8")
    target.chmod(0o600)
    binding.proxy_grant_ref.unlink()
    binding.proxy_grant_ref.symlink_to(target)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        build_authority_launch(base_dir, account, "codex", [])


def test_binding_rejects_hard_linked_grant(authority, tmp_path: Path) -> None:
    base_dir, account, binding, _origin = authority
    original = tmp_path / "original.grant"
    binding.proxy_grant_ref.unlink()
    original.write_text("synthetic", encoding="utf-8")
    original.chmod(0o600)
    binding.proxy_grant_ref.hardlink_to(original)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        build_authority_launch(base_dir, account, "codex", [])


def test_binding_rejects_intermediate_symlink_in_grant_path(
    authority, tmp_path: Path
) -> None:
    base_dir, account, binding, _origin = authority
    safe = tmp_path / "safe"
    safe.mkdir(mode=0o700)
    outside = tmp_path / "outside"
    outside.mkdir(mode=0o700)
    grant = outside / "grant"
    grant.write_text("synthetic", encoding="utf-8")
    grant.chmod(0o600)
    (safe / "link").symlink_to(outside, target_is_directory=True)
    rebound = Account(
        ref=account.ref,
        alias=account.alias,
        account_home=account.account_home,
        email=account.email,
        plan=account.plan,
        account_id=account.account_id,
        authority_binding=AuthorityBinding(
            mode=binding.mode,
            authority_name=binding.authority_name,
            route_id=binding.route_id,
            provider=binding.provider,
            proxy_grant_ref=(safe / "link" / "grant").absolute(),
        ),
    )

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        build_authority_launch(base_dir, rebound, "codex", [])


def test_endpoint_configuration_refuses_writable_or_symlinked_file(
    authority, tmp_path: Path
) -> None:
    base_dir, _account, binding, _origin = authority
    endpoints = base_dir / "authority_endpoints.json"
    endpoints.chmod(0o622)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        resolve_authority_endpoint(base_dir, binding)

    target = tmp_path / "endpoints.json"
    target.write_text(endpoints.read_text(encoding="utf-8"), encoding="utf-8")
    target.chmod(0o600)
    endpoints.unlink()
    endpoints.symlink_to(target)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        resolve_authority_endpoint(base_dir, binding)


def test_endpoint_configuration_refuses_cleartext_non_loopback(
    authority,
) -> None:
    base_dir, _account, binding, _origin = authority
    endpoints = base_dir / "authority_endpoints.json"
    endpoints.write_text(
        json.dumps(
            {"authorities": {"workstation": {"origin": "http://192.0.2.1:31415"}}}
        ),
        encoding="utf-8",
    )
    endpoints.chmod(0o600)

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        resolve_authority_endpoint(base_dir, binding)


def test_gateway_health_retains_last_verified_snapshot_as_stale(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, _binding, _origin = authority

    class _Registry:
        def __init__(self, path: Path) -> None:
            self.base_dir = path

    snapshots = iter(
        (
            AccountSnapshot(
                status=HealthStatus.OK,
                primary_used_pct=23,
                secondary_used_pct=7,
                checked_at=10.0,
                detail="Gateway: ready",
            ),
            AccountSnapshot(
                status=HealthStatus.UNKNOWN,
                primary_used_pct=None,
                secondary_used_pct=None,
                checked_at=20.0,
                detail="Gateway: unavailable",
            ),
        )
    )
    monkeypatch.setattr(
        provider_services,
        "gateway_health_snapshot",
        lambda *_args, **_kwargs: next(snapshots),
    )
    adapter = AuthorityHealthAdapter(_Registry(base_dir))  # type: ignore[arg-type]

    assert adapter.fetch(account).status == HealthStatus.OK
    stale = adapter.fetch(account)

    assert stale.status == HealthStatus.UNKNOWN
    assert stale.primary_used_pct == 23
    assert stale.secondary_used_pct == 7
    assert stale.checked_at == 10.0
    assert stale.detail == "Gateway: unavailable; last verified status is stale"


def test_gateway_health_cache_does_not_cross_route_rebinding(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, binding, _origin = authority

    class _Registry:
        def __init__(self, path: Path) -> None:
            self.base_dir = path

    snapshots = iter(
        (
            AccountSnapshot(HealthStatus.OK, 23, 7, checked_at=10.0, detail="Gateway: ready"),
            AccountSnapshot(
                HealthStatus.UNKNOWN,
                None,
                None,
                checked_at=20.0,
                detail="Gateway: unavailable",
            ),
        )
    )
    monkeypatch.setattr(
        provider_services,
        "gateway_health_snapshot",
        lambda *_args, **_kwargs: next(snapshots),
    )
    adapter = AuthorityHealthAdapter(_Registry(base_dir))  # type: ignore[arg-type]
    assert adapter.fetch(account).status == HealthStatus.OK
    rebound = Account(
        ref=account.ref,
        alias=account.alias,
        account_home=account.account_home,
        email=account.email,
        plan=account.plan,
        account_id=account.account_id,
        authority_binding=AuthorityBinding(
            mode=binding.mode,
            authority_name=binding.authority_name,
            route_id="route-other",
            provider=binding.provider,
            proxy_grant_ref=binding.proxy_grant_ref,
        ),
    )

    unavailable = adapter.fetch(rebound)

    assert unavailable.status == HealthStatus.UNKNOWN
    assert unavailable.primary_used_pct is None
    assert unavailable.checked_at == 20.0
    assert unavailable.detail == "Gateway: unavailable"


def test_dark_binding_preserves_native_health_and_adds_gateway_status(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, account, _binding, _origin = authority

    class _Registry:
        def __init__(self, path: Path) -> None:
            self.base_dir = path

    class _NativeHealth:
        def fetch(self, _account: Account, timeout_secs: float = 10.0) -> AccountSnapshot:
            del timeout_secs
            return AccountSnapshot(
                HealthStatus.OK,
                12,
                34,
                checked_at=5.0,
                detail="Native: ready",
            )

    monkeypatch.setattr(
        provider_services,
        "gateway_health_snapshot",
        lambda *_args, **_kwargs: AccountSnapshot(
            HealthStatus.UNKNOWN,
            None,
            None,
            checked_at=20.0,
            detail="Gateway: unavailable",
        ),
    )
    native_health = _NativeHealth()
    services = ProviderServices(
        tool=AccountRegistryKind.CODEX,
        registry=_Registry(base_dir),  # type: ignore[arg-type]
        lifecycle=object(),  # type: ignore[arg-type]
        health=native_health,
    )
    selected = ProviderServiceMap((services,)).for_account(account)

    snapshot = selected.health.fetch(account)

    assert snapshot.status == HealthStatus.OK
    assert snapshot.primary_used_pct == 12
    assert snapshot.secondary_used_pct == 34
    assert snapshot.checked_at == 5.0
    assert snapshot.detail == "Native: ready; Gateway: unavailable"


def test_resolve_authority_data_plane_returns_exact_responses_route_and_verified_expiry(
    authority,
) -> None:
    base_dir, _account, binding, origin = authority
    active = AuthorityBinding(
        mode=AuthorityMode.SUBROUTER,
        authority_name=binding.authority_name,
        route_id=binding.route_id,
        provider=binding.provider,
        proxy_grant_ref=binding.proxy_grant_ref,
    )

    data_plane = authority_client.resolve_authority_data_plane(base_dir, active)

    assert data_plane.responses_url == f"{origin}/r/route-fixture/v1/responses"
    assert data_plane.proxy_key == "synthetic-route-key"
    assert data_plane.grant_expires_at_ms > 4_000_000_000_000
    assert data_plane.sanitized_status.state == "ready"
    assert data_plane.sanitized_status.grant_expires_at_ms == data_plane.grant_expires_at_ms
    assert _StatusHandler.requests == [
        ("/r/route-fixture/_subrouter/status", "Bearer synthetic-route-key")
    ]


def test_resolve_authority_data_plane_refuses_dark_or_quiesced_before_reading_grant(
    authority, monkeypatch: pytest.MonkeyPatch
) -> None:
    base_dir, _account, binding, _origin = authority
    monkeypatch.setattr(
        authority_client,
        "_read_proxy_key",
        lambda *_args, **_kwargs: pytest.fail("invalid binding read grant material"),
    )

    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        authority_client.resolve_authority_data_plane(base_dir, binding)

    quiesced = AuthorityBinding(
        mode=AuthorityMode.SUBROUTER,
        authority_name=binding.authority_name,
        route_id=binding.route_id,
        provider=binding.provider,
        proxy_grant_ref=binding.proxy_grant_ref,
        quiesced=True,
    )
    with pytest.raises(AuthorityConfigurationError, match="migration required"):
        authority_client.resolve_authority_data_plane(base_dir, quiesced)