test(issue-20): state isolation regression tests

S1/S2 per-test defaults and the mam_sandbox override, S3 submit with WORKSPACE_ROOT on a checkout leaves its .mam alone, S6 a nested pytest started with a pane's binding sees none of it (import-time copy included) and leaves the target untouched, S7 lib.sh takes the session registry from a test's WORKSPACE_ROOT.

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

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

@@ -27,6 +27,10 @@ from typing import Any, Dict, List, Optional, Set, Tuple
 
 from lib_py.loop import checkpoint, eager, parser
 
+UNAVAILABLE_MARKER = "!!! CHANGE SET UNAVAILABLE !!!"  # diff_collect.sh MAM_DIFF_UNAVAILABLE_MARKER
+SELF_REVIEW_PROMPT = ("작업 완료 상태에 대해 스스로 검증(Self-Review)하여 결함이 없음을 확인하고 종결해주세요. "
+                      "리뷰 리포트 마지막에 단독 행으로 반드시 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 명시해주세요.")
+
 # ANSI colors matching run_loop.sh
 _COLOR_INFO = "\033[1;34m[*]\033[0m"
 _COLOR_SUCCESS = "\033[1;32m[✓]\033[0m"
@@ -531,6 +535,11 @@ class LoopRunner:
             res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, cwd=src)
             if res.returncode == 0:
                 return res.stdout
+            if head_commit:
+                # the snapshot's evidence is not replaced by anything else (B13):
+                # the collector's UNAVAILABLE output is passed on as it is
+                return res.stdout if UNAVAILABLE_MARKER in res.stdout else (
+                    f"{UNAVAILABLE_MARKER}\nreason: diff collector failed (rc {res.returncode})\n")
         # Fallback to direct git diff
         try:
             if head_commit:
@@ -538,6 +547,8 @@ class LoopRunner:
             else:
                 cmd = ["git", "diff", self.base_commit] if self.base_commit else ["git", "diff", "HEAD"]
             res = subprocess.run(cmd, capture_output=True, text=True, cwd=src)
+            if head_commit and res.returncode != 0:
+                return f"{UNAVAILABLE_MARKER}\nreason: git diff failed (rc {res.returncode})\n"
             return res.stdout
         except Exception:
             return ""
@@ -550,9 +561,11 @@ class LoopRunner:
     def _context_active(self) -> bool:
         return self.worktree_given or self.base_ref is not None
 
-    def _worktree_context(self, remote: bool = False) -> str:
+    def _worktree_context(self, remote: bool = False, purpose: str = "work") -> str:
         """The source-root block put in front of every delegated prompt (D6);
-        empty without --worktree/--base, so prompts stay byte-identical."""
+        empty without --worktree/--base, so prompts stay byte-identical.
+        purpose: 'work' (creator/planner), 'review' (a verdict job, whose
+        target is the [리뷰 identity] path) or 'adjudication' (a re-hearing)."""
         if not self._context_active():
             return ""
         src = self.worktree_root
@@ -570,8 +583,21 @@ class LoopRunner:
         # a remote reviewer (diff-only) cannot reach this host's paths (B6)
         src_line = ("- 소스 경로: (오케스트레이터 호스트 전용 — 원격 리뷰어는 접근할 수 없음)\n" if remote
                     else f"- 소스 경로: {src}\n")
-        tail = ("이 잡의 검증 대상은 아래 SHA 와 diff 다.\n\n" if remote else
-                "이 잡의 코드 작업·검증 대상은 위 소스 경로다. 브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n")
+        if purpose == "review" and remote:
+            tail = "이 잡의 검증 대상은 아래 SHA 와 diff 다.\n\n"
+        elif purpose == "review" and self.review_isolation == "worktree":
+            # one verification target per prompt (loop B review B-2)
+            tail = ("리뷰 검증은 아래 [리뷰 identity] 의 검증 경로에서만 한다. 소스 경로는 creator 의 작업 트리이며 "
+                    "리뷰어는 거기서 실행·수정하지 않는다. 브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n")
+        elif purpose == "review":
+            tail = ("리뷰 검증은 아래 [리뷰 identity] 의 검증 경로(= 위 소스 경로)에서 하며, 거기서 수정·커밋하지 않는다. "
+                    "브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n")
+        elif purpose == "adjudication" and remote:
+            tail = "이 잡은 아래 반론의 재심이다. 판정 근거는 직전 리뷰에서 받은 SHA·diff 와 아래 반론 본문이다.\n\n"
+        elif remote:
+            tail = "이 잡의 검증 대상은 아래 SHA 와 diff 다.\n\n"
+        else:
+            tail = "이 잡의 코드 작업·검증 대상은 위 소스 경로다. 브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n"
         return ("[워크트리 컨텍스트]\n"
                 + src_line
                 + f"- 브랜치: {branch}\n"
@@ -581,9 +607,9 @@ class LoopRunner:
                 + tail)
 
     def _dispatch(self, agent_session: str, agent_type: str, role: str, prompt: str,
-                  remote: bool = False) -> Optional[str]:
+                  remote: bool = False, purpose: str = "work") -> 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(remote) + prompt)
+        return self.delegate_job(agent_session, agent_type, role, self._worktree_context(remote, purpose) + prompt)
 
     def _change_set_state(self) -> str:
         """'empty' | 'present' | 'unknown' for the source root against base,
