import io
import json
import signal
import threading
import time
from pathlib import Path

import pytest

import codex_state
from health_client import (
    AccountHealthClient,
    AccountSnapshot,
    ExtraUsageSummary,
    HealthStatus,
    NamedLimit,
    SpendSummary,
)

_RESOLVE_CODEX_EXECUTABLE = AccountHealthClient._codex_executable


class _FakeProcess:
    def __init__(self, stdout_lines=None):
        self.stdout = io.StringIO("".join(stdout_lines or []))
        self.stdin = io.StringIO()
        self.pid = 4_999_999
        self.terminated = False
        self.signals = []
        self.wait_calls = []

    def send_signal(self, sig):
        self.signals.append(sig)
        if sig == signal.SIGTERM:
            self.terminated = True

    def terminate(self):
        self.send_signal(signal.SIGTERM)

    def kill(self):
        self.send_signal(signal.SIGKILL)

    def wait(self, timeout=None):
        self.wait_calls.append(timeout)
        return 0


@pytest.fixture(autouse=True)
def _fixed_codex_executable(monkeypatch):
    monkeypatch.setattr(
        AccountHealthClient,
        "_codex_executable",
        staticmethod(lambda: "/usr/bin/codex"),
    )


@pytest.fixture(autouse=True)
def _no_real_killpg(monkeypatch):
    def _refuse(pgid, sig):
        raise ProcessLookupError(pgid)

    monkeypatch.setattr("health_client.os.killpg", _refuse)


@pytest.fixture(autouse=True)
def _no_user_scope(monkeypatch):
    monkeypatch.setattr(AccountHealthClient, "_scope_available", False)


def test_fetch_returns_ok_snapshot_from_rate_limit_result(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "primary": {"usedPercent": 42, "resetSeconds": 300},
                        "secondary": {"usedPercent": 7, "resetsAt": 1_800_000_000},
                    },
                }
            )
            + "\n",
        ]
    )

    def fake_popen(argv, **kwargs):
        assert argv == [AccountHealthClient._codex_executable(), "app-server", "--stdio"]
        assert kwargs["env"]["CODEX_HOME"] == str(tmp_path)
        return process

    monkeypatch.setattr("health_client.subprocess.Popen", fake_popen)
    monkeypatch.setattr("health_client.time.time", lambda: 1_700_000_000.0)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.OK,
        primary_used_pct=42,
        secondary_used_pct=7,
        primary_reset_at=1_700_000_300.0,
        secondary_reset_at=1_800_000_000.0,
    )
    assert process.stdin.getvalue().splitlines() == [
        '{"id":1,"method":"initialize","params":{"clientInfo":{"name":"codex-tray","version":"1.0"}}}',
        '{"id":2,"method":"account/rateLimits/read","params":{}}',
    ]
    assert process.terminated is True


def test_fetch_parses_iso_timestamp_resets(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "primary": {
                            "usedPercent": 42,
                            "resetsAt": "2026-07-01T12:00:00Z",
                        },
                        "secondary": {
                            "usedPercent": 7,
                            "resetAt": "2026-07-07T00:00:00Z",
                        },
                    },
                }
            )
            + "\n",
        ]
    )

    def fake_popen(argv, **kwargs):
        assert argv == [AccountHealthClient._codex_executable(), "app-server", "--stdio"]
        assert kwargs["env"]["CODEX_HOME"] == str(tmp_path)
        return process

    monkeypatch.setattr("health_client.subprocess.Popen", fake_popen)
    monkeypatch.setattr("health_client.time.time", lambda: 1_700_000_000.0)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.OK,
        primary_used_pct=42,
        secondary_used_pct=7,
        primary_reset_at=1782907200.0,
        secondary_reset_at=1783382400.0,
    )


def test_fetch_returns_ok_snapshot_from_nested_rate_limits_result(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "rateLimits": {
                            "primary": {"usedPercent": 65},
                            "secondary": {"usedPercent": 59},
                        }
                    },
                }
            )
            + "\n",
        ]
    )
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.OK,
        primary_used_pct=65,
        secondary_used_pct=59,
    )


def test_fetch_returns_unknown_when_rate_limits_are_not_loaded(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps({"id": 2, "result": {"rateLimits": {}}}) + "\n",
        ]
    )
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.UNKNOWN,
        primary_used_pct=None,
        secondary_used_pct=None,
    )


def test_fetch_maps_duration_aware_weekly_only_window_to_secondary(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "rateLimits": {
                            "primary": {
                                "usedPercent": 86,
                                "windowDurationMins": 10_080,
                                "resetsAt": 1_800_000_000,
                            },
                            "secondary": None,
                        }
                    },
                }
            )
            + "\n",
        ]
    )
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.OK,
        primary_used_pct=None,
        secondary_used_pct=86,
        primary_reset_at=None,
        secondary_reset_at=1_800_000_000.0,
    )


