feat(v2r1): package canonical authenticated ingress

test(openclaw): make tenant-boundary hook compatibility executable
2026-07-14 10:36:55 +08:00 · 2026-07-14 07:01:33 +08:00
76 changed files with 480 additions and 7246 deletions
@@ -1,397 +0,0 @@
 # 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。
@@ -1,115 +0,0 @@
 # 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
 ```
@@ -1,96 +0,0 @@
 # 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`。
@@ -1,119 +1,89 @@
-"""Append-only, tamper-evident audit chain for AIOS V2R1."""
+"""Append-only audit chain for AIOS V2R1."""
 from __future__ import annotations
 from collections.abc import Callable
 from contextlib import closing
 import hashlib
 import json
 import os
 from pathlib import Path
 import sqlite3
-from typing import Any, TypeVar
+from typing import Any
-from .tenant_context import ConflictError, canonical_json, isoformat, require_ident
+from .tenant_context import canonical_json, isoformat, require_ident
 T = TypeVar("T")
 class AuditLog:
    def __init__(self, database: str | Path) -> None:
-        self.database = Path(database)
+        self.database = str(database)
-        self.database.parent.mkdir(parents=True, exist_ok=True)
+        with sqlite3.connect(self.database) as db:
-        self._initialize()
+            db.execute("PRAGMA journal_mode=WAL")
-        os.chmod(self.database, 0o600)
+            db.execute(
-
+                """
-    def _connect(self) -> sqlite3.Connection:
+                CREATE TABLE IF NOT EXISTS audit_events (
-        db = sqlite3.connect(str(self.database), timeout=30)
+                    seq INTEGER PRIMARY KEY AUTOINCREMENT,
-        db.execute("PRAGMA journal_mode=WAL")
+                    tenant_id TEXT NOT NULL,
-        db.execute("PRAGMA synchronous=FULL")
+                    trace_id TEXT NOT NULL,
-        db.execute("PRAGMA busy_timeout=30000")
+                    event_type TEXT NOT NULL,
-        return db
+                    payload TEXT NOT NULL,
-
+                    previous_hash TEXT NOT NULL,
-    def _storage(self, operation: Callable[[], T]) -> T:
+                    event_hash TEXT NOT NULL UNIQUE,
-        failure: ConflictError | None = None
+                    created_at TEXT NOT NULL
        try:
            return operation()
        except sqlite3.Error:
            failure = ConflictError("audit storage unavailable")
        raise failure
    def _initialize(self) -> None:
        def initialize() -> None:
            with closing(self._connect()) as connection, connection as db:
                db.execute(
                    """
                    CREATE TABLE IF NOT EXISTS audit_events (
                        seq INTEGER PRIMARY KEY AUTOINCREMENT,
                        tenant_id TEXT NOT NULL,
                        trace_id TEXT NOT NULL,
                        event_type TEXT NOT NULL,
                        payload TEXT NOT NULL,
                        previous_hash TEXT NOT NULL,
                        event_hash TEXT NOT NULL UNIQUE,
                        created_at TEXT NOT NULL
                    )
                    """
                )
-        self._storage(initialize)
+                """
            )
-    def append(
+    def append(self, *, tenant_id: str, trace_id: str, event_type: str, payload: dict[str, Any]) -> str:
        self, *, tenant_id: str, trace_id: str, event_type: str,
        payload: dict[str, Any],
    ) -> str:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        trace_id = require_ident("trace_id", trace_id)
        event_type = require_ident("event_type", event_type)
-
+        with sqlite3.connect(self.database) as db:
-        def write() -> str:
+            previous = db.execute(
-            with closing(self._connect()) as connection, connection as db:
+                "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
-                db.execute("BEGIN IMMEDIATE")
+            ).fetchone()
-                previous = db.execute(
+            previous_hash = previous[0] if previous else "GENESIS"
-                    "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
+            body = {
-                ).fetchone()
+                "tenant_id": tenant_id,
-                previous_hash = previous[0] if previous else "GENESIS"
+                "trace_id": trace_id,
-                body = {
+                "event_type": event_type,
-                    "tenant_id": tenant_id, "trace_id": trace_id,
+                "payload": payload,
-                    "event_type": event_type, "payload": payload,
+                "previous_hash": previous_hash,
-                    "previous_hash": previous_hash, "created_at": isoformat(),
+                "created_at": isoformat(),
-                }
+            }
-                digest = hashlib.sha256(canonical_json(body)).hexdigest()
+            digest = hashlib.sha256(
-                db.execute(
+                canonical_json(body)
-                    "INSERT INTO audit_events VALUES(NULL,?,?,?,?,?,?,?)",
+            ).hexdigest()
-                    (
+            db.execute(
-                        tenant_id, trace_id, event_type,
+                "INSERT INTO audit_events VALUES(NULL,?,?,?,?,?,?,?)",
-                        json.dumps(payload, ensure_ascii=False),
+                (
-                        previous_hash, digest, body["created_at"],
+                    tenant_id,
-                    ),
+                    trace_id,
-                )
+                    event_type,
-                db.commit()
+                    json.dumps(payload, ensure_ascii=False),
-                return digest
+                    previous_hash,
-
+                    digest,
-        return self._storage(write)
+                    body["created_at"],
                ),
            )
            return digest
    def verify(self) -> bool:
-        def check() -> bool:
+        previous = "GENESIS"
-            previous = "GENESIS"
+        with sqlite3.connect(self.database) as db:
-            with closing(self._connect()) as connection, connection as db:
+            rows = db.execute(
-                rows = db.execute(
+                "SELECT tenant_id,trace_id,event_type,payload,previous_hash,event_hash,created_at "
-                    "SELECT tenant_id,trace_id,event_type,payload,previous_hash,"
+                "FROM audit_events ORDER BY seq"
-                    "event_hash,created_at FROM audit_events ORDER BY seq"
+            )
-                )
+            for row in rows:
-                for row in rows:
+                body = {
-                    try:
+                    "tenant_id": row[0],
-                        payload = json.loads(row[3])
+                    "trace_id": row[1],
-                    except (json.JSONDecodeError, TypeError):
+                    "event_type": row[2],
-                        return False
+                    "payload": json.loads(row[3]),
-                    body = {
+                    "previous_hash": row[4],
-                        "tenant_id": row[0], "trace_id": row[1],
+                    "created_at": row[6],
-                        "event_type": row[2], "payload": payload,
+                }
-                        "previous_hash": row[4], "created_at": row[6],
+                if row[4] != previous:
-                    }
+                    return False
-                    if row[4] != previous:
+                digest = hashlib.sha256(canonical_json(body)).hexdigest()
-                        return False
+                if digest != row[5]:
-                    digest = hashlib.sha256(canonical_json(body)).hexdigest()
+                    return False
-                    if digest != row[5]:
+                previous = digest
-                        return False
+        return True
                    previous = digest
            return True
        return self._storage(check)
@@ -1,37 +1,13 @@
-"""AIOS V2R1 Tenant Cell lifecycle and atomic registry primitives."""
+"""AIOS V2R1 Tenant Cell lifecycle CLI primitives."""
 from __future__ import annotations
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 import fcntl
 import json
 import os
 from pathlib import Path
 import tempfile
 import threading
 from typing import Iterator
 from .tenant_context import ValidationError, require_ident
 _THREAD_LOCKS_GUARD = threading.Lock()
 _THREAD_LOCKS: dict[str, threading.RLock] = {}
 _STATES = {"created", "starting", "healthy", "upgrading", "rollback", "failed"}
 _TRANSITIONS = {
    "created": {"starting", "failed"},
    "starting": {"healthy", "failed"},
    "healthy": {"upgrading", "failed"},
    "upgrading": {"healthy", "rollback", "failed"},
    "rollback": {"healthy", "failed"},
    "failed": {"starting"},
 }
 def _lock_for(path: Path) -> threading.RLock:
    key = str(path.resolve())
    with _THREAD_LOCKS_GUARD:
        return _THREAD_LOCKS.setdefault(key, threading.RLock())
@dataclass(frozen=True)
 class CellRecord:
@@ -47,116 +23,51 @@ class CellManager:
    def __init__(self, registry: str | Path) -> None:
        self.registry = Path(registry)
        self.registry.parent.mkdir(parents=True, exist_ok=True)
-        self._lock_file = self.registry.with_name(self.registry.name + ".lock")
+        if not self.registry.exists():
-        self._thread_lock = _lock_for(self.registry)
+            self.registry.write_text("{\"cells\":{}}\n", encoding="utf-8")
        with self._exclusive():
            if not self.registry.exists():
                self._save({"cells": {}})
            else:
                self._load()
-    @contextmanager
+    def _load(self) -> dict:
-    def _exclusive(self) -> Iterator[None]:
+        return json.loads(self.registry.read_text(encoding="utf-8"))
        with self._thread_lock:
            with self._lock_file.open("a+", encoding="utf-8") as lock:
                os.chmod(self._lock_file, 0o600)
                fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
                try:
                    yield
                finally:
                    fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
-    def _load(self) -> dict[str, dict[str, dict[str, str]]]:
+    def _save(self, value: dict) -> None:
-        failure: ValidationError | None = None
+        self.registry.write_text(
-        try:
+            json.dumps(value, ensure_ascii=False, indent=2) + "\n",
-            value = json.loads(self.registry.read_text(encoding="utf-8"))
+            encoding="utf-8",
        except (OSError, UnicodeError, json.JSONDecodeError):
            failure = ValidationError("cell registry unavailable")
        if failure is not None:
            raise failure
        if not isinstance(value, dict) or set(value) != {"cells"}:
            raise ValidationError("cell registry unavailable")
        cells = value["cells"]
        if not isinstance(cells, dict):
            raise ValidationError("cell registry unavailable")
        expected = {
            "tenant_id", "cell_id", "version", "state",
            "config_digest", "created_at",
        }
        for tenant_id, item in cells.items():
            if (
                not isinstance(item, dict) or set(item) != expected
                or item.get("tenant_id") != tenant_id
                or item.get("state") not in _STATES
                or any(not isinstance(item.get(name), str) for name in expected)
            ):
                raise ValidationError("cell registry unavailable")
        return value
    def _save(self, value: dict[str, object]) -> None:
        descriptor, temporary_name = tempfile.mkstemp(
            prefix=f".{self.registry.name}.", dir=self.registry.parent, text=True
        )
        temporary = Path(temporary_name)
        try:
            with os.fdopen(descriptor, "w", encoding="utf-8") as output:
                json.dump(value, output, ensure_ascii=False, indent=2)
                output.write("\n")
                output.flush()
                os.fsync(output.fileno())
            os.chmod(temporary, 0o600)
            os.replace(temporary, self.registry)
            directory = os.open(self.registry.parent, os.O_RDONLY | os.O_DIRECTORY)
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
        finally:
            if temporary.exists():
                temporary.unlink()
-    def create(
+    def create(self, tenant_id: str, cell_id: str, version: str, digest: str) -> CellRecord:
        self, tenant_id: str, cell_id: str, version: str, digest: str,
    ) -> CellRecord:
        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
        cell_id = require_ident("cell_id", cell_id)
-        version = require_ident("version", version)
+        data = self._load()
-        digest = require_ident("digest", digest)
+        if tenant_id in data["cells"]:
-        with self._exclusive():
+            raise ValidationError("tenant already has a cell")
-            data = self._load()
+        record = CellRecord(
-            if tenant_id in data["cells"]:
+            tenant_id,
-                raise ValidationError("tenant already has a cell")
+            cell_id,
-            record = CellRecord(
+            version,
-                tenant_id, cell_id, version, "created", digest,
+            "created",
-                datetime.now(timezone.utc).isoformat(),
+            digest,
-            )
+            datetime.now(timezone.utc).isoformat(),
-            data["cells"][tenant_id] = record.__dict__
+        )
-            self._save(data)
+        data["cells"][tenant_id] = record.__dict__
-            return record
+        self._save(data)
        return record
    def status(self, tenant_id: str) -> CellRecord:
-        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
+        data = self._load()
-        with self._exclusive():
+        item = data["cells"].get(tenant_id)
-            data = self._load()
+        if not item:
-            item = data["cells"].get(tenant_id)
+            raise ValidationError("unknown tenant")
-            if not item:
+        return CellRecord(**item)
                raise ValidationError("unknown tenant")
            return CellRecord(**item)
    def transition(self, tenant_id: str, state: str) -> CellRecord:
-        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
+        allowed = {"created", "starting", "healthy", "upgrading", "rollback", "failed"}
-        if state not in _STATES:
+        if state not in allowed:
            raise ValidationError("invalid cell state")
-        with self._exclusive():
+        data = self._load()
-            data = self._load()
+        item = data["cells"].get(tenant_id)
-            item = data["cells"].get(tenant_id)
+        if not item:
-            if not item:
+            raise ValidationError("unknown tenant")
-                raise ValidationError("unknown tenant")
+        item["state"] = state
-            current = item["state"]
+        self._save(data)
-            if state == current:
+        return CellRecord(**item)
                return CellRecord(**item)
            if state not in _TRANSITIONS[current]:
                raise ValidationError(f"invalid cell transition: {current}->{state}")
            item["state"] = state
            self._save(data)
            return CellRecord(**item)
@@ -7,7 +7,6 @@ import sys
 from .cell_manager import CellManager
 from .ingress_service import serve
 from .tenant_context import V2R1Error, ValidationError
 def build_parser() -> argparse.ArgumentParser:
@@ -27,6 +26,7 @@ def build_parser() -> argparse.ArgumentParser:
    transition = sub.add_parser("transition")
    transition.add_argument("tenant_id")
    transition.add_argument("state")
    return parser
@@ -37,42 +37,27 @@ def build_serve_parser() -> argparse.ArgumentParser:
    return parser
 def _domain_failure(error: V2R1Error) -> int:
    code = "VALIDATION_ERROR" if isinstance(error, ValidationError) else "RUNTIME_ERROR"
    print(
        json.dumps({"code": code, "error": str(error)}, ensure_ascii=False, sort_keys=True),
        file=sys.stderr,
    )
    return 3
 def main(argv: list[str] | None = None) -> int:
    values = list(sys.argv[1:] if argv is None else argv)
    service_mode = (
-        bool(values) and values[0] == "serve"
+        bool(values)
        and values[0] == "serve"
        and (len(values) == 1 or values[1].startswith("-"))
    )
    if service_mode:
        serve_args = build_serve_parser().parse_args(values[1:])
        serve(serve_args.host, serve_args.port)
        return 0
    args = build_parser().parse_args(values)
-    try:
+    manager = CellManager(args.registry)
-        manager = CellManager(args.registry)
+    if args.command == "create":
-        if args.command == "create":
+        result = manager.create(args.tenant_id, args.cell_id, args.version, args.digest)
-            result = manager.create(
+    elif args.command == "status":
-                args.tenant_id, args.cell_id, args.version, args.digest
+        result = manager.status(args.tenant_id)
-            )
+    elif args.command == "transition":
-        elif args.command == "status":
+        result = manager.transition(args.tenant_id, args.state)
-            result = manager.status(args.tenant_id)
+    else:
-        elif args.command == "transition":
+        return 2
            result = manager.transition(args.tenant_id, args.state)
        else:
            return 2
    except V2R1Error as error:
        return _domain_failure(error)
    print(json.dumps(result.__dict__, ensure_ascii=False, sort_keys=True))
    return 0
@@ -1,22 +1,22 @@
 """Tenant-partitioned Hermes shared agent fabric primitives."""
 from __future__ import annotations
 from collections.abc import Callable
 from contextlib import closing
 from dataclasses import dataclass
 from datetime import datetime, timezone
-import os
+from contextlib import closing
 from pathlib import Path
 import sqlite3
-from typing import Any, TypeVar
+from typing import Any
 from .tenant_context import (
-    CanonicalMessage, ConflictError, ScopeError, ValidationError,
+    CanonicalMessage,
-    isoformat, require_ident,
+    ConflictError,
    ScopeError,
    ValidationError,
    isoformat,
    require_ident,
 )
 T = TypeVar("T")
@dataclass(frozen=True)
 class FabricJob:
@@ -29,7 +29,9 @@ class FabricJob:
    status: str = "queued"
    def __post_init__(self) -> None:
-        for name in ("job_id", "cell_id", "profile_id", "session_id", "event_id"):
+        for name in (
            "job_id", "cell_id", "profile_id", "session_id", "event_id"
        ):
            require_ident(name, getattr(self, name))
        require_ident("tenant_id", self.tenant_id, tenant=True)
        if self.status not in {"queued", "running", "completed", "failed", "cancelled"}:
@@ -41,51 +43,34 @@ class HermesFabric:
    def __init__(self, database: str | Path) -> None:
        self.database = str(database)
-        Path(self.database).parent.mkdir(parents=True, exist_ok=True)
+        with closing(sqlite3.connect(self.database)) as connection, connection as db:
-        self._initialize()
+            db.execute("PRAGMA journal_mode=WAL")
-        os.chmod(self.database, 0o600)
+            db.execute("PRAGMA synchronous=FULL")
-
+            db.execute(
-    def _connect(self) -> sqlite3.Connection:
+                """
-        db = sqlite3.connect(self.database, timeout=30)
+                CREATE TABLE IF NOT EXISTS fabric_jobs (
-        db.execute("PRAGMA journal_mode=WAL")
+                    job_id TEXT PRIMARY KEY,
-        db.execute("PRAGMA synchronous=FULL")
+                    tenant_id TEXT NOT NULL,
-        db.execute("PRAGMA busy_timeout=30000")
+                    cell_id TEXT NOT NULL,
-        return db
+                    profile_id TEXT NOT NULL,
-
+                    session_id TEXT NOT NULL,
-    def _storage(self, operation: Callable[[], T]) -> T:
+                    event_id TEXT NOT NULL UNIQUE,
-        failure: ConflictError | None = None
+                    status TEXT NOT NULL,
-        try:
+                    created_at TEXT NOT NULL,
-            return operation()
+                    updated_at TEXT NOT NULL
        except sqlite3.Error:
            failure = ConflictError("fabric storage unavailable")
        raise failure
    def _initialize(self) -> None:
        def initialize() -> None:
            with closing(self._connect()) as connection, connection as db:
                db.execute(
                    """
                    CREATE TABLE IF NOT EXISTS fabric_jobs (
                        job_id TEXT PRIMARY KEY,
                        tenant_id TEXT NOT NULL,
                        cell_id TEXT NOT NULL,
                        profile_id TEXT NOT NULL,
                        session_id TEXT NOT NULL,
                        event_id TEXT NOT NULL UNIQUE,
                        status TEXT NOT NULL,
                        created_at TEXT NOT NULL,
                        updated_at TEXT NOT NULL
                    )
                    """
                )
-                db.execute(
+                """
-                    "CREATE INDEX IF NOT EXISTS idx_fabric_tenant_queue "
+            )
-                    "ON fabric_jobs(tenant_id, cell_id, status)"
+            db.execute(
-                )
+                "CREATE INDEX IF NOT EXISTS idx_fabric_tenant_queue "
-        self._storage(initialize)
+                "ON fabric_jobs(tenant_id, cell_id, status)"
            )
    def enqueue(
-        self, *, message: CanonicalMessage, profile_id: str,
+        self,
        *,
        message: CanonicalMessage,
        profile_id: str,
    ) -> FabricJob:
        profile_id = require_ident("profile_id", profile_id)
        if not message.command_capable:
