"""``openai4s stop`` against a daemon that outlives the shutdown grace period.

``cmd_stop`` sent SIGTERM, polled for five seconds, then discarded the poll's
outcome: it cleared the pidfile, printed "daemon stopped" and returned 0
unconditionally. A daemon held past the grace period by an in-flight cell (a
measured single-cell shutdown ran 4.8s) was thereby orphaned — ``status`` and a
second ``stop`` read the missing pidfile as "not running", the port stayed
bound, and the next ``serve`` crashed into it with a bare "Address already in
use" *after* printing its success banner, because the banner used to precede
the bind.

These tests drive the real ``cmd_stop`` against a real child process that
ignores SIGTERM. The state files must outlive the process they describe: on
timeout the pidfile survives, the exit code is non-zero, and ``--force``
escalates to SIGKILL. The poll is the real ``_wait_pid_exit``, bounded through
the real ``--timeout`` flag, not a stub.

The opposite failure is covered too: a daemon that takes a few seconds longer
than the SIGTERM grace to finish its teardown is a clean stop, so ``stop``
keeps polling (default ``STOP_TIMEOUT_S``) with a progress line instead of
returning 2 for a shutdown that completes by itself.
"""

from __future__ import annotations

import importlib
import os
import signal
import socket
import subprocess
import sys
import threading
import time

import pytest

# ``openai4s.cli.__init__`` re-exports a FUNCTION named ``main`` that shadows
# the module of the same name; resolve the module the way the sibling CLI
# tests do.
cli_main = importlib.import_module("openai4s.cli.main")


def _spawn_sigterm_ignoring_child() -> subprocess.Popen:
    """A stand-in for a daemon whose shutdown outlasts the stop grace period.

    Prints ``ready`` only after the SIGTERM handler is installed, so the test
    cannot signal it during the interpreter's default-disposition window.
    """
    proc = subprocess.Popen(
        [
            sys.executable,
            "-c",
            "import signal, sys, time\n"
            "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n"
            "print('ready', flush=True)\n"
            "time.sleep(120)\n",
        ],
        stdout=subprocess.PIPE,
        text=True,
    )
    assert proc.stdout is not None and proc.stdout.readline().strip() == "ready"
    return proc


@pytest.fixture
def slow_daemon():
    """A SIGTERM-ignoring child registered in the pidfile.

    The reaper thread keeps the child from lingering as a zombie once it is
    SIGKILLed; ``_pid_alive`` reads a zombie as exited on Linux and macOS, and
    the unreaped case is pinned on its own below. Tests shorten the wait through
    the real ``--timeout`` flag rather than by replacing the poll.
    """
    cfg = cli_main.get_config()
    proc = _spawn_sigterm_ignoring_child()
    threading.Thread(target=proc.wait, daemon=True).start()
    cfg.pidfile.write_text(str(proc.pid), "utf-8")
    try:
        yield cfg, proc
    finally:
        if proc.poll() is None:
            proc.kill()
            proc.wait(timeout=10)


def test_stop_timeout_keeps_pidfile_and_fails(slow_daemon, capsys):
    cfg, proc = slow_daemon

    rc = cli_main.main(["stop", "--timeout", "0.1"])

    out, err = capsys.readouterr()
    assert rc == 2
    assert proc.poll() is None, "the child must still be running to prove the timeout"
    assert cfg.pidfile.read_text("utf-8").strip() == str(proc.pid)
    assert "daemon stopped" not in out
    assert "still shutting down" in err
    assert "--force" in err


def test_stop_timeout_leaves_status_and_retry_able_to_see_the_daemon(
    slow_daemon, capsys
):
    """The point of keeping the pidfile: the CLI converges instead of lying.

    After the failed stop, the daemon is still visible; once the process is
    really gone, the next ``stop`` observes that and clears the state files.
    """
    cfg, proc = slow_daemon

    assert cli_main.main(["stop", "--timeout", "0.1"]) == 2
    proc.kill()
    proc.wait(timeout=10)
    for _ in range(100):  # the reaper thread races us to wait(); give it time
        if not cli_main._pid_alive(proc.pid):
            break
        time.sleep(0.05)

    rc = cli_main.main(["stop"])

    out, _ = capsys.readouterr()
    assert rc == 1
    assert "daemon: not running" in out
    assert not cfg.pidfile.exists()


