batch: executable autonomy caps, and three planner bugs found by use

agents/conductors/batch/AGENTS.md

@@ -99,6 +99,57 @@ pyauto-brain batch plan --json
 
 Stdlib-only, offline, writes nothing.
 
+## Running a batch by hand (there is no dispatcher yet, and that is fine)
+
+Phase 5 — automatic fan-out — is deliberately unbuilt. Everything below works
+today, costs about two minutes of tapping, and is the honest way to find out
+what a dispatcher actually needs before writing one.
+
+**In the slot (about an hour, once a day):**
+
+1. `pyauto-brain batch plan` — read the BatchDecision. Edit it by removing
+   members you do not want; do not add ones it rejected without reading why.
+2. **Acknowledge the Heart reason set for the shift**, if Heart is YELLOW
+   (`pyauto-brain vitals`). Write it verbatim into the batch record — a grant
+   recorded loosely is how a scoped one becomes standing, which doctrine voids
+   (`AUTONOMY.md`, "Leg 4 under a batch launch").
+3. Open the batch record `PyAutoMind/batches/<YYYY-MM-DD>-<am|pm>.md` from the
+   schema in that folder's `AGENTS.md`, and write the members, the planned
+   review-minutes and the reason set **before** dispatching. That file is what
+   makes the launch auditable afterwards.
+4. **Dispatch**: paste each line the decision prints into **its own session**.
+   One session per member — they must not share one, because a single session
+   would serialise them and carry one member's context into the next.
+5. Go and do something else.
+
+**Next slot:**
+
+6. Read the PRs. Failures first, then anything labelled `decision-taken`, then
+   clean ones. For each: merge (`/prm <PR>`), or say one line of what you want
+   changed and let a session draft the follow-up prompt into `queue.md`.
+7. Append the outcome to the batch record — `delivered:`, and especially
+   **`review-minutes-actual:`**. That last number is the only calibration the
+   estimate will ever get, and everything the planner does rests on it.
+8. Plan the next batch.
+
+**The one leg you must not skip.** A batch launch requires the independent
+adversary (`AUTONOMY.md` leg 5). Until `batch collect` exists, run it by hand
+before reading a PR:
+
+```
+pyauto-brain review --task <name> --witness "<the prompt's Witness:>" --adversary
+```
+
+in a session **using a different model from the one that wrote the branch**. A
+self-run adversary leg is an absent leg, not a weak one.
+
+**What to expect from batch 1.** Small. The planner picks against a 45-minute
+review budget, and with almost no prompt carrying a `Witness:` nearly everything
+grades `judge` at 20 minutes — so a slot holds two or three. That number is the
+honest state of the backlog rather than a limit of the machinery, and the way to
+move it is to write witnesses, which costs zero review-minutes and is the best
+possible fill work.
+
 ## Not built yet
 
 `slice` (the decomposition pass doctrine has named since inception) and

agents/conductors/batch/_batch.py

@@ -35,8 +35,9 @@
 BRAIN = Path(__file__).resolve().parents[3]
 sys.path.insert(0, str(BRAIN / "agents" / "faculties" / "sizing"))
 from _sizing import (  # noqa: E402
-    BODY_MAP_PATH, LIBRARY_REPOS, effective_consequence, effective_difficulty,
-    effective_review_minutes, effective_unattended, parse_prompt, priority_rank,
+    BODY_MAP_PATH, LIBRARY_REPOS, effective_autonomy, effective_consequence,
+    effective_difficulty, effective_review_minutes, effective_unattended,
+    parse_prompt, priority_rank,
 )
 
 # One slot's worth of the human's attention. Not a task count — see the module
@@ -74,7 +75,10 @@ def grade(path: Path, mind: Path) -> dict:
     tier, why, _ = effective_consequence(p, factors)
     ready, ready_why, _ = effective_unattended(p, level, factors, derived)
     minutes, _ = effective_review_minutes(p, tier, level)
+    autonomy, cap, declared_autonomy = effective_autonomy(p, level)
     return {
+        "autonomy": autonomy, "autonomy_cap": cap,
+        "declared_autonomy": declared_autonomy,
         "path": p["path"], "repos": p["repos"], "work_type": p["work_type"],
         "difficulty": level, "score": score, "consequence": tier,
         "witness": p.get("witness"), "review_minutes": minutes,
@@ -85,22 +89,52 @@ def grade(path: Path, mind: Path) -> dict:
     }
 
 
