"""Tests for factory stop / reconcile control plane."""

from __future__ import annotations

import os
import shlex
import sqlite3
import subprocess
import sys
import time
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from adw_modules import agent_pi
from adw_modules.control import (
    argv_matches_recorded_subsequence,
    finalize_session_failed,
    open_trace_db,
    read_proc_argv,
    read_proc_start_ticks,
    reconcile_runs,
    recorded_command_argv,
    stop_run,
)
from adw_modules.data_types import PiRequest
from adw_modules.tracer import connect_db, Tracer

ROOT = Path(__file__).resolve().parents[1]
FACTORY_BIN = ROOT / "bin" / "factory"
STOP_PARENT = Path(__file__).resolve().parent / "fixtures" / "stop_parent.py"
STOP_CHILD = Path(__file__).resolve().parent / "fixtures" / "stop_child.py"
STOP_DECOY = Path(__file__).resolve().parent / "fixtures" / "stop_decoy.py"
SESSION_PROBE = Path(__file__).resolve().parent / "fixtures" / "adw_session_probe.py"
PROC_ARGV_PROBE = Path(__file__).resolve().parent / "fixtures" / "proc_argv_identity_probe.py"
CONFIG_TEMPLATE = Path(__file__).resolve().parent / "fixtures" / "test-sssf.config.yaml"


def _init_git_repo(path: Path) -> None:
    subprocess.run(["git", "init", "-q", "-b", "main", str(path)], check=True)
    subprocess.run(
        ["git", "-C", str(path), "config", "user.email", "factory@test"],
        check=True,
    )
    subprocess.run(
        ["git", "-C", str(path), "config", "user.name", "factory"],
        check=True,
    )


def _factory_env(tmp_path: Path) -> dict[str, str]:
    factory_root = tmp_path / "factory-root"
    factory_root.mkdir()
    for name in ROOT.glob("adw_*.py"):
        (factory_root / name.name).symlink_to(name)
    (factory_root / "adw_modules").symlink_to(ROOT / "adw_modules")
    (factory_root / "prompt_engineering").symlink_to(ROOT / "prompt_engineering")
    (factory_root / "pyproject.toml").symlink_to(ROOT / "pyproject.toml")
    if (ROOT / "uv.lock").exists():
        (factory_root / "uv.lock").symlink_to(ROOT / "uv.lock")
    (factory_root / "bin").mkdir()
    (factory_root / "bin" / "factory").symlink_to(FACTORY_BIN)

    data_dir = tmp_path / "factory-data"
    db_path = tmp_path / "sssf.db"
    config_text = CONFIG_TEMPLATE.read_text()
    config_text = config_text.replace("PLACEHOLDER_DATA_DIR", str(data_dir))
    config_text = config_text.replace("PLACEHOLDER_DB", str(db_path))
    (factory_root / "sssf.config.yaml").write_text(config_text)

    return {
        "FACTORY_ROOT": str(factory_root),
        "FACTORY_TEST_DB": str(db_path),
        "FACTORY_DATA_DIR": str(data_dir),
        "HOME": str(tmp_path / "home"),
        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
    }


def _session_row(db_path: Path, adw_id: str) -> tuple[str, str | None]:
    conn = sqlite3.connect(db_path)
    try:
        row = conn.execute(
            "SELECT status, ended_at FROM sessions WHERE adw_id=?",
            (adw_id,),
        ).fetchone()
        assert row is not None
        return row[0], row[1]
    finally:
        conn.close()


def _open_process_count(db_path: Path, adw_id: str) -> int:
    conn = sqlite3.connect(db_path)
    try:
        row = conn.execute(
            "SELECT count(*) FROM processes WHERE adw_id=? AND ended_at IS NULL",
            (adw_id,),
        ).fetchone()
        return int(row[0])
    finally:
        conn.close()


