test(issue-21): review N2 - T2-8 runs in mam_sandbox, so lib.sh's herdr shim never lands in the caller's control root

run_loop.sh sources lib.sh before it refuses --worktree/--base/--review-only
under MAM_LOOP_PY=0, and sourcing lib.sh runs _init_herdr_isolation, which
rewrites $WORKSPACE_ROOT/.mam/shim/herdr. T2-8 copied the caller's
environment as is, so that root was the caller's control root (normally the
real checkout). The test now requests mam_sandbox, pins WORKSPACE_ROOT to it
explicitly, runs from it, and asserts the shim was written there; the four
parametrized refusal cases and their three assertions are unchanged. No
product code changes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017R3Eg7MEpSWnm1oB9ZPP6G

.agents/skills/lib.sh

@@ -775,14 +775,15 @@ except Exception:
     pass
 " 2>/dev/null || echo "")
 
-      if [ -n "$idle_pane" ]; then
+      # An idle pane is reused only when no environment has to be set: the
+      # pane's shell is already running, and `agent start ... -- <args>` hands
+      # every token after -- to the agent executable (--kind), so an
+      # `env K=V` prefix would reach the agent as arguments, not as its
+      # environment (Issue #21 R1). With env flags, fall through to
+      # pane split / workspace create, which take --env.
+      if [ -n "$idle_pane" ] && [ -z "$env_flags" ]; then
         existing_ws="$cand_ws"
         target_pane="$idle_pane"
-        if [ -n "$env_flags" ] && [ -n "$final_cmd" ]; then
-          idle_env="${env_flags//--env /}"
-          idle_env="${idle_env:+$idle_env }HERDR_WORKSPACE_ID=$existing_ws"
-          final_cmd="env $idle_env $final_cmd"
-        fi
         break
       fi
 
@@ -2693,3 +2694,19 @@ handle_startup_dialogs() {
 
 # Auto-initialize the herdr translation shim wrapper on library source
 _init_herdr_isolation
+
+# ---------------------------------------------------------------------------
+# Issue #21 (item 3): control-plane binding of a session row. The logic lives
+# in lib_py/control_plane.py; these are thin shell entry points.
+# mam_control_plane_json <workspace> [<row control_plane JSON>]
+mam_control_plane_json() {
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.control_plane json "$@"
+}
+
+# mam_control_plane_prefix <json>: the leading K=V tokens for new-session, or
+# nothing when no forward is needed. rc 3: a forward is needed but a path
+# contains whitespace (lib.sh splits the --env flags unquoted).
+mam_control_plane_prefix() {
+  [ -n "${1:-}" ] || return 0
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.control_plane prefix "$1"
+}

.agents/skills/lib_py/control_plane.py

@@ -0,0 +1,95 @@
+"""Control-plane binding of a session row (Issue #21, item 3).
+
+The binding is the four absolute paths a session's tools must agree on:
+WORKSPACE_ROOT, AGENT_SESSIONS_YAML, MAM_JOBS_DIR, DELEGATE_JOB_LOGS_DIR.
+create/resume persist it on the row (`control_plane`) and forward it to the
+pane only when it differs from what `--workspace` alone would imply, through
+the leading `K=V` tokens lib.sh's new-session already turns into `--env`.
+
+CLI (used by lib.sh):
+  python -m lib_py.control_plane json <workspace> [<row control_plane JSON>]
+      -> {4 keys, "forward": bool, "whitespace": bool, "source": "row"|"env"}
+  python -m lib_py.control_plane prefix <json>
+      -> "WORKSPACE_ROOT=... AGENT_SESSIONS_YAML=... ... " or "" (no forward)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shlex
+import sys
+from typing import Dict, Optional
+
+KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir")
+ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML",
+       "mam_jobs_dir": "MAM_JOBS_DIR", "delegate_job_logs_dir": "DELEGATE_JOB_LOGS_DIR"}
+
+
+def _abs(p: str) -> str:
+    return os.path.realpath(os.path.expanduser(p))
+
+
+def defaults(workspace: str) -> Dict[str, str]:
+    """D(ws): the binding `--workspace` implies on its own."""
+    ws = _abs(workspace)
+    mam = os.path.join(ws, ".mam")
+    return {"workspace_root": ws, "agent_sessions_yaml": os.path.join(mam, "agent-sessions.yaml"),
+            "mam_jobs_dir": os.path.join(mam, "jobs"), "delegate_job_logs_dir": os.path.join(mam, "delegate_job_logs")}
+
+
+def from_env(env=None) -> Dict[str, str]:
+    env = os.environ if env is None else env
+    b = defaults(env.get("WORKSPACE_ROOT") or os.getcwd())
+    for k, var in ENV.items():
+        v = (env.get(var) or "").strip()
+        if v:
+            b[k] = _abs(v)
+    return b
+
+
+def describe(workspace: str, row: Optional[dict] = None) -> dict:
+    """The binding to use: the row's persisted one if complete (D11), else the
+    caller's env. A row binding that differs from the env is reported on stderr."""
+    env_b = from_env()
+    source = "env"
+    b = env_b
+    if isinstance(row, dict) and all(isinstance(row.get(k), str) and row.get(k) for k in KEYS):
+        b = {k: row[k] for k in KEYS}
+        source = "row"
+        if b != env_b:
+            print("WARNING: using the control-plane binding recorded in the row "
+                  f"({json.dumps(b, sort_keys=True)}) instead of the caller's environment", file=sys.stderr)
+    return {**b, "forward": b != defaults(workspace),
+            "whitespace": any(any(c.isspace() for c in v) for v in b.values()), "source": source}
+
+
+def prefix(desc: dict) -> str:
+    if not desc.get("forward"):
+        return ""
+    return "".join(f"{ENV[k]}={shlex.quote(desc[k])} " for k in KEYS)
+
+
+def main(argv=None) -> int:
+    argv = list(sys.argv[1:] if argv is None else argv)
+    if len(argv) >= 2 and argv[0] == "json":
+        row = None
+        if len(argv) >= 3 and argv[2].strip():
+            try:
+                row = json.loads(argv[2])
+            except json.JSONDecodeError:
+                row = None
+        print(json.dumps(describe(argv[1], row)))
+        return 0
+    if len(argv) == 2 and argv[0] == "prefix":
+        desc = json.loads(argv[1])
+        if desc.get("forward") and desc.get("whitespace"):
+            return 3  # lib.sh splits --env flags unquoted: refuse rather than break silently
+        sys.stdout.write(prefix(desc))
+        return 0
+    print(__doc__, file=sys.stderr)
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.agents/skills/lib_py/loop/runner.py

@@ -13,6 +13,7 @@ import glob
 import json
 import os
 import re
+import shlex
 import shutil
 import signal
 import sqlite3
@@ -20,7 +21,7 @@ import subprocess
 import sys
 import time
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 from lib_py.loop import checkpoint, eager, parser
 
@@ -58,6 +59,20 @@ class LoopRunner:
         self.workspace_root = Path(os.environ.get("MAM_REAL_ROOT") or os.environ.get("WORKSPACE_ROOT") or Path.cwd()).resolve()
         self.mam_dir = self.workspace_root / ".mam"
         self.checkpoint_file = self.mam_dir / "loop-checkpoint.json"
+        # Issue #21: the jobs dir honours MAM_JOBS_DIR (and the wrapper is told
+        # the same dir explicitly). The source root (--worktree) is where the
+        # diff, base and head come from; everything else - wrapper, registry,
+        # checkpoint, loop guard, promotion, cleanup - stays on the control
+        # root (workspace_root).
+        # A relative MAM_JOBS_DIR is taken against the control root, as the
+        # wrapper (run with cwd=workspace_root) takes it.
+        self.jobs_dir = (self.workspace_root / Path(os.environ.get("MAM_JOBS_DIR") or self.mam_dir / "jobs")).resolve()
+        self.worktree_root: Path = self.workspace_root
+        self.worktree_given = False
+        self.base_ref: Optional[str] = None
+        self.base_sha = ""
+        self.base_source = "head"
+        self.review_only = False
         self.created_jobs: List[str] = []
         self.failed_submit_jobs: Set[str] = set()
         self.current_plan = ""
@@ -93,6 +108,9 @@ class LoopRunner:
             "  --max-rebut N          Max rebuttal attempts per review iteration (default: 1, 0: disabled)",
             "  --verbose              Print detailed execution timeline traces",
             "  --cleanup              Purge temporary job directories upon success",
+            "  --worktree <path>      Source worktree for diff/base/head (control plane stays here)",
+            "  --base <ref>           Pin the review diff base (must be an ancestor of the source HEAD)",
+            "  --review-only          Skip implementation; review <base>..HEAD (requires --base)",
         ]
         print("\n".join(lines))
 
@@ -161,6 +179,22 @@ class LoopRunner:
             elif arg == "--cleanup":
                 self.cleanup = True
                 i += 1
+            elif arg == "--worktree":
+                if i + 1 >= len(args):
+                    self.print_usage()
+                    sys.exit(1)
+                self.worktree_root = Path(args[i + 1])
+                self.worktree_given = True
+                i += 2
+            elif arg == "--base":
+                if i + 1 >= len(args):
+                    self.print_usage()
+                    sys.exit(1)
+                self.base_ref = args[i + 1]
+                i += 2
+            elif arg == "--review-only":
+                self.review_only = True
+                i += 1
             else:
                 print(f"Unknown option: {arg}")
                 self.print_usage()
@@ -183,6 +217,67 @@ class LoopRunner:
         if self.plan_talk_turns > 0 and not self.plan_mode:
             log_warn("--plan-talk was specified but --plan mode is not enabled. Discussion turns will be ignored.")
 
+        # Issue #21 (D3): CLI --worktree wins over MAM_LOOP_WORKTREE. --base has
+        # no env fallback (D17): a stale sha left in a shell would still pass
+        # the ancestor check and silently widen the next loop's review range.
+        if not self.worktree_given and os.environ.get("MAM_LOOP_WORKTREE", "").strip():
+            self.worktree_root = Path(os.environ["MAM_LOOP_WORKTREE"].strip())
+            self.worktree_given = True
+        if self.worktree_given:
+            self.worktree_root = self._validate_worktree(self.worktree_root)
+        if self.review_only and self.base_ref is None:
+            print("ERROR: --review-only requires --base <ref>.", file=sys.stderr)
+            sys.exit(1)
+        if self.review_only and self.plan_mode:
+            print("ERROR: --review-only cannot be combined with --plan.", file=sys.stderr)
+            sys.exit(1)
+        if self.base_ref is not None:
+            self.base_sha = self._resolve_base(self.base_ref)
+            self.base_source = "cli"
+
+    @staticmethod
+    def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
+        return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True)
+
+    def _git_common_dir(self, cwd: Path) -> str:
+        res = self._git(cwd, "rev-parse", "--git-common-dir")
+        if res.returncode != 0 or not res.stdout.strip():
+            return ""
+        d = res.stdout.strip()
+        # the main checkout answers ".git", relative to cwd (not to this process)
+        return os.path.realpath(d if os.path.isabs(d) else os.path.join(str(cwd), d))
+
+    def _validate_worktree(self, p: Path) -> Path:
+        p = Path(os.path.abspath(str(p)))
+        if not p.is_dir():
+            log_error(f"--worktree {p}: no such directory.")
+            sys.exit(1)
+        top = self._git(p, "rev-parse", "--show-toplevel")
+        if top.returncode != 0 or not top.stdout.strip():
+            log_error(f"--worktree {p}: not a git working tree.")
+            sys.exit(1)
+        if os.path.realpath(top.stdout.strip()) != os.path.realpath(p):
+            log_error(f"--worktree {p}: not the top of its working tree ({top.stdout.strip()}).")
+            sys.exit(1)
+        mine, ctl = self._git_common_dir(p), self._git_common_dir(self.workspace_root)
+        if not mine or mine != ctl:
+            log_error(f"--worktree {p}: not a worktree of the repository at {self.workspace_root}.")
+            sys.exit(1)
+        return Path(os.path.realpath(p))
+
+    def _resolve_base(self, ref: str) -> str:
+        """--base (D14): a full sha that is an ancestor of the source HEAD."""
+        res = self._git(self.worktree_root, "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}")
+        sha = res.stdout.strip() if res.returncode == 0 else ""
+        if not sha:
+            log_error(f"--base {ref}: not a commit in {self.worktree_root}.")
+            sys.exit(1)
+        if self._git(self.worktree_root, "merge-base", "--is-ancestor", sha, "HEAD").returncode != 0:
+            log_error(f"--base {ref} ({sha[:12]}) is not an ancestor of HEAD in {self.worktree_root}; "
+                      "refusing to review an unrelated range.")
+            sys.exit(1)
+        return sha
+
     def load_sessions_state(self) -> Dict[str, Any]:
         """Load registered sessions from MAM_STATE_JSON or agent-sessions.db/yaml."""
         raw_env = os.environ.get("MAM_STATE_JSON")
@@ -296,6 +391,7 @@ class LoopRunner:
                 "--agent", agent_type,
                 "--type", "direct",
                 "--role", role,
+                "--registry-dir", str(self.jobs_dir),
                 "--prompt", p_text,
             ] + self.contract_args()
             env = dict(os.environ)
@@ -346,7 +442,7 @@ class LoopRunner:
         if self.verbose:
             log_info(f"Monitoring job '{job_id}' for status changes (timeout: {max_wait}s)...")
 
-        job_file = self.mam_dir / "jobs" / f"{job_id}.json"
+        job_file = self.jobs_dir / f"{job_id}.json"
         start_time = time.time()
         missing_count = 0
         unknown_count = 0
@@ -388,7 +484,7 @@ class LoopRunner:
 
     def find_report(self, job_id: str) -> Optional[Path]:
         """Locate report-final.md for a job."""
-        job_dir = self.mam_dir / "jobs" / job_id
+        job_dir = self.jobs_dir / job_id
         matches = list(job_dir.glob("*/report-final.md")) + list(job_dir.glob("report-final.md"))
         if matches:
             return matches[0]
@@ -396,20 +492,94 @@ class LoopRunner:
 
     def collect_changes_diff(self) -> str:
         """Collect git diff since base commit."""
+        # The collector is the control root's copy (never the worktree's
+        # unreviewed one); the tree it diffs is the source root (Issue #21).
         diff_script = self.workspace_root / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"
+        src = str(self.worktree_root)
         if diff_script.exists():
-            cmd = f"source '{diff_script}' && mam_collect_changes_diff '{self.workspace_root}' '{self.base_commit}'"
-            res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, cwd=str(self.workspace_root))
+            cmd = (f"source {shlex.quote(str(diff_script))} && "
+                   f"mam_collect_changes_diff {shlex.quote(src)} {shlex.quote(self.base_commit)}")
+            res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, cwd=src)
             if res.returncode == 0:
                 return res.stdout
         # Fallback to direct git diff
         try:
             cmd = ["git", "diff", self.base_commit] if self.base_commit else ["git", "diff", "HEAD"]
-            res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(self.workspace_root))
+            res = subprocess.run(cmd, capture_output=True, text=True, cwd=src)
             return res.stdout
         except Exception:
             return ""
 
