test(agent-team): cover transitions, topology, dashboard API + rework status_page tests
Add test_transitions.py (recorder idempotency under replay, terminal close, fail-soft, the B4 _instrument signature-preservation guarantee, end-to-end graph drive), test_topology.py (meta coverage, tree grouping, edge classification, phase->node map), and test_dashboard.py (/api/state contract + per-node state, /api/topology, /api/task timeline+cost+partial+thread_id validation, all via TestClient, fastapi-skipif guarded). Rework test_status_page.py to the data layer (drop retired render_html/SVG tests). conftest exposes tests/ for sibling imports. Frontend: Vitest+RTL for layout, TaskList filters/selection, TaskDrawer timeline.
This commit is contained in:
parent
677f3d3c71
commit
e3137e33e6
5 changed files with 367 additions and 179 deletions
|
|
@ -11,3 +11,9 @@ from pathlib import Path
|
|||
_PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_PROJECT_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_PROJECT_ROOT))
|
||||
|
||||
# Also expose the tests/ dir so a test module can import shared helpers from a
|
||||
# sibling test module by bare name (e.g. ``from test_status_page import ...``).
|
||||
_TESTS_DIR = Path(__file__).resolve().parent
|
||||
if str(_TESTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(_TESTS_DIR))
|
||||
|
|
|
|||
118
agent-team/tests/test_dashboard.py
Normal file
118
agent-team/tests/test_dashboard.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
"""Tests for the read-only dashboard FastAPI app (``agent_team.dashboard``).
|
||||
|
||||
Exercises the three JSON endpoints against a seeded temp ledger via FastAPI's
|
||||
TestClient: /api/state (contract preserved + per-node live state), /api/topology,
|
||||
and /api/task/{id} (timeline assembly, thread_id validation, partial fallback,
|
||||
cost join). Guarded so the suite still runs where FastAPI is absent.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from agent_team.db import TransitionRecorder, connect, init_db
|
||||
from agent_team.dashboard import task_detail
|
||||
|
||||
# Reuse the seeded-ledger helper from the status-page tests (sibling module,
|
||||
# importable by bare name under pytest's default prepend import mode).
|
||||
from test_status_page import seed_ledger
|
||||
|
||||
_HAS_FASTAPI = importlib.util.find_spec("fastapi") is not None
|
||||
pytestmark = pytest.mark.skipif(_HAS_FASTAPI is False, reason="fastapi not installed")
|
||||
|
||||
|
||||
def _client(db: Path):
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from agent_team.dashboard import make_dashboard_app
|
||||
|
||||
return TestClient(make_dashboard_app(db_path=db))
|
||||
|
||||
|
||||
def _seed_budget(db: Path, thread_id: str, stage: str, usd: float) -> None:
|
||||
conn = connect(db)
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO budget_ledger (thread_id, stage, model, billing_mode, "
|
||||
"usd_cost, recorded_at, day_bucket) VALUES (?,?,?,?,?,?,?)",
|
||||
(
|
||||
thread_id,
|
||||
stage,
|
||||
"claude",
|
||||
"subscription",
|
||||
usd,
|
||||
"2026-06-23T00:00:00+00:00",
|
||||
"2026-06-23",
|
||||
),
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_api_state_contract_and_nodes(tmp_path: Path) -> None:
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
seed_ledger(db)
|
||||
r = _client(db).get("/api/state")
|
||||
assert r.status_code == 200
|
||||
payload = r.json()
|
||||
# Existing contract preserved.
|
||||
assert payload["ok"] is True
|
||||
assert {"stages", "tasks", "summary", "budget"} <= set(payload)
|
||||
# New per-node live state, keyed by topology node id.
|
||||
assert "nodes" in payload
|
||||
# thread-waiting is at clarify and awaiting human; thread-active at plan.
|
||||
assert payload["nodes"]["clarify"]["state"] == "awaiting_human"
|
||||
assert payload["nodes"]["plan"]["state"] == "active"
|
||||
|
||||
|
||||
def test_api_topology(tmp_path: Path) -> None:
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
init_db(db)
|
||||
payload = _client(db).get("/api/topology").json()
|
||||
ids = {n["id"] for n in payload["nodes"]}
|
||||
assert {"intake", "clarify", "plan", "review"} <= ids
|
||||
assert payload["edges"]
|
||||
|
||||
|
||||
def test_api_task_timeline_and_cost(tmp_path: Path) -> None:
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
seed_ledger(db)
|
||||
rec = TransitionRecorder(db)
|
||||
rec.record_entry(thread_id="thread-active", to_phase="intake", status="active")
|
||||
rec.record_entry(thread_id="thread-active", to_phase="plan", status="active")
|
||||
_seed_budget(db, "thread-active", "plan", 0.5)
|
||||
|
||||
payload = _client(db).get("/api/task/thread-active").json()
|
||||
assert payload["ok"] is True
|
||||
assert payload["partial"] is False
|
||||
steps = {s["to_phase"]: s for s in payload["timeline"]}
|
||||
assert "intake" in steps and "plan" in steps
|
||||
assert steps["plan"]["cost_usd"] == pytest.approx(0.5)
|
||||
assert payload["total_usd"] == pytest.approx(0.5)
|
||||
|
||||
|
||||
def test_api_task_partial_when_no_transitions(tmp_path: Path) -> None:
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
seed_ledger(db) # checkpoints exist, but no task_transitions rows
|
||||
payload = _client(db).get("/api/task/thread-active").json()
|
||||
assert payload["ok"] is True
|
||||
assert payload["partial"] is True
|
||||
# Best-effort single step reconstructed from the checkpoint phase.
|
||||
assert payload["timeline"]
|
||||
assert payload["timeline"][0]["note"]
|
||||
|
||||
|
||||
def test_api_task_rejects_bad_thread_id(tmp_path: Path) -> None:
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
init_db(db)
|
||||
cli = _client(db)
|
||||
assert cli.get("/api/task/has space").status_code == 400
|
||||
assert cli.get("/api/task/" + "x" * 65).status_code == 400
|
||||
|
||||
|
||||
def test_task_detail_missing_ledger_is_failsafe(tmp_path: Path) -> None:
|
||||
out = task_detail(tmp_path / "nope.sqlite", "abc")
|
||||
assert out["ok"] is False
|
||||
|
|
@ -1,10 +1,12 @@
|
|||
"""Tests for the LAN-only read-only status dashboard (``agent_team.status_page``).
|
||||
"""Tests for the status dashboard DATA LAYER (``agent_team.status_page``).
|
||||
|
||||
No live socket is bound: :func:`render_html` is exercised against fabricated
|
||||
:class:`Snapshot` objects, and the read-only reader (:func:`build_snapshot`) is
|
||||
exercised against a temp SQLite ledger seeded with a couple of rows (a real
|
||||
``SqliteSaver`` checkpoint plus ``pending_questions`` / ``budget_ledger`` rows).
|
||||
``conftest.py`` already puts ``agent-team/`` on ``sys.path``.
|
||||
After the WebUI makeover, ``status_page`` no longer renders HTML — the dashboard
|
||||
is a React/Vite SPA served by ``agent_team.dashboard`` (see ``test_dashboard.py``
|
||||
for the FastAPI endpoints). What remains here is the read-only reader
|
||||
(:func:`build_snapshot`) and the ``/api/state`` payload builder
|
||||
(:func:`snapshot_to_dict`), exercised against fabricated :class:`Snapshot`
|
||||
objects and a temp SQLite ledger seeded with real rows. ``conftest.py`` puts
|
||||
``agent-team/`` on ``sys.path``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -21,12 +23,9 @@ from agent_team.status_page import (
|
|||
Snapshot,
|
||||
TaskView,
|
||||
build_snapshot,
|
||||
render_html,
|
||||
snapshot_to_dict,
|
||||
)
|
||||
|
||||
# --- pure render_html tests (no DB, no socket). ------------------------------
|
||||
|
||||
|
||||
def _sample_snapshot() -> Snapshot:
|
||||
return Snapshot(
|
||||
|
|
@ -74,85 +73,16 @@ def _sample_snapshot() -> Snapshot:
|
|||
)
|
||||
|
||||
|
||||
def test_render_html_returns_complete_document() -> None:
|
||||
html_out = render_html(_sample_snapshot())
|
||||
assert html_out.startswith("<!doctype html>")
|
||||
assert html_out.rstrip().endswith("</html>")
|
||||
# Auto-refresh + offline (no external CDN / http references).
|
||||
assert '<meta http-equiv="refresh" content="10">' in html_out
|
||||
assert "http://" not in html_out and "https://" not in html_out
|
||||
assert "<style>" in html_out # inline CSS only
|
||||
|
||||
|
||||
def test_render_html_shows_counts_and_waiting() -> None:
|
||||
snap = _sample_snapshot()
|
||||
html_out = render_html(snap)
|
||||
assert snap.active_count == 2 # active + waiting_human
|
||||
assert snap.parked_count == 1
|
||||
assert len(snap.waiting_tasks) == 1
|
||||
# The waiting task and its phase are surfaced.
|
||||
assert "abc123de" in html_out
|
||||
assert "remediate CVE-2026-0001 in payments-svc" in html_out
|
||||
assert "Waiting on the human gate" in html_out
|
||||
# Budget panel present with a total.
|
||||
assert "Budget" in html_out
|
||||
assert "claude-opus-4" in html_out
|
||||
|
||||
|
||||
def test_render_html_escapes_task_descriptions() -> None:
|
||||
snap = Snapshot(
|
||||
generated_at="now",
|
||||
db_path="/x.sqlite",
|
||||
ok=True,
|
||||
tasks=[
|
||||
TaskView(
|
||||
thread_id="x",
|
||||
short_id="x",
|
||||
task="<script>alert('xss')</script>",
|
||||
current_phase="intake",
|
||||
status="active",
|
||||
)
|
||||
],
|
||||
)
|
||||
html_out = render_html(snap)
|
||||
assert "<script>alert" not in html_out
|
||||
assert "<script>" in html_out
|
||||
|
||||
|
||||
def test_render_html_no_data_page_on_not_ok() -> None:
|
||||
snap = Snapshot(
|
||||
generated_at="now",
|
||||
db_path="/missing.sqlite",
|
||||
ok=False,
|
||||
error="ledger not found",
|
||||
)
|
||||
html_out = render_html(snap)
|
||||
assert "No data" in html_out
|
||||
assert "ledger not found" in html_out
|
||||
assert html_out.rstrip().endswith("</html>")
|
||||
|
||||
|
||||
def test_render_html_omits_budget_when_unavailable() -> None:
|
||||
snap = Snapshot(
|
||||
generated_at="now",
|
||||
db_path="/x.sqlite",
|
||||
ok=True,
|
||||
tasks=[],
|
||||
budget_available=False,
|
||||
)
|
||||
html_out = render_html(snap)
|
||||
assert "Budget" not in html_out
|
||||
|
||||
|
||||
# --- read-only reader tests against a temp seeded SQLite ledger. -------------
|
||||
|
||||
|
||||
def _seed_db(db_path: Path) -> None:
|
||||
def seed_ledger(db_path: Path) -> None:
|
||||
"""Create the agent-team schema and seed a couple of rows.
|
||||
|
||||
Seeds one pending question (open) and one budget row via the foundation
|
||||
Seeds one open pending question and one budget row via the foundation
|
||||
schema, and one LangGraph checkpoint per thread via the real ``SqliteSaver``
|
||||
so the reader's checkpoint path is exercised faithfully.
|
||||
so the reader's checkpoint path is exercised faithfully. Imported by
|
||||
``test_dashboard.py`` as well.
|
||||
"""
|
||||
from agent_team.db.schema import init_db
|
||||
from langgraph.checkpoint.base import empty_checkpoint
|
||||
|
|
@ -183,7 +113,6 @@ def _seed_db(db_path: Path) -> None:
|
|||
)
|
||||
conn.commit()
|
||||
|
||||
# Seed checkpoints for two threads via the real saver.
|
||||
with closing(sqlite3.connect(str(db_path), check_same_thread=False)) as conn:
|
||||
saver = SqliteSaver(conn)
|
||||
for tid, values in (
|
||||
|
|
@ -216,13 +145,12 @@ def _seed_db(db_path: Path) -> None:
|
|||
|
||||
def test_build_snapshot_reads_seeded_ledger(tmp_path: Path) -> None:
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
_seed_db(db)
|
||||
seed_ledger(db)
|
||||
|
||||
snap = build_snapshot(db)
|
||||
|
||||
assert snap.ok is True
|
||||
assert snap.error is None
|
||||
# Two checkpointed threads enumerated.
|
||||
by_id = {t.thread_id: t for t in snap.tasks}
|
||||
assert set(by_id) == {"thread-waiting", "thread-active"}
|
||||
|
||||
|
|
@ -230,35 +158,26 @@ def test_build_snapshot_reads_seeded_ledger(tmp_path: Path) -> None:
|
|||
assert waiting.task == "remediate CVE"
|
||||
assert waiting.current_phase == "clarify"
|
||||
assert waiting.status == "waiting_human"
|
||||
assert waiting.waiting is True # has an open pending_question
|
||||
assert waiting.waiting is True
|
||||
assert waiting.waiting_since == "2026-06-23T11:50:00+00:00"
|
||||
|
||||
active = by_id["thread-active"]
|
||||
assert active.waiting is False
|
||||
assert active.status == "active"
|
||||
|
||||
# Queue summary + question counts.
|
||||
assert snap.active_count == 2 # active + waiting_human
|
||||
assert snap.parked_count == 0
|
||||
assert snap.question_counts.get("open") == 1
|
||||
|
||||
# Budget read.
|
||||
assert snap.budget_available is True
|
||||
assert snap.spend_total_usd == pytest.approx(0.25)
|
||||
assert snap.recent_spend[0]["model"] == "claude-opus-4"
|
||||
|
||||
# The whole thing renders without error.
|
||||
html_out = render_html(snap)
|
||||
assert "remediate CVE" in html_out
|
||||
|
||||
|
||||
def test_build_snapshot_missing_db_is_fail_safe(tmp_path: Path) -> None:
|
||||
snap = build_snapshot(tmp_path / "does-not-exist.sqlite")
|
||||
assert snap.ok is False
|
||||
assert snap.error and "not found" in snap.error
|
||||
# Still renders a friendly page rather than raising.
|
||||
html_out = render_html(snap)
|
||||
assert "No data" in html_out
|
||||
|
||||
|
||||
def test_build_snapshot_never_writes(tmp_path: Path) -> None:
|
||||
|
|
@ -268,43 +187,11 @@ def test_build_snapshot_never_writes(tmp_path: Path) -> None:
|
|||
assert not missing.exists() # mode=ro did not create it
|
||||
|
||||
|
||||
# --- pipeline-map + /api/state JSON contract tests ---------------------------
|
||||
|
||||
|
||||
def test_render_html_includes_svg_map_and_poller() -> None:
|
||||
"""The page must carry the inline SVG map and the inline JS poller."""
|
||||
html_out = render_html(_sample_snapshot())
|
||||
# Inline SVG map (hand-rolled, not fetched).
|
||||
assert '<svg class="map"' in html_out
|
||||
assert 'data-stage="clarify"' in html_out
|
||||
# Inline poller that fetches the JSON sidecar — no external libraries.
|
||||
assert "<script>" in html_out
|
||||
assert "fetch('/api/state'" in html_out
|
||||
assert "setInterval(poll" in html_out
|
||||
# Tooltip mount + live clock the poller updates.
|
||||
assert 'id="tip"' in html_out
|
||||
assert 'id="clock"' in html_out
|
||||
# Legend present.
|
||||
assert "awaiting human" in html_out
|
||||
# Still fully offline: no CDN / external references.
|
||||
assert "http://" not in html_out and "https://" not in html_out
|
||||
# The no-JS fallback is the meta-refresh, now inside <noscript>.
|
||||
assert "<noscript>" in html_out
|
||||
assert '<meta http-equiv="refresh" content="10">' in html_out
|
||||
|
||||
|
||||
def test_render_html_renders_map_even_with_no_data() -> None:
|
||||
"""A not-ok snapshot still renders the map + poller so it goes live later."""
|
||||
snap = Snapshot(generated_at="now", db_path="/x.sqlite", ok=False, error="boom")
|
||||
html_out = render_html(snap)
|
||||
assert '<svg class="map"' in html_out
|
||||
assert "fetch('/api/state'" in html_out
|
||||
assert "No data" in html_out
|
||||
assert html_out.rstrip().endswith("</html>")
|
||||
# --- /api/state JSON payload (snapshot_to_dict) contract ---------------------
|
||||
|
||||
|
||||
def test_snapshot_to_dict_shape_and_grouping() -> None:
|
||||
"""The /api/state payload has the expected keys + correct stage grouping."""
|
||||
"""The /api/state payload keeps its keys + correct stage grouping."""
|
||||
payload = snapshot_to_dict(_sample_snapshot())
|
||||
assert payload["ok"] is True
|
||||
assert set(payload) >= {
|
||||
|
|
@ -317,32 +204,22 @@ def test_snapshot_to_dict_shape_and_grouping() -> None:
|
|||
"summary",
|
||||
"budget",
|
||||
}
|
||||
# One stage entry per declared STAGE, in declared order.
|
||||
assert [s["key"] for s in payload["stages"]] == [s.key for s in STAGES]
|
||||
|
||||
by_key = {s["key"]: s for s in payload["stages"]}
|
||||
# The waiting clarify task lands on the clarify node and flags awaiting_human.
|
||||
clarify = by_key["clarify"]
|
||||
assert clarify["count"] == 1
|
||||
assert clarify["state"] == "awaiting_human"
|
||||
assert clarify["tasks"][0]["short_id"] == "abc123de"
|
||||
# Each stage carries its model/agent role for the per-node label.
|
||||
assert clarify["agent"] == "Claude (sub)"
|
||||
assert by_key["review"]["agent"] == "GPT-4.1 (cross)"
|
||||
assert by_key["build"]["agent"] == "DeepSeek (fast)"
|
||||
# The active plan task lands on the plan node as 'active'.
|
||||
assert by_key["plan"]["state"] == "active"
|
||||
assert by_key["plan"]["count"] == 1
|
||||
# The parked task maps onto the (terminal-ish) parked side — no stage owns
|
||||
# the 'parked' phase, so no pipeline node claims it.
|
||||
# The parked task maps onto no pipeline node.
|
||||
assert all(
|
||||
t["thread_id"] != "dead0000beef" for s in payload["stages"] for t in s["tasks"]
|
||||
)
|
||||
# An idle stage with no tasks.
|
||||
assert by_key["intake"]["state"] == "idle"
|
||||
assert by_key["intake"]["count"] == 0
|
||||
|
||||
# Summary mirrors the snapshot counts.
|
||||
assert payload["summary"]["active"] == 2
|
||||
assert payload["summary"]["waiting"] == 1
|
||||
assert payload["summary"]["open_questions"] == 1
|
||||
|
|
@ -350,20 +227,17 @@ def test_snapshot_to_dict_shape_and_grouping() -> None:
|
|||
|
||||
|
||||
def test_snapshot_to_dict_not_ok_is_serializable() -> None:
|
||||
"""A not-ok snapshot still serializes to a valid, JSON-dumpable payload."""
|
||||
snap = Snapshot(generated_at="now", db_path="/x.sqlite", ok=False, error="boom")
|
||||
payload = snapshot_to_dict(snap)
|
||||
assert payload["ok"] is False
|
||||
assert payload["error"] == "boom"
|
||||
# Stages still present (all idle) so the client can render the empty map.
|
||||
assert [s["key"] for s in payload["stages"]] == [s.key for s in STAGES]
|
||||
assert all(s["count"] == 0 for s in payload["stages"])
|
||||
# Round-trips through json.
|
||||
assert json.loads(json.dumps(payload))["ok"] is False
|
||||
|
||||
|
||||
def test_descriptions_escaped_in_html_and_json_seed() -> None:
|
||||
"""Hostile descriptions must not break out of HTML *or* the inline JSON."""
|
||||
def test_snapshot_to_dict_carries_raw_description_for_textcontent() -> None:
|
||||
"""The JSON sidecar carries the raw description; the SPA renders it safely."""
|
||||
snap = Snapshot(
|
||||
generated_at="now",
|
||||
db_path="/x.sqlite",
|
||||
|
|
@ -378,39 +252,6 @@ def test_descriptions_escaped_in_html_and_json_seed() -> None:
|
|||
)
|
||||
],
|
||||
)
|
||||
html_out = render_html(snap)
|
||||
# No raw script tag survives anywhere in the document (table render escapes
|
||||
# it; the inline JSON seed escapes < and > to \\uXXXX).
|
||||
assert "<script>alert(1)" not in html_out
|
||||
assert "</script><script>" not in html_out
|
||||
assert "\\u003cscript\\u003e" in html_out # JSON seed escaped form
|
||||
|
||||
# The JSON sidecar itself is valid JSON carrying the raw description (the
|
||||
# consumer renders it via textContent, never as markup).
|
||||
payload = snapshot_to_dict(snap)
|
||||
body = json.dumps(payload)
|
||||
parsed = json.loads(body)
|
||||
parsed = json.loads(json.dumps(payload))
|
||||
assert parsed["tasks"][0]["task"] == "</script><script>alert(1)</script>"
|
||||
|
||||
|
||||
def test_build_snapshot_api_payload_against_seeded_ledger(tmp_path: Path) -> None:
|
||||
"""End-to-end: read a seeded ledger, then assert the /api/state payload."""
|
||||
db = tmp_path / "agent_team.sqlite"
|
||||
_seed_db(db)
|
||||
|
||||
payload = snapshot_to_dict(build_snapshot(db))
|
||||
|
||||
assert payload["ok"] is True
|
||||
by_key = {s["key"]: s for s in payload["stages"]}
|
||||
# thread-waiting (clarify, open question) -> clarify node, awaiting_human.
|
||||
assert by_key["clarify"]["state"] == "awaiting_human"
|
||||
assert by_key["clarify"]["count"] == 1
|
||||
assert by_key["clarify"]["tasks"][0]["thread_id"] == "thread-waiting"
|
||||
# thread-active (plan, active) -> plan node, active.
|
||||
assert by_key["plan"]["state"] == "active"
|
||||
assert by_key["plan"]["count"] == 1
|
||||
# Budget surfaced.
|
||||
assert payload["budget"]["available"] is True
|
||||
assert payload["budget"]["total_usd"] == pytest.approx(0.25)
|
||||
# Whole payload is JSON-serializable.
|
||||
assert json.loads(json.dumps(payload))["summary"]["total"] == 2
|
||||
|
|
|
|||
60
agent-team/tests/test_topology.py
Normal file
60
agent-team/tests/test_topology.py
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
"""Tests for the LangGraph-introspected pipeline topology (``agent_team.topology``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from agent_team.topology import (
|
||||
NODE_META,
|
||||
PHASE_TO_NODE,
|
||||
build_topology,
|
||||
missing_meta,
|
||||
node_for_phase,
|
||||
)
|
||||
|
||||
|
||||
def test_every_graph_node_has_meta() -> None:
|
||||
"""A newly added graph node must fail loudly until its NODE_META is filled in."""
|
||||
assert missing_meta() == []
|
||||
|
||||
|
||||
def test_topology_has_expected_nodes_and_trees() -> None:
|
||||
topo = build_topology()
|
||||
ids = {n["id"] for n in topo["nodes"]}
|
||||
# The maximal P3+ wiring nodes, derived from the real graph.
|
||||
assert {"intake", "clarify", "plan", "review", "build_node", "verify_node"} <= ids
|
||||
# __start__/__end__ are not drawn.
|
||||
assert "__start__" not in ids and "__end__" not in ids
|
||||
# intake roots the core tree; the SDLC pipeline branches off it.
|
||||
tree_ids = {t["id"] for t in topo["trees"]}
|
||||
assert {"core", "sdlc"} <= tree_ids
|
||||
by_id = {n["id"]: n for n in topo["nodes"]}
|
||||
assert by_id["intake"]["tree"] == "core"
|
||||
assert by_id["clarify"]["kind"] == "gate" # human gate lives at clarify
|
||||
# P3 nodes render but are flagged inert.
|
||||
assert by_id["build_node"]["gated"] is True
|
||||
|
||||
|
||||
def test_edges_classified_spine_branch_loopback() -> None:
|
||||
topo = build_topology()
|
||||
kinds = {(e["from"], e["to"]): e["kind"] for e in topo["edges"]}
|
||||
assert kinds[("intake", "clarify")] == "spine"
|
||||
assert kinds[("clarify", "plan")] == "spine"
|
||||
# review -> plan is the revise loop; verify -> build_node is the build loop.
|
||||
assert kinds[("review", "plan")] == "loopback"
|
||||
assert kinds[("verify_node", "build_node")] == "loopback"
|
||||
# No edge references the framework terminals.
|
||||
nodes = {n["id"] for n in topo["nodes"]}
|
||||
for e in topo["edges"]:
|
||||
assert e["from"] in nodes and e["to"] in nodes
|
||||
|
||||
|
||||
def test_phase_to_node_mapping() -> None:
|
||||
assert node_for_phase("build") == "build_node"
|
||||
assert node_for_phase("verify") == "verify_node"
|
||||
assert node_for_phase("clarify") == "clarify"
|
||||
# Terminal/exception phases map to no node.
|
||||
assert node_for_phase("done") is None
|
||||
assert node_for_phase("parked") is None
|
||||
assert node_for_phase("") is None
|
||||
# Every mapped target is a real meta node.
|
||||
for node_id in PHASE_TO_NODE.values():
|
||||
assert node_id in NODE_META
|
||||
163
agent-team/tests/test_transitions.py
Normal file
163
agent-team/tests/test_transitions.py
Normal file
|
|
@ -0,0 +1,163 @@
|
|||
"""Tests for the task_transitions recorder + graph instrumentation.
|
||||
|
||||
Covers (plan refs): idempotent record_entry under resume replay (BLOCK-1), the
|
||||
terminal close that fills exited_at on the last node (N2), fail-soft writes that
|
||||
never break the pipeline, the B4 signature-preservation guarantee of the
|
||||
``_instrument`` wrapper, and an end-to-end graph drive that records rows.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
from agent_team.db import connect, init_db, read_transitions
|
||||
from agent_team.db.schema import assert_task_transitions_ready
|
||||
from agent_team.db.transitions import TransitionRecorder
|
||||
|
||||
|
||||
def _rows(db: Path, thread_id: str) -> list[dict[str, object]]:
|
||||
conn = connect(db)
|
||||
try:
|
||||
return read_transitions(conn, thread_id)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_record_entry_chains_and_closes_previous(tmp_path: Path) -> None:
|
||||
db = tmp_path / "led.sqlite"
|
||||
init_db(db)
|
||||
rec = TransitionRecorder(db)
|
||||
|
||||
rec.record_entry(thread_id="t1", to_phase="intake", status="active")
|
||||
rec.record_entry(thread_id="t1", to_phase="clarify", status="active")
|
||||
|
||||
rows = _rows(db, "t1")
|
||||
assert [r["to_phase"] for r in rows] == ["intake", "clarify"]
|
||||
# First row closed (exited_at set) when the second landed; chain carried.
|
||||
assert rows[0]["exited_at"] is not None
|
||||
assert rows[1]["exited_at"] is None
|
||||
assert rows[1]["from_phase"] == "intake"
|
||||
|
||||
|
||||
def test_record_entry_is_idempotent_under_replay(tmp_path: Path) -> None:
|
||||
"""Re-entering the SAME open node (resume replay) does not duplicate a row."""
|
||||
db = tmp_path / "led.sqlite"
|
||||
init_db(db)
|
||||
rec = TransitionRecorder(db)
|
||||
|
||||
rec.record_entry(thread_id="t1", to_phase="clarify", status="active")
|
||||
rec.record_entry(thread_id="t1", to_phase="clarify", status="active") # replay
|
||||
rec.record_entry(thread_id="t1", to_phase="clarify", status="active") # replay
|
||||
|
||||
rows = _rows(db, "t1")
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["to_phase"] == "clarify"
|
||||
assert rows[0]["exited_at"] is None # still open, single row
|
||||
|
||||
|
||||
def test_close_terminal_stamps_exit_on_last_open_row(tmp_path: Path) -> None:
|
||||
db = tmp_path / "led.sqlite"
|
||||
init_db(db)
|
||||
rec = TransitionRecorder(db)
|
||||
|
||||
rec.record_entry(thread_id="t1", to_phase="plan", status="active")
|
||||
rec.close_terminal(thread_id="t1", status="done")
|
||||
|
||||
rows = _rows(db, "t1")
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["exited_at"] is not None
|
||||
assert rows[0]["status"] == "done"
|
||||
|
||||
|
||||
def test_writes_are_fail_soft(tmp_path: Path) -> None:
|
||||
"""A ledger error is swallowed — recording must never raise into the graph."""
|
||||
# Point the recorder at a path that is a directory, so connect/execute fails.
|
||||
bad = tmp_path / "a_dir"
|
||||
bad.mkdir()
|
||||
rec = TransitionRecorder(bad)
|
||||
# Must not raise.
|
||||
rec.record_entry(thread_id="t1", to_phase="intake", status="active")
|
||||
rec.close_terminal(thread_id="t1", status="done")
|
||||
|
||||
|
||||
def test_read_transitions_missing_table_returns_empty(tmp_path: Path) -> None:
|
||||
db = tmp_path / "empty.sqlite"
|
||||
# A bare DB with no agent-team schema.
|
||||
sqlite3.connect(str(db)).close()
|
||||
conn = connect(db)
|
||||
try:
|
||||
assert read_transitions(conn, "t1") == []
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_assert_task_transitions_ready(tmp_path: Path) -> None:
|
||||
db = tmp_path / "led.sqlite"
|
||||
init_db(db)
|
||||
# Does not raise on a properly migrated DB.
|
||||
assert_task_transitions_ready(db)
|
||||
|
||||
|
||||
# --- B4: _instrument signature preservation + graph integration --------------
|
||||
|
||||
|
||||
def test_instrument_preserves_signature_and_forwards_config() -> None:
|
||||
"""The wrapper must present fn's signature (so LangGraph injects config) and
|
||||
forward every argument verbatim (plan B4)."""
|
||||
from agent_team.graph import _instrument
|
||||
|
||||
seen: dict[str, object] = {}
|
||||
|
||||
def node_with_config(state: dict, config: dict) -> dict: # type: ignore[type-arg]
|
||||
seen["state"] = state
|
||||
seen["config"] = config
|
||||
return {"status": "active"}
|
||||
|
||||
class _Rec:
|
||||
def record_entry(self, **_kw: object) -> None: ...
|
||||
def close_terminal(self, **_kw: object) -> None: ...
|
||||
|
||||
wrapped = _instrument("plan", node_with_config, _Rec())
|
||||
# Signature still advertises BOTH params, so LangGraph passes config.
|
||||
assert list(inspect.signature(wrapped).parameters) == ["state", "config"]
|
||||
# And the wrapper forwards them.
|
||||
wrapped({"thread_id": "t"}, {"configurable": {}})
|
||||
assert seen["config"] == {"configurable": {}}
|
||||
|
||||
|
||||
def test_instrument_none_recorder_is_identity() -> None:
|
||||
from agent_team.graph import _instrument
|
||||
|
||||
def node(state: dict) -> dict: # type: ignore[type-arg]
|
||||
return {}
|
||||
|
||||
assert _instrument("intake", node, None) is node
|
||||
|
||||
|
||||
def test_graph_drive_records_transitions(tmp_path: Path) -> None:
|
||||
"""End-to-end: a recorder-instrumented graph records rows as a task runs."""
|
||||
from langgraph.checkpoint.memory import MemorySaver
|
||||
|
||||
from agent_team.graph import build_graph, resume_task, start_task
|
||||
|
||||
db = tmp_path / "led.sqlite"
|
||||
init_db(db)
|
||||
rec = TransitionRecorder(db)
|
||||
graph = build_graph(MemorySaver(), transition_recorder=rec)
|
||||
|
||||
tid, _ = start_task(graph, transport="cli", task="do a thing")
|
||||
# Ran INTAKE -> CLARIFY (suspended on the human gate).
|
||||
rows = _rows(db, tid)
|
||||
phases = [r["to_phase"] for r in rows]
|
||||
assert "intake" in phases
|
||||
assert "clarify" in phases
|
||||
|
||||
resume_task(graph, thread_id=tid, answer="scope it")
|
||||
rows = _rows(db, tid)
|
||||
phases = [r["to_phase"] for r in rows]
|
||||
assert "plan" in phases
|
||||
# Terminal plan node (status DONE) closed its row.
|
||||
plan_row = [r for r in rows if r["to_phase"] == "plan"][-1]
|
||||
assert plan_row["exited_at"] is not None
|
||||
Reference in a new issue