feat: autonomous orchestrators p3d — shift digest, proof packets, one comment

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 <record> --mind <PyAutoMind>` 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 <x>`, 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` (`<url>`, or `merged <url>` for an in-turn tiered
+merge), and `<task>.proof.md` moves to `<slot>/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 `<slot>/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 <file>` 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

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 <file>` — 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 <file> 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 <file>` — read-only; exit 0 valid, 1 refused."""
     if not a.check:
-        print("batch brief: --check <file> is the only brief verb yet "
-              "(dispatch and digest are phase 3)", file=sys.stderr)
+        print("batch brief: --check <file> is the only brief verb "
+              "(the digest is `batch digest --record <file>`)", file=sys.stderr)
         return RC_USAGE
     res = _brief.check(Path(a.check), mind)
     if a.as_json:

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,14 +324,39 @@ 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:
             rejected.append((r["path"], f"lane {r.get('lane')} not in the brief's lanes"))
             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"

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 <url>` 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

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 []