feat(spec): add reusable implementation review verification helper

harness/agents/spec/skills/gitea-implementation-spec-review/SKILL.md

@@ -187,6 +187,25 @@ Canonical body를 렌더링하기 전에 trusted workflow directory에 JSON obje
 - Verdict와 native state 조합은 `approve/APPROVED`, `request_changes/REQUEST_CHANGES`, `blocked_spec/COMMENT`만 허용한다.
 - Initial version에서는 inline review comment를 만들지 않고 `comments: []`를 명시한다.
 
+## 반복 기계 검증 helper
+
+반복 실행의 기계 검증은 [scripts/review_support.py](scripts/review_support.py)를 사용한다. Python 표준 라이브러리만 필요하며 다음 호출은 `<harness-root>`에서 `terminal`로 실행한다.
+
+```python
+terminal(command="python3 harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py --help")
+terminal(command="python3 harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py validate-result --result <run>/result.json --allowed-paths <run>/allowed-paths.json")
+terminal(command="python3 harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py render --result <run>/result.json --allowed-paths <run>/allowed-paths.json")
+```
+
+- `<run>`은 `<harness-root>/harness/.runtime/implementation-spec-review/<run-id>`의 trusted workflow directory다. `allowed-paths.json`은 main이 authoritative changed-file set과 pinned base/head의 실제 경로에서 만든 JSON string array이며 결과 artifact가 스스로 선언한 allowlist를 신뢰하지 않는다.
+- `validate-result`, `render`, `parse`는 기존 strict schema, canonical body, digest·correlation과 round-trip을 검증한다. `parse`에는 `--body`와 `--allowed-paths`를 전달한다. 구분자·markup 때문에 모호한 list 항목만 JSON string quoting/escaping으로 원래 값을 보존하며 구분자·markup이 없는 기존 항목의 본문 형식은 유지한다. 출력은 stdout으로 반환하므로 기존 artifact 보존 규칙에 따라 main이 저장한다.
+- `check-cache --contract-path <읽은 repository-cache.md> --owner <owner> --repository <repository> --issue <number>`는 기존 canonical cache의 경로별 permission·symlink만 검사한다. Cache 생성·repair·lock·Git fetch와 bare/remote/ref 확인은 기존 절차에 남는다.
+- `collect_pages`는 1부터 시작하는 page 번호로 main의 read-only callback을 호출하고 반환된 `items`, `total_count`, `has_more`를 검증한다. API adapter는 provider의 실제 metadata와 확인된 effective page size를 이 envelope와 `page_size`에 대응하고 null collection은 count와 종료 근거로 empty임이 증명될 때만 전달한다. 저장된 page envelope 배열은 `check-pages --pages <file> --key <id|filename|sha> [--page-size <number>] [--max-pages <number>]`로 검증한다.
+- `verify-publication --before <file> --after <file> --exact-review <file> --result <file> --allowed-paths <file> --reviewer-id <id> --inline-comments <file> --expected-pr-url <url> [--provider-contract <file>]`는 single native create의 exact review·collection·timeline과 제한된 field projection을 대조한다. `expected_pr_url`은 snapshot에서 자동 추출하지 않고 main이 canonical authenticated access로 따로 확인한 exact PR URL을 제공하며 host·설치 prefix를 포함해 exact-match한다. Timeline은 ID·type·review ID·author·body·timestamp binding과 기존 event 보존을 검증하며 provider 전체 wire schema 확인은 기존 main 절차에 남는다. Snapshot은 `issue`, `issue_comments`, `pr`, `pr_comments`, `reviews`, `timeline`, `files`를 포함한다. Provider 계약을 생략하면 기존 review 변경을 허용하지 않는다. 현재 projection 계약은 `version.version=1.27.3`, boolean `supersession_same_author_only`·`already_dismissed_review_timestamp_projection`이며 main이 서버 버전별 authoritative 근거를 검증한 뒤 제공한다. `official` 정책은 main이 따로 판정하고 확인된 boolean `expected_official`을 계약에 제공한 경우 helper도 대조한다. 동일 `PENDING` ID submit 복구는 기존 절차로 검증하며 이 helper의 single-create 검증으로 대체하지 않는다.
+- `check-unchanged --before <file> --after <file>`는 read-only replay의 exact snapshot 동일성을 확인한다. Helper 성공은 호출자가 실제 write를 하지 않았다는 증명이나 최종 리뷰 완료를 단독으로 의미하지 않는다.
+- Helper는 credential을 발견하거나 network·외부 write를 수행하지 않고 acceptance 의미·evidence 유효성·verdict를 결정하지 않는다. 실제 secret 부재는 main이 별도로 확인하며 알려진 민감값은 Python API의 `sensitive_values`로 메모리 안에서만 전달한다. 입력 오류는 nonzero로 중단하고 입력 본문·credential을 error에 출력하지 않는다.
+- 회귀 검증은 `terminal(command="PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s harness/tests/spec -p 'test_implementation_spec_review_support.py' -v")`로 실행한다. Schema/body/provider 처리 변경 시 같은 schema의 저장된 실제 wire artifact로 render·parse byte 일치와 publication·read-only replay도 검증한다. Helper 검증을 위해 새 외부 write를 하지 않으며 과거 artifact replay를 fresh authenticated readback으로 보고하지 않는다.
+
 ## 절차
 
 1. Exact target identifiers를 검증하고 `operations/repository-cache.md`를 새로 읽는다.
@@ -199,7 +218,7 @@ Canonical body를 렌더링하기 전에 trusted workflow directory에 JSON obje
 7. Cache path·ownership·permission·bare·remote를 검증하고 canonical base repository origin에서 prune/fetch한다. Gitea default branch SHA와 local ref를 일치시키고 PR base commit object가 cache에 존재하는지 검증한다.
 8. Same-repository PR이면 head object도 cache에서 검증하고 exact base-to-head tree를 Gitea changed-file set·diff와 대조한다. Fork PR이면 target cache에 head object가 있다고 가정하지 않고 authoritative head repository·branch SHA와 Gitea changed-file set·diff를 대조한다. Diff 직렬화의 byte 차이만으로 실패시키지 않고 변경 경로와 exact base/head blob 및 hunk가 동일 변경을 나타내는지 검증한다. 임의 ref, checkout, clone 또는 worktree를 만들지 않는다.
 9. Repository lock을 해제하되 Issue lock은 유지한다.
-10. Acceptance matrix 작성 전에 기존 evidence의 source revision과 reviewed head 사이 전체 diff·실행 입력 변경을 확인하고, 필수 검증별 재사용 가능 범위와 현재 head에 새로 결합해야 할 근거를 먼저 요약한다. 재사용 여부는 위 Acceptance matrix의 기존 evidence binding 기준으로 판단하며 새 gate나 일괄 재실행 요구를 추가하지 않는다. 이 초기 요약만으로 검토를 끝내지 않고 Approved Discovery에서 acceptance matrix를 만들고 모든 changed file, 문서, source, test와 exact-head-bound status evidence를 대응시킨다.
+10. Acceptance matrix 작성 전에 기존 evidence의 source revision과 reviewed head 사이 전체 diff·실행 입력 변경을 확인하고, 필수 검증별 재사용 가능 범위와 현재 head에 새로 결합해야 할 근거를 먼저 요약한다. 재사용 여부는 위 Acceptance matrix의 기존 evidence binding 기준으로 판단하며 새 gate나 일괄 재실행 요구를 추가하지 않는다. 재검토에서는 이전 finding별 승인 계약 근거와 최소 보완 요청을 새 구현·evidence의 확인 위치에 대응하고, 해소 또는 잔여 사유를 기존 criterion의 `evidence`와 `reason`에 기록한다. 이미 충족된 검증을 source revision 차이만으로 다시 요구하지 않되, 변경 영향으로 새로 판정할 수 없어진 필수 조건은 그 근거와 최소 보완 범위를 특정한다. 이 초기 요약만으로 검토를 끝내지 않고 Approved Discovery에서 acceptance matrix를 만들고 모든 changed file, 문서, source, test와 exact-head-bound status evidence를 대응시킨다.
 11. 독립 code-review worker 결과가 있으면 exact head에 바인딩됐고 authoritative하게 읽을 수 있을 때만 보조 evidence로 사용한다. 그 결과를 spec verdict로 그대로 복사하지 않는다.
 12. Verdict 규칙으로 하나의 strict result artifact와 canonical review body를 trusted workflow directory에 생성한다.
 13. Exact key/type, duplicate key, marker field, digest, deterministic correlation, criterion cardinality, path binding, verdict/state 조합, credential 부재와 render → parse → render byte equality를 검증한다.

harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py