def test_fetch_reads_nested_rate_limits_payload_with_epoch_resets(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "rateLimits": {
                            "primary": {"usedPercent": 90, "resetsAt": 1782919351},
                            "secondary": {"usedPercent": 48, "resetsAt": 1783389459},
                        }
                    },
                }
            )
            + "\n",
        ]
    )
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.OK,
        primary_used_pct=90,
        secondary_used_pct=48,
        primary_reset_at=1782919351.0,
        secondary_reset_at=1783389459.0,
    )


def test_fetch_returns_broken_on_json_rpc_error(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 2, "error": {"code": -32603, "message": "token_revoked"}})
            + "\n"
        ]
    )
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.BROKEN,
        primary_used_pct=None,
        secondary_used_pct=None,
    )


def test_fetch_returns_unknown_on_timeout(monkeypatch, tmp_path):
    process = _FakeProcess(stdout_lines=[])
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)
    timeouts = iter([0.0, 11.0])
    monkeypatch.setattr("health_client.time.monotonic", lambda: next(timeouts))

    snapshot = AccountHealthClient().fetch(tmp_path, timeout_secs=10.0)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.UNKNOWN,
        primary_used_pct=None,
        secondary_used_pct=None,
    )
    assert process.terminated is True


class _NeverEmitsStdout:
    """Simulates a subprocess.stdout that never yields a newline-terminated
    line (the subprocess is alive but produces no output)."""

    def __init__(self) -> None:
        self._unblock = threading.Event()

    def readline(self):
        self._unblock.wait()
        return ""

    def release(self) -> None:
        self._unblock.set()


class _NeverTerminatingProcess:
    def __init__(self) -> None:
        self.stdout = _NeverEmitsStdout()
        self.stdin = io.StringIO()
        self.pid = 4_999_999
        self.terminated = False

    def send_signal(self, sig):
        if sig == signal.SIGTERM:
            self.terminate()

    def kill(self):
        self.send_signal(signal.SIGKILL)

    def terminate(self):
        self.terminated = True
        self.stdout.release()

    def wait(self, timeout=None):
        return 0

    def poll(self):
        return None


def test_fetch_times_out_when_subprocess_never_emits_a_line(monkeypatch, tmp_path):
    process = _NeverTerminatingProcess()
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    start = time.monotonic()
    snapshot = AccountHealthClient().fetch(tmp_path, timeout_secs=0.2)
    elapsed = time.monotonic() - start

    assert snapshot == AccountSnapshot(
        status=HealthStatus.UNKNOWN,
        primary_used_pct=None,
        secondary_used_pct=None,
    )
    assert elapsed < 5.0
    assert process.terminated is True


def test_fetch_returns_unknown_on_spawn_failure(monkeypatch, tmp_path):
    def fake_popen(*args, **kwargs):
        raise FileNotFoundError("codex missing")

    monkeypatch.setattr("health_client.subprocess.Popen", fake_popen)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.UNKNOWN,
        primary_used_pct=None,
        secondary_used_pct=None,
    )


def test_write_cache_writes_json_atomically(monkeypatch, tmp_path):
    cache_path = tmp_path / "health_cache.json"
    cache_path.write_text('{"existing": true}', encoding="utf-8")
    snapshots = {
        "rafa": AccountSnapshot(
            status=HealthStatus.OK,
            primary_used_pct=13,
            secondary_used_pct=27,
            primary_reset_at=1_700_000_300.0,
            secondary_reset_at=1_800_000_000.0,
        )
    }
    captured = {}
    real_replace = Path.replace

    def fake_replace(self, target):
        captured["target_before_replace"] = Path(target).read_text(encoding="utf-8")
        captured["temp_payload"] = self.read_text(encoding="utf-8")
        return real_replace(self, target)

    monkeypatch.setattr(Path, "replace", fake_replace)

    AccountHealthClient().write_cache(cache_path, snapshots)

    data = json.loads(cache_path.read_text(encoding="utf-8"))
    assert data == {
        "rafa": {
            "status": "ok",
            "primary_used_pct": 13,
            "secondary_used_pct": 27,
            "primary_reset_at": 1_700_000_300.0,
            "secondary_reset_at": 1_800_000_000.0,
            "checked_at": None,
            "detail": None,
            "named_limits": [],
            "extra_usage": None,
            "spend": None,
        }
    }
    assert captured["target_before_replace"] == '{"existing": true}'
    assert json.loads(captured["temp_payload"]) == data