+    def _source_head(self) -> str:
+        res = self._git(self.worktree_root, "rev-parse", "HEAD")
+        out = res.stdout if isinstance(res.stdout, str) else ""
+        return out.strip() if res.returncode == 0 else ""
+
+    def _context_active(self) -> bool:
+        return self.worktree_given or self.base_ref is not None
+
+    def _worktree_context(self) -> str:
+        """The source-root block put in front of every delegated prompt (D6);
+        empty without --worktree/--base, so prompts stay byte-identical."""
+        if not self._context_active():
+            return ""
+        src = self.worktree_root
+        branch = self._git(src, "branch", "--show-current").stdout.strip() or "(detached)"
+        head = self._source_head()
+        base = self.base_commit or self.base_sha
+        origin = "--base" if self.base_source == "cli" else "루프 시작 시 HEAD"
+        count = "?"
+        if base and head:
+            r = self._git(src, "rev-list", "--count", f"{base}..HEAD")
+            if r.returncode == 0 and r.stdout.strip():
+                count = r.stdout.strip()
+        st = self._git(src, "status", "--porcelain")
+        dirty = "알 수 없음" if st.returncode != 0 else ("있음" if st.stdout.strip() else "없음")
+        return ("[워크트리 컨텍스트]\n"
+                f"- 소스 경로: {src}\n"
+                f"- 브랜치: {branch}\n"
+                f"- base: {base or '(Phase 2 시작 시 HEAD 로 정해짐)'}  (출처: {origin})\n"
+                f"- head: {head}  (base..head 커밋: {count}개, 미커밋 변경: {dirty})\n"
+                f"- 제어 평면(잡 브리프·리포트·레지스트리): {self.workspace_root}\n"
+                "이 잡의 코드 작업·검증 대상은 위 소스 경로다. 브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n")
+
+    def _dispatch(self, agent_session: str, agent_type: str, role: str, prompt: str) -> Optional[str]:
+        """delegate_job with the source-root context block in front of the prompt."""
+        return self.delegate_job(agent_session, agent_type, role, self._worktree_context() + prompt)
+
+    def _change_set_state(self) -> str:
+        """'empty' | 'present' | 'unknown' for the source root against base,
+        judged by git exit codes only (D16)."""
+        if not self.base_commit:
+            return "unknown"
+        try:
+            d = self._git(self.worktree_root, "diff", "--quiet", self.base_commit)
+            if d.returncode == 1:
+                return "present"
+            if d.returncode != 0:
+                return "unknown"
+            u = self._git(self.worktree_root, "ls-files", "-o", "--exclude-standard", "-z")
+            if u.returncode != 0:
+                return "unknown"
+            return "present" if u.stdout else "empty"
+        except Exception:
+            return "unknown"
+
+    def _change_set_allows_review(self) -> bool:
+        """Fail-closed guard (D16), on only with --worktree/--base: never ask
+        for a review of an empty or undeterminable change set."""
+        if not self._context_active():
+            return True
+        state = self._change_set_state()
+        if state == "empty":
+            log_error(f"empty change set since base {self.base_commit}; refusing to request reviews")
+            return False
+        if state != "present":
+            log_error(f"cannot determine the change set since base {self.base_commit or '(none)'} in "
+                      f"{self.worktree_root}; refusing to request reviews on no evidence")
+            return False
+        return True
+
     def _record_checkpoint(self, iteration: int = 1, review_status: Optional[Dict[str, str]] = None) -> None:
         data = {
             "iteration": iteration,
@@ -418,6 +588,10 @@ class LoopRunner:
             "plan_job_id": self.plan_job_id,
             "review_status": review_status or {},
             "base_commit": self.base_commit,
+            "base_source": self.base_source,
+            "review_only": self.review_only,
+            "worktree_root": str(self.worktree_root),
+            "head_commit": self._source_head() if self._context_active() else "",
             "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
         }
         checkpoint.save_checkpoint(self.checkpoint_file, data)
@@ -506,7 +680,7 @@ class LoopRunner:
             log_info("Requesting initial implementation plan from Planner...")
             p_type = self.resolve_agent_type(planner_session, state)
             plan_prompt = f"태스크 목표를 바탕으로 구체적인 구현 계획서를 작성해주세요. 목표: {self.task}"
-            p_jid = self.delegate_job(planner_session, p_type, "Planner", plan_prompt)
+            p_jid = self._dispatch(planner_session, p_type, "Planner", plan_prompt)
             if not p_jid:
                 log_error("Failed to register planner job.")
                 return 1
@@ -528,7 +702,7 @@ class LoopRunner:
             for turn in range(1, self.plan_talk_turns + 1):
                 log_info(f"Discussion Turn {turn}/{self.plan_talk_turns}: Creator challenging the plan...")
                 c_prompt = f"Planner가 제시한 다음 계획서를 꼼꼼히 검토하고, 실제 구현 시 마주할 수 있는 맹점이나 제약사항 1가지를 발굴하여 Planner에게 이의를 제기(Challenge)해주세요. 계획서:\n{self.current_plan}"
-                c_jid = self.delegate_job(self.creator, creator_type, "Worker", c_prompt)
+                c_jid = self._dispatch(self.creator, creator_type, "Worker", c_prompt)
                 if not c_jid or not self.wait_for_job(c_jid):
                     log_error("Creator critique step failed.")
                     return 1
@@ -538,7 +712,7 @@ class LoopRunner:
 
                 log_info("Planner refining plan with Creator's feedback...")
                 refine_prompt = f"작업자(Creator)로부터 다음 이의제기 피드백을 받았습니다. 피드백을 반영하여 계획서를 정교하게 업데이트(Refine)하여 다시 출력해주세요. 피드백:\n{critique_text}\n기존 계획서:\n{self.current_plan}"
-                ref_jid = self.delegate_job(planner_session, p_type, "Planner", refine_prompt)
+                ref_jid = self._dispatch(planner_session, p_type, "Planner", refine_prompt)
                 if not ref_jid or not self.wait_for_job(ref_jid):
                     log_error("Planner refinement step failed.")
                     return 1
@@ -560,28 +734,34 @@ class LoopRunner:
         # PHASE 2: IMPLEMENTATION (CREATION)
         # =======================================================================
         log_info("=== Phase 2: Code Implementation ===")
-        try:
-            res_base = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, cwd=str(self.workspace_root))
-            self.base_commit = res_base.stdout.strip()
-        except Exception:
-            self.base_commit = ""
-
-        if self.current_plan:
-            exec_prompt = f"이미 수립된 다음 계획서에 입각하여 작업자의 판단하에 코드를 구현하고 작업 목표를 완성해주세요. 계획서:\n{self.current_plan}\n\n작업 목표: {self.task}"
+        if self.base_sha:
+            self.base_commit = self.base_sha  # --base (D14)
         else:
-            exec_prompt = f"계획서가 존재하지 않으므로, 작업자(Creator)의 판단하에 스스로 구현 계획 및 설계를 수립한 뒤, 이를 바탕으로 코드를 구현하고 다음 작업 목표를 완성해주세요. 작업 목표: {self.task}"
+            try:
+                res_base = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, cwd=str(self.worktree_root))
+                self.base_commit = res_base.stdout.strip()
+            except Exception:
+                self.base_commit = ""
 
         creator_type = self.resolve_agent_type(self.creator, state)
-        exec_jid = self.delegate_job(self.creator, creator_type, "Worker", exec_prompt)
-        if not exec_jid:
-            log_error("Failed to register Creator execution job.")
-            return 1
-        log_info(f"Creator Job ID: {exec_jid}")
+        if self.review_only:
+            log_info(f"Phase 2 skipped (--review-only): reviewing {self.base_commit}..{self._source_head()}")
+        else:
+            if self.current_plan:
+                exec_prompt = f"이미 수립된 다음 계획서에 입각하여 작업자의 판단하에 코드를 구현하고 작업 목표를 완성해주세요. 계획서:\n{self.current_plan}\n\n작업 목표: {self.task}"
+            else:
+                exec_prompt = f"계획서가 존재하지 않으므로, 작업자(Creator)의 판단하에 스스로 구현 계획 및 설계를 수립한 뒤, 이를 바탕으로 코드를 구현하고 다음 작업 목표를 완성해주세요. 작업 목표: {self.task}"
 
-        if not self.wait_for_job(exec_jid):
-            log_error("Creator execution failed.")
-            return 1
-        log_success("Initial implementation finished.")
+            exec_jid = self._dispatch(self.creator, creator_type, "Worker", exec_prompt)
+            if not exec_jid:
+                log_error("Failed to register Creator execution job.")
+                return 1
+            log_info(f"Creator Job ID: {exec_jid}")
+
+            if not self.wait_for_job(exec_jid):
+                log_error("Creator execution failed.")
+                return 1
+            log_success("Initial implementation finished.")
 
         # =======================================================================
         # PHASE 3: VERIFICATION LOOP (PEER REVIEW)
@@ -605,12 +785,14 @@ class LoopRunner:
             log_info(f"Review Loop Iteration {loop_count}/{self.max_loop}...")
             rebut_budget = self.max_rebut
             self._record_checkpoint(iteration=loop_count)
+            if not self._change_set_allows_review():
+                return 1
 
             if not reviewers:
                 # Creator Self-Review
                 log_warn("No reviewers specified. Conducting Creator Self-Review...")
                 sr_prompt = "작업 완료 상태에 대해 스스로 검증(Self-Review)하여 결함이 없음을 확인하고 종결해주세요. 리뷰 리포트 마지막에 단독 행으로 반드시 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 명시해주세요."
-                sr_jid = self.delegate_job(self.creator, creator_type, "Reviewer", sr_prompt)
+                sr_jid = self._dispatch(self.creator, creator_type, "Reviewer", sr_prompt)
                 if not sr_jid:
                     log_error("Failed to register Self-Review job.")
                     return 1
@@ -634,7 +816,7 @@ class LoopRunner:
                 log_info(f"Requesting code review from Reviewer '{rev}'...")
                 r_type = self.resolve_agent_type(rev, state)
                 r_prompt = f"다음 구현 사항(작업 목표: {self.task}) 및 누적 변경분(git diff)에 대해 린트, 동작성, 유실 등의 관점에서 교차 코드 리뷰를 수행해주세요. 확인 후 최종 Verdict로 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 리뷰 리포트 마지막에 단독 행으로 명시적으로 작성해주세요. 만약 단순 버그 수정으로는 부족하고 설계 변경/재작업 수준의 재계획이 필요하다고 판단되면, 리포트 아무 곳에나 단독 행으로 '[ESCALATE: PLANNER]' 태그도 함께 남겨주세요. 변경분:\n{changes_diff}"
-                r_jid = self.delegate_job(rev, r_type, "Reviewer", r_prompt)
+                r_jid = self._dispatch(rev, r_type, "Reviewer", r_prompt)
                 if not r_jid:
                     log_error(f"Failed to register review job for '{rev}'.")
                     return 1
@@ -685,7 +867,7 @@ class LoopRunner:
                 log_warn("Feedback involves complex code modifications. Diverting to Planner to revise plan...")
                 p_type = self.resolve_agent_type(planner_session, state)
                 refine_prompt = f"리뷰어들로부터 다음과 같이 정교한 코드 수정 피드백이 도착했습니다. 해당 피드백을 수렴하여 구현 계획서(Plan)를 갱신(Refine)하여 다시 작성해주세요. 피드백:\n{feedback_aggregate}\n기존 계획서:\n{self.current_plan}"
-                ref_jid = self.delegate_job(planner_session, p_type, "Planner", refine_prompt)
+                ref_jid = self._dispatch(planner_session, p_type, "Planner", refine_prompt)
                 if not ref_jid or not self.wait_for_job(ref_jid):
                     log_error("Failed to register plan refinement job.")
                     return 1
@@ -700,7 +882,7 @@ class LoopRunner:
             if self.max_rebut > 0:
                 correction_prompt += f"\n\n[이의제기 채널] 위 피드백 중 타당하지 않다고 판단되는 항목이 있다면, 그 항목은 구현하지 말고 반론을 제기하십시오. 반론 시 리포트에 단독 행으로 '[REBUT: <리뷰어_세션명>]' 태그를 남기고, 해당 리뷰어의 지적 중 어느 항목을 왜 거부하는지 근거를 함께 서술하십시오. 타당한 지적은 정상적으로 반영하십시오. 반론할 항목이 없으면 태그를 남기지 마십시오. 대상 리뷰어 세션명: {' '.join(fail_revs) if fail_revs else '(없음)'}"
 
-            correct_jid = self.delegate_job(self.creator, creator_type, "Worker", correction_prompt)
+            correct_jid = self._dispatch(self.creator, creator_type, "Worker", correction_prompt)
             if not correct_jid or not self.wait_for_job(correct_jid):
                 log_error("Failed to register Creator correction job.")
                 return 1
@@ -728,7 +910,7 @@ class LoopRunner:
                         log_info(f"Re-adjudication: returning rebuttal to Reviewer '{rt}'...")
                         rt_type = self.resolve_agent_type(rt, state)
                         adj_prompt = f"작업자(Creator)가 귀하의 리뷰 지적 중 일부를 타당하지 않다고 판단하여 반론을 제기했습니다. 반론을 검토하고 재심(re-adjudication)하십시오. 반론이 타당하여 귀하의 지적을 철회한다면 리포트 마지막에 단독 행으로 '[ADJUDICATION: SUSTAINED]' 를, 반론이 타당하지 않아 지적을 유지한다면 '[ADJUDICATION: OVERRULED]' 를 명시하십시오. 반론 내용:\n{rebuttal_text}"
-                        adj_jid = self.delegate_job(rt, rt_type, "Reviewer", adj_prompt)
+                        adj_jid = self._dispatch(rt, rt_type, "Reviewer", adj_prompt)
                         if not adj_jid or not self.wait_for_job(adj_jid):
                             log_warn(f"Re-adjudication job for '{rt}' failed; objection stands.")
                             overruled_revs.append(rt)
@@ -747,7 +929,7 @@ class LoopRunner:
                             log_info(f"Deadlock on {' '.join(overruled_revs)}; escalating to Planner for arbitration...")
                             p_type = self.resolve_agent_type(planner_session, state)
                             arb_prompt = f"작업자(Creator)와 리뷰어의 의견이 충돌하여 교착 상태입니다. 계획서 관점에서 어느 쪽이 타당한지 재정(arbitration)하십시오. 리포트 마지막에 단독 행으로 '[ARBITRATION: CREATOR]' 또는 '[ARBITRATION: REVIEWER]' 를 명시하십시오. 리뷰어 지적:\n{feedback_aggregate}\n작업자 반론:\n{rebuttal_text}\n기존 계획서:\n{self.current_plan}"
-                            arb_jid = self.delegate_job(planner_session, p_type, "Planner", arb_prompt)
+                            arb_jid = self._dispatch(planner_session, p_type, "Planner", arb_prompt)
                             if arb_jid and self.wait_for_job(arb_jid):
                                 arb_file = self.find_report(arb_jid)
                                 if arb_file and parser.has_tag_line(arb_file, "ARBITRATION: CREATOR"):
@@ -761,7 +943,7 @@ class LoopRunner:
                             log_warn("Ruling stands with the reviewer(s); Creator must comply.")
                             rebut_budget = 0
                             comply_prompt = f"귀하의 반론은 재심 결과 기각되었습니다. 더 이상 반론하지 말고 다음 리뷰 피드백을 그대로 반영하여 코드를 수정하십시오. 피드백:\n{feedback_aggregate}"
-                            comp_jid = self.delegate_job(self.creator, creator_type, "Worker", comply_prompt)
+                            comp_jid = self._dispatch(self.creator, creator_type, "Worker", comply_prompt)
                             if comp_jid:
                                 self.wait_for_job(comp_jid)
 
@@ -788,10 +970,10 @@ class LoopRunner:
         if self.cleanup:
             log_info("Cleaning up temporary job directories created during this loop run...")
             for job in self.created_jobs:
-                job_dir = self.mam_dir / "jobs" / job
+                job_dir = self.jobs_dir / job
                 if job_dir.exists():
                     shutil.rmtree(job_dir, ignore_errors=True)
-                sub_out = self.mam_dir / "jobs" / f"{job}.subscriber.out"
+                sub_out = self.jobs_dir / f"{job}.subscriber.out"
                 if sub_out.exists():
                     sub_out.unlink(missing_ok=True)
                 if self.verbose:

.agents/skills/multi-agent-mux-create/scripts/create_session.sh

@@ -202,6 +202,23 @@ if [ -z "$CMD_FULL" ]; then
   esac
 fi
 