@@ -0,0 +1,666 @@
+#!/usr/bin/env python3
+"""Pure, standard-library support for implementation-spec review artifacts."""
+from __future__ import annotations
+import argparse, grp, hashlib, json, math, os, pwd, re, stat, sys
+from typing import NoReturn
+from pathlib import Path
+
+SCHEMA = "gitea-agent.implementation-spec-review-result/v1"
+TOP = ("schema", "target", "contract_sha256", "verdict", "criteria", "scope", "conclusion")
+TARGET = ("owner", "repository", "issue", "pull", "head")
+CRITERION = ("criterion", "contract", "implementation_paths", "test_paths", "evidence", "result", "reason")
+SCOPE = ("missing", "excess", "spec_conflicts")
+VERDICTS = {"approve", "request_changes", "blocked_spec"}
+RESULTS = {"satisfied", "unsatisfied", "indeterminate"}
+HEX40 = re.compile(r"^[0-9a-f]{40}$")
+HEX64 = re.compile(r"^[0-9a-f]{64}$")
+IDENT = re.compile(r"^[^\s/\\\x00<>`=@]+$")
+CONTROL = re.compile(r"[\x00-\x1f\x7f-\x9f\u2028\u2029]")
+MARKUP = re.compile(r"[`<>\[\]{}]|(?:^|\s)[#*!|]|[a-z][a-z0-9+.-]*://|www\.", re.I)
+
+class ValidationError(ValueError):
+    pass
+
+def _fail(msg="invalid review artifact") -> NoReturn:
+    raise ValidationError(msg)
+
+def _obj(v, keys, label):
+    if type(v) is not dict or set(v) != set(keys) or len(v) != len(keys): _fail("invalid " + label)
+    return v
+
+def _str(v, label, lo=1, hi=4000, human=False):
+    if type(v) is not str or not lo <= len(v) <= hi: _fail("invalid " + label)
+    if CONTROL.search(v) or v != v.rstrip(): _fail("invalid " + label)
+    try:
+        v.encode("utf-8")
+    except UnicodeError:
+        _fail("invalid " + label)
+    if human and (MARKUP.search(v) or "@" in v): _fail("invalid " + label)
+    return v
+
+def _array(v, label, hi=512, human=False):
+    if type(v) is not list or len(v) > hi: _fail("invalid " + label)
+    out=[]; seen=set()
+    for x in v:
+        _str(x, label + " item", 1, 4000, human)
+        if x in seen: _fail("duplicate " + label)
+        seen.add(x); out.append(x)
+    return out
+
+def loads_json(text):
+    if type(text) is not str: _fail("invalid JSON")
+    def hook(pairs):
+        d={}
+        for k,v in pairs:
+            if k in d: _fail("duplicate JSON key")
+            d[k]=v
+        return d
+    try:
+        return json.loads(text, object_pairs_hook=hook, parse_constant=lambda x: _fail("nonfinite JSON"),
+                          parse_float=lambda x: float(x) if math.isfinite(float(x)) else _fail("nonfinite JSON"))
+    except ValidationError: raise
+    except Exception: _fail("invalid JSON")
+
+def _path(p):
+    _str(p, "path", 1, 1024)
+    if p.startswith("/") or "\\" in p or "\x00" in p: _fail("invalid path")
+    bits=p.split("/")
+    if any(x in ("", ".", "..") for x in bits): _fail("invalid path")
+    return p
+
+def validate_result(result, allowed_paths, sensitive_values=()):
+    _obj(result, TOP, "result schema")
+    if result["schema"] != SCHEMA: _fail("invalid schema")
+    t=_obj(result["target"], TARGET, "target")
+    for k in ("owner", "repository"):
+        _str(t[k], k, 1, 255); 
+        if not IDENT.fullmatch(t[k]) or t[k] in (".", ".."): _fail("invalid target")
+    for k in ("issue", "pull"):
+        if type(t[k]) is not int or t[k] < 1: _fail("invalid target")
+    if type(t["head"]) is not str or not HEX40.fullmatch(t["head"]): _fail("invalid target")
+    if type(result["contract_sha256"]) is not str or not HEX64.fullmatch(result["contract_sha256"]): _fail("invalid contract hash")
+    if type(result["verdict"]) is not str or result["verdict"] not in VERDICTS: _fail("invalid verdict")
+    if type(allowed_paths) not in (list, tuple, set, frozenset): _fail("invalid allowed paths")
+    for path in allowed_paths:
+        _path(path)
+    allowed = set(allowed_paths)
+    if len(allowed) != len(allowed_paths): _fail("duplicate allowed path")
+    criteria=result["criteria"]
+    if type(criteria) is not list or not 1 <= len(criteria) <= 512: _fail("invalid criteria")
+    ids=set()
+    for c in criteria:
+        _obj(c, CRITERION, "criterion")
+        _str(c["criterion"], "criterion", 1, 200, True)
+        if c["criterion"] in ids: _fail("duplicate criterion")
+        ids.add(c["criterion"])
+        for k in ("contract", "evidence", "reason"): _str(c[k], k, 1, 4000, True)
+        for k in ("implementation_paths", "test_paths"):
+            for p in _array(c[k], k):
+                _path(p)
+                if p not in allowed: _fail("path outside allowed set")
+        if type(c["result"]) is not str or c["result"] not in RESULTS: _fail("invalid criterion result")
+    s=_obj(result["scope"], SCOPE, "scope")
+    for k in SCOPE: _array(s[k], k, 512, True)
+    _str(result["conclusion"], "conclusion", 1, 4000, True)
+    human = [result["conclusion"]] + [c[k] for c in criteria for k in ("criterion", "contract", "evidence", "reason")]
+    human += [x for k in SCOPE for x in s[k]]
+    for secret in sensitive_values:
+        if type(secret) is not str: _fail("invalid sensitive values")
+        if secret and any(secret in value for value in human):
+            _fail("sensitive value in human data")
+    states={c["result"] for c in criteria}
+    if result["verdict"] == "approve" and (states != {"satisfied"} or any(s[k] for k in SCOPE)): _fail("verdict contradiction")
+    if result["verdict"] == "request_changes" and not ({"unsatisfied", "indeterminate"} & states): _fail("verdict contradiction")
+    if result["verdict"] == "blocked_spec" and not (s["spec_conflicts"] or "indeterminate" in states): _fail("verdict contradiction")
+    return result
+
+def _sha(b): return hashlib.sha256(b).hexdigest()
+def _human_paths(xs):
+    if not xs:
+        return "없음"
+    # Quote complete list elements only when plain comma-separated prose is ambiguous.
+    unsafe = "<>@`[]{}*|~"
+    if any(", " in x or "; " in x or x == "없음" or '"' in x or any(c in x for c in unsafe) for x in xs):
+        def quoted(value):
+            text = json.dumps(value, ensure_ascii=False)
+            for char in unsafe:
+                text = text.replace(char, "\\u%04x" % ord(char))
+            return text
+        return ", ".join(quoted(x) for x in xs)
+    return ", ".join(xs)
+
+
+def _parse_list(text):
+    if text == "없음":
+        return []
+    if not text.startswith('"'):
+        return text.split(", ")
+    decoder = json.JSONDecoder()
+    out = []
+    while text:
+        try:
+            value, end = decoder.raw_decode(text)
+        except ValueError:
+            _fail("invalid list encoding")
+        if type(value) is not str:
+            _fail("invalid list encoding")
+        out.append(value)
+        text = text[end:]
+        if text:
+            if not text.startswith(", "):
+                _fail("invalid list encoding")
+            text = text[2:]
+            if not text:
+                _fail("invalid list encoding")
+    return out
+
+
+def _parse_evidence(text):
+    # Find the first path/evidence separator outside quoted path elements.
+    quoted = escaped = False
+    for i, char in enumerate(text):
+        if quoted:
+            if escaped:
+                escaped = False
+            elif char == "\\":
+                escaped = True
+            elif char == '"':
+                quoted = False
+        elif char == '"':
+            quoted = True
+        elif text[i:i + 2] == "; ":
+            return _parse_list(text[:i]), text[i + 2:]
+    _fail("invalid criterion block")
+
+def _marker(r, digest):
+    t=r["target"]
+    raw="\x00".join([t["owner"],t["repository"],str(t["issue"]),str(t["pull"]),t["head"],r["contract_sha256"],digest]).encode()
+    corr=_sha(raw)
+    return ("<!-- gitea-agent:implementation-spec-review/v1 owner=%s repository=%s issue=%d pr=%d head=%s contract-sha256=%s result-sha256=%s correlation=%s -->" %
+            (t["owner"],t["repository"],t["issue"],t["pull"],t["head"],r["contract_sha256"],digest,corr))
+
+def render_review(result, allowed_paths, sensitive_values=()):
+    r=validate_result(result, allowed_paths, sensitive_values); t=r["target"]
+    lines=["## 구현 명세 적합성 검토", "", f"- Source Issue: #{t['issue']}", f"- Pull Request: #{t['pull']}",
+           f"- Reviewed head: `{t['head']}`", f"- Contract SHA-256: `{r['contract_sha256']}`", f"- Verdict: `{r['verdict']}`", "",
+           "### Acceptance criteria 판정", ""]
+    for c in r["criteria"]:
+        lines += [f"- `{c['criterion']}` — `{c['result']}`", f"  - 계약: {c['contract']}",
+                  f"  - 구현 근거: {_human_paths(c['implementation_paths'])}; {c['evidence']}", f"  - 테스트 근거: {_human_paths(c['test_paths'])}",
+                  f"  - 판정 근거: {c['reason']}", ""]
+    s=r["scope"]
+    lines += ["### 범위 판정", "", f"- 누락: {_human_paths(s['missing'])}", f"- 범위 초과: {_human_paths(s['excess'])}",
+              f"- 명세 모순: {_human_paths(s['spec_conflicts'])}", "", "### 결론", "", r["conclusion"], ""]
+    body="\n".join(lines)
+    return body + "\n" + _marker(r, _sha(body.encode("utf-8"))) + "\n"
+
+def parse_review(body, allowed_paths, sensitive_values=()):
+    if type(body) is not str or not body.endswith("\n"): _fail("invalid review body")
+    lines=body.splitlines()
+    if not lines or not lines[-1].startswith("<!-- gitea-agent:implementation-spec-review/v1 ") or not lines[-1].endswith(" -->"): _fail("invalid review marker")
+    marker=lines[-1]; m=re.fullmatch(r"<!-- gitea-agent:implementation-spec-review/v1 owner=([^ ]+) repository=([^ ]+) issue=(\d+) pr=(\d+) head=([0-9a-f]{40}) contract-sha256=([0-9a-f]{64}) result-sha256=([0-9a-f]{64}) correlation=([0-9a-f]{64}) -->", marker)
+    if not m: _fail("invalid review marker")
+    # Parse the stable prose emitted by render_review, then validate by re-rendering.
+    try:
+        if lines[0] != "## 구현 명세 적합성 검토" or lines[8] != "### Acceptance criteria 판정": _fail("invalid review body")
+        issue=int(lines[2][len("- Source Issue: #") :]); pull=int(lines[3][len("- Pull Request: #"):])
+        head=lines[4][len("- Reviewed head: `"):-1]; contract=lines[5][len("- Contract SHA-256: `"):-1]; verdict=lines[6][len("- Verdict: `"):-1]
+        i=9; criteria=[]
+        while i < len(lines) and lines[i] != "### 범위 판정":
+            if not lines[i]: i+=1; continue
+            cm=re.fullmatch(r"- `([^`]+)` — `([^`]+)`", lines[i]);
+            if not cm: _fail("invalid criterion block")
+            c={"criterion":cm.group(1),"result":cm.group(2)}
+            for key,prefix in (("contract","  - 계약: "),("implementation_paths","  - 구현 근거: "),("test_paths","  - 테스트 근거: "),("reason","  - 판정 근거: ")):
+                i+=1
+                if i>=len(lines) or not lines[i].startswith(prefix): _fail("invalid criterion block")
+                val=lines[i][len(prefix):]
+                if key == "implementation_paths":
+                    c[key], c["evidence"] = _parse_evidence(val)
+                else:
+                    c[key] = _parse_list(val) if key.endswith("paths") else val
+            c={"criterion":c["criterion"],"contract":c["contract"],"implementation_paths":c["implementation_paths"],"test_paths":c["test_paths"],"evidence":c["evidence"],"result":c["result"],"reason":c["reason"]}
+            criteria.append(c); i+=1
+        if i+6 >= len(lines) or lines[i+1] != "": _fail("invalid scope block")
+        scope={}; i += 1
+        for key,prefix in (("missing","- 누락: "),("excess","- 범위 초과: "),("spec_conflicts","- 명세 모순: ")):
+            i += 1
+            if not lines[i].startswith(prefix): _fail("invalid scope block")
+            scope[key] = _parse_list(lines[i][len(prefix):])
+        i+=2
+        if lines[i] != "### 결론" or lines[i+1] != "": _fail("invalid conclusion")
+        conclusion=lines[i+2]
+    except (IndexError, ValueError): _fail("invalid review body")
+    r={"schema":SCHEMA,"target":{"owner":m.group(1),"repository":m.group(2),"issue":issue,"pull":pull,"head":head},"contract_sha256":contract,"verdict":verdict,"criteria":criteria,"scope":scope,"conclusion":conclusion}
+    validate_result(r, allowed_paths, sensitive_values)
+    canonical=render_review(r, allowed_paths, sensitive_values)
+    if canonical != body or _sha(canonical[:canonical.rfind("<!--") - 1].encode()) != m.group(7): _fail("review digest mismatch")
+    if _marker(r,m.group(7)) != marker: _fail("review marker mismatch")
+    return r
+
+def _lstat(path, kind=None):
+    try: st=os.lstat(path)
+    except OSError: _fail("cache path unavailable")
+    if stat.S_ISLNK(st.st_mode): _fail("cache symlink")
+    if kind and not kind(st.st_mode): _fail("cache path type")
+    return st
+
+def _safe_name(x):
+    if type(x) is not str or not x or x in (".", "..") or "/" in x or "\\" in x or CONTROL.search(x):
+        _fail("invalid cache identifier")
+
+
+def _role_group_ok(st, uid):
+    if not (st.st_mode & stat.S_IWGRP):
+        return True
+    try:
+        group = grp.getgrgid(st.st_gid)
+        role = pwd.getpwuid(uid)
+        accounts = pwd.getpwall()
+        if group.gr_gid != st.st_gid or role.pw_uid != uid or not accounts:
+            return False
+        members = set(group.gr_mem)
+        members.update(account.pw_name for account in accounts if account.pw_gid == st.st_gid)
+        if role.pw_gid == st.st_gid:
+            members.add(role.pw_name)
+        return members == {role.pw_name}
+    except (KeyError, OSError):
+        return False
+
+
+def _optional_stat(path, kind):
+    try:
+        os.lstat(path)
+    except FileNotFoundError:
+        return None
+    except OSError:
+        _fail("cache path unavailable")
+    return _lstat(path, kind)
+
+
+def validate_cache_paths(contract_path, owner, repository, issue, uid=None):
+    """Inspect existing paths only; acquire locks and verify Git separately."""
+    _safe_name(owner)
+    _safe_name(repository)
+    if type(issue) is not int or issue < 1:
+        _fail("invalid cache issue")
+    uid = os.getuid() if uid is None else uid
+    if type(uid) is not int or uid < 0:
+        _fail("invalid role UID")
+    try:
+        raw = os.fspath(contract_path)
+    except TypeError:
+        _fail("invalid cache contract path")
+    if type(raw) is not str or not raw.startswith("/") or any(x in (".", "..", "") for x in raw.split("/")[1:]):
+        _fail("invalid cache contract path")
+    cp = Path(raw)
+    if tuple(cp.parts[-3:]) != ("harness", "operations", "repository-cache.md"):
+        _fail("invalid cache contract path")
+    root = cp.parent.parent.parent
+    for ancestor in reversed(cp.parents):
+        _lstat(ancestor, stat.S_ISDIR)
+    _lstat(cp, stat.S_ISREG)
+    # NFS permits a different root owner, but never a world-writable root.
+    if os.lstat(root).st_mode & stat.S_IWOTH:
+        _fail("invalid harness root")
+    rs = root / "repositories"
+    rst = _lstat(rs, stat.S_ISDIR)
+    if rst.st_uid != uid or rst.st_mode & stat.S_IWOTH or not _role_group_ok(rst, uid):
+        _fail("invalid repositories directory")
+    base = rs / owner / repository
+    for path in (rs / owner, base, base / "locks"):
+        st = _lstat(path, stat.S_ISDIR)
+        if st.st_uid != uid or stat.S_IMODE(st.st_mode) != 0o700:
+            _fail("invalid cache directory")
+    git = base / "repository.git"
+    gst = _lstat(git, stat.S_ISDIR)
+    if gst.st_uid != uid or gst.st_mode & stat.S_IWOTH:
+        _fail("invalid repository cache")
+    # Bootstrap paths are optional for an already-existing cache.
+    bootstrap_root = rs / ".locks"
+    if _optional_stat(bootstrap_root, stat.S_ISDIR) is not None:
+        for path in (bootstrap_root, bootstrap_root / owner):
+            st = _optional_stat(path, stat.S_ISDIR)
+            if st is not None and (st.st_uid != uid or stat.S_IMODE(st.st_mode) != 0o700):
+                _fail("invalid bootstrap directory")
+        bootstrap = bootstrap_root / owner / (repository + ".bootstrap.lock")
+        st = _optional_stat(bootstrap, stat.S_ISREG)
+        if st is not None and (st.st_uid != uid or stat.S_IMODE(st.st_mode) != 0o600):
+            _fail("invalid bootstrap lock")
+    out = {"root": str(root), "repositories": str(rs), "owner": str(rs / owner),
+           "repository": str(base), "repository_git": str(git), "locks": str(base / "locks")}
+    for name in (f"issue-{issue}.lock", "repository.lock"):
+        path = base / "locks" / name
+        st = _optional_stat(path, stat.S_ISREG)
+        if st is not None:
+            if st.st_uid != uid or stat.S_IMODE(st.st_mode) != 0o600:
+                _fail("invalid lock file")
+            out[name] = str(path)
+    return out
+
+def collect_pages(fetch_page, key, page_size=50, max_pages=10000):
+    """Consume provider-verified envelopes with one-based page numbers."""
+    if key not in ("id", "filename", "sha") or type(key) is not str:
+        _fail("invalid identity key")
+    if type(page_size) is not int or page_size < 1 or type(max_pages) is not int or max_pages < 1:
+        _fail("invalid pagination limits")
+    all_items, seen, total = [], set(), None
+    for page in range(1, max_pages + 1):
+        try:
+            env = fetch_page(page)
+        except Exception:
+            _fail("page read failed")
+        if type(env) is not dict or "items" not in env or set(env) - {"items", "total_count", "has_more"}:
+            _fail("invalid page envelope")
+        declared, more = env.get("total_count"), env.get("has_more")
+        if declared is not None and (type(declared) is not int or declared < 0):
+            _fail("invalid page count")
+        if more is not None and type(more) is not bool:
+            _fail("invalid page metadata")
+        if declared is not None:
+            if total is not None and total != declared:
+                _fail("conflicting page count")
+            total = declared
+        items = env["items"]
+        if items is None and declared == 0 and more is False:
+            items = []
+        if type(items) is not list or len(items) > page_size:
+            _fail("invalid page items")
+        for item in items:
+            if type(item) is not dict or key not in item:
+                _fail("invalid page identity")
+            ident = item[key]
+            if key == "id":
+                _positive_id(ident)
+            elif key == "filename":
+                _path(ident)
+            elif type(ident) is not str or not HEX40.fullmatch(ident):
+                _fail("invalid page identity")
+            if ident in seen:
+                _fail("duplicate page identity")
+            seen.add(ident)
+            all_items.append(item)
+        if total is not None and len(all_items) > total:
+            _fail("page count exceeded")
+        if more is True:
+            if len(items) != page_size or (total is not None and len(all_items) >= total):
+                _fail("truncated or inconsistent continuation")
+            continue
+        done = more is False or (more is None and (
+            (total is not None and len(all_items) == total) or len(items) < page_size))
+        if done:
+            if total is not None and len(all_items) != total:
+                _fail("truncated pages")
+            return all_items
+    _fail("pagination limit reached")
+
+def _json_bytes(value):
+    def check(item):
+        if type(item) is dict:
+            if any(type(key) is not str for key in item):
+                _fail("invalid JSON object")
+            for child in item.values():
+                check(child)
+        elif type(item) is list:
+            for child in item:
+                check(child)
+        elif item is not None and type(item) not in (str, int, float, bool):
+            _fail("invalid JSON value")
+    check(value)
+    try:
+        return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False,
+                          separators=(",", ":")).encode("utf-8")
+    except (TypeError, ValueError, UnicodeError):
+        _fail("invalid JSON value")
+
+
+def _same(a, b):
+    return _json_bytes(a) == _json_bytes(b)
+
+
+def verify_unchanged(before, after):
+    if not _same(before, after):
+        _fail("snapshot changed")
+    return True
+
+
+def _positive_id(value):
+    if type(value) is not int or value < 1:
+        _fail("invalid identity")
+    return value
+
+
+def _user_id(record):
+    if type(record) is not dict or type(record.get("user")) is not dict:
+        _fail("invalid author identity")
+    return _positive_id(record["user"].get("id"))
+
+
+def _index(collection, key="id"):
+    if type(collection) is not list:
+        _fail("invalid collection")
+    out = {}
+    for item in collection:
+        if type(item) is not dict or key not in item:
+            _fail("invalid collection identity")
+        ident = item[key]
+        if key == "id":
+            _positive_id(ident)
+        elif key == "filename":
+            _path(ident)
+            _str(item.get("status"), "file status", 1, 100)
+        if ident in out:
+            _fail("duplicate collection identity")
+        out[ident] = item
+    return out
+
+
+def _timestamp(value):
+    from datetime import datetime
+    if type(value) is not str or not re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})", value):
+        _fail("invalid review timestamp")
+    try:
+        datetime.fromisoformat(value)
+    except ValueError:
+        _fail("invalid review timestamp")
+    return value
+
+
+def _without(value, fields):
+    return {key: child for key, child in value.items() if key not in fields}
+
+
+def _target_snapshot(snapshot, target):
+    fields = ("issue", "issue_comments", "pr", "pr_comments", "reviews", "timeline", "files")
+    _obj(snapshot, fields, "publication snapshot")
+    issue, pr = snapshot["issue"], snapshot["pr"]
+    if type(issue) is not dict or type(pr) is not dict:
+        _fail("invalid source snapshot")
+    if _positive_id(issue.get("number")) != target["issue"] or _positive_id(pr.get("number")) != target["pull"]:
+        _fail("snapshot target mismatch")
+    fullname = target["owner"] + "/" + target["repository"]
+    if type(issue.get("repository")) is not dict or issue["repository"].get("full_name") != fullname:
+        _fail("snapshot repository mismatch")
+    if type(pr.get("base")) is not dict or type(pr["base"].get("repo")) is not dict or pr["base"]["repo"].get("full_name") != fullname:
+        _fail("snapshot repository mismatch")
+    if type(pr.get("head")) is not dict or pr["head"].get("sha") != target["head"]:
+        _fail("snapshot head mismatch")
+    if pr.get("state") != "open" or pr.get("merged") is not False:
+        _fail("snapshot PR state mismatch")
+    from urllib.parse import urlsplit
+    url = pr.get("html_url")
+    if type(url) is not str:
+        _fail("invalid PR URL")
+    try:
+        parts = urlsplit(url)
+    except ValueError:
+        _fail("invalid PR URL")
+    if parts.scheme not in ("https", "http") or not parts.hostname or parts.username or parts.password or parts.query or parts.fragment or not parts.path.endswith("/" + fullname + "/pulls/" + str(target["pull"])):
+        _fail("snapshot PR URL mismatch")
+    for key in ("issue_comments", "pr_comments", "reviews", "timeline"):
+        _index(snapshot[key])
+    _index(snapshot["files"], "filename")
+    return url
+
+
+def _projection_contract(contract):
+    if contract is None:
+        return False, False, None
+    if type(contract) is not dict or type(contract.get("version")) is not dict or contract["version"].get("version") != "1.27.3":
+        _fail("unsupported provider projection contract")
+    flags = []
+    for key in ("supersession_same_author_only", "already_dismissed_review_timestamp_projection"):
+        value = contract.get(key, False)
+        if type(value) is not bool:
+            _fail("invalid provider projection contract")
+        flags.append(value)
+    official = contract.get("expected_official")
+    if official is not None and type(official) is not bool:
+        _fail("invalid provider review policy")
+    return flags[0], flags[1], official
+
+
+def verify_publication(before, after, exact_review, result, allowed_paths, reviewer_id, inline_comments, provider_contract=None, *, expected_pr_url):
+    """Verify saved readbacks of one native create; never authenticate or publish."""
+    validate_result(result, allowed_paths)
+    _positive_id(reviewer_id)
+    if type(inline_comments) is not list or inline_comments:
+        _fail("inline comments are not supported")
+    target = result["target"]
+    _str(expected_pr_url, "canonical PR URL", 1, 4000)
+    url = _target_snapshot(before, target)
+    if url != expected_pr_url or _target_snapshot(after, target) != expected_pr_url:
+        _fail("snapshot PR URL mismatch")
+    if _user_id(before["pr"]) == reviewer_id:
+        _fail("reviewer is PR author")
+    supersede, retouch, expected_official = _projection_contract(provider_contract)
+    old_reviews, reviews = _index(before["reviews"]), _index(after["reviews"])
+    if type(exact_review) is not dict:
+        _fail("invalid exact review")
+    rid = _positive_id(exact_review.get("id"))
+    if set(reviews) - set(old_reviews) != {rid} or set(old_reviews) - set(reviews):
+        _fail("review cardinality mismatch")
+    if [x for x in reviews if x != rid] != list(old_reviews):
+        _fail("previous review order changed")
+    er = reviews[rid]
+    if not _same(er, exact_review):
+        _fail("exact review readback mismatch")
+    expected = {"approve": "APPROVED", "request_changes": "REQUEST_CHANGES", "blocked_spec": "COMMENT"}[result["verdict"]]
+    if er.get("body") != render_review(result, allowed_paths) or er.get("commit_id") != target["head"] or er.get("state") != expected:
+        _fail("review binding mismatch")
+    if _user_id(er) != reviewer_id or er.get("pull_request_url") != url:
+        _fail("review identity mismatch")
+    if er.get("stale") is not False or er.get("dismissed") is not False or type(er.get("official")) is not bool:
+        _fail("review status mismatch")
+    if type(er.get("comments_count")) is not int or er["comments_count"] != 0:
+        _fail("review comment count mismatch")
+    if expected_official is not None and er["official"] != expected_official:
+        _fail("review policy mismatch")
+    submitted = _timestamp(er.get("submitted_at"))
+    _timestamp(er.get("updated_at"))
+    old_events, events = _index(before["timeline"]), _index(after["timeline"])
+    additions = set(events) - set(old_events)
+    if len(additions) != 1 or set(old_events) - set(events) or not _same(after["timeline"][:len(before["timeline"])], before["timeline"]):
+        _fail("timeline cardinality or preservation mismatch")
+    event = after["timeline"][-1]
+    if event["id"] not in additions or _positive_id(event.get("review_id")) != rid or event.get("type") != "review":
+        _fail("timeline review binding mismatch")
+    if _user_id(event) != reviewer_id or event.get("body") != er["body"] or event.get("pull_request_url") != url:
+        _fail("timeline author or body mismatch")
+    created = _timestamp(event.get("created_at"))
+    event_updated = _timestamp(event.get("updated_at"))
+    # Bind the event by exact record ID/author/body, not coincident wall clocks.
+    causal_times = {submitted, created, event_updated}
+    for section in ("issue", "pr"):
+        b, a = before[section], after[section]
+        permitted = {"updated_at"} if section == "issue" else {"updated_at", "review_comments", "requested_reviewers"}
+        if not _same(_without(b, permitted), _without(a, permitted)) or set(b) != set(a):
+            _fail("source or PR drift")
+        if not _same(b.get("updated_at"), a.get("updated_at")) and a.get("updated_at") not in causal_times:
+            _fail("timestamp projection mismatch")
+    bpr, apr = before["pr"], after["pr"]
+    old_count, count = bpr.get("review_comments"), apr.get("review_comments")
+    if type(old_count) is not int or old_count < 0 or type(count) is not int or count != old_count + 1:
+        _fail("review count projection mismatch")
+    old_requested = _index(bpr.get("requested_reviewers"))
+    requested = _index(apr.get("requested_reviewers"))
+    if not _same(bpr["requested_reviewers"], apr["requested_reviewers"]):
+        if reviewer_id in old_requested or set(requested) - set(old_requested) != {reviewer_id} or set(old_requested) - set(requested):
+            _fail("requested reviewers projection mismatch")
+        if not _same([x for x in apr["requested_reviewers"] if x["id"] != reviewer_id], bpr["requested_reviewers"]):
+            _fail("previous requested reviewers changed")
+    for collection in ("issue_comments", "pr_comments"):
+        if not _same(before[collection], after[collection]):
+            _fail("comment collection changed")
+    if not _same(_index(before["files"], "filename"), _index(after["files"], "filename")):
+        _fail("changed file set drift")
+    projected = []
+    for old_id, old in old_reviews.items():
+        current = reviews[old_id]
+        if _same(old, current):
+            continue
+        if _user_id(old) != reviewer_id or old.get("pull_request_url") != url or old.get("state") not in ("APPROVED", "REQUEST_CHANGES"):
+            _fail("previous review drift")
+        fields = {"official", "dismissed", "updated_at"}
+        if set(old) != set(current) or not _same(_without(old, fields), _without(current, fields)) or current.get("updated_at") not in causal_times:
+            _fail("previous review drift")
+        superseded = supersede and old.get("official") is True and old.get("dismissed") is False
+        retouched = retouch and old.get("official") is False and old.get("dismissed") is True
+        if not (superseded or retouched) or current.get("official") is not False or current.get("dismissed") is not True:
+            _fail("previous review projection not authorized")
+        projected.append(old_id)
+    return {"review_id": rid, "state": expected, "review_records_added": 1,
+            "timeline_events_added": 1, "projected_review_ids": projected, "read_only": True}
+
+def _read(path):
+    try:
+        with open(path, encoding="utf-8", newline="") as stream:
+            return stream.read()
+    except (OSError, ValueError, UnicodeError):
+        _fail("input unavailable")
+def _json_file(path): return loads_json(_read(path))
+def _emit_error(exc):
+    sys.stderr.write(json.dumps({"error":"invalid input"}, ensure_ascii=False)+"\n"); return 2
+
+class _Parser(argparse.ArgumentParser):
+    def error(self, message):
+        _fail("invalid command arguments")
+
+
+def main(argv=None):
+    p=_Parser(prog="review_support", add_help=True); sub=p.add_subparsers(dest="cmd",required=True)
+    def common(name):
+        q=sub.add_parser(name); q.add_argument("--result",required=True); q.add_argument("--allowed-paths",required=True); return q
+    common("validate-result"); common("render")
+    q=sub.add_parser("parse"); q.add_argument("--body",required=True); q.add_argument("--allowed-paths",required=True)
+    q=sub.add_parser("check-cache"); q.add_argument("--contract-path",required=True); q.add_argument("--owner",required=True); q.add_argument("--repository",required=True); q.add_argument("--issue",required=True,type=int)
+    q=sub.add_parser("check-pages"); q.add_argument("--pages",required=True); q.add_argument("--key",required=True)
+    q.add_argument("--page-size", type=int, default=50); q.add_argument("--max-pages", type=int, default=10000)
+    q=sub.add_parser("check-unchanged"); q.add_argument("--before",required=True); q.add_argument("--after",required=True)
+    q=sub.add_parser("verify-publication");
+    for n in ("before","after","exact-review","result","allowed-paths","inline-comments"): q.add_argument("--"+n,required=True)
+    q.add_argument("--reviewer-id",required=True,type=int); q.add_argument("--provider-contract")
+    q.add_argument("--expected-pr-url", required=True)
+    try:
+        a=p.parse_args(argv)
+        if a.cmd in ("validate-result","render"): r=_json_file(a.result); paths=_json_file(a.allowed_paths); validate_result(r,paths); out=r if a.cmd=="validate-result" else render_review(r,paths); print(json.dumps(out,ensure_ascii=False,sort_keys=True) if a.cmd=="validate-result" else out,end="" if a.cmd=="render" else "\n")
+        elif a.cmd=="parse": print(json.dumps(parse_review(_read(a.body),_json_file(a.allowed_paths)),ensure_ascii=False,sort_keys=True))
+        elif a.cmd=="check-cache": print(json.dumps(validate_cache_paths(a.contract_path,a.owner,a.repository,a.issue),sort_keys=True))
+        elif a.cmd=="check-pages":
+            pages=_json_file(a.pages)
+            if type(pages) is not list: _fail("invalid pages")
+            used = []
+            def f(i):
+                used.append(i)
+                return pages[i - 1]
+            items = collect_pages(f, a.key, a.page_size, a.max_pages)
+            if len(used) != len(pages): _fail("extra saved pages")
+            print(json.dumps(items,ensure_ascii=False,sort_keys=True))
+        elif a.cmd=="check-unchanged": print(json.dumps({"unchanged":verify_unchanged(_json_file(a.before),_json_file(a.after))}))
+        else:
+            print(json.dumps(verify_publication(_json_file(a.before),_json_file(a.after),_json_file(a.exact_review),_json_file(a.result),_json_file(a.allowed_paths),a.reviewer_id,_json_file(a.inline_comments),_json_file(a.provider_contract) if a.provider_contract else None, expected_pr_url=a.expected_pr_url),sort_keys=True))
+        return 0
+    except (ValidationError, OSError, ValueError, TypeError): return _emit_error(None)
+
+if __name__ == "__main__": sys.exit(main())

