test(agent-team): cover live pipeline map + /api/state JSON contract

- render_html includes the inline SVG map + inline fetch('/api/state') poller
  + tooltip mount + live clock, stays offline (no CDN), keeps the <noscript>
  meta-refresh fallback, and renders the map even on a not-ok snapshot.
- snapshot_to_dict: expected top-level keys, one entry per STAGE in order,
  per-phase grouping, awaiting-human classification (open pending_question),
  per-node model/agent role labels, summary counts, not-ok serializability.
- Descriptions escaped in BOTH the HTML and the JSON-in-script seed
  (</script><script> breakout neutralised; \u003c form present); the raw
  /api/state JSON round-trips the description for textContent rendering.
- End-to-end snapshot_to_dict over the seeded ledger.

agent-team/tests/test_status_page.py

@@ -9,6 +9,7 @@ exercised against a temp SQLite ledger seeded with a couple of rows (a real
 
 from __future__ import annotations
 
+import json
 import sqlite3
 from contextlib import closing
 from pathlib import Path
@@ -16,10 +17,12 @@ from pathlib import Path
 import pytest
 
 from agent_team.status_page import (
+    STAGES,
     Snapshot,
     TaskView,
     build_snapshot,
     render_html,
+    snapshot_to_dict,
 )
 
 # --- pure render_html tests (no DB, no socket). ------------------------------
@@ -263,3 +266,151 @@ def test_build_snapshot_never_writes(tmp_path: Path) -> None:
     missing = tmp_path / "nope.sqlite"
     build_snapshot(missing)
     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>")
+
+
+def test_snapshot_to_dict_shape_and_grouping() -> None:
+    """The /api/state payload has the expected keys + correct stage grouping."""
+    payload = snapshot_to_dict(_sample_snapshot())
+    assert payload["ok"] is True
+    assert set(payload) >= {
+        "ok",
+        "generated_at",
+        "stages",
+        "tasks",
+        "waiting",
+        "question_counts",
+        "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.
+    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
+    assert payload["summary"]["total"] == 3
+
+
+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."""
+    snap = Snapshot(
+        generated_at="now",
+        db_path="/x.sqlite",
+        ok=True,
+        tasks=[
+            TaskView(
+                thread_id="x",
+                short_id="x",
+                task="</script><script>alert(1)</script>",
+                current_phase="plan",
+                status="active",
+            )
+        ],
+    )
+    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)
+    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