feat(batch): batch watch — live read-only view of a running shift

agents/conductors/batch/AGENTS.md

@@ -243,6 +243,20 @@ writer. Every refusal is final for the shift and named in the digest
 (`awaiting /prm — close-out refused: <why>`). Same-model close-out: AUTONOMY
 "Stage-2 adversary amendment — 2026-11-04".
 
+**Watching a shift live (`batch watch`, PyAutoBrain#575).** `batch watch
+--record <brief> [--once] [--interval N]` (`_watch.py`) draws a running shift
+in the terminal: SHADOW or LIVE, time left until `review-at:`, STOP and
+`digest.posted`, each task's state with its pick reason, PR, CI health and
+re-dispatch count, the budgets (live workers against `max-workers:`, sessions
+against `max-sessions:`, tokens, merges against `merge-cap:`, crashes) and the last five ticks. It redraws every `--interval` seconds (at
+least 30); `--once` prints one snapshot. Task wording is `_digest.task_state`,
+the digest's own, plus the states only a held lease shows (`running`, `blocked
+on a person`, `crashed — awaiting re-dispatch`). It is **read-only**: no write
+to the slot, Mind or GitHub, no lock — safe beside the timer. Its reads are the
+watcher's own: one `claude agents --json` per refresh, only on a live shift
+holding real worker leases, and `gh pr checks` for unmerged PRs, cached for the
+interval.
+
 **Workers start in a trusted folder.** `claude --bg` refuses an untrusted one
 at once ("Workspace not trusted", exit 1) — the spawn fails and leaves no lease.
 The watcher launches from the workspace root; task worktrees under the
@@ -272,6 +286,9 @@ pyauto-brain batch collect --fetch               # laptop: `gh pr view` per PR
 pyauto-brain batch collect --integration         # laptop: merge every member's head into one worktree root
 pyauto-brain batch collect --evidence ev.json --apply  # the one form that writes
 pyauto-brain batch collect --out report.md --json
+
+pyauto-brain batch watch --record <brief>         # live view of a running shift; read-only
+pyauto-brain batch watch --record <brief> --once  # one snapshot
 ```
 
 **One board: the development one.** The science board this conductor once

agents/conductors/batch/_batch.py

@@ -2777,7 +2777,7 @@ def organ_for(kind: str, root: Path, slot: str) -> dict:
 
 def main(argv=None) -> int:
     ap = argparse.ArgumentParser(prog="batch")
-    ap.add_argument("verb", choices=["plan", "collect", "brief", "digest"], nargs="?",
+    ap.add_argument("verb", choices=["plan", "collect", "brief", "digest", "watch"], nargs="?",
                     default="plan")
     ap.add_argument("--mind", type=Path, default=BODY_MAP_PATH.parent)
     ap.add_argument("--budget", type=int, default=DEFAULT_REVIEW_BUDGET,
@@ -2797,11 +2797,16 @@ def main(argv=None) -> int:
                          "first, then its backlog policy)")
     # digest
     ap.add_argument("--record", default="", metavar="FILE",
-                    help="digest: print the digest of this shift brief record "
-                         "(read-only — the watcher alone writes and posts it)")
+                    help="digest / watch: the shift brief record (read-only — "
+                         "the watcher alone writes and posts the digest)")
     ap.add_argument("--home", type=Path, default=None,
-                    help="digest: the watcher's state directory (default: "
-                         "$PYAUTO_ORCHESTRATOR_HOME, else .orchestrator)")
+                    help="digest / watch: the watcher's state directory "
+                         "(default: $PYAUTO_ORCHESTRATOR_HOME, else .orchestrator)")
+    # watch
+    ap.add_argument("--once", action="store_true",
+                    help="watch: print one snapshot and exit")
+    ap.add_argument("--interval", type=float, default=30.0,
+                    help="watch: seconds between refreshes (at least 30)")
     # collect only
     ap.add_argument("--slot", default="",
                     help="the batch record to collect (default: the newest)")
@@ -2837,6 +2842,8 @@ def main(argv=None) -> int:
             return _main_brief(mind, a)
         if a.verb == "digest":
             return _main_digest(mind, a)
+        if a.verb == "watch":
+            return _main_watch(a)
         return _main_plan(mind, a, lane)
     except BatchUsageError as e:
         print(e, file=sys.stderr)
