diff --git a/agents/conductors/batch/AGENTS.md b/agents/conductors/batch/AGENTS.md index 01f6089..f59eb2a 100644 --- a/agents/conductors/batch/AGENTS.md +++ b/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: `). Same-model close-out: AUTONOMY "Stage-2 adversary amendment — 2026-11-04". +**Watching a shift live (`batch watch`, PyAutoBrain#575).** `batch watch +--record [--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 # live view of a running shift; read-only +pyauto-brain batch watch --record --once # one snapshot ``` **One board: the development one.** The science board this conductor once diff --git a/agents/conductors/batch/_batch.py b/agents/conductors/batch/_batch.py index 965978e..63e5769 100644 --- a/agents/conductors/batch/_batch.py +++ b/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 [--once] [--interval N]` — a live view of a + running shift; writes nothing anywhere (`_watch`).""" + if not a.record: + print("batch watch: --record 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 ` — read-only; exit 0 valid, 1 refused.""" if not a.check: diff --git a/agents/conductors/batch/_digest.py b/agents/conductors/batch/_digest.py index 881b36a..1ada1bb 100644 --- a/agents/conductors/batch/_digest.py +++ b/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: diff --git a/agents/conductors/batch/_watch.py b/agents/conductors/batch/_watch.py new file mode 100644 index 0000000..c7b9c93 --- /dev/null +++ b/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 diff --git a/tests/test_batch_watch.py b/tests/test_batch_watch.py new file mode 100644 index 0000000..5a4e1bf --- /dev/null +++ b/tests/test_batch_watch.py @@ -0,0 +1,235 @@ +"""`batch watch` (PyAutoBrain#575): a live, read-only view of a running shift. + +One group per clause of the prompt's `Witness:`: the view shows every task's +state, pick reason and PR / CI health plus the shift's budgets; `--once` prints +one snapshot; it writes nothing (the slot is byte-identical afterwards); and its +task wording is the digest's own. +""" + +from __future__ import annotations + +import datetime as _dt +import hashlib +import io +import json +import subprocess +import sys +from pathlib import Path + +BRAIN = Path(__file__).resolve().parents[1] +BATCH = BRAIN / "agents" / "conductors" / "batch" +sys.path.insert(0, str(BRAIN / "agents" / "faculties" / "sizing")) +sys.path.insert(0, str(BATCH)) +import _brief # noqa: E402 +import _digest # noqa: E402 +import _orchestrator as orch # noqa: E402 +import _watch # noqa: E402 + +NOW = _dt.datetime(2026, 11, 4, 9, 30, tzinfo=_dt.timezone.utc) +PR_OPEN = "https://github.com/example-org/example-repo/pull/11" +PR_CLOSING = "https://github.com/example-org/example-repo/pull/12" +PR_MERGED = "https://github.com/example-org/example-repo/pull/13" + +BRIEF = """\ +# Batch 2026-11-04 am +- dispatched: 2026-11-04T09:00Z +- review-at: 2026-11-04T11:00Z +- issue: https://github.com/example-org/example-mind/issues/9999 +- brief: + - goals: + - 1. live shift + - tasks: + - draft/feature/pyautobrain/named.md + - backlog: scoped + - scoped: + - tiers: notify, glance + - review-minutes: 30 + - lanes: any local-dev + - merge-cap: 2 + - token-budget: 5000 + - max-workers: 3 +{extra}""" + + +class Agents: + def __init__(self, states): + self.states, self.calls = states, 0 + + def state(self, agent_id): + self.calls += 1 + return self.states.get(agent_id) + + +def _health(url): + return {PR_OPEN: "RUNNING"}.get(url, "FAILED") + + +def _record(tmp_path, extra=""): + brief = tmp_path / "2026-11-04-am.md" + brief.write_text(BRIEF.format(extra=extra)) + return brief + + +def _live_slot(tmp_path) -> Path: + slot = tmp_path / "home" / "2026-11-04-am" + (slot / "leases").mkdir(parents=True) + pick = lambda why, tier="glance": {"prompt": "draft/x.md", "why": why, "tier": tier} # noqa: E731 + counters = orch.read_counters(slot) + counters.update({ + "sessions": 4, "tokens": 1234, "crashes": 1, "glance_auto_merges": 1, + "picks": {"named": pick("named"), "running": pick("scoped tier glance"), + "blocked": pick("scoped tier glance"), "pr_open": pick("scoped tier notify", "notify"), + "closing": pick("scoped tier glance"), "merged": pick("scoped tier glance"), + "parked": pick("scoped tier glance"), "crashed": pick("scoped tier glance")}, + "outcomes": {"pr_open": {"prs": [PR_OPEN], "merged": []}, + "closing": {"prs": [PR_CLOSING], "merged": []}, + "merged": {"prs": [PR_MERGED], "merged": [PR_MERGED]}}, + "closeouts": {PR_CLOSING: {"task": "closing", "state": "running"}, + PR_MERGED: {"task": "merged", "state": "merged", "model": "opus"}}, + "parked": ["parked"], "pending_redispatch": ["crashed"], + "redispatched": {"running": 1}, + }) + (slot / orch.COUNTERS_FILE).write_text(json.dumps(counters)) + (slot / "leases" / "running.json").write_text(json.dumps( + {"task": "running", "agent_id": "a1", "started": "2026-11-04T09:05:00Z"})) + (slot / "leases" / "blocked.json").write_text(json.dumps( + {"task": "blocked", "agent_id": "a2", "started": "2026-11-04T09:06:00Z", + "stalled_since": "2026-11-04T09:20:00Z"})) + (slot / orch.LOG_FILE).write_text("\n".join(json.dumps( + {"action": "dispatch", "at": f"2026-11-04T09:{m:02d}:00Z", "spawned": ["running"], + "proposed": [], "reaped": [], "refused": []}) for m in range(0, 30, 3)) + "\n") + return slot + + +def _tree_digest(root: Path) -> dict: + return {str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in sorted(root.rglob("*")) if p.is_file()} + + +# --- what it shows ------------------------------------------------------------ + +def test_live_snapshot_shows_every_task_state_reason_health_and_budget(tmp_path): + brief, slot = _record(tmp_path), _live_slot(tmp_path) + agents = Agents({"a1": "working", "a2": "blocked"}) + text = _watch.render(_watch.snapshot(brief, slot, NOW, agents=agents, health=_health)) + assert "Shift 2026-11-04-am — LIVE · 1h 30m left" in text + assert "- named — not started — glance — why: named in the brief" in text + assert "- running — running (agent a1, working, since 2026-11-04T09:05:00Z)" in text + assert "re-dispatched 1×" in text + assert "- blocked — blocked on a person since 2026-11-04T09:20:00Z" in text + assert f"- pr_open — {PR_OPEN} — RUNNING — notify — why: scoped tier notify" in text + assert f"{PR_CLOSING} — close-out running" in text + assert f"{PR_MERGED} — merged unattended (close-out, opus)" in text + assert "- parked — parked: awaiting-input" in text + assert "- crashed — crashed — awaiting re-dispatch" in text + assert "- workers: 2 live of 3" in text + assert "- sessions: 4 of 6" in text # the max-sessions default + assert "- tokens: 1234 of 5000" in text + assert "- merges: 1 of 2 (1 unattended)" in text + assert "- workers crashed: 1" in text + # The last five ticks only, oldest first. + assert text.count("dispatch — proposed 0, spawned 1") == 5 + assert "09:27Z dispatch" in text and "09:12Z" not in text + + +def test_shadow_snapshot_reads_like_the_shadow_digest(tmp_path): + brief = _record(tmp_path, extra=" - shadow: yes\n") + slot = tmp_path / "home" / "2026-11-04-am" + slot.mkdir(parents=True) + counters = orch.read_counters(slot) + counters["picks"] = {"a": {"prompt": "draft/a.md", "why": "scoped tier notify", + "tier": "notify", "proposed": True}} + (slot / orch.COUNTERS_FILE).write_text(json.dumps(counters)) + text = _watch.render(_watch.snapshot(brief, slot, NOW, health=_health)) + assert "SHADOW" in text + assert "- a — proposed (shadow) — notify — why: scoped tier notify" in text + assert "Last ticks\n none" in text + + +def test_no_tick_yet_and_closed_and_stop(tmp_path): + brief = _record(tmp_path) + slot = tmp_path / "home" / "2026-11-04-am" + text = _watch.render(_watch.snapshot(brief, slot, NOW, health=_health)) + assert _watch.NO_TICK in text + assert not slot.exists() # looking never creates the slot + + slot = _live_slot(tmp_path) + (slot / orch.STOP_FILE).write_text("") + (slot / orch.DIGEST_MARKER).write_text("") + late = NOW + _dt.timedelta(hours=2) + text = _watch.render(_watch.snapshot(brief, slot, late, agents=Agents({}), health=_health)) + assert "closed — review-at has passed" in text + assert "STOP present" in text and "digest posted" in text + + +def test_time_left_formats(): + at = NOW + _dt.timedelta(minutes=7) + assert _watch._time_left(NOW, at) == "7m left" + assert _watch._time_left(NOW, at + _dt.timedelta(hours=1)) == "1h 07m left" + assert _watch._time_left(at, NOW).startswith("closed") + + +# --- read-only --------------------------------------------------------------- + +def test_once_writes_nothing_to_the_slot(tmp_path): + brief, slot = _record(tmp_path), _live_slot(tmp_path) + before = _tree_digest(tmp_path) + out = io.StringIO() + rc = _watch.watch(brief, slot, once=True, out=out, clock=lambda: NOW, + agents=Agents({"a1": "working", "a2": "working"}), health=_health) + assert rc == 0 and "Tasks" in out.getvalue() + assert _tree_digest(tmp_path) == before + + +def test_cli_once_through_pyauto_brain(tmp_path): + brief = _record(tmp_path, extra=" - shadow: yes\n") + home = tmp_path / "home" + res = subprocess.run([sys.executable, str(BATCH / "_batch.py"), "watch", + "--record", str(brief), "--home", str(home), "--once"], + capture_output=True, text=True) + assert res.returncode == 0, res.stderr + assert "SHADOW" in res.stdout and _watch.NO_TICK in res.stdout + assert not home.exists() + res = subprocess.run([sys.executable, str(BATCH / "_batch.py"), "watch"], + capture_output=True, text=True) + assert res.returncode == 2 and "--record" in res.stderr + + +# --- shared wording ------------------------------------------------------------ + +def test_digest_and_watch_word_every_task_the_same(tmp_path): + """With no lease held (as at review-at), each watch line carries the digest + line's state verbatim.""" + brief, slot = _record(tmp_path), _live_slot(tmp_path) + for f in (slot / "leases").glob("*.json"): + f.unlink() + counters = json.loads((slot / orch.COUNTERS_FILE).read_text()) + counters["pending_redispatch"] = [] + (slot / orch.COUNTERS_FILE).write_text(json.dumps(counters)) + block = _brief.parse(brief.read_text())["brief"] + digest = _digest.build(block, slot, orch.read_counters(slot), health=_health) + snap = _watch.snapshot(brief, slot, NOW, health=_health) + for row in snap["tasks"]: + assert f"- {row['task']} — {row['state']}" in digest, row + + +def test_refresh_interval_floor_and_cached_health(): + calls = [] + t = [0.0] + h = _watch.CachedHealth(30, health=lambda u: calls.append(u) or "RUNNING", clock=lambda: t[0]) + h(PR_OPEN), h(PR_OPEN) + t[0] = 31 + h(PR_OPEN) + assert calls == [PR_OPEN, PR_OPEN] + + slept = [] + + def sleep(s): + slept.append(s) + raise KeyboardInterrupt + + tmp = Path(__import__("tempfile").mkdtemp()) + brief = _record(tmp) + assert _watch.watch(brief, tmp / "none", interval=1, out=io.StringIO(), sleep=sleep, + clock=lambda: NOW, health=_health) == 0 + assert slept == [_watch.MIN_INTERVAL_S]