"""LIVE Windows E2E for the gateway control socket named-pipe transport.

Runs ONLY on a real Windows host (the on-demand ``windows-venv-e2e.yml``
lane). Spawns a REAL child process that binds the REAL named pipe via the
proactor event loop with the DEFAULT verb handlers, then drives the real
sync client and the real fleet consumers against it — no mocks anywhere.

Proves, on windows-latest:
  1. `GatewayControlServer` binds ``\\\\.\\pipe\\hermes-gateway-<hash>`` via
     ``loop.start_serving_pipe`` and answers ``identify``/``status``.
  2. The sync client's pipe transport (open/write/read/busy-retry) works
     against a live server and returns the child's true pid + code identity.
  3. ``collect_fleet_versions()`` prefers the socket (``source: socket``).
  4. After the server process is force-killed, the client returns None
     (FileNotFoundError on the pipe — no stale-file hazard on Windows) and
     consumers fall back to the state-file layer, which classifies only a
     live-verified gateway PID (#110420).
"""

from __future__ import annotations

import json
import os
import queue
import subprocess
import sys
import threading
import time
from pathlib import Path

import pytest

from tests.live_process_fixtures import sleeper_script_path

pytestmark = pytest.mark.platforms("windows")  # live Windows named-pipe E2E

PROJECT_ROOT = Path(__file__).resolve().parents[2]


def _wait_until(predicate, timeout: float = 15.0, interval: float = 0.05) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if predicate():
            return True
        time.sleep(interval)
    return bool(predicate())


def _readline(stream, timeout: float = 60.0) -> str:
    """``stream.readline()`` bounded by *timeout* (a hung child fails, not hangs)."""
    got: queue.Queue = queue.Queue()
    threading.Thread(target=lambda: got.put(stream.readline()), daemon=True).start()
    try:
        return got.get(timeout=timeout)
    except queue.Empty:
        return ""


def _argv_visible(pid: int, marker: str) -> bool:
    """True once the process table shows *pid* with *marker* in its argv."""
    import psutil

    try:
        return marker in " ".join(psutil.Process(pid).cmdline())
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        return False

_CHILD_CODE = r"""
import asyncio, os, sys
sys.path.insert(0, sys.argv[1])
os.environ["HERMES_HOME"] = sys.argv[2]
from gateway.control_socket import GatewayControlServer

async def main():
    server = GatewayControlServer()
    ok = await server.start()
    # Print our REAL pid: on Windows uv venvs, python.exe is a trampoline
    # that spawns the actual interpreter as a child, so Popen.pid is the
    # shim, not the server process. (That spawner-view-vs-reality gap is
    # the exact bug class the control socket exists to eliminate.)
    print(f"SERVER_STARTED {os.getpid()}" if ok else "SERVER_FAILED", flush=True)
    if not ok:
        return
    await asyncio.sleep(120)

asyncio.run(main())
"""