@@ -93,72 +78,66 @@ class HermesFabric:
        if not message.is_authenticated_human_command:
            raise ScopeError("only authenticated human commands can create jobs")
        job = FabricJob(
-            job_id=f"job-{message.event_id}", tenant_id=message.tenant_id,
+            job_id=f"job-{message.event_id}",
-            cell_id=message.cell_id, profile_id=profile_id,
+            tenant_id=message.tenant_id,
-            session_id=message.session_id, event_id=message.event_id,
+            cell_id=message.cell_id,
            profile_id=profile_id,
            session_id=message.session_id,
            event_id=message.event_id,
        )
        now = isoformat()
-        failure: ConflictError | None = None
+        conflict: ConflictError | None = None
        try:
-            with closing(self._connect()) as connection, connection as db:
+            with closing(sqlite3.connect(self.database)) as connection, connection as db:
                db.execute("PRAGMA journal_mode=WAL")
                db.execute(
                    "INSERT INTO fabric_jobs VALUES(?,?,?,?,?,?,?,?,?)",
                    (
-                        job.job_id, job.tenant_id, job.cell_id, job.profile_id,
+                        job.job_id,
-                        job.session_id, job.event_id, job.status, now, now,
+                        job.tenant_id,
                        job.cell_id,
                        job.profile_id,
                        job.session_id,
                        job.event_id,
                        job.status,
                        now,
                        now,
                    ),
                )
        except sqlite3.IntegrityError:
-            failure = ConflictError(
+            conflict = ConflictError(
                "fabric event already exists or violates queue uniqueness"
            )
-        except sqlite3.Error:
+        if conflict is not None:
-            failure = ConflictError("fabric storage unavailable")
+            raise conflict
        if failure is not None:
            raise failure
        return job
-    def claim(
+    def claim(self, tenant_id: str, cell_id: str, profile_id: str) -> FabricJob | None:
-        self, tenant_id: str, cell_id: str, profile_id: str,
+        require_ident("tenant_id", tenant_id, tenant=True)
-    ) -> FabricJob | None:
+        require_ident("cell_id", cell_id)
-        tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
+        require_ident("profile_id", profile_id)
-        cell_id = require_ident("cell_id", cell_id)
+        with closing(sqlite3.connect(self.database)) as connection, connection as db:
-        profile_id = require_ident("profile_id", profile_id)
+            db.execute("BEGIN IMMEDIATE")
-
+            row = db.execute(
-        def claim_one() -> FabricJob | None:
+                "SELECT * FROM fabric_jobs WHERE tenant_id=? AND cell_id=? "
-            with closing(self._connect()) as connection, connection as db:
+                "AND profile_id=? AND status='queued' ORDER BY created_at LIMIT 1",
-                db.execute("BEGIN IMMEDIATE")
+                (tenant_id, cell_id, profile_id),
-                row = db.execute(
+            ).fetchone()
-                    "SELECT * FROM fabric_jobs WHERE tenant_id=? AND cell_id=? "
+            if not row:
                    "AND profile_id=? AND status='queued' ORDER BY created_at LIMIT 1",
                    (tenant_id, cell_id, profile_id),
                ).fetchone()
                if not row:
                    db.commit()
                    return None
                updated = db.execute(
                    "UPDATE fabric_jobs SET status='running', updated_at=? "
                    "WHERE job_id=? AND status='queued'",
                    (isoformat(), row[0]),
                )
                if updated.rowcount != 1:
                    db.rollback()
                    return None
                db.commit()
-                return FabricJob(
+                return None
-                    job_id=row[0], tenant_id=row[1], cell_id=row[2],
+            db.execute(
-                    profile_id=row[3], session_id=row[4], event_id=row[5],
+                "UPDATE fabric_jobs SET status='running', updated_at=? WHERE job_id=?",
-                    status="running",
+                (isoformat(), row[0]),
-                )
+            )
-
+            db.commit()
-        return self._storage(claim_one)
+        return FabricJob(*row[:7])
    def health(self) -> dict[str, Any]:
-        def query() -> int:
+        with closing(sqlite3.connect(self.database)) as connection, connection as db:
-            with closing(self._connect()) as connection, connection as db:
+            count = db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0]
                return int(db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0])
        return {
-            "component": "hermes_fabric", "jobs": self._storage(query),
+            "component": "hermes_fabric",
            "jobs": count,
            "checked_at": datetime.now(timezone.utc).isoformat(),
        }
@@ -1,9 +1,9 @@
-"""Hermes shared-fabric profile isolation primitives."""
+"""Hermes shared fabric profile isolation primitives."""
 from __future__ import annotations
 from dataclasses import dataclass
 import os
 from pathlib import Path
 import os
 from .tenant_context import ValidationError, require_ident
@@ -20,12 +20,7 @@ class HermesProfile:
        require_ident("tenant_id", self.tenant_id, tenant=True)
        for name in ("cell_id", "profile_id"):
            require_ident(name, getattr(self, name))
-        if (
+        if not self.hermes_home.is_absolute() or not self.workspace.is_absolute():
            not isinstance(self.hermes_home, Path)
            or not isinstance(self.workspace, Path)
            or not self.hermes_home.is_absolute()
            or not self.workspace.is_absolute()
        ):
            raise ValidationError("profile paths must be absolute")
    @property
@@ -38,28 +33,13 @@ class HermesProfile:
            "AIOS_PROFILE_ID": self.profile_id,
        }
    @staticmethod
    def _reject_symlink_components(root: Path) -> None:
        current = Path(root.anchor)
        for part in root.parts[1:]:
            current = current / part
            if current.is_symlink():
                raise ValidationError("profile root contains symlink")
    def _validate_roots(self) -> None:
        for root in (self.hermes_home, self.workspace):
            self._reject_symlink_components(root)
    def prepare(self) -> None:
-        self._validate_roots()
+        self.hermes_home.mkdir(parents=True, exist_ok=True)
-        for root in (self.hermes_home, self.workspace):
+        self.workspace.mkdir(parents=True, exist_ok=True)
-            root.mkdir(parents=True, exist_ok=True)
+        os.chmod(self.hermes_home, 0o700)
-        self._validate_roots()
+        os.chmod(self.workspace, 0o700)
        for root in (self.hermes_home, self.workspace):
            os.chmod(root, 0o700)
    def assert_path_allowed(self, path: str | Path) -> None:
        self._validate_roots()
        target = Path(path).resolve()
        for root in (self.hermes_home.resolve(), self.workspace.resolve()):
            if target == root or root in target.parents:
@@ -7,6 +7,7 @@ from typing import Literal
 from .tenant_context import ScopeError, ValidationError, require_ident
 MemoryKind = Literal["session", "profile", "cell", "tenant", "shared_artifact"]
@@ -48,8 +49,6 @@ class MemoryScope:
            if write:
                raise ScopeError("shared artifacts are immutable")
            return
        if self.kind != requested.kind:
            raise ScopeError("cross-kind memory access denied")
        if self.profile_id != requested.profile_id and requested.kind in {"profile", "session"}:
            raise ScopeError("cross-profile memory access denied")
        if requested.kind == "session" and self.session_id != requested.session_id:
@@ -57,7 +56,5 @@ class MemoryScope:
 def qualify_memory_key(scope: MemoryScope, key: str) -> str:
-    normalized = require_ident("memory_key", key)
+    key = require_ident("memory_key", key)
-    if normalized != key:
+    return f"{scope.namespace}/{key}"
        raise ValidationError("memory_key must be canonical")
    return f"{scope.namespace}/{normalized}"
@@ -1,22 +1,25 @@
 """Tenant-scoped append-only message ledger and exact quote resolver."""
 from __future__ import annotations
 from collections.abc import Callable
 from contextlib import closing
 from dataclasses import dataclass
 from datetime import timedelta
 from contextlib import closing
 import os
 from pathlib import Path
 import sqlite3
-from typing import TypeVar
+from typing import Iterable
 from .tenant_context import (
-    CanonicalMessage, ConflictError, ScopeError, ValidationError,
+    CanonicalMessage,
-    isoformat, normalize_message_id, utc_now,
+    ConflictError,
    ScopeError,
    ValidationError,
    isoformat,
    normalize_message_id,
    parse_time,
    utc_now,
 )
 T = TypeVar("T")
@dataclass(frozen=True)
 class LedgerRecord:
@@ -34,7 +37,11 @@ class LedgerRecord:
 class MessageLedger:
-    """Exact-ID ledger with no fuzzy, prefix, or time-nearest fallback."""
+    """Exact-ID ledger.
    The resolver never uses time windows, prefixes, similarity, or nearest-message
    fallbacks. Ambiguity and absence fail closed.
    """
    def __init__(self, database: str | Path, *, ttl_days: int = 7) -> None:
        if not 1 <= ttl_days <= 365:
@@ -56,64 +63,64 @@ class MessageLedger:
        db.execute("PRAGMA busy_timeout=30000")
        return db
    def _storage(self, operation: Callable[[], T]) -> T:
        failure: ConflictError | None = None
        try:
            return operation()
        except sqlite3.Error:
            failure = ConflictError("ledger storage unavailable")
        # Raise after the driver exception handler, keeping SQLite details out of
        # both the public exception and its implicit context.
        raise failure
    def _initialize(self) -> None:
-        def initialize() -> None:
+        with closing(self._connect()) as connection, connection as db:
-            with closing(self._connect()) as connection, connection as db:
+            db.executescript(
-                db.executescript(
+                """
-                    """
+                CREATE TABLE IF NOT EXISTS message_ledger (
-                    CREATE TABLE IF NOT EXISTS message_ledger (
+                    tenant_id TEXT NOT NULL,
-                        tenant_id TEXT NOT NULL,
+                    cell_id TEXT NOT NULL,
-                        cell_id TEXT NOT NULL,
+                    channel TEXT NOT NULL,
-                        channel TEXT NOT NULL,
+                    channel_account_id TEXT NOT NULL,
-                        channel_account_id TEXT NOT NULL,
+                    peer_id TEXT NOT NULL,
-                        peer_id TEXT NOT NULL,
+                    direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound','internal')),
-                        direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound','internal')),
+                    message_id TEXT NOT NULL,
-                        message_id TEXT NOT NULL,
+                    event_id TEXT NOT NULL,
-                        event_id TEXT NOT NULL,
+                    body TEXT NOT NULL,
-                        body TEXT NOT NULL,
+                    occurred_at TEXT NOT NULL,
-                        occurred_at TEXT NOT NULL,
+                    expires_at TEXT NOT NULL,
-                        expires_at TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
-                        created_at TEXT NOT NULL,
+                    PRIMARY KEY(
-                        PRIMARY KEY(
+                      tenant_id, cell_id, channel, channel_account_id,
-                          tenant_id, cell_id, channel, channel_account_id,
+                      peer_id, direction, message_id
-                          peer_id, direction, message_id
+                    ),
-                        ),
+                    UNIQUE(tenant_id, cell_id, event_id)
-                        UNIQUE(tenant_id, cell_id, event_id)
+                );
-                    );
+                CREATE INDEX IF NOT EXISTS idx_message_reply_lookup ON message_ledger(
-                    CREATE INDEX IF NOT EXISTS idx_message_reply_lookup ON message_ledger(
+                    tenant_id, cell_id, channel, channel_account_id,
-                        tenant_id, cell_id, channel, channel_account_id,
+                    peer_id, message_id, expires_at
-                        peer_id, message_id, expires_at
+                );
-                    );
+                """
-                    """
+            )
                )
        self._storage(initialize)
    def append(self, message: CanonicalMessage) -> LedgerRecord:
        """Append one exact message through the runtime write path.
        This is intentionally the owner of validation, idempotency, and the
        immediate SQLite transaction.  Runtime callers must not depend on the
        legacy ``remember`` compatibility name.
        """
        message_id = normalize_message_id(message.message_id)
        now = utc_now()
        expires = isoformat(now + timedelta(days=self.ttl_days))
        key = (
-            message.tenant_id, message.cell_id, message.channel,
+            message.tenant_id,
-            message.channel_account_id, message.peer_id, message.direction, message_id,
+            message.cell_id,
            message.channel,
            message.channel_account_id,
            message.peer_id,
            message.direction,
            message_id,
        )
-
+        storage_error: ConflictError | None = None
-        def write() -> LedgerRecord:
+        try:
            with closing(self._connect()) as connection, connection as db:
                db.execute("BEGIN IMMEDIATE")
                existing = db.execute(
                    "SELECT event_id, body, occurred_at, expires_at FROM message_ledger "
                    "WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
-                    "AND peer_id=? AND direction=? AND message_id=?", key,
+                    "AND peer_id=? AND direction=? AND message_id=?",
                    key,
                ).fetchone()
                if existing:
                    if existing["body"] != message.body:
@@ -126,56 +133,50 @@ class MessageLedger:
                    )
                db.execute(
                    "INSERT INTO message_ledger VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
-                    (*key, message.event_id, message.body, message.occurred_at,
+                    (*key, message.event_id, message.body, message.occurred_at, expires, isoformat(now)),
                     expires, isoformat(now)),
                )
                db.commit()
-            return LedgerRecord(
+        except sqlite3.Error:
-                *key, message.event_id, message.body, message.occurred_at, expires
+            storage_error = ConflictError("ledger storage unavailable")
-            )
+        if storage_error is not None:
-
+            # Raise outside the sqlite exception handler so callers cannot inspect
-        return self._storage(write)
+            # a driver exception through __cause__ or __context__.
            raise storage_error
        return LedgerRecord(*key, message.event_id, message.body, message.occurred_at, expires)
    def remember(self, message: CanonicalMessage) -> LedgerRecord:
-        """Backward-compatible API wrapper outside the runtime write path."""
+        """Backward-compatible API wrapper for callers outside the runtime path."""
        return self.append(message)
    def matches(self, message: CanonicalMessage) -> bool:
        """Whether this exact message is already durably represented."""
        message_id = normalize_message_id(message.message_id)
