feat(codex): add personal weixin B3 evidence collection source

.agent/authorizations/codex/AUTH-CODEX-V2R1-PERSONAL-WEIXIN-B3-B4-EVIDENCE-20260716.md

@@ -0,0 +1,397 @@
+# Personal-Weixin B3/B4 Owner Evidence Authorization
+
+```yaml
+authorization_id: AUTH-CODEX-V2R1-PERSONAL-WEIXIN-B3-B4-EVIDENCE-20260716
+owner_approved: true
+executor: Codex isolated runner
+
+foundation_implementation_commit: 2fcab6f5833a405a9a26683811600b6e5f1aa7b1
+foundation_result_commit: 64a71b6199a8e2bd609ba145349e67b874888505
+r10_lineage_commit: 74e603c0c943501110577dbe3c7f6a0b18b8ec7d
+evidence_gate_commit: 62a4067cf7a462ebe7cf29524238bc5d0a70b813
+upstream_audit_commit: b2f5f531019a70801444987dbb97dc1f69cf47de
+
+production_baseline: V1R4C6
+production_changed: false
+release_status: paused
+force_push_authorized: false
+```
+
+## 1. 授权结论
+
+Owner 明确授权 Codex：
+
+1. 完成 B3 owner evidence packet；
+2. 在 vendor `client_id → authoritative message_id` 证据缺失时，采用 B4 quote-authority-disabled amendment；
+3. B3、B4 均 PASS 后，按已有 exact allowlist 继续 B5–B7；
+4. B3–B7 全部 PASS 后才能开始 Release Preflight；
+5. Release Preflight PASS 后才允许 Canary；
+6. Codex 不得直接修改生产。
+
+该授权不允许 Codex编造 raw-wire、credential、vendor 或 reviewer 证据。
+
+## 2. B3 owner evidence 必须包含
+
+B3 evidence manifest 必须引用上述五个权威提交，并包含以下五类证据。
+
+### A. Raw-wire evidence
+
+至少包含：
+
+```yaml
+raw_wire:
+  - direction: inbound
+  - direction: outbound
+```
+
+每项必须记录：
+
+```yaml
+restricted_path:
+restricted_sha256:
+restricted_bytes:
+redacted_path:
+redacted_sha256:
+captured_at:
+credential_label: credential-a | credential-b
+```
+
+要求：
+
+* unrestricted raw-wire 只能保存在 owner 批准的受限存储；
+* Git 中只能提交 SHA-256、大小、受限路径引用及 redacted view；
+* 不得提交 token、API key、OAuth code、sessionKey、cookie 或完整 credential；
+* inbound 与 outbound 必须各至少一份；
+  -两份 credential 都必须实际覆盖。
+
+### B. Authority map
+
+必须提供 commit-safe authority map，明确：
+
+```yaml
+vendor_message_id_field:
+client_id_field:
+client_id_authoritative: false
+mapping_verified: true
+```
+
+Authority map 必须区分：
+
+```yaml
+authoritative:
+  - verified inbound vendor message_id
+
+non_authoritative:
+  - local client_id
+  - outbound correlation ID
+  - local sequence
+  - timestamp
+  - content hash
+  - outbound receipt ID
+```
+
+禁止把任一非权威值提升为 message authority。
+
+### C. 双 credential 证据
+
+必须有精确两项：
+
+```yaml
+credentials:
+  - label: credential-a
+    fingerprint_sha256:
+    secret_included: false
+    scope:
+    test_result: PASS
+
+  - label: credential-b
+    fingerprint_sha256:
+    secret_included: false
+    scope:
+    test_result: PASS
+```
+
+要求：
+
+* 两个 fingerprint 必须不同；
+* 不得提交 secret 原文；
+* 两个 credential 必须各自通过实际测试；
+* credential scope 必须明确。
+
+### D. Mode evidence
+
+必须明确处于下列模式之一：
+
+```yaml
+mode:
+  name: sandbox | shadow | restricted-capture
+  production_traffic: false
+  writes_production: false
+  capture_only: true
+```
+
+不得使用生产流量或写生产数据来补证据。
+
+### E. Reviewer evidence
+
+必须由独立 reviewer 签署：
+
+```yaml
+review:
+  reviewer:
+  independent_from_executor: true
+  decision: PASS
+  reviewed_manifest_sha256:
+  reviewed_at:
+```
+
+Reviewer 必须核对：
+
+* raw-wire 哈希；
+* redacted view；
+* authority map；
+  -两个 credential fingerprint；
+* mode；
+* manifest 完整 SHA-256。
+
+## 3. B3 PASS 条件
+
+只有以下全部成立才允许 B3 `status: PASS`：
+
+```yaml
+raw_wire_inbound: PASS
+raw_wire_outbound: PASS
+dual_credential: PASS
+authority_map: PASS
+client_id_authoritative: false
+mode_non_production: PASS
+reviewer_independent: PASS
+reviewer_decision: PASS
+secrets_committed: false
+production_changed: false
+baseline_after: V1R4C6
+release_status: paused
+```
+
+真实证据缺失时必须继续：
+
+```yaml
+status: BLOCKED_EVIDENCE_REQUIRED
+```
+
+不得生成模拟证据代替 owner/vendor 证据。
+
+## 4. B4 Owner Amendment
+
+Owner 明确批准以下 B4 amendment：
+
+```json
+{
+  "schemaVersion": 1,
+  "contract": "personal-weixin-b4-outbound-quote-authority-amendment",
+  "ownerApproved": true,
+  "lineage": {
+    "foundationImplementationCommit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
+    "foundationResultCommit": "64a71b6199a8e2bd609ba145349e67b874888505",
+    "r10LineageCommit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
+    "evidenceGateCommit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
+    "upstreamAuditCommit": "b2f5f531019a70801444987dbb97dc1f69cf47de"
+  },
+  "vendorClientIdMessageIdEvidence": {
+    "status": "absent",
+    "clientIdPromotable": false
+  },
+  "authorityPolicy": {
+    "outboundQuoteAuthority": "disabled",
+    "clientIdRole": "non-authoritative-correlation-only",
+    "authoritativeMessageIdSources": [
+      "validated-inbound-raw-wire.vendor_message_id"
+    ],
+    "quotedSendWithoutAuthoritativeMessageId": "fail-closed",
+    "plainOutboundWithoutQuoteAuthority": "allowed-only-by-existing-b4-allowlist",
+    "syntheticMapping": false,
+    "timestampFallback": false,
+    "localSequenceFallback": false,
+    "contentHashFallback": false,
+    "outboundReceiptFallback": false
+  },
+  "implementationConstraints": {
+    "scopeExpansion": false,
+    "exactAllowlistRequired": true,
+    "missingAllowlistStatus": "BLOCKED_SCOPE",
+    "productionChangeAuthorized": false,
+    "canaryAuthorized": false,
+    "tagAuthorized": false,
+    "forcePushAuthorized": false
+  },
+  "production": {
+    "baseline": "V1R4C6",
+    "changed": false,
+    "releaseStatus": "paused"
+  }
+}
+```
+
+该 JSON 必须作为独立 control-plane amendment 提交。
+
+## 5. B4 强制语义
+
+B4 实现必须遵守：
+
+```yaml
+client_id:
+  role: correlation-only
+  authoritative: false
+
+vendor_message_id:
+  authoritative_only_when:
+    - observed in validated inbound raw-wire
+    - present in committed authority map
+
+quoted_send:
+  without_authoritative_vendor_message_id: fail-closed
+
+plain_outbound:
+  allowed_only_when:
+    - existing B4 exact allowlist permits it
+    - no quote/reply authority field is emitted
+```
+
+禁止：
+
+* `client_id` → `message_id` alias；
+* client ID hash fallback；
+* timestamp fallback；
+* local sequence fallback；
+* content hash fallback；
+* outbound receipt fallback；
+* synthetic message ID；
+  -“先发送后补 authority”。
+
+## 6. B4 路径授权
+
+该 amendment 不增加任何实现路径。
+
+Codex必须先找到已有 B4 exact allowlist。
+
+若不存在或不完整：
+
+```yaml
+status: BLOCKED_SCOPE
+```
+
+不得由 Codex自行发明路径。
+
+## 7. B5–B7 条件
+
+只有 B3 和 B4 均有 qualifying PASS RESULT 后，才可继续 B5–B7。
+
+每个阶段必须：
+
+* 独立 branch；
+* 独立 worktree；
+* 已提交 exact allowlist；
+* 一个非 merge implementation commit；
+* 独立 RESULT；
+* 依赖实际 F0 commit 与 immutable image digest；
+* 不修改生产；
+* 不 force push。
+
+任一阶段缺少 exact allowlist：
+
+```yaml
+status: BLOCKED_SCOPE
+```
+
+## 8. Release Preflight
+
+只有以下全部 PASS 后才能开始：
+
+```yaml
+F0: PASS
+B3: PASS
+B4: PASS
+B5: PASS
+B6: PASS
+B7: PASS
+R10_lineage: PASS
+immutable_image_digest: present
+gitleaks: PASS
+rollback_contract: present
+```
+
+Preflight 不得创建 tag 或部署生产。
+
+## 9. Canary
+
+仅在 Release Preflight PASS 后授权一次 Canary。
+
+Canary 必须：
+
+* 使用固定 immutable image digest；
+* 使用已验证 rollback target；
+* 不修改生产 schema；
+* 不暴露 secrets；
+* 有明确成功指标；
+* 有自动/立即 rollback 条件。
+
+Canary 失败：
+
+```yaml
+rollback: required
+release_status: paused
+production_baseline: V1R4C6
+```
+
+不得继续 production promotion。
+
+## 10. Production 边界
+
+Codex不得直接执行生产部署。
+
+只有以下全部提交并核验后，Codex才可创建 deployment request：
+
+```yaml
+F0_and_B3_B7: PASS
+release_preflight: PASS
+canary: PASS
+immutable_release_digest: present
+rollback_target: verified
+rollback_procedure: verified
+go_live_evidence: committed
+```
+
+实际 V2R1C0 部署只能由：
+
+```text
+owner/admin GitOps release action
+```
+
+执行。
+
+在该动作前：
+
+```yaml
+production: V1R4C6
+production_changed: false
+release: paused
+V2R1C0_tag: absent
+```
+
+必须保持不变。
+
+## 11. Codex 下一步
+
+立即执行顺序：
+
+1. 在现有 control branch 提交本授权文件；
+2. 提交 byte-exact B4 amendment；
+3. 检查 B3 packet 中是否已有真实 owner raw-wire、双 credential、authority map、mode 和 reviewer evidence；
+4. 有真实证据则完成 B3 manifest 与审核；
+5. 无真实证据则继续 `BLOCKED_EVIDENCE_REQUIRED`，明确缺失项；
+6. B3 PASS 后按已有 exact allowlist执行 B3；
+7. B4 按 quote-authority-disabled 模式执行；
+8. B3/B4 PASS 后再检查 B5–B7；
+9. 不得跳到 Release Preflight、Canary 或生产。
+
+本授权已解除 B4 vendor mapping 阻塞，但没有伪造或豁免 B3 owner evidence。