@@ -653,7 +679,17 @@ class LoopRunner:
 
     def _worktree_tree(self) -> Optional[str]:
         """The source worktree's state as a tree, through a temporary index
-        (the real index, status and refs are untouched); None on failure."""
+        (the real index, status and refs are untouched); None on failure.
+        A dirty submodule is a failure: the tree records only its gitlink, so
+        its uncommitted content would be neither reviewed nor pinned."""
+        dirty_subs = self._dirty_submodules()
+        if dirty_subs is None:
+            log_error(f"review snapshot: cannot check the submodules of {self.worktree_root}")
+            return None
+        if dirty_subs:
+            log_error(f"review snapshot: submodule(s) with uncommitted or untracked content: {' '.join(dirty_subs)}; "
+                      "the snapshot records only their commit - commit or clean them first")
+            return None
         fd, idx = tempfile.mkstemp(prefix="mam-review-index-")
         os.close(fd)
         os.unlink(idx)
@@ -673,6 +709,18 @@ class LoopRunner:
                 except OSError:
                     pass
 
+    def _dirty_submodules(self) -> Optional[List[str]]:
+        """Checked-out submodules (recursively) with uncommitted or untracked
+        content; None when the check itself fails. Each submodule is asked
+        directly with --ignore-submodules=none, so no submodule.*.ignore or
+        diff.ignoreSubmodules setting can hide it."""
+        script = ('s=$(git status --porcelain --ignore-submodules=none) || exit 1; '
+                  '[ -z "$s" ] || printf "%s\\n" "$displaypath"')
+        r = self._git(self.worktree_root, "submodule", "foreach", "--recursive", "--quiet", script)
+        if r.returncode != 0:
+            return None
+        return [line for line in r.stdout.splitlines() if line.strip()]
+
     def _gitlinks(self, treeish: str) -> Optional[Set[str]]:
         r = self._git(self.worktree_root, "ls-tree", "-r", "-z", treeish)
         if r.returncode != 0:
@@ -735,11 +783,18 @@ class LoopRunner:
     def _close_review_tree(self, path: Optional[str]) -> None:
         if not path or path not in self._review_paths:
             return
-        self._git(self.worktree_root, "worktree", "remove", "--force", path)
-        self._git(self.worktree_root, "worktree", "prune")
+        r = self._git(self.worktree_root, "worktree", "remove", "--force", path)
         if os.path.isdir(path):
             shutil.rmtree(path, ignore_errors=True)
+        if r.returncode != 0:
+            # only when remove failed: prune drops this entry (its directory is
+            # gone now) - and any other stale one, which is why it is not routine
+            self._git(self.worktree_root, "worktree", "prune")
         self._review_paths.remove(path)
+        try:
+            os.rmdir(os.path.dirname(path))  # TMPDIR/mam-review/<loop_id>, once empty
+        except OSError:
+            pass
 
     def _close_all_review_trees(self) -> None:
         for p in list(self._review_paths):
@@ -1003,6 +1058,13 @@ class LoopRunner:
 
         final_passed_reviewers: Dict[str, Tuple[str, Path]] = {}
         identity = self._context_active()
+        # Identity mode: a creator self-review is a verdict like any other, so it
+        # takes the reviewer path - snapshot diff, evidence check, B10 re-check
+        # and sidecar (loop B review B-R2/B-R3). A NOT PASS re-reviews, as before.
+        self_review = identity and not reviewers
+        if self_review:
+            log_warn("No reviewers specified. Conducting Creator Self-Review...")
+            reviewers = [self.creator]
 
         for loop_count in range(1, self.max_loop + 1):
             log_info(f"Review Loop Iteration {loop_count}/{self.max_loop}...")
@@ -1029,29 +1091,13 @@ class LoopRunner:
             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_remote = self.creator in self._remote_recipients
-                if identity:
-                    sr_prompt = self._identity_block(snap, review_path, sr_remote) + sr_prompt
-                sr_jid = self._dispatch(self.creator, creator_type, "Reviewer", sr_prompt, remote=sr_remote)
+                sr_prompt = SELF_REVIEW_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
                 self.wait_for_job(sr_jid)
                 sr_report = self.find_report(sr_jid)
-                if identity:
-                    if self.review_isolation == "worktree":
-                        self._close_review_tree(review_path)
-                    sr_stale = self._round_is_stale(snap)
-                    sr_verdict = "PASS" if (sr_report and parser.has_verdict(sr_report, "PASS")) else "NOT PASS"
-                    self._record_round(loop_count, snap, review_path, {self.creator: sr_jid},
-                                       {self.creator: sr_verdict}, sr_stale)
-                    if sr_stale:
-                        log_warn("source changed during review; the self-review is discarded, re-reviewing")
-                        if loop_count == self.max_loop:
-                            log_error("the source changed during review in the last iteration; no verdict to accept")
-                            return 1
-                        continue
                 if sr_report and parser.has_verdict(sr_report, "PASS") and not parser.has_verdict(sr_report, "NOT PASS"):
                     log_success("Self-Review PASS.")
                     break
@@ -1077,11 +1123,15 @@ class LoopRunner:
             for rev in reviewers:
                 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}"
+                if self_review:
+                    r_prompt = f"{SELF_REVIEW_PROMPT} 변경분:\n{changes_diff}"
+                else:
+                    r_prompt = f"다음 구현 사항(작업 목표: {self.task}) 및 누적 변경분(git diff)에 대해 린트, 동작성, 유실 등의 관점에서 교차 코드 리뷰를 수행해주세요. 확인 후 최종 Verdict로 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 리뷰 리포트 마지막에 단독 행으로 명시적으로 작성해주세요. 만약 단순 버그 수정으로는 부족하고 설계 변경/재작업 수준의 재계획이 필요하다고 판단되면, 리포트 아무 곳에나 단독 행으로 '[ESCALATE: PLANNER]' 태그도 함께 남겨주세요. 변경분:\n{changes_diff}"
                 r_remote = rev in self._remote_recipients
                 if identity:
                     r_prompt = self._identity_block(snap, review_path, r_remote) + r_prompt
-                r_jid = self._dispatch(rev, r_type, "Reviewer", r_prompt, remote=r_remote)
+                r_jid = self._dispatch(rev, r_type, "Reviewer", r_prompt, remote=r_remote,
+                                       purpose="review" if identity else "work")
                 if not r_jid:
                     log_error(f"Failed to register review job for '{rev}'.")
                     return 1
@@ -1143,6 +1193,13 @@ class LoopRunner:
                 log_success("All reviewers issued [VERDICT: PASS]. Loop completed successfully.")
                 break
 
+            if self_review:
+                log_warn("Self-Review NOT PASS.")
+                if loop_count == self.max_loop:
+                    log_error("Reached max loop count. Self-review loop aborted with failures.")
+                    return 1
+                continue
+
             if loop_count == self.max_loop:
                 log_error(f"Reached max loop count ({self.max_loop}). Review loop aborted with failures.")
                 return 1
@@ -1198,7 +1255,7 @@ class LoopRunner:
                         rt_type = self.resolve_agent_type(rt, state)
                         adj_prompt = f"작업자(Creator)가 귀하의 리뷰 지적 중 일부를 타당하지 않다고 판단하여 반론을 제기했습니다. 반론을 검토하고 재심(re-adjudication)하십시오. 반론이 타당하여 귀하의 지적을 철회한다면 리포트 마지막에 단독 행으로 '[ADJUDICATION: SUSTAINED]' 를, 반론이 타당하지 않아 지적을 유지한다면 '[ADJUDICATION: OVERRULED]' 를 명시하십시오. 반론 내용:\n{rebuttal_text}"
                         adj_jid = self._dispatch(rt, rt_type, "Reviewer", adj_prompt,
-                                                 remote=rt in self._remote_recipients)
+                                                 remote=rt in self._remote_recipients, purpose="adjudication")
                         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)

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

@@ -58,7 +58,9 @@ process probing happens for it). The row still needs `role` and
 `status: running` to be picked up by `--all-reviewer`. Remote rows need the
 default Python loop runner: the legacy shell runner (`MAM_LOOP_PY=0`)
 addresses every session as `herdr:<name>` and does not support
-`transport: remote`.
+`transport: remote`. With `--worktree`/`--base` (review identity), remote reviewers are
+refused by default - they cannot reach the review tree on this host; `--remote-review diff-only`
+admits them with the snapshot diff as their only evidence, recorded as `evidence: diff-only`.
 
 ## Roles & Responsibilities
 
@@ -198,9 +200,13 @@ sequenceDiagram
 | **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 에게 갑니다. |
+| **Review isolation** | `--review-isolation worktree\|env` | `--worktree`/`--base` 에서만 의미. `worktree`(기본): 라운드마다 스냅샷의 detached 리뷰 워크트리(TMPDIR)에서 검증합니다. `env`: 소스 루트에서 `PYTHONDONTWRITEBYTECODE=1 … -p no:cacheprovider --basetemp=<밖>` 지침으로 검증합니다. 리뷰 워크트리를 열 수 없으면 조용히 낮추지 않고 exit 1. `env` 에서 리뷰어가 ignore 되지 않는 산출물을 소스 루트에 남기면 라운드가 stale 로 판정돼 재리뷰가 반복됩니다(fail-closed) — 산출물은 `--basetemp` 등으로 소스 밖에 둡니다. |
+| **Remote reviewers** | `--remote-review refuse\|diff-only` | `--worktree`/`--base` 에서만 의미. 원격 리뷰어(`transport: remote`)는 이 호스트의 리뷰 트리에 닿을 수 없으므로 기본으로 **어떤 잡도 위임하기 전에** 거부(exit 1)합니다. `diff-only` 는 스냅샷 SHA 와 diff 만을 근거로 받아들이고 그렇게 기록합니다. |
 
 `--worktree` 또는 `--base` 가 있으면 매 리뷰 라운드(셀프 리뷰 포함) 전에 소스 루트의 변경분을 git 종료 코드로 판정합니다. 변경분이 비었거나 판정할 수 없으면 리뷰를 요청하지 않고 exit 1 로 끝납니다(fail-closed). 둘 다 없으면 기존 동작 그대로입니다.
 
+**리뷰 identity (Issue #21 항목 5).** `--worktree`/`--base` 가 있으면 매 리뷰 라운드를 `(loop_id, base, head, tree)` 에 묶습니다. 소스의 미커밋·untracked 상태까지 담은 스냅샷(ref 없는 커밋)을 만들고, 리뷰어가 받는 diff 도 그 스냅샷에서 만듭니다(`base..snapshot`). 라운드가 끝난 뒤 소스가 바뀌었으면 그 라운드의 PASS 는 버리고 수정 잡 없이 다시 리뷰합니다(마지막 반복이면 exit 1). 승격 직전에도 한 번 더 확인하며, 승격된 리포트 옆에 `report-<jid>.identity.json`(loop_id·base·head·tree·스냅샷 커밋·리뷰어·`transport`·`evidence`: `review-tree`/`source-tree`/`diff-only`)을 둡니다. 추적되지 않는 중첩 git 저장소는 스냅샷에 담을 수 없어 거부합니다. 서브모듈 안의 미커밋·untracked 내용도 스냅샷에는 gitlink 커밋만 남으므로 거부합니다(라운드 시작 시 exit 1, 리뷰 중에 생기면 그 라운드는 stale). `submodule.*.ignore` 설정으로 가려지지 않습니다. 리뷰어가 없으면 creator 셀프 리뷰가 같은 리뷰어 경로(스냅샷 diff·증거 검사·승격 직전 재확인·사이드카)를 거치고, NOT PASS 면 수정 잡 없이 다시 리뷰합니다. 리뷰 잡의 `[워크트리 컨텍스트]` 는 검증 대상을 `[리뷰 identity]` 의 검증 경로 하나로만 지시합니다.
+
 구현이 루프 밖(워크트리의 creator 에게 `delegate-job --type direct`)에서 끝났다면, **구현 위임 전에** base 를 기록해 두고 리뷰만 돌립니다:
 
 ```bash

docs/design/worktree-creator-model.md

@@ -33,7 +33,7 @@ MAM 은 **워크스페이스 = `.agents/skills` 가 들어 있는 체크아웃
 
 ### C. (제안) creator 만 워크트리, 리뷰어 풀은 공유, 제어 평면은 하나
 - **creator = 워크트리 전용 인스턴스.** 제품 코드를 편집·커밋하는 유일한 에이전트이므로 cwd 가 곧 이슈 브랜치여야 한다. 오케스트레이터 커밋과 루프의 리포트 승격(`runner.py` 의 Promotion 블록 → 제어 루트 `.agents/reports/`)은 별도 소유다. 이슈마다 `creator-<agent>-<issue>` 로 만들고 끝나면 정지(대화 보존, 필요 시 재개). 재개하려면 같은 워크트리 경로가 있어야 한다(대화 키가 cwd).
-- **리뷰어 풀 = 본 체크아웃 공유.** 리뷰어는 소스를 편집하지 않는다. 그러나 테스트 실행은 쓰기를 동반하므로 '읽기 전용'이 아니다. 검증 격리(head SHA detached 워크트리 또는 격리 옵션)와 (loop_id, base, head) 판정 결속은 항목 5 에서 정한다. 1단계 브리프는 러너가 경로·base·head·커밋 수를 자동 삽입한다(`--worktree`/`--base`). 리뷰가 병목이면 리뷰어 인스턴스를 추가한다.
+- **리뷰어 풀 = 본 체크아웃 공유.** 리뷰어는 소스를 편집하지 않는다. 그러나 테스트 실행은 쓰기를 동반하므로 '읽기 전용'이 아니다. `--worktree`/`--base` 에서는 러너가 매 라운드를 스냅샷 `(loop_id, base, head, tree)` 에 묶고(항목 5), 리뷰어는 그 스냅샷의 detached 리뷰 워크트리(기본) 또는 env 격리 지침으로 소스 루트에서 검증한다. 리뷰 중 소스가 바뀐 라운드의 승인은 버리고 다시 리뷰하며, 승격 직전에도 다시 확인한다. 원격 리뷰어는 리뷰 트리에 닿을 수 없어 기본 거부, `--remote-review diff-only` 로만 받는다. 리뷰가 병목이면 리뷰어 인스턴스를 추가한다.
 - **오케스트레이터 커밋 = 워크트리.** 본 체크아웃은 리뷰어의 작업 공간이므로 오케스트레이터는 그곳에서 브랜치를 바꾸지 않는다.
 - **제어 평면 = 본 체크아웃의 `.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` 는 워크트리다.
 
@@ -100,9 +100,9 @@ bash "$MAIN/.agents/skills/multi-agent-mux-stop/scripts/worktree_remove.sh" --pa
 | 2 | 루프 러너 `--worktree` / `--base` / `--review-only`, 빈 변경분 fail-closed 가드, 러너의 `MAM_JOBS_DIR` 존중 | 완료(v4.6.0) |
 | 3 | 제어 평면 바인딩 영속(`control_plane`)·명시 전파(create/resume), stop·status·reconcile 의 `MAM_JOBS_DIR`·행 바인딩 존중 | 완료(v4.6.0). 루프 A 에서 바인딩 계산 실패 처리·조기 종료 경로 영속 보강 |
 | 4 | 워크트리 준비·해체 헬퍼, resume 을 원래 cwd 로만 허용, create 의 reconcile/watchdog 기동 위치, `MAM_NO_SUBSCRIBER_DAEMON`(prepare 가 설정). reconcile 의 루트 밖 cwd 재발견 거부는 제한으로 남김 | 루프 A (이 브랜치) |
-| 5 | 리뷰 identity: (loop_id, base_sha, head_sha) 판정 기록·무효화, 리뷰어 검증 격리(detached 워크트리) | 대기 |
+| 5 | 리뷰 identity: (loop_id, base_sha, head_sha) 판정 기록·무효화, 리뷰어 검증 격리(detached 워크트리) | 루프 B (이 브랜치): 스냅샷 identity·스냅샷 diff·stale 재리뷰·승격 직전 재확인·`identity.json`(evidence), 원격 리뷰어 거부/diff-only. 리뷰 후속: dirty 서브모듈 거부(스냅샷·stale 검사), identity 모드 셀프 리뷰도 리뷰어 경로로 기록·승격, 리뷰 프롬프트의 검증 대상 단일화 |
 | 6 | 이 설계 문서 정정 | 완료(v4.6.0), 루프 A 에서 갱신 |
-| 7 | 수용 사례 고정(실제 herdr 의 pane env 전파 포함) | 대기 |
+| 7 | 수용 사례 고정(실제 herdr 의 pane env 전파 포함) | 루프 B (이 브랜치): 수용 사례 8건 테스트 고정(`tests/test_issue21_acceptance.py`, A1~A8). pane env 전파는 mock herdr 의 `--env` 호출 단언까지이며, 실제 herdr pane 안의 값 확인과 시험 운영은 단계 C |
 | 8 | 다중 루프 네임스페이스·리뷰어 스케줄러 | 보류 |
 
 ## 6. 리뷰어에게 묻는 것
@@ -129,6 +129,8 @@ bash "$MAIN/.agents/skills/multi-agent-mux-stop/scripts/worktree_remove.sh" --pa
 | status | `MAM_JOBS_DIR`·행 `mam_jobs_dir`·감사 로그(`DELEGATE_JOB_LOGS_DIR`·행 `delegate_job_logs_dir`)·기존 후보 순 조회 | 행 cwd 로 대화 키 | — |
 | monitor/reconcile | `MAM_JOBS_DIR` 우선(정적 가드로만 검증) | 기존 행은 행 cwd 로 검증 | 루트 밖 cwd 의 미등록 세션은 자동 등록 안 됨 → create/prepare 로만 등록 (남은 제한) |
 | 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`)는 새 옵션 거부 |
+| loop | `MAM_JOBS_DIR` 존중, 래퍼에 `--registry-dir` 명시. `--worktree` 로 소스 루트 분리 | diff·base·head 만 소스 루트. **구현이 루프 밖이면 `--base <구현 전 sha> --review-only` 필수**. `--worktree`/`--base` 가 있으면 빈 변경분 거부 | 루프 하나(8). 원격 리뷰어는 identity 모드에서 거부(기본) 또는 diff-only(`--remote-review diff-only`); 원격 리뷰어가 트리를 검증하게 하는 스냅샷 공유(공유 원격의 `refs/mam/review/<loop_id>/<n>`)는 범위 밖. 레거시 셸 루프(`MAM_LOOP_PY=0`)는 새 옵션 거부 |
 
-실제 herdr 가 `pane split --env` / `workspace create --env` 로 pane 환경을 넘기는지는 1단계에서 mock 수준으로만 검증했다. 실제 전파는 항목 7 의 수용 사례로 고정한다.
+실제 herdr 가 `pane split --env` / `workspace create --env` 로 pane 환경을 넘기는지는 mock 수준으로만 검증했다. 수용 사례 A2 는 러너가 바인딩 4개를 `--env` 로 싣고 `agent start --` 뒤에는 싣지 않음을 고정한다. 실제 herdr pane 안에서 값이 보이는지는 단계 C 시험 운영에서 확인한다.
+
+**호환성 메모 (루프 B)**: 1단계부터 있던 `--worktree`/`--base` + 원격 리뷰어 조합은 이전에는 원격 리뷰어가 도달할 수 없는 로컬 경로를 프롬프트에 담아 진행했다. 이제는 어떤 잡도 위임하기 전에 기본으로 거부하며, `--remote-review diff-only` 로 opt-in 할 수 있다. 옵션 없는 루프는 그대로다. 스냅샷 커밋은 ref 가 없어 gc 가 정리하고(기본 2주), 리뷰 워크트리는 라운드마다 지운다.

tests/_mam_state_guard.py

@@ -0,0 +1,157 @@
+"""Issue #20 - helpers for the session guard over real control planes.
+
+A pytest process can reach two kinds of `.mam`: this checkout's own, and the
+one the agent pane's control-plane binding named before conftest's
+pytest_configure dropped it (a worktree creator's pane carries
+WORKSPACE_ROOT, AGENT_SESSIONS_YAML, MAM_JOBS_DIR and DELEGATE_JOB_LOGS_DIR,
+all at the live main checkout). The session guard snapshots their job
+registries and audit-log roots and, at the end, reports the new entries no
+live delegation explains. Pure functions, unit-tested in
+test_issue20_state_isolation.py (S4, S5); the name does not start with
+`test_`, so pytest never collects it.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+
+SESSION_PREFIXES = ("herdr:", "remote:")
+# Top-level items of `herdr_sessions:` as atomic_dump_yaml writes them
+# (`- name: <session>` at column 0); indented `- name:` lines belong to nested
+# lists such as mcp_attachments. A regex keeps the guard free of PyYAML.
+_SESSION_LINE = re.compile(r"""^- name:\s*['"]?([^'"\s]+)""")
+
+
+def _listdir(path):
+    try:
+        return os.listdir(path)
+    except OSError:
+        return []
+
+
+def _load(path):
+    try:
+        with open(path, encoding="utf-8") as fh:
+            data = json.load(fh)
+    except (OSError, ValueError):
+        return {}
+    return data if isinstance(data, dict) else {}
+
+
+def watch_roots(repo_root, recorded):
+    """`.mam` directories to watch, realpath-deduplicated, in order: this
+    checkout's own (always, so a leak that creates it is still seen), then
+    those the recorded binding names - `$WORKSPACE_ROOT/.mam` and the parents
+    of MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR - when they exist."""
+    roots = [os.path.realpath(os.path.join(str(repo_root), ".mam"))]
+    candidates = []
+    if recorded.get("WORKSPACE_ROOT"):
+        candidates.append(os.path.join(recorded["WORKSPACE_ROOT"], ".mam"))
+    for var in ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR"):
+        if recorded.get(var):
+            candidates.append(os.path.dirname(os.path.abspath(recorded[var])))
+    for path in candidates:
+        real = os.path.realpath(path)
+        if os.path.isdir(real) and real not in roots:
+            roots.append(real)
+    return roots
+
+
+def state_dirs(roots, recorded):
+    """(registry dirs, audit-log dirs): `<root>/jobs` and
+    `<root>/delegate_job_logs` of every root, plus a recorded MAM_JOBS_DIR /
+    DELEGATE_JOB_LOGS_DIR that is not one of those."""
+    jobs = [os.path.join(r, "jobs") for r in roots]
+    logs = [os.path.join(r, "delegate_job_logs") for r in roots]
+    for var, dirs in (("MAM_JOBS_DIR", jobs), ("DELEGATE_JOB_LOGS_DIR", logs)):
+        if recorded.get(var):
+            real = os.path.realpath(recorded[var])
+            if os.path.isdir(real) and real not in dirs:
+                dirs.append(real)
+    return jobs, logs
+
+
+def snapshot(roots, recorded):
+    """Registry records (`*.json`) and audit-log entries per watched dir; a
+    missing dir counts as empty."""
+    jobs, logs = state_dirs(roots, recorded)
+    return {
+        "jobs": {d: {n for n in _listdir(d) if n.endswith(".json")} for d in jobs},
+        "logs": {d: set(_listdir(d)) for d in logs},
+    }
+
+
+def session_names(yaml_path):
+    """Session names registered in an agent-sessions.yaml; empty if unreadable."""
+    try:
+        with open(yaml_path, encoding="utf-8", errors="replace") as fh:
+            return {m.group(1) for m in map(_SESSION_LINE.match, fh) if m}
+    except OSError:
+        return set()
+
+
+def _bare(session):
+    for prefix in SESSION_PREFIXES:
+        if session.startswith(prefix):
+            return session[len(prefix):]
+    return session
+
+
+def leaks(roots, before, recorded):
+    """New entries since `before` that no live delegation explains.
+
+    A real delegation writes its registry record before its audit log and
+    targets a registered session (registry.register_job), so a new entry is a
+    leak when
+      (a) it is an audit log whose job has no record in any watched registry, or
+      (b) its record's agent_session is missing or names no session in the
+          watched agent-sessions.yaml files.
+    Returns [{"path", "reason", "agent_session", "prompt"}] sorted by path."""
+    now = snapshot(roots, recorded)
+    yamls = [os.path.join(r, "agent-sessions.yaml") for r in roots]
+    if recorded.get("AGENT_SESSIONS_YAML"):
+        yamls.append(recorded["AGENT_SESSIONS_YAML"])
+    known = set()
+    for path in yamls:
+        known |= session_names(path)
+    records = {}
+    for d, names in now["jobs"].items():
+        for name in names:
+            records.setdefault(name[:-len(".json")], os.path.join(d, name))
+
+    found = []
+
+    def judge(path, meta, reason=None):
+        session = meta.get("agent_session")
+        if reason is None:
+            if not session:
+                reason = "no agent_session"
+            elif _bare(session) not in known:
+                reason = "agent_session names no registered session"
+            else:
+                return
+        found.append({"path": path, "reason": reason, "agent_session": session,
+                      "prompt": str(meta.get("prompt", ""))[:60]})
+
+    for d, names in now["jobs"].items():
+        for name in names - before["jobs"].get(d, set()):
+            path = os.path.join(d, name)
+            judge(path, _load(path))
+    for d, names in now["logs"].items():
+        for name in names - before["logs"].get(d, set()):
+            path = os.path.join(d, name)
+            if name in records:
+                judge(path, _load(records[name]))
+            else:
+                judge(path, _load(os.path.join(path, "meta.json")), "audit log without a registry record")
+    return sorted(found, key=lambda f: f["path"])
+
+
+def describe(found):
+    """Failure message for the session guard."""
+    lines = ["Issue #20: {} entr{} made during this session in a real control plane "
+             "(left in place; check and remove by hand):".format(len(found), "y" if len(found) == 1 else "ies")]
+    lines += ["  {path}  [{reason}]  agent_session={agent_session!r}  prompt={prompt!r}".format(**f)
+              for f in found]
+    return "\n".join(lines)

