#!/usr/bin/env python3
"""Fake ``agy --output-format stream-json`` for ``nvsh.agent.agy.AgyAgent``.

Reads a transcript spec from ``$NVSH_FAKE_EVENTS`` (a JSON file) and replays
it verbatim, the same "record real CLI output, replay it in tests" pattern
as ``tests/fakes/codex``/``tests/fakes/claude``/``tests/fakes/qwen`` -- the
difference is the spec here carries whole recorded NDJSON *lines* (agy's
own wire shapes), not an abstract event-kind list, since
``tests/test_agent_agy.py`` needs to prove ``AgyAgent`` parses agy's actual
wire format.

Spec shape, one of:

* Cold (no ``--input-format`` on argv): a single turn --
  ``{"stdout": [<ndjson line>, ...], "stderr": [<line>, ...], "exit_code": N}``.
* Warm (``--input-format stream-json`` on argv): one turn per line read
  from stdin -- ``{"turns": [{"stdout": [...], "stderr": [...],
  "exit_code": N}, ...]}``. Each stdin line consumed plays back the next
  turn's stdout/stderr; the process exits once ``turns`` is exhausted (or
  immediately, with the last turn's ``exit_code``, if stdin closes first).

``$NVSH_FAKE_ARGV_LOG``, when set, gets one JSON line appended per process
invocation (``argv[1:]``) -- how ``tests/test_agent_agy.py`` proves the
exact argv shape (criterion 1) and counts how many processes a "warm"
session actually spawned (should be one, across many turns) without
inspecting ``AgyAgent`` internals.

Nothing here imports ``nvsh`` -- it is spawned as a real subprocess, exactly
like the real ``agy`` binary would be.

A few more knobs, for the reliable-agent-stop tests:

* ``NVSH_FAKE_IGNORE_CANCEL=1`` -- agy has no protocol-level cancel at all
  (real cancellation is signal-only), so "ignoring cancel" here means the
  turn simply never ends on its own: once the scripted turns are exhausted,
  the process blocks reading stdin forever instead of exiting. It also
  makes this fake ignore ``SIGINT``/``SIGTERM`` (see ``NVSH_FAKE_SIGNAL_LOG``
  below) instead of exiting on them, standing in for a harness whose stop
  signal is swallowed -- only ``SIGKILL``
  (``nvsh.agent._subprocess.kill_tree``'s last rung) can still end it.
* ``NVSH_FAKE_GRANDCHILD=1`` -- start a ``sleep 600`` child at launch and
  write ``{"harness": <this pid>, "grandchild": <sleep's pid>}`` to
  ``$NVSH_FAKE_PID_FILE``.
* ``NVSH_FAKE_SIGNAL_LOG=<path>`` -- append one JSON line
  (``{"signal": <int>}``) to *path* whenever this process receives
  ``SIGINT`` or ``SIGTERM``, so a test can prove a warm ``AgyAgent.cancel()``
  really reached the child instead of only flipping a local flag (agy has
  no protocol cancel to send). Without ``NVSH_FAKE_IGNORE_CANCEL``, the
  fake exits right after logging, the way a real ``agy`` asked to stop
  would.
* ``NVSH_FAKE_READY_FILE=<path>`` -- touched once signal handling is
  actually installed, so a test can wait for it instead of racing this
  process's own interpreter startup before sending a signal.
* ``NVSH_FAKE_LATE_LINE=<ndjson line>`` -- written to stdout from inside
  the ``SIGINT``/``SIGTERM`` handler itself, standing in for a real agy
  that keeps emitting output from the turn being cancelled for a moment
  after the signal arrives (it has no protocol-level cancel to make that
  synchronous). Combine with ``NVSH_FAKE_IGNORE_CANCEL=1`` to keep the
  warm process alive afterward, so a test can prove the *next* turn never
  sees this line.
* A turn's own ``"sleep_before"`` (seconds) makes ``_play_turn`` sleep
  before writing that turn's output -- long enough that a test can be sure
  the fake is still "working" the turn when it sends a signal, instead of
  racing a turn that finishes on its own first.
* ``NVSH_FAKE_TURN_STARTED_FILE=<path>`` (warm mode only) -- touched right
  after each turn's stdin line is read, before ``_play_turn`` runs (so
  before any ``sleep_before``). A test that calls ``cancel()`` right after
  starting a background ``run()`` races that background thread's own
  write -- ``AgyAgent.start()`` already spawned the child, so waiting for
  ``self._proc`` alone proves nothing about whether *this* turn's prompt
  has actually been written yet. Waiting on this file first proves the
  write landed before ``cancel()`` fires.
"""

from __future__ import annotations

import json
import os
import signal
import subprocess  # nosec B404 - fixed argv below, test fixture only
import sys
import time
from pathlib import Path


