hud.py

"""hud.py — the Aporia swarm HUD (task 32). A small always-on-top panel showing swarm state at a
glance so Aaron has live visibility between debriefs.

Design: swarm-hud-design.md (task 31, ratified by Aaron 2026-07-04). Clean-room inspiration:
    Ringside by Nate B Jones (inspiration; clean-room, no source consulted) — the IDEA of an
    always-on-top swarm glance panel, not one line of its code (New Brain fact 173).

v1 is STRICTLY DISPLAY-ONLY (explicit trust-ramp decision). This process opens estate files
READ-ONLY and writes NOTHING — no buttons, no approve actions, no control path of any kind.
Approvals happen in Aaron's sessions; making this a control surface is a later decision.

COLORBLIND PALETTE — LAW (Aaron is colorblind; New Brain facts 182-186):
    pass/success = BLUE (never green) · needs-Aaron = PURPLE (never orange/amber) ·
    fail = RED · running = GRAY hollow ring.
    EVERY state also carries a shape/icon cue (check / x / hollow dot) — never color alone.

Tech: pywebview — a frameless, always-on-top window rendering local HTML, refreshed ~2s. The
window chrome is pywebview's; the panel is the HTML below. One pip dependency (pywebview).

Data sources (all read-only):
    dispatch/*.json        — the task records (id, title, type)
    dispatch_log.jsonl     — dispatch OUTCOMES (done / escalated / returned / not_dispatched / waiting)
    worker_status.jsonl    — worker LIVENESS events per task, written by the orchestrator's additive
                             lifecycle side-channel (task 32 spine change): 'running' lands before the
                             dispatch blocks, then a terminal 'finished' (or 'requeued' on a rate
                             limit). A task is LIVE iff its latest event is 'running'.
    field_log.jsonl +      — the ladder's abstain queue (the "Open questions" vitals count comes from
    ue_field_log.jsonl       here: unique facts with verdict 'abstain', per ladder.py abstained_nulls).
    New Brain Postgres     — ONE count (open conflicts) over a STRUCTURALLY READ-ONLY connection
                             (default_transaction_read_only=on — the display-only law is enforced by
                             the connection itself, not by discipline). Aaron-approved 2026-07-05;
                             config imported from new_brain_store.py (the one DB seam). Degrades to
                             '-' whenever the DB is unreachable; the panel never blocks on it.

Run:  .venv/Scripts/python.exe hud.py
"""

from __future__ import annotations

import html
import json
import threading
import time
from datetime import datetime, timezone
from pathlib import Path

import webview

HERE = Path(__file__).resolve().parent
DISPATCH_DIR = HERE / "dispatch"
DISPATCH_LOG = HERE / "dispatch_log.jsonl"
WORKER_STATUS = HERE / "worker_status.jsonl"

REFRESH_SECONDS = 2.0
RECENT_GATES_MAX = 8            # how many recent gate chips to show (newest first)


# ---------------------------------------------------------------------------
# read layer — READ-ONLY, and DEFENSIVE: files may be mid-append, rotated, absent, or empty at any
# tick. Every reader tolerates a torn/partial last line and a missing file; a bad read yields empty,
# never a crash (the HUD must survive log rotation/append while it runs — gate 2).
# ---------------------------------------------------------------------------
def _read_jsonl(path: Path) -> list[dict]:
    """Read a .jsonl file into a list of dicts, skipping any line that does not parse (a line being
    appended right now can be half-written; a rotation can truncate the file). Missing file -> []."""
    try:
        text = path.read_text(encoding="utf-8")
    except (FileNotFoundError, OSError):
        return []
    out = []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            out.append(json.loads(line))
        except json.JSONDecodeError:
            continue                                          # torn/partial line — skip, don't crash
    return out