def test_stop_force_escalates_to_sigkill(slow_daemon, capsys):
    cfg, proc = slow_daemon
    # The SIGTERM half times out after --timeout (the child ignores it). After
    # SIGKILL the post-kill poll is the full TERM_GRACE_S ladder; a zombie
    # counts as exited, so this does not depend on when the reaper thread
    # gets scheduled.
    rc = cli_main.main(["stop", "--force", "--timeout", "0.1"])

    out, err = capsys.readouterr()
    assert rc == 0
    assert "daemon stopped" in out
    assert "still shutting down" not in err
    assert not cfg.pidfile.exists()
    assert proc.poll() is not None, "SIGKILL must actually have taken the child down"


def test_stop_of_a_promptly_exiting_daemon_still_reports_success(monkeypatch, capsys):
    """The happy path the fix must not regress: SIGTERM honoured, exit 0."""
    cfg = cli_main.get_config()
    proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"])
    threading.Thread(target=proc.wait, daemon=True).start()
    cfg.pidfile.write_text(str(proc.pid), "utf-8")
    try:
        rc = cli_main.main(["stop"])
    finally:
        if proc.poll() is None:
            proc.kill()
            proc.wait(timeout=10)

    assert rc == 0
    assert "daemon stopped" in capsys.readouterr().out
    assert not cfg.pidfile.exists()


_ZOMBIE_AWARE = pytest.mark.skipif(
    not (sys.platform.startswith("linux") or sys.platform == "darwin"),
    reason="zombie state is read from /proc (Linux) or kern.proc (macOS)",
)


def _wait_for_zombie(proc: subprocess.Popen, *, seconds: float = 10.0) -> None:
    """Block until ``proc`` has exited but is deliberately still unreaped.

    The pid must stay allocated — ``os.kill(pid, 0)`` still succeeds — or the
    test would prove nothing about zombies, only that the exit happened.
    """
    state = ""
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        state = subprocess.run(
            ["ps", "-o", "stat=", "-p", str(proc.pid)],
            capture_output=True,
            text=True,
        ).stdout.strip()
        if state.startswith("Z"):
            break
        time.sleep(0.02)
    else:  # pragma: no cover - diagnostic
        pytest.fail(f"pid {proc.pid} never became a zombie (last state {state!r})")
    os.kill(proc.pid, 0)  # still allocated: nothing has reaped it


@_ZOMBIE_AWARE
def test_pid_alive_reads_an_unreaped_zombie_as_exited():
    """A zombie has exited; only its pid is still allocated.

    ``os.kill(pid, 0)`` succeeds on it, so a stop polling that answer waited
    its whole timeout for a daemon that was already gone whenever the daemon's
    parent was not reaping it — the release smoke spawned ``serve`` as its own
    child and then blocked in ``subprocess.run(stop)``, which returned 2 after
    30s.
    """
    proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"])
    try:
        assert cli_main._pid_alive(proc.pid), "a running child is alive"
        proc.kill()
        _wait_for_zombie(proc)
        assert not cli_main._pid_alive(proc.pid)
    finally:
        if proc.poll() is None:
            proc.kill()
        proc.wait(timeout=10)


@_ZOMBIE_AWARE
def test_stop_of_a_daemon_nothing_reaps_returns_promptly(capsys):
    """A supervisor that spawns ``serve`` as a non-reaping child and then runs
    ``openai4s stop`` gets a clean stop, not a full timeout and exit 2."""
    cfg = cli_main.get_config()
    proc = subprocess.Popen(
        [
            sys.executable,
            "-c",
            "import time\nprint('ready', flush=True)\ntime.sleep(120)\n",
        ],
        stdout=subprocess.PIPE,
        text=True,
    )
    assert proc.stdout is not None and proc.stdout.readline().strip() == "ready"
    cfg.pidfile.write_text(str(proc.pid), "utf-8")
    try:
        started = time.monotonic()
        rc = cli_main.main(["stop", "--timeout", "10"])
        elapsed = time.monotonic() - started
        out, err = capsys.readouterr()
        # stop observed the exit while the child was still unreaped.
        os.kill(proc.pid, 0)
    finally:
        if proc.poll() is None:
            proc.kill()
        proc.wait(timeout=10)

    assert rc == 0, err
    assert "daemon stopped" in out
    assert "still shutting down" not in err
    assert not cfg.pidfile.exists()
    assert elapsed < 5, f"stop polled an exited daemon for {elapsed:.1f}s"


