diff --git a/AUTONOMY.md b/AUTONOMY.md index 71cc038..0813d34 100644 --- a/AUTONOMY.md +++ b/AUTONOMY.md @@ -614,6 +614,30 @@ and a second push in one shift is itself a digest finding. *Revert condition:* two pushes in one shift that the human judged unnecessary retire push for that orchestrator. +## The watcher's network surface — 2026-11-03 + +Approved by the human 2026-11-03 with the phase-3a plan of the +autonomous-orchestrators epic. The hard invariant below says any further +network surface in the Brain is a new decision; this is that decision for the +shift watcher (`bin/orchestrator_watch.py`, `agents/conductors/batch/_orchestrator.py`; +the exemption it runs under is `PyAutoMind/policy/shift_watcher.md`). + +- **What it may reach, and nothing else:** read-only `gh pr checks` for PRs its + own leases name; read-only `claude agents --json` for the workers it started; + and **one comment per shift** on the brief's `issue:` (the digest, phase 3d). + It starts worker sessions (`claude --bg`, phase 3b) — that is dispatch, ruled + by "Shift briefs and unattended `glance`", not a network read. +- **Opt-in and laptop-only.** It runs only from a systemd user timer the human + installs, only while a brief is open, and never while `STOP` exists. No + conductor, faculty or skill calls it; no session arms it. +- **Never:** a merge, a push, a label, a PR, a review, a second comment in the + same shift, or any write to a repository. Merges stay with the `/prm` + close-out the section above grants. + +*Revert condition:* one network call outside the three named above — or a +second issue comment in one shift — retires the watcher to fake mode until the +cause is found and this section is re-approved. + ## Human override for Heart RED (development only) Heart RED still parks every autonomous run and blocks every release. After the @@ -871,6 +895,12 @@ Codex hook list. name instead — and every pushed ref carries a `sweep-after:` date on the batch record, after which `bin/branch_sweep.sh` may delete it. + A third exception, added 2026-11-03: the **shift watcher** — read-only + `gh pr checks` and `claude agents --json`, and one issue comment per shift — + opt-in through a human-installed timer and laptop-only; the full terms and + their revert condition are "The watcher's network surface — 2026-11-03" + above. + Any *further* network surface in the Brain is a new decision, not covered by this line. - The `Autonomy:` header is a model's own estimate. The caps, the explicit @@ -893,5 +923,7 @@ Codex hook list. (2026-11-03): the brief as the launch act, the backlog-dip scope, the unattended `glance` close-out guards and the merge cap, one Mind writer per shift, drafts-only decomposition, the push test. +- `batch` (the watcher) — "The watcher's network surface" (2026-11-03): what + the watcher may reach, and that it is opt-in, laptop-only and never writes. Skills must link here rather than copying the tables. diff --git a/agents/conductors/batch/AGENTS.md b/agents/conductors/batch/AGENTS.md index 3d8ca03..c18938f 100644 --- a/agents/conductors/batch/AGENTS.md +++ b/agents/conductors/batch/AGENTS.md @@ -146,6 +146,26 @@ never the *authority*. A shift brief written with `review-at:` is that act in writing (2026-11-03): the human dispatches by writing it, and its scope is the whole grant (`AUTONOMY.md`, "Shift briefs and unattended `glance`"). +## The watcher (phase 3a — fake workers only) + +`bin/orchestrator_watch.py --brief --mind ` runs **one +tick**: `STOP` first, then the open-brief gate, the tick lock, the lease reap, +the plan inside the brief, the caps (`max-workers:`, default 2; +`max-sessions:`, default 6) and an exclusive-create lease per spawn. State lives +under `$PYAUTO_ORCHESTRATOR_HOME` (default `.orchestrator` under the worktree +root), one directory per brief: `STOP`, `leases/`, `counters.json`, `tick.log`. +Until phase 3b it spawns only stub workers (`PYAUTO_ORCHESTRATOR_FAKE=1`); a real +dispatch is refused. Doctrine: `AUTONOMY.md` "The watcher's network surface — +2026-11-03"; the exemption it runs under, `PyAutoMind/policy/shift_watcher.md`. + +**Installing the timer is the human's act** — no session runs these. Copy +`bin/systemd/pyauto-orchestrator.{service,timer}` to `~/.config/systemd/user/`, +write `~/.config/pyauto/orchestrator.env` with `PYAUTO_BRAIN`, `PYAUTO_MIND`, +`PYAUTO_SHIFT_BRIEF` (and `PYAUTO_ORCHESTRATOR_FAKE=1` until 3b), then +`systemctl --user daemon-reload && systemctl --user enable --now +pyauto-orchestrator.timer`. To stop a shift, `touch /STOP`; to remove the +watcher, `systemctl --user disable --now pyauto-orchestrator.timer`. + ## Running ``` diff --git a/agents/conductors/batch/_brief.py b/agents/conductors/batch/_brief.py index 39c51a0..8588984 100644 --- a/agents/conductors/batch/_brief.py +++ b/agents/conductors/batch/_brief.py @@ -22,6 +22,8 @@ ) BACKLOG_POLICIES = ("none", "fill-idle", "scoped") +#: Optional usage caps (phase 3a) and their defaults when a brief omits them. +CAP_DEFAULTS = {"max-workers": 2, "max-sessions": 6} SCOPED_KEYS = ("epics", "tags", "tiers") _ITEM_RE = re.compile(r"^(\s*)-\s+(.*?)\s*$") @@ -121,6 +123,12 @@ def excluded_of(brief: dict) -> tuple[set[str], dict[str, list[str]]]: return paths, keyed +def cap_of(brief: dict, key: str) -> int: + """A usage cap from the brief, or its default when the brief omits it.""" + value = _nonneg_int(brief.get(key, {}).get("value", "")) + return CAP_DEFAULTS[key] if value is None else value + + def lanes_of(brief: dict) -> list[str]: node = brief.get("lanes") return _csv(node["value"]) if node else [] @@ -212,6 +220,10 @@ def check(path: Path, mind: Path) -> dict: errors.append(f"brief.{key}: not a whole number " f"({brief[key]['value']!r})") + for key in CAP_DEFAULTS: + if key in brief and _nonneg_int(brief[key]["value"]) is None: + errors.append(f"brief.{key}: not a whole number ({brief[key]['value']!r})") + lanes = lanes_of(brief) if not lanes: errors.append("brief.lanes: missing — any subset of " diff --git a/agents/conductors/batch/_orchestrator.py b/agents/conductors/batch/_orchestrator.py new file mode 100644 index 0000000..54a8da2 --- /dev/null +++ b/agents/conductors/batch/_orchestrator.py @@ -0,0 +1,375 @@ +"""The shift watcher's core — one deterministic tick, no LLM in it. + +Doctrine: `PyAutoMind/policy/shift_watcher.md` (the one exemption to "sessions +end at their deliverable") and `AUTONOMY.md` "Shift briefs and unattended +`glance`" plus "The watcher's network surface". A tick, in this order: + +1. `STOP` in the slot directory → do nothing, before reading anything else; +2. no open brief (`batch brief --check` fails, or `review-at:` has passed) → + do nothing; +3. take the tick lock (exclusive create) — an overlapping tick backs off; +4. reap leases — a finished worker is released, a dead one is counted and its + task may be dispatched again **once** per shift; +5. ask the planner what the brief admits (`plan(..., brief=...)`); +6. for each candidate: the caps, then an exclusive-create lease, then spawn. + +Phase 3a runs fake workers only (`PYAUTO_ORCHESTRATOR_FAKE=1`); real +`claude --bg` workers are phase 3b, the Mind writer 3c, the digest 3d. +Nothing here commits, merges, posts or arms a timer: the systemd timer that +calls `bin/orchestrator_watch.py` is installed by the human. +""" + +from __future__ import annotations + +import datetime as _dt +import json +import os +import subprocess +import sys +from pathlib import Path + +import _brief # noqa: E402 — the batch conductor puts this directory on sys.path + +STOP_FILE = "STOP" +LOCK_FILE = "tick.lock" +COUNTERS_FILE = "counters.json" +LOG_FILE = "tick.log" +#: A tick lock older than this belongs to a tick that died; it is broken. +LOCK_TTL_S = 600 +#: A worker whose process is gone keeps its lease this long before it is +#: declared dead — room for a worker that writes its `.done` marker late. +LEASE_GRACE_S = 120 +#: A task's lease may be re-dispatched after a crash this many times a shift. +REDISPATCH_LIMIT = 1 + +FAKE_ENV = "PYAUTO_ORCHESTRATOR_FAKE" +HOME_ENV = "PYAUTO_ORCHESTRATOR_HOME" + + +class OrchestratorError(RuntimeError): + pass + + +# --- where state lives --------------------------------------------------------- + +def orchestrator_home() -> Path: + """`$PYAUTO_ORCHESTRATOR_HOME`, else `.orchestrator` under the worktree root. + + No path is written here: the worktree root comes from the shared resolver, + so an adopting fork changes its environment, not this file. + """ + env = os.environ.get(HOME_ENV) + if env: + return Path(env) + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from _pyauto_root import pyauto_wt_root # noqa: E402 + return pyauto_wt_root() / ".orchestrator" + + +def slot_dir(brief_path: Path, home: Path | None = None) -> Path: + """One directory per shift, named after the brief record.""" + return (home or orchestrator_home()) / Path(brief_path).stem + + +def _now() -> _dt.datetime: + return _dt.datetime.now(_dt.timezone.utc) + + +def _iso(t: _dt.datetime) -> str: + return t.astimezone(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --- counters and the log -------------------------------------------------------- + +def read_counters(slot: Path) -> dict: + try: + c = json.loads((slot / COUNTERS_FILE).read_text()) + except (OSError, ValueError): + c = {} + c.setdefault("sessions", 0) + c.setdefault("crashes", 0) + c.setdefault("redispatched", {}) + c.setdefault("done", []) + c.setdefault("abandoned", []) + c.setdefault("refusals", []) + c.setdefault("pending_redispatch", []) + return c + + +def _write_counters(slot: Path, c: dict) -> None: + tmp = slot / (COUNTERS_FILE + ".tmp") + tmp.write_text(json.dumps(c, indent=2, sort_keys=True)) + os.replace(tmp, slot / COUNTERS_FILE) + + +def _log(slot: Path, now: _dt.datetime, result: dict) -> None: + with open(slot / LOG_FILE, "a", encoding="utf-8") as fh: + fh.write(json.dumps({"at": _iso(now), **result}, sort_keys=True) + "\n") + + +# --- processes ------------------------------------------------------------------ + +def _alive(pid: int) -> bool: + """Whether a worker process is still running (a zombie is not).""" + try: + done, _status = os.waitpid(pid, os.WNOHANG) + if done == pid: + return False + except ChildProcessError: + pass # not our child: another tick spawned it + except OSError: + return False + try: + os.kill(pid, 0) + except OSError: + return False + try: + state = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0] + return state != "Z" + except (OSError, IndexError): + return True + + +# The stub a fake worker runs: sleep, then write the done marker — or exit 1 +# without it, which is what a crash looks like to the watcher. +_FAKE_WORKER = ( + "import sys, time, pathlib\n" + "lease_dir, task, mode, secs = sys.argv[1:5]\n" + "time.sleep(float(secs))\n" + "if mode == 'crash':\n" + " sys.exit(1)\n" + "pathlib.Path(lease_dir, task + '.done').write_text('done')\n" +) + + +class FakeSpawner: + """Stub workers for a dry shift (`PYAUTO_ORCHESTRATOR_FAKE=1`).""" + + def __init__(self, seconds: float = 0.5, crash: set[str] | None = None): + self.seconds = seconds + self.crash = set(crash or ()) + + def spawn(self, task: str, prompt_path: str, lease_dir: Path) -> dict: + mode = "crash" if task in self.crash else "ok" + proc = subprocess.Popen( + [sys.executable, "-c", _FAKE_WORKER, str(lease_dir), task, mode, + str(self.seconds)], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + return {"pid": proc.pid, "session_id": f"fake-{proc.pid}", "worktree": None} + + +class RealSpawner: + """`claude --bg` workers from the session template — phase 3b.""" + + def spawn(self, task: str, prompt_path: str, lease_dir: Path) -> dict: + raise OrchestratorError("real dispatch arrives in phase 3b; set " + f"{FAKE_ENV}=1 for a dry shift") + + +def default_spawner(): + return FakeSpawner() if os.environ.get(FAKE_ENV) == "1" else RealSpawner() + + +# --- locks and leases ------------------------------------------------------------ + +def _create_exclusive(path: Path, payload: dict) -> bool: + """Write `payload` to `path` only if nobody has; True when this call won.""" + try: + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) + except FileExistsError: + return False + with os.fdopen(fd, "w") as fh: + json.dump(payload, fh, sort_keys=True) + return True + + +def _take_lock(slot: Path, now: _dt.datetime) -> bool: + lock = slot / LOCK_FILE + me = {"pid": os.getpid(), "at": now.timestamp()} + if _create_exclusive(lock, me): + return True + try: + held = json.loads(lock.read_text()) + stale = (now.timestamp() - float(held.get("at", 0)) > LOCK_TTL_S + or not _alive(int(held.get("pid", 0)))) + except (OSError, ValueError): + stale = False # being written right now: someone else is ticking + if stale: + try: + lock.unlink() + except FileNotFoundError: + pass + return _create_exclusive(lock, me) + return False + + +def _drop_lock(slot: Path) -> None: + try: + (slot / LOCK_FILE).unlink() + except FileNotFoundError: + pass + + +def task_id(prompt_path: str) -> str: + return Path(prompt_path).stem + + +def leases(slot: Path) -> dict[str, dict]: + out = {} + for f in sorted((slot / "leases").glob("*.json")): + try: + out[f.stem] = json.loads(f.read_text()) + except (OSError, ValueError): + continue + return out + + +def _reap(slot: Path, counters: dict, now: _dt.datetime, grace_s: float) -> list[dict]: + """Release finished leases and declare dead ones; returns what happened.""" + events = [] + lease_dir = slot / "leases" + for task, lease in leases(slot).items(): + done_marker = lease_dir / f"{task}.done" + path = lease_dir / f"{task}.json" + if done_marker.exists(): + path.unlink(missing_ok=True) + done_marker.unlink(missing_ok=True) + if task not in counters["done"]: + counters["done"].append(task) + events.append({"task": task, "event": "done"}) + continue + if _alive(int(lease.get("pid", 0))): + lease["heartbeat"] = _iso(now) + lease["expires"] = _iso(now + _dt.timedelta(seconds=grace_s)) + path.write_text(json.dumps(lease, sort_keys=True)) + continue + expires = _dt.datetime.fromisoformat(lease["expires"].replace("Z", "+00:00")) + if now < expires: + continue # gone, but inside the grace for a late `.done` + path.unlink(missing_ok=True) + counters["crashes"] += 1 + again = counters["redispatched"].get(task, 0) < REDISPATCH_LIMIT + if again and task not in counters["pending_redispatch"]: + counters["pending_redispatch"].append(task) + if not again and task not in counters["abandoned"]: + counters["abandoned"].append(task) + events.append({"task": task, "event": "dead", + "redispatch": "allowed" if again else "spent — abandoned"}) + return events + + +# --- the tick -------------------------------------------------------------------- + +def brief_open(brief_path: Path, mind: Path, now: _dt.datetime) -> tuple[bool, str, dict | None]: + """(open?, why, the parsed brief block).""" + res = _brief.check(brief_path, mind) + if not res["ok"]: + return False, "brief refused: " + "; ".join(res["errors"]), None + rec = _brief.parse(brief_path.read_text(encoding="utf-8")) + at = _brief._iso(rec["top"]["review-at"]["value"]) + if now >= at: + return False, f"review-at {rec['top']['review-at']['value']} has passed", None + return True, "open", rec["brief"] + + +def tick(brief_path: Path, mind: Path, *, home: Path | None = None, + now: _dt.datetime | None = None, spawner=None, + grace_s: float = LEASE_GRACE_S, records: list[dict] | None = None) -> dict: + """One watcher tick. Returns what it did; also appended to `tick.log`.""" + now = now or _now() + brief_path = Path(brief_path) + slot = slot_dir(brief_path, home) + + # The kill switch is the FIRST read: nothing else is touched while it exists. + if (slot / STOP_FILE).exists(): + return {"action": "stopped", "reason": f"{STOP_FILE} present"} + + is_open, why, brief = brief_open(brief_path, mind, now) + if not is_open: + return {"action": "idle", "reason": why} + + (slot / "leases").mkdir(parents=True, exist_ok=True) + if not _take_lock(slot, now): + return {"action": "busy", "reason": "another tick holds the lock"} + try: + result = _tick_locked(slot, brief, mind, now, spawner or default_spawner(), + grace_s, records) + finally: + _drop_lock(slot) + _log(slot, now, result) + return result + + +def _tick_locked(slot, brief, mind, now, spawner, grace_s, records) -> dict: + counters = read_counters(slot) + reaped = _reap(slot, counters, now, grace_s) + live = leases(slot) + max_workers = _brief.cap_of(brief, "max-workers") + max_sessions = _brief.cap_of(brief, "max-sessions") + + import _batch # noqa: E402 — deferred: _batch imports this module's sibling + if records is None: + records = _batch.survey(mind) + d = _batch.plan(records, session_lane="local-dev", brief=brief) + spawned, refused = [], [] + for member in d["members"]: + task = task_id(member["path"]) + if task in live or task in counters["done"] or task in counters["abandoned"]: + continue + if len(live) >= max_workers: + refused.append({"task": task, "why": f"max-workers {max_workers} reached"}) + continue + if counters["sessions"] >= max_sessions: + refused.append({"task": task, "why": f"max-sessions {max_sessions} reached"}) + continue + lease = {"task": task, "prompt": member["path"], "started": _iso(now), + "heartbeat": _iso(now), + "expires": _iso(now + _dt.timedelta(seconds=grace_s)), + "pid": 0, "session_id": None, "worktree": None} + lease_path = slot / "leases" / f"{task}.json" + if not _create_exclusive(lease_path, lease): + continue # a lease appeared since we read them: never a second worker + try: + lease.update(spawner.spawn(task, member["path"], slot / "leases")) + except Exception: + lease_path.unlink(missing_ok=True) + raise + lease_path.write_text(json.dumps(lease, sort_keys=True)) + if task in counters["pending_redispatch"]: + counters["pending_redispatch"].remove(task) + counters["redispatched"][task] = counters["redispatched"].get(task, 0) + 1 + counters["sessions"] += 1 + live[task] = lease + spawned.append({"task": task, "pid": lease["pid"]}) + counters["refusals"] += [dict(r, at=_iso(now)) for r in refused] + _write_counters(slot, counters) + return {"action": "ticked", "reaped": reaped, "spawned": spawned, + "refused": refused, "live": sorted(live)} + + +def main(argv=None) -> int: + import argparse + ap = argparse.ArgumentParser( + prog="orchestrator_watch", + description="One shift-watcher tick (run by a systemd user timer the " + "human installs). Fake workers only until phase 3b.") + ap.add_argument("--brief", required=True, type=Path, + help="the shift brief record (PyAutoMind/batches/.md)") + ap.add_argument("--mind", required=True, type=Path) + ap.add_argument("--home", type=Path, default=None, + help=f"state directory (default ${HOME_ENV}, else " + ".orchestrator under the worktree root)") + ap.add_argument("--json", action="store_true", dest="as_json") + a = ap.parse_args(argv) + try: + res = tick(a.brief, a.mind.resolve(), home=a.home) + except OrchestratorError as e: + print(f"orchestrator_watch: {e}", file=sys.stderr) + return 2 + print(json.dumps(res, indent=2) if a.as_json else + f"{res['action']}: " + (res.get("reason") or + f"spawned {len(res['spawned'])}, " + f"refused {len(res['refused'])}, " + f"live {len(res['live'])}")) + return 0 diff --git a/bin/orchestrator_watch.py b/bin/orchestrator_watch.py new file mode 100755 index 0000000..8ca9168 --- /dev/null +++ b/bin/orchestrator_watch.py @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""One shift-watcher tick — the entry the systemd user timer calls. + +See `agents/conductors/batch/_orchestrator.py` for what a tick does, and +`skills/batch/batch.md` "The watcher" for installing the timer (the human's +act; no session installs it). +""" +import sys +from pathlib import Path + +BATCH = Path(__file__).resolve().parents[1] / "agents" / "conductors" / "batch" +sys.path.insert(0, str(BATCH.parents[1] / "faculties" / "sizing")) +sys.path.insert(0, str(BATCH)) + +from _orchestrator import main # noqa: E402 + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bin/systemd/pyauto-orchestrator.service b/bin/systemd/pyauto-orchestrator.service new file mode 100644 index 0000000..99a4637 --- /dev/null +++ b/bin/systemd/pyauto-orchestrator.service @@ -0,0 +1,10 @@ +# Template — copy to ~/.config/systemd/user/ (the human's act; see +# skills/batch/batch.md "The watcher"). Paths come from the environment file, +# never from this unit, so the template carries no workspace path. +[Unit] +Description=PyAuto shift watcher — one tick (deterministic, no LLM) + +[Service] +Type=oneshot +EnvironmentFile=%h/.config/pyauto/orchestrator.env +ExecStart=/usr/bin/env python3 ${PYAUTO_BRAIN}/bin/orchestrator_watch.py --brief ${PYAUTO_SHIFT_BRIEF} --mind ${PYAUTO_MIND} diff --git a/bin/systemd/pyauto-orchestrator.timer b/bin/systemd/pyauto-orchestrator.timer new file mode 100644 index 0000000..8dc34d7 --- /dev/null +++ b/bin/systemd/pyauto-orchestrator.timer @@ -0,0 +1,13 @@ +# Template — see pyauto-orchestrator.service. The timer keeps time so no +# Claude session has to (PyAutoMind/policy/shift_watcher.md). A `STOP` file in +# the slot directory turns every tick into a no-op without touching this unit. +[Unit] +Description=PyAuto shift watcher — tick every 2 minutes + +[Timer] +OnBootSec=2min +OnUnitActiveSec=2min +Unit=pyauto-orchestrator.service + +[Install] +WantedBy=timers.target diff --git a/skills/batch/batch.md b/skills/batch/batch.md index 7cfa69b..036f69a 100644 --- a/skills/batch/batch.md +++ b/skills/batch/batch.md @@ -131,6 +131,11 @@ tasks first, then only the backlog its policy covers, against its they arrive with the watcher in phase 3 of the autonomous-orchestrators epic, and until then the steps above are the dispatch. +## The watcher (phase 3a — fake workers only) + +One deterministic tick per timer firing; the timer is installed by the human, never a session. +What a tick does, its state, caps and the install commands: `agents/conductors/batch/AGENTS.md` "The watcher". + ## When they come back — collect, review, close (2026-08-31) **Collect first, before the human reads anything.** It is one command: diff --git a/tests/test_autonomy_doctrine.py b/tests/test_autonomy_doctrine.py index 66325c6..941cab5 100644 --- a/tests/test_autonomy_doctrine.py +++ b/tests/test_autonomy_doctrine.py @@ -139,3 +139,34 @@ def test_consumers_and_invariant_name_the_shift_section(): assert "Shift briefs and unattended `glance`" in consumers invariants = _norm(AUTONOMY[AUTONOMY.index("## Hard invariants"):]) assert "\"shift briefs and unattended glance\" (2026-11-03) grant them" in invariants + + +# --- The watcher's network surface (2026-11-03, PyAutoBrain#555) ------------- + +WATCHER_HEADER = "## The watcher's network surface — 2026-11-03" + + +def _watcher_section(): + assert AUTONOMY.count(WATCHER_HEADER) == 1 + rest = AUTONOMY[AUTONOMY.index(WATCHER_HEADER) + len(WATCHER_HEADER):] + return _norm(rest[:rest.find("\n## ")]) + + +def test_watcher_section_names_exactly_its_surface_and_a_revert_condition(): + sec = _watcher_section() + for surface in ("gh pr checks", "claude agents --json", "one comment per shift"): + assert surface in sec, surface + assert "revert condition:" in sec + assert "laptop-only" in sec and "never" in sec + + +def test_the_network_invariant_points_at_the_watcher_section(): + invariants = AUTONOMY[AUTONOMY.index("## Hard invariants"):] + assert "A third exception, added 2026-11-03" in invariants + assert "The watcher's network surface — 2026-11-03" in invariants + assert "Any *further* network surface in the Brain is a new decision" in invariants + + +def test_watcher_section_is_a_consumer_entry(): + consumers = AUTONOMY[AUTONOMY.index("## Consumers"):] + assert "The watcher's network surface" in consumers diff --git a/tests/test_orchestrator_watcher.py b/tests/test_orchestrator_watcher.py new file mode 100644 index 0000000..1289702 --- /dev/null +++ b/tests/test_orchestrator_watcher.py @@ -0,0 +1,264 @@ +"""Autonomous orchestrators phase 3a (PyAutoBrain#555): the shift watcher's core. + +A dry shift with fake workers — one test group per clause of the prompt's +`Witness:`: (a) overlapping ticks never double-dispatch, (b) a dead worker is +re-dispatched once per shift, (c) `STOP` is the first read, (d) the usage caps +refuse the N+1th spawn, (e) no open brief dispatches nothing. +""" + +from __future__ import annotations + +import datetime as _dt +import json +import os +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +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 _orchestrator as orch # noqa: E402 + +NOW = _dt.datetime(2026, 11, 4, 9, 0, tzinfo=_dt.timezone.utc) + +BRIEF = """\ +# Batch 2026-11-04 am +- dispatched: 2026-11-04T08:30Z +- review-at: 2026-11-04T11:30Z +- brief: + - goals: + - 1. dry shift + - tasks: +{tasks} + - backlog: none + - review-minutes: 45 + - lanes: any local-dev + - merge-cap: 0 + - token-budget: 1000 +{caps}""" + +PROMPT = """\ +# {name} + +Type: feature +Target: PyAutoBrain +Repos: +- PyAutoBrain +Difficulty: small +Autonomy: supervised +Priority: normal +Status: formalised +Consequence: glance +Witness: a unit test +Review-minutes: 3 +Unattended: ready + +Body. +""" + + +def _shift(tmp_path: Path, names=("t1", "t2"), caps="") -> tuple[Path, Path, Path]: + """(mind, brief, home) for a dry shift naming `names`.""" + mind = tmp_path / "mind" + folder = mind / "draft" / "feature" / "pyautobrain" + folder.mkdir(parents=True) + for n in names: + (folder / f"{n}.md").write_text(PROMPT.format(name=n)) + tasks = "\n".join(f" - draft/feature/pyautobrain/{n}.md" for n in names) + brief = tmp_path / "shift.md" + brief.write_text(BRIEF.format(tasks=tasks, caps=caps)) + return mind, brief, tmp_path / "home" + + +def _wait_exit(slot: Path, timeout=10.0): + """Until every leased fake worker has exited.""" + end = time.time() + timeout + while time.time() < end: + if all(not orch._alive(int(l["pid"])) for l in orch.leases(slot).values()): + return + time.sleep(0.05) + raise AssertionError("fake workers did not exit") + + +# --- (a) overlapping ticks ------------------------------------------------------- + +def test_overlapping_tick_processes_dispatch_one_worker_per_task(tmp_path): + mind, brief, home = _shift(tmp_path) + script = ( + "import sys, datetime as d\n" + f"sys.path[:0] = [{str(BRAIN / 'agents' / 'faculties' / 'sizing')!r}, {str(BATCH)!r}]\n" + "import _orchestrator as o\n" + "now = d.datetime(2026, 11, 4, 9, 0, tzinfo=d.timezone.utc)\n" + f"r = o.tick({str(brief)!r}, o.Path({str(mind)!r}), home=o.Path({str(home)!r})," + " now=now, spawner=o.FakeSpawner(seconds=3))\n" + "print(len(r.get('spawned', [])))\n") + procs = [subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE, + text=True) for _ in range(4)] + spawned = sum(int(p.communicate(timeout=60)[0].strip() or 0) for p in procs) + slot = orch.slot_dir(brief, home) + assert spawned == 2 + assert sorted(orch.leases(slot)) == ["t1", "t2"] + assert orch.read_counters(slot)["sessions"] == 2 + + +def test_a_held_lock_makes_the_second_tick_back_off(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + slot.mkdir(parents=True) + (slot / orch.LOCK_FILE).write_text(json.dumps({"pid": os.getpid(), + "at": NOW.timestamp()})) + assert orch.tick(brief, mind, home=home, now=NOW, + spawner=orch.FakeSpawner())["action"] == "busy" + + +def test_an_existing_lease_is_never_spawned_twice(tmp_path): + mind, brief, home = _shift(tmp_path) + first = orch.tick(brief, mind, home=home, now=NOW, + spawner=orch.FakeSpawner(seconds=3)) + second = orch.tick(brief, mind, home=home, now=NOW, + spawner=orch.FakeSpawner(seconds=3)) + assert len(first["spawned"]) == 2 and second["spawned"] == [] + + +# --- (b) crash recovery ---------------------------------------------------------- + +def test_a_dead_worker_is_redispatched_once_then_abandoned(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1",)) + slot = orch.slot_dir(brief, home) + crashy = orch.FakeSpawner(seconds=0.05, crash={"t1"}) + later = lambda s: NOW + _dt.timedelta(seconds=s) # noqa: E731 + + assert [s["task"] for s in orch.tick(brief, mind, home=home, now=NOW, + spawner=crashy, grace_s=0)["spawned"]] == ["t1"] + _wait_exit(slot) + r = orch.tick(brief, mind, home=home, now=later(1), spawner=crashy, grace_s=0) + assert r["reaped"] == [{"task": "t1", "event": "dead", "redispatch": "allowed"}] + assert [s["task"] for s in r["spawned"]] == ["t1"] # the one re-dispatch + _wait_exit(slot) + r = orch.tick(brief, mind, home=home, now=later(2), spawner=crashy, grace_s=0) + assert r["reaped"][0]["redispatch"] == "spent — abandoned" + assert r["spawned"] == [] + c = orch.read_counters(slot) + assert c["crashes"] == 2 and c["redispatched"] == {"t1": 1} + assert c["abandoned"] == ["t1"] and c["sessions"] == 2 + + +def test_a_dead_lease_waits_out_its_grace(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1",)) + slot = orch.slot_dir(brief, home) + crashy = orch.FakeSpawner(seconds=0.05, crash={"t1"}) + orch.tick(brief, mind, home=home, now=NOW, spawner=crashy, grace_s=60) + _wait_exit(slot) + r = orch.tick(brief, mind, home=home, now=NOW + _dt.timedelta(seconds=30), + spawner=crashy, grace_s=60) + assert r["reaped"] == [] and r["spawned"] == [] # still inside the grace + + +def test_a_finished_worker_is_released_and_never_redispatched(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1",)) + slot = orch.slot_dir(brief, home) + orch.tick(brief, mind, home=home, now=NOW, spawner=orch.FakeSpawner(seconds=0.05)) + _wait_exit(slot) + r = orch.tick(brief, mind, home=home, now=NOW + _dt.timedelta(seconds=5), + spawner=orch.FakeSpawner(seconds=0.05), grace_s=0) + assert r["reaped"] == [{"task": "t1", "event": "done"}] and r["spawned"] == [] + assert orch.read_counters(slot)["done"] == ["t1"] + + +# --- (c) the kill switch ----------------------------------------------------------- + +def test_stop_is_read_before_anything_else(tmp_path): + home = tmp_path / "home" + missing_brief = tmp_path / "never_written.md" # reading it would raise + slot = orch.slot_dir(missing_brief, home) + slot.mkdir(parents=True) + (slot / orch.STOP_FILE).write_text("") + r = orch.tick(missing_brief, tmp_path / "no_mind", home=home, now=NOW, + spawner=orch.FakeSpawner()) + assert r == {"action": "stopped", "reason": "STOP present"} + assert sorted(p.name for p in slot.iterdir()) == ["STOP"] + + +def test_stop_halts_a_running_shift_at_the_next_tick(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1", "t2"), caps=" - max-workers: 1\n") + slot = orch.slot_dir(brief, home) + orch.tick(brief, mind, home=home, now=NOW, spawner=orch.FakeSpawner(seconds=3)) + (slot / orch.STOP_FILE).write_text("") + assert orch.tick(brief, mind, home=home, now=NOW, + spawner=orch.FakeSpawner())["action"] == "stopped" + assert orch.read_counters(slot)["sessions"] == 1 + + +# --- (d) usage caps --------------------------------------------------------------- + +def test_max_workers_refuses_the_next_spawn(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1", "t2", "t3")) + r = orch.tick(brief, mind, home=home, now=NOW, spawner=orch.FakeSpawner(seconds=3)) + assert len(r["spawned"]) == 2 + assert r["refused"] == [{"task": "t3", "why": "max-workers 2 reached"}] + + +def test_max_sessions_refuses_past_the_shift_total(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1", "t2"), + caps=" - max-workers: 5\n - max-sessions: 1\n") + r = orch.tick(brief, mind, home=home, now=NOW, spawner=orch.FakeSpawner(seconds=3)) + assert len(r["spawned"]) == 1 + assert r["refused"][0]["why"] == "max-sessions 1 reached" + slot = orch.slot_dir(brief, home) + assert orch.read_counters(slot)["refusals"][0]["why"] == "max-sessions 1 reached" + + +def test_a_cap_that_is_not_a_number_refuses_the_brief(tmp_path): + mind, brief, home = _shift(tmp_path, caps=" - max-workers: lots\n") + r = orch.tick(brief, mind, home=home, now=NOW, spawner=orch.FakeSpawner()) + assert r["action"] == "idle" and "brief.max-workers" in r["reason"] + + +# --- (e) no open brief -------------------------------------------------------------- + +def test_a_brief_past_review_at_dispatches_nothing(tmp_path): + mind, brief, home = _shift(tmp_path) + r = orch.tick(brief, mind, home=home, now=NOW + _dt.timedelta(hours=3), + spawner=orch.FakeSpawner()) + assert r["action"] == "idle" and "has passed" in r["reason"] + assert not (orch.slot_dir(brief, home) / "leases").exists() + + +def test_a_refused_brief_dispatches_nothing(tmp_path): + mind, brief, home = _shift(tmp_path) + brief.write_text(brief.read_text().replace("- review-at: 2026-11-04T11:30Z\n", "")) + r = orch.tick(brief, mind, home=home, now=NOW, spawner=orch.FakeSpawner()) + assert r["action"] == "idle" and "review-at: missing" in r["reason"] + + +# --- fake mode only, and the entry -------------------------------------------------- + +def test_real_dispatch_is_refused_until_3b(tmp_path, monkeypatch): + monkeypatch.delenv(orch.FAKE_ENV, raising=False) + mind, brief, home = _shift(tmp_path, names=("t1",)) + with pytest.raises(orch.OrchestratorError, match="phase 3b"): + orch.tick(brief, mind, home=home, now=NOW) + # The lease it could not fill is not left behind to block the task. + assert orch.leases(orch.slot_dir(brief, home)) == {} + + +def test_home_comes_from_the_environment(tmp_path, monkeypatch): + monkeypatch.setenv(orch.HOME_ENV, str(tmp_path / "h")) + assert orch.slot_dir(Path("batches/2026-11-04-am.md")) == tmp_path / "h" / "2026-11-04-am" + + +def test_entry_script_runs_one_idle_tick(tmp_path): + mind, brief, home = _shift(tmp_path) + brief.write_text(brief.read_text().replace("2026-11-04T", "2020-01-01T")) + out = subprocess.run( + [sys.executable, str(BRAIN / "bin" / "orchestrator_watch.py"), "--brief", + str(brief), "--mind", str(mind), "--home", str(home), "--json"], + capture_output=True, text=True, env={**os.environ, orch.FAKE_ENV: "1"}) + assert out.returncode == 0, out.stderr + # Real time is long past this brief's review-at, so the tick is idle. + assert json.loads(out.stdout)["action"] == "idle"