"""Tests for durable human-in-the-loop decisions."""

from __future__ import annotations

import json
import sqlite3
import threading
import time
from pathlib import Path

import pytest

from adw_modules.control import main as control_main, reconcile_runs, stop_run
from adw_modules.data_types import (
    AgentCall,
    DecisionOption,
    HumanDecisionRequest,
    PlanOutput,
)
from adw_modules.hitl import (
    DecisionCanceled,
    DecisionTimeout,
    TooManyHumanDecisions,
    ask_human,
    resolve_plan_decisions,
    wait_for_decision,
)
from adw_modules.tracer import Tracer, connect_db
from types import SimpleNamespace

LEGACY_SCHEMA = """
CREATE TABLE sessions (
  adw_id        TEXT PRIMARY KEY,
  request       TEXT,
  status        TEXT,
  engineer      TEXT,
  started_at    TEXT, ended_at TEXT,
  total_tokens  INTEGER, total_cost REAL
);
"""

OPTIONS = [
    {"value": "yes", "label": "Yes", "recommended": True},
    {"value": "no", "label": "No"},
]


def _tracer(tmp_path: Path) -> Tracer:
    return Tracer(tmp_path / "sssf.db", tmp_path / "events.jsonl")


def _decision_events(tracer: Tracer, kind: str) -> list[dict]:
    rows = tracer.conn.execute(
        "SELECT payload_json FROM events WHERE type=? ORDER BY started_at",
        (kind,),
    ).fetchall()
    return [json.loads(row[0]) for row in rows]


def test_migration_creates_decisions_table_on_fresh_db(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tables = {
            row[0]
            for row in tracer.conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'",
            )
        }
        indexes = {
            row[0]
            for row in tracer.conn.execute(
                "SELECT name FROM sqlite_master WHERE type='index'",
            )
        }
    finally:
        tracer.conn.close()

    assert "decisions" in tables
    assert "idx_decisions_status" in indexes


def test_migration_creates_decisions_table_on_existing_db(tmp_path: Path) -> None:
    db = tmp_path / "legacy.db"
    events = tmp_path / "events.jsonl"

    conn = sqlite3.connect(db, isolation_level=None)
    try:
        conn.executescript(LEGACY_SCHEMA)
        conn.execute(
            "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?)",
            ("legacy1", "fix bug", "success", "alice",
             "2026-01-01T00:00:00Z", "2026-01-01T01:00:00Z", 100, 0.01),
        )
    finally:
        conn.close()

    tracer = Tracer(db, events)
    try:
        columns = {
            row[1] for row in tracer.conn.execute("PRAGMA table_info(decisions)")
        }
    finally:
        tracer.conn.close()

    assert "decision_id" in columns
    assert "status" in columns