# A fixture ADW's interpreter startup and first db write take >5s under full-suite
# load, so these waits are liveness deadlines, not performance assertions.
PROCESS_RECORD_TIMEOUT_S = 60.0


def _wait_for_live_processes(
    db_path: Path, adw_id: str, minimum: int, timeout_s: float = PROCESS_RECORD_TIMEOUT_S
) -> None:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        try:
            if _open_process_count(db_path, adw_id) >= minimum:
                return
        except sqlite3.OperationalError:
            pass
        time.sleep(0.02)
    raise AssertionError(f"expected at least {minimum} live process rows for {adw_id}")


def _wait_for_parent_exit(proc: subprocess.Popen[str], timeout_s: float = PROCESS_RECORD_TIMEOUT_S) -> None:
    if proc.stdin is not None:
        proc.stdin.close()
    deadline = time.monotonic() + timeout_s
    while proc.poll() is None and time.monotonic() < deadline:
        time.sleep(0.02)
    if proc.poll() is None:
        proc.kill()
        proc.wait(timeout=5)
        raise AssertionError("stop_parent did not exit after factory stop")


def _run_stop_parent(
    repo: Path,
    env: dict[str, str],
    adw_id: str,
    order_file: Path,
    child_mode: str = "normal",
) -> subprocess.Popen[str]:
    run_env = {
        **os.environ,
        **env,
        "ADW_ID": adw_id,
        "ORDER_FILE": str(order_file),
        "CHILD_MODE": child_mode,
        "PYTHONPATH": env["FACTORY_ROOT"],
    }
    return subprocess.Popen(
        [sys.executable, str(STOP_PARENT)],
        cwd=repo,
        env=run_env,
        text=True,
        stdin=subprocess.PIPE,
    )


def _run_factory_stop(
    adw_id: str,
    repo: Path,
    env: dict[str, str],
) -> subprocess.CompletedProcess[str]:
    run_env = {k: v for k, v in {**os.environ, **env}.items() if k != "VIRTUAL_ENV"}
    return subprocess.run(
        [str(FACTORY_BIN), "stop", adw_id],
        cwd=repo,
        env=run_env,
        text=True,
        capture_output=True,
    )


@pytest.fixture
def stop_env(tmp_path: Path) -> dict[str, str]:
    repo = tmp_path / "project"
    repo.mkdir()
    _init_git_repo(repo)
    env = _factory_env(tmp_path)
    env["FACTORY_REPO"] = str(repo)
    return env


def _process_command(db_path: Path, adw_id: str, kind: str, name: str = "") -> str:
    conn = sqlite3.connect(db_path)
    try:
        row = conn.execute(
            "SELECT command FROM processes"
            " WHERE adw_id=? AND kind=? AND name=? AND ended_at IS NULL"
            " ORDER BY id DESC LIMIT 1",
            (adw_id, kind, name),
        ).fetchone()
        assert row is not None
        return row[0]
    finally:
        conn.close()


def test_argv_subsequence_matches_absolute_recorded_paths() -> None:
    trusted_child = "/trusted/stop_child.py"
    recorded = shlex.join(
        [sys.executable, trusted_child, "/trusted/order.txt", "child-a"],
    )
    proc = [sys.executable, trusted_child, "/trusted/order.txt", "child-a", "3"]
    assert recorded_command_argv(recorded) == [
        sys.executable,
        trusted_child,
        "/trusted/order.txt",
        "child-a",
    ]
    assert argv_matches_recorded_subsequence(proc, recorded)


def test_argv_subsequence_rejects_same_basename_different_paths() -> None:
    recorded = shlex.join(["/trusted/x.py", "child-a"])
    proc = [sys.executable, "/attacker/x.py", "child-a"]
    assert not argv_matches_recorded_subsequence(proc, recorded)