tests/conftest.py

@@ -109,6 +109,69 @@ def _pid_alive(pid):
     return bool(state) and not state.startswith("Z")
 
 
+# ==============================================================================
+# Issue #20: test state stays out of every real control plane.
+#
+# An agent pane can carry a control-plane binding: a worktree creator's pane
+# gets WORKSPACE_ROOT, AGENT_SESSIONS_YAML, MAM_JOBS_DIR and
+# DELEGATE_JOB_LOGS_DIR pointing at the live main checkout (create/resume
+# forward the row's control_plane), and each of them wins over a test's own
+# WORKSPACE_ROOT (mqtt_common.default_registry_dir/get_logs_dir, lib.sh's
+# AGENT_SESSIONS_YAML). pytest_configure records the binding and drops it
+# before collection - a session fixture is too late for a module that copies
+# os.environ at import (test_orc_onboard.SCRUBBED_ENV). WORKSPACE_ROOT is
+# dropped only when it names another checkout; naming this one, it is what
+# lib.sh derives anyway. Every test then defaults to a registry and an
+# audit-log root under its own tmp_path, and a session guard (next to the #19
+# one) reports new entries in the recorded control planes.
+# ==============================================================================
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+_TESTS_DIR = str(Path(__file__).resolve().parent)
+if _TESTS_DIR not in sys.path:
+    sys.path.insert(0, _TESTS_DIR)
+import _mam_state_guard  # noqa: E402
+
+_BINDING_VARS = ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
+_RECORDED_BINDING = {}
+_DROPPED_BINDING = pytest.StashKey[dict]()
+
+
+def pytest_configure(config):
+    _RECORDED_BINDING.clear()
+    _RECORDED_BINDING.update({k: os.environ[k] for k in _BINDING_VARS if os.environ.get(k)})
+    dropped = {k: os.environ.pop(k) for k in ("AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
+               if k in os.environ}
+    ws = os.environ.get("WORKSPACE_ROOT")
+    if ws and os.path.realpath(ws) != str(_REPO_ROOT):
+        dropped["WORKSPACE_ROOT"] = os.environ.pop("WORKSPACE_ROOT")
+    config.stash[_DROPPED_BINDING] = dropped
+
+
+def pytest_unconfigure(config):
+    # Back for callers that run pytest.main() in-process.
+    os.environ.update(config.stash.get(_DROPPED_BINDING, {}))
+
+
+@pytest.fixture(scope="session")
+def _ambient_control_plane():
+    """The binding pytest_configure recorded (and dropped), as a copy."""
+    return dict(_RECORDED_BINDING)
+
+
+@pytest.fixture(autouse=True)
+def _isolate_mam_state_dirs(_ambient_control_plane, tmp_path, monkeypatch):
+    """Default registry and audit-log roots under this test's tmp_path; only
+    the env is set, nothing is created. mam_sandbox and a test's own
+    setenv/delenv run after this autouse fixture and win, and the shared
+    function-scope monkeypatch undoes them in LIFO order. A hidden directory,
+    so tests that use tmp_path/"jobs" as their own registry or list tmp_path
+    are unaffected. Takes _ambient_control_plane to state the order, not to
+    rely on it."""
+    state = tmp_path / ".mam-test-state"
+    monkeypatch.setenv("MAM_JOBS_DIR", str(state / "jobs"))
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(state / "delegate_job_logs"))
+
+
 @pytest.fixture
 def mam_sandbox(tmp_path, monkeypatch, request):
     """
@@ -1381,10 +1444,27 @@ def _drop_ambient_mam_version():
             os.environ["MAM_VERSION"] = previous
 
 
+@pytest.fixture(scope="session", autouse=True)
+def mam_state_leak_guard(_ambient_control_plane):
+    """Issue #20: no test may leave a record in a real `.mam/jobs` or an audit
+    log in a real `.mam/delegate_job_logs` - this checkout's, or the control
+    plane the ambient binding named before pytest_configure dropped it.
+    Delegations the live system runs meanwhile are told apart by
+    _mam_state_guard.leaks (record first, registered session). The per-test
+    defaults are the fix, this is the regression alarm; it reports and never
+    deletes - these are real control planes. The session registry yaml/db is
+    not hashed: reconcile and create write it during a run."""
+    roots = _mam_state_guard.watch_roots(_REPO_ROOT, _ambient_control_plane)
+    before = _mam_state_guard.snapshot(roots, _ambient_control_plane)
+    yield
+    found = _mam_state_guard.leaks(roots, before, _ambient_control_plane)
+    assert not found, _mam_state_guard.describe(found)
+
+
 @pytest.fixture(scope="session", autouse=True)
 def subscriber_daemon_leak_guard(tmp_path_factory):
     """Issue #19: no `reconcile.sh --subscribe` daemon anchored in this session's