@pytest.mark.parametrize(
    ("state", "alive"),
    [(b"S", True), (b"R", True), (b"D", True), (b"Z", False), (b"X", False)],
)
def test_the_linux_branch_reads_state_after_the_last_parenthesis(
    monkeypatch, state, alive
):
    """Forces the procfs branch, which macOS never takes.

    The comm field is attacker-shaped: a process may name itself so that a
    whitespace split lands on a fake state. ``os.getpid()`` keeps the
    ``os.kill(pid, 0)`` half genuinely true, so only the state decides.
    """
    pid = os.getpid()
    stat = b"%d (evil) Z 1 2) " % pid + state + b" 1 %d %d 0 -1" % (pid, pid)
    real_open = open

    def fake_open(path, *args, **kwargs):
        if path == f"/proc/{pid}/stat":
            import io

            return io.BytesIO(stat)
        return real_open(path, *args, **kwargs)

    monkeypatch.setattr(cli_main, "open", fake_open, raising=False)
    assert cli_main._pid_alive(pid) is alive


def test_stop_keeps_waiting_past_the_sigterm_grace_for_a_daemon_still_exiting(
    monkeypatch, capsys
):
    """A daemon that exits a little after the SIGTERM grace is a clean stop.

    With a first-kernel bootstrap or several live kernels in teardown, a real
    daemon measured 8-9s from SIGTERM to exit. `stop` polled only the 5s
    TERM_GRACE_S and returned 2, so a script treating non-zero as failure got
    a false alarm and was pushed toward `--force` for a shutdown that
    finished by itself seconds later. The grace is shrunk here so the child
    can outlive it quickly; the default overall wait stays STOP_TIMEOUT_S.
    """
    monkeypatch.setattr(cli_main, "TERM_GRACE_S", 0.3)
    cfg = cli_main.get_config()
    proc = subprocess.Popen(
        [
            sys.executable,
            "-c",
            "import os, signal, threading, time\n"
            "def on_term(*_):\n"
            "    threading.Timer(1.2, lambda: os._exit(0)).start()\n"
            "signal.signal(signal.SIGTERM, on_term)\n"
            "print('ready', flush=True)\n"
            "time.sleep(120)\n",
        ],
        stdout=subprocess.PIPE,
        text=True,
    )
    assert proc.stdout is not None and proc.stdout.readline().strip() == "ready"
    threading.Thread(target=proc.wait, daemon=True).start()
    cfg.pidfile.write_text(str(proc.pid), "utf-8")
    try:
        started = time.monotonic()
        rc = cli_main.main(["stop"])
        elapsed = time.monotonic() - started
    finally:
        if proc.poll() is None:
            proc.kill()
            proc.wait(timeout=10)

    out, err = capsys.readouterr()
    assert rc == 0, err
    assert "daemon stopped" in out
    assert f"daemon (pid {proc.pid}) is shutting down" in err
    assert "still shutting down" not in err
    assert not cfg.pidfile.exists()
    assert elapsed < 10


def test_stop_waits_thirty_seconds_by_default_and_documents_timeout(capsys):
    parser = cli_main.build_parser()
    assert parser.parse_args(["stop"]).timeout == cli_main.STOP_TIMEOUT_S == 30.0
    assert parser.parse_args(["stop", "--timeout", "2.5"]).timeout == 2.5
    assert cli_main.STOP_TIMEOUT_S > cli_main.TERM_GRACE_S

    with pytest.raises(SystemExit) as stopped:
        cli_main.main(["stop", "--help"])
    assert stopped.value.code == 0
    assert "--timeout" in capsys.readouterr().out


