"""Tests for the control-plane subcommands (spens/control.py).

No real docker: every docker-touching helper is stubbed, mirroring how
test_runner.py stubs ``spens.runner._run``.
"""

from __future__ import annotations

import json
import subprocess

import pytest
from spens import control
from spens.sinks.file import read_state


def _make_session(root, session_id: str, state: str, exit_code=None) -> None:
    d = root / "sessions" / session_id
    d.mkdir(parents=True, exist_ok=True)
    (d / "state.json").write_text(json.dumps({
        "state": state,
        "session_id": session_id,
        "exit_code": exit_code,
        "agent_container": f"spens-agent-{session_id}",
        "interceptor_container": f"spens-interceptor-{session_id}",
        "started_at": "t0",
        "updated_at": "t0",
        "prompt": "p",
        "env": "node-20",
        "agent": "opencode",
    }))


@pytest.fixture
def stopped_containers(monkeypatch: pytest.MonkeyPatch) -> list[str]:
    stopped: list[str] = []
    monkeypatch.setattr(control, "_docker_stop", lambda c: stopped.append(c) or True)
    monkeypatch.setattr(control, "_agent_exit_code", lambda c: 0)
    return stopped


# -- status -----------------------------------------------------------------


def test_status_reads_state_json(tmp_path, capsys) -> None:
    _make_session(tmp_path, "abc123", "finished", exit_code=0)
    assert control.cmd_status("abc123", tmp_path) == 0
    payload = json.loads(capsys.readouterr().out)
    assert payload == {
        "state": "finished",
        "exit_code": 0,
        "agent_container": "spens-agent-abc123",
        "updated_at": "t0",
    }


def test_status_unknown_session_is_nonzero(tmp_path, capsys) -> None:
    assert control.cmd_status("missing", tmp_path) == 1
    assert json.loads(capsys.readouterr().out) == {"state": "unknown"}


def test_status_unreadable_state_is_unknown(tmp_path, capsys) -> None:
    d = tmp_path / "sessions" / "broken"
    d.mkdir(parents=True)
    (d / "state.json").write_text("{not json")
    assert control.cmd_status("broken", tmp_path) == 1
    assert json.loads(capsys.readouterr().out) == {"state": "unknown"}


# -- cancel -----------------------------------------------------------------


def test_cancel_stops_containers_and_marks_canceled(
    tmp_path, capsys, stopped_containers
) -> None:
    _make_session(tmp_path, "run1", "running")
    assert control.cmd_cancel("run1", tmp_path) == 0
    # agent first, then interceptor; names come from state.json
    assert stopped_containers == ["spens-agent-run1", "spens-interceptor-run1"]
    state = read_state(tmp_path / "sessions" / "run1")
    assert state["state"] == "canceled"
    rows = [json.loads(l) for l in
            (tmp_path / "sessions" / "run1" / "events.jsonl").read_text().splitlines()]
    assert [r["event"] for r in rows] == ["canceled"]
    assert rows[0]["schema_version"] == 1
    assert rows[0]["session_id"] == "run1"
    assert rows[0]["data"]["reason"]
    payload = json.loads(capsys.readouterr().out)
    assert payload["state"] == "canceled"


def test_cancel_is_idempotent(tmp_path, capsys, stopped_containers) -> None:
    _make_session(tmp_path, "run1", "running")
    assert control.cmd_cancel("run1", tmp_path) == 0
    events_file = tmp_path / "sessions" / "run1" / "events.jsonl"
    first = events_file.read_text()
    assert control.cmd_cancel("run1", tmp_path) == 0
    assert events_file.read_text() == first  # no duplicate canceled event
    assert read_state(tmp_path / "sessions" / "run1")["state"] == "canceled"


def test_cancel_never_overwrites_an_existing_terminal_state(
    tmp_path, capsys, stopped_containers
) -> None:
    _make_session(tmp_path, "done1", "finished", exit_code=0)
    assert control.cmd_cancel("done1", tmp_path) == 0
    state = read_state(tmp_path / "sessions" / "done1")
    assert state["state"] == "finished"
    # no canceled event was appended either
    assert not (tmp_path / "sessions" / "done1" / "events.jsonl").exists()
    # already-canceled sessions likewise stay canceled
    _make_session(tmp_path, "cx1", "canceled")
    assert control.cmd_cancel("cx1", tmp_path) == 0
    assert read_state(tmp_path / "sessions" / "cx1")["state"] == "canceled"


def test_cancel_first_writer_wins_against_running_session(
    tmp_path, stopped_containers
) -> None:
    """An external cancel lands before the session's own terminal event."""
    from spens.events import Emitter
    from spens.sinks import FileSink

    _make_session(tmp_path, "race1", "running")
    assert control.cmd_cancel("race1", tmp_path) == 0

    session_dir = tmp_path / "sessions" / "race1"
    emitter = Emitter("race1", [FileSink(session_dir, {"session_id": "race1"})])
    emitter.emit("agent_exited", exit_code=137)
    emitter.emit("finished", exit_code=137, summary_path="/s")

    state = read_state(session_dir)
    assert state["state"] == "canceled"      # terminal state not overwritten
    assert state["exit_code"] == 137         # final exit code still recorded