-def _epic_of(text: str) -> str:
+# A prompt whose own header says the work is finished. Mirrors intake's
+# DONE_STATUSES: a session that ships the work and writes the outcome into
+# `Status:` but leaves the file in `draft/` is a known, recorded failure mode,
+# and the file keeps rendering as pickable backlog until someone retires it.
+# Dispatching one wastes a whole shift re-doing finished work.
+DONE_STATUSES = ("shipped", "superseded", "absorbed", "complete", "completed",
+                 "done", "retired")
+
+
+def _header_of(text: str, key: str) -> str:
     for line in text.splitlines()[:30]:
-        if line.lower().startswith("epic:"):
+        if line.lower().startswith(f"{key}:"):
             return line.split(":", 1)[1].strip()
     return ""
 
 
+def _epic_of(text: str) -> str:
+    return _header_of(text, "epic")
+
+
+def _is_done(text: str) -> bool:
+    status = _header_of(text, "status").lower()
+    return any(status.startswith(d) for d in DONE_STATUSES)
+
+
+def _phase_of(text: str) -> float:
+    """`Phase: <n>` as a sortable number; phase-less members sort last."""
+    raw = _header_of(text, "phase")
+    try:
+        return float(raw.rstrip("abcdefgh") or "inf")
+    except ValueError:
+        return float("inf")
+
+
 def survey(mind: Path) -> list[dict]:
     """Grade every backlog prompt. Epic membership rides along for the
     one-slice-per-epic rule."""
     out = []
     for f in sorted(mind.glob("draft/**/*.md")):
         if f.name == "README.md":
             continue
+        text = f.read_text(encoding="utf-8", errors="replace")
         g = grade(f, mind)
-        g["epic"] = _epic_of(f.read_text(encoding="utf-8", errors="replace"))
+        g["epic"] = _epic_of(text)
+        g["phase"] = _phase_of(text)
+        g["done"] = _is_done(text)
         out.append(g)
     return out
 
