feat: autonomous orchestrators p4a — shadow shifts + the pilot ledger

agents/conductors/batch/AGENTS.md

@@ -214,6 +214,17 @@ NO-CHECKS). `glance-auto-merges`, `reverts`, `pushes`, `drafts-filed` and
 writes nothing; `batch brief --check` refuses a digest missing a section, a
 counter, a pick's `why:` or a started task's packet.
 
+**Shadow shifts and the pilot ledger (phase 4a).** Under `shadow: yes` a tick
+runs the same plan inside the brief but, instead of spawning, records each
+proposal in `picks` with `proposed: true` (`_propose`, capped by
+`max-sessions:`): no lease, no worker, no session counted. The digest marks
+them `proposed (shadow)` and its packet section says nothing was dispatched;
+`brief --check` asks no packet of a proposal. `finish_shift` also appends one
+row to `PyAutoMind/batches/pilot/pilot.md` (`_digest.pilot_row`: the digest's
+counters plus slot and mode; `review-minutes-actual` and `overrules` land as
+`?` for the human) in the same Mind commit as the digest. A shift whose
+`digest.posted` exists writes nothing more — no second append, row or comment.
+
 **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/_brief.py

@@ -138,6 +138,11 @@ def worker_mode_of(brief: dict) -> str:
     return brief.get("worker-mode", {}).get("value") or DEFAULT_WORKER_MODE
 
 
+def is_shadow(brief: dict) -> bool:
+    """A shadow shift (phase 4a) plans and records proposals but spawns nothing."""
+    return brief.get("shadow", {}).get("value", "").lower() == "yes"
+
+
 def is_sandbox(brief: dict) -> bool:
     return brief.get("sandbox", {}).get("value", "").lower() == "yes"
 
@@ -242,8 +247,9 @@ def check(path: Path, mind: Path) -> dict:
                       + ", ".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")
+    for key in ("sandbox", "shadow"):
+        if key in brief and brief[key]["value"].lower() not in ("yes", "no"):
+            errors.append(f"brief.{key}: {brief[key]['value']} — yes or no")
 
     lanes = lanes_of(brief)
     if not lanes:

agents/conductors/batch/_digest.py

@@ -33,6 +33,20 @@
                   "packets-sent": "packets_sent"}
 PROOFS_DIR = "proofs"
 NO_PICKS = "- none"
+#: What a shadow shift's line and packet section say instead of work (phase 4a).
+PROPOSED = "proposed (shadow)"
+NO_PACKETS_SHADOW = "- none — shadow shift: nothing was dispatched"
+
+#: The pilot ledger (phase 4a). A subfolder, because every `batches/*.md` is
+#: read as a batch record (lifecycle.py, `_batch.records`, the calibration rule).
+PILOT_PATH = "batches/pilot/pilot.md"
+PILOT_COLUMNS = ("slot", "mode", "review-minutes-planned", "review-minutes-actual",
+                 "merges", "glance-auto-merges", "reverts", "overrules", "pushes",
+                 "drafts-filed", "tokens", "workers-crashed")
+#: Cells only the human can fill, at check-in.
+PILOT_HUMAN = ("review-minutes-actual", "overrules")
+PILOT_HEADER = ("| " + " | ".join(PILOT_COLUMNS) + " |\n"
+                + "|" + "|".join("---" for _ in PILOT_COLUMNS) + "|\n")
 
 _BULLET_RE = re.compile(r"^- (\S+)")
 
@@ -78,6 +92,8 @@ def pr_health(url: str, run=None) -> str:
 
 def _line(task: str, counters: dict, tier: str, health) -> str:
     out = counters.get("outcomes", {}).get(task)
+    if counters.get("picks", {}).get(task, {}).get("proposed"):
+        return f"- {task} — {PROPOSED} — {tier}"
     if task in counters.get("parked", []):
         return f"- {task} — parked: awaiting-input (blocked on a person)"
     if task in counters.get("abandoned", []):
@@ -124,17 +140,33 @@ def build(brief: dict, slot: Path, counters: dict, *, tiers: dict | None = None,
                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]
+    lines += ["\n\n".join(_packet(t, slot, counters) for t in started)
+              or (NO_PACKETS_SHADOW if _brief.is_shadow(brief) else NO_PICKS)]
 