def test_cancel_unknown_session_is_nonzero(tmp_path, capsys) -> None:
    assert control.cmd_cancel("missing", tmp_path) == 1


def test_cancel_missing_state_is_nonzero(tmp_path, capsys) -> None:
    d = tmp_path / "sessions" / "empty"
    d.mkdir(parents=True)
    assert control.cmd_cancel("empty", tmp_path) == 1


def test_cancel_reports_docker_failure(tmp_path, monkeypatch, capsys) -> None:
    _make_session(tmp_path, "run2", "running")
    monkeypatch.setattr(control, "_docker_stop", lambda c: False)
    assert control.cmd_cancel("run2", tmp_path) == 1
    # and the session was NOT marked canceled (docker failed)
    assert read_state(tmp_path / "sessions" / "run2")["state"] == "running"


def test_cancel_warns_when_agent_was_sigkilled(
    tmp_path, stopped_containers, monkeypatch
) -> None:
    _make_session(tmp_path, "kill1", "running")
    monkeypatch.setattr(control, "_agent_exit_code", lambda c: 137)
    assert control.cmd_cancel("kill1", tmp_path) == 0
    rows = [json.loads(l) for l in
            (tmp_path / "sessions" / "kill1" / "events.jsonl").read_text().splitlines()]
    assert [r["event"] for r in rows] == ["canceled", "warning"]
    assert "may still be present" in rows[1]["data"]["message"]


# -- _docker_stop (real function, stubbed docker binary) ---------------------


def test_docker_stop_tolerates_missing_container(monkeypatch) -> None:
    class Result:
        returncode = 1
        stderr = "Error response from daemon: No such container: foo"

    monkeypatch.setattr(
        subprocess, "run", lambda cmd, **kw: Result()
    )
    assert control._docker_stop("foo") is True


def test_docker_stop_hard_failure_is_false(monkeypatch) -> None:
    class Result:
        returncode = 1
        stderr = "docker daemon is not running"

    monkeypatch.setattr(
        subprocess, "run", lambda cmd, **kw: Result()
    )
    assert control._docker_stop("foo") is False


# -- attach -----------------------------------------------------------------


def test_attach_exits_with_session_exit_code(tmp_path, monkeypatch, capsys) -> None:
    _make_session(tmp_path, "fin1", "finished", exit_code=7)
    # no docker logs thread: session already terminal
    assert control.cmd_attach("fin1", tmp_path) == 7
    assert capsys.readouterr().out == ""  # no events file -> nothing printed


def test_attach_unknown_session_is_nonzero(tmp_path) -> None:
    assert control.cmd_attach("missing", tmp_path) == 1


def test_final_exit_code_mapping() -> None:
    assert control._final_exit_code({"exit_code": 7, "state": "finished"}) == 7
    assert control._final_exit_code({"exit_code": None, "state": "canceled"}) == 130
    assert control._final_exit_code({"exit_code": None, "state": "error"}) == 1
    assert control._final_exit_code({"exit_code": None, "state": "running"}) == 0


def test_attach_follows_events_until_terminal(tmp_path, monkeypatch, capsys) -> None:
    import threading

    from spens.events import Emitter
    from spens.sinks import FileSink

    _make_session(tmp_path, "live1", "running")
    session_dir = tmp_path / "sessions" / "live1"
    monkeypatch.setattr(control, "ATTACH_POLL_SECS", 0.05)
    monkeypatch.setattr(
        control, "_follow_docker_logs", lambda container: None
    )

    emitter = Emitter("live1", [FileSink(session_dir, {"session_id": "live1"})])
    emitter.emit("started")

    def finish_later():
        emitter.emit("agent_started", container="spens-agent-live1")
        emitter.emit("agent_exited", exit_code=0)
        emitter.emit("finished", exit_code=0)

    thread = threading.Timer(0.2, finish_later)
    thread.start()
    try:
        assert control.cmd_attach("live1", tmp_path) == 0
    finally:
        thread.join()
    out = capsys.readouterr().out
    seen = [json.loads(line)["event"] for line in out.splitlines()]
    assert seen == ["started", "agent_started", "agent_exited", "finished"]


# -- dispatch ---------------------------------------------------------------


def test_run_control_command_dispatch(tmp_path, monkeypatch, stopped_containers) -> None:
    _make_session(tmp_path, "d1", "finished", exit_code=2)
    assert control.run_control_command(
        "status", ["d1", "--spens-dir", str(tmp_path)]
    ) == 0
    with pytest.raises(SystemExit):
        control.run_control_command("status", [])  # missing session id