harness/tests/spec/test_implementation_spec_review_support.py

@@ -0,0 +1,462 @@
+import copy
+import importlib.util
+import json
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+MODPATH = ROOT / "harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py"
+spec = importlib.util.spec_from_file_location("review_support", MODPATH)
+mod = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(mod)
+
+
+def result(verdict="approve"):
+    return {
+        "schema": "gitea-agent.implementation-spec-review-result/v1",
+        "target": {"owner": "acme", "repository": "repo", "issue": 3, "pull": 7,
+                   "head": "a" * 40},
+        "contract_sha256": "b" * 64,
+        "verdict": verdict,
+        "criteria": [{"criterion": "REQ-1", "contract": "계약", "implementation_paths": ["src/a.py"],
+                      "test_paths": ["tests/test_a.py"], "evidence": "증거", "result": "satisfied",
+                      "reason": "충족"}],
+        "scope": {"missing": [], "excess": [], "spec_conflicts": []},
+        "conclusion": "모든 기준을 충족했다.",
+    }
+
+
+def publication_fixture():
+    r = result(); paths = {"src/a.py", "tests/test_a.py"}
+    body = mod.render_review(r, paths)
+    url = "https://example.test/acme/repo/pulls/7"
+    user = {"id": 42, "login": "spec-agent"}
+    review = {"id": 11, "body": body, "commit_id": r["target"]["head"], "state": "APPROVED",
+              "user": user, "official": True, "stale": False, "dismissed": False,
+              "comments_count": 0, "submitted_at": "2026-10-03T01:24:00Z",
+              "updated_at": "2026-10-03T01:24:00Z", "pull_request_url": url}
+    before = {"issue": {"id": 3, "number": 3, "repository": {"full_name": "acme/repo"},
+                        "updated_at": "2026-10-03T00:00:00Z", "title": "contract"},
+              "issue_comments": [], "pr": {"id": 7, "number": 7, "html_url": url,
+              "url": url, "base": {"repo": {"full_name": "acme/repo"}, "sha": "c" * 40},
+              "head": {"sha": "a" * 40}, "user": {"id": 99}, "state": "open", "merged": False,
+              "review_comments": 0, "updated_at": "2026-10-03T00:00:00Z", "requested_reviewers": []},
+              "pr_comments": [], "reviews": [], "timeline": [],
+              "files": [{"filename": "src/a.py", "status": "modified", "additions": 1}]}
+    after = copy.deepcopy(before); after["reviews"] = [review]
+    after["pr"]["updated_at"] = review["submitted_at"]; after["pr"]["review_comments"] = 1
+    after["timeline"] = [{"id": 101, "review_id": 11, "user": user, "body": body,
+                          "type": "review", "created_at": review["submitted_at"],
+                          "updated_at": review["submitted_at"], "pull_request_url": url}]
+    return before, after, review, r, paths
+
+
+def verify_fixture(before, after, review, r, paths, contract=None):
+    return mod.verify_publication(before, after, review, r, paths, 42, [], contract,
+                                  expected_pr_url="https://example.test/acme/repo/pulls/7")
+
+
+class SupportTests(unittest.TestCase):
+    def test_loads_json_rejects_duplicate_and_nonfinite(self):
+        with self.assertRaises(mod.ValidationError):
+            mod.loads_json('{"a":1,"a":2}')
+        with self.assertRaises(mod.ValidationError):
+            mod.loads_json('{"a":NaN}')
+
+    def test_validate_and_render_roundtrip(self):
+        r = result()
+        self.assertEqual(mod.validate_result(r, {"src/a.py", "tests/test_a.py"}), r)
+        body = mod.render_review(r, {"src/a.py", "tests/test_a.py"})
+        parsed = mod.parse_review(body, {"src/a.py", "tests/test_a.py"})
+        self.assertEqual(parsed, r)
+        self.assertEqual(mod.render_review(parsed, {"src/a.py", "tests/test_a.py"}), body)
+
+    def test_validation_rejects_bool_path_traversal_bad_verdict(self):
+        r = result()
+        r["target"]["issue"] = True
+        with self.assertRaises(mod.ValidationError):
+            mod.validate_result(r, {"src/a.py"})
+        r = result(); r["criteria"][0]["implementation_paths"] = ["../x"]
+        with self.assertRaises(mod.ValidationError):
+            mod.validate_result(r, {"../x"})
+        r = result("request_changes")
+        with self.assertRaises(mod.ValidationError):
+            mod.validate_result(r, {"src/a.py", "tests/test_a.py"})
+
+    def test_collect_pages_requires_identity_and_reads_all_pages(self):
+        pages = [[{"id": 1}, {"id": 2}], [{"id": 3}]]
+        def fetch(n):
+            return {"items": pages[n - 1], "total_count": 3, "has_more": n == 1}
+        self.assertEqual(mod.collect_pages(fetch, "id", page_size=2), pages[0] + pages[1])
+        def bad(n):
+            return {"items": [{"id": 1}], "total_count": 2, "has_more": True}
+        with self.assertRaises(mod.ValidationError):
+            mod.collect_pages(bad, "id", max_pages=1)
+
+    def test_verify_unchanged(self):
+        self.assertTrue(mod.verify_unchanged({"a": [1]}, {"a": [1]}))
+        with self.assertRaises(mod.ValidationError):
+            mod.verify_unchanged({"a": 1}, {"a": 2})
+
+    def test_cache_validation_and_rejects_symlink(self):
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            root = Path(td); ops = root / "harness" / "operations"; ops.mkdir(parents=True)
+            contract = ops / "repository-cache.md"; contract.write_text("cache", encoding="utf-8")
+            rs = root / "repositories"; owner = rs / "acme"; repo = owner / "repo"
+            (repo / "locks").mkdir(parents=True)
+            (repo / "repository.git").mkdir()
+            for p in (rs, owner, repo, repo / "locks"): os.chmod(p, 0o700)
+            info = mod.validate_cache_paths(contract, "acme", "repo", 3)
+            self.assertTrue(info["repository_git"].endswith("repository.git"))
+            (repo / "locks" / "issue-3.lock").symlink_to(contract)
+            with self.assertRaises(mod.ValidationError):
+                mod.validate_cache_paths(contract, "acme", "repo", 3)
+
+    def test_publication_accepts_one_exact_review_and_rejects_competing_event(self):
+        before, after, review, r, paths = publication_fixture()
+        self.assertEqual(verify_fixture(before, after, review, r, paths)["review_id"], 11)
+        bad = copy.deepcopy(after)
+        bad["timeline"].append({"id": 102, "review_id": 12, "user": {"id": 42}, "body": review["body"]})
+        with self.assertRaises(mod.ValidationError):
+            verify_fixture(before, bad, review, r, paths)
+
+    def test_cli_check_pages_and_unchanged(self):
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            td = Path(td); env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
+            pages = td / "pages.json"; pages.write_text(json.dumps([
+                {"items": [{"id": 1}], "total_count": 1, "has_more": False}],), encoding="utf-8")
+            base = [sys.executable, str(MODPATH)]
+            p = subprocess.run(base + ["check-pages", "--pages", str(pages), "--key", "id"], capture_output=True, text=True, env=env)
+            self.assertEqual(p.returncode, 0, p.stderr)
+            b = td / "before.json"; a = td / "after.json"
+            b.write_text('{"x":1}', encoding="utf-8"); a.write_text('{"x":1}', encoding="utf-8")
+            p = subprocess.run(base + ["check-unchanged", "--before", str(b), "--after", str(a)], capture_output=True, text=True, env=env)
+            self.assertEqual(p.returncode, 0, p.stderr); self.assertIn('"unchanged": true', p.stdout)
+
+    def test_cli_validate_render_parse(self):
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            td = Path(td); rf = td / "r.json"; af = td / "a.json"; bf = td / "b.md"
+            rf.write_text(json.dumps(result(), ensure_ascii=False), encoding="utf-8")
+            af.write_text(json.dumps(["src/a.py", "tests/test_a.py"]), encoding="utf-8")
+            env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
+            base = [sys.executable, str(MODPATH)]
+            p = subprocess.run(base + ["validate-result", "--result", str(rf), "--allowed-paths", str(af)],
+                               text=True, capture_output=True, env=env)
+            self.assertEqual(p.returncode, 0, p.stderr)
+            p = subprocess.run(base + ["render", "--result", str(rf), "--allowed-paths", str(af)],
+                               text=True, capture_output=True, env=env)
+            self.assertEqual(p.returncode, 0, p.stderr); bf.write_text(p.stdout, encoding="utf-8")
+            p = subprocess.run(base + ["parse", "--body", str(bf), "--allowed-paths", str(af)],
+                               text=True, capture_output=True, env=env)
+            self.assertEqual(p.returncode, 0, p.stderr)
+            self.assertEqual(json.loads(p.stdout), result())
+
+
+    def test_result_codec_preserves_existing_native_format(self):
+        r = result()
+        paths = {"src/a.py", "tests/test_a.py"}
+        body = mod.render_review(r, paths)
+        self.assertIn("  - 구현 근거: src/a.py; 증거\n", body)
+        self.assertIn("모든 기준을 충족했다.\n\n<!--", body)
+        self.assertEqual(mod.parse_review(body, paths), r)
+        unordered = json.loads(json.dumps(r, sort_keys=True))
+        self.assertEqual(mod.render_review(unordered, paths), body)
+
+    def test_strict_codec_rejects_unrenderable_and_sensitive_human_values(self):
+        paths = {"src/a.py", "tests/test_a.py"}
+        for field, value in (("verdict", []), ("conclusion", "trailing "),
+                             ("conclusion", "line\u2028break"), ("conclusion", "ftp://example.test")):
+            r = result(); r[field] = value
+            with self.subTest(field=field, value=value), self.assertRaises(mod.ValidationError):
+                mod.validate_result(r, paths)
+        for field in ("criterion", "contract", "evidence", "reason"):
+            r = result(); r["criteria"][0][field] = "secretvalue"
+            with self.subTest(field=field), self.assertRaises(mod.ValidationError):
+                mod.validate_result(r, paths, sensitive_values=("secretvalue",))
+        r = result(); r["verdict"] = "blocked_spec"; r["scope"]["spec_conflicts"] = ["secretvalue"]
+        with self.assertRaises(mod.ValidationError):
+            mod.validate_result(r, paths, sensitive_values=("secretvalue",))
+        with self.assertRaises(mod.ValidationError):
+            mod.loads_json('{"n":1e999}')
+        with self.assertRaises(mod.ValidationError):
+            mod.validate_result(result(), "src/a.py")
+
+    def test_parser_rejects_forged_marker_and_body_drift(self):
+        paths = {"src/a.py", "tests/test_a.py"}
+        body = mod.render_review(result(), paths)
+        for bad in (body.replace("issue=3", "issue=4"), body.replace("증거", "다른 증거"),
+                    body.replace("\n", "\r\n"), body + "\n", body.replace("correlation=", "other=")):
+            with self.subTest(body=bad[-100:]), self.assertRaises(mod.ValidationError):
+                mod.parse_review(bad, paths)
+
+    def test_codec_roundtrips_delimiter_bearing_paths_and_prose(self):
+        r = result(); r["criteria"][0]["implementation_paths"] = ["src/a, b.py", "src/c; d.py"]
+        r["criteria"][0]["evidence"] = "확인; 모두 — 일치"
+        r["verdict"] = "blocked_spec"; r["scope"]["spec_conflicts"] = ["하나, 둘", "세 가지"]
+        paths = {"src/a, b.py", "src/c; d.py", "tests/test_a.py"}
+        self.assertEqual(mod.parse_review(mod.render_review(r, paths), paths), r)
+
+
+    def test_cache_enforces_root_boundary_uid_and_dangling_locks(self):
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            root = Path(td); ops = root / "harness" / "operations"; ops.mkdir(parents=True)
+            contract = ops / "repository-cache.md"; contract.write_text("cache", encoding="utf-8")
+            rs = root / "repositories"; owner = rs / "acme"; repo = owner / "repo"
+            (repo / "locks").mkdir(parents=True); (repo / "repository.git").mkdir()
+            for path in (rs, owner, repo, repo / "locks"):
+                path.chmod(0o700)
+            os.chmod(root, 0o777)
+            with self.assertRaises(mod.ValidationError):
+                mod.validate_cache_paths(contract, "acme", "repo", 3)
+            os.chmod(root, 0o700)
+            with self.assertRaises(mod.ValidationError):
+                mod.validate_cache_paths(contract, "acme", "repo", 3, uid=os.getuid() + 10000)
+            (repo / "locks" / "issue-3.lock").symlink_to(root / "missing")
+            with self.assertRaises(mod.ValidationError):
+                mod.validate_cache_paths(contract, "acme", "repo", 3)
+            (repo / "locks" / "issue-3.lock").unlink()
+            alias = root / "alias"; alias.symlink_to(ops.parent, target_is_directory=True)
+            with self.assertRaises(mod.ValidationError):
+                mod.validate_cache_paths(alias / "operations" / "repository-cache.md", "acme", "repo", 3)
+
+    def test_cache_group_membership_includes_other_primary_accounts(self):
+        from types import SimpleNamespace
+        from unittest.mock import patch
+        role = SimpleNamespace(pw_name="role", pw_uid=os.getuid(), pw_gid=111)
+        other = SimpleNamespace(pw_name="other", pw_uid=os.getuid() + 1, pw_gid=111)
+        group = SimpleNamespace(gr_gid=111, gr_mem=["role"])
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            root = Path(td); ops = root / "harness" / "operations"; ops.mkdir(parents=True)
+            contract = ops / "repository-cache.md"; contract.write_text("cache")
+            rs = root / "repositories"; repo = rs / "acme" / "repo"
+            (repo / "locks").mkdir(parents=True); (repo / "repository.git").mkdir()
+            for path in (rs / "acme", repo, repo / "locks"):
+                path.chmod(0o700)
+            rs.chmod(0o770)
+            role.pw_gid = other.pw_gid = group.gr_gid = rs.stat().st_gid
+            with patch.object(mod.grp, "getgrgid", return_value=group), \
+                 patch.object(mod.pwd, "getpwuid", return_value=role), \
+                 patch.object(mod.pwd, "getpwall", return_value=[role, other]):
+                with self.assertRaises(mod.ValidationError):
+                    mod.validate_cache_paths(contract, "acme", "repo", 3)
+
+
+    def test_publication_accepts_gitea_nested_identity(self):
+        before, after, review, r, paths = publication_fixture()
+        self.assertEqual(verify_fixture(before, after, review, r, paths)["review_id"], 11)
+
+    def test_publication_requires_exact_readback_not_only_review_id(self):
+        before, after, review, r, paths = publication_fixture()
+        for field, value in (("body", "changed"), ("official", False), ("submitted_at", "bad")):
+            exact = copy.deepcopy(review); exact[field] = value
+            with self.subTest(field=field), self.assertRaises(mod.ValidationError):
+                verify_fixture(before, after, exact, r, paths)
+
+    def test_publication_rejects_unrelated_drift_and_bad_projections(self):
+        before, after, review, r, paths = publication_fixture()
+        cases = []
+        for field, value in (("updated_at", "2026-10-03T10:00:00Z"), ("review_comments", 4),
+                             ("requested_reviewers", [{"id": 43}]), ("head", {"sha": "d" * 40})):
+            bad = copy.deepcopy(after); bad["pr"][field] = value; cases.append(bad)
+        for field, value in (("updated_at", "2026-10-03T10:00:00Z"), ("title", "changed")):
+            bad = copy.deepcopy(after); bad["issue"][field] = value; cases.append(bad)
+        bad = copy.deepcopy(after); bad["reviews"][0]["stale"] = 0; cases.append(bad)
+        bad = copy.deepcopy(after); bad["reviews"][0]["state"] = "PENDING"; cases.append(bad)
+        bad = copy.deepcopy(after); bad["reviews"].append(copy.deepcopy(review)); cases.append(bad)
+        bad = copy.deepcopy(after); bad["timeline"][0]["type"] = "comment"; cases.append(bad)
+        bad = copy.deepcopy(after); bad["timeline"][0]["review_id"] = True; cases.append(bad)
+        for bad in cases:
+            with self.subTest(snapshot=cases.index(bad)), self.assertRaises(mod.ValidationError):
+                verify_fixture(before, bad, bad["reviews"][0], r, paths)
+
+    def test_publication_binds_snapshot_repository_issue_and_head(self):
+        before, after, review, r, paths = publication_fixture()
+        for section, field, value in (("issue", "number", 4), ("issue", "repository", {"full_name": "wrong/repo"}),
+                                       ("pr", "head", {"sha": "d" * 40}), ("pr", "number", 8)):
+            b = copy.deepcopy(before); a = copy.deepcopy(after)
+            b[section][field] = a[section][field] = value
+            with self.subTest(section=section, field=field), self.assertRaises(mod.ValidationError):
+                verify_fixture(b, a, review, r, paths)
+
+    def test_publication_preserves_previous_reviews_and_bounds_supersession(self):
+        before, after, review, r, paths = publication_fixture()
+        old = copy.deepcopy(review); old.update(id=10, body="previous", commit_id="d" * 40,
+                                             state="REQUEST_CHANGES", updated_at="2026-10-03T00:00:00Z")
+        before["reviews"] = [old]; after["reviews"].insert(0, copy.deepcopy(old))
+        verify_fixture(before, after, review, r, paths)
+        after["reviews"][0].update(official=False, dismissed=True, updated_at=review["submitted_at"])
+        with self.assertRaises(mod.ValidationError):
+            verify_fixture(before, after, review, r, paths)
+        contract = {"version": {"version": "1.27.3"}, "supersession_same_author_only": True,
+                    "already_dismissed_review_timestamp_projection": True}
+        verify_fixture(before, after, review, r, paths, contract)
+        for mutation in ("body", "user", "missing"):
+            a = copy.deepcopy(after)
+            if mutation == "missing": a["reviews"].pop(0)
+            elif mutation == "user": a["reviews"][0]["user"] = {"id": 43}
+            else: a["reviews"][0]["body"] = "tampered"
+            with self.subTest(mutation=mutation), self.assertRaises(mod.ValidationError):
+                verify_fixture(before, a, review, r, paths, contract)
+        old["official"] = False; old["dismissed"] = True
+        after["reviews"][0] = copy.deepcopy(old); after["reviews"][0]["updated_at"] = review["submitted_at"]
+        verify_fixture(before, after, review, r, paths, contract)
+        contract["already_dismissed_review_timestamp_projection"] = False
+        with self.assertRaises(mod.ValidationError):
+            verify_fixture(before, after, review, r, paths, contract)
+
+    def test_publication_preserves_collection_identity_and_normalizes_only_files_order(self):
+        before, after, review, r, paths = publication_fixture()
+        second = {"filename": "tests/test_a.py", "status": "added", "additions": 2}
+        before["files"].append(second); after["files"].insert(0, copy.deepcopy(second))
+        verify_fixture(before, after, review, r, paths)
+        after["files"].append(copy.deepcopy(second))
+        with self.assertRaises(mod.ValidationError):
+            verify_fixture(before, after, review, r, paths)
+
+    def test_read_only_replay_uses_recursive_exact_json_types(self):
+        with self.assertRaises(mod.ValidationError):
+            mod.verify_unchanged({"nested": [True]}, {"nested": [1]})
+
+
+    def test_pagination_is_one_based_and_rejects_short_continuations(self):
+        calls = []
+        pages = [{"items": [{"id": 1}, {"id": 2}], "total_count": 3, "has_more": True},
+                 {"items": [{"id": 3}], "total_count": 3, "has_more": False}]
+        def fetch(page):
+            calls.append(page)
+            return pages[page - 1]
+        self.assertEqual(mod.collect_pages(fetch, "id", page_size=2), [{"id": 1}, {"id": 2}, {"id": 3}])
+        self.assertEqual(calls, [1, 2])
+        with self.assertRaises(mod.ValidationError):
+            mod.collect_pages(lambda _: {"items": [{"id": 1}], "total_count": 3, "has_more": True}, "id", page_size=2)
+
+    def test_pagination_validates_all_declared_metadata_and_identity_types(self):
+        bad_pages = [
+            {"items": [{"id": 1}], "total_count": 0, "has_more": True},
+            {"items": [{"id": 1}, {"id": 2}], "total_count": 2, "has_more": True},
+            {"items": [{"id": []}], "has_more": False},
+            {"items": [{"id": 1.5}], "has_more": False},
+            {"items": [{"id": 0}], "has_more": False},
+            {"items": [{"id": True}], "has_more": False},
+            {"items": [{"id": 1}, {"id": 1}], "has_more": False},
+            {"items": None, "has_more": False},
+            {"items": [], "has_more": True},
+        ]
+        for page in bad_pages:
+            with self.subTest(page=bad_pages.index(page)), self.assertRaises(mod.ValidationError):
+                mod.collect_pages(lambda _: page, "id", page_size=2, max_pages=2)
+        pages = [{"items": [{"id": 1}], "total_count": 2, "has_more": True},
+                 {"items": [{"id": 2}], "total_count": 3, "has_more": False}]
+        with self.assertRaises(mod.ValidationError):
+            mod.collect_pages(lambda i: pages[i - 1], "id", page_size=1)
+        self.assertEqual(mod.collect_pages(lambda _: {"items": None, "total_count": 0, "has_more": False}, "id"), [])
+
+    def test_cli_errors_are_sanitized_without_tracebacks_or_echoed_values(self):
+        env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
+        base = [sys.executable, str(MODPATH)]
+        for args in (["unknown-secret-value"], ["check-cache", "--issue", "secret-value"],
+                     ["check-unchanged", "--before", "secret-value", "--after", "missing"]):
+            p = subprocess.run(base + args, capture_output=True, text=True, env=env)
+            self.assertNotEqual(p.returncode, 0)
+            self.assertNotIn("secret-value", p.stderr + p.stdout)
+            self.assertNotIn("Traceback", p.stderr + p.stdout)
+            self.assertEqual(json.loads(p.stderr), {"error": "invalid input"})
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            path = Path(td) / "pages.json"
+            path.write_text(json.dumps([{ "items": [{"id": 1}], "has_more": True, "total_count": 2 }]))
+            p = subprocess.run(base + ["check-pages", "--pages", str(path), "--key", "id"],
+                               capture_output=True, text=True, env=env)
+            self.assertNotEqual(p.returncode, 0)
+            self.assertNotIn("Traceback", p.stderr)
+            self.assertEqual(json.loads(p.stderr), {"error": "invalid input"})
+
+
+    def test_target_identifiers_preserve_dots_but_reject_marker_breaking_whitespace(self):
+        r = result(); paths = {"src/a.py", "tests/test_a.py"}
+        r["target"].update(owner="acme.name", repository="repo.name")
+        body = mod.render_review(r, paths)
+        self.assertEqual(mod.parse_review(body, paths), r)
+        r["target"]["owner"] = "acme name"
+        with self.assertRaises(mod.ValidationError):
+            mod.validate_result(r, paths)
+
+    def test_rendered_paths_cannot_inject_markdown_html_or_mentions(self):
+        r = result(); path = "src/<script>@user`[link].py"
+        r["criteria"][0]["implementation_paths"] = [path]
+        paths = {path, "tests/test_a.py"}
+        body = mod.render_review(r, paths)
+        self.assertNotIn("<script>", body)
+        self.assertNotIn("@user", body)
+        self.assertNotIn("`[link]", body)
+        self.assertEqual(mod.parse_review(body, paths), r)
+
+    def test_strict_types_limits_and_duplicate_nested_keys(self):
+        paths = {"src/a.py", "tests/test_a.py"}
+        for bad_json in ('{"outer":{"x":1,"x":2}}', '{"x":Infinity}', '{"x":-Infinity}'):
+            with self.subTest(bad_json=bad_json), self.assertRaises(mod.ValidationError):
+                mod.loads_json(bad_json)
+        cases = []
+        r = result(); r["target"]["pull"] = False; cases.append(r)
+        r = result(); r["criteria"].append(copy.deepcopy(r["criteria"][0])); cases.append(r)
+        r = result(); r["criteria"][0]["contract"] = "a" * 4001; cases.append(r)
+        r = result(); r["criteria"][0]["criterion"] = "a" * 201; cases.append(r)
+        r = result(); r["criteria"][0]["result"] = {}; cases.append(r)
+        r = result(); r["criteria"][0]["test_paths"] *= 2; cases.append(r)
+        r = result(); r["criteria"][0]["test_paths"] = ["/absolute"]; cases.append(r)
+        r = result(); r["criteria"][0]["test_paths"] = ["tests/../file.py"]; cases.append(r)
+        r = result(); r["extra"] = None; cases.append(r)
+        r = result(); del r["scope"]; cases.append(r)
+        for value in cases:
+            with self.subTest(case=cases.index(value)), self.assertRaises(mod.ValidationError):
+                mod.validate_result(value, paths)
+
+    def test_cli_verify_publication_and_rejects_raw_crlf_body(self):
+        before, after, review, r, paths = publication_fixture()
+        env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
+        with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td:
+            root = Path(td)
+            for name, value in (("before", before), ("after", after), ("exact", review),
+                                ("result", r), ("paths", sorted(paths)), ("inline", [])):
+                (root / (name + ".json")).write_text(json.dumps(value, ensure_ascii=False))
+            cmd = [sys.executable, str(MODPATH), "verify-publication", "--before", str(root / "before.json"),
+                   "--after", str(root / "after.json"), "--exact-review", str(root / "exact.json"),
+                   "--result", str(root / "result.json"), "--allowed-paths", str(root / "paths.json"),
+                   "--inline-comments", str(root / "inline.json"), "--reviewer-id", "42",
+                   "--expected-pr-url", "https://example.test/acme/repo/pulls/7"]
+            p = subprocess.run(cmd, capture_output=True, text=True, env=env)
+            self.assertEqual(p.returncode, 0, p.stderr)
+            self.assertEqual(json.loads(p.stdout)["review_id"], 11)
+            body_path = root / "body.md"; body_path.write_bytes(review["body"].replace("\n", "\r\n").encode())
+            p = subprocess.run([sys.executable, str(MODPATH), "parse", "--body", str(body_path),
+                                "--allowed-paths", str(root / "paths.json")], capture_output=True, text=True, env=env)
+            self.assertNotEqual(p.returncode, 0)
+            self.assertEqual(json.loads(p.stderr), {"error": "invalid input"})
+
+
+    def test_publication_accepts_causal_event_timestamp_without_forcing_clock_equality(self):
+        before, after, review, r, paths = publication_fixture()
+        event = after["timeline"][0]
+        event["created_at"] = event["updated_at"] = "2026-10-03T01:24:01Z"
+        after["pr"]["updated_at"] = event["created_at"]
+        verify_fixture(before, after, review, r, paths)
+
+
+    def test_publication_requires_independent_canonical_pr_url_anchor(self):
+        before, after, review, r, paths = publication_fixture()
+        for foreign in ("https://evil.test/acme/repo/pulls/7", "https://example.test/prefix/acme/repo/pulls/7"):
+            b = copy.deepcopy(before); a = copy.deepcopy(after)
+            b["pr"]["html_url"] = b["pr"]["url"] = foreign
+            a["pr"]["html_url"] = a["pr"]["url"] = foreign
+            a["reviews"][0]["pull_request_url"] = foreign
+            a["timeline"][0]["pull_request_url"] = foreign
+            with self.subTest(url=foreign), self.assertRaises(mod.ValidationError):
+                verify_fixture(b, a, a["reviews"][0], r, paths)
+
+
+if __name__ == "__main__":
+    unittest.main()