-
+        with closing(self._connect()) as connection, connection as db:
-        def query() -> bool:
+            row = db.execute(
-            with closing(self._connect()) as connection, connection as db:
+                "SELECT event_id, body FROM message_ledger WHERE tenant_id=? AND "
-                row = db.execute(
+                "cell_id=? AND channel=? AND channel_account_id=? AND peer_id=? "
-                    "SELECT event_id, body FROM message_ledger WHERE tenant_id=? AND "
+                "AND direction=? AND message_id=?",
-                    "cell_id=? AND channel=? AND channel_account_id=? AND peer_id=? "
+                (
-                    "AND direction=? AND message_id=?",
+                    message.tenant_id, message.cell_id, message.channel,
-                    (
+                    message.channel_account_id, message.peer_id, message.direction,
-                        message.tenant_id, message.cell_id, message.channel,
+                    message_id,
-                        message.channel_account_id, message.peer_id, message.direction,
+                ),
-                        message_id,
+            ).fetchone()
-                    ),
+        return bool(row and row["event_id"] == message.event_id and row["body"] == message.body)
                ).fetchone()
            return bool(
                row and row["event_id"] == message.event_id
                and row["body"] == message.body
            )
        return self._storage(query)
    def remove(self, record: LedgerRecord) -> None:
-        def delete() -> None:
+        """Compensate a newly appended record if a later ingress step fails."""
-            with closing(self._connect()) as connection, connection as db:
+        with closing(self._connect()) as connection, connection as db:
-                db.execute(
+            db.execute(
-                    "DELETE FROM message_ledger WHERE tenant_id=? AND cell_id=? AND "
+                "DELETE FROM message_ledger WHERE tenant_id=? AND cell_id=? AND "
-                    "channel=? AND channel_account_id=? AND peer_id=? AND direction=? "
+                "channel=? AND channel_account_id=? AND peer_id=? AND direction=? "
-                    "AND message_id=? AND event_id=?",
+                "AND message_id=? AND event_id=?",
-                    (
+                (
-                        record.tenant_id, record.cell_id, record.channel,
+                    record.tenant_id, record.cell_id, record.channel,
-                        record.channel_account_id, record.peer_id, record.direction,
+                    record.channel_account_id, record.peer_id, record.direction,
-                        record.message_id, record.event_id,
+                    record.message_id, record.event_id,
-                    ),
+                ),
-                )
+            )
        self._storage(delete)
    def resolve_quote(self, message: CanonicalMessage) -> str:
        if message.quoted_body is not None:
@@ -183,22 +184,24 @@ class MessageLedger:
        if message.reply_to_message_id is None:
            raise ScopeError("quoted_text_unavailable")
        reply_id = normalize_message_id(
-            message.reply_to_message_id, field_name="reply_to_message_id"
+            message.reply_to_message_id,
            field_name="reply_to_message_id",
        )
-
+        with closing(self._connect()) as connection, connection as db:
-        def query() -> list[sqlite3.Row]:
+            rows = db.execute(
-            with closing(self._connect()) as connection, connection as db:
+                "SELECT direction, body, event_id FROM message_ledger "
-                return db.execute(
+                "WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
-                    "SELECT direction, body, event_id FROM message_ledger "
+                "AND peer_id=? AND message_id=? AND expires_at>?",
-                    "WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
+                (
-                    "AND peer_id=? AND message_id=? AND expires_at>?",
+                    message.tenant_id,
-                    (
+                    message.cell_id,
-                        message.tenant_id, message.cell_id, message.channel,
+                    message.channel,
-                        message.channel_account_id, message.peer_id, reply_id, isoformat(),
+                    message.channel_account_id,
-                    ),
+                    message.peer_id,
-                ).fetchall()
+                    reply_id,
-
+                    isoformat(),
-        rows = self._storage(query)
+                ),
            ).fetchall()
        if len(rows) == 1:
            return str(rows[0]["body"])
        if len(rows) > 1:
@@ -206,22 +209,16 @@ class MessageLedger:
        raise ScopeError("quoted_text_unavailable")
    def purge_expired(self) -> int:
-        def purge() -> int:
+        with closing(self._connect()) as connection, connection as db:
-            with closing(self._connect()) as connection, connection as db:
+            cursor = db.execute(
-                cursor = db.execute(
+                "DELETE FROM message_ledger WHERE expires_at<=?", (isoformat(),)
-                    "DELETE FROM message_ledger WHERE expires_at<=?", (isoformat(),)
+            )
-                )
+            return cursor.rowcount
                return cursor.rowcount
        return self._storage(purge)
    def integrity_check(self) -> str:
-        def check() -> str:
+        with closing(self._connect()) as connection, connection as db:
-            with closing(self._connect()) as connection, connection as db:
+            return str(db.execute("PRAGMA integrity_check").fetchone()[0])
                return str(db.execute("PRAGMA integrity_check").fetchone()[0])
        return self._storage(check)
    def count(self) -> int:
-        def query() -> int:
+        with closing(self._connect()) as connection, connection as db:
-            with closing(self._connect()) as connection, connection as db:
+            return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])
                return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])
        return self._storage(query)
@@ -1,10 +1,14 @@
-"""Tenant and Cell scoped, value-free secret reference model."""
+"""Tenant and Cell scoped secret reference model.
 This module never stores secret values. It stores ownership and retrieval scope
 metadata for integration with a real secret broker.
 """
 from __future__ import annotations
 from dataclasses import dataclass
-from typing import Any, Literal
+from typing import Literal
-from .tenant_context import ScopeError, ValidationError, require_ident
+from .tenant_context import ValidationError, require_ident
@dataclass(frozen=True)
@@ -18,44 +22,12 @@ class SecretRef:
    def __post_init__(self) -> None:
        for name in ("secret_id", "cell_id", "version"):
-            value = getattr(self, name)
+            require_ident(name, getattr(self, name))
-            if require_ident(name, value) != value:
+        require_ident("tenant_id", self.tenant_id, tenant=True)
                raise ValidationError(f"{name} must be canonical")
        if require_ident("tenant_id", self.tenant_id, tenant=True) != self.tenant_id:
            raise ValidationError("tenant_id must be canonical")
        if self.kind not in {"channel", "model", "tool", "runtime"}:
            raise ValidationError("invalid secret kind")
-        self._validate_broker_path()
+        if not self.broker_path.startswith("secret://"):
    def _validate_broker_path(self) -> None:
        if not isinstance(self.broker_path, str) or not self.broker_path.startswith("secret://"):
            raise ValidationError("secret broker path required")
        suffix = self.broker_path[len("secret://"):]
        if (
            not suffix or "?" in suffix or "#" in suffix or chr(92) in suffix
            or any(part in {"", ".", ".."} for part in suffix.split("/"))
        ):
            raise ValidationError("invalid secret broker path")
        for part in suffix.split("/"):
            if require_ident("secret broker path segment", part) != part:
                raise ValidationError("invalid secret broker path")
    def can_be_used_by(self, *, tenant_id: str, cell_id: str) -> bool:
        return self.tenant_id == tenant_id and self.cell_id == cell_id
    def resolve(self, broker: Any) -> str:
        """Resolve through one explicit broker boundary without leaking provider errors."""
        failure: ScopeError | None = None
        value: object = None
        try:
            reader = getattr(broker, "read_secret")
            if not callable(reader):
                raise TypeError("invalid broker")
            value = reader(self)
            if not isinstance(value, str) or not value:
                raise ValueError("invalid broker result")
        except Exception:
            failure = ScopeError("secret broker unavailable")
        if failure is not None:
            raise failure
        return value
@@ -1,39 +0,0 @@
 # Personal Weixin v1 security contract
 Status: normative design contract for R10 B6/B7. It authorizes no runtime or production change.
 ## Trust boundaries and ownership
 The browser is untrusted. The Portal authenticates the human user and resolves tenant identity from the server session. The Portal adapter is least-privileged: it may request account-scoped operations but must not store provider credentials, tokens, raw iLink state, or OpenClaw protocol state. OpenClaw owns provider credentials and protocol state. The Gateway transport authenticator, not an event-body boolean or identity field, establishes whether an inbound event is authentic.
 Every operation is keyed by server-resolved `(tenantId, accountId)`. Entitlements are allow-only, server-owned records. A default account is display preference, never authorization. The browser-visible `accountId` is an untrusted reference to a server-preallocated account slot: the server MUST resolve it through the authenticated user's tenant and current entitlement on list, readiness, login start, login wait, status, inbound challenge, canary, and rollback checks. Start and every wait poll re-run that resolution. Missing, arbitrary, stale, cross-tenant, or no-longer-entitled values fail closed before account existence or state is disclosed; there is no default/first/configured-account fallback.
 ## Login state and secret handling
 `sessionKey` is a required, nonempty opaque value returned by OpenClaw. The adapter MUST forward its exact bytes unchanged. It MUST NOT trim, normalize, case-fold, derive, regenerate, alias, or substitute the value. A missing or empty `sessionKey` fails closed and MUST NOT fall back to `accountId` or any identifier. Durable state stores only a one-way session-key digest where correlation is required; raw `sessionKey` is not persisted.
 `verifyCode` is accepted only as an authenticated, CSRF-protected wait POST test/input field, matches 4-8 decimal digits, is forwarded once, and is discarded. `challengeCode` is accepted only as an authenticated Gateway event test/input field, is compared through a one-way digest, and is discarded. Neither secret may be persisted in raw form or appear in URLs, responses, logs, traces, metrics labels, analytics, audit payloads, exceptions, or background-job arguments. Golden fixtures containing these values are synthetic test vectors only and MUST NOT be copied into production telemetry or responses.
 QR data URLs are transient no-store response data with a 1 MiB bound and MUST NOT be logged or persisted. Canonical login outcomes are `qr_pending`, `need_verifycode`, `invalid_verifycode`, `expired`, `failed`, `already_connected`, and `confirmed`. Any unknown upstream status maps fail-closed to canonical `failed` with `errorCode=unknown_upstream_status`; raw unknown status is bounded evidence only. Upstream `binded_redirect` maps only to `already_connected` with `connected=false`, `alreadyConnected=true`, and `terminal=true`, never to `confirmed`.
 ## Canonical readiness
 Readiness is one live, authenticated snapshot from the Gateway/OpenClaw boundary with one unified response shape for `ready`, `not_ready`, and `error`. It MUST include Gateway reachability/error, exact plugin id/version/digest, selected account identity, configured/enabled/running/last-error state, current tenant/account entitlement, reasons, observation time, and freshness deadline. `ready=true` is legal only when every required field is present, the snapshot is still fresh, expected plugin version/digest match, the selected account is entitled/configured/enabled/running with null error, and the Gateway is reachable. Missing fields, an old snapshot, digest mismatch, no entitlement, not-running state, unauthenticated evidence, or any contradiction returns `ready=false` with `code=not_ready` or `code=error` and at least one stable reason. Cached, synthetic, file-only, default-account, or service-up-only evidence cannot produce ready.
 ## Authenticated inbound-event challenge
 The event body contains exactly `eventId`, `accountId`, `senderId`, `chatId`, `challengeCode`, and `observedAt`. It does not and cannot assert authentication, tenant id, Portal user id, or session key. Those values are absent by contract. Transport authentication MUST cover HTTP method, path, timestamp header, nonce header, and raw body digest; enforce a bounded clock window; durably reserve nonce/event replay keys for at least that window; compare signatures in constant time; and complete before body values reach logs or mutable business state.
 After transport authentication, the server hashes the ephemeral `challengeCode` and resolves the pending durable challenge. That server-created record already owns tenant id, Portal user id, entitled account id, session-key digest, issued/expiry/consumed times, and state (`issued`, `consumed`, or `expired`). The body `accountId` must exactly match the record and current entitlement; it never supplies tenant/user/session authority. The first valid authenticated event atomically binds its provider `senderId` and `chatId`.
 Issuance and consumption MUST run in a database transaction with row locking or an equivalent compare-and-swap primitive. `issued -> consumed` is single-use and atomic. An exact replay of the same authenticated event is idempotent. A changed event payload, sender, chat, account, durable tenant/user/account/session binding, expired or already-consumed record, duplicate event with different payload, missing entitlement, or unauthenticated transport fails closed with zero mutation. Raw challenge codes and raw session keys are never stored.
 ## Unsupported unbind
 End-to-end unbind is unsupported in R10. Every public or internal unbind entry point MUST return the exact schema response with HTTP 501, `ok=false`, `supported=false`, `code=unbind_not_supported`, and `mutationCount=0`. The 501 decision occurs before token deletion, account update, session or challenge mutation, Gateway/OpenClaw call, business-state audit mutation, or background-job enqueue.
 ## Web, canary, and rollback controls
 Only exact B6/B7 allowlisted paths in the R10 amendment may change. Portal endpoints require authenticated Django sessions, CSRF protection for mutations, server-scoped object lookup, IDOR tests, rate limits, request-size limits, `Cache-Control: no-store`, and structured redaction of session keys, verification codes, challenge codes, QR data, tokens, and provider payloads. Stable public errors never expose upstream bodies or cross-tenant existence.
 The canary uses one explicitly entitled tenant/account pair and proves restart, concurrent waits, atomic challenge consumption, exact replay, altered replay denial, cross-tenant denial, all canonical statuses/readiness shapes, and zero-mutation 501 behavior. Promotion requires byte-identical artifacts and digests from the verified canary. Rollback restores the immediately preceding byte-identical artifacts/configuration and re-runs canonical readiness; database rollback is forbidden unless an explicitly approved manifest proves backward incompatibility and supplies a separate recovery plan.
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"tenant.account.entitlement","tenantId":"tenant-canary","accountId":"account-canary","effect":"allow","policyVersion":"policy-v1","issuedAtMs":1784068000000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":true,"code":"ready","reasons":[],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":null},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"field_missing","field":"plugin.digestSha256"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"snapshot_stale","field":"freshUntilMs"}],"observedAtMs":1784060000000,"freshUntilMs":1784060060000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"plugin_digest_mismatch","field":"plugin.digestSha256"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":false}],"ready":false,"code":"not_ready","reasons":[{"code":"account_not_entitled","field":"accounts[0].entitled"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":false,"lastError":"plugin_not_running","isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"account_not_running","field":"accounts[0].running"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":false,"errorCode":"gateway_unreachable"},"plugin":{"id":"openclaw-weixin","loaded":false,"version":null,"digestSha256":null},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":false,"lastError":"gateway_unreachable","isDefault":false,"entitled":true}],"ready":false,"code":"error","reasons":[{"code":"gateway_error","field":"gateway.reachable"}],"observedAtMs":1784068060000,"freshUntilMs":1784068060000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.start.request","accountId":"account-canary","force":false,"timeoutMs":30000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.wait.request","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","verifyCode":"4321","currentQrDataUrl":"data:image/png;base64,iVBORw0KGgo=","timeoutMs":30000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"qr_pending","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":false,"needsVerifyCode":false,"qrDataUrl":"data:image/png;base64,iVBORw0KGgo="}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"need_verifycode","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":false,"needsVerifyCode":true}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"invalid_verifycode","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":false,"needsVerifyCode":true,"errorCode":"invalid_verifycode"}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"expired","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"errorCode":"login_expired"}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"failed","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"errorCode":"upstream_failed"}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"already_connected","upstreamStatus":"binded_redirect","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":true,"terminal":true,"needsVerifyCode":false}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"confirmed","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":true,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"userId":"sender-canary"}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"failed","upstreamStatus":"future_unknown_state","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"errorCode":"unknown_upstream_status"}
@@ -1 +0,0 @@
 {"eventId":"event-20260715-A1","accountId":"account-canary","senderId":"sender-canary","chatId":"chat-canary","challengeCode":"A1B2C3D4","observedAt":1784068120000}
@@ -1 +0,0 @@
 {"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.unbind.result","httpStatus":501,"ok":false,"supported":false,"code":"unbind_not_supported","mutationCount":0}