+    values = counter_values(brief, counters)
+    lines += ["", "### Counters"] + [f"- {k}: {values[k]}" for k in COUNTERS]
+    return "\n".join(lines) + "\n"
+
+
+def counter_values(brief: dict, counters: dict) -> dict:
+    """Every digest counter's value, from the brief and `counters.json`."""
     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"
+    return values
+
+
+def pilot_row(slot_name: str, brief: dict, counters: dict) -> str:
+    """One `batches/pilot/pilot.md` row; the human's cells land as `?`."""
+    v = counter_values(brief, counters)
+    cells = {**v, "slot": slot_name, "tokens": counters.get("tokens", 0),
+             "mode": "shadow" if _brief.is_shadow(brief) else "live"}
+    return "| " + " | ".join("?" if c in PILOT_HUMAN else str(cells[c])
+                             for c in PILOT_COLUMNS) + " |\n"
 
 
 # --- validate ------------------------------------------------------------------
@@ -183,6 +215,7 @@ def validate(text: str) -> list[str]:
         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 PROPOSED 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/_orchestrator.py

@@ -512,6 +512,12 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents,
     if records is None:
         records = _batch.survey(mind)
     d = _batch.plan(records, session_lane="local-dev", brief=brief)
+    if _brief.is_shadow(brief):
+        proposed = _propose(d["members"], brief, counters, max_sessions)
+        _write_counters(slot, counters)
+        return {"action": "shadow", "reaped": reaped, "spawned": [],
+                "proposed": proposed, "refused": [], "live": sorted(live),
+                "ledger": ledger}
     spawned, refused = [], []
     for member in d["members"]:
         task = task_id(member["path"])
