diff --git a/agents/conductors/batch/AGENTS.md b/agents/conductors/batch/AGENTS.md index c18938f..872ef55 100644 --- a/agents/conductors/batch/AGENTS.md +++ b/agents/conductors/batch/AGENTS.md @@ -158,6 +158,30 @@ Until phase 3b it spawns only stub workers (`PYAUTO_ORCHESTRATOR_FAKE=1`); a rea dispatch is refused. Doctrine: `AUTONOMY.md` "The watcher's network surface — 2026-11-03"; the exemption it runs under, `PyAutoMind/policy/shift_watcher.md`. +**Real workers (phase 3b).** Without `PYAUTO_ORCHESTRATOR_FAKE=1` each spawn is +`claude --bg --name shift: --permission-mode `, the +prompt filled from `worker_template.md` (the `/start_dev --auto` line, +the end-at-deliverable contract, the brief path, "never commit to PyAutoMind +yourself", and the `/.done` marker). The lease records the short +id `claude --bg` prints (`backgrounded · · `; `--session-id` is not +honoured by `--bg`). Each tick reads `claude agents --json --all` once: +`working` → heartbeat; `blocked` (waiting on a person) → `stalled_since`, and +after `STALL_GRACE_S` (15 min) the task is **parked** — never killed, never +re-dispatched, named in the digest; `done` with the marker → released, without +it → crashed (re-dispatched once). A finished worker's tokens (input + output + +cache creation) are summed from its local transcript, named by the full +`sessionId`, and the brief's `token-budget:` refuses the next spawn once spent. +`worker-mode:` defaults to `auto` (human, 2026-11-03); `dontAsk` and +`acceptEdits` are allowed, `bypassPermissions` never. **Until phase 3c lands, +real dispatch needs `sandbox: yes` in the brief** — one Mind writer is doctrine +and a real worker would otherwise commit to Mind itself. + +**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 +`-wt` root are not trusted by default, so trust that root once by hand if +workers are ever launched there. + **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`, diff --git a/agents/conductors/batch/_brief.py b/agents/conductors/batch/_brief.py index 8588984..d95eab5 100644 --- a/agents/conductors/batch/_brief.py +++ b/agents/conductors/batch/_brief.py @@ -24,6 +24,11 @@ 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} +#: Permission modes a real worker may run in (phase 3b). `auto` is the default +#: the human chose 2026-11-03: a worker that still needs a person shows +#: `blocked` and is parked. `bypassPermissions` is never a worker mode. +WORKER_MODES = ("auto", "dontAsk", "acceptEdits") +DEFAULT_WORKER_MODE = "auto" SCOPED_KEYS = ("epics", "tags", "tiers") _ITEM_RE = re.compile(r"^(\s*)-\s+(.*?)\s*$") @@ -129,6 +134,14 @@ def cap_of(brief: dict, key: str) -> int: return CAP_DEFAULTS[key] if value is None else value +def worker_mode_of(brief: dict) -> str: + return brief.get("worker-mode", {}).get("value") or DEFAULT_WORKER_MODE + + +def is_sandbox(brief: dict) -> bool: + return brief.get("sandbox", {}).get("value", "").lower() == "yes" + + def lanes_of(brief: dict) -> list[str]: node = brief.get("lanes") return _csv(node["value"]) if node else [] @@ -224,6 +237,14 @@ def check(path: Path, mind: Path) -> dict: 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})") + if "worker-mode" in brief and brief["worker-mode"]["value"] not in WORKER_MODES: + errors.append(f"brief.worker-mode: {brief['worker-mode']['value']} — one of " + + ", ".join(WORKER_MODES) + + (" (bypassPermissions is never a worker mode)" + if brief["worker-mode"]["value"] == "bypassPermissions" else "")) + if "sandbox" in brief and brief["sandbox"]["value"].lower() not in ("yes", "no"): + errors.append(f"brief.sandbox: {brief['sandbox']['value']} — yes or no") + 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 index 54a8da2..037c5b7 100644 --- a/agents/conductors/batch/_orchestrator.py +++ b/agents/conductors/batch/_orchestrator.py @@ -13,8 +13,11 @@ 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. +Workers are fake stubs (`PYAUTO_ORCHESTRATOR_FAKE=1`, phase 3a) or real +`claude --bg` sessions from one session template (phase 3b). A real worker's +liveness is its `claude agents --json` state; one left `blocked` past a grace +is parked for a human — never killed, never re-dispatched. Until phase 3c (one +Mind writer) lands, real workers run only under a `sandbox: yes` brief. Nothing here commits, merges, posts or arms a timer: the systemd timer that calls `bin/orchestrator_watch.py` is installed by the human. """ @@ -24,6 +27,7 @@ import datetime as _dt import json import os +import re import subprocess import sys from pathlib import Path @@ -41,6 +45,11 @@ LEASE_GRACE_S = 120 #: A task's lease may be re-dispatched after a crash this many times a shift. REDISPATCH_LIMIT = 1 +#: A real worker `blocked` (waiting on a human) this long is parked: its lease +#: is released, the task is not re-dispatched, and the digest names it. +STALL_GRACE_S = 900 +#: The worker prompt every real worker starts from (never a bare session). +WORKER_TEMPLATE = Path(__file__).with_name("worker_template.md") FAKE_ENV = "PYAUTO_ORCHESTRATOR_FAKE" HOME_ENV = "PYAUTO_ORCHESTRATOR_HOME" @@ -93,6 +102,8 @@ def read_counters(slot: Path) -> dict: c.setdefault("abandoned", []) c.setdefault("refusals", []) c.setdefault("pending_redispatch", []) + c.setdefault("parked", []) + c.setdefault("tokens", 0) return c @@ -159,16 +170,115 @@ def spawn(self, task: str, prompt_path: str, lease_dir: Path) -> dict: return {"pid": proc.pid, "session_id": f"fake-{proc.pid}", "worktree": None} +def worker_prompt(task: str, prompt_path: str, brief_path: Path, lease_dir: Path) -> str: + """The session template filled in for one task.""" + return WORKER_TEMPLATE.read_text(encoding="utf-8").format( + task=task, task_path=prompt_path, brief_path=brief_path, lease_dir=lease_dir) + + +def session_command(task: str, prompt_path: str, brief_path: Path, lease_dir: Path, + mode: str) -> list[str]: + """The exact argv that starts one worker — named, moded, templated. + + No `--session-id`: `claude --bg` (2.1.316) assigns its own, so the worker is + tracked by the short id the launch prints.""" + if mode not in _brief.WORKER_MODES: + raise OrchestratorError(f"worker mode {mode!r} is not one of " + + ", ".join(_brief.WORKER_MODES)) + return ["claude", "--bg", "--name", f"shift:{task}", "--permission-mode", mode, + worker_prompt(task, prompt_path, brief_path, lease_dir)] + + +#: `claude --bg` prints `backgrounded · <8-hex id> · ` on its first line. +_BG_ID_RE = re.compile(r"backgrounded\W+([0-9a-f]{8})\b") + + class RealSpawner: - """`claude --bg` workers from the session template — phase 3b.""" + """`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") + real = True + def __init__(self, brief_path: Path, mode: str, cwd: Path | None = None, run=None): + self.brief_path, self.mode, self.cwd = Path(brief_path), mode, cwd + self.run = run or subprocess.run -def default_spawner(): - return FakeSpawner() if os.environ.get(FAKE_ENV) == "1" else RealSpawner() + def spawn(self, task: str, prompt_path: str, lease_dir: Path) -> dict: + argv = session_command(task, prompt_path, self.brief_path, lease_dir, self.mode) + res = self.run(argv, cwd=self.cwd, capture_output=True, text=True, timeout=120) + # An untrusted folder is refused here, fast ("Workspace not trusted") — + # never a silent stall. Workers start from a folder the human trusted. + if res.returncode != 0: + raise OrchestratorError(f"claude --bg failed for {task}: " + + (res.stderr or res.stdout).strip()[:300]) + m = _BG_ID_RE.search(res.stdout) + if not m: + raise OrchestratorError(f"claude --bg printed no session id for {task}: " + + res.stdout.strip()[:200]) + return {"pid": 0, "agent_id": m.group(1), "session_id": None, "mode": self.mode} + + +def default_spawner(brief_path: Path, brief: dict): + if os.environ.get(FAKE_ENV) == "1": + return FakeSpawner() + cwd = None + try: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from _pyauto_root import pyauto_root # noqa: E402 + cwd = pyauto_root() + except Exception: + cwd = None + return RealSpawner(brief_path, _brief.worker_mode_of(brief), cwd=cwd) + + +# --- real-worker liveness and usage ------------------------------------------------ + +class AgentsView: + """`claude agents --json --all`, read once per tick, keyed by the short id + `claude --bg` printed (its `id` field).""" + + def __init__(self, sessions: list[dict] | None = None, run=None): + self._sessions, self._run = sessions, run or subprocess.run + + def _load(self) -> list[dict]: + if self._sessions is None: + try: + res = self._run(["claude", "agents", "--json", "--all"], + capture_output=True, text=True, timeout=60) + self._sessions = json.loads(res.stdout) if res.returncode == 0 else [] + except (OSError, ValueError, subprocess.SubprocessError): + self._sessions = [] + return self._sessions + + def _find(self, agent_id: str) -> dict | None: + return next((s for s in self._load() if s.get("id") == agent_id), None) + + def state(self, agent_id: str) -> str | None: + """The CLI's state word — `working` while it runs (2.1.316), `blocked` + waiting on a person, `done` — or None if unlisted. Anything but + `blocked` / `done` / None is read as alive.""" + s = self._find(agent_id) + return None if s is None else (s.get("state") or "running") + + def session_of(self, agent_id: str) -> str | None: + """The full session id — what the transcript file is named by.""" + s = self._find(agent_id) + return s.get("sessionId") if s else None + + +def transcript_tokens(session_id: str, root: Path | None = None) -> int: + """Tokens a finished session spent: input + output + cache creation, summed + from its local transcript (a file read — no network).""" + root = root or Path.home() / ".claude" / "projects" + total = 0 + for f in root.glob(f"*/{session_id}.jsonl"): + for line in f.read_text(encoding="utf-8", errors="replace").splitlines(): + try: + usage = (json.loads(line).get("message") or {}).get("usage") or {} + except (ValueError, AttributeError): + continue + total += sum(int(usage.get(k) or 0) for k in + ("input_tokens", "output_tokens", "cache_creation_input_tokens")) + return total # --- locks and leases ------------------------------------------------------------ @@ -225,13 +335,22 @@ def leases(slot: Path) -> dict[str, dict]: 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.""" +def _reap(slot: Path, counters: dict, now: _dt.datetime, grace_s: float, + agents: AgentsView | None = None, tokens=transcript_tokens) -> list[dict]: + """Release finished leases, park stalled ones and declare dead ones.""" events = [] lease_dir = slot / "leases" + agents = agents or AgentsView() for task, lease in leases(slot).items(): done_marker = lease_dir / f"{task}.done" path = lease_dir / f"{task}.json" + real = bool(lease.get("agent_id")) and not lease.get("pid") + state = agents.state(lease["agent_id"]) if real else None + finished = done_marker.exists() or state == "done" or ( + real and state is None and now >= _parse_iso(lease["expires"])) + if real and finished: + sid = agents.session_of(lease["agent_id"]) or lease.get("session_id") + counters["tokens"] += tokens(sid) if sid else 0 if done_marker.exists(): path.unlink(missing_ok=True) done_marker.unlink(missing_ok=True) @@ -239,13 +358,27 @@ def _reap(slot: Path, counters: dict, now: _dt.datetime, grace_s: float) -> list counters["done"].append(task) events.append({"task": task, "event": "done"}) continue - if _alive(int(lease.get("pid", 0))): + if real and state == "blocked": + since = lease.setdefault("stalled_since", _iso(now)) + if (now - _parse_iso(since)).total_seconds() >= STALL_GRACE_S: + # Waiting on a human is not a crash: never killed, never + # re-dispatched — the digest names it and the human answers. + path.unlink(missing_ok=True) + if task not in counters["parked"]: + counters["parked"].append(task) + events.append({"task": task, "event": "parked", + "why": f"blocked since {since}"}) + else: + path.write_text(json.dumps(lease, sort_keys=True)) + continue + alive = (state not in (None, "done")) if real else _alive(int(lease.get("pid", 0))) + if alive: + lease.pop("stalled_since", None) 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: + if now < _parse_iso(lease["expires"]) and state != "done": continue # gone, but inside the grace for a late `.done` path.unlink(missing_ok=True) counters["crashes"] += 1 @@ -259,6 +392,10 @@ def _reap(slot: Path, counters: dict, now: _dt.datetime, grace_s: float) -> list return events +def _parse_iso(value: str) -> _dt.datetime: + return _dt.datetime.fromisoformat(value.replace("Z", "+00:00")) + + # --- the tick -------------------------------------------------------------------- def brief_open(brief_path: Path, mind: Path, now: _dt.datetime) -> tuple[bool, str, dict | None]: @@ -275,7 +412,8 @@ def brief_open(brief_path: Path, mind: Path, now: _dt.datetime) -> tuple[bool, s 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: + grace_s: float = LEASE_GRACE_S, records: list[dict] | None = None, + agents: AgentsView | None = None, tokens=transcript_tokens) -> dict: """One watcher tick. Returns what it did; also appended to `tick.log`.""" now = now or _now() brief_path = Path(brief_path) @@ -293,17 +431,25 @@ def tick(brief_path: Path, mind: Path, *, home: Path | None = None, 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) + result = _tick_locked(slot, brief, mind, now, + spawner or default_spawner(brief_path, brief), + grace_s, records, agents, tokens) finally: _drop_lock(slot) _log(slot, now, result) return result -def _tick_locked(slot, brief, mind, now, spawner, grace_s, records) -> dict: +def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents, + tokens) -> dict: counters = read_counters(slot) - reaped = _reap(slot, counters, now, grace_s) + # One Mind writer (AUTONOMY.md, 2026-11-03) is phase 3c. Until it lands a + # real worker would commit to PyAutoMind itself, so real dispatch is fenced + # to briefs that declare themselves a sandbox. + if getattr(spawner, "real", False) and not _brief.is_sandbox(brief): + raise OrchestratorError("real dispatch needs `sandbox: yes` in the brief " + "until phase 3c (one Mind writer) lands") + reaped = _reap(slot, counters, now, grace_s, agents, tokens) live = leases(slot) max_workers = _brief.cap_of(brief, "max-workers") max_sessions = _brief.cap_of(brief, "max-sessions") @@ -315,7 +461,8 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records) -> dict: 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"]: + if task in live or task in counters["done"] or task in counters["abandoned"] \ + or task in counters["parked"]: continue if len(live) >= max_workers: refused.append({"task": task, "why": f"max-workers {max_workers} reached"}) @@ -323,6 +470,11 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records) -> dict: if counters["sessions"] >= max_sessions: refused.append({"task": task, "why": f"max-sessions {max_sessions} reached"}) continue + budget = _brief._nonneg_int(brief.get("token-budget", {}).get("value", "")) + if budget is not None and counters["tokens"] >= budget: + refused.append({"task": task, "why": f"token-budget {budget} spent " + f"({counters['tokens']})"}) + continue lease = {"task": task, "prompt": member["path"], "started": _iso(now), "heartbeat": _iso(now), "expires": _iso(now + _dt.timedelta(seconds=grace_s)), diff --git a/agents/conductors/batch/worker_template.md b/agents/conductors/batch/worker_template.md new file mode 100644 index 0000000..922ff65 --- /dev/null +++ b/agents/conductors/batch/worker_template.md @@ -0,0 +1,17 @@ +You are a worker in a shift the human launched by writing a shift brief: +{brief_path} + +Use the start-dev skill. {task_path} --auto + +The terms of this shift (AUTONOMY.md "Shift briefs and unattended `glance`"): + +- Work only on {task_path}. The brief above is the launch; read it at the ship + step for its `heart-ack:` and `expected-effects:`. +- End at your deliverable: a PR, a posted verdict, or a park with a question + on the issue. Never arm anything that outlives your turn (no send_later, + ScheduleWakeup, CronCreate, RemoteTrigger or /loop) — the watcher keeps time. +- Never commit to PyAutoMind yourself: write any ledger line you would have + committed as a file in {lease_dir}/ and the orchestrator commits it. +- When you finish, write {lease_dir}/{task}.done containing the URLs of the PRs + you opened (or the issue you parked on), one per line. Without that file the + watcher counts your session as crashed. diff --git a/tests/test_orchestrator_dispatch.py b/tests/test_orchestrator_dispatch.py new file mode 100644 index 0000000..14dfc93 --- /dev/null +++ b/tests/test_orchestrator_dispatch.py @@ -0,0 +1,370 @@ +"""Autonomous orchestrators phase 3b (PyAutoBrain#557): real dispatch. + +Real workers are `claude --bg` sessions started from one session template. In +these tests the `claude` binary is a recorded fake and liveness comes from a +fake `claude agents --json`, so no session is started; the one live check is +opt-in (`PYAUTO_ORCHESTRATOR_LIVE=1`). +""" + +from __future__ import annotations + +import datetime as _dt +import json +import os +import subprocess +import sys +import uuid +from pathlib import Path +from types import SimpleNamespace + +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 _brief # noqa: E402 +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. sandbox shift + - tasks: +{tasks} + - backlog: none + - review-minutes: 45 + - lanes: any local-dev + - merge-cap: 0 + - token-budget: {budget} + - sandbox: yes +{extra}""" + +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, names=("t1",), budget=100000, extra=""): + 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, budget=budget, extra=extra)) + return mind, brief, tmp_path / "home" + + +class FakeClaude: + """Records every `claude --bg` argv; prints a short id like the real CLI.""" + + def __init__(self, rc=0): + self.calls, self.rc = [], rc + + def __call__(self, argv, **kw): + self.calls.append((argv, kw)) + name = argv[argv.index("--name") + 1] if "--name" in argv else "x" + out = (f"backgrounded \u00b7 {len(self.calls):08x} \u00b7 {name}\n" + f" claude agents list sessions\n" + f" claude stop {len(self.calls):08x} stop this session\n") + return SimpleNamespace(returncode=self.rc, stdout="" if self.rc else out, + stderr="Workspace not trusted." if self.rc else "") + + +def _agents(state_by_task: dict, slot: Path) -> orch.AgentsView: + """A `claude agents --json` view giving each leased task the named state.""" + sessions = [] + for task, lease in orch.leases(slot).items(): + if task in state_by_task and state_by_task[task] != "absent": + sessions.append({"id": lease["agent_id"], + "sessionId": f"{lease['agent_id']}-full-session", + "kind": "background", "state": state_by_task[task]}) + return orch.AgentsView(sessions) + + +def _tick(brief, mind, home, now, claude, agents=None, tokens=lambda s: 0, grace_s=0): + return orch.tick(brief, mind, home=home, now=now, grace_s=grace_s, + spawner=orch.RealSpawner(brief, "auto", run=claude), + agents=agents or orch.AgentsView([]), tokens=tokens) + + +# --- the session template: never bare ------------------------------------------- + +def test_session_command_is_never_bare(tmp_path): + argv = orch.session_command("t1", "draft/feature/pyautobrain/t1.md", + tmp_path / "shift.md", tmp_path / "leases", "auto") + assert argv[:2] == ["claude", "--bg"] + assert argv[argv.index("--name") + 1] == "shift:t1" + assert "--session-id" not in argv # --bg assigns its own (CLI 2.1.316) + assert argv[argv.index("--permission-mode") + 1] == "auto" + prompt = argv[-1] + for must in ("Use the start-dev skill. draft/feature/pyautobrain/t1.md --auto", + str(tmp_path / "shift.md"), + "Never commit to PyAutoMind yourself", + str(tmp_path / "leases" / "t1.done"), + "Never arm anything that outlives your turn"): + assert must in prompt, must + + +def test_bypass_permissions_is_never_a_worker_mode(tmp_path): + with pytest.raises(orch.OrchestratorError, match="not one of"): + orch.session_command("t1", "p.md", tmp_path, tmp_path, "bypassPermissions") + mind, brief, _ = _shift(tmp_path, extra=" - worker-mode: bypassPermissions\n") + res = _brief.check(brief, mind) + assert not res["ok"] + assert any("never a worker mode" in e for e in res["errors"]) + + +def test_default_worker_mode_is_auto_and_overridable(): + assert _brief.worker_mode_of({}) == "auto" + assert _brief.worker_mode_of({"worker-mode": {"value": "dontAsk"}}) == "dontAsk" + + +# --- dispatch -------------------------------------------------------------------- + +def test_real_spawn_records_the_session_in_the_lease(tmp_path): + mind, brief, home = _shift(tmp_path) + claude = FakeClaude() + r = _tick(brief, mind, home, NOW, claude) + assert [s["task"] for s in r["spawned"]] == ["t1"] + lease = orch.leases(orch.slot_dir(brief, home))["t1"] + assert lease["agent_id"] == "00000001" and lease["mode"] == "auto" + assert lease["pid"] == 0 and lease["session_id"] is None + + +def test_a_failed_spawn_leaves_no_lease(tmp_path): + mind, brief, home = _shift(tmp_path) + with pytest.raises(orch.OrchestratorError, match="claude --bg failed"): + _tick(brief, mind, home, NOW, FakeClaude(rc=1)) + assert orch.leases(orch.slot_dir(brief, home)) == {} + + +def test_real_dispatch_outside_a_sandbox_brief_is_refused(tmp_path): + mind, brief, home = _shift(tmp_path) + brief.write_text(brief.read_text().replace(" - sandbox: yes\n", "")) + with pytest.raises(orch.OrchestratorError, match="sandbox: yes"): + _tick(brief, mind, home, NOW, FakeClaude()) + + +# --- liveness: running, blocked → parked, done, gone --------------------------------- + +def test_a_running_worker_keeps_its_lease(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude) + later = NOW + _dt.timedelta(minutes=30) + r = _tick(brief, mind, home, later, claude, agents=_agents({"t1": "running"}, slot)) + assert r["reaped"] == [] and r["spawned"] == [] and r["live"] == ["t1"] + assert orch.leases(slot)["t1"]["heartbeat"] == orch._iso(later) + + +def test_a_blocked_worker_is_parked_after_the_grace_and_never_redispatched(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude) + blocked = lambda: _agents({"t1": "blocked"}, slot) # noqa: E731 + r = _tick(brief, mind, home, NOW + _dt.timedelta(minutes=1), claude, agents=blocked()) + assert r["reaped"] == [] and orch.leases(slot)["t1"]["stalled_since"] + past = NOW + _dt.timedelta(minutes=1, seconds=orch.STALL_GRACE_S) + r = _tick(brief, mind, home, past, claude, agents=blocked()) + assert r["reaped"][0]["event"] == "parked" + c = orch.read_counters(slot) + assert c["parked"] == ["t1"] and c["crashes"] == 0 + r = _tick(brief, mind, home, past + _dt.timedelta(minutes=5), claude) + assert r["spawned"] == [] and len(claude.calls) == 1 # never killed, never re-sent + + +def test_a_worker_that_unblocks_clears_its_stall(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude) + _tick(brief, mind, home, NOW + _dt.timedelta(minutes=1), claude, + agents=_agents({"t1": "blocked"}, slot)) + _tick(brief, mind, home, NOW + _dt.timedelta(minutes=2), claude, + agents=_agents({"t1": "running"}, slot)) + assert "stalled_since" not in orch.leases(slot)["t1"] + + +def test_done_with_its_marker_is_released_and_its_tokens_counted(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude) + (slot / "leases" / "t1.done").write_text("https://example.invalid/pr/1\n") + seen = [] + r = _tick(brief, mind, home, NOW + _dt.timedelta(minutes=5), claude, + agents=_agents({"t1": "done"}, slot), + tokens=lambda s: seen.append(s) or 1234) + assert r["reaped"] == [{"task": "t1", "event": "done"}] + assert seen == ["00000001-full-session"] # the transcript is named by sessionId + c = orch.read_counters(slot) + assert c["done"] == ["t1"] and c["tokens"] == 1234 and c["crashes"] == 0 + + +def test_done_without_its_marker_is_a_crash(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude) + r = _tick(brief, mind, home, NOW + _dt.timedelta(minutes=5), claude, + agents=_agents({"t1": "done"}, slot), tokens=lambda s: 10) + assert r["reaped"][0]["event"] == "dead" + assert [s["task"] for s in r["spawned"]] == ["t1"] # the one re-dispatch + assert orch.read_counters(slot)["crashes"] == 1 + + +def test_a_vanished_worker_waits_out_the_lease_grace(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude, grace_s=120) + gone = lambda: _agents({"t1": "absent"}, slot) # noqa: E731 + r = _tick(brief, mind, home, NOW + _dt.timedelta(seconds=60), claude, agents=gone(), + grace_s=120) + assert r["reaped"] == [] + r = _tick(brief, mind, home, NOW + _dt.timedelta(seconds=200), claude, agents=gone(), + grace_s=120) + assert r["reaped"][0]["event"] == "dead" + + +# --- the token budget ------------------------------------------------------------- + +def test_a_spent_token_budget_refuses_the_next_spawn(tmp_path): + mind, brief, home = _shift(tmp_path, names=("t1", "t2"), budget=1000, + extra=" - max-workers: 1\n") + slot = orch.slot_dir(brief, home) + claude = FakeClaude() + _tick(brief, mind, home, NOW, claude) + (slot / "leases" / "t1.done").write_text("") + r = _tick(brief, mind, home, NOW + _dt.timedelta(minutes=5), claude, + agents=_agents({"t1": "done"}, slot), tokens=lambda s: 1500) + assert r["spawned"] == [] + assert r["refused"] == [{"task": "t2", "why": "token-budget 1000 spent (1500)"}] + + +def test_transcript_tokens_sums_input_output_and_cache_creation(tmp_path): + sid = str(uuid.uuid4()) + d = tmp_path / "projects" / "-some-project" + d.mkdir(parents=True) + lines = [ + {"type": "user", "message": {"role": "user"}}, + {"message": {"usage": {"input_tokens": 10, "output_tokens": 5, + "cache_creation_input_tokens": 100, + "cache_read_input_tokens": 9999}}}, + {"message": {"usage": {"input_tokens": 1, "output_tokens": 2}}}, + "not json", + ] + (d / f"{sid}.jsonl").write_text("\n".join( + x if isinstance(x, str) else json.dumps(x) for x in lines)) + assert orch.transcript_tokens(sid, root=tmp_path / "projects") == 118 + + +def test_agents_view_reads_the_cli_once(tmp_path): + calls = [] + + def run(argv, **kw): + calls.append(argv) + return SimpleNamespace(returncode=0, stdout=json.dumps( + [{"id": "a1", "sessionId": "a1-x", "state": "blocked"}, {"id": "a2"}])) + view = orch.AgentsView(run=run) + assert view.state("a1") == "blocked" + assert view.state("a2") == "running" # a listed session with no state is live + assert view.state("a3") is None + assert view.session_of("a1") == "a1-x" + assert calls == [["claude", "agents", "--json", "--all"]] + + +def test_brief_validates_sandbox_and_worker_mode(tmp_path): + mind, brief, _ = _shift(tmp_path, extra=" - worker-mode: dontAsk\n") + assert _brief.check(brief, mind)["ok"] + brief.write_text(brief.read_text().replace("sandbox: yes", "sandbox: maybe")) + assert any(e.startswith("brief.sandbox") for e in _brief.check(brief, mind)["errors"]) + + +# --- the live check (opt-in: starts one real, tiny worker) ------------------------------ + +def _trusted_root() -> Path: + """`claude --bg` refuses an untrusted folder (it fails fast, it does not + stall), so the live worker starts where real ones do: the workspace root.""" + if os.environ.get("PYAUTO_ORCHESTRATOR_LIVE_CWD"): + return Path(os.environ["PYAUTO_ORCHESTRATOR_LIVE_CWD"]) + sys.path.insert(0, str(BRAIN / "agents")) + from _pyauto_root import pyauto_root # noqa: E402 + return pyauto_root() + + +def test_an_untrusted_folder_fails_the_spawn_fast(tmp_path): + # Recorded from the CLI (2.1.316): exit 1, "Workspace not trusted". The + # spawner turns it into an error and leaves no lease — never a silent stall. + mind, brief, home = _shift(tmp_path) + untrusted = FakeClaude(rc=1) + with pytest.raises(orch.OrchestratorError, match="claude --bg failed"): + _tick(brief, mind, home, NOW, untrusted) + assert orch.leases(orch.slot_dir(brief, home)) == {} + + +@pytest.mark.skipif(os.environ.get("PYAUTO_ORCHESTRATOR_LIVE") != "1", + reason="starts a real claude --bg session; opt in with " + "PYAUTO_ORCHESTRATOR_LIVE=1") +def test_live_sandbox_worker(): + """One real worker on a no-op prompt, launched the way RealSpawner launches: + its printed id is parsed, `claude agents --json` follows it to `done`, + `claude logs` reads it (attach's non-interactive twin), and its transcript + yields a non-zero token count. Removed afterwards.""" + import time + res = subprocess.run(["claude", "--bg", "--name", "shift:live-witness", + "--permission-mode", "auto", + "Reply with the single word DONE and stop."], + cwd=_trusted_root(), capture_output=True, text=True, + timeout=120) + assert res.returncode == 0, res.stderr + m = orch._BG_ID_RE.search(res.stdout) + assert m, res.stdout + agent = m.group(1) + states = [] + try: + deadline = time.time() + 180 + while time.time() < deadline: + view = orch.AgentsView() + state = view.state(agent) + if not states or states[-1] != state: + states.append(state) + if state in ("done", "blocked"): + break + time.sleep(5) + print("live states:", states) + assert states[-1] == "done", states + logs = subprocess.run(["claude", "logs", agent], capture_output=True, + text=True, timeout=60) + assert logs.returncode == 0 + tokens = orch.transcript_tokens(view.session_of(agent)) + print("live tokens:", tokens) + assert tokens > 0 + finally: + subprocess.run(["claude", "stop", agent], capture_output=True, timeout=60) + subprocess.run(["claude", "rm", agent], capture_output=True, timeout=60) diff --git a/tests/test_orchestrator_watcher.py b/tests/test_orchestrator_watcher.py index 1289702..c58b24f 100644 --- a/tests/test_orchestrator_watcher.py +++ b/tests/test_orchestrator_watcher.py @@ -238,10 +238,12 @@ def test_a_refused_brief_dispatches_nothing(tmp_path): # --- fake mode only, and the entry -------------------------------------------------- -def test_real_dispatch_is_refused_until_3b(tmp_path, monkeypatch): +def test_real_dispatch_needs_a_sandbox_brief_until_3c(tmp_path, monkeypatch): + # 3b replaced 3a's blanket refusal: real workers exist, but only a + # `sandbox: yes` brief may start one until the one-Mind-writer path (3c). monkeypatch.delenv(orch.FAKE_ENV, raising=False) mind, brief, home = _shift(tmp_path, names=("t1",)) - with pytest.raises(orch.OrchestratorError, match="phase 3b"): + with pytest.raises(orch.OrchestratorError, match="sandbox: yes"): 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)) == {}