feat: securely prepare verification roots

harness/agents/impl/skills/gitea-issue-implementation/SKILL.md

@@ -251,9 +251,12 @@ Main agent는 구현 범위와 최종 결과에 책임을 진다. 독립적인 r
 
 검증 실행 기록은 공통 [Repository cache 운영 지침](../../../../operations/repository-cache.md)의 정본 경로 `<harness-root>/repositories/<owner>/<repo>/worktrees/issue-<number>/verification/<run-id>`에 보존한다. Product `checkout`과 artifact root를 분리하며, 회차별 evidence를 제품 `docs/verification/`에 새로 커밋하지 않는다. 기존 정본 문서·테스트·fixture·추적 파일은 자동 이동·삭제·ignore하지 않는다.
 
-인증된 Issue/cache/worktree admission을 마친 caller가 broker Issue lease 아래 새 run directory를 mode `0700`으로 만들고 계획 파일을 mode `0600`으로 준비한다. Stage별 새 run-id를 사용한다. Helper는 directory/lease/cache를 생성·획득·복구하지 않는다. `<harness-root>`는 실제로 읽은 contract path로 고정하고 다음 변수는 그 root와 인증된 Issue/run identity에서 결정한다. 아래 명령은 harness root에서 호출한다.
+인증된 Issue/cache/worktree admission을 마친 caller가 broker Issue lease 아래 evidence `run` 전에 live lease의 exact harness root·Issue identity를 인증하고 root를 준비한다. Stage별 새 run-id를 사용한다. 아래 prepare는 `verification/`과 `<run-id>/`를 각각 별도 생성하고 각 directory를 실행 UID 소유 mode `0700`으로 만들며 즉시 `lstat`한다. Symlink, owner·mode 불일치와 collision은 거부하고 기존 malformed root를 자동 chmod하거나 repair하지 않는다. 이 경로에는 중간 parent가 umask를 따르는 `Path.mkdir(parents=True, mode=0o700)`를 사용하지 않는다. Helper는 lease/cache를 획득·복구하지 않는다. `<harness-root>`는 실제로 읽은 contract path로 고정하고 다음 변수는 그 root와 인증된 Issue/run identity에서 결정한다. 아래 명령은 harness root에서 호출한다.
 
 ```sh
+python3 harness/lib/verification_root.py prepare \
+  --root "$HARNESS_ROOT" --owner "$OWNER" --repository "$REPOSITORY" \
+  --issue "$ISSUE" --run-id "$RUN_ID" --handle "$HANDLE"
 python3 harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py run \
   --contract-path "$HARNESS_ROOT/harness/operations/repository-cache.md" \
   --repo-root "$HARNESS_ROOT/repositories/$OWNER/$REPOSITORY/worktrees/issue-$ISSUE/checkout" \

harness/agents/impl/skills/gitea-issue-rework/SKILL.md

@@ -321,9 +321,12 @@ Reviewer가 rework head를 새 snapshot으로 다시 검토해야 하며 이 com
 
 검증 실행 기록은 공통 [Repository cache 운영 지침](../../../../operations/repository-cache.md)의 정본 경로 `<harness-root>/repositories/<owner>/<repo>/worktrees/issue-<number>/verification/<run-id>`에 보존한다. Product `checkout`과 artifact root를 분리하며, 회차별 evidence를 제품 `docs/verification/`에 새로 커밋하지 않는다. 기존 정본 문서·테스트·fixture·추적 파일은 자동 이동·삭제·ignore하지 않는다.
 
-인증된 Issue/cache/worktree admission을 마친 caller가 broker Issue lease 아래 새 run directory를 mode `0700`으로 만들고 계획 파일을 mode `0600`으로 준비한다. Stage별 새 run-id를 사용한다. Helper는 directory/lease/cache를 생성·획득·복구하지 않는다. `<harness-root>`는 실제로 읽은 contract path로 고정하고 다음 변수는 그 root와 인증된 Issue/run identity에서 결정한다. 아래 명령은 harness root에서 호출한다.
+인증된 Issue/cache/worktree admission을 마친 caller가 broker Issue lease 아래 evidence `run` 전에 live lease의 exact harness root·Issue identity를 인증하고 root를 준비한다. Stage별 새 run-id를 사용한다. 아래 prepare는 `verification/`과 `<run-id>/`를 각각 별도 생성하고 각 directory를 실행 UID 소유 mode `0700`으로 만들며 즉시 `lstat`한다. Symlink, owner·mode 불일치와 collision은 거부하고 기존 malformed root를 자동 chmod하거나 repair하지 않는다. 이 경로에는 중간 parent가 umask를 따르는 `Path.mkdir(parents=True, mode=0o700)`를 사용하지 않는다. Helper는 lease/cache를 획득·복구하지 않는다. `<harness-root>`는 실제로 읽은 contract path로 고정하고 다음 변수는 그 root와 인증된 Issue/run identity에서 결정한다. 아래 명령은 harness root에서 호출한다.
 
 ```sh
