feat(harness): add structural workflow guard gates

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

@@ -153,6 +153,8 @@ Discovery가 `신규 작성` 또는 `수정`으로 판정한 문서만 작성한
 
 ### 7. Test-first 구현
 
+[공통 workflow guard 계약](../../../../operations/repository-cache.md#shared-workflow-guard-gates)을 baseline 조사·첫 편집·독립 검토·결과 채택 단계에 적용한다. 첫 편집 전에 테스트 ID/skip/expected-failure와 영향을 받는 기존 API·CLI·상태·cleanup 의미를 기록하고 `check-tests --input JSON` 및 `check-contracts --input JSON`으로 검증한다. 읽기 전용 worker 시작 전에 `python3 <harness-root>/harness/lib/workflow_guard.py snapshot --repo-root ROOT --paths-json INPUT --output SNAPSHOT`으로 immutable protected scope를 고정하고, 결과 채택 전에 `compare-source --repo-root ROOT --before SNAPSHOT --paths-json INPUT`을 실행한다. 쓰기 worker의 승인된 편집은 실제 diff로 검토하며 최종 독립 검토 중에는 보호 범위 mutator를 실행하지 않는다. 기록은 제품 checkout 밖 정본 verification run에 보존한다. 변경된 테스트의 의미 보존과 named finding의 실제 RED/GREEN 원인은 main agent가 확인한다. 필요한 gate·근거가 미해결이면 완료로 판정하지 않는다.
+
 1. Acceptance matrix에서 변경 전 실패해야 하는 정상, 실패, 경계와 보안 test를 먼저 추가한다.
    - Test file과 test function 이름은 추적 번호가 아니라 안정적인 기능·동작·불변조건을 표현한다. Issue, PR, review, comment 번호나 일시적인 finding ID를 이름에 넣지 않는다. 추적성은 acceptance matrix, commit, PR과 evidence metadata에 기록한다.
 2. 새 test가 의도한 이유로 실패하는지 실행해 확인한다. 환경 오류나 잘못된 fixture 실패를 RED evidence로 사용하지 않는다.

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

@@ -71,6 +71,8 @@ Repository file을 수정하기 전에 다음을 모두 검증한다.
 
 ## Snapshot과 finding matrix
 
+첫 write 전에 canonical collection의 test inventory와 영향을 받는 API·CLI·필드·상태·cleanup 의미를 snapshot/finding matrix에 연결하고 [공통 workflow guard 계약](../../../../operations/repository-cache.md#shared-workflow-guard-gates)을 적용한다. `python3 <harness-root>/harness/lib/workflow_guard.py check-tests --input JSON`과 `check-contracts --input JSON`을 결과 채택 전에 실행한다. RED/GREEN records must be named: 각 유효한 finding은 current test ID와 실제 실행 기록에 결합하며 환경/import 오류나 반증된 지적에 가짜 RED를 만들지 않는다. 읽기 전용 검토에는 위임 전 source snapshot과 채택 전 비교를 요구하고, 승인된 쓰기 worker의 편집에는 diff·의미 검토를 적용한다. 기록은 제품 checkout 밖 정본 verification run에 보존하며 helper schema·증거를 검증한다. gate failure는 운영 근거 무효이지 자동 제품 결함이 아니다.
+
 첫 write 전에 다음 immutable snapshot을 보존한다.
 
 - Issue title, body, state, `updated_at`, labels, assignees, milestone, 전체 comments와 body SHA-256

harness/agents/review/skills/gitea-code-review/SKILL.md

@@ -38,6 +38,8 @@ Bootstrap 이후 code review는 canonical [Repository cache 운영 지침](../..
 
 ## 리뷰 관점
 
+읽기 전용 위임과 최종 결과 채택에는 [공통 workflow guard 계약](../../../../operations/repository-cache.md#shared-workflow-guard-gates)을 적용한다. 사용이 허용된 source에서 위임 전 보호 범위를 snapshot하고 채택 전 compare-source를 완료한다. 최종 독립 검토 중 보호 범위 mutator를 병렬 실행하지 않는다. `check-tests --input JSON`으로 기존 ID·status·finding 연결을 확인하고 변경된 assertion·fixture·mock의 의미 검출력을 직접 검토한다. API·CLI·필드·상태·cleanup 보존과 실제 public-response 관찰을 대조한다. 기록은 제품 checkout 밖 허용된 정본 artifact에 보존하고 시작 baseline 누락·source drift이면 PASS를 채택하거나 자동 복구하지 않는다. 운영 근거 실패와 제품 finding을 구분한다.
+
 - **정확성·회귀:** 조건 분기, 경계값, 상태 전이, 데이터 처리와 기존 호출자에 대한 부작용.
 - **보안:** 인증·인가, 입력 검증, injection·경로 탈출, secret 노출, cookie·origin·proxy trust와 fail-closed/no-call/no-write 경계.
 - **동시성·복구:** race, 원자성, 잠금 순서, timeout·retry·중복 처리, 부분 실패와 자원 정리.

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

@@ -71,6 +71,8 @@ Issue comment·body·title·state·label·assignee·milestone, repository file,
 
 ## Acceptance matrix
 
+Verdict 확정 전에 immutable provider/Git blob·contract·PR snapshot에 결합된 complete baseline inventory를 [공통 workflow guard 계약](../../../../operations/repository-cache.md#shared-workflow-guard-gates)과 대조하고 `python3 <harness-root>/harness/lib/workflow_guard.py check-contracts --input JSON`을 실행한다. 기존 의미는 실제 current 항목·검증 근거에 대응하며 의도적 변경에는 승인 근거가 필요하다. 미확인 관찰·missing·ambiguous를 개수나 승인 문자열로 통과시키지 않는다. API·CLI·필드·상태·cleanup 의미까지 확인하되 새 제품 요구사항을 만들지 않는다. Gate failure invalidates workflow evidence: 운영 근거 무효와 제품 계약 위반을 분리한다. 기존 provider/blob source binding, evidence 경로/schema, moving implementation checkout 사용 금지와 제품 독립 실행 제한은 그대로이며 local snapshot helper가 그 권한을 확대하지 않는다.
+
 Discovery의 포함 범위, 제외 범위, 제약 조건, 확정된 결정과 `문서 영향 및 산출물`에서 판정 가능한 항목을 추출한다. 각 항목을 다음 필드로 기록한다.
 
 ```text

harness/lib/workflow_guard.py

@@ -0,0 +1,402 @@
+"""Structural workflow guards; evidence claims are shape-only, not authenticated.
+
+CLI (exactly four commands): snapshot --repo-root ROOT --paths-json INPUT
+--output SNAPSHOT; compare-source --repo-root ROOT --before SNAPSHOT
+--paths-json INPUT; check-tests/check-contracts --input JSON. Exit 0 is PASS,
+1 detected drift/violations, 2 rejected input. APIs raise GuardError for rejected
+input or return status/violations. JSON inputs are regular files bounded to 1 MiB;
+unknown/missing/duplicate object keys and boolean-as-integer fields block.
+ROOT is a literal canonical absolute Git root, without symlink ancestors. Paths
+are a nonempty unique repository-relative string array, with no dot/traversal
+segments or backslash aliases. Protected files must exist and be regular.
+Snapshot output is new/no-clobber, mode 0600, outside ROOT, in an already-existing
+executing-UID-owned 0700 directory. No permissions or directories are repaired.
+
+check_tests input has exactly these required keys:
+ baseline_ids, current_ids: arrays of unique nonempty test-ID strings;
+ baseline_status, current_status: exact ID-keyed maps with values
+   {skip: bool, expected_failure: bool};
+ baseline_collection, current_collection: objects with exactly
+   {source_revision: lowercase 40-hex, command_id: nonempty string,
+    argv: nonempty array of nonempty strings, inventory_sha256: lowercase 64-hex,
+    evidence: nonempty reference string}.
+The digest is SHA256 over UTF-8 bytes of json.dumps(
+ {'ids': sorted(ids), 'statuses': {id: statuses[id] for id in sorted(ids)}},
+ sort_keys=True, separators=(',', ':')) with default JSON escaping.
+ authorized_changes: array of exact objects
+   {old_id: baseline ID, kind: 'delete'|'rename'|'status',
+    replacement_ids: unique current-ID array, reason: nonempty string,
+    approval: nonempty reference, evidence: nonempty semantic-review reference}.
+   delete requires old ID absent and empty replacements; rename requires old ID
+   absent and at least one actual replacement; status requires old ID present
+   and replacements exactly [old_id]. Old IDs and replacement targets cannot
+   repeat between approvals. Ordinary unskipped regression additions need no
+   deletion approval. Every removed ID and newly introduced skip/xfail requires
+   a compatible approval. Do not invent replacement coverage.
+ changed_tests: array, unique by test_id, of exact objects
+   {test_id: current ID, baseline_sha256: lowercase 64-hex,
+    current_sha256: lowercase 64-hex, review_evidence: nonempty reference}.
+ findings: array, unique by id, of exact objects
+   {id: nonempty string, test_id: current ID,
+    classification: 'assertion'|'approved_behavior', red: execution,
+    green: execution}. Each execution has exactly
+   {test_id: finding's test_id, source_revision: lowercase 40-hex,
+    scope_sha256: lowercase 64-hex, command_id: nonempty reference,
+    argv: nonempty string array, exit_code: int, expected_exit_code: same int,
+    failure_kind: string, evidence: nonempty reference}.
+   RED exit is positive/nonzero, failure_kind assertion or approved_behavior;
+   GREEN exit is 0, failure_kind pass. Import/environment failures are not RED.
+
+check_contracts input has exactly baseline, current, mappings, stages, precedence:
+ baseline: nonempty array, unique semantic_key, of exact objects
+   {semantic_key: nonempty string, baseline_revision: lowercase 40-hex,
+    path: safe repository-relative path, meaning: nonempty string};
+ current: array, unique semantic_key, of exact objects
+   {semantic_key: nonempty string, meaning: nonempty string,
+    observation: 'observed'|'not_applicable'|'unknown', evidence: nonempty reference}.
+   unknown blocks even with evidence; not_applicable needs genuine justification.
+ mappings: one exact object per baseline semantic_key
+   {semantic_key: baseline key, current_keys: unique current-key array,
+    status: 'preserved'|'intentionally_changed'|'missing'|'ambiguous',
+    authorization: string, verification: string}.
+   Resolved rows require nonempty current_keys/verification. intentionally_changed
+   additionally requires nonempty authorization; preserved may use empty
+   authorization. missing/ambiguous always block. All current keys are covered.
+   Substitutions and merges may use new current keys; several old rows can map
+   to one current row without dropping any old meaning or verification.
+ stages: nonempty array of unique nonempty strings;
+ precedence: array of unique [prerequisite, dependent] declared-stage pairs.
+   Initialization graph must be acyclic, not a graph of allowed lifecycle loops.
+
+Snapshot JSON is generated by snapshot, not hand-declared: schema integer 1,
+repo canonical root, head full SHA, sorted paths, files keyed by exact paths.
+Each file object has present true, type regular, sha256 64-hex, and nonnegative
+integer device/inode/mode/uid/nlink. Comparison checks HEAD, path set and file
+content/identity metadata; it does not prove absence of edit-then-restore or
+out-of-scope writes and is not a sandbox.
+
+No evidence/approval reference is opened or authenticated by check_tests or
+check_contracts. Main verifies actual source HEAD/blob/scope digests, approvals,
+argv/environment/output/exit and intended named failure using existing immutable
+verification records. Declared hashes are not proof of source identity. Meaning
+preservation and omitted edits require independent review. Runnable nested JSON
+examples and main-agent binding procedure are in repository-cache.md, Shared
+workflow guard gates / Nested input contracts; synthetic fixtures are not evidence.
+"""
+import argparse, hashlib, json, os, re, stat, subprocess
+from functools import wraps
+from pathlib import Path
+
+class GuardError(ValueError): pass
+
+_HEX40 = re.compile(r'^[0-9a-f]{40}$')
+_HEX64 = re.compile(r'^[0-9a-f]{64}$')
+
+def _fail(message='invalid workflow guard input'):
+    raise GuardError(message)
+
+def _keys(obj, keys):
+    if not isinstance(obj, dict) or set(obj) != set(keys): _fail('invalid object fields')
+
+def _str(v): return isinstance(v, str) and bool(v.strip())
+def _hex(v, pattern): return isinstance(v, str) and pattern.fullmatch(v) is not None
+
+def _guard_errors(function):
+    @wraps(function)
+    def checked(*args, **kwargs):
+        try:
+            return function(*args, **kwargs)
+        except GuardError:
+            raise
+        except (OSError, ValueError, TypeError, KeyError, RecursionError):
+            _fail('workflow guard rejected input')
+    return checked
+
+def _directory(raw):
+    if not isinstance(raw, str) or not os.path.isabs(raw) or raw != os.path.normpath(raw):
+        _fail('noncanonical directory path')
+    if raw != '/' and any(part in ('', '.', '..') for part in raw.split('/')[1:]):
+        _fail('noncanonical directory path')
+    fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        for part in raw.split('/')[1:]:
+            if not part:
+                continue
+            nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
+            os.close(fd)
+            fd = nxt
+        return fd
+    except BaseException:
+        os.close(fd)
+        raise
+
+def _json(path):
+    def pairs(items):
+        out = {}
+        for k,v in items:
+            if k in out: _fail('duplicate JSON key')
+            out[k]=v
+        return out
+    try:
+        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        try:
+            if not stat.S_ISREG(os.fstat(fd).st_mode):
+                _fail('JSON input must be regular')
+            with os.fdopen(fd, 'rb') as f:
+                fd = None
+                raw = f.read(1048577)
+        finally:
+            if fd is not None:
+                os.close(fd)
+        if len(raw)>1048576: _fail('JSON input too large')
+        return json.loads(raw.decode('utf-8'), object_pairs_hook=pairs,
+                          parse_constant=lambda _: _fail('invalid JSON number'))
+    except GuardError: raise
+    except (OSError, UnicodeError, ValueError, TypeError, RecursionError): _fail('invalid JSON input')
+
+def _paths(paths):
+    if not isinstance(paths,list) or not paths or any(not isinstance(p,str) or not p for p in paths): _fail('invalid paths')
+    for p in paths:
+        if p.startswith('/') or '\\' in p or any(x in ('','.','..') for x in p.split('/')): _fail('unsafe path')
+    if len(set(paths)) != len(paths): _fail('duplicate path')
+    return sorted(paths)
+
+def _root(root):
+    raw=os.fspath(root)
+    if not isinstance(raw,str) or not os.path.isabs(raw) or raw.endswith('/') or '//' in raw or '/./' in raw or '/../' in raw: _fail('invalid repository root')
+    fd=None
+    try:
+        fd=_directory(raw)
+        p=subprocess.run(['git','-C',raw,'rev-parse','--show-toplevel','--is-bare-repository'],check=True,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True)
+        lines=p.stdout.splitlines()
+        if len(lines)!=2 or lines[1]!='false' or os.path.realpath(raw)!=lines[0]: _fail('root is not repository top-level')
+        return os.path.realpath(raw)
+    except (OSError, subprocess.SubprocessError): _fail('cannot identify repository root')
+    finally:
+        if fd is not None: os.close(fd)
+
+def _head(root):
+    try:
+        h=subprocess.run(['git','-C',root,'rev-parse','--verify','HEAD'],check=True,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True).stdout.strip()
+        if not _hex(h,_HEX40): _fail('invalid HEAD')
+        return h
+    except (OSError,subprocess.SubprocessError): _fail('cannot identify HEAD')
+
+def _open_file(root,rel):
+    fd=_directory(root)
+    try:
+        parts=rel.split('/')
+        for part in parts[:-1]:
+            nxt=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=fd); os.close(fd); fd=nxt
+        named = os.stat(parts[-1], dir_fd=fd, follow_symlinks=False)
+        if not stat.S_ISREG(named.st_mode):
+            _fail('protected path is not a regular file')
+        f = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd)
+        try:
+            opened = os.fstat(f)
+            if not stat.S_ISREG(opened.st_mode) or (named.st_dev, named.st_ino) != (opened.st_dev, opened.st_ino):
+                _fail('protected file changed')
+            return f, opened
+        except BaseException:
+            os.close(f)
+            raise
+    finally: os.close(fd)
+
+def _record(root, rel):
+    fd,st=_open_file(root,rel)
+    try:
+        h=hashlib.sha256()
+        while True:
+            b=os.read(fd,131072)
+            if not b: break
+            h.update(b)
+        end=os.fstat(fd)
+        if (st.st_dev,st.st_ino,st.st_size,st.st_mtime_ns)!=(end.st_dev,end.st_ino,end.st_size,end.st_mtime_ns): _fail('protected file changed during read')
+        return {'present':True,'type':'regular','sha256':h.hexdigest(),'device':st.st_dev,'inode':st.st_ino,'mode':stat.S_IMODE(st.st_mode),'uid':st.st_uid,'nlink':st.st_nlink}
+    finally: os.close(fd)
+
+def _out_parent(output,root):
+    raw=os.path.abspath(os.fspath(output)); parent,name=os.path.split(raw)
+    if not name or os.path.commonpath([root,raw])==root: _fail('output must be outside repository')
+    cur=os.open('/' if raw.startswith('/') else '.',os.O_RDONLY|os.O_DIRECTORY)
+    try:
+        for part in [x for x in parent.split('/') if x]:
+            n=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=cur); os.close(cur); cur=n
+        st=os.fstat(cur)
+        if st.st_uid!=os.getuid() or stat.S_IMODE(st.st_mode)!=0o700: _fail('output parent must be owned 0700 directory')
+        return cur,name
+    except (OSError, GuardError):
+        os.close(cur); _fail('unsafe output parent')
+
+@_guard_errors
+def snapshot(repo_root,paths,output):
+    root=_root(repo_root); paths=_paths(paths); head=_head(root)
+    files={p:_record(root,p) for p in paths}
+    if _head(root)!=head: _fail('HEAD changed during capture')
+    parent,name=_out_parent(output,root)
+    fd = None
+    try:
+        fd=os.open(name,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600,dir_fd=parent)
+        os.fchmod(fd,0o600); st=os.fstat(fd); named=os.stat(name,dir_fd=parent,follow_symlinks=False)
+        if (st.st_dev,st.st_ino)!=(named.st_dev,named.st_ino): _fail('output changed')
+        data={'schema':1,'repo':root,'head':head,'paths':paths,'files':files}
+        with os.fdopen(fd,'w',encoding='utf8') as f:
+            fd = None
+            json.dump(data,f,sort_keys=True); f.write('\n')
+    except (OSError,ValueError,TypeError): _fail('cannot safely create snapshot')
+    finally:
+        if fd is not None: os.close(fd)
+        os.close(parent)
+    return {'status':'pass','head':head,'paths':paths}
+
+@_guard_errors
+def compare_source(repo_root,before,paths):
+    root=_root(repo_root); paths=_paths(paths); b=_json(before)
+    _keys(b,('schema','repo','head','paths','files'))
+    if type(b['schema']) is not int or b['schema']!=1 or b['repo']!=root or not _hex(b['head'],_HEX40) or b['paths']!=paths or not isinstance(b['files'],dict) or set(b['files'])!=set(paths): _fail('malformed source baseline')
+    for x in b['files'].values():
+        _keys(x,('present','type','sha256','device','inode','mode','uid','nlink'))
+        if x['present'] is not True or x['type']!='regular' or not _hex(x['sha256'],_HEX64) or any(type(x[k]) is not int or x[k]<0 for k in ('device','inode','mode','uid','nlink')): _fail('malformed file baseline')
+    violations=[]
+    if _head(root)!=b['head']: violations.append('HEAD changed')
+    for p in paths:
+        try:
+            r=_record(root,p)
+            if r!=b['files'][p]: violations.append('source changed: '+p)
+        except (GuardError, OSError): violations.append('source missing or changed: '+p)
+    return {'status':'block' if violations else 'pass','violations':violations}
+
+def _ids(d,k):
+    x=d.get(k)
+    if not isinstance(x,list) or any(not _str(i) for i in x) or len(set(x))!=len(x): _fail('invalid IDs')
+    return x,set(x)
+
+def _status(s):
+    _keys(s,('skip','expected_failure'))
+    if type(s['skip']) is not bool or type(s['expected_failure']) is not bool: _fail('invalid status')
+
+def _collection(c,ids,status):
+    _keys(c,('source_revision','command_id','argv','inventory_sha256','evidence'))
+    if not _hex(c['source_revision'],_HEX40) or not _str(c['command_id']) or not isinstance(c['argv'],list) or not c['argv'] or any(not _str(a) for a in c['argv']) or not _str(c['evidence']): _fail('invalid collection provenance')
+    digest=hashlib.sha256(json.dumps({'ids':sorted(ids),'statuses':{i:status[i] for i in sorted(ids)}},sort_keys=True,separators=(',',':')).encode()).hexdigest()
+    if c['inventory_sha256']!=digest: _fail('collection digest mismatch')
+
+@_guard_errors
+def check_tests(data):
+    _keys(data,('baseline_ids','current_ids','baseline_status','current_status','baseline_collection','current_collection','authorized_changes','changed_tests','findings'))
+    old,olds=_ids(data,'baseline_ids'); new,news=_ids(data,'current_ids')
+    violations=[]
+    for prefix,ids in (('baseline',olds),('current',news)):
+        statuses=data[prefix+'_status']
+        if not isinstance(statuses,dict) or set(statuses)!=ids: _fail('status inventory mismatch')
+        for s in statuses.values(): _status(s)
+        _collection(data[prefix+'_collection'],ids,statuses)
+    auth=data['authorized_changes']
+    if not isinstance(auth,list): _fail('invalid authorizations')
+    approvals={}; targets=set()
+    for a in auth:
+        _keys(a,('old_id','kind','replacement_ids','reason','approval','evidence'))
+        if not _str(a['old_id']) or a['old_id'] not in olds or a['old_id'] in approvals or not isinstance(a['replacement_ids'],list) or any(not _str(x) or x not in news for x in a['replacement_ids']) or len(set(a['replacement_ids']))!=len(a['replacement_ids']) or any(not _str(a[k]) for k in ('reason','approval','evidence')): _fail('invalid authorization')
+        kind, old_id, replacements = a['kind'], a['old_id'], a['replacement_ids']
+        if kind == 'delete':
+            if old_id in news or replacements: _fail('invalid deletion authorization')
+        elif kind == 'rename':
+            if old_id in news or not replacements: _fail('invalid rename authorization')
+        elif kind == 'status':
+            if old_id not in news or replacements != [old_id]: _fail('invalid status authorization')
+        else: _fail('invalid authorization kind')
+        if targets.intersection(a['replacement_ids']): _fail('reused replacement')
+        targets.update(a['replacement_ids']); approvals[a['old_id']]=a
+    for i in olds-news:
+        if i not in approvals: violations.append('removed test lacks authorization')
+    for i in news:
+        if data['current_status'][i]['skip'] and (i not in olds or not data['baseline_status'][i]['skip']) and not any(i in a['replacement_ids'] or a['old_id']==i for a in auth): violations.append('new skip lacks authorization')
+        if data['current_status'][i]['expected_failure'] and (i not in olds or not data['baseline_status'][i]['expected_failure']) and not any(i in a['replacement_ids'] or a['old_id']==i for a in auth): violations.append('new expected failure lacks authorization')
+    changed=data['changed_tests']
+    if not isinstance(changed,list): _fail('invalid changed tests')
+    seen=set()
+    for c in changed:
+        _keys(c,('test_id','baseline_sha256','current_sha256','review_evidence'))
+        i=c['test_id']
+        if i not in news or i in seen or not _hex(c['baseline_sha256'],_HEX64) or not _hex(c['current_sha256'],_HEX64) or not _str(c['review_evidence']): _fail('invalid changed test review')
+        seen.add(i)
+    findings=data['findings']
+    if not isinstance(findings,list): _fail('invalid findings')
+    fids=set()
+    for f in findings:
+        _keys(f,('id','test_id','classification','red','green'))
+        if not _str(f['id']) or f['id'] in fids or f['test_id'] not in news or f['classification'] not in ('assertion','approved_behavior'): _fail('invalid finding')
+        fids.add(f['id'])
+        for stage in ('red','green'):
+            r=f[stage]; _keys(r,('test_id','source_revision','scope_sha256','command_id','argv','exit_code','expected_exit_code','failure_kind','evidence'))
+            if r['test_id']!=f['test_id'] or not _hex(r['source_revision'],_HEX40) or not _hex(r['scope_sha256'],_HEX64) or not _str(r['command_id']) or not isinstance(r['argv'],list) or not r['argv'] or any(not _str(a) for a in r['argv']) or type(r['exit_code']) is not int or r['expected_exit_code']!=r['exit_code'] or type(r['expected_exit_code']) is not int or not _str(r['evidence']): _fail('invalid finding execution')
+            if stage=='red' and (r['exit_code']<=0 or r['failure_kind'] not in ('assertion','approved_behavior')): _fail('invalid RED evidence')
+            if stage=='green' and (r['exit_code']!=0 or r['failure_kind']!='pass'): _fail('invalid GREEN evidence')
+    return {'status':'block' if violations else 'pass','violations':violations}
+
+@_guard_errors
+def check_contracts(data):
+    _keys(data,('baseline','current','mappings','stages','precedence'))
+    b,c,m=data['baseline'],data['current'],data['mappings']
+    if not all(isinstance(x,list) for x in (b,c,m)) or not b: _fail('invalid or empty contracts')
+    def unique_rows(rows,base):
+        out={}
+        for r in rows:
+            fields=('semantic_key','baseline_revision','path','meaning') if base else ('semantic_key','meaning','observation','evidence')
+            _keys(r,fields); k=r['semantic_key']
+            if not _str(k) or k in out or not _str(r['meaning']): _fail('invalid contract row')
+            if base:
+                if not _hex(r['baseline_revision'],_HEX40) or not isinstance(r['path'],str) or not r['path'] or r['path'].startswith('/') or any(x in ('','.','..') for x in r['path'].split('/')): _fail('invalid baseline contract')
+            else:
+                if r['observation'] not in ('observed','unknown','not_applicable') or not isinstance(r['evidence'],str) or not r['evidence'].strip() or r['observation']=='unknown': _fail('unverified current contract')
+            out[k]=r
+        return out
+    old=unique_rows(b,True); new=unique_rows(c,False)
+    violations=[]; maps={}
+    for r in m:
+        _keys(r,('semantic_key','current_keys','status','authorization','verification'))
+        k=r['semantic_key']
+        if k not in old or k in maps or not isinstance(r['current_keys'],list) or any(not isinstance(x,str) for x in r['current_keys']) or len(set(r['current_keys']))!=len(r['current_keys']) or any(x not in new for x in r['current_keys']) or not isinstance(r['authorization'],str) or not isinstance(r['verification'],str): _fail('invalid mapping')
+        maps[k]=r
+    if set(maps)!=set(old): _fail('incomplete mappings')
+    claimed=set()
+    for k,r in maps.items():
+        status=r['status']; keys=r['current_keys']
+        if status not in ('preserved','intentionally_changed','missing','ambiguous'): _fail('invalid mapping status')
+        if status in ('missing','ambiguous'): violations.append('unresolved contract: '+k)
+        elif not keys or not r['verification'].strip(): violations.append('contract mapping lacks verification: '+k)
+        if status=='intentionally_changed' and not r['authorization'].strip(): violations.append('contract change lacks authorization: '+k)
+        claimed.update(keys)
+    if claimed != set(new): _fail('unmapped current contract')
+    stages=data['stages']; edges=data['precedence']
+    if not isinstance(stages,list) or not stages or any(not _str(x) for x in stages) or len(set(stages))!=len(stages): _fail('invalid stages')
+    if not isinstance(edges,list): _fail('invalid precedence')
+    graph={x:[] for x in stages}; indegree={x:0 for x in stages}; seen=set()
+    for e in edges:
+        if not isinstance(e,list) or len(e)!=2 or any(not isinstance(x,str) or x not in graph for x in e) or tuple(e) in seen: _fail('invalid precedence edge')
+        seen.add(tuple(e)); graph[e[0]].append(e[1]); indegree[e[1]]+=1
+    queue=[x for x in stages if indegree[x]==0]; n=0
+    while queue:
+        x=queue.pop(); n+=1
+        for y in graph[x]:
+            indegree[y]-=1
+            if indegree[y]==0: queue.append(y)
+    if n!=len(stages): violations.append('precedence cycle')
+    return {'status':'block' if violations else 'pass','violations':violations}
+
+def main(argv=None):
+    p=argparse.ArgumentParser(); s=p.add_subparsers(dest='cmd',required=True)
+    x=s.add_parser('snapshot'); x.add_argument('--repo-root',required=True); x.add_argument('--paths-json',required=True); x.add_argument('--output',required=True)
+    x=s.add_parser('compare-source'); x.add_argument('--repo-root',required=True); x.add_argument('--before',required=True); x.add_argument('--paths-json',required=True)
+    for name in ('check-tests','check-contracts'):
+        x=s.add_parser(name); x.add_argument('--input',required=True)
+    a=p.parse_args(argv)
+    try:
+        if a.cmd=='snapshot': result=snapshot(a.repo_root,_json(a.paths_json),a.output)
+        elif a.cmd=='compare-source': result=compare_source(a.repo_root,a.before,_json(a.paths_json))
+        elif a.cmd=='check-tests': result=check_tests(_json(a.input))
+        else: result=check_contracts(_json(a.input))
+        print(json.dumps(result,sort_keys=True)); return 0 if result['status']=='pass' else 1
+    except (GuardError,OSError,ValueError,TypeError,RecursionError):
+        print(json.dumps({'status':'block','violations':['workflow guard rejected input']},sort_keys=True)); return 2
+if __name__=='__main__': raise SystemExit(main())

harness/operations/repository-cache.md

@@ -321,6 +321,152 @@ Issue worktree workflow의 완료 보고에는 다음도 포함한다.
 
 ## Canonical external Issue verification artifacts
 
+### Shared workflow guard gates
+
+Workflow guard는 기존 위임·test-first·계약 보존 규칙의 구조적 완료 gate다. 쓰기를 막는 sandbox, 기록된 주장의 진실성을 인증하는 수단, 자연어 의미 판정기 또는 자동 복구 도구가 아니다. 보호 경로 밖 변경이나 변경 후 원상 복구까지 없었다고 증명하지 않는다. 실제 승인 근거·관찰·assertion 검출력은 main agent와 독립 reviewer가 확인한다.
+
+지원 CLI는 아래 네 가지다. `ROOT`는 사용이 이미 승인된 exact Git checkout root이며 raw 절대경로의 alias·symlink를 허용하지 않는다. `INPUT`과 JSON input은 이미 존재하는 bounded 기록이고 snapshot 출력의 parent는 실행 UID 소유 mode `0700`이어야 한다. Helper는 directory를 만들거나 기존 권한을 repair하지 않고 새 snapshot만 mode `0600`·no-clobber로 생성한다.
+
+```sh
+python3 <harness-root>/harness/lib/workflow_guard.py snapshot --repo-root ROOT --paths-json INPUT --output SNAPSHOT
+python3 <harness-root>/harness/lib/workflow_guard.py compare-source --repo-root ROOT --before SNAPSHOT --paths-json INPUT
+python3 <harness-root>/harness/lib/workflow_guard.py check-tests --input JSON
+python3 <harness-root>/harness/lib/workflow_guard.py check-contracts --input JSON
+```
+
+- **기록 위치:** 제품 checkout 밖의 기존 `verification/<run-id>`에 snapshot·inventory·command/result를 보존한다. 먼저 `verification_root.py prepare`로 해당 run의 identity·lease·권한을 확인하고 기존 `evidence_manifest.py run/validate`의 command/claim 기록에 연결한다. 경로·v2 manifest schema를 바꾸거나 generic run record를 추가하지 않는다. Spec의 native-review 운영 artifact는 기존 trusted `harness/.runtime` 예외를 유지한다. Helper의 snapshot 출력은 검사 대상 Git root 밖이어야 하므로 harness 자체 유지보수에서 내부 진단 위치를 검사 대상 root의 예외로 자동 허용하지 않는다.
+- **읽기 전용 위임:** 위임 전에 보호 경로 JSON array와 HEAD, 파일 존재·종류·identity·digest를 immutable protected scope로 고정한다. 결과 채택 전에 같은 경로 집합으로 비교한다. 시작 baseline 누락·source 변경·경로 집합 drift는 검토 근거를 무효화하며 PASS를 채택하거나 자동 restore하지 않는다. 쓰기 worker는 승인된 변경 allowlist 안에서 수정하고 main agent가 실제 diff·test를 확인한다. 허용된 편집을 읽기 전용 위반으로 취급하지 않는다. 최종 독립 검토 중에는 보호 대상 mutator를 병렬 실행하지 않는다. Spec reviewer의 immutable provider/Git blob binding과 moving checkout 사용 금지는 그대로이며 이 gate가 별도 checkout·제품 test 실행 권한을 주지 않는다.
+- **기존 테스트:** 첫 편집 전에 canonical collection의 ID·skip·expected-failure 상태와 collection command/source/output 근거를 기록한다. 새 regression 추가는 허용한다. 기존 ID 삭제·rename·새 skip/xfail에는 exact old-to-current mapping, 승인 근거와 대체/의미 검토 근거가 필요하다. 변경된 assertion·fixture·mock의 의미 검토도 별도로 연결한다. 같은 개수·이름·exit 0만으로 검출력 보존을 판정하지 않는다. 수집하지 않은 framework나 case를 수집 완료라고 기록하지 않는다.
+- **Finding 재현:** 유효한 finding마다 current inventory의 named test와 구조화된 RED/GREEN 참조를 연결한다. 각 참조는 test ID, source revision/scope digest, command ID/argv, 실제·예상 exit와 evidence를 담는다. RED는 의도한 assertion/승인 동작의 nonzero 실패이고 GREEN은 exit 0이며 import·환경 오류를 RED로 인정하지 않는다. 반증된 지적에는 가짜 RED를 만들지 않는다. Helper는 참조 형식·연결·exit 조건을 확인하고 main agent는 실제 기록을 읽어 source binding과 실패 원인을 검증한다.
+- **계약·단계:** 변경 영향 범위의 기존 requirement뿐 아니라 API/CLI/응답 필드/상태 전이/cleanup 의미를 baseline inventory로 고정한다. 각 항목은 실제 current 항목과 검증 근거에 정확히 대응하거나 승인된 의도적 변경을 설명해야 한다. `missing`·`ambiguous`는 승인 문자열로 통과시킬 수 없고, 실제 public-response 관찰이 미확인이면 observed로 바꾸지 않는다. 초기화 선행 조건의 모순·미지 stage·순환 precedence를 거부하되 정상 renew 등 lifecycle loop를 초기화 precedence로 모델링해 금지하지 않는다.
+- **판정:** helper의 `status=pass`·exit 0은 제출된 구조적 gate가 통과했다는 뜻이며 제품 승인이나 evidence 인증을 대신하지 않는다. `status=block`/nonzero이면 해당 운영 근거를 채택하지 않고 누락·변경·판단 불가를 보고한다. 운영 gate 실패만으로 새로운 제품 acceptance criterion이나 자동 `REQUEST_CHANGES`를 만들지 않는다.
+
+Input schema와 엄격한 type/key 규칙은 `harness/lib/workflow_guard.py`의 API 계약을 따른다. `paths-json`은 중복 없는 기존 repository-relative protected 파일 array다. `check-tests`는 `baseline_ids/current_ids`, `baseline_status/current_status`, source·collection argv·inventory digest·evidence를 가진 `baseline_collection/current_collection`, `authorized_changes`, `changed_tests`, `findings`를 연결한다. `check-contracts`는 `baseline/current/mappings`와 `stages/precedence`를 연결한다. 파일별 의미를 임의 추론하지 말고 helper의 실제 schema와 기록을 검증한다. 이 문서의 정적 contract test 통과는 agent가 실행 중 절차를 준수했다는 증명이 아니다.
+
+#### Nested input contracts
+
+All object keys below are required and exact; unknown/missing keys, duplicate JSON keys, duplicate list identities, incorrect scalar types and non-regular/unbounded JSON inputs block. Empty strings are allowed only where explicitly stated. References are nonempty strings identifying existing verification command/result or approval records, not their secret contents. Full source revision is lowercase 40-hex; byte/scope digests are lowercase 64-hex. Integer fields reject booleans. CLI exit codes are 0 for structural PASS, 1 for detected violations/drift, 2 for rejected input; APIs raise `GuardError` for rejected input or return a status/violations result.
+
+**Test input:** exactly `baseline_ids`, `current_ids`, `baseline_status`, `current_status`, `baseline_collection`, `current_collection`, `authorized_changes`, `changed_tests`, `findings`.
+
+- ID arrays contain unique nonempty strings. Each status object has exactly that inventory's ID keys; each value is `{"skip": boolean, "expected_failure": boolean}`. No omitted status is silently treated as false.
+- Each collection object is `{source_revision: full SHA, command_id: reference, argv: nonempty array of nonempty strings, inventory_sha256: digest, evidence: reference}`. Digest is SHA256 over UTF-8 bytes of `json.dumps({"ids": sorted(ids), "statuses": {id: statuses[id] for id in sorted(ids)}}, sort_keys=True, separators=(',', ':'))`; use Python's default JSON escaping. Main verifies collection actually covered the declared frameworks/cases.
+- `authorized_changes` is an array of `{old_id: baseline ID, kind: "delete" | "rename" | "status", replacement_ids: unique current ID array, reason: nonempty string, approval: reference, evidence: semantic-review reference}`. A delete requires the old ID absent and an empty replacement; a rename requires the old ID absent and at least one actual current replacement; status requires the old ID still present and exactly `[old_id]`. Each old ID has at most one approval row. No replacement is silently invented or reused by another approval. Unclassified empty mappings block. Ordinary added regression IDs do not require removal approval. Every removed ID and newly introduced skip/xfail is covered by a compatible explicit approval; new skipped tests need explicit replacement coverage, not an unrelated deletion.
+- `changed_tests` is an array of `{test_id: current ID, baseline_sha256: digest, current_sha256: digest, review_evidence: reference}`, unique by test ID. Main inventories actual assertion/fixture/mock edits; helper cannot discover an omitted changed-test declaration or prove semantic preservation.
+- `findings` is an array of `{id: unique nonempty string, test_id: current ID, classification: "assertion" | "approved_behavior", red: execution, green: execution}`. Each execution is `{test_id: same finding test, source_revision: full SHA, scope_sha256: digest, command_id: reference, argv: nonempty string array, exit_code: integer, expected_exit_code: same integer, failure_kind: string, evidence: reference}`. RED requires positive nonzero exit and failure_kind `assertion` or `approved_behavior`; GREEN requires exit 0 and failure_kind `pass`. Import/environment failure is invalid RED. Main inspects actual logs and ties each command to the named test and its exact tested source; expected/actual code equality alone does not prove that.
+
+**Contract input:** exactly `baseline`, `current`, `mappings`, `stages`, `precedence`.
+
+- `baseline` is a nonempty array of `{semantic_key: unique nonempty string, baseline_revision: full SHA, path: repository-relative path without empty/dot/traversal segments, meaning: nonempty string}`. It inventories old semantics, not merely current document headings.
+- `current` is an array of `{semantic_key: unique nonempty string, meaning: nonempty string, observation: "observed" | "not_applicable" | "unknown", evidence: reference}`. `unknown` blocks even with an evidence string; justified not_applicable is not a fabricated public-response observation. Main must check actual source/static contract and public response as applicable.
+- `mappings` is one row per baseline semantic key: `{semantic_key: baseline key, current_keys: unique current-key array, status: "preserved" | "intentionally_changed" | "missing" | "ambiguous", authorization: string, verification: string}`. Resolved rows require nonempty current_keys and verification; intentionally_changed additionally requires explicit authorization. Authorization may be empty for preserved rows. Missing/ambiguous always block even with authorization. New current keys and approved merges are permitted with complete old-to-current coverage: every old key retains its own row and semantic verification, and every current key is referenced. Several old rows may reference one merged current row; this is not permission to omit old meanings. Main checks preservation/substitution/merge semantics and real approvals independently.
+- `stages` is a nonempty unique string array; `precedence` is an array of unique `[prerequisite, dependent]` string pairs, both declared stages. Initialization precedence must be acyclic. Do not encode allowed renew/lifecycle loops as startup prerequisites.
+
+**Binding procedure (shape-only, not evidence authentication):** collection inventory digest binds submitted ID/status data, not its truth. Helper does not open/authenticate evidence/approval references or compare their source_revision/scope_sha256 to a checkout. Before adopting PASS, main reads the referenced existing immutable command/result and approval records, verifies exact HEAD/provider blob and relevant source digests, actual argv/environment/exit/output and named RED failure cause, then independently reviews changed test/contract semantics. Missing/unreadable/wrong-source evidence is `evidence_gap`, not a product defect by itself. Product execution remains in existing `verification/<run-id>` with v2 manifest run/validate and exact-head readback; spec uses immutable authoritative blobs and its existing trusted runtime exception, never a moving implementation checkout or newly mandated product execution.
+
+The following two inputs are runnable **synthetic structure examples only**; the `a` revision and synthetic evidence strings are not genuine verification records and cannot be adopted as product evidence. They demonstrate the exact nested shape and actual calculated inventory digest. For production, replace them with verified collection, source and evidence values before gate adoption.
+
+```json
+{
+  "baseline_ids": [
+    "case_a"
+  ],
+  "current_ids": [
+    "case_a"
+  ],
+  "baseline_status": {
+    "case_a": {
+      "skip": false,
+      "expected_failure": false
+    }
+  },
+  "current_status": {
+    "case_a": {
+      "skip": false,
+      "expected_failure": false
+    }
+  },
+  "baseline_collection": {
+    "source_revision": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "command_id": "fixture-collection",
+    "argv": [
+      "python3",
+      "-B",
+      "-m",
+      "unittest",
+      "case_a"
+    ],
+    "inventory_sha256": "b2fac8d5f9e294a3d2c642955deab1bbf3bf8c9cf696526be56f2661b7eba455",
+    "evidence": "synthetic-collection-record"
+  },
+  "current_collection": {
+    "source_revision": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "command_id": "fixture-collection",
+    "argv": [
+      "python3",
+      "-B",
+      "-m",
+      "unittest",
+      "case_a"
+    ],
+    "inventory_sha256": "b2fac8d5f9e294a3d2c642955deab1bbf3bf8c9cf696526be56f2661b7eba455",
+    "evidence": "synthetic-collection-record"
+  },
+  "authorized_changes": [],
+  "changed_tests": [],
+  "findings": []
+}
+```
+
+```json
+{
+  "baseline": [
+    {
+      "semantic_key": "ready-fields",
+      "baseline_revision": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+      "path": "src/broker.py",
+      "meaning": "Public ready fields are handle/socket."
+    }
+  ],
+  "current": [
+    {
+      "semantic_key": "ready-fields",
+      "meaning": "Public ready fields are handle/socket.",
+      "observation": "observed",
+      "evidence": "synthetic-response-record"
+    }
+  ],
+  "mappings": [
+    {
+      "semantic_key": "ready-fields",
+      "current_keys": [
+        "ready-fields"
+      ],
+      "status": "preserved",
+      "authorization": "",
+      "verification": "synthetic-contract-record"
+    }
+  ],
+  "stages": [
+    "runtime",
+    "ready",
+    "admission"
+  ],
+  "precedence": [
+    [
+      "runtime",
+      "ready"
+    ],
+    [
+      "ready",
+      "admission"
+    ]
+  ]
+}
+```
+
+Invalid examples: rename with `replacement_ids: []` blocks (not an implicit deletion); `expected_exit_code: true` blocks; unknown observation or missing contract with nonempty authorization blocks; duplicate current contract rows and precedence cycles block. Static documentation tests establish documentation/schema consistency only, not agent runtime compliance.
+
 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 결과에 맞춰 자동 정규화하지 않는다.

harness/tests/common/test_verification_contract_docs.py

@@ -123,6 +123,68 @@ class VerificationContractDocumentationTests(unittest.TestCase):
         self.assertIn("trusted `harness/.runtime`", section)
         self.assertIn("Producer evidence와 reviewer 실행을 구분", section)
 
+    def test_workflow_guard_records_reuse_existing_verification_location(self):
+        text = CACHE.read_text()
+        guard = text.split('### Shared workflow guard gates', 1)[1].split('Issue별 검증 실행 기록의 정본', 1)[0]
+        self.assertNotIn('or verification root', guard)
+        for path in (SKILLS['implementation'], SKILLS['rework']):
+            self.assertNotIn('outside product checkout and verification root', path.read_text())
+        self.assertIn('verification/<run-id>', guard)
+
+    def test_workflow_guard_readonly_scope_does_not_ban_approved_worker_edits(self):
+        text = CACHE.read_text()
+        guard = text.split('### Shared workflow guard gates', 1)[1].split('Issue별 검증 실행 기록의 정본', 1)[0]
+        self.assertIn('읽기 전용', guard)
+        self.assertIn('쓰기 worker는 승인된 변경 allowlist', guard)
+
+    def test_workflow_guard_snapshot_protects_scope_before_delegation(self):
+        implementation = SKILLS["implementation"].read_text()
+        self.assertIn("workflow_guard.py snapshot", implementation)
+        self.assertIn("compare-source --repo-root", implementation)
+        self.assertIn("immutable protected scope", implementation)
+
+    def test_workflow_guard_inventory_is_checked_before_adopting_results(self):
+        rework = SKILLS["rework"].read_text()
+        self.assertIn("workflow_guard.py check-tests", rework)
+        self.assertIn("test inventory", rework)
+        self.assertIn("RED/GREEN records must be named", rework)
+
+    def test_workflow_guard_contract_preservation_precedes_spec_verdict(self):
+        spec = SKILLS["review"].read_text()
+        self.assertIn("workflow_guard.py check-contracts", spec)
+        self.assertIn("baseline inventory", spec)
+        self.assertIn("Gate failure invalidates workflow evidence", spec)
+
+    def test_workflow_guard_nested_schema_is_actionable_and_claims_are_not_authenticated(self):
+        text = CACHE.read_text()
+        section = text.split('### Shared workflow guard gates', 1)[1].split('Issue별 검증 실행 기록의 정본', 1)[0]
+        for token in ('#### Nested input contracts', 'baseline_collection', 'inventory_sha256',
+                      'baseline_status', 'expected_failure', 'replacement_ids', 'kind',
+                      'delete', 'rename', 'status', 'review_evidence', 'scope_sha256',
+                      'expected_exit_code', 'failure_kind', 'semantic_key', 'current_keys',
+                      'observation', 'intentionally_changed', 'sort_keys=True',
+                      'separators', 'UTF-8', 'shape-only', 'one row per baseline semantic key'):
+            with self.subTest(token=token):
+                self.assertIn(token, section)
+        import json
+        blocks = re.findall(r'```json\n(.*?)```', section, re.S)
+        self.assertEqual(len(blocks), 2, 'provide runnable test and contract input examples')
+        self.assertEqual(set(json.loads(blocks[0])), {'baseline_ids', 'current_ids', 'baseline_status',
+            'current_status', 'baseline_collection', 'current_collection', 'authorized_changes', 'changed_tests', 'findings'})
+        self.assertEqual(set(json.loads(blocks[1])), {'baseline', 'current', 'mappings', 'stages', 'precedence'})
+
+    def test_mapping_schema_types_use_raw_source_not_redacted_tool_rendering(self):
+        # Authorization-looking prose can be masked by read-tool transport.
+        # Inspect the approved public schema in raw source; never disable redaction.
+        declaration = 'authorization: string, verification: string'
+        masked = 'authorization: ' + '*' * 3 + ' verification'
+        helper = ROOT / 'harness/lib/workflow_guard.py'
+        for path in (helper, CACHE):
+            text = path.read_text()
+            with self.subTest(path=path.name):
+                self.assertEqual(text.count(declaration), 1)
+                self.assertNotIn(masked, text)
+
     def test_rework_no_longer_excludes_repo_evidence_from_clean_tree(self):
         text = SKILLS["rework"].read_text()
         self.assertNotIn("plan과 manifest를 제외한 clean tree", text)

harness/tests/common/test_workflow_guard.py

@@ -0,0 +1,420 @@
+import copy
+import hashlib
+import json
+import os
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from harness.lib import workflow_guard as guard
+
+
+def inventory_digest(ids, statuses):
+    value = {'ids': sorted(ids), 'statuses': {key: statuses[key] for key in sorted(ids)}}
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def test_inventory(ids=('case_a',), current=None):
+    current = list(ids if current is None else current)
+    ids = list(ids)
+    old_status = {key: {'skip': False, 'expected_failure': False} for key in ids}
+    new_status = {key: {'skip': False, 'expected_failure': False} for key in current}
+    def collection(names, statuses):
+        return {'source_revision': 'a' * 40, 'command_id': 'collect-tests',
+                'argv': ['python3', '-B', '-m', 'unittest', 'discover'],
+                'inventory_sha256': inventory_digest(names, statuses),
+                'evidence': 'fixture-collection-record'}
+    return {'baseline_ids': ids, 'current_ids': current,
+            'baseline_status': old_status, 'current_status': new_status,
+            'baseline_collection': collection(ids, old_status),
+            'current_collection': collection(current, new_status),
+            'authorized_changes': [], 'changed_tests': [], 'findings': []}
+
+
+def contract_inventory():
+    return {'baseline': [{'semantic_key': 'ready-fields', 'baseline_revision': 'a' * 40,
+                          'path': 'src/broker.py', 'meaning': 'Public ready fields are handle/socket.'}],
+            'current': [{'semantic_key': 'ready-fields', 'meaning': 'Public ready fields are handle/socket.',
+                         'observation': 'observed', 'evidence': 'fixture-response-check'}],
+            'mappings': [{'semantic_key': 'ready-fields', 'current_keys': ['ready-fields'],
+                          'status': 'preserved', 'authorization': '', 'verification': 'fixture-static-contract'}],
+            'stages': ['runtime', 'ready', 'admission'],
+            'precedence': [['runtime', 'ready'], ['ready', 'admission']]}
+
+
+def finding_record():
+    def execution(code, kind):
+        return {'test_id': 'case_a', 'source_revision': 'a' * 40,
+                'scope_sha256': 'b' * 64, 'command_id': 'named-regression',
+                'argv': ['python3', '-B', '-m', 'unittest', 'case_a'],
+                'exit_code': code, 'expected_exit_code': code,
+                'failure_kind': kind, 'evidence': 'fixture-execution-record'}
+    return {'id': 'finding-a', 'test_id': 'case_a', 'classification': 'assertion',
+            'red': execution(1, 'assertion'), 'green': execution(0, 'pass')}
+
+
+class WorkflowGuardTests(unittest.TestCase):
+    def repository(self, directory):
+        root = Path(directory) / 'repo'
+        root.mkdir(mode=0o700)
+        subprocess.run(['git', 'init', '-q', str(root)], check=True)
+        subprocess.run(['git', '-C', str(root), '-c', 'user.name=Fixture',
+                        '-c', 'user.email=fixture@example.invalid', 'commit',
+                        '--allow-empty', '-qm', 'baseline'], check=True)
+        (root / 'src').mkdir(mode=0o700)
+        (root / 'src/a.py').write_text('one')
+        return root
+
+    def assert_block(self, function, value):
+        try:
+            result = function(value)
+        except guard.GuardError:
+            return
+        self.assertEqual(result['status'], 'block')
+
+    def test_snapshot_compare_and_change_detection(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            out = Path(td) / 'snapshot.json'
+            guard.snapshot(root, ['src/a.py'], out)
+            self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'pass')
+            (root / 'src/a.py').write_text('two')
+            self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'block')
+
+    def test_rejects_traversal_and_malformed_baseline(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            with self.assertRaises(guard.GuardError):
+                guard.snapshot(root, ['../outside'], Path(td) / 'x.json')
+            bad = Path(td) / 'bad.json'
+            bad.write_text('{}')
+            with self.assertRaises(guard.GuardError):
+                guard.compare_source(root, bad, ['src/a.py'])
+
+    def test_test_inventory_requires_identity_and_review_evidence(self):
+        self.assert_block(guard.check_tests, {'baseline_ids': ['a'], 'current_ids': ['b']})
+
+    def test_contract_inventory_requires_full_coverage(self):
+        self.assert_block(guard.check_contracts, {'baseline': [{'semantic_key': 'x'}], 'current': [], 'mappings': []})
+
+    def test_snapshot_rejects_literal_root_alias_and_symlink_root(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            alias = Path(td) / 'alias'
+            alias.symlink_to(root, target_is_directory=True)
+            for value in (str(root) + '/./', alias):
+                with self.subTest(root=str(value)), self.assertRaises(guard.GuardError):
+                    guard.snapshot(value, ['src/a.py'], Path(td) / 'snapshot.json')
+
+    def test_snapshot_rejects_symlink_output_parent(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            actual = Path(td) / 'private'
+            actual.mkdir(mode=0o700)
+            alias = Path(td) / 'alias'
+            alias.symlink_to(actual, target_is_directory=True)
+            with self.assertRaises(guard.GuardError):
+                guard.snapshot(root, ['src/a.py'], alias / 'snapshot.json')
+            self.assertFalse((actual / 'snapshot.json').exists())
+
+    def test_snapshot_exact_mode_survives_restrictive_umask(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            out = Path(td) / 'snapshot.json'
+            before = os.umask(0o777)
+            try:
+                guard.snapshot(root, ['src/a.py'], out)
+            finally:
+                os.umask(before)
+            self.assertEqual(out.stat().st_mode & 0o777, 0o600)
+
+    def test_snapshot_rejects_boolean_schema_baseline(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            out = Path(td) / 'snapshot.json'
+            guard.snapshot(root, ['src/a.py'], out)
+            value = json.loads(out.read_text())
+            value['schema'] = True
+            out.write_text(json.dumps(value))
+            with self.assertRaises(guard.GuardError):
+                guard.compare_source(root, out, ['src/a.py'])
+
+    def test_snapshot_rejects_repository_subdirectory_root(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            with self.assertRaises(guard.GuardError):
+                guard.snapshot(root / 'src', ['a.py'], Path(td) / 'snapshot.json')
+
+    def test_snapshot_detects_same_bytes_inode_replacement(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            out = Path(td) / 'snapshot.json'
+            guard.snapshot(root, ['src/a.py'], out)
+            replacement = root / 'src/new.py'
+            replacement.write_text('one')
+            replacement.replace(root / 'src/a.py')
+            self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'block')
+
+    def test_new_regression_tests_do_not_require_deletion_authorization(self):
+        self.assertEqual(guard.check_tests(test_inventory(current=['case_a', 'case_b']))['status'], 'pass')
+
+    def test_collected_inventory_requires_provenance(self):
+        value = test_inventory()
+        del value['baseline_collection']
+        self.assert_block(guard.check_tests, value)
+
+    def test_collected_inventory_digest_must_match_ids_and_statuses(self):
+        value = test_inventory()
+        value['baseline_collection']['inventory_sha256'] = '0' * 64
+        self.assert_block(guard.check_tests, value)
+
+    def test_new_skip_requires_explicit_approval(self):
+        value = test_inventory()
+        value['current_status']['case_a']['skip'] = True
+        value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status'])
+        self.assert_block(guard.check_tests, value)
+
+    def test_new_expected_failure_requires_explicit_approval(self):
+        value = test_inventory()
+        value['current_status']['case_a']['expected_failure'] = True
+        value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status'])
+        self.assert_block(guard.check_tests, value)
+
+    def test_rename_requires_real_replacement_and_semantic_evidence(self):
+        value = test_inventory(current=['case_b'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': ['ghost'],
+                                        'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_approved_rename_preserves_mapped_test(self):
+        value = test_inventory(current=['case_b'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': ['case_b'],
+                                        'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assertEqual(guard.check_tests(value)['status'], 'pass')
+
+    def test_modified_test_requires_semantic_review(self):
+        value = test_inventory()
+        value['changed_tests'] = [{'test_id': 'case_a', 'baseline_sha256': 'a' * 64,
+                                    'current_sha256': 'b' * 64, 'review_evidence': ''}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_finding_truthy_strings_are_not_red_green_records(self):
+        value = test_inventory()
+        value['findings'] = [{'id': 'finding-a', 'test_id': 'case_a', 'classification': 'assertion',
+                              'red': 'claimed-red', 'green': 'claimed-green'}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_red_success_exit_cannot_demonstrate_failure(self):
+        value = test_inventory()
+        finding = finding_record()
+        finding['red']['exit_code'] = finding['red']['expected_exit_code'] = 0
+        value['findings'] = [finding]
+        self.assert_block(guard.check_tests, value)
+
+    def test_green_failure_exit_cannot_demonstrate_resolution(self):
+        value = test_inventory()
+        finding = finding_record()
+        finding['green']['exit_code'] = finding['green']['expected_exit_code'] = 1
+        value['findings'] = [finding]
+        self.assert_block(guard.check_tests, value)
+
+    def test_environment_failure_is_not_valid_red(self):
+        value = test_inventory()
+        finding = finding_record()
+        finding['red']['failure_kind'] = 'environment'
+        value['findings'] = [finding]
+        self.assert_block(guard.check_tests, value)
+
+    def test_finding_must_name_present_regression_test(self):
+        value = test_inventory()
+        finding = finding_record()
+        finding['test_id'] = 'ghost'
+        value['findings'] = [finding]
+        self.assert_block(guard.check_tests, value)
+
+    def test_named_red_green_records_with_provenance_pass(self):
+        value = test_inventory()
+        value['findings'] = [finding_record()]
+        self.assertEqual(guard.check_tests(value)['status'], 'pass')
+
+    def test_missing_and_ambiguous_contracts_cannot_be_approved_away(self):
+        for status in ('missing', 'ambiguous'):
+            with self.subTest(status=status):
+                value = contract_inventory()
+                value['mappings'][0].update(status=status, authorization='fixture-approval')
+                self.assert_block(guard.check_contracts, value)
+
+    def test_preserved_contract_requires_existing_current_mapping(self):
+        value = contract_inventory()
+        value['current'] = []
+        self.assert_block(guard.check_contracts, value)
+
+    def test_unknown_observation_is_not_actual_public_response_evidence(self):
+        value = contract_inventory()
+        value['current'][0].update(observation='unknown', evidence='')
+        self.assert_block(guard.check_contracts, value)
+
+    def test_preserved_contract_requires_verification(self):
+        value = contract_inventory()
+        value['mappings'][0]['verification'] = ''
+        self.assert_block(guard.check_contracts, value)
+
+    def test_precedence_cycle_blocks_startup_order(self):
+        value = contract_inventory()
+        value['precedence'].append(['admission', 'runtime'])
+        self.assert_block(guard.check_contracts, value)
+
+    def test_complete_observed_contract_and_acyclic_order_pass(self):
+        self.assertEqual(guard.check_contracts(contract_inventory())['status'], 'pass')
+
+    def test_duplicate_current_contract_is_not_coverage(self):
+        value = contract_inventory()
+        value['current'].append(copy.deepcopy(value['current'][0]))
+        self.assert_block(guard.check_contracts, value)
+
+    def test_invalid_stage_types_fail_closed_without_unhandled_typeerror(self):
+        value = contract_inventory()
+        value['stages'] = [['unhashable']]
+        self.assert_block(guard.check_contracts, value)
+
+    def guard_cli(self, *args):
+        command = ['python3', '-B', str(Path(guard.__file__).resolve()), *map(str, args)]
+        try:
+            return subprocess.run(command, capture_output=True, text=True, timeout=3)
+        except subprocess.TimeoutExpired:
+            self.fail('guard did not fail closed within 3 seconds')
+
+    def test_snapshot_fifo_protected_file_fails_fast(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            os.mkfifo(root / 'src/pipe')
+            paths = Path(td) / 'paths.json'
+            paths.write_text('["src/pipe"]')
+            result = self.guard_cli('snapshot', '--repo-root', root, '--paths-json', paths,
+                                    '--output', Path(td) / 'snapshot.json')
+            self.assertEqual(result.returncode, 2)
+            self.assertEqual(json.loads(result.stdout)['status'], 'block')
+
+    def test_compare_source_fifo_protected_file_fails_fast(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            out = Path(td) / 'snapshot.json'
+            guard.snapshot(root, ['src/a.py'], out)
+            (root / 'src/a.py').unlink()
+            os.mkfifo(root / 'src/a.py')
+            paths = Path(td) / 'paths.json'
+            paths.write_text('["src/a.py"]')
+            result = self.guard_cli('compare-source', '--repo-root', root, '--paths-json', paths,
+                                    '--before', out)
+            self.assertEqual(result.returncode, 1)  # Valid baseline; observed source-type drift.
+            self.assertEqual(json.loads(result.stdout)['status'], 'block')
+
+    def test_snapshot_rejects_root_ending_dotdot(self):
+        with tempfile.TemporaryDirectory() as td:
+            root = self.repository(td)
+            with self.assertRaises(guard.GuardError):
+                guard.snapshot(str(root / 'src') + '/..', ['src/a.py'], Path(td) / 'snapshot.json')
+
+    def test_snapshot_rejects_ancestor_symlink_root(self):
+        with tempfile.TemporaryDirectory() as td:
+            self.repository(td)
+            alias = Path(td) / 'alias'
+            alias.symlink_to(td, target_is_directory=True)
+            with self.assertRaises(guard.GuardError):
+                guard.snapshot(alias / 'repo', ['src/a.py'], Path(td) / 'snapshot.json')
+
+    def test_json_fifo_input_fails_fast(self):
+        with tempfile.TemporaryDirectory() as td:
+            pipe = Path(td) / 'pipe.json'
+            os.mkfifo(pipe)
+            result = self.guard_cli('check-tests', '--input', pipe)
+            self.assertEqual(result.returncode, 2)
+            self.assertEqual(json.loads(result.stdout)['status'], 'block')
+
+    def test_deeply_nested_json_blocks_without_traceback(self):
+        with tempfile.TemporaryDirectory() as td:
+            bad = Path(td) / 'bad.json'
+            bad.write_text('[' * 2000 + ']' * 2000)
+            result = self.guard_cli('check-tests', '--input', bad)
+            self.assertEqual(result.returncode, 2)
+            self.assertNotIn('Traceback', result.stderr)
+            self.assertEqual(json.loads(result.stdout)['status'], 'block')
+
+    def test_rename_empty_replacement_cannot_stand_in_for_deletion(self):
+        value = test_inventory(current=['case_b'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': [],
+            'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_unclassified_empty_replacement_cannot_stand_in_for_deletion(self):
+        value = test_inventory(current=['case_b'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'replacement_ids': [],
+            'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_explicit_approved_deletion_is_allowed(self):
+        value = test_inventory(ids=['case_a', 'case_b'], current=['case_b'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'delete', 'replacement_ids': [],
+            'reason': 'approved obsolete case deletion', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assertEqual(guard.check_tests(value)['status'], 'pass')
+
+    def test_status_approval_cannot_authorize_removal(self):
+        value = test_inventory(current=['case_b'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'status', 'replacement_ids': ['case_b'],
+            'reason': 'approved status update', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_explicit_status_approval_maps_same_current_id(self):
+        value = test_inventory()
+        value['current_status']['case_a']['skip'] = True
+        value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status'])
+        value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'status', 'replacement_ids': ['case_a'],
+            'reason': 'approved skip', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assertEqual(guard.check_tests(value)['status'], 'pass')
+
+    def test_unhashable_authorization_id_fails_closed_api(self):
+        value = test_inventory()
+        value['authorized_changes'] = [{'old_id': ['case_a'], 'replacement_ids': [],
+            'reason': 'fixture', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
+        self.assert_block(guard.check_tests, value)
+
+    def test_approved_contract_substitution_can_map_new_current_key(self):
+        value = contract_inventory()
+        value['current'][0]['semantic_key'] = 'published-ready-fields'
+        value['mappings'][0].update(current_keys=['published-ready-fields'],
+            status='intentionally_changed', authorization='fixture-approved-substitution')
+        self.assertEqual(guard.check_contracts(value)['status'], 'pass')
+
+    def test_approved_contract_merge_retains_each_old_semantic_mapping(self):
+        value = contract_inventory()
+        value['baseline'].append({'semantic_key': 'state-fields', 'baseline_revision': 'a' * 40,
+            'path': 'src/broker.py', 'meaning': 'Public state is held.'})
+        value['current'][0].update(semantic_key='combined-fields', meaning='Public fields are handle/socket and state held.')
+        value['mappings'] = [{'semantic_key': key, 'current_keys': ['combined-fields'],
+            'status': 'intentionally_changed', 'authorization': 'fixture-approved-merge',
+            'verification': 'fixture-checked-each-old-meaning'} for key in ('ready-fields', 'state-fields')]
+        self.assertEqual(guard.check_contracts(value)['status'], 'pass')
+
+    def test_schema_docstring_lists_nested_fields_and_exit_rules(self):
+        for field in ('baseline_status', 'current_status', 'skip', 'expected_failure',
+                      'authorized_changes', 'old_id', 'kind', 'replacement_ids', 'changed_tests',
+                      'baseline_sha256', 'current_sha256', 'review_evidence', 'findings', 'scope_sha256',
+                      'expected_exit_code', 'failure_kind', 'baseline_revision', 'semantic_key',
+                      'current_keys', 'observation', 'authorization', 'verification', 'precedence',
+                      "sort_keys=True", "separators=(',', ':')", 'RED', 'GREEN', 'delete', 'rename', 'status'):
+            with self.subTest(field=field):
+                self.assertIn(field, guard.__doc__ or '')
+
+    def test_declared_evidence_shape_does_not_authenticate_records(self):
+        # Deliberately synthetic fixture claims: the main agent must check actual evidence separately.
+        self.assertEqual(guard.check_tests(test_inventory())['status'], 'pass')
+
+    def test_unknown_inventory_fields_are_rejected(self):
+        value = test_inventory()
+        value['pretend_verified'] = True
+        self.assert_block(guard.check_tests, value)
+
+
+if __name__ == '__main__':
+    unittest.main()