@@ -1,986 +0,0 @@
 {
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://aios.imruyi.top/contracts/personal-weixin/v1/golden-contract.schema.json",
  "title": "AIOS Personal Weixin Golden Contract v1",
  "oneOf": [
    {
      "$ref": "#/$defs/entitlement"
    },
    {
      "$ref": "#/$defs/readiness"
    },
    {
      "$ref": "#/$defs/startRequest"
    },
    {
      "$ref": "#/$defs/waitRequest"
    },
    {
      "$ref": "#/$defs/loginResult"
    },
    {
      "$ref": "#/$defs/inboundEvent"
    },
    {
      "$ref": "#/$defs/unbindResult"
    }
  ],
  "$defs": {
    "identifier": {
      "type": "string",
      "minLength": 1,
      "maxLength": 128,
      "pattern": "^[A-Za-z0-9._~:@-]+$"
    },
    "sessionKey": {
      "type": "string",
      "minLength": 1,
      "maxLength": 256,
      "pattern": "^[A-Za-z0-9._~:-]+$"
    },
    "schemaVersion": {
      "const": "aios.personal-weixin.golden.v1"
    },
    "entitlement": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "schemaVersion",
        "kind",
        "tenantId",
        "accountId",
        "effect",
        "policyVersion",
        "issuedAtMs"
      ],
      "properties": {
        "schemaVersion": {
          "$ref": "#/$defs/schemaVersion"
        },
        "kind": {
          "const": "tenant.account.entitlement"
        },
        "tenantId": {
          "$ref": "#/$defs/identifier"
        },
        "accountId": {
          "$ref": "#/$defs/identifier"
        },
        "effect": {
          "const": "allow"
        },
        "policyVersion": {
          "$ref": "#/$defs/identifier"
        },
        "issuedAtMs": {
          "type": "integer",
          "minimum": 0
        }
      }
    },
    "readinessReason": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "code",
        "field"
      ],
      "properties": {
        "code": {
          "enum": [
            "field_missing",
            "snapshot_stale",
            "plugin_digest_mismatch",
            "account_not_entitled",
            "account_not_running",
            "gateway_error"
          ]
        },
        "field": {
          "type": "string",
          "minLength": 1,
          "maxLength": 128
        }
      }
    },
    "readiness": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "schemaVersion",
        "kind",
        "tenantId",
        "selectedAccountId",
        "gateway",
        "plugin",
        "accounts",
        "ready",
        "code",
        "reasons",
        "observedAtMs",
        "freshUntilMs"
      ],
      "properties": {
        "schemaVersion": {
          "$ref": "#/$defs/schemaVersion"
        },
        "kind": {
          "const": "personal.weixin.readiness"
        },
        "tenantId": {
          "$ref": "#/$defs/identifier"
        },
        "selectedAccountId": {
          "$ref": "#/$defs/identifier"
        },
        "gateway": {
          "type": "object",
          "additionalProperties": false,
          "required": [
            "reachable",
            "errorCode"
          ],
          "properties": {
            "reachable": {
              "type": "boolean"
            },
            "errorCode": {
              "type": [
                "string",
                "null"
              ],
              "minLength": 1,
              "maxLength": 128
            }
          }
        },
        "plugin": {
          "type": "object",
          "additionalProperties": false,
          "required": [
            "id",
            "loaded",
            "version",
            "digestSha256"
          ],
          "properties": {
            "id": {
              "const": "openclaw-weixin"
            },
            "loaded": {
              "type": "boolean"
            },
            "version": {
              "type": [
                "string",
                "null"
              ],
              "minLength": 1,
              "maxLength": 64
            },
            "digestSha256": {
              "anyOf": [
                {
                  "type": "string",
                  "pattern": "^[a-f0-9]{64}$"
                },
                {
                  "type": "null"
                }
              ]
            }
          }
        },
        "accounts": {
          "type": "array",
          "minItems": 1,
          "maxItems": 100,
          "items": {
            "type": "object",
            "additionalProperties": false,
            "required": [
              "accountId",
              "name",
              "configured",
              "enabled",
              "running",
              "lastError",
              "isDefault",
              "entitled"
            ],
            "properties": {
              "accountId": {
                "$ref": "#/$defs/identifier"
              },
              "name": {
                "type": "string",
                "minLength": 1,
                "maxLength": 128
              },
              "configured": {
                "type": "boolean"
              },
              "enabled": {
                "type": "boolean"
              },
              "running": {
                "type": "boolean"
              },
              "lastError": {
                "type": [
                  "string",
                  "null"
                ],
                "minLength": 1,
                "maxLength": 256
              },
              "isDefault": {
                "type": "boolean"
              },
              "entitled": {
                "type": "boolean"
              }
            }
          }
        },
        "ready": {
          "type": "boolean"
        },
        "code": {
          "enum": [
            "ready",
            "not_ready",
            "error"
          ]
        },
        "reasons": {
          "type": "array",
          "maxItems": 32,
          "items": {
            "$ref": "#/$defs/readinessReason"
          }
        },
        "observedAtMs": {
          "type": "integer",
          "minimum": 0
        },
        "freshUntilMs": {
          "type": "integer",
          "minimum": 0
        }
      },
      "allOf": [
        {
          "if": {
            "properties": {
              "ready": {
                "const": true
              }
            },
            "required": [
              "ready"
            ]
          },
          "then": {
            "properties": {
              "code": {
                "const": "ready"
              },
              "reasons": {
                "maxItems": 0
              },
              "gateway": {
                "properties": {
                  "reachable": {
                    "const": true
                  },
                  "errorCode": {
                    "const": null
                  }
                }
              },
              "plugin": {
                "properties": {
                  "loaded": {
                    "const": true
                  },
                  "version": {
                    "type": "string"
                  },
                  "digestSha256": {
                    "type": "string",
                    "pattern": "^[a-f0-9]{64}$"
                  }
                }
              },
              "accounts": {
                "items": {
                  "properties": {
                    "configured": {
                      "const": true
                    },
                    "enabled": {
                      "const": true
                    },
                    "running": {
                      "const": true
                    },
                    "lastError": {
                      "const": null
                    },
                    "entitled": {
                      "const": true
                    }
                  }
                }
              }
            }
          },
          "else": {
            "properties": {
              "code": {
                "enum": [
                  "not_ready",
                  "error"
                ]
              },
              "reasons": {
                "minItems": 1
              }
            }
          }
        }
      ]
    },
    "startRequest": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "schemaVersion",
        "kind",
        "accountId",
        "force",
        "timeoutMs"
      ],
      "properties": {
        "schemaVersion": {
          "$ref": "#/$defs/schemaVersion"
        },
        "kind": {
          "const": "web.login.start.request"
        },
        "accountId": {
          "$ref": "#/$defs/identifier"
        },
        "force": {
          "const": false
        },
        "timeoutMs": {
          "type": "integer",
          "minimum": 1000,
          "maximum": 120000
        }
      }
    },
    "waitRequest": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "schemaVersion",
        "kind",
        "accountId",
        "sessionKey",
        "timeoutMs"
      ],
      "properties": {
        "schemaVersion": {
          "$ref": "#/$defs/schemaVersion"
        },
        "kind": {
          "const": "web.login.wait.request"
        },
        "accountId": {
          "$ref": "#/$defs/identifier"
        },
        "sessionKey": {
          "$ref": "#/$defs/sessionKey"
        },
        "verifyCode": {
          "type": "string",
          "pattern": "^[0-9]{4,8}$"
        },
        "currentQrDataUrl": {
          "type": "string",
          "minLength": 1,
          "maxLength": 1048576,
          "pattern": "^data:image/png;base64,"
        },
        "timeoutMs": {
          "type": "integer",
          "minimum": 1000,
          "maximum": 120000
        }
      }
    },
    "loginResult": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "schemaVersion",
        "kind",
        "status",
        "accountId",
        "sessionKey",
        "connected",
        "alreadyConnected",
        "terminal",
        "needsVerifyCode"
      ],
      "properties": {
        "schemaVersion": {
          "$ref": "#/$defs/schemaVersion"
        },
        "kind": {
          "const": "web.login.result"
        },
        "status": {
          "enum": [
            "qr_pending",
            "need_verifycode",
            "invalid_verifycode",
            "expired",
            "failed",
            "already_connected",
            "confirmed"
          ]
        },
        "accountId": {
          "$ref": "#/$defs/identifier"
        },
        "sessionKey": {
          "$ref": "#/$defs/sessionKey"
        },
        "connected": {
          "type": "boolean"
        },
        "alreadyConnected": {
          "type": "boolean"
        },
        "terminal": {
          "type": "boolean"
        },
        "needsVerifyCode": {
          "type": "boolean"
        },
        "qrDataUrl": {
          "type": "string",
          "minLength": 1,
          "maxLength": 1048576,
          "pattern": "^data:image/png;base64,"
        },
        "userId": {
          "$ref": "#/$defs/identifier"
        },
        "errorCode": {
          "enum": [
            "invalid_verifycode",
            "login_expired",
            "upstream_failed",
            "unknown_upstream_status"
          ]
        },
        "upstreamStatus": {
          "type": "string",
          "minLength": 1,
          "maxLength": 128
        }
      },
      "allOf": [
        {
          "if": {
            "properties": {
              "status": {
                "const": "qr_pending"
              }
            }
          },
          "then": {
            "required": [
              "qrDataUrl"
            ],
            "properties": {
              "connected": {
                "const": false
              },
              "alreadyConnected": {
                "const": false
              },
              "terminal": {
                "const": false
              },
              "needsVerifyCode": {
                "const": false
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "userId"
                  ]
                },
                {
                  "required": [
                    "errorCode"
                  ]
                },
                {
                  "required": [
                    "upstreamStatus"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "properties": {
              "status": {
                "const": "need_verifycode"
              }
            }
          },
          "then": {
            "properties": {
              "connected": {
                "const": false
              },
              "alreadyConnected": {
                "const": false
              },
              "terminal": {
                "const": false
              },
              "needsVerifyCode": {
                "const": true
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "userId"
                  ]
                },
                {
                  "required": [
                    "errorCode"
                  ]
                },
                {
                  "required": [
                    "upstreamStatus"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "properties": {
              "status": {
                "const": "invalid_verifycode"
              }
            }
          },
          "then": {
            "required": [
              "errorCode"
            ],
            "properties": {
              "errorCode": {
                "const": "invalid_verifycode"
              },
              "connected": {
                "const": false
              },
              "alreadyConnected": {
                "const": false
              },
              "terminal": {
                "const": false
              },
              "needsVerifyCode": {
                "const": true
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "userId"
                  ]
                },
                {
                  "required": [
                    "upstreamStatus"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "properties": {
              "status": {
                "const": "expired"
              }
            }
          },
          "then": {
            "required": [
              "errorCode"
            ],
            "properties": {
              "errorCode": {
                "const": "login_expired"
              },
              "connected": {
                "const": false
              },
              "alreadyConnected": {
                "const": false
              },
              "terminal": {
                "const": true
              },
              "needsVerifyCode": {
                "const": false
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "userId"
                  ]
                },
                {
                  "required": [
                    "upstreamStatus"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "properties": {
              "status": {
                "const": "failed"
              }
            }
          },
          "then": {
            "required": [
              "errorCode"
            ],
            "properties": {
              "connected": {
                "const": false
              },
              "alreadyConnected": {
                "const": false
              },
              "terminal": {
                "const": true
              },
              "needsVerifyCode": {
                "const": false
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "userId"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "required": [
              "upstreamStatus"
            ],
            "properties": {
              "status": {
                "const": "failed"
              }
            }
          },
          "then": {
            "properties": {
              "errorCode": {
                "const": "unknown_upstream_status"
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "userId"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "required": [
              "status",
              "errorCode"
            ],
            "properties": {
              "status": {
                "const": "failed"
              },
              "errorCode": {
                "const": "unknown_upstream_status"
              }
            }
          },
          "then": {
            "required": [
              "upstreamStatus"
            ]
          }
        },
        {
          "if": {
            "required": [
              "status",
              "errorCode"
            ],
            "properties": {
              "status": {
                "const": "failed"
              },
              "errorCode": {
                "const": "upstream_failed"
              }
            }
          },
          "then": {
            "not": {
              "required": [
                "upstreamStatus"
              ]
            }
          }
        },
        {
          "if": {
            "properties": {
              "status": {
                "const": "already_connected"
              }
            }
          },
          "then": {
            "required": [
              "upstreamStatus"
            ],
            "properties": {
              "upstreamStatus": {
                "const": "binded_redirect"
              },
              "connected": {
                "const": false
              },
              "alreadyConnected": {
                "const": true
              },
              "terminal": {
                "const": true
              },
              "needsVerifyCode": {
                "const": false
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "userId"
                  ]
                },
                {
                  "required": [
                    "errorCode"
                  ]
                }
              ]
            }
          }
        },
        {
          "if": {
            "properties": {
              "status": {
                "const": "confirmed"
              }
            }
          },
          "then": {
            "required": [
              "userId"
            ],
            "properties": {
              "connected": {
                "const": true
              },
              "alreadyConnected": {
                "const": false
              },
              "terminal": {
                "const": true
              },
              "needsVerifyCode": {
                "const": false
              }
            },
            "not": {
              "anyOf": [
                {
                  "required": [
                    "qrDataUrl"
                  ]
                },
                {
                  "required": [
                    "errorCode"
                  ]
                },
                {
                  "required": [
                    "upstreamStatus"
                  ]
                }
              ]
            }
          }
        }
      ]
    },
    "inboundEvent": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "eventId",
        "accountId",
        "senderId",
        "chatId",
        "challengeCode",
        "observedAt"
      ],
      "properties": {
        "eventId": {
          "$ref": "#/$defs/identifier"
        },
        "accountId": {
          "$ref": "#/$defs/identifier"
        },
        "senderId": {
          "$ref": "#/$defs/identifier"
        },
        "chatId": {
          "$ref": "#/$defs/identifier"
        },
        "challengeCode": {
          "type": "string",
          "minLength": 6,
          "maxLength": 64,
          "pattern": "^[A-Z0-9]+$"
        },
        "observedAt": {
          "type": "integer",
          "minimum": 0
        }
      }
    },
    "unbindResult": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "schemaVersion",
        "kind",
        "httpStatus",
        "ok",
        "supported",
        "code",
        "mutationCount"
      ],
      "properties": {
        "schemaVersion": {
          "$ref": "#/$defs/schemaVersion"
        },
        "kind": {
          "const": "web.unbind.result"
        },
        "httpStatus": {
          "const": 501
        },
        "ok": {
          "const": false
        },
        "supported": {
          "const": false
        },
        "code": {
          "const": "unbind_not_supported"
        },
        "mutationCount": {
          "const": 0
        }
      }
    }
  }
 }
@@ -1,92 +0,0 @@
 # 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`.
@@ -1,5 +0,0 @@
 #!/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 "$@"
@@ -1,21 +0,0 @@
 {
  "$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"}
  }
 }
@@ -1,5 +0,0 @@
 #!/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 "$@"
@@ -1,5 +0,0 @@
 #!/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 "$@"
@@ -1,5 +0,0 @@
 #!/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 "$@"
@@ -1,50 +0,0 @@
 {
  "$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"
    }
  }
 }
@@ -1,11 +0,0 @@
 {
  "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"
 }
@@ -1,5 +0,0 @@
 #!/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 "$@"