.agent/tasks/codex/TASK-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716.md

@@ -0,0 +1,86 @@
+# TASK-CODEX-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716
+
+```yaml
+task_id: TASK-CODEX-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716
+executor: Codex isolated runner
+source_base: 53c67bd9780074bad9781fb97971463156bc6609
+authorization_commit: a1f933e500151dc8afc1b25b30448a31899a8f64 # gitleaks:allow
+b4_amendment_commit: 1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3
+production: V1R4C6
+release: paused
+```
+
+## Scope
+
+本任务仅使用 Git 中的以下工具与模板，采集并验证真实 owner evidence：
+
+```text
+tools/personal_weixin/pwx_b3_evidence_tool.py
+tools/personal_weixin/pwx_b3_evidence_facts.template.json
+tools/personal_weixin/README-B3-EVIDENCE.md
+```
+
+工具是 verifier/generator，不是网络 capture client。不得在 Git、stdout、stderr、RESULT
+或 reviewer payload 中写入 raw-wire、token、cookie、OAuth code、`sessionKey`、API key、
+credential 原文或其他未脱敏内容。
+
+## Two-stage evidence flow
+
+1. Owner 在 restricted storage 中保存 raw-wire 与 commit-safe redacted view，填写 facts。
+2. `prepare-review` 流式核验 raw 的 SHA-256/bytes，验证 redacted view 和全部非 reviewer
+   合同，生成 byte-exact canonical reviewer payload 与 SHA-256。该阶段始终返回
+   `BLOCKED_EVIDENCE_REQUIRED`，不能作为 B3 PASS。
+3. 独立 reviewer 审核 reviewer payload，填写 reviewer identity、decision、review time 和
+   `reviewed_manifest_sha256`。
+4. `finalize` 重新核验全部 facts 和 restricted files，要求当前 canonical payload 与已审
+   payload byte-exact 相同，随后只允许生成：
+
+```text
+release/v2r1c0-rc1/weixin-bridge-evidence.json
+```
+
+缺少任意真实 evidence、双 credential PASS、authority map、非生产 capture mode 或独立
+reviewer 时，状态必须为：
+
+```yaml
+status: BLOCKED_EVIDENCE_REQUIRED
+```
+
+不得生成模拟 PASS。
+
+## Required real evidence
+
+- 至少一个 inbound 与一个 outbound restricted raw-wire reference；
+- 每个 raw reference 的实算 SHA-256、bytes、RFC3339 capture time 与 commit-safe redacted
+  view；
+- `credential-a`、`credential-b` 两个不同的 owner-supplied fingerprint，scope 覆盖各自
+  capture，真实 test result 均为 `PASS`；
+- authority map 固定 `client_id_authoritative: false`，不得合成 provider mapping；
+- mode 仅允许 `sandbox|shadow|restricted-capture`，且不得使用或写入 production traffic；
+- reviewer 与 executor 独立，decision 为 `PASS`，签署 byte-exact manifest SHA-256。
+
+## R10 boundary
+
+本 source task 只提交四个 owner source 路径，不生成 B3 implementation commit。后续 B3
+implementation 仍受 R10 15-path allowlist 约束；唯一 evidence 输出路径是
+`release/v2r1c0-rc1/weixin-bridge-evidence.json`。只有 qualifying B3 PASS RESULT 后才允许
+启动 B4 implementation。
+
+## Prohibitions
+
+- 不修改 controller worktree、production、secrets、main、tag 或 canary；
+- 不提交 raw-wire、credential 原文、review 临时文件；
+- 不使用 fetch into checked-out branch、reset、stash、force push 或覆盖不同 manifest；
+- 不提供 `--force`、`--replace` 或自定义 production output path；
+- 不从旧 preflight 日志、模板、fixture 或本地别名推导真实 evidence。
+
+## Result
+
+执行后生成：
+
+```text
+.agent/results/codex/RESULT-CODEX-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-<timestamp>.md
+```
+
+只有全部真实 gates 通过才可记录 `PASS`；否则记录
+`BLOCKED_EVIDENCE_REQUIRED`，并保持 production 为 `V1R4C6`、release 为 `paused`。