@@ -587,6 +593,11 @@ def finish_shift(brief_path: Path, mind: Path, slot: Path, now: _dt.datetime, *,
     watcher's network surface" — a second comment retires the watcher). Nothing
     is read back from GitHub to check.
     """
+    if (slot / DIGEST_MARKER).exists():
+        # A finished shift writes nothing more: no second record append, no
+        # second pilot row, no second comment.
+        return {"action": "digest", "ledger": None, "packet": None,
+                "post": {"posted": False, "error": "already posted this shift"}}
     rec = _brief.parse(Path(brief_path).read_text(encoding="utf-8"))
     brief, top = rec["brief"] or {}, rec["top"]
     counters = read_counters(slot)
@@ -598,6 +609,11 @@ def finish_shift(brief_path: Path, mind: Path, slot: Path, now: _dt.datetime, *,
     if rel is not None:
         _mind_writer.queue_op(slot / _mind_writer.LEDGER_DIR,
                               {"op": "append", "path": rel, "text": "\n" + digest})
+        # The pilot ledger (phase 4a): one row per shift, in the same commit.
+        header = "" if (Path(mind) / _digest.PILOT_PATH).is_file() else _digest.PILOT_HEADER
+        _mind_writer.queue_op(slot / _mind_writer.LEDGER_DIR,
+                              {"op": "append", "path": _digest.PILOT_PATH,
+                               "text": header + _digest.pilot_row(slot.name, brief, counters)})
         try:
             result["ledger"] = _mind_writer.commit_ledger(
                 slot, mind, f"shift {slot.name}: digest")
@@ -634,6 +650,24 @@ def _write_packet(slot: Path, digest: str) -> Path:
     return out
 
 
+def _propose(members: list[dict], brief: dict, counters: dict, max_sessions: int) -> list[str]:
+    """A shadow shift's tick (phase 4a): record what a live shift would have
+    dispatched — no lease, no worker — up to the brief's `max-sessions:`."""
+    out = []
+    for member in members:
+        task = task_id(member["path"])
+        if task in counters["picks"]:
+            continue
+        if len(counters["picks"]) >= max_sessions:
+            break
+        counters["picks"][task] = {"prompt": member["path"],
+                                   "why": _brief.pick_reason(member, brief),
+                                   "tier": member.get("consequence") or "?",
+                                   "proposed": True}
+        out.append(task)
+    return out
+
+
 def main(argv=None) -> int:
     import argparse
     ap = argparse.ArgumentParser(

skills/batch/batch.md

@@ -135,7 +135,9 @@ and the one issue comment are the watcher's, at `review-at:`.
 
 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:`.
+and posted once on the brief's `issue:`. A `shadow: yes` brief plans and digests
+but spawns nothing — the pilot's baseline; every shift adds one row to
+`PyAutoMind/batches/pilot/pilot.md`, whose `?` cells are the human's at check-in.
 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)

tests/test_orchestrator_shadow.py

@@ -0,0 +1,143 @@
+"""Autonomous orchestrators phase 4a (PyAutoBrain#570): shadow shifts + the pilot ledger.
+
+The pilot's baseline is two shadow shifts: the orchestrator proposes and
+digests, the human works as today. A shadow tick plans inside the brief and
+records what it would have dispatched — no lease, no worker — and the shift's
+`finish_shift` appends one row to `batches/pilot/pilot.md`.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+
+BRAIN = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BRAIN / "tests"))
+import test_orchestrator_digest as t  # noqa: E402 — the 3d fixtures
+from test_orchestrator_digest import _brief, _digest, orch  # noqa: E402
+
+SHADOW = "  - shadow: yes\n"
+
+
+class NoSpawn:
+    def spawn(self, *a, **kw):
+        raise AssertionError("a shadow shift must never spawn")
+
+
+def _shadow(tmp_path: Path, extra=SHADOW, mind_root=None):
+    mind = mind_root or t._mind(tmp_path / "mind")
+    brief = (mind / "batches" / "2026-11-04-shadow.md") if mind_root else tmp_path / "shift.md"
+    brief.parent.mkdir(parents=True, exist_ok=True)
+    brief.write_text(t.BRIEF.format(issue=t.ISSUE, packet="") + extra)
+    return mind, brief, tmp_path / "home"
+
+
+def test_a_shadow_tick_proposes_with_reasons_and_spawns_nothing(tmp_path):
+    mind, brief, home = _shadow(tmp_path)
+    r = orch.tick(brief, mind, home=home, now=t.NOW, spawner=NoSpawn())
+    assert r["action"] == "shadow" and r["spawned"] == []
+    assert sorted(r["proposed"]) == ["named", "picked"]
+    slot = orch.slot_dir(brief, home)
+    assert orch.leases(slot) == {}
+    picks = orch.read_counters(slot)["picks"]
+    assert picks["named"] == {"prompt": "draft/feature/pyautobrain/named.md",
+                              "why": "named task", "tier": "glance", "proposed": True}
+    assert picks["picked"]["why"] == "scoped epic e1"
+    assert orch.read_counters(slot)["sessions"] == 0
+
+
+def test_a_second_shadow_tick_proposes_nothing_new(tmp_path):
+    mind, brief, home = _shadow(tmp_path)
+    orch.tick(brief, mind, home=home, now=t.NOW, spawner=NoSpawn())
+    r = orch.tick(brief, mind, home=home, now=t.NOW + _dt.timedelta(minutes=1),
+                  spawner=NoSpawn())
+    assert r["proposed"] == []
+
+
+def test_proposals_stop_at_max_sessions(tmp_path):
+    mind, brief, home = _shadow(tmp_path, extra=SHADOW + "  - max-sessions: 1\n")
+    r = orch.tick(brief, mind, home=home, now=t.NOW, spawner=NoSpawn())
+    assert r["proposed"] == ["named"]
+
+
+def test_the_shadow_digest_lists_proposals_and_validates(tmp_path):
+    mind, brief, home = _shadow(tmp_path)
+    orch.tick(brief, mind, home=home, now=t.NOW, spawner=NoSpawn())
+    slot = orch.slot_dir(brief, home)
+    d = _digest.build(_brief.parse(brief.read_text())["brief"], slot,
+                      orch.read_counters(slot), health=t._health)
+    assert "- named — proposed (shadow) — glance" in d
+    assert "- picked — proposed (shadow) — glance — why: scoped epic e1" in d
+    assert "- never — not started" in d
+    assert _digest.NO_PACKETS_SHADOW in d
+    for key in _digest.COUNTERS:
+        assert f"- {key}: " in d
+    rec = tmp_path / "with_digest.md"
+    rec.write_text(brief.read_text() + "\n" + d)
+    res = _brief.check(rec, mind)
+    assert res["ok"], res["errors"]
+
+
+@pytest.mark.parametrize("value,ok", [("yes", True), ("no", True), ("maybe", False)])
+def test_brief_check_validates_shadow(tmp_path, value, ok):
+    mind, brief, _ = _shadow(tmp_path, extra=f"  - shadow: {value}\n")
+    res = _brief.check(brief, mind)
+    assert res["ok"] is ok
+    if not ok:
+        assert "brief.shadow: maybe — yes or no" in res["errors"]
+
+
+def test_pilot_row_has_one_cell_per_column_and_leaves_the_humans_blank(tmp_path):
+    mind, brief, home = _shadow(tmp_path)
+    orch.tick(brief, mind, home=home, now=t.NOW, spawner=NoSpawn())
+    slot = orch.slot_dir(brief, home)
+    row = _digest.pilot_row(slot.name, _brief.parse(brief.read_text())["brief"],
+                            orch.read_counters(slot))
+    cells = [c.strip() for c in row.strip().strip("|").split("|")]
+    assert len(cells) == len(_digest.PILOT_COLUMNS)
+    got = dict(zip(_digest.PILOT_COLUMNS, cells))
+    assert got["mode"] == "shadow" and got["slot"] == "shift"
+    assert got["review-minutes-actual"] == "?" and got["overrules"] == "?"
+    assert got["review-minutes-planned"] == "45" and got["tokens"] == "0"
+
+
+@pytest.mark.skipif(not t.LEDGER_MERGE.is_file() or shutil.which("git") is None,
+                    reason="needs git and PyAutoMind's scripts/ledger_merge.py")
+def test_a_shadow_shift_lands_exactly_one_pilot_row_on_mind_main(tmp_path, monkeypatch):
+    monkeypatch.setattr(sys, "dont_write_bytecode", False)
+    bare = tmp_path / "remote.git"
+    t.git(tmp_path, "init", "-q", "--bare", "-b", "main", str(bare))
+    canon = tmp_path / "canon"
+    t.git(tmp_path, "clone", "-q", str(bare), str(canon))
+    t.git(canon, "config", "user.name", "Test")
+    t.git(canon, "config", "user.email", "test@example.invalid")
+    t._mind(canon)
+    (canon / "scripts").mkdir()
+    shutil.copy(t.LEDGER_MERGE, canon / "scripts" / "ledger_merge.py")
+    (canon / "batches" / "pilot").mkdir(parents=True)
+    (canon / "batches" / "pilot" / "pilot.md").write_text("# Pilot\n\n" + _digest.PILOT_HEADER)
+    mind, brief, home = _shadow(tmp_path, mind_root=canon)
+    t.git(canon, "add", "-A")
+    t.git(canon, "commit", "-q", "-m", "seed")
+    t.git(canon, "push", "-q", "origin", "HEAD:main")
+
+    orch.tick(brief, canon, home=home, now=t.NOW, spawner=NoSpawn())
+    r = orch.tick(brief, canon, home=home, now=t.AFTER, poster=t.Poster(), health=t._health)
+    assert r["ledger"]["committed"], r
+    pilot = t.git(bare, "show", "main:batches/pilot/pilot.md").stdout
+    rows = [ln for ln in pilot.splitlines() if ln.startswith("| 2026-11-04-shadow ")]
+    assert len(rows) == 1 and "| shadow |" in rows[0]
+    assert pilot.count("| slot |") == 1  # the header was not appended twice
+    landed = t.git(bare, "show", "main:batches/2026-11-04-shadow.md").stdout
+    assert "proposed (shadow)" in landed and _digest.validate(landed) == []
+
+    # A finished shift writes nothing more.
+    slot = orch.slot_dir(brief, home)
+    again = orch.finish_shift(brief, canon, slot, t.AFTER, poster=t.Poster(), health=t._health)
+    assert again["ledger"] is None
+    assert _digest.validate(t.git(bare, "show", "main:batches/2026-11-04-shadow.md").stdout) == []
+    assert t.git(bare, "show", "main:batches/pilot/pilot.md").stdout == pilot