orchestrator/agent-team/tests/test_run_team.py

"""Unit tests for the ``run-team.py`` operator CLI (design §3.3.1, §7.1 P1).

``run-team.py`` is a hyphenated entry script (per the design's "entry CLI
``run-team.py``"), so it cannot be imported by normal ``import`` syntax. These
tests load it via :mod:`importlib` from its file path and exercise the manual
ledger path against the FOUNDATION ``agent_team.db.schema`` ledger.

The tests assert the §3.3.1 manual-path contract: list open/parked questions,
answer-on-behalf / force-expire / supersede gated behind ``--confirm`` and
audit-logged, first-answer-wins semantics inherited from the foundation
compare-and-set, and read-only commands needing no confirmation.
"""

from __future__ import annotations

import importlib.util
import io
import json
from pathlib import Path
from types import ModuleType

import pytest

from agent_team.db.schema import QUESTION_STATES, connect, init_db

# Path to the hyphenated entry CLI (sibling of the agent_team package).
_CLI_PATH = Path(__file__).resolve().parents[1] / "run-team.py"


def _load_cli() -> ModuleType:
    """Import ``run-team.py`` from its file path as a module."""
    spec = importlib.util.spec_from_file_location("run_team_cli", _CLI_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


@pytest.fixture(scope="module")
def cli() -> ModuleType:
    """The loaded run-team CLI module (loaded once per test module)."""
    return _load_cli()


@pytest.fixture()
def db_path(tmp_path: Path) -> Path:
    """A fresh, initialized ledger DB for each test."""
    path = tmp_path / "state" / "agent_team.sqlite"
    init_db(path)
    return path


@pytest.fixture()
def audit_log(tmp_path: Path) -> Path:
    """Path to a per-test audit log (not created until first destructive op)."""
    return tmp_path / "state" / "audit.log.jsonl"


def _insert_question(
    db_path: Path,
    *,
    question_id: str,
    thread_id: str = "thread-a",
    turn: int = 0,
    status: str = "open",
    transport: str = "slack",
    deadline_at: str | None = None,
) -> None:
    """Insert a pending_questions row directly for test setup."""
    conn = connect(db_path)
    try:
        conn.execute(
            "INSERT INTO pending_questions "
            "(question_id, thread_id, turn, status, transport, posted_at, "
            "deadline_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                question_id,
                thread_id,
                turn,
                status,
                transport,
                "2026-06-17T00:00:00+00:00",
                deadline_at,
            ),
        )
    finally:
        conn.close()


def _status_of(db_path: Path, question_id: str) -> str | None:
    conn = connect(db_path)
    try:
        row = conn.execute(
            "SELECT status FROM pending_questions WHERE question_id = ?",
            (question_id,),
        ).fetchone()
    finally:
        conn.close()
    return None if row is None else row["status"]


def _run(
    cli: ModuleType,
    db_path: Path,
    audit_log: Path,
    *args: str,
) -> tuple[int, str]:
    """Invoke ``main`` with the standard global flags, capturing stdout."""
    out = io.StringIO()
    argv = ["--db", str(db_path), "--audit-log", str(audit_log), *args]
    code = cli.main(argv, out=out)
    return code, out.getvalue()


# --------------------------------------------------------------------------- #
# Foundation-import / structural assertions
# --------------------------------------------------------------------------- #


def test_cli_file_exists_and_is_hyphenated() -> None:
    assert _CLI_PATH.name == "run-team.py"
    assert _CLI_PATH.is_file()


def test_cli_imports_foundation_contracts_verbatim(cli: ModuleType) -> None:
    # The CLI must import the foundation, not redefine it.
    from agent_team.db import schema as foundation_schema

    assert cli.answer_question is foundation_schema.answer_question
    assert cli.expire_question is foundation_schema.expire_question
    assert cli.supersede_question is foundation_schema.supersede_question
    assert cli.connect is foundation_schema.connect
    assert cli.init_db is foundation_schema.init_db


def test_build_parser_has_no_side_effects(cli: ModuleType) -> None:
    parser = cli.build_parser()
    assert parser.prog == "run-team.py"


# --------------------------------------------------------------------------- #
# init-db
# --------------------------------------------------------------------------- #


def test_init_db_creates_ledger_tables(cli: ModuleType, tmp_path: Path) -> None:
    db_path = tmp_path / "state" / "fresh.sqlite"
    audit_log = tmp_path / "audit.jsonl"
    code, out = _run(cli, db_path, audit_log, "init-db")
    assert code == 0
    assert db_path.exists()
    conn = connect(db_path)
    try:
        names = {
            r["name"]
            for r in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            ).fetchall()
        }
    finally:
        conn.close()
    assert "pending_questions" in names
    assert "budget_ledger" in names