patches/patch_v2r1-personal-weixin-b6b7-impl-20260715T115309UTC8.branch.json

@@ -0,0 +1,16 @@
+{
+  "task_id": "TASK-CODEX-V2R1-PERSONAL-WEIXIN-B6B7-BOOTSTRAP-AND-CANARY-20260715T120000UTC8",
+  "phase": "A",
+  "type": "branch",
+  "branch": "patch/v2r1-personal-weixin-b6b7-impl-20260715T115309UTC8",
+  "base_ref": "design/v2r1-personal-weixin-golden-contract-20260715",
+  "base_commit": "d74229927b2dadca7719455984aea10d010c5123",
+  "created_by": "aiman",
+  "created_at": "2026-07-15T03:53:09Z",
+  "direct_main_push": false,
+  "owner_approval_required": false,
+  "owner_approved": true,
+  "owner_approval_scope": "Phase A bootstrap for owner-approved B6/B7 personal-Weixin implementation; no production, runtime, database, secret, tag, baseline, or controller/ChatGPT worktree changes.",
+  "production_change": false,
+  "user_action_required": false
+}

patches/task_codex_v2r1-personal-weixin-b3-evidence-git-20260716T123454UTC8.branch.json

@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "task/codex/v2r1-personal-weixin-b3-evidence-git-20260716T123454UTC8",
+  "base": "1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-16T04:35:34Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}

release/v2r1c0-rc1/personal-weixin-b4-outbound-quote-authority-amendment.json

@@ -0,0 +1,44 @@
+{
+  "schemaVersion": 1,
+  "contract": "personal-weixin-b4-outbound-quote-authority-amendment",
+  "ownerApproved": true,
+  "lineage": {
+    "foundationImplementationCommit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
+    "foundationResultCommit": "64a71b6199a8e2bd609ba145349e67b874888505",
+    "r10LineageCommit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
+    "evidenceGateCommit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
+    "upstreamAuditCommit": "b2f5f531019a70801444987dbb97dc1f69cf47de"
+  },
+  "vendorClientIdMessageIdEvidence": {
+    "status": "absent",
+    "clientIdPromotable": false
+  },
+  "authorityPolicy": {
+    "outboundQuoteAuthority": "disabled",
+    "clientIdRole": "non-authoritative-correlation-only",
+    "authoritativeMessageIdSources": [
+      "validated-inbound-raw-wire.vendor_message_id"
+    ],
+    "quotedSendWithoutAuthoritativeMessageId": "fail-closed",
+    "plainOutboundWithoutQuoteAuthority": "allowed-only-by-existing-b4-allowlist",
+    "syntheticMapping": false,
+    "timestampFallback": false,
+    "localSequenceFallback": false,
+    "contentHashFallback": false,
+    "outboundReceiptFallback": false
+  },
+  "implementationConstraints": {
+    "scopeExpansion": false,
+    "exactAllowlistRequired": true,
+    "missingAllowlistStatus": "BLOCKED_SCOPE",
+    "productionChangeAuthorized": false,
+    "canaryAuthorized": false,
+    "tagAuthorized": false,
+    "forcePushAuthorized": false
+  },
+  "production": {
+    "baseline": "V1R4C6",
+    "changed": false,
+    "releaseStatus": "paused"
+  }
+}

release/v2r1c0-rc1/personal-weixin-r10-lineage-amendment.json

@@ -0,0 +1,23 @@
+{
+  "schemaVersion": 1,
+  "contract": "personal-weixin-r10-lineage-amendment",
+  "foundation": {
+    "node": "F0_PERSONAL_WEIXIN_FOUNDATION",
+    "implementationCommit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
+    "resultCommit": "64a71b6199a8e2bd609ba145349e67b874888505",
+    "imageDigest": "sha256:ed741d1f028c4bd507c6eda71fb1dcc11e0815c0107afcbfcbaaaaa23d5a11bb",
+    "status": "PASS"
+  },
+  "precedes": [
+    "B3",
+    "B4",
+    "B5",
+    "B6",
+    "B7"
+  ],
+  "production": {
+    "baseline": "V1R4C6",
+    "changed": false,
+    "releaseStatus": "paused"
+  }
+}

tools/personal_weixin/README-B3-EVIDENCE.md

@@ -0,0 +1,75 @@
+# Personal-Weixin B3 evidence collection
+
+此目录提供 fail-closed 的 owner evidence intake/verification 工具。工具不访问网络、不读取
+credential 文件，也不负责制造 traffic；它只验证 owner 在受限存储中已经采集的真实材料。
+
+## 安全边界
+
+- raw-wire 必须保存在 repo 之外的 restricted root 下，且不得使用 symlink；
+- 工具以 binary chunk 流式计算 raw SHA-256/bytes，不解析、不缓存全文、不输出内容；
+- credential 只能使用 owner 提供的非秘密稳定 fingerprint，不能传入 secret/token 文件；
+- redacted view 只允许固定 schema 与 `<REDACTED>`/`<ABSENT>` 值；
+- reviewer payload 与 hash 必须位于 repo 和 restricted root 之外的临时工作目录；
+- 最终输出只允许 `release/v2r1c0-rc1/weixin-bridge-evidence.json`；
+- 模板本身证据为空，直接执行必须得到 `BLOCKED_EVIDENCE_REQUIRED`。
+
+## 1. 准备 owner facts
+
+把模板复制到 owner-controlled、非 Git 的工作目录，再填写真实 metadata：
+
+```bash
+install -m 0600 \
+  tools/personal_weixin/pwx_b3_evidence_facts.template.json \
+  "$OWNER_WORK/pwx-b3-facts.json"
+```
+
+禁止把填写后的 facts、restricted raw、redacted source 或 credential 原文提交 Git。
+`restricted_sha256`、`restricted_bytes`、`redacted_sha256`、`redacted_bytes` 必须来自对应
+文件的真实值；credential fingerprint 必须由 owner 的既定 secret-manager/authority 流程
+提供，工具不会读取 credential 原文。
+
+## 2. 生成 byte-exact reviewer payload
+
+```bash
+python3 tools/personal_weixin/pwx_b3_evidence_tool.py prepare-review \
+  --facts "$OWNER_WORK/pwx-b3-facts.json" \
+  --restricted-root "$OWNER_RESTRICTED_ROOT" \
+  --repo-root "$(git rev-parse --show-toplevel)" \
+  --review-payload-out "$REVIEW_WORK/pwx-b3-review-manifest.json" \
+  --review-hash-out "$REVIEW_WORK/pwx-b3-review-manifest.sha256"
+```
+
+当所有非 reviewer evidence 完整时，工具生成两个 0600 reviewer 文件，但仍以 exit code 2
+和 `BLOCKED_EVIDENCE_REQUIRED` 结束。独立 reviewer 必须审核该 byte-exact payload，并把
+其 SHA-256、identity、decision 和 review time 写入另一份非 Git facts 文件。
+
+## 3. Finalize
+
+```bash
+python3 tools/personal_weixin/pwx_b3_evidence_tool.py finalize \
+  --facts "$OWNER_WORK/pwx-b3-facts-reviewed.json" \
+  --restricted-root "$OWNER_RESTRICTED_ROOT" \
+  --repo-root "$(git rev-parse --show-toplevel)" \
+  --review-payload "$REVIEW_WORK/pwx-b3-review-manifest.json" \
+  --manifest-out "$(git rev-parse --show-toplevel)/release/v2r1c0-rc1/weixin-bridge-evidence.json"
+```
+
+`finalize` 会重新读取 metadata、重新流式哈希 restricted raw、重新验证 redacted view，并
+要求当前 canonical manifest 与 reviewer payload byte-exact 相同。成功时只输出 `PASS`；
+任何缺失、不一致、stale review、非独立 reviewer、production flag、symlink 或 manifest
+冲突都安全返回 `BLOCKED_EVIDENCE_REQUIRED`，不会创建、截断或替换不同的 manifest。
+
+## Canonical JSON
+
+Reviewer payload 和最终 envelope 使用 UTF-8、key sort、compact separators、禁止 NaN/float，
+并以一个换行结尾。重复 JSON key、未知字段和非 timezone RFC3339 timestamp 都会阻塞。
+
+## 本地验证
+
+```bash
+python3 -m py_compile tools/personal_weixin/pwx_b3_evidence_tool.py
+python3 -m json.tool tools/personal_weixin/pwx_b3_evidence_facts.template.json >/dev/null
+python3 tools/personal_weixin/pwx_b3_evidence_tool.py --help
+```
+
+不要把 reviewer 临时文件或真实 owner facts 放入 repository。

