Adam Moussa
15a416d31a
Consolidates the 18 leaf modules from the r720-plane2-scaffold workflow onto the foundation commit. Full suite: 535 passed, 1 skipped; ruff + format clean. Built (pre-deployment scaffold only — nothing provisioned/enabled): - LangGraph pipeline graph.py (INTAKE->CLARIFY->PLAN, interrupt()/resume, checkpointer-injectable) - nodes: clarifier (98% gate), planner, review_loop (GPT-4.1), builders->candidate diff, verifier - §3.3.1 HITL: ledger ops, resume_worker, deadline_timer, recovery sweep, responder - transports: slack / github / claude_code adapters - ci_gate (pure-code pass/fail), operator_cli, run-team.py entry, P1 sim harness - ci/agent-team-apply-verify.yml (split untrusted/privileged jobs) — authored, disabled KNOWN OPEN FINDINGS (verifier/cross-review, not yet fixed — see follow-up): - builders denylist: 4 execution-proven bypasses (delete, mode-change, copy-to, out-of-scope delete) - §3.3.1 CAS: BEGIN IMMEDIATE outside try/except; shared-connection txn nesting unsafe under concurrency - operator_cli: missing re-deliver/force-resume; audit-after-mutate ordering gap - ci yaml: GPT-4.1 cross-review PASS w/ 4 FIX items (symlink path escape, etc.) - P1 sim harness models the ledger layer, not real LangGraph interrupt/resume; P1 exit criteria not yet truly proven Deploy-gated (NOT done): IAM/step-ca/Roles Anywhere/confluence-bot provisioning, /sh-security-review sign-off, live Slack/CI, rsync, live dry-runs, Adam approval.
513 lines
16 KiB
Python
513 lines
16 KiB
Python
"""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
|