@pytest.mark.parametrize("value", ["0", "-1", "nan", "inf", "soon"])
def test_stop_rejects_a_timeout_that_is_not_a_positive_finite_number(value, capsys):
    with pytest.raises(SystemExit) as stopped:
        cli_main.main(["stop", "--timeout", value])
    assert stopped.value.code == 2
    assert "argument --timeout: must be a positive number of seconds" in (
        capsys.readouterr().err
    )


def test_serve_bind_failure_prints_no_success_banner(monkeypatch, capsys):
    """``serve`` used to print "listening"/"web UI ready" before the bind, so a
    port collision surfaced as a traceback after a success banner. The banner
    now requires the bind; the collision is one clear error and exit 1."""
    placeholder = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    placeholder.bind(("127.0.0.1", 0))
    placeholder.listen(1)
    port = placeholder.getsockname()[1]
    monkeypatch.setenv("OPENAI4S_NO_OPEN", "1")
    # Config.port resolves OPENAI4S_PORT at class-definition time, so setenv
    # after import cannot steer it; patch the live singleton instead.
    cfg = cli_main.get_config()
    monkeypatch.setattr(cfg, "port", port)

    try:
        rc = cli_main.main(["serve", "--no-open"])
    finally:
        placeholder.close()

    out, err = capsys.readouterr()
    assert rc == 1
    assert "listening" not in out
    assert "web UI ready" not in out
    assert "address already in use" in err
    assert f"{port}" in err
    assert not cfg.pidfile.exists(), "a failed bind must not leave daemon state"


def test_stop_survives_the_pid_exiting_between_check_and_signal(monkeypatch, capsys):
    """The aliveness check and the SIGTERM are not atomic; a pid that vanishes
    in between is a completed stop, not a traceback.

    ``_pid_alive`` is forced True so the real ``os.kill`` — aimed at a pid no
    platform allocates — raises the genuine ``ProcessLookupError``.
    """
    cfg = cli_main.get_config()
    cfg.pidfile.write_text("99999999", "utf-8")
    monkeypatch.setattr(cli_main, "_pid_alive", lambda pid: True)
    monkeypatch.setattr(cli_main, "_wait_pid_exit", lambda pid, **kw: True)

    rc = cli_main.main(["stop"])

    assert rc == 0
    assert "daemon stopped" in capsys.readouterr().out
    assert not cfg.pidfile.exists()


def test_sigterm_handler_preserves_state_files_until_teardown_completes():
    """The daemon must not delete its own pidfile at SIGTERM-arrival.

    The old inline handler ran ``_clear_state`` the instant the signal
    landed — before the (possibly >5s) runner/kernel teardown — so a stop
    that timed out told the user to retry against a pidfile the daemon had
    already removed: the retry and ``stop --force`` both saw "not running"
    while the port stayed bound, and SIGKILL was never sent.  State is
    cleared in the serve loop's finally, after teardown.
    """
    cfg = cli_main.get_config()
    cfg.pidfile.write_text(str(os.getpid()), "utf-8")

    with pytest.raises(KeyboardInterrupt):
        cli_main._sigterm_to_keyboard_interrupt(signal.SIGTERM, None)

    assert cfg.pidfile.read_text("utf-8").strip() == str(os.getpid())


def test_serve_pidfile_covers_the_build_window(monkeypatch, capsys):
    """The singleton pidfile must exist during the whole boot, not post-bind.

    Written after ``build_server``, a second ``serve`` racing a slow store
    open/migration passed the already-running check and booted the same data
    dir concurrently, while ``stop``/``status`` called the mid-boot daemon
    "not running".  A failed build still clears the state files.
    """
    import errno as errno_mod

    import openai4s.server as server_pkg

    cfg = cli_main.get_config()
    seen: dict[str, bool] = {}

    def probe(_cfg):
        seen["pidfile_during_build"] = cfg.pidfile.exists()
        raise OSError(errno_mod.EADDRINUSE, "address already in use")

    monkeypatch.setattr(server_pkg, "build_server", probe)
    monkeypatch.setenv("OPENAI4S_NO_OPEN", "1")

    rc = cli_main.main(["serve", "--no-open"])

    out, err = capsys.readouterr()
    assert rc == 1
    assert seen["pidfile_during_build"] is True
    assert "address already in use" in err
    assert "listening" not in out
    assert not cfg.pidfile.exists(), "a failed build must not leave daemon state"