def spawn_grandchild_if_requested() -> None:
    """``NVSH_FAKE_GRANDCHILD=1``: start a sleeping child and record both pids."""
    if os.environ.get("NVSH_FAKE_GRANDCHILD") != "1":
        return
    pid_file = os.environ.get("NVSH_FAKE_PID_FILE")
    child = subprocess.Popen(["sleep", "600"])  # nosec B603 B607 - fixed argv
    if pid_file:
        Path(pid_file).write_text(
            json.dumps({"harness": os.getpid(), "grandchild": child.pid}), encoding="utf-8"
        )


def hang_if_ignoring_cancel() -> None:
    """``NVSH_FAKE_IGNORE_CANCEL=1``: never let the turn end on its own."""
    if os.environ.get("NVSH_FAKE_IGNORE_CANCEL") != "1":
        return
    while sys.stdin.readline():
        pass


def install_signal_log() -> None:
    """``NVSH_FAKE_SIGNAL_LOG``: log ``SIGINT``/``SIGTERM``, then act on it.

    agy has no protocol-level cancel (see module docstring), so
    ``AgyAgent.cancel()``/``force_stop()`` can only ever reach a warm child
    through a signal. Logging receipt here is what
    ``tests/test_agent_agy.py`` checks to prove that reached the child
    instead of nvsh only flipping a local flag. Ignoring the signal instead
    of exiting on it (``NVSH_FAKE_IGNORE_CANCEL=1``) stands in for a
    harness whose own stop handling swallows it.
    """
    log_path = os.environ.get("NVSH_FAKE_SIGNAL_LOG")
    ignore = os.environ.get("NVSH_FAKE_IGNORE_CANCEL") == "1"
    late_line = os.environ.get("NVSH_FAKE_LATE_LINE")
    if not log_path and not ignore and not late_line:
        return  # nothing to log, default signal disposition is fine as-is

    def _handler(signum: int, _frame: object) -> None:
        if log_path:
            with open(log_path, "a", encoding="utf-8") as fh:
                fh.write(json.dumps({"signal": int(signum)}) + "\n")
        if late_line:
            sys.stdout.write(late_line.rstrip("\n") + "\n")
            sys.stdout.flush()
        if not ignore:
            os._exit(0)

    signal.signal(signal.SIGINT, _handler)
    signal.signal(signal.SIGTERM, _handler)


def mark_signal_handling_ready() -> None:
    """``NVSH_FAKE_READY_FILE``: touch *path* once signal handling is live.

    Sending a signal right after ``Popen()`` returns races the child's own
    interpreter startup -- until it has actually run
    :func:`install_signal_log`, ``SIGINT``/``SIGTERM`` fall to the OS
    default (kill outright, logging nothing), which a test could easily
    mistake for the fake correctly ignoring/logging the signal. This marker
    gives a test something concrete to wait on instead of a fixed sleep.
    """
    path = os.environ.get("NVSH_FAKE_READY_FILE")
    if path:
        Path(path).write_text("1", encoding="utf-8")


def _load_spec() -> dict:
    events_path = os.environ.get("NVSH_FAKE_EVENTS")
    if not events_path:
        return {}
    with open(events_path, encoding="utf-8") as fh:
        return json.load(fh)


def _log_argv(argv: list[str]) -> None:
    log_path = os.environ.get("NVSH_FAKE_ARGV_LOG")
    if not log_path:
        return
    with open(log_path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps(argv) + "\n")


def _play_turn(turn: dict) -> int:
    delay = turn.get("sleep_before")
    if delay:
        time.sleep(float(delay))
    for line in turn.get("stdout", []):
        sys.stdout.write(line.rstrip("\n") + "\n")
    sys.stdout.flush()
    for line in turn.get("stderr", []):
        sys.stderr.write(line.rstrip("\n") + "\n")
    sys.stderr.flush()
    return int(turn.get("exit_code", 0))


def main(argv: list[str]) -> int:
    _log_argv(argv)
    install_signal_log()
    mark_signal_handling_ready()
    spawn_grandchild_if_requested()
    spec = _load_spec()
    warm = "--input-format" in argv

    if not warm:
        exit_code = _play_turn(spec)
        hang_if_ignoring_cancel()
        return exit_code

    turn_started_file = os.environ.get("NVSH_FAKE_TURN_STARTED_FILE")
    turns = list(spec.get("turns", []))
    exit_code = 0
    for turn in turns:
        line = sys.stdin.readline()
        if not line:
            break
        if turn_started_file:
            Path(turn_started_file).write_text("1", encoding="utf-8")
        exit_code = _play_turn(turn)
    hang_if_ignoring_cancel()
    return exit_code


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))
