orchestrator/agent-team/tests/test_operator_cli.py

"""Unit tests for agent_team.operator_cli (§3.3.1, §6.6).

Covers the two design-mandated invariants — every destructive action is
audit-logged, and destructive actions require an explicit confirmation flag —
plus the ledger lifecycle effects (which delegate to the committed
compare-and-set helpers) and the argparse entrypoint.
"""

from __future__ import annotations

import json
import stat
from pathlib import Path

import pytest

from agent_team.db.schema import connect, init_db
from agent_team.operator_cli import (
    DESTRUCTIVE_ACTIONS,
    AuditEntry,
    AuditLog,
    ConfirmationRequired,
    OperatorCli,
    QuestionNotFound,
    build_parser,
    main,
)


# --------------------------------------------------------------------------- #
# fixtures / helpers
# --------------------------------------------------------------------------- #


def _insert_open_question(
    db_path: Path,
    qid: str,
    *,
    thread_id: str = "thread-1",
    turn: int = 0,
    status: str = "open",
    transport: str = "slack",
    channel_ref: str | None = "slack-ts-1",
) -> None:
    conn = connect(db_path)
    try:
        conn.execute(
            "INSERT INTO pending_questions "
            "(question_id, thread_id, turn, status, transport, channel_ref, "
            " posted_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                qid,
                thread_id,
                turn,
                status,
                transport,
                channel_ref,
                "2026-06-17T00:00:00",
            ),
        )
    finally:
        conn.close()


@pytest.fixture()
def db_path(tmp_path: Path) -> Path:
    db = tmp_path / "agent_team.sqlite"
    init_db(db)
    return db


@pytest.fixture()
def audit_path(tmp_path: Path) -> Path:
    return tmp_path / "audit" / "operator.jsonl"


@pytest.fixture()
def cli(db_path: Path, audit_path: Path) -> OperatorCli:
    operator = OperatorCli(db_path, audit_path, actor="tester")
    yield operator
    operator.close()


def _status(db_path: Path, qid: str) -> str:
    conn = connect(db_path)
    try:
        return conn.execute(
            "SELECT status FROM pending_questions WHERE question_id=?", (qid,)
        ).fetchone()["status"]
    finally:
        conn.close()


# --------------------------------------------------------------------------- #
# AuditEntry / AuditLog
# --------------------------------------------------------------------------- #


def test_audit_entry_to_dict_is_json_safe() -> None:
    entry = AuditEntry(
        timestamp="2026-06-17T00:00:00+00:00",
        actor="tester",
        action="force-expire",
        phase="attempt",
        confirmed=True,
        question_id="q1",
        thread_id="t1",
        detail={"k": "v"},
    )
    data = json.loads(entry.to_json())
    assert data["action"] == "force-expire"
    assert data["confirmed"] is True
    assert data["detail"] == {"k": "v"}


def test_audit_entry_is_frozen() -> None:
    entry = AuditEntry(
        timestamp="t", actor="a", action="x", phase="attempt", confirmed=False
    )
    with pytest.raises(Exception):
        entry.actor = "other"  # type: ignore[misc]


def test_audit_log_append_is_jsonl_and_appends(audit_path: Path) -> None:
    log = AuditLog(audit_path)
    log.append(
        AuditEntry(
            timestamp="t1", actor="a", action="x", phase="attempt", confirmed=True
        )
    )
    log.append(
        AuditEntry(
            timestamp="t2", actor="a", action="y", phase="outcome", confirmed=True
        )
    )
    records = log.read_all()
    assert [r["action"] for r in records] == ["x", "y"]
    # raw file is one JSON object per line
    lines = audit_path.read_text().splitlines()
    assert len(lines) == 2
    assert json.loads(lines[0])["timestamp"] == "t1"


def test_audit_log_file_is_mode_600(audit_path: Path) -> None:
    log = AuditLog(audit_path)
    log.append(
        AuditEntry(
            timestamp="t", actor="a", action="x", phase="attempt", confirmed=True
        )
    )
    mode = stat.S_IMODE(audit_path.stat().st_mode)
    assert mode == 0o600


def test_audit_log_read_all_empty_when_absent(audit_path: Path) -> None:
    assert AuditLog(audit_path).read_all() == []


# --------------------------------------------------------------------------- #
# confirmation gating (§3.3.1: explicit confirmation flag)
# --------------------------------------------------------------------------- #


@pytest.mark.parametrize("action", sorted(DESTRUCTIVE_ACTIONS))
def test_destructive_actions_set_matches_design(action: str) -> None:
    assert action in {"force-expire", "answer-on-behalf", "force-resume"}