def test_fetch_ignores_invalid_reset_timestamps(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "primary": {"usedPercent": 42, "resetsAt": {"bad": True}},
                        "secondary": {"usedPercent": 7, "resetAt": ["bad"]},
                    },
                }
            )
            + "\n"
        ]
    )
    monkeypatch.setattr("health_client.subprocess.Popen", lambda *args, **kwargs: process)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.OK,
        primary_used_pct=42,
        secondary_used_pct=7,
        primary_reset_at=None,
        secondary_reset_at=None,
    )


def test_read_cache_returns_valid_snapshots_and_skips_bad_entries(tmp_path):
    cache_path = tmp_path / "health_cache.json"
    cache_path.write_text(
        json.dumps(
            {
                "rafa": {
                    "status": "ok",
                    "primary_used_pct": 13,
                    "secondary_used_pct": 27,
                    "primary_reset_at": 1_700_000_300.0,
                    "secondary_reset_at": 1_800_000_000.0,
                },
                "bad-status": {
                    "status": "missing",
                    "primary_used_pct": 1,
                    "secondary_used_pct": 2,
                },
                "bad-pct": {
                    "status": "ok",
                    "primary_used_pct": "13",
                    "secondary_used_pct": None,
                },
            }
        ),
        encoding="utf-8",
    )

    snapshots = AccountHealthClient().read_cache(cache_path)

    assert snapshots == {
        "rafa": AccountSnapshot(
            HealthStatus.OK,
            13,
            27,
            primary_reset_at=1_700_000_300.0,
            secondary_reset_at=1_800_000_000.0,
        ),
        "bad-pct": AccountSnapshot(HealthStatus.OK, None, None),
    }


def test_read_cache_returns_empty_on_missing_or_invalid_file(tmp_path):
    assert AccountHealthClient().read_cache(tmp_path / "missing.json") == {}

    invalid_path = tmp_path / "invalid.json"
    invalid_path.write_text("not-json", encoding="utf-8")

    assert AccountHealthClient().read_cache(invalid_path) == {}


def test_write_and_read_cache_round_trips_extended_snapshot_fields(tmp_path):
    cache_path = tmp_path / "health_cache.json"
    snapshots = {
        "rafa": AccountSnapshot(
            status=HealthStatus.OK,
            primary_used_pct=13,
            secondary_used_pct=27,
            primary_reset_at=1_700_000_300.0,
            secondary_reset_at=1_800_000_000.0,
            checked_at=1_700_000_123.0,
            detail="limits unavailable",
            named_limits=(
                NamedLimit("five_hour", "primary", 13, 1_700_000_300.0, True),
                NamedLimit("seven_day", "secondary", 27, 1_800_000_000.0, True),
            ),
            extra_usage=ExtraUsageSummary(
                used=3.0,
                limit=10.0,
                unit="hours",
                display="3 / 10 hours",
            ),
            spend=SpendSummary(
                amount=12.5,
                limit=50.0,
                currency="USD",
                display="$12.50 / $50",
            ),
        )
    }

    client = AccountHealthClient()
    client.write_cache(cache_path, snapshots)

    assert client.read_cache(cache_path) == snapshots


def _incomplete_backfill(monkeypatch, *, warming: bool) -> None:
    monkeypatch.setattr("health_client.codex_state.needs_warmup", lambda home: True)
    monkeypatch.setattr("health_client.codex_state.warmup_in_progress", lambda home: warming)


def test_fetch_reports_preparing_while_a_backfill_warmup_runs(monkeypatch, tmp_path):
    _incomplete_backfill(monkeypatch, warming=True)
    monkeypatch.setattr(
        "health_client.subprocess.Popen",
        lambda *args, **kwargs: pytest.fail("probe must not race the running backfill"),
    )

    snapshot = AccountHealthClient().fetch(tmp_path, allow_warmup=True)

    assert snapshot == AccountSnapshot(
        status=HealthStatus.UNKNOWN,
        primary_used_pct=None,
        secondary_used_pct=None,
        detail="preparing account data",
    )


def test_fetch_gives_a_first_backfill_the_whole_warmup_budget(monkeypatch, tmp_path):
    _incomplete_backfill(monkeypatch, warming=False)
    released: list[Path] = []
    monkeypatch.setattr(
        "health_client.codex_state.release_orphaned_claim",
        lambda home: released.append(home) or True,
    )
    budgets: list[float] = []

    def fake_probe(self, codex_home, timeout_secs):
        budgets.append(timeout_secs)
        assert (codex_home / ".systray-backfill-worker").exists()
        return AccountSnapshot(HealthStatus.OK, 1, 2)

    monkeypatch.setattr(AccountHealthClient, "_probe", fake_probe)

    snapshot = AccountHealthClient().fetch(tmp_path, timeout_secs=10.0, allow_warmup=True)

    assert released == [tmp_path]
    assert budgets == [codex_state.WARMUP_TIMEOUT_S]
    assert snapshot.status is HealthStatus.OK
    assert not (tmp_path / ".systray-backfill-worker").exists()