-    pytest basetemp may survive the suite. Mirrors the #15 registry guard: the
+    pytest basetemp may survive the suite. Mirrors the #20 state guard: the
     per-sandbox finalizer is the fix, this is the regression alarm. Leftovers
     are reaped after being reported so a failing run still cleans up."""
     yield

tests/test_issue14_delivery_ack.py

@@ -517,15 +517,19 @@ def test_wrapper_idle_grace_never_outlives_the_wall_clock(tmp_path):
 # publish_event: a heartbeat reaches events.log before the wire publish
 # ---------------------------------------------------------------------------
 def _job_for_inprocess_publish(tmp_path, monkeypatch):
-    """A registered job in a private registry. The audit log resolves against
-    the cwd, so it lands under tmp_path too, and no real .mam.env is read."""
-    from lib_py.delegation import registry
+    """A registered job in a private registry. Its audit log goes to the
+    per-test default under tmp_path (conftest, Issue #20): deleting
+    DELEGATE_JOB_LOGS_DIR here, to fall back to the cwd, fell through to an
+    ambient WORKSPACE_ROOT instead - the real .mam/delegate_job_logs (PR #22
+    rec 4). No real .mam.env is read."""
+    from lib_py.delegation import mqtt_common, registry
 
     monkeypatch.chdir(tmp_path)
-    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
     monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
     reg_dir = tmp_path / "jobs"
     jid = registry.register_job("issue 14 heartbeat order", registry_dir=str(reg_dir))
+    logs = Path(mqtt_common.get_logs_dir())
+    assert tmp_path in logs.parents and (logs / jid).is_dir(), logs
     return reg_dir, jid
 
 

tests/test_issue15_registry_isolation.py

@@ -16,12 +16,13 @@ import json
 import os
 import subprocess
 import sys
+import uuid
 from pathlib import Path
 
+import _mam_state_guard
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 SKILLS = REPO_ROOT / ".agents" / "skills"
-REAL_JOBS = REPO_ROOT / ".mam" / "jobs"
-REAL_LOGS = REPO_ROOT / ".mam" / "delegate_job_logs"
 
 sys.path.insert(0, str(SKILLS))
 from lib_py.delegation import mqtt_common  # noqa: E402
@@ -67,8 +68,10 @@ def test_logs_dir_follows_the_same_order(monkeypatch, tmp_path):
 
 def test_cli_defaults_are_resolved_per_process(monkeypatch, tmp_path):
     """registry.py / publish_event.py build their argparse defaults at call
-    time, so a process started with WORKSPACE_ROOT set registers there."""
+    time, so a process started with WORKSPACE_ROOT set registers there - and
+    seeds its audit log there too."""
     monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
+    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
     env = dict(os.environ, WORKSPACE_ROOT=str(tmp_path), PYTHONPATH=f"{SKILLS}:{os.environ.get('PYTHONPATH', '')}")
     env.pop("MAM_JOBS_DIR", None)
     res = subprocess.run(
@@ -80,6 +83,7 @@ def test_cli_defaults_are_resolved_per_process(monkeypatch, tmp_path):
     records = list((tmp_path / ".mam" / "jobs").glob("*.json"))
     assert len(records) == 1, records
     assert json.loads(records[0].read_text())["agent_session"] == "herdr:issue15-probe"
+    assert (tmp_path / ".mam" / "delegate_job_logs" / records[0].stem).is_dir()
 
 
 # ---------------------------------------------------------------------------
@@ -95,26 +99,65 @@ def _create_session_from_repo_root(sandbox: Path, session: str) -> subprocess.Co
     )
 
 
-def _assert_registered_in_sandbox_only(sandbox: Path, session: str, before_jobs: set, before_logs: set, res):
+def _real_state_dirs(recorded: dict) -> list:
+    """Every real registry and audit-log root this run could leak into: this
+    checkout's .mam and the control plane the ambient binding named (conftest
+    records it before dropping it)."""
+    jobs, logs = _mam_state_guard.state_dirs(_mam_state_guard.watch_roots(REPO_ROOT, recorded), recorded)
+    return jobs + logs
+
+
+def _entries(dirs: list) -> set:
+    return {(d, name) for d in dirs for name in _listing(Path(d))}
+
+
+def _unique(prefix: str) -> str:
+    """A session name no concurrent run shares (Issue #20)."""
+    return f"{prefix}-{uuid.uuid4().hex[:8]}"
+
+
+def _assert_registered_in_sandbox_only(sandbox: Path, session: str, real_dirs: list, before: set, res):
+    """Judged on this test's own job ids and session only (Issue #20): live
+    delegations and other test runs may add entries to the real .mam
+    meanwhile, so comparing whole listings gave false failures."""
     assert res.returncode == 0, res.stderr + res.stdout
-    assert _listing(REAL_JOBS) == before_jobs, "onboarding job leaked into the repo's .mam/jobs"
-    assert _listing(REAL_LOGS) == before_logs, "audit log leaked into the repo's .mam/delegate_job_logs"
-    records = [json.loads(p.read_text()) for p in (sandbox / ".mam" / "jobs").glob("*.json")]
-    assert any(r.get("agent_session") == f"herdr:{session}" for r in records), records
+    records = {p.stem: json.loads(p.read_text()) for p in (sandbox / ".mam" / "jobs").glob("*.json")}
+    own = [jid for jid, r in records.items() if r.get("agent_session") == f"herdr:{session}"]
+    assert own, records
+    for d in real_dirs:
+        for jid in own:
+            assert not (Path(d) / f"{jid}.json").exists(), f"onboarding job {jid} leaked into {d}"
+            assert not (Path(d) / jid).exists(), f"audit log {jid} leaked into {d}"
+    # A job that never reached the sandbox has an id we do not know: look for
+    # the session in whatever appeared in the real roots meanwhile.
+    for d, name in _entries(real_dirs) - before:
+        entry = Path(d) / name
+        meta = entry if name.endswith(".json") else entry / "meta.json"
+        try:
+            session_of = json.loads(meta.read_text()).get("agent_session")
+        except (OSError, ValueError):
+            continue
+        assert session_of != f"herdr:{session}", f"onboarding job for {session} leaked: {entry}"
 
 
-def test_onboarding_job_stays_in_sandbox_with_explicit_env(mam_sandbox, mock_herdr, mock_agents):
-    before_jobs, before_logs = _listing(REAL_JOBS), _listing(REAL_LOGS)
-    res = _create_session_from_repo_root(mam_sandbox, "issue15-explicit-creator-claude")
-    _assert_registered_in_sandbox_only(mam_sandbox, "issue15-explicit-creator-claude", before_jobs, before_logs, res)
+def test_onboarding_job_stays_in_sandbox_with_explicit_env(mam_sandbox, mock_herdr, mock_agents,
+                                                           _ambient_control_plane):
+    session = _unique("issue15-explicit-creator-claude")
+    real_dirs = _real_state_dirs(_ambient_control_plane)
+    before = _entries(real_dirs)
+    res = _create_session_from_repo_root(mam_sandbox, session)
+    _assert_registered_in_sandbox_only(mam_sandbox, session, real_dirs, before, res)
 
 
-def test_onboarding_job_follows_workspace_root_alone(mam_sandbox, mock_herdr, mock_agents, monkeypatch):
+def test_onboarding_job_follows_workspace_root_alone(mam_sandbox, mock_herdr, mock_agents, monkeypatch,
+                                                     _ambient_control_plane):
     """Without the explicit MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR overrides the
     WORKSPACE_ROOT that lib.sh exports must be enough to keep the record out
     of the repo's registry."""
     monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
     monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
-    before_jobs, before_logs = _listing(REAL_JOBS), _listing(REAL_LOGS)
-    res = _create_session_from_repo_root(mam_sandbox, "issue15-wsroot-creator-claude")
-    _assert_registered_in_sandbox_only(mam_sandbox, "issue15-wsroot-creator-claude", before_jobs, before_logs, res)
+    session = _unique("issue15-wsroot-creator-claude")
+    real_dirs = _real_state_dirs(_ambient_control_plane)
+    before = _entries(real_dirs)
+    res = _create_session_from_repo_root(mam_sandbox, session)
+    _assert_registered_in_sandbox_only(mam_sandbox, session, real_dirs, before, res)

tests/test_issue18_liveness_contract.py

@@ -24,8 +24,8 @@ sandbox); each stays under a minute.
 from __future__ import annotations
 
 import json