def _failing_build(monkeypatch, exc: OSError):
    import openai4s.server as server_pkg

    monkeypatch.setattr(
        server_pkg, "build_server", lambda cfg: (_ for _ in ()).throw(exc)
    )
    monkeypatch.setenv("OPENAI4S_NO_OPEN", "1")


def test_serve_privileged_port_prints_one_clear_error(monkeypatch, capsys):
    """EACCES on a privileged port gets the same one-line treatment as a
    port collision, naming OPENAI4S_PORT — not a bare traceback."""
    import errno as errno_mod

    cfg = cli_main.get_config()
    monkeypatch.setattr(cfg, "port", 80)
    _failing_build(monkeypatch, OSError(errno_mod.EACCES, "permission denied"))

    rc = cli_main.main(["serve", "--no-open"])

    out, err = capsys.readouterr()
    assert rc == 1
    assert "listening" not in out
    assert "permission denied" in err and "OPENAI4S_PORT" in err
    assert not cfg.pidfile.exists()


def test_serve_unwritable_data_dir_eacces_still_raises(monkeypatch):
    """EACCES with an unprivileged port is not a port problem: build_server
    does more than bind, and a read-only data dir must stay a traceback
    pointing at the real path instead of blaming OPENAI4S_PORT."""
    import errno as errno_mod

    cfg = cli_main.get_config()
    assert cfg.port >= 1024
    _failing_build(monkeypatch, OSError(errno_mod.EACCES, "permission denied"))

    with pytest.raises(OSError):
        cli_main.main(["serve", "--no-open"])
    assert not cfg.pidfile.exists(), "a failed build must not leave daemon state"


def test_serve_unavailable_address_prints_one_clear_error(monkeypatch, capsys):
    """EADDRNOTAVAIL (a mis-set OPENAI4S_HOST) names the env var to fix."""
    import errno as errno_mod

    cfg = cli_main.get_config()
    _failing_build(
        monkeypatch, OSError(errno_mod.EADDRNOTAVAIL, "cannot assign address")
    )

    rc = cli_main.main(["serve", "--no-open"])

    out, err = capsys.readouterr()
    assert rc == 1
    assert "listening" not in out
    assert "OPENAI4S_HOST" in err
    assert not cfg.pidfile.exists()


def test_serve_drives_the_shared_run_server_lifecycle(monkeypatch, capsys):
    """`openai4s serve` and `serve_app` must share one service loop.

    cmd_serve used to inline a copy of serve_app's serve/teardown, so a
    lifecycle fix could land on one path and silently miss the daemon every
    real user runs.  This pins the CLI to gateway.run_server."""
    import openai4s.server as server_pkg

    cfg = cli_main.get_config()
    sentinel = object()
    driven: list[object] = []

    monkeypatch.setattr(server_pkg, "build_server", lambda _cfg: sentinel)
    monkeypatch.setattr(server_pkg, "run_server", driven.append)
    monkeypatch.setenv("OPENAI4S_NO_OPEN", "1")

    rc = cli_main.main(["serve", "--no-open"])

    assert rc == 0
    assert driven == [sentinel]
    assert "listening" in capsys.readouterr().out
    assert not cfg.pidfile.exists(), "state clears after the loop returns"


def test_stop_grace_period_is_the_shared_process_group_budget():
    """The 5s SIGTERM grace is TERM_GRACE_S, not a re-derived constant: tuning
    the job-stop budget must tune the daemon stop with it. `stop` keeps
    polling past it, up to --timeout, before it reports failure."""
    from openai4s.execution import process_group

    assert cli_main.TERM_GRACE_S is process_group.TERM_GRACE_S
    # The default attempts derive from the shared budget (attempts × interval
    # == TERM_GRACE_S).  A dead pid exercises the default path without
    # sleeping: the first poll observes the exit.
    assert cli_main._wait_pid_exit(99999999) is True


