fix(codex): harden runner provenance and secret-ref boundary

- Add _reject_symlink_chain() protection against ancestor symlink escapes
- Implement dirfd-based atomic writes with O_DIRECTORY | O_NOFOLLOW
- Add mandatory --trust-policy parameter with strict validation
- Integrate SSH signature verification via ssh-keygen -Y verify
- Authority field hard fail (present: false or missing -> BLOCKED)
- Self-audit rejection (reviewer != executor/capture/credential)
- Update template with trust_policy_key field
- Update README with trust policy and dirfd documentation
- Update task file with B3 BLOCKED_EVIDENCE_REQUIRED status

Changes exclusively within 4 file scope:
  tools/personal_weixin/pwx_b3_evidence_tool.py
  tools/personal_weixin/pwx_b3_evidence_facts.template.json
  tools/personal_weixin/README-B3-EVIDENCE.md
  .agent/tasks/codex/TASK-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716.md

.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-CODEX-AIOS-SPARK-RUNNER-PROVISIONING-20260717T103500UTC8.md

@@ -0,0 +1,115 @@
+# TASK-CODEX-AIOS-SPARK-RUNNER-PROVISIONING-20260717T103500UTC8
+
+## Authorization and boundary
+
+Owner-led preparation task for Codex to implement an AIOS-side managed Spark isolated runner through GitOps. This task does not authorize direct host mutation, production deployment, main-branch push, release, tag creation, or bypass of owner/admin approval.
+
+Current protected state:
+
+```yaml
+baseline: V1R4C6
+release: paused
+production_release: NOT_STARTED
+repair3_base: 8d9c6da90fc8caa69b691d243c6fefb40b6b0575
+```
+
+## Required provisioning deliverables
+
+Codex must inspect repository conventions and prepare an append-only provisioning patch containing all of the following, using existing repository paths and patterns where available:
+
+1. A fail-closed provisioning or installation script.
+2. A fixed runner artifact reference by immutable digest or versioned repository path.
+3. SHA-256 or signature verification before installation.
+4. A managed systemd unit or approved container definition.
+5. A dedicated least-privilege run user and group.
+6. A dedicated work directory outside controller and production worktrees.
+7. A managed heartbeat file or endpoint updated only by the runner.
+8. A runtime `secret_ref`; no plaintext secret may enter Git, logs, task files, environment dumps, or RESULT files.
+9. Sandboxing and resource limits appropriate to repository conventions.
+10. Cleanup for temporary branches and worktrees created by runner acceptance tests.
+
+## Required configuration inputs
+
+The implementation must fail closed until owner/admin supplies or approves:
+
+```yaml
+runner_artifact_uri_or_repo_path: gitea://aios/aios.git#tools/aios_spark_runner
+runner_artifact_version_or_digest: generated_from_final_implementation_commit
+runner_artifact_sha256_or_signature: generated_sha256_of_deterministic_artifact
+runner_secret_ref: null
+runner_run_user: aios-spark-runner
+runner_run_group: aios-spark-runner
+runner_workdir: /var/lib/aios-spark-runner
+runner_heartbeat_path_or_endpoint: /run/aios-spark-runner/heartbeat.json
+service_name_or_container_name: aios-spark-runner.service
+```
+
+Placeholders must never be interpreted as runnable defaults.
+
+## Security requirements
+
+- No download from an unapproved or mutable URL.
+- No `curl | sh`, shell/tmux/nohup daemon, or interactive session used as a service.
+- No controller worktree mutation.
+- No direct main push.
+- No force push, reset, amend, rebase, or Gitea database mutation.
+- No secret disclosure.
+- Service must use the dedicated account, explicit working directory, restrictive umask, bounded resources, restart policy, and managed logs.
+- Runner must create isolated temporary Git worktrees and must not reuse the controller or production worktree.
+- Repository writes remain subject to existing GitOps branch and review controls.
+
+## Required validation
+
+Before any Repair 3 execution, owner/admin must merge and apply the provisioning patch through the approved GitOps mechanism, then Codex must verify:
+
+```yaml
+artifact_integrity: PASS
+managed_service_present: PASS
+managed_service_active: PASS
+runner_process_owned_by_expected_user: PASS
+heartbeat_age_under_5_minutes: PASS
+repo_read_8d9c6da: PASS
+temporary_worktree_create: PASS
+temporary_worktree_read_only_probe: PASS
+temporary_worktree_remove: PASS
+temporary_branch_remove: PASS
+controller_unchanged: PASS
+main_unchanged: PASS
+production_unchanged: PASS
+```
+
+Success output:
+
+```text
+RUNNER_RECOVERED=YES artifact=<immutable-ref> service=<name> user=<user> heartbeat=<timestamp> repo_read=PASS temp_worktree=PASS cleanup=PASS
+```
+
+Failure output:
+
+```text
+RUNNER_RECOVERED=NO status=BLOCKED_OWNER_RUNNER_PROVISIONING missing=<items> required_admin_action=<exact action>
+```
+
+## Repair 3 continuation gate
+
+Only after `RUNNER_RECOVERED=YES` may Codex create a new isolated Repair 3 worktree from `8d9c6da90fc8caa69b691d243c6fefb40b6b0575` and implement the separately approved four-path B3 security repair. All B3, B4, merge, tag, canary, and production gates remain independently mandatory.
+
+## GitOps implementation status
+
+```yaml
+runner_kind: AIOS isolated Python worker
+artifact_source: repository-owned deterministic build
+artifact_transport: local AIOS path only
+artifact_integrity: SHA-256 required before mutation
+managed_service: aios-spark-runner.service
+run_user: aios-spark-runner
+run_group: aios-spark-runner
+workdir: /var/lib/aios-spark-runner
+heartbeat: /run/aios-spark-runner/heartbeat.json
+runner_secret_ref: null
+runner_code: COMPLETE
+runner_install: NOT_STARTED
+runner_recovered: false
+blocker: BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE
+repair3_started: false
+```

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

@@ -0,0 +1,96 @@
+# 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
+```
+
+## Status
+
+**B3 status**: `BLOCKED_EVIDENCE_REQUIRED` (evidence collection not started, owner must prepare facts and produce restricted raw data)
+**B2 status**: `NOT_APPLICABLE` (reserved, not yet reached)
+**Production baseline**: V1R4C6 (the most recent approved production release)
+
+## 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
+   合同，并验证 trust policy（root 所有、mode 0600、无 symlink、allowed-signers sha256 匹配），
+   通过 `ssh-keygen -Y verify` 验证 capture 和 credential 的 SSH 签名。
+   生成 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，验证 trust policy，
+   通过 `ssh-keygen -Y verify` 验证 capture、credential 和 reviewer 的 SSH 签名，
+   要求当前 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`。

deploy/spark-runner/README.md

@@ -0,0 +1,92 @@
+# AIOS managed isolated Python runner
+
+This package provisions the AIOS "Spark runner": a dedicated, systemd-managed
+Python worker for isolated Git worktree execution. It is not Apache Spark.
+
+## Trust boundary
+
+The repository fixes the source to
+`gitea://aios/aios.git#tools/aios_spark_runner`. The artifact builder accepts
+only a full lowercase commit OID that resolves to an actual Git commit and
+emits a deterministic tarball, an exact `sha256:<digest>` binding, a signed
+provenance manifest, and a non-secret configuration. The manifest contains
+`source_commit`, `artifact_digest`, `sha256`, `signature`, and `created_at`.
+The signature is Ed25519 and must verify against an independently provisioned,
+root-owned trusted public key; neither config nor manifest text is a trust root. The
+installer reads only local files, rejects mutable URLs, verifies the actual
+artifact SHA-256 before any install-root mutation, and rejects symlinked or
+group/world-writable inputs.
+
+`runner_secret_ref` is a reference only. It must match
+`secret://runtime/aios-spark-runner/gateway/<version>` and be resolvable by
+the existing gateway secret broker. Production installation also requires an
+approved broker availability adapter; none is invented or shipped here.
+Secret values are never accepted in configuration, arguments, artifacts,
+logs, receipts, or Git. If the reference or broker adapter is absent or
+invalid, provisioning fails before installation with:
+
+`BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE`
+
+There is deliberately no runnable secret default, network-piped installer,
+interactive shell, terminal multiplexer, background shell, or unmanaged daemon fallback.
+
+## Build
+
+From a clean checkout of the approved implementation commit:
+
+```bash
+deploy/spark-runner/build-artifact.sh \
+  --commit <full-implementation-commit> \
+  --signing-key /owner/offline/runner-artifact-ed25519.pem \
+  --source-repo /approved/aios/worktree \
+  --output /tmp/aios-spark-runner-dist
+```
+
+Transfer is unnecessary when the output is built on the AIOS host. The
+provision command verifies the manifest binding and actual digest; mismatch is
+fatal.
+
+## Provision and rollback
+
+Owner/admin first fills only `runner_secret_ref` in the emitted configuration
+using an approved, scope-matching reference. Production provisioning requires
+UID 0 and root `/`; tests are restricted to explicit roots below `/tmp`.
+
+```bash
+deploy/spark-runner/provision.sh \
+  --artifact /approved/path/aios-spark-runner-<commit>.tar.gz \
+  --manifest /approved/path/aios-spark-runner-<commit>.manifest.json \
+  --trusted-public-key /etc/aios-spark-runner/trusted-artifact-key.pem \
+  --source-repo /data/aios-gitops/repo.git \
+  --config /approved/path/aios-spark-runner-<commit>.json
+```
+
+The transaction installs a real directory at
+`/opt/aios-spark-runner/current`, creates the dedicated account with
+systemd-sysusers, creates private state/runtime directories with
+systemd-tmpfiles, writes a rollback receipt, and enables the sandboxed service.
+Rollback uses only that authoritative receipt. The receipt records
+`install_transaction_id`, `previous_artifact`, `rollback_target`, and
+`rollback_verified`; failed installs restore automatically and explicit rollback is
+idempotent:
+
+```bash
+deploy/spark-runner/rollback.sh \
+  --receipt /var/lib/aios-spark-runner/install-receipt.json
+```
+
+## Acceptance
+
+`acceptance.sh` verifies the heartbeat schema, source commit, artifact digest,
+live PID, and freshness, then creates a unique temporary branch/worktree from
+the requested immutable commit. It checks a clean exact OID and removes both
+worktree and branch without touching main or controller.
+
+The heartbeat is atomically replaced at exactly
+`/run/aios-spark-runner/heartbeat.json`, mode 0600, with these fields only:
+`schema_version`, `service`, `pid`, `started_at`, `heartbeat_at`,
+`artifact_digest`, `source_commit`, and `status`.
+
+Until an owner-approved, gateway-scope-matching `secret://` reference exists,
+the correct operational state remains
+`RUNNER_RECOVERED=NO` and `RUNNER_INSTALL=NOT_STARTED`.

deploy/spark-runner/acceptance.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+umask 0077
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd -P)"
+exec python3 "$ROOT/tools/aios_spark_runner/runnerctl.py" acceptance "$@"

deploy/spark-runner/artifact-manifest.schema.json

@@ -0,0 +1,21 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://aios.internal/schemas/aios-spark-runner-artifact-v2.json",
+  "title": "AIOS Spark runner verified artifact provenance",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "source_commit",
+    "artifact_digest",
+    "sha256",
+    "signature",
+    "created_at"
+  ],
+  "properties": {
+    "source_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "artifact_digest": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"},
+    "sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "signature": {"type": "string", "pattern": "^ed25519:[A-Za-z0-9+/]+={0,2}$"},
+    "created_at": {"type": "string", "format": "date-time"}
+  }
+}

deploy/spark-runner/build-artifact.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+umask 0077
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd -P)"
+exec python3 "$ROOT/tools/aios_spark_runner/runnerctl.py" build-artifact "$@"

deploy/spark-runner/provision.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+umask 0077
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd -P)"
+exec python3 "$ROOT/tools/aios_spark_runner/runnerctl.py" provision "$@"

deploy/spark-runner/rollback.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+umask 0077
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd -P)"
+exec python3 "$ROOT/tools/aios_spark_runner/runnerctl.py" rollback "$@"

deploy/spark-runner/runner-config.schema.json

@@ -0,0 +1,50 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://aios.internal/schemas/aios-spark-runner-v1.json",
+  "title": "AIOS managed isolated Python runner configuration",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "runner_artifact_uri_or_repo_path",
+    "runner_artifact_version_or_digest",
+    "runner_artifact_sha256_or_signature",
+    "runner_secret_ref",
+    "runner_run_user",
+    "runner_run_group",
+    "runner_workdir",
+    "runner_heartbeat_path_or_endpoint",
+    "service_name_or_container_name"
+  ],
+  "properties": {
+    "runner_artifact_uri_or_repo_path": {
+      "const": "gitea://aios/aios.git#tools/aios_spark_runner"
+    },
+    "runner_artifact_version_or_digest": {
+      "type": "string",
+      "pattern": "^[0-9a-f]{40}$"
+    },
+    "runner_artifact_sha256_or_signature": {
+      "type": "string",
+      "pattern": "^sha256:[0-9a-f]{64}$"
+    },
+    "runner_secret_ref": {
+      "type": ["string", "null"],
+      "pattern": "^secret://runtime/aios-spark-runner/gateway/[A-Za-z0-9][A-Za-z0-9._-]*$"
+    },
+    "runner_run_user": {
+      "const": "aios-spark-runner"
+    },
+    "runner_run_group": {
+      "const": "aios-spark-runner"
+    },
+    "runner_workdir": {
+      "const": "/var/lib/aios-spark-runner"
+    },
+    "runner_heartbeat_path_or_endpoint": {
+      "const": "/run/aios-spark-runner/heartbeat.json"
+    },
+    "service_name_or_container_name": {
+      "const": "aios-spark-runner.service"
+    }
+  }
+}

deploy/spark-runner/runner-config.template.json

@@ -0,0 +1,11 @@
+{
+  "runner_artifact_uri_or_repo_path": "gitea://aios/aios.git#tools/aios_spark_runner",
+  "runner_artifact_version_or_digest": null,
+  "runner_artifact_sha256_or_signature": null,
+  "runner_secret_ref": null,
+  "runner_run_user": "aios-spark-runner",
+  "runner_run_group": "aios-spark-runner",
+  "runner_workdir": "/var/lib/aios-spark-runner",
+  "runner_heartbeat_path_or_endpoint": "/run/aios-spark-runner/heartbeat.json",
+  "service_name_or_container_name": "aios-spark-runner.service"
+}

deploy/spark-runner/uninstall.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+umask 0077
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd -P)"
+exec python3 "$ROOT/tools/aios_spark_runner/runnerctl.py" uninstall "$@"

deploy/systemd/aios-spark-runner.service

