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.
574 lines
18 KiB
Python
574 lines
18 KiB
Python
"""Unit tests for agent_team.responder (§3.3, §3.3.1).
|
|
|
|
Covers the notify+resume seam end to end against the real ``pending_questions``
|
|
ledger (an on-disk SQLite DB via the foundation ``init_db``/``connect``):
|
|
|
|
* notify: ledger-row-first ordering, channel_ref persisted, lost-post leaves an
|
|
open row with no ref;
|
|
* submit_answer: first-answer-wins accept + enqueue, duplicate/late no-op,
|
|
answer-after-expiry loses the compare-and-set;
|
|
* ResumeWorker: turn-guard skip → superseded, happy-path resume, single-flight
|
|
per-thread serialization, concurrent different threads;
|
|
* deadline_sweep: overdue open → expired, race vs answer;
|
|
* recover_open_questions: answered rows re-enqueued.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from agent_team.db.schema import (
|
|
answer_question,
|
|
connect,
|
|
expire_question,
|
|
init_db,
|
|
)
|
|
from agent_team.responder import (
|
|
AnswerOutcome,
|
|
GraphHandle,
|
|
ResumeJob,
|
|
ResumeWorker,
|
|
deadline_sweep,
|
|
notify_question,
|
|
recover_open_questions,
|
|
submit_answer,
|
|
)
|
|
from agent_team.transport.base import NormalizedAnswer, QuestionSet, Transport
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures + fakes.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def conn(tmp_path: Path) -> sqlite3.Connection:
|
|
"""A real ledger-backed connection (foundation schema)."""
|
|
db_path = tmp_path / "agent-team.db"
|
|
init_db(db_path)
|
|
connection = connect(db_path)
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
class FakeTransport(Transport):
|
|
"""In-memory transport recording posts and parsing dict answers.
|
|
|
|
``post_question`` embeds the ``question_id`` in the returned ref (Slack
|
|
``ts`` analogue). ``post_fails`` toggles the lost-post path.
|
|
"""
|
|
|
|
def __init__(self, *, post_fails: bool = False) -> None:
|
|
self.posts: list[dict[str, Any]] = []
|
|
self.post_fails = post_fails
|
|
|
|
def post_question(
|
|
self, *, thread_id, question_id, turn, question_set, deadline
|
|
) -> str:
|
|
if self.post_fails:
|
|
raise RuntimeError("transport unreachable")
|
|
ref = f"slack-ts-{question_id}"
|
|
self.posts.append(
|
|
{
|
|
"thread_id": thread_id,
|
|
"question_id": question_id,
|
|
"turn": turn,
|
|
"deadline": deadline,
|
|
"ref": ref,
|
|
}
|
|
)
|
|
return ref
|
|
|
|
def parse_answer(self, raw) -> tuple[str, Any, str]:
|
|
na = NormalizedAnswer(
|
|
question_id=raw["callback_id"], answer=raw["value"], via="slack"
|
|
)
|
|
return na.question_id, na.answer, na.via
|
|
|
|
|
|
class FakeGraph:
|
|
"""Structural ``GraphHandle``: configurable interrupt turn + resume recorder."""
|
|
|
|
def __init__(self, turns: dict[str, int | None] | None = None) -> None:
|
|
# thread_id -> turn it is interrupted on (None = not interrupted).
|
|
self.turns: dict[str, int | None] = turns or {}
|
|
self.resumed: list[tuple[str, Any]] = []
|
|
self._resume_hook = None
|
|
|
|
def interrupted_turn(self, thread_id: str) -> int | None:
|
|
return self.turns.get(thread_id)
|
|
|
|
def resume(self, thread_id: str, answer: Any) -> Any:
|
|
if self._resume_hook is not None:
|
|
self._resume_hook(thread_id)
|
|
self.resumed.append((thread_id, answer))
|
|
return {"resumed": thread_id}
|
|
|
|
|
|
def _question_set(
|
|
*, thread_id: str = "t1", question_id: str = "q1", turn: int = 0
|
|
) -> QuestionSet:
|
|
return QuestionSet(
|
|
thread_id=thread_id,
|
|
question_id=question_id,
|
|
turn=turn,
|
|
questions=["proceed?"],
|
|
context={"repo": "x"},
|
|
)
|
|
|
|
|
|
def _row(conn: sqlite3.Connection, qid: str) -> sqlite3.Row:
|
|
return conn.execute(
|
|
"SELECT * FROM pending_questions WHERE question_id=?", (qid,)
|
|
).fetchone()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Protocol conformance.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_fake_graph_satisfies_protocol() -> None:
|
|
assert isinstance(FakeGraph(), GraphHandle)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# notify_question — delivery + lost-post (§3.3.1).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_notify_writes_open_row_then_stores_ref(conn: sqlite3.Connection) -> None:
|
|
transport = FakeTransport()
|
|
qs = _question_set()
|
|
|
|
ref = notify_question(conn, transport, qs, deadline="2026-06-18T00:00:00+00:00")
|
|
|
|
assert ref == "slack-ts-q1"
|
|
row = _row(conn, "q1")
|
|
assert row["status"] == "open"
|
|
assert row["channel_ref"] == "slack-ts-q1"
|
|
assert row["thread_id"] == "t1"
|
|
assert row["turn"] == 0
|
|
assert row["transport"] == "FakeTransport"
|
|
assert row["deadline_at"] == "2026-06-18T00:00:00+00:00"
|
|
assert row["posted_at"] is not None
|
|
# The post carried the question_id so an answer can map back.
|
|
assert transport.posts[0]["question_id"] == "q1"
|
|
|
|
|
|
def test_notify_row_is_written_before_post(conn: sqlite3.Connection) -> None:
|
|
"""The durable row must exist even while the post is in flight."""
|
|
seen: dict[str, Any] = {}
|
|
|
|
class CheckingTransport(FakeTransport):
|
|
def post_question(self, *, question_id, **kw): # type: ignore[override]
|
|
# At post time the ledger row must already be persisted as open.
|
|
seen["row"] = _row(conn, question_id)
|
|
return super().post_question(question_id=question_id, **kw)
|
|
|
|
notify_question(
|
|
conn, CheckingTransport(), _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
assert seen["row"] is not None
|
|
assert seen["row"]["status"] == "open"
|
|
|
|
|
|
def test_notify_lost_post_leaves_open_row_without_ref(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
transport = FakeTransport(post_fails=True)
|
|
|
|
ref = notify_question(
|
|
conn, transport, _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
|
|
# Post failed: no ref returned, row stays open with no ref for reconcile.
|
|
assert ref is None
|
|
row = _row(conn, "q1")
|
|
assert row["status"] == "open"
|
|
assert row["channel_ref"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# submit_answer — first-answer-wins (§3.3.1).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_submit_answer_first_wins_enqueues_resume(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
transport = FakeTransport()
|
|
notify_question(
|
|
conn, transport, _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
|
|
enqueued: list[ResumeJob] = []
|
|
outcome = submit_answer(
|
|
conn,
|
|
transport,
|
|
{"callback_id": "q1", "value": "yes"},
|
|
enqueue_resume=enqueued.append,
|
|
)
|
|
|
|
assert isinstance(outcome, AnswerOutcome)
|
|
assert outcome.accepted is True
|
|
assert outcome.question_id == "q1"
|
|
assert outcome.via == "slack"
|
|
assert outcome.job is not None
|
|
assert outcome.job == ResumeJob(
|
|
thread_id="t1", question_id="q1", turn=0, answer="yes"
|
|
)
|
|
assert enqueued == [outcome.job]
|
|
|
|
row = _row(conn, "q1")
|
|
assert row["status"] == "answered"
|
|
assert row["answer_json"] == '"yes"'
|
|
assert row["answered_via"] == "slack"
|
|
assert row["answered_at"] is not None
|
|
|
|
|
|
def test_submit_answer_duplicate_is_noop(conn: sqlite3.Connection) -> None:
|
|
transport = FakeTransport()
|
|
notify_question(
|
|
conn, transport, _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
|
|
enqueued: list[ResumeJob] = []
|
|
first = submit_answer(
|
|
conn,
|
|
transport,
|
|
{"callback_id": "q1", "value": "yes"},
|
|
enqueue_resume=enqueued.append,
|
|
)
|
|
second = submit_answer(
|
|
conn,
|
|
transport,
|
|
{"callback_id": "q1", "value": "no"},
|
|
enqueue_resume=enqueued.append,
|
|
)
|
|
|
|
assert first.accepted is True
|
|
assert second.accepted is False
|
|
assert second.job is None
|
|
# Only the first answer enqueued a resume; the duplicate is ignored.
|
|
assert len(enqueued) == 1
|
|
# The stored answer is the first one, never overwritten by the duplicate.
|
|
assert _row(conn, "q1")["answer_json"] == '"yes"'
|
|
|
|
|
|
def test_submit_answer_after_expiry_loses_race(conn: sqlite3.Connection) -> None:
|
|
transport = FakeTransport()
|
|
notify_question(
|
|
conn, transport, _question_set(), deadline="2000-01-01T00:00:00+00:00"
|
|
)
|
|
# Question times out first.
|
|
assert expire_question(conn, question_id="q1") is True
|
|
|
|
enqueued: list[ResumeJob] = []
|
|
outcome = submit_answer(
|
|
conn,
|
|
transport,
|
|
{"callback_id": "q1", "value": "yes"},
|
|
enqueue_resume=enqueued.append,
|
|
)
|
|
|
|
assert outcome.accepted is False
|
|
assert enqueued == []
|
|
assert _row(conn, "q1")["status"] == "expired"
|
|
|
|
|
|
def test_submit_answer_unknown_question_is_noop(conn: sqlite3.Connection) -> None:
|
|
transport = FakeTransport()
|
|
enqueued: list[ResumeJob] = []
|
|
outcome = submit_answer(
|
|
conn,
|
|
transport,
|
|
{"callback_id": "nope", "value": "x"},
|
|
enqueue_resume=enqueued.append,
|
|
)
|
|
assert outcome.accepted is False
|
|
assert enqueued == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResumeWorker — single-flight, turn-guarded (§3.3.1).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_resume_happy_path(conn: sqlite3.Connection) -> None:
|
|
notify_question(
|
|
conn, FakeTransport(), _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
answer_question(conn, question_id="q1", answer_json='"yes"', answered_via="slack")
|
|
|
|
graph = FakeGraph(turns={"t1": 0})
|
|
worker = ResumeWorker(conn, graph)
|
|
job = ResumeJob(thread_id="t1", question_id="q1", turn=0, answer="yes")
|
|
|
|
assert worker.run(job) is True
|
|
assert graph.resumed == [("t1", "yes")]
|
|
# Still 'answered' — the worker does not mutate the ledger on success.
|
|
assert _row(conn, "q1")["status"] == "answered"
|
|
|
|
|
|
def test_resume_stale_turn_supersedes_and_skips(conn: sqlite3.Connection) -> None:
|
|
notify_question(
|
|
conn,
|
|
FakeTransport(),
|
|
_question_set(turn=2),
|
|
deadline="2026-06-18T00:00:00+00:00",
|
|
)
|
|
answer_question(conn, question_id="q1", answer_json='"yes"', answered_via="slack")
|
|
|
|
# Graph already advanced to turn 3 (or is not on turn 2 any more).
|
|
graph = FakeGraph(turns={"t1": 3})
|
|
worker = ResumeWorker(conn, graph)
|
|
job = ResumeJob(thread_id="t1", question_id="q1", turn=2, answer="yes")
|
|
|
|
assert worker.run(job) is False
|
|
assert graph.resumed == []
|
|
assert _row(conn, "q1")["status"] == "superseded"
|
|
|
|
|
|
def test_resume_not_interrupted_supersedes_and_skips(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
notify_question(
|
|
conn, FakeTransport(), _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
answer_question(conn, question_id="q1", answer_json='"yes"', answered_via="slack")
|
|
|
|
# Thread not currently interrupted (None).
|
|
graph = FakeGraph(turns={"t1": None})
|
|
worker = ResumeWorker(conn, graph)
|
|
job = ResumeJob(thread_id="t1", question_id="q1", turn=0, answer="yes")
|
|
|
|
assert worker.run(job) is False
|
|
assert graph.resumed == []
|
|
assert _row(conn, "q1")["status"] == "superseded"
|
|
|
|
|
|
def test_resume_redelivered_job_does_not_double_apply(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
"""A second (redelivered) job for the same answered turn supersedes-skips."""
|
|
notify_question(
|
|
conn, FakeTransport(), _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
answer_question(conn, question_id="q1", answer_json='"yes"', answered_via="slack")
|
|
|
|
graph = FakeGraph(turns={"t1": 0})
|
|
worker = ResumeWorker(conn, graph)
|
|
job = ResumeJob(thread_id="t1", question_id="q1", turn=0, answer="yes")
|
|
|
|
assert worker.run(job) is True
|
|
|
|
# The real graph advances after a successful resume; model that.
|
|
graph.turns["t1"] = 1
|
|
assert worker.run(job) is False
|
|
# Resume applied exactly once.
|
|
assert graph.resumed == [("t1", "yes")]
|
|
|
|
|
|
def test_resume_single_flight_serializes_same_thread(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
"""Two jobs for one thread never resume concurrently (per-thread lock)."""
|
|
notify_question(
|
|
conn, FakeTransport(), _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
answer_question(conn, question_id="q1", answer_json='"yes"', answered_via="slack")
|
|
|
|
graph = FakeGraph(turns={"t1": 0})
|
|
concurrency = {"current": 0, "max": 0}
|
|
lock = threading.Lock()
|
|
|
|
def hook(_thread_id: str) -> None:
|
|
with lock:
|
|
concurrency["current"] += 1
|
|
concurrency["max"] = max(concurrency["max"], concurrency["current"])
|
|
time.sleep(0.02)
|
|
with lock:
|
|
concurrency["current"] -= 1
|
|
|
|
graph._resume_hook = hook
|
|
worker = ResumeWorker(conn, graph)
|
|
job = ResumeJob(thread_id="t1", question_id="q1", turn=0, answer="yes")
|
|
|
|
threads = [threading.Thread(target=worker.run, args=(job,)) for _ in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Same thread_id => never more than one in-flight resume at a time.
|
|
assert concurrency["max"] == 1
|
|
|
|
|
|
def test_resume_different_threads_run_concurrently(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
"""Different thread_ids are NOT serialized against each other."""
|
|
graph = FakeGraph(turns={f"t{i}": 0 for i in range(4)})
|
|
barrier = threading.Barrier(4, timeout=2.0)
|
|
reached = {"ok": True}
|
|
|
|
def hook(_thread_id: str) -> None:
|
|
try:
|
|
barrier.wait()
|
|
except threading.BrokenBarrierError:
|
|
reached["ok"] = False
|
|
|
|
graph._resume_hook = hook
|
|
worker = ResumeWorker(conn, graph)
|
|
jobs = [
|
|
ResumeJob(thread_id=f"t{i}", question_id=f"q{i}", turn=0, answer="y")
|
|
for i in range(4)
|
|
]
|
|
|
|
threads = [threading.Thread(target=worker.run, args=(j,)) for j in jobs]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# All four reached the barrier together => they ran concurrently.
|
|
assert reached["ok"] is True
|
|
assert len(graph.resumed) == 4
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# deadline_sweep — overdue open -> expired (§3.3.1).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_deadline_sweep_expires_only_overdue_open(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
transport = FakeTransport()
|
|
notify_question(
|
|
conn,
|
|
transport,
|
|
_question_set(question_id="overdue"),
|
|
deadline="2000-01-01T00:00:00+00:00",
|
|
)
|
|
notify_question(
|
|
conn,
|
|
transport,
|
|
_question_set(question_id="future"),
|
|
deadline="2099-01-01T00:00:00+00:00",
|
|
)
|
|
|
|
expired = deadline_sweep(conn, now="2026-06-17T00:00:00+00:00")
|
|
|
|
assert expired == ["overdue"]
|
|
assert _row(conn, "overdue")["status"] == "expired"
|
|
assert _row(conn, "future")["status"] == "open"
|
|
|
|
|
|
def test_deadline_sweep_skips_already_answered(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
transport = FakeTransport()
|
|
notify_question(
|
|
conn,
|
|
transport,
|
|
_question_set(question_id="ans"),
|
|
deadline="2000-01-01T00:00:00+00:00",
|
|
)
|
|
answer_question(conn, question_id="ans", answer_json='"yes"', answered_via="slack")
|
|
|
|
expired = deadline_sweep(conn, now="2026-06-17T00:00:00+00:00")
|
|
|
|
# Already answered => the deadline race was already lost; not expired.
|
|
assert expired == []
|
|
assert _row(conn, "ans")["status"] == "answered"
|
|
|
|
|
|
def test_deadline_sweep_ignores_null_deadline(conn: sqlite3.Connection) -> None:
|
|
conn.execute(
|
|
"INSERT INTO pending_questions "
|
|
"(question_id, thread_id, turn, status, transport) "
|
|
"VALUES ('q', 't', 0, 'open', 'slack')"
|
|
)
|
|
assert deadline_sweep(conn, now="2099-01-01T00:00:00+00:00") == []
|
|
assert _row(conn, "q")["status"] == "open"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# recover_open_questions — startup answered->resume replay (§3.3.1).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_recover_reenqueues_answered_rows(conn: sqlite3.Connection) -> None:
|
|
transport = FakeTransport()
|
|
notify_question(
|
|
conn,
|
|
transport,
|
|
_question_set(thread_id="ta", question_id="qa"),
|
|
deadline="2026-06-18T00:00:00+00:00",
|
|
)
|
|
notify_question(
|
|
conn,
|
|
transport,
|
|
_question_set(thread_id="tb", question_id="qb", turn=1),
|
|
deadline="2026-06-18T00:00:00+00:00",
|
|
)
|
|
notify_question(
|
|
conn,
|
|
transport,
|
|
_question_set(thread_id="tc", question_id="qc"),
|
|
deadline="2026-06-18T00:00:00+00:00",
|
|
)
|
|
|
|
# qa, qb answered before a crash; qc still open.
|
|
answer_question(conn, question_id="qa", answer_json='"yes"', answered_via="slack")
|
|
answer_question(
|
|
conn, question_id="qb", answer_json='{"k": 1}', answered_via="github"
|
|
)
|
|
|
|
enqueued: list[ResumeJob] = []
|
|
jobs = recover_open_questions(conn, enqueue_resume=enqueued.append)
|
|
|
|
assert jobs == enqueued
|
|
by_thread = {j.thread_id: j for j in jobs}
|
|
assert set(by_thread) == {"ta", "tb"}
|
|
assert by_thread["ta"] == ResumeJob(
|
|
thread_id="ta", question_id="qa", turn=0, answer="yes"
|
|
)
|
|
assert by_thread["tb"] == ResumeJob(
|
|
thread_id="tb", question_id="qb", turn=1, answer={"k": 1}
|
|
)
|
|
|
|
|
|
def test_recover_is_idempotent_via_turn_guard(conn: sqlite3.Connection) -> None:
|
|
"""Re-enqueued recover jobs no-op when the graph already advanced."""
|
|
notify_question(
|
|
conn, FakeTransport(), _question_set(), deadline="2026-06-18T00:00:00+00:00"
|
|
)
|
|
answer_question(conn, question_id="q1", answer_json='"yes"', answered_via="slack")
|
|
|
|
enqueued: list[ResumeJob] = []
|
|
recover_open_questions(conn, enqueue_resume=enqueued.append)
|
|
assert len(enqueued) == 1
|
|
|
|
# Graph already past turn 0 (resume happened before the crash record cleared).
|
|
graph = FakeGraph(turns={"t1": 1})
|
|
worker = ResumeWorker(conn, graph)
|
|
assert worker.run(enqueued[0]) is False
|
|
assert graph.resumed == []
|
|
assert _row(conn, "q1")["status"] == "superseded"
|
|
|
|
|
|
def test_recover_empty_ledger(conn: sqlite3.Connection) -> None:
|
|
enqueued: list[ResumeJob] = []
|
|
assert recover_open_questions(conn, enqueue_resume=enqueued.append) == []
|
|
assert enqueued == []
|