diff --git a/harness/agents/review/skills/gitea-code-review/SKILL.md b/harness/agents/review/skills/gitea-code-review/SKILL.md
index e343223..90a54ba 100644
--- a/harness/agents/review/skills/gitea-code-review/SKILL.md
+++ b/harness/agents/review/skills/gitea-code-review/SKILL.md
@@ -1,7 +1,7 @@
 ---
 name: gitea-code-review
 description: Spec pass된 Gitea PR의 코드 품질과 안전성을 검토한다.
-version: 0.1.0
+version: 0.1.1
 author: genichin, Hermes Agent
 license: MIT
 platforms: [linux]
@@ -67,6 +67,8 @@ Finding마다 `F-001` 같은 ID, 파일·행, 발생 조건, 영향·심각도,
 
 ## Native review 게시와 재검토
 
+게시 단계는 [공통 Native review adapter](references/native-review.md)와 canonical `scripts/native_review.py`의 검증된 API를 사용한다. 매 실행마다 gate·게시·검증 bootstrap을 다시 생성하거나 지원하지 않는 CLI·legacy JSON schema를 추정하지 않는다. 기존 role-authenticated transport와 현재 보유한 Issue lease를 연결하고, durable journal을 유지해 timeout·partial 이후에는 POST 재시도 없이 readback만 수행한다. `provider_contract`는 확인된 Gitea 자동 supersession/retouch 및 commit `files` 순서 변화만 제한적으로 허용하며 head·본문·기존 리뷰 내용을 비교에서 제외하지 않는다. Snapshot/plan schema, 실제 지원 interface 및 read-only replay와 live 게시의 차이는 참조 문서대로 확인한다.
+
 1. Exact PR 리뷰 요청은 진입 조건을 충족한 native review 하나의 게시만 승인한다. 단순 조사·역할 보고 요청은 write를 승인하지 않는다. `approve → APPROVED`, `request_changes → REQUEST_CHANGES`로 매핑하고 `blocked`는 게시 없이 보고한다.
 2. 게시 직전 Issue 계약, PR base·head, spec pass와 review baseline을 재확인한다. Drift면 중단한다. Body에는 결과 artifact의 요약과 아래 marker를 넣는다.
    `<!-- gitea-agent:code-review/v1 owner=<owner> repository=<repository> issue=<issue-number> pr=<pull-number> head=<full-head-sha> contract-sha256=<digest> -->`