-import re
 import os
+import re
 import signal
 import subprocess
 import sys
@@ -1220,9 +1220,16 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path
         stdout, stderr = proc.communicate(timeout=180)
         jid = _wait_job_id(reg)
         if proc.returncode != 0:
+            if ("no CONNACK from broker" in stderr
+                    and "published started" not in runner_log.read_text(errors="replace")):
+                pytest.skip("broker unreachable when publishing the dispatch (no CONNACK); the runner never saw the job")
             _skip_if_broker_did_not_deliver(reg, runner_log)
         assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
-        assert "subscriber received started" in stderr or "advanced to 'running'" in stderr
+        # delivery is confirmed by the protocol: the subscriber saw `started`, or
+        # the registry left pending - to running, or straight to completed when
+        # the broker delivered only the retained terminal event
+        assert ("subscriber received started" in stderr or "advanced to 'running'" in stderr
+                or "advanced to 'completed'" in stderr), stderr
         assert _job(reg, jid)["status"] == "completed", \
             f"the subscriber mirrors the remote terminal into the registry\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
         assert _job(reg, jid)["transport"] == "remote"

tests/test_issue20_state_isolation.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python3
+"""Issue #20 - test state stays out of every real control plane.
+
+conftest records the control-plane binding an agent pane carries and drops it
+before collection (pytest_configure), gives every test a registry and an
+audit-log root under its own tmp_path (_isolate_mam_state_dirs), and a
+session guard reports new entries in a real `.mam/jobs` or
+`.mam/delegate_job_logs` (mam_state_leak_guard, helpers in _mam_state_guard).
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+import _mam_state_guard as guard
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+
+
+def _fake_control_plane(root: Path, sessions=()) -> Path:
+    """A checkout-shaped tree with a `.mam` registry, audit-log root and session yaml."""
+    mam = root / ".mam"
+    (mam / "jobs").mkdir(parents=True)
+    (mam / "delegate_job_logs").mkdir()
+    rows = "".join(f"- name: {s}\n  status: running\n  mcp_attachments:\n  - name: stitch\n" for s in sessions)
+    (mam / "agent-sessions.yaml").write_text("herdr_sessions:\n" + rows if rows else "herdr_sessions: []\n")
+    return root
+
+
+def _record(mam: Path, jid: str, session) -> None:
+    rec = {"job_id": jid, "prompt": f"prompt of {jid}"}
+    if session:
+        rec["agent_session"] = session
+    (mam / "jobs" / f"{jid}.json").write_text(json.dumps(rec))
+
+
+def _log(mam: Path, jid: str, session=None) -> None:
+    d = mam / "delegate_job_logs" / jid
+    d.mkdir()
+    if session:
+        (d / "meta.json").write_text(json.dumps({"job_id": jid, "agent_session": session}))
+
+
+# ---------------------------------------------------------------------------
+# the session guard (S4, S5)
+# ---------------------------------------------------------------------------
+def test_s4_guard_flags_only_entries_no_live_delegation_explains(tmp_path):
+    main = _fake_control_plane(tmp_path / "main", sessions=("creator-real",))
+    mam = main / ".mam"
+    _record(mam, "old00001", "herdr:nobody")        # before the snapshot: not judged
+    _log(mam, "old00001")
+    recorded = {"WORKSPACE_ROOT": str(main)}
+    roots = guard.watch_roots(tmp_path / "repo", recorded)
+    before = guard.snapshot(roots, recorded)
+
+    _record(mam, "live0001", "herdr:creator-real")  # a live delegation: record first, registered session
+    _log(mam, "live0001", "herdr:creator-real")
+    _record(mam, "live0002", "remote:creator-real")
+    _log(mam, "orphan01", "herdr:creator-real")     # (a) audit log with no record, even for a known session
+    _record(mam, "stray001", "herdr:nobody")         # (b) unknown session
+    _record(mam, "stitch01", "herdr:stitch")         # (b) `stitch` is an mcp attachment, not a session
+    _record(mam, "anon0001", None)                   # (b) no session at all
+
+    found = {Path(f["path"]).name: f for f in guard.leaks(roots, before, recorded)}
+    assert sorted(found) == ["anon0001.json", "orphan01", "stitch01.json", "stray001.json"], found
+    assert found["orphan01"]["reason"] == "audit log without a registry record"
+    assert found["anon0001.json"]["reason"] == "no agent_session"
+    assert found["stray001.json"]["reason"] == "agent_session names no registered session"
+    assert found["stray001.json"]["prompt"] == "prompt of stray001"
+    assert "orphan01" in guard.describe(list(found.values()))
+
+
+def test_s5_watch_roots_follow_the_recorded_binding(tmp_path):
+    repo = _fake_control_plane(tmp_path / "repo")
+    main = _fake_control_plane(tmp_path / "main")
+    (tmp_path / "main-link").symlink_to(main)
+    own, other = str((repo / ".mam").resolve()), str((main / ".mam").resolve())
+
+    assert guard.watch_roots(repo, {}) == [own]
+    # WORKSPACE_ROOT naming this checkout is counted once
+    assert guard.watch_roots(repo, {"WORKSPACE_ROOT": str(repo)}) == [own]
+    # a worktree creator's pane: all four at another checkout, reached through a symlink too
+    pane = {"WORKSPACE_ROOT": str(tmp_path / "main-link"),
+            "AGENT_SESSIONS_YAML": str(main / ".mam" / "agent-sessions.yaml"),
+            "MAM_JOBS_DIR": str(main / ".mam" / "jobs"),
+            "DELEGATE_JOB_LOGS_DIR": str(main / ".mam" / "delegate_job_logs")}
+    assert guard.watch_roots(repo, pane) == [own, other]
+    # a recorded root that does not exist is skipped; the checkout's own .mam never is
+    bare = tmp_path / "bare"
+    assert guard.watch_roots(bare, {"WORKSPACE_ROOT": str(tmp_path / "gone")}) == [str((bare / ".mam").resolve())]
+    # a registry outside any <root>/jobs is watched directly
+    odd = tmp_path / "elsewhere" / "registry"
+    odd.mkdir(parents=True)
+    rec = {"MAM_JOBS_DIR": str(odd)}
+    jobs, logs = guard.state_dirs(guard.watch_roots(repo, rec), rec)
+    assert str(odd.resolve()) in jobs and str(odd.resolve()) not in logs
+
+
+# ---------------------------------------------------------------------------
+# per-test defaults and the dropped binding (S1-S3, S6, S7)
+# ---------------------------------------------------------------------------
+# Copied at import, like test_orc_onboard.SCRUBBED_ENV: the binding has to be
+# gone before collection, not merely before the first test (S6).
+_IMPORT_ENV = dict(os.environ)
+PROBE_ENV = "MAM_ISSUE20_PROBE"
+BINDING = ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
+
+
+def _tree(root: Path) -> dict:
+    return {str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() if p.is_file() else None
+            for p in sorted(root.rglob("*"))}
+
+
+def _binding(root: Path) -> dict:
+    """The four variables a worktree creator's pane carries, aimed at `root`."""
+    mam = root / ".mam"
+    return {"WORKSPACE_ROOT": str(root), "AGENT_SESSIONS_YAML": str(mam / "agent-sessions.yaml"),
+            "MAM_JOBS_DIR": str(mam / "jobs"), "DELEGATE_JOB_LOGS_DIR": str(mam / "delegate_job_logs")}
+
+
+def test_s1_every_test_defaults_to_state_dirs_under_tmp_path(tmp_path, monkeypatch):
+    from lib_py.delegation import mqtt_common
+
+    state = tmp_path / ".mam-test-state"
+    assert os.environ["MAM_JOBS_DIR"] == str(state / "jobs")
+    assert os.environ["DELEGATE_JOB_LOGS_DIR"] == str(state / "delegate_job_logs")
+    # a WORKSPACE_ROOT on a checkout with a .mam no longer decides where state goes
+    monkeypatch.setenv("WORKSPACE_ROOT", str(_fake_control_plane(tmp_path / "repo")))
+    assert mqtt_common.default_registry_dir() == str(state / "jobs")
+    assert mqtt_common.get_logs_dir() == str(state / "delegate_job_logs")
+
+
+def test_s2_mam_sandbox_overrides_the_default(mam_sandbox):
+    assert os.environ["MAM_JOBS_DIR"] == str(mam_sandbox / ".mam" / "jobs")
+    assert os.environ["DELEGATE_JOB_LOGS_DIR"] == str(mam_sandbox / ".mam" / "delegate_job_logs")
+
+
+def test_s3_submit_with_workspace_root_on_a_checkout_leaves_its_mam_alone(tmp_path):
+    """The issue's acceptance check in small: WORKSPACE_ROOT names a checkout
+    with a real-looking .mam and the test sets nothing else."""
+    repo = _fake_control_plane(tmp_path / "repo", sessions=("creator-real",))
+    before = _tree(repo)
+    env = dict(os.environ, WORKSPACE_ROOT=str(repo), MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"),
+               PYTHONPATH=f"{SKILLS}:{os.environ.get('PYTHONPATH', '')}")
+    res = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "issue 20 probe",
+         "--agent", "claude-code", "--agent-session", "herdr:issue20-probe"],
+        capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), timeout=60,
+    )
+    assert res.returncode == 0, res.stderr
+    state = tmp_path / ".mam-test-state"
+    records = list((state / "jobs").glob("*.json"))
+    assert len(records) == 1, records
+    assert (state / "delegate_job_logs" / records[0].stem).is_dir()
+    assert _tree(repo) == before
+
+
+def test_s6_a_pane_binding_is_dropped_before_collection(tmp_path):
+    """A pytest started with a worktree creator pane's binding - all four at
+    another checkout F - sees none of it, not even in a module-level copy of
+    os.environ, and leaves F untouched."""
+    fake_main = _fake_control_plane(tmp_path / "main", sessions=("creator-real",))
+    before = _tree(fake_main)
+    env = dict(os.environ, **_binding(fake_main), **{PROBE_ENV: "1"})
+    # An explicit basetemp: the inner run must not clean up the outer one's.
+    res = subprocess.run(
+        [sys.executable, "-m", "pytest", f"{__file__}::test_s6_probe", "-q", "-p", "no:cacheprovider",
+         f"--basetemp={tmp_path / 'inner'}"],
+        capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), timeout=300,
+    )
+    assert res.returncode == 0, res.stdout[-4000:] + res.stderr[-2000:]
+    assert "1 passed" in res.stdout, res.stdout[-2000:]
+    assert _tree(fake_main) == before
+
+
+@pytest.mark.skipif(not os.environ.get(PROBE_ENV), reason="the inner half of S6 (nested pytest) only")
+def test_s6_probe(tmp_path):
+    assert not [k for k in BINDING if k in _IMPORT_ENV], {k: _IMPORT_ENV[k] for k in BINDING if k in _IMPORT_ENV}
+    assert "AGENT_SESSIONS_YAML" not in os.environ
+    assert "WORKSPACE_ROOT" not in os.environ  # it named another checkout
+    state = tmp_path / ".mam-test-state"
+    assert os.environ["MAM_JOBS_DIR"] == str(state / "jobs")
+    assert os.environ["DELEGATE_JOB_LOGS_DIR"] == str(state / "delegate_job_logs")
+
+
+def test_s7_lib_sh_takes_the_session_registry_from_a_tests_workspace_root(tmp_path):
+    """An ambient AGENT_SESSIONS_YAML (a worktree pane's) would win over the
+    test's WORKSPACE_ROOT in lib.sh; conftest dropped it before collection."""
+    assert "AGENT_SESSIONS_YAML" not in os.environ
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    script = f'source "{SKILLS / "lib.sh"}" >/dev/null 2>&1; printf %s "$AGENT_SESSIONS_YAML"'
+    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                         env=dict(os.environ, WORKSPACE_ROOT=str(repo)), timeout=60)
+    assert res.stdout == str(repo / ".mam" / "agent-sessions.yaml"), (res.stdout, res.stderr)

