feat: autonomous orchestrators p3c — one Mind writer (ledger ops, temp-worktree commits, worker hook)

agents/conductors/batch/AGENTS.md

@@ -172,9 +172,25 @@ it → crashed (re-dispatched once). A finished worker's tokens (input + output
 cache creation) are summed from its local transcript, named by the full
 `sessionId`, and the brief's `token-budget:` refuses the next spawn once spent.
 `worker-mode:` defaults to `auto` (human, 2026-11-03); `dontAsk` and
-`acceptEdits` are allowed, `bypassPermissions` never. **Until phase 3c lands,
-real dispatch needs `sandbox: yes` in the brief** — one Mind writer is doctrine
-and a real worker would otherwise commit to Mind itself.
+`acceptEdits` are allowed, `bypassPermissions` never.
+
+**One Mind writer (phase 3c).** Workers never commit to PyAutoMind. Each Mind
+change — the `active.md` claim, the `draft/ → active/` move, the close-out — is
+queued as an op file in `<slot>/ledger/` by `bin/shift_ledger.py`
+(`append | write | move | remove | drop-section`; every path must be *ledger*
+by PyAutoMind's own `scripts/ledger_merge.py`, default deny). Each tick, after
+the reap, `_mind_writer.commit_ledger` applies all pending ops in a temporary
+detached worktree on `origin/main`, commits once and pushes; if `main` moved it
+replays the ops on the new `origin/main` (up to three tries). It never resets,
+never forces, and never touches the canonical checkout's working tree.
+Committed op files move to `ledger/committed/`; an op that cannot apply stays
+pending and the tick reports it. **Enforced, not asked:** a worker runs with
+`PYAUTO_SHIFT_WORKER=1`, `PYAUTO_SHIFT_LEDGER_DIR`, and `core.hooksPath` (via
+`GIT_CONFIG_*`) pointing at `bin/worker_git_hooks/`, whose `pre-commit`
+refuses any commit in a PyAutoMind repo — `prompt_sync_push`, `lifecycle close`
+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.
 
 **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.

agents/conductors/batch/_mind_writer.py

@@ -0,0 +1,186 @@
+"""One Mind writer per shift (AUTONOMY.md, 2026-11-03) — the orchestrator's side.
+
+Workers never commit to PyAutoMind. They describe each change as an *op file*
+in the shift's `ledger/` directory (`bin/shift_ledger.py` writes them); the
+orchestrator applies every pending op in a **temporary detached worktree** on
+`origin/main`, commits once, and pushes. A rejected push means someone else
+moved `main` (another session, the dashboard self-heal bot): the ops are
+replayed on the new `origin/main` — a rebase of operations, not of diffs — at
+most `PUSH_TRIES` times. The canonical checkout's working tree is never read
+or written, nothing is reset, and nothing is forced.
+
+Ops (one JSON object per file, applied in file-name order):
+
+    {"op": "append",       "path": "active.md", "text": "..."}
+    {"op": "write",        "path": "complete/2026/11/x.md", "text": "..."}   # must not exist
+    {"op": "move",         "path": "draft/a/b/x.md", "to": "active/x.md"}
+    {"op": "remove",       "path": "active/x.md"}
+    {"op": "drop-section", "path": "active.md", "section": "task-name"}
+
+Every path must be *ledger* by PyAutoMind's own rule (`scripts/ledger_merge.py`
+`is_ledger_path`, default deny), so a worker cannot route code through here.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import sys
+import tempfile
+import types
+from pathlib import Path
+
+OPS = ("append", "write", "move", "remove", "drop-section")
+PUSH_TRIES = 3
+LEDGER_DIR = "ledger"
+COMMITTED_DIR = "committed"
+
+
+class MindWriteError(RuntimeError):
+    pass
+
+
+def _ledger_merge(mind_repo: Path):
+    """PyAutoMind's own `ledger_merge` module, from the checkout we write to.
+
+    Compiled from source rather than imported: an import writes
+    `scripts/__pycache__/` into the canonical checkout, and this module's
+    promise is that the canonical working tree is never written."""
+    path = Path(mind_repo) / "scripts" / "ledger_merge.py"
+    if not path.is_file():
+        raise MindWriteError(f"no scripts/ledger_merge.py under {mind_repo}")
+    name = "_mind_ledger_merge"
+    mod = types.ModuleType(name)
+    mod.__file__ = str(path)
+    sys.modules[name] = mod  # dataclasses and friends look the module up by name
+    try:
+        exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), mod.__dict__)
+    finally:
+        sys.modules.pop(name, None)
+    return mod
+
+
+def validate_op(op: dict, lm) -> list[str]:
+    """Problems with one op; empty when it may be applied."""
+    errs = []
+    kind = op.get("op")
+    if kind not in OPS:
+        return [f"op: {kind!r} is not one of {', '.join(OPS)}"]
+    paths = [op.get("path")] + ([op.get("to")] if kind == "move" else [])
+    for p in paths:
+        if not isinstance(p, str) or not p:
+            errs.append(f"{kind}: a path is missing")
+        elif p.startswith("/") or not lm.is_ledger_path(p):
+            errs.append(f"{kind}: {p} is not a ledger path (scripts/ledger_merge.py)")
+    if kind in ("append", "write") and not isinstance(op.get("text"), str):
+        errs.append(f"{kind}: text is missing")
+    if kind == "drop-section":
+        if not op.get("section"):
+            errs.append("drop-section: section is missing")
+        elif op.get("path") not in lm.ENTRY_MERGED_FILES:
+            errs.append(f"drop-section: {op.get('path')} is not a `## slug` registry")
+    return errs
+
+
+def apply_op(tree: Path, op: dict, lm) -> None:
+    """Apply one validated op inside `tree`; raises MindWriteError if it cannot."""
+    target = tree / op["path"]
+    kind = op["op"]
+    if kind == "append":
+        target.parent.mkdir(parents=True, exist_ok=True)
+        old = target.read_text(encoding="utf-8") if target.exists() else ""
+        sep = "" if not old or old.endswith("\n") else "\n"
+        target.write_text(old + sep + op["text"], encoding="utf-8")
+    elif kind == "write":
+        if target.exists():
+            raise MindWriteError(f"write: {op['path']} already exists")
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(op["text"], encoding="utf-8")
+    elif kind == "move":
+        dest = tree / op["to"]
+        if not target.exists():
+            raise MindWriteError(f"move: {op['path']} does not exist")
+        if dest.exists():
+            raise MindWriteError(f"move: {op['to']} already exists")
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        _git(tree, "mv", op["path"], op["to"])
+    elif kind == "remove":
+        if not target.exists():
+            raise MindWriteError(f"remove: {op['path']} does not exist")
+        _git(tree, "rm", "-q", op["path"])
+    elif kind == "drop-section":
+        preamble, entries = lm.split_entries(target.read_text(encoding="utf-8"))
+        kept = [(s, b) for s, b in entries if s != f"## {op['section']}"]
+        if len(kept) == len(entries):
+            raise MindWriteError(f"drop-section: no `## {op['section']}` in {op['path']}")
+        target.write_text(lm.join_entries(preamble, kept), encoding="utf-8")
+
+
+def _git(cwd: Path, *args, check=True) -> subprocess.CompletedProcess:
+    res = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
+    if check and res.returncode != 0:
+        raise MindWriteError(f"git {' '.join(args)}: {(res.stderr or res.stdout).strip()}")
+    return res
+
+
+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 []
+
+
+def commit_ledger(slot: Path, mind_repo: Path, message: str,
+                  remote: str = "origin", branch: str = "main") -> dict:
+    """Commit every pending op in `<slot>/ledger/` to PyAutoMind as one commit.
+
+    Returns {"committed": sha | None, "ops": n, "tries": k}. Ops that fail
+    validation are refused before anything is touched; an op that cannot apply
+    on the current `origin/main` stops the run and stays pending.
+    """
+    mind_repo = Path(mind_repo)
+    files = pending_ops(slot)
+    if not files:
+        return {"committed": None, "ops": 0, "tries": 0}
+    lm = _ledger_merge(mind_repo)
+    ops = []
+    for f in files:
+        try:
+            op = json.loads(f.read_text(encoding="utf-8"))
+        except ValueError as e:
+            raise MindWriteError(f"{f.name}: not JSON ({e})") from e
+        errs = validate_op(op, lm)
+        if errs:
+            raise MindWriteError(f"{f.name}: " + "; ".join(errs))
+        ops.append(op)
+
+    for attempt in range(1, PUSH_TRIES + 1):
+        _git(mind_repo, "fetch", "-q", remote, branch)
+        tmp = Path(tempfile.mkdtemp(prefix="mind-writer-"))
+        tree = tmp / "tree"
+        try:
+            _git(mind_repo, "worktree", "add", "-q", "--detach", str(tree),
+                 f"{remote}/{branch}")
+            for op in ops:
+                apply_op(tree, op, lm)
+            _git(tree, "add", "-A")
+            if not _git(tree, "status", "--porcelain").stdout.strip():
+                sha = None  # every op was already true on main
+            else:
+                _git(tree, "commit", "-q", "-m", message)
+                sha = _git(tree, "rev-parse", "HEAD").stdout.strip()
+                pushed = _git(tree, "push", "-q", remote, f"HEAD:{branch}", check=False)
+                if pushed.returncode != 0:
+                    continue  # main moved: replay the ops on the new origin/main
+            done = Path(slot) / LEDGER_DIR / COMMITTED_DIR
+            done.mkdir(parents=True, exist_ok=True)
+            for f in files:
+                f.replace(done / f.name)
+            return {"committed": sha, "ops": len(ops), "tries": attempt}
+        finally:
+            # Our own throwaway tree. A failed apply can leave it dirty, so
+            # delete the directory and prune the registration, no force flag.
+            if _git(mind_repo, "worktree", "remove", str(tree), check=False).returncode:
+                shutil.rmtree(tmp, ignore_errors=True)
+                _git(mind_repo, "worktree", "prune", check=False)
+            shutil.rmtree(tmp, ignore_errors=True)
+    raise MindWriteError(f"push rejected {PUSH_TRIES} times; ops left pending")

agents/conductors/batch/_orchestrator.py

@@ -16,8 +16,10 @@
 Workers are fake stubs (`PYAUTO_ORCHESTRATOR_FAKE=1`, phase 3a) or real
 `claude --bg` sessions from one session template (phase 3b). A real worker's
 liveness is its `claude agents --json` state; one left `blocked` past a grace
-is parked for a human — never killed, never re-dispatched. Until phase 3c (one
-Mind writer) lands, real workers run only under a `sandbox: yes` brief.
+is parked for a human — never killed, never re-dispatched. Workers never
+commit to PyAutoMind (phase 3c): they queue ledger ops (`bin/shift_ledger.py`)
+and each tick commits them from a temporary worktree (`_mind_writer.py`); a
+git hook injected through the worker's environment refuses a direct commit.
 Nothing here commits, merges, posts or arms a timer: the systemd timer that
 calls `bin/orchestrator_watch.py` is installed by the human.
 """
@@ -33,6 +35,7 @@
 from pathlib import Path
 
 import _brief  # noqa: E402 — the batch conductor puts this directory on sys.path
+import _mind_writer  # noqa: E402
 
 STOP_FILE = "STOP"
 LOCK_FILE = "tick.lock"
@@ -50,6 +53,10 @@
 STALL_GRACE_S = 900
 #: The worker prompt every real worker starts from (never a bare session).
 WORKER_TEMPLATE = Path(__file__).with_name("worker_template.md")
+#: Hooks every worker's git runs with (core.hooksPath, set through GIT_CONFIG_*):
+#: its pre-commit refuses a commit in PyAutoMind and chains to the repo's own.
+WORKER_HOOKS = Path(__file__).resolve().parents[3] / "bin" / "worker_git_hooks"
+SHIFT_LEDGER = Path(__file__).resolve().parents[3] / "bin" / "shift_ledger.py"
 
 FAKE_ENV = "PYAUTO_ORCHESTRATOR_FAKE"
 HOME_ENV = "PYAUTO_ORCHESTRATOR_HOME"
@@ -173,7 +180,23 @@ def spawn(self, task: str, prompt_path: str, lease_dir: Path) -> dict:
 def worker_prompt(task: str, prompt_path: str, brief_path: Path, lease_dir: Path) -> str:
     """The session template filled in for one task."""
     return WORKER_TEMPLATE.read_text(encoding="utf-8").format(
-        task=task, task_path=prompt_path, brief_path=brief_path, lease_dir=lease_dir)
+        task=task, task_path=prompt_path, brief_path=brief_path, lease_dir=lease_dir,
+        ledger_dir=Path(lease_dir).parent / _mind_writer.LEDGER_DIR,
+        shift_ledger=SHIFT_LEDGER)
+
+
+def worker_env(lease_dir: Path, base: dict | None = None) -> dict:
+    """The environment a worker runs in: marked as a shift worker, pointed at
+    the shift's ledger, and with git's hooks path overridden so a commit in
+    PyAutoMind is refused whatever tool makes it."""
+    env = dict(os.environ if base is None else base)
+    n = int(env.get("GIT_CONFIG_COUNT", "0") or 0)
+    env.update({"PYAUTO_SHIFT_WORKER": "1",
+                "PYAUTO_SHIFT_LEDGER_DIR": str(Path(lease_dir).parent / _mind_writer.LEDGER_DIR),
+                "GIT_CONFIG_COUNT": str(n + 1),
+                f"GIT_CONFIG_KEY_{n}": "core.hooksPath",
+                f"GIT_CONFIG_VALUE_{n}": str(WORKER_HOOKS)})
+    return env
 
 
 def session_command(task: str, prompt_path: str, brief_path: Path, lease_dir: Path,
@@ -204,7 +227,8 @@ def __init__(self, brief_path: Path, mode: str, cwd: Path | None = None, run=Non
 
     def spawn(self, task: str, prompt_path: str, lease_dir: Path) -> dict:
         argv = session_command(task, prompt_path, self.brief_path, lease_dir, self.mode)
-        res = self.run(argv, cwd=self.cwd, capture_output=True, text=True, timeout=120)
+        res = self.run(argv, cwd=self.cwd, capture_output=True, text=True, timeout=120,
+                       env=worker_env(lease_dir))
         # An untrusted folder is refused here, fast ("Workspace not trusted") —
         # never a silent stall. Workers start from a folder the human trusted.
         if res.returncode != 0:
@@ -443,13 +467,16 @@ def tick(brief_path: Path, mind: Path, *, home: Path | None = None,
 def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents,
                  tokens) -> dict:
     counters = read_counters(slot)
-    # One Mind writer (AUTONOMY.md, 2026-11-03) is phase 3c. Until it lands a
-    # real worker would commit to PyAutoMind itself, so real dispatch is fenced
-    # to briefs that declare themselves a sandbox.
-    if getattr(spawner, "real", False) and not _brief.is_sandbox(brief):
-        raise OrchestratorError("real dispatch needs `sandbox: yes` in the brief "
-                                "until phase 3c (one Mind writer) lands")
     reaped = _reap(slot, counters, now, grace_s, agents, tokens)
+    # One Mind writer: whatever the workers queued is committed here, by the
+    # orchestrator alone, from a temporary worktree — never by a worker.
+    ledger = {"committed": None, "ops": 0}
+    if _mind_writer.pending_ops(slot):
+        try:
+            ledger = _mind_writer.commit_ledger(
+                slot, mind, f"shift {slot.name}: ledger from workers")
+        except _mind_writer.MindWriteError as e:
+            ledger = {"committed": None, "error": str(e)}
     live = leases(slot)
     max_workers = _brief.cap_of(brief, "max-workers")
     max_sessions = _brief.cap_of(brief, "max-sessions")
@@ -497,7 +524,7 @@ def _tick_locked(slot, brief, mind, now, spawner, grace_s, records, agents,
     counters["refusals"] += [dict(r, at=_iso(now)) for r in refused]
     _write_counters(slot, counters)
     return {"action": "ticked", "reaped": reaped, "spawned": spawned,
-            "refused": refused, "live": sorted(live)}
+            "refused": refused, "live": sorted(live), "ledger": ledger}
 
 
 def main(argv=None) -> int:

agents/conductors/batch/worker_template.md

@@ -10,8 +10,10 @@ The terms of this shift (AUTONOMY.md "Shift briefs and unattended `glance`"):
 - End at your deliverable: a PR, a posted verdict, or a park with a question
   on the issue. Never arm anything that outlives your turn (no send_later,
   ScheduleWakeup, CronCreate, RemoteTrigger or /loop) — the watcher keeps time.
-- Never commit to PyAutoMind yourself: write any ledger line you would have
-  committed as a file in {lease_dir}/ and the orchestrator commits it.
+- Never commit to PyAutoMind yourself — a hook refuses it. Queue each Mind
+  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.

bin/shift_ledger.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""Hand a PyAutoMind change to the shift orchestrator instead of committing it.
+
+A shift worker never commits to PyAutoMind (AUTONOMY.md "One Mind writer per
+shift"); it writes one op file per change here and the orchestrator commits
+them all from a temporary worktree. Paths are PyAutoMind-relative and must be
+ledger paths (`draft/`, `active/`, `complete/`, `batches/`, the registries).
+
+    shift_ledger.py append  active.md --text-file entry.md
+    shift_ledger.py write   complete/2026/11/x.md --text-file record.md
+    shift_ledger.py move    draft/feature/x/y.md active/y.md
+    shift_ledger.py remove  active/y.md
+    shift_ledger.py drop-section active.md my-task
+
+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"
+sys.path.insert(0, str(BATCH))
+import _mind_writer as mw  # noqa: E402
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(prog="shift_ledger", description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("op", choices=mw.OPS)
+    ap.add_argument("path")
+    ap.add_argument("arg", nargs="?", help="move: destination; drop-section: section")
+    ap.add_argument("--text-file", type=Path)
+    ap.add_argument("--text")
+    ap.add_argument("--mind", type=Path, default=os.environ.get("PYAUTO_MIND"),
+                    help="PyAutoMind checkout to validate paths against (read only)")
+    a = ap.parse_args(argv)
+    ledger = os.environ.get("PYAUTO_SHIFT_LEDGER_DIR")
+    if not ledger:
+        print("shift_ledger: PYAUTO_SHIFT_LEDGER_DIR is not set — this is for "
+              "shift workers only", file=sys.stderr)
+        return 2
+    op = {"op": a.op, "path": a.path}
+    if a.op == "move":
+        op["to"] = a.arg
+    if a.op == "drop-section":
+        op["section"] = a.arg
+    if a.op in ("append", "write"):
+        op["text"] = a.text_file.read_text(encoding="utf-8") if a.text_file else (a.text or "")
+    if a.mind:
+        errs = mw.validate_op(op, mw._ledger_merge(a.mind))
+        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}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())