def test_force_expire_without_confirm_raises_and_no_mutation(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1")
    with pytest.raises(ConfirmationRequired):
        cli.force_expire("q1")
    assert _status(db_path, "q1") == "open"  # not mutated


def test_answer_on_behalf_without_confirm_raises_and_no_mutation(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1")
    with pytest.raises(ConfirmationRequired):
        cli.answer_on_behalf("q1", {"approve": True})
    assert _status(db_path, "q1") == "open"


def test_force_resume_without_confirm_raises_and_no_mutation(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1", status="answered")
    with pytest.raises(ConfirmationRequired):
        cli.force_resume("q1")
    assert _status(db_path, "q1") == "answered"


def test_refused_action_is_audit_logged(cli: OperatorCli, db_path: Path) -> None:
    _insert_open_question(db_path, "q1")
    with pytest.raises(ConfirmationRequired):
        cli.force_expire("q1")
    records = cli.audit_log.read_all()
    phases = [(r["action"], r["phase"]) for r in records]
    # attempt + refusal outcome both recorded
    assert ("force-expire", "attempt") in phases
    assert ("force-expire", "outcome") in phases
    refusal = [r for r in records if r["phase"] == "outcome"][0]
    assert refusal["confirmed"] is False
    assert "refused" in refusal["detail"]


# --------------------------------------------------------------------------- #
# destructive actions: audit-logged on success (§3.3.1: audit-logged)
# --------------------------------------------------------------------------- #


def test_force_expire_confirmed_mutates_and_audits(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1")
    result = cli.force_expire("q1", confirm=True)
    assert result.ok is True
    assert _status(db_path, "q1") == "expired"
    records = cli.audit_log.read_all()
    actions = [(r["action"], r["phase"], r["confirmed"]) for r in records]
    assert ("force-expire", "attempt", True) in actions
    assert ("force-expire", "outcome", True) in actions
    outcome = [r for r in records if r["phase"] == "outcome"][0]
    assert outcome["question_id"] == "q1"
    assert outcome["thread_id"] == "thread-1"
    assert outcome["detail"]["changed"] is True


def test_answer_on_behalf_confirmed_writes_first_answer_wins(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1")
    result = cli.answer_on_behalf("q1", {"approve": True}, confirm=True)
    assert result.ok is True
    conn = connect(db_path)
    try:
        row = conn.execute(
            "SELECT status, answer_json, answered_via FROM pending_questions "
            "WHERE question_id='q1'"
        ).fetchone()
    finally:
        conn.close()
    assert row["status"] == "answered"
    assert json.loads(row["answer_json"]) == {"approve": True}
    assert row["answered_via"] == "operator:tester"
    # answer payload + via captured in audit attempt detail
    attempt = [
        r
        for r in cli.audit_log.read_all()
        if r["action"] == "answer-on-behalf" and r["phase"] == "attempt"
    ][0]
    assert attempt["detail"]["answer"] == {"approve": True}
    assert attempt["detail"]["via"] == "operator:tester"


def test_answer_on_behalf_late_loses_compare_and_set(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1", status="expired")
    result = cli.answer_on_behalf("q1", "yes", confirm=True)
    assert result.ok is False  # already closed -> ignored
    assert _status(db_path, "q1") == "expired"


def test_force_resume_confirmed_supersedes_and_records_intent(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1", status="answered")
    result = cli.force_resume("q1", confirm=True)
    assert result.ok is True
    assert result.detail["resume_requested"] is True
    assert result.detail["superseded"] is True
    assert _status(db_path, "q1") == "superseded"
    outcome = [
        r
        for r in cli.audit_log.read_all()
        if r["action"] == "force-resume" and r["phase"] == "outcome"
    ][0]
    assert outcome["detail"]["resume_requested"] is True
    assert outcome["thread_id"] == "thread-1"


def test_force_resume_with_no_open_question_still_records_intent(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1", status="expired")
    result = cli.force_resume("q1", confirm=True)
    assert result.ok is True
    assert result.detail["superseded"] is False
    assert result.detail["resume_requested"] is True


# --------------------------------------------------------------------------- #
# redeliver (non-destructive, still audited)
# --------------------------------------------------------------------------- #


def test_redeliver_clears_channel_ref_and_audits(
    cli: OperatorCli, db_path: Path
) -> None:
    _insert_open_question(db_path, "q1", channel_ref="slack-ts-99")
    result = cli.redeliver("q1")
    assert result.ok is True
    assert result.detail["prior_channel_ref"] == "slack-ts-99"
    conn = connect(db_path)
    try:
        ref = conn.execute(
            "SELECT channel_ref FROM pending_questions WHERE question_id='q1'"
        ).fetchone()["channel_ref"]
    finally:
        conn.close()
    assert ref is None
    actions = [r["action"] for r in cli.audit_log.read_all()]
    assert actions.count("redeliver") == 2  # attempt + outcome


def test_redeliver_non_open_is_noop(cli: OperatorCli, db_path: Path) -> None:
    _insert_open_question(db_path, "q1", status="answered")
    result = cli.redeliver("q1")
    assert result.ok is False


def test_redeliver_does_not_require_confirm(cli: OperatorCli, db_path: Path) -> None:
    # redeliver is NOT in the destructive set, so no confirm needed
    assert "redeliver" not in DESTRUCTIVE_ACTIONS
    _insert_open_question(db_path, "q1")
    cli.redeliver("q1")  # must not raise


# --------------------------------------------------------------------------- #
# listing (read-only)
# --------------------------------------------------------------------------- #


def test_list_questions_defaults_and_filters(cli: OperatorCli, db_path: Path) -> None:
    _insert_open_question(db_path, "q1", thread_id="t1", status="open")
    _insert_open_question(db_path, "q2", thread_id="t2", status="expired")
    _insert_open_question(db_path, "q3", thread_id="t1", status="open")

    open_rows = cli.list_questions(statuses=["open"])
    assert {r["question_id"] for r in open_rows} == {"q1", "q3"}

    t1_rows = cli.list_questions(statuses=["open", "expired"], thread_id="t1")
    assert {r["question_id"] for r in t1_rows} == {"q1", "q3"}

    all_rows = cli.list_questions()
    assert len(all_rows) == 3


def test_list_is_not_audited_as_mutation(cli: OperatorCli, db_path: Path) -> None:
    _insert_open_question(db_path, "q1")
    cli.list_questions(statuses=["open"])
    # listing does not write audit rows
    assert cli.audit_log.read_all() == []


# --------------------------------------------------------------------------- #
# missing question
# --------------------------------------------------------------------------- #


def test_force_expire_missing_question_raises_after_attempt_logged(
    cli: OperatorCli,
) -> None:
    with pytest.raises(QuestionNotFound):
        cli.force_expire("nope", confirm=True)
    # the attempt is logged even though the question doesn't exist
    attempts = [r for r in cli.audit_log.read_all() if r["phase"] == "attempt"]
    assert any(r["question_id"] == "nope" for r in attempts)


# --------------------------------------------------------------------------- #
# argparse entrypoint (main)
# --------------------------------------------------------------------------- #


def test_build_parser_requires_db_and_audit_log() -> None:
    parser = build_parser()
    with pytest.raises(SystemExit):
        parser.parse_args(["list"])  # missing --db/--audit-log


def test_main_list_returns_zero(db_path: Path, audit_path: Path, capsys) -> None:
    _insert_open_question(db_path, "q1")
    rc = main(
        [
            "--db",
            str(db_path),
            "--audit-log",
            str(audit_path),
            "list",
            "--status",
            "open",
        ]
    )
    assert rc == 0
    out = json.loads(capsys.readouterr().out)
    assert out[0]["question_id"] == "q1"


def test_main_force_expire_requires_confirm_flag(
    db_path: Path, audit_path: Path, capsys
) -> None:
    _insert_open_question(db_path, "q1")
    rc = main(
        ["--db", str(db_path), "--audit-log", str(audit_path), "force-expire", "q1"]
    )
    assert rc == 2  # ConfirmationRequired exit code
    assert "refused" in capsys.readouterr().err
    assert _status(db_path, "q1") == "open"


def test_main_force_expire_with_confirm(db_path: Path, audit_path: Path) -> None:
    _insert_open_question(db_path, "q1")
    rc = main(
        [
            "--db",
            str(db_path),
            "--audit-log",
            str(audit_path),
            "--actor",
            "adam",
            "force-expire",
            "q1",
            "--confirm",
        ]
    )
    assert rc == 0
    assert _status(db_path, "q1") == "expired"
    records = AuditLog(audit_path).read_all()
    assert any(r["actor"] == "adam" for r in records)


def test_main_answer_on_behalf_parses_json_answer(
    db_path: Path, audit_path: Path
) -> None:
    _insert_open_question(db_path, "q1")
    rc = main(
        [
            "--db",
            str(db_path),
            "--audit-log",
            str(audit_path),
            "answer-on-behalf",
            "q1",
            '{"approve": true}',
            "--confirm",
        ]
    )
    assert rc == 0
    conn = connect(db_path)
    try:
        answer_json = conn.execute(
            "SELECT answer_json FROM pending_questions WHERE question_id='q1'"
        ).fetchone()["answer_json"]
    finally:
        conn.close()
    assert json.loads(answer_json) == {"approve": True}


def test_main_answer_on_behalf_raw_string_answer(
    db_path: Path, audit_path: Path
) -> None:
    _insert_open_question(db_path, "q1")
    rc = main(
        [
            "--db",
            str(db_path),
            "--audit-log",
            str(audit_path),
            "answer-on-behalf",
            "q1",
            "approve",
            "--confirm",
        ]
    )
    assert rc == 0
    conn = connect(db_path)
    try:
        answer_json = conn.execute(
            "SELECT answer_json FROM pending_questions WHERE question_id='q1'"
        ).fetchone()["answer_json"]
    finally:
        conn.close()
    assert json.loads(answer_json) == "approve"


def test_main_missing_question_returns_error_code(
    db_path: Path, audit_path: Path, capsys
) -> None:
    rc = main(
        [
            "--db",
            str(db_path),
            "--audit-log",
            str(audit_path),
            "force-resume",
            "ghost",
            "--confirm",
        ]
    )
    assert rc == 3
    assert "error" in capsys.readouterr().err


def test_main_redeliver_returns_one_when_noop(db_path: Path, audit_path: Path) -> None:
    _insert_open_question(db_path, "q1", status="answered")
    rc = main(["--db", str(db_path), "--audit-log", str(audit_path), "redeliver", "q1"])
    assert rc == 1  # not open -> ok=False -> exit 1