tests/test_issue21_acceptance.py

@@ -0,0 +1,278 @@
+"""Issue #21 item 7: the eight acceptance cases (A1-A8) of the worktree
+creator model, one test (or pair) per case.
+
+Everything runs against mock herdr / stubbed delegation inside mam_sandbox or
+tmp_path; nothing touches the real .mam. The real herdr pane (whether a pane
+sees the --env values) and the trial operation belong to stage C - A2 pins
+what the scripts hand to herdr.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import time
+from pathlib import Path
+
+from test_issue21_loop_b import LOCAL_ROWS, _drive, _reviews, _sidecars, _state
+from test_issue21_phase2_loop_a import (REM, _calls, _prepare, _resume, _rows, _run, _skill, _wt,  # noqa: F401
+                                        ctl)
+from test_issue21_worktree_prereqs import (CP_ENV, WRAPPER, _assert_no_env_in_agent_args, _binding, _dry_submit,
+                                           _env, _env_flags, _injected, _kill_timer, _only_job_id)
+from test_issue21_worktree_prereqs import _git as _g
+from test_issue21_worktree_prereqs import wt_repo  # noqa: F401
+
+CREATE = ("multi-agent-mux-create", "scripts", "create_session.sh")
+HS = "multi-agent-mux"  # listed as running by mock herdr: an existing server
+
+
+def _raw_calls(state_file) -> list:
+    return json.loads(Path(state_file).read_text()).get("calls", [])
+
+
+def _sibling(main: Path, state_file):
+    """A1 setup: a session on the existing server in the control root, then a
+    sibling worktree's creator on the same server. Returns (wt, since, rows before)."""
+    res = _run(main, CREATE, "--workspace", str(main), "--agent", "claude", "--role", "reviewer",
+               "--herdr-session", HS, "--no-onboard")
+    assert res.returncode == 0, res.stdout + res.stderr
+    before = _rows(main)
+    since = len(_raw_calls(state_file))
+    wt = _wt(main)
+    res = _prepare(main, wt, "--creator-agent", "claude", "--herdr-session", HS)
+    assert res.returncode == 0, res.stdout + res.stderr
+    return wt, since, before
+
+
+# A1 -------------------------------------------------------------------------
+
+def test_a1_sibling_worktree_on_an_existing_herdr_server(ctl, mock_herdr, mock_agents):
+    main = ctl
+    wt, since, before = _sibling(main, mock_herdr)
+    new = _raw_calls(mock_herdr)[since:]
+    assert not [c for c in new if c[-1:] == ["server"]], "no server bootstrap on an existing server"
+    rows = _rows(main)
+    assert rows[0] == before[0], "the existing row is untouched"
+    creator = [r for r in rows if r.get("role") == "creator"]
+    assert len(rows) == 2 and len(creator) == 1
+    assert rows[0].get("herdr_session") == creator[0].get("herdr_session") == HS
+    assert os.path.realpath(creator[0]["pane"]["cwd"]) == os.path.realpath(wt)
+    assert creator[0]["control_plane"]["workspace_root"] == os.path.realpath(main)
+
+
+# A2 -------------------------------------------------------------------------
+
+def test_a2_pane_env_carries_the_binding_not_the_agent_args(ctl, mock_herdr, mock_agents):
+    main = ctl
+    wt, since, _ = _sibling(main, mock_herdr)
+    flags = _env_flags(mock_herdr, since)
+    for k, v in zip(CP_ENV, _binding(main).values()):
+        assert f"{k}={v}" in flags, flags
+    spawns = [c for c in _calls(mock_herdr) if c[:2] in (["pane", "split"], ["workspace", "create"])]
+    assert spawns and any("--env" in c for c in spawns), "the binding rides on the pane-creating call"
+    _assert_no_env_in_agent_args(mock_herdr, since)
+
+
+# A3 -------------------------------------------------------------------------
+
+def test_a3_brief_start_and_report_route_to_the_control_root(tmp_path):
+    main, wt = tmp_path / "main", tmp_path / "wt"
+    main.mkdir()
+    _g(main, "init", "-q")
+    (main / "a.txt").write_text("a\n")
+    _g(main, "add", "a.txt")
+    _g(main, "commit", "-q", "-m", "init")
+    _g(main, "worktree", "add", "-q", "-b", "feat", str(wt))
+    real = os.path.realpath(main)
+    res = _dry_submit(main, _env(tmp_path, {"WORKSPACE_ROOT": main}), "--workdir", str(main))
+    assert res.returncode == 0, res.stderr
+    reg = main / ".mam" / "jobs"
+    jid = _only_job_id(reg)
+    assert f"{real}/.mam/jobs/{jid}/brief.md" in res.stdout
+    start = _injected(res.stdout, "start")
+    assert f"--registry-dir {real}/.mam/jobs " in start and start.startswith(f"DELEGATE_JOB_LOGS_DIR={real}/")
+    # the brief (not written by --dry-run) derives its report path from that same
+    # absolute registry dir, so the report lands under MAIN as well
+    wrapper = WRAPPER.read_text()
+    assert "- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/$AGENT-reports/report-final.md" in wrapper
+    try:
+        r1 = subprocess.run(["bash", "-c", start + " --attempts 1"], capture_output=True, text=True,
+                            env=_env(tmp_path), cwd=str(wt), timeout=120)  # no control env, broker unreachable
+        assert r1.returncode == 2, r1.stderr
+        assert json.loads((reg / f"{jid}.json").read_text())["status"] != "pending"
+    finally:
+        _kill_timer(reg, jid)
+    assert not (wt / ".mam").exists(), "nothing lands in the worktree"
+
+
+# A4 -------------------------------------------------------------------------
+
+def _main_state(main: Path) -> tuple:
+    branch = _g(main, "branch", "--show-current")
+    return (_g(main, "rev-parse", "HEAD"), branch, _g(main, "status", "--porcelain"),
+            _g(main, "rev-parse", f"refs/heads/{branch}"), _g(main, "worktree", "list", "--porcelain"))
+
+
+def test_a4_main_checkout_is_unchanged_by_the_whole_cycle(ctl, monkeypatch):
+    main = ctl
+    # what prepare adds to info/exclude is added up front, so status compares like for like
+    excl = main / ".git" / "info" / "exclude"
+    excl.write_text(excl.read_text() + "/.venv\n.pytest_cache/\n")
+    scratch = main.parent / f"{main.name}-a4"
+    scratch.mkdir()
+    pre = _main_state(main)
+    wt = _wt(main)
+    res = _prepare(main, wt)
+    assert res.returncode == 0, res.stdout + res.stderr
+    base = _g(wt, "rev-parse", "HEAD")
+    (wt / "impl.txt").write_text("impl line\n")
+    _g(wt, "add", "impl.txt")
+    _g(wt, "commit", "-q", "-m", "impl")
+    _state(monkeypatch, LOCAL_ROWS)
+    rc, calls, r = _drive(monkeypatch, scratch, ["--worktree", str(wt), "--base", base, "--review-only"])
+    assert rc == 0 and len(_reviews(calls)) == 1
+    res = _run(main, REM, "--path", str(wt))
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert _main_state(main) == pre, "HEAD, branch, status, branch tip and worktree list are unchanged"
+    assert "mam-review" not in _g(main, "worktree", "list", "--porcelain")
+
+
+# A5 -------------------------------------------------------------------------
+
+def _review_path(prompt: str) -> Path:
+    m = re.search(r"^- 검증 경로: (.+)$", prompt, re.M)
+    assert m, prompt[:800]
+    return Path(m.group(1))
+
+
+def _commit_h2(wt: Path) -> str:
+    (wt / "impl.txt").write_text("impl line\nsecond commit\n")
+    _g(wt, "commit", "-q", "-am", "h2")
+    return _g(wt, "rev-parse", "HEAD")
+
+
+def test_a5_fixed_sha_review_while_the_creator_keeps_committing(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    base = _g(wt, "rev-parse", "HEAD")
+    (wt / "impl.txt").write_text("impl line\n")
+    _g(wt, "add", "impl.txt")
+    _g(wt, "commit", "-q", "-m", "h1")
+    h1 = _g(wt, "rev-parse", "HEAD")
+    seen = {}
+
+    def reviewer(r, n, prompt):
+        if seen:
+            return
+        path = _review_path(prompt)
+        seen["head"] = _g(path, "rev-parse", "HEAD")
+        seen["symref"] = subprocess.run(["git", "-C", str(path), "symbolic-ref", "-q", "HEAD"]).returncode
+        seen["content"] = (path / "impl.txt").read_text()
+        seen["h2"] = _commit_h2(wt)
+
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
+                          on_review=reviewer, max_loop=2)
+    assert rc == 0
+    assert (seen["head"], seen["symref"], seen["content"]) == (h1, 1, "impl line\n"), "round 1 saw H1, detached"
+    assert [role for _, role, _ in calls] == ["Reviewer", "Reviewer"], "no correction job for a stale round"
+    assert f"- head: {seen['h2']}" in _reviews(calls)[1][1]
+    rounds = r._rounds
+    assert rounds[0]["stale"] is True and rounds[1]["head_sha"] == seen["h2"]
+    (side,) = _sidecars(main)
+    data = json.loads(side.read_text())
+    assert (data["head_sha"], data["evidence"]) == (seen["h2"], "review-tree")
+
+
+def test_a5b_a_stale_last_round_is_not_accepted(wt_repo, monkeypatch, tmp_path, capsys):
+    main, wt = wt_repo
+    base = _g(wt, "rev-parse", "HEAD")
+    (wt / "impl.txt").write_text("impl line\n")
+    _g(wt, "add", "impl.txt")
+    _g(wt, "commit", "-q", "-m", "h1")
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
+                          on_review=lambda *_: _commit_h2(wt), max_loop=1)
+    assert rc == 1 and "changed during review" in capsys.readouterr().err
+    assert _sidecars(main) == []
+
+
+# A6 -------------------------------------------------------------------------
+
+def test_a6_row_recovery_and_resume_only_at_the_original_cwd(ctl, mock_herdr, mock_agents):
+    main = ctl
+    wt = _wt(main)
+    res = _prepare(main, wt, "--creator-agent", "claude")
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(main)
+    name, binding = row["name"], row["control_plane"]
+    # the creator died: gone from herdr, the row terminated (reconcile drift A)
+    st = json.loads(Path(mock_herdr).read_text())
+    st["agents"] = {}
+    Path(mock_herdr).write_text(json.dumps(st))
+    from test_issue21_phase2_loop_a import _mutate
+    _mutate(main, "d['herdr_sessions'][0]['status'] = 'terminated'")
+    res = _resume(main, main, name)
+    assert res.returncode == 2 and (str(wt) in res.stderr or os.path.realpath(wt) in res.stderr), res.stderr
+    since = len(_raw_calls(mock_herdr))
+    res = _resume(main, wt, name)
+    assert res.returncode == 0, res.stdout + res.stderr
+    (row,) = _rows(main)
+    assert row["status"] == "running" and row["control_plane"] == binding
+    flags = _env_flags(mock_herdr, since)
+    assert f"WORKSPACE_ROOT={binding['workspace_root']}" in flags, flags
+
+
+# A7 -------------------------------------------------------------------------
+
+def test_a7_reviewer_test_output_stays_out_of_the_source_and_the_sandbox(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    base = _g(wt, "rev-parse", "HEAD")
+    (wt / "impl.txt").write_text("impl line\n")
+    status0 = _g(wt, "status", "--porcelain")
+    paths = []
+
+    def reviewer_runs_tests(r, n, prompt):
+        path = _review_path(prompt)
+        paths.append(path)
+        (path / ".pytest_cache").mkdir()
+        (path / ".pytest_cache" / "v").write_text("x\n")
+        (path / "__pycache__").mkdir()
+        (path / "__pycache__" / "x.pyc").write_bytes(b"\0")
+        (path / "tmp-output.txt").write_text("output\n")
+
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
+                          on_review=reviewer_runs_tests)
+    assert rc == 0 and len(paths) == 1
+    assert _g(wt, "status", "--porcelain") == status0, "the source worktree did not change"
+    assert not paths[0].exists() and str(paths[0]) not in _g(main, "worktree", "list", "--porcelain")
+    prompt = _reviews(calls)[0][1]
+    assert "PYTHONDONTWRITEBYTECODE=1" in prompt and "-p no:cacheprovider" in prompt
+    assert not (wt / ".mam").exists() and not (wt / ".agents").exists(), "state stays in the control root"
+    assert len(_sidecars(main)) == 1
+
+
+# A8 -------------------------------------------------------------------------
+
+def test_a8_second_loop_is_refused_before_a_scheduler_exists(ctl):
+    main = ctl
+    wt2 = _wt(main, "wt2")
+    _g(main, "worktree", "add", "-q", "--detach", str(wt2))
+    holder = subprocess.Popen(["sleep", "60"])
+    try:
+        lstart = subprocess.run(["ps", "-p", str(holder.pid), "-o", "lstart="], capture_output=True,
+                                text=True).stdout.strip()
+        marker = main.parent / f"{main.name}-loop-marker"
+        content = f"pid={holder.pid}\nlstart={lstart}\nstarted={time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}\n"
+        marker.write_text(content)
+        env = {**os.environ, "WORKSPACE_ROOT": str(main), "MAM_LOOP_MARKER": str(marker), "MAM_LOOP_NO_FREEZE": "1"}
+        env.pop("MAM_LOOP_PY", None)
+        res = subprocess.run(["bash", _skill(main, "multi-agent-mux-loop", "scripts", "run_loop.sh"),
+                              "--creator", "c", "--task", "t", "--worktree", str(wt2), "--base", "HEAD",
+                              "--review-only", "--reviewer", "r"],
+                             capture_output=True, text=True, env=env, cwd=str(main), timeout=120)
+        assert res.returncode == 1 and "already running" in (res.stdout + res.stderr), res.stdout + res.stderr
+        assert list((main / ".mam" / "jobs").glob("*.json")) == [], "no job was registered"
+        assert marker.read_text() == content, "the live holder's marker is kept"
+    finally:
+        holder.kill()
+        holder.wait()