+python3 harness/lib/verification_root.py prepare \
+  --root "$HARNESS_ROOT" --owner "$OWNER" --repository "$REPOSITORY" \
+  --issue "$ISSUE" --run-id "$RUN_ID" --handle "$HANDLE"
 python3 harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py run \
   --contract-path "$HARNESS_ROOT/harness/operations/repository-cache.md" \
   --repo-root "$HARNESS_ROOT/repositories/$OWNER/$REPOSITORY/worktrees/issue-$ISSUE/checkout" \

harness/lib/managed_lock_broker.py

@@ -191,7 +191,9 @@ class Broker:
         self.token = secrets.token_hex(32)
         self.deadline = time.monotonic() + args.ttl_seconds
         self.set = ManagedLockSet.for_issue(self.root, args.owner, args.repository, args.issue, timeout=args.acquire_timeout)
-        self.meta = {"owner": args.owner, "repository": args.repository, "issue": args.issue, "role": args.role, "run_id": args.run_id, "pid": os.getpid(), "deadline": self.deadline}
+        self.meta = {"root": str(self.root), "owner": args.owner, "repository": args.repository,
+                     "issue": args.issue, "role": args.role, "run_id": args.run_id,
+                     "pid": os.getpid(), "deadline": self.deadline}
         self.state = {**self.meta, "socket": str(self.sock_path), "status": "held", "token": self.token}
         self.server: socket.socket | None = None
         self.socket_identity: tuple[int, int] | None = None
@@ -274,7 +276,7 @@ class Broker:
         action = req.get("action")
         if action == "status":
             return {"status": self.state["status"], "deadline": self.deadline,
-                    "repository_held": self.set.repository.held,
+                    "repository_held": self.set.repository.held, "root": self.state["root"],
                     "owner": self.state["owner"], "repository": self.state["repository"],
                     "issue": self.state["issue"], "role": self.state["role"],
                     "pid": self.state["pid"]}