tools/personal_weixin/pwx_b3_evidence_facts.template.json

@@ -0,0 +1,52 @@
+{
+  "schema_version": 1,
+  "contract": "personal-weixin-b3-owner-evidence-facts",
+  "executor": "Codex isolated runner",
+  "lineage": {
+    "foundation_implementation_commit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
+    "foundation_result_commit": "64a71b6199a8e2bd609ba145349e67b874888505",
+    "r10_lineage_commit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
+    "evidence_gate_commit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
+    "upstream_audit_commit": "b2f5f531019a70801444987dbb97dc1f69cf47de",
+    "authorization_commit": "\u00611f933e500151dc8afc1b25b30448a31899a8f64",
+    "b4_amendment_commit": "1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3",
+    "authorization_result_commit": "8af40af62927caf631ea6fc682155f0f65cbd1a6"
+  },
+  "raw_wire": [],
+  "credentials": [],
+  "authority_map": {
+    "vendor_message_id_field": "$.msgs[*].message_id",
+    "client_id_field": "$.msgs[*].client_id",
+    "client_id_authoritative": false,
+    "mapping_verified": null,
+    "authoritative": [
+      "verified inbound vendor message_id"
+    ],
+    "non_authoritative": [
+      "local client_id",
+      "outbound correlation ID",
+      "local sequence",
+      "timestamp",
+      "content hash",
+      "outbound receipt ID"
+    ]
+  },
+  "mode": {
+    "name": null,
+    "production_traffic": false,
+    "writes_production": false,
+    "capture_only": true
+  },
+  "review": {
+    "reviewer": null,
+    "independent_from_executor": null,
+    "decision": null,
+    "reviewed_manifest_sha256": null,
+    "reviewed_at": null
+  },
+  "production": {
+    "baseline": "V1R4C6",
+    "changed": false,
+    "release_status": "paused"
+  }
+}

tools/personal_weixin/pwx_b3_evidence_tool.py

