feat(batch): readable shift digest picks + shadow row measures the human baseline

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
+`[<title>](<Mind blob link>) (`<slug>`; <size> · <repos>)`. 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

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 — `## <n>. <label>`. The digits matter: the
 #: file documents its own schema in a fenced block whose heading is the literal
 #: `## <n>. <label>`, 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
 
 

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)

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/<owner>/<mind>/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

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
 

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