diff --git a/agents/conductors/batch/AGENTS.md b/agents/conductors/batch/AGENTS.md
index 1cdeee4..4823209 100644
--- a/agents/conductors/batch/AGENTS.md
+++ b/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
diff --git a/agents/conductors/batch/_batch.py b/agents/conductors/batch/_batch.py
index c40c893..965978e 100644
--- a/agents/conductors/batch/_batch.py
+++ b/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:
diff --git a/agents/conductors/batch/_brief.py b/agents/conductors/batch/_brief.py
index d95eab5..1c20a65 100644
--- a/agents/conductors/batch/_brief.py
+++ b/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,10 +324,7 @@ 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:
@@ -331,3 +332,31 @@ def scope(records: list[dict], brief: dict) -> tuple[list[dict], list[tuple[str,
             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"
diff --git a/agents/conductors/batch/_digest.py b/agents/conductors/batch/_digest.py
new file mode 100644
index 0000000..4cf8120
--- /dev/null
+++ b/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
diff --git a/agents/conductors/batch/_mind_writer.py b/agents/conductors/batch/_mind_writer.py
index e087b83..04d0ebb 100644
--- a/agents/conductors/batch/_mind_writer.py
+++ b/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 []
diff --git a/agents/conductors/batch/_orchestrator.py b/agents/conductors/batch/_orchestrator.py
index a6e2a4b..2d28864 100644
--- a/agents/conductors/batch/_orchestrator.py
+++ b/agents/conductors/batch/_orchestrator.py
@@ -35,12 +35,16 @@
 from pathlib import Path
 
 import _brief  # noqa: E402 — the batch conductor puts this directory on sys.path
+import _digest  # noqa: E402
 import _mind_writer  # noqa: E402
 
 STOP_FILE = "STOP"
 LOCK_FILE = "tick.lock"
 COUNTERS_FILE = "counters.json"
 LOG_FILE = "tick.log"
+DIGEST_FILE = "digest.md"
+#: Created exclusively before the one post: its existence means "never again".
+DIGEST_MARKER = "digest.posted"
 #: A tick lock older than this belongs to a tick that died; it is broken.
 LOCK_TTL_S = 600
 #: A worker whose process is gone keeps its lease this long before it is
@@ -111,6 +115,8 @@ def read_counters(slot: Path) -> dict:
     c.setdefault("pending_redispatch", [])
     c.setdefault("parked", [])
     c.setdefault("tokens", 0)
+    c.setdefault("picks", {})      # task → {prompt, why, tier}, at dispatch (3d)
+    c.setdefault("outcomes", {})   # task → {prs, merged}, from its `.done` (3d)
     return c
 
 
@@ -376,6 +382,13 @@ def _reap(slot: Path, counters: dict, now: _dt.datetime, grace_s: float,
             sid = agents.session_of(lease["agent_id"]) or lease.get("session_id")
             counters["tokens"] += tokens(sid) if sid else 0
         if done_marker.exists():
+            # The digest reads these at review-at, long after the marker is gone.
+            counters["outcomes"][task] = _digest.parse_done(
+                done_marker.read_text(encoding="utf-8", errors="replace"))
+            proof = lease_dir / f"{task}.proof.md"
+            if proof.exists():
+                (slot / _digest.PROOFS_DIR).mkdir(exist_ok=True)
+                proof.replace(slot / _digest.PROOFS_DIR / f"{task}.md")
             path.unlink(missing_ok=True)
             done_marker.unlink(missing_ok=True)
             if task not in counters["done"]:
@@ -437,8 +450,11 @@ def brief_open(brief_path: Path, mind: Path, now: _dt.datetime) -> tuple[bool, s
 def tick(brief_path: Path, mind: Path, *, home: Path | None = None,
          now: _dt.datetime | None = None, spawner=None,
          grace_s: float = LEASE_GRACE_S, records: list[dict] | None = None,
-         agents: AgentsView | None = None, tokens=transcript_tokens) -> dict:
-    """One watcher tick. Returns what it did; also appended to `tick.log`."""
+         agents: AgentsView | None = None, tokens=transcript_tokens,
+         poster=None, health=None) -> dict:
+    """One watcher tick. Returns what it did; also appended to `tick.log`.
+
+    After `review-at:` the one remaining act is the digest (phase 3d), once."""
     now = now or _now()
     brief_path = Path(brief_path)
     slot = slot_dir(brief_path, home)
@@ -449,6 +465,17 @@ def tick(brief_path: Path, mind: Path, *, home: Path | None = None,
 
     is_open, why, brief = brief_open(brief_path, mind, now)
     if not is_open:
+        if why.startswith("review-at") and (slot / "leases").is_dir() \
+                and not (slot / DIGEST_MARKER).exists():
+            if not _take_lock(slot, now):
+                return {"action": "busy", "reason": "another tick holds the lock"}
+            try:
+                result = finish_shift(brief_path, mind, slot, now, poster=poster,
+                                      health=health)
+            finally:
+                _drop_lock(slot)
+            _log(slot, now, result)
+            return result
         return {"action": "idle", "reason": why}
 
     (slot / "leases").mkdir(parents=True, exist_ok=True)
@@ -515,6 +542,9 @@ 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 "?"}
         if task in counters["pending_redispatch"]:
             counters["pending_redispatch"].remove(task)
             counters["redispatched"][task] = counters["redispatched"].get(task, 0) + 1
@@ -527,6 +557,83 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents,
             "refused": refused, "live": sorted(live), "ledger": ledger}
 
 
+# --- the digest (phase 3d) ----------------------------------------------------------
+
+def gh_comment(issue_url: str, body_file: Path, run=None) -> dict:
+    """The watcher's one write: a comment on the brief's `issue:`."""
+    run = run or subprocess.run
+    res = run(["gh", "issue", "comment", issue_url, "--body-file", str(body_file)],
+              capture_output=True, text=True, timeout=120)
+    if res.returncode != 0:
+        return {"posted": False, "error": (res.stderr or res.stdout).strip()[:300]}
+    return {"posted": True, "url": res.stdout.strip().splitlines()[-1] if res.stdout.strip() else ""}
+
+
+def _record_rel(brief_path: Path, mind: Path) -> str | None:
+    """The record's path inside Mind, for the one-Mind-writer append; None when
+    the record does not live in that checkout (a test brief, a scratch file)."""
+    try:
+        return Path(brief_path).resolve().relative_to(Path(mind).resolve()).as_posix()
+    except ValueError:
+        return None
+
+
+def finish_shift(brief_path: Path, mind: Path, slot: Path, now: _dt.datetime, *,
+                 poster=None, health=None) -> dict:
+    """Write the digest into the record and post it once on the brief's issue.
+
+    The marker is created **before** the post, exclusively: a crash between the
+    two leaves a shift with no comment, never one with two (AUTONOMY.md "The
+    watcher's network surface" — a second comment retires the watcher). Nothing
+    is read back from GitHub to check.
+    """
+    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)
+    (slot / DIGEST_FILE).write_text(digest, encoding="utf-8")
+    result = {"action": "digest", "ledger": None, "post": None, "packet": None}
+
+    rel = _record_rel(brief_path, mind)
+    if rel is not None:
+        _mind_writer.queue_op(slot / _mind_writer.LEDGER_DIR,
+                              {"op": "append", "path": rel, "text": "\n" + digest})
+        try:
+            result["ledger"] = _mind_writer.commit_ledger(
+                slot, mind, f"shift {slot.name}: digest")
+        except _mind_writer.MindWriteError as e:
+            result["ledger"] = {"committed": None, "error": str(e)}
+
+    if top.get("packet", {}).get("value", "").lower() == "yes":
+        result["packet"] = str(_write_packet(slot, digest))
+
+    issue = top.get("issue", {}).get("value", "")
+    claim = {"at": _iso(now), "issue": issue}
+    if not _create_exclusive(slot / DIGEST_MARKER, claim):
+        result["post"] = {"posted": False, "error": "already posted this shift"}
+        return result
+    if not issue.startswith("http"):
+        result["post"] = {"posted": False, "error": "the brief names no issue:"}
+    else:
+        result["post"] = (poster or gh_comment)(issue, slot / DIGEST_FILE)
+    (slot / DIGEST_MARKER).write_text(json.dumps({**claim, **result["post"]},
+                                                 sort_keys=True))
+    return result
+
+
+def _write_packet(slot: Path, digest: str) -> Path:
+    """The opt-in HTML packet (`packet: yes`): the digest on one page."""
+    import html
+    css = Path(__file__).with_name("packet.css")
+    style = css.read_text(encoding="utf-8") if css.is_file() else ""
+    page = ("<!doctype html><meta charset=\"utf-8\"><title>Shift digest "
+            f"{html.escape(slot.name)}</title><style>{style}</style>"
+            f"<main><pre>{html.escape(digest)}</pre></main>\n")
+    out = slot / "digest.html"
+    out.write_text(page, encoding="utf-8")
+    return out
+
+
 def main(argv=None) -> int:
     import argparse
     ap = argparse.ArgumentParser(
diff --git a/agents/conductors/batch/worker_template.md b/agents/conductors/batch/worker_template.md
index fc22004..c9f6097 100644
--- a/agents/conductors/batch/worker_template.md
+++ b/agents/conductors/batch/worker_template.md
@@ -14,6 +14,12 @@ The terms of this shift (AUTONOMY.md "Shift briefs and unattended `glance`"):
   change with `{shift_ledger} append|write|move|remove|drop-section ...`
   (ops land in {ledger_dir}/) and the orchestrator commits them. This covers
   your `active.md` claim, the `draft/ → active/` move and your close-out.
-- When you finish, write {lease_dir}/{task}.done containing the URLs of the PRs
-  you opened (or the issue you parked on), one per line. Without that file the
-  watcher counts your session as crashed.
+- When you finish, write {lease_dir}/{task}.proof.md first: one proof packet in
+  the format of PyAutoMind/batches/AGENTS.md "Proof packet", headed
+  `#### {task} — <PR url>`, with the Witness: output trimmed to the lines that
+  decide it. It goes into the shift digest verbatim; leave it out and the
+  digest says you wrote none.
+- Then write {lease_dir}/{task}.done containing the URLs of the PRs you opened
+  (or the issue you parked on), one per line — `merged <url>` for a PR you
+  merged in-turn under its tier. Without that file the watcher counts your
+  session as crashed.
diff --git a/bin/shift_ledger.py b/bin/shift_ledger.py
index b8cad8f..69024b0 100755
--- a/bin/shift_ledger.py
+++ b/bin/shift_ledger.py
@@ -15,10 +15,8 @@
 The ledger directory is `$PYAUTO_SHIFT_LEDGER_DIR` (set for every worker).
 """
 import argparse
-import json
 import os
 import sys
-import time
 from pathlib import Path
 
 BATCH = Path(__file__).resolve().parents[1] / "agents" / "conductors" / "batch"
@@ -54,13 +52,8 @@ def main(argv=None) -> int:
         if errs:
             print("shift_ledger: refused — " + "; ".join(errs), file=sys.stderr)
             return 1
-    d = Path(ledger)
-    d.mkdir(parents=True, exist_ok=True)
-    name = f"{time.time_ns():020d}-{os.getpid()}-{a.op}.json"
-    tmp = d / (name + ".tmp")
-    tmp.write_text(json.dumps(op, indent=2), encoding="utf-8")
-    tmp.replace(d / name)
-    print(f"shift_ledger: queued {a.op} {a.path} → {d / name}")
+    queued = mw.queue_op(Path(ledger), op)
+    print(f"shift_ledger: queued {a.op} {a.path} → {queued}")
     return 0
 
 
diff --git a/skills/batch/batch.md b/skills/batch/batch.md
index 036f69a..7bb9882 100644
--- a/skills/batch/batch.md
+++ b/skills/batch/batch.md
@@ -127,13 +127,15 @@ brief against the `batches/AGENTS.md` schema (every error names its key;
 exit 1 refuses it), and `batch plan --brief <file>` composes inside it: named
 tasks first, then only the backlog its policy covers, against its
 `review-minutes:`. Run the check before writing a brief's `review-at:` into
-`batches/`. `batch dispatch` and `batch digest` are **planned — not code yet**:
-they arrive with the watcher in phase 3 of the autonomous-orchestrators epic,
-and until then the steps above are the dispatch.
+`batches/`. Dispatch is the watcher's (below), not a verb. `batch digest
+--record <file>` prints a shift's digest and writes nothing; the record append
+and the one issue comment are the watcher's, at `review-at:`.
 
-## The watcher (phase 3a — fake workers only)
+## The watcher (phases 3a–3d)
 
 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:`.
 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)
diff --git a/tests/test_orchestrator_digest.py b/tests/test_orchestrator_digest.py
new file mode 100644
index 0000000..9011822
--- /dev/null
+++ b/tests/test_orchestrator_digest.py
@@ -0,0 +1,383 @@
+"""Autonomous orchestrators phase 3d (PyAutoBrain#568): the shift digest.
+
+One test group per clause of the prompt's `Witness:`: the digest is appended
+to the shift record in the `PyAutoMind/batches/AGENTS.md` format (brief work vs
+backlog picks, each pick with its reason, one proof packet per task, every
+counter); `batch brief --check` accepts it and refuses one missing a counter
+or a pick reason; and it is posted exactly once — a second run posts nothing.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import os
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+BRAIN = Path(__file__).resolve().parents[1]
+BATCH = BRAIN / "agents" / "conductors" / "batch"
+sys.path.insert(0, str(BRAIN / "agents" / "faculties" / "sizing"))
+sys.path.insert(0, str(BATCH))
+import _brief  # noqa: E402
+import _digest  # noqa: E402
+import _orchestrator as orch  # noqa: E402
+
+NOW = _dt.datetime(2026, 11, 4, 9, 0, tzinfo=_dt.timezone.utc)
+AFTER = NOW + _dt.timedelta(hours=3)
+ISSUE = "https://github.com/example-org/example-mind/issues/9999"
+
+BRIEF = """\
+# Batch 2026-11-04 am
+- dispatched: 2026-11-04T08:30Z
+- review-at: 2026-11-04T11:30Z
+- issue: {issue}
+{packet}- brief:
+  - goals:
+    - 1. dry shift
+  - tasks:
+    - draft/feature/pyautobrain/named.md
+    - draft/feature/pyautobrain/never.md
+  - backlog: scoped
+  - scoped:
+    - epics: e1
+  - review-minutes: 45
+  - lanes: any local-dev
+  - merge-cap: 1
+  - token-budget: 1000
+  - max-workers: 3
+"""
+
+PROMPT = """\
+# {name}
+
+Type: feature
+Target: PyAutoBrain
+Repos:
+- PyAutoBrain
+Difficulty: small
+Autonomy: supervised
+Priority: normal
+Status: formalised
+{epic}Consequence: glance
+Witness: a unit test
+Review-minutes: 3
+Unattended: ready
+
+Body.
+"""
+
+PROOF = """\
+#### named — https://github.com/o/r/pull/1
+- claim: the thing is true
+- witness: `pytest -q`
+    3 passed
+- not checked: nothing else
+- revert: `git revert abc`
+"""
+
+
+class FilesSpawner:
+    """A worker that has already finished: `.done` (and, for some, a proof)."""
+
+    def __init__(self, done: dict[str, str], proofs: dict[str, str]):
+        self.done, self.proofs = done, proofs
+
+    def spawn(self, task, prompt_path, lease_dir):
+        Path(lease_dir).mkdir(parents=True, exist_ok=True)
+        if task in self.proofs:
+            (Path(lease_dir) / f"{task}.proof.md").write_text(self.proofs[task])
+        (Path(lease_dir) / f"{task}.done").write_text(self.done.get(task, ""))
+        return {"pid": 0, "session_id": None, "worktree": None}
+
+
+def _mind(root: Path) -> Path:
+    folder = root / "draft" / "feature" / "pyautobrain"
+    folder.mkdir(parents=True)
+    (folder / "named.md").write_text(PROMPT.format(name="named", epic=""))
+    (folder / "picked.md").write_text(PROMPT.format(name="picked", epic="Epic: e1\n"))
+    (folder / "outside.md").write_text(PROMPT.format(name="outside", epic=""))
+    return root
+
+
+def _run_shift(tmp_path: Path, brief: Path, mind: Path, home: Path):
+    """Dispatch at NOW, reap one tick later — both workers finished."""
+    sp = FilesSpawner(
+        done={"named": "merged https://github.com/o/r/pull/1\n",
+              "picked": "https://github.com/o/r/pull/2\n"},
+        proofs={"named": PROOF})
+    r = orch.tick(brief, mind, home=home, now=NOW, spawner=sp)
+    assert sorted(s["task"] for s in r["spawned"]) == ["named", "picked"], r
+    orch.tick(brief, mind, home=home, now=NOW + _dt.timedelta(minutes=1), spawner=sp)
+
+
+def _shift(tmp_path: Path, packet=""):
+    mind = _mind(tmp_path / "mind")
+    brief = tmp_path / "shift.md"
+    brief.write_text(BRIEF.format(issue=ISSUE, packet=packet))
+    home = tmp_path / "home"
+    _run_shift(tmp_path, brief, mind, home)
+    return mind, brief, home
+
+
+class Poster:
+    def __init__(self):
+        self.calls = []
+
+    def __call__(self, issue, body_file):
+        self.calls.append((issue, Path(body_file).read_text()))
+        return {"posted": True, "url": issue + "#issuecomment-1"}
+
+
+def _health(url):
+    return "HEALTHY"
+
+
+# --- the digest's content -----------------------------------------------------------
+
+def test_digest_separates_brief_work_from_backlog_picks_with_reasons(tmp_path):
+    mind, brief, home = _shift(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 "outside" not in counters["picks"]
+
+    d = _digest.build(_brief.parse(brief.read_text())["brief"], slot, counters,
+                      health=_health)
+    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
+    assert "- never — not started" in work
+    assert "- picked — https://github.com/o/r/pull/2 — HEALTHY — glance — why: scoped epic e1" in picks
+    assert "named" not in picks
+
+
+def test_one_proof_packet_per_task_and_an_honest_stub_when_missing(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    slot = orch.slot_dir(brief, home)
+    d = _digest.build(_brief.parse(brief.read_text())["brief"], slot,
+                      orch.read_counters(slot), health=_health)
+    packets = d.split("### Proof packets")[1].split("### Counters")[0]
+    assert "- claim: the thing is true" in packets      # the worker's, verbatim
+    assert "#### picked — https://github.com/o/r/pull/2" in packets
+    assert "the worker wrote no proof packet" in packets
+    assert "#### never" not in packets                  # never started, no packet
+
+
+def test_every_counter_is_printed(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    slot = orch.slot_dir(brief, home)
+    d = _digest.build(_brief.parse(brief.read_text())["brief"], slot,
+                      orch.read_counters(slot), health=_health)
+    counters = d.split("### Counters")[1]
+    for key in _digest.COUNTERS:
+        assert f"- {key}: " in counters, key
+    assert "- merges: 1" in counters
+    assert "- tokens: 0 of 1000" in counters
+    assert "- review-minutes-planned: 45" in counters
+    assert "- glance-auto-merges: 0" in counters
+
+
+# --- brief --check over the digest -----------------------------------------------
+
+def _with_digest(tmp_path, brief, home, mutate=lambda d: d):
+    slot = orch.slot_dir(brief, home)
+    d = _digest.build(_brief.parse(brief.read_text())["brief"], slot,
+                      orch.read_counters(slot), health=_health)
+    rec = tmp_path / "record_with_digest.md"
+    rec.write_text(brief.read_text() + "\n" + mutate(d))
+    return rec
+
+
+def test_brief_check_accepts_a_whole_digest(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    res = _brief.check(_with_digest(tmp_path, brief, home), mind)
+    assert res["ok"], res["errors"]
+
+
+def test_brief_check_refuses_a_missing_counter(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    rec = _with_digest(tmp_path, brief, home,
+                       lambda d: d.replace("- reverts: 0\n", ""))
+    res = _brief.check(rec, mind)
+    assert not res["ok"] and "digest: counter reverts missing" in res["errors"]
+
+
+def test_brief_check_refuses_a_pick_without_a_reason(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    rec = _with_digest(tmp_path, brief, home,
+                       lambda d: d.replace(" — why: scoped epic e1", ""))
+    res = _brief.check(rec, mind)
+    assert not res["ok"]
+    assert any("backlog pick picked has no `why:`" in e for e in res["errors"])
+
+
+def test_brief_check_refuses_a_task_without_a_packet(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    rec = _with_digest(tmp_path, brief, home,
+                       lambda d: d.replace("#### picked", "#### someone-else"))
+    assert "digest: picked has no proof packet" in _brief.check(rec, mind)["errors"]
+
+
+# --- posted exactly once ---------------------------------------------------------
+
+def test_the_digest_is_posted_once_and_a_second_run_posts_nothing(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    slot = orch.slot_dir(brief, home)
+    poster = Poster()
+    r = orch.tick(brief, mind, home=home, now=AFTER, poster=poster, health=_health)
+    assert r["action"] == "digest" and r["post"]["posted"], r
+    assert len(poster.calls) == 1 and poster.calls[0][0] == ISSUE
+    assert poster.calls[0][1].startswith("## Digest")
+    assert (slot / orch.DIGEST_MARKER).exists()
+
+    again = orch.tick(brief, mind, home=home, now=AFTER + _dt.timedelta(minutes=5),
+                      poster=poster, health=_health)
+    assert again["action"] == "idle" and len(poster.calls) == 1
+    direct = orch.finish_shift(brief, mind, slot, AFTER, poster=poster, health=_health)
+    assert direct["post"] == {"posted": False, "error": "already posted this shift"}
+    assert len(poster.calls) == 1
+
+
+def test_a_failed_post_is_never_retried(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    calls = []
+
+    def failing(issue, body_file):
+        calls.append(issue)
+        return {"posted": False, "error": "HTTP 502"}
+
+    orch.tick(brief, mind, home=home, now=AFTER, poster=failing, health=_health)
+    orch.tick(brief, mind, home=home, now=AFTER + _dt.timedelta(minutes=5),
+              poster=failing, health=_health)
+    assert calls == [ISSUE]  # a second comment would retire the watcher
+
+
+def test_a_shift_that_never_ran_has_no_digest(tmp_path):
+    mind = _mind(tmp_path / "mind")
+    brief = tmp_path / "shift.md"
+    brief.write_text(BRIEF.format(issue=ISSUE, packet=""))
+    poster = Poster()
+    r = orch.tick(brief, mind, home=tmp_path / "home", now=AFTER, poster=poster)
+    assert r["action"] == "idle" and poster.calls == []
+
+
+def test_stop_still_wins_after_review_at(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    (orch.slot_dir(brief, home) / orch.STOP_FILE).write_text("")
+    poster = Poster()
+    assert orch.tick(brief, mind, home=home, now=AFTER, poster=poster)["action"] == "stopped"
+    assert poster.calls == []
+
+
+def test_the_html_packet_is_opt_in(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    r = orch.tick(brief, mind, home=home, now=AFTER, poster=Poster(), health=_health)
+    assert r["packet"] is None
+    assert not (orch.slot_dir(brief, home) / "digest.html").exists()
+
+    mind2, brief2, home2 = _shift(tmp_path / "b", packet="- packet: yes\n")
+    r2 = orch.tick(brief2, mind2, home=home2, now=AFTER, poster=Poster(), health=_health)
+    page = Path(r2["packet"]).read_text()
+    assert "## Digest" in page and "<pre>" in page
+
+
+# --- the record append, through the one Mind writer ------------------------------
+
+LEDGER_MERGE = BRAIN.parent / "PyAutoMind" / "scripts" / "ledger_merge.py"
+if not LEDGER_MERGE.is_file():
+    LEDGER_MERGE = Path(os.environ.get("PYAUTO_MIND", "/nonexistent")) / "scripts" / "ledger_merge.py"
+
+
+def git(cwd, *args):
+    res = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
+    assert res.returncode == 0, (args, res.stderr)
+    return res
+
+
+@pytest.mark.skipif(not LEDGER_MERGE.is_file() or shutil.which("git") is None,
+                    reason="needs git and PyAutoMind's scripts/ledger_merge.py")
+def test_the_digest_lands_in_the_record_on_mind_main(tmp_path, monkeypatch):
+    monkeypatch.setattr(sys, "dont_write_bytecode", False)
+    bare = tmp_path / "remote.git"
+    git(tmp_path, "init", "-q", "--bare", "-b", "main", str(bare))
+    canon = tmp_path / "canon"
+    git(tmp_path, "clone", "-q", str(bare), str(canon))
+    git(canon, "config", "user.name", "Test")
+    git(canon, "config", "user.email", "test@example.invalid")
+    _mind(canon)
+    (canon / "scripts").mkdir()
+    shutil.copy(LEDGER_MERGE, canon / "scripts" / "ledger_merge.py")
+    (canon / "batches").mkdir()
+    brief = canon / "batches" / "2026-11-04-am.md"
+    brief.write_text(BRIEF.format(issue=ISSUE, packet=""))
+    git(canon, "add", "-A")
+    git(canon, "commit", "-q", "-m", "seed")
+    git(canon, "push", "-q", "origin", "HEAD:main")
+
+    home = tmp_path / "home"
+    _run_shift(tmp_path, brief, canon, home)
+    status_before = git(canon, "status", "--porcelain").stdout
+    r = orch.tick(brief, canon, home=home, now=AFTER, poster=Poster(), health=_health)
+    assert r["ledger"]["committed"], r
+    landed = git(bare, "show", "main:batches/2026-11-04-am.md").stdout
+    assert "## Digest" in landed and "- reverts: 0" in landed
+    assert git(canon, "status", "--porcelain").stdout == status_before
+
+    # What landed is what `batch brief --check` accepts.
+    fetched = tmp_path / "landed.md"
+    fetched.write_text(landed)
+    assert _digest.validate(landed) == []
+
+
+# --- the small parts ---------------------------------------------------------------
+
+def test_parse_done_reads_urls_and_merged_lines():
+    out = _digest.parse_done("merged https://x/pull/1\nhttps://x/pull/2\n\nnoise\n")
+    assert out == {"prs": ["https://x/pull/1", "https://x/pull/2"],
+                   "merged": ["https://x/pull/1"]}
+
+
+@pytest.mark.parametrize("rc,err,want", [(0, "", "HEALTHY"), (8, "", "RUNNING"),
+                                         (1, "", "FAILED"),
+                                         (1, "no checks reported", "NO-CHECKS")])
+def test_pr_health_reads_the_gh_exit_code(rc, err, want):
+    def run(argv, **kw):
+        assert argv[:3] == ["gh", "pr", "checks"]
+        return subprocess.CompletedProcess(argv, rc, "", err)
+    assert _digest.pr_health("https://github.com/o/r/pull/1", run=run) == want
+    assert _digest.pr_health("https://github.com/o/r/issues/3", run=run) == "PARKED-ON-ISSUE"
+
+
+@pytest.mark.parametrize("backlog,record,want", [
+    ("fill-idle", {"path": "x/a.md"}, "backlog: fill-idle (the planner's ranking)"),
+    ("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"),
+])
+def test_pick_reason_names_the_rule(backlog, record, want):
+    brief = {"backlog": {"value": backlog, "children": []},
+             "scoped": {"value": "", "children": [
+                 {"text": "tags: jax", "children": []},
+                 {"text": "tiers: notify", "children": []}]},
+             "tasks": {"value": "", "children": [{"text": "x/named.md", "children": []}]}}
+    assert _brief.pick_reason(record, brief) == want
+
+
+def test_batch_digest_verb_prints_and_writes_nothing(tmp_path):
+    mind, brief, home = _shift(tmp_path)
+    slot = orch.slot_dir(brief, home)
+    before = sorted(p.name for p in slot.rglob("*"))
+    env = {**os.environ, "PATH": "/nonexistent"}  # no gh: health says UNKNOWN
+    res = subprocess.run([sys.executable, str(BATCH / "_batch.py"), "digest",
+                          "--record", str(brief), "--home", str(home),
+                          "--mind", str(mind)],
+                         capture_output=True, text=True, env=env)
+    assert res.returncode == 0, res.stderr
+    assert res.stdout.startswith("## Digest") and "- merges: 1" in res.stdout
+    assert sorted(p.name for p in slot.rglob("*")) == before
+    assert not (slot / orch.DIGEST_MARKER).exists()