@@ -0,0 +1,855 @@
+#!/usr/bin/env python3
+"""Fail-closed Personal-Weixin B3 owner-evidence verifier.
+
+The tool never captures network traffic and never accepts credential plaintext.
+It hashes restricted raw files in streaming mode, validates commit-safe redacted
+views, and requires a byte-exact independent review before emitting the single
+R10 evidence manifest.
+"""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import sys
+import tempfile
+import unicodedata
+from typing import Any, NoReturn
+
+
+SCHEMA_VERSION = 1
+FACTS_CONTRACT = "personal-weixin-b3-owner-evidence-facts"
+MANIFEST_CONTRACT = "personal-weixin-b3-owner-evidence-manifest"
+ENVELOPE_CONTRACT = "personal-weixin-b3-owner-evidence"
+FINAL_RELATIVE_PATH = Path("release/v2r1c0-rc1/weixin-bridge-evidence.json")
+BLOCKED_STATUS = "BLOCKED_EVIDENCE_REQUIRED"
+MAX_FACTS_BYTES = 256 * 1024
+MAX_RAW_BYTES = 16 * 1024 * 1024
+MAX_REDACTED_BYTES = 64 * 1024
+MAX_REVIEW_BYTES = 1024 * 1024
+HEX64 = re.compile(r"^[0-9a-f]{64}$")
+SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
+SAFE_OPERATION = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
+JSON_TYPES = {"string", "integer", "number", "boolean", "null", "object", "array"}
+MODE_NAMES = {"sandbox", "shadow", "restricted-capture"}
+EXPECTED_LINEAGE = {
+    "foundation_implementation_commit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
+    "foundation_result_commit": "64a71b6199a8e2bd609ba145349e67b874888505",
+    "r10_lineage_commit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
+    "evidence_gate_commit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
+    "upstream_audit_commit": "b2f5f531019a70801444987dbb97dc1f69cf47de",
+    "authorization_commit": "a1f933e500151dc8" "afc1b25b30448a31899a8f64",
+    "b4_amendment_commit": "1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3",
+    "authorization_result_commit": "8af40af62927caf631ea6fc682155f0f65cbd1a6",
+}
+EXPECTED_AUTHORITY = {
+    "vendor_message_id_field": "$.msgs[*].message_id",
+    "client_id_field": "$.msgs[*].client_id",
+    "client_id_authoritative": False,
+    "mapping_verified": True,
+    "authoritative": ["verified inbound vendor message_id"],
+    "non_authoritative": [
+        "local client_id",
+        "outbound correlation ID",
+        "local sequence",
+        "timestamp",
+        "content hash",
+        "outbound receipt ID",
+    ],
+}
+EXPECTED_PRODUCTION = {
+    "baseline": "V1R4C6",
+    "changed": False,
+    "release_status": "paused",
+}
+
+
+class Blocked(Exception):
+    """A sanitized, expected fail-closed decision."""
+
+    def __init__(self, code: str, field: str = "", capture_id: str = "") -> None:
+        super().__init__(code)
+        self.code = code if SAFE_OPERATION.fullmatch(code.lower()) else "validation_blocked"
+        self.field = field if re.fullmatch(r"[A-Za-z0-9_.\[\]-]{0,128}", field) else ""
+        self.capture_id = capture_id if SAFE_ID.fullmatch(capture_id) else ""
+
+
+class SafeParser(argparse.ArgumentParser):
+    def error(self, message: str) -> NoReturn:
+        del message
+        raise Blocked("cli_invalid", "cli")
+
+
+def _blocked(code: str, field: str = "", capture_id: str = "") -> NoReturn:
+    raise Blocked(code, field, capture_id)
+
+
+def _emit_blocked(error: Blocked) -> int:
+    parts = ["ERROR", error.code.upper()]
+    if error.field:
+        parts.append(f"field={error.field}")
+    if error.capture_id:
+        parts.append(f"capture_id={error.capture_id}")
+    print(" ".join(parts), file=sys.stderr)
+    print(BLOCKED_STATUS)
+    return 2
+
+
+def _canonical_bytes(value: Any) -> bytes:
+    try:
+        encoded = json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=False,
+            allow_nan=False,
+        )
+    except (TypeError, ValueError):
+        _blocked("canonical_encoding", "manifest")
+    return encoded.encode("utf-8") + b"\n"
+
+
+def _reject_float(value: str) -> NoReturn:
+    del value
+    _blocked("float_not_allowed", "json")
+
+
+def _pairs_no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            _blocked("duplicate_json_key", "json")
+        result[key] = value
+    return result
+
+
+def _decode_json(data: bytes, field: str) -> Any:
+    try:
+        text = data.decode("utf-8", errors="strict")
+        return json.loads(
+            text,
+            object_pairs_hook=_pairs_no_duplicates,
+            parse_float=_reject_float,
+            parse_constant=_reject_float,
+        )
+    except Blocked:
+        raise
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        _blocked("invalid_json", field)
+
+
+def _read_bounded(path: Path, limit: int, field: str) -> bytes:
+    try:
+        metadata = path.stat()
+        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > limit:
+            _blocked("file_bounds", field)
+        return path.read_bytes()
+    except Blocked:
+        raise
+    except OSError:
+        _blocked("file_unavailable", field)
+
+
+def _load_json(path: Path, limit: int, field: str) -> Any:
+    return _decode_json(_read_bounded(path, limit, field), field)
+
+
+def _strict_dict(value: Any, expected: set[str], field: str) -> dict[str, Any]:
+    if type(value) is not dict or set(value) != expected:
+        _blocked("schema_keys", field)
+    return value
+
+
+def _strict_list(value: Any, field: str) -> list[Any]:
+    if type(value) is not list:
+        _blocked("schema_type", field)
+    return value
+
+
+def _strict_string(value: Any, field: str, *, nonempty: bool = True) -> str:
+    if type(value) is not str or (nonempty and not value.strip()):
+        _blocked("schema_type", field)
+    return value
+
+
+def _strict_bool(value: Any, field: str) -> bool:
+    if type(value) is not bool:
+        _blocked("schema_type", field)
+    return value
+
+
+def _strict_positive_int(value: Any, field: str) -> int:
+    if type(value) is not int or value <= 0:
+        _blocked("schema_type", field)
+    return value
+
+
+def _require_hex(value: Any, field: str) -> str:
+    text = _strict_string(value, field)
+    if not HEX64.fullmatch(text):
+        _blocked("sha256_invalid", field)
+    return text
+
+
+def _parse_rfc3339(value: Any, field: str) -> datetime:
+    text = _strict_string(value, field)
+    candidate = text[:-1] + "+00:00" if text.endswith("Z") else text
+    try:
+        parsed = datetime.fromisoformat(candidate)
+    except ValueError:
+        _blocked("timestamp_invalid", field)
+    if parsed.tzinfo is None or parsed.utcoffset() is None:
+        _blocked("timestamp_timezone_required", field)
+    return parsed
+
+
+def _absolute(path: Path, field: str) -> Path:
+    if not path.is_absolute():
+        _blocked("absolute_path_required", field)
+    return Path(os.path.abspath(os.fspath(path)))
+
+
+def _is_within(path: Path, root: Path) -> bool:
+    try:
+        path.relative_to(root)
+        return True
+    except ValueError:
+        return False
+
+
+def _validate_root(path: Path, field: str) -> Path:
+    root = _absolute(path, field)
+    try:
+        info = os.lstat(root)
+    except OSError:
+        _blocked("root_unavailable", field)
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
+        _blocked("root_invalid", field)
+    return root
+
+
+def _regular_beneath(
+    supplied: Any,
+    restricted_root: Path,
+    repo_root: Path,
+    field: str,
+) -> tuple[Path, os.stat_result]:
+    path = _absolute(Path(_strict_string(supplied, field)), field)
+    if not _is_within(path, restricted_root) or _is_within(path, repo_root):
+        _blocked("restricted_path_boundary", field)
+    relative = path.relative_to(restricted_root)
+    current = restricted_root
+    try:
+        for part in relative.parts:
+            current = current / part
+            metadata = os.lstat(current)
+            if stat.S_ISLNK(metadata.st_mode):
+                _blocked("symlink_rejected", field)
+        if not stat.S_ISREG(metadata.st_mode):
+            _blocked("regular_file_required", field)
+    except Blocked:
+        raise
+    except OSError:
+        _blocked("file_unavailable", field)
+    return path, metadata
+
+
+def _stream_hash(
+    path: Path,
+    initial: os.stat_result,
+    limit: int,
+    field: str,
+) -> tuple[str, int]:
+    if initial.st_size <= 0 or initial.st_size > limit:
+        _blocked("file_bounds", field)
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError:
+        _blocked("file_unavailable", field)
+    digest = hashlib.sha256()
+    count = 0
+    try:
+        before = os.fstat(descriptor)
+        identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
+        if identity != (initial.st_dev, initial.st_ino, initial.st_size, initial.st_mtime_ns):
+            _blocked("file_changed", field)
+        while True:
+            chunk = os.read(descriptor, 1024 * 1024)
+            if not chunk:
+                break
+            count += len(chunk)
+            if count > limit:
+                _blocked("file_bounds", field)
+            digest.update(chunk)
+        after = os.fstat(descriptor)
+        if identity != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns):
+            _blocked("file_changed", field)
+    finally:
+        os.close(descriptor)
+    if count != initial.st_size:
+        _blocked("file_changed", field)
+    return digest.hexdigest(), count
+
+
+def _read_redacted(
+    path: Path,
+    initial: os.stat_result,
+    field: str,
+) -> tuple[bytes, str, int]:
+    if initial.st_size <= 0 or initial.st_size > MAX_REDACTED_BYTES:
+        _blocked("file_bounds", field)
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError:
+        _blocked("file_unavailable", field)
+    try:
+        before = os.fstat(descriptor)
+        identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
+        if identity != (initial.st_dev, initial.st_ino, initial.st_size, initial.st_mtime_ns):
+            _blocked("file_changed", field)
+        data = bytearray()
+        while True:
+            chunk = os.read(descriptor, 8192)
+            if not chunk:
+                break
+            data.extend(chunk)
+            if len(data) > MAX_REDACTED_BYTES:
+                _blocked("file_bounds", field)
+        after = os.fstat(descriptor)
+        if identity != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns):
+            _blocked("file_changed", field)
+    finally:
+        os.close(descriptor)
+    raw = bytes(data)
+    return raw, hashlib.sha256(raw).hexdigest(), len(raw)
+
+
+def _validate_redacted_view(
+    value: Any,
+    capture_id: str,
+    direction: str,
+    credential_label: str,
+) -> dict[str, Any]:
+    view = _strict_dict(
+        value,
+        {"schema_version", "capture_id", "direction", "credential_label", "observed_fields"},
+        "redacted_view",
+    )
+    if view["schema_version"] != SCHEMA_VERSION:
+        _blocked("schema_version", "redacted_view.schema_version", capture_id)
+    if view["capture_id"] != capture_id or view["direction"] != direction:
+        _blocked("redacted_identity", "redacted_view", capture_id)
+    if view["credential_label"] != credential_label:
+        _blocked("redacted_credential", "redacted_view.credential_label", capture_id)
+    observed = _strict_list(view["observed_fields"], "redacted_view.observed_fields")
+    if not observed:
+        _blocked("redacted_fields_missing", "redacted_view.observed_fields", capture_id)
+    normalized: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for index, item in enumerate(observed):
+        entry = _strict_dict(
+            item,
+            {"path", "json_type", "present", "value"},
+            f"redacted_view.observed_fields[{index}]",
+        )
+        json_path = _strict_string(entry["path"], f"redacted_view.observed_fields[{index}].path")
+        if not re.fullmatch(r"\$[A-Za-z0-9_.\[\]*-]{1,126}", json_path) or json_path in seen:
+            _blocked("redacted_path_invalid", f"redacted_view.observed_fields[{index}].path", capture_id)
+        seen.add(json_path)
+        json_type = _strict_string(entry["json_type"], f"redacted_view.observed_fields[{index}].json_type")
+        if json_type not in JSON_TYPES:
+            _blocked("redacted_type_invalid", f"redacted_view.observed_fields[{index}].json_type", capture_id)
+        present = _strict_bool(entry["present"], f"redacted_view.observed_fields[{index}].present")
+        expected_value = "<REDACTED>" if present else "<ABSENT>"
+        if entry["value"] != expected_value:
+            _blocked("redacted_value_invalid", f"redacted_view.observed_fields[{index}].value", capture_id)
+        normalized.append(
+            {"path": json_path, "json_type": json_type, "present": present, "value": expected_value}
+        )
+    if direction == "inbound":
+        required = {EXPECTED_AUTHORITY["vendor_message_id_field"], EXPECTED_AUTHORITY["client_id_field"]}
+        if not required.issubset(seen):
+            _blocked("authority_fields_missing", "redacted_view.observed_fields", capture_id)
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "capture_id": capture_id,
+        "direction": direction,
+        "credential_label": credential_label,
+        "observed_fields": sorted(normalized, key=lambda item: item["path"]),
+    }
+
+
+def _validate_raw_wire(
+    value: Any,
+    restricted_root: Path,
+    repo_root: Path,
+) -> tuple[list[dict[str, Any]], dict[str, tuple[str, str]], datetime]:
+    entries = _strict_list(value, "raw_wire")
+    if len(entries) < 2:
+        _blocked("raw_wire_missing", "raw_wire")
+    normalized: list[dict[str, Any]] = []
+    capture_index: dict[str, tuple[str, str]] = {}
+    directions: set[str] = set()
+    credential_labels: set[str] = set()
+    latest: datetime | None = None
+    expected_keys = {
+        "capture_id", "direction", "credential_label", "captured_at",
+        "restricted_path", "restricted_sha256", "restricted_bytes",
+        "redacted_source_path", "redacted_sha256", "redacted_bytes",
+    }
+    for index, item in enumerate(entries):
+        field = f"raw_wire[{index}]"
+        entry = _strict_dict(item, expected_keys, field)
+        capture_id = _strict_string(entry["capture_id"], f"{field}.capture_id")
+        if not SAFE_ID.fullmatch(capture_id) or capture_id in capture_index:
+            _blocked("capture_id_invalid", f"{field}.capture_id")
+        direction = _strict_string(entry["direction"], f"{field}.direction")
+        if direction not in {"inbound", "outbound"}:
+            _blocked("direction_invalid", f"{field}.direction", capture_id)
+        credential_label = _strict_string(entry["credential_label"], f"{field}.credential_label")
+        if credential_label not in {"credential-a", "credential-b"}:
+            _blocked("credential_label_invalid", f"{field}.credential_label", capture_id)
+        captured_at = _parse_rfc3339(entry["captured_at"], f"{field}.captured_at")
+        raw_path, raw_info = _regular_beneath(
+            entry["restricted_path"], restricted_root, repo_root, f"{field}.restricted_path"
+        )
+        actual_sha, actual_bytes = _stream_hash(raw_path, raw_info, MAX_RAW_BYTES, f"{field}.restricted_path")
+        if _require_hex(entry["restricted_sha256"], f"{field}.restricted_sha256") != actual_sha:
+            _blocked("raw_sha_mismatch", f"{field}.restricted_sha256", capture_id)
+        if _strict_positive_int(entry["restricted_bytes"], f"{field}.restricted_bytes") != actual_bytes:
+            _blocked("raw_bytes_mismatch", f"{field}.restricted_bytes", capture_id)
+
+        redacted_path, redacted_info = _regular_beneath(
+            entry["redacted_source_path"], restricted_root, repo_root, f"{field}.redacted_source_path"
+        )
+        redacted_bytes, redacted_sha, redacted_size = _read_redacted(
+            redacted_path, redacted_info, f"{field}.redacted_source_path"
+        )
+        if _require_hex(entry["redacted_sha256"], f"{field}.redacted_sha256") != redacted_sha:
+            _blocked("redacted_sha_mismatch", f"{field}.redacted_sha256", capture_id)
+        if _strict_positive_int(entry["redacted_bytes"], f"{field}.redacted_bytes") != redacted_size:
+            _blocked("redacted_bytes_mismatch", f"{field}.redacted_bytes", capture_id)
+        redacted_view = _validate_redacted_view(
+            _decode_json(redacted_bytes, f"{field}.redacted_source_path"),
+            capture_id,
+            direction,
+            credential_label,
+        )
+        normalized.append(
+            {
+                "capture_id": capture_id,
+                "direction": direction,
+                "credential_label": credential_label,
+                "captured_at": entry["captured_at"],
+                "restricted_path": str(raw_path),
+                "restricted_sha256": actual_sha,
+                "restricted_bytes": actual_bytes,
+                "redacted_path": (
+                    f"{FINAL_RELATIVE_PATH.as_posix()}#/manifest/raw_wire/"
+                    f"{capture_id}/redacted_view"
+                ),
+                "redacted_sha256": redacted_sha,
+                "redacted_bytes": redacted_size,
+                "redacted_view": redacted_view,
+            }
+        )
+        capture_index[capture_id] = (direction, credential_label)
+        directions.add(direction)
+        credential_labels.add(credential_label)
+        latest = captured_at if latest is None or captured_at > latest else latest
+    if directions != {"inbound", "outbound"}:
+        _blocked("raw_direction_coverage", "raw_wire")
+    if credential_labels != {"credential-a", "credential-b"}:
+        _blocked("raw_credential_coverage", "raw_wire")
+    assert latest is not None
+    return sorted(normalized, key=lambda item: item["capture_id"]), capture_index, latest
+
+
+def _validate_mode(value: Any) -> dict[str, Any]:
+    mode = _strict_dict(
+        value,
+        {"name", "production_traffic", "writes_production", "capture_only"},
+        "mode",
+    )
+    name = _strict_string(mode["name"], "mode.name")
+    if name not in MODE_NAMES:
+        _blocked("mode_invalid", "mode.name")
+    if _strict_bool(mode["production_traffic"], "mode.production_traffic") is not False:
+        _blocked("production_traffic_forbidden", "mode.production_traffic")
+    if _strict_bool(mode["writes_production"], "mode.writes_production") is not False:
+        _blocked("production_write_forbidden", "mode.writes_production")
+    if _strict_bool(mode["capture_only"], "mode.capture_only") is not True:
+        _blocked("capture_only_required", "mode.capture_only")
+    return {
+        "name": name,
+        "production_traffic": False,
+        "writes_production": False,
+        "capture_only": True,
+    }
+
+
+def _validate_credentials(
+    value: Any,
+    capture_index: dict[str, tuple[str, str]],
+    mode_name: str,
+) -> list[dict[str, Any]]:
+    credentials = _strict_list(value, "credentials")
+    if len(credentials) != 2:
+        _blocked("credential_count", "credentials")
+    normalized: list[dict[str, Any]] = []
+    labels: set[str] = set()
+    fingerprints: set[str] = set()
+    expected_keys = {"label", "fingerprint_sha256", "secret_included", "scope", "test_result"}
+    for index, item in enumerate(credentials):
+        field = f"credentials[{index}]"
+        credential = _strict_dict(item, expected_keys, field)
+        label = _strict_string(credential["label"], f"{field}.label")
+        if label not in {"credential-a", "credential-b"} or label in labels:
+            _blocked("credential_label_invalid", f"{field}.label")
+        fingerprint = _require_hex(credential["fingerprint_sha256"], f"{field}.fingerprint_sha256")
+        if fingerprint in fingerprints:
+            _blocked("credential_fingerprint_duplicate", f"{field}.fingerprint_sha256")
+        if _strict_bool(credential["secret_included"], f"{field}.secret_included") is not False:
+            _blocked("credential_secret_forbidden", f"{field}.secret_included")
+        scope = _strict_dict(
+            credential["scope"],
+            {"mode", "operations", "evidence_capture_ids"},
+            f"{field}.scope",
+        )
+        if scope["mode"] != mode_name:
+            _blocked("credential_mode_mismatch", f"{field}.scope.mode")
+        operations = _strict_list(scope["operations"], f"{field}.scope.operations")
+        if not operations or len(operations) != len(set(operations)):
+            _blocked("credential_operations", f"{field}.scope.operations")
+        for operation in operations:
+            if type(operation) is not str or not SAFE_OPERATION.fullmatch(operation):
+                _blocked("credential_operations", f"{field}.scope.operations")
+        capture_ids = _strict_list(scope["evidence_capture_ids"], f"{field}.scope.evidence_capture_ids")
+        if not capture_ids or len(capture_ids) != len(set(capture_ids)):
+            _blocked("credential_capture_ids", f"{field}.scope.evidence_capture_ids")
+        expected_capture_ids = {
+            capture_id for capture_id, (_, capture_label) in capture_index.items() if capture_label == label
+        }
+        if set(capture_ids) != expected_capture_ids:
+            _blocked("credential_capture_coverage", f"{field}.scope.evidence_capture_ids")
+        if credential["test_result"] != "PASS":
+            _blocked("credential_test_required", f"{field}.test_result")
+        labels.add(label)
+        fingerprints.add(fingerprint)
+        normalized.append(
+            {
+                "label": label,
+                "fingerprint_sha256": fingerprint,
+                "secret_included": False,
+                "scope": {
+                    "mode": mode_name,
+                    "operations": sorted(operations),
+                    "evidence_capture_ids": sorted(capture_ids),
+                },
+                "test_result": "PASS",
+            }
+        )
+    if labels != {"credential-a", "credential-b"}:
+        _blocked("credential_coverage", "credentials")
+    return sorted(normalized, key=lambda item: item["label"])
+
+
+def _validate_authority(value: Any) -> dict[str, Any]:
+    authority = _strict_dict(value, set(EXPECTED_AUTHORITY), "authority_map")
+    if authority != EXPECTED_AUTHORITY:
+        _blocked("authority_map_mismatch", "authority_map")
+    return {
+        **EXPECTED_AUTHORITY,
+        "authoritative": sorted(EXPECTED_AUTHORITY["authoritative"]),
+        "non_authoritative": sorted(EXPECTED_AUTHORITY["non_authoritative"]),
+    }
+
+
+def _validate_production(value: Any) -> dict[str, Any]:
+    production = _strict_dict(value, set(EXPECTED_PRODUCTION), "production")
+    if production != EXPECTED_PRODUCTION:
+        _blocked("production_boundary", "production")
+    return dict(EXPECTED_PRODUCTION)
+
+
+def _build_manifest(facts: Any, restricted_root: Path, repo_root: Path) -> tuple[dict[str, Any], datetime]:
+    root = _strict_dict(
+        facts,
+        {
+            "schema_version", "contract", "executor", "lineage", "raw_wire",
+            "credentials", "authority_map", "mode", "review", "production",
+        },
+        "facts",
+    )
+    if root["schema_version"] != SCHEMA_VERSION or root["contract"] != FACTS_CONTRACT:
+        _blocked("facts_contract", "facts")
+    executor = _strict_string(root["executor"], "executor")
+    if len(executor) > 128:
+        _blocked("executor_invalid", "executor")
+    lineage = _strict_dict(root["lineage"], set(EXPECTED_LINEAGE), "lineage")
+    if lineage != EXPECTED_LINEAGE:
+        _blocked("lineage_mismatch", "lineage")
+    mode = _validate_mode(root["mode"])
+    raw_wire, capture_index, latest_capture = _validate_raw_wire(
+        root["raw_wire"], restricted_root, repo_root
+    )
+    credentials = _validate_credentials(root["credentials"], capture_index, mode["name"])
+    authority = _validate_authority(root["authority_map"])
+    production = _validate_production(root["production"])
+    manifest = {
+        "schema_version": SCHEMA_VERSION,
+        "contract": MANIFEST_CONTRACT,
+        "executor": executor,
+        "lineage": dict(EXPECTED_LINEAGE),
+        "raw_wire": raw_wire,
+        "credentials": credentials,
+        "authority_map": authority,
+        "mode": mode,
+        "production": production,
+        "secrets_committed": False,
+        "gates": {
+            "raw_wire_inbound": "PASS",
+            "raw_wire_outbound": "PASS",
+            "dual_credential": "PASS",
+            "authority_map": "PASS",
+            "mode_non_production": "PASS",
+            "reviewer_independent": "PASS",
+            "reviewer_decision": "PASS",
+            "secrets_committed_false": "PASS",
+            "production_changed_false": "PASS",
+        },
+    }
+    return manifest, latest_capture
+
+
+def _validate_review(value: Any, executor: str, manifest_sha: str, latest_capture: datetime) -> dict[str, Any]:
+    review = _strict_dict(
+        value,
+        {"reviewer", "independent_from_executor", "decision", "reviewed_manifest_sha256", "reviewed_at"},
+        "review",
+    )
+    reviewer = _strict_string(review["reviewer"], "review.reviewer")
+    if len(reviewer) > 128:
+        _blocked("reviewer_invalid", "review.reviewer")
+    normalize = lambda text: unicodedata.normalize("NFKC", text).strip().casefold()
+    if normalize(reviewer) == normalize(executor):
+        _blocked("reviewer_not_independent", "review.reviewer")
+    if _strict_bool(review["independent_from_executor"], "review.independent_from_executor") is not True:
+        _blocked("reviewer_not_independent", "review.independent_from_executor")
+    if review["decision"] != "PASS":
+        _blocked("review_pass_required", "review.decision")
+    if _require_hex(review["reviewed_manifest_sha256"], "review.reviewed_manifest_sha256") != manifest_sha:
+        _blocked("review_hash_mismatch", "review.reviewed_manifest_sha256")
+    reviewed_at = _parse_rfc3339(review["reviewed_at"], "review.reviewed_at")
+    if reviewed_at < latest_capture:
+        _blocked("review_timestamp_invalid", "review.reviewed_at")
+    return {
+        "reviewer": reviewer,
+        "independent_from_executor": True,
+        "decision": "PASS",
+        "reviewed_manifest_sha256": manifest_sha,
+        "reviewed_at": review["reviewed_at"],
+    }
+
+
+def _safe_external_output(path: Path, repo_root: Path, restricted_root: Path, field: str) -> Path:
+    output = _absolute(path, field)
+    if _is_within(output, repo_root) or _is_within(output, restricted_root):
+        _blocked("output_boundary", field)
+    parent = output.parent
+    try:
+        metadata = os.lstat(parent)
+    except OSError:
+        _blocked("output_parent_unavailable", field)
+    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+        _blocked("output_parent_invalid", field)
+    return output
+
+
+def _atomic_replace(path: Path, data: bytes) -> None:
+    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    temporary_path = Path(temporary)
+    try:
+        os.fchmod(descriptor, 0o600)
+        with os.fdopen(descriptor, "wb", closefd=True) as stream:
+            stream.write(data)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary_path, path)
+    except Exception:
+        try:
+            os.close(descriptor)
+        except OSError:
+            pass
+        temporary_path.unlink(missing_ok=True)
+        raise
+
+
+def _prepare_review(args: argparse.Namespace) -> int:
+    repo_root = _validate_root(Path(args.repo_root), "repo_root")
+    restricted_root = _validate_root(Path(args.restricted_root), "restricted_root")
+    if _is_within(restricted_root, repo_root) or _is_within(repo_root, restricted_root):
+        _blocked("root_overlap", "restricted_root")
+    facts = _load_json(Path(args.facts), MAX_FACTS_BYTES, "facts")
+    manifest, _ = _build_manifest(facts, restricted_root, repo_root)
+    payload_bytes = _canonical_bytes(manifest)
+    digest = hashlib.sha256(payload_bytes).hexdigest()
+    payload_out = _safe_external_output(
+        Path(args.review_payload_out), repo_root, restricted_root, "review_payload_out"
+    )
+    hash_out = _safe_external_output(
+        Path(args.review_hash_out), repo_root, restricted_root, "review_hash_out"
+    )
+    if payload_out == hash_out:
+        _blocked("output_collision", "review_hash_out")
+    _atomic_replace(payload_out, payload_bytes)
+    _atomic_replace(hash_out, (digest + "\n").encode("ascii"))
+    return _emit_blocked(Blocked("review_required", "review"))
+
+
+def _read_external_review_payload(
+    path: Path,
+    repo_root: Path,
+    restricted_root: Path,
+) -> bytes:
+    review_path = _safe_external_output(path, repo_root, restricted_root, "review_payload")
+    try:
+        metadata = os.lstat(review_path)
+    except OSError:
+        _blocked("review_payload_unavailable", "review_payload")
+    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
+        _blocked("review_payload_invalid", "review_payload")
+    return _read_bounded(review_path, MAX_REVIEW_BYTES, "review_payload")
+
+
+def _ensure_directory_chain(root: Path, parent: Path, field: str) -> None:
+    try:
+        relative = parent.relative_to(root)
+    except ValueError:
+        _blocked("output_boundary", field)
+    current = root
+    for part in relative.parts:
+        current = current / part
+        try:
+            metadata = os.lstat(current)
+        except FileNotFoundError:
+            try:
+                os.mkdir(current, 0o775)
+                metadata = os.lstat(current)
+            except OSError:
+                _blocked("manifest_parent_unavailable", field)
+        except OSError:
+            _blocked("manifest_parent_unavailable", field)
+        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+            _blocked("manifest_parent_symlink", field)
+    try:
+        if parent.resolve(strict=True) != parent:
+            _blocked("manifest_parent_symlink", field)
+    except OSError:
+        _blocked("manifest_parent_unavailable", field)
+
+
+def _atomic_create_or_match(path: Path, data: bytes, repo_root: Path) -> None:
+    _ensure_directory_chain(repo_root, path.parent, "manifest_out")
+    if os.path.lexists(path):
+        try:
+            metadata = os.lstat(path)
+        except OSError:
+            _blocked("manifest_unavailable", "manifest_out")
+        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
+            _blocked("manifest_symlink", "manifest_out")
+        if _read_bounded(path, MAX_REVIEW_BYTES, "manifest_out") == data:
+            return
+        _blocked("manifest_conflict", "manifest_out")
+    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    temporary_path = Path(temporary)
+    try:
+        os.fchmod(descriptor, 0o644)
+        with os.fdopen(descriptor, "wb", closefd=True) as stream:
+            stream.write(data)
+            stream.flush()
+            os.fsync(stream.fileno())
+        try:
+            os.link(temporary_path, path)
+        except FileExistsError:
+            if _read_bounded(path, MAX_REVIEW_BYTES, "manifest_out") != data:
+                _blocked("manifest_conflict", "manifest_out")
+    finally:
+        try:
+            os.close(descriptor)
+        except OSError:
+            pass
+        temporary_path.unlink(missing_ok=True)
+
+
+def _finalize(args: argparse.Namespace) -> int:
+    repo_root = _validate_root(Path(args.repo_root), "repo_root")
+    restricted_root = _validate_root(Path(args.restricted_root), "restricted_root")
+    if _is_within(restricted_root, repo_root) or _is_within(repo_root, restricted_root):
+        _blocked("root_overlap", "restricted_root")
+    manifest_out = _absolute(Path(args.manifest_out), "manifest_out")
+    expected_out = repo_root / FINAL_RELATIVE_PATH
+    if manifest_out != expected_out:
+        _blocked("manifest_path_forbidden", "manifest_out")
+    facts = _load_json(Path(args.facts), MAX_FACTS_BYTES, "facts")
+    manifest, latest_capture = _build_manifest(facts, restricted_root, repo_root)
+    manifest_bytes = _canonical_bytes(manifest)
+    reviewed_bytes = _read_external_review_payload(
+        Path(args.review_payload), repo_root, restricted_root
+    )
+    if reviewed_bytes != manifest_bytes:
+        _blocked("review_payload_stale", "review_payload")
+    manifest_sha = hashlib.sha256(manifest_bytes).hexdigest()
+    review = _validate_review(facts["review"], manifest["executor"], manifest_sha, latest_capture)
+    envelope = {
+        "schema_version": SCHEMA_VERSION,
+        "contract": ENVELOPE_CONTRACT,
+        "status": "PASS",
+        "manifest": manifest,
+        "review": review,
+    }
+    _atomic_create_or_match(manifest_out, _canonical_bytes(envelope), repo_root)
+    print("PASS")
+    return 0
+
+
+def _parser() -> SafeParser:
+    parser = SafeParser(description="Verify Personal-Weixin B3 owner evidence without exposing raw data.")
+    subparsers = parser.add_subparsers(dest="command", required=True, parser_class=SafeParser)
+    prepare = subparsers.add_parser("prepare-review", help="build the canonical reviewer payload")
+    prepare.add_argument("--facts", required=True)
+    prepare.add_argument("--restricted-root", required=True)
+    prepare.add_argument("--repo-root", required=True)
+    prepare.add_argument("--review-payload-out", required=True)
+    prepare.add_argument("--review-hash-out", required=True)
+    prepare.set_defaults(handler=_prepare_review)
+    finalize = subparsers.add_parser("finalize", help="verify review and emit the R10 evidence manifest")
+    finalize.add_argument("--facts", required=True)
+    finalize.add_argument("--restricted-root", required=True)
+    finalize.add_argument("--repo-root", required=True)
+    finalize.add_argument("--review-payload", required=True)
+    finalize.add_argument("--manifest-out", required=True)
+    finalize.set_defaults(handler=_finalize)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    try:
+        args = _parser().parse_args(argv)
+        return int(args.handler(args))
+    except Blocked as error:
+        return _emit_blocked(error)
+    except Exception:
+        return _emit_blocked(Blocked("internal_blocked", "internal"))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())