def test_fetch_never_waits_on_a_backfill_for_a_synchronous_caller(monkeypatch, tmp_path):
    _incomplete_backfill(monkeypatch, warming=False)
    monkeypatch.setattr(
        "health_client.subprocess.Popen",
        lambda *args, **kwargs: pytest.fail("a cli caller must not run the warm-up probe"),
    )

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot.detail == "preparing account data"
    assert snapshot.status is HealthStatus.UNKNOWN


def test_codex_executable_skips_the_agent_session_shim(monkeypatch, tmp_path):
    shim_dir = tmp_path / "shim"
    real_dir = tmp_path / "real"
    shim_dir.mkdir()
    real_dir.mkdir()
    shim_body = shim_dir / "_tmpjail-shim.sh"
    shim_body.write_text("#!/bin/sh\n", encoding="utf-8")
    shim_body.chmod(0o755)
    (shim_dir / "codex").symlink_to(shim_body)
    real = real_dir / "codex"
    real.write_text("#!/bin/sh\n", encoding="utf-8")
    real.chmod(0o755)
    monkeypatch.setenv("PATH", f"{shim_dir}:{real_dir}")

    assert _RESOLVE_CODEX_EXECUTABLE() == str(real.resolve())


def test_codex_executable_returns_an_absolute_path_for_empty_path_entries(
    monkeypatch, tmp_path
):
    shim_dir = tmp_path / "shim"
    shim_dir.mkdir()
    shim_body = shim_dir / "_tmpjail-shim.sh"
    shim_body.write_text("#!/bin/sh\n", encoding="utf-8")
    shim_body.chmod(0o755)
    (shim_dir / "codex").symlink_to(shim_body)
    current = tmp_path / "codex"
    current.write_text("#!/bin/sh\n", encoding="utf-8")
    current.chmod(0o755)
    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("PATH", f"{shim_dir}:")

    assert _RESOLVE_CODEX_EXECUTABLE() == str(current.resolve())


def test_probe_wraps_codex_in_named_scope_and_kills_it(monkeypatch, tmp_path):
    process = _FakeProcess(
        stdout_lines=[
            json.dumps({"id": 1, "result": {"ok": True}}) + "\n",
            json.dumps(
                {
                    "id": 2,
                    "result": {
                        "primary": {"usedPercent": 1},
                        "secondary": {"usedPercent": 2},
                    },
                }
            )
            + "\n",
        ]
    )
    spawned = {}
    scope_kills = []

    def fake_popen(argv, **kwargs):
        spawned["argv"] = argv
        return process

    def fake_run(argv, **kwargs):
        scope_kills.append(argv)

        class _Result:
            returncode = 0

        return _Result()

    monkeypatch.setattr(AccountHealthClient, "_scope_available", True)
    monkeypatch.setattr("health_client.subprocess.Popen", fake_popen)
    monkeypatch.setattr("health_client.subprocess.run", fake_run)

    snapshot = AccountHealthClient().fetch(tmp_path)

    assert snapshot.status is HealthStatus.OK
    argv = spawned["argv"]
    assert argv[:6] == [
        "systemd-run",
        "--user",
        "--scope",
        "--collect",
        "--quiet",
        "--slice=agent.slice",
    ]
    assert argv[6].startswith("--unit=codex-probe-")
    assert "MemoryMax=2G" in argv
    assert "MemoryHigh=1G" in argv
    assert argv[-3:] == [AccountHealthClient._codex_executable(), "app-server", "--stdio"]
    assert len(scope_kills) == 1
    kill_argv = scope_kills[0]
    assert kill_argv[:5] == [
        "systemctl",
        "--user",
        "kill",
        "--kill-whom=all",
        "--signal=SIGKILL",
    ]
    assert kill_argv[5].startswith("codex-probe-")
    assert kill_argv[5].endswith(".scope")


def test_user_scope_available_requires_working_user_manager(monkeypatch):
    monkeypatch.setattr(AccountHealthClient, "_scope_available", None)
    monkeypatch.setattr("health_client.shutil.which", lambda name: None)
    assert AccountHealthClient._user_scope_available() is False

    monkeypatch.setattr(AccountHealthClient, "_scope_available", None)
    monkeypatch.setattr("health_client.shutil.which", lambda name: f"/usr/bin/{name}")

    class _Result:
        returncode = 0

    monkeypatch.setattr("health_client.subprocess.run", lambda *a, **k: _Result())
    assert AccountHealthClient._user_scope_available() is True
    assert AccountHealthClient._scope_available is True