def _state_cfg(tmp_path):
    from types import SimpleNamespace

    return SimpleNamespace(
        pidfile=tmp_path / "daemon.pid",
        statefile=tmp_path / "daemon.json",
        host="127.0.0.1",
        port=8760,
    )


def test_serve_clear_spares_a_pidfile_owned_by_another_booter(tmp_path):
    """A startup that lost the race must not delete the winner's state.

    Two concurrent `serve` runs could both pass the old read-then-write check;
    if one then failed its build, its unconditional `_clear_state` deleted the
    *winner's* pidfile, stranding a live daemon that `status`/`stop` then read
    as "not running". The serve-side clear now only removes state it owns.
    """
    cfg = _state_cfg(tmp_path)
    cfg.pidfile.write_text("424242", "utf-8")  # another process now owns it
    cfg.statefile.write_text('{"pid": 424242}', "utf-8")

    cli_main._clear_state(cfg, only_if_owned_by=os.getpid())
    assert cfg.pidfile.read_text("utf-8").strip() == "424242", "must not delete it"

    # `stop` clears unconditionally — it removes another process's state on
    # purpose, after confirming that process is gone.
    cli_main._clear_state(cfg)
    assert not cfg.pidfile.exists() and not cfg.statefile.exists()


def test_acquire_singleton_is_atomic_single_winner_and_reclaims_stale(tmp_path):
    """`_acquire_singleton` is the atomic claim that closes the boot race.

    O_EXCL makes "create the pidfile or fail" one step, so a second live
    claimant is refused rather than both booting the same data dir; a stale
    pidfile (its recorded pid is gone) is reclaimed exactly once.
    """
    cfg = _state_cfg(tmp_path)

    assert cli_main._acquire_singleton(cfg) is True
    assert cfg.pidfile.read_text("utf-8").strip() == str(os.getpid())
    # Our own pid is alive, so a second claim against the same pidfile fails.
    assert cli_main._acquire_singleton(cfg) is False

    # A stale pidfile (dead pid) is reclaimed and rewritten with our pid.
    cfg.pidfile.write_text("99999999", "utf-8")
    assert cli_main._acquire_singleton(cfg) is True
    assert cfg.pidfile.read_text("utf-8").strip() == str(os.getpid())


def test_serve_arms_the_sigterm_handler_before_the_build(monkeypatch, capsys):
    """The SIGTERM handler must be armed before the (slow) build, and a failed
    startup must put back the disposition it found.

    Installed only after the build returned, a signal arriving during a slow
    store open/migration took the interpreter's default disposition —
    terminate — orphaning the pidfile the boot had already claimed. Asserted on
    the ordering of the real ``signal.signal`` calls rather than by signalling
    this process: an asynchronously delivered SIGTERM raises KeyboardInterrupt
    at whatever bytecode boundary it lands on, which escapes the test and
    aborts the whole pytest session.
    """
    import errno as errno_mod

    calls: list[tuple[str, object]] = []
    real_signal = signal.signal

    def recording_signal(signum, handler):
        if signum == signal.SIGTERM:
            calls.append(("sigterm", handler))
        return real_signal(signum, handler)

    def build(_cfg):
        calls.append(("build", None))
        raise OSError(errno_mod.EADDRINUSE, "address already in use")

    import openai4s.server as server_pkg

    monkeypatch.setattr(server_pkg, "build_server", build)
    monkeypatch.setenv("OPENAI4S_NO_OPEN", "1")

    # Whatever this process's disposition happens to be — an earlier serve test
    # in this module may already have armed the same handler, so identity alone
    # cannot tell "armed" from "restored".
    before = signal.getsignal(signal.SIGTERM)
    monkeypatch.setattr(cli_main.signal, "signal", recording_signal)
    try:
        assert cli_main.main(["serve", "--no-open"]) == 1
    finally:
        real_signal(signal.SIGTERM, before)
    capsys.readouterr()

    assert [step for step, _ in calls] == ["sigterm", "build", "sigterm"]
    assert calls[0][1] is cli_main._sigterm_to_keyboard_interrupt, "armed before build"
    assert calls[2][1] is before, "a failed startup restores what it found"