adam/orchestrator
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
This repository has been archived on 2026-08-04. You can view files and clone it, but cannot push or open issues or pull requests.
orchestrator/agent-team/tests/test_operator_cli.py
Adam Moussa a6275b4000
fix(agent-team): harden listener respawn/close, broaden handle_event guard, channel_ref partial-unique (#29) 
Three robustness/hardening fixes surfaced by /sh-security-review on the
agent-team listener/coordinator surface. None alter AUTHZ-01 allowlist
behavior or the first-answer-wins compare-and-set semantics.

1. Respawn close() leak (CWE-772). The watchdog _supervise_slack_listener
   respawned the inbound Slack listener without tearing down the dead one,
   leaking a Socket Mode WebSocket / SDK thread set per flap. Now the dead
   listener is closed before respawn (new _close_dead_listener, idempotent),
   AND _run_listener has a finally that always closes the listener so a
   crashed serve() releases its socket. SlackListener.close() is idempotent,
   so the belt-and-braces close stays a safe no-op.

2. Broadened exception guard in handle_event (CWE-248). The submit block
   only caught ValueError; the accept path (submit_answer ->
   _question_turn/_question_thread) can raise KeyError on a concurrently
   mutated row, and the CAS can raise sqlite3.Error. An uncaught exception
   would escape into the Bolt dispatch. Added a separate `except Exception`
   that logs at WARNING (not silent, not debug) and returns None. The
   existing ValueError-as-debug behavior is unchanged; authorization still
   runs first, so the trust boundary is not widened.

3. channel_ref partial-unique index (defense-in-depth). Added
   uq_pending_questions_open_channel_ref — a PARTIAL UNIQUE index on
   (channel_ref) WHERE channel_ref IS NOT NULL AND status='open' — so two
   OPEN rows can never share a non-null channel_ref (a thread_ts can never
   map to two open questions). Installed in init_db AND unconditionally in
   migrate (idempotent IF NOT EXISTS) so existing v1 DBs gain it. NULLs and
   closed rows are excluded; mirrored verbatim into schema.sql.

Tests: +8 (was 960, now 968). New: schema partial-unique reject/null/closed/
migrate cases; handle_event KeyError + sqlite3.Error swallow cases;
coordinator close-before-respawn + run_listener-closes-on-crash. Fixed the
operator-cli test fixture to use a per-question channel_ref (it previously
inserted multiple open rows sharing one ref, which the new index correctly
rejects).
2026-06-22 16:14:22 -04:00

519 lines
16 KiB
Python
Raw Blame History

"""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 = None,
) -> None:
conn = connect(db_path)
# Default to a per-question channel_ref so two OPEN rows never collide on the
# partial unique index uq_pending_questions_open_channel_ref (production never
# assigns the same channel_ref to two open questions — each gets its own
# posted message ts). Tests that need a specific ref still pass one.
if channel_ref is None:
channel_ref = f"slack-ts-{qid}"
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
Reference in a new issue View git blame Copy permalink