@@ -2864,6 +2871,22 @@ def _main_digest(mind: Path, a) -> int:
     return RC_OK
 
 
+def _main_watch(a) -> int:
+    """`batch watch --record <file> [--once] [--interval N]` — a live view of a
+    running shift; writes nothing anywhere (`_watch`)."""
+    if not a.record:
+        print("batch watch: --record <file> is required", file=sys.stderr)
+        return RC_USAGE
+    import _orchestrator
+    import _watch
+    record = Path(a.record)
+    if not record.is_file():
+        print(f"batch watch: no such brief record {record}", file=sys.stderr)
+        return RC_USAGE
+    return _watch.watch(record, _orchestrator.slot_dir(record, a.home),
+                        interval=a.interval, once=a.once)
+
+
 def _main_brief(mind: Path, a) -> int:
     """`batch brief --check <file>` — read-only; exit 0 valid, 1 refused."""
     if not a.check:

agents/conductors/batch/_digest.py

@@ -90,18 +90,28 @@ def pr_health(url: str, run=None) -> str:
 
 # --- build ---------------------------------------------------------------------
 
-def _line(task: str, counters: dict, tier: str, health) -> str:
+def task_state(task: str, counters: dict, health, *, live: bool = False,
+               lease: dict | None = None, agents=None) -> tuple[str, bool]:
+    """(what a task is doing, whether its tier follows) — the one wording the
+    digest and `batch watch` share, so the live view and the digest never
+    disagree. Only the live view passes `live=True` (with the task's `lease`
+    and the `agents` view): at `review-at:` the watcher has already reaped
+    every lease, and the digest's wording stays exactly as it was."""
     out = counters.get("outcomes", {}).get(task)
     if counters.get("picks", {}).get(task, {}).get("proposed"):
-        return f"- {task} — {PROPOSED} — {tier}"
+        return PROPOSED, True
     if task in counters.get("parked", []):
-        return f"- {task} — parked: awaiting-input (blocked on a person)"
+        return "parked: awaiting-input (blocked on a person)", False
     if task in counters.get("abandoned", []):
-        return f"- {task} — abandoned: crashed twice, not re-dispatched"
+        return "abandoned: crashed twice, not re-dispatched", False
     if out is None:
-        return f"- {task} — not started"
+        if live and lease is not None:
+            return _lease_state(lease, agents), True
+        if live and task in counters.get("pending_redispatch", []):
+            return "crashed — awaiting re-dispatch", True
+        return "not started", False
     if not out["prs"]:
-        return f"- {task} — finished, no PR or issue named in its .done"
+        return "finished, no PR or issue named in its .done", False
     parts = []
     closeouts = counters.get("closeouts", {})
     for url in out["prs"]:
@@ -116,7 +126,28 @@ def _line(task: str, counters: dict, tier: str, health) -> str:
         else:
             state = health(url)
         parts.append(f"{url} — {state}")
-    return f"- {task} — " + "; ".join(parts) + f" — {tier}"
+    return "; ".join(parts), True
+
+
+def _lease_state(lease: dict, agents) -> str:
+    """A held lease, as the live view words it (the digest never sees one)."""
+    since = lease.get("started") or lease.get("heartbeat") or "?"
+    if lease.get("stalled_since"):
+        return f"blocked on a person since {lease['stalled_since']}"
+    agent = lease.get("agent_id")
+    if not agent:
+        return f"running (pid {lease.get('pid', '?')}, since {since})"
+    state = agents.state(agent) if agents is not None else None
+    if state == "blocked":
+        return f"blocked on a person (agent {agent})"
+    if state is None:
+        return f"running? (agent {agent} not listed — inside the grace for a late .done)"
+    return f"running (agent {agent}, {state}, since {since})"
+
+
+def _line(task: str, counters: dict, tier: str, health) -> str:
+    state, tiered = task_state(task, counters, health)
+    return f"- {task} — {state}" + (f" — {tier}" if tiered else "")
 
 
 def _packet(task: str, slot: Path, counters: dict) -> str:

agents/conductors/batch/_watch.py