@@ -125,9 +159,33 @@ def plan(records: list[dict], *, budget: int = DEFAULT_REVIEW_BUDGET,
     else:
         effective_budget, pressure = budget, "clear"
 
+    # An epic's members are worked IN ORDER, so only its lowest un-shipped
+    # phase is startable. Without this the planner can propose phase 6 while
+    # phase 3 is still open — the one-slice-per-epic rule below caps how many
+    # run, not which.
+    next_phase: dict[str, float] = {}
+    for r in records:
+        if r.get("epic") and not r.get("done"):
+            e = r["epic"]
+            next_phase[e] = min(next_phase.get(e, float("inf")),
+                                r.get("phase", float("inf")))
+
     pool = []
     for r in records:
-        if r["unattended"] != "ready":
+        # Readiness says the work FITS one run; autonomy says the run may
+        # FINISH it. A batch that ignores the second fills a shift with tasks
+        # that all stop at the ship checkpoint and come back as questions —
+        # which is the failure the whole epic exists to remove.
+        if r.get("done"):
+            rejected.append((r["path"], "Status: says the work is already done"))
+        elif r.get("epic") and r.get("phase", float("inf")) > next_phase.get(
+                r["epic"], float("inf")):
+            rejected.append((r["path"],
+                             f"epic {r['epic']} phase {r.get('phase')} is not next"))
+        elif r["autonomy"] != "safe":
+            rejected.append((r["path"],
+                             f"autonomy {r['autonomy']} — would park at ship"))
+        elif r["unattended"] != "ready":
             rejected.append((r["path"], f"unattended: {r['unattended']}"))
         elif r["blocked"]:
             rejected.append((r["path"], "declares Blocked-by:"))
@@ -166,6 +224,7 @@ def plan(records: list[dict], *, budget: int = DEFAULT_REVIEW_BUDGET,
         libs.update(x for x in r["repos"] if x in LIBRARY_REPOS)
 
     return {
+        "dispatch": [_dispatch_payload(r) for r in members],
         "session_lane": session_lane,
         "review_budget": budget,
         "effective_budget": effective_budget,
@@ -180,6 +239,17 @@ def plan(records: list[dict], *, budget: int = DEFAULT_REVIEW_BUDGET,
     }
 
 
+def _dispatch_payload(r: dict) -> str:
+    """The exact text a human pastes into one unattended session.
+
+    Written out rather than left to be composed at dispatch time: the launch is
+    the human's act (AUTONOMY.md, "What a batch launch is"), so the thing they
+    perform should carry no decisions — everything decided was decided when they
+    approved the batch.
+    """
+    return f"/start_dev {r['path']} --auto"
+
+
 def emit(d: dict) -> None:
     lane = d["session_lane"]
     print("== BatchDecision ==")
@@ -223,6 +293,11 @@ def emit(d: dict) -> None:
     for why, n in sorted(counts.items(), key=lambda kv: -kv[1])[:8]:
         print(f"  {n:>4}  {why}")
     print()
+    if d["members"]:
+        print("To dispatch: paste ONE of these into its own session —")
+        for line in d["dispatch"]:
+            print(f"  {line}")
+        print()
     print("This is a PROPOSAL. Approving it in the slot is what launches the")
     print("batch — membership is fixed at approval and the grant expires with")
     print("the shift (AUTONOMY.md, \"What a batch launch is\").")

agents/faculties/sizing/_sizing.py

@@ -922,6 +922,54 @@ def effective_review_minutes(p: dict, tier: str, level: str):
     return (declared if declared is not None else derived), derived
 
 
+# The per-work-type autonomy caps, mirrored from `AUTONOMY.md` "Per-work-type
+# caps". Until 2026-08-30 these lived ONLY as prose there, applied by whichever
+# agent happened to read the doctrine — so nothing could compute what a prompt's
+# autonomy actually resolves to, and the batch planner proposed work that would
+# park at the ship checkpoint the moment it ran. A rule every consumer has to
+# re-read and re-apply by hand is a rule that gets applied differently by each
+# of them; this is the same lesson `effective_difficulty` already carries.
+#
+# Editing this table is a doctrine change: edit AUTONOMY.md first, then mirror.
+WORK_TYPE_AUTONOMY_CAPS = {
+    "refactor": "safe", "test": "safe", "maintenance": "safe",
+    "bug": "supervised", "research": "supervised", "experiment": "supervised",
+    "release": "human-required", "human_review": "human-required",
+    "triage": "human-required",
+}
+_AUTONOMY_RANK = {"safe": 0, "supervised": 1, "human-required": 2}
+
+
+def autonomy_cap(work_type: str, difficulty: str) -> str:
+    """The most autonomy this work-type may ever have, whatever it declares."""
+    if work_type in ("feature", "docs"):
+        # Raised 2026-07-09 on calibration evidence; `large` and above still
+        # checkpoints, because size is where the heuristic is least trustworthy.
+        return "safe" if difficulty in ("small", "medium") else "supervised"
+    return WORK_TYPE_AUTONOMY_CAPS.get(work_type, "human-required")
+
+
+def effective_autonomy(p: dict, difficulty: str):
+    """(level, cap, declared) — what a run's autonomy ACTUALLY resolves to.
+
+    `min(header, cap)` in the doctrine's sense: the more restrictive of the two
+    wins, and a missing header means `human-required` rather than a default.
+
+    This is the difference between a task that reaches PR-open unattended and
+    one that parks on a question the moment it gets there, and it is not the
+    same question as `Unattended:`. Readiness asks whether the work FITS one
+    run; this asks whether the run is allowed to finish it without a human.
+    A batch planner that reads only the first will fill a shift with tasks that
+    all stop at the ship checkpoint.
+    """
+    declared = p.get("declared_autonomy")
+    cap = autonomy_cap(p.get("work_type", "?"), difficulty)
+    if not declared:
+        return "human-required", cap, None
+    level = max(declared, cap, key=lambda v: _AUTONOMY_RANK[v])
+    return level, cap, declared
+
+
 def effective_difficulty(p: dict):
     """(level, score, factors, derived_level) — the DECLARED level wins.
 

tests/test_batch_plan.py

@@ -20,12 +20,15 @@
 
 
 def rec(path, *, minutes=20, tier="judge", ready="ready", repos=(), epic="",
-        lane="any", blocked=False, priority="normal"):
+        lane="any", blocked=False, priority="normal", autonomy="safe",
+        done=False, phase=float("inf")):
     return {"path": path, "repos": list(repos), "work_type": "feature",
             "difficulty": "medium", "score": 4, "consequence": tier,
             "witness": None, "review_minutes": minutes, "unattended": ready,
             "why": "", "ready_why": [], "epic": epic, "lane": lane,
-            "priority": priority, "blocked": blocked}
+            "priority": priority, "blocked": blocked, "autonomy": autonomy,
+            "autonomy_cap": "safe", "declared_autonomy": autonomy,
+            "done": done, "phase": phase}
 
 
 def paths(d):
@@ -128,3 +131,49 @@ def test_lane_detection_reads_the_environment_not_a_flag():
     """A session that could be told where it is could plan local-dev work it
     cannot run. Probed from `gh`, the same signal the organism already uses."""
     assert _batch.detect_lane() in ("local-dev", "web-github")
+
+
+def test_only_safe_work_is_dispatched():
+    """Readiness says the work FITS one run; autonomy says the run may FINISH
+    it. A batch that reads only the first fills a shift with tasks that all
+    stop at the ship checkpoint and come back as questions — which is the
+    failure the epic exists to remove. Found by running the planner against the
+    live backlog and reading what it picked."""
+    d = _batch.plan([rec("a.md", autonomy="safe"),
+                     rec("b.md", autonomy="supervised"),
+                     rec("c.md", autonomy="human-required")], budget=100)
+    assert paths(d) == ["a.md"]
+    assert "would park at ship" in why(d, "b.md")
+
+
+def test_a_prompt_that_says_it_is_done_is_never_dispatched():
+    """A session that ships the work and writes the outcome into `Status:` but
+    leaves the file in draft/ is a recorded failure mode; the file keeps
+    rendering as pickable backlog. Dispatching one wastes a whole shift
+    re-doing finished work."""
+    d = _batch.plan([rec("a.md", done=True), rec("b.md")], budget=100)
+    assert paths(d) == ["b.md"]
+    assert "already done" in why(d, "a.md")
+
+
+def test_an_epic_offers_only_its_next_phase():
+    """Members are worked in order. One-slice-per-epic caps how many run; this
+    decides WHICH — without it the planner can propose phase 6 while phase 3 is
+    still open."""
+    d = _batch.plan([rec("late.md", epic="euclid", phase=6),
+                     rec("next.md", epic="euclid", phase=3)], budget=100)
+    assert paths(d) == ["next.md"]
+    assert "is not next" in why(d, "late.md")
+
+
+def test_a_shipped_phase_does_not_block_the_one_after_it():
+    d = _batch.plan([rec("done.md", epic="e", phase=1, done=True),
+                     rec("next.md", epic="e", phase=2)], budget=100)
+    assert paths(d) == ["next.md"]
+
+
+def test_dispatch_payloads_carry_no_decisions():
+    """The launch is the human's act, so what they perform should carry nothing
+    still to be decided — everything was decided when they approved the batch."""
+    d = _batch.plan([rec("draft/x/y/z.md")], budget=100)
+    assert d["dispatch"] == ["/start_dev draft/x/y/z.md --auto"]

tests/test_sizing_review_cost.py

@@ -25,6 +25,7 @@
 from _sizing import (  # noqa: E402
     CONSEQUENCE_TIERS,
     UNATTENDED_LEVELS,
+    effective_autonomy,
     effective_consequence,
     effective_difficulty,
     effective_review_minutes,
@@ -180,3 +181,32 @@ def test_golden_sample_is_unchanged(tmp_path):
         GOLDEN.parent.mkdir(parents=True, exist_ok=True)
         GOLDEN.write_text(json.dumps(got, indent=2, sort_keys=True) + "\n")
     assert got == json.loads(GOLDEN.read_text())
+
+
+def test_autonomy_cap_is_executable_not_prose(tmp_path):
+    """Until 2026-08-30 the per-work-type caps lived only in AUTONOMY.md, applied
+    by whichever agent happened to read it — so nothing could compute what a
+    prompt's autonomy actually resolves to, and the batch planner proposed work
+    that would park the moment it ran."""
+    from _sizing import autonomy_cap, effective_autonomy
+
+    assert autonomy_cap("refactor", "large") == "safe"
+    assert autonomy_cap("bug", "small") == "supervised"
+    assert autonomy_cap("release", "small") == "human-required"
+    # feature/docs were raised on calibration evidence, but only to `medium`.
+    assert autonomy_cap("feature", "medium") == "safe"
+    assert autonomy_cap("feature", "large") == "supervised"
+
+
+def test_the_more_restrictive_of_header_and_cap_wins(tmp_path):
+    body = HEADERLESS_DOC.replace("Type: docs", "Type: bug").replace(
+        "Priority: normal", "Autonomy: safe\nPriority: normal")
+    p = _prompt(tmp_path, body, "draft/bug/pyautobrain/t.md")
+    level, _s, _f, _d = effective_difficulty(p)
+    assert effective_autonomy(p, level) == ("supervised", "supervised", "safe")
+
+
+def test_a_missing_header_is_human_required_not_a_default(tmp_path):
+    p = _prompt(tmp_path, HEADERLESS_DOC.replace("Difficulty: small", ""))
+    level, _s, _f, _d = effective_difficulty(p)
+    assert effective_autonomy(p, level)[0] == "human-required"