def test_malformed_recorded_command_refuses_safely(tmp_path: Path) -> None:
    proc = [sys.executable, "/tmp/stop_child.py", "child-a"]
    assert recorded_command_argv('"unclosed') == []
    assert not argv_matches_recorded_subsequence(proc, '"unclosed')
    result = stop_run(tmp_path / "sssf.db", "missing-run")
    assert result.exit_code == 2


def test_session_ensure_parent_command_matches_proc_argv(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "session-probe"
    config_path = Path(stop_env["FACTORY_ROOT"]) / "sssf.config.yaml"
    run_env = {
        **os.environ,
        **stop_env,
        "PYTHONPATH": stop_env["FACTORY_ROOT"],
    }
    proc = subprocess.Popen(
        [
            sys.executable,
            str(SESSION_PROBE),
            "--config",
            str(config_path),
            "--adw-id",
            adw_id,
            "--hold",
        ],
        cwd=repo,
        env=run_env,
        text=True,
        stdin=subprocess.PIPE,
    )
    assert proc.stdin is not None
    deadline = time.monotonic() + PROCESS_RECORD_TIMEOUT_S
    while proc.poll() is None and time.monotonic() < deadline:
        try:
            if _open_process_count(db_path, adw_id) >= 1:
                break
        except sqlite3.OperationalError:
            pass
        time.sleep(0.02)
    assert proc.poll() is None

    recorded = _process_command(db_path, adw_id, "adw")
    live_argv = read_proc_argv(proc.pid)
    assert live_argv is not None
    assert argv_matches_recorded_subsequence(live_argv, recorded)

    proc.terminate()
    proc.wait(timeout=5)
    if proc.stdin is not None:
        proc.stdin.close()


def test_agent_pi_records_complete_command_identity(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    recorded_commands: list[str] = []

    request = PiRequest(
        prompt="secret user prompt must not be stored",
        system_prompt="secret system prompt may trail in live argv",
        model="openai/gpt-5.6-terra",
        thinking="off",
        session_id="sssf-stop-test-agent",
        session_dir=str(Path(stop_env["FACTORY_DATA_DIR"]) / "pi-sessions"),
        raw_output_path=str(Path(stop_env["FACTORY_DATA_DIR"]) / "raw.jsonl"),
        cwd=str(repo),
    )

    fake_line = (
        '{"type":"message_end","message":{"role":"assistant",'
        '"content":[{"type":"text","text":"{\\"status\\":\\"success\\",'
        '\\"summary\\":\\"ok\\",\\"artifacts\\":[]}"]}],'
        '"usage":{"input":1,"output":1,"totalTokens":2,"cost":{"total":0.0}},'
        '"stopReason":"stop"}}\n'
    )
    fake_proc = MagicMock()
    fake_proc.pid = 4242
    fake_proc.stdout = iter([fake_line])
    fake_proc.stderr = MagicMock()
    fake_proc.stderr.read.return_value = ""
    fake_proc.wait.return_value = 0

    with patch("adw_modules.agent_pi.resolve_model", return_value=("openai", "gpt-5.6-terra")), \
         patch("adw_modules.agent_pi.context_window", return_value=128_000), \
         patch("adw_modules.agent_pi.subprocess.Popen", return_value=fake_proc) as popen:
        agent_pi.run(
            request,
            on_spawn=lambda _pid, command: recorded_commands.append(command),
        )
        called_cmd = popen.call_args[0][0]

    expected_identity = shlex.join(called_cmd)
    assert len(recorded_commands) == 1
    assert recorded_commands[0] == expected_identity
    assert request.prompt in recorded_commands[0]
    assert argv_matches_recorded_subsequence(called_cmd, recorded_commands[0])


def test_argv_subsequence_rejects_broad_substring() -> None:
    proc = [sys.executable, str(STOP_DECOY.resolve()), "9"]
    assert not argv_matches_recorded_subsequence(
        proc,
        shlex.join([str(STOP_CHILD.resolve()), "child-a"]),
    )


def test_read_proc_start_ticks_handles_parentheses_in_comm() -> None:
    suffix = ["S", *[str(value) for value in range(4, 22)], "987654"]
    with patch(
        "adw_modules.control.Path.read_text",
        return_value=f"123 (pi worker ) name (x)) {' '.join(suffix)}\n",
    ):
        assert read_proc_start_ticks(123) == 987654


def test_stop_children_before_parent_and_finalizes_db(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-order"
    order_file = Path(stop_env["FACTORY_DATA_DIR"]) / "order.txt"
    order_file.parent.mkdir(parents=True, exist_ok=True)

    parent = _run_stop_parent(repo, stop_env, adw_id, order_file)
    _wait_for_live_processes(db_path, adw_id, 3)

    result = _run_factory_stop(adw_id, repo, stop_env)
    assert result.returncode == 0, result.stderr + result.stdout
    _wait_for_parent_exit(parent)

    order = order_file.read_text().splitlines()
    parent_index = order.index("parent")
    assert "child-a" in order
    assert "child-b" in order
    assert order.index("child-a") < parent_index
    assert order.index("child-b") < parent_index

    status, ended_at = _session_row(db_path, adw_id)
    assert status == "fail"
    assert ended_at is not None
    assert _open_process_count(db_path, adw_id) == 0


def test_stop_command_mismatch_protects_live_decoy(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-mismatch"
    data_dir = Path(stop_env["FACTORY_DATA_DIR"])

    decoy = subprocess.Popen(
        [sys.executable, str(STOP_DECOY)],
        stdin=subprocess.PIPE,
        text=True,
    )
    assert decoy.stdin is not None
    _ = decoy.stdin

    tracer = Tracer(db_path, data_dir / "sessions" / adw_id / "events.jsonl")
    tracer.session_start(adw_id, "test", adw_name="mismatch", repo=str(repo))
    tracer.process_start(adw_id, "agent", "decoy", decoy.pid, "stop_child.py child-a")
    tracer.conn.close()

    result = _run_factory_stop(adw_id, repo, stop_env)
    assert result.returncode == 1
    assert "mismatch" in result.stderr.lower() or "identity" in result.stdout.lower()

    assert decoy.poll() is None
    status, _ = _session_row(db_path, adw_id)
    assert status == "running"
    assert _open_process_count(db_path, adw_id) == 1

    decoy.stdin.close()
    decoy.wait(timeout=5)


def test_stop_matching_start_ticks_ignores_rewritten_argv(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-start-ticks-match"
    data_dir = Path(stop_env["FACTORY_DATA_DIR"])

    decoy = subprocess.Popen(
        [sys.executable, str(STOP_DECOY)],
        stdin=subprocess.PIPE,
        text=True,
    )
    assert decoy.stdin is not None
    start_ticks = read_proc_start_ticks(decoy.pid)
    assert start_ticks is not None

    tracer = Tracer(db_path, data_dir / "sessions" / adw_id / "events.jsonl")
    tracer.session_start(adw_id, "test", adw_name="rewritten", repo=str(repo))
    tracer.process_start(
        adw_id, "agent", "pi", decoy.pid, "pi -p --mode json original prompt",
        start_ticks=start_ticks,
    )
    tracer.conn.close()

    result = _run_factory_stop(adw_id, repo, stop_env)
    assert result.returncode == 0, result.stderr + result.stdout
    decoy.wait(timeout=5)

    status, ended_at = _session_row(db_path, adw_id)
    assert status == "fail"
    assert ended_at is not None


def test_stop_mismatched_start_ticks_refuses_live_pid(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-start-ticks-mismatch"
    data_dir = Path(stop_env["FACTORY_DATA_DIR"])

    decoy = subprocess.Popen(
        [sys.executable, str(STOP_DECOY)],
        stdin=subprocess.PIPE,
        text=True,
    )
    assert decoy.stdin is not None
    live_start_ticks = read_proc_start_ticks(decoy.pid)
    assert live_start_ticks is not None
    recorded_start_ticks = live_start_ticks + 1

    tracer = Tracer(db_path, data_dir / "sessions" / adw_id / "events.jsonl")
    tracer.session_start(adw_id, "test", adw_name="reused", repo=str(repo))
    tracer.process_start(
        adw_id, "agent", "pi", decoy.pid, "x" * 5_000,
        start_ticks=recorded_start_ticks,
    )
    tracer.conn.close()

    result = _run_factory_stop(adw_id, repo, stop_env)
    assert result.returncode == 1
    assert f"start_ticks recorded={recorded_start_ticks} live={live_start_ticks}" in result.stderr
    assert "x" * 100 not in result.stderr
    assert decoy.poll() is None

    status, _ = _session_row(db_path, adw_id)
    assert status == "running"
    decoy.stdin.close()
    decoy.wait(timeout=5)


def test_stop_sigkill_escalation(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-sigkill"
    order_file = Path(stop_env["FACTORY_DATA_DIR"]) / "sigkill-order.txt"
    order_file.parent.mkdir(parents=True, exist_ok=True)

    parent = _run_stop_parent(repo, stop_env, adw_id, order_file, child_mode="ignore-term")
    _wait_for_live_processes(db_path, adw_id, 3)

    result = _run_factory_stop(adw_id, repo, stop_env)
    assert result.returncode == 0, result.stderr + result.stdout
    _wait_for_parent_exit(parent)

    status, ended_at = _session_row(db_path, adw_id)
    assert status == "fail"
    assert ended_at is not None
    assert _open_process_count(db_path, adw_id) == 0


def test_stop_idempotent_on_already_stopped(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-idempotent"

    tracer = Tracer(
        db_path,
        Path(stop_env["FACTORY_DATA_DIR"]) / "sessions" / adw_id / "events.jsonl",
    )
    tracer.session_start(adw_id, "test", adw_name="done", repo=str(repo))
    tracer.conn.close()
    conn = open_trace_db(db_path)
    try:
        finalize_session_failed(conn, adw_id)
    finally:
        conn.close()

    result = _run_factory_stop(adw_id, repo, stop_env)
    assert result.returncode == 0
    assert "already stopped" in result.stdout


def test_stop_unknown_adw_id_is_actionable(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    result = _run_factory_stop("missing-run", repo, stop_env)
    assert result.returncode == 2
    assert "unknown adw_id" in result.stdout


def test_reconcile_marks_absent_running_session_failed(stop_env: dict[str, str]) -> None:
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "reconcile-absent"
    data_dir = Path(stop_env["FACTORY_DATA_DIR"])

    tracer = Tracer(db_path, data_dir / "sessions" / adw_id / "events.jsonl")
    tracer.session_start(adw_id, "test", adw_name="crash", repo=stop_env["FACTORY_REPO"])
    tracer.process_start(adw_id, "adw", "", 424242, "stop_parent.py")
    tracer.conn.close()

    messages = reconcile_runs(db_path)
    assert any("reconciled" in line and adw_id in line for line in messages)

    status, ended_at = _session_row(db_path, adw_id)
    assert status == "fail"
    assert ended_at is not None
    assert _open_process_count(db_path, adw_id) == 0


def test_reconcile_skips_ambiguous_live_mismatch(stop_env: dict[str, str]) -> None:
    repo = Path(stop_env["FACTORY_REPO"])
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "reconcile-ambiguous"
    data_dir = Path(stop_env["FACTORY_DATA_DIR"])

    decoy = subprocess.Popen(
        [sys.executable, str(STOP_DECOY)],
        stdin=subprocess.PIPE,
        text=True,
    )
    assert decoy.stdin is not None
    _ = decoy.stdin

    tracer = Tracer(db_path, data_dir / "sessions" / adw_id / "events.jsonl")
    tracer.session_start(adw_id, "test", adw_name="live", repo=str(repo))
    tracer.process_start(adw_id, "agent", "decoy", decoy.pid, "stop_child.py child-a")
    tracer.conn.close()

    messages = reconcile_runs(db_path)
    assert any("skipped" in line and adw_id in line for line in messages)

    status, _ = _session_row(db_path, adw_id)
    assert status == "running"
    assert _open_process_count(db_path, adw_id) == 1

    decoy.stdin.close()
    decoy.wait(timeout=5)


def test_stop_children_before_parent_ordering_stable(stop_env: dict[str, str]) -> None:
    """Repeat the stop-order integration to catch sibling-order flakes."""
    for _ in range(10):
        repo = Path(stop_env["FACTORY_REPO"])
        db_path = Path(stop_env["FACTORY_TEST_DB"])
        adw_id = f"stop-order-{_}"
        order_file = Path(stop_env["FACTORY_DATA_DIR"]) / f"order-{_}.txt"
        order_file.parent.mkdir(parents=True, exist_ok=True)

        parent = _run_stop_parent(repo, stop_env, adw_id, order_file)
        _wait_for_live_processes(db_path, adw_id, 3)

        result = _run_factory_stop(adw_id, repo, stop_env)
        assert result.returncode == 0, result.stderr + result.stdout
        _wait_for_parent_exit(parent)

        order = order_file.read_text().splitlines()
        parent_index = order.index("parent")
        assert order.index("child-a") < parent_index
        assert order.index("child-b") < parent_index


def test_read_proc_argv_round_trip_direct_launch() -> None:
    proc = subprocess.Popen(
        [sys.executable, str(PROC_ARGV_PROBE), "hold"],
        stdout=subprocess.PIPE,
        text=True,
    )
    try:
        assert proc.stdout is not None
        recorded = proc.stdout.readline().strip()
        assert recorded

        live_argv = read_proc_argv(proc.pid)
        assert live_argv is not None
        assert argv_matches_recorded_subsequence(live_argv, recorded)

        pytest_main = Path(pytest.__file__).resolve().parent / "__main__.py"
        module_launch_argv = [sys.executable, "-m", "pytest", "tests", "-q"]
        argv0_recorded = shlex.join(
            [sys.executable, str(pytest_main), "tests", "-q"],
        )
        assert not argv_matches_recorded_subsequence(module_launch_argv, argv0_recorded)
    finally:
        if proc.poll() is None:
            proc.terminate()
        proc.wait(timeout=5)
        if proc.stdout is not None:
            proc.stdout.close()


def test_stop_run_direct_api(stop_env: dict[str, str]) -> None:
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    result = stop_run(db_path, "does-not-exist")
    assert result.exit_code == 2


def test_stop_cancels_pending_decisions(stop_env: dict[str, str]) -> None:
    db_path = Path(stop_env["FACTORY_TEST_DB"])
    adw_id = "stop-hitl"
    data_dir = Path(stop_env["FACTORY_DATA_DIR"])

    tracer = Tracer(db_path, data_dir / "sessions" / adw_id / "events.jsonl")
    try:
        tracer.session_start(adw_id, "test", adw_name="hitl", repo=stop_env["FACTORY_REPO"])
        decision_id = tracer.decision_request(
            adw_id,
            "Continue?",
            [{"value": "yes", "label": "Yes"}],
        )
    finally:
        tracer.conn.close()

    conn = open_trace_db(db_path)
    try:
        finalize_session_failed(conn, adw_id)
    finally:
        conn.close()

    result = _run_factory_stop(adw_id, Path(stop_env["FACTORY_REPO"]), stop_env)
    assert result.returncode == 0

    reader = connect_db(db_path)
    try:
        row = reader.execute(
            "SELECT status FROM decisions WHERE decision_id=?",
            (decision_id,),
        ).fetchone()
    finally:
        reader.close()
    assert row == ("canceled",)