harness/lib/verification_root.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+"""Prepare canonical, private verification roots for an Issue run."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import os
+import re
+import stat
+import sys
+import time
+from pathlib import Path
+
+if __package__ in (None, ""):
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from harness.lib import managed_lock_broker as broker
+
+
+class VerificationError(ValueError):
+    """The lease, identifiers, or cache layout is not safe to use."""
+
+
+_PRIVATE_MODE = 0o700
+_OWNER_REPOSITORY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,254}")
+_RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}")
+_OPEN_DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
+
+
+def _identifier(value: object, label: str, pattern: re.Pattern[str]) -> str:
+    if not isinstance(value, str) or not pattern.fullmatch(value) or value in {".", ".."}:
+        raise VerificationError(f"invalid {label}")
+    return value
+
+
+def _issue_number(value: object) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        raise VerificationError("issue must be positive")
+    return value
+
+
+def _absolute(value: os.PathLike[str] | str, label: str) -> Path:
+    raw = os.fspath(value)
+    path = Path(raw)
+    if not isinstance(raw, str) or not path.is_absolute() or "\x00" in raw:
+        raise VerificationError(f"{label} must be an absolute path")
+    if any(part in {"", ".", ".."} for part in raw.split("/")[1:]):
+        raise VerificationError(f"invalid {label}")
+    return path
+
+
+def _open_absolute_directory(path: Path, label: str) -> int:
+    """Open every component without following symlinks."""
+    parts = path.parts
+    if not parts or parts[0] != "/":
+        raise VerificationError(f"invalid {label}")
+    try:
+        fd = os.open(parts[0], _OPEN_DIR_FLAGS)
+        for part in parts[1:]:
+            nxt = os.open(part, _OPEN_DIR_FLAGS, dir_fd=fd)
+            os.close(fd)
+            fd = nxt
+        return fd
+    except OSError as exc:
+        try:
+            os.close(fd)
+        except (OSError, UnboundLocalError):
+            pass
+        raise VerificationError(f"unsafe {label}") from exc
+
+
+def _require_private_fd(fd: int, label: str) -> None:
+    st = os.fstat(fd)
+    if (
+        not stat.S_ISDIR(st.st_mode)
+        or st.st_uid != os.getuid()
+        or stat.S_IMODE(st.st_mode) != _PRIVATE_MODE
+    ):
+        raise VerificationError(f"unsafe {label}")
+
+
+def _open_private_child(parent_fd: int, name: str, label: str) -> int:
+    try:
+        fd = os.open(name, _OPEN_DIR_FLAGS, dir_fd=parent_fd)
+    except OSError as exc:
+        raise VerificationError(f"unsafe {label}") from exc
+    try:
+        _require_private_fd(fd, label)
+        return fd
+    except BaseException:
+        os.close(fd)
+        raise
+
+
+def _same_inode(left: os.stat_result, right: os.stat_result) -> bool:
+    return (left.st_dev, left.st_ino) == (right.st_dev, right.st_ino)
+
+
+def _remove_created_directory(parent_fd: int, name: str, created: os.stat_result) -> None:
+    """Remove only the still-empty directory inode created by this call."""
+    try:
+        current = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
+        if not _same_inode(current, created):
+            return
+        os.rmdir(name, dir_fd=parent_fd)
+    except OSError:
+        pass
+
+
+def _create_private_child(parent_fd: int, name: str, label: str) -> int:
+    try:
+        os.mkdir(name, _PRIVATE_MODE, dir_fd=parent_fd)
+        created = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
+    except FileExistsError as exc:
+        raise VerificationError(f"existing {label}") from exc
+    except OSError as exc:
+        raise VerificationError(f"cannot create {label}") from exc
+    try:
+        if not stat.S_ISDIR(created.st_mode):
+            raise VerificationError(f"changed {label}")
+        fd = _open_private_child(parent_fd, name, label)
+        # Bind both the opened descriptor and the current name to the inode
+        # observed immediately after mkdir. A cooperating same-UID process is
+        # excluded by the authenticated Issue lease; replacement is fail-closed.
+        named = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
+        opened = os.fstat(fd)
+        if not (_same_inode(created, opened) and _same_inode(created, named)):
+            os.close(fd)
+            raise VerificationError(f"changed {label}")
+        return fd
+    except BaseException:
+        # Never chmod, replace, or remove an entry substituted after creation.
+        _remove_created_directory(parent_fd, name, created)
+        raise
+
+
+def _lease(handle: Path, root: Path, owner: str, repository: str, issue: int) -> None:
+    try:
+        metadata = broker.read_handle_metadata(handle)
+        response = broker.request_handle(handle, "status")
+    except Exception as exc:
+        raise VerificationError("broker lease unavailable") from exc
+    if not isinstance(metadata, dict) or not isinstance(response, dict):
+        raise VerificationError("invalid broker response")
+    if response.get("error") or response.get("status") not in {"held", "repository-held"}:
+        raise VerificationError("broker lease not held")
+    deadline = response.get("deadline")
+    if (
+        isinstance(deadline, bool)
+        or not isinstance(deadline, (int, float))
+        or not math.isfinite(deadline)
+        or deadline <= time.monotonic()
+    ):
+        raise VerificationError("broker lease expired")
+    for source in (metadata, response):
+        if (
+            source.get("root") != str(root)
+            or source.get("owner") != owner
+            or source.get("repository") != repository
+            or source.get("issue") != issue
+        ):
+            raise VerificationError("broker lease binding mismatch")
+    if metadata.get("role") != response.get("role") or response.get("role") not in broker.ROLES:
+        raise VerificationError("broker lease role mismatch")
+
+
+def prepare(
+    root: os.PathLike[str] | str,
+    owner: str,
+    repository: str,
+    issue: int,
+    run_id: str,
+    handle: os.PathLike[str] | str,
+) -> Path:
+    """Authenticate a lease and no-clobber-create one private run directory."""
+    owner = _identifier(owner, "owner", _OWNER_REPOSITORY)
+    repository = _identifier(repository, "repository", _OWNER_REPOSITORY)
+    run_id = _identifier(run_id, "run_id", _RUN_ID)
+    issue = _issue_number(issue)
+    root = _absolute(root, "root")
+    handle = _absolute(handle, "handle")
+    _lease(handle, root, owner, repository, issue)
+
+    issue_path = root / "repositories" / owner / repository / "worktrees" / f"issue-{issue}"
+    issue_fd = _open_absolute_directory(issue_path, "Issue directory")
+    verification_fd = None
+    run_fd = None
+    created_verification_inode = None
+    try:
+        _require_private_fd(issue_fd, "Issue directory")
+        try:
+            verification_fd = _open_private_child(issue_fd, "verification", "verification directory")
+        except VerificationError as exc:
+            try:
+                os.stat("verification", dir_fd=issue_fd, follow_symlinks=False)
+            except FileNotFoundError:
+                verification_fd = _create_private_child(issue_fd, "verification", "verification directory")
+                created_verification_inode = os.fstat(verification_fd)
+            else:
+                raise exc
+
+        try:
+            os.stat(run_id, dir_fd=verification_fd, follow_symlinks=False)
+        except FileNotFoundError:
+            pass
+        except OSError as exc:
+            raise VerificationError("cannot inspect run directory") from exc
+        else:
+            raise VerificationError("existing run directory")
+
+        try:
+            run_fd = _create_private_child(verification_fd, run_id, "run directory")
+        except BaseException:
+            if created_verification_inode is not None:
+                _remove_created_directory(
+                    issue_fd, "verification", created_verification_inode
+                )
+            raise
+        return issue_path / "verification" / run_id
+    finally:
+        for fd in (run_fd, verification_fd, issue_fd):
+            if fd is not None:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="verification_root")
+    sub = parser.add_subparsers(dest="command", required=True)
+    command = sub.add_parser("prepare")
+    command.add_argument("--root", required=True)
+    command.add_argument("--owner", required=True)
+    command.add_argument("--repository", required=True)
+    command.add_argument("--issue", required=True, type=int)
+    command.add_argument("--run-id", required=True)
+    command.add_argument("--handle", required=True)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    try:
+        path = prepare(args.root, args.owner, args.repository, args.issue, args.run_id, args.handle)
+        print(json.dumps({"path": str(path), "verification_mode": "0700", "run_mode": "0700"}, separators=(",", ":")))
+        return 0
+    except (VerificationError, OSError, ValueError) as exc:
+        print(json.dumps({"error": str(exc)[:200]}, separators=(",", ":")), file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

harness/operations/repository-cache.md

@@ -309,7 +309,8 @@ Issue worktree workflow의 완료 보고에는 다음도 포함한다.
 Issue별 검증 실행 기록의 정본은 `<harness-root>/repositories/<owner>/<repo>/worktrees/issue-<number>/verification/<run-id>`이다. `<harness-root>`는 실제로 읽은 `harness/operations/repository-cache.md`에서 결정한다. 제품 source는 형제 `checkout/`에 있고 verification은 제품 Git tree 밖에 있다. 구현·재작업의 회차별 plan, evidence manifest, command output, RED/GREEN 결과, finding matrix와 head별 provenance를 제품 `docs/verification/`에 새로 누적하거나 evidence-only commit을 만들지 않는다. 승인된 제품 요구사항이 명시적으로 요구하는 정본 문서와 지속적인 테스트·synthetic fixture·재현 스크립트는 제품에 유지할 수 있다. 기존 파일을 자동 이동·삭제하거나 광범위 `.gitignore` 변경으로 처리하지 않는다.
 
 - 경로 identity는 인증된 exact owner, repository와 양의 Issue 번호에서 결정한다. 기존 cache identifier 규칙을 따르며 `issue-<number>`는 선행 0 없는 십진수다. `<run-id>`는 `[A-Za-z0-9][A-Za-z0-9_.-]{0,127}` 형식의 새 식별자다. Stage, source revision 또는 실행을 바꾸면 새 run-id를 사용한다. Dot segment, 절대 artifact file 경로, traversal, symlink와 임의 temporary/artifact root는 거부한다. `<owner>/<repo>`를 성공한 lookup 결과에 맞춰 자동 정규화하지 않는다.
-- Caller가 인증된 Issue identity와 cache/worktree admission을 확인하고 **broker Issue lease** 아래 `verification/`과 새 run directory를 만든다. Directory는 실행 UID 소유 mode `0700`, plan·manifest·output 파일은 실행 UID 소유 mode `0600`으로 만들고 즉시 `lstat`으로 확인한다. 기존 cache bootstrap, repository lock과 lease 계약은 바꾸지 않는다. Artifact helper는 lease·Gitea identity·Git registration을 대신 인증하거나 directory를 만들고 cache/권한을 복구하지 않는다.
+- evidence `run` 전에 다음 명령으로 live broker lease와 exact harness root·Issue identity를 인증한다: `python3 harness/lib/verification_root.py prepare --root <harness-root> --owner <owner> --repository <repository> --issue <issue> --run-id <run-id> --handle <handle>`. 이 helper는 `verification/`과 `<run-id>/`를 각각 별도 생성하고, 각 directory가 실행 UID 소유 mode `0700`인지 생성 직후 다시 `lstat`한다. Symlink, owner·mode 불일치와 path collision은 거부하며, 기존 malformed root를 자동 chmod하거나 repair하지 않는다. `Path.mkdir(parents=True, mode=0o700)`는 중간 parent가 umask를 따르므로 이 경로에 사용하지 않는다.
+- Caller가 인증된 Issue identity와 cache/worktree admission을 확인하고 **broker Issue lease** 아래 위 prepare를 실행한다. Directory는 실행 UID 소유 mode `0700`, plan·manifest·output 파일은 실행 UID 소유 mode `0600`으로 만들고 즉시 `lstat`으로 확인한다. 기존 cache bootstrap, repository lock과 lease 계약은 바꾸지 않는다. Artifact helper는 lease·Gitea identity·Git registration을 대신 인증하거나 directory를 만들고 cache/권한을 복구하지 않는다.
 - `repo_root`는 canonical `checkout`, `artifact_root`는 exact `verification/<run-id>`로 분리한다. Plan의 `hash_paths`는 repository-relative source 경로이고 plan/output/manifest 인자는 run-relative artifact 경로다. 두 namespace는 이름이 같아도 서로 다른 파일이다. Source tree의 tracked·untracked·ignored 파일을 함께 검사하고 evidence 경로 exclusion을 허용하지 않는다. Python은 `-B` 또는 allowlist에 선언한 `PYTHONDONTWRITEBYTECODE=1`, pytest는 필요하면 `-p no:cacheprovider`로 checkout 안의 자동 cache 생성을 막는다. 테스트 임시 산출물·dependency environment도 checkout 밖 승인된 경로를 사용한다. Helper가 ignored cache를 자동 삭제하거나 clean 상태로 간주하지 않는다.
 - 새 실행은 `evidence-manifest/v2`로 source full SHA, tracked source blob SHA·byte digest, owner/repository/Issue/run-id, plan path·digest, 실제 argv·exit code·환경·bounded output·시간과 claim pointer를 결합한다. Existing manifest는 no-clobber 원자 게시로 덮어쓰지 않는다. 완료 기록은 불변으로 취급하고 실패·재실행도 새 run-id에 보존한다. Helper의 `validate` 성공은 local artifact 정합성일 뿐 원격 PR head 또는 provider readback 검증을 대체하지 않는다.
 - Baseline, integration, precommit과 remote-head 검증은 각 stage에서 검증한 **그 commit의 clean source**에 결합한다. Dirty 개발 tree는 보존하되 helper가 성공 증거로 게시하지 않는다. Final tracked 제품 산출물을 commit한 뒤 전달할 exact commit에서 검증하고, 실패하면 push하지 않는다. 모든 PR/결과 보고는 같은 immutable record의 command/result를 사용한다.

harness/tests/common/test_managed_lock_broker.py

@@ -226,6 +226,8 @@ class ManagedLockBrokerTests(unittest.TestCase):
             try:
                 result = broker.request_handle(handle, "status")
                 self.assertEqual(result["status"], "held")
+                self.assertEqual(result["root"], str(root))
+                self.assertEqual(broker.read_handle_metadata(handle)["root"], str(root))
                 self.assertNotIn("token", json.dumps(result))
                 cli = subprocess.run([sys.executable, str(BROKER), "request", "--handle", str(handle), "--action", "status"],
                                      capture_output=True, text=True)

harness/tests/common/test_verification_contract_docs.py

@@ -6,6 +6,7 @@ from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[3]
 SCRIPT = ROOT / "harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py"
+PREPARE_SCRIPT = ROOT / "harness/lib/verification_root.py"
 CACHE = ROOT / "harness/operations/repository-cache.md"
 SKILLS = {
     "implementation": ROOT / "harness/agents/impl/skills/gitea-issue-implementation/SKILL.md",
@@ -39,13 +40,26 @@ class VerificationContractDocumentationTests(unittest.TestCase):
             assert match is not None
             block = match.group(1).replace("\\\n", " ")
             commands = [line for line in block.splitlines() if line.startswith("python3 ")]
-            self.assertEqual(len(commands), 2)
+            self.assertEqual(len(commands), 3)
-            for op, command in zip(("run", "validate"), commands):
+            for op, command in zip(("prepare", "run", "validate"), commands):
                 with self.subTest(skill=path.parent.name, op=op):
+                    declared = set(re.findall(r"--[a-z-]+", command))
+                    if op == "prepare":
+                        self.assertIn("harness/lib/verification_root.py", command)
+                        help_result = subprocess.run(
+                            [sys.executable, str(PREPARE_SCRIPT), op, "--help"],
+                            text=True,
+                            capture_output=True,
+                        )
+                        self.assertEqual(help_result.returncode, 0, help_result.stderr)
+                        supported = set(re.findall(r"--[a-z-]+", help_result.stdout))
+                        required = {"--root", "--owner", "--repository", "--issue", "--run-id", "--handle"}
+                        self.assertEqual(declared, required)
+                        self.assertTrue(declared <= supported)
+                    else:
                         help_result = subprocess.run([sys.executable, str(SCRIPT), op, "--help"], text=True, capture_output=True)
                         self.assertEqual(help_result.returncode, 0)
                         supported = set(re.findall(r"--[a-z-]+", help_result.stdout))
-                    declared = set(re.findall(r"--[a-z-]+", command))
                         required = {"--contract-path", "--repo-root", "--artifact-root"} | ({"--plan", "--output"} if op == "run" else {"--manifest"})
                         self.assertEqual(declared, required)
                         self.assertTrue(declared <= supported)
@@ -53,6 +67,14 @@ class VerificationContractDocumentationTests(unittest.TestCase):
                     self.assertIn("validate-legacy", section)
                     self.assertIn("exclusion 없음", section)
 
+    def test_prepare_contract_binds_root_and_forbids_recursive_parent_creation(self):
+        for path in (CACHE, SKILLS["implementation"], SKILLS["rework"]):
+            with self.subTest(path=path.name):
+                text = path.read_text(encoding="utf-8")
+                self.assertIn("exact harness root", text)
+                self.assertIn("Path.mkdir(parents=True, mode=0o700)", text)
+                self.assertIn("자동 chmod", text)
+
     def test_checkout_cleanup_preserves_verification_without_recursive_delete(self):
         text = CACHE.read_text()
         cleanup = text.split("## Worktree 실패 처리와 제거", 1)[1].split("## Credential 안전", 1)[0]

harness/tests/common/test_verification_root.py

@@ -0,0 +1,342 @@
+import contextlib
+import io
+import json
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from harness.lib import verification_root
+
+
+class VerificationRootTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name) / "cache"
+        self.issue = self.root / "repositories" / "alice" / "ledger" / "worktrees" / "issue-7"
+        self.issue.mkdir(parents=True, mode=0o700)
+        for path in (
+            self.root / "repositories",
+            self.root / "repositories/alice",
+            self.root / "repositories/alice/ledger",
+            self.root / "repositories/alice/ledger/worktrees",
+            self.issue,
+        ):
+            os.chmod(path, 0o700)
+        # The trusted harness/NFS root is not required to be role-owned 0700.
+        os.chmod(self.root, 0o770)
+        self.handle = self.root / "runtime" / "handle.json"
+        self.handle.parent.mkdir(mode=0o700)
+        self.meta = {
+            "root": str(self.root),
+            "owner": "alice",
+            "repository": "ledger",
+            "issue": 7,
+            "role": "gitea-impl-agent",
+        }
+        self.status = {
+            "status": "held",
+            "root": str(self.root),
+            "owner": "alice",
+            "repository": "ledger",
+            "issue": 7,
+            "role": "gitea-impl-agent",
+            "deadline": 10**20,
+        }
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    @contextlib.contextmanager
+    def broker(self, *, metadata=None, status=None):
+        with (
+            patch.object(
+                verification_root.broker,
+                "read_handle_metadata",
+                return_value=self.meta if metadata is None else metadata,
+            ),
+            patch.object(
+                verification_root.broker,
+                "request_handle",
+                return_value=self.status if status is None else status,
+            ),
+        ):
+            yield
+
+    def test_umask_does_not_make_verification_parent_world_readable(self):
+        old = os.umask(0o022)
+        try:
+            with self.broker():
+                result = verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        finally:
+            os.umask(old)
+        verification = self.issue / "verification"
+        self.assertEqual(result, verification / "run-1")
+        self.assertEqual(stat.S_IMODE(verification.stat().st_mode), 0o700)
+        self.assertEqual(stat.S_IMODE((verification / "run-1").stat().st_mode), 0o700)
+
+    def test_existing_secure_parent_accepts_a_fresh_run(self):
+        verification = self.issue / "verification"
+        verification.mkdir(mode=0o700)
+        (verification / "prior-run").mkdir(mode=0o700)
+        with self.broker():
+            result = verification_root.prepare(
+                self.root, "alice", "ledger", 7, "new-run", self.handle
+            )
+        self.assertEqual(result, verification / "new-run")
+        self.assertTrue((verification / "prior-run").is_dir())
+        self.assertEqual(stat.S_IMODE(result.stat().st_mode), 0o700)
+
+    def test_bad_lease_is_rejected_before_filesystem_mutation(self):
+        before = list(self.issue.iterdir())
+        with self.broker(status={"error": "expired"}):
+            with self.assertRaises(verification_root.VerificationError):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        self.assertEqual(list(self.issue.iterdir()), before)
+
+    def test_missing_deadline_and_role_mismatch_fail_before_mutation(self):
+        status = dict(self.status)
+        status.pop("deadline")
+        with self.broker(status=status):
+            with self.assertRaisesRegex(verification_root.VerificationError, "expired"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        status = dict(self.status, role="gitea-review-agent")
+        with self.broker(status=status):
+            with self.assertRaisesRegex(verification_root.VerificationError, "role"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        self.assertFalse((self.issue / "verification").exists())
+
+    def test_existing_insecure_verification_is_not_repaired(self):
+        verification = self.issue / "verification"
+        verification.mkdir(mode=0o755)
+        with self.broker():
+            with self.assertRaises(verification_root.VerificationError):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        self.assertEqual(stat.S_IMODE(verification.stat().st_mode), 0o755)
+        self.assertFalse((verification / "run-1").exists())
+
+    def test_insecure_issue_directory_is_rejected_without_repair(self):
+        os.chmod(self.issue, 0o755)
+        with self.broker():
+            with self.assertRaisesRegex(verification_root.VerificationError, "Issue directory"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        self.assertEqual(stat.S_IMODE(self.issue.stat().st_mode), 0o755)
+        self.assertFalse((self.issue / "verification").exists())
+
+    def test_symlink_and_existing_run_are_rejected_without_mutation(self):
+        verification = self.issue / "verification"
+        verification.mkdir(mode=0o700)
+        existing = verification / "run-1"
+        existing.mkdir(mode=0o700)
+        with self.broker():
+            with self.assertRaisesRegex(verification_root.VerificationError, "existing run"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        self.assertTrue(existing.is_dir())
+        existing.rmdir()
+        verification.rmdir()
+        verification.symlink_to(self.issue, target_is_directory=True)
+        with self.broker():
+            with self.assertRaises(verification_root.VerificationError):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-2", self.handle
+                )
+        self.assertTrue(verification.is_symlink())
+
+    def test_binding_and_identifier_validation(self):
+        bad = dict(self.meta, owner="mallory")
+        with self.broker(metadata=bad):
+            with self.assertRaisesRegex(verification_root.VerificationError, "binding"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        wrong_root = dict(self.status, root=str(self.root / "other"))
+        with self.broker(status=wrong_root):
+            with self.assertRaisesRegex(verification_root.VerificationError, "binding"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "run-1", self.handle
+                )
+        for value in ("../x", "a/b", "", ".", "..", ".hidden", "x" * 129):
+            with self.subTest(value=value), self.assertRaises(verification_root.VerificationError):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, value, self.handle
+                )
+
+    def test_real_broker_cli_prepares_canonical_private_root(self):
+        locks = self.root / "repositories/alice/ledger/locks"
+        locks.mkdir(mode=0o700)
+        (locks / "repository.lock").touch(mode=0o600)
+        broker_script = Path(verification_root.broker.__file__)
+        socket_path = self.handle.parent / "broker.sock"
+        command = [
+            sys.executable,
+            str(broker_script),
+            "serve",
+            "--root",
+            str(self.root),
+            "--owner",
+            "alice",
+            "--repository",
+            "ledger",
+            "--issue",
+            "7",
+            "--role",
+            "gitea-impl-agent",
+            "--run-id",
+            "broker-run",
+            "--handle",
+            str(self.handle),
+            "--socket",
+            str(socket_path),
+            "--ttl-seconds",
+            "30",
+            "--acquire-timeout",
+            "1",
+        ]
+        proc = subprocess.Popen(
+            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
+        )
+        assert proc.stdout is not None and proc.stderr is not None
+        try:
+            ready = json.loads(proc.stdout.readline())
+            self.assertEqual(ready["status"], "held")
+            helper = subprocess.run(
+                [
+                    sys.executable,
+                    str(Path(verification_root.__file__)),
+                    "prepare",
+                    "--root",
+                    str(self.root),
+                    "--owner",
+                    "alice",
+                    "--repository",
+                    "ledger",
+                    "--issue",
+                    "7",
+                    "--run-id",
+                    "real-run",
+                    "--handle",
+                    str(self.handle),
+                ],
+                text=True,
+                capture_output=True,
+            )
+            self.assertEqual(helper.returncode, 0, helper.stderr)
+            record = json.loads(helper.stdout)
+            run = Path(record["path"])
+            self.assertEqual(run, self.issue / "verification/real-run")
+            self.assertEqual(stat.S_IMODE(run.parent.stat().st_mode), 0o700)
+            self.assertEqual(stat.S_IMODE(run.stat().st_mode), 0o700)
+            self.assertEqual(
+                verification_root.broker.request_handle(self.handle, "release")["status"],
+                "released",
+            )
+            proc.wait(timeout=3)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait()
+            proc.stdout.close()
+            proc.stderr.close()
+
+    def test_created_inode_replacement_fails_without_removing_replacement(self):
+        verification = self.issue / "verification"
+        verification.mkdir(mode=0o700)
+        real_open = verification_root._open_private_child
+
+        def replace_before_open(parent_fd, name, label):
+            os.rename(name, f"{name}.original", src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
+            os.mkdir(name, 0o700, dir_fd=parent_fd)
+            return real_open(parent_fd, name, label)
+
+        with self.broker(), patch.object(
+            verification_root, "_open_private_child", side_effect=replace_before_open
+        ):
+            with self.assertRaisesRegex(verification_root.VerificationError, "changed run directory"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "race-run", self.handle
+                )
+        self.assertTrue((verification / "race-run").is_dir())
+
+    def test_replaced_new_verification_parent_is_not_removed_on_run_failure(self):
+        real_create = verification_root._create_private_child
+        calls = 0
+
+        def replace_parent_before_run(parent_fd, name, label):
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                return real_create(parent_fd, name, label)
+            os.rename(
+                self.issue / "verification",
+                self.issue / "verification.original",
+            )
+            (self.issue / "verification").mkdir(mode=0o700)
+            raise verification_root.VerificationError("forced run failure")
+
+        with self.broker(), patch.object(
+            verification_root, "_create_private_child", side_effect=replace_parent_before_run
+        ):
+            with self.assertRaisesRegex(verification_root.VerificationError, "forced run failure"):
+                verification_root.prepare(
+                    self.root, "alice", "ledger", 7, "race-parent", self.handle
+                )
+        self.assertTrue((self.issue / "verification").is_dir())
+        self.assertTrue((self.issue / "verification.original").is_dir())
+
+    def test_cli_is_directly_executable_and_emits_verified_modes(self):
+        script = Path(verification_root.__file__)
+        help_result = subprocess.run(
+            [str(script), "prepare", "--help"],
+            text=True,
+            capture_output=True,
+        )
+        self.assertEqual(help_result.returncode, 0, help_result.stderr)
+        for option in ("--root", "--owner", "--repository", "--issue", "--run-id", "--handle"):
+            self.assertIn(option, help_result.stdout)
+
+        output = io.StringIO()
+        with self.broker(), contextlib.redirect_stdout(output):
+            code = verification_root.main(
+                [
+                    "prepare",
+                    "--root",
+                    str(self.root),
+                    "--owner",
+                    "alice",
+                    "--repository",
+                    "ledger",
+                    "--issue",
+                    "7",
+                    "--run-id",
+                    "run-cli",
+                    "--handle",
+                    str(self.handle),
+                ]
+            )
+        self.assertEqual(code, 0)
+        record = json.loads(output.getvalue())
+        self.assertEqual(record["verification_mode"], "0700")
+        self.assertEqual(record["run_mode"], "0700")
+        self.assertEqual(Path(record["path"]), self.issue / "verification/run-cli")
+
+
+if __name__ == "__main__":
+    unittest.main()