diff --git a/harness/agents/review/skills/gitea-code-review/references/native-review.md b/harness/agents/review/skills/gitea-code-review/references/native-review.md
new file mode 100644
index 0000000..698ba0d
--- /dev/null
+++ b/harness/agents/review/skills/gitea-code-review/references/native-review.md
@@ -0,0 +1,98 @@
+# Native code-review publication adapter
+
+Use the canonical harness source at `<HARNESS_ROOT>/harness/agents/review/skills/gitea-code-review/scripts/native_review.py`. Do not copy the helper into a standalone profile: it reuses public JSON, complete-pagination, cache-path and canonical spec-review codecs from the adjacent harness spec skill. The spec publisher/renderer and spec source files are not changed.
+
+This adapter is a publication guard, not a substitute for the main reviewer's complete diff, surrounding-code, test and runtime review. Keep the existing approval/blocker rubric. A `blocked` review is local only and has no Native write payload.
+
+## Supported interface
+
+- Python API: `build_payload(plan)`, `validate_snapshot(plan, snapshot)`, `verify_publication(plan, before, after, exact)`, and `publish(plan, baseline, api, lease, journal_directory)`.
+- REST bridge: `ApiAdapter(authenticated_request, origin=..., credential_source=...)`.
+- Existing lease bridge: `HeldIssueLease(issue_fd, absolute_cache_policy, deadline_monotonic)`.
+- Read-only CLI: `preflight --plan PLAN --baseline SNAPSHOT` and `verify --plan PLAN --before BEFORE --after AFTER --exact EXACT`.
+- There is intentionally **no live `publish` CLI**. Live publication uses the Python API with the workflow's existing authenticated transport and still-held Issue lock. Never invent a `capture`, `build`, `publish`, or `submit` command or guess a legacy helper schema.
+
+Use Python 3 stdlib; no package installation is needed. Discover an absolute canonical `HARNESS_ROOT` using the repository-cache contract first. Invoke the canonical helper, not a relocated skill copy.
+
+## Plan schema: `gitea-agent.code-review-publication/v1`
+
+All fields are explicit; extra/missing fields fail closed.
+
+- `schema`: the literal schema above.
+- `target`: `owner`, `repository`, positive integer `issue`, positive integer `pull`, lowercase full 40-hex `head` and `base`.
+- `contract_sha256`: lowercase 64-hex SHA-256 of the exact, unmodified UTF-8 Discovery Issue body.
+- `reviewer`: exact `{id, login}` from authenticated `/user`; login must be `gitea-review-agent`.
+- `spec_reviewer`: exact `{id, login}` from authoritative `/users/gitea-spec-agent`; never infer an account ID from historical comments.
+- `spec_review_id`: the single nonstale, nondismissed, canonical Native `APPROVED` spec review on this head and contract.
+- `spec_paths`: unique, safe repository-relative paths in the approved spec's evidence scope. Supply source-head-verified surrounding paths as well as changed paths; an approved spec may legitimately reference unchanged tests or callers. Do **not** restrict this set to the PR diff or read another role profile's artifacts to reconstruct it. The main review/admission must establish this scope.
+- `verdict`: exactly `approve` or `request_changes`. Do not normalize malformed identifiers, hashes, states or verdicts.
+- `body`: finalized UTF-8 review body, beginning with the exact code-review marker and containing exactly one `- Verdict: \`<verdict>\`` line. `marker(plan)` creates the binding line:
+  `<!-- gitea-agent:code-review/v1 owner=<owner> repository=<repository> issue=<issue> pr=<pull> head=<head> contract-sha256=<digest> -->`.
+- `provider_contract`: three explicit booleans `supersede_same_author`, `retouch_dismissed`, `unordered_commit_files`, and `evidence`, a list of redacted, verifiable provider-contract/observation references. Disable exceptions lacking evidence; never set every flag simply to make a failing check pass.
+
+Owner/repository currently support ASCII alphanumeric, underscore, dot and hyphen identifiers, with an alphanumeric/underscore first character. Unsupported names are rejected, not rewritten. Inline comments in a new review are not supported: the Native payload always explicitly sets `comments: []`.
+
+## Trusted transport and lease prerequisites
+
+Keep `gitea-access.md` and `repository-cache.md` authoritative. Before calling `publish`:
+
+1. Admit the canonical, existing role-owned repository cache and impl-created, registered detached/locked worktree. Complete the independent code review and collect evidence on the exact source head.
+2. Authenticate **one** transport from the current role's `$HERMES_HOME/.env` `GITEA_TOKEN`, or an explicitly authorized role-scoped external secret source. No ambient-token fallback, cross-profile credentials, credential-bearing Git URLs or anonymous downgrade is allowed.
+3. Validate the exact fixed origin `https://gitea.jsfamily.quest/api/v1` and `GET /user` account ID/login. The transport must verify TLS, reject redirects, bound network timeouts and **not retry mutation requests**. The bridge's provenance labels are assertions supplied by this already-validated caller, not a mechanism to load or certify an arbitrary client.
+4. Keep the Issue `fcntl.flock` exclusive descriptor owned by this process open throughout publication/readback. Pass a live monotonic deadline, not an old PID/`ready.json` flag. `HeldIssueLease` checks exact canonical Issue lock inode, owner/mode, worktree state/registration, clean HEAD and deadline. It never releases or closes the caller's descriptor. The caller must do so in its normal `finally` cleanup. A custom supervisor lease must implement the same invariants: `check(plan)` returns `None` only on success and otherwise raises; boolean/dict success flags are unsupported.
+5. Put the journal outside every source checkout in a trusted artifact parent. Use the **same journal for every retry/resume of this plan**, not a new directory after a timeout. The adapter creates mode 0700 directories and 0600 atomic/fsynced artifacts, rejects symlinks/hardlinks/foreign or unsafe modes, and holds a separate nonblocking journal mutex. Redact secrets before constructing plans/artifacts; do not log request headers, raw transport exceptions or token values. Journals contain complete source snapshots and must remain private.
+
+The current helper is strict about worktree child directories: group/world-writable checkout paths are rejected. Do not change a real worktree's permissions to force publication; stop and reconcile an admission/publisher compatibility conflict. The built-in lease bridge is Linux-only: it requires an existing whole-file exclusive `FLOCK` owned by this process in `/proc/self/fdinfo`, and never acquires or reacquires that lock. It uses a root-owned, non-group/world-writable `/usr/bin/git` with a fixed minimal environment, not the caller's `PATH`.
+
+These guards assume a cooperative, trusted role-local filesystem and transport. They do not isolate a malicious process with the same UID or pin every filesystem ancestor across subprocess path resolution. At-most-once creation is scoped to the retained journal and held canonical Issue lease, **not** arbitrary alternate journal paths, unrelated publishers, or multiple hosts. Do not share one lease across simultaneous publication callers or change the journal after an uncertain attempt.
+
+The request callback is `authenticated_request(relative_api_path, payload=None) -> (decoded_json, headers, integer_status)`. `None` means GET; a dict means exactly one POST. It always uses the fixed origin and original role credential, with no authorization fallback. It is provided by the existing role-local transport; do not regenerate the full gate/publisher per review run.
+
+`ApiAdapter.snapshot(target)` collects repository/Issue/PR metadata, identity/spec identity, all changed files, commits, reviews, historical review comments, Issue/PR comments and head statuses. Use this same collector for the baseline and subsequent snapshots. Every collection records contiguous 50-item pages, declared totals and Link headers; duplicates, inconsistent totals and truncated enumeration fail closed. A saved older collector snapshot may be replayed if it provides these complete facts, but differently shaped collectors must not be mixed for live source-drift comparison.
+
+## Live integration
+
+After absolute root discovery, role-authenticated transport capability validation, complete review, and Issue lease acquisition, load the module once using `importlib.util.spec_from_file_location` and retain it. These are existing workflow objects, **not** placeholders for an alternative credential source:
+
+```python
+api = native_review.ApiAdapter(
+    authenticated_request,
+    origin="https://gitea.jsfamily.quest/api/v1",
+    credential_source="$HERMES_HOME/.env:GITEA_TOKEN",
+)
+lease = native_review.HeldIssueLease(issue_fd, absolute_cache_policy, deadline_monotonic)
+baseline = api.snapshot(plan["target"])
+native_review.validate_snapshot(plan, baseline)
+result = native_review.publish(plan, baseline, api, lease, journal_directory)
+```
+
+Preserve the original baseline until `publish` returns verified or the uncertainty is manually reconciled. The helper rechecks authoritative role/head/base/contract/default branch/source Issue linkage/spec approval and complete source snapshots immediately before creating. It explicitly sends `commit_id`, `body`, `event`, `comments` and performs at most one `create_review` call per durable attempt.
+
+A matching canonical review is exact-read and verified as a write-free Native no-op; conflicting/multiple/current-head pending reviews stop rather than overwrite or duplicate. Reposting a different body/verdict on the same binding is **not** an automatic update path.
+
+A durable `intent.json` is fsynced **before** the POST. After timeout, partial success, readback error, process crash or lease expiry, resume the same journal and perform only authoritative reads. If there is no unique match, retain the attempt and report uncertainty; never delete the journal, switch directories or silently POST again. A definitive failure may require an explicitly authorized manual recovery decision, outside this helper.
+
+## Verification semantics
+
+- Confirm the exact new review ID, role ID/login, reviewed head, body bytes, state, target PR URL, empty inline comments, typed flags and timezone-bearing timestamps against the complete review collection.
+- Confirm precisely one new review and unchanged historical author/body/head/state/comments. No unrelated native review, comment, repository, head, branch, Issue contract or source change is silently accepted.
+- Validate a source Issue repository binding when the provider includes it. A PR head may legitimately belong to a fork: the fixed base repository, exact head SHA and unchanged complete PR metadata bind it, rather than incorrectly requiring every head repository to equal the base.
+- `official` is a provider merge-policy flag, not an additional verdict or role-identity requirement. Current accepted entry/publication rules require canonical role/head/contract/state and nondismissed/nonstale status; they do not require `official=True`. Changing that policy requires an explicit acceptance-criteria change.
+- When explicitly supported by provider evidence, same-author official-review supersession or retouching an already-dismissed same-author formal review may change only `official`, `dismissed`, `updated_at`. Require the specific expected false/true flag transitions and a timestamp equal to this exact new review's submitted/updated timestamp. This is not a global mutable-field mask. Timestamp/pattern agreement is provider-contract consistency evidence, not proof of a unique server-side causal transaction; a coincident manual mutation matching the full permitted pattern cannot be distinguished without additional authoritative event evidence.
+- Only commit `files` arrays are conditionally sorted by unique filename. File record contents, commit order, parent order and other arrays are still compared without blanket normalization. Changed-file records are compared by unique filename.
+- Source Issue/PR timestamp, PR review-count and requested-reviewer changes are narrowly tied to the exact added review; unrelated derived-state mutation fails.
+- The CLI is **saved-artifact replay**, not fresh network verification or approval authority. Its `read_only: true` output must not be presented as a newly published review. In live results `wrote` refers to a Native POST by that invocation; local journal writes still occur for no-ops.
+
+## Read-only commands and regression tests
+
+```sh
+python3 "$HARNESS_ROOT/harness/agents/review/skills/gitea-code-review/scripts/native_review.py" preflight \
+  --plan "$RUN/plan.json" --baseline "$RUN/before.json"
+python3 "$HARNESS_ROOT/harness/agents/review/skills/gitea-code-review/scripts/native_review.py" verify \
+  --plan "$RUN/plan.json" --before "$RUN/before.json" --after "$RUN/after.json" --exact "$RUN/exact.json"
+PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s harness/tests/review -p 'test_native_review.py' -v
+PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s harness/tests/spec -v
+PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s harness/tests/impl -v
+```
+
+Run the test commands from the canonical harness repository root. The existing spec suite requires `/opt/data/cache/scratch`: prepare that directory only if absent after confirming a nonsymlink, role-owned, non-group/world-writable parent, and require mode 0700 for the scratch child. Preserve existing shared parent permissions; do not chmod/chown them to satisfy a test. Use isolated/local runner variables and properly quoted command arguments rather than reusing mutable interpreter globals across review steps. Keep local stub/loopback transport checks, real temporary Git/worktree/lease checks, and saved production-wire replay clearly distinguished from a live Gitea mutation test. Introducing this helper does not authorize changing any existing product PR verdict, source, Issue state, merge or deployment.
diff --git a/harness/agents/review/skills/gitea-code-review/scripts/native_review.py b/harness/agents/review/skills/gitea-code-review/scripts/native_review.py
new file mode 100644
index 0000000..a190b07
--- /dev/null
+++ b/harness/agents/review/skills/gitea-code-review/scripts/native_review.py
@@ -0,0 +1,759 @@
+#!/usr/bin/env python3
+"""Bounded code-review publication using the caller's authenticated adapter."""
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import re
+from pathlib import Path
+
+# Reuse public JSON/page/cache and canonical spec codecs, not the spec-only publisher.
+_SPEC_PATH = (Path(__file__).resolve().parents[4] /
+              'spec/skills/gitea-implementation-spec-review/scripts/review_support.py')
+_spec = importlib.util.spec_from_file_location('gitea_native_spec_support', _SPEC_PATH)
+_support = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_support)
+ValidationError = _support.ValidationError
+loads_json = _support.loads_json
+collect_pages = _support.collect_pages
+ORIGIN = 'https://gitea.jsfamily.quest'
+SCHEMA = 'gitea-agent.code-review-publication/v1'
+
+
+def fail(message):
+    raise ValidationError(message)
+
+
+def positive_id(value):
+    if type(value) is not int or value < 1:
+        fail('invalid positive identity')
+    return value
+
+
+def digest(value, length):
+    if type(value) is not str or not re.fullmatch('[0-9a-f]{' + str(length) + '}', value):
+        fail('invalid digest')
+    return value
+
+
+def marker(plan):
+    t = plan['target']
+    return (f'<!-- gitea-agent:code-review/v1 owner={t["owner"]} repository={t["repository"]} '
+            f'issue={t["issue"]} pr={t["pull"]} head={t["head"]} '
+            f'contract-sha256={plan["contract_sha256"]} -->')
+
+
+def validate_plan(plan):
+    keys = {'schema', 'target', 'contract_sha256', 'reviewer', 'spec_reviewer',
+            'spec_review_id', 'spec_paths', 'verdict', 'body', 'provider_contract'}
+    if type(plan) is not dict or set(plan) != keys or plan['schema'] != SCHEMA:
+        fail('invalid publication plan')
+    t = plan['target']
+    if type(t) is not dict or set(t) != {'owner', 'repository', 'issue', 'pull', 'head', 'base'}:
+        fail('invalid target schema')
+    for k in ['owner', 'repository']:
+        value = t[k]
+        if type(value) is not str or not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,254}', value):
+            fail('invalid target identifier')
+    for k in ['issue', 'pull']:
+        positive_id(t[k])
+    for k in ['head', 'base']:
+        digest(t[k], 40)
+    digest(plan['contract_sha256'], 64)
+    for k, login in [('reviewer', 'gitea-review-agent'), ('spec_reviewer', 'gitea-spec-agent')]:
+        who = plan[k]
+        if type(who) is not dict or set(who) != {'id', 'login'} or who['login'] != login:
+            fail('invalid role identity')
+        positive_id(who['id'])
+    if plan['reviewer']['id'] == plan['spec_reviewer']['id']:
+        fail('reviewer independence violation')
+    positive_id(plan['spec_review_id'])
+    paths = plan['spec_paths']
+    if type(paths) is not list or any(type(x) is not str or not x or x.startswith('/')
+                                        or any(p in ['', '.', '..'] for p in x.split('/'))
+                                        or any(ord(c) < 32 for c in x) for x in paths) or len(set(paths)) != len(paths):
+        fail('invalid declared spec path scope')
+    if type(plan['verdict']) is not str or plan['verdict'] not in ['approve', 'request_changes']:
+        fail('unsupported publication verdict')
+    body = plan['body']
+    if type(body) is not str or not 1 <= len(body) <= 250000:
+        fail('invalid body')
+    try:
+        body.encode('utf-8')
+    except UnicodeError:
+        fail('invalid body encoding')
+    if (not body.startswith(marker(plan) + '\n') or body.count('<!-- gitea-agent:code-review/v1 ') != 1
+            or [x for x in body.splitlines() if x.startswith('- Verdict:')] != [f'- Verdict: `{plan["verdict"]}`']):
+        fail('body binding mismatch')
+    provider = plan['provider_contract']
+    fields = {'supersede_same_author', 'retouch_dismissed', 'unordered_commit_files', 'evidence'}
+    if type(provider) is not dict or set(provider) != fields:
+        fail('invalid provider contract')
+    flags = [provider[k] for k in fields - {'evidence'}]
+    if any(type(x) is not bool for x in flags) or type(provider['evidence']) is not list:
+        fail('invalid provider projection flags')
+    if any(type(x) is not str or not x.strip() for x in provider['evidence']):
+        fail('invalid provider evidence')
+    if any(flags) and not provider['evidence']:
+        fail('provider projection requires evidence')
+    return plan
+
+
+def build_payload(plan):
+    validate_plan(plan)
+    return {'commit_id': plan['target']['head'], 'body': plan['body'],
+            'event': {'approve': 'APPROVED', 'request_changes': 'REQUEST_CHANGES'}[plan['verdict']],
+            'comments': []}
+
+
+def canonical(value):
+    # Reuse duplicate/nonfinite validation before fingerprinting or comparison.
+    try:
+        text = json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False, separators=(',', ':'))
+        loads_json(text)
+        return text.encode('utf-8')
+    except (TypeError, ValueError, UnicodeError):
+        fail('invalid JSON artifact')
+
+
+def indexed(items, key='id'):
+    if type(items) is not list:
+        fail('invalid collection')
+    out = {}
+    for item in items:
+        if type(item) is not dict or key not in item:
+            fail('invalid collection item')
+        identity = item[key]
+        if key == 'id':
+            positive_id(identity)
+        elif key == 'sha':
+            digest(identity, 40)
+        elif (type(identity) is not str or not identity or identity.startswith('/')
+              or '\\' in identity or any(p in ['', '.', '..'] for p in identity.split('/'))):
+            fail('invalid filename')
+        if identity in out:
+            fail('duplicate collection identity')
+        out[identity] = item
+    return out
+
+
+def check_pages(items, pages):
+    if type(pages) is not list or not pages:
+        fail('missing complete pagination evidence')
+    total = None; counted = 0
+    for number, page in enumerate(pages, 1):
+        if type(page) is not dict or set(page) != {'page', 'count', 'declared_total', 'link'}:
+            fail('invalid pagination record')
+        if type(page['page']) is not int or page['page'] != number:
+            fail('invalid page sequence')
+        count = page['count']; declared = page['declared_total']; link = page['link']
+        if type(count) is not int or not 0 <= count <= 50:
+            fail('invalid page size')
+        if declared is not None:
+            if type(declared) is not int or declared < 0 or (total is not None and declared != total):
+                fail('conflicting declared total')
+            total = declared
+        if link is not None and type(link) is not str:
+            fail('invalid Link header')
+        if number < len(pages) and count != 50:
+            fail('invalid pagination continuation')
+        counted += count
+    last = pages[-1]
+    if counted != len(items) or (total is not None and total != counted):
+        fail('collection cardinality mismatch')
+    if (last['link'] and re.search(r'rel=["\']?next', last['link'])) or (total is None and last['count'] == 50):
+        fail('truncated pagination')
+
+
+def role_identity(value, expected):
+    if type(value) is not dict or type(value.get('id')) is not int or value.get('id') != expected['id'] or value.get('login') != expected['login']:
+        fail('role identity mismatch')
+    if value.get('origin', ORIGIN + '/api/v1') != ORIGIN + '/api/v1':
+        fail('origin mismatch')
+
+
+def validate_snapshot(plan, snapshot):
+    validate_plan(plan)
+    required = {'identity', 'spec_user', 'repository', 'issue', 'pr', 'files', 'commits',
+                'reviews', 'review_comments', 'issue_comments', 'pr_comments', 'statuses', 'pagination'}
+    if type(snapshot) is not dict or not required <= set(snapshot):
+        fail('incomplete snapshot')
+    canonical(snapshot)
+    t = plan['target']; full = t['owner'] + '/' + t['repository']
+    role_identity(snapshot['identity'], plan['reviewer'])
+    role_identity(snapshot['spec_user'], plan['spec_reviewer'])
+    repo = snapshot['repository']; issue = snapshot['issue']; pr = snapshot['pr']
+    if any(type(x) is not dict for x in [repo, issue, pr, snapshot['pagination']]):
+        fail('invalid target snapshot shape')
+    positive_id(issue.get('number')); positive_id(pr.get('number'))
+    if 'repository' in issue and (type(issue['repository']) is not dict or issue['repository'].get('full_name') != full):
+        fail('source Issue repository mismatch')
+    for side in ['base', 'head']:
+        if (type(pr.get(side)) is not dict or type(pr[side].get('repo')) is not dict
+                or type(pr[side]['repo'].get('full_name')) is not str):
+            fail('invalid PR repository binding')
+    if (repo.get('full_name') != full or issue.get('number') != t['issue']
+            or pr.get('number') != t['pull'] or pr.get('html_url') != f'{ORIGIN}/{full}/pulls/{t["pull"]}'
+            or issue.get('state') != 'open' or pr.get('state') != 'open' or pr.get('merged') is not False):
+        fail('target or lifecycle mismatch')
+    if type(issue.get('body')) is not str or hashlib.sha256(issue['body'].encode('utf-8')).hexdigest() != plan['contract_sha256']:
+        fail('contract drift')
+    if snapshot.get('contract_sha256', plan['contract_sha256']) != plan['contract_sha256']:
+        fail('contract digest mismatch')
+    if (pr['base']['sha'] != t['base'] or pr['head']['sha'] != t['head']
+            or pr['base']['repo']['full_name'] != full or pr['base']['ref'] != repo['default_branch']):
+        fail('head/base/default branch drift')
+    author = positive_id(pr['user']['id'])
+    if author in [plan['reviewer']['id'], plan['spec_reviewer']['id']]:
+        fail('reviewer independence violation')
+    links = re.findall(r'(?im)^(?:closes|fixes|resolves)[ \t]+[^\r\n]+$', pr.get('body', ''))
+    if links != ['Closes #' + str(t['issue'])]:
+        fail('source issue linkage mismatch')
+    for name, key in [('files', 'filename'), ('commits', 'sha'), ('reviews', 'id'),
+                      ('issue_comments', 'id'), ('pr_comments', 'id'), ('statuses', 'id')]:
+        indexed(snapshot[name], key)
+        check_pages(snapshot[name], snapshot['pagination'].get(name))
+    review_map = indexed(snapshot['reviews'])
+    comments = snapshot['review_comments']
+    if type(comments) is not dict or set(comments) != {str(rid) for rid in review_map}:
+        fail('incomplete review comment collection')
+    for rid, review in review_map.items():
+        if type(review.get('user')) is not dict:
+            fail('invalid historical reviewer identity')
+        positive_id(review['user'].get('id'))
+        digest(review.get('commit_id'), 40)
+        if (type(review.get('body')) is not str or review.get('state') not in ['COMMENT', 'PENDING', 'APPROVED', 'REQUEST_CHANGES']
+                or any(type(review.get(k)) is not bool for k in ['official', 'dismissed', 'stale'])):
+            fail('invalid native review shape')
+        indexed(comments[str(rid)])
+        check_pages(comments[str(rid)], snapshot['pagination'].get('review_comments_' + str(rid)))
+        if type(review.get('comments_count')) is not int or review['comments_count'] != len(comments[str(rid)]):
+            fail('review comment cardinality mismatch')
+    approved = [r for r in snapshot['reviews'] if r.get('user', {}).get('id') == plan['spec_reviewer']['id']
+                and r.get('commit_id') == t['head'] and r.get('state') == 'APPROVED'
+                and r.get('stale') is False and r.get('dismissed') is False]
+    if len(approved) != 1 or approved[0]['id'] != plan['spec_review_id']:
+        fail('missing or ambiguous current-head spec approval')
+    role_identity(approved[0]['user'], plan['spec_reviewer'])
+    parsed = _support.parse_review(approved[0]['body'], set(plan['spec_paths']))
+    if (parsed['verdict'] != 'approve' or parsed['contract_sha256'] != plan['contract_sha256']
+            or parsed['target'] != {k: t[k] for k in ['owner', 'repository', 'issue', 'pull', 'head']}):
+        fail('spec approval binding mismatch')
+    return snapshot
+
+
+def normalized_commits(commits, enabled):
+    import copy
+    out = copy.deepcopy(commits)
+    for row in out:
+        files = row.get('files', [])
+        indexed(files, 'filename')
+        if enabled and 'files' in row:
+            row['files'] = sorted(files, key=lambda f: f['filename'])
+    return out
+
+
+def same_source(plan, before, after, new_review=None):
+    import copy
+    validate_snapshot(plan, before); validate_snapshot(plan, after)
+    a = copy.deepcopy(before); b = copy.deepcopy(after)
+    mutable = {'reviews', 'review_comments', 'pagination'}
+    # Provider-derived changes are tied to this exact review, never globally ignored.
+    if new_review is not None:
+        times = {new_review['submitted_at'], new_review['updated_at']}
+        for section in ['issue', 'pr']:
+            if a[section].get('updated_at') != b[section].get('updated_at'):
+                if b[section].get('updated_at') not in times:
+                    fail('unrelated timestamp mutation')
+                b[section]['updated_at'] = a[section].get('updated_at')
+        if 'review_comments' in a['pr']:
+            old = a['pr']['review_comments']; new = b['pr'].get('review_comments')
+            if type(old) is not int or type(new) is not int or new != old + 1:
+                fail('PR review count projection mismatch')
+            b['pr']['review_comments'] = old
+        requested_before = indexed(a['pr'].get('requested_reviewers', []))
+        requested_after = indexed(b['pr'].get('requested_reviewers', []))
+        extra = set(requested_after) - set(requested_before)
+        if set(requested_before) - set(requested_after) or extra - {plan['reviewer']['id']}:
+            fail('unrelated requested reviewer mutation')
+        if any(canonical(row) != canonical(requested_after[rid]) for rid, row in requested_before.items()):
+            fail('requested reviewer content changed')
+        if extra:
+            role_identity(requested_after[plan['reviewer']['id']], plan['reviewer'])
+        b['pr']['requested_reviewers'] = a['pr'].get('requested_reviewers', [])
+    for data in [a, b]:
+        data['files'] = sorted(data['files'], key=lambda f: f['filename'])
+        data['commits'] = normalized_commits(data['commits'], plan['provider_contract']['unordered_commit_files'])
+    if canonical({k: v for k, v in a.items() if k not in mutable}) != canonical({k: v for k, v in b.items() if k not in mutable}):
+        fail('source snapshot drift')
+
+
+def exact_binding(plan, snapshot, exact):
+    payload = build_payload(plan)
+    if type(exact) is not dict:
+        fail('invalid exact review')
+    rid = positive_id(exact.get('id'))
+    role_identity(exact.get('user'), plan['reviewer'])
+    if (exact.get('body') != payload['body'] or exact.get('commit_id') != payload['commit_id']
+            or exact.get('state') != payload['event'] or exact.get('stale') is not False
+            or exact.get('dismissed') is not False or type(exact.get('official')) is not bool
+            or type(exact.get('comments_count')) is not int or exact['comments_count'] != 0
+            or exact.get('pull_request_url') != snapshot['pr']['html_url']):
+        fail('exact review binding mismatch')
+    if snapshot['review_comments'].get(str(rid)) != []:
+        fail('inline comments are unsupported')
+    if canonical(indexed(snapshot['reviews']).get(rid)) != canonical(exact):
+        fail('exact review differs from complete collection')
+    for key in ['submitted_at', 'updated_at']:
+        from datetime import datetime
+        try:
+            stamp = datetime.fromisoformat(exact[key].replace('Z', '+00:00'))
+            if stamp.utcoffset() is None:
+                fail('unbound review timestamp')
+        except (KeyError, ValueError, TypeError, AttributeError):
+            fail('invalid review timestamp')
+    return rid
+
+
+def verify_publication(plan, before, after, exact):
+    rid = exact_binding(plan, validate_snapshot(plan, after), exact)
+    same_source(plan, before, after, exact)
+    old = indexed(before['reviews']); new = indexed(after['reviews'])
+    if set(new) - set(old) != {rid} or set(old) - set(new):
+        fail('review cardinality mismatch')
+    current = [r for r in after['reviews'] if r['user']['id'] == plan['reviewer']['id']
+               and r['commit_id'] == plan['target']['head'] and r['state'] in ['PENDING', 'APPROVED', 'REQUEST_CHANGES']
+               and not r['dismissed']]
+    if len(current) != 1 or current[0]['id'] != rid:
+        fail('competing current-head review')
+    projected = []
+    for old_id, review in old.items():
+        fresh = new[old_id]
+        if before['review_comments'][str(old_id)] != after['review_comments'][str(old_id)]:
+            fail('historical review comments changed')
+        if canonical(review) == canonical(fresh):
+            continue
+        permitted = {'official', 'dismissed', 'updated_at'}
+        if (review['user']['id'] != plan['reviewer']['id'] or review['state'] not in ['APPROVED', 'REQUEST_CHANGES']
+                or set(review) != set(fresh)
+                or canonical({k: v for k, v in review.items() if k not in permitted}) != canonical({k: v for k, v in fresh.items() if k not in permitted})):
+            fail('historical review immutable content changed')
+        provider = plan['provider_contract']
+        superseded = provider['supersede_same_author'] and review.get('official') is True and review.get('dismissed') is False
+        retouched = provider['retouch_dismissed'] and review.get('official') is False and review.get('dismissed') is True
+        if (not (superseded or retouched) or fresh.get('official') is not False or fresh.get('dismissed') is not True
+                or fresh.get('updated_at') not in {exact['submitted_at'], exact['updated_at']}):
+            fail('unauthorized historical review projection')
+        projected.append(old_id)
+    return {'verified': True, 'review_id': rid, 'state': exact['state'], 'head': plan['target']['head'],
+            'projected_review_ids': projected, 'review_records_added': 1,
+            'commit_file_order_normalized': before['commits'] != after['commits'], 'read_only': True}
+
+
+class UncertainPublication(ValidationError):
+    """A durable attempt exists; a second create is prohibited."""
+
+
+class Journal:
+    def __init__(self, directory):
+        import os, stat
+        self.path = Path(directory)
+        if not self.path.is_absolute():
+            fail('journal must be an absolute trusted path')
+        for p in [self.path, *self.path.parents]:
+            if p.is_symlink():
+                fail('symlinked journal path')
+        self.path.mkdir(mode=0o700, exist_ok=True)
+        st = self.path.lstat()
+        if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) != 0o700:
+            fail('unsafe journal directory')
+        self.fd = None
+
+    def __enter__(self):
+        import fcntl, os, stat
+        self.fd = os.open(self.path / 'publication.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+        try:
+            st = os.fstat(self.fd)
+            if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) != 0o600 or st.st_nlink != 1:
+                fail('unsafe journal lock')
+            fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BaseException:
+            os.close(self.fd); self.fd = None
+            raise
+        return self
+
+    def __exit__(self, *args):
+        import os
+        os.close(self.fd)
+
+    def read(self, name):
+        import os, stat
+        p = self.path / name
+        try:
+            fd = os.open(p, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        except FileNotFoundError:
+            return None
+        except OSError:
+            raise ValidationError('journal artifact unavailable') from None
+        with os.fdopen(fd, 'r', encoding='utf-8') as artifact:
+            st = os.fstat(artifact.fileno())
+            if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) != 0o600 or st.st_nlink != 1:
+                fail('unsafe journal artifact')
+            return loads_json(artifact.read())
+
+    def write(self, name, value):
+        import os, tempfile
+        self.read(name)  # Reject existing symlink/hardlink/foreign artifacts.
+        fd, tmp = tempfile.mkstemp(dir=self.path, prefix='.native-review-')
+        try:
+            with os.fdopen(fd, 'wb') as out:
+                out.write(canonical(value)); out.flush(); os.fsync(out.fileno())
+            os.replace(tmp, self.path / name)
+            directory = os.open(self.path, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
+        finally:
+            if os.path.exists(tmp):
+                os.unlink(tmp)
+
+
+def candidates(plan, snapshot):
+    own = [r for r in snapshot['reviews'] if r['user']['id'] == plan['reviewer']['id']
+           and r['commit_id'] == plan['target']['head'] and r['state'] in ['PENDING', 'APPROVED', 'REQUEST_CHANGES']]
+    matching = [r for r in own if marker(plan) in r.get('body', '')]
+    if len(matching) > 1 or any(r not in matching for r in own):
+        fail('duplicate or conflicting current-head review')
+    if matching and (matching[0]['body'] != plan['body'] or matching[0]['state'] != build_payload(plan)['event']):
+        fail('conflicting canonical review')
+    return matching
+
+
+def read_api(api, method, *args):
+    try:
+        return getattr(api, method)(*args)
+    except Exception:
+        raise ValidationError('authoritative API read unavailable; do not retry POST') from None
+
+
+def check_lease(lease, plan):
+    try:
+        result = lease.check(plan)
+    except Exception:
+        raise ValidationError('held Issue lease invalid or unavailable') from None
+    if result is not None:
+        fail('unsupported lease check result')
+
+
+def publish(plan, baseline, api, lease, journal_directory):
+    """Use one trusted role transport and an already-held Issue lease; never retry POST."""
+    import copy
+    validate_snapshot(plan, baseline)
+    payload = build_payload(plan)
+    if any(not callable(getattr(api, k, None)) for k in ['identity', 'snapshot', 'create_review', 'read_review']):
+        fail('unsupported adapter capabilities')
+    if not callable(getattr(lease, 'check', None)):
+        fail('missing held Issue lease')
+    fingerprint = hashlib.sha256(canonical(plan)).hexdigest()
+    check_lease(lease, plan)
+    with Journal(journal_directory) as journal:
+        intent = journal.read('intent.json')
+        before = baseline
+        if intent is not None:
+            if type(intent) is not dict or intent.get('fingerprint') != fingerprint:
+                fail('journal belongs to a different publication')
+            before = journal.read('before.json')
+            if before is None or hashlib.sha256(canonical(before)).hexdigest() != intent.get('before_sha256'):
+                fail('journal baseline integrity mismatch')
+            validate_snapshot(plan, before)
+            expected_intent = {'schema': 'gitea-agent.native-review-intent/v1', 'fingerprint': fingerprint,
+                               'before_sha256': hashlib.sha256(canonical(before)).hexdigest(),
+                               'head': plan['target']['head'], 'event': payload['event'],
+                               'body_sha256': hashlib.sha256(payload['body'].encode()).hexdigest()}
+            if canonical(intent) != canonical(expected_intent):
+                fail('journal intent schema or binding mismatch')
+        check_lease(lease, plan)
+        role_identity(read_api(api, 'identity'), plan['reviewer'])
+        fresh = validate_snapshot(plan, read_api(api, 'snapshot', plan['target']))
+        matching = candidates(plan, fresh)
+        if matching:
+            exact = read_api(api, 'read_review', plan['target'], matching[0]['id'])
+            if exact['id'] in indexed(before['reviews']):
+                same_source(plan, before, fresh)
+                if canonical(before['reviews']) != canonical(fresh['reviews']) or before['review_comments'] != fresh['review_comments']:
+                    fail('no-op baseline drift')
+                rid = exact_binding(plan, fresh, exact)
+                result = {'verified': True, 'review_id': rid, 'state': exact['state'], 'head': plan['target']['head'], 'review_records_added': 0}
+            else:
+                result = verify_publication(plan, before, fresh, exact)
+            check_lease(lease, plan)
+            result.update(mode='no_op', read_only=False, wrote=False)
+            journal.write('verified.json', result)
+            return result
+        if intent is not None:
+            raise UncertainPublication('durable attempt has no authoritative match; do not retry POST')
+        same_source(plan, baseline, fresh)
+        if canonical(baseline['reviews']) != canonical(fresh['reviews']) or baseline['review_comments'] != fresh['review_comments']:
+            fail('review baseline drift')
+        check_lease(lease, plan)
+        # A second source read immediately precedes intent/create; do not mutate a stale baseline.
+        role_identity(read_api(api, 'identity'), plan['reviewer'])
+        latest = validate_snapshot(plan, read_api(api, 'snapshot', plan['target']))
+        same_source(plan, fresh, latest)
+        if canonical(fresh['reviews']) != canonical(latest['reviews']) or fresh['review_comments'] != latest['review_comments']:
+            fail('review baseline drift')
+        before = copy.deepcopy(latest)
+        journal.write('before.json', before)
+        journal.write('intent.json', {'schema': 'gitea-agent.native-review-intent/v1', 'fingerprint': fingerprint,
+                                     'before_sha256': hashlib.sha256(canonical(before)).hexdigest(),
+                                     'head': plan['target']['head'], 'event': payload['event'],
+                                     'body_sha256': hashlib.sha256(payload['body'].encode()).hexdigest()})
+        check_lease(lease, plan)
+        uncertain = False
+        response_id = None
+        try:
+            created = api.create_review(plan['target'], payload)
+            if type(created) is dict:
+                response_id = positive_id(created.get('id'))
+            else:
+                uncertain = True
+        except Exception:
+            # Never expose raw transport exception text (which may contain credentials).
+            uncertain = True
+        check_lease(lease, plan)
+        try:
+            role_identity(read_api(api, 'identity'), plan['reviewer'])
+            after = validate_snapshot(plan, read_api(api, 'snapshot', plan['target']))
+            matching = candidates(plan, after)
+            if len(matching) != 1:
+                raise UncertainPublication('create outcome not uniquely found; do not retry POST')
+            if response_id is not None and response_id != matching[0]['id']:
+                fail('create response and collection mismatch')
+            exact = read_api(api, 'read_review', plan['target'], matching[0]['id'])
+            result = verify_publication(plan, before, after, exact)
+        except ValidationError:
+            raise
+        except Exception:
+            raise UncertainPublication('authoritative readback unavailable; retained attempt; do not retry POST') from None
+        check_lease(lease, plan)
+        result.update(mode='reconciled' if uncertain else 'created', read_only=False, wrote=True)
+        journal.write('verified.json', result)
+        return result
+
+
+class ApiAdapter:
+    """REST bridge: request(path, payload=None) -> (JSON, headers, status).
+
+    The caller supplies one already-authenticated, no-redirect, TLS-verifying,
+    role-local transport. This class never loads/falls back to credentials.
+    """
+    def __init__(self, request, *, origin, credential_source):
+        if not callable(request) or origin != ORIGIN + '/api/v1':
+            fail('invalid authenticated transport origin')
+        if credential_source not in ['$HERMES_HOME/.env:GITEA_TOKEN', 'role-scoped-external-secret:GITEA_TOKEN']:
+            fail('unverified role credential source')
+        self.request = request
+        self.origin = origin
+        self.credential_source = credential_source
+
+    def call(self, path, payload=None):
+        try:
+            data, headers, status = self.request(path, payload)
+        except Exception:
+            raise ValidationError('authenticated API request failed') from None
+        expected = [200, 201] if payload is not None else [200]
+        if type(status) is not int or status not in expected or not hasattr(headers, 'items'):
+            fail('authenticated API status/shape mismatch')
+        return data, {str(k).lower(): v for k, v in headers.items()}
+
+    def identity(self):
+        data, _ = self.call('/user')
+        if type(data) is not dict or data.get('login') != 'gitea-review-agent':
+            fail('unexpected authenticated role')
+        positive_id(data.get('id'))
+        return {'id': data['id'], 'login': data['login'], 'origin': self.origin,
+                'credential_source': self.credential_source}
+
+    def prefix(self, target):
+        for key in ['owner', 'repository']:
+            if type(target.get(key)) is not str or not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,254}', target[key]):
+                fail('invalid request target')
+        positive_id(target['issue']); positive_id(target['pull'])
+        return '/repos/' + target['owner'] + '/' + target['repository']
+
+    def collection(self, endpoint, key):
+        records = []
+        def fetch(page):
+            data, headers = self.call(endpoint + '?limit=50&page=' + str(page))
+            raw_total = headers.get('x-total-count')
+            if raw_total is not None and (type(raw_total) is not str or not re.fullmatch(r'\d+', raw_total)):
+                fail('invalid declared collection total')
+            total = int(raw_total) if raw_total is not None else None
+            if data is None and total == 0:
+                data = []
+            if type(data) is not list:
+                fail('invalid API collection')
+            link = headers.get('link')
+            if link is not None and type(link) is not str:
+                fail('invalid collection Link header')
+            more = True if link and re.search(r'rel=["\']?next', link) else None
+            if more is None and total is not None:
+                more = page * 50 < total
+            records.append({'page': page, 'count': len(data), 'declared_total': total, 'link': link})
+            envelope = {'items': data, 'has_more': more}
+            if total is not None:
+                envelope['total_count'] = total
+            return envelope
+        items = collect_pages(fetch, key, page_size=50, max_pages=1000)
+        check_pages(items, records)
+        return items, records
+
+    def snapshot(self, target):
+        identity = self.identity()  # Always first authoritative read.
+        prefix = self.prefix(target)
+        repo, _ = self.call(prefix)
+        issue, _ = self.call(prefix + '/issues/' + str(target['issue']))
+        pr, _ = self.call(prefix + '/pulls/' + str(target['pull']))
+        spec_user, _ = self.call('/users/gitea-spec-agent')
+        if type(spec_user) is not dict or spec_user.get('login') != 'gitea-spec-agent':
+            fail('invalid spec identity')
+        out = {'identity': identity, 'spec_user': {'id': positive_id(spec_user.get('id')), 'login': spec_user['login']},
+               'repository': repo, 'issue': issue, 'pr': pr, 'pagination': {}, 'review_comments': {}}
+        pull = prefix + '/pulls/' + str(target['pull'])
+        endpoints = [('files', pull + '/files', 'filename'), ('commits', pull + '/commits', 'sha'),
+                     ('reviews', pull + '/reviews', 'id'),
+                     ('issue_comments', prefix + '/issues/' + str(target['issue']) + '/comments', 'id'),
+                     ('pr_comments', prefix + '/issues/' + str(target['pull']) + '/comments', 'id'),
+                     ('statuses', prefix + '/statuses/' + digest(target['head'], 40), 'id')]
+        for name, endpoint, key in endpoints:
+            out[name], out['pagination'][name] = self.collection(endpoint, key)
+        for review in out['reviews']:
+            rid = str(positive_id(review['id']))
+            out['review_comments'][rid], out['pagination']['review_comments_' + rid] = self.collection(pull + '/reviews/' + rid + '/comments', 'id')
+        return out
+
+    def create_review(self, target, payload):
+        if type(payload) is not dict or set(payload) != {'commit_id', 'body', 'event', 'comments'} or payload['comments'] != []:
+            fail('invalid create payload')
+        data, _ = self.call(self.prefix(target) + '/pulls/' + str(target['pull']) + '/reviews', payload)
+        return data
+
+    def read_review(self, target, review_id):
+        positive_id(review_id)
+        data, _ = self.call(self.prefix(target) + '/pulls/' + str(target['pull']) + '/reviews/' + str(review_id))
+        return data
+
+
+class HeldIssueLease:
+    """Validate a caller-owned Issue lock FD; never close/release the caller's FD."""
+    def __init__(self, issue_fd, policy_path, deadline_monotonic):
+        self.issue_fd = issue_fd
+        self.policy_path = policy_path
+        self.deadline_monotonic = deadline_monotonic
+
+    def check(self, plan):
+        import math, os, stat, subprocess, time
+        validate_plan(plan)
+        if (type(self.issue_fd) is not int or type(self.deadline_monotonic) not in [int, float]
+                or not math.isfinite(self.deadline_monotonic) or time.monotonic() >= self.deadline_monotonic):
+            fail('invalid or expired Issue lease')
+        t = plan['target']
+        info = _support.validate_cache_paths(self.policy_path, t['owner'], t['repository'], t['issue'])
+        lock = Path(info['locks']) / f'issue-{t["issue"]}.lock'
+        try:
+            current = os.fstat(self.issue_fd); disk = lock.lstat()
+            if (not stat.S_ISREG(current.st_mode) or not stat.S_ISREG(disk.st_mode)
+                    or current.st_uid != os.getuid() or stat.S_IMODE(current.st_mode) != 0o600
+                    or current.st_nlink != 1 or (current.st_dev, current.st_ino) != (disk.st_dev, disk.st_ino)):
+                fail('invalid Issue lock descriptor')
+            # Linux fdinfo reports locks on this open file description. Do not
+            # acquire/reacquire here: an opened but unheld FD is not a lease.
+            records = [line.split() for line in Path(f'/proc/self/fdinfo/{self.issue_fd}').read_text().splitlines()]
+            device_inode = f'{os.major(current.st_dev):02x}:{os.minor(current.st_dev):02x}:{current.st_ino}'
+            if not any(len(row) == 9 and row[0] == 'lock:'
+                       and row[2:6] == ['FLOCK', 'ADVISORY', 'WRITE', str(os.getpid())]
+                       and row[6:] == [device_inode, '0', 'EOF'] for row in records):
+                fail('Issue lock is not already held by this process')
+        except OSError:
+            raise ValidationError('Issue lock unavailable') from None
+        wt = Path(info['repository']) / 'worktrees' / f'issue-{t["issue"]}' / 'checkout'
+        for directory in [wt.parent.parent, wt.parent, wt]:
+            st = directory.lstat()
+            if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid() or st.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+                fail('unsafe existing worktree path')
+        state_file = wt.parent / 'state.json'; st = state_file.lstat()
+        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) != 0o600:
+            fail('unsafe worktree state')
+        state = loads_json(state_file.read_text(encoding='utf-8'))
+        if (type(state) is not dict or state.get('schema') != 'gitea-agent.worktree/v1' or state.get('owner') != t['owner']
+                or state.get('repository') != t['repository'] or type(state.get('issue')) is not int
+                or state['issue'] != t['issue'] or state.get('created_by_role') != 'gitea-impl-agent'
+                or state.get('mode') != 'detached'):
+            fail('worktree registration identity mismatch')
+        st = (wt / '.git').lstat()
+        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid():
+            fail('invalid worktree Git pointer')
+        git_binary = '/usr/bin/git'
+        binary_stat = Path(git_binary).lstat()
+        if not stat.S_ISREG(binary_stat.st_mode) or binary_stat.st_uid != 0 or binary_stat.st_mode & 0o022:
+            fail('untrusted system Git binary')
+        env = {'PATH': '/usr/bin:/bin', 'GIT_CONFIG_GLOBAL': '/dev/null',
+               'GIT_CONFIG_NOSYSTEM': '1', 'GIT_OPTIONAL_LOCKS': '0'}
+        def git(*args):
+            try:
+                process = subprocess.run([git_binary, '-c', 'core.fsmonitor=false', '-c', 'core.hooksPath=/dev/null',
+                                          '-c', 'safe.directory=' + str(wt), *args],
+                                         capture_output=True, text=True, env=env, timeout=15)
+            except Exception:
+                raise ValidationError('read-only Git check unavailable') from None
+            if process.returncode or len(process.stdout) > 1000000:
+                fail('read-only Git check failed')
+            return process.stdout
+        if git('-C', str(wt), 'rev-parse', 'HEAD').strip() != t['head']:
+            fail('worktree HEAD mismatch')
+        if git('-C', str(wt), 'status', '--porcelain=v1', '--untracked-files=all').strip():
+            fail('existing worktree is not clean')
+        if git('-C', str(wt), 'rev-parse', '--path-format=absolute', '--git-common-dir').strip() != info['repository_git']:
+            fail('worktree common Git directory mismatch')
+        records = git('--git-dir=' + info['repository_git'], 'worktree', 'list', '--porcelain', '-z').split('\0\0')
+        matches = [r.split('\0') for r in records if r.startswith('worktree ' + str(wt) + '\0')]
+        if len(matches) != 1 or 'detached' not in matches[0] or not any(x == 'locked' or x.startswith('locked ') for x in matches[0]):
+            fail('worktree detached/locked registration mismatch')
+        if time.monotonic() >= self.deadline_monotonic:
+            fail('Issue lease expired during source checks')
+
+
+def main(argv=None):
+    import argparse, sys
+    parser = argparse.ArgumentParser(description='Read-only native review artifact preflight/replay. Live publication is the Python publish API.')
+    commands = parser.add_subparsers(dest='command', required=True)
+    pre = commands.add_parser('preflight')
+    pre.add_argument('--plan', required=True); pre.add_argument('--baseline', required=True)
+    verify = commands.add_parser('verify')
+    for name in ['plan', 'before', 'after', 'exact']:
+        verify.add_argument('--' + name, required=True)
+    args = parser.parse_args(argv)
+    def read(path):
+        return loads_json(Path(path).read_text(encoding='utf-8'))
+    try:
+        plan = read(args.plan)
+        if args.command == 'preflight':
+            snapshot = validate_snapshot(plan, read(args.baseline))
+            out = {'ready': True, 'read_only': True, 'artifact_validation_only': True,
+                   'head': plan['target']['head'], 'event': build_payload(plan)['event'],
+                   'reviewer_id': plan['reviewer']['id'], 'spec_review_id': plan['spec_review_id'],
+                   'snapshot_sha256': hashlib.sha256(canonical(snapshot)).hexdigest()}
+        else:
+            out = verify_publication(plan, read(args.before), read(args.after), read(args.exact))
+        print(json.dumps(out, ensure_ascii=False))
+        return 0
+    except Exception:
+        # No arbitrary artifact or transport exception is echoed to the terminal.
+        print(json.dumps({'error': 'invalid or unavailable review artifacts', 'read_only': True}), file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/harness/tests/review/test_native_review.py b/harness/tests/review/test_native_review.py
new file mode 100644
index 0000000..501a590
--- /dev/null
+++ b/harness/tests/review/test_native_review.py
@@ -0,0 +1,569 @@
+import copy
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / 'harness/agents/review/skills/gitea-code-review/scripts/native_review.py'
+SPEC_SCRIPT = ROOT / 'harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py'
+
+
+def load(path, name):
+    spec = importlib.util.spec_from_file_location(name, path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+support = load(SPEC_SCRIPT, 'spec_support_fixture')
+
+
+def plan_fixture():
+    target = {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pull': 7,
+              'head': 'a' * 40, 'base': 'c' * 40}
+    digest = __import__('hashlib').sha256(b'approved contract').hexdigest()
+    marker = ('<!-- gitea-agent:code-review/v1 owner=acme repository=repo issue=3 pr=7 '
+              f'head={target["head"]} contract-sha256={digest} -->')
+    return {'schema': 'gitea-agent.code-review-publication/v1', 'target': target,
+            'contract_sha256': digest, 'reviewer': {'id': 42, 'login': 'gitea-review-agent'},
+            'spec_reviewer': {'id': 5, 'login': 'gitea-spec-agent'}, 'spec_review_id': 9, 'spec_paths': ['src/a.py'],
+            'verdict': 'approve', 'body': marker + '\n- Verdict: `approve`\n검토 완료.\n',
+            'provider_contract': {'supersede_same_author': True, 'retouch_dismissed': True,
+                                  'unordered_commit_files': True,
+                                  'evidence': ['saved observed Gitea projection and ordering']}}
+
+
+def publication_fixture():
+    p = plan_fixture(); t = p['target']
+    spec_result = {'schema': support.SCHEMA,
+                   'target': {k: t[k] for k in ['owner', 'repository', 'issue', 'pull', 'head']},
+                   'contract_sha256': p['contract_sha256'], 'verdict': 'approve',
+                   'criteria': [{'criterion': 'REQ-1', 'contract': '계약',
+                                 'implementation_paths': ['src/a.py'], 'test_paths': [],
+                                 'evidence': '직접 검증', 'result': 'satisfied', 'reason': '충족'}],
+                   'scope': {'missing': [], 'excess': [], 'spec_conflicts': []}, 'conclusion': '충족'}
+    url = 'https://gitea.jsfamily.quest/acme/repo/pulls/7'
+    old_time = '2026-10-04T00:00:00Z'; new_time = '2026-10-04T00:01:00Z'
+    def review(rid, user, body, head, state):
+        return {'id': rid, 'user': copy.deepcopy(user), 'body': body, 'commit_id': head, 'state': state,
+                'official': True, 'dismissed': False, 'stale': head != t['head'],
+                'submitted_at': old_time, 'updated_at': old_time, 'comments_count': 0,
+                'pull_request_url': url}
+    sr = review(9, p['spec_reviewer'], support.render_review(spec_result, {'src/a.py'}),
+                t['head'], 'APPROVED')
+    previous = review(8, p['reviewer'], 'old review body', 'd' * 40, 'REQUEST_CHANGES')
+    s = {'identity': copy.deepcopy(p['reviewer']), 'spec_user': copy.deepcopy(p['spec_reviewer']),
+         'repository': {'full_name': 'acme/repo', 'default_branch': 'main'},
+         'issue': {'number': 3, 'body': 'approved contract', 'state': 'open',
+                   'updated_at': old_time, 'repository': {'full_name': 'acme/repo'}},
+         'pr': {'number': 7, 'html_url': url, 'body': 'Closes #3', 'user': {'id': 99},
+                'state': 'open', 'merged': False, 'updated_at': old_time,
+                'review_comments': 2, 'requested_reviewers': [],
+                'base': {'sha': t['base'], 'ref': 'main', 'repo': {'full_name': 'acme/repo'}},
+                'head': {'sha': t['head'], 'ref': 'feature', 'repo': {'full_name': 'acme/repo'}}},
+         'files': [{'filename': 'src/a.py', 'status': 'modified'}],
+         'commits': [{'sha': t['head'], 'files': [{'filename': 'src/a.py', 'status': 'modified'},
+                                              {'filename': 'docs/x.md', 'status': 'added'}]}],
+         'reviews': [previous, sr], 'review_comments': {'8': [], '9': []},
+         'issue_comments': [], 'pr_comments': [], 'statuses': [], 'pagination': {}}
+    exact = review(10, p['reviewer'], p['body'], t['head'], 'APPROVED')
+    exact['submitted_at'] = exact['updated_at'] = new_time
+    after = copy.deepcopy(s); after['reviews'].append(exact); after['review_comments']['10'] = []
+    after['pr']['updated_at'] = new_time; after['pr']['review_comments'] += 1
+    for snap in [s, after]:
+        for name in ['files', 'commits', 'reviews', 'issue_comments', 'pr_comments', 'statuses']:
+            snap['pagination'][name] = [{'page': 1, 'count': len(snap[name]),
+                                        'declared_total': len(snap[name]), 'link': None}]
+        for rid, comments in snap['review_comments'].items():
+            snap['pagination']['review_comments_' + rid] = [{'page': 1, 'count': len(comments),
+                                                           'declared_total': len(comments), 'link': None}]
+    return p, s, after, copy.deepcopy(exact)
+
+
+class FakeAPI:
+    def __init__(self, plan, before, after, exact):
+        self.plan = plan; self.before = before; self.after = after; self.exact = exact
+        self.created = False; self.posts = 0; self.timeout = False; self.apply = True
+        self.events = []
+
+    def identity(self):
+        self.events.append('identity')
+        return copy.deepcopy(self.plan['reviewer'])
+
+    def snapshot(self, target):
+        self.events.append('snapshot')
+        return copy.deepcopy(self.after if self.created else self.before)
+
+    def create_review(self, target, payload):
+        self.events.append('create'); self.posts += 1
+        if self.apply: self.created = True
+        if self.timeout: raise TimeoutError('secret transport detail must not escape')
+        return copy.deepcopy(self.exact)
+
+    def read_review(self, target, rid):
+        self.events.append('exact')
+        return copy.deepcopy(self.exact)
+
+
+class FakeLease:
+    def __init__(self): self.valid = True; self.checks = 0
+    def check(self, plan):
+        self.checks += 1
+        if not self.valid: raise support.ValidationError('expired lease')
+
+
+class NativeReviewTests(unittest.TestCase):
+    def module(self):
+        self.assertTrue(SCRIPT.is_file(), 'shared native review adapter is missing')
+        return load(SCRIPT, 'native_review_test')
+
+    def test_payload_is_exact_bound_native_create(self):
+        m = self.module()
+        p = plan_fixture()
+        self.assertEqual(m.build_payload(p), {'commit_id': p['target']['head'],
+                         'body': p['body'], 'event': 'APPROVED', 'comments': []})
+        bad = copy.deepcopy(p); bad['target']['pull'] = True
+        with self.assertRaises(m.ValidationError):
+            m.build_payload(bad)
+        bad = copy.deepcopy(p); bad['body'] = bad['body'].replace('a' * 40, 'd' * 40)
+        with self.assertRaises(m.ValidationError):
+            m.build_payload(bad)
+    def test_verify_exact_target_and_bound_provider_changes(self):
+        m = self.module()
+        self.assertTrue(callable(getattr(m, 'verify_publication', None)), 'readback verifier is missing')
+        p, before, after, exact = publication_fixture()
+        previous = after['reviews'][0]
+        previous.update(official=False, dismissed=True, updated_at=exact['submitted_at'])
+        after['commits'][0]['files'].reverse()
+        result = m.verify_publication(p, before, after, exact)
+        self.assertEqual(result['review_id'], 10)
+        self.assertEqual(result['projected_review_ids'], [8])
+        self.assertTrue(result['commit_file_order_normalized'])
+        for mutation in ['head', 'body', 'unrelated_review', 'commit_file']:
+            bad = copy.deepcopy(after)
+            if mutation == 'head': bad['pr']['head']['sha'] = 'e' * 40
+            elif mutation == 'body': bad['reviews'][-1]['body'] += ' altered'
+            elif mutation == 'unrelated_review': bad['reviews'][1]['body'] += ' altered'
+            else: bad['commits'][0]['files'][0]['status'] = 'removed'
+            with self.subTest(mutation=mutation), self.assertRaises(m.ValidationError):
+                m.verify_publication(p, before, bad, exact)
+    def test_publish_once_and_reconcile_durable_attempt_without_reposting(self):
+        m = self.module()
+        self.assertTrue(callable(getattr(m, 'publish', None)), 'shared publisher is missing')
+        p, before, after, exact = publication_fixture()
+        api = FakeAPI(p, before, after, exact); lease = FakeLease()
+        with tempfile.TemporaryDirectory() as td:
+            result = m.publish(p, before, api, lease, td)
+            self.assertEqual(result['mode'], 'created')
+            self.assertEqual(api.posts, 1)
+            self.assertEqual(m.publish(p, before, api, lease, td)['mode'], 'no_op')
+            self.assertEqual(api.posts, 1)
+            self.assertNotIn('body', json.loads((Path(td) / 'intent.json').read_text()))
+        for applies in [True, False]:
+            api = FakeAPI(p, before, after, exact); api.timeout = True; api.apply = applies
+            with self.subTest(applies=applies), tempfile.TemporaryDirectory() as td:
+                if applies:
+                    self.assertEqual(m.publish(p, before, api, lease, td)['mode'], 'reconciled')
+                    self.assertEqual(m.publish(p, before, api, lease, td)['mode'], 'no_op')
+                else:
+                    for _ in range(2):
+                        with self.assertRaises(m.UncertainPublication) as error:
+                            m.publish(p, before, api, lease, td)
+                        self.assertNotIn('secret transport detail', str(error.exception))
+                self.assertEqual(api.posts, 1)
+    def test_api_adapter_checks_identity_first_and_collects_all_pages(self):
+        m = self.module()
+        self.assertTrue(callable(getattr(m, 'ApiAdapter', None)), 'REST bridge is missing')
+        p, before, after, exact = publication_fixture()
+        calls = []
+        def request(path, payload=None):
+            calls.append((path, payload))
+            if path == '/user': return p['reviewer'], {}, 200
+            if path == '/users/gitea-spec-agent': return p['spec_reviewer'], {}, 200
+            if path == '/repos/acme/repo': return before['repository'], {}, 200
+            if path == '/repos/acme/repo/issues/3': return before['issue'], {}, 200
+            if path == '/repos/acme/repo/pulls/7': return before['pr'], {}, 200
+            from urllib.parse import parse_qs, urlsplit
+            parsed = urlsplit(path); page = int(parse_qs(parsed.query).get('page', ['1'])[0])
+            for name, suffix in [('files', '/pulls/7/files'), ('commits', '/pulls/7/commits'),
+                                 ('reviews', '/pulls/7/reviews'), ('issue_comments', '/issues/3/comments'),
+                                 ('pr_comments', '/issues/7/comments'), ('statuses', '/statuses/' + p['target']['head'])]:
+                if parsed.path == '/repos/acme/repo' + suffix:
+                    if name == 'reviews' and payload is not None: return exact, {}, 201
+                    return before[name], {'x-total-count': str(len(before[name]))}, 200
+            if parsed.path.endswith('/comments'): return [], {'X-Total-Count': '0'}, 200
+            raise AssertionError('unexpected request')
+        adapter = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1',
+                               credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
+        snapshot = adapter.snapshot(p['target'])
+        m.validate_snapshot(p, snapshot)
+        self.assertEqual(calls[0][0], '/user')
+        self.assertEqual(adapter.create_review(p['target'], m.build_payload(p))['id'], 10)
+        self.assertEqual(calls[-1][1]['comments'], [])
+        with self.assertRaises(m.ValidationError):
+            m.ApiAdapter(request, origin='https://other.test/api/v1', credential_source='unknown')
+    def test_held_lease_checks_real_registered_worktree_and_expiry(self):
+        import fcntl
+        m = self.module()
+        self.assertTrue(callable(getattr(m, 'HeldIssueLease', None)), 'real Issue lease guard is missing')
+        with tempfile.TemporaryDirectory() as td:
+            root = Path(td); policy = root / 'harness/operations/repository-cache.md'
+            policy.parent.mkdir(parents=True); policy.write_text('canonical cache fixture')
+            rs = root / 'repositories'; owner = rs / 'acme'; cache = owner / 'repo'; locks = cache / 'locks'
+            for directory in [rs, owner, cache, locks]: directory.mkdir(mode=0o700)
+            bare = cache / 'repository.git'; seed = root / 'seed'
+            def git(*args):
+                out = subprocess.run(['git', '-c', 'core.hooksPath=/dev/null', *map(str, args)], capture_output=True, text=True,
+                                     env={'PATH': os.environ['PATH'], 'GIT_CONFIG_GLOBAL': '/dev/null', 'GIT_CONFIG_NOSYSTEM': '1'})
+                self.assertEqual(out.returncode, 0, out.stderr)
+                return out.stdout.strip()
+            git('init', '--bare', bare); git('init', seed)
+            git('-C', seed, 'config', 'user.name', 'Synthetic'); git('-C', seed, 'config', 'user.email', 'fixture@example.test')
+            (seed / 'a.txt').write_text('synthetic')
+            git('-C', seed, 'add', 'a.txt'); git('-C', seed, 'commit', '-m', 'fixture')
+            git('--git-dir=' + str(bare), 'fetch', seed, 'HEAD:refs/heads/main')
+            wt = cache / 'worktrees/issue-3/checkout'; wt.parent.mkdir(parents=True, mode=0o700)
+            git('--git-dir=' + str(bare), 'worktree', 'add', '--detach', '--lock', wt, 'main')
+            head = git('-C', wt, 'rev-parse', 'HEAD')
+            p = plan_fixture(); p['target']['head'] = head; p['body'] = p['body'].replace('a' * 40, head)
+            state = wt.parent / 'state.json'
+            state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
+                                         'issue': 3, 'created_by_role': 'gitea-impl-agent', 'mode': 'detached'}))
+            state.chmod(0o600)
+            path = locks / 'issue-3.lock'; fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600)
+            try:
+                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                lease = m.HeldIssueLease(fd, policy, time.monotonic() + 30)
+                lease.check(p)
+                fcntl.flock(fd, fcntl.LOCK_UN)
+                with self.assertRaises(m.ValidationError): lease.check(p)
+                self.assertNotIn('lock:', Path(f'/proc/self/fdinfo/{fd}').read_text())
+                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                from unittest.mock import patch
+                fakebin = root / 'fakebin'; fakebin.mkdir()
+                marker = root / 'path-git-executed'
+                fakegit = fakebin / 'git'
+                fakegit.write_text(f'#!/bin/sh\n/usr/bin/touch "{marker}"\nexit 1\n'); fakegit.chmod(0o700)
+                with patch.dict(os.environ, {'PATH': str(fakebin)}): lease.check(p)
+                self.assertFalse(marker.exists())
+                lease.deadline_monotonic = time.monotonic() - 1
+                with self.assertRaises(m.ValidationError): lease.check(p)
+                lease.deadline_monotonic = time.monotonic() + 30
+                (wt / 'a.txt').write_text('dirty')
+                with self.assertRaises(m.ValidationError): lease.check(p)
+            finally:
+                os.close(fd)
+    def test_cli_preflight_and_saved_wire_verification_are_read_only(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        with tempfile.TemporaryDirectory() as td:
+            root = Path(td)
+            for name, value in [('plan', p), ('before', before), ('after', after), ('exact', exact)]:
+                (root / (name + '.json')).write_text(json.dumps(value, ensure_ascii=False))
+            base = [sys.executable, str(SCRIPT)]
+            env = {'PATH': os.environ['PATH'], 'PYTHONDONTWRITEBYTECODE': '1'}
+            pre = subprocess.run(base + ['preflight', '--plan', str(root / 'plan.json'),
+                                        '--baseline', str(root / 'before.json')], capture_output=True, text=True, env=env)
+            self.assertEqual(pre.returncode, 0, pre.stderr)
+            self.assertIn('"ready": true', pre.stdout, 'read-only preflight CLI is missing')
+            verified = subprocess.run(base + ['verify', '--plan', str(root / 'plan.json'),
+                                               '--before', str(root / 'before.json'), '--after', str(root / 'after.json'),
+                                               '--exact', str(root / 'exact.json')], capture_output=True, text=True, env=env)
+            self.assertEqual(verified.returncode, 0, verified.stderr)
+            self.assertTrue(json.loads(verified.stdout)['read_only'])
+            self.assertEqual(json.loads(verified.stdout)['review_id'], 10)
+            unsupported = subprocess.run(base + ['publish'], capture_output=True, text=True, env=env)
+            self.assertNotEqual(unsupported.returncode, 0)
+    def test_spec_approval_may_reference_unchanged_surrounding_test_paths(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        p['spec_paths'] = ['src/a.py', 'tests/existing.py']
+        spec_result = support.parse_review(before['reviews'][1]['body'], {'src/a.py'})
+        spec_result['criteria'][0]['test_paths'] = ['tests/existing.py']
+        before['reviews'][1]['body'] = support.render_review(spec_result, set(p['spec_paths']))
+        try:
+            m.validate_snapshot(p, before)
+        except m.ValidationError:
+            self.fail('spec approval must use its declared scope, not only changed files')
+    def test_malformed_historical_actor_and_missing_exact_review_flags_fail_closed(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        bad = copy.deepcopy(before); bad['reviews'][0]['user'] = {'id': True, 'login': 'x'}
+        with self.assertRaises(m.ValidationError):
+            m.validate_snapshot(p, bad)
+        p, before, after, exact = publication_fixture()
+        del exact['official']; del after['reviews'][-1]['official']
+        with self.assertRaises(m.ValidationError):
+            m.verify_publication(p, before, after, exact)
+    def test_invalid_plan_bindings_are_not_normalized(self):
+        m = self.module()
+        mutations = [lambda p: p['target'].__setitem__('issue', True),
+                     lambda p: p['target'].__setitem__('head', 'A' * 40),
+                     lambda p: p['target'].__setitem__('owner', 'acme/other'),
+                     lambda p: p.__setitem__('verdict', 'blocked'),
+                     lambda p: p.__setitem__('body', p['body'] + '- Verdict: `approve`\n'),
+                     lambda p: p['provider_contract'].__setitem__('evidence', []),
+                     lambda p: p.__setitem__('spec_paths', ['../secret']),
+                     lambda p: p.__setitem__('spec_paths', ['src/a.py', 'src/a.py']),
+                     lambda p: p['reviewer'].__setitem__('id', p['spec_reviewer']['id'])]
+        for i, mutate in enumerate(mutations):
+            with self.subTest(case=i):
+                p = plan_fixture(); mutate(p)
+                with self.assertRaises(m.ValidationError): m.build_payload(p)
+
+    def test_wrong_role_and_expired_lease_prevent_any_post(self):
+        m = self.module()
+        for kind in ['wrong_role', 'expired']:
+            with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
+                p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact); lease = FakeLease()
+                if kind == 'expired': lease.valid = False
+                else: api.identity = lambda: p['spec_reviewer']
+                with self.assertRaises(m.ValidationError): m.publish(p, before, api, lease, td)
+                self.assertEqual(api.posts, 0)
+                self.assertFalse((Path(td) / 'intent.json').exists())
+
+    def test_fresh_head_contract_and_source_linkage_drift_prevent_post(self):
+        m = self.module()
+        changes = [lambda s: s['pr']['head'].__setitem__('sha', 'b' * 40),
+                   lambda s: s['issue'].__setitem__('body', 'changed contract'),
+                   lambda s: s['pr'].__setitem__('body', 'Closes #4'),
+                   lambda s: s['pr']['base'].__setitem__('ref', 'other'),
+                   lambda s: s['pr'].__setitem__('merged', True)]
+        for i, change in enumerate(changes):
+            with self.subTest(case=i), tempfile.TemporaryDirectory() as td:
+                p, before, after, exact = publication_fixture(); api = FakeAPI(p, copy.deepcopy(before), after, exact)
+                change(api.before)
+                with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), td)
+                self.assertEqual(api.posts, 0)
+
+    def test_pagination_total_duplicate_and_truncation_are_hard_errors(self):
+        m = self.module()
+        changes = [lambda s: s['pagination']['files'][0].__setitem__('declared_total', 2),
+                   lambda s: s['pagination']['reviews'][0].__setitem__('link', '<x>; rel="next"'),
+                   lambda s: s['pagination']['commits'][0].__setitem__('count', True),
+                   lambda s: s['files'].append(copy.deepcopy(s['files'][0])),
+                   lambda s: s['review_comments'].pop('8')]
+        for i, change in enumerate(changes):
+            with self.subTest(case=i):
+                p, before, after, exact = publication_fixture(); change(before)
+                with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
+
+    def test_spec_approval_body_head_role_and_state_are_bound(self):
+        m = self.module()
+        changes = [lambda s: s['reviews'][1].__setitem__('state', 'REQUEST_CHANGES'),
+                   lambda s: s['reviews'][1].__setitem__('commit_id', 'b' * 40),
+                   lambda s: s['reviews'][1].__setitem__('dismissed', True),
+                   lambda s: s['reviews'][1].__setitem__('body', s['reviews'][1]['body'].replace('충족', '임의 변경', 1)),
+                   lambda s: s['reviews'][1]['user'].__setitem__('id', 23)]
+        for i, change in enumerate(changes):
+            with self.subTest(case=i):
+                p, before, after, exact = publication_fixture(); change(before)
+                with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
+
+    def test_provider_projection_requires_explicit_bound_evidence(self):
+        m = self.module()
+        for kind in ['disabled', 'wrong_timestamp', 'wrong_author']:
+            with self.subTest(kind=kind):
+                p, before, after, exact = publication_fixture()
+                after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
+                if kind == 'disabled': p['provider_contract']['supersede_same_author'] = False
+                elif kind == 'wrong_timestamp': after['reviews'][0]['updated_at'] = '2026-10-04T00:02:00Z'
+                else:
+                    before['reviews'][0]['user'] = {'id': 72, 'login': 'other'}
+                    after['reviews'][0]['user'] = {'id': 72, 'login': 'other'}
+                with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+
+    def test_order_normalization_does_not_mask_file_contents_or_parent_order(self):
+        m = self.module()
+        p, before, after, exact = publication_fixture()
+        after['commits'][0]['files'].reverse()
+        self.assertTrue(m.verify_publication(p, before, after, exact)['commit_file_order_normalized'])
+        p['provider_contract']['unordered_commit_files'] = False
+        with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+        p, before, after, exact = publication_fixture()
+        after['commits'][0]['files'][0]['status'] = 'removed'
+        with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+        p, before, after, exact = publication_fixture()
+        before['commits'][0]['parents'] = [{'sha': 'e' * 40}, {'sha': 'f' * 40}]
+        after['commits'][0]['parents'] = list(reversed(before['commits'][0]['parents']))
+        with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+
+    def test_historical_head_state_body_and_comments_remain_immutable(self):
+        m = self.module()
+        changes = [lambda s: s['reviews'][0].__setitem__('body', 'altered'),
+                   lambda s: s['reviews'][0].__setitem__('commit_id', 'e' * 40),
+                   lambda s: s['reviews'][0].__setitem__('state', 'APPROVED')]
+        for i, change in enumerate(changes):
+            with self.subTest(case=i):
+                p, before, after, exact = publication_fixture(); change(after)
+                with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+        p, before, after, exact = publication_fixture()
+        for snap in [before, after]:
+            snap['review_comments']['8'] = [{'id': 71, 'body': 'original'}]
+            snap['reviews'][0]['comments_count'] = 1
+            snap['pagination']['review_comments_8'][0].update(count=1, declared_total=1)
+        after['review_comments']['8'][0]['body'] = 'altered'
+        with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+
+    def test_existing_canonical_review_is_write_free_no_op(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        api = FakeAPI(p, before, after, exact); api.created = True
+        with tempfile.TemporaryDirectory() as td:
+            result = m.publish(p, after, api, FakeLease(), td)
+            self.assertEqual(result['mode'], 'no_op'); self.assertFalse(result['wrote'])
+            self.assertEqual(api.posts, 0)
+            self.assertFalse((Path(td) / 'intent.json').exists())
+
+    def test_duplicate_or_conflicting_canonical_review_is_not_reposted(self):
+        m = self.module()
+        for kind in ['body', 'duplicate']:
+            with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
+                p, before, after, exact = publication_fixture()
+                if kind == 'body': after['reviews'][-1]['body'] += 'other'
+                else:
+                    extra = copy.deepcopy(exact); extra['id'] = 11; after['reviews'].append(extra)
+                    after['review_comments']['11'] = []
+                    after['pagination']['reviews'][0].update(count=4, declared_total=4)
+                    after['pagination']['review_comments_11'] = [{'page': 1, 'count': 0, 'declared_total': 0, 'link': None}]
+                api = FakeAPI(p, before, after, exact); api.created = True
+                with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), td)
+                self.assertEqual(api.posts, 0)
+
+    def test_readback_failure_and_changed_journal_plan_never_retry_post(self):
+        m = self.module(); p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
+        api.read_review = lambda target, rid: (_ for _ in ()).throw(TimeoutError('DO_NOT_ECHO_TRANSPORT_SECRET'))
+        with tempfile.TemporaryDirectory() as td:
+            for _ in range(2):
+                with self.assertRaises(Exception) as raised: m.publish(p, before, api, FakeLease(), td)
+                self.assertNotIn('DO_NOT_ECHO_TRANSPORT_SECRET', str(raised.exception))
+            self.assertEqual(api.posts, 1)
+            different = copy.deepcopy(p); different['body'] += 'different'
+            with self.assertRaises(m.ValidationError): m.publish(different, before, api, FakeLease(), td)
+            self.assertEqual(api.posts, 1)
+
+    def test_journal_symlinks_hardlinks_and_changed_baseline_are_rejected(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        with tempfile.TemporaryDirectory() as td:
+            root = Path(td); link = root / 'link'; link.symlink_to(root, target_is_directory=True)
+            with self.assertRaises(m.ValidationError): m.Journal(link)
+            api = FakeAPI(p, before, after, exact); api.timeout = True; api.apply = False
+            journal = root / 'journal'
+            with self.assertRaises(m.UncertainPublication): m.publish(p, before, api, FakeLease(), journal)
+            saved = json.loads((journal / 'before.json').read_text()); saved['issue']['body'] = 'changed'
+            (journal / 'before.json').write_text(json.dumps(saved))
+            with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), journal)
+            self.assertEqual(api.posts, 1)
+            with m.Journal(root / 'hardlinks') as j:
+                j.write('a.json', {'safe': True}); os.link(j.path / 'a.json', j.path / 'b.json')
+                with self.assertRaises(m.ValidationError): j.read('a.json')
+
+    def test_transport_origin_credential_source_and_exception_text_are_guarded(self):
+        m = self.module()
+        with self.assertRaises(m.ValidationError): m.ApiAdapter(lambda *args: None, origin='https://other/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
+        with self.assertRaises(m.ValidationError): m.ApiAdapter(lambda *args: None, origin=m.ORIGIN + '/api/v1', credential_source='other-profile')
+        def bad(*args): raise TimeoutError('DO_NOT_ECHO_TRANSPORT_SECRET')
+        api = m.ApiAdapter(bad, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
+        with self.assertRaises(m.ValidationError) as raised: api.identity()
+        self.assertNotIn('DO_NOT_ECHO_TRANSPORT_SECRET', str(raised.exception))
+        self.assertTrue(raised.exception.__suppress_context__)
+    def test_false_lease_result_does_not_authorize_publication(self):
+        m = self.module(); p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
+        lease = FakeLease(); lease.check = lambda plan: False
+        with tempfile.TemporaryDirectory() as td:
+            with self.assertRaises(m.ValidationError): m.publish(p, before, api, lease, td)
+            self.assertEqual(api.posts, 0)
+    def test_live_rest_bridge_over_loopback_publishes_once_then_no_ops(self):
+        import threading
+        from http.server import BaseHTTPRequestHandler, HTTPServer
+        from urllib.request import Request, urlopen
+        m = self.module(); p, before, after, exact = publication_fixture()
+        after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
+        after['commits'][0]['files'].reverse()
+        state = {'posted': False, 'posts': 0}
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args): pass
+            def respond(self, value, status=200, total=None):
+                body = json.dumps(value, ensure_ascii=False).encode()
+                self.send_response(status); self.send_header('Content-Type', 'application/json')
+                if total is not None: self.send_header('X-Total-Count', str(total))
+                self.send_header('Content-Length', str(len(body))); self.end_headers(); self.wfile.write(body)
+            def do_GET(self):
+                path = self.path.split('?')[0]; s = after if state['posted'] else before
+                if path == '/user': return self.respond(p['reviewer'])
+                if path == '/users/gitea-spec-agent': return self.respond(p['spec_reviewer'])
+                prefix = '/repos/acme/repo'; pull = prefix + '/pulls/7'
+                singles = {prefix: s['repository'], prefix + '/issues/3': s['issue'], pull: s['pr']}
+                if path in singles: return self.respond(singles[path])
+                if path == pull + '/reviews/10': return self.respond(exact)
+                arrays = {pull + '/files': s['files'], pull + '/commits': s['commits'], pull + '/reviews': s['reviews'],
+                          prefix + '/issues/3/comments': s['issue_comments'], prefix + '/issues/7/comments': s['pr_comments'],
+                          prefix + '/statuses/' + p['target']['head']: s['statuses']}
+                for rid, comments in s['review_comments'].items(): arrays[pull + '/reviews/' + rid + '/comments'] = comments
+                if path in arrays: return self.respond(arrays[path], total=len(arrays[path]))
+                self.respond({}, 404)
+            def do_POST(self):
+                self.server.received = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+                state['posts'] += 1; state['posted'] = True
+                self.respond(exact, 201)
+        server = HTTPServer(('127.0.0.1', 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start()
+        def request(path, payload=None):
+            wire = None if payload is None else json.dumps(payload, ensure_ascii=False).encode()
+            req = Request('http://127.0.0.1:' + str(server.server_port) + path, data=wire,
+                          headers={'Content-Type': 'application/json'})
+            with urlopen(req, timeout=5) as response:
+                return json.loads(response.read()), dict(response.headers), response.status
+        try:
+            # Controlled local fixture transport, not a real role credential/origin assertion.
+            api = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
+            baseline = api.snapshot(p['target'])
+            with tempfile.TemporaryDirectory() as td:
+                result = m.publish(p, baseline, api, FakeLease(), td)
+                self.assertEqual(result['mode'], 'created'); self.assertEqual(result['projected_review_ids'], [8])
+                self.assertTrue(result['commit_file_order_normalized'])
+                second = m.publish(p, baseline, api, FakeLease(), td)
+                self.assertEqual(second['mode'], 'no_op'); self.assertFalse(second['wrote'])
+                self.assertEqual(state['posts'], 1); self.assertEqual(server.received, m.build_payload(p))
+        finally:
+            server.shutdown(); server.server_close(); thread.join(timeout=5)
+            self.assertFalse(thread.is_alive())
+    def test_expired_lease_does_not_create_new_journal_directory(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        lease = FakeLease(); lease.valid = False
+        with tempfile.TemporaryDirectory() as td:
+            directory = Path(td) / 'not-created'
+            with self.assertRaises(m.ValidationError): m.publish(p, before, FakeAPI(p, before, after, exact), lease, directory)
+            self.assertFalse(directory.exists())
+
+    def test_journal_intent_fields_are_validated_on_resume(self):
+        m = self.module()
+        for field in ['schema', 'head', 'event', 'body_sha256']:
+            with self.subTest(field=field), tempfile.TemporaryDirectory() as td:
+                p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
+                api.timeout = True; api.apply = False
+                with self.assertRaises(m.UncertainPublication): m.publish(p, before, api, FakeLease(), td)
+                path = Path(td) / 'intent.json'; intent = json.loads(path.read_text()); intent[field] = 'corrupt'
+                path.write_text(json.dumps(intent)); api.created = True
+                with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), td)
+                self.assertEqual(api.posts, 1)
+
+    def test_issue_repository_mismatch_is_not_accepted(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        before['issue']['repository']['full_name'] = 'other/repo'
+        with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
+    def test_fork_head_and_nonofficial_native_states_preserve_accepted_contract(self):
+        m = self.module(); p, before, after, exact = publication_fixture()
+        for snapshot in [before, after]:
+            snapshot['pr']['head']['repo']['full_name'] = 'contributor/repo'
+            snapshot['reviews'][1]['official'] = False
+        after['reviews'][-1]['official'] = exact['official'] = False
+        self.assertTrue(m.verify_publication(p, before, after, exact)['verified'])
+        after['pr']['head']['repo']['full_name'] = 'different/repo'
+        with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
+
+
+if __name__ == '__main__':
+    unittest.main()