+# Issue #21 (item 3): the control-plane binding. Persisted on the row always;
+# forwarded to the pane (leading K=V tokens that lib.sh's new-session turns
+# into --env) only when it differs from what --workspace alone implies.
+# lib.sh splits those flags unquoted (and, with env flags, never reuses an
+# idle pane - `agent start --` would pass them to the agent as arguments), so a
+# binding with whitespace that would have to be forwarded is refused.
+CP_JSON="$(mam_control_plane_json "$WORKSPACE" 2>/dev/null || true)"
+CP_PREFIX=""
+if [ -n "$CP_JSON" ]; then
+  cp_rc=0
+  CP_PREFIX="$(mam_control_plane_prefix "$CP_JSON")" || cp_rc=$?
+  if [ "$cp_rc" -eq 3 ]; then
+    echo "ERROR: control-plane path contains whitespace; cannot forward to the pane" >&2
+    exit 2
+  fi
+fi
+
 spawn() {
   if [ -z "$HERDR_SERVER_OPT" ]; then
     if [ -z "${HERDR_SESSION_NAME:-}" ] || [ "$HERDR_SESSION_NAME" = "default" ]; then
@@ -213,20 +230,21 @@ spawn() {
       if { [ -x "$WRAPPER" ] && [ "$(basename "$WRAPPER")" != "claude" ]; } || [ "$USE_WRAPPER" = "1" ]; then
         SESSION_UUID=""
         CMD_FULL="$("$(_delegate_py_bin)" -m lib_py.agents spawn-spec "$AGENT" "$RESOLVED_BIN" "" "true" 2>/dev/null || echo "${RESOLVED_BIN} --dangerously-skip-permissions")"
+        [ -z "$CP_PREFIX" ] || echo "WARNING: the ~/.local/bin wrapper path bypasses herdr; the control-plane binding is recorded but not forwarded" >&2
         nohup "$WRAPPER" >/dev/null 2>&1 &
         disown
       else
-        HERDR_SESSION_NAME="$HERDR_SESSION_NAME" _herdr new-session -d -s "$SESSION_NAME" -x 140 -y 40 -c "$WORKSPACE" "$CMD_FULL"
+        HERDR_SESSION_NAME="$HERDR_SESSION_NAME" _herdr new-session -d -s "$SESSION_NAME" -x 140 -y 40 -c "$WORKSPACE" "$CP_PREFIX$CMD_FULL"
       fi
       ;;
     agy|hermes|grok|codex)
-      HERDR_SESSION_NAME="$HERDR_SESSION_NAME" _herdr new-session -d -s "$SESSION_NAME" -x 140 -y 40 -c "$WORKSPACE" "$CMD_FULL"
+      HERDR_SESSION_NAME="$HERDR_SESSION_NAME" _herdr new-session -d -s "$SESSION_NAME" -x 140 -y 40 -c "$WORKSPACE" "$CP_PREFIX$CMD_FULL"
       ;;
     opencode)
       if [ -z "${OPENCODE_PERMISSION:-}" ]; then
         export OPENCODE_PERMISSION='{"*":"allow"}'
       fi
-      HERDR_SESSION_NAME="$HERDR_SESSION_NAME" _herdr new-session -d -s "$SESSION_NAME" -x 140 -y 40 -c "$WORKSPACE" "$CMD_FULL"
+      HERDR_SESSION_NAME="$HERDR_SESSION_NAME" _herdr new-session -d -s "$SESSION_NAME" -x 140 -y 40 -c "$WORKSPACE" "$CP_PREFIX$CMD_FULL"
       ;;
     *) echo "ERROR: --agent must be claude, agy, hermes, grok, opencode, or codex, got: $AGENT" >&2; exit 2 ;;
   esac
@@ -391,7 +409,7 @@ atomic_dump_yaml "$AGENT_SESSIONS_YAML" \
   MAM_WS_LABEL="$MAM_WS_LABEL" \
   SESSION_UUID="$SESSION_UUID" \
   LAST_VISIBLE_STATUS_OVERRIDE="${LAST_VISIBLE_STATUS_OVERRIDE:-}" \
-  DELEGATE_JOB_ID="$DELEGATE_JOB_ID" ROLE="$ROLE" <<'PYEOF'
+  DELEGATE_JOB_ID="$DELEGATE_JOB_ID" ROLE="$ROLE" CONTROL_PLANE_JSON="$CP_JSON" <<'PYEOF'
 name = os.environ['SESSION_NAME']
 agent = os.environ['AGENT']
 role = os.environ['ROLE']
@@ -487,6 +505,15 @@ last_vis_override = os.environ.get('LAST_VISIBLE_STATUS_OVERRIDE', '')
 if last_vis_override:
     entry['last_visible_status'] = last_vis_override
 
+# Issue #21: the control-plane binding this row was created against.
+_cp_raw = os.environ.get('CONTROL_PLANE_JSON', '')
+if _cp_raw:
+    try:
+        import json as _json
+        _cp = _json.loads(_cp_raw)
+        entry['control_plane'] = {k: _cp[k] for k in ('workspace_root', 'agent_sessions_yaml', 'mam_jobs_dir', 'delegate_job_logs_dir')}
+    except Exception:
+        pass
 sessions.append(entry)
 
 snap = d.setdefault('snapshot', {})
@@ -503,7 +530,20 @@ if [ -n "$DELEGATE_JOB_ID" ]; then
   
   # Construct instructions matching multi-agent-mux-delegate-job format
   py_bin="$(_delegate_py_bin)"
-  pub="$py_bin .agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py --job $DELEGATE_JOB_ID"
+  # Issue #21: absolute script, registry and audit-log paths (shell-quoted), so
+  # an agent whose cwd is a git worktree can run the command as written.
+  pub_script="$WORKSPACE_ROOT/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py"
+  [ -f "$pub_script" ] || pub_script="$(_delegate_script publish_event.py)"
+  pub_script="$(cd -P "$(dirname "$pub_script")" && pwd -P)/$(basename "$pub_script")"
+  pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}"
+  pub_reg="${pub_reg:-.mam/jobs}"
+  mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)"
+  pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' 2>/dev/null || true)"
+  pub="$(
+    if [ -n "$pub_logs" ]; then printf 'DELEGATE_JOB_LOGS_DIR=%q ' "$pub_logs"; fi
+    printf '%q ' "$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$DELEGATE_JOB_ID"
+  )"
+  pub="${pub% }"
   instructions="Your job_id is \"$DELEGATE_JOB_ID\" (the one just registered for THIS delegation).
 
 On start run:        $pub --event started.

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -68,7 +68,42 @@ pick_python() {
   echo "$py_bin"
 }
 
