diff --git a/agents/conductors/batch/AGENTS.md b/agents/conductors/batch/AGENTS.md
index f59eb2a..08d5454 100644
--- a/agents/conductors/batch/AGENTS.md
+++ b/agents/conductors/batch/AGENTS.md
@@ -214,6 +214,20 @@ 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.
+**Picks are named for a human (PyAutoBrain#579).** Each pick
+(`_orchestrator.pick_record`, shared by dispatch and shadow) keeps the prompt's
+title, size and repos, and the digest names it as
+`[
]() (``; · )`. The link is built
+from the brief's `issue:` repository; a pick recorded without a title falls
+back to the bare slug, and `validate` reads the backticked slug either way. The
+`why:` is the brief's admission rule plus `_brief.rank_note`, which is the
+planner's actual order: `queue.md #k`, or `not in queue.md · ~N review-min ·
+priority P` (then the path, so a tie is alphabetical), plus `epic E phase X`.
+Shadow shift 1 (2026-11-04) showed why: six bare slugs, each `why: scoped tier
+notify`, could not be reviewed. A shadow digest prints its counters as one line
+(`- all counters 0 — shadow shift, nothing dispatched · review-minutes-planned: N`),
+which `validate` accepts in place of the nine.
+
**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
@@ -222,7 +236,11 @@ 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
+`?` for the human) in the same Mind commit as the digest. In a **shadow** row
+`merges` is a `?` too: the orchestrator merges nothing, so the baseline is the
+human's own work in the window (the epic: "the human works as today, minutes
+are logged"), with `review-minutes-actual` the human's own review time and
+`merges` their own merges, both counted at check-in. A shift whose
`digest.posted` exists writes nothing more — no second append, row or comment.
**Unattended `glance` close-out (phase 4b).** A `glance` worker under a brief
diff --git a/agents/conductors/batch/_batch.py b/agents/conductors/batch/_batch.py
index 63e5769..8380101 100644
--- a/agents/conductors/batch/_batch.py
+++ b/agents/conductors/batch/_batch.py
@@ -168,10 +168,19 @@ def survey(mind: Path) -> list[dict]:
g["phase"] = _phase_of(text)
g["done"] = _is_done(text)
g["themes"] = _themes_of(text)
+ g["title"] = _title_of(text)
out.append(g)
return out
+def _title_of(text: str) -> str:
+ """The prompt's first `# ` heading — what a human reads the pick by."""
+ for line in text.splitlines():
+ if line.startswith("# "):
+ return line[2:].strip()
+ return ""
+
+
#: A `queue.md` section heading — `## . `. The digits matter: the
#: file documents its own schema in a fenced block whose heading is the literal
#: `## . `, and a reader that took it would rank a placeholder.
@@ -332,6 +341,7 @@ def plan(records: list[dict], *, budget: int = DEFAULT_REVIEW_BUDGET,
# the question is what fits in it.
for r in pool:
r["queue_rank"] = queue_rank(r, queue)
+ r["queued"] = r["queue_rank"] < len(queue)
pool.sort(key=lambda r: (r["queue_rank"], r["review_minutes"],
priority_rank(r), r["path"]))
@@ -2862,12 +2872,14 @@ def _main_digest(mind: Path, a) -> int:
import _digest
import _orchestrator
record = Path(a.record)
- brief = _brief.parse(record.read_text(encoding="utf-8"))["brief"]
+ rec = _brief.parse(record.read_text(encoding="utf-8"))
+ brief = rec["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="")
+ issue = (rec["top"].get("issue") or {}).get("value", "")
+ print(_digest.build(brief, slot, _orchestrator.read_counters(slot), issue=issue), end="")
return RC_OK
diff --git a/agents/conductors/batch/_brief.py b/agents/conductors/batch/_brief.py
index 9e65534..e06d22f 100644
--- a/agents/conductors/batch/_brief.py
+++ b/agents/conductors/batch/_brief.py
@@ -353,7 +353,8 @@ def _scoped_match(record: dict, scoped: dict[str, list[str]]) -> str:
def pick_reason(record: dict, brief: dict) -> str:
- """Why the brief admits this record — the digest's one-line `why:`.
+ """Why the brief admits this record, then why it ranked where it did — 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.
@@ -362,7 +363,31 @@ def pick_reason(record: dict, brief: dict) -> str:
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"
+ rule = "backlog: fill-idle"
+ elif backlog == "scoped":
+ rule = _scoped_match(record, scoped_of(brief))
+ if not rule:
+ return "NOT RECORDED"
+ else:
+ return "NOT RECORDED"
+ rank = rank_note(record)
+ return f"{rule} · {rank}" if rank else rule
+
+
+def rank_note(record: dict) -> str:
+ """The planner's order, as it applied to this record (`_batch.plan`):
+ queue.md position first; below the queue, cheapest review first, then
+ priority, then the path — so a tie is alphabetical, and says so."""
+ parts = []
+ if record.get("queued"):
+ parts.append(f"queue.md #{record['queue_rank'] + 1}")
+ elif "queue_rank" in record:
+ parts.append("not in queue.md")
+ if record.get("review_minutes") is not None:
+ parts.append(f"~{record['review_minutes']} review-min")
+ parts.append(f"priority {record.get('priority') or 'normal'}")
+ if record.get("epic"):
+ phase = record.get("phase")
+ parts.append(f"epic {record['epic']}" + (f" phase {phase:g}" if isinstance(phase, (int, float))
+ and phase != float("inf") else ""))
+ return " · ".join(parts)
diff --git a/agents/conductors/batch/_digest.py b/agents/conductors/batch/_digest.py
index 1ada1bb..e8d5a4b 100644
--- a/agents/conductors/batch/_digest.py
+++ b/agents/conductors/batch/_digest.py
@@ -36,6 +36,11 @@
#: 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"
+#: A shadow digest's counters, in one line: every one is 0 by construction.
+SHADOW_COUNTERS = "- all counters 0 — shadow shift, nothing dispatched"
+#: A pick line names its task in backticks after its title (`label`).
+_SLUG_RE = re.compile(r"\(`([^`]+)`")
+_ISSUE_RE = re.compile(r"^(https://github\.com/[^/]+/[^/]+)/issues/\d+")
#: The pilot ledger (phase 4a). A subfolder, because every `batches/*.md` is
#: read as a batch record (lifecycle.py, `_batch.records`, the calibration rule).
@@ -45,6 +50,10 @@
"drafts-filed", "tokens", "workers-crashed")
#: Cells only the human can fill, at check-in.
PILOT_HUMAN = ("review-minutes-actual", "overrules")
+#: In a shadow shift the orchestrator merges nothing, so `merges` is the human's
+#: own in the window — the baseline the live shifts are measured against
+#: (epic: "the human works as today, minutes are logged"), counted at check-in.
+PILOT_HUMAN_SHADOW = ("merges",)
PILOT_HEADER = ("| " + " | ".join(PILOT_COLUMNS) + " |\n"
+ "|" + "|".join("---" for _ in PILOT_COLUMNS) + "|\n")
@@ -145,9 +154,31 @@ def _lease_state(lease: dict, agents) -> str:
return f"running (agent {agent}, {state}, since {since})"
-def _line(task: str, counters: dict, tier: str, health) -> str:
+def mind_blob_base(issue_url: str) -> str:
+ """`https://github.com///blob/main/` from the brief's `issue:`
+ (a Mind issue), so a pick links its prompt; "" when it cannot tell."""
+ m = _ISSUE_RE.match(issue_url or "")
+ return f"{m.group(1)}/blob/main/" if m else ""
+
+
+def label(task: str, pick: dict | None, base: str = "") -> str:
+ """How a pick is named to a human: its title, linked to its prompt, the
+ task slug in backticks, then size and repos. A pick recorded before titles
+ were (or with none) falls back to the bare slug."""
+ pick = pick or {}
+ title = pick.get("title")
+ if not title:
+ return task
+ prompt = pick.get("prompt") or ""
+ head = f"[{title}]({base}{prompt})" if base and prompt else f"**{title}**"
+ details = " · ".join(x for x in (pick.get("size"), ", ".join(pick.get("repos") or [])) if x)
+ return f"{head} (`{task}`" + (f"; {details}" if details else "") + ")"
+
+
+def _line(task: str, counters: dict, tier: str, health, base: str = "") -> str:
state, tiered = task_state(task, counters, health)
- return f"- {task} — {state}" + (f" — {tier}" if tiered else "")
+ name = label(task, counters.get("picks", {}).get(task), base)
+ return f"- {name} — {state}" + (f" — {tier}" if tiered else "")
def _packet(task: str, slot: Path, counters: dict) -> str:
@@ -162,7 +193,7 @@ def _packet(task: str, slot: Path, counters: dict) -> str:
def build(brief: dict, slot: Path, counters: dict, *, tiers: dict | None = None,
- health=None) -> str:
+ health=None, issue: str = "") -> str:
"""The `## Digest` block for one shift, as Markdown."""
tiers = tiers or {}
health = health or pr_health
@@ -171,10 +202,11 @@ def build(brief: dict, slot: Path, counters: dict, *, tiers: dict | None = None,
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)
+ base = mind_blob_base(issue)
+ lines += [_line(t, counters, tiers.get(t, picks.get(t, {}).get("tier", "?")), health, base)
for t in named] or [NO_PICKS]
lines += ["", "### Backlog picks"]
- lines += [_line(t, counters, tiers.get(t, picks[t].get("tier", "?")), health)
+ lines += [_line(t, counters, tiers.get(t, picks[t].get("tier", "?")), health, base)
+ f" — why: {picks[t].get('why') or 'NOT RECORDED'}" for t in backlog] or [NO_PICKS]
started = [t for t in named + backlog
@@ -185,7 +217,11 @@ def build(brief: dict, slot: Path, counters: dict, *, tiers: dict | None = None,
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]
+ lines += ["", "### Counters"]
+ if _brief.is_shadow(brief):
+ lines.append(f"{SHADOW_COUNTERS} · review-minutes-planned: {values['review-minutes-planned']}")
+ else:
+ lines += [f"- {k}: {values[k]}" for k in COUNTERS]
return "\n".join(lines) + "\n"
@@ -206,7 +242,8 @@ def pilot_row(slot_name: str, brief: dict, counters: dict) -> str:
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])
+ human = PILOT_HUMAN + (PILOT_HUMAN_SHADOW if cells["mode"] == "shadow" else ())
+ return "| " + " | ".join("?" if c in human else str(cells[c])
for c in PILOT_COLUMNS) + " |\n"
@@ -227,6 +264,16 @@ def _sections(text: str) -> dict[str, list[str]]:
return out
+def _slug(line: str) -> str:
+ """The task a digest bullet is about: the backticked slug a titled pick
+ carries, else the bullet's first word (a pick recorded without a title)."""
+ m = _BULLET_RE.match(line)
+ if not m:
+ return ""
+ s = _SLUG_RE.search(line.split(" — ", 1)[0])
+ return s.group(1) if s else m.group(1)
+
+
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:
@@ -237,26 +284,27 @@ def validate(text: str) -> list[str]:
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:
+ slug = _slug(line)
+ if slug 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:` — "
+ errs.append(f"digest: backlog pick {slug} 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:
+ shadow = any(line.startswith(SHADOW_COUNTERS) for line in secs.get("### Counters", []))
+ for key in (() if shadow else 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 \
+ slug = _slug(line)
+ if slug 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")
+ and f"#### {slug}" not in packets:
+ errs.append(f"digest: {slug} has no proof packet")
return errs
diff --git a/agents/conductors/batch/_orchestrator.py b/agents/conductors/batch/_orchestrator.py
index 80f919c..dea9b3f 100644
--- a/agents/conductors/batch/_orchestrator.py
+++ b/agents/conductors/batch/_orchestrator.py
@@ -606,9 +606,7 @@ 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 "?"}
+ counters["picks"][task] = pick_record(member, brief)
if task in counters["pending_redispatch"]:
counters["pending_redispatch"].remove(task)
counters["redispatched"][task] = counters["redispatched"].get(task, 0) + 1
@@ -660,7 +658,8 @@ def finish_shift(brief_path: Path, mind: Path, slot: Path, now: _dt.datetime, *,
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)
+ issue = (top.get("issue") or {}).get("value", "")
+ digest = _digest.build(brief, slot, counters, health=health, issue=issue)
(slot / DIGEST_FILE).write_text(digest, encoding="utf-8")
result = {"action": "digest", "ledger": None, "post": None, "packet": None}
@@ -807,6 +806,18 @@ def stage2_row(now: _dt.datetime, task: str, pr: str, slot_name: str, gate: str,
return "| " + " | ".join(" ".join(str(c).replace("|", "/").split()) for c in cells) + " |"
+def pick_record(member: dict, brief: dict, *, proposed: bool = False) -> dict:
+ """What `counters.json` keeps about one pick — enough for the digest and
+ `batch watch` to show it by title, size and repo, not by file name."""
+ rec = {"prompt": member["path"], "why": _brief.pick_reason(member, brief),
+ "tier": member.get("consequence") or "?",
+ "title": member.get("title") or "", "size": member.get("difficulty") or "",
+ "repos": list(member.get("repos") or [])}
+ if proposed:
+ rec["proposed"] = True
+ return rec
+
+
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:`."""
@@ -817,10 +828,7 @@ def _propose(members: list[dict], brief: dict, counters: dict, max_sessions: int
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}
+ counters["picks"][task] = pick_record(member, brief, proposed=True)
out.append(task)
return out
diff --git a/agents/conductors/batch/_watch.py b/agents/conductors/batch/_watch.py
index c7b9c93..63436f9 100644
--- a/agents/conductors/batch/_watch.py
+++ b/agents/conductors/batch/_watch.py
@@ -106,6 +106,7 @@ def snapshot(brief_path: Path, slot: Path, now: _dt.datetime, *, agents=None,
state, _ = _digest.task_state(task, counters, health, live=True,
lease=held.get(task), agents=agents)
rows.append({"task": task, "state": state,
+ "label": _digest.label(task, picks.get(task)),
"tier": picks.get(task, {}).get("tier", "?"),
"why": "named in the brief" if task in named
else picks.get(task, {}).get("why") or "NOT RECORDED",
@@ -158,7 +159,7 @@ def render(snap: dict) -> str:
if not snap["started"]:
return "\n".join(lines + [NO_TICK]) + "\n"
lines.append("Tasks")
- lines += [f"- {r['task']} — {r['state']} — {r['tier']} — why: {r['why']}"
+ lines += [f"- {r['label']} — {r['state']} — {r['tier']} — why: {r['why']}"
+ (f" — re-dispatched {r['redispatched']}×" if r["redispatched"] else "")
for r in snap["tasks"]] or ["- none yet"]
b = snap["budgets"]
diff --git a/tests/test_orchestrator_digest.py b/tests/test_orchestrator_digest.py
index 41e5ba8..c2d8ee8 100644
--- a/tests/test_orchestrator_digest.py
+++ b/tests/test_orchestrator_digest.py
@@ -157,18 +157,22 @@ def test_digest_separates_brief_work_from_backlog_picks_with_reasons(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 counters["picks"]["picked"]["why"].startswith("scoped epic e1 · not in queue.md")
assert "outside" not in counters["picks"]
d = _digest.build(_brief.parse(brief.read_text())["brief"], slot, counters,
- health=_health)
+ health=_health, issue=ISSUE)
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
+ # Picks read by title, linked to the prompt on the Mind the issue lives in.
+ blob = "https://github.com/example-org/example-mind/blob/main/"
+ assert (f"- [named]({blob}draft/feature/pyautobrain/named.md) (`named`; small · pyautobrain)"
+ " — https://github.com/o/r/pull/1 — merged — glance") in work
assert "- never — not started" in work
# Green and glance: since phase 4b its close-out is running at review time.
- assert "- picked — https://github.com/o/r/pull/2 — close-out running — glance — why: scoped epic e1" in picks
- assert "named" not in picks
+ assert ("(`picked`; small · pyautobrain) — https://github.com/o/r/pull/2 — close-out running"
+ " — glance — why: scoped epic e1 · not in queue.md") in picks
+ assert "`named`" not in picks
def test_one_proof_packet_per_task_and_an_honest_stub_when_missing(tmp_path):
@@ -369,7 +373,11 @@ def run(argv, **kw):
@pytest.mark.parametrize("backlog,record,want", [
- ("fill-idle", {"path": "x/a.md"}, "backlog: fill-idle (the planner's ranking)"),
+ ("fill-idle", {"path": "x/a.md"}, "backlog: fill-idle"),
+ ("fill-idle", {"path": "x/a.md", "queue_rank": 2, "queued": True}, "backlog: fill-idle · queue.md #3"),
+ ("scoped", {"path": "x/a.md", "consequence": "notify", "queue_rank": 4, "queued": False,
+ "review_minutes": 3, "priority": "high", "epic": "e1", "phase": 2.0},
+ "scoped tier notify · not in queue.md · ~3 review-min · priority high · epic e1 phase 2"),
("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"),
diff --git a/tests/test_orchestrator_shadow.py b/tests/test_orchestrator_shadow.py
index 198faa2..3070859 100644
--- a/tests/test_orchestrator_shadow.py
+++ b/tests/test_orchestrator_shadow.py
@@ -45,8 +45,9 @@ def test_a_shadow_tick_proposes_with_reasons_and_spawns_nothing(tmp_path):
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"
+ "why": "named task", "tier": "glance", "proposed": True,
+ "title": "named", "size": "small", "repos": ["pyautobrain"]}
+ assert picks["picked"]["why"].startswith("scoped epic e1 · not in queue.md")
assert orch.read_counters(slot)["sessions"] == 0
@@ -70,12 +71,13 @@ def test_the_shadow_digest_lists_proposals_and_validates(tmp_path):
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 "- **named** (`named`; small · pyautobrain) — proposed (shadow) — glance" in d
+ assert "- **picked** (`picked`; small · pyautobrain) — proposed (shadow) — glance — why: scoped epic e1 · not in queue.md" 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
+ # Every counter is 0 in a shadow shift: one line, not nine.
+ assert _digest.SHADOW_COUNTERS + " · review-minutes-planned: 45" in d
+ assert "- merges: " not in d
rec = tmp_path / "with_digest.md"
rec.write_text(brief.read_text() + "\n" + d)
res = _brief.check(rec, mind)
@@ -102,6 +104,8 @@ def test_pilot_row_has_one_cell_per_column_and_leaves_the_humans_blank(tmp_path)
got = dict(zip(_digest.PILOT_COLUMNS, cells))
assert got["mode"] == "shadow" and got["slot"] == "shift"
assert got["review-minutes-actual"] == "?" and got["overrules"] == "?"
+ # Shadow merges are the human's own in the window — the baseline.
+ assert got["merges"] == "?"
assert got["review-minutes-planned"] == "45" and got["tokens"] == "0"
@@ -141,3 +145,48 @@ def test_a_shadow_shift_lands_exactly_one_pilot_row_on_mind_main(tmp_path, monke
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
+
+
+# --- readable picks (PyAutoBrain#579) ------------------------------------------
+
+def test_titled_picks_validate_and_a_missing_why_names_the_slug(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, issue=t.ISSUE)
+ assert _digest.validate(d) == []
+ import re
+ bad = re.sub(r" — why: scoped epic e1[^\n]*", " — why: NOT RECORDED", d)
+ assert any("backlog pick picked has no `why:`" in e for e in _digest.validate(bad))
+ # The one-line shadow counters are accepted only as that line: a live
+ # digest without its counters is still refused.
+ live = d.replace(_digest.SHADOW_COUNTERS, "- something else")
+ assert any("counter merges missing" in e for e in _digest.validate(live))
+
+
+def test_an_old_untitled_digest_still_validates():
+ """Shift 1 (2026-11-04) was digested before picks had titles."""
+ old = ("## Digest\n\n### Brief work\n- none\n\n### Backlog picks\n"
+ "- batch_slice — proposed (shadow) — notify — why: scoped tier notify\n\n"
+ "### Proof packets\n- none — shadow shift: nothing was dispatched\n\n### Counters\n"
+ + "".join(f"- {k}: 0\n" for k in _digest.COUNTERS))
+ assert _digest.validate(old) == []
+ assert _digest.label("batch_slice", {"prompt": "draft/x.md"}) == "batch_slice"
+
+
+def test_a_live_pilot_row_counts_the_orchestrators_merges(tmp_path):
+ brief = {"review-minutes": {"value": "30"}, "token-budget": {"value": "100"}}
+ counters = orch.read_counters(tmp_path)
+ counters["outcomes"] = {"a": {"prs": ["u"], "merged": ["u"]}}
+ cells = dict(zip(_digest.PILOT_COLUMNS,
+ [c.strip() for c in _digest.pilot_row("s", brief, counters).strip().strip("|").split("|")]))
+ assert cells["mode"] == "live" and cells["merges"] == "1"
+
+
+def test_batch_watch_shows_titles(tmp_path):
+ import _watch
+ mind, brief, home = _shadow(tmp_path)
+ orch.tick(brief, mind, home=home, now=t.NOW, spawner=NoSpawn())
+ text = _watch.render(_watch.snapshot(brief, orch.slot_dir(brief, home), t.NOW, health=t._health))
+ assert "- **named** (`named`; small · pyautobrain) — proposed (shadow)" in text