@@ -0,0 +1,63 @@
+[Unit]
+Description=AIOS managed isolated Python runner
+After=network-online.target
+Wants=network-online.target
+ConditionFileNotEmpty=/etc/aios-spark-runner/runner.json
+StartLimitIntervalSec=60
+StartLimitBurst=5
+
+[Service]
+Type=simple
+User=aios-spark-runner
+Group=aios-spark-runner
+WorkingDirectory=/var/lib/aios-spark-runner
+UMask=0077
+ExecStart=/opt/aios-spark-runner/current/bin/aios-spark-runner run --config /etc/aios-spark-runner/runner.json
+Restart=on-failure
+RestartSec=5s
+TimeoutStartSec=30s
+TimeoutStopSec=30s
+KillSignal=SIGTERM
+SyslogIdentifier=aios-spark-runner
+StandardOutput=journal
+StandardError=journal
+
+NoNewPrivileges=true
+MemoryDenyWriteExecute=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectKernelLogs=true
+ProtectControlGroups=true
+ProtectClock=true
+ProtectHostname=true
+ProtectProc=invisible
+ProcSubset=pid
+RestrictSUIDSGID=true
+LockPersonality=true
+RestrictRealtime=true
+RestrictNamespaces=true
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+CapabilityBoundingSet=
+AmbientCapabilities=
+RemoveIPC=true
+RuntimeDirectory=aios-spark-runner
+RuntimeDirectoryMode=0700
+StateDirectory=aios-spark-runner
+StateDirectoryMode=0700
+ReadWritePaths=/var/lib/aios-spark-runner /run/aios-spark-runner
+InaccessiblePaths=/data/aios-gitops/worktrees/controller
+ReadOnlyPaths=/data/aios-gitops/repo.git
+InaccessiblePaths=-/data/aios-gitops/worktrees/production
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+MemoryMax=512M
+TasksMax=128
+LimitNOFILE=4096
+
+[Install]
+WantedBy=multi-user.target

deploy/sysusers.d/aios-spark-runner.conf

@@ -0,0 +1,2 @@
+g aios-spark-runner -
+u aios-spark-runner - "AIOS isolated runner" /var/lib/aios-spark-runner /usr/sbin/nologin

deploy/tmpfiles.d/aios-spark-runner.conf

@@ -0,0 +1,3 @@
+d /var/lib/aios-spark-runner 0700 aios-spark-runner aios-spark-runner - -
+d /run/aios-spark-runner 0700 aios-spark-runner aios-spark-runner - -
+d /etc/aios-spark-runner 0750 root aios-spark-runner - -

docs/superpowers/plans/2026-07-17-aios-spark-runner-provisioning.md

@@ -0,0 +1,118 @@
+# AIOS Managed Spark Runner Provisioning Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox syntax for tracking.
+
+**Goal:** Add a fail-closed, auditable Python 3 isolated AIOS runner with deterministic packaging, hardened systemd deployment, atomic heartbeat, acceptance, rollback, and full negative tests.
+
+**Architecture:** Keep runtime behavior in `tools/aios_spark_runner`; keep deployment policy under `deploy/spark-runner`, `deploy/systemd`, `deploy/sysusers.d`, and `deploy/tmpfiles.d`. A deterministic build produces a tar artifact plus manifest and SHA-256; provisioning validates all bindings before atomic installation. Missing `runner_secret_ref` permits build and offline tests but blocks host installation and service start.
+
+**Tech Stack:** Python 3.11 standard library, Bash, systemd, pytest, JSON Schema-compatible validation, Git worktrees, SHA-256, Gitleaks.
+
+## Global Constraints
+
+- Base commit is `54b6c1efe4f61bf73ae655d53aba01726b834249`.
+- Never use the accident worktree, controller worktree, or production worktree.
+- Never run `git reset`, `git commit --amend`, `git rebase`, force push, direct main push, or direct bare-ref mutation.
+- Runner identity is `aios-spark-runner:aios-spark-runner`; shell login and sudo are forbidden.
+- Work directory is `/var/lib/aios-spark-runner`; runtime directory is `/run/aios-spark-runner`; heartbeat is `/run/aios-spark-runner/heartbeat.json`.
+- Service is `aios-spark-runner.service`; logs use journald; controller and production worktree access is forbidden.
+- Artifact source URI is `gitea://aios/aios.git#tools/aios_spark_runner`.
+- Artifact source version is the complete implementation commit OID supplied to the build command.
+- Artifact SHA-256 is calculated from the final deterministic artifact and copied into the generated manifest/config.
+- `runner_secret_ref` has no default. Only a validated `secret://` reference is accepted; absence yields `BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE`.
+- Tests use only `/tmp` roots and never claim host recovery.
+
+---
+
+### Task 1: Executable contracts and RED
+
+**Files:**
+- Create: `tests/test_aios_spark_runner_provisioning.py`
+
+**Interfaces:**
+- Consumes: repository files at the clean base.
+- Produces: behavior contracts for config validation, artifact binding, safe paths, heartbeat, worktree probe, provisioning, rollback, and systemd policy.
+
+- [ ] Add tests that import `tools.aios_spark_runner.runnerctl` and assert missing implementation.
+- [ ] Add negative cases for empty fields, `REQUIRED`, mutable URI/tag, malformed OID/SHA, missing or malformed secret reference, symlinks, permissive modes, wrong owner metadata, digest mismatch, and signature mismatch.
+- [ ] Add heartbeat tests requiring atomic replace, schema version 1, actual PID matching, owner/mode checks, freshness, artifact digest, and source commit.
+- [ ] Add `/tmp` Git bare-repo tests for create/read-only/remove worktree and branch cleanup while controller/main snapshots remain unchanged.
+- [ ] Run `PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' python3 -m pytest tests/test_aios_spark_runner_provisioning.py -q`.
+- [ ] Record RED only when tests collect successfully and fail because runner deliverables are absent.
+
+### Task 2: Runtime and configuration
+
+**Files:**
+- Create: `tools/aios_spark_runner/__init__.py`
+- Create: `tools/aios_spark_runner/runnerctl.py`
+- Create: `deploy/spark-runner/runner-config.schema.json`
+- Create: `deploy/spark-runner/runner-config.template.json`
+
+**Interfaces:**
+- `load_config(path: Path, require_secret_ref: bool = True) -> RunnerConfig`
+- `validate_artifact(path: Path, expected_sha256: str) -> str`
+- `write_heartbeat(config: RunnerConfig, heartbeat_path: Path, pid: int, started_at: str) -> None`
+- `check_heartbeat(config: RunnerConfig, heartbeat_path: Path, max_age_seconds: int) -> dict[str, object]`
+- CLI commands: `validate-config`, `run`, `health`, and `worktree-probe`.
+
+- [ ] Implement strict allowlisted JSON parsing and reject unknown fields or non-canonical values.
+- [ ] Require absolute non-symlink paths and reject symlink ancestors before open/replace.
+- [ ] Implement parent-dirfd atomic heartbeat writes with mode `0600`, fsync, and inode/device recheck.
+- [ ] Keep logs value-free; never log config contents or resolved credentials.
+- [ ] Make `run` fail closed when `runner_secret_ref` is absent.
+
+### Task 3: Deterministic artifact and transactional provisioning
+
+**Files:**
+- Create: `deploy/spark-runner/build-artifact.sh`
+- Create: `deploy/spark-runner/provision.sh`
+- Create: `deploy/spark-runner/rollback.sh`
+- Create: `deploy/spark-runner/uninstall.sh`
+- Create: `deploy/spark-runner/acceptance.sh`
+
+**Interfaces:**
+- `build-artifact.sh --source-commit <40-hex> --output-dir <absolute-path>` emits one tar, manifest JSON, config JSON, and SHA-256 file.
+- `provision.sh --artifact <absolute-path> --config <absolute-path> [--root /]` validates first and installs atomically.
+- `rollback.sh --root <absolute-path> --receipt <absolute-path>` restores the recorded pre-install state.
+- `acceptance.sh --config <absolute-path> --repo <bare-repo> --commit <40-hex>` verifies service/heartbeat and temporary worktree cleanup.
+
+- [ ] Use `set -euo pipefail`, restrictive umask, canonical absolute paths, no network download, and no `curl | sh`.
+- [ ] Build deterministic tar members with fixed ordering, ownership, mode, and timestamp.
+- [ ] Validate SHA-256 and optional signature before any destination mutation.
+- [ ] Stage under the destination parent, fsync, atomically replace, and leave a rollback receipt.
+- [ ] On failure, remove staged content and preserve the previous installation.
+- [ ] Refuse controller/production paths and any group/world-writable or symlink configuration.
+
+### Task 4: Hardened systemd integration
+
+**Files:**
+- Create: `deploy/systemd/aios-spark-runner.service`
+- Create: `deploy/sysusers.d/aios-spark-runner.conf`
+- Create: `deploy/tmpfiles.d/aios-spark-runner.conf`
+
+**Interfaces:**
+- Unit starts `/opt/aios-spark-runner/current/bin/aios-spark-runner` as the dedicated user.
+- Config path is `/etc/aios-spark-runner/runner.json`; it contains only non-secret values and `secret://` references.
+- sysusers creates a system account with no login shell.
+- tmpfiles creates `/var/lib/aios-spark-runner`, `/run/aios-spark-runner`, and `/etc/aios-spark-runner` with restrictive ownership/modes.
+
+- [ ] Include all mandatory hardening directives from the approved task.
+- [ ] Permit writes only to state/runtime paths and deny controller/production paths.
+- [ ] Restrict address families and system calls to the Python/Git execution contract.
+- [ ] Set bounded memory, process, task, and file limits plus restart policy.
+- [ ] Verify with `systemd-analyze verify` and `systemd-analyze security --offline=yes`.
+
+### Task 5: Operations documentation and full verification
+
+**Files:**
+- Create: `deploy/spark-runner/README.md`
+- Modify: `.agent/tasks/codex/TASK-CODEX-AIOS-SPARK-RUNNER-PROVISIONING-20260717T103500UTC8.md`
+
+- [ ] Document build, offline validation, approved secret-provider handoff, installation, health, rollback, and uninstall without secret values.
+- [ ] Record the resolved eight non-secret fields and the blocked `runner_secret_ref` boundary.
+- [ ] Run Python compile, `bash -n`, JSON parsing/schema matrix, focused pytest, full pytest, systemd verify/security, diff check, and final full Gitleaks.
+- [ ] Treat collection errors, skips, timeouts, transport damage, warnings-as-errors failures, or unexecuted checks as non-PASS.
+- [ ] Build the final artifact from the implementation commit, record its digest, and push the append-only branch.
+- [ ] Require `WORKTREE_OID == BARE_OID == GITEA_OID`.
+- [ ] Obtain an independent RESULT-only review whose last commit adds exactly one result file.
+- [ ] If no approved `runner_secret_ref` exists, stop before installation with `OPENCLAW_GATE0=BLOCKED reason=GATEWAY_SECRET_REF_UNAVAILABLE`.

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
+}

patches/v2r1-personal-weixin-b3-repair3-owner-20260717T102830UTC8.json

@@ -0,0 +1,9 @@
+{
+  "id": "patch-1784255342812-558b2dc6",
+  "patch_name": "v2r1-personal-weixin-b3-repair3-owner-20260717T102830UTC8",
+  "branch": "patch/chatgpt/v2r1-personal-weixin-b3-repair3-owner-20260717T102830UTC8",
+  "risk": "high",
+  "description": "Owner-led Repair 3 implementation branch. ChatGPT defines code contract and verification gates; Codex is authorized to implement only the four approved B3 evidence paths in an isolated worktree. No production, B3 evidence, B4, merge, tag, canary, or release progression until independent acceptance.",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-17T02:29:02Z"
+}

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"
+  }
+}

tests/test_aios_spark_runner_provisioning.py