-REGISTRY_DIR_DEFAULT=".mam/jobs"
+REGISTRY_DIR_DEFAULT="${MAM_JOBS_DIR:-.mam/jobs}"
+
+# Issue #21: the injected brief path, the start/success/error commands and the
+# Output Report Path must work from any cwd - a creator in a git worktree runs
+# them from there. Called right after `cd "$WORKDIR"`, so a relative registry
+# still resolves against WORKDIR, as before.
+_abs_registry_dir() {
+  mkdir -p "$REGISTRY_DIR"
+  REGISTRY_DIR="$(cd -P "$REGISTRY_DIR" && pwd -P)"
+  export SKS_REGISTRY_DIR="$REGISTRY_DIR"
+}
+
+# The audit-log root this wrapper's own subscriber resolves (same resolver),
+# pinned into the injected commands so an agent publishing from another cwd
+# does not fall back to <cwd>/.mam/delegate_job_logs.
+_logs_dir_abs() {
+  "$PY" -c 'import os, sys; sys.path.insert(0, sys.argv[1]); from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' "$SCRIPT_DIR/.."
+}
+
+# _pub_cmd <extra...>: the publish command as one shell-quoted string.
+_pub_cmd() {
+  local s
+  s="$(
+    if [ -n "${LOGS_DIR_ABS:-}" ]; then printf 'DELEGATE_JOB_LOGS_DIR=%q ' "$LOGS_DIR_ABS"; fi
+    printf '%q ' "$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$JOB_ID" "$@"
+  )"
+  printf '%s' "${s% }"
+}
+
+# _instructions <pub> [iteration]: the prompt injected into the agent (direct
+# and loop jobs share it; test18 pins parts of this wording).
+_instructions() {
+  local pub="$1" iter="${2:-}" brief="$REGISTRY_DIR/$JOB_ID/brief.md"
+  printf 'Your job_id is "%s". Detailed task requirements, instructions, and target output paths%s are documented in the task brief file at: %s. Please READ and follow %s to complete your work. Commands: start='"'"'%s --event started --heartbeat %s'"'"' (run it FIRST, within %s s; it also starts an automatic heartbeat timer), success='"'"'%s --event completed --detail <summary>'"'"', error='"'"'%s --event error --detail <reason>'"'"'.' \
+    "$JOB_ID" "${iter:+ for iteration $iter}" "$brief" "$brief" "$pub" "$HEARTBEAT_SEC" "$ACK_TIMEOUT" "$pub" "$pub"
+}
 
 usage() {
   cat <<'EOF'
@@ -150,8 +185,10 @@ cmd_submit() {
   parse_opts "$@"
   [[ -n "$PROMPT" ]] || { echo "submit requires --prompt" >&2; exit 1; }
   PY="$(pick_python)"
+  [[ "$PY" == */* && "$PY" != /* ]] && PY="$(pwd)/$PY"
   cd "$WORKDIR"
-  mkdir -p "$REGISTRY_DIR"
+  _abs_registry_dir
+  LOGS_DIR_ABS="$(_logs_dir_abs 2>/dev/null || true)"
 
   # 1) register job (prints the new job id)
   JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \
@@ -175,7 +212,7 @@ cmd_submit() {
 - **Role**: $DELEGATE_ROLE
 - **Timeout**: $TIMEOUT s (Idle: $IDLE_TIMEOUT s)
 - **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s
-- **Output Report Path**: .mam/jobs/$JOB_ID/$AGENT-reports/report-final.md
+- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/$AGENT-reports/report-final.md
 
 ## 🔎 Task Description
 $PROMPT
@@ -195,8 +232,9 @@ EOF
     local pub_extra=()
     [ -n "$_disp_tok" ] && pub_extra+=(--expect-dispatch-token "$_disp_tok")
     [ -n "$_attempt_id" ] && pub_extra+=(--attempt-id "$_attempt_id")
-    local pub="$PY $SCRIPT_DIR/scripts/publish_event.py --registry-dir $REGISTRY_DIR --job $JOB_ID ${pub_extra[*]}"
-    local instructions="Your job_id is \"$JOB_ID\". Detailed task requirements, instructions, and target output paths are documented in the task brief file at: .mam/jobs/$JOB_ID/brief.md. Please READ and follow .mam/jobs/$JOB_ID/brief.md to complete your work. Commands: start='$pub --event started --heartbeat $HEARTBEAT_SEC' (run it FIRST, within $ACK_TIMEOUT s; it also starts an automatic heartbeat timer), success='$pub --event completed --detail <summary>', error='$pub --event error --detail <reason>'."
+    local pub instructions
+    pub="$(_pub_cmd ${pub_extra[@]+"${pub_extra[@]}"})"
+    instructions="$(_instructions "$pub")"
 
     if [[ "$DRY_RUN" == "1" ]]; then
       run_agent "$JOB_ID" "$instructions"
@@ -349,7 +387,7 @@ EOF
 - **Role**: $display_role
 - **Iteration**: $iteration
 - **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s
-- **Output Report Path**: .mam/jobs/$JOB_ID/${clean_session}-reports/report-final.md
+- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/${clean_session}-reports/report-final.md
  
 ## 🔎 Task Description
 $current_prompt
@@ -451,8 +489,9 @@ EOF
       # Format instruction block
       local loop_pub_extra=(--attempt-id "$attempt")
       [ -n "$_disp_tok" ] && loop_pub_extra+=(--expect-dispatch-token "$_disp_tok")
-      local pub="$PY $SCRIPT_DIR/scripts/publish_event.py --registry-dir $REGISTRY_DIR --job $JOB_ID ${loop_pub_extra[*]}"
-      local instructions="Your job_id is \"$JOB_ID\". Detailed task requirements, instructions, and target output paths for iteration $iteration are documented in the task brief file at: .mam/jobs/$JOB_ID/brief.md. Please READ and follow .mam/jobs/$JOB_ID/brief.md to complete your work. Commands: start='$pub --event started --heartbeat $HEARTBEAT_SEC' (run it FIRST, within $ACK_TIMEOUT s; it also starts an automatic heartbeat timer), success='$pub --event completed --detail <summary>', error='$pub --event error --detail <reason>'."
+      local pub instructions
+      pub="$(_pub_cmd "${loop_pub_extra[@]}")"
+      instructions="$(_instructions "$pub" "$iteration")"
 
       # Trigger agent
       if ! deliver_prompt "$JOB_ID" "$instructions" "$current_session" "$logf"; then

.agents/skills/multi-agent-mux-loop/SKILL.md

@@ -195,6 +195,21 @@ sequenceDiagram
 | **Iterative Loop** | `--max-loop M` | NOT PASS 판정 시 최대 `M`회까지 Creator가 자체 수정합니다. `--plan` 모드에서 리뷰어가 리포트에 `[ESCALATE: PLANNER]` 태그를 남기면 설계 변경 수준으로 판단하여 Planner에게 계획 갱신을 위임합니다 (린트는 리뷰어가 검토 관점 중 하나로 확인할 뿐, 별도의 자동 게이트는 아닙니다). |
 | **Rebuttal** | `--max-rebut N` | 이터레이션당 Creator 반론 횟수 (기본 1, `0`이면 끔). |
 | **Control** | `--verbose` / `--cleanup` | 상세 로그 / 성공 시 임시 job 디렉터리 삭제. |
+| **Source root** | `--worktree <path>` (env `MAM_LOOP_WORKTREE`, CLI 우선) | diff 수집·base·head·빈 변경분 판정만 이 git 워크트리 기준으로 합니다. wrapper·레지스트리·체크포인트·loop guard·리포트 승격·cleanup 은 루프를 띄운 본 체크아웃(제어 루트)에 남습니다. 모든 위임 프롬프트 앞에 `[워크트리 컨텍스트]` 블록(경로·브랜치·base·head·커밋 수·미커밋 변경)이 붙습니다. Python 루프 전용(Issue #21). |
+| **Diff base** | `--base <ref>` (CLI 전용, env 없음) | 리뷰 diff 의 base 를 고정합니다. 소스 HEAD 의 조상이어야 하며 아니면 exit 1. 생략 시 Phase 2 시작 시점의 소스 HEAD 입니다. |
+| **Review only** | `--review-only` | Phase 2(구현 위임)를 건너뛰고 `<base>..HEAD` 를 리뷰합니다. `--base` 필수, `--plan` 과 함께 쓸 수 없습니다. 수정·순응 잡은 여전히 creator 에게 갑니다. |
+
+`--worktree` 또는 `--base` 가 있으면 매 리뷰 라운드(셀프 리뷰 포함) 전에 소스 루트의 변경분을 git 종료 코드로 판정합니다. 변경분이 비었거나 판정할 수 없으면 리뷰를 요청하지 않고 exit 1 로 끝납니다(fail-closed). 둘 다 없으면 기존 동작 그대로입니다.
+
+구현이 루프 밖(워크트리의 creator 에게 `delegate-job --type direct`)에서 끝났다면, **구현 위임 전에** base 를 기록해 두고 리뷰만 돌립니다:
+
+```bash
+BASE=$(git -C "$WT" rev-parse HEAD)          # 구현 위임 전에 기록
+# ... creator 가 $WT 에서 구현·커밋 ...
+bash .agents/skills/multi-agent-mux-loop/scripts/run_loop.sh \
+  --worktree "$WT" --base "$BASE" --review-only \
+  --creator "<worktree-creator>" --reviewer "<reviewer>" --task "Issue #N review"
+```
 
 ---
 
@@ -232,5 +247,5 @@ bash .agents/skills/multi-agent-mux-loop/scripts/run_loop.sh \
 - **Incorrect Verdict format (앵커링 파서 하드닝)**: 리뷰 리포트 파일 내에서 `[VERDICT: PASS]` 또는 `[VERDICT: NOT PASS]` 토큰은 반드시 리포트의 **마지막에 단독 행**으로 기재되어야 합니다. 코드 인용이나 변경 diff 내에 등장하는 토큰은 매칭 대상에서 완전 배제됩니다.
 - **Fail-closed on missing verdict**: 최종 Verdict 토큰이 누락되거나 리포트 픽업에 실패하면, 파서는 **경고 후 통과시키는 것이 아니라** 안전을 위해 즉시 `NOT PASS`로 판정(fail-closed)하고 교정 사이클을 수행합니다. 리뷰어에게는 반드시 리포트 끝에 단독 행으로 토큰을 찍도록 지시해야 합니다.
 - **Session Availability**: `run_loop.sh` 기동 전에 참조되는 Planner, Target Agent, Reviewer 세션들이 모두 herdr 세션으로 기동되어 (`status.sh` 기준 `alive` 및 `running`) 있어야 합니다.
-- **동시 루프 기동 금지 (NFS Lock Shadowing)**: 동일한 작업 트리 내에서 다수의 `run_loop.sh` 제어기를 동시에 기동하면 SQLite DB 갱신 경합 및 YAML 데이터 오염이 발생합니다. 하나의 루프가 끝날 때까지 다른 루프를 병렬로 기동하지 마십시오.
+- **동시 루프 기동 금지 (NFS Lock Shadowing)**: 동일한 작업 트리 내에서 다수의 `run_loop.sh` 제어기를 동시에 기동하면 SQLite DB 갱신 경합 및 YAML 데이터 오염이 발생합니다. 하나의 루프가 끝날 때까지 다른 루프를 병렬로 기동하지 마십시오. (실제로는 loop lock 이 두 번째 루프를 exit 1 로 거부한다 — run_loop.sh 의 `mam_acquire_loop_lock` 분기)
 - **원자적 아카이빙 (Promotion)**: 루프 성공 종료 시 최종 계획서와 검증 리포트들은 `.agents/reports/<session_name>/` 디렉토리로 원자적으로 덮어쓰기(`mv -f`)되어 보존됩니다. 해당 경로의 리포트들로 VCS 추적성을 확보해야 합니다.

.agents/skills/multi-agent-mux-loop/scripts/run_loop.sh

@@ -32,6 +32,9 @@ TARGET_AGENT=""
 TASK=""
 REVIEWER_LIST=""
 PLANNER_SESSION_OVERRIDE=""
+LOOP_WORKTREE=""
+LOOP_BASE=""
+LOOP_REVIEW_ONLY=0
 
 # Print usage instructions
 usage() {
@@ -47,6 +50,9 @@ usage() {
   echo "  --max-rebut N          Max rebuttal attempts per review iteration (default: 1, 0: disabled)"
   echo "  --verbose              Print detailed execution timeline traces"
   echo "  --cleanup              Purge temporary job directories upon success"
+  echo "  --worktree <path>      Source worktree for diff/base/head (control plane stays here)"
+  echo "  --base <ref>           Pin the review diff base (must be an ancestor of the source HEAD)"
+  echo "  --review-only          Skip implementation; review <base>..HEAD (requires --base)"
   exit 1
 }
 
@@ -83,6 +89,9 @@ while [[ "$#" -gt 0 ]]; do
       ;;
     --planner) PLANNER_SESSION_OVERRIDE="$2"; shift 2 ;;
     --task) TASK="$2"; shift 2 ;;
+    --worktree) LOOP_WORKTREE="$2"; shift 2 ;;
+    --base) LOOP_BASE="$2"; shift 2 ;;
+    --review-only) LOOP_REVIEW_ONLY=1; shift ;;
     -h|--help) usage ;;
     *) echo "Unknown option: $1"; usage ;;
   esac
@@ -104,6 +113,15 @@ if [ -n "${PLANNER_SESSION_OVERRIDE:-}" ] && [ "$PLAN_MODE" = false ]; then
   exit 1
 fi
 
+# Issue #21 (D12): the legacy shell loop assumes a cwd-relative .mam/jobs and
+# diffs the checkout it runs in; the worktree/base options are Python-only.
+# Checked before the freeze and the loop lock (log_* are not defined yet).
+if [ "${MAM_LOOP_PY:-1}" = "0" ] && { [ -n "$LOOP_WORKTREE" ] || [ -n "$LOOP_BASE" ] \
+     || [ "$LOOP_REVIEW_ONLY" = "1" ] || [ -n "${MAM_LOOP_WORKTREE:-}" ]; }; then
+  echo "ERROR: --worktree/--base/--review-only require the Python loop (MAM_LOOP_PY=1)" >&2
+  exit 1
+fi
+
 # Anchor Python interpreter before freeze so freeze child processes find the project venv (O-4)
 if [ -z "${AGENT_PYTHON_BIN:-}" ]; then
   if [ -x "$REPO_ROOT/.venv/bin/python" ]; then

.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh

@@ -100,7 +100,7 @@ if [ "$SUBSCRIBE" = "1" ]; then
   # The MQTT subscribe loop exits 3 to signal "broker unavailable → poll instead".
   set +e
   YAML_PATH="$AGENT_SESSIONS_YAML" HOME_DIR="$HOME_DIR" CLAUDE_PROJECT_DIR="$CLAUDE_PROJECT_DIR" LOCAL_BIN="$LOCAL_BIN" \
-    WORKSPACE_ROOT="$WORKSPACE_ROOT" SUB_TIMEOUT="$SUB_TIMEOUT" SUB_IDLE_TIMEOUT="$SUB_IDLE_TIMEOUT" \
+    WORKSPACE_ROOT="$WORKSPACE_ROOT" MAM_JOBS_DIR="${MAM_JOBS_DIR:-}" SUB_TIMEOUT="$SUB_TIMEOUT" SUB_IDLE_TIMEOUT="$SUB_IDLE_TIMEOUT" \
     SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" \
     "$PYBIN" - <<'PYEOF' &
 import os, sys, json, time, subprocess
@@ -236,7 +236,9 @@ def on_message(_client, _userdata, msg):
         if not jid or not event:
             return
         
-        if workspace_root:
+        if os.environ.get('MAM_JOBS_DIR'):
+            registry_dir = os.environ['MAM_JOBS_DIR']  # Issue #21: MAM_JOBS_DIR wins
+        elif workspace_root:
             registry_dir = os.path.join(workspace_root, '.mam', 'jobs')
         else:
             yaml_dir = os.path.dirname(yaml_path) if yaml_path else ""

.agents/skills/multi-agent-mux-resume/scripts/resume_session.sh

@@ -150,6 +150,29 @@ else
   fi
 fi
 
+# Issue #21 (item 3, D11): the control-plane binding recorded on the row wins
+# over the caller's environment (a difference is reported on stderr); a legacy
+# row without one uses the caller's binding, which update_yaml_resumed.sh then
+# records. Forwarded to the pane only when it differs from --workspace's own.
+ROW_CP="$(MAM_STATE_JSON="$(load_state_json)" SESSION_NAME="$SESSION_NAME" python3 -c "
+import json, os
+d = json.loads(os.environ.get('MAM_STATE_JSON') or '{}')
+for s in d.get('herdr_sessions', []):
+    if s.get('name') == os.environ['SESSION_NAME'] and isinstance(s.get('control_plane'), dict):
+        print(json.dumps(s['control_plane']))
+        break
+" 2>/dev/null || true)"
+CP_JSON="$(mam_control_plane_json "$WORKSPACE" "$ROW_CP" || true)"
+CP_PREFIX=""
+if [ -n "$CP_JSON" ]; then
+  cp_rc=0
+  CP_PREFIX="$(mam_control_plane_prefix "$CP_JSON")" || cp_rc=$?
+  if [ "$cp_rc" -eq 3 ]; then
+    echo "ERROR: control-plane path contains whitespace; cannot forward to the pane" >&2
+    exit 2
+  fi
+fi
+
 if [ "${DRY_RUN:-0}" = "1" ]; then
   echo "[dry-run] would spawn: $CMD_FULL"
   exit 0
@@ -162,7 +185,7 @@ if [ "$AGENT" = "opencode" ]; then
 fi
 
 # 4. Spawn new agent session (delegates to herdr translation shim)
-_herdr new-session -d -s "$SESSION_NAME" -c "$WORKSPACE" "$CMD_FULL"
+_herdr new-session -d -s "$SESSION_NAME" -c "$WORKSPACE" "$CP_PREFIX$CMD_FULL"
 if [ "$AGENT" = "claude" ]; then
   # auto-handle trust / bypass dialogs
   handle_startup_dialogs "$SESSION_NAME" 20
@@ -172,7 +195,7 @@ fi
 sleep 2
 
 # 5. Update agent-sessions.yaml: status running, last_visible_status
-bash "$(dirname "${BASH_SOURCE[0]}")/update_yaml_resumed.sh" \
+CONTROL_PLANE_JSON="$CP_JSON" bash "$(dirname "${BASH_SOURCE[0]}")/update_yaml_resumed.sh" \
   --session "$SESSION_NAME" --uuid "$UUID" --agent "$AGENT" --workspace "$WORKSPACE" \
   --herdr-session "$HERDR_SESSION_NAME" --cmd-full "$CMD_FULL" \
   ${HERDR_WORKSPACE_OPT:+--herdr-workspace "$HERDR_WORKSPACE_OPT"}

.agents/skills/multi-agent-mux-resume/scripts/update_yaml_resumed.sh

@@ -163,6 +163,16 @@ else:
         target['herdr_workspace'] = wsl
 
 target['status'] = 'running'
+# Issue #21 (D11): a row keeps the control-plane binding it was created with;
+# a legacy row without one gets the binding this resume used.
+_cp_raw = os.environ.get('CONTROL_PLANE_JSON', '')
+if _cp_raw and not isinstance(target.get('control_plane'), dict):
+    try:
+        import json as _json
+        _cp = _json.loads(_cp_raw)
+        target['control_plane'] = {k: _cp[k] for k in ('workspace_root', 'agent_sessions_yaml', 'mam_jobs_dir', 'delegate_job_logs_dir')}
+    except Exception:
+        pass
 if not uuid:
     # ISSUE-3 (challenge c4b0a075): a Class A fresh-spawn resume re-arms
     # discovery from scratch - the resumed agent has not spoken yet, so any

.agents/skills/multi-agent-mux-status/scripts/status.sh

@@ -160,7 +160,14 @@ def get_job_status(s):
         return ('-', '-')
 
     project_root = os.environ.get('PROJECT_ROOT', '.')
-    candidates = [
+    # Issue #21: MAM_JOBS_DIR and the registry the row was created against
+    # (control_plane) first, then the matching audit logs, then the old ones.
+    cp = s.get('control_plane') or {}
+    candidates = [os.path.join(r, f"{jid}.json")
+                  for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
+    candidates += [os.path.join(r, jid, 'status.json')
+                   for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r]
+    candidates += [
         os.path.join('.mam', 'jobs', f"{jid}.json"),
         os.path.join(project_root, '.mam', 'jobs', f"{jid}.json"),
         os.path.join(project_root, '.mam', 'delegate_job_logs', jid, 'status.json'),
@@ -361,7 +368,14 @@ def get_job_status(s):
     project_root = os.environ.get('PROJECT_ROOT', '.')
     # Candidate locations (review item 6: project-root-relative, no hardcoded abs paths):
     #   1) cwd-relative registry  2) project-root registry  3) project-root audit log
-    candidates = [
+    # Issue #21: MAM_JOBS_DIR and the registry the row was created against
+    # (control_plane) first, then the matching audit logs, then the old ones.
+    cp = s.get('control_plane') or {}
+    candidates = [os.path.join(r, f"{jid}.json")
+                  for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
+    candidates += [os.path.join(r, jid, 'status.json')
+                   for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r]
+    candidates += [
         os.path.join('.mam', 'jobs', f"{jid}.json"),
         os.path.join(project_root, '.mam', 'jobs', f"{jid}.json"),
         os.path.join(project_root, '.mam', 'delegate_job_logs', jid, 'status.json'),

.agents/skills/multi-agent-mux-stop/scripts/stop_session.sh

@@ -132,7 +132,9 @@ for s in d.get('herdr_sessions', []):
     if s.get('name') == name:
         cwd = (s.get('pane') or {}).get('cwd', '')
         jid = s.get('delegate_job_id', '') or ''
-        print(json.dumps({'cwd': cwd, 'job_id': jid}))
+        cp = s.get('control_plane') or {}
+        print(json.dumps({'cwd': cwd, 'job_id': jid, 'mam_jobs_dir': cp.get('mam_jobs_dir', '') or '',
+                          'delegate_job_logs_dir': cp.get('delegate_job_logs_dir', '') or ''}))
         sys.exit(0)
 sys.exit(7)
 ") || {
@@ -142,6 +144,19 @@ sys.exit(7)
 
 TARGET_CWD=$(printf '%s' "$MAPPED_DATA" | python3 -c 'import sys,json; print(json.load(sys.stdin).get("cwd",""))')
 DELEGATE_JOB_ID=$(printf '%s' "$MAPPED_DATA" | python3 -c 'import sys,json; print(json.load(sys.stdin).get("job_id",""))')
+# Issue #21: a row created against another control plane names its registry.
+# Order: MAM_JOBS_DIR > the row's control_plane.mam_jobs_dir > WORKSPACE_ROOT >
+# .mam/jobs; exported so delegate_publish_event (lib.sh) uses the same one.
+ROW_JOBS_DIR=$(printf '%s' "$MAPPED_DATA" | python3 -c 'import sys,json; print(json.load(sys.stdin).get("mam_jobs_dir",""))')
+if [ -z "${MAM_JOBS_DIR:-}" ] && [ -n "$ROW_JOBS_DIR" ]; then
+  export MAM_JOBS_DIR="$ROW_JOBS_DIR"
+fi
+# Same for the audit log (R3): the cancellation history belongs next to the
+# registry the row names, not under $WORKSPACE_ROOT/.mam/delegate_job_logs.
+ROW_LOGS_DIR=$(printf '%s' "$MAPPED_DATA" | python3 -c 'import sys,json; print(json.load(sys.stdin).get("delegate_job_logs_dir",""))')
+if [ -z "${DELEGATE_JOB_LOGS_DIR:-}" ] && [ -n "$ROW_LOGS_DIR" ]; then
+  export DELEGATE_JOB_LOGS_DIR="$ROW_LOGS_DIR"
+fi
 DELEGATE_JOB_TOKEN=""
 DELEGATE_JOB_ATTEMPT=""
 if [ -n "$DELEGATE_JOB_ID" ]; then

docs/design/worktree-creator-model.md

@@ -1,6 +1,6 @@
 # 워크트리 운영 모델 제안 — creator 만 워크트리, 리뷰어는 공유 (Design Proposal)
 
-- 상태: 제안 (리뷰 요청)
+- 상태: 리뷰 반영(4건, docs/design/reviews/) — 1단계 선행 작업 구현(Issue #21 항목 1·2·3·6)
 - 작성: 오케스트레이터 세션 `dbe6dd10-a210-4b46-b613-58a1860499b6`, 2026-09-14
 - 배경 이슈: #17 항목 4(셸 계층 축소)·#18 사후 분석(체크아웃 전환 사고)
 
@@ -13,7 +13,7 @@ MAM 은 **워크스페이스 = `.agents/skills` 가 들어 있는 체크아웃
 | `WORKSPACE_ROOT` | `lib.sh:26` — `${WORKSPACE_ROOT:-$SKILL_DIR/../..}` |
 | `.mam/` (agent-sessions.yaml, jobs 레지스트리, 감사 로그, monitor.lock) | `$WORKSPACE_ROOT/.mam` (gitignore 대상) |
 | herdr 서버 이름 | `resolve_herdr_session` — 워크스페이스 경로 슬러그, `--herdr-session` 으로 override |
-| 모니터 토픽 지문 | `reconcile.sh:288-290` — `sha256(abspath(workspace_root))[:12]` |
+| 모니터 토픽 지문 | `reconcile.sh:288-290` — `sha256(abspath(workspace_root))[:12]`. 모니터 구독에만 쓰인다. 잡 토픽은 레코드의 topic_prefix `python/mqtt/jobs/<job_id>`(mqtt_common.py:137, :351), HMAC 은 레코드의 auth_token 이다 |
 | claude 대화 저장 키 | `~/.claude/projects/<cwd 경로 키>` — resume 는 같은 디렉터리여야 함 |
 
 그 결과 두 가지 문제가 있다.
@@ -24,63 +24,104 @@ MAM 은 **워크스페이스 = `.agents/skills` 가 들어 있는 체크아웃
 ## 2. 후보안
 
 ### A. 모두 워크트리별 (오케스트레이터 + creator + 리뷰어)
-- 장점: 완전 격리. 현재 코드가 그대로 동작한다(워크트리 = 새 워크스페이스).
-- 단점: 이슈마다 에이전트 5개를 새로 띄운다 → 토큰·요청 한도가 이슈 수에 비례(codex 는 이번 세션에서 이미 한도 도달). 리뷰어 대화 문맥이 이슈마다 초기화된다. `.mam`·모니터·토픽 지문이 이슈마다 흩어져 관측이 분산된다.
+- 장점: 런타임 분리가 가장 강하다(단 git 워크트리는 refs·객체 저장소를 공유하고, `.venv`·캐시·자격증명·브로커도 공유 자원이다). 워크트리 준비·수명 처리는 별도로 필요하다.
+- 단점: 이슈마다 에이전트를 새로 띄운다. 용량 추정: 이슈당 5개 에이전트는 운영 정책이고, 토큰 사용은 살아 있는 세션 수가 아니라 실제 작업량에 비례한다. 리뷰어 대화 문맥이 이슈마다 초기화된다. `.mam`·모니터·토픽 지문이 이슈마다 흩어져 관측이 분산된다.
 
 ### B. 오케스트레이터 커밋만 워크트리, 에이전트는 본 체크아웃
 - 장점: 오케스트레이터 커밋은 격리된다.
 - 단점: **병렬성이 없다.** 에이전트는 대화 하나짜리 프로세스라 잡을 하나씩만 처리하고, 두 루프가 같은 creator·리뷰어를 두고 `session_busy`(rc 5) 로 경합해 직렬화된다. creator 는 여전히 본 체크아웃에서 브랜치를 옮겨 커밋해야 하므로 #18 사고 유형이 그대로 남는다. 리뷰어가 검증하는 트리가 브랜치 트리가 아니다.
 
 ### C. (제안) creator 만 워크트리, 리뷰어 풀은 공유, 제어 평면은 하나
-- **creator = 워크트리 전용 인스턴스.** 편집·커밋하는 유일한 에이전트이므로 cwd 가 곧 이슈 브랜치여야 한다. 이슈마다 `creator-<agent>-<issue>` 로 만들고 끝나면 정지(대화 보존, 필요 시 재개).
-- **리뷰어 풀 = 본 체크아웃 공유.** 읽기 전용. 브리프가 워크트리 경로·base·head 를 명시하고 리뷰어는 그 경로에서 `git diff <base>...<head>` 와 테스트를 실행한다. 리뷰가 병목이면 리뷰어 인스턴스를 추가한다.
+- **creator = 워크트리 전용 인스턴스.** 제품 코드를 편집·커밋하는 유일한 에이전트이므로 cwd 가 곧 이슈 브랜치여야 한다. 오케스트레이터 커밋과 루프의 리포트 승격(`runner.py` 의 Promotion 블록 → 제어 루트 `.agents/reports/`)은 별도 소유다. 이슈마다 `creator-<agent>-<issue>` 로 만들고 끝나면 정지(대화 보존, 필요 시 재개). 재개하려면 같은 워크트리 경로가 있어야 한다(대화 키가 cwd).
+- **리뷰어 풀 = 본 체크아웃 공유.** 리뷰어는 소스를 편집하지 않는다. 그러나 테스트 실행은 쓰기를 동반하므로 '읽기 전용'이 아니다. 검증 격리(head SHA detached 워크트리 또는 격리 옵션)와 (loop_id, base, head) 판정 결속은 항목 5 에서 정한다. 1단계 브리프는 러너가 경로·base·head·커밋 수를 자동 삽입한다(`--worktree`/`--base`). 리뷰가 병목이면 리뷰어 인스턴스를 추가한다.
 - **오케스트레이터 커밋 = 워크트리.** 본 체크아웃은 리뷰어의 작업 공간이므로 오케스트레이터는 그곳에서 브랜치를 바꾸지 않는다.
-- **제어 평면 = 본 체크아웃의 `.mam` 하나.** 워크트리에서 creator 를 만들 때 `WORKSPACE_ROOT`·`AGENT_SESSIONS_YAML`·`MAM_JOBS_DIR`·`DELEGATE_JOB_LOGS_DIR` 를 본 체크아웃으로 넘기면 같은 yaml 에 등록되고(행의 `pane.cwd` 는 워크트리), 잡 레지스트리·토픽 지문·모니터가 한 곳에 모인다.
+- **제어 평면 = 본 체크아웃의 `.mam` 하나.** create/resume 이 바인딩(`WORKSPACE_ROOT`·`AGENT_SESSIONS_YAML`·`MAM_JOBS_DIR`·`DELEGATE_JOB_LOGS_DIR`)을 행(`control_plane`)에 영속하고, `--workspace` 기본값과 다르면 pane 에 명시 전파한다(항목 3). 호출자 env 만으로는 pane 상속이 보장되지 않는다(lib.sh:667-680 — new-session 은 선행 `K=V` 토큰만 `--env` 로 바꾼다). 행의 `pane.cwd` 는 워크트리다.
 
-## 3. C 안의 동작 방식 (현재 코드 기준)
+## 3. C 안의 동작 방식 (부분 지원 + 1단계 수정 반영)
 
 ```bash
-MAIN=/path/to/multi-agent-mux            # 본 체크아웃: 리뷰어 풀 + .mam
-WT=/path/to/multi-agent-mux-wt/issue-20  # 이슈 워크트리
+MAIN=/path/to/multi-agent-mux            # 본 체크아웃: 리뷰어 풀 + .mam (제어 루트)
+WT=/path/to/multi-agent-mux-wt/issue-20  # 이슈 워크트리 (소스 루트)
 git -C "$MAIN" worktree add -b fix/issue-20 "$WT" main
-ln -s "$MAIN/.venv" "$WT/.venv"; git -C "$WT" submodule update --init   # 워크트리 준비(.venv, nats-docker)
+ln -s "$MAIN/.venv" "$WT/.venv"; git -C "$WT" submodule update --init   # 워크트리 준비(.venv, nats-docker) — 헬퍼는 항목 4
 
-# creator 를 워크트리 cwd 로, 그러나 본 체크아웃의 .mam 에 등록
+# creator 를 워크트리 cwd 로, 그러나 본 체크아웃의 .mam 에 등록.
+# 바인딩은 행의 control_plane 에 남고, --workspace 기본값과 다르므로 pane 에 --env 로 전파된다.
 WORKSPACE_ROOT="$MAIN" AGENT_SESSIONS_YAML="$MAIN/.mam/agent-sessions.yaml" \
 MAM_JOBS_DIR="$MAIN/.mam/jobs" DELEGATE_JOB_LOGS_DIR="$MAIN/.mam/delegate_job_logs" \
   bash "$MAIN/.agents/skills/multi-agent-mux-create/scripts/create_session.sh" \
     --workspace "$WT" --agent claude --role Creator --session creator-claude-issue20 \
     --herdr-session multi-agent-mux --herdr-workspace mam-agent
 
-# 루프는 본 체크아웃에서, creator 이름만 워크트리용으로
+# (a) 루프가 구현까지 한다: base 는 Phase 2 시작 시점의 워크트리 HEAD
 cd "$MAIN" && bash .agents/skills/multi-agent-mux-loop/scripts/run_loop.sh \
-  --creator creator-claude-issue20 --all-reviewer \
-  --task "[이슈 #20 @ $WT, branch fix/issue-20, base <sha>] ... 리뷰어는 $WT 에서 git diff <base>...HEAD 와 pytest 를 실행하라 ..."
+  --worktree "$WT" --creator creator-claude-issue20 --all-reviewer --task "Issue #20 ..."
+
+# (b) 구현은 루프 밖(1단계 기본): base 를 구현 위임 **전에** 기록한다
+BASE=$(git -C "$WT" rev-parse HEAD)
+bash "$MAIN/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job" submit --type direct \
+  --agent claude --agent-session herdr:creator-claude-issue20 --workdir "$MAIN" --prompt "Issue #20 ..."
+# ... creator 가 $WT 에서 구현·커밋 ...
+cd "$MAIN" && bash .agents/skills/multi-agent-mux-loop/scripts/run_loop.sh \
+  --worktree "$WT" --base "$BASE" --review-only --creator creator-claude-issue20 --all-reviewer --task "Issue #20 review"
+
+# 정지·재개: --workspace 는 원래 워크트리 경로. 재개는 행에 영속된 바인딩을 다시 전파한다
+bash "$MAIN/.agents/skills/multi-agent-mux-stop/scripts/stop_session.sh" --session creator-claude-issue20
+bash "$MAIN/.agents/skills/multi-agent-mux-resume/scripts/resume_session.sh" \
+  --workspace "$WT" --agent claude --session creator-claude-issue20
 ```
 
-- 브리프 규약: 첫 줄에 `워크트리 경로 / 브랜치 / base sha` 를 적고, 리뷰어에게 "그 경로에서 검증"을 명시한다. 잡 브리프 파일(`.mam/jobs/<id>/brief.md`)은 본 체크아웃에 있으므로 creator 는 `$MAIN/.mam/jobs/<id>/brief.md` 를 읽는다(주입 문구의 상대경로 `.mam/jobs/...` 는 creator cwd 기준으로는 존재하지 않는다 → **주입 문구를 절대경로로 바꾸는 작은 변경이 필요**).
-- 루프 러너의 `base_commit`/diff 수집(`runner.py:399-401`)은 `workspace_root`(본 체크아웃) 기준이다. C 안에서는 워크트리 경로를 기준으로 diff 를 모아야 하므로 **`--worktree <path>` 옵션(또는 `MAM_LOOP_WORKTREE` env)이 필요**하다.
+- `--task` 에 base 를 적지 말 것. 러너가 넣는 base 와 둘이 된다. 리뷰어가 받는 base 는 러너의 `[워크트리 컨텍스트]` 블록 하나뿐이어야 한다.
+- 모니터(`reconcile.sh`)는 본 체크아웃에서만 띄운다.
+- 브리프·발행 명령(`--registry-dir`, 감사 로그 루트 `DELEGATE_JOB_LOGS_DIR`)·Output Report Path 를 제어 평면 절대경로로 바꿨다(항목 1). creator 는 워크트리 cwd 에서 주입된 명령을 그대로 실행하면 된다.
+- 루프 러너 `--worktree` 가 추가됐다: diff·base·head 만 소스 루트 기준이고, wrapper·레지스트리·체크포인트·승격은 제어 루트다. `--worktree`/`--base` 가 있으면 빈 변경분(또는 판정 불가)에 리뷰를 요청하지 않는다(fail-closed). `--base` 는 소스 HEAD 의 조상이어야 하고 env 폴백이 없다.
 - 오케스트레이터 커밋 규칙: 커밋 전 `git branch --show-current` 확인(메모리 규칙), 본 체크아웃에서는 `main` 유지.
 
 ## 4. 기대 효과와 한계
 
 - 병렬화 이득은 **구현 단계**에서 크다: 워크트리마다 creator 가 동시에 작업하고, 전체 스위트도 워크트리마다 동시에 돈다.
 - **리뷰 단계는 풀 크기만큼만 병렬**이다. 이슈 3개를 동시에 돌려도 리뷰어 하나는 잡을 순서대로 처리한다(래퍼는 busy 세션에 rc 5 → 러너가 `MAM_SESSION_BUSY_RETRY_S` 뒤 1회 재시도 → 그래도 busy 면 실패). 다중 루프 동시 기동에는 **리뷰어 큐잉 또는 리뷰어 인스턴스 증설**이 전제다.
-- 루프 SKILL.md 의 "동시 루프 기동 금지(NFS lock shadowing)" 는 **같은 `.mam` 을 쓰는 루프**에 대한 경고다. C 안은 `.mam` 을 공유하므로, 다중 루프 동시 기동 전에 이 제약(체크포인트 파일·SQLite 갱신 경합)을 해소해야 한다.
-- 워크트리 준비 비용(.venv 링크, 서브모듈, `.mam.env` 복사)은 스크립트 한 줄로 흡수 가능하다.
+- 같은 `.mam` 의 두 번째 루프는 경고가 아니라 **거부**다: loop lock 이 exit 1 로 막는다(`run_loop.sh` 의 `mam_acquire_loop_lock` 분기). 체크포인트·loop guard 도 파일 하나다(`runner.py` 의 `checkpoint_file`, `MAM_LOOP_MARKER`). YAML/SQLite 는 flock·`BEGIN IMMEDIATE` 로 트랜잭션 안전(atomic_yaml.py)하므로, 문제는 DB 손상이 아니라 **공유 논리 상태와 리뷰어 소유권**이다.
+- **1단계 운영 형태 = 구현 병렬, 리뷰 루프 하나**: creator 구현은 루프 밖(delegate-job direct)에서 워크트리마다 병렬로 한다. 리뷰는 `run_loop.sh --worktree <WT> --base <구현 전 sha> --review-only` 로 한 번에 하나씩 한다. base 를 구현 위임 **전에** 기록하지 않으면 리뷰를 시작할 수 없다(가드가 빈 변경분을 거부한다). busy 처리는 큐가 아니다(1회 재시도). 다중 루프는 항목 8(리뷰어 스케줄러) 이후다.
+- 준비 헬퍼는 항목 4 다. `.mam.env` 는 복사 대신 중앙 `MAM_ENV_FILE` 정책으로 하고, `.venv` 심링크는 gitignore 되지 않으므로 `.git/info/exclude` 로 처리한다.
 
-## 5. 필요한 변경 (작업 항목)
+## 5. 필요한 변경 (Issue #21 작업 항목)
 
-1. 주입 문구·브리프의 잡 경로를 절대경로로(`$REGISTRY_DIR/<id>/brief.md`).
-2. 루프 러너에 `--worktree <path>` (diff 수집·`base_commit`·리뷰 브리프의 경로 자동 삽입).
-3. `create_session.sh` 문서에 "제어 평면 override" 레시피 추가(`WORKSPACE_ROOT`/`AGENT_SESSIONS_YAML`/`MAM_JOBS_DIR`/`DELEGATE_JOB_LOGS_DIR`), 워크트리 준비 헬퍼(`scripts/worktree_prepare.sh`).
-4. 다중 루프 동시 기동 시 `.mam` 경합 해소(체크포인트 파일 이름에 루프 id 포함, 리뷰어 busy 시 대기 큐).
-5. 리뷰어 브리프 템플릿에 워크트리 경로·base·head 필드 고정.
+| # | 항목 | 상태 |
+|---|---|---|
+| 1 | 래퍼 주입 문구·시작 명령(`--registry-dir`, 감사 로그 루트)·Output Report Path 절대경로화, `MAM_JOBS_DIR` 존중 | 이 브랜치 |
+| 2 | 루프 러너 `--worktree` / `--base` / `--review-only`, 빈 변경분 fail-closed 가드, 러너의 `MAM_JOBS_DIR` 존중 | 이 브랜치 |
+| 3 | 제어 평면 바인딩 영속(`control_plane`)·명시 전파(create/resume), stop·status·reconcile 의 `MAM_JOBS_DIR`·행 바인딩 존중 | 이 브랜치 |
+| 4 | 워크트리 준비·해체 헬퍼, resume 을 원래 cwd 로만 허용, create 의 reconcile/watchdog 기동 위치, `MAM_NO_SUBSCRIBER_DAEMON` 기본화, reconcile 의 루트 밖 cwd 재발견 거부(#17-4 셸 축소와 함께) | 대기 |
+| 5 | 리뷰 identity: (loop_id, base_sha, head_sha) 판정 기록·무효화, 리뷰어 검증 격리(detached 워크트리) | 대기 |
+| 6 | 이 설계 문서 정정 | 이 브랜치 |
+| 7 | 수용 사례 고정(실제 herdr 의 pane env 전파 포함) | 대기 |
+| 8 | 다중 루프 네임스페이스·리뷰어 스케줄러 | 보류 |
 
 ## 6. 리뷰어에게 묻는 것
 
+답변: docs/design/reviews/ 4건, 종합은 Issue #21.
+
 1. C 안의 역할 분리(쓰는 쪽만 격리, 읽는 쪽 공유)가 두 문제(병렬성, 체크아웃 전환 사고)를 실제로 해소하는가? 놓친 경로가 있는가(예: 리뷰어가 테스트 중 파일을 만드는 경우, 모니터 reconcile 이 워크트리 cwd 행을 어떻게 다루는지).
 2. 제어 평면 공유(env override)가 현재 코드에서 안전한가? 특히 `pane.cwd` 가 워크트리인 행을 status/monitor/stop/resume 스킬이 올바르게 처리하는가.
 3. A/B 안 대비 C 안의 비용·이득 평가에 동의하는가? 다른 대안이 있는가?
 4. §5 작업 항목의 누락·과잉. 특히 다중 루프 동시 기동의 `.mam` 경합을 먼저 풀어야 하는지, 아니면 "루프는 하나씩, 구현만 병렬" 로 시작해도 되는지.
 5. #17 항목 4(셸 계층 축소)와의 순서: 이 모델을 먼저 문서·레시피로 굳히고 항목 4 에서 코드로 흡수하는 순서가 맞는가.
+
+## 7. 루트 정의와 스킬별 호환성
+
+- **제어 루트**: `.mam` 을 가진 본 체크아웃. 레지스트리·감사 로그·세션 yaml·체크포인트·loop guard·리포트 승격이 여기 있다.
+- **소스 루트**: 이슈 워크트리. creator 의 cwd, 루프의 diff·base·head 대상.
+- **런타임 코드 루트**: 실제로 실행되는 `.agents/skills` 사본. 루프는 제어 루트의 사본을 실행하며, 워크트리의 미검토 `diff_collect.sh` 를 실행하지 않는다.
+
+| 스킬 | 제어 평면 바인딩 | 워크트리 cwd 행 | 1단계 이후 남은 제한 (항목) |
+|---|---|---|---|
+| create | env 에서 결정 → 행 `control_plane` 에 항상 영속. `--workspace` 기본값과 다르면 새 pane 의 생성 호출(`pane split`/`workspace create`)에 `--env` 로 전파한다. 전파가 필요하면 idle pane 을 재사용하지 않는다(`agent start -- <args>` 는 `--` 뒤를 에이전트 인자로 넘기므로 `env K=V` 를 쓸 수 없다). 전파가 필요한데 경로에 공백이 있으면 exit 2. `~/.local/bin` wrapper 경로는 기록만 하고 경고 | pane.cwd = `--workspace` | create 가 `$WORKSPACE` 사본으로 reconcile/watchdog 기동, `MAM_NO_SUBSCRIBER_DAEMON` 미기본 (4) |
+| resume | 행 바인딩 우선(호출자 env 와 다르면 stderr 경고) 재전파. 바인딩 없는 기존 행은 호출자 바인딩으로 채움 | `--workspace` 는 원래 워크트리 경로여야 함 | 워크트리 삭제 시 재개 불가, 종료 헬퍼 (4) |
+| stop | `MAM_JOBS_DIR` > 행의 `mam_jobs_dir` > `$WORKSPACE_ROOT/.mam/jobs` 순으로 잡 조회·발행. 감사 로그도 `DELEGATE_JOB_LOGS_DIR` > 행의 `delegate_job_logs_dir` 순 | 행 cwd 로 uuid 캡처·purge | 해체 순서 헬퍼 (4) |
+| status | `MAM_JOBS_DIR`·행 `mam_jobs_dir`·감사 로그(`DELEGATE_JOB_LOGS_DIR`·행 `delegate_job_logs_dir`)·기존 후보 순 조회 | 행 cwd 로 대화 키 | — |
+| monitor/reconcile | `MAM_JOBS_DIR` 우선(정적 가드로만 검증) | 기존 행은 행 cwd 로 검증 | 루트 밖 cwd 의 미등록 세션은 자동 등록 안 됨 → create 로만 등록 (4) |
+| delegate-job | `--registry-dir` > `MAM_JOBS_DIR` > `.mam/jobs`(`--workdir` 기준), 모두 절대화. 발행 명령이 감사 로그 루트(`DELEGATE_JOB_LOGS_DIR`)를 고정 | 문구·명령·리포트 경로 절대 | 작은따옴표가 든 경로는 지원 범위 밖 |
+| loop | `MAM_JOBS_DIR` 존중, 래퍼에 `--registry-dir` 명시. `--worktree` 로 소스 루트 분리 | diff·base·head 만 소스 루트. **구현이 루프 밖이면 `--base <구현 전 sha> --review-only` 필수**. `--worktree`/`--base` 가 있으면 빈 변경분 거부 | 루프 하나(8), 리뷰 identity·검증 격리(5), 레거시 셸 루프(`MAM_LOOP_PY=0`)는 새 옵션 거부 |
+
+실제 herdr 가 `pane split --env` / `workspace create --env` 로 pane 환경을 넘기는지는 1단계에서 mock 수준으로만 검증했다. 실제 전파는 항목 7 의 수용 사례로 고정한다.

tests/test_herdr_shim_contract.py

@@ -516,10 +516,13 @@ def test_h24_new_session_reuses_idle_pane_without_split(mam_sandbox, mock_herdr,
     assert not split_calls, f"Expected no pane split call when an idle pane was available: {split_calls}"
 
 
-def test_h25_new_session_idle_pane_preserves_env_and_ws_id(mam_sandbox, mock_herdr, mock_agents):
-    """Finding A: when an idle pane is reused, caller-provided KEY=VAL environment
-    flags (and HERDR_WORKSPACE_ID) must be prepended via env to final_cmd so they
-    are not silently lost when pane split / workspace create is bypassed."""
+def test_h25_new_session_with_env_flags_skips_the_idle_pane_and_uses_env_on_split(mam_sandbox, mock_herdr, mock_agents):
+    """Finding A, revised for Issue #21 R1: caller-provided KEY=VAL environment
+    flags (and HERDR_WORKSPACE_ID) must not be silently lost - and must not be
+    put after `agent start --` either, because herdr passes those tokens to the
+    agent executable (--kind) as arguments. With env flags an idle pane is not
+    reused; the env rides on the pane-creation call (pane split / workspace
+    create --env). Reuse without env flags stays covered by h24."""
     tmp_path = mam_sandbox
     state = _load_mock_state(mock_herdr)
     state["panes"] = [{
@@ -538,17 +541,22 @@ def test_h25_new_session_idle_pane_preserves_env_and_ws_id(mam_sandbox, mock_her
     assert res.returncode == 0, f"Stderr: {res.stderr}\nStdout: {res.stdout}"
 
     final_state = _load_mock_state(mock_herdr)
-    agents = final_state.get("agents", {})
-    matched = [d for n, d in agents.items() if "test-creator-claude" in n]
-    assert matched, agents
-    assert matched[0].get("pane_id") == "w1:p_idle"
-    cmd = matched[0].get("command", "")
-    assert "MY_TEST_FLAG=active" in cmd, f"Expected MY_TEST_FLAG=active in agent command: {cmd}"
-    assert "HERDR_WORKSPACE_ID=w1" in cmd, f"Expected HERDR_WORKSPACE_ID=w1 in agent command: {cmd}"
+    # the shim may prefix `--session <name>`; compare the subcommands themselves
+    new_calls = [c[2:] if c[:1] == ["--session"] else c for c in final_state.get("calls", [])[n_calls:]]
+    creates = [c for c in new_calls if c[:2] in (["pane", "split"], ["workspace", "create"])]
+    assert creates, f"expected a pane split / workspace create carrying the env: {new_calls}"
+    env_vals = [c[i + 1] for c in creates for i, t in enumerate(c[:-1]) if t == "--env"]
+    assert "MY_TEST_FLAG=active" in env_vals and "HERDR_WORKSPACE_ID=w1" in env_vals, env_vals
 
-    new_calls = final_state.get("calls", [])[n_calls:]
-    split_calls = [c for c in new_calls if len(c) >= 2 and c[0] == "pane" and c[1] == "split"]
-    assert not split_calls, f"Expected no pane split call when an idle pane was available: {split_calls}"
+    matched = [d for n, d in final_state.get("agents", {}).items() if "test-creator-claude" in n]
+    assert matched and matched[0].get("pane_id") != "w1:p_idle", matched
+
+    starts = [c for c in new_calls if c[:2] == ["agent", "start"]]
+    assert starts, new_calls
+    for c in starts:
+        args = c[c.index("--") + 1:] if "--" in c else []
+        assert "env" not in args, args
+        assert not any(a.startswith(("MY_TEST_FLAG=", "HERDR_WORKSPACE_ID=")) for a in args), args
 
 
 def test_h26_daemon_spawn_uses_process_group_detachment(mam_sandbox, mock_herdr, mock_agents):

tests/test_issue21_worktree_prereqs.py

@@ -0,0 +1,780 @@
+"""Issue #21 phase-1 prerequisites for the worktree operating model.
+
+  §1 (item 1) delegate-job wrapper: absolute brief / start-command registry /
+     audit-log root / Output Report Path, runnable from any cwd.
+  §2 (item 2) loop runner --worktree / --base / --review-only and the
+     fail-closed empty change-set guard; the control plane stays in the main
+     checkout.
+  §3 (item 3) control-plane binding persisted on the session row and forwarded
+     to the pane; status / reconcile / stop honour MAM_JOBS_DIR.
+
+Every test writes only under tmp_path / mam_sandbox (Issue #20).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+RUN_LOOP_SH = SKILLS / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
+
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
+GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid",
+          "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"}
+
+
+def _env(home: Path, extra=None, drop=()) -> dict:
+    env = dict(os.environ)
+    env["DELEGATE_JOB_PYTHON"] = sys.executable
+    env["AGENT_PYTHON_BIN"] = sys.executable
+    env["HOME"] = str(home)
+    env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
+    env["PYTHONPATH"] = str(SKILLS)
+    for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD",
+              "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT"):
+        env.pop(k, None)
+    env.update(UNREACHABLE)
+    if extra:
+        env.update({k: str(v) for k, v in extra.items()})
+    for k in drop:
+        env.pop(k, None)
+    return env
+
+
+def _only_job_id(reg: Path) -> str:
+    ids = [p.stem for p in reg.glob("*.json")]
+    assert len(ids) == 1, ids
+    return ids[0]
+
+
+def _kill_timer(reg: Path, jid: str) -> None:
+    try:
+        pid = int((reg / f"{jid}.heartbeat.pid").read_text().strip())
+        os.kill(pid, signal.SIGKILL)
+    except (OSError, ValueError):
+        pass
+
+
+def _dry_submit(cwd: Path, env: dict, *args) -> subprocess.CompletedProcess:
+    return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--prompt", "p21",
+                           "--dry-run", *args],
+                          capture_output=True, text=True, env=env, cwd=str(cwd), timeout=60)
+
+
+def _injected(stdout: str, name: str) -> str:
+    m = re.search(name + r"='(.*?)'", stdout)
+    assert m, stdout
+    return m.group(1)
+
+
+# ---------------------------------------------------------------------------
+# §1 item 1 — the wrapper's injected paths are absolute
+# ---------------------------------------------------------------------------
+
+def test_t1_1_injected_paths_are_absolute(tmp_path):
+    a = tmp_path / "A"
+    a.mkdir()
+    real_a = os.path.realpath(a)
+    env = _env(tmp_path, {"WORKSPACE_ROOT": a, "DELEGATE_JOB_LOGS_DIR": a / "logs"})
+    res = _dry_submit(a, env, "--workdir", str(a))
+    assert res.returncode == 0, res.stderr
+    jid = _only_job_id(a / ".mam" / "jobs")
+    brief = f"{real_a}/.mam/jobs/{jid}/brief.md"
+    assert res.stdout.count(brief) == 2, res.stdout
+    start = _injected(res.stdout, "start")
+    assert f"--registry-dir {real_a}/.mam/jobs " in start
+    assert start.startswith(f"DELEGATE_JOB_LOGS_DIR={real_a}/logs "), start
+    # no cwd-relative registry path is left anywhere in the injected text
+    assert res.stdout.count(f".mam/jobs/{jid}") == res.stdout.count(f"{real_a}/.mam/jobs/{jid}")
+
+
+def test_t1_2_start_and_success_commands_run_from_another_cwd(tmp_path):
+    """The core regression: a registry and an audit-log root with a space in
+    them, the commands executed as written from a different cwd with no
+    workspace env - the job advances, the audit trail stays in the control
+    root, and nothing is written under the other cwd."""
+    ctl = tmp_path / "ctl root"
+    reg, logs, wt = ctl / "jobs", ctl / "logs", tmp_path / "wt"
+    for d in (reg, logs, wt):
+        d.mkdir(parents=True)
+    env = _env(tmp_path, {"WORKSPACE_ROOT": ctl, "DELEGATE_JOB_LOGS_DIR": logs})
+    res = _dry_submit(ctl, env, "--registry-dir", str(reg), "--workdir", str(ctl),
+                      "--heartbeat", "7", "--ack-timeout", "11")
+    assert res.returncode == 0, res.stderr
+    jid = _only_job_id(reg)
+    start = _injected(res.stdout, "start")
+    success = _injected(res.stdout, "success").replace("<summary>", "done")
+    agent_env = _env(tmp_path)  # no WORKSPACE_ROOT / MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR
+    try:
+        r1 = subprocess.run(["bash", "-c", start + " --attempts 1"], capture_output=True, text=True,
+                            env=agent_env, cwd=str(wt), timeout=120)
+        # rc 2 = broker unreachable; the event is recorded locally first (test18 contract)
+        assert r1.returncode == 2, r1.stderr
+        assert json.loads((reg / f"{jid}.json").read_text())["status"] != "pending"
+        r2 = subprocess.run(["bash", "-c", success + " --attempts 1"], capture_output=True, text=True,
+                            env=agent_env, cwd=str(wt), timeout=120)
+        assert r2.returncode == 2, r2.stderr
+        assert json.loads((reg / f"{jid}.json").read_text())["status"] == "completed"
+    finally:
+        _kill_timer(reg, jid)
+    audit = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+    published = [json.dumps(e) for e in audit if e.get("event") == "published"]
+    assert len(published) == 2, audit
+    assert '"started"' in published[0] and '"completed"' in published[1], published
+    assert not (wt / ".mam").exists(), "the agent's cwd must not grow a .mam"
+
+
+def test_t1_3_mam_jobs_dir_is_the_default_registry(tmp_path):
+    x, y = tmp_path / "X", tmp_path / "Y"
+    env = _env(tmp_path, {"MAM_JOBS_DIR": x, "WORKSPACE_ROOT": tmp_path, "DELEGATE_JOB_LOGS_DIR": tmp_path / "logs"})
+    res = _dry_submit(tmp_path, env, "--workdir", str(tmp_path))
+    assert res.returncode == 0, res.stderr
+    assert len(list(x.glob("*.json"))) == 1
+    assert not (tmp_path / ".mam" / "jobs").exists()
+
+    res = _dry_submit(tmp_path, env, "--workdir", str(tmp_path), "--registry-dir", str(y))
+    assert res.returncode == 0, res.stderr
+    assert len(list(y.glob("*.json"))) == 1, "an explicit --registry-dir wins over MAM_JOBS_DIR"
+    assert len(list(x.glob("*.json"))) == 1
+
+
+def test_t1_4_wrapper_source_guard():
+    """The loop job type never builds its text under --dry-run (it returns
+    early), so both paths are pinned to the shared helpers here."""
+    src = WRAPPER.read_text()
+    assert ".mam/jobs/$JOB_ID" not in src
+    assert src.count('instructions="$(_instructions "$pub")"') == 1
+    assert src.count('instructions="$(_instructions "$pub" "$iteration")"') == 1
+    assert src.count('pub="$(_pub_cmd ') == 2
+    report_lines = [l for l in src.splitlines() if "**Output Report Path**" in l]
+    assert len(report_lines) == 2 and all("$REGISTRY_DIR/$JOB_ID/" in l for l in report_lines)
+
+
+def _sent_texts(state) -> list:
+    out = []
+
+    def walk(o):
+        if isinstance(o, dict):
+            for k, v in o.items():
+                if k == "sent_text" and isinstance(v, str):
+                    out.append(v)
+                else:
+                    walk(v)
+        elif isinstance(o, list):
+            for v in o:
+                walk(v)
+    walk(state)
+    return out
+
+
+def test_t1_5_onboarding_publish_command_is_absolute(mam_sandbox, mock_herdr, mock_agents):
+    tmp = mam_sandbox
+    script = tmp / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
+    res = subprocess.run(["bash", str(script), "--workspace", str(tmp), "--agent", "claude",
+                          "--role", "creator", "--onboard"],
+                         capture_output=True, text=True, cwd=str(tmp), timeout=180)
+    assert res.returncode == 0, res.stdout + res.stderr
+    texts = [t for t in _sent_texts(json.loads(Path(mock_herdr).read_text())) if "publish_event.py" in t]
+    assert texts, "the onboarding instructions were not injected"
+    text = "".join(texts)
+    real = os.path.realpath(tmp)
+    assert f" {real}/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py " in text, text
+    assert f"--registry-dir {real}/.mam/jobs " in text, text
+    assert f"DELEGATE_JOB_LOGS_DIR={real}/.mam/delegate_job_logs " in text, text
+    assert " .agents/skills/" not in text
+
+
+# ---------------------------------------------------------------------------
+# §2 item 2 — loop runner --worktree / --base / --review-only, empty-set guard
+# ---------------------------------------------------------------------------
+
+def _git(cwd: Path, *args: str) -> str:
+    return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True, check=True,
+                          env={**os.environ, **GIT_ID}).stdout.strip()
+
+
+@pytest.fixture
+def wt_repo(tmp_path, monkeypatch):
+    """main checkout (control root) + a linked worktree on branch feat."""
+    main, wt = tmp_path / "main", tmp_path / "wt"
+    main.mkdir()
+    _git(main, "init", "-q")
+    (main / "a.txt").write_text("a\n")
+    _git(main, "add", "a.txt")
+    _git(main, "commit", "-q", "-m", "init")
+    _git(main, "worktree", "add", "-q", "-b", "feat", str(wt))
+    for k, v in GIT_ID.items():
+        monkeypatch.setenv(k, v)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(main))
+    monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
+    monkeypatch.setenv("MAM_JOBS_DIR", str(main / ".mam" / "jobs"))
+    monkeypatch.delenv("MAM_LOOP_WORKTREE", raising=False)
+    monkeypatch.delenv("MAM_LOOP_BASE", raising=False)
+    monkeypatch.setenv("MAM_STATE_JSON", json.dumps({"herdr_sessions": [
+        {"name": "creator-claude-01", "agent": "claude", "role": "creator", "status": "running"},
+        {"name": "reviewer-claude-01", "agent": "claude", "role": "reviewer", "status": "running"},
+    ]}))
+    return main, wt
+
+
+def _runner(*extra):
+    from lib_py.loop import runner as lr
+    return lr.LoopRunner(["--creator", "creator-claude-01", "--task", "t21", *extra])
+
+
+def _drive(monkeypatch, tmp_path, argv, on_worker=None):
+    """run() end to end with the delegation stubbed: every job 'completes' and
+    every report is a PASS. Returns (rc, [(role, prompt)], runner)."""
+    from lib_py.loop import runner as lr
+    monkeypatch.setattr(lr.signal, "signal", lambda *a, **k: None)
+    r = lr.LoopRunner(["--creator", "creator-claude-01", "--task", "t21",
+                       "--reviewer", "reviewer-claude-01", "--max-loop", "1", *argv])
+    calls = []
+    report = tmp_path / "pass-report.md"
+    report.write_text("looks fine\n\n[VERDICT: PASS]\n")
+
+    def fake_delegate(session, agent_type, role, prompt):
+        calls.append((role, prompt))
+        if role == "Worker" and on_worker:
+            on_worker(r)
+        return f"j{len(calls)}"
+
+    monkeypatch.setattr(r, "delegate_job", fake_delegate)
+    monkeypatch.setattr(r, "wait_for_job", lambda *a, **k: True)
+    monkeypatch.setattr(r, "find_report", lambda jid: report)
+    return r.run(), calls, r
+
+
+def _roles(calls, role):
+    return [p for (ro, p) in calls if ro == role]
+
+
+def test_t2_1_diff_and_head_come_from_the_worktree(wt_repo):
+    main, wt = wt_repo
+    (wt / "a.txt").write_text("a\nworktree change\n")
+    (main / "b.txt").write_text("main-only change\n")
+    r = _runner("--worktree", str(wt))
+    r.parse_args()
+    r.base_commit = r._source_head()
+    assert r._source_head() == _git(wt, "rev-parse", "HEAD")
+    diff = r.collect_changes_diff()
+    assert "worktree change" in diff
+    assert "main-only change" not in diff and "b.txt" not in diff
+
+
+def test_t2_2_invalid_worktrees_are_refused(wt_repo, tmp_path):
+    main, wt = wt_repo
+    (wt / "sub").mkdir()
+    (tmp_path / "plain").mkdir()
+    other = tmp_path / "other"
+    other.mkdir()
+    _git(other, "init", "-q")
+    for bad in (tmp_path / "missing", tmp_path / "plain", wt / "sub", other):
+        with pytest.raises(SystemExit) as e:
+            _runner("--worktree", str(bad)).parse_args()
+        assert e.value.code == 1, bad
+
+
+def test_t2_3_delegation_stays_on_the_control_root(wt_repo, monkeypatch):
+    main, wt = wt_repo
+    from lib_py.loop import runner as lr
+    monkeypatch.setenv("MAM_HEARTBEAT_SEC", "20")
+    monkeypatch.setenv("MAM_HEARTBEAT_MISSES", "3")
+    monkeypatch.setenv("MAM_ACK_TIMEOUT_S", "45")
+    r = _runner("--worktree", str(wt))
+    r.parse_args()
+    seen = []
+
+    def fake_run(cmd, **kw):
+        seen.append((cmd, kw))
+
+        class P:
+            returncode = 0
+            stdout = "registered job: cafe21\n"
+            stderr = ""
+        return P()
+
+    monkeypatch.setattr(lr.subprocess, "run", fake_run)
+    assert r.delegate_job("creator-claude-01", "claude", "Worker", "go") == "cafe21"
+    cmd, kw = seen[0]
+    assert os.path.realpath(kw["cwd"]) == os.path.realpath(main)
+    assert cmd[1].endswith("multi-agent-mux-delegate-job/multi-agent-mux-delegate-job")
+    assert not cmd[1].startswith(os.path.realpath(wt))
+    assert cmd[cmd.index("--registry-dir") + 1] == str(main / ".mam" / "jobs")
+    assert cmd[-6:] == ["--heartbeat", "20", "--heartbeat-misses", "3", "--ack-timeout", "45"]
+
+
+def test_t2_4_checkpoint_lives_in_the_control_root(wt_repo):
+    main, wt = wt_repo
+    r = _runner("--worktree", str(wt))
+    r.parse_args()
+    r._record_checkpoint()
+    data = json.loads((main / ".mam" / "loop-checkpoint.json").read_text())
+    assert data["worktree_root"] == os.path.realpath(wt)
+    assert data["base_source"] == "head" and data["review_only"] is False
+    assert data["head_commit"] == _git(wt, "rev-parse", "HEAD")
+    assert not (wt / ".mam").exists()
+
+
+def test_t2_5_prompts_carry_the_worktree_context(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    head0 = _git(wt, "rev-parse", "HEAD")
+
+    def implement(_r):
+        (wt / "a.txt").write_text("a\nimplemented in wt\n")
+
+    rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt)], on_worker=implement)
+    assert rc == 0
+    worker, review = _roles(calls, "Worker")[0], _roles(calls, "Reviewer")[0]
+    for p in (worker, review):
+        assert p.startswith("[워크트리 컨텍스트]\n")
+        assert f"- 소스 경로: {os.path.realpath(wt)}\n" in p
+        assert f"- base: {head0}  (출처: 루프 시작 시 HEAD)" in p
+        assert f"- head: {head0}  (base..head 커밋: 0개" in p
+    assert "미커밋 변경: 있음" in review and "implemented in wt" in review
+
+
+def test_t2_5b_prompts_unchanged_without_options(wt_repo, monkeypatch, tmp_path):
+    main, _ = wt_repo
+    monkeypatch.setenv("WORKSPACE_ROOT", str(main))
+
+    def implement(_r):
+        (main / "a.txt").write_text("a\nimplemented in main\n")
+
+    rc, calls, _ = _drive(monkeypatch, tmp_path, [], on_worker=implement)
+    assert rc == 0
+    assert all("[워크트리 컨텍스트]" not in p for _, p in calls)
+    assert _roles(calls, "Worker")[0].startswith("계획서가 존재하지 않으므로")
+    assert _roles(calls, "Reviewer")[0].startswith("다음 구현 사항(작업 목표: t21)")
+
+
+def test_t2_6_worktree_env_fallback_and_no_base_env(wt_repo, tmp_path):
+    main, wt = wt_repo
+    wt2 = tmp_path / "wt2"
+    _git(main, "worktree", "add", "-q", "-b", "feat2", str(wt2))
+    os.environ["MAM_LOOP_WORKTREE"] = str(wt)
+    os.environ["MAM_LOOP_BASE"] = _git(main, "rev-parse", "HEAD")
+    try:
+        r = _runner()
+        r.parse_args()
+        assert r.worktree_root == Path(os.path.realpath(wt)) and r.worktree_given
+        assert r.base_ref is None and r.base_sha == "" and r.base_source == "head", "MAM_LOOP_BASE is ignored (D17)"
+        r = _runner("--worktree", str(wt2))
+        r.parse_args()
+        assert r.worktree_root == Path(os.path.realpath(wt2)), "the CLI wins over MAM_LOOP_WORKTREE"
+    finally:
+        os.environ.pop("MAM_LOOP_WORKTREE", None)
+        os.environ.pop("MAM_LOOP_BASE", None)
+
+
+def test_t2_7_runner_reads_mam_jobs_dir(wt_repo, tmp_path, monkeypatch):
+    x = tmp_path / "elsewhere" / "jobs"
+    (x / "jid21" / "rev-reports").mkdir(parents=True)
+    (x / "jid21.json").write_text(json.dumps({"status": "completed"}))
+    (x / "jid21" / "rev-reports" / "report-final.md").write_text("r\n")
+    monkeypatch.setenv("MAM_JOBS_DIR", str(x))
+    r = _runner()
+    r.parse_args()
+    assert r.wait_for_job("jid21", max_wait=2) is True
+    assert r.find_report("jid21") == x / "jid21" / "rev-reports" / "report-final.md"
+
+
+@pytest.mark.parametrize("opts,env", [
+    (["--worktree", "/nowhere"], {}),
+    (["--base", "HEAD"], {}),
+    (["--review-only"], {}),
+    ([], {"MAM_LOOP_WORKTREE": "/nowhere"}),
+])
+def test_t2_8_legacy_shell_loop_refuses_the_new_options(mam_sandbox, opts, env):
+    """run_loop.sh sources lib.sh before the refusal, and lib.sh writes its herdr
+    shim under $WORKSPACE_ROOT/.mam/shim: that root is the sandbox, never the
+    caller's control root (#20, review N2)."""
+    marker = mam_sandbox / "loop-marker"
+    run_env = {**os.environ, "WORKSPACE_ROOT": str(mam_sandbox), "MAM_LOOP_PY": "0", "MAM_LOOP_NO_FREEZE": "1",
+               "MAM_LOOP_MARKER": str(marker), **env}
+    res = subprocess.run(["bash", str(RUN_LOOP_SH), "--creator", "c", "--task", "t", *opts],
+                         capture_output=True, text=True, env=run_env, cwd=str(mam_sandbox), timeout=60)
+    assert res.returncode == 1
+    assert "require the Python loop (MAM_LOOP_PY=1)" in res.stderr
+    assert not marker.exists(), "refused before the loop lock"
+    assert (mam_sandbox / ".mam" / "shim" / "herdr").is_file(), "lib.sh's shim belongs in the sandbox"
+
+
+def _commit_impl(wt: Path) -> str:
+    """Implementation committed in the worktree outside the loop; returns the pre-implementation sha."""
+    before = _git(wt, "rev-parse", "HEAD")
+    (wt / "impl.txt").write_text("impl line\n")
+    _git(wt, "add", "impl.txt")
+    _git(wt, "commit", "-q", "-m", "impl")
+    return before
+
+
+def test_t2_9a_committed_impl_without_base_is_refused(wt_repo, monkeypatch, tmp_path, capsys):
+    main, wt = wt_repo
+    _commit_impl(wt)
+    rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt)])
+    assert rc == 1
+    assert _roles(calls, "Reviewer") == []
+    assert "empty change set since base" in capsys.readouterr().err
+
+
+def test_t2_9b_review_only_reviews_the_committed_range(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    before = _commit_impl(wt)
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", before, "--review-only"])
+    assert rc == 0
+    assert _roles(calls, "Worker") == [], "Phase 2 is skipped"
+    review = _roles(calls, "Reviewer")[0]
+    assert "+impl line" in review
+    assert f"- base: {before}  (출처: --base)" in review
+    assert "(base..head 커밋: 1개, 미커밋 변경: 없음)" in review
+
+
+def test_t2_10_base_must_be_an_ancestor(wt_repo):
+    main, wt = wt_repo
+    _git(main, "commit", "-q", "--allow-empty", "-m", "side")  # on main's branch, not in feat
+    side = _git(main, "rev-parse", "HEAD")
+    for ref in (side, "no-such-ref"):
+        with pytest.raises(SystemExit) as e:
+            _runner("--worktree", str(wt), "--base", ref).parse_args()
+        assert e.value.code == 1, ref
+
+
+def test_t2_11_review_only_needs_base_and_no_plan(wt_repo):
+    main, wt = wt_repo
+    head = _git(wt, "rev-parse", "HEAD")
+    for extra in (["--review-only"], ["--review-only", "--base", head, "--plan"]):
+        with pytest.raises(SystemExit) as e:
+            _runner("--worktree", str(wt), *extra).parse_args()
+        assert e.value.code == 1, extra
+
+
+def test_t2_12_guard_is_off_without_options(wt_repo, monkeypatch, tmp_path):
+    """Compatibility: no --worktree/--base, nothing changed - the review is
+    still requested exactly as before."""
+    rc, calls, _ = _drive(monkeypatch, tmp_path, [])
+    assert rc == 0
+    assert len(_roles(calls, "Reviewer")) == 1
+
+
+def test_t2_13_base_without_review_only_still_implements(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    before = _commit_impl(wt)
+    rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", before])
+    assert rc == 0
+    assert len(_roles(calls, "Worker")) == 1
+    assert "+impl line" in _roles(calls, "Reviewer")[0]
+
+
+def test_t2_14_undeterminable_change_set_is_refused(wt_repo, monkeypatch, tmp_path, capsys):
+    main, wt = wt_repo
+
+    def corrupt(r):
+        r.base_commit = "0" * 40  # not an object: git diff exits 128
+
+    rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt)], on_worker=corrupt)
+    assert rc == 1
+    assert _roles(calls, "Reviewer") == []
+    assert "refusing to request reviews on no evidence" in capsys.readouterr().err
+
+
+# ---------------------------------------------------------------------------
+# §3 item 3 — control-plane binding persisted on the row, forwarded to the pane
+# ---------------------------------------------------------------------------
+
+CP_ENV = ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
+
+
+def _binding(root: Path) -> dict:
+    r = os.path.realpath(root)
+    return {"workspace_root": r, "agent_sessions_yaml": f"{r}/.mam/agent-sessions.yaml",
+            "mam_jobs_dir": f"{r}/.mam/jobs", "delegate_job_logs_dir": f"{r}/.mam/delegate_job_logs"}
+
+
+def _rows(tmp: Path) -> list:
+    import yaml
+    return (yaml.safe_load((tmp / ".mam" / "agent-sessions.yaml").read_text()) or {}).get("herdr_sessions") or []
+
+
+def _env_flags(mock_state_file, since=0) -> list:
+    calls = json.loads(Path(mock_state_file).read_text()).get("calls", [])[since:]
+    return [c[i + 1] for c in calls for i, t in enumerate(c[:-1]) if t == "--env"]
+
+
+def _agent_args(mock_state_file, since=0) -> list:
+    """The tokens after `--` of every `herdr agent start` call: herdr passes
+    them to the canonical agent executable (--kind), never to a shell."""
+    calls = json.loads(Path(mock_state_file).read_text()).get("calls", [])[since:]
+    starts = [c for c in calls if any(c[i:i + 2] == ["agent", "start"] for i in range(len(c) - 1))]
+    return [c[c.index("--") + 1:] if "--" in c else [] for c in starts]
+
+
+def _assert_no_env_in_agent_args(mock_state_file, since=0) -> None:
+    arg_lists = _agent_args(mock_state_file, since)
+    assert arg_lists, "no agent start was recorded"
+    for args in arg_lists:
+        assert "env" not in args, args
+        assert not [a for a in args if a.startswith(tuple(f"{k}=" for k in CP_ENV))], args
+
+
+def _ncalls(mock_state_file) -> int:
+    return len(json.loads(Path(mock_state_file).read_text()).get("calls", []))
+
+
+def _skill(tmp: Path, *parts) -> str:
+    return str(tmp / ".agents" / "skills" / Path(*parts))
+
+
+def _create(tmp: Path, ws: Path, env=None) -> subprocess.CompletedProcess:
+    return subprocess.run(["bash", _skill(tmp, "multi-agent-mux-create", "scripts", "create_session.sh"),
+                           "--workspace", str(ws), "--agent", "claude", "--role", "creator", "--no-onboard"],
+                          capture_output=True, text=True, cwd=str(tmp), env=env or dict(os.environ), timeout=180)
+
+
+def _mutate(tmp: Path, code: str) -> None:
+    res = subprocess.run(["bash", "-c", f"source {_skill(tmp, 'lib.sh')} && atomic_dump_yaml {tmp / '.mam' / 'agent-sessions.yaml'}"],
+                         input=code, capture_output=True, text=True)
+    assert res.returncode == 0, res.stderr
+
+
+def test_t3_0_control_plane_module(tmp_path, monkeypatch):
+    from lib_py import control_plane as cp
+    for k in CP_ENV:
+        monkeypatch.delenv(k, raising=False)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
+    d = cp.describe(str(tmp_path))
+    assert d["forward"] is False and d["source"] == "env" and cp.prefix(d) == ""
+    d = cp.describe(str(tmp_path / "wt"))
+    assert d["forward"] is True
+    assert cp.prefix(d).startswith(f"WORKSPACE_ROOT={os.path.realpath(tmp_path)} ")
+    row = _binding(tmp_path / "other")
+    d = cp.describe(str(tmp_path / "wt"), row)
+    assert d["source"] == "row" and d["workspace_root"] == row["workspace_root"]
+    monkeypatch.setenv("MAM_JOBS_DIR", str(tmp_path / "a b"))
+    d = cp.describe(str(tmp_path))
+    assert d["forward"] and d["whitespace"]
+    assert cp.main(["prefix", json.dumps(d)]) == 3
+
+
+def test_t3_1_create_persists_and_forwards_a_foreign_binding(mam_sandbox, mock_herdr, mock_agents):
+    tmp = mam_sandbox
+    wt = tmp / "wt"
+    wt.mkdir()
+    res = _create(tmp, wt)
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(tmp)
+    assert row["control_plane"] == _binding(tmp)
+    flags = _env_flags(mock_herdr)
+    for k, v in zip(CP_ENV, _binding(tmp).values()):
+        assert f"{k}={v}" in flags, flags
+    assert "WORKSPACE_ROOT=" not in row["pane"]["cmd_full"]
+    assert "WORKSPACE_ROOT=" not in row["start_command"]
+    _assert_no_env_in_agent_args(mock_herdr)
+
+
+def test_t3_2_default_binding_is_recorded_not_forwarded(mam_sandbox, mock_herdr, mock_agents):
+    tmp = mam_sandbox
+    res = _create(tmp, tmp)
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(tmp)
+    assert row["control_plane"] == _binding(tmp)
+    assert not [f for f in _env_flags(mock_herdr) if f.startswith(CP_ENV)], "no forward for the default binding"
+
+
+def test_t3_5_whitespace_binding_that_needs_forwarding_is_refused(mam_sandbox, mock_herdr, mock_agents):
+    tmp = mam_sandbox
+    env = {**os.environ, "MAM_JOBS_DIR": str(tmp / "jobs dir")}
+    res = _create(tmp, tmp, env)
+    assert res.returncode == 2, res.stdout + res.stderr
+    assert "control-plane path contains whitespace" in res.stderr
+    assert _rows(tmp) == []
+
+
+def test_t3_3_4_stop_keeps_the_binding_and_resume_reforwards_it(mam_sandbox, mock_herdr, mock_agents):
+    tmp = mam_sandbox
+    wt = tmp / "wt"
+    wt.mkdir()
+    assert _create(tmp, wt).returncode == 0
+    (row,) = _rows(tmp)
+    name = row["name"]
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-stop", "scripts", "stop_session.sh"),
+                          "--session", name, "--agent", "claude"], capture_output=True, text=True, timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(tmp)
+    assert row["control_plane"] == _binding(tmp), "stop edits the row in place"
+
+    env = {k: v for k, v in os.environ.items() if k not in ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")}
+    since = _ncalls(mock_herdr)
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-resume", "scripts", "resume_session.sh"),
+                          "--workspace", str(wt), "--agent", "claude", "--session", name],
+                         capture_output=True, text=True, env=env, timeout=180)
+    assert res.returncode == 0, res.stdout + res.stderr
+    # The stopped session left an idle pane behind. A binding that must be
+    # forwarded is never put in front of `agent start --` (herdr passes those
+    # tokens to the agent as arguments, review R1): resume takes a new pane,
+    # whose creation call carries the binding as --env.
+    flags = _env_flags(mock_herdr, since)
+    for k, v in zip(CP_ENV, _binding(tmp).values()):
+        assert f"{k}={v}" in flags, (res.stderr[-1500:], flags)
+    _assert_no_env_in_agent_args(mock_herdr, since)
+    (row,) = _rows(tmp)
+    assert row["control_plane"] == _binding(tmp)
+
+
+def test_t3_4b_stop_finds_the_job_in_the_rows_registry(mam_sandbox, mock_herdr, mock_agents, monkeypatch):
+    from lib_py.delegation import registry
+    tmp = mam_sandbox
+    ctl_jobs = tmp / "ctl" / "jobs"
+    jid = registry.register_job("t3-4", registry_dir=str(ctl_jobs), agent_session="herdr:t34-creator-claude")
+    binding = {**_binding(tmp), "mam_jobs_dir": os.path.realpath(ctl_jobs)}
+    _mutate(tmp, f"""
+d['herdr_sessions'] = [{{
+    'name': 't34-creator-claude', 'status': 'running', 'role': 'creator',
+    'delegate_job_id': {jid!r}, 'pane': {{'cwd': {str(tmp)!r}, 'cmd': 'claude'}},
+    'control_plane': {binding!r},
+}}]
+""")
+    env = {k: v for k, v in os.environ.items() if k != "MAM_JOBS_DIR"}
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-stop", "scripts", "stop_session.sh"),
+                          "--session", "t34-creator-claude", "--agent", "claude"],
+                         capture_output=True, text=True, env=env, timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert (ctl_jobs / f"{jid}.events.log").exists(), "stop published into the row's registry"
+    assert not (tmp / ".mam" / "jobs" / f"{jid}.json").exists()
+
+
+def test_t3_6_status_reads_mam_jobs_dir_and_the_row_binding(mam_sandbox, monkeypatch):
+    tmp = mam_sandbox
+    x, y = tmp.parent / f"x-{tmp.name}", tmp / "row-jobs"
+    for d, jid in ((x, "job21x"), (y, "job21y")):
+        d.mkdir(parents=True, exist_ok=True)
+        (d / f"{jid}.json").write_text('{"status": "completed"}')
+    status_sh = _skill(tmp, "multi-agent-mux-status", "scripts", "status.sh")
+    try:
+        _mutate(tmp, f"""
+d['herdr_sessions'] = [{{'name': 's21-creator-claude', 'status': 'running', 'delegate_job_id': 'job21x',
+                         'pane': {{'cwd': {str(tmp)!r}}}}}]
+""")
+        env = {**os.environ, "MAM_JOBS_DIR": str(x)}
+        res = subprocess.run(["bash", status_sh, "--json"], capture_output=True, text=True, env=env, timeout=120)
+        assert res.returncode == 0, res.stderr
+        assert json.loads(res.stdout)["sessions_detail"][0]["job_status"] == "completed"
+
+        _mutate(tmp, f"""
+d['herdr_sessions'] = [{{'name': 's21-creator-claude', 'status': 'running', 'delegate_job_id': 'job21y',
+                         'pane': {{'cwd': {str(tmp)!r}}}, 'control_plane': {{'mam_jobs_dir': {str(y)!r}}}}}]
+""")
+        env = {k: v for k, v in os.environ.items() if k != "MAM_JOBS_DIR"}
+        res = subprocess.run(["bash", status_sh, "--json"], capture_output=True, text=True, env=env, timeout=120)
+        assert json.loads(res.stdout)["sessions_detail"][0]["job_status"] == "completed"
+    finally:
+        import shutil
+        shutil.rmtree(x, ignore_errors=True)
+
+
+def test_t3_7_reconcile_prefers_mam_jobs_dir_static_guard():
+    """Weak by design: the subscribe loop is a heredoc run only against a live
+    broker, so this pins the source text."""
+    src = (SKILLS / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh").read_text()
+    assert "if os.environ.get('MAM_JOBS_DIR'):\n            registry_dir = os.environ['MAM_JOBS_DIR']" in src
+    assert 'MAM_JOBS_DIR="${MAM_JOBS_DIR:-}"' in src
+
+
+def test_t3_8_legacy_row_without_binding_is_filled_on_resume(mam_sandbox, mock_herdr, mock_agents):
+    tmp = mam_sandbox
+    _mutate(tmp, f"""
+d['herdr_sessions'] = [{{
+    'name': 'legacy-creator-claude', 'status': 'stopped', 'role': 'creator',
+    'herdr_session': 'default', 'herdr_server': 'default',
+    'pane': {{'cwd': {str(tmp)!r}, 'cmd': 'claude'}},
+    'claude_session_id_own': '11111111-2222-4333-8444-555555555555',
+    'session_id_source': 'assigned', 'session_id_verified': False,
+}}]
+""")
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-resume", "scripts", "resume_session.sh"),
+                          "--workspace", str(tmp), "--agent", "claude", "--session", "legacy-creator-claude"],
+                         capture_output=True, text=True, timeout=180)
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(tmp)
+    assert row["control_plane"] == _binding(tmp)
+
+
+def test_t3_4c_stop_keeps_the_audit_trail_in_the_rows_log_root(mam_sandbox, mock_herdr, mock_agents, monkeypatch):
+    """Review R3: with no MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR in the caller,
+    stop cancels the job in the row's registry and writes its history to the
+    row's audit-log root - not to $WORKSPACE_ROOT/.mam/delegate_job_logs."""
+    from lib_py.delegation import registry
+    tmp = mam_sandbox
+    jobs, logs = tmp / "custom-jobs", tmp / "custom-logs"
+    # the job was registered under that binding (its `registered` audit entry lives there)
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(logs))
+    jid = registry.register_job("t3-4c", registry_dir=str(jobs), agent_session="herdr:logs-creator-claude")
+    binding = {**_binding(tmp), "mam_jobs_dir": str(jobs), "delegate_job_logs_dir": str(logs)}
+    _mutate(tmp, f"""
+d['herdr_sessions'] = [{{
+    'name': 'logs-creator-claude', 'status': 'running', 'role': 'creator',
+    'delegate_job_id': {jid!r}, 'pane': {{'cwd': {str(tmp)!r}, 'cmd': 'claude'}},
+    'control_plane': {binding!r},
+}}]
+""")
+    env = {k: v for k, v in os.environ.items() if k not in ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")}
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-stop", "scripts", "stop_session.sh"),
+                          "--session", "logs-creator-claude", "--agent", "claude"],
+                         capture_output=True, text=True, env=env, timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert json.loads((jobs / f"{jid}.json").read_text())["status"] == "cancelled"
+    assert not (tmp / ".mam" / "delegate_job_logs" / jid).exists(), "history misrouted to the default log root"
+    audit = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+    assert {"progress", "cancelled"} <= {e.get("source_event") for e in audit}, audit
+    assert any(e.get("event") == "status_changed" for e in audit), audit
+
+
+def test_t2_15_relative_mam_jobs_dir_means_the_same_dir_to_runner_and_wrapper(tmp_path, monkeypatch):
+    """Review R2: the wrapper runs with cwd=workspace_root and absolutises a
+    relative registry there; the runner must look in that same directory."""
+    from lib_py.loop import runner as lr
+    ctl, caller = tmp_path / "control", tmp_path / "caller"
+    ctl.mkdir()
+    caller.mkdir()
+    monkeypatch.chdir(caller)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(ctl))
+    monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
+    monkeypatch.setenv("MAM_JOBS_DIR", ".mam/jobs")
+    r = _runner()
+    jobs = Path(os.path.realpath(ctl)) / ".mam" / "jobs"
+    assert r.jobs_dir == jobs and r.jobs_dir.is_absolute()
+    report = jobs / "j21" / "reviewer-reports" / "report-final.md"
+    report.parent.mkdir(parents=True)
+    report.write_text("[VERDICT: PASS]\n")
+    (jobs / "j21.json").write_text(json.dumps({"status": "completed"}))
+    assert r.find_report("j21") == report
+    assert r.wait_for_job("j21", max_wait=2) is True
+    seen = []
+
+    def fake_run(cmd, **kw):
+        seen.append(cmd)
+
+        class P:
+            returncode = 0
+            stdout = "registered job: j22\n"
+            stderr = ""
+        return P()
+
+    monkeypatch.setattr(lr.subprocess, "run", fake_run)
+    r.delegate_job("creator-claude-01", "claude", "Worker", "go")
+    assert seen[0][seen[0].index("--registry-dir") + 1] == str(jobs), "the wrapper is told the same absolute dir"
+    assert not (caller / ".mam").exists()