diff --git a/agent-team/tests/conftest.py b/agent-team/tests/conftest.py
index 900c650..236b80e 100644
--- a/agent-team/tests/conftest.py
+++ b/agent-team/tests/conftest.py
@@ -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))
diff --git a/agent-team/tests/test_dashboard.py b/agent-team/tests/test_dashboard.py
new file mode 100644
index 0000000..14a936a
--- /dev/null
+++ b/agent-team/tests/test_dashboard.py
@@ -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
diff --git a/agent-team/tests/test_status_page.py b/agent-team/tests/test_status_page.py
index 700bdaf..0b03067 100644
--- a/agent-team/tests/test_status_page.py
+++ b/agent-team/tests/test_status_page.py
@@ -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 "&lt;script&gt;" 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
diff --git a/agent-team/tests/test_topology.py b/agent-team/tests/test_topology.py
new file mode 100644
index 0000000..623a090
--- /dev/null
+++ b/agent-team/tests/test_topology.py
@@ -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
diff --git a/agent-team/tests/test_transitions.py b/agent-team/tests/test_transitions.py
new file mode 100644
index 0000000..25e7c5a
--- /dev/null
+++ b/agent-team/tests/test_transitions.py
@@ -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