def test_init_db_is_idempotent(cli: ModuleType, tmp_path: Path) -> None:
    db_path = tmp_path / "state" / "fresh.sqlite"
    audit_log = tmp_path / "audit.jsonl"
    assert _run(cli, db_path, audit_log, "init-db")[0] == 0
    assert _run(cli, db_path, audit_log, "init-db")[0] == 0


# --------------------------------------------------------------------------- #
# list / show (read-only, no confirmation)
# --------------------------------------------------------------------------- #


def test_list_open_default(cli: ModuleType, db_path: Path, audit_log: Path) -> None:
    _insert_question(db_path, question_id="q-open", status="open")
    _insert_question(db_path, question_id="q-exp", status="expired")
    code, out = _run(cli, db_path, audit_log, "list")
    assert code == 0
    payload = json.loads(out)
    ids = {row["question_id"] for row in payload}
    assert ids == {"q-open"}


def test_list_all(cli: ModuleType, db_path: Path, audit_log: Path) -> None:
    _insert_question(db_path, question_id="q-open", status="open")
    _insert_question(db_path, question_id="q-exp", status="expired")
    code, out = _run(cli, db_path, audit_log, "list", "--all")
    assert code == 0
    ids = {row["question_id"] for row in json.loads(out)}
    assert ids == {"q-open", "q-exp"}