@@ -0,0 +1,781 @@
+from __future__ import annotations
+
+import importlib.util
+import importlib
+import json
+import os
+from pathlib import Path
+import shutil
+import stat
+import re
+import subprocess
+import sys
+import time
+from types import ModuleType
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE_PATH = ROOT / "tools/aios_spark_runner/runnerctl.py"
+DEPLOY = ROOT / "deploy/spark-runner"
+SOURCE_COMMIT = subprocess.check_output(
+    ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True
+).strip()
+SOURCE_URI = "gitea://aios/aios.git#tools/aios_spark_runner"
+
+
+def runnerctl() -> ModuleType:
+    assert MODULE_PATH.is_file(), "runner implementation is missing"
+    spec = importlib.util.spec_from_file_location("aios_spark_runner_runnerctl_test", MODULE_PATH)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def write_config(
+    path: Path,
+    *,
+    artifact_sha: str = "b" * 64,
+    secret_ref: str | None = "secret://runtime/aios-spark-runner/gateway/v1",
+    **changes: object,
+) -> dict[str, object]:
+    data: dict[str, object] = {
+        "runner_artifact_uri_or_repo_path": SOURCE_URI,
+        "runner_artifact_version_or_digest": SOURCE_COMMIT,
+        "runner_artifact_sha256_or_signature": f"sha256:{artifact_sha}",
+        "runner_secret_ref": secret_ref,
+        "runner_run_user": "aios-spark-runner",
+        "runner_run_group": "aios-spark-runner",
+        "runner_workdir": "/var/lib/aios-spark-runner",
+        "runner_heartbeat_path_or_endpoint": "/run/aios-spark-runner/heartbeat.json",
+        "service_name_or_container_name": "aios-spark-runner.service",
+    }
+    data.update(changes)
+    path.write_text(json.dumps(data), encoding="utf-8")
+    path.chmod(0o600)
+    return data
+
+
+def signing_keys(tmp_path: Path) -> tuple[Path, Path]:
+    private = tmp_path / "artifact-signing-key.pem"
+    public = tmp_path / "artifact-signing-key.pub.pem"
+    subprocess.run(
+        ["openssl", "genpkey", "-algorithm", "ED25519", "-out", str(private)],
+        check=True,
+    )
+    subprocess.run(
+        ["openssl", "pkey", "-in", str(private), "-pubout", "-out", str(public)],
+        check=True,
+    )
+    private.chmod(0o600)
+    public.chmod(0o600)
+    return private, public
+
+
+def build_artifact(
+    tmp_path: Path, *, commit: str = SOURCE_COMMIT
+) -> tuple[Path, Path, Path, Path]:
+    module = runnerctl()
+    tmp_path.mkdir(mode=0o700, parents=True, exist_ok=True)
+    output = tmp_path / "dist"
+    private, public = signing_keys(tmp_path)
+    result = module.build_artifact(
+        commit, output, signing_key=private, source_repo=ROOT
+    )
+    return (
+        Path(result["artifact"]),
+        Path(result["config"]),
+        Path(result["manifest"]),
+        public,
+    )
+
+
+def test_required_delivery_paths_exist() -> None:
+    expected = [
+        "tools/aios_spark_runner/__init__.py",
+        "tools/aios_spark_runner/runnerctl.py",
+        "tools/aios_spark_runner/broker.py",
+        "tools/aios_spark_runner/provenance.py",
+        "deploy/spark-runner/build-artifact.sh",
+        "deploy/spark-runner/provision.sh",
+        "deploy/spark-runner/rollback.sh",
+        "deploy/spark-runner/uninstall.sh",
+        "deploy/spark-runner/acceptance.sh",
+        "deploy/spark-runner/runner-config.schema.json",
+        "deploy/spark-runner/artifact-manifest.schema.json",
+        "deploy/spark-runner/runner-config.template.json",
+        "deploy/spark-runner/README.md",
+        "deploy/systemd/aios-spark-runner.service",
+        "deploy/sysusers.d/aios-spark-runner.conf",
+        "deploy/tmpfiles.d/aios-spark-runner.conf",
+    ]
+    missing = [path for path in expected if not (ROOT / path).is_file()]
+    assert not missing, missing
+
+
+def test_template_locks_nonsecret_defaults_and_blocks_secret() -> None:
+    path = DEPLOY / "runner-config.template.json"
+    assert path.is_file()
+    data = json.loads(path.read_text(encoding="utf-8"))
+    assert data == {
+        "runner_artifact_uri_or_repo_path": SOURCE_URI,
+        "runner_artifact_version_or_digest": None,
+        "runner_artifact_sha256_or_signature": None,
+        "runner_secret_ref": None,
+        "runner_run_user": "aios-spark-runner",
+        "runner_run_group": "aios-spark-runner",
+        "runner_workdir": "/var/lib/aios-spark-runner",
+        "runner_heartbeat_path_or_endpoint": "/run/aios-spark-runner/heartbeat.json",
+        "service_name_or_container_name": "aios-spark-runner.service",
+    }
+    assert "REQUIRED" not in path.read_text(encoding="utf-8")
+
+
+@pytest.mark.parametrize(
+    ("changes", "match"),
+    [
+        ({"runner_artifact_uri_or_repo_path": "https://example.invalid/latest"}, "artifact URI"),
+        ({"runner_artifact_version_or_digest": "latest"}, "source commit"),
+        ({"runner_artifact_version_or_digest": "A" * 40}, "source commit"),
+        ({"runner_artifact_sha256_or_signature": "sha256:1234"}, "SHA-256"),
+        ({"runner_artifact_sha256_or_signature": "signature:fixture"}, "signature"),
+        ({"runner_run_user": "root"}, "run user"),
+        ({"runner_run_group": "root"}, "run group"),
+        ({"runner_workdir": "relative"}, "workdir"),
+        ({"runner_heartbeat_path_or_endpoint": "/tmp/heartbeat"}, "heartbeat"),
+        ({"service_name_or_container_name": "runner.service"}, "service"),
+        ({"extra": "value"}, "unknown"),
+    ],
+)
+def test_config_rejects_mutable_placeholder_or_unknown_values(
+    tmp_path: Path, changes: dict[str, object], match: str
+) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    write_config(path, **changes)
+    with pytest.raises(module.ConfigError, match=match):
+        module.load_config(path, expected_uid=os.getuid())
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        None,
+        "",
+        "REQUIRED",
+        "secret://",
+        "env://RUNNER_TOKEN",
+        "secret://runtime/aios-spark-runner/gateway/../admin",
+        "secret://other/aios-spark-runner/gateway/v1",
+    ],
+)
+def test_missing_or_invalid_secret_ref_fails_closed(tmp_path: Path, value: str | None) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    write_config(path, secret_ref=value)
+    with pytest.raises(module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"):
+        module.load_config(path, expected_uid=os.getuid())
+
+def test_json_schema_locks_exact_gateway_secret_scope() -> None:
+    schema = json.loads(
+        (DEPLOY / "runner-config.schema.json").read_text(encoding="utf-8")
+    )
+    assert schema["additionalProperties"] is False
+    assert set(schema["required"]) == set(schema["properties"])
+    pattern = schema["properties"]["runner_secret_ref"]["pattern"]
+    assert re.fullmatch(
+        pattern, "secret://runtime/aios-spark-runner/gateway/v1"
+    )
+    for rejected in [
+        "secret://runtime/aios-spark-runner/gateway/../admin",
+        "secret://runtime/aios-spark-runner/gateway/",
+        "secret://other/aios-spark-runner/gateway/v1",
+    ]:
+        assert re.fullmatch(pattern, rejected) is None
+
+
+def test_offline_config_can_validate_without_resolving_secret(tmp_path: Path) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    write_config(path, secret_ref=None)
+    config = module.load_config(path, require_secret_ref=False, expected_uid=os.getuid())
+    assert config.runner_secret_ref is None
+    assert "secret" not in repr(config).lower()
+
+
+def test_inline_secret_fields_are_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    data = write_config(path)
+    data["token"] = "must-not-be-accepted"
+    path.write_text(json.dumps(data), encoding="utf-8")
+    path.chmod(0o600)
+    with pytest.raises(module.ConfigError, match="unknown"):
+        module.load_config(path, expected_uid=os.getuid())
+
+
+def test_config_leaf_symlink_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    real = tmp_path / "real.json"
+    link = tmp_path / "link.json"
+    write_config(real)
+    link.symlink_to(real)
+    with pytest.raises(module.ConfigError, match="symlink"):
+        module.load_config(link, expected_uid=os.getuid())
+
+
+def test_config_ancestor_symlink_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    real_dir = tmp_path / "real"
+    real_dir.mkdir()
+    link_dir = tmp_path / "link"
+    link_dir.symlink_to(real_dir, target_is_directory=True)
+    path = real_dir / "runner.json"
+    write_config(path)
+    with pytest.raises(module.ConfigError, match="symlink"):
+        module.load_config(link_dir / "runner.json", expected_uid=os.getuid())
+
+
+def test_permissive_config_mode_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    write_config(path)
+    path.chmod(0o666)
+    with pytest.raises(module.ConfigError, match="writable"):
+        module.load_config(path, expected_uid=os.getuid())
+
+
+def test_runtime_config_mode_is_exactly_enforced(tmp_path: Path) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    write_config(path)
+    path.chmod(0o644)
+    with pytest.raises(module.ConfigError, match="mode"):
+        module.load_config(path, expected_uid=os.getuid())
+
+
+def test_wrong_config_owner_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    path = tmp_path / "runner.json"
+    write_config(path)
+    with pytest.raises(module.ConfigError, match="owner"):
+        module.load_config(path, expected_uid=os.getuid() + 1)
+
+
+def test_artifact_hash_is_verified(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact = tmp_path / "artifact.tar.gz"
+    artifact.write_bytes(b"runner")
+    artifact.chmod(0o600)
+    digest = module.sha256_file(artifact)
+    assert module.validate_artifact(artifact, f"sha256:{digest}", expected_uid=os.getuid()) == digest
+
+
+def test_artifact_hash_mismatch_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact = tmp_path / "artifact.tar.gz"
+    artifact.write_bytes(b"runner")
+    artifact.chmod(0o600)
+    with pytest.raises(module.ConfigError, match="digest mismatch"):
+        module.validate_artifact(artifact, f"sha256:{'0' * 64}", expected_uid=os.getuid())
+
+
+def test_artifact_symlink_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact = tmp_path / "artifact.tar.gz"
+    link = tmp_path / "link.tar.gz"
+    artifact.write_bytes(b"runner")
+    artifact.chmod(0o600)
+    link.symlink_to(artifact)
+    with pytest.raises(module.ConfigError, match="symlink"):
+        module.validate_artifact(link, f"sha256:{'0' * 64}", expected_uid=os.getuid())
+
+def test_artifact_ancestor_symlink_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    real = tmp_path / "real"
+    real.mkdir()
+    artifact = real / "artifact.tar.gz"
+    artifact.write_bytes(b"runner")
+    artifact.chmod(0o600)
+    link = tmp_path / "linked"
+    link.symlink_to(real, target_is_directory=True)
+    with pytest.raises(module.ConfigError, match="symlink"):
+        module.validate_artifact(
+            link / artifact.name, f"sha256:{'0' * 64}", expected_uid=os.getuid()
+        )
+
+
+def test_heartbeat_output_ancestor_symlink_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    config_path = tmp_path / "runner.json"
+    write_config(config_path)
+    config = module.load_config(config_path, expected_uid=os.getuid())
+    real = tmp_path / "real-run"
+    real.mkdir(mode=0o700)
+    link = tmp_path / "linked-run"
+    link.symlink_to(real, target_is_directory=True)
+    with pytest.raises(module.ConfigError, match="symlink"):
+        module.write_heartbeat(config, link / "heartbeat.json", os.getpid(), module.utc_now())
+    assert not (real / "heartbeat.json").exists()
+
+
+def test_heartbeat_is_atomic_schema_bound_and_private(tmp_path: Path) -> None:
+    module = runnerctl()
+    config_path = tmp_path / "runner.json"
+    write_config(config_path)
+    config = module.load_config(config_path, expected_uid=os.getuid())
+    heartbeat = tmp_path / "run" / "heartbeat.json"
+    heartbeat.parent.mkdir(mode=0o700)
+    started = module.utc_now()
+    module.write_heartbeat(config, heartbeat, os.getpid(), started)
+    data = json.loads(heartbeat.read_text(encoding="utf-8"))
+    assert data == {
+        "schema_version": 1,
+        "service": "aios-spark-runner",
+        "pid": os.getpid(),
+        "started_at": started,
+        "heartbeat_at": data["heartbeat_at"],
+        "artifact_digest": "b" * 64,
+        "source_commit": SOURCE_COMMIT,
+        "status": "healthy",
+    }
+    assert stat.S_IMODE(heartbeat.stat().st_mode) == 0o600
+    assert not list(heartbeat.parent.glob(".heartbeat.json.*"))
+
+
+def test_runtime_config_write_fsyncs_file_and_directory(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    module = runnerctl()
+    config_path = tmp_path / "runner.json"
+    write_config(config_path)
+    config = module.load_config(config_path, expected_uid=os.getuid())
+    heartbeat = tmp_path / "run" / "heartbeat.json"
+    heartbeat.parent.mkdir(mode=0o700)
+    securefs = importlib.import_module("tools.aios_spark_runner.securefs")
+    original = securefs.os.fsync
+    calls: list[int] = []
+
+    def recording_fsync(fd: int) -> None:
+        calls.append(fd)
+        original(fd)
+
+    monkeypatch.setattr(securefs.os, "fsync", recording_fsync)
+    module.write_heartbeat(config, heartbeat, os.getpid(), module.utc_now())
+    assert len(calls) >= 2
+
+
+def test_heartbeat_freshness_and_pid_are_verified(tmp_path: Path) -> None:
+    module = runnerctl()
+    config_path = tmp_path / "runner.json"
+    write_config(config_path)
+    config = module.load_config(config_path, expected_uid=os.getuid())
+    heartbeat = tmp_path / "heartbeat.json"
+    module.write_heartbeat(config, heartbeat, os.getpid(), module.utc_now())
+    result = module.check_heartbeat(
+        config, heartbeat, max_age_seconds=30, expected_uid=os.getuid()
+    )
+    assert result["status"] == "healthy"
+    data = json.loads(heartbeat.read_text(encoding="utf-8"))
+    data["heartbeat_at"] = "2000-01-01T00:00:00Z"
+    heartbeat.write_text(json.dumps(data), encoding="utf-8")
+    heartbeat.chmod(0o600)
+    with pytest.raises(module.ConfigError, match="stale"):
+        module.check_heartbeat(config, heartbeat, max_age_seconds=30, expected_uid=os.getuid())
+
+
+def test_heartbeat_wrong_pid_or_digest_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    config_path = tmp_path / "runner.json"
+    write_config(config_path)
+    config = module.load_config(config_path, expected_uid=os.getuid())
+    heartbeat = tmp_path / "heartbeat.json"
+    module.write_heartbeat(config, heartbeat, os.getpid(), module.utc_now())
+    data = json.loads(heartbeat.read_text(encoding="utf-8"))
+    data["pid"] = 999999999
+    heartbeat.write_text(json.dumps(data), encoding="utf-8")
+    heartbeat.chmod(0o600)
+    with pytest.raises(module.ConfigError, match="PID"):
+        module.check_heartbeat(config, heartbeat, max_age_seconds=30, expected_uid=os.getuid())
+    data["pid"] = os.getpid()
+    data["artifact_digest"] = "0" * 64
+    heartbeat.write_text(json.dumps(data), encoding="utf-8")
+    heartbeat.chmod(0o600)
+    with pytest.raises(module.ConfigError, match="artifact digest"):
+        module.check_heartbeat(config, heartbeat, max_age_seconds=30, expected_uid=os.getuid())
+
+
+def test_worktree_probe_creates_reads_and_cleans_branch(tmp_path: Path) -> None:
+    module = runnerctl()
+    source = tmp_path / "source"
+    bare = tmp_path / "repo.git"
+    source.mkdir()
+    subprocess.run(["git", "init", "-q", "-b", "main"], cwd=source, check=True)
+    subprocess.run(["git", "config", "user.name", "Test"], cwd=source, check=True)
+    subprocess.run(["git", "config", "user.email", "test@example.invalid"], cwd=source, check=True)
+    (source / "README").write_text("fixture\n", encoding="utf-8")
+    subprocess.run(["git", "add", "README"], cwd=source, check=True)
+    subprocess.run(["git", "commit", "-qm", "fixture"], cwd=source, check=True)
+    commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=source, text=True).strip()
+    subprocess.run(["git", "clone", "-q", "--bare", str(source), str(bare)], check=True)
+    before = subprocess.check_output(
+        ["git", f"--git-dir={bare}", "rev-parse", "refs/heads/main"], text=True
+    ).strip()
+    result = module.probe_worktree(bare, commit, tmp_path / "probes")
+    after = subprocess.check_output(
+        ["git", f"--git-dir={bare}", "rev-parse", "refs/heads/main"], text=True
+    ).strip()
+    assert result == {"repo_read": "PASS", "temp_worktree": "PASS", "cleanup": "PASS"}
+    assert before == after
+    assert not (tmp_path / "probes").exists()
+    refs = subprocess.check_output(
+        ["git", f"--git-dir={bare}", "for-each-ref", "--format=%(refname)", "refs/heads/probe-"],
+        text=True,
+    )
+    assert refs == ""
+
+
+def test_build_artifact_is_deterministic_and_bound(tmp_path: Path) -> None:
+    module = runnerctl()
+    private, public = signing_keys(tmp_path)
+    first = module.build_artifact(
+        SOURCE_COMMIT, tmp_path / "one", signing_key=private, source_repo=ROOT
+    )
+    second = module.build_artifact(
+        SOURCE_COMMIT, tmp_path / "two", signing_key=private, source_repo=ROOT
+    )
+    assert Path(first["artifact"]).read_bytes() == Path(second["artifact"]).read_bytes()
+    assert first["sha256"] == second["sha256"]
+    checksum = Path(first["sha256_file"])
+    assert checksum.read_text(encoding="ascii") == (
+        f"{first['sha256']}  {Path(first['artifact']).name}\n"
+    )
+    assert stat.S_IMODE(checksum.stat().st_mode) == 0o600
+    assert Path(first["sha256_file"]).name.endswith(".tar.gz.sha256")
+    manifest = json.loads(Path(first["manifest"]).read_text(encoding="utf-8"))
+    assert manifest["source_commit"] == SOURCE_COMMIT
+    assert set(manifest) == {
+        "source_commit", "artifact_digest", "sha256", "signature", "created_at"
+    }
+    assert manifest["artifact_digest"] == f"sha256:{first['sha256']}"
+    assert manifest["sha256"] == first["sha256"]
+    verified = module.verify_manifest(
+        Path(first["artifact"]),
+        Path(first["manifest"]),
+        public,
+        ROOT,
+        expected_uid=os.getuid(),
+    )
+    assert verified == manifest
+    config = json.loads(Path(first["config"]).read_text(encoding="utf-8"))
+    assert config["runner_artifact_sha256_or_signature"] == f"sha256:{first['sha256']}"
+    assert config["runner_secret_ref"] is None
+
+
+def test_provision_missing_secret_fails_without_output(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact, config, manifest, public = build_artifact(tmp_path)
+    root = tmp_path / "root"
+    with pytest.raises(module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"):
+        module.provision(
+            artifact, manifest, public, ROOT, config, root,
+            test_mode=True, expected_uid=os.getuid()
+        )
+    assert not (root / "opt/aios-spark-runner/current").exists()
+
+def test_production_provision_requires_broker_availability_probe(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact, config_path, manifest, public = build_artifact(tmp_path)
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+    data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
+    config_path.write_text(json.dumps(data), encoding="utf-8")
+    config_path.chmod(0o600)
+    root = tmp_path / "must-not-exist"
+    with pytest.raises(
+        module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"
+    ):
+        module.provision(
+            artifact,
+            manifest,
+            public,
+            ROOT,
+            config_path,
+            root,
+            test_mode=False,
+            expected_uid=os.getuid(),
+        )
+    assert not root.exists()
+
+
+def test_gateway_secret_ref_probe_is_value_free_and_exact(tmp_path: Path) -> None:
+    module = runnerctl()
+    probe_path = tmp_path / "secret-ref-probe"
+    probe_path.write_text(
+        "#!/bin/sh\n"
+        "[ \"$1\" = check ] || exit 2\n"
+        "IFS= read -r ref\n"
+        "[ \"$ref\" = secret://runtime/aios-spark-runner/gateway/v1 ] || exit 3\n"
+        "printf 'AVAILABLE\\n'\n",
+        encoding="utf-8",
+    )
+    probe_path.chmod(0o700)
+    probe = module.make_secret_ref_probe(
+        probe_path, expected_uid=os.getuid(), timeout_seconds=2
+    )
+    assert probe("secret://runtime/aios-spark-runner/gateway/v1") is True
+    assert probe("secret://runtime/aios-spark-runner/gateway/v2") is False
+
+
+@pytest.mark.parametrize("unsafe", ["symlink", "writable", "relative"])
+def test_gateway_probe_executable_boundary_is_fail_closed(
+    tmp_path: Path, unsafe: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    module = runnerctl()
+    executable = tmp_path / "probe"
+    executable.write_text("#!/bin/sh\nprintf 'AVAILABLE\\n'\n", encoding="utf-8")
+    executable.chmod(0o700)
+    candidate = executable
+    if unsafe == "symlink":
+        candidate = tmp_path / "probe-link"
+        candidate.symlink_to(executable)
+    elif unsafe == "writable":
+        executable.chmod(0o722)
+    else:
+        monkeypatch.chdir(tmp_path)
+        candidate = Path("probe")
+    with pytest.raises(module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"):
+        module.make_secret_ref_probe(candidate, expected_uid=os.getuid())
+
+
+def test_forged_provenance_is_rejected(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact, _, manifest, public = build_artifact(tmp_path)
+    original = json.loads(manifest.read_text(encoding="utf-8"))
+    for field, value, match in [
+        ("sha256", "0" * 64, "SHA-256"),
+        ("artifact_digest", f"sha256:{'0' * 64}", "digest"),
+        ("signature", "ed25519:" + "A" * 88, "signature"),
+        ("source_commit", "f" * 40, "actual Git commit"),
+    ]:
+        forged = dict(original)
+        forged[field] = value
+        manifest.write_text(json.dumps(forged), encoding="utf-8")
+        manifest.chmod(0o600)
+        with pytest.raises(module.ConfigError, match=match):
+            module.verify_manifest(
+                artifact, manifest, public, ROOT, expected_uid=os.getuid()
+            )
+    manifest.write_text(json.dumps(original), encoding="utf-8")
+    manifest.chmod(0o600)
+
+
+def test_artifact_manifest_schema_requires_full_provenance() -> None:
+    schema = json.loads(
+        (DEPLOY / "artifact-manifest.schema.json").read_text(encoding="utf-8")
+    )
+    required = {
+        "source_commit", "artifact_digest", "sha256", "signature", "created_at"
+    }
+    assert set(schema["required"]) == required
+    assert set(schema["properties"]) == required
+    assert schema["additionalProperties"] is False
+
+def test_provision_preserves_existing_parent_permissions_and_normalizes_config(
+    tmp_path: Path,
+) -> None:
+    module = runnerctl()
+    artifact, config_path, manifest, public = build_artifact(tmp_path)
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+    data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
+    config_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    config_path.chmod(0o600)
+    root = tmp_path / "root"
+    root.mkdir(mode=0o700)
+    opt = root / "opt"
+    opt.mkdir(mode=0o755)
+    result = module.provision(
+        artifact, manifest, public, ROOT, config_path, root,
+        test_mode=True, expected_uid=os.getuid()
+    )
+    assert stat.S_IMODE(opt.stat().st_mode) == 0o755
+    installed = json.loads(
+        (root / "etc/aios-spark-runner/runner.json").read_text(encoding="utf-8")
+    )
+    assert set(installed) == set(data)
+    assert installed["runner_secret_ref"] == data["runner_secret_ref"]
+    module.rollback(root, Path(result["receipt"]), test_mode=True)
+
+
+def test_provision_is_idempotent_and_rollback_removes_first_install(tmp_path: Path) -> None:
+    module = runnerctl()
+    artifact, config_path, manifest, public = build_artifact(tmp_path)
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+    data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
+    config_path.write_text(json.dumps(data), encoding="utf-8")
+    config_path.chmod(0o600)
+    root = tmp_path / "root"
+    first = module.provision(
+        artifact, manifest, public, ROOT, config_path, root,
+        test_mode=True, expected_uid=os.getuid()
+    )
+    current = root / "opt/aios-spark-runner/current"
+    assert current.is_dir() and not current.is_symlink()
+    marker = current / "manifest.json"
+    before = marker.read_bytes()
+    second = module.provision(
+        artifact, manifest, public, ROOT, config_path, root,
+        test_mode=True, expected_uid=os.getuid()
+    )
+    assert marker.read_bytes() == before
+    assert first["artifact_sha256"] == second["artifact_sha256"]
+    receipt = Path(second["receipt"])
+    first_rollback = module.rollback(root, receipt, test_mode=True)
+    second_rollback = module.rollback(root, receipt, test_mode=True)
+    assert first_rollback["rollback_verified"] == "PASS"
+    assert second_rollback == first_rollback
+    assert not current.exists()
+    transaction = json.loads(receipt.read_text(encoding="utf-8"))
+    assert transaction["install_transaction_id"]
+    assert transaction["previous_artifact"] is None
+    assert transaction["rollback_target"] is None
+    assert transaction["rollback_verified"] is True
+    assert not list((root / "var/lib/aios-spark-runner").glob("rollback-*"))
+
+
+def test_failed_upgrade_automatically_restores_previous_artifact(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    module = runnerctl()
+    artifact, config_path, manifest, public = build_artifact(tmp_path / "build")
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+    data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
+    config_path.write_text(json.dumps(data), encoding="utf-8")
+    config_path.chmod(0o600)
+    root = tmp_path / "root"
+    first = module.provision(
+        artifact, manifest, public, ROOT, config_path, root,
+        test_mode=True, expected_uid=os.getuid()
+    )
+    receipt_before = Path(first["receipt"]).read_bytes()
+    marker = root / "opt/aios-spark-runner/current/manifest.json"
+    marker_data = json.loads(marker.read_text(encoding="utf-8"))
+    marker_data["artifact_digest"] = "0" * 64
+    marker.write_text(json.dumps(marker_data), encoding="utf-8")
+    marker.chmod(0o600)
+    installed_before = marker.read_bytes()
+    installer_globals = module.provision.__globals__
+    original_write = installer_globals["atomic_write"]
+    failed = False
+
+    def fail_once(path: Path, payload: bytes, *, mode: int = 0o600) -> None:
+        nonlocal failed
+        if not failed and str(path).endswith("etc/aios-spark-runner/runner.json"):
+            failed = True
+            raise RuntimeError("injected install failure")
+        original_write(path, payload, mode=mode)
+
+    monkeypatch.setitem(installer_globals, "atomic_write", fail_once)
+    with pytest.raises(RuntimeError, match="injected install failure"):
+        module.provision(
+            artifact, manifest, public, ROOT, config_path, root,
+            test_mode=True, expected_uid=os.getuid()
+        )
+    assert marker.read_bytes() == installed_before
+    assert Path(first["receipt"]).read_bytes() == receipt_before
+
+def test_systemd_unit_has_mandatory_sandbox_and_identity() -> None:
+    unit = ROOT / "deploy/systemd/aios-spark-runner.service"
+    assert unit.is_file()
+    text = unit.read_text(encoding="utf-8")
+    for directive in [
+        "User=aios-spark-runner",
+        "Group=aios-spark-runner",
+        "WorkingDirectory=/var/lib/aios-spark-runner",
+        "UMask=0077",
+        "NoNewPrivileges=true",
+        "PrivateTmp=true",
+        "ProtectSystem=strict",
+        "MemoryDenyWriteExecute=true",
+        "ProtectHome=true",
+        "ProtectKernelTunables=true",
+        "ProtectKernelModules=true",
+        "ProtectControlGroups=true",
+        "RestrictSUIDSGID=true",
+        "LockPersonality=true",
+        "RestrictRealtime=true",
+        "CapabilityBoundingSet=",
+        "AmbientCapabilities=",
+        "RuntimeDirectory=aios-spark-runner",
+        "SystemCallFilter=@system-service",
+        "ReadOnlyPaths=/data/aios-gitops/repo.git",
+        "StateDirectory=aios-spark-runner",
+        "ReadWritePaths=/var/lib/aios-spark-runner /run/aios-spark-runner",
+        "InaccessiblePaths=/data/aios-gitops/worktrees/controller",
+    ]:
+        assert directive in text
+    assert "User=root" not in text
+    assert "Environment=" not in text
+
+
+def test_sysusers_tmpfiles_and_unit_identity_are_consistent() -> None:
+    sysusers = (ROOT / "deploy/sysusers.d/aios-spark-runner.conf").read_text(encoding="utf-8")
+    tmpfiles = (ROOT / "deploy/tmpfiles.d/aios-spark-runner.conf").read_text(encoding="utf-8")
+    assert "aios-spark-runner" in sysusers
+    assert "/usr/sbin/nologin" in sysusers
+    assert "d /var/lib/aios-spark-runner 0700 aios-spark-runner aios-spark-runner" in tmpfiles
+    assert "d /run/aios-spark-runner 0700 aios-spark-runner aios-spark-runner" in tmpfiles
+    assert "d /etc/aios-spark-runner 0750 root aios-spark-runner" in tmpfiles
+
+
+def test_shell_scripts_are_fail_closed_and_do_not_download_or_daemonize() -> None:
+    for name in ["build-artifact.sh", "provision.sh", "rollback.sh", "uninstall.sh", "acceptance.sh"]:
+        path = DEPLOY / name
+        assert path.is_file()
+        text = path.read_text(encoding="utf-8")
+        assert "set -euo pipefail" in text
+        assert "curl | sh" not in text
+        assert "nohup" not in text
+        assert "tmux" not in text
+        assert "git reset" not in text
+        assert "git rebase" not in text
+        assert "git commit --amend" not in text
+        assert "push main" not in text
+
+def test_build_artifact_shell_entrypoint_executes_real_cli(tmp_path: Path) -> None:
+    output = tmp_path / "shell-dist"
+    private, _ = signing_keys(tmp_path)
+    result = subprocess.run(
+        [
+            str(DEPLOY / "build-artifact.sh"),
+            "--commit",
+            SOURCE_COMMIT,
+            "--signing-key",
+            str(private),
+            "--source-repo",
+            str(ROOT),
+            "--output",
+            str(output),
+        ],
+        cwd=ROOT,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    assert result.returncode == 0, result.stderr
+    response = json.loads(result.stdout)
+    assert Path(response["artifact"]).is_file()
+    assert Path(response["sha256_file"]).is_file()
+
+
+def test_readme_keeps_install_blocked_without_secret_ref() -> None:
+    path = DEPLOY / "README.md"
+    assert path.is_file()
+    text = path.read_text(encoding="utf-8")
+    assert "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE" in text
+    assert "secret://" in text
+    assert "curl | sh" not in text
+    assert "tmux" not in text
+    assert "nohup" not in text

tools/aios_spark_runner/__init__.py

@@ -0,0 +1,3 @@
+"""Managed AIOS isolated Python runner provisioning package."""
+
+__all__ = []

tools/aios_spark_runner/artifact.py

@@ -0,0 +1,213 @@
+"""Deterministic, repository-owned runner artifact builder."""
+from __future__ import annotations
+
+import gzip
+import io
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tarfile
+
+try:
+    from tools.aios_spark_runner.config import (
+        HEARTBEAT,
+        RUN_GROUP,
+        RUN_USER,
+        SERVICE,
+        SOURCE_URI,
+        WORKDIR,
+    )
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.provenance import git_commit_created_at, sign_fields
+    from tools.aios_spark_runner.securefs import (
+        atomic_json,
+        atomic_write,
+        read_repository_bytes,
+        sha256_file,
+    )
+except ModuleNotFoundError:
+    from aios_spark_runner.config import (
+        HEARTBEAT,
+        RUN_GROUP,
+        RUN_USER,
+        SERVICE,
+        SOURCE_URI,
+        WORKDIR,
+    )
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.provenance import git_commit_created_at, sign_fields
+    from aios_spark_runner.securefs import (
+        atomic_json,
+        atomic_write,
+        read_repository_bytes,
+        sha256_file,
+    )
+
+
+_COMMIT = re.compile(r"^[0-9a-f]{40}$")
+_PACKAGE_FILES = [
+    "__init__.py",
+    "errors.py",
+    "securefs.py",
+    "broker.py",
+    "provenance.py",
+    "config.py",
+    "runtime.py",
+    "artifact.py",
+    "extract.py",
+    "installer.py",
+    "runnerctl.py",
+]
+_STATIC_FILES = {
+    "share/systemd/aios-spark-runner.service": "deploy/systemd/aios-spark-runner.service",
+    "share/sysusers.d/aios-spark-runner.conf": "deploy/sysusers.d/aios-spark-runner.conf",
+    "share/tmpfiles.d/aios-spark-runner.conf": "deploy/tmpfiles.d/aios-spark-runner.conf",
+    "share/runner-config.schema.json": "deploy/spark-runner/runner-config.schema.json",
+    "share/artifact-manifest.schema.json": "deploy/spark-runner/artifact-manifest.schema.json",
+}
+_WRAPPER = """#!/usr/bin/env python3
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "lib"))
+from aios_spark_runner.runnerctl import main
+raise SystemExit(main())
+"""
+
+
+def _archive(entries: dict[str, tuple[bytes, int]], destination: Path) -> None:
+    with destination.open("wb") as raw:
+        with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed:
+            with tarfile.open(
+                fileobj=compressed, mode="w", format=tarfile.GNU_FORMAT
+            ) as archive:
+                for name in sorted(entries):
+                    payload, mode = entries[name]
+                    info = tarfile.TarInfo(name)
+                    info.size = len(payload)
+                    info.mode = mode
+                    info.uid = 0
+                    info.gid = 0
+                    info.uname = ""
+                    info.gname = ""
+                    info.mtime = 0
+                    archive.addfile(info, io.BytesIO(payload))
+
+
+def _config(commit: str, digest: str) -> dict[str, object]:
+    return {
+        "runner_artifact_uri_or_repo_path": SOURCE_URI,
+        "runner_artifact_version_or_digest": commit,
+        "runner_artifact_sha256_or_signature": f"sha256:{digest}",
+        "runner_secret_ref": None,
+        "runner_run_user": RUN_USER,
+        "runner_run_group": RUN_GROUP,
+        "runner_workdir": WORKDIR,
+        "runner_heartbeat_path_or_endpoint": HEARTBEAT,
+        "service_name_or_container_name": SERVICE,
+    }
+
+
+def _git_blob(repo: Path, commit: str, relative: str) -> bytes:
+    repo = Path(repo)
+    command = ["git", "-C", str(repo)] if (repo / ".git").exists() else [
+        "git", f"--git-dir={repo}"
+    ]
+    try:
+        return subprocess.check_output(
+            [*command, "show", f"{commit}:{relative}"], stderr=subprocess.PIPE
+        )
+    except subprocess.CalledProcessError as exc:
+        raise ConfigError("artifact source is absent from source_commit") from exc
+
+
+def _verify_source_binding(root: Path, repo: Path, commit: str) -> None:
+    sources = [
+        *(f"tools/aios_spark_runner/{name}" for name in _PACKAGE_FILES),
+        *_STATIC_FILES.values(),
+    ]
+    for relative in sources:
+        actual = read_repository_bytes(root / relative, expected_uid=os.getuid())
+        if actual != _git_blob(repo, commit, relative):
+            raise ConfigError(
+                f"artifact source does not match source_commit: {relative}"
+            )
+
+
+def build_artifact(
+    commit: str,
+    output: Path,
+    *,
+    signing_key: Path,
+    source_repo: Path | None = None,
+) -> dict[str, str]:
+    if not _COMMIT.fullmatch(commit):
+        raise ConfigError("artifact source commit must be a full immutable OID")
+    output = Path(output)
+    if not output.is_absolute():
+        raise ConfigError("artifact output must be absolute")
+    output.mkdir(mode=0o700, parents=True, exist_ok=False)
+    root = Path(__file__).resolve().parents[2]
+    source_repo = root if source_repo is None else Path(source_repo)
+    created_at = git_commit_created_at(source_repo, commit)
+    _verify_source_binding(root, source_repo, commit)
+    package = root / "tools/aios_spark_runner"
+    entries: dict[str, tuple[bytes, int]] = {
+        "bin/aios-spark-runner": (_WRAPPER.encode("utf-8"), 0o755),
+        "manifest.json": (
+            (
+                json.dumps(
+                    {"schema_version": 1, "source_commit": commit},
+                    sort_keys=True,
+                    separators=(",", ":"),
+                )
+                + "\n"
+            ).encode("utf-8"),
+            0o600,
+        ),
+    }
+    for name in _PACKAGE_FILES:
+        source = package / name
+        if not source.is_file():
+            raise ConfigError(f"artifact source file is missing: {source}")
+        entries[f"lib/aios_spark_runner/{name}"] = (
+            read_repository_bytes(source, expected_uid=os.getuid()), 0o644
+        )
+    for target, source_name in _STATIC_FILES.items():
+        source = root / source_name
+        if not source.is_file():
+            raise ConfigError(f"artifact static file is missing: {source}")
+        entries[target] = (read_repository_bytes(source, expected_uid=os.getuid()), 0o644)
+
+    artifact = output / f"aios-spark-runner-{commit}.tar.gz"
+    _archive(entries, artifact)
+    artifact.chmod(0o600)
+    digest = sha256_file(artifact, expected_uid=None)
+    config = output / f"aios-spark-runner-{commit}.json"
+    manifest = output / f"aios-spark-runner-{commit}.manifest.json"
+    checksum = output / f"{artifact.name}.sha256"
+    atomic_json(config, _config(commit, digest))
+    provenance = {
+        "source_commit": commit,
+        "artifact_digest": f"sha256:{digest}",
+        "sha256": digest,
+        "created_at": created_at,
+    }
+    atomic_json(
+        manifest,
+        {
+            **provenance,
+            "signature": sign_fields(
+                provenance, Path(signing_key), expected_uid=os.getuid()
+            ),
+        },
+    )
+    atomic_write(checksum, f"{digest}  {artifact.name}\n".encode("ascii"))
+    return {
+        "artifact": str(artifact),
+        "config": str(config),
+        "manifest": str(manifest),
+        "sha256_file": str(checksum),
+        "sha256": digest,
+    }

tools/aios_spark_runner/broker.py

@@ -0,0 +1,93 @@
+"""Fail-closed gateway secret-reference availability adapter.
+
+This module never resolves or returns secret material. The owner-installed
+probe receives only the secret reference on stdin and must emit exactly
+``AVAILABLE`` followed by a newline on stdout.
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import stat
+import subprocess
+from typing import Callable
+
+try:
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.securefs import reject_symlink_chain
+except ModuleNotFoundError:
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.securefs import reject_symlink_chain
+
+DEFAULT_SECRET_REF_PROBE = Path("/usr/libexec/aios-spark-runner/secret-ref-probe")
+_BLOCKED = "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"
+_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
+_CLOEXEC = getattr(os, "O_CLOEXEC", 0)
+
+
+def _validated_probe(path: Path, *, expected_uid: int = 0) -> int:
+    path = Path(path)
+    if not path.is_absolute():
+        raise ConfigError(_BLOCKED)
+    reject_symlink_chain(path)
+    before = os.lstat(path)
+    if not stat.S_ISREG(before.st_mode):
+        raise ConfigError(_BLOCKED)
+    if before.st_uid != expected_uid:
+        raise ConfigError(_BLOCKED)
+    if before.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+        raise ConfigError(_BLOCKED)
+    if not before.st_mode & stat.S_IXUSR:
+        raise ConfigError(_BLOCKED)
+    fd = os.open(path, os.O_RDONLY | _NOFOLLOW | _CLOEXEC)
+    try:
+        after = os.fstat(fd)
+        if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
+            raise ConfigError(_BLOCKED)
+        if not stat.S_ISREG(after.st_mode) or after.st_uid != expected_uid:
+            raise ConfigError(_BLOCKED)
+        if after.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+            raise ConfigError(_BLOCKED)
+        if not after.st_mode & stat.S_IXUSR:
+            raise ConfigError(_BLOCKED)
+        return fd
+    except Exception:
+        os.close(fd)
+        raise
+
+
+def make_secret_ref_probe(
+    executable: Path = DEFAULT_SECRET_REF_PROBE,
+    *,
+    expected_uid: int = 0,
+    timeout_seconds: int = 10,
+) -> Callable[[str], bool]:
+    """Build a value-free availability probe for an approved secret reference."""
+    try:
+        executable_fd = _validated_probe(Path(executable), expected_uid=expected_uid)
+    except (ConfigError, OSError):
+        raise ConfigError(_BLOCKED) from None
+    if timeout_seconds < 1 or timeout_seconds > 30:
+        raise ConfigError(_BLOCKED)
+
+    def probe(secret_ref: str) -> bool:
+        if not isinstance(secret_ref, str) or not secret_ref.startswith("secret://"):
+            return False
+        try:
+            result = subprocess.run(
+                [f"/proc/self/fd/{executable_fd}", "check"],
+                input=secret_ref + "\n",
+                text=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                timeout=timeout_seconds,
+                check=False,
+                cwd="/",
+                env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
+                pass_fds=(executable_fd,),
+            )
+        except (OSError, subprocess.SubprocessError):
+            return False
+        return result.returncode == 0 and result.stdout == "AVAILABLE\n" and result.stderr == ""
+
+    return probe

tools/aios_spark_runner/config.py

@@ -0,0 +1,144 @@
+"""Strict non-secret runner configuration."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import hmac
+from pathlib import Path
+import re
+
+try:
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.securefs import read_private_json, sha256_file
+except ModuleNotFoundError:
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.securefs import read_private_json, sha256_file
+
+
+SOURCE_URI = "gitea://aios/aios.git#tools/aios_spark_runner"
+RUN_USER = "aios-spark-runner"
+RUN_GROUP = "aios-spark-runner"
+WORKDIR = "/var/lib/aios-spark-runner"
+HEARTBEAT = "/run/aios-spark-runner/heartbeat.json"
+SERVICE = "aios-spark-runner.service"
+_KEYS = {
+    "runner_artifact_uri_or_repo_path",
+    "runner_artifact_version_or_digest",
+    "runner_artifact_sha256_or_signature",
+    "runner_secret_ref",
+    "runner_run_user",
+    "runner_run_group",
+    "runner_workdir",
+    "runner_heartbeat_path_or_endpoint",
+    "service_name_or_container_name",
+}
+_COMMIT = re.compile(r"^[0-9a-f]{40}$")
+_SHA = re.compile(r"^sha256:([0-9a-f]{64})$")
+_SECRET_REF = re.compile(
+    r"^secret://runtime/aios-spark-runner/gateway/[A-Za-z0-9][A-Za-z0-9._-]*$"
+)
+
+
+@dataclass(frozen=True)
+class RunnerConfig:
+    runner_artifact_uri_or_repo_path: str
+    runner_artifact_version_or_digest: str
+    runner_artifact_sha256_or_signature: str
+    runner_secret_ref: str | None = field(repr=False)
+    runner_run_user: str
+    runner_run_group: str
+    runner_workdir: str
+    runner_heartbeat_path_or_endpoint: str
+    service_name_or_container_name: str
+
+    @property
+    def artifact_digest(self) -> str:
+        return self.runner_artifact_sha256_or_signature.removeprefix("sha256:")
+
+
+def _exact(value: object, expected: str, label: str) -> str:
+    if value != expected:
+        raise ConfigError(f"invalid {label}: expected approved value")
+    return expected
+
+
+def config_as_dict(config: RunnerConfig) -> dict[str, object]:
+    return {
+        "runner_artifact_uri_or_repo_path": config.runner_artifact_uri_or_repo_path,
+        "runner_artifact_version_or_digest": config.runner_artifact_version_or_digest,
+        "runner_artifact_sha256_or_signature": config.runner_artifact_sha256_or_signature,
+        "runner_secret_ref": config.runner_secret_ref,
+        "runner_run_user": config.runner_run_user,
+        "runner_run_group": config.runner_run_group,
+        "runner_workdir": config.runner_workdir,
+        "runner_heartbeat_path_or_endpoint": config.runner_heartbeat_path_or_endpoint,
+        "service_name_or_container_name": config.service_name_or_container_name,
+    }
+
+
+def load_config(
+    path: Path,
+    *,
+    require_secret_ref: bool = True,
+    expected_uid: int | None = 0,
+    expected_modes: tuple[int, ...] = (0o600, 0o640),
+) -> RunnerConfig:
+    data = read_private_json(
+        Path(path), expected_uid=expected_uid, expected_modes=expected_modes
+    )
+    unknown = set(data) - _KEYS
+    missing = _KEYS - set(data)
+    if unknown:
+        raise ConfigError(f"unknown configuration fields: {sorted(unknown)}")
+    if missing:
+        raise ConfigError(f"missing configuration fields: {sorted(missing)}")
+
+    uri = _exact(data["runner_artifact_uri_or_repo_path"], SOURCE_URI, "artifact URI")
+    commit = data["runner_artifact_version_or_digest"]
+    if not isinstance(commit, str) or not _COMMIT.fullmatch(commit):
+        raise ConfigError("invalid immutable source commit")
+    binding = data["runner_artifact_sha256_or_signature"]
+    if isinstance(binding, str) and binding.startswith("signature:"):
+        raise ConfigError("signature verification is not configured; SHA-256 is mandatory")
+    if not isinstance(binding, str) or not _SHA.fullmatch(binding):
+        raise ConfigError("invalid artifact SHA-256 binding")
+
+    secret_ref = data["runner_secret_ref"]
+    if require_secret_ref and (
+        not isinstance(secret_ref, str) or not _SECRET_REF.fullmatch(secret_ref)
+    ):
+        raise ConfigError("BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE")
+    if secret_ref is not None and (
+        not isinstance(secret_ref, str) or not _SECRET_REF.fullmatch(secret_ref)
+    ):
+        raise ConfigError("BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE")
+
+    return RunnerConfig(
+        runner_artifact_uri_or_repo_path=uri,
+        runner_artifact_version_or_digest=commit,
+        runner_artifact_sha256_or_signature=binding,
+        runner_secret_ref=secret_ref,
+        runner_run_user=_exact(data["runner_run_user"], RUN_USER, "run user"),
+        runner_run_group=_exact(data["runner_run_group"], RUN_GROUP, "run group"),
+        runner_workdir=_exact(data["runner_workdir"], WORKDIR, "workdir"),
+        runner_heartbeat_path_or_endpoint=_exact(
+            data["runner_heartbeat_path_or_endpoint"], HEARTBEAT, "heartbeat path"
+        ),
+        service_name_or_container_name=_exact(
+            data["service_name_or_container_name"], SERVICE, "service name"
+        ),
+    )
+
+
+def validate_artifact(
+    path: Path,
+    binding: str,
+    *,
+    expected_uid: int | None = 0,
+) -> str:
+    match = _SHA.fullmatch(binding)
+    if not match:
+        raise ConfigError("invalid artifact SHA-256 binding")
+    actual = sha256_file(Path(path), expected_uid=expected_uid)
+    if not hmac.compare_digest(actual, match.group(1)):
+        raise ConfigError("artifact digest mismatch")
+    return actual

tools/aios_spark_runner/errors.py

@@ -0,0 +1,5 @@
+"""Stable fail-closed errors for runner provisioning."""
+
+
+class ConfigError(RuntimeError):
+    """Configuration, integrity, or runtime validation failed."""

tools/aios_spark_runner/extract.py

@@ -0,0 +1,74 @@
+"""Validated archive extraction with no links or traversal."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import io
+import os
+from pathlib import Path, PurePosixPath
+import tarfile
+
+try:
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.securefs import (
+        atomic_write,
+        read_private_bytes,
+        reject_symlink_chain,
+    )
+except ModuleNotFoundError:
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.securefs import (
+        atomic_write,
+        read_private_bytes,
+        reject_symlink_chain,
+    )
+
+
+def extract_artifact(
+    artifact: Path,
+    destination: Path,
+    *,
+    expected_digest: str,
+    expected_uid: int | None,
+) -> None:
+    artifact = Path(artifact)
+    destination = Path(destination)
+    reject_symlink_chain(destination)
+    seen: set[str] = set()
+    artifact_bytes = read_private_bytes(artifact, expected_uid=expected_uid)
+    if not hmac.compare_digest(hashlib.sha256(artifact_bytes).hexdigest(), expected_digest):
+        raise ConfigError("artifact digest mismatch during extraction")
+
+    try:
+        archive = tarfile.open(fileobj=io.BytesIO(artifact_bytes), mode="r:gz")
+    except (tarfile.TarError, OSError) as exc:
+        raise ConfigError("runner artifact is not a valid tar.gz") from exc
+    with archive:
+        members = archive.getmembers()
+        for member in members:
+            pure = PurePosixPath(member.name)
+            if pure.is_absolute() or ".." in pure.parts or not pure.parts:
+                raise ConfigError("unsafe artifact member path")
+            if member.name in seen:
+                raise ConfigError("duplicate artifact member")
+            seen.add(member.name)
+            if not member.isfile():
+                raise ConfigError("artifact links and special files are forbidden")
+        for member in sorted(members, key=lambda item: item.name):
+            target = destination.joinpath(*PurePosixPath(member.name).parts)
+            target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            reject_symlink_chain(target.parent)
+            stream = archive.extractfile(member)
+            if stream is None:
+                raise ConfigError("artifact member could not be read")
+            payload = stream.read()
+            if len(payload) != member.size:
+                raise ConfigError("artifact member size mismatch")
+            mode = 0o755 if member.name == "bin/aios-spark-runner" else 0o600
+            atomic_write(target, payload, mode=mode)
+    for directory in sorted(
+        (item for item in destination.rglob("*") if item.is_dir()),
+        key=lambda item: len(item.parts),
+        reverse=True,
+    ):
+        os.chmod(directory, 0o700)

tools/aios_spark_runner/installer.py

@@ -0,0 +1,446 @@
+"""Transactional install, rollback, and uninstall operations."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import stat
+import time
+import uuid
+from typing import Callable
+
+try:
+    from tools.aios_spark_runner.config import (
+        config_as_dict,
+        load_config,
+        validate_artifact,
+    )
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.extract import extract_artifact
+    from tools.aios_spark_runner.provenance import verify_manifest
+    from tools.aios_spark_runner.broker import make_secret_ref_probe
+    from tools.aios_spark_runner.securefs import (
+        atomic_json,
+        atomic_write,
+        read_private_bytes,
+        read_private_json,
+        reject_symlink_chain,
+    )
+except ModuleNotFoundError:
+    from aios_spark_runner.config import config_as_dict, load_config, validate_artifact
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.extract import extract_artifact
+    from aios_spark_runner.provenance import verify_manifest
+    from aios_spark_runner.broker import make_secret_ref_probe
+    from aios_spark_runner.securefs import (
+        atomic_json,
+        atomic_write,
+        read_private_bytes,
+        read_private_json,
+        reject_symlink_chain,
+    )
+
+
+_TARGETS = {
+    "etc/aios-spark-runner/runner.json": ("config", 0o640),
+    "etc/systemd/system/aios-spark-runner.service": (
+        "share/systemd/aios-spark-runner.service",
+        0o644,
+    ),
+    "usr/lib/sysusers.d/aios-spark-runner.conf": (
+        "share/sysusers.d/aios-spark-runner.conf",
+        0o644,
+    ),
+    "usr/lib/tmpfiles.d/aios-spark-runner.conf": (
+        "share/tmpfiles.d/aios-spark-runner.conf",
+        0o644,
+    ),
+}
+
+
+def _validate_root(root: Path, test_mode: bool) -> Path:
+    root = Path(root)
+    if not root.is_absolute():
+        raise ConfigError("install root must be absolute")
+    if test_mode:
+        if len(root.parts) < 2 or root.parts[1] != "tmp":
+            raise ConfigError("test installation is restricted to /tmp")
+    elif root != Path("/") or os.geteuid() != 0:
+        raise ConfigError("production installation requires root and /")
+    if root.exists():
+        reject_symlink_chain(root)
+        if not root.is_dir():
+            raise ConfigError("install root must be a directory")
+    else:
+        reject_symlink_chain(root, leaf_may_be_missing=True)
+        root.mkdir(mode=0o700)
+    return root
+
+
+def _ensure(root: Path, relative: str, mode: int = 0o700) -> Path:
+    current = root
+    parts = Path(relative).parts
+    for index, part in enumerate(parts):
+        current = current / part
+        if current.exists():
+            reject_symlink_chain(current)
+            info = os.lstat(current)
+            if info.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+                raise ConfigError(f"writable install directory rejected: {current}")
+            if not current.is_dir():
+                raise ConfigError(f"install path is not a directory: {current}")
+        else:
+            current.mkdir(mode=mode if index == len(parts) - 1 else 0o755)
+    return current
+
+
+def _receipt_path(root: Path) -> Path:
+    return root / "var/lib/aios-spark-runner/install-receipt.json"
+
+
+def _marker(current: Path, expected_uid: int) -> dict[str, object] | None:
+    path = current / "manifest.json"
+    if not path.is_file():
+        return None
+    return read_private_json(path, expected_uid=expected_uid)
+
+
+def _run_host_actions() -> None:
+    subprocess.run(
+        ["systemd-sysusers", "/usr/lib/sysusers.d/aios-spark-runner.conf"],
+        check=True,
+    )
+    subprocess.run(
+        ["systemd-tmpfiles", "--create", "/usr/lib/tmpfiles.d/aios-spark-runner.conf"],
+        check=True,
+    )
+    shutil.chown(
+        "/etc/aios-spark-runner/runner.json",
+        user="root",
+        group="aios-spark-runner",
+    )
+    os.chmod("/etc/aios-spark-runner/runner.json", 0o640)
+    subprocess.run(["systemctl", "daemon-reload"], check=True)
+    subprocess.run(
+        ["systemctl", "enable", "--now", "aios-spark-runner.service"],
+        check=True,
+    )
+
+def _fsync_directory(path: Path) -> None:
+    reject_symlink_chain(path)
+    fd = os.open(
+        path,
+        os.O_RDONLY
+        | getattr(os, "O_DIRECTORY", 0)
+        | getattr(os, "O_NOFOLLOW", 0),
+    )
+    try:
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def _restore_target_backups(
+    root: Path,
+    backups: dict[str, str | None],
+    *,
+    expected_uid: int,
+) -> None:
+    for relative, backup_name in backups.items():
+        if not isinstance(relative, str):
+            raise ConfigError("rollback target is invalid")
+        relative_path = Path(relative)
+        if relative_path.is_absolute() or ".." in relative_path.parts:
+            raise ConfigError("rollback target is invalid")
+        if backup_name is not None and not isinstance(backup_name, str):
+            raise ConfigError("rollback backup entry is invalid")
+        target = root / relative_path
+        if target.exists():
+            reject_symlink_chain(target)
+            if not target.is_file():
+                raise ConfigError("rollback target is not a regular file")
+            target.unlink()
+        if isinstance(backup_name, str):
+            source = root / backup_name
+            payload = read_private_bytes(source, expected_uid=expected_uid)
+            mode = 0o640 if relative.endswith("runner.json") else 0o644
+            atomic_write(target, payload, mode=mode)
+
+
+def provision(
+    artifact: Path,
+    manifest: Path,
+    trusted_public_key: Path,
+    source_repo: Path,
+    config_path: Path,
+    root: Path,
+    *,
+    test_mode: bool = False,
+    expected_uid: int | None = 0,
+    secret_ref_probe: Callable[[str], bool] | None = None,
+) -> dict[str, str]:
+    config = load_config(
+        Path(config_path), require_secret_ref=True, expected_uid=expected_uid
+    )
+    digest = validate_artifact(
+        Path(artifact),
+        config.runner_artifact_sha256_or_signature,
+        expected_uid=expected_uid,
+    )
+    provenance = verify_manifest(
+        Path(artifact),
+        Path(manifest),
+        Path(trusted_public_key),
+        Path(source_repo),
+        expected_uid=expected_uid,
+    )
+    if provenance["source_commit"] != config.runner_artifact_version_or_digest:
+        raise ConfigError("artifact source_commit does not match configuration")
+    if provenance["sha256"] != digest:
+        raise ConfigError("artifact provenance digest does not match configuration")
+    if not test_mode:
+        if config.runner_secret_ref is None:
+            raise ConfigError("BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE")
+        if secret_ref_probe is None:
+            secret_ref_probe = make_secret_ref_probe()
+        try:
+            available = secret_ref_probe(config.runner_secret_ref)
+        except Exception:
+            raise ConfigError(
+                "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"
+            ) from None
+        if available is not True:
+            raise ConfigError("BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE")
+
+    root = _validate_root(Path(root), test_mode)
+    install_base = _ensure(root, "opt/aios-spark-runner")
+    state = _ensure(root, "var/lib/aios-spark-runner")
+    current = install_base / "current"
+    receipt_path = _receipt_path(root)
+    uid = os.getuid() if expected_uid is None else expected_uid
+
+    if current.exists():
+        reject_symlink_chain(current)
+        if not current.is_dir():
+            raise ConfigError("current installation must be a real directory")
+        marker = _marker(current, uid)
+        if marker and marker.get("artifact_digest") == digest:
+            if not receipt_path.is_file():
+                raise ConfigError("idempotent install receipt is missing")
+            return {"artifact_sha256": digest, "receipt": str(receipt_path)}
+
+    staging = install_base / f".staging-{uuid.uuid4().hex}"
+    transaction_id = uuid.uuid4().hex
+    backup = state / f"rollback-{transaction_id}"
+    staging.mkdir(mode=0o700)
+    backup.mkdir(mode=0o700)
+    previous_current: str | None = None
+    target_backups: dict[str, str | None] = {}
+    installed_current = False
+    previous_artifact = _marker(current, uid) if current.exists() else None
+    previous_receipt = (
+        read_private_bytes(receipt_path, expected_uid=uid)
+        if receipt_path.is_file()
+        else None
+    )
+    try:
+        extract_artifact(
+            Path(artifact),
+            staging,
+            expected_digest=digest,
+            expected_uid=expected_uid,
+        )
+        atomic_json(
+            staging / "manifest.json",
+            {
+                "schema_version": 1,
+                "source_commit": config.runner_artifact_version_or_digest,
+                "artifact_digest": digest,
+            },
+        )
+        if current.exists():
+            previous = backup / "current"
+            os.replace(current, previous)
+            previous_current = str(previous.relative_to(root))
+        os.replace(staging, current)
+        _fsync_directory(install_base)
+        installed_current = True
+
+        for relative, (source_name, mode) in _TARGETS.items():
+            target = root / relative
+            parent_mode = (
+                0o750 if relative == "etc/aios-spark-runner/runner.json" else 0o755
+            )
+            _ensure(root, str(Path(relative).parent), mode=parent_mode)
+            if target.exists():
+                reject_symlink_chain(target)
+                prior = backup / relative
+                prior.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+                atomic_write(prior, read_private_bytes(target, expected_uid=None))
+                target_backups[relative] = str(prior.relative_to(root))
+            else:
+                target_backups[relative] = None
+            if source_name == "config":
+                payload = (
+                    json.dumps(
+                        config_as_dict(config), sort_keys=True, separators=(",", ":")
+                    ) + "\n"
+                ).encode("utf-8")
+            else:
+                payload = read_private_bytes(current / source_name, expected_uid=uid)
+            atomic_write(target, payload, mode=mode)
+
+        receipt = {
+            "schema_version": 2,
+            "install_transaction_id": transaction_id,
+            "artifact_digest": digest,
+            "source_commit": config.runner_artifact_version_or_digest,
+            "previous_artifact": previous_artifact,
+            "rollback_target": previous_artifact,
+            "rollback_verified": False,
+            "previous_current": previous_current,
+            "target_backups": target_backups,
+            "backup_root": str(backup.relative_to(root)),
+        }
+        atomic_json(receipt_path, receipt)
+        if not test_mode:
+            _run_host_actions()
+        return {"artifact_sha256": digest, "receipt": str(receipt_path)}
+    except Exception:
+        if receipt_path.exists():
+            receipt_path.unlink()
+        if previous_receipt is not None:
+            atomic_write(receipt_path, previous_receipt, mode=0o600)
+        _restore_target_backups(root, target_backups, expected_uid=uid)
+        if installed_current and current.exists():
+            shutil.rmtree(current)
+        if previous_current:
+            previous = root / previous_current
+            if previous.exists():
+                os.replace(previous, current)
+        if staging.exists():
+            shutil.rmtree(staging)
+        if backup.exists():
+            shutil.rmtree(backup)
+        _fsync_directory(install_base)
+        raise
+
+
+def _rollback_target_matches(root: Path, target: object, expected_uid: int) -> bool:
+    current = root / "opt/aios-spark-runner/current"
+    if target is None:
+        return not current.exists()
+    if not isinstance(target, dict):
+        return False
+    marker = _marker(current, expected_uid) if current.is_dir() else None
+    return bool(
+        marker
+        and marker.get("source_commit") == target.get("source_commit")
+        and marker.get("artifact_digest") == target.get("artifact_digest")
+    )
+
+
+def rollback(root: Path, receipt: Path, *, test_mode: bool = False) -> dict[str, str]:
+    root = _validate_root(Path(root), test_mode)
+    expected = _receipt_path(root)
+    receipt = Path(receipt)
+    if receipt != expected:
+        raise ConfigError("rollback receipt path is not authoritative")
+    uid = os.getuid() if test_mode else 0
+    data = read_private_json(receipt, expected_uid=uid)
+    if data.get("schema_version") != 2:
+        raise ConfigError("rollback receipt schema mismatch")
+    transaction_id = data.get("install_transaction_id")
+    if not isinstance(transaction_id, str) or not transaction_id:
+        raise ConfigError("rollback transaction id is invalid")
+    target = data.get("rollback_target")
+    if data.get("rollback_verified") is True:
+        if not _rollback_target_matches(root, target, uid):
+            raise ConfigError("previously verified rollback target changed")
+        return {
+            "rollback": "PASS",
+            "rollback_verified": "PASS",
+            "install_transaction_id": transaction_id,
+        }
+    if not test_mode:
+        subprocess.run(
+            ["systemctl", "disable", "--now", "aios-spark-runner.service"],
+            check=True,
+        )
+    current = root / "opt/aios-spark-runner/current"
+    if current.exists():
+        reject_symlink_chain(current)
+        if current.is_symlink() or not current.is_dir():
+            raise ConfigError("unsafe current installation")
+        shutil.rmtree(current)
+    previous = data.get("previous_current")
+    if isinstance(previous, str):
+        source = root / previous
+        reject_symlink_chain(source)
+        os.replace(source, current)
+
+    backups = data.get("target_backups")
+    if not isinstance(backups, dict):
+        raise ConfigError("rollback receipt target map is invalid")
+    backup_name = data.get("backup_root")
+    backup_relative = Path(backup_name) if isinstance(backup_name, str) else None
+    if (
+        backup_relative is None
+        or backup_relative.is_absolute()
+        or ".." in backup_relative.parts
+    ):
+        raise ConfigError("rollback backup root is invalid")
+    _restore_target_backups(root, backups, expected_uid=uid)
+    backup_root = root / backup_relative
+    reject_symlink_chain(backup_root)
+    if not backup_root.is_dir():
+        raise ConfigError("rollback backup root is not a directory")
+    shutil.rmtree(backup_root)
+    _fsync_directory(root / "opt/aios-spark-runner")
+    if not _rollback_target_matches(root, target, uid):
+        raise ConfigError("rollback target verification failed")
+    data["rollback_verified"] = True
+    atomic_json(receipt, data)
+    if not test_mode:
+        subprocess.run(["systemctl", "daemon-reload"], check=True)
+        if target is not None:
+            subprocess.run(
+                ["systemctl", "enable", "--now", "aios-spark-runner.service"],
+                check=True,
+            )
+            heartbeat = root / "run/aios-spark-runner/heartbeat.json"
+            deadline = time.monotonic() + 30
+            heartbeat_data: dict[str, object] | None = None
+            while time.monotonic() < deadline:
+                try:
+                    candidate = read_private_json(heartbeat, expected_uid=None)
+                except (ConfigError, FileNotFoundError):
+                    candidate = None
+                if candidate and (
+                    candidate.get("source_commit") == target.get("source_commit")
+                    and candidate.get("artifact_digest") == target.get("artifact_digest")
+                ):
+                    heartbeat_data = candidate
+                    break
+                time.sleep(0.25)
+            if heartbeat_data is None:
+                raise ConfigError("rollback heartbeat version mismatch")
+    return {
+        "rollback": "PASS",
+        "rollback_verified": "PASS",
+        "install_transaction_id": transaction_id,
+    }
+
+
+def uninstall(root: Path, *, test_mode: bool = False) -> dict[str, str]:
+    root = _validate_root(Path(root), test_mode)
+    receipt = _receipt_path(root)
+    if receipt.is_file():
+        return rollback(root, receipt, test_mode=test_mode)
+    current = root / "opt/aios-spark-runner/current"
+    if current.exists():
+        reject_symlink_chain(current)
+        shutil.rmtree(current)
+    return {"uninstall": "PASS"}

tools/aios_spark_runner/provenance.py

@@ -0,0 +1,203 @@
+"""Cryptographically verified provenance for runner artifacts."""
+from __future__ import annotations
+
+import base64
+from datetime import datetime
+import hmac
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+
+try:
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.securefs import (
+        read_private_json,
+        reject_symlink_chain,
+        sha256_file,
+    )
+except ModuleNotFoundError:
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.securefs import (
+        read_private_json,
+        reject_symlink_chain,
+        sha256_file,
+    )
+
+
+_OID = re.compile(r"^[0-9a-f]{40}$")
+_SHA = re.compile(r"^[0-9a-f]{64}$")
+_SIGNATURE = re.compile(r"^ed25519:([A-Za-z0-9+/]+={0,2})$")
+_FIELDS = {
+    "source_commit",
+    "artifact_digest",
+    "sha256",
+    "signature",
+    "created_at",
+}
+_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
+
+
+def _git(repo: Path, *args: str) -> str:
+    repo = Path(repo)
+    reject_symlink_chain(repo)
+    command = ["git", "-C", str(repo)] if (repo / ".git").exists() else [
+        "git", f"--git-dir={repo}"
+    ]
+    try:
+        return subprocess.check_output(
+            [*command, *args], text=True, stderr=subprocess.PIPE
+        ).strip()
+    except subprocess.CalledProcessError as exc:
+        raise ConfigError("source_commit is not an actual Git commit") from exc
+
+
+def git_commit_created_at(repo: Path, commit: str) -> str:
+    if not _OID.fullmatch(commit):
+        raise ConfigError("source_commit must be a full Git OID")
+    _git(repo, "cat-file", "-e", f"{commit}^{{commit}}")
+    created_at = _git(repo, "show", "-s", "--format=%cI", commit)
+    try:
+        datetime.fromisoformat(created_at)
+    except ValueError as exc:
+        raise ConfigError("Git commit timestamp is invalid") from exc
+    return created_at
+
+
+def _payload(fields: dict[str, str]) -> bytes:
+    return (
+        json.dumps(fields, sort_keys=True, separators=(",", ":")) + "\n"
+    ).encode("utf-8")
+
+
+def _validated_key_fd(path: Path, *, expected_uid: int | None) -> int:
+    path = Path(path)
+    if not path.is_absolute():
+        raise ConfigError("provenance key path must be absolute")
+    reject_symlink_chain(path)
+    before = os.lstat(path)
+    if not stat.S_ISREG(before.st_mode):
+        raise ConfigError("provenance key must be a regular file")
+    if expected_uid is not None and before.st_uid != expected_uid:
+        raise ConfigError("provenance key owner mismatch")
+    if before.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+        raise ConfigError("provenance key mode is unsafe")
+    fd = os.open(path, os.O_RDONLY | _NOFOLLOW)
+    after = os.fstat(fd)
+    if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
+        os.close(fd)
+        raise ConfigError("provenance key changed during secure open")
+    return fd
+
+
+def _openssl(
+    payload: bytes,
+    key: Path,
+    *,
+    expected_uid: int | None,
+) -> bytes:
+    fd = _validated_key_fd(key, expected_uid=expected_uid)
+    if not hasattr(os, "memfd_create"):
+        os.close(fd)
+        raise ConfigError("anonymous provenance input unavailable")
+    payload_fd = os.memfd_create("aios-provenance-payload", flags=0)
+    os.write(payload_fd, payload)
+    os.lseek(payload_fd, 0, os.SEEK_SET)
+    command = [
+        "openssl", "pkeyutl", "-sign", "-rawin",
+        "-inkey", f"/proc/self/fd/{fd}",
+        "-in", f"/proc/self/fd/{payload_fd}",
+    ]
+    try:
+        result = subprocess.run(
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            pass_fds=(fd, payload_fd),
+            check=False,
+        )
+        if result.returncode != 0:
+            raise ConfigError("artifact signature verification failed")
+        return result.stdout
+    except OSError as exc:
+        raise ConfigError("OpenSSL provenance verification unavailable") from exc
+    finally:
+        os.close(fd)
+        os.close(payload_fd)
+
+
+def sign_fields(
+    fields: dict[str, str], signing_key: Path, *, expected_uid: int | None
+) -> str:
+    signature = _openssl(
+        _payload(fields), signing_key, expected_uid=expected_uid
+    )
+    return "ed25519:" + base64.b64encode(signature).decode("ascii")
+
+
+def verify_manifest(
+    artifact: Path,
+    manifest: Path,
+    trusted_public_key: Path,
+    source_repo: Path,
+    *,
+    expected_uid: int | None,
+) -> dict[str, str]:
+    data = read_private_json(
+        Path(manifest), expected_uid=expected_uid, expected_modes=(0o600,)
+    )
+    if set(data) != _FIELDS or not all(isinstance(data[key], str) for key in _FIELDS):
+        raise ConfigError("artifact manifest schema mismatch")
+    commit = data["source_commit"]
+    created_at = git_commit_created_at(Path(source_repo), commit)
+    if data["created_at"] != created_at:
+        raise ConfigError("artifact created_at does not match Git commit")
+    actual = sha256_file(Path(artifact), expected_uid=expected_uid)
+    if not _SHA.fullmatch(data["sha256"]):
+        raise ConfigError("artifact manifest SHA-256 is invalid")
+    if data["artifact_digest"] != f"sha256:{actual}":
+        raise ConfigError("artifact digest mismatch")
+    if not hmac.compare_digest(data["sha256"], actual):
+        raise ConfigError("artifact SHA-256 mismatch")
+    match = _SIGNATURE.fullmatch(data["signature"])
+    if not match:
+        raise ConfigError("artifact signature format is invalid")
+    try:
+        signature = base64.b64decode(match.group(1), validate=True)
+    except ValueError as exc:
+        raise ConfigError("artifact signature format is invalid") from exc
+    fields = {key: data[key] for key in _FIELDS - {"signature"}}
+    # OpenSSL verification uses temporary anonymous descriptors only; trusted
+    # key material never comes from the runner config or manifest.
+    key_fd = _validated_key_fd(trusted_public_key, expected_uid=expected_uid)
+    if not hasattr(os, "memfd_create"):
+        os.close(key_fd)
+        raise ConfigError("anonymous provenance input unavailable")
+    signature_fd = os.memfd_create("aios-provenance-signature", flags=0)
+    payload_fd = os.memfd_create("aios-provenance-payload", flags=0)
+    try:
+        os.write(signature_fd, signature)
+        os.lseek(signature_fd, 0, os.SEEK_SET)
+        os.write(payload_fd, _payload(fields))
+        os.lseek(payload_fd, 0, os.SEEK_SET)
+        result = subprocess.run(
+            [
+                "openssl", "pkeyutl", "-verify", "-rawin", "-pubin",
+                "-inkey", f"/proc/self/fd/{key_fd}",
+                "-sigfile", f"/proc/self/fd/{signature_fd}",
+                "-in", f"/proc/self/fd/{payload_fd}",
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            pass_fds=(key_fd, signature_fd, payload_fd),
+            check=False,
+        )
+        if result.returncode != 0:
+            raise ConfigError("artifact signature verification failed")
+    finally:
+        os.close(key_fd)
+        os.close(signature_fd)
+        os.close(payload_fd)
+    return data

tools/aios_spark_runner/runnerctl.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""CLI and public API for the managed AIOS isolated Python runner."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import pwd
+import sys
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+try:
+    from tools.aios_spark_runner.artifact import build_artifact
+    from tools.aios_spark_runner.config import RunnerConfig, load_config, validate_artifact
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.installer import provision, rollback, uninstall
+    from tools.aios_spark_runner.provenance import verify_manifest
+    from tools.aios_spark_runner.broker import make_secret_ref_probe
+    from tools.aios_spark_runner.runtime import (
+        check_heartbeat,
+        probe_worktree,
+        run_forever,
+        utc_now,
+        write_heartbeat,
+    )
+    from tools.aios_spark_runner.securefs import sha256_file
+except ModuleNotFoundError:
+    from aios_spark_runner.artifact import build_artifact
+    from aios_spark_runner.config import RunnerConfig, load_config, validate_artifact
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.installer import provision, rollback, uninstall
+    from aios_spark_runner.provenance import verify_manifest
+    from aios_spark_runner.broker import make_secret_ref_probe
+    from aios_spark_runner.runtime import (
+        check_heartbeat,
+        probe_worktree,
+        run_forever,
+        utc_now,
+        write_heartbeat,
+    )
+    from aios_spark_runner.securefs import sha256_file
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="aios-spark-runner")
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    build = commands.add_parser("build-artifact")
+    build.add_argument("--commit", required=True)
+    build.add_argument("--output", type=Path, required=True)
+    build.add_argument("--signing-key", type=Path, required=True)
+    build.add_argument("--source-repo", type=Path, default=Path.cwd())
+
+    install = commands.add_parser("provision")
+    install.add_argument("--artifact", type=Path, required=True)
+    install.add_argument("--manifest", type=Path, required=True)
+    install.add_argument("--trusted-public-key", type=Path, required=True)
+    install.add_argument("--source-repo", type=Path, required=True)
+    install.add_argument("--config", type=Path, required=True)
+    install.add_argument("--root", type=Path, default=Path("/"))
+    install.add_argument("--test-mode", action="store_true")
+
+    undo = commands.add_parser("rollback")
+    undo.add_argument("--root", type=Path, default=Path("/"))
+    undo.add_argument("--receipt", type=Path, required=True)
+    undo.add_argument("--test-mode", action="store_true")
+
+    remove = commands.add_parser("uninstall")
+    remove.add_argument("--root", type=Path, default=Path("/"))
+    remove.add_argument("--test-mode", action="store_true")
+
+    accept = commands.add_parser("acceptance")
+    accept.add_argument("--config", type=Path, required=True)
+    accept.add_argument("--bare-repo", type=Path, required=True)
+    accept.add_argument("--commit", required=True)
+    accept.add_argument("--probe-parent", type=Path, required=True)
+    accept.add_argument("--heartbeat", type=Path, required=True)
+    accept.add_argument("--max-age-seconds", type=int, default=300)
+
+    run = commands.add_parser("run")
+    run.add_argument("--config", type=Path, required=True)
+    run.add_argument("--interval-seconds", type=int, default=30)
+    return parser
+
+
+def _execute(args: argparse.Namespace) -> object:
+    if args.command == "build-artifact":
+        return build_artifact(
+            args.commit,
+            args.output,
+            signing_key=args.signing_key,
+            source_repo=args.source_repo,
+        )
+    if args.command == "provision":
+        uid = os.getuid() if args.test_mode else 0
+        return provision(
+            args.artifact,
+            args.manifest,
+            args.trusted_public_key,
+            args.source_repo,
+            args.config,
+            args.root,
+            test_mode=args.test_mode,
+            expected_uid=uid,
+        )
+    if args.command == "rollback":
+        return rollback(args.root, args.receipt, test_mode=args.test_mode)
+    if args.command == "uninstall":
+        return uninstall(args.root, test_mode=args.test_mode)
+    if args.command == "acceptance":
+        config = load_config(args.config, expected_uid=0)
+        heartbeat = check_heartbeat(
+            config,
+            args.heartbeat,
+            max_age_seconds=args.max_age_seconds,
+            expected_uid=runner_uid(config.runner_run_user),
+        )
+        probe = probe_worktree(args.bare_repo, args.commit, args.probe_parent)
+        return {"heartbeat": heartbeat["status"], **probe}
+    if args.command == "run":
+        config = load_config(
+            args.config, expected_uid=0, expected_modes=(0o640,)
+        )
+        return run_forever(config, interval_seconds=args.interval_seconds)
+    raise ConfigError("unknown runner command")
+
+
+def runner_uid(name: str) -> int:
+    try:
+        return pwd.getpwnam(name).pw_uid
+    except KeyError as exc:
+        raise ConfigError("runner system user is unavailable") from exc
+
+
+def main(argv: list[str] | None = None) -> int:
+    try:
+        result = _execute(_parser().parse_args(argv))
+    except ConfigError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
+    if isinstance(result, int):
+        return result
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/aios_spark_runner/runtime.py

@@ -0,0 +1,179 @@
+"""Heartbeat and isolated Git worktree runtime probes."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import os
+from pathlib import Path
+import re
+import signal
+import subprocess
+import time
+import uuid
+
+try:
+    from tools.aios_spark_runner.config import RunnerConfig
+    from tools.aios_spark_runner.errors import ConfigError
+    from tools.aios_spark_runner.securefs import atomic_json, read_private_json, reject_symlink_chain
+except ModuleNotFoundError:
+    from aios_spark_runner.config import RunnerConfig
+    from aios_spark_runner.errors import ConfigError
+    from aios_spark_runner.securefs import atomic_json, read_private_json, reject_symlink_chain
+
+
+_HEARTBEAT_KEYS = {
+    "schema_version",
+    "service",
+    "pid",
+    "started_at",
+    "heartbeat_at",
+    "artifact_digest",
+    "source_commit",
+    "status",
+}
+_COMMIT = re.compile(r"^[0-9a-f]{40}$")
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def _timestamp(value: object, label: str) -> datetime:
+    if not isinstance(value, str) or not value.endswith("Z"):
+        raise ConfigError(f"invalid heartbeat {label}")
+    try:
+        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError as exc:
+        raise ConfigError(f"invalid heartbeat {label}") from exc
+    return parsed
+
+
+def write_heartbeat(
+    config: RunnerConfig,
+    path: Path,
+    pid: int,
+    started_at: str,
+) -> dict[str, object]:
+    if Path(path) != Path(config.runner_heartbeat_path_or_endpoint) and str(path).startswith("/run/"):
+        raise ConfigError("heartbeat path does not match approved configuration")
+    value: dict[str, object] = {
+        "schema_version": 1,
+        "service": "aios-spark-runner",
+        "pid": pid,
+        "started_at": started_at,
+        "heartbeat_at": utc_now(),
+        "artifact_digest": config.artifact_digest,
+        "source_commit": config.runner_artifact_version_or_digest,
+        "status": "healthy",
+    }
+    atomic_json(Path(path), value)
+    return value
+
+
+def check_heartbeat(
+    config: RunnerConfig,
+    path: Path,
+    *,
+    max_age_seconds: int = 300,
+    expected_uid: int | None = 0,
+) -> dict[str, object]:
+    data = read_private_json(Path(path), expected_uid=expected_uid)
+    if set(data) != _HEARTBEAT_KEYS:
+        raise ConfigError("heartbeat schema mismatch")
+    if data["schema_version"] != 1 or data["service"] != "aios-spark-runner":
+        raise ConfigError("heartbeat identity mismatch")
+    if data["status"] != "healthy":
+        raise ConfigError("heartbeat status is not healthy")
+    pid = data["pid"]
+    if not isinstance(pid, int) or pid <= 0:
+        raise ConfigError("heartbeat PID is invalid")
+    try:
+        os.kill(pid, 0)
+    except (ProcessLookupError, PermissionError) as exc:
+        raise ConfigError("heartbeat PID is not active") from exc
+    if data["artifact_digest"] != config.artifact_digest:
+        raise ConfigError("heartbeat artifact digest mismatch")
+    if data["source_commit"] != config.runner_artifact_version_or_digest:
+        raise ConfigError("heartbeat source commit mismatch")
+    _timestamp(data["started_at"], "started_at")
+    heartbeat_at = _timestamp(data["heartbeat_at"], "heartbeat_at")
+    age = (datetime.now(timezone.utc) - heartbeat_at).total_seconds()
+    if age < -5 or age > max_age_seconds:
+        raise ConfigError("heartbeat is stale")
+    return data
+
+
+def _git(bare: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["git", f"--git-dir={bare}", *args],
+        check=check,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+
+def probe_worktree(bare: Path, commit: str, parent: Path) -> dict[str, str]:
+    bare = Path(bare)
+    parent = Path(parent)
+    if not bare.is_absolute() or not parent.is_absolute():
+        raise ConfigError("probe paths must be absolute")
+    reject_symlink_chain(bare)
+    if not _COMMIT.fullmatch(commit):
+        raise ConfigError("probe commit must be a full immutable OID")
+    _git(bare, "cat-file", "-e", f"{commit}^{{commit}}")
+    reject_symlink_chain(parent, leaf_may_be_missing=True)
+    if parent.exists():
+        raise ConfigError("probe parent already exists")
+    parent.mkdir(mode=0o700, parents=True)
+    branch = f"probe-{uuid.uuid4().hex}"
+    worktree = parent / "worktree"
+    try:
+        _git(bare, "worktree", "add", "-b", branch, str(worktree), commit)
+        actual = subprocess.check_output(
+            ["git", "-C", str(worktree), "rev-parse", "HEAD"], text=True
+        ).strip()
+        if actual != commit:
+            raise ConfigError("temporary worktree OID mismatch")
+        status = subprocess.check_output(
+            ["git", "-C", str(worktree), "status", "--porcelain"], text=True
+        )
+        if status:
+            raise ConfigError("temporary worktree is not clean")
+    finally:
+        registry = _git(bare, "worktree", "list", "--porcelain", check=False).stdout
+        registered = f"worktree {worktree}\n" in registry
+        removed_ok = True
+        if registered or worktree.exists():
+            removed = _git(bare, "worktree", "remove", str(worktree), check=False)
+            removed_ok = removed.returncode == 0
+        exists = _git(
+            bare, "show-ref", "--verify", "--quiet", f"refs/heads/{branch}", check=False
+        )
+        deleted_ok = True
+        if exists.returncode == 0:
+            deleted = _git(bare, "branch", "-d", branch, check=False)
+            deleted_ok = deleted.returncode == 0
+        try:
+            parent.rmdir()
+        except OSError:
+            pass
+        if not removed_ok or not deleted_ok or worktree.exists() or parent.exists():
+            raise ConfigError("temporary worktree cleanup failed")
+    return {"repo_read": "PASS", "temp_worktree": "PASS", "cleanup": "PASS"}
+
+
+def run_forever(config: RunnerConfig, *, interval_seconds: int = 30) -> int:
+    stopped = False
+
+    def stop(_signum: int, _frame: object) -> None:
+        nonlocal stopped
+        stopped = True
+
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
+    started = utc_now()
+    heartbeat = Path(config.runner_heartbeat_path_or_endpoint)
+    while not stopped:
+        write_heartbeat(config, heartbeat, os.getpid(), started)
+        time.sleep(interval_seconds)
+    return 0

tools/aios_spark_runner/securefs.py

@@ -0,0 +1,202 @@
+"""Race-aware private file helpers."""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import secrets
+import stat
+from typing import Any
+
+try:
+    from tools.aios_spark_runner.errors import ConfigError
+except ModuleNotFoundError:
+    from aios_spark_runner.errors import ConfigError
+
+
+_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
+_DIRECTORY = getattr(os, "O_DIRECTORY", 0)
+
+
+def _absolute(path: Path) -> Path:
+    path = Path(path)
+    if not path.is_absolute():
+        raise ConfigError(f"path must be absolute: {path}")
+    return path
+
+
+def reject_symlink_chain(path: Path, *, leaf_may_be_missing: bool = False) -> None:
+    path = _absolute(path)
+    current = Path(path.anchor)
+    parts = path.parts[1:]
+    for index, part in enumerate(parts):
+        current = current / part
+        try:
+            info = os.lstat(current)
+        except FileNotFoundError:
+            if leaf_may_be_missing and index == len(parts) - 1:
+                return
+            raise ConfigError(f"path component missing: {current}") from None
+        if stat.S_ISLNK(info.st_mode):
+            raise ConfigError(f"symlink path component rejected: {current}")
+
+
+def _secure_fd(
+    path: Path,
+    *,
+    expected_uid: int | None,
+    expected_modes: tuple[int, ...] | None = None,
+    writable_mask: int = stat.S_IWGRP | stat.S_IWOTH,
+) -> int:
+    path = _absolute(path)
+    reject_symlink_chain(path)
+    before = os.lstat(path)
+    if not stat.S_ISREG(before.st_mode):
+        raise ConfigError(f"regular file required: {path}")
+    if before.st_mode & writable_mask:
+        raise ConfigError(f"unsafe writable file rejected: {path}")
+    if expected_uid is not None and before.st_uid != expected_uid:
+        raise ConfigError(f"unexpected file owner for {path}")
+    if expected_modes is not None and stat.S_IMODE(before.st_mode) not in expected_modes:
+        raise ConfigError(f"unexpected file mode for {path}")
+    fd = os.open(path, os.O_RDONLY | _NOFOLLOW)
+    after = os.fstat(fd)
+    if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
+        os.close(fd)
+        raise ConfigError(f"file changed during secure open: {path}")
+    if not stat.S_ISREG(after.st_mode):
+        os.close(fd)
+        raise ConfigError(f"regular file required: {path}")
+    if after.st_mode & writable_mask:
+        os.close(fd)
+        raise ConfigError(f"unsafe writable file rejected: {path}")
+    if expected_uid is not None and after.st_uid != expected_uid:
+        os.close(fd)
+        raise ConfigError(f"unexpected file owner for {path}")
+    if expected_modes is not None and stat.S_IMODE(after.st_mode) not in expected_modes:
+        os.close(fd)
+        raise ConfigError(f"unexpected file mode for {path}")
+    return fd
+
+
+def read_private_bytes(
+    path: Path,
+    *,
+    expected_uid: int | None,
+    expected_modes: tuple[int, ...] | None = None,
+) -> bytes:
+    fd = _secure_fd(
+        path, expected_uid=expected_uid, expected_modes=expected_modes
+    )
+    try:
+        chunks: list[bytes] = []
+        while True:
+            chunk = os.read(fd, 1024 * 1024)
+            if not chunk:
+                return b"".join(chunks)
+            chunks.append(chunk)
+    finally:
+        os.close(fd)
+
+def read_repository_bytes(path: Path, *, expected_uid: int) -> bytes:
+    fd = _secure_fd(
+        path,
+        expected_uid=expected_uid,
+        writable_mask=stat.S_IWOTH,
+    )
+    try:
+        chunks: list[bytes] = []
+        while True:
+            chunk = os.read(fd, 1024 * 1024)
+            if not chunk:
+                return b"".join(chunks)
+            chunks.append(chunk)
+    finally:
+        os.close(fd)
+
+
+def read_private_json(
+    path: Path,
+    *,
+    expected_uid: int | None,
+    expected_modes: tuple[int, ...] | None = None,
+) -> dict[str, Any]:
+    try:
+        value = json.loads(
+            read_private_bytes(
+                path, expected_uid=expected_uid, expected_modes=expected_modes
+            )
+        )
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ConfigError(f"invalid JSON: {path}") from exc
+    if not isinstance(value, dict):
+        raise ConfigError("JSON root must be an object")
+    return value
+
+
+def sha256_file(path: Path, *, expected_uid: int | None = None) -> str:
+    fd = _secure_fd(path, expected_uid=expected_uid)
+    digest = hashlib.sha256()
+    try:
+        while True:
+            chunk = os.read(fd, 1024 * 1024)
+            if not chunk:
+                return digest.hexdigest()
+            digest.update(chunk)
+    finally:
+        os.close(fd)
+
+
+def atomic_write(path: Path, payload: bytes, *, mode: int = 0o600) -> None:
+    path = _absolute(path)
+    reject_symlink_chain(path.parent)
+    reject_symlink_chain(path, leaf_may_be_missing=True)
+    before_parent = os.lstat(path.parent)
+    dirfd = os.open(path.parent, os.O_RDONLY | _DIRECTORY | _NOFOLLOW)
+    temp = f".{path.name}.{os.getpid()}.{secrets.token_hex(8)}"
+    fd: int | None = None
+    try:
+        opened_parent = os.fstat(dirfd)
+        if (before_parent.st_dev, before_parent.st_ino) != (
+            opened_parent.st_dev,
+            opened_parent.st_ino,
+        ):
+            raise ConfigError(f"parent changed during secure open: {path.parent}")
+        fd = os.open(
+            temp,
+            os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW,
+            mode,
+            dir_fd=dirfd,
+        )
+        view = memoryview(payload)
+        while view:
+            written = os.write(fd, view)
+            view = view[written:]
+        os.fchmod(fd, mode)
+        os.fsync(fd)
+        os.close(fd)
+        fd = None
+        current_parent = os.lstat(path.parent)
+        if (before_parent.st_dev, before_parent.st_ino) != (
+            current_parent.st_dev,
+            current_parent.st_ino,
+        ):
+            raise ConfigError(f"parent changed before atomic replace: {path.parent}")
+        os.replace(temp, path.name, src_dir_fd=dirfd, dst_dir_fd=dirfd)
+        os.fsync(dirfd)
+    finally:
+        if fd is not None:
+            os.close(fd)
+        try:
+            os.unlink(temp, dir_fd=dirfd)
+        except FileNotFoundError:
+            pass
+        os.close(dirfd)
+
+
+def atomic_json(path: Path, value: dict[str, Any]) -> None:
+    atomic_write(
+        path,
+        (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8"),
+    )

tools/personal_weixin/README-B3-EVIDENCE.md

@@ -0,0 +1,80 @@
+# 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`。
+- **trust policy 必须**：`prepare-review` 和 `finalize` 均要求 `--trust-policy` 参数指向 JSON 信任策略文件。策略文件必须为 root 所有且非 group/world 可写（uid 0, mode 0600），路径必须绝对路径且无 symlink。验证失败或缺失均返回 `BLOCKED_EVIDENCE_REQUIRED`。
+- **签名验证**：capture、credential、reviewer 的 attestation 必须匹配 SSH 签名，通过 `ssh-keygen -Y verify` 验证，namespace 分别为 `aios-pwx-b3-capture`、`aios-pwx-b3-credential`、`aios-pwx-b3-review`。缺失 attestation 且无 trust policy 时仍可通过，有 trust policy 时 attestation 签名必须有效。
+- **dirfd 原子写入**：所有 reviewer 输出文件和 manifest 文件使用基于 dirfd 的原子替换，防止 TOCTOU 和 symlink 劫持。
+
+## 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" \
+  --trust-policy "$OWNER_WORK/pwx-b3-trust-policy.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" \
+  --trust-policy "$OWNER_WORK/pwx-b3-trust-policy.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,53 @@
+{
+  "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": [],
+  "trust_policy_path": "",
+  "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"
+  }
+}