@@ -0,0 +1,193 @@
+"""`batch watch` — a live terminal view of a running shift (PyAutoBrain#575).
+
+The shift watcher (`_orchestrator`) keeps a shift's state in its slot directory
+— `counters.json`, the lease files, `tick.log` — and says nothing until the
+digest at `review-at:`. This verb draws that state while the shift runs: time
+left, each task's state and pick reason, PR / CI health, the budgets, and the
+last few ticks.
+
+It is strictly **read-only**: it writes nothing to the slot, to Mind or to
+GitHub, takes no lock and leaves no comment, so it is safe beside the ticking
+timer. Its only reads beyond local files are the two the watcher itself is
+allowed (AUTONOMY.md "The watcher's network surface — 2026-11-03"): one
+`claude agents --json` per refresh, only on a live shift holding real worker
+leases, and `gh pr checks` for PRs not yet merged, cached for the interval.
+
+Task wording comes from `_digest.task_state`, the digest's own, so the live
+view and the digest never disagree.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import json
+import sys
+import time
+from pathlib import Path
+
+import _brief  # noqa: E402 — the batch conductor puts this directory on sys.path
+import _digest  # noqa: E402
+import _orchestrator  # noqa: E402
+
+#: The refresh interval floor, in seconds: `gh pr checks` is not hammered.
+MIN_INTERVAL_S = 30
+TICK_LINES = 5
+NO_TICK = "no tick yet — the watcher has not run for this brief"
+
+
+class CachedHealth:
+    """`_digest.pr_health`, remembered per PR for `ttl_s` seconds."""
+
+    def __init__(self, ttl_s: float, health=None, clock=time.monotonic):
+        self._ttl, self._health, self._clock = ttl_s, health or _digest.pr_health, clock
+        self._seen: dict[str, tuple[float, str]] = {}
+
+    def __call__(self, url: str) -> str:
+        hit = self._seen.get(url)
+        if hit and self._clock() - hit[0] < self._ttl:
+            return hit[1]
+        verdict = self._health(url)
+        self._seen[url] = (self._clock(), verdict)
+        return verdict
+
+
+def _value(block: dict | None, key: str, default="?"):
+    return (block or {}).get(key, {}).get("value", default)
+
+
+def _ticks(slot: Path, n: int) -> list[dict]:
+    try:
+        lines = (slot / _orchestrator.LOG_FILE).read_text(encoding="utf-8").splitlines()
+    except OSError:
+        return []
+    out = []
+    for line in lines[-n:]:
+        try:
+            out.append(json.loads(line))
+        except ValueError:
+            out.append({"action": "unreadable", "raw": line[:80]})
+    return out
+
+
+def _time_left(now: _dt.datetime, review_at: _dt.datetime | None) -> str:
+    if review_at is None:
+        return "no review-at"
+    secs = int((review_at - now).total_seconds())
+    if secs <= 0:
+        return "closed — review-at has passed"
+    h, m = divmod(secs // 60, 60)
+    return (f"{h}h {m:02d}m" if h else f"{m}m") + " left"
+
+
+def snapshot(brief_path: Path, slot: Path, now: _dt.datetime, *, agents=None,
+             health=None, tick_lines: int = TICK_LINES) -> dict:
+    """Everything one refresh draws, as a dict. Reads only."""
+    rec = _brief.parse(Path(brief_path).read_text(encoding="utf-8"))
+    top, brief = rec["top"], rec["brief"] or {}
+    shadow = _brief.is_shadow(brief)
+    try:
+        review_at = _brief._iso(_value(top, "review-at", ""))
+    except (ValueError, TypeError):
+        review_at = None
+
+    started = slot.is_dir()
+    counters = _orchestrator.read_counters(slot) if started else {}
+    held = _orchestrator.leases(slot) if started else {}
+    picks = counters.get("picks", {})
+    named = [Path(p).stem for p in _brief.tasks_of(brief)]
+    tasks = list(dict.fromkeys(named + list(picks) + list(held)))
+
+    if agents is None and not shadow and any(l.get("agent_id") for l in held.values()):
+        agents = _orchestrator.AgentsView()  # one `claude agents --json` per refresh
+    health = health or _digest.pr_health
+
+    rows = []
+    for task in tasks:
+        state, _ = _digest.task_state(task, counters, health, live=True,
+                                      lease=held.get(task), agents=agents)
+        rows.append({"task": task, "state": state,
+                     "tier": picks.get(task, {}).get("tier", "?"),
+                     "why": "named in the brief" if task in named
+                            else picks.get(task, {}).get("why") or "NOT RECORDED",
+                     "redispatched": counters.get("redispatched", {}).get(task, 0)})
+
+    values = _digest.counter_values(brief, counters)
+    merges = values["merges"]
+    return {
+        "slot": slot.name,
+        "brief": str(brief_path),
+        "mode": "SHADOW" if shadow else "LIVE",
+        "now": _orchestrator._iso(now),
+        "dispatched": _value(top, "dispatched"),
+        "review_at": _value(top, "review-at"),
+        "time_left": _time_left(now, review_at),
+        "stop": (slot / _orchestrator.STOP_FILE).exists(),
+        "digest_posted": (slot / _orchestrator.DIGEST_MARKER).exists(),
+        "started": started,
+        "tasks": rows,
+        "budgets": {
+            "workers": f"{len(held)} live of {_brief.cap_of(brief, 'max-workers')}",
+            "sessions": f"{counters.get('sessions', 0)} of {_brief.cap_of(brief, 'max-sessions')}",
+            "tokens": values["tokens"],
+            "merges": f"{merges} of {_value(brief, 'merge-cap')}"
+                      f" ({values['glance-auto-merges']} unattended)",
+            "crashes": counters.get("crashes", 0),
+        },
+        "ticks": _ticks(slot, tick_lines) if started else [],
+    }
+
+
+def _tick_line(t: dict) -> str:
+    at = (t.get("at") or "?")[11:16] + "Z" if t.get("at") else "?"
+    if t.get("reason"):
+        return f"  {at} {t.get('action', '?')} — {t['reason']}"
+    counts = ", ".join(f"{k} {len(t[k])}" for k in ("proposed", "spawned", "reaped", "refused")
+                       if isinstance(t.get(k), list))
+    return f"  {at} {t.get('action', '?')}" + (f" — {counts}" if counts else "")
+
+
+def render(snap: dict) -> str:
+    flags = [snap["mode"], snap["time_left"]]
+    if snap["stop"]:
+        flags.append("STOP present — the watcher does nothing")
+    if snap["digest_posted"]:
+        flags.append("digest posted")
+    lines = [f"Shift {snap['slot']} — " + " · ".join(flags),
+             f"  {snap['dispatched']} → {snap['review_at']}   (as of {snap['now']})",
+             f"  brief: {snap['brief']}", ""]
+    if not snap["started"]:
+        return "\n".join(lines + [NO_TICK]) + "\n"
+    lines.append("Tasks")
+    lines += [f"- {r['task']} — {r['state']} — {r['tier']} — why: {r['why']}"
+              + (f" — re-dispatched {r['redispatched']}×" if r["redispatched"] else "")
+              for r in snap["tasks"]] or ["- none yet"]
+    b = snap["budgets"]
+    lines += ["", "Budgets",
+              f"- workers: {b['workers']}", f"- sessions: {b['sessions']}",
+              f"- tokens: {b['tokens']}",
+              f"- merges: {b['merges']}", f"- workers crashed: {b['crashes']}",
+              "", "Last ticks"]
+    lines += [_tick_line(t) for t in snap["ticks"]] or ["  none"]
+    return "\n".join(lines) + "\n"
+
+
+def watch(brief_path: Path, slot: Path, *, interval: float = MIN_INTERVAL_S,
+          once: bool = False, out=None, sleep=time.sleep, clock=None,
+          agents=None, health=None) -> int:
+    """Draw the shift; with `once`, one snapshot and return. Ctrl-C ends it."""
+    out = out or sys.stdout
+    clock = clock or (lambda: _dt.datetime.now(_dt.timezone.utc))
+    interval = max(float(interval), MIN_INTERVAL_S)
+    health = health or CachedHealth(interval)
+    try:
+        while True:
+            text = render(snapshot(brief_path, slot, clock(), agents=agents, health=health))
+            if once:
+                out.write(text)
+                return 0
+            out.write("\033[H\033[2J" + text + f"\n(refresh every {int(interval)} s — Ctrl-C to stop)\n")
+            out.flush()
+            sleep(interval)
+    except KeyboardInterrupt:
+        out.write("\n")
+        return 0