def _read_task_records(dispatch_dir: Path = DISPATCH_DIR) -> dict[str, dict]:
    """task_id -> record for every dispatch/*.json. A record that fails to parse is skipped (one bad
    file never blanks the HUD). Absent dir -> {}."""
    records: dict[str, dict] = {}
    try:
        paths = sorted(dispatch_dir.glob("*.json"))
    except OSError:
        return records
    for p in paths:
        try:
            rec = json.loads(p.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            continue
        tid = rec.get("id") or p.stem
        records[tid] = rec
    return records


# ---------------------------------------------------------------------------
# live vitals (Aaron-approved 2026-07-05) — the two counts the file sources alone couldn't give.
#   Open questions: the ladder's abstain/triage queue. Ground truth is CODE, not the DB: the New Brain
#     schema has no queue table (new_brain_schema.sql: tags/tag_aliases/facts/conflicts/documents);
#     ladder.py:51+588 derives the queue from FIELD_LOGS — unique facts with verdict 'abstain'. The
#     HUD counts the same way, from the same files, read-only.
#   Open conflicts: `SELECT count(*) FROM conflicts WHERE status='open'` (new_brain_store.stats,
#     new_brain_store.py:132) over a STRUCTURALLY READ-ONLY connection: default_transaction_read_only=on
#     makes Postgres itself refuse any write on this connection — display-only enforced by the server,
#     not by discipline. Config is IMPORTED from new_brain_store (the one DB seam; NB_DB_* env scheme).
#     A background thread polls every DB_POLL_SECONDS into a cache; render() only ever reads the cache,
#     so a dead/slow DB costs the poller a bounded timeout and the panel NOTHING. Any failure -> '-'.
# ---------------------------------------------------------------------------
FIELD_LOGS = [HERE / "field_log.jsonl", HERE / "ue_field_log.jsonl"]   # ladder.py:51 — the abstain queue
DB_POLL_SECONDS = 10.0          # DB poll cadence (render's ~2s tick reads the cache, never the DB)
DB_CONNECT_TIMEOUT = 2          # a dead DB costs the background poller <=2s per cycle, the panel nothing

_db_cache = {"open_conflicts": "-"}   # written only by the poll thread; read by _vitals


def _nb_db_config() -> dict | None:
    """The New Brain connection config, IMPORTED from new_brain_store.py (its DB dict is the
    NB_DB_HOST/PORT/NAME/USER/PASSWORD env scheme — reused, never re-invented). None when the store
    module can't load (missing dep, moved file): vitals then stay '-', the HUD runs regardless."""
    try:
        from new_brain_store import DB
        return dict(DB)
    except Exception:
        return None


def _poll_open_conflicts(cfg: dict, connect=None) -> int | str:
    """ONE poll of the open-conflicts count over a fresh, STRUCTURALLY READ-ONLY connection
    (default_transaction_read_only=on — the server rejects any write attempt on this connection, so
    the display-only law is enforced by Postgres itself). Query mirrors new_brain_store.stats()
    (new_brain_store.py:132). ANY failure — DB down, timeout, missing table, bad auth — returns '-'
    (unknown), never a fabricated count and never an exception. `connect` is an injectable seam so
    tests can prove both directions without a live DB."""
    try:
        if connect is None:
            import psycopg
            connect = psycopg.connect
        with connect(connect_timeout=DB_CONNECT_TIMEOUT, autocommit=True,
                     options="-c default_transaction_read_only=on", **cfg) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT count(*) FROM conflicts WHERE status='open'")
                return int(cur.fetchone()[0])
    except Exception:
        return "-"


def _db_poll_loop() -> None:
    """Background poller: refresh the open-conflicts cache every DB_POLL_SECONDS, forever. Config is
    resolved once; each cycle uses a fresh short-timeout connection (no stale-connection state)."""
    cfg = _nb_db_config()
    while True:
        _db_cache["open_conflicts"] = _poll_open_conflicts(cfg) if cfg else "-"
        time.sleep(DB_POLL_SECONDS)


def start_db_poller() -> None:
    """Start the vitals DB poller as a daemon thread (dies with the window). Called from main() —
    importing hud (tests) starts no thread and opens no connection."""
    threading.Thread(target=_db_poll_loop, daemon=True, name="hud-db-poller").start()


def _open_questions_count() -> int:
    """The 'Open questions' vitals count — the abstain/triage queue, derived exactly as
    ladder.abstained_nulls() does (ladder.py:588): unique `fact` values with verdict 'abstain' across
    the field logs. Missing logs contribute nothing (the ladder treats them the same); torn lines are
    skipped by _read_jsonl. Pure read-only file derivation — no DB involved."""
    seen = set()
    for log in FIELD_LOGS:
        for rec in _read_jsonl(log):
            if not isinstance(rec, dict):
                continue
            for r in rec.get("results", []):
                if isinstance(r, dict) and r.get("verdict") == "abstain":
                    seen.add(r.get("fact"))
    return len(seen)


# ---------------------------------------------------------------------------
# derive layer — turn the three raw sources into the six-section view model. Pure functions of the
# file contents at one instant; no state kept between ticks (so rotation just changes the next frame).
# ---------------------------------------------------------------------------
def _running_workers(status_events: list[dict]) -> list[dict]:
    """A task is RUNNING iff its most recent liveness event is 'running' — any later terminal event
    ('finished', or 'requeued' on a rate limit) clears it. Events are append-ordered in
    worker_status.jsonl, so the last event per task_id is authoritative."""
    last: dict[str, dict] = {}
    for ev in status_events:
        tid = ev.get("task_id")
        if tid is not None:
            last[tid] = ev
    return [ev for ev in last.values() if ev.get("phase") == "running"]


# outcome statuses that mean a gate verdict landed (feed the "Recent gates" band)
_PASS_STATUSES = {"done"}
_FAIL_STATUSES = {"escalated", "returned", "returned_to_spawner"}
# outcomes that mean Aaron owes a decision (feed the "Needs your yes" band): a batched dispatch that
# has not been approved, and a gate that climbed to him. "waiting" = subscription queue-wait (vitals).
_NEEDS_YES_STATUSES = {"not_dispatched"}


def _recent_gates(outcomes: list[dict]) -> list[dict]:
    """The most recent gate verdicts, newest first: a 'done' is a BLUE pass chip; an escalated/returned
    is a RED fail chip that NAMES its escalation (failure reads as HANDLED, not alarming — design)."""
    chips = []
    for rec in outcomes:
        status = rec.get("status")
        if status in _PASS_STATUSES:
            chips.append({"task_id": rec.get("task_id"), "state": "pass", "detail": "gates passed"})
        elif status in _FAIL_STATUSES:
            esc = rec.get("escalated_to")
            detail = (f"escalated -> {esc}" if esc else
                      "returned to spawner" if status in ("returned", "returned_to_spawner") else status)
            chips.append({"task_id": rec.get("task_id"), "state": "fail", "detail": detail})
    chips.reverse()                                           # newest first
    return chips[:RECENT_GATES_MAX]


def _needs_yes(outcomes: list[dict], task_records: dict[str, dict]) -> list[dict]:
    """Batched merge reviews + supervised checkpoints Aaron owes a yes on. v1 file-derived signal: the
    LATEST outcome per task is 'not_dispatched' (a batched dispatch awaiting his yes) — a later 'done'
    for the same task clears it. Titles come from the task records when present."""
    latest: dict[str, dict] = {}
    for rec in outcomes:
        tid = rec.get("task_id")
        if tid is not None:
            latest[tid] = rec
    items = []
    for tid, rec in latest.items():
        if rec.get("status") in _NEEDS_YES_STATUSES:
            title = (task_records.get(tid) or {}).get("title", "")
            items.append({"task_id": tid, "title": title})
    return items


def _vitals(outcomes: list[dict], task_records: dict[str, dict]) -> dict:
    """The vitals strip (live counts Aaron-approved 2026-07-05).
      - open_questions: the abstain/triage queue counted from the field logs (_open_questions_count —
                        same derivation as ladder.abstained_nulls; a number, missing logs count 0)
      - subscription:   count of tasks whose latest outcome is 'waiting' (subscription queue-wait)
      - open_conflicts: the cached read-only DB poll (_db_cache) — a live count when the New Brain DB
                        answers, '-' (unknown, never fabricated) when it doesn't or the poller is off."""
    latest: dict[str, dict] = {}
    for rec in outcomes:
        tid = rec.get("task_id")
        if tid is not None:
            latest[tid] = rec
    waiting = sum(1 for r in latest.values() if r.get("status") == "waiting")
    return {"open_questions": _open_questions_count(), "subscription": waiting,
            "open_conflicts": _db_cache["open_conflicts"]}


def build_view() -> dict:
    """The whole view model for one frame — a pure snapshot of the three files right now. Robust to any
    of them being absent/rotated/mid-append (each reader degrades to empty)."""
    outcomes = _read_jsonl(DISPATCH_LOG)
    status_events = _read_jsonl(WORKER_STATUS)
    task_records = _read_task_records()

    running = _running_workers(status_events)
    for r in running:                                         # attach elapsed + title for display
        r["title"] = r.get("title") or (task_records.get(r.get("task_id")) or {}).get("title", "")
        r["elapsed"] = _elapsed_str(r.get("ts"))

    gates = _recent_gates(outcomes)
    needs = _needs_yes(outcomes, task_records)
    vitals = _vitals(outcomes, task_records)

    return {
        "running": running,
        "queued": max(0, len(task_records) - len(running)),   # coarse: records not currently running
        "done": sum(1 for r in outcomes if r.get("status") == "done"),
        "needs_yes": needs,
        "recent_gates": gates,
        "vitals": vitals,
        "updated": datetime.now(timezone.utc).isoformat(),
    }


def _elapsed_str(iso_ts: str | None) -> str:
    """Human elapsed since an ISO timestamp (worker start). Unknown/unparseable -> ''."""
    if not iso_ts:
        return ""
    try:
        started = datetime.fromisoformat(iso_ts)
    except (ValueError, TypeError):
        return ""
    if started.tzinfo is None:
        started = started.replace(tzinfo=timezone.utc)
    secs = int((datetime.now(timezone.utc) - started).total_seconds())
    if secs < 0:
        secs = 0
    if secs < 60:
        return f"{secs}s"
    if secs < 3600:
        return f"{secs // 60}m {secs % 60}s"
    return f"{secs // 3600}h {(secs % 3600) // 60}m"


# ---------------------------------------------------------------------------
# render layer — the panel HTML/CSS. The PALETTE and the SHAPE/ICON cues are the law; they live here
# as CSS variables + per-state glyphs so no state is ever distinguished by color alone.
# ---------------------------------------------------------------------------
# Palette (facts 182-186). Blue = good (NOT green); purple = needs-Aaron (NOT orange/amber); red = fail;
# gray hollow ring = running. Each also carries a glyph so the state survives with color stripped out.
PALETTE_CSS = """
:root{
  --bg:#12141a; --panel:#171a22; --edge:#262b36; --text:#e6e9f0; --muted:#8b93a7;
  --blue:#4c8dff;        /* pass / success — NEVER green */
  --purple:#a97cff;      /* needs-Aaron / attention — NEVER orange or amber */
  --red:#ff5c6c;         /* fail */
  --running:#8b93a7;     /* running — a GRAY hollow ring, so blue stays unambiguously 'good' */
}
"""

# glyph cues carried by every state (never color alone): check=pass, x=fail, hollow dot=running,
# filled dot=needs-Aaron. Kept as literal characters so no icon font is required (offline, self-contained).
GLYPH_PASS = "&#10003;"      # check
GLYPH_FAIL = "&#10007;"      # x
GLYPH_NEEDS = "&#9679;"      # filled dot (attention)
GLYPH_RUNNING = "&#9711;"    # large hollow circle (running ring)


PAGE_TEMPLATE = """<!doctype html>
<html><head><meta charset="utf-8"><style>
""" + PALETTE_CSS + """
*{box-sizing:border-box;margin:0;padding:0}
html,body{background:var(--bg);color:var(--text);
  font:12px/1.4 "Segoe UI",system-ui,sans-serif;-webkit-user-select:none;user-select:none;overflow:hidden}
#app{padding:8px 10px}
.title{display:flex;align-items:center;gap:8px;padding-bottom:6px;border-bottom:1px solid var(--edge)}
.title .dot{width:9px;height:9px;border-radius:50%;background:var(--blue)}
.title .name{font-weight:600;letter-spacing:.2px}
.title .counts{margin-left:auto;color:var(--muted);font-variant-numeric:tabular-nums}
.title .pin{color:var(--muted);font-size:11px}
.section{margin-top:8px}
.section h2{font-size:10px;text-transform:uppercase;letter-spacing:.7px;color:var(--muted);margin-bottom:4px}
.band-yes{background:rgba(169,124,255,.12);border:1px solid var(--purple);border-radius:6px;padding:6px 8px}
.band-yes h2{color:var(--purple)}
.yes-row{display:flex;align-items:center;gap:7px;padding:2px 0}
.yes-row .g{color:var(--purple)}
.mono{font-family:"Cascadia Mono",Consolas,monospace;font-size:11px}
.worker{display:flex;align-items:center;gap:8px;padding:3px 0}
.worker .ring{color:var(--running);font-size:12px}
.worker .model{color:var(--muted)}
.worker .elapsed{margin-left:auto;color:var(--muted);font-variant-numeric:tabular-nums}
.chips{display:flex;flex-wrap:wrap;gap:5px}
.chip{display:inline-flex;align-items:center;gap:5px;border-radius:5px;padding:3px 7px;font-size:11px;
  border:1px solid var(--edge)}
.chip.pass{color:var(--blue);border-color:var(--blue);background:rgba(76,141,255,.10)}
.chip.fail{color:var(--red);border-color:var(--red);background:rgba(255,92,108,.10)}
.chip .g{font-weight:700}
.vitals{display:flex;gap:14px;color:var(--muted);font-variant-numeric:tabular-nums}
.vitals b{color:var(--text);font-weight:600}
.empty{color:var(--muted);font-style:italic}
.footer{margin-top:9px;padding-top:6px;border-top:1px solid var(--edge);display:flex;
  color:var(--muted);font-size:10px}
.footer .age{margin-left:auto;font-variant-numeric:tabular-nums}
</style></head>
<body><div id="app">__RENDER__</div>
<script>
async function tick(){
  try{
    const html = await window.pywebview.api.render();
    document.getElementById('app').innerHTML = html;
  }catch(e){ /* mid-refresh; try again next tick */ }
}
window.addEventListener('pywebviewready', ()=>{ setInterval(tick, __REFRESH_MS__); tick(); });
</script>
</body></html>"""


def _esc(s) -> str:
    return html.escape(str(s if s is not None else ""))


def render_app(view: dict) -> str:
    """Render the six sections to the #app inner HTML. Pure function of the view model; called every
    tick from JS via the exposed api.render(). No control affordances are emitted — display only."""
    running = view["running"]
    counts = f'{len(running)} running &middot; {view["queued"]} queued &middot; {view["done"]} done'

    parts = []
    # 1. title bar
    parts.append(
        '<div class="title"><span class="dot"></span><span class="name">Aporia swarm</span>'
        f'<span class="counts">{counts}</span><span class="pin">&#128204;</span></div>')

    # 2. "Needs your yes" band (purple) — collapses entirely when empty
    if view["needs_yes"]:
        rows = []
        for item in view["needs_yes"]:
            label = _esc(item.get("title") or item.get("task_id"))
            rows.append(f'<div class="yes-row"><span class="g">{GLYPH_NEEDS}</span>'
                        f'<span class="mono">{_esc(item.get("task_id"))}</span>'
                        f'<span>{label}</span></div>')
        parts.append('<div class="section band-yes">'
                     f'<h2>Needs your yes ({len(view["needs_yes"])})</h2>' + "".join(rows) + '</div>')

    # 3. running workers — gray hollow-ring markers, task id (mono), model, elapsed
    parts.append('<div class="section"><h2>Running</h2>')
    if running:
        for w in running:
            parts.append(
                '<div class="worker">'
                f'<span class="ring">{GLYPH_RUNNING}</span>'
                f'<span class="mono">{_esc(w.get("task_id"))}</span>'
                f'<span class="model">{_esc(w.get("model") or "?")}</span>'
                f'<span class="elapsed">{_esc(w.get("elapsed"))}</span></div>')
    else:
        parts.append('<div class="empty">no workers running</div>')
    parts.append('</div>')

    # 4. recent gates — blue pass chips / red fail chips (fail names its escalation)
    parts.append('<div class="section"><h2>Recent gates</h2>')
    if view["recent_gates"]:
        chips = []
        for c in view["recent_gates"]:
            glyph = GLYPH_PASS if c["state"] == "pass" else GLYPH_FAIL
            chips.append(
                f'<span class="chip {c["state"]}"><span class="g">{glyph}</span>'
                f'<span class="mono">{_esc(c.get("task_id"))}</span>'
                f'<span>{_esc(c.get("detail"))}</span></span>')
        parts.append('<div class="chips">' + "".join(chips) + '</div>')
    else:
        parts.append('<div class="empty">no gate results yet</div>')
    parts.append('</div>')

    # 5. vitals strip — open questions / subscription / open conflicts
    v = view["vitals"]
    parts.append(
        '<div class="section"><h2>Vitals</h2><div class="vitals">'
        f'<span>Open questions <b>{_esc(v["open_questions"])}</b></span>'
        f'<span>Subscription <b>{_esc(v["subscription"])}</b></span>'
        f'<span>Open conflicts <b>{_esc(v["open_conflicts"])}</b></span>'
        '</div></div>')

    # 6. footer — display-only disclaimer + last-update age
    parts.append(
        '<div class="footer"><span>Display only &mdash; approvals happen in your sessions</span>'
        f'<span class="age">updated {_esc(_elapsed_str(view["updated"]) or "0s")} ago</span></div>')

    return "".join(parts)


# ---------------------------------------------------------------------------
# js bridge — the ONLY thing exposed to the page is a read-only render(). No write/approve/launch/cancel
# method exists on this object, so the page has NO path to mutate any estate file (display-only law).
# ---------------------------------------------------------------------------
class HudApi:
    """The window's JS bridge. render() re-reads the files and returns fresh HTML — that is the whole
    surface. There is deliberately NO method that writes, approves, launches, or cancels anything."""

    def render(self) -> str:
        try:
            return render_app(build_view())
        except Exception as e:                                # a bad frame must not kill the window
            return f'<div class="empty">HUD read error (retrying): {_esc(e)}</div>'


def main() -> int:
    start_db_poller()                                         # live open-conflicts vitals (read-only DB)
    api = HudApi()
    page = (PAGE_TEMPLATE
            .replace("__RENDER__", render_app(build_view()))
            .replace("__REFRESH_MS__", str(int(REFRESH_SECONDS * 1000))))
    webview.create_window("Aporia swarm", html=page, js_api=api,
                          frameless=True, on_top=True, easy_drag=True,
                          width=400, height=460, x=40, y=40,
                          background_color="#12141a")
    webview.start()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())