@@ -1,63 +0,0 @@
 [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
@@ -1,2 +0,0 @@
 g aios-spark-runner -
 u aios-spark-runner - "AIOS isolated runner" /var/lib/aios-spark-runner /usr/sbin/nologin
@@ -1,3 +0,0 @@
 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 - -
@@ -18,14 +18,12 @@
 - All OpenClaw loader work uses an isolated temporary or candidate config/state directory.
 - `ACCEPTED` is terminal. Legacy handling is reachable only when bridge mode is explicitly `disabled`.
 - Never claim exactly-once Hermes delivery or use time/approximation as message authority.
 - D0 is a non-blocker documentation correction. Its reviewed commit is the R9 `START`; it does not consume a blocker label or relax any B3/B4/B5 evidence gate.
 - B2 is both the canonical-package blocker and the executable completeness gate for Ledger, Memory, Secret, Audit, Hermes Profile, Cell Manager/CLI, and Fabric. Existing correct behavior is recorded as characterization; a production edit requires an observed contract RED first.
 After docs review, initialize the stable per-worktree ledger from the then-current tip rather than a hardcoded SHA:
 ```bash
 test -z "$(git status --porcelain)"
-test "$(git branch --show-current)" = "integration/v2r1-runtime-20260715-r9"
+test "$(git branch --show-current)" = "integration/v2r1-runtime-20260714"
 start_sha="$(git rev-parse HEAD)"
 start_parent="$(git rev-parse HEAD^)"
 start_tree="$(git rev-parse HEAD^{tree})"
@@ -172,24 +170,8 @@ test "$tree" = "$(git rev-parse HEAD^{tree})"
 - Create: `tests/v2r1/test_ingress_service.py`
 - Create: `tests/test_v2r1_package_identity.py`
 - Modify: the exact 23 Appendix A test files, and no unlisted legacy import file
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/message_ledger.py`
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/memory_scope.py`
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/secret_scope.py`
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/audit.py`
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/hermes_profile.py`
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/cell_manager.py`
 - Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/hermes_fabric.py`
 - Expand: `tests/v2r1/test_message_ledger_append.py`
 - Create: `tests/v2r1/test_memory_scope.py`
 - Create: `tests/v2r1/test_secret_scope.py`
 - Create: `tests/v2r1/test_audit.py`
 - Create: `tests/v2r1/test_hermes_profile.py`
 - Create: `tests/v2r1/test_cell_manager.py`
 - Expand: `tests/v2r1/test_hermes_fabric.py`
-**Exact contract:** V2R1 imports only as `aios_v2r1`; no `active_runtime` shim/alias/namespace extension/copied tree. Wheel includes `aios_v2r1`, excludes `active_runtime` and legacy `app`. Legacy repository tests change `active_runtime.app.*` to `app.*` and run with `PYTHONPATH='active-runtime:active-runtime/app'`. Mutation routes return bounded `SERVICE_NOT_CONFIGURED` until B3/B4 processors exist. Existing cell CLI stays compatible and `serve` is separate. B2 changes exactly 43 paths: the original 29 package/import paths plus the seven named runtime modules and seven named matrix-test paths above.
+**Exact contract:** V2R1 imports only as `aios_v2r1`; no `active_runtime` shim/alias/namespace extension/copied tree. Wheel includes `aios_v2r1`, excludes `active_runtime` and legacy `app`. Legacy repository tests change `active_runtime.app.*` to `app.*` and run with `PYTHONPATH='active-runtime:active-runtime/app'`. Mutation routes return bounded `SERVICE_NOT_CONFIGURED` until B3/B4 processors exist. Existing cell CLI stays compatible and `serve` is separate.
 **Executable completeness matrix:** Message Ledger proves exact-ID query with no fallback, idempotency, collision conflict, tenant/cell/direction isolation, expiry/purge, restart, concurrency, and sanitized SQLite failures. Memory proves every namespace, cross-scope denial, shared-artifact read-only access, qualified keys, traversal rejection, and disabled search. Secret proves reference/path/scope/kind/version/identifier validation, no value leakage, and fail-closed broker boundaries. Audit proves genesis, tamper/delete/reorder detection, Unicode canonicalization, concurrency, restart, validation, tenant isolation, and sanitized SQLite failures. Hermes Profile proves exact environment, absolute roots, `0700` modes, traversal/symlink denial, scope, idempotent prepare, and no inherited environment. Cell Manager/CLI proves one cell per tenant, an explicit state graph, atomic/concurrent persistence, corrupt-registry fail-closed behavior, stable CLI exit codes/JSON, and restart. Fabric proves claimed jobs return `running` consistently with durable state, tenant/cell/profile isolation, one concurrent claimant, rejection of non-claimable states, restart, and sanitized SQLite failures.
 - [ ] Rehydrate the stable blocker ledger using the global snippet and require its latest label is `B1`.
 - [ ] Rehydrate the exact explorer manifest path and compare sorted actual/expected lists between exact marker lines. The marker names are assembled from split literals so this command cannot match itself:
@@ -214,21 +196,16 @@ cmp -s "$actual" "$expected"
 Expected: exact match to 23 paths. Drift requires another docs review.
- [ ] Add package/service/CLI tests and the seven-component matrix before any production repair. Run each focused case and record the exact command plus its expected contract failure. If current behavior already satisfies the contract, record it as characterization and do not manufacture a failure or edit that production module.
+- [ ] Add package/service/CLI tests; run initial RED:
 - [ ] Run the aggregate initial RED/characterization command:
 ```bash
 PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' \
-  python3 -m pytest tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py \
+  python3 -m pytest tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py -q
    tests/v2r1/test_message_ledger_append.py tests/v2r1/test_memory_scope.py \
    tests/v2r1/test_secret_scope.py tests/v2r1/test_audit.py \
    tests/v2r1/test_hermes_profile.py tests/v2r1/test_cell_manager.py \
    tests/v2r1/test_hermes_fabric.py -q
 ```
-Expected: package metadata, ingress service, and identity tests begin RED because their implementation is absent. Each matrix case is either a recorded contract RED or a recorded characterization; no production repair precedes its corresponding RED.
+Expected RED: package metadata, ingress service, and identity tests do not exist.
- [ ] Implement package/disabled processor boundary and replace imports in exactly Appendix A. Apply only the minimum runtime repairs demonstrated by matrix REDs; do not delete or weaken a matrix test. Prove no shim/old import:
+- [ ] Implement package/disabled processor boundary and replace imports in exactly Appendix A. Prove no shim/old import:
 ```bash
 test ! -e active_runtime
@@ -238,15 +215,11 @@ test -z "$(git grep -l -E '^(from|import) active_runtime\.app' -- tests || true)
 - [ ] Run full collect and full suite:
 ```bash
 PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' \
  python3 -m pytest tests/v2r1 tests/integration tests/test_v2r1_package_identity.py -q
 PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' python3 -m pytest --collect-only -q
 PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' python3 -m pytest -q
 PYTHONDONTWRITEBYTECODE=1 python3 -m compileall -q active-runtime/app/aios_v2r1
 (cd openclaw/plugins/aios-tenant-boundary && npm test)
 ```
-Expected GREEN: the focused carried suite reports at least `160 passed`, including every matrix bullet; full collection has zero import errors; the complete pytest suite, 15 Node tests, and Python compileall pass. Re-run B1's real isolated default/opt-in loader sequence and require the same hook evidence. Pre-existing failures still block B2.
+Expected GREEN: zero collection/import errors and complete suite passes; pre-existing failures still block B2.
 - [ ] Build and inspect isolated wheel:
@@ -268,29 +241,22 @@ PY
 Expected GREEN: one wheel, one V2R1 namespace, installed identity `aios_v2r1`.
- [ ] Stage the exact 43 paths (six package paths, Appendix A's 23 import paths, seven matrix runtime modules, and seven matrix-test paths), commit, append ledger:
+- [ ] Stage six named paths plus exact explorer list, commit, append ledger:
 ```bash
 actual="$(git rev-parse --git-path v2r1-active-runtime-imports.actual)"
 git add active-runtime/pyproject.toml active-runtime/app/aios_v2r1/__init__.py \
  active-runtime/app/aios_v2r1/ingress_service.py active-runtime/app/aios_v2r1/cli.py \
-  tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py \
+  tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py
  active-runtime/app/aios_v2r1/message_ledger.py active-runtime/app/aios_v2r1/memory_scope.py \
  active-runtime/app/aios_v2r1/secret_scope.py active-runtime/app/aios_v2r1/audit.py \
  active-runtime/app/aios_v2r1/hermes_profile.py active-runtime/app/aios_v2r1/cell_manager.py \
  active-runtime/app/aios_v2r1/hermes_fabric.py tests/v2r1/test_message_ledger_append.py \
  tests/v2r1/test_memory_scope.py tests/v2r1/test_secret_scope.py tests/v2r1/test_audit.py \
  tests/v2r1/test_hermes_profile.py tests/v2r1/test_cell_manager.py tests/v2r1/test_hermes_fabric.py
 while IFS= read -r path; do git add -- "$path"; done <"$actual"
 git commit -m "feat(v2r1): package canonical authenticated ingress"
 ```
 ### Review B2
- [ ] Require `git diff --check`, exactly 43 paths (the original 29 plus the exact 14 matrix paths), exact-list/no-shim checks, focused matrix/carried tests, full collect/full pytest, Node 15, real loader default/opt-in, compileall, and wheel identity.
+- [ ] Require `git diff --check`, exactly 29 paths (six named plus Appendix A), exact-list/no-shim checks, full collect/full pytest, and wheel identity.
 - [ ] Require B2 parent equals B1 in ledger.
 - [ ] Prove mutation routes remain fail-closed until authenticated processors are installed.
 - [ ] Prove every production matrix edit has a recorded pre-edit RED and every already-correct case is recorded as characterization. The partial R9 ledger must contain exactly `START,B1,B2`; B3/B4/B5 remain absent.
 - [ ] After all prior B2 review checks pass, append B2 and verify the recorded commit identity:
 ```bash
@@ -1,188 +0,0 @@
 # V2R1 R10 personal Weixin release-plan amendment
 Status: reviewed design-only amendment. It adds D1, B6, and B7 to the R9 linear release plan. It does not authorize runtime, candidate, production, baseline, tag, or Fleet mutation.
 Authority remains `docs/superpowers/plans/2026-07-14-v2r1-runtime-release-plan.md` at B2 tip `68d27879103990aac692207a8512b3725c6d0d41`. R9 `START` is `a9a0b4ae396e33c9bfc35a5f5694e5e354c3a47c`; accepted B1 is `57103c11d21192a50febe846d30cee6fc5da4ac1`; accepted B2 is `68d27879103990aac692207a8512b3725c6d0d41`. Their trees and path sets are immutable. This amendment does not relabel, squash, reorder, or absorb any of the original five blockers.
 ## Exact linear ledger
 The only valid ancestry is:
 ```text
 START -> B1 -> B2 -> D1 -> B3 -> B4 -> B5 -> B6 -> B7
 ```
 `D1` is a non-blocker design dependency whose parent is exactly B2. `B3` parent is D1; every later blocker parent is the immediately preceding row. `START..B7` contains exactly eight non-merge commits: B1, B2, D1, B3, B4, B5, B6, B7. B1-B5 remain five distinct original blocker commits; B6 and B7 are distinct new blocker commits. No cherry-pick collapse, squash, fixup, merge commit, history rewrite, or force push is allowed.
 B3/B4 retain every R9 evidence gate. In particular, unresolved `EVIDENCE_REQUIRED` for the exact provider `client_id -> message_id` binding stops B4 with no implementation, commit, or ledger append. B5 can land only after accepted B4. Even after B5 lands, every forward `stage`, `verify-stage`, or `promote` operation remains blocked until accepted B6 and B7 commits exist and all gates below pass.
 ## D1 design-only commit
 D1 changes exactly these 23 paths and no others:
 ```text
 docs/superpowers/plans/2026-07-15-v2r1-personal-weixin-r10-release-plan-amendment.md
 contracts/personal-weixin/v1/SECURITY.md
 contracts/personal-weixin/v1/golden-contract.schema.json
 contracts/personal-weixin/v1/fixtures/01-tenant-account-entitlement.json
 contracts/personal-weixin/v1/fixtures/02-readiness-ready.json
 contracts/personal-weixin/v1/fixtures/03-readiness-field-missing.json
 contracts/personal-weixin/v1/fixtures/04-readiness-stale.json
 contracts/personal-weixin/v1/fixtures/05-readiness-plugin-digest-mismatch.json
 contracts/personal-weixin/v1/fixtures/06-readiness-account-not-entitled.json
 contracts/personal-weixin/v1/fixtures/07-readiness-account-not-running.json
 contracts/personal-weixin/v1/fixtures/08-readiness-error.json
 contracts/personal-weixin/v1/fixtures/09-web-login-start-request.json
 contracts/personal-weixin/v1/fixtures/10-web-login-wait-request.json
 contracts/personal-weixin/v1/fixtures/11-qr-pending.json
 contracts/personal-weixin/v1/fixtures/12-need-verify-code.json
 contracts/personal-weixin/v1/fixtures/13-invalid-verify-code.json
 contracts/personal-weixin/v1/fixtures/14-expired.json
 contracts/personal-weixin/v1/fixtures/15-failed.json
 contracts/personal-weixin/v1/fixtures/16-already-connected.json
 contracts/personal-weixin/v1/fixtures/17-confirmed.json
 contracts/personal-weixin/v1/fixtures/18-unknown-upstream-fail-closed.json
 contracts/personal-weixin/v1/fixtures/19-authenticated-inbound-event.json
 contracts/personal-weixin/v1/fixtures/20-unbind-unsupported.json
 ```
 D1 validation requires JSON parse, Draft 2020-12 schema validation of every fixture, negative mutation tests, exact 23-path comparison, `git diff --check`, placeholder/secret scan, and confirmation that no runtime/deploy/release-active path changed.
 ## Preserved R9 blocker allowlists
 B1 and B2 are already accepted at the immutable SHAs above. Their actual commit path sets, parent SHAs, and trees are the allowlists; compare with `git diff-tree --no-commit-id --name-only -r <sha>` and reject any mismatch. B1 has six actual paths under `openclaw/plugins/aios-tenant-boundary/`; B2 has exactly 43 actual package/import/matrix paths. No later blocker may amend them.
 B3 changes exactly these 15 R9 paths:
 ```text
 active-runtime/app/aios_v2r1/identity.py
 active-runtime/app/aios_v2r1/tenant_context.py
 active-runtime/app/aios_v2r1/ingress_service.py
 active-runtime/app/aios_v2r1/bridge_auth.py
 active-runtime/app/aios_v2r1/weixin_projection.py
 fixes/patch_openclaw_weixin_v2r1_ingress.py
 tests/v2r1/test_bridge_auth.py
 tests/v2r1/test_weixin_projection.py
 tests/integration/test_v2r1_bridge_transport.py
 tests/test_weixin_v2r1_ingress_bridge.py
 tests/fixtures/weixin_v2r1/text.json
 tests/fixtures/weixin_v2r1/quoted-text.json
 tests/fixtures/weixin_v2r1/image-with-text.json
 tests/fixtures/weixin_v2r1/unresolved-reference.json
 release/v2r1c0-rc1/weixin-bridge-evidence.json
 ```
 B4 changes exactly these 19 R9 paths and remains evidence-stopped until both R9 evidence files are accepted:
 ```text
 active-runtime/app/aios_v2r1/outbox.py
 active-runtime/app/aios_v2r1/hermes_http.py
 active-runtime/app/aios_v2r1/hermes_worker.py
 active-runtime/app/aios_v2r1/outbound_receipt.py
 active-runtime/app/aios_v2r1/ingress_service.py
 active-runtime/app/aios_v2r1/message_ledger.py
 tests/v2r1/test_outbox.py
 tests/v2r1/test_hermes_http.py
 tests/v2r1/test_outbound_receipt.py
 tests/integration/test_v2r1_delivery_flow.py
 fixes/patch_openclaw_weixin_outbound_quotes.py
 fixes/patch_openclaw_weixin_outbound_receipts.py
 openclaw/plugins/aios-tenant-boundary/dist/receipt-client.js
 openclaw/plugins/aios-tenant-boundary/dist/outbound-receipt-journal.js
 openclaw/plugins/aios-tenant-boundary/test/receipt-client.test.js
 tests/test_weixin_outbound_quote_timestamp.py
 tests/test_weixin_outbound_receipt_boundary.py
 release/v2r1c0-rc1/hermes-http-contract-evidence.json
 release/v2r1c0-rc1/weixin-outbound-binding-evidence.json
 ```
 B5 changes exactly these 14 R9 paths:
 ```text
 deploy/v2r1/release_ctl.py
 deploy/v2r1/capture_live_preflight.py
 deploy/v2r1/install.sh
 deploy/v2r1/validate.sh
 deploy/v2r1/healthcheck.sh
 deploy/v2r1/rollback.sh
 deploy/v2r1/aios-v2r1-candidate-gateway.service
 deploy/v2r1/aios-v2r1-production-gateway.service
 deploy/v2r1/aios-v2r1-ingress.service
 deploy/v2r1/aios-v2r1-hermes-worker.service
 deploy/v2r1/openclaw-candidate.json5
 release/v2r1c0-rc1/FLEET_UNSUPPORTED
 tests/deploy/test_v2r1_release_contract.py
 release/v2r1c0-rc1/RELEASE.md
 ```
 ## B6: evidence-bound OpenClaw personal-Weixin candidate
 B6 changes exactly these 13 paths:
 ```text
 openclaw/candidates/personal-weixin-r10/README.md
 openclaw/candidates/personal-weixin-r10/manifest.json
 openclaw/candidates/personal-weixin-r10/src/golden-contract.js
 openclaw/candidates/personal-weixin-r10/src/gateway-inbound-auth.js
 openclaw/candidates/personal-weixin-r10/plugin/openclaw.plugin.json
 openclaw/candidates/personal-weixin-r10/plugin/index.js
 fixes/patch_openclaw_personal_weixin_login.py
 fixes/patch_openclaw_personal_weixin_gateway_auth.py
 openclaw/candidates/personal-weixin-r10/test/golden-contract.test.js
 openclaw/candidates/personal-weixin-r10/test/gateway-auth.test.js
 tests/personal_weixin/test_golden_fixtures.py
 tests/integration/test_personal_weixin_gateway_contract.py
 release/v2r1c0-rc1/personal-weixin-b6-evidence.json
 ```
 Before B6 code, evidence records exact installed upstream source commit, every patch target path/digest, plugin loader version/digest, Gateway auth owner, credential files, and candidate-only state/config. Stop on inference or drift. Tests are RED before implementation and prove: required nonempty opaque `sessionKey`; no account fallback; exact canonical statuses and readiness shapes; unknown status fail-closed; 4-8 digit one-shot `verifyCode` with no persistence/log/response; exact event body with no authentication/tenant/user/session assertions; transport signature/skew/nonce/replay checks; exact 501 zero mutation; byte-idempotent patches; build/runtime inspect in isolated candidate state. The B6 evidence JSON is produced only after these gates pass.
 ## B7: tenant-scoped Portal adapter and durable challenge
 B7 changes exactly these 13 paths:
 ```text
 active-runtime/app/aios_v2r1/personal_weixin/official_login.py
 active-runtime/app/aios_v2r1/personal_weixin/adapter.py
 portal/personal_weixin/views.py
 portal/personal_weixin/urls.py
 portal/personal_weixin/services.py
 portal/personal_weixin/templates/personal_weixin/bind.html
 tests/personal_weixin/test_adapter.py
 tests/personal_weixin/test_portal_views.py
 tests/integration/test_personal_weixin_golden_flow.py
 release/v2r1c0-rc1/personal-weixin-source-baseline.json
 release/v2r1c0-rc1/personal-weixin-source-baseline.sha256
 release/v2r1c0-rc1/personal-weixin-canary-contract.json
 release/v2r1c0-rc1/personal-weixin-rollback-contract.json
 ```
 Tests are RED before B7 implementation and prove: Django authentication/CSRF; server-session tenant resolution; browser `accountId` treated only as a preallocated-slot reference and current entitlement rechecked on every list/readiness/start/wait/status; exact `sessionKey` forwarding without fallback; live canonical readiness; authenticated Gateway transport context separate from the six-field event body; durable challenge record owning tenant/user/account/session digest; hash-only secret storage; transactional/CAS single consumption; exact replay idempotence; altered replay/concurrency/expiry/cross-tenant denial with zero mutation; no provider token/raw iLink/verify code/challenge code persistence or leakage; size/rate/no-store controls; restart; and unbind 501 before all mutations/calls/jobs.
 ## Canonical contract gates
 `sessionKey` is nonempty and exact. Canonical results are:
 ```text
 qr_pending         connected=false alreadyConnected=false terminal=false needsVerifyCode=false
 need_verifycode    connected=false alreadyConnected=false terminal=false needsVerifyCode=true
 invalid_verifycode connected=false alreadyConnected=false terminal=false needsVerifyCode=true
 expired            connected=false alreadyConnected=false terminal=true  needsVerifyCode=false
 failed             connected=false alreadyConnected=false terminal=true  needsVerifyCode=false
 already_connected  connected=false alreadyConnected=true  terminal=true  needsVerifyCode=false
 confirmed           connected=true  alreadyConnected=false terminal=true  needsVerifyCode=false
 ```
 `binded_redirect` maps to `already_connected`. An unknown upstream status maps to canonical `failed/errorCode=unknown_upstream_status`. Readiness always uses the same complete shape: missing fields, stale evidence, plugin digest mismatch, absent entitlement, not-running account, and Gateway error all produce `ready=false`; only a live authenticated all-green snapshot produces `ready=true/code=ready`.
 The inbound event body is exactly `eventId/accountId/senderId/chatId/challengeCode/observedAt`; transport authentication and server-resolved tenant/user/session context are not body fields. Fixture secrets are test inputs only. Raw `verifyCode` and `challengeCode` never enter persistence, logs, responses, URLs, analytics, traces, metrics labels, or job arguments. Raw `sessionKey` appears only where the golden login request/result contract requires it; it is never persisted or logged and is forwarded byte-exact without fallback.
 ## Merge, canary, promotion, and rollback gates
 For each D1/B3-B7 row: require the exact parent row SHA, one non-merge commit, exact sorted path allowlist, clean `diff --check`, focused and carried tests, reviewer acceptance, then append `label/SHA/parent/tree` to the worktree-local ledger. No row may start from an unaccepted predecessor.
 After B7, rebuild once from the accepted B7 tree. Stage one isolated candidate using B5 controls and no active configuration. Canary exactly one explicitly entitled `(tenant-canary, account-canary)` pair. Before traffic, require live Gateway/plugin/account readiness with expected digests and freshness. Exercise all golden outcomes, invalid and unknown fail-closed paths, restart, concurrent waits, atomic challenge consumption, exact replay, altered replay, expiry, unauthenticated event, tenant/account mismatch, cross-tenant IDOR, secret redaction, and exact end-to-end unbind 501 with zero durable/runtime/upstream/job mutation. Any missing evidence, extra account, digest drift, stale snapshot, noncanonical status, secret observation, or mutation fails the canary and blocks promotion.
 Promotion accepts only the exact artifact/config/plugin/source digests verified by canary; it never rebuilds, repatches, switches branches, edits `VERSION`/`baseline/current.json`, creates a tag, invokes Fleet, or changes unrelated services. Candidate and production artifacts must be byte-identical. B5 live preflight is recaptured and reconciled immediately before `verify-stage` and `promote`.
 Rollback is armed before canary with the immediately preceding production artifact/config/plugin digests, units, credentials metadata, and health contract. On any gate failure, STOP blocks forward actions; invoke only B5's recorded rollback-only path. Restore byte-identical prior artifacts/config, never the candidate rebuild; do not delete STOP automatically; do not roll back the database absent a separately approved incompatibility plan. Verify previous digests, listeners, units, secret paths/modes, canonical readiness, and tenant/account isolation. If verification fails, record `ROLLBACK_FAILED`, keep STOP, and escalate without forward mutation.
@@ -1,118 +0,0 @@
 # 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`.
@@ -19,7 +19,6 @@ Alternative C remains the selected architecture because active full-execution au
 - B3 stops unless the installed Weixin payload shape, patch target hashes, authority-map ownership, and transport signing boundary are evidenced.
 - B4 stops unless Hermes request/response schema, router ownership, downstream timeout, idempotency behavior, and exact outbound message binding are evidenced.
 - B5 stops unless B3 and B4 are accepted and the reconciled sentinel plus the V1R4C6 live-runtime snapshot are current for the candidate digest.
 - The R9 D0 documentation correction becomes ledger `START` without consuming a blocker. It expands B2 executable coverage only; all authenticated transport, evidence, terminal, isolation, and release gates remain unchanged.
 The present outbound evidence is insufficient, so B4 and B5 are currently stopped. This document defines the fail-closed path; it does not claim the missing evidence exists.
@@ -131,12 +130,6 @@ The release package has one V2R1 import identity: `aios_v2r1`. There is no `acti
 `active-runtime/pyproject.toml` packages `aios_v2r1` from `active-runtime/app`. Repository tests use `PYTHONPATH='active-runtime:active-runtime/app'`: legacy modules import as `app.*`, while V2R1 imports as `aios_v2r1.*`. The implementation plan enumerates all 23 known `active_runtime.app` test imports and replaces them explicitly. Full collection, full pytest, and wheel inspection must prove the wheel contains one V2R1 namespace and that source and installed imports resolve to `aios_v2r1` only.
 Packaging acceptance is also the executable completeness gate for seven preserved runtime components. Message Ledger must prove exact-ID/no-fallback semantics, idempotency/conflict, tenant/cell/direction isolation, expiry/purge/restart/concurrency, and sanitized SQLite errors. Memory must prove every namespace, cross-scope denial, shared-artifact read-only behavior, qualified-key traversal rejection, and disabled search. Secret must prove reference, broker path, scope, kind, version, identifier, no-value-leakage, and fail-closed broker behavior. Audit must prove genesis and tamper/delete/reorder detection, Unicode canonicalization, concurrency, restart, validation, tenant isolation, and sanitized SQLite errors. Hermes Profile must prove exact environment, absolute `0700` roots, traversal/symlink denial, scope, idempotence, and no environment inheritance. Cell Manager/CLI must prove one cell per tenant, an explicit state graph, atomic concurrent persistence, corrupt-registry fail-closed behavior, stable JSON/exit codes, and restart. Fabric must prove that claims return `running` consistently with durable state, isolation, one concurrent claimant, non-claimable states, restart, and sanitized SQLite errors.
 The original 29 B2 paths are extended by exactly fourteen reviewed paths: seven production modules (`message_ledger.py`, `memory_scope.py`, `secret_scope.py`, `audit.py`, `hermes_profile.py`, `cell_manager.py`, and `hermes_fabric.py`) and seven matrix tests (`test_message_ledger_append.py`, `test_memory_scope.py`, `test_secret_scope.py`, `test_audit.py`, `test_hermes_profile.py`, `test_cell_manager.py`, and `test_hermes_fabric.py`). `cli.py` remains one of the original package paths. A production module changes only after its focused test has been observed failing for the intended contract; already-correct behavior is a characterization and must not be weakened merely to manufacture a RED.
 The B2 review requires exactly 43 changed paths, at least 160 passing carried V2R1/integration/package tests with every matrix bullet represented, zero full-collection errors, a passing complete pytest suite, the 15 Node tests, the real isolated loader in default and opt-in modes, compileall, wheel identity, and a clean tree. These coverage additions do not authorize B3 bridge, B4 Hermes HTTP/outbound, B5 deployment, active service, main, tag, `VERSION`, or baseline mutation.
 ## Exact projection, quote, and attachment authority
 `WeixinProjection` runs in Python after transport authentication. It preserves numeric Weixin IDs from the JSON token as decimal strings and rejects a value that has already passed through unsafe JavaScript-number conversion.
@@ -291,4 +284,4 @@ Preparation or release stops when any of these is true:
 ## Acceptance
-The design is review-complete when the plan contains exactly five blocker commits, dynamically anchors their parent chain at B1 start, enumerates all legacy import edits, and has executable RED/GREEN/review gates. R9 preparation additionally requires D0 as `START`, B1 as its child, B2 as one 43-path child commit, the full seven-component executable matrix, and a partial ledger containing exactly `START,B1,B2`. Runtime release acceptance additionally requires all B3/B4/B5 evidence gates, the full Python collection and suite, the real Node/OpenClaw loader harness, exact receipt semantics, isolated candidate staging, deterministic artifacts, current live evidence, and rehearsed constrained rollback.
+The design is review-complete when the plan contains exactly five blocker commits, dynamically anchors their parent chain at B1 start, enumerates all legacy import edits, and has executable RED/GREEN/review gates. Runtime release acceptance additionally requires all B3/B4/B5 evidence gates, the full Python collection and suite, the real Node/OpenClaw loader harness, exact receipt semantics, isolated candidate staging, deterministic artifacts, current live evidence, and rehearsed constrained rollback.
@@ -1,16 +0,0 @@
 {
  "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
 }
@@ -1,9 +0,0 @@
 {
  "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
 }
@@ -1,9 +0,0 @@
 {
  "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"
 }
@@ -1,44 +0,0 @@
 {
  "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"
  }
 }
@@ -1,23 +0,0 @@
 {
  "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"
  }
 }
@@ -1,781 +0,0 @@
 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
@@ -1,101 +0,0 @@
 """Executable completeness contract for the append-only audit chain."""
 from __future__ import annotations
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 import os
 import sqlite3
 import pytest
 from aios_v2r1.audit import AuditLog
 from aios_v2r1.tenant_context import ConflictError, ValidationError
 def append(log: AuditLog, index: int = 0, *, tenant_id: str = "tenant-a") -> str:
    return log.append(tenant_id=tenant_id, trace_id=f"trace-{index}",
                      event_type="tool_call",
                      payload={"index": index, "text": "虾马 café"})
 def test_first_event_is_anchored_to_genesis(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    log = AuditLog(database)
    append(log)
    with sqlite3.connect(database) as db:
        assert db.execute("SELECT previous_hash FROM audit_events").fetchone()[0] == "GENESIS"
    assert log.verify()
 def test_unicode_payload_is_canonical_across_restart(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    digest = append(AuditLog(database))
    reopened = AuditLog(database)
    assert reopened.verify()
    with sqlite3.connect(database) as db:
        stored = db.execute("SELECT event_hash,payload FROM audit_events").fetchone()
    assert stored[0] == digest
    assert "虾马" in stored[1]
 def test_payload_tamper_is_detected(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    log = AuditLog(database)
    append(log)
    with sqlite3.connect(database) as db:
        db.execute("UPDATE audit_events SET payload='{}'")
    assert not log.verify()
 def test_middle_event_deletion_is_detected(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    log = AuditLog(database)
    for index in range(3):
        append(log, index)
    with sqlite3.connect(database) as db:
        db.execute("DELETE FROM audit_events WHERE seq=2")
    assert not log.verify()
 def test_event_reordering_is_detected(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    log = AuditLog(database)
    append(log, 1)
    append(log, 2)
    with sqlite3.connect(database) as db:
        db.execute("UPDATE audit_events SET seq=100 WHERE seq=1")
    assert not log.verify()
 def test_concurrent_appends_form_one_valid_chain(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    log = AuditLog(database)
    with ThreadPoolExecutor(max_workers=16) as executor:
        digests = list(executor.map(lambda index: append(log, index), range(24)))
    assert len(set(digests)) == 24
    assert log.verify()
    with sqlite3.connect(database) as db:
        assert db.execute("SELECT count(*) FROM audit_events").fetchone()[0] == 24
@pytest.mark.parametrize("kwargs", [
    {"tenant_id": "Tenant-A"}, {"trace_id": "../trace"},
    {"event_type": "tool/call"}])
 def test_audit_identifiers_are_validated(tmp_path: Path, kwargs: dict[str, str]) -> None:
    values = {"tenant_id": "tenant-a", "trace_id": "trace-a",
              "event_type": "tool_call", "payload": {"ok": True}}
    values.update(kwargs)
    with pytest.raises(ValidationError):
        AuditLog(tmp_path / "audit.sqlite3").append(**values)
 def test_tenants_share_no_hash_identity_for_equal_payloads(tmp_path: Path) -> None:
    log = AuditLog(tmp_path / "audit.sqlite3")
    first = append(log, tenant_id="tenant-a")
    second = append(log, tenant_id="tenant-b")
    assert first != second
    assert log.verify()
 def test_database_is_owner_only(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    AuditLog(database)
    assert os.stat(database).st_mode & 0o777 == 0o600
 def test_sqlite_failures_are_sanitized(tmp_path: Path) -> None:
    database = tmp_path / "audit.sqlite3"
    log = AuditLog(database)
    database.write_bytes(b"not-a-sqlite-database")
    with pytest.raises(ConflictError, match="audit storage unavailable") as raised:
        log.verify()
    assert not isinstance(raised.value, sqlite3.Error)
    assert "sqlite" not in str(raised.value).lower()
    assert raised.value.__cause__ is None
    assert raised.value.__context__ is None
@@ -1,114 +0,0 @@
 """Executable completeness contract for CellManager and its CLI."""
 from __future__ import annotations
 from concurrent.futures import ThreadPoolExecutor
 import json
 import os
 from pathlib import Path
 from unittest.mock import patch
 import pytest
 from aios_v2r1.cell_manager import CellManager
 from aios_v2r1.cli import main
 from aios_v2r1.tenant_context import ValidationError
 def test_one_cell_per_tenant(tmp_path: Path) -> None:
    manager = CellManager(tmp_path / "cells.json")
    manager.create("tenant-a", "cell-a", "v1", "digest-a")
    with pytest.raises(ValidationError, match="already"):
        manager.create("tenant-a", "cell-b", "v1", "digest-b")
 def test_explicit_state_graph_accepts_the_recovery_cycle(tmp_path: Path) -> None:
    manager = CellManager(tmp_path / "cells.json")
    manager.create("tenant-a", "cell-a", "v1", "digest-a")
    for state in ("starting", "healthy", "upgrading", "rollback", "healthy"):
        assert manager.transition("tenant-a", state).state == state
 def test_explicit_state_graph_rejects_jump_from_created_to_healthy(tmp_path: Path) -> None:
    manager = CellManager(tmp_path / "cells.json")
    manager.create("tenant-a", "cell-a", "v1", "digest-a")
    with pytest.raises(ValidationError, match="transition"):
        manager.transition("tenant-a", "healthy")
 def test_persistence_uses_atomic_replace(tmp_path: Path) -> None:
    manager = CellManager(tmp_path / "cells.json")
    with patch("aios_v2r1.cell_manager.os.replace", wraps=os.replace) as replaced:
        manager.create("tenant-a", "cell-a", "v1", "digest-a")
    assert replaced.called
 def test_concurrent_distinct_tenant_creates_are_not_lost(tmp_path: Path) -> None:
    registry = tmp_path / "cells.json"
    manager = CellManager(registry)
    def create(index: int) -> None:
        manager.create(f"tenant-{index}", f"cell-{index}", "v1", f"digest-{index}")
    with ThreadPoolExecutor(max_workers=16) as executor:
        list(executor.map(create, range(24)))
    data = json.loads(registry.read_text(encoding="utf-8"))
    assert len(data["cells"]) == 24
 def test_concurrent_same_tenant_has_exactly_one_winner(tmp_path: Path) -> None:
    manager = CellManager(tmp_path / "cells.json")
    def create(index: int) -> str:
        try:
            manager.create("tenant-a", f"cell-{index}", "v1", f"digest-{index}")
            return "created"
        except ValidationError:
            return "rejected"
    with ThreadPoolExecutor(max_workers=16) as executor:
        results = list(executor.map(create, range(16)))
    assert results.count("created") == 1
    assert results.count("rejected") == 15
 def test_corrupt_registry_fails_closed_with_domain_error(tmp_path: Path) -> None:
    registry = tmp_path / "cells.json"
    manager = CellManager(registry)
    registry.write_text('{"cells":{"tenant-a":', encoding="utf-8")
    with pytest.raises(ValidationError, match="registry unavailable") as raised:
        manager.status("tenant-a")
    assert "JSON" not in type(raised.value).__name__
    assert raised.value.__cause__ is None
    assert raised.value.__context__ is None
 def test_restart_preserves_cell_status(tmp_path: Path) -> None:
    registry = tmp_path / "cells.json"
    CellManager(registry).create("tenant-a", "cell-a", "v1", "digest-a")
    assert CellManager(registry).status("tenant-a").cell_id == "cell-a"
 def test_cli_create_returns_stable_json(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
    registry = tmp_path / "cells.json"
    assert main([str(registry), "create", "tenant-a", "cell-a", "v1", "digest-a"]) == 0
    output = json.loads(capsys.readouterr().out)
    assert output["tenant_id"] == "tenant-a"
    assert output["state"] == "created"
 def test_cli_domain_failure_returns_code_3_and_json(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
    registry = tmp_path / "cells.json"
    assert main([str(registry), "status", "missing-tenant"]) == 3
    captured = capsys.readouterr()
    assert captured.out == ""
    assert json.loads(captured.err) == {
        "code": "VALIDATION_ERROR", "error": "unknown tenant"}
 def test_cli_corrupt_registry_is_sanitized(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
    registry = tmp_path / "cells.json"
    CellManager(registry)
    registry.write_text("TOP-SECRET invalid json", encoding="utf-8")
    assert main([str(registry), "status", "tenant-a"]) == 3
    error = json.loads(capsys.readouterr().err)
    assert error["code"] == "VALIDATION_ERROR"
    assert "TOP-SECRET" not in error["error"]
@pytest.mark.parametrize("args", [
    ("Tenant-A", "cell-a", "v1", "digest-a"),
    ("tenant-a", "../cell", "v1", "digest-a"),
    ("tenant-a", "cell-a", "../v1", "digest-a"),
    ("tenant-a", "cell-a", "v1", "../digest")])
 def test_create_validates_all_identifiers(
    tmp_path: Path, args: tuple[str, str, str, str]
 ) -> None:
    with pytest.raises(ValidationError):
        CellManager(tmp_path / "cells.json").create(*args)
@@ -1,114 +1,88 @@
-"""Executable completeness contract for the tenant-partitioned Hermes fabric."""
+"""Gate 1 contract tests for the Hermes shared agent fabric."""
 from __future__ import annotations
-from concurrent.futures import ThreadPoolExecutor
+
 from dataclasses import replace
 from pathlib import Path
 import sqlite3
-import pytest
+from tempfile import TemporaryDirectory
 import unittest
 from aios_v2r1.hermes_fabric import HermesFabric
 from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
-def authenticated_command(event_id: str, **changes: object) -> CanonicalMessage:
+
-    value = CanonicalMessage(
+def authenticated_command(
-        tenant_id="tenant-a", cell_id="cell-a", principal_id="user-a",
+    event_id: str,
-        principal_type="authenticated_user", origin_type="channel_user",
+    *,
-        channel="weixin", channel_account_id="account-a", peer_id="peer-a",
+    command_capable: bool = True,
-        conversation_id="conversation-a", session_id="session-a", task_id=None,
+    origin_type: str = "channel_user",
-        event_id=event_id, message_id=f"message-{event_id}",
+) -> CanonicalMessage:
-        reply_to_message_id=None, direction="inbound", command_capable=True,
+    return CanonicalMessage(
        tenant_id="tenant-a",
        cell_id="cell-a",
        principal_id="user-a",
        principal_type="authenticated_user",
        origin_type=origin_type,
        channel="weixin",
        channel_account_id="account-a",
        peer_id="peer-a",
        conversation_id="conversation-a",
        session_id="session-a",
        task_id=None,
        event_id=event_id,
        message_id="1234567890123456789",
        reply_to_message_id=None,
        direction="inbound",
        command_capable=command_capable,
        occurred_at="2026-07-13T00:00:00+00:00",
-        received_at="2026-07-13T00:00:01+00:00", body="/run",
+        received_at="2026-07-13T00:00:01+00:00",
-        quoted_body=None, adapter_name="weixin", adapter_version="v2")
+        body="/run",
-    return replace(value, **changes)
+        quoted_body=None,
        adapter_name="weixin",
        adapter_version="v2",
    )
@pytest.fixture
 def fabric(tmp_path: Path) -> HermesFabric:
    return HermesFabric(tmp_path / "fabric.sqlite3")
-def test_duplicate_event_returns_sanitized_conflict(fabric: HermesFabric) -> None:
+class HermesFabricContractTests(unittest.TestCase):
-    command = authenticated_command("event-a")
+    def test_duplicate_event_returns_sanitized_conflict(self) -> None:
-    assert fabric.enqueue(message=command, profile_id="profile-a").event_id == "event-a"
+        with TemporaryDirectory() as directory:
-    with pytest.raises(ConflictError) as raised:
+            fabric = HermesFabric(f"{directory}/fabric.sqlite3")
-        fabric.enqueue(message=command, profile_id="profile-a")
+            message = authenticated_command("event-a")
    assert not isinstance(raised.value, sqlite3.IntegrityError)
    assert "sqlite" not in str(raised.value).lower()
    assert raised.value.__cause__ is None
    assert raised.value.__context__ is None
-@pytest.mark.parametrize("changes", [
+            first = fabric.enqueue(message=message, profile_id="profile-a")
-    {"command_capable": False}, {"origin_type": "heartbeat"},
+            self.assertEqual(first.event_id, "event-a")
    {"principal_type": "service"}, {"direction": "internal"}])
 def test_non_claimable_provenance_cannot_enqueue(
    fabric: HermesFabric, changes: dict[str, object]
 ) -> None:
    with pytest.raises(ScopeError):
        fabric.enqueue(message=authenticated_command("event-denied", **changes),
                       profile_id="profile-a")
-def test_claim_returns_running_and_matches_durable_state(fabric: HermesFabric) -> None:
+            with self.assertRaises(ConflictError) as raised:
-    fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
+                fabric.enqueue(message=message, profile_id="profile-a")
    claimed = fabric.claim("tenant-a", "cell-a", "profile-a")
    assert claimed is not None
    assert claimed.status == "running"
    with sqlite3.connect(fabric.database) as db:
        assert db.execute("SELECT status FROM fabric_jobs WHERE job_id=?",
                          (claimed.job_id,)).fetchone()[0] == "running"
-@pytest.mark.parametrize(("tenant_id", "cell_id", "profile_id"), [
+        error = raised.exception
-    ("tenant-b", "cell-a", "profile-a"), ("tenant-a", "cell-b", "profile-a"),
+        self.assertNotIsInstance(error, sqlite3.IntegrityError)
-    ("tenant-a", "cell-a", "profile-b")])
+        self.assertNotIn("sqlite", str(error).lower())
-def test_claim_scope_is_exact(
+        self.assertIsNone(error.__cause__)
-    fabric: HermesFabric, tenant_id: str, cell_id: str, profile_id: str
+        self.assertIsNone(error.__context__)
 ) -> None:
    fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
    assert fabric.claim(tenant_id, cell_id, profile_id) is None
    assert fabric.claim("tenant-a", "cell-a", "profile-a") is not None
-def test_only_one_concurrent_claimant_wins(fabric: HermesFabric) -> None:
+    def test_non_command_event_cannot_enqueue(self) -> None:
-    fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
+        with TemporaryDirectory() as directory:
-    with ThreadPoolExecutor(max_workers=16) as executor:
+            fabric = HermesFabric(f"{directory}/fabric.sqlite3")
        claimed = list(executor.map(
            lambda _: fabric.claim("tenant-a", "cell-a", "profile-a"), range(16)))
    winners = [job for job in claimed if job is not None]
    assert len(winners) == 1
    assert winners[0].status == "running"
-@pytest.mark.parametrize("status", ["running", "completed", "failed", "cancelled"])
+            with self.assertRaises(ScopeError):
-def test_nonqueued_states_are_not_claimable(
+                fabric.enqueue(
-    fabric: HermesFabric, status: str
+                    message=authenticated_command(
-) -> None:
+                        "event-non-command", command_capable=False
-    job = fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
+                    ),
-    with sqlite3.connect(fabric.database) as db:
+                    profile_id="profile-a",
-        db.execute("UPDATE fabric_jobs SET status=? WHERE job_id=?", (status, job.job_id))
+                )
    assert fabric.claim("tenant-a", "cell-a", "profile-a") is None
-def test_restart_preserves_queue_and_claims_once(tmp_path: Path) -> None:
+    def test_heartbeat_provenance_cannot_enqueue(self) -> None:
-    database = tmp_path / "fabric.sqlite3"
+        with TemporaryDirectory() as directory:
-    HermesFabric(database).enqueue(message=authenticated_command("event-a"),
+            fabric = HermesFabric(f"{directory}/fabric.sqlite3")
                                   profile_id="profile-a")
    reopened = HermesFabric(database)
    claimed = reopened.claim("tenant-a", "cell-a", "profile-a")
    assert claimed is not None and claimed.status == "running"
    assert reopened.claim("tenant-a", "cell-a", "profile-a") is None
-def test_concurrent_duplicate_enqueue_has_one_winner(fabric: HermesFabric) -> None:
+            with self.assertRaises(ScopeError):
-    command = authenticated_command("event-a")
+                fabric.enqueue(
-    def enqueue(_: int) -> str:
+                    message=authenticated_command(
-        try:
+                        "event-heartbeat", origin_type="heartbeat"
-            fabric.enqueue(message=command, profile_id="profile-a")
+                    ),
-            return "queued"
+                    profile_id="profile-a",
-        except ConflictError:
+                )
            return "conflict"
    with ThreadPoolExecutor(max_workers=16) as executor:
        results = list(executor.map(enqueue, range(16)))
    assert results.count("queued") == 1
    assert results.count("conflict") == 15
-def test_sqlite_failures_are_sanitized(fabric: HermesFabric) -> None:
+
-    Path(fabric.database).write_bytes(b"not-a-sqlite-database")
+if __name__ == "__main__":
-    with pytest.raises(ConflictError, match="fabric storage unavailable") as raised:
+    unittest.main()
        fabric.health()
    assert not isinstance(raised.value, sqlite3.Error)
    assert "sqlite" not in str(raised.value).lower()
    assert raised.value.__cause__ is None
    assert raised.value.__context__ is None
@@ -1,88 +0,0 @@
 """Executable completeness contract for Hermes profile isolation."""
 from __future__ import annotations
 from pathlib import Path
 import os
 import pytest
 from aios_v2r1.hermes_profile import HermesProfile
 from aios_v2r1.tenant_context import ValidationError
 def profile(tmp_path: Path, **changes: object) -> HermesProfile:
    values: dict[str, object] = {
        "tenant_id": "tenant-a", "cell_id": "cell-a", "profile_id": "profile-a",
        "hermes_home": tmp_path / "home", "workspace": tmp_path / "workspace"}
    values.update(changes)
    return HermesProfile(**values)  # type: ignore[arg-type]
 def test_environment_is_exact_and_has_no_inherited_values(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
    monkeypatch.setenv("HOST_SECRET", "must-not-inherit")
    value = profile(tmp_path).environment
    assert value == {
        "HERMES_HOME": str(tmp_path / "home"), "HOME": str(tmp_path / "home"),
        "AIOS_TENANT_ID": "tenant-a", "AIOS_CELL_ID": "cell-a",
        "AIOS_PROFILE_ID": "profile-a"}
    assert "HOST_SECRET" not in value
@pytest.mark.parametrize("changes", [
    {"hermes_home": Path("relative-home")},
    {"workspace": Path("relative-workspace")}])
 def test_profile_roots_must_be_absolute(
    tmp_path: Path, changes: dict[str, Path]
 ) -> None:
    with pytest.raises(ValidationError, match="absolute"):
        profile(tmp_path, **changes)
 def test_prepare_sets_both_roots_to_0700(tmp_path: Path) -> None:
    value = profile(tmp_path)
    value.prepare()
    assert os.stat(value.hermes_home).st_mode & 0o777 == 0o700
    assert os.stat(value.workspace).st_mode & 0o777 == 0o700
 def test_prepare_is_idempotent(tmp_path: Path) -> None:
    value = profile(tmp_path)
    value.prepare()
    marker = value.workspace / "marker"
    marker.write_text("kept", encoding="utf-8")
    value.prepare()
    assert marker.read_text(encoding="utf-8") == "kept"
 def test_parent_traversal_is_denied(tmp_path: Path) -> None:
    value = profile(tmp_path)
    value.prepare()
    with pytest.raises(ValidationError, match="escapes"):
        value.assert_path_allowed(value.workspace / ".." / "outside")
 def test_symlink_inside_profile_cannot_escape(tmp_path: Path) -> None:
    value = profile(tmp_path)
    value.prepare()
    outside = tmp_path / "outside"
    outside.mkdir()
    (value.workspace / "link").symlink_to(outside, target_is_directory=True)
    with pytest.raises(ValidationError, match="escapes"):
        value.assert_path_allowed(value.workspace / "link" / "secret")
@pytest.mark.parametrize("root_name", ["home", "workspace"])
 def test_prepare_rejects_symlink_roots(tmp_path: Path, root_name: str) -> None:
    outside = tmp_path / f"outside-{root_name}"
    outside.mkdir()
    symlink = tmp_path / root_name
    symlink.symlink_to(outside, target_is_directory=True)
    changes = {"hermes_home" if root_name == "home" else "workspace": symlink}
    with pytest.raises(ValidationError, match="symlink"):
        profile(tmp_path, **changes).prepare()
 def test_other_profile_scope_is_denied(tmp_path: Path) -> None:
    first = profile(tmp_path / "first")
    second = profile(tmp_path / "second", tenant_id="tenant-b",
                     cell_id="cell-b", profile_id="profile-b")
    first.prepare()
    second.prepare()
    with pytest.raises(ValidationError, match="escapes"):
        first.assert_path_allowed(second.workspace / "data")
 def test_paths_within_either_root_are_allowed(tmp_path: Path) -> None:
    value = profile(tmp_path)
    value.prepare()
    value.assert_path_allowed(value.hermes_home / "config.json")
    value.assert_path_allowed(value.workspace / "task.txt")
@@ -1,53 +0,0 @@
 """Executable completeness contract for memory namespaces and access."""
 from __future__ import annotations
 import pytest
 from aios_v2r1.memory_scope import MemoryScope, qualify_memory_key
 from aios_v2r1.tenant_context import ScopeError, ValidationError
 def scope(kind: str = "session", **changes: object) -> MemoryScope:
    values: dict[str, object] = {
        "tenant_id": "tenant-a", "cell_id": "cell-a", "profile_id": "profile-a",
        "session_id": "session-a", "kind": kind}
    values.update(changes)
    return MemoryScope(**values)  # type: ignore[arg-type]
@pytest.mark.parametrize(("kind", "session_id", "expected"), [
    ("session", "session-a", "tenant/tenant-a/cell/cell-a/profile/profile-a/session/session-a/memory/session"),
    ("profile", None, "tenant/tenant-a/cell/cell-a/profile/profile-a/memory/profile"),
    ("cell", None, "tenant/tenant-a/cell/cell-a/memory/cell"),
    ("tenant", None, "tenant/tenant-a/cell/cell-a/memory/tenant"),
    ("shared_artifact", None, "tenant/tenant-a/cell/cell-a/memory/shared_artifact")])
 def test_every_namespace_is_explicit(kind: str, session_id: str | None, expected: str) -> None:
    assert scope(kind, session_id=session_id).namespace == expected
@pytest.mark.parametrize("requested", [
    scope(tenant_id="tenant-b"), scope(cell_id="cell-b"),
    scope(profile_id="profile-b"), scope(session_id="session-b")])
 def test_cross_scope_access_is_denied(requested: MemoryScope) -> None:
    with pytest.raises(ScopeError):
        scope().assert_access(requested)
 def test_cross_kind_access_is_denied() -> None:
    with pytest.raises(ScopeError, match="cross-kind"):
        scope("session").assert_access(scope("profile", session_id=None))
 def test_shared_artifacts_are_read_only() -> None:
    owner = scope()
    shared = scope("shared_artifact", session_id=None)
    owner.assert_access(shared)
    with pytest.raises(ScopeError, match="immutable"):
        owner.assert_access(shared, write=True)
 def test_qualified_key_contains_the_full_namespace() -> None:
    assert qualify_memory_key(scope(), "facts.v1") == (
        "tenant/tenant-a/cell/cell-a/profile/profile-a/session/session-a/"
        "memory/session/facts.v1")
@pytest.mark.parametrize("key", ["../escape", "a/b", ".", "..", " space ", "x?query"])
 def test_memory_key_traversal_and_malformed_keys_are_rejected(key: str) -> None:
    with pytest.raises(ValidationError):
        qualify_memory_key(scope(), key)
 def test_memory_search_surface_is_disabled() -> None:
    assert not hasattr(scope(), "search")
    assert not hasattr(scope(), "search_all")
@@ -1,120 +1,73 @@
-"""Executable completeness contract for the exact-ID message ledger."""
+"""Contract tests for the runtime-facing direct MessageLedger append path."""
 from __future__ import annotations
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import replace
 from pathlib import Path
 import sqlite3
 import pytest
 from aios_v2r1.message_ledger import MessageLedger
 from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
-def message(**changes: object) -> CanonicalMessage:
+from concurrent.futures import ThreadPoolExecutor
-    value = CanonicalMessage(
+from tempfile import TemporaryDirectory
-        tenant_id="tenant-a", cell_id="cell-a", principal_id="user-a",
+import unittest
 from aios_v2r1.message_ledger import MessageLedger
 from aios_v2r1.tenant_context import CanonicalMessage, ConflictError
 def message(*, tenant_id: str = "tenant-a", cell_id: str = "cell-a",
            event_id: str = "event-a", message_id: str = "message-a",
            body: str = "hello", reply_to_message_id: str | None = None) -> CanonicalMessage:
    return CanonicalMessage(
        tenant_id=tenant_id, cell_id=cell_id, principal_id="user-a",
        principal_type="authenticated_user", origin_type="channel_user",
        channel="weixin", channel_account_id="account-a", peer_id="peer-a",
        conversation_id="conversation-a", session_id="session-a", task_id=None,
-        event_id="event-a", message_id="message-a", reply_to_message_id=None,
+        event_id=event_id, message_id=message_id, reply_to_message_id=reply_to_message_id,
        direction="inbound", command_capable=True,
        occurred_at="2026-07-13T00:00:00+00:00",
-        received_at="2026-07-13T00:00:01+00:00", body="hello",
+        received_at="2026-07-13T00:00:01+00:00", body=body,
-        quoted_body=None, adapter_name="weixin", adapter_version="v2")
+        quoted_body=None, adapter_name="weixin", adapter_version="v2",
-    return replace(value, **changes)
+    )
@pytest.fixture
 def ledger(tmp_path: Path) -> MessageLedger:
    return MessageLedger(tmp_path / "ledger.sqlite3")
-def test_append_writes_directly_without_using_remember(ledger: MessageLedger) -> None:
+class MessageLedgerAppendContractTests(unittest.TestCase):
-    ledger.remember = lambda _: (_ for _ in ()).throw(AssertionError("remember called"))  # type: ignore[method-assign]
+    def setUp(self) -> None:
-    assert ledger.append(message()).message_id == "message-a"
+        self.directory = TemporaryDirectory()
-    assert ledger.count() == 1
+        self.addCleanup(self.directory.cleanup)
        self.ledger = MessageLedger(f"{self.directory.name}/ledger.sqlite3")
-def test_append_rejects_same_message_id_with_different_content(ledger: MessageLedger) -> None:
+    def test_append_writes_directly_without_using_remember(self) -> None:
-    ledger.append(message(body="first"))
+        # A compatibility wrapper may remain, but the runtime append path may not
-    with pytest.raises(ConflictError):
+        # route through it: this catches append() -> remember() regressions.
-        ledger.append(message(event_id="event-b", body="second"))
+        self.ledger.remember = lambda _: (_ for _ in ()).throw(AssertionError("remember called"))  # type: ignore[method-assign]
-    assert ledger.count() == 1
+        record = self.ledger.append(message())
        self.assertEqual("message-a", record.message_id)
        self.assertEqual(1, self.ledger.count())
-def test_append_is_idempotent_when_only_event_id_changes(ledger: MessageLedger) -> None:
+    def test_append_rejects_same_message_id_with_different_content(self) -> None:
-    stored = ledger.append(message(event_id="event-a", body="same body"))
+        self.ledger.append(message(body="first"))
-    replay = ledger.append(message(event_id="event-b", body="same body"))
+        with self.assertRaises(ConflictError):
-    assert replay == stored
+            self.ledger.append(message(event_id="event-b", body="second"))
-    assert replay.event_id == "event-a"
+        self.assertEqual(1, self.ledger.count())
    assert ledger.count() == 1
-@pytest.mark.parametrize("candidate", ["message", "message-a-extra", "MESSAGE-A"])
+    def test_append_is_idempotent_when_only_event_id_changes(self) -> None:
-def test_quote_lookup_has_no_prefix_similarity_or_case_fallback(
+        stored = self.ledger.append(message(event_id="event-a", body="same body"))
-    ledger: MessageLedger, candidate: str
+        replay = self.ledger.append(message(event_id="event-b", body="same body"))
 ) -> None:
    ledger.append(message(message_id="message-a"))
    with pytest.raises(ScopeError, match="quoted_text_unavailable"):
        ledger.resolve_quote(message(event_id="reply-event", message_id="reply-message",
                                     reply_to_message_id=candidate))
-def test_explicit_quoted_body_never_falls_back_to_storage(ledger: MessageLedger) -> None:
+        self.assertEqual(stored, replay)
-    assert ledger.resolve_quote(message(quoted_body="exact transport quote")) == "exact transport quote"
+        self.assertEqual("event-a", replay.event_id)
-    assert ledger.count() == 0
+        self.assertEqual(1, self.ledger.count())
-@pytest.mark.parametrize("scope_change", [{"tenant_id": "tenant-b"}, {"cell_id": "cell-b"}])
+    def test_append_keeps_tenants_isolated(self) -> None:
-def test_quote_lookup_is_tenant_and_cell_isolated(
+        self.ledger.append(message())
-    ledger: MessageLedger, scope_change: dict[str, str]
+        self.ledger.append(message(tenant_id="tenant-b", cell_id="cell-b"))
-) -> None:
+        self.assertEqual(2, self.ledger.count())
-    ledger.append(message())
+        self.assertEqual("hello", self.ledger.resolve_quote(
-    with pytest.raises(ScopeError, match="quoted_text_unavailable"):
+            message(tenant_id="tenant-b", cell_id="cell-b", event_id="event-c",
-        ledger.resolve_quote(message(**scope_change, event_id="reply-event",
+                    message_id="message-c", body="reply", reply_to_message_id="message-a")
-                                     message_id="reply-message",
+        ))
                                     reply_to_message_id="message-a"))
-def test_same_id_across_directions_is_ambiguous_not_guessed(ledger: MessageLedger) -> None:
+    def test_concurrent_identical_append_is_one_record(self) -> None:
-    ledger.append(message(direction="inbound", body="in"))
+        with ThreadPoolExecutor(max_workers=16) as executor:
-    ledger.append(message(direction="outbound", event_id="event-b", body="out"))
+            records = list(executor.map(lambda _: self.ledger.append(message()), range(16)))
-    with pytest.raises(ConflictError, match="ambiguous"):
+        self.assertEqual(1, self.ledger.count())
-        ledger.resolve_quote(message(event_id="event-c", message_id="message-c",
+        self.assertEqual({"message-a"}, {record.message_id for record in records})
                                     reply_to_message_id="message-a"))
 def test_expired_record_is_not_resolved(ledger: MessageLedger) -> None:
    ledger.append(message())
    with sqlite3.connect(ledger.database) as db:
        db.execute("UPDATE message_ledger SET expires_at='2000-01-01T00:00:00+00:00'")
    with pytest.raises(ScopeError, match="quoted_text_unavailable"):
        ledger.resolve_quote(message(event_id="event-b", message_id="message-b",
                                     reply_to_message_id="message-a"))
-def test_purge_removes_only_expired_records(ledger: MessageLedger) -> None:
+if __name__ == "__main__":
-    ledger.append(message())
+    unittest.main()
    ledger.append(message(event_id="event-b", message_id="message-b"))
    with sqlite3.connect(ledger.database) as db:
        db.execute("UPDATE message_ledger SET expires_at='2000-01-01T00:00:00+00:00' "
                   "WHERE message_id='message-a'")
    assert ledger.purge_expired() == 1
    assert ledger.count() == 1
 def test_restart_preserves_exact_resolution(tmp_path: Path) -> None:
    database = tmp_path / "ledger.sqlite3"
    MessageLedger(database).append(message())
    reopened = MessageLedger(database)
    assert reopened.resolve_quote(message(event_id="event-b", message_id="message-b",
                                          reply_to_message_id="message-a")) == "hello"
 def test_concurrent_identical_append_is_one_record(ledger: MessageLedger) -> None:
    with ThreadPoolExecutor(max_workers=16) as executor:
        records = list(executor.map(lambda _: ledger.append(message()), range(16)))
    assert ledger.count() == 1
    assert {record.message_id for record in records} == {"message-a"}
 def test_concurrent_distinct_appends_are_not_lost(ledger: MessageLedger) -> None:
    def append(index: int) -> None:
        ledger.append(message(event_id=f"event-{index}", message_id=f"message-{index}"))
    with ThreadPoolExecutor(max_workers=16) as executor:
        list(executor.map(append, range(16)))
    assert ledger.count() == 16
 def test_sqlite_failures_are_sanitized(ledger: MessageLedger) -> None:
    ledger.database.write_bytes(b"not-a-sqlite-database")
    with pytest.raises(ConflictError, match="ledger storage unavailable") as raised:
        ledger.count()
    assert not isinstance(raised.value, sqlite3.Error)
    assert "sqlite" not in str(raised.value).lower()
    assert raised.value.__cause__ is None
    assert raised.value.__context__ is None
@@ -1,80 +0,0 @@
 """Executable completeness contract for scoped secret references."""
 from __future__ import annotations
 from dataclasses import fields
 import pytest
 from aios_v2r1.secret_scope import SecretRef
 from aios_v2r1.tenant_context import ScopeError, ValidationError
 def secret(**changes: object) -> SecretRef:
    values: dict[str, object] = {
        "secret_id": "channel-key", "tenant_id": "tenant-a", "cell_id": "cell-a",
        "kind": "channel", "version": "v1",
        "broker_path": "secret://tenant-a/cell-a/channel-key/v1"}
    values.update(changes)
    return SecretRef(**values)  # type: ignore[arg-type]
@pytest.mark.parametrize("path", [
    "secret://tenant-a/cell-a/channel-key/v1",
    "secret://models/provider/key-v2",
    "secret://runtime/cell-a/signing_key"])
 def test_valid_broker_paths_are_references(path: str) -> None:
    assert secret(broker_path=path).broker_path == path
@pytest.mark.parametrize("path", [
    "", "vault://tenant-a/key", "secret://", "secret:///key",
    "secret://tenant-a/../tenant-b/key", "secret://tenant-a//key",
    "secret://tenant-a/key?version=1", "secret://tenant-a/key#fragment"])
 def test_broker_path_traversal_and_malformed_paths_are_rejected(path: str) -> None:
    with pytest.raises(ValidationError, match="broker path"):
        secret(broker_path=path)
@pytest.mark.parametrize(("tenant_id", "cell_id", "allowed"), [
    ("tenant-a", "cell-a", True), ("tenant-b", "cell-a", False),
    ("tenant-a", "cell-b", False)])
 def test_secret_scope_is_exact(tenant_id: str, cell_id: str, allowed: bool) -> None:
    assert secret().can_be_used_by(tenant_id=tenant_id, cell_id=cell_id) is allowed
@pytest.mark.parametrize("kind", ["channel", "model", "tool", "runtime"])
 def test_secret_kind_allowlist(kind: str) -> None:
    assert secret(kind=kind).kind == kind
 def test_invalid_secret_kind_is_rejected() -> None:
    with pytest.raises(ValidationError, match="kind"):
        secret(kind="admin")
@pytest.mark.parametrize("changes", [
    {"secret_id": "../key"}, {"secret_id": " key "},
    {"version": "../v2"}, {"version": ""}, {"tenant_id": "Tenant-A"},
    {"cell_id": "cell/a"}])
 def test_secret_identifiers_and_version_are_validated(changes: dict[str, str]) -> None:
    with pytest.raises(ValidationError):
        secret(**changes)
 def test_reference_never_contains_a_secret_value_field() -> None:
    assert "value" not in {field.name for field in fields(SecretRef)}
    assert "super-secret-value" not in repr(secret())
 def test_broker_resolution_uses_only_the_validated_reference() -> None:
    seen: list[SecretRef] = []
    class Broker:
        def read_secret(self, reference: SecretRef) -> str:
            seen.append(reference)
            return "resolved-value"
    reference = secret()
    assert reference.resolve(Broker()) == "resolved-value"
    assert seen == [reference]
@pytest.mark.parametrize("broker", [None, object()])
 def test_missing_or_invalid_broker_fails_closed(broker: object | None) -> None:
    with pytest.raises(ScopeError, match="secret broker unavailable"):
        secret().resolve(broker)
 def test_broker_error_is_sanitized_without_secret_leakage() -> None:
    class BrokenBroker:
        def read_secret(self, reference: SecretRef) -> str:
            raise RuntimeError("provider leaked super-secret-value")
    with pytest.raises(ScopeError, match="secret broker unavailable") as raised:
        secret().resolve(BrokenBroker())
    assert "super-secret-value" not in str(raised.value)
    assert raised.value.__cause__ is None
    assert raised.value.__context__ is None
@@ -1,3 +0,0 @@
 """Managed AIOS isolated Python runner provisioning package."""
 __all__ = []
@@ -1,213 +0,0 @@
 """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,
    }
@@ -1,93 +0,0 @@
 """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
@@ -1,144 +0,0 @@
 """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
@@ -1,5 +0,0 @@
 """Stable fail-closed errors for runner provisioning."""
 class ConfigError(RuntimeError):
    """Configuration, integrity, or runtime validation failed."""
@@ -1,74 +0,0 @@
 """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)
@@ -1,446 +0,0 @@
 """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"}
@@ -1,203 +0,0 @@
 """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
@@ -1,150 +0,0 @@
 #!/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())
@@ -1,179 +0,0 @@
 """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
@@ -1,202 +0,0 @@
 """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"),
    )
@@ -1,80 +0,0 @@
 # 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。
@@ -1,53 +0,0 @@
 {
  "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"
  }
 }
Author	SHA1	Message	Date
openclaw	97808b70a7	feat(v2r1): package canonical authenticated ingress	2026-07-14 10:36:55 +08:00
openclaw	910a16a998	test(openclaw): make tenant-boundary hook compatibility executable	2026-07-14 07:01:33 +08:00
		`@@ -1 +0,0 @@`
			`{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"tenant.account.entitlement","tenantId":"tenant-canary","accountId":"account-canary","effect":"allow","policyVersion":"policy-v1","issuedAtMs":1784068000000}`
		`@@ -1 +0,0 @@`
			{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":true,"code":"ready","reasons":[],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
		`@@ -1 +0,0 @@`
			`{"eventId":"event-20260715-A1","accountId":"account-canary","senderId":"sender-canary","chatId":"chat-canary","challengeCode":"A1B2C3D4","observedAt":1784068120000}`
		`@@ -1,2 +0,0 @@`
			`g aios-spark-runner -`
			`u aios-spark-runner - "AIOS isolated runner" /var/lib/aios-spark-runner /usr/sbin/nologin`
		`@@ -1,3 +0,0 @@`
			`"""Managed AIOS isolated Python runner provisioning package."""`

			`__all__ = []`