def test_request_pending_answer_roundtrip(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request(
            "adw1", "Proceed?", OPTIONS, phase="plan",
        )
        pending = tracer.decisions_pending("adw1")
        assert len(pending) == 1
        assert pending[0]["decision_id"] == decision_id
        assert pending[0]["status"] == "pending"

        assert tracer.decision_answer(decision_id, value="yes", answered_by="test")
        answered = tracer.decision_get(decision_id)
        assert answered is not None
        assert answered["status"] == "answered"
        assert answered["answer_value"] == "yes"
        assert answered["answered_by"] == "test"
        assert answered["answered_at"] is not None

        requested = _decision_events(tracer, "decision_requested")
        answered_events = _decision_events(tracer, "decision_answered")
        assert requested[0]["decision_id"] == decision_id
        assert requested[0]["question"] == "Proceed?"
        assert answered_events[0]["decision_id"] == decision_id
        assert answered_events[0]["answer_value"] == "yes"
    finally:
        tracer.conn.close()


def test_double_answer_returns_false_and_leaves_row_unchanged(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request("adw1", "Proceed?", OPTIONS)
        assert tracer.decision_answer(decision_id, value="yes")
        first = tracer.decision_get(decision_id)
        assert tracer.decision_answer(decision_id, value="no") is False
        second = tracer.decision_get(decision_id)
        assert second == first
    finally:
        tracer.conn.close()


def test_invalid_choice_raises_and_row_stays_pending(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request("adw1", "Proceed?", OPTIONS)
        with pytest.raises(ValueError, match="invalid choice"):
            tracer.decision_answer(decision_id, value="maybe")
        row = tracer.decision_get(decision_id)
        assert row is not None
        assert row["status"] == "pending"
    finally:
        tracer.conn.close()


def test_free_text_answer_stores_text(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request(
            "adw1", "Explain?", [], free_text=True,
        )
        assert tracer.decision_answer(decision_id, text="because reasons")
        row = tracer.decision_get(decision_id)
        assert row is not None
        assert row["answer_text"] == "because reasons"
        assert row["answer_value"] is None
    finally:
        tracer.conn.close()


def test_options_required_when_free_text_false(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        with pytest.raises(ValueError, match="options required"):
            tracer.decision_request("adw1", "Proceed?", [])
    finally:
        tracer.conn.close()


def test_wait_for_decision_returns_after_concurrent_answer(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request("adw1", "Proceed?", OPTIONS)

        def answer_later() -> None:
            time.sleep(0.05)
            other = Tracer(tracer.db_path, tracer.events_jsonl)
            try:
                other.decision_answer(decision_id, value="yes", answered_by="human")
            finally:
                other.conn.close()

        thread = threading.Thread(target=answer_later)
        thread.start()
        result = wait_for_decision(tracer, decision_id, poll_sec=0.05)
        thread.join(timeout=5)
        assert result["status"] == "answered"
        assert result["answer_value"] == "yes"
    finally:
        tracer.conn.close()


def test_wait_for_decision_raises_decision_canceled(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request("adw1", "Proceed?", OPTIONS)
        tracer.decisions_cancel("adw1")
        with pytest.raises(DecisionCanceled):
            wait_for_decision(tracer, decision_id, poll_sec=0.05)
    finally:
        tracer.conn.close()


def test_wait_for_decision_timeout_raises_decision_timeout(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request("adw1", "Proceed?", OPTIONS)
        with pytest.raises(DecisionTimeout):
            wait_for_decision(
                tracer, decision_id, poll_sec=0.05, timeout_sec=0.15,
            )
    finally:
        tracer.conn.close()


def test_ask_human_roundtrip(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    events = tmp_path / "events.jsonl"
    tracer = Tracer(db, events)
    try:
        tracer.session_start("adw1", "alice")
    finally:
        tracer.conn.close()

    results: list[dict] = []

    def ask_later() -> None:
        worker = Tracer(db, events)
        try:
            results.append(
                ask_human(worker, "adw1", "Proceed?", OPTIONS, poll_sec=0.05),
            )
        finally:
            worker.conn.close()

    def answer_when_pending() -> None:
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            reader = connect_db(db)
            try:
                row = reader.execute(
                    "SELECT decision_id FROM decisions"
                    " WHERE adw_id=? AND status='pending' LIMIT 1",
                    ("adw1",),
                ).fetchone()
            finally:
                reader.close()
            if row:
                other = Tracer(db, events)
                try:
                    other.decision_answer(row[0], value="no")
                finally:
                    other.conn.close()
                return
            time.sleep(0.02)
        raise AssertionError("decision never became pending")

    ask_thread = threading.Thread(target=ask_later)
    answer_thread = threading.Thread(target=answer_when_pending)
    ask_thread.start()
    answer_thread.start()
    ask_thread.join(timeout=5)
    answer_thread.join(timeout=5)
    assert results[0]["answer_value"] == "no"


def test_stop_cancels_session_pending_decisions(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    events = tmp_path / "events.jsonl"
    tracer = Tracer(db, events)
    adw_id = "stop-decisions"
    try:
        tracer.session_start(adw_id, "alice")
        decision_id = tracer.decision_request(adw_id, "Continue?", OPTIONS)
    finally:
        tracer.conn.close()

    result = stop_run(db, adw_id)
    assert result.exit_code == 0

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


def test_reconcile_cancels_dead_session_pending_decisions(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    events = tmp_path / "events.jsonl"
    adw_id = "reconcile-decisions"
    tracer = Tracer(db, events)
    try:
        tracer.session_start(adw_id, "alice")
        tracer.process_start(adw_id, "adw", "", 424242, "stop_parent.py")
        decision_id = tracer.decision_request(adw_id, "Continue?", OPTIONS)
    finally:
        tracer.conn.close()

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

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


def test_control_answer_success_and_not_pending_exit_codes(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    events = tmp_path / "events.jsonl"
    config = tmp_path / "sssf.config.yaml"
    config.write_text(
        "defaults:\n  data_dir: data\n"
        f"observability:\n  db: {db}\n"
        "agents: []\n"
    )

    tracer = Tracer(db, events)
    try:
        tracer.session_start("adw1", "alice")
        decision_id = tracer.decision_request("adw1", "Proceed?", OPTIONS)
    finally:
        tracer.conn.close()

    ok = control_main(["answer", decision_id, "--choice", "yes", "--config", str(config)])
    assert ok == 0

    again = control_main(["answer", decision_id, "--choice", "no", "--config", str(config)])
    assert again == 1

    bad = control_main(["answer", decision_id, "--choice", "nope", "--config", str(config)])
    assert bad == 1


def _plan_output(**kwargs) -> PlanOutput:
    return PlanOutput(status="success", summary="planned", **kwargs)


def _decision_request(**kwargs) -> HumanDecisionRequest:
    options = kwargs.pop("options", [
        DecisionOption(value="a", label="Option A"),
        DecisionOption(value="b", label="Option B"),
    ])
    return HumanDecisionRequest(
        question=kwargs.pop("question", "Which way?"),
        options=options,
        **kwargs,
    )


class _FakePhaseHandle:
    def __init__(self, outputs: list[PlanOutput]):
        self.phase = SimpleNamespace(name="plan", params=SimpleNamespace(name="plan"))
        self._outputs = list(outputs)
        self.calls: list[AgentCall] = []

    def call(self, call: AgentCall) -> PlanOutput:
        self.calls.append(call)
        if not self._outputs:
            raise AssertionError("no more fake planner outputs")
        return self._outputs.pop(0)


def _fake_run(tracer: Tracer, adw_id: str = "adw1") -> SimpleNamespace:
    return SimpleNamespace(tracer=tracer, adw_id=adw_id)


def test_resolve_plan_decisions_no_human_decision_single_call(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        expected = _plan_output(commit_message="plan it")
        ph = _FakePhaseHandle([expected])
        run = _fake_run(tracer)
        call = AgentCall(output_type=PlanOutput, prompt="build feature x")

        result = resolve_plan_decisions(run, ph, call)

        assert result is expected
        assert len(ph.calls) == 1
        assert ph.calls[0].prompt == "build feature x"
    finally:
        tracer.conn.close()


def test_resolve_plan_decisions_blocks_until_answered(tmp_path: Path) -> None:
    db = tmp_path / "sssf.db"
    events = tmp_path / "events.jsonl"
    tracer = Tracer(db, events)
    try:
        tracer.session_start("adw1", "alice")
    finally:
        tracer.conn.close()

    first = _plan_output(human_decision=_decision_request(
        question="Paid or free?",
        options=[
            DecisionOption(value="paid", label="Paid"),
            DecisionOption(value="free", label="Free"),
        ],
    ))
    second = _plan_output(commit_message="scoped")
    ph = _FakePhaseHandle([first, second])
    call = AgentCall(output_type=PlanOutput, prompt="launch pricing")
    results: list[PlanOutput] = []

    def resolve_later() -> None:
        worker = Tracer(db, events)
        try:
            results.append(
                resolve_plan_decisions(_fake_run(worker), ph, call, poll_sec=0.05),
            )
        finally:
            worker.conn.close()

    def answer_when_pending() -> None:
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            reader = connect_db(db)
            try:
                row = reader.execute(
                    "SELECT decision_id FROM decisions"
                    " WHERE adw_id=? AND status='pending' LIMIT 1",
                    ("adw1",),
                ).fetchone()
            finally:
                reader.close()
            if row:
                other = Tracer(db, events)
                try:
                    other.decision_answer(row[0], value="paid", answered_by="owner")
                finally:
                    other.conn.close()
                return
            time.sleep(0.02)
        raise AssertionError("decision never became pending")

    resolve_thread = threading.Thread(target=resolve_later)
    answer_thread = threading.Thread(target=answer_when_pending)
    resolve_thread.start()
    answer_thread.start()
    resolve_thread.join(timeout=5)
    answer_thread.join(timeout=5)

    assert results[0] is second
    assert len(ph.calls) == 2
    assert "## Human decision" in ph.calls[1].prompt
    assert "Q: Paid or free?" in ph.calls[1].prompt
    assert "A: paid" in ph.calls[1].prompt


def test_resolve_plan_decisions_canceled_while_waiting(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        first = _plan_output(human_decision=_decision_request())
        ph = _FakePhaseHandle([first])
        run = _fake_run(tracer)
        call = AgentCall(output_type=PlanOutput, prompt="scope it")

        def cancel_later() -> None:
            time.sleep(0.05)
            other = Tracer(tracer.db_path, tracer.events_jsonl)
            try:
                other.decisions_cancel("adw1")
            finally:
                other.conn.close()

        thread = threading.Thread(target=cancel_later)
        thread.start()
        with pytest.raises(DecisionCanceled):
            resolve_plan_decisions(run, ph, call, poll_sec=0.05)
        thread.join(timeout=5)
    finally:
        tracer.conn.close()


def test_resolve_plan_decisions_max_rounds_exceeded(tmp_path: Path) -> None:
    tracer = _tracer(tmp_path)
    try:
        tracer.session_start("adw1", "alice")
        pending = _plan_output(human_decision=_decision_request())
        ph = _FakePhaseHandle([pending, pending, pending])
        run = _fake_run(tracer)
        call = AgentCall(output_type=PlanOutput, prompt="scope it")

        def answer_loop() -> None:
            for _ in range(2):
                deadline = time.monotonic() + 5.0
                while time.monotonic() < deadline:
                    reader = connect_db(tracer.db_path)
                    try:
                        row = reader.execute(
                            "SELECT decision_id FROM decisions"
                            " WHERE adw_id=? AND status='pending' LIMIT 1",
                            ("adw1",),
                        ).fetchone()
                    finally:
                        reader.close()
                    if row:
                        other = Tracer(tracer.db_path, tracer.events_jsonl)
                        try:
                            other.decision_answer(row[0], value="a")
                        finally:
                            other.conn.close()
                        break
                    time.sleep(0.02)
                else:
                    raise AssertionError("decision never became pending")

        thread = threading.Thread(target=answer_loop)
        thread.start()
        with pytest.raises(TooManyHumanDecisions):
            resolve_plan_decisions(run, ph, call, max_rounds=2, poll_sec=0.05)
        thread.join(timeout=5)
        assert len(ph.calls) == 3
    finally:
        tracer.conn.close()