From 33b8df8ed098025809e79f3c951cb838779603fa Mon Sep 17 00:00:00 2001 From: Jammy2211 Date: Wed, 4 Nov 2026 08:54:46 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20autonomous=20orchestrators=20p3d=20?= =?UTF-8?q?=E2=80=94=20shift=20digest,=20proof=20packets,=20one=20comment?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first tick after review-at writes the ## Digest (brief work vs backlog picks with the brief rule each was picked under, one proof packet per started task or an honest stub, every counter) into the record through the one Mind writer and posts it once on the brief's issue. digest.posted is created exclusively before the post: a crash leaves no comment, never two; nothing is read back. batch digest --record prints it; batch brief --check refuses a digest missing a section, counter, pick reason or packet. Co-Authored-By: Claude Opus 5.5 --- agents/conductors/batch/AGENTS.md | 24 +- agents/conductors/batch/_batch.py | 36 +- agents/conductors/batch/_brief.py | 37 +- agents/conductors/batch/_digest.py | 188 ++++++++++ agents/conductors/batch/_mind_writer.py | 13 + agents/conductors/batch/_orchestrator.py | 111 +++++- agents/conductors/batch/worker_template.md | 12 +- bin/shift_ledger.py | 11 +- skills/batch/batch.md | 10 +- tests/test_orchestrator_digest.py | 383 +++++++++++++++++++++ 10 files changed, 799 insertions(+), 26 deletions(-) create mode 100644 agents/conductors/batch/_digest.py create mode 100644 tests/test_orchestrator_digest.py diff --git a/agents/conductors/batch/AGENTS.md b/agents/conductors/batch/AGENTS.md index 1cdeee4..4823209 100644 --- a/agents/conductors/batch/AGENTS.md +++ b/agents/conductors/batch/AGENTS.md @@ -146,7 +146,7 @@ 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) +## The watcher (phases 3a–3d) `bin/orchestrator_watch.py --brief --mind ` runs **one tick**: `STOP` first, then the open-brief gate, the tick lock, the lease reap, @@ -192,6 +192,28 @@ or a raw `git commit` alike — and otherwise chains to the repo's own hook. With that in place real dispatch no longer needs `sandbox: yes` (human, 2026-11-03); the key stays as a test-shift marker. +**The digest (phase 3d).** Every spawn records, in `counters.json` `picks`, +the prompt, its tier and **why the brief admitted it** (`_brief.pick_reason`: +`named task`, `scoped epic|tag|tier `, or `backlog: fill-idle`) — at +dispatch, because by `review-at:` the prompt may have moved to `complete/`. +The reap keeps what the worker left before deleting its marker: the `.done` +lines go to `outcomes` (``, or `merged ` for an in-turn tiered +merge), and `.proof.md` moves to `/proofs/`. The first tick after +`review-at:` runs `finish_shift` once: `_digest.build` writes the `## Digest` +(`PyAutoMind/batches/AGENTS.md` format — brief work, backlog picks each with +its `why:`, one proof packet per started task or a stub that says none was +written, every counter) to `/digest.md`, appends it to the record through +the one Mind writer, writes `digest.html` only under `packet: yes`, then +**exclusive-creates `digest.posted` before** the one `gh issue comment` on the +brief's `issue:`. A crash between the two leaves no comment, never two; a failed +post is not retried; nothing is read back from GitHub. Health per PR is one +read-only `gh pr checks` (exit 0 → HEALTHY, 8 → RUNNING, else FAILED or +NO-CHECKS). `glance-auto-merges`, `reverts`, `pushes`, `drafts-filed` and +`packets-sent` read `counters.json` and stay 0 until phases 4–5 move them +(human, 2026-11-04). `batch digest --record ` prints the digest and +writes nothing; `batch brief --check` refuses a digest missing a section, a +counter, a pick's `why:` or a started task's packet. + **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 diff --git a/agents/conductors/batch/_batch.py b/agents/conductors/batch/_batch.py index c40c893..965978e 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"], nargs="?", + ap.add_argument("verb", choices=["plan", "collect", "brief", "digest"], nargs="?", default="plan") ap.add_argument("--mind", type=Path, default=BODY_MAP_PATH.parent) ap.add_argument("--budget", type=int, default=DEFAULT_REVIEW_BUDGET, @@ -2795,6 +2795,13 @@ def main(argv=None) -> int: ap.add_argument("--brief", default="", metavar="FILE", help="plan: compose inside this shift brief (named tasks " "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)") + ap.add_argument("--home", type=Path, default=None, + help="digest: the watcher's state directory (default: " + "$PYAUTO_ORCHESTRATOR_HOME, else .orchestrator)") # collect only ap.add_argument("--slot", default="", help="the batch record to collect (default: the newest)") @@ -2828,17 +2835,40 @@ def main(argv=None) -> int: return _main_collect(mind, a) if a.verb == "brief": return _main_brief(mind, a) + if a.verb == "digest": + return _main_digest(mind, a) return _main_plan(mind, a, lane) except BatchUsageError as e: print(e, file=sys.stderr) return RC_USAGE +def _main_digest(mind: Path, a) -> int: + """`batch digest --record ` — print the shift's digest; writes nothing. + + The record append and the one issue comment are the watcher's, at + `review-at:` (`_orchestrator.finish_shift`); this verb is how a person or a + session reads the digest before then.""" + if not a.record: + print("batch digest: --record is required", file=sys.stderr) + return RC_USAGE + import _digest + import _orchestrator + record = Path(a.record) + brief = _brief.parse(record.read_text(encoding="utf-8"))["brief"] + if brief is None: + print(f"batch digest: {record} has no brief: block", file=sys.stderr) + return RC_USAGE + slot = _orchestrator.slot_dir(record, a.home) + print(_digest.build(brief, slot, _orchestrator.read_counters(slot)), end="") + return RC_OK + + def _main_brief(mind: Path, a) -> int: """`batch brief --check ` — read-only; exit 0 valid, 1 refused.""" if not a.check: - print("batch brief: --check is the only brief verb yet " - "(dispatch and digest are phase 3)", file=sys.stderr) + print("batch brief: --check is the only brief verb " + "(the digest is `batch digest --record `)", file=sys.stderr) return RC_USAGE res = _brief.check(Path(a.check), mind) if a.as_json: diff --git a/agents/conductors/batch/_brief.py b/agents/conductors/batch/_brief.py index d95eab5..1c20a65 100644 --- a/agents/conductors/batch/_brief.py +++ b/agents/conductors/batch/_brief.py @@ -289,6 +289,10 @@ def check(path: Path, mind: Path) -> dict: errors.append(f"review-minutes-actual: missing on the previous " f"record {prev.name} — measure it before a new brief") + # The digest, once collect has appended one (phase 3d). + import _digest # noqa: E402 — deferred: _digest imports this module + errors += _digest.validate(path.read_text(encoding="utf-8", errors="replace")) + return {"ok": not errors, "errors": errors, "warnings": warnings} @@ -320,10 +324,7 @@ def scope(records: list[dict], brief: dict) -> tuple[list[dict], list[tuple[str, if backlog == "none": rejected.append((r["path"], "outside the brief (backlog: none)")) continue - if backlog == "scoped" and not ( - r.get("epic") in scoped.get("epics", []) - or themes & set(scoped.get("tags", [])) - or r.get("consequence") in scoped.get("tiers", [])): + if backlog == "scoped" and not _scoped_match(r, scoped): rejected.append((r["path"], "outside the brief's scoped backlog")) continue if (r.get("lane") or "any") not in lanes: @@ -331,3 +332,31 @@ def scope(records: list[dict], brief: dict) -> tuple[list[dict], list[tuple[str, continue kept.append(r) return kept, rejected, named + + +def _scoped_match(record: dict, scoped: dict[str, list[str]]) -> str: + """The `scoped:` rule a record matches, as the digest names it; "" if none.""" + if record.get("epic") in scoped.get("epics", []): + return f"scoped epic {record['epic']}" + tags = sorted(set(record.get("themes") or []) & set(scoped.get("tags", []))) + if tags: + return f"scoped tag {tags[0]}" + if record.get("consequence") in scoped.get("tiers", []): + return f"scoped tier {record['consequence']}" + return "" + + +def pick_reason(record: dict, brief: dict) -> str: + """Why the brief admits this record — the digest's one-line `why:`. + + Recorded at dispatch, when the record is in hand: by `review-at:` the + prompt may have moved to `complete/` and its headers with it. + """ + if Path(record["path"]).name in {Path(t).name for t in tasks_of(brief)}: + return "named task" + backlog = brief.get("backlog", {}).get("value", "none") + if backlog == "fill-idle": + return "backlog: fill-idle (the planner's ranking)" + if backlog == "scoped": + return _scoped_match(record, scoped_of(brief)) or "NOT RECORDED" + return "NOT RECORDED" diff --git a/agents/conductors/batch/_digest.py b/agents/conductors/batch/_digest.py new file mode 100644 index 0000000..4cf8120 --- /dev/null +++ b/agents/conductors/batch/_digest.py @@ -0,0 +1,188 @@ +"""The shift digest (phase 3d) — what the human reads at `review-at:`. + +The format is `PyAutoMind/batches/AGENTS.md` "Digest" and "Proof packet". The +digest is built from what the watcher itself recorded during the shift — the +brief, `counters.json` (dispatch picks and their reasons, outcomes, crashes, +tokens) and the proof packets the workers wrote — plus one read-only +`gh pr checks` per PR, the only GitHub read the watcher's network surface +allows (AUTONOMY.md "The watcher's network surface — 2026-11-03"). It never +reads a session's own account of itself beyond the `.done` and proof files the +worker contract asks for, and it invents nothing: a task with no proof packet +gets a stub that says so. + +`validate` is the digest half of `batch brief --check`. +""" + +from __future__ import annotations + +import re +import subprocess +from pathlib import Path + +import _brief # noqa: E402 + +DIGEST_HEADING = "## Digest" +SECTIONS = ("### Brief work", "### Backlog picks", "### Proof packets", "### Counters") +#: Every counter, always printed — a zero is a fact, an absent line is a finding. +COUNTERS = ("merges", "glance-auto-merges", "reverts", "pushes", "drafts-filed", + "packets-sent", "workers-crashed", "tokens", "review-minutes-planned") +#: Counters only phase 4 (unattended merges) and phase 5 (RAL packets) move; +#: they read from `counters.json` and are 0 until then (human, 2026-11-04). +LATER_COUNTERS = {"glance-auto-merges": "glance_auto_merges", "reverts": "reverts", + "pushes": "pushes", "drafts-filed": "drafts_filed", + "packets-sent": "packets_sent"} +PROOFS_DIR = "proofs" +NO_PICKS = "- none" + +_BULLET_RE = re.compile(r"^- (\S+)") + + +# --- outcomes ------------------------------------------------------------------ + +def parse_done(text: str) -> dict: + """A worker's `.done` file: one PR or issue URL per line, `merged ` for + a PR it merged in-turn under its tier.""" + prs, merged = [], [] + for line in text.splitlines(): + words = line.split() + if not words: + continue + if words[0] == "merged" and len(words) > 1: + merged.append(words[1]) + prs.append(words[1]) + elif words[0].startswith("http"): + prs.append(words[0]) + return {"prs": prs, "merged": merged} + + +def pr_health(url: str, run=None) -> str: + """One read-only `gh pr checks`. Its exit code is the verdict: 0 passed, + 8 pending; anything else failed — or names why it could not say.""" + if "/pull/" not in url: + return "PARKED-ON-ISSUE" + run = run or subprocess.run + try: + res = run(["gh", "pr", "checks", url], capture_output=True, text=True, timeout=60) + except (OSError, subprocess.TimeoutExpired): + return "UNKNOWN" + if res.returncode == 0: + return "HEALTHY" + if res.returncode == 8: + return "RUNNING" + if "no checks" in (res.stderr or res.stdout or "").lower(): + return "NO-CHECKS" + return "FAILED" + + +# --- build --------------------------------------------------------------------- + +def _line(task: str, counters: dict, tier: str, health) -> str: + out = counters.get("outcomes", {}).get(task) + if task in counters.get("parked", []): + return f"- {task} — parked: awaiting-input (blocked on a person)" + if task in counters.get("abandoned", []): + return f"- {task} — abandoned: crashed twice, not re-dispatched" + if out is None: + return f"- {task} — not started" + if not out["prs"]: + return f"- {task} — finished, no PR or issue named in its .done" + parts = [] + for url in out["prs"]: + state = "merged" if url in out["merged"] else health(url) + parts.append(f"{url} — {state}") + return f"- {task} — " + "; ".join(parts) + f" — {tier}" + + +def _packet(task: str, slot: Path, counters: dict) -> str: + proof = slot / PROOFS_DIR / f"{task}.md" + if proof.is_file(): + return proof.read_text(encoding="utf-8").strip() + prs = counters.get("outcomes", {}).get(task, {}).get("prs") or ["no PR"] + return (f"#### {task} — {prs[0]}\n" + "- claim: not supplied — the worker wrote no proof packet\n" + "- not checked: everything; read the PR itself\n" + "- revert: see the PR") + + +def build(brief: dict, slot: Path, counters: dict, *, tiers: dict | None = None, + health=None) -> str: + """The `## Digest` block for one shift, as Markdown.""" + tiers = tiers or {} + health = health or pr_health + picks = counters.get("picks", {}) + named = [Path(p).stem for p in _brief.tasks_of(brief)] + backlog = [t for t in picks if t not in named] + + lines = [DIGEST_HEADING, "", "### Brief work"] + lines += [_line(t, counters, tiers.get(t, picks.get(t, {}).get("tier", "?")), health) + for t in named] or [NO_PICKS] + lines += ["", "### Backlog picks"] + lines += [_line(t, counters, tiers.get(t, picks[t].get("tier", "?")), health) + + f" — why: {picks[t].get('why') or 'NOT RECORDED'}" for t in backlog] or [NO_PICKS] + + started = [t for t in named + backlog + if t in counters.get("outcomes", {}) or t in counters.get("parked", []) + or t in counters.get("abandoned", [])] + lines += ["", "### Proof packets"] + lines += ["\n\n".join(_packet(t, slot, counters) for t in started) or NO_PICKS] + + budget = brief.get("token-budget", {}).get("value", "?") + merged = sum(len(o["merged"]) for o in counters.get("outcomes", {}).values()) + values = {"merges": merged, "workers-crashed": counters.get("crashes", 0), + "tokens": f"{counters.get('tokens', 0)} of {budget}", + "review-minutes-planned": brief.get("review-minutes", {}).get("value", "?")} + for key, field in LATER_COUNTERS.items(): + values[key] = counters.get(field, 0) + lines += ["", "### Counters"] + [f"- {k}: {values[k]}" for k in COUNTERS] + return "\n".join(lines) + "\n" + + +# --- validate ------------------------------------------------------------------ + +def _sections(text: str) -> dict[str, list[str]]: + body = text[text.index(DIGEST_HEADING) + len(DIGEST_HEADING):] + nxt = body.find("\n## ") + body = body if nxt < 0 else body[:nxt] + out: dict[str, list[str]] = {} + current = None + for line in body.splitlines(): + if line.startswith("### "): + current = line.strip() + out[current] = [] + elif current is not None: + out[current].append(line) + return out + + +def validate(text: str) -> list[str]: + """Errors in a record's `## Digest`; empty when there is none, or it is whole.""" + if DIGEST_HEADING not in text: + return [] + errs = [] + secs = _sections(text) + for s in SECTIONS: + if s not in secs: + errs.append(f"digest: {s[4:]} section missing") + for line in secs.get("### Backlog picks", []): + m = _BULLET_RE.match(line) + if m and line.strip() != NO_PICKS: + why = line.split(" — why: ", 1) + if len(why) < 2 or not why[1].strip() or why[1].strip() == "NOT RECORDED": + errs.append(f"digest: backlog pick {m.group(1)} has no `why:` — " + "every pick names the brief rule it was picked under") + seen = {} + for line in secs.get("### Counters", []): + m = re.match(r"^- ([a-z-]+):\s*(\S.*)?$", line) + if m: + seen[m.group(1)] = (m.group(2) or "").strip() + for key in COUNTERS: + if not seen.get(key): + errs.append(f"digest: counter {key} missing") + packets = "\n".join(secs.get("### Proof packets", [])) + for sec in ("### Brief work", "### Backlog picks"): + for line in secs.get(sec, []): + m = _BULLET_RE.match(line) + if m and line.strip() != NO_PICKS and "not started" not in line \ + and f"#### {m.group(1)}" not in packets: + errs.append(f"digest: {m.group(1)} has no proof packet") + return errs diff --git a/agents/conductors/batch/_mind_writer.py b/agents/conductors/batch/_mind_writer.py index e087b83..04d0ebb 100644 --- a/agents/conductors/batch/_mind_writer.py +++ b/agents/conductors/batch/_mind_writer.py @@ -124,6 +124,19 @@ def _git(cwd: Path, *args, check=True) -> subprocess.CompletedProcess: return res +def queue_op(ledger_dir: Path, op: dict) -> Path: + """Write one op file into a ledger directory, atomically, in arrival order.""" + import os + import time + d = Path(ledger_dir) + d.mkdir(parents=True, exist_ok=True) + name = f"{time.time_ns():020d}-{os.getpid()}-{op['op']}.json" + tmp = d / (name + ".tmp") + tmp.write_text(json.dumps(op, indent=2), encoding="utf-8") + tmp.replace(d / name) + return d / name + + def pending_ops(slot: Path) -> list[Path]: d = Path(slot) / LEDGER_DIR return sorted(f for f in d.glob("*.json")) if d.is_dir() else [] diff --git a/agents/conductors/batch/_orchestrator.py b/agents/conductors/batch/_orchestrator.py index a6e2a4b..2d28864 100644 --- a/agents/conductors/batch/_orchestrator.py +++ b/agents/conductors/batch/_orchestrator.py @@ -35,12 +35,16 @@ from pathlib import Path import _brief # noqa: E402 — the batch conductor puts this directory on sys.path +import _digest # noqa: E402 import _mind_writer # noqa: E402 STOP_FILE = "STOP" LOCK_FILE = "tick.lock" COUNTERS_FILE = "counters.json" LOG_FILE = "tick.log" +DIGEST_FILE = "digest.md" +#: Created exclusively before the one post: its existence means "never again". +DIGEST_MARKER = "digest.posted" #: 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 @@ -111,6 +115,8 @@ def read_counters(slot: Path) -> dict: c.setdefault("pending_redispatch", []) c.setdefault("parked", []) c.setdefault("tokens", 0) + c.setdefault("picks", {}) # task → {prompt, why, tier}, at dispatch (3d) + c.setdefault("outcomes", {}) # task → {prs, merged}, from its `.done` (3d) return c @@ -376,6 +382,13 @@ def _reap(slot: Path, counters: dict, now: _dt.datetime, grace_s: float, sid = agents.session_of(lease["agent_id"]) or lease.get("session_id") counters["tokens"] += tokens(sid) if sid else 0 if done_marker.exists(): + # The digest reads these at review-at, long after the marker is gone. + counters["outcomes"][task] = _digest.parse_done( + done_marker.read_text(encoding="utf-8", errors="replace")) + proof = lease_dir / f"{task}.proof.md" + if proof.exists(): + (slot / _digest.PROOFS_DIR).mkdir(exist_ok=True) + proof.replace(slot / _digest.PROOFS_DIR / f"{task}.md") path.unlink(missing_ok=True) done_marker.unlink(missing_ok=True) if task not in counters["done"]: @@ -437,8 +450,11 @@ 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, - agents: AgentsView | None = None, tokens=transcript_tokens) -> dict: - """One watcher tick. Returns what it did; also appended to `tick.log`.""" + agents: AgentsView | None = None, tokens=transcript_tokens, + poster=None, health=None) -> dict: + """One watcher tick. Returns what it did; also appended to `tick.log`. + + After `review-at:` the one remaining act is the digest (phase 3d), once.""" now = now or _now() brief_path = Path(brief_path) slot = slot_dir(brief_path, home) @@ -449,6 +465,17 @@ def tick(brief_path: Path, mind: Path, *, home: Path | None = None, is_open, why, brief = brief_open(brief_path, mind, now) if not is_open: + if why.startswith("review-at") and (slot / "leases").is_dir() \ + and not (slot / DIGEST_MARKER).exists(): + if not _take_lock(slot, now): + return {"action": "busy", "reason": "another tick holds the lock"} + try: + result = finish_shift(brief_path, mind, slot, now, poster=poster, + health=health) + finally: + _drop_lock(slot) + _log(slot, now, result) + return result return {"action": "idle", "reason": why} (slot / "leases").mkdir(parents=True, exist_ok=True) @@ -515,6 +542,9 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents, lease_path.unlink(missing_ok=True) raise lease_path.write_text(json.dumps(lease, sort_keys=True)) + counters["picks"][task] = {"prompt": member["path"], + "why": _brief.pick_reason(member, brief), + "tier": member.get("consequence") or "?"} if task in counters["pending_redispatch"]: counters["pending_redispatch"].remove(task) counters["redispatched"][task] = counters["redispatched"].get(task, 0) + 1 @@ -527,6 +557,83 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents, "refused": refused, "live": sorted(live), "ledger": ledger} +# --- the digest (phase 3d) ---------------------------------------------------------- + +def gh_comment(issue_url: str, body_file: Path, run=None) -> dict: + """The watcher's one write: a comment on the brief's `issue:`.""" + run = run or subprocess.run + res = run(["gh", "issue", "comment", issue_url, "--body-file", str(body_file)], + capture_output=True, text=True, timeout=120) + if res.returncode != 0: + return {"posted": False, "error": (res.stderr or res.stdout).strip()[:300]} + return {"posted": True, "url": res.stdout.strip().splitlines()[-1] if res.stdout.strip() else ""} + + +def _record_rel(brief_path: Path, mind: Path) -> str | None: + """The record's path inside Mind, for the one-Mind-writer append; None when + the record does not live in that checkout (a test brief, a scratch file).""" + try: + return Path(brief_path).resolve().relative_to(Path(mind).resolve()).as_posix() + except ValueError: + return None + + +def finish_shift(brief_path: Path, mind: Path, slot: Path, now: _dt.datetime, *, + poster=None, health=None) -> dict: + """Write the digest into the record and post it once on the brief's issue. + + The marker is created **before** the post, exclusively: a crash between the + two leaves a shift with no comment, never one with two (AUTONOMY.md "The + watcher's network surface" — a second comment retires the watcher). Nothing + is read back from GitHub to check. + """ + rec = _brief.parse(Path(brief_path).read_text(encoding="utf-8")) + brief, top = rec["brief"] or {}, rec["top"] + counters = read_counters(slot) + digest = _digest.build(brief, slot, counters, health=health) + (slot / DIGEST_FILE).write_text(digest, encoding="utf-8") + result = {"action": "digest", "ledger": None, "post": None, "packet": None} + + rel = _record_rel(brief_path, mind) + if rel is not None: + _mind_writer.queue_op(slot / _mind_writer.LEDGER_DIR, + {"op": "append", "path": rel, "text": "\n" + digest}) + try: + result["ledger"] = _mind_writer.commit_ledger( + slot, mind, f"shift {slot.name}: digest") + except _mind_writer.MindWriteError as e: + result["ledger"] = {"committed": None, "error": str(e)} + + if top.get("packet", {}).get("value", "").lower() == "yes": + result["packet"] = str(_write_packet(slot, digest)) + + issue = top.get("issue", {}).get("value", "") + claim = {"at": _iso(now), "issue": issue} + if not _create_exclusive(slot / DIGEST_MARKER, claim): + result["post"] = {"posted": False, "error": "already posted this shift"} + return result + if not issue.startswith("http"): + result["post"] = {"posted": False, "error": "the brief names no issue:"} + else: + result["post"] = (poster or gh_comment)(issue, slot / DIGEST_FILE) + (slot / DIGEST_MARKER).write_text(json.dumps({**claim, **result["post"]}, + sort_keys=True)) + return result + + +def _write_packet(slot: Path, digest: str) -> Path: + """The opt-in HTML packet (`packet: yes`): the digest on one page.""" + import html + css = Path(__file__).with_name("packet.css") + style = css.read_text(encoding="utf-8") if css.is_file() else "" + page = ("Shift digest " + f"{html.escape(slot.name)}" + f"
{html.escape(digest)}
\n") + out = slot / "digest.html" + out.write_text(page, encoding="utf-8") + return out + + def main(argv=None) -> int: import argparse ap = argparse.ArgumentParser( diff --git a/agents/conductors/batch/worker_template.md b/agents/conductors/batch/worker_template.md index fc22004..c9f6097 100644 --- a/agents/conductors/batch/worker_template.md +++ b/agents/conductors/batch/worker_template.md @@ -14,6 +14,12 @@ The terms of this shift (AUTONOMY.md "Shift briefs and unattended `glance`"): change with `{shift_ledger} append|write|move|remove|drop-section ...` (ops land in {ledger_dir}/) and the orchestrator commits them. This covers your `active.md` claim, the `draft/ → active/` move and your close-out. -- 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. +- When you finish, write {lease_dir}/{task}.proof.md first: one proof packet in + the format of PyAutoMind/batches/AGENTS.md "Proof packet", headed + `#### {task} — `, with the Witness: output trimmed to the lines that + decide it. It goes into the shift digest verbatim; leave it out and the + digest says you wrote none. +- Then write {lease_dir}/{task}.done containing the URLs of the PRs you opened + (or the issue you parked on), one per line — `merged ` for a PR you + merged in-turn under its tier. Without that file the watcher counts your + session as crashed. diff --git a/bin/shift_ledger.py b/bin/shift_ledger.py index b8cad8f..69024b0 100755 --- a/bin/shift_ledger.py +++ b/bin/shift_ledger.py @@ -15,10 +15,8 @@ The ledger directory is `$PYAUTO_SHIFT_LEDGER_DIR` (set for every worker). """ import argparse -import json import os import sys -import time from pathlib import Path BATCH = Path(__file__).resolve().parents[1] / "agents" / "conductors" / "batch" @@ -54,13 +52,8 @@ def main(argv=None) -> int: if errs: print("shift_ledger: refused — " + "; ".join(errs), file=sys.stderr) return 1 - d = Path(ledger) - d.mkdir(parents=True, exist_ok=True) - name = f"{time.time_ns():020d}-{os.getpid()}-{a.op}.json" - tmp = d / (name + ".tmp") - tmp.write_text(json.dumps(op, indent=2), encoding="utf-8") - tmp.replace(d / name) - print(f"shift_ledger: queued {a.op} {a.path} → {d / name}") + queued = mw.queue_op(Path(ledger), op) + print(f"shift_ledger: queued {a.op} {a.path} → {queued}") return 0 diff --git a/skills/batch/batch.md b/skills/batch/batch.md index 036f69a..7bb9882 100644 --- a/skills/batch/batch.md +++ b/skills/batch/batch.md @@ -127,13 +127,15 @@ brief against the `batches/AGENTS.md` schema (every error names its key; exit 1 refuses it), and `batch plan --brief ` composes inside it: named tasks first, then only the backlog its policy covers, against its `review-minutes:`. Run the check before writing a brief's `review-at:` into -`batches/`. `batch dispatch` and `batch digest` are **planned — not code yet**: -they arrive with the watcher in phase 3 of the autonomous-orchestrators epic, -and until then the steps above are the dispatch. +`batches/`. Dispatch is the watcher's (below), not a verb. `batch digest +--record ` prints a shift's digest and writes nothing; the record append +and the one issue comment are the watcher's, at `review-at:`. -## The watcher (phase 3a — fake workers only) +## The watcher (phases 3a–3d) One deterministic tick per timer firing; the timer is installed by the human, never a session. +After `review-at:` its one remaining act is the digest: written into the record +and posted once on the brief's `issue:`. 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) diff --git a/tests/test_orchestrator_digest.py b/tests/test_orchestrator_digest.py new file mode 100644 index 0000000..9011822 --- /dev/null +++ b/tests/test_orchestrator_digest.py @@ -0,0 +1,383 @@ +"""Autonomous orchestrators phase 3d (PyAutoBrain#568): the shift digest. + +One test group per clause of the prompt's `Witness:`: the digest is appended +to the shift record in the `PyAutoMind/batches/AGENTS.md` format (brief work vs +backlog picks, each pick with its reason, one proof packet per task, every +counter); `batch brief --check` accepts it and refuses one missing a counter +or a pick reason; and it is posted exactly once — a second run posts nothing. +""" + +from __future__ import annotations + +import datetime as _dt +import os +import shutil +import subprocess +import sys +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 _brief # noqa: E402 +import _digest # noqa: E402 +import _orchestrator as orch # noqa: E402 + +NOW = _dt.datetime(2026, 11, 4, 9, 0, tzinfo=_dt.timezone.utc) +AFTER = NOW + _dt.timedelta(hours=3) +ISSUE = "https://github.com/example-org/example-mind/issues/9999" + +BRIEF = """\ +# Batch 2026-11-04 am +- dispatched: 2026-11-04T08:30Z +- review-at: 2026-11-04T11:30Z +- issue: {issue} +{packet}- brief: + - goals: + - 1. dry shift + - tasks: + - draft/feature/pyautobrain/named.md + - draft/feature/pyautobrain/never.md + - backlog: scoped + - scoped: + - epics: e1 + - review-minutes: 45 + - lanes: any local-dev + - merge-cap: 1 + - token-budget: 1000 + - max-workers: 3 +""" + +PROMPT = """\ +# {name} + +Type: feature +Target: PyAutoBrain +Repos: +- PyAutoBrain +Difficulty: small +Autonomy: supervised +Priority: normal +Status: formalised +{epic}Consequence: glance +Witness: a unit test +Review-minutes: 3 +Unattended: ready + +Body. +""" + +PROOF = """\ +#### named — https://github.com/o/r/pull/1 +- claim: the thing is true +- witness: `pytest -q` + 3 passed +- not checked: nothing else +- revert: `git revert abc` +""" + + +class FilesSpawner: + """A worker that has already finished: `.done` (and, for some, a proof).""" + + def __init__(self, done: dict[str, str], proofs: dict[str, str]): + self.done, self.proofs = done, proofs + + def spawn(self, task, prompt_path, lease_dir): + Path(lease_dir).mkdir(parents=True, exist_ok=True) + if task in self.proofs: + (Path(lease_dir) / f"{task}.proof.md").write_text(self.proofs[task]) + (Path(lease_dir) / f"{task}.done").write_text(self.done.get(task, "")) + return {"pid": 0, "session_id": None, "worktree": None} + + +def _mind(root: Path) -> Path: + folder = root / "draft" / "feature" / "pyautobrain" + folder.mkdir(parents=True) + (folder / "named.md").write_text(PROMPT.format(name="named", epic="")) + (folder / "picked.md").write_text(PROMPT.format(name="picked", epic="Epic: e1\n")) + (folder / "outside.md").write_text(PROMPT.format(name="outside", epic="")) + return root + + +def _run_shift(tmp_path: Path, brief: Path, mind: Path, home: Path): + """Dispatch at NOW, reap one tick later — both workers finished.""" + sp = FilesSpawner( + done={"named": "merged https://github.com/o/r/pull/1\n", + "picked": "https://github.com/o/r/pull/2\n"}, + proofs={"named": PROOF}) + r = orch.tick(brief, mind, home=home, now=NOW, spawner=sp) + assert sorted(s["task"] for s in r["spawned"]) == ["named", "picked"], r + orch.tick(brief, mind, home=home, now=NOW + _dt.timedelta(minutes=1), spawner=sp) + + +def _shift(tmp_path: Path, packet=""): + mind = _mind(tmp_path / "mind") + brief = tmp_path / "shift.md" + brief.write_text(BRIEF.format(issue=ISSUE, packet=packet)) + home = tmp_path / "home" + _run_shift(tmp_path, brief, mind, home) + return mind, brief, home + + +class Poster: + def __init__(self): + self.calls = [] + + def __call__(self, issue, body_file): + self.calls.append((issue, Path(body_file).read_text())) + return {"posted": True, "url": issue + "#issuecomment-1"} + + +def _health(url): + return "HEALTHY" + + +# --- the digest's content ----------------------------------------------------------- + +def test_digest_separates_brief_work_from_backlog_picks_with_reasons(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + counters = orch.read_counters(slot) + assert counters["picks"]["named"]["why"] == "named task" + assert counters["picks"]["picked"]["why"] == "scoped epic e1" + assert "outside" not in counters["picks"] + + d = _digest.build(_brief.parse(brief.read_text())["brief"], slot, counters, + health=_health) + work = d.split("### Brief work")[1].split("### Backlog picks")[0] + picks = d.split("### Backlog picks")[1].split("### Proof packets")[0] + assert "- named — https://github.com/o/r/pull/1 — merged — glance" in work + assert "- never — not started" in work + assert "- picked — https://github.com/o/r/pull/2 — HEALTHY — glance — why: scoped epic e1" in picks + assert "named" not in picks + + +def test_one_proof_packet_per_task_and_an_honest_stub_when_missing(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + d = _digest.build(_brief.parse(brief.read_text())["brief"], slot, + orch.read_counters(slot), health=_health) + packets = d.split("### Proof packets")[1].split("### Counters")[0] + assert "- claim: the thing is true" in packets # the worker's, verbatim + assert "#### picked — https://github.com/o/r/pull/2" in packets + assert "the worker wrote no proof packet" in packets + assert "#### never" not in packets # never started, no packet + + +def test_every_counter_is_printed(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + d = _digest.build(_brief.parse(brief.read_text())["brief"], slot, + orch.read_counters(slot), health=_health) + counters = d.split("### Counters")[1] + for key in _digest.COUNTERS: + assert f"- {key}: " in counters, key + assert "- merges: 1" in counters + assert "- tokens: 0 of 1000" in counters + assert "- review-minutes-planned: 45" in counters + assert "- glance-auto-merges: 0" in counters + + +# --- brief --check over the digest ----------------------------------------------- + +def _with_digest(tmp_path, brief, home, mutate=lambda d: d): + slot = orch.slot_dir(brief, home) + d = _digest.build(_brief.parse(brief.read_text())["brief"], slot, + orch.read_counters(slot), health=_health) + rec = tmp_path / "record_with_digest.md" + rec.write_text(brief.read_text() + "\n" + mutate(d)) + return rec + + +def test_brief_check_accepts_a_whole_digest(tmp_path): + mind, brief, home = _shift(tmp_path) + res = _brief.check(_with_digest(tmp_path, brief, home), mind) + assert res["ok"], res["errors"] + + +def test_brief_check_refuses_a_missing_counter(tmp_path): + mind, brief, home = _shift(tmp_path) + rec = _with_digest(tmp_path, brief, home, + lambda d: d.replace("- reverts: 0\n", "")) + res = _brief.check(rec, mind) + assert not res["ok"] and "digest: counter reverts missing" in res["errors"] + + +def test_brief_check_refuses_a_pick_without_a_reason(tmp_path): + mind, brief, home = _shift(tmp_path) + rec = _with_digest(tmp_path, brief, home, + lambda d: d.replace(" — why: scoped epic e1", "")) + res = _brief.check(rec, mind) + assert not res["ok"] + assert any("backlog pick picked has no `why:`" in e for e in res["errors"]) + + +def test_brief_check_refuses_a_task_without_a_packet(tmp_path): + mind, brief, home = _shift(tmp_path) + rec = _with_digest(tmp_path, brief, home, + lambda d: d.replace("#### picked", "#### someone-else")) + assert "digest: picked has no proof packet" in _brief.check(rec, mind)["errors"] + + +# --- posted exactly once --------------------------------------------------------- + +def test_the_digest_is_posted_once_and_a_second_run_posts_nothing(tmp_path): + mind, brief, home = _shift(tmp_path) + slot = orch.slot_dir(brief, home) + poster = Poster() + r = orch.tick(brief, mind, home=home, now=AFTER, poster=poster, health=_health) + assert r["action"] == "digest" and r["post"]["posted"], r + assert len(poster.calls) == 1 and poster.calls[0][0] == ISSUE + assert poster.calls[0][1].startswith("## Digest") + assert (slot / orch.DIGEST_MARKER).exists() + + again = orch.tick(brief, mind, home=home, now=AFTER + _dt.timedelta(minutes=5), + poster=poster, health=_health) + assert again["action"] == "idle" and len(poster.calls) == 1 + direct = orch.finish_shift(brief, mind, slot, AFTER, poster=poster, health=_health) + assert direct["post"] == {"posted": False, "error": "already posted this shift"} + assert len(poster.calls) == 1 + + +def test_a_failed_post_is_never_retried(tmp_path): + mind, brief, home = _shift(tmp_path) + calls = [] + + def failing(issue, body_file): + calls.append(issue) + return {"posted": False, "error": "HTTP 502"} + + orch.tick(brief, mind, home=home, now=AFTER, poster=failing, health=_health) + orch.tick(brief, mind, home=home, now=AFTER + _dt.timedelta(minutes=5), + poster=failing, health=_health) + assert calls == [ISSUE] # a second comment would retire the watcher + + +def test_a_shift_that_never_ran_has_no_digest(tmp_path): + mind = _mind(tmp_path / "mind") + brief = tmp_path / "shift.md" + brief.write_text(BRIEF.format(issue=ISSUE, packet="")) + poster = Poster() + r = orch.tick(brief, mind, home=tmp_path / "home", now=AFTER, poster=poster) + assert r["action"] == "idle" and poster.calls == [] + + +def test_stop_still_wins_after_review_at(tmp_path): + mind, brief, home = _shift(tmp_path) + (orch.slot_dir(brief, home) / orch.STOP_FILE).write_text("") + poster = Poster() + assert orch.tick(brief, mind, home=home, now=AFTER, poster=poster)["action"] == "stopped" + assert poster.calls == [] + + +def test_the_html_packet_is_opt_in(tmp_path): + mind, brief, home = _shift(tmp_path) + r = orch.tick(brief, mind, home=home, now=AFTER, poster=Poster(), health=_health) + assert r["packet"] is None + assert not (orch.slot_dir(brief, home) / "digest.html").exists() + + mind2, brief2, home2 = _shift(tmp_path / "b", packet="- packet: yes\n") + r2 = orch.tick(brief2, mind2, home=home2, now=AFTER, poster=Poster(), health=_health) + page = Path(r2["packet"]).read_text() + assert "## Digest" in page and "
" in page
+
+
+# --- the record append, through the one Mind writer ------------------------------
+
+LEDGER_MERGE = BRAIN.parent / "PyAutoMind" / "scripts" / "ledger_merge.py"
+if not LEDGER_MERGE.is_file():
+    LEDGER_MERGE = Path(os.environ.get("PYAUTO_MIND", "/nonexistent")) / "scripts" / "ledger_merge.py"
+
+
+def git(cwd, *args):
+    res = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
+    assert res.returncode == 0, (args, res.stderr)
+    return res
+
+
+@pytest.mark.skipif(not LEDGER_MERGE.is_file() or shutil.which("git") is None,
+                    reason="needs git and PyAutoMind's scripts/ledger_merge.py")
+def test_the_digest_lands_in_the_record_on_mind_main(tmp_path, monkeypatch):
+    monkeypatch.setattr(sys, "dont_write_bytecode", False)
+    bare = tmp_path / "remote.git"
+    git(tmp_path, "init", "-q", "--bare", "-b", "main", str(bare))
+    canon = tmp_path / "canon"
+    git(tmp_path, "clone", "-q", str(bare), str(canon))
+    git(canon, "config", "user.name", "Test")
+    git(canon, "config", "user.email", "test@example.invalid")
+    _mind(canon)
+    (canon / "scripts").mkdir()
+    shutil.copy(LEDGER_MERGE, canon / "scripts" / "ledger_merge.py")
+    (canon / "batches").mkdir()
+    brief = canon / "batches" / "2026-11-04-am.md"
+    brief.write_text(BRIEF.format(issue=ISSUE, packet=""))
+    git(canon, "add", "-A")
+    git(canon, "commit", "-q", "-m", "seed")
+    git(canon, "push", "-q", "origin", "HEAD:main")
+
+    home = tmp_path / "home"
+    _run_shift(tmp_path, brief, canon, home)
+    status_before = git(canon, "status", "--porcelain").stdout
+    r = orch.tick(brief, canon, home=home, now=AFTER, poster=Poster(), health=_health)
+    assert r["ledger"]["committed"], r
+    landed = git(bare, "show", "main:batches/2026-11-04-am.md").stdout
+    assert "## Digest" in landed and "- reverts: 0" in landed
+    assert git(canon, "status", "--porcelain").stdout == status_before
+
+    # What landed is what `batch brief --check` accepts.
+    fetched = tmp_path / "landed.md"
+    fetched.write_text(landed)
+    assert _digest.validate(landed) == []
+
+
+# --- the small parts ---------------------------------------------------------------
+
+def test_parse_done_reads_urls_and_merged_lines():
+    out = _digest.parse_done("merged https://x/pull/1\nhttps://x/pull/2\n\nnoise\n")
+    assert out == {"prs": ["https://x/pull/1", "https://x/pull/2"],
+                   "merged": ["https://x/pull/1"]}
+
+
+@pytest.mark.parametrize("rc,err,want", [(0, "", "HEALTHY"), (8, "", "RUNNING"),
+                                         (1, "", "FAILED"),
+                                         (1, "no checks reported", "NO-CHECKS")])
+def test_pr_health_reads_the_gh_exit_code(rc, err, want):
+    def run(argv, **kw):
+        assert argv[:3] == ["gh", "pr", "checks"]
+        return subprocess.CompletedProcess(argv, rc, "", err)
+    assert _digest.pr_health("https://github.com/o/r/pull/1", run=run) == want
+    assert _digest.pr_health("https://github.com/o/r/issues/3", run=run) == "PARKED-ON-ISSUE"
+
+
+@pytest.mark.parametrize("backlog,record,want", [
+    ("fill-idle", {"path": "x/a.md"}, "backlog: fill-idle (the planner's ranking)"),
+    ("scoped", {"path": "x/a.md", "themes": ["jax"]}, "scoped tag jax"),
+    ("scoped", {"path": "x/a.md", "consequence": "notify"}, "scoped tier notify"),
+    ("scoped", {"path": "x/named.md"}, "named task"),
+])
+def test_pick_reason_names_the_rule(backlog, record, want):
+    brief = {"backlog": {"value": backlog, "children": []},
+             "scoped": {"value": "", "children": [
+                 {"text": "tags: jax", "children": []},
+                 {"text": "tiers: notify", "children": []}]},
+             "tasks": {"value": "", "children": [{"text": "x/named.md", "children": []}]}}
+    assert _brief.pick_reason(record, brief) == want
+
+
+def test_batch_digest_verb_prints_and_writes_nothing(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    slot = orch.slot_dir(brief, home)
+    before = sorted(p.name for p in slot.rglob("*"))
+    env = {**os.environ, "PATH": "/nonexistent"}  # no gh: health says UNKNOWN
+    res = subprocess.run([sys.executable, str(BATCH / "_batch.py"), "digest",
+                          "--record", str(brief), "--home", str(home),
+                          "--mind", str(mind)],
+                         capture_output=True, text=True, env=env)
+    assert res.returncode == 0, res.stderr
+    assert res.stdout.startswith("## Digest") and "- merges: 1" in res.stdout
+    assert sorted(p.name for p in slot.rglob("*")) == before
+    assert not (slot / orch.DIGEST_MARKER).exists()