def test_list_parked_excludes_open(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q-open", status="open")
    _insert_question(db_path, question_id="q-ans", status="answered")
    _insert_question(db_path, question_id="q-exp", status="expired")
    _insert_question(db_path, question_id="q-sup", status="superseded")
    code, out = _run(cli, db_path, audit_log, "list", "--parked")
    assert code == 0
    ids = {row["question_id"] for row in json.loads(out)}
    assert ids == {"q-ans", "q-exp", "q-sup"}
    assert "q-open" not in ids


def test_list_status_filter(cli: ModuleType, db_path: Path, audit_log: Path) -> None:
    _insert_question(db_path, question_id="q-open", status="open")
    _insert_question(db_path, question_id="q-exp", status="expired")
    code, out = _run(cli, db_path, audit_log, "list", "--status", "expired")
    assert code == 0
    ids = {row["question_id"] for row in json.loads(out)}
    assert ids == {"q-exp"}


def test_list_empty_returns_empty_array(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    code, out = _run(cli, db_path, audit_log, "list")
    assert code == 0
    assert json.loads(out) == []


def test_show_existing(cli: ModuleType, db_path: Path, audit_log: Path) -> None:
    _insert_question(db_path, question_id="q1", thread_id="t1", turn=3)
    code, out = _run(cli, db_path, audit_log, "show", "q1")
    assert code == 0
    row = json.loads(out)
    assert row["question_id"] == "q1"
    assert row["thread_id"] == "t1"
    assert row["turn"] == 3


def test_show_missing_returns_1(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    code, _ = _run(cli, db_path, audit_log, "show", "nope")
    assert code == 1


# --------------------------------------------------------------------------- #
# Destructive actions require --confirm and are audit-logged
# --------------------------------------------------------------------------- #


def test_expire_without_confirm_refuses_and_does_not_mutate(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "expire", "q1")
    assert code == 1
    # Unchanged: the guard fired before touching the ledger.
    assert _status_of(db_path, "q1") == "open"
    assert not audit_log.exists()


def test_answer_without_confirm_refuses(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "answer", "q1", "--answer", "yes")
    assert code == 1
    assert _status_of(db_path, "q1") == "open"


def test_supersede_without_confirm_refuses(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "supersede", "q1")
    assert code == 1
    assert _status_of(db_path, "q1") == "open"


def test_expire_with_confirm_flips_status_and_audits(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, out = _run(
        cli, db_path, audit_log, "--operator", "adam", "expire", "q1", "--confirm"
    )
    assert code == 0
    assert _status_of(db_path, "q1") == "expired"
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    # Attempt is recorded BEFORE the mutation, outcome after, so a mutation can
    # never land without a trail (§3.3.1).
    assert len(entries) == 2
    assert entries[0]["phase"] == "attempt"
    assert "applied" not in entries[0]
    assert entries[-1]["phase"] == "outcome"
    assert entries[-1]["action"] == "expire"
    assert entries[-1]["question_id"] == "q1"
    assert entries[-1]["operator"] == "adam"
    assert entries[-1]["applied"] is True
    assert "ts" in entries[-1]


def test_answer_with_confirm_flips_status_records_via(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, out = _run(
        cli,
        db_path,
        audit_log,
        "--operator",
        "adam",
        "answer",
        "q1",
        "--answer",
        '{"choice": "B"}',
        "--confirm",
    )
    assert code == 0
    assert _status_of(db_path, "q1") == "answered"
    conn = connect(db_path)
    try:
        row = conn.execute(
            "SELECT answer_json, answered_via FROM pending_questions "
            "WHERE question_id = ?",
            ("q1",),
        ).fetchone()
    finally:
        conn.close()
    assert row["answer_json"] == '{"choice": "B"}'
    assert row["answered_via"] == "cli:adam"


def test_answer_explicit_via_overrides_default(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(
        cli,
        db_path,
        audit_log,
        "answer",
        "q1",
        "--answer",
        "x",
        "--via",
        "slack:U123",
        "--confirm",
    )
    assert code == 0
    conn = connect(db_path)
    try:
        row = conn.execute(
            "SELECT answered_via FROM pending_questions WHERE question_id = ?",
            ("q1",),
        ).fetchone()
    finally:
        conn.close()
    assert row["answered_via"] == "slack:U123"


def test_supersede_with_confirm_flips_status(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "supersede", "q1", "--confirm")
    assert code == 0
    assert _status_of(db_path, "q1") == "superseded"


# --------------------------------------------------------------------------- #
# First-answer-wins / no-op semantics inherited from the foundation
# --------------------------------------------------------------------------- #


def test_answer_already_expired_is_noop_returns_1_but_audits(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="expired")
    code, _ = _run(
        cli, db_path, audit_log, "answer", "q1", "--answer", "x", "--confirm"
    )
    assert code == 1
    # Status unchanged (compare-and-set lost), but the attempt is audited.
    assert _status_of(db_path, "q1") == "expired"
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    assert entries[-1]["action"] == "answer"
    assert entries[-1]["applied"] is False


def test_expire_missing_question_is_noop_returns_1(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    code, _ = _run(cli, db_path, audit_log, "expire", "ghost", "--confirm")
    assert code == 1
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    assert entries[-1]["applied"] is False


def test_double_answer_second_is_noop(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    first, _ = _run(
        cli, db_path, audit_log, "answer", "q1", "--answer", "a", "--confirm"
    )
    second, _ = _run(
        cli, db_path, audit_log, "answer", "q1", "--answer", "b", "--confirm"
    )
    assert first == 0
    assert second == 1  # first-answer-wins; second is a no-op
    conn = connect(db_path)
    try:
        row = conn.execute(
            "SELECT answer_json FROM pending_questions WHERE question_id = ?",
            ("q1",),
        ).fetchone()
    finally:
        conn.close()
    assert row["answer_json"] == "a"  # original answer preserved


# --------------------------------------------------------------------------- #
# Audit log durability (append-only, multiple actions)
# --------------------------------------------------------------------------- #


def test_audit_log_appends_across_actions(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    _insert_question(db_path, question_id="q2", status="open")
    _run(cli, db_path, audit_log, "expire", "q1", "--confirm")
    _run(cli, db_path, audit_log, "answer", "q2", "--answer", "y", "--confirm")
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    # Each destructive action writes an attempt + an outcome record (append-only).
    assert len(entries) == 4
    actions = [e["action"] for e in entries]
    assert actions == ["expire", "expire", "answer", "answer"]
    outcomes = [e["action"] for e in entries if e["phase"] == "outcome"]
    assert outcomes == ["expire", "answer"]


def test_audit_entries_are_valid_json_lines(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    _run(cli, db_path, audit_log, "expire", "q1", "--confirm")
    content = audit_log.read_text()
    assert content.endswith("\n")
    for line in content.splitlines():
        json.loads(line)  # raises if any line is not valid JSON


# --------------------------------------------------------------------------- #
# argparse-level usage errors
# --------------------------------------------------------------------------- #


def test_no_subcommand_is_usage_error(cli: ModuleType) -> None:
    with pytest.raises(SystemExit) as exc:
        cli.main([])
    assert exc.value.code == 2


def test_unknown_status_choice_is_usage_error(cli: ModuleType) -> None:
    with pytest.raises(SystemExit) as exc:
        cli.main(["list", "--status", "bogus"])
    assert exc.value.code == 2


def test_answer_requires_answer_flag(cli: ModuleType) -> None:
    with pytest.raises(SystemExit) as exc:
        cli.main(["answer", "q1", "--confirm"])
    assert exc.value.code == 2


def test_parked_states_derived_from_foundation(cli: ModuleType) -> None:
    # The parked-context states are exactly the non-open foundation states.
    assert set(cli._PARKED_STATES) == set(QUESTION_STATES) - {"open"}


# --------------------------------------------------------------------------- #
# re-deliver + force-resume (design-named operator verbs, §3.3.1 / §6.6)
# --------------------------------------------------------------------------- #


def _set_channel_ref(db_path: Path, question_id: str, ref: str) -> None:
    conn = connect(db_path)
    try:
        conn.execute(
            "UPDATE pending_questions SET channel_ref = ? WHERE question_id = ?",
            (ref, question_id),
        )
    finally:
        conn.close()


def _channel_ref_of(db_path: Path, question_id: str) -> str | None:
    conn = connect(db_path)
    try:
        row = conn.execute(
            "SELECT channel_ref FROM pending_questions WHERE question_id = ?",
            (question_id,),
        ).fetchone()
    finally:
        conn.close()
    return None if row is None else row["channel_ref"]


def test_redeliver_clears_channel_ref_and_audits(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    _set_channel_ref(db_path, "q1", "slack:123.456")
    code, out = _run(cli, db_path, audit_log, "redeliver", "q1")
    assert code == 0
    assert _channel_ref_of(db_path, "q1") is None
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    # Non-destructive but audited: attempt + outcome, no --confirm needed.
    assert [e["phase"] for e in entries] == ["attempt", "outcome"]
    assert entries[-1]["action"] == "redeliver"
    assert entries[-1]["applied"] is True
    assert entries[-1]["prior_channel_ref"] == "slack:123.456"


def test_redeliver_needs_no_confirm(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    # redeliver is not in the destructive set, so it runs without --confirm.
    assert "redeliver" not in cli._DESTRUCTIVE_ACTIONS


def test_redeliver_non_open_is_noop_returns_1(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="answered")
    code, _ = _run(cli, db_path, audit_log, "redeliver", "q1")
    assert code == 1
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    assert entries[-1]["applied"] is False


def test_force_resume_reopens_expired_parked_question(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    # The parked case: an expired question is RE-OPENED so it can be answered,
    # NOT superseded (superseding would make it permanently un-resumable).
    _insert_question(db_path, question_id="q1", status="expired")
    code, out = _run(
        cli, db_path, audit_log, "--operator", "adam", "force-resume", "q1", "--confirm"
    )
    assert code == 0
    assert _status_of(db_path, "q1") == "open"  # reopened, not superseded
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    assert [e["phase"] for e in entries] == ["attempt", "outcome"]
    assert entries[-1]["action"] == "force-resume"
    assert entries[-1]["resume_requested"] is True
    assert entries[-1]["applied"] is True
    assert entries[-1]["operator"] == "adam"


def test_force_resume_does_not_supersede_answered_row(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    # Regression: force-resume must NOT flip an answered-but-unresumed row out of
    # the state the recovery sweep resumes from. It stays 'answered'.
    _insert_question(db_path, question_id="q1", status="answered")
    code, _ = _run(cli, db_path, audit_log, "force-resume", "q1", "--confirm")
    assert code == 0
    assert _status_of(db_path, "q1") == "answered"  # untouched, still resumable


def test_force_resume_on_open_is_noop(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "force-resume", "q1", "--confirm")
    assert code == 1  # an open (not parked) question has nothing to force
    assert _status_of(db_path, "q1") == "open"


def test_force_resume_without_confirm_refuses(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "force-resume", "q1")
    assert code == 1
    assert _status_of(db_path, "q1") == "open"  # unmutated
    assert not audit_log.exists()  # refused before any audit (confirm-check first)


def test_force_resume_is_in_destructive_set(cli: ModuleType) -> None:
    assert "force-resume" in cli._DESTRUCTIVE_ACTIONS


def test_operator_defaults_to_os_login_not_empty(
    cli: ModuleType, db_path: Path, audit_log: Path
) -> None:
    # AUTHZ regression: --operator defaulted to "" → non-attributable audit.
    # Omitting it must record a real (non-empty) operator identity.
    _insert_question(db_path, question_id="q1", status="open")
    code, _ = _run(cli, db_path, audit_log, "expire", "q1", "--confirm")
    assert code == 0
    entries = [json.loads(line) for line in audit_log.read_text().splitlines()]
    assert entries[-1]["operator"]  # non-empty
    assert cli._default_operator()  # helper never returns empty


# --------------------------------------------------------------------------- #
# Audit-before-mutate: an unwritable audit path aborts BEFORE the ledger mutates
# (regression: previously the row was mutated, then the audit append crashed,
# leaving a mutation with no record and an uncaught traceback)
# --------------------------------------------------------------------------- #


def test_unwritable_audit_path_aborts_before_mutation(
    cli: ModuleType, db_path: Path, tmp_path: Path
) -> None:
    _insert_question(db_path, question_id="q1", status="open")
    # Point the audit log at a path whose parent is a FILE, so the atomic write
    # of the attempt record fails with OSError before the mutation runs.
    blocker = tmp_path / "not-a-dir"
    blocker.write_text("x")
    bad_audit = blocker / "audit.jsonl"
    code, _ = _run(cli, db_path, bad_audit, "expire", "q1", "--confirm")
    assert code == 1  # clean failure, not an uncaught traceback
    assert _status_of(db_path, "q1") == "open"  # NOT mutated — no trail, no change