tests/test_issue21_loop_b.py

@@ -25,6 +25,7 @@ REMOTE_ROW = {"name": "reviewer-far-01", "agent": "claude", "role": "reviewer",
               "transport": "remote"}
 LOCAL_ROWS = [{"name": "creator-claude-01", "agent": "claude", "role": "creator", "status": "running"},
               {"name": "reviewer-claude-01", "agent": "claude", "role": "reviewer", "status": "running"}]
+WORK_TAIL = "코드 작업·검증 대상은 위 소스 경로다"
 
 
 def _state(monkeypatch, rows):
@@ -195,6 +196,7 @@ def test_l6_env_isolation_reviews_the_source_tree(wt_repo, monkeypatch, tmp_path
     assert rc == 0 and counts == [n], "env isolation opens no review worktree"
     prompt = _reviews(calls)[0][1]
     assert f"- 검증 경로: {os.path.realpath(wt)}" in prompt and "PYTHONDONTWRITEBYTECODE=1" in prompt
+    assert WORK_TAIL not in prompt and "검증 경로(= 위 소스 경로)" in prompt, "one verification target (B-2)"
     data = json.loads(_sidecars(main)[0].read_text())
     assert (data["evidence"], data["isolation"]) == ("source-tree", "env")
 
@@ -287,6 +289,7 @@ def test_l11_diff_only_remote_reviewer_gets_shas_and_the_snapshot_diff(wt_repo,
     assert f"- head: {snap['head_sha']}" in remote and f"- tree: {snap['tree_sha']}" in remote
     assert "diff 만으로" in remote and "+uncommitted change" in remote and "+impl line" in remote
     assert "- 검증 경로: " in local and "mam-review" in local
+    assert WORK_TAIL not in local and "[리뷰 identity] 의 검증 경로에서만" in local, "one verification target (B-2)"
     assert snap["reviewers"]["reviewer-far-01"]["evidence"] == "diff-only"
     assert snap["reviewers"]["reviewer-claude-01"]["evidence"] == "review-tree"
     side = {json.loads(p.read_text())["reviewer"]: json.loads(p.read_text()) for p in _sidecars(main)}
@@ -332,12 +335,14 @@ def test_l13_snapshot_diff_is_commit_to_commit(wt_repo, monkeypatch, tmp_path):
     # the two-argument form is unchanged: same output as the script before this change
     old = subprocess.run(["git", "-C", str(SKILLS.parent.parent), "show", "25f4be3:.agents/skills/multi-agent-mux-loop/scripts/diff_collect.sh"],
                          capture_output=True, text=True)
-    old_sh = tmp_path / "diff_collect_old.sh"
-    old_sh.write_text(old.stdout)
     new2 = _diff_collect(wt, base)
-    old2 = subprocess.run(["bash", "-c", f"source '{old_sh}' && mam_collect_changes_diff '{wt}' '{base}'"],
-                          capture_output=True, text=True)
-    assert (new2.returncode, new2.stdout) == (old2.returncode, old2.stdout)
+    assert new2.returncode == 0 and "+modified" in new2.stdout and "+untracked" in new2.stdout
+    if old.returncode == 0 and old.stdout:  # needs git history (absent in a `git archive` export)
+        old_sh = tmp_path / "diff_collect_old.sh"
+        old_sh.write_text(old.stdout)
+        old2 = subprocess.run(["bash", "-c", f"source '{old_sh}' && mam_collect_changes_diff '{wt}' '{base}'"],
+                              capture_output=True, text=True)
+        assert (new2.returncode, new2.stdout) == (old2.returncode, old2.stdout)
 
 
 def test_l13b_prompt_diff_comes_from_the_snapshot(wt_repo, monkeypatch, tmp_path):
@@ -395,3 +400,127 @@ def test_l15_truncated_diff_refuses_diff_only_but_not_local(wt_repo, monkeypatch
     assert rc == 1 and _reviews(calls) == [] and "would have no evidence" in capsys.readouterr().err
     rc, calls, r = _drive(monkeypatch, tmp_path, argv, reviewers="reviewer-claude-01")
     assert rc == 0 and len(_reviews(calls)) == 1
+
+
+# ---------------------------------------------------------------------------
+# loop B review follow-up: B-2, B-R1..B-R3, N1, N3, N5
+# ---------------------------------------------------------------------------
+
+def _add_submodule(wt: Path, tmp_path: Path) -> None:
+    src = tmp_path / "sm-src"
+    src.mkdir()
+    _git(src, "init", "-q")
+    (src / "x.txt").write_text("original\n")
+    _git(src, "add", "x.txt")
+    _git(src, "commit", "-q", "-m", "initial")
+    _git(wt, "-c", "protocol.file.allow=always", "submodule", "add", "-q", str(src), "sm")
+    _git(wt, "commit", "-q", "-m", "add sm")
+
+
+@pytest.mark.parametrize("when", ["before", "before-ignored", "untracked", "during"])
+def test_l17_dirty_submodule_is_refused_or_makes_the_round_stale(wt_repo, monkeypatch, tmp_path, capsys, when):
+    main, wt = wt_repo
+    base = _impl(wt)
+    _add_submodule(wt, tmp_path)
+    if when == "before-ignored":  # no ignore setting hides the dirty content
+        _git(wt, "config", "submodule.sm.ignore", "all")
+        _git(wt, "config", "diff.ignoreSubmodules", "all")
+
+    def dirty(*_):
+        (wt / "sm" / "x.txt").write_text("unreviewed work\n")
+
+    if when.startswith("before"):
+        dirty()
+    elif when == "untracked":
+        (wt / "sm" / "new.txt").write_text("untracked in the submodule\n")
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
+                          on_review=dirty if when == "during" else None)
+    assert rc == 1 and _sidecars(main) == []
+    err = capsys.readouterr().err
+    assert "submodule(s) with uncommitted or untracked content: sm" in err
+    if when == "during":
+        assert len(_reviews(calls)) == 1 and r._rounds[0]["stale"] is True
+    else:
+        assert _reviews(calls) == [], "refused before any review"
+
+
+def test_l17b_clean_submodule_is_reviewed(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    base = _impl(wt)
+    _add_submodule(wt, tmp_path)
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"])
+    assert rc == 0 and len(_sidecars(main)) == 1
+
+
+def test_l18_self_review_is_a_recorded_verdict_and_not_pass_re_reviews(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    base = _impl(wt)
+    report = tmp_path / "pass-report.md"
+
+    def verdicts(r, n, prompt):
+        report.write_text("[VERDICT: NOT PASS]\n" if n == 1 else "[VERDICT: PASS]\n")
+
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
+                          on_review=verdicts, reviewers=None, max_loop=2)
+    assert rc == 0
+    assert [(s, role) for s, role, _ in calls] == [("creator-claude-01", "Reviewer")] * 2, \
+        "a NOT PASS self-review is re-reviewed, no correction job"
+    prompt = _reviews(calls)[1][1]
+    assert "Self-Review" in prompt and "+impl line" in prompt and "[리뷰 identity]" in prompt
+    assert r._rounds[0]["verdicts"] == {"creator-claude-01": "NOT PASS"}
+    (side,) = _sidecars(main)
+    data = json.loads(side.read_text())
+    assert (data["reviewer"], data["loop_id"], data["head_sha"]) == ("creator-claude-01", r.loop_id,
+                                                                    _git(wt, "rev-parse", "HEAD"))
+    assert data["evidence"] == "review-tree"
+
+
+def test_l18b_self_review_without_options_is_unchanged(wt_repo, monkeypatch, tmp_path):
+    main, _ = wt_repo
+    (main / "a.txt").write_text("a\nchanged\n")
+    rc, calls, r = _drive(monkeypatch, tmp_path, [], reviewers=None)
+    (review,) = _reviews(calls)
+    assert rc == 0 and review[0] == "creator-claude-01"
+    assert review[1].startswith("작업 완료 상태에 대해 스스로 검증") and "변경분" not in review[1]
+    assert _sidecars(main) == []
+
+
+def test_l19_remote_self_review_gets_the_snapshot_diff(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    base = _impl(wt)
+    _state(monkeypatch, [dict(LOCAL_ROWS[0], transport="remote")])
+    rc, calls, r = _drive(monkeypatch, tmp_path,
+                          ["--worktree", str(wt), "--base", base, "--review-only", "--remote-review", "diff-only"],
+                          reviewers=None)
+    assert rc == 0
+    prompt = _reviews(calls)[0][1]
+    assert "diff --git" in prompt and "+impl line" in prompt and "검증 경로" not in prompt
+    data = json.loads(_sidecars(main)[0].read_text())
+    assert (data["reviewer"], data["evidence"]) == ("creator-claude-01", "diff-only")
+
+
+def test_l20_snapshot_diff_failure_keeps_the_unavailable_marker(wt_repo):
+    main, wt = wt_repo
+    dst = main / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts"
+    dst.mkdir(parents=True)
+    (dst / "diff_collect.sh").write_text("mam_collect_changes_diff() { return 1; }\n")
+    from lib_py.loop.runner import UNAVAILABLE_MARKER, LoopRunner
+    r = LoopRunner(["--creator", "creator-claude-01", "--task", "t", "--worktree", str(wt)])
+    r.parse_args()
+    r.base_commit = _git(wt, "rev-parse", "HEAD")
+    assert UNAVAILABLE_MARKER in r.collect_changes_diff(r.base_commit)
+    assert UNAVAILABLE_MARKER not in r.collect_changes_diff(), "the live-tree form keeps its git fallback"
+
+
+def test_l21_closing_a_review_tree_keeps_other_worktree_entries_and_its_parent_dir(wt_repo, monkeypatch, tmp_path):
+    main, wt = wt_repo
+    base = _impl(wt)
+    other = tmp_path / "detached-disk"
+    _git(main, "worktree", "add", "-q", "--detach", str(other))
+    import shutil
+    import tempfile
+    shutil.rmtree(other)  # a worktree whose directory is gone (a detached disk)
+    rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"])
+    assert rc == 0
+    assert str(other) in _git(main, "worktree", "list", "--porcelain"), "the loop does not prune others' entries"
+    assert not (Path(tempfile.gettempdir()) / "mam-review" / r.loop_id).exists()