@pytest.fixture()
def live_server(tmp_path: Path):
    home = tmp_path / ".hermes"
    home.mkdir()
    proc = subprocess.Popen(
        [sys.executable, "-c", _CHILD_CODE, str(PROJECT_ROOT), str(home)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        cwd=str(PROJECT_ROOT),
    )
    line = _readline(proc.stdout).strip()
    if not line.startswith("SERVER_STARTED"):
        err = proc.stderr.read() if proc.poll() is not None else ""
        _kill_tree(proc)
        pytest.fail(f"pipe server child failed to start: {line!r} {err}")
    server_pid = int(line.split()[1])
    yield proc, home, server_pid
    _kill_tree(proc)


def _kill_tree(proc: subprocess.Popen) -> None:
    """Kill the spawned child AND its descendants (uv trampoline shims)."""
    if proc.poll() is None:
        subprocess.run(
            ["taskkill", "/PID", str(proc.pid), "/T", "/F"],
            capture_output=True,
        )
        proc.wait()


def test_named_pipe_identify_status_and_fleet_consumer(live_server, monkeypatch):
    proc, home, server_pid = live_server
    from gateway.control_socket import identify_gateway, query_gateway_control

    ident = identify_gateway(home, timeout=5.0)
    assert ident is not None, "identify returned None against a live pipe server"
    # Compare against the server's SELF-reported pid, not Popen.pid — uv's
    # Windows trampoline makes the spawner's view wrong (see _CHILD_CODE).
    assert ident["pid"] == server_pid
    assert ident["protocol"] == 1
    assert ident["kind"] == "hermes-gateway"
    assert ident["supervisor"] in {"systemd", "launchd", "desktop", "external", "manual"}

    status = query_gateway_control(home, "status", timeout=5.0)
    assert status is not None
    assert status["answering_pid"] == server_pid

    # Real fleet consumer prefers the pipe
    import hermes_cli.update_receipt as ur

    monkeypatch.setattr(
        "hermes_cli.version_info.get_code_identity",
        lambda refresh=False: {"sha": ident.get("code_sha") or "X", "version": "t"},
    )
    monkeypatch.setattr("hermes_cli.profiles._get_default_hermes_home", lambda: home)
    monkeypatch.setattr(
        "hermes_cli.profiles._get_profiles_root", lambda: home / "no-profiles"
    )
    fleet = ur.collect_fleet_versions()
    assert len(fleet) == 1, fleet
    assert fleet[0]["source"] == "socket"
    assert fleet[0]["pid"] == server_pid


@pytest.mark.spawns_gateway_lookalike
def test_pipe_gone_after_kill_falls_back(live_server, monkeypatch):
    proc, home, server_pid = live_server
    from gateway.control_socket import identify_gateway

    assert identify_gateway(home, timeout=5.0) is not None
    _kill_tree(proc)

    # taskkill /T returns once the tree is signalled; the pipe disappears when
    # the kernel tears the server's handles down.
    assert _wait_until(lambda: identify_gateway(home, timeout=0.5) is None)

    # Consumer falls back to the state file. That file is a claim, not an
    # identity: its sha classifies a row only when live_gateway_pid_for_home
    # verifies the PID as this home's gateway via its live command line
    # (#110420). A real process wearing a `gateway run` argv stands in for
    # a gateway that lost its pipe.
    import hermes_cli.update_receipt as ur

    monkeypatch.setattr(
        "hermes_cli.version_info.get_code_identity",
        lambda refresh=False: {"sha": "NEW", "version": "t"},
    )
    monkeypatch.setattr("hermes_cli.profiles._get_default_hermes_home", lambda: home)
    monkeypatch.setattr(
        "hermes_cli.profiles._get_profiles_root", lambda: home / "no-profiles"
    )

    def _write_state(pid: int) -> None:
        (home / "gateway_state.json").write_text(
            json.dumps(
                {
                    "pid": pid, "gateway_state": "running",
                    "code_sha": "OLD", "kind": "hermes-gateway",
                }
            ),
            encoding="utf-8",
        )

    standin = subprocess.Popen(
        # Use the real interpreter: a Windows venv's python.exe can be a shim
        # whose PID differs from the process running the command line. The
        # stand-in runs a SCRIPT, not `-c`: gateway identity is no longer
        # inferred from inline `-c` source (#107002).
        [getattr(sys, "_base_executable"), sleeper_script_path(), "hermes", "gateway", "run"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        assert _wait_until(lambda: _argv_visible(standin.pid, "gateway")), "stand-in argv never visible"
        _write_state(standin.pid)
        fleet = ur.collect_fleet_versions()
        assert len(fleet) == 1, fleet
        assert "source" not in fleet[0]
        assert fleet[0]["pid"] == standin.pid
        assert fleet[0]["code_sha"] == "OLD"
        assert fleet[0]["state"] == "stale"
    finally:
        _kill_tree(standin)

    # A live NON-gateway writer (this pytest process) stays visible, but its
    # self-reported sha must never classify the row.
    _write_state(os.getpid())
    fleet = ur.collect_fleet_versions()
    assert len(fleet) == 1, fleet
    assert "source" not in fleet[0]
    assert fleet[0]["pid"] == os.getpid()
    assert fleet[0]["code_sha"] is None
    assert fleet[0]["state"] == "unknown"