feat(acceptance): 添加失败关闭的第三阶段验收运行器

.gitea/workflows/ci.yml

@@ -30,6 +30,7 @@ jobs:
         working-directory: backend
       - run: uv run pytest
         working-directory: backend
+      - run: python scripts/phase3-production-acceptance.py --list-cases --json
 
   service-test:
     runs-on: ubuntu-latest

backend/tests/test_phase3_acceptance_runner.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from argparse import Namespace
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SPEC = importlib.util.spec_from_file_location("phase3_acceptance", ROOT / "scripts" / "phase3_acceptance.py")
+assert SPEC and SPEC.loader
+runner = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(runner)
+
+
+def config(path: Path, data_root: Path, **changes) -> Path:
+    payload = {
+        "schema": 1,
+        "run_id": "runner-test-001",
+        "isolated": True,
+        "allow_destructive": True,
+        "platform_profile": "windows-11-x64",
+        "data_root": str(data_root),
+        "seed": 20260908,
+        "service_urls": {},
+        "artifacts": {},
+        "secret_env": {},
+    }
+    payload.update(changes)
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    return path
+
+
+def test_manifest_contains_every_documented_case_once():
+    assert len(runner.ALL_CASES) == 30
+    assert len(set(runner.ALL_CASES)) == 30
+    assert runner.select_cases("sync-client", "s-08") == ("S-08",)
+    with pytest.raises(runner.AcceptanceError, match="CASE_NOT_IN_SUITE"):
+        runner.select_cases("sidecar", "S-08")
+
+
+def test_config_rejects_personal_vault_plaintext_secrets_and_unconfirmed_roots(tmp_path):
+    with pytest.raises(runner.AcceptanceError, match="PERSONAL_VAULT"):
+        runner.load_config(config(tmp_path / "vault.json", runner.VAULT_ROOT))
+    with pytest.raises(runner.AcceptanceError, match="PLAINTEXT_SECRET"):
+        runner.load_config(config(tmp_path / "secret.json", tmp_path / "data", password="do-not-store-this"))
+    existing = tmp_path / "existing"
+    existing.mkdir()
+    (existing / "keep.txt").write_text("keep", encoding="utf-8")
+    loaded = runner.load_config(config(tmp_path / "existing.json", existing))
+    with pytest.raises(runner.AcceptanceError, match="NOT_EMPTY_OR_MARKED"):
+        runner.prepare_isolated_root(loaded)
+    assert (existing / "keep.txt").read_text(encoding="utf-8") == "keep"
+
+
+def test_missing_driver_is_a_junit_failure_and_never_a_skip(tmp_path):
+    data_root = tmp_path / "isolated"
+    config_path = config(tmp_path / "config.json", data_root)
+    report = tmp_path / "report"
+    args = Namespace(
+        suite="sidecar", case="A-01", config=str(config_path), report_dir=str(report),
+        list_cases=False, json=False,
+    )
+    assert runner.execute(args, {}) == 1
+    result = json.loads((report / "cases" / "A-01.json").read_text(encoding="utf-8"))
+    summary = json.loads((report / "summary.json").read_text(encoding="utf-8"))
+    junit = (report / "junit.xml").read_text(encoding="utf-8")
+    assert result["status"] == "NOT_IMPLEMENTED"
+    assert summary["status"] == "NOT_PASSED"
+    assert summary["contains_credentials"] is False
+    assert '<failure type="NOT_IMPLEMENTED">' in junit
+    assert "skipped=\"0\"" in junit
+
+
+def test_driver_result_must_supply_assertions_metrics_and_zero_exit(tmp_path):
+    driver = tmp_path / "driver.py"
+    driver.write_text("", encoding="utf-8")
+    original_root = runner.ROOT
+    try:
+        runner.ROOT = tmp_path
+        case = {
+            "driver": "driver.py",
+            "required_metrics": ("peak_rss_bytes",),
+        }
+        result_root = tmp_path / "result"
+        result_root.mkdir()
+        result = runner.run_case("A-01", tmp_path / "config.json", {"data_root": str(tmp_path / "data"), "secret_env": {}}, result_root, {}, {"A-01": case})
+        assert result["status"] == "FAILED"
+        assert "RESULT_ASSERTIONS_MISSING" in result["runner_errors"]
+        assert "RESULT_METRIC_MISSING:peak_rss_bytes" in result["runner_errors"]
+    finally:
+        runner.ROOT = original_root
+
+
+def test_valid_driver_passes_and_its_log_is_redacted(tmp_path):
+    driver = tmp_path / "driver.py"
+    driver.write_text(
+        """import argparse, json, os
+from pathlib import Path
+p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--output');a=p.parse_args()
+print(os.environ['TEST_ACCEPTANCE_SECRET'])
+Path(a.output).write_text(json.dumps({
+ 'schema':1,'case_id':os.environ['OPENNEXUS_ACCEPTANCE_CASE_ID'],'status':'PASSED','reason':'',
+ 'assertions':[{'name':'independent oracle','status':'PASSED','evidence':'fixture'}],
+ 'metrics':{'peak_rss_bytes':123,'max_process_count':1,'denied_access_count':0},
+ 'files':[],'revisions':[]}), encoding='utf-8')
+""",
+        encoding="utf-8",
+    )
+    original_root = runner.ROOT
+    try:
+        runner.ROOT = tmp_path
+        result_root = tmp_path / "result"
+        result_root.mkdir()
+        result = runner.run_case(
+            "A-01",
+            tmp_path / "config.json",
+            {
+                "data_root": str(tmp_path / "data"), "secret_env": {"fixture": "TEST_ACCEPTANCE_SECRET"},
+                "platform_profile": "windows-11-x64", "artifacts": {},
+            },
+            result_root,
+            {"TEST_ACCEPTANCE_SECRET": "planted-secret-value"},
+            {"A-01": {"driver": "driver.py", "required_metrics": ("peak_rss_bytes",)}},
+        )
+        assert result["status"] == "PASSED"
+        log = (result_root / "logs" / "A-01.log").read_text(encoding="utf-8")
+        assert "planted-secret-value" not in log
+        assert "[REDACTED]" in log
+    finally:
+        runner.ROOT = original_root

docs/architecture/第三阶段生产化工程规划与验收目标.md

@@ -182,7 +182,7 @@ cargo test --manifest-path frontend/src-tauri/Cargo.toml --lib --locked
 cargo clippy --manifest-path frontend/src-tauri/Cargo.toml --lib --locked -- -D warnings
 ```
 
-以下验收runner和配置是**P0/R0待交付接口，现在不存在，不能报告已运行**：仓库根执行`python scripts/phase3-production-acceptance.py --suite <suite> --config <isolated-config> --report-dir <output>`；suite取`sidecar/credentials/sandbox/extensions/sync-client/sync-service/e2e/all`，支持`--case <ID>`单例复跑。配置由环境模板产生，含临时数据根、服务URL、安装包路径、平台profile、seed及秘密的环境引用，不能含生产凭据。runner必须拒绝仓库个人Vault及未标记隔离环境，缺依赖/安装包/权限时非零退出；必测项skip也非零，不能只打印成功。
+验收入口已经建立：仓库根执行`python scripts/phase3-production-acceptance.py --suite <suite> --config <isolated-config> --report-dir <output>`；suite取`sidecar/credentials/sandbox/extensions/sync-client/sync-service/e2e/all`，支持`--case <ID>`单例复跑。配置由[隔离模板](../../scripts/phase3-acceptance-config.example.json)产生，含临时数据根、服务URL、安装包路径、平台profile、seed及秘密的环境引用，不能含生产凭据。runner拒绝仓库个人Vault、未标记隔离环境和非空报告目录；缺case driver、依赖/安装包/权限或必测项skip均以非零退出，不能只打印成功。当前各验收 ID 的完整生产 driver 仍须逐项接入，未登记项明确报告 `NOT_IMPLEMENTED`，不因 runner 接口存在而视为已验收；使用和结果契约见[生产验收 Runner](../development/OpenNexus生产验收Runner.md)。
 
 runner输出JUnit、逐ID JSON、耗时/峰值内存、进程树/拒绝访问计数、文件摘要与revision清单及脱敏日志。退出0仅表示该suite所有适用ID通过；`all`要求显式平台profile并检查缺失项。下表同一行内所有断言均必测，每类输入保留独立子用例；测试代码硬编码预期协议向量或独立oracle，不能用被测函数自证。P0需生成ID清单并校验报告完整性。
 

docs/development/OpenNexus生产化实施进度-2026-09-08.md

@@ -772,3 +772,12 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
 - 实际 Python Core + Rust Host 集成覆盖创建、20 次相同重放、改变输入冲突、列表、跨 Vault 拒绝、CAS 更新、过期摘要拒绝、Agent 选择运行、删除及关闭前 journal 计数。实际 Uvicorn Sync + 两个 Rust Workspace 客户端对用户 Skill 进行了 60 轮同时修改，保留本地/采用远端/创建文本副本各 20 轮，并验证旧摘要拒绝、重复解决、两端收敛、重开稳定；第三个默认范围客户端可接收用户 Skill，同时仍排除未选择的人设与布局。
 - 最终 desktop Rust 全目标 148 通过、12 ignored（library 131、Host 10、其余集成 7），Clippy `--all-targets -D warnings` 通过，日志 `.build/user-skill-rust-full-final.log`、`.build/user-skill-clippy-current.log`。后端 908 项通过、1 项依赖弃用警告，日志 `.build/user-skill-python-full-current.log`。前端 103 个测试文件、536 项通过，类型检查和生产构建通过，日志 `.build/user-skill-frontend-full-final2.log`、`.build/user-skill-frontend-build-final2.log`。
 - 证据来自本机隔离服务和客户端进程，尚未替代两台独立硬件上的桌面 UI、跨发布版本或每个持久化边界强制终止验收。Provider 非秘密参数、安装清单、对话等规划内逻辑数据仍未适配；完整生产化目标保持未完成。
+
+
+## 增量：生产验收 Runner 的失败闭合入口
+
+- 新增规划约定的 `scripts/phase3-production-acceptance.py`，固定登记 sidecar/credentials/sandbox/extensions/sync-client/sync-service/e2e 七个 suite 和 30 个验收 ID，支持 `--case` 单例与机器可读清单。CI 的 backend job 会验证清单入口，新单元测试覆盖清单唯一性和 suite 归属。
+- 配置必须显式确认隔离和破坏性测试，提供唯一 run、平台 profile、固定 seed 及绝对临时数据根；首次只接受空目录并写入绑定标记。runner 拒绝个人 Vault、仓库或 Git 路径、符号链接、非空未标记目录、明文秘密字段、带认证信息/query/fragment 的服务 URL，以及未显式声明的测试 HTTP。秘密只按环境变量名引用，缺值直接失败。
+- 每个 case 只调用仓库固定登记的 driver，配置不能注入命令。driver 的 JSON 必须含独立断言、文件/revision 清单和该 ID 要求的指标；非零退出、超时、坏 schema、缺指标、缺安装产物/权限/秘密引用、必测 skip 或结果秘密命中均失败。输出含逐 ID JSON、JUnit、固定 ID 清单、脱敏限长日志、commit、锁文件/安装产物/config 摘要；已有非空报告目录不覆盖。
+- 5 项 runner 测试通过，覆盖个人 Vault/明文秘密/非空目录拒绝，未登记 ID 生成 JUnit failure 且退出 1，driver 缺断言/指标失败，以及测试秘密在日志中被替换。实际 CLI 清单检查通过；使用模板运行 A-01 得到预期 `NOT_IMPLEMENTED`、summary `NOT_PASSED` 和退出 1，证明缺实现不会误报成功。包含该测试的后端全套 913 项通过、1 项依赖弃用警告，日志 `.build/acceptance-runner-python.log`；Python 编译与 Markdown 链接检查通过。
+- 当前仅完成 P0 runner 与报告协议，30 项完整生产 driver 尚未登记；因此没有任何 A/B/C/D/S/E ID 因本增量变为通过。后续逐项接入时仍须满足规划表的全部断言、实机/生产依赖和量化指标，完整生产化目标保持未完成。

docs/development/OpenNexus生产验收Runner.md

@@ -0,0 +1,27 @@
+# OpenNexus 生产验收 Runner
+
+`scripts/phase3-production-acceptance.py` 是第三阶段 A/B/C/D/S/E 验收 ID 的统一入口。它只把仓库登记的逐 ID driver 结果计为生产验收；普通单元测试、隔离冒烟和历史人工汇总不会自动换算成通过。
+
+查看固定的 30 项清单：
+
+```powershell
+python scripts/phase3-production-acceptance.py --list-cases --json
+```
+
+复制 `scripts/phase3-acceptance-config.example.json` 到仓库外，设置唯一 `run_id` 和平台 profile。`data_root` 可直接使用绝对路径，也可像模板一样通过环境变量引用。第一次运行只接受不存在或空目录并写入与 `run_id` 绑定的隔离标记；后续只能由同一 run 复用。个人 Vault、仓库或 `.git` 的父目录/子目录、符号链接、含明文密码/令牌的配置均会被拒绝。HTTP 服务只允许在配置中明确标记为测试服务，URL 不得包含用户信息、query 或 fragment。
+
+执行单例时应使用新的空报告目录：
+
+```powershell
+$env:OPENNEXUS_ACCEPTANCE_DATA_ROOT = 'D:\opennexus-acceptance\run-001'
+python scripts/phase3-production-acceptance.py `
+  --suite sync-service --case S-04 `
+  --config D:\opennexus-acceptance\run-001.json `
+  --report-dir D:\opennexus-acceptance\reports\run-001-s04
+```
+
+每个 driver 由仓库内 `CASE_DRIVERS` 固定登记，接收 `--config` 与 `--output`，并通过环境获得 case ID、数据根和报告根。结果必须是 schema 1 JSON，包含匹配的 `case_id`、`PASSED/FAILED/SKIPPED/NOT_APPLICABLE` 状态、至少一项独立断言、该 ID 要求的非负整数指标，以及 `files` 和 `revisions` 清单。只有 driver 退出 0、结果为 `PASSED`、所有必填指标齐全且结果不含配置引用的秘密值时，该 ID 才通过。缺 driver、必测 skip、非零退出、超时、缺安装包/权限/秘密环境引用、坏 schema 或秘密命中都会让 suite 非零退出。
+
+报告目录包含 `summary.json`、`case-manifest.json`、`junit.xml`、`cases/<ID>.json` 和脱敏的 `logs/<ID>.log`。摘要记录 commit、各锁文件 SHA-256、配置摘要与已提供安装产物摘要。日志将仓库、数据根、报告根、用户主目录和配置声明的秘密值替换为占位符，并限制为 10 MiB。报告目录必须为空，避免单例复跑覆盖原始证据。
+
+当前 runner 与失败闭合行为已实现，各生产验收 ID 的完整 driver 尚未登记。因此现在运行任一 ID 会生成 `NOT_IMPLEMENTED` 证据并退出 1；这用于阻止误报，不是验收通过。

scripts/phase3-acceptance-config.example.json

@@ -0,0 +1,15 @@
+{
+  "schema": 1,
+  "run_id": "replace-with-unique-run-id",
+  "isolated": true,
+  "allow_destructive": true,
+  "platform_profile": "windows-11-x64",
+  "data_root_env": "OPENNEXUS_ACCEPTANCE_DATA_ROOT",
+  "seed": 20260908,
+  "allow_http_test_services": true,
+  "service_urls": {
+    "sync": "http://127.0.0.1:18080"
+  },
+  "artifacts": {},
+  "secret_env": {}
+}

scripts/phase3-production-acceptance.py

@@ -0,0 +1,7 @@
+"""Documented command entry point for the production acceptance runner."""
+
+from phase3_acceptance import main
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/phase3_acceptance.py

@@ -0,0 +1,498 @@
+"""Fail-closed runner for the OpenNexus phase-three production acceptance IDs."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+from xml.etree import ElementTree
+
+ROOT = Path(__file__).resolve().parents[1]
+VAULT_ROOT = (ROOT / "backend" / "data" / "vault").resolve()
+SCHEMA = 1
+CASE_SUITES = {
+    "sidecar": tuple(f"A-{number:02d}" for number in range(1, 5)),
+    "credentials": tuple(f"B-{number:02d}" for number in range(1, 5)),
+    "sandbox": tuple(f"C-{number:02d}" for number in range(1, 5)),
+    "extensions": tuple(f"D-{number:02d}" for number in range(1, 5)),
+    "sync-client": ("S-01", "S-02", "S-03", "S-08"),
+    "sync-service": ("S-04", "S-05", "S-06", "S-07", "S-09"),
+    "e2e": tuple(f"E-{number:02d}" for number in range(1, 6)),
+}
+ALL_CASES = tuple(case for cases in CASE_SUITES.values() for case in cases)
+# A case becomes executable only when a repository-owned driver is registered here.
+# Component/unit test commands are deliberately not treated as production acceptance.
+CASE_DRIVERS: dict[str, dict[str, Any]] = {}
+ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,127}")
+RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{2,63}")
+SENSITIVE_KEY = re.compile(r"(?:password|passwd|secret|token|api[_-]?key|credential)", re.I)
+MAX_LOG_BYTES = 10 * 1024 * 1024
+
+
+class AcceptanceError(ValueError):
+    """A stable, user-actionable runner configuration error."""
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _inside(path: Path, parent: Path) -> bool:
+    try:
+        path.relative_to(parent)
+        return True
+    except ValueError:
+        return False
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _contains_plain_secret(value: Any, path: tuple[str, ...] = ()) -> bool:
+    if isinstance(value, dict):
+        for key, item in value.items():
+            nested = path + (str(key),)
+            if SENSITIVE_KEY.search(str(key)) and str(key) != "secret_env" and (not path or path[-1] != "secret_env"):
+                return True
+            if _contains_plain_secret(item, nested):
+                return True
+    elif isinstance(value, list):
+        return any(_contains_plain_secret(item, path) for item in value)
+    return False
+
+
+def _resolve_data_root(config: dict[str, Any], environ: dict[str, str]) -> Path:
+    direct = config.get("data_root")
+    reference = config.get("data_root_env")
+    if bool(direct) == bool(reference):
+        raise AcceptanceError("CONFIG_DATA_ROOT_REQUIRED: set exactly one of data_root or data_root_env")
+    if reference:
+        if not isinstance(reference, str) or not ENV_NAME.fullmatch(reference):
+            raise AcceptanceError("CONFIG_DATA_ROOT_ENV_INVALID")
+        direct = environ.get(reference)
+        if not direct:
+            raise AcceptanceError(f"CONFIG_ENV_MISSING: {reference}")
+    path = Path(str(direct)).expanduser()
+    if not path.is_absolute():
+        raise AcceptanceError("CONFIG_DATA_ROOT_NOT_ABSOLUTE")
+    resolved = path.resolve(strict=False)
+    if resolved == ROOT or _inside(ROOT, resolved):
+        raise AcceptanceError("CONFIG_DATA_ROOT_CONTAINS_REPOSITORY")
+    if _inside(resolved, VAULT_ROOT) or _inside(VAULT_ROOT, resolved):
+        raise AcceptanceError("CONFIG_PERSONAL_VAULT_FORBIDDEN")
+    if _inside(resolved, (ROOT / ".git").resolve()):
+        raise AcceptanceError("CONFIG_GIT_DIRECTORY_FORBIDDEN")
+    if path.exists() and path.is_symlink():
+        raise AcceptanceError("CONFIG_DATA_ROOT_SYMLINK_FORBIDDEN")
+    return resolved
+
+
+def load_config(path: Path, environ: dict[str, str] | None = None) -> dict[str, Any]:
+    environ = os.environ if environ is None else environ
+    try:
+        config = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        raise AcceptanceError("CONFIG_UNREADABLE") from error
+    if not isinstance(config, dict) or config.get("schema") != SCHEMA:
+        raise AcceptanceError("CONFIG_SCHEMA_UNSUPPORTED")
+    if _contains_plain_secret(config):
+        raise AcceptanceError("CONFIG_PLAINTEXT_SECRET_FORBIDDEN")
+    allowed = {
+        "schema", "run_id", "isolated", "allow_destructive", "platform_profile",
+        "data_root", "data_root_env", "seed", "allow_http_test_services",
+        "service_urls", "artifacts", "secret_env",
+    }
+    unknown = sorted(set(config) - allowed)
+    if unknown:
+        raise AcceptanceError(f"CONFIG_FIELD_UNKNOWN: {','.join(unknown)}")
+    if config.get("isolated") is not True or config.get("allow_destructive") is not True:
+        raise AcceptanceError("CONFIG_ISOLATION_CONFIRMATION_REQUIRED")
+    run_id = config.get("run_id")
+    if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
+        raise AcceptanceError("CONFIG_RUN_ID_INVALID")
+    profile = config.get("platform_profile")
+    if not isinstance(profile, str) or not profile.strip():
+        raise AcceptanceError("CONFIG_PLATFORM_PROFILE_REQUIRED")
+    seed = config.get("seed")
+    if not isinstance(seed, int) or isinstance(seed, bool) or not 0 <= seed <= 2**63 - 1:
+        raise AcceptanceError("CONFIG_SEED_INVALID")
+    secret_env = config.get("secret_env", {})
+    if not isinstance(secret_env, dict):
+        raise AcceptanceError("CONFIG_SECRET_ENV_INVALID")
+    for alias, reference in secret_env.items():
+        if not isinstance(alias, str) or not isinstance(reference, str) or not ENV_NAME.fullmatch(reference):
+            raise AcceptanceError("CONFIG_SECRET_ENV_INVALID")
+        if not environ.get(reference):
+            raise AcceptanceError(f"CONFIG_ENV_MISSING: {reference}")
+    urls = config.get("service_urls", {})
+    if not isinstance(urls, dict):
+        raise AcceptanceError("CONFIG_SERVICE_URLS_INVALID")
+    for name, value in urls.items():
+        if not isinstance(name, str) or not isinstance(value, str):
+            raise AcceptanceError("CONFIG_SERVICE_URL_INVALID")
+        parsed = urlsplit(value)
+        if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
+            raise AcceptanceError("CONFIG_SERVICE_URL_INVALID")
+        if parsed.query or parsed.fragment:
+            raise AcceptanceError("CONFIG_SERVICE_URL_INVALID")
+        if parsed.scheme == "http" and config.get("allow_http_test_services") is not True:
+            raise AcceptanceError("CONFIG_HTTP_SERVICE_REQUIRES_TEST_FLAG")
+    artifacts = config.get("artifacts", {})
+    if not isinstance(artifacts, dict) or any(not isinstance(name, str) or not isinstance(value, str) for name, value in artifacts.items()):
+        raise AcceptanceError("CONFIG_ARTIFACTS_INVALID")
+    normalized = dict(config)
+    normalized["data_root"] = str(_resolve_data_root(config, environ))
+    normalized.pop("data_root_env", None)
+    return normalized
+
+
+def prepare_isolated_root(config: dict[str, Any]) -> Path:
+    root = Path(config["data_root"])
+    marker = root / ".opennexus-acceptance-isolated.json"
+    expected = {"schema": SCHEMA, "run_id": config["run_id"]}
+    if root.exists():
+        if root.is_symlink():
+            raise AcceptanceError("CONFIG_DATA_ROOT_SYMLINK_FORBIDDEN")
+        if not root.is_dir():
+            raise AcceptanceError("CONFIG_DATA_ROOT_NOT_DIRECTORY")
+        children = list(root.iterdir())
+        if children:
+            if not marker.is_file():
+                raise AcceptanceError("CONFIG_DATA_ROOT_NOT_EMPTY_OR_MARKED")
+            try:
+                actual = json.loads(marker.read_text(encoding="utf-8"))
+            except (OSError, json.JSONDecodeError) as error:
+                raise AcceptanceError("CONFIG_ISOLATION_MARKER_INVALID") from error
+            if actual != expected:
+                raise AcceptanceError("CONFIG_ISOLATION_MARKER_MISMATCH")
+    else:
+        root.mkdir(parents=True)
+    marker.write_text(json.dumps(expected, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return root
+
+
+def select_cases(suite: str, case: str | None) -> tuple[str, ...]:
+    selected = ALL_CASES if suite == "all" else CASE_SUITES[suite]
+    if case is None:
+        return selected
+    normalized = case.upper()
+    if normalized not in ALL_CASES:
+        raise AcceptanceError("CASE_UNKNOWN")
+    if normalized not in selected:
+        raise AcceptanceError("CASE_NOT_IN_SUITE")
+    return (normalized,)
+
+
+def _redactor(config: dict[str, Any], report_root: Path, environ: dict[str, str]):
+    replacements = [
+        (str(ROOT), "$REPO"),
+        (config["data_root"], "$DATA"),
+        (str(report_root), "$REPORT"),
+        (str(Path.home()), "$HOME"),
+    ]
+    for reference in config.get("secret_env", {}).values():
+        value = environ.get(reference, "")
+        if value:
+            replacements.append((value, "[REDACTED]"))
+
+    def redact(text: str) -> str:
+        for source, target in sorted(replacements, key=lambda item: len(item[0]), reverse=True):
+            text = text.replace(source, target).replace(source.replace("\\", "/"), target)
+        return text
+
+    return redact
+
+
+def _base_result(case_id: str, status: str, reason: str) -> dict[str, Any]:
+    return {
+        "schema": SCHEMA,
+        "case_id": case_id,
+        "status": status,
+        "reason": reason,
+        "assertions": [],
+        "metrics": {
+            "peak_rss_bytes": None,
+            "max_process_count": None,
+            "denied_access_count": None,
+        },
+        "files": [],
+        "revisions": [],
+        "command": None,
+    }
+
+
+def _has_administrator_permission() -> bool:
+    if os.name == "nt":
+        try:
+            import ctypes
+
+            return bool(ctypes.windll.shell32.IsUserAnAdmin())
+        except (AttributeError, OSError):
+            return False
+    return hasattr(os, "geteuid") and os.geteuid() == 0
+
+
+def _validate_driver_result(case_id: str, result: Any, required_metrics: tuple[str, ...]) -> list[str]:
+    errors: list[str] = []
+    if not isinstance(result, dict) or result.get("schema") != SCHEMA:
+        return ["RESULT_SCHEMA_INVALID"]
+    if result.get("case_id") != case_id:
+        errors.append("RESULT_CASE_ID_MISMATCH")
+    if result.get("status") not in {"PASSED", "FAILED", "SKIPPED", "NOT_APPLICABLE"}:
+        errors.append("RESULT_STATUS_INVALID")
+    assertions = result.get("assertions")
+    if not isinstance(assertions, list) or not assertions:
+        errors.append("RESULT_ASSERTIONS_MISSING")
+    elif any(not isinstance(item, dict) or item.get("status") not in {"PASSED", "FAILED"} for item in assertions):
+        errors.append("RESULT_ASSERTION_INVALID")
+    metrics = result.get("metrics")
+    if not isinstance(metrics, dict):
+        errors.append("RESULT_METRICS_MISSING")
+    else:
+        for name in required_metrics:
+            value = metrics.get(name)
+            if not isinstance(value, int) or isinstance(value, bool) or value < 0:
+                errors.append(f"RESULT_METRIC_MISSING:{name}")
+    for name in ("files", "revisions"):
+        if not isinstance(result.get(name), list):
+            errors.append(f"RESULT_{name.upper()}_INVALID")
+    return errors
+
+
+def run_case(
+    case_id: str,
+    config_path: Path,
+    config: dict[str, Any],
+    report_root: Path,
+    environ: dict[str, str] | None = None,
+    drivers: dict[str, dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    environ = dict(os.environ if environ is None else environ)
+    drivers = CASE_DRIVERS if drivers is None else drivers
+    started = time.monotonic()
+    started_utc = _utc_now()
+    definition = drivers.get(case_id)
+    if definition is None:
+        result = _base_result(case_id, "NOT_IMPLEMENTED", "No repository-owned production acceptance driver is registered.")
+        result.update({"started_utc": started_utc, "finished_utc": _utc_now(), "duration_ms": 0, "exit_code": None})
+        return result
+    if definition.get("requires_admin") and not _has_administrator_permission():
+        result = _base_result(case_id, "BLOCKED", "Administrator permission required by this acceptance case is unavailable.")
+        result.update({"started_utc": started_utc, "finished_utc": _utc_now(), "duration_ms": 0, "exit_code": None})
+        return result
+    required_platforms = tuple(definition.get("platform_profiles", ()))
+    if required_platforms and config["platform_profile"] not in required_platforms:
+        result = _base_result(case_id, "BLOCKED", "Configured platform profile is not supported by this acceptance driver.")
+        result["supported_platform_profiles"] = list(required_platforms)
+        result.update({"started_utc": started_utc, "finished_utc": _utc_now(), "duration_ms": 0, "exit_code": None})
+        return result
+    driver = (ROOT / str(definition["driver"])).resolve()
+    if not _inside(driver, ROOT) or not driver.is_file():
+        result = _base_result(case_id, "BLOCKED", "Registered acceptance driver is missing or outside the repository.")
+        result.update({"started_utc": started_utc, "finished_utc": _utc_now(), "duration_ms": 0, "exit_code": None})
+        return result
+    missing = [name for name in definition.get("required_artifacts", ()) if not Path(config.get("artifacts", {}).get(name, "")).is_file()]
+    if missing:
+        result = _base_result(case_id, "BLOCKED", "Required artifacts are missing.")
+        result["missing_artifacts"] = missing
+        result.update({"started_utc": started_utc, "finished_utc": _utc_now(), "duration_ms": 0, "exit_code": None})
+        return result
+    missing_secrets = [name for name in definition.get("required_secret_env", ()) if name not in config.get("secret_env", {})]
+    if missing_secrets:
+        result = _base_result(case_id, "BLOCKED", "Required secret environment references are missing.")
+        result["missing_secret_env"] = missing_secrets
+        result.update({"started_utc": started_utc, "finished_utc": _utc_now(), "duration_ms": 0, "exit_code": None})
+        return result
+    result_path = report_root / "driver-results" / f"{case_id}.json"
+    result_path.parent.mkdir(parents=True, exist_ok=True)
+    command = [sys.executable, str(driver), "--config", str(config_path), "--output", str(result_path)]
+    child_env = dict(environ)
+    child_env.update({
+        "OPENNEXUS_ACCEPTANCE_CASE_ID": case_id,
+        "OPENNEXUS_ACCEPTANCE_DATA_ROOT": config["data_root"],
+        "OPENNEXUS_ACCEPTANCE_REPORT_ROOT": str(report_root),
+    })
+    timed_out = False
+    try:
+        completed = subprocess.run(
+            command,
+            cwd=ROOT,
+            env=child_env,
+            capture_output=True,
+            text=True,
+            timeout=int(definition.get("timeout_seconds", 1800)),
+            check=False,
+        )
+        exit_code = completed.returncode
+        output = completed.stdout + ("\n" if completed.stdout and completed.stderr else "") + completed.stderr
+    except subprocess.TimeoutExpired as error:
+        timed_out = True
+        exit_code = None
+        output = (error.stdout or "") + (error.stderr or "")
+        if isinstance(output, bytes):
+            output = output.decode("utf-8", errors="replace")
+    redact = _redactor(config, report_root, environ)
+    encoded = redact(output).encode("utf-8", errors="replace")[:MAX_LOG_BYTES]
+    (report_root / "logs").mkdir(exist_ok=True)
+    (report_root / "logs" / f"{case_id}.log").write_bytes(encoded)
+    if timed_out:
+        result = _base_result(case_id, "FAILED", "Acceptance driver timed out.")
+    elif not result_path.is_file():
+        result = _base_result(case_id, "FAILED", "Acceptance driver did not produce a result.")
+    else:
+        try:
+            result = json.loads(result_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            result = _base_result(case_id, "FAILED", "Acceptance driver result is unreadable.")
+    errors = _validate_driver_result(case_id, result, tuple(definition.get("required_metrics", ())))
+    if not isinstance(result, dict):
+        result = _base_result(case_id, "FAILED", "Acceptance driver result must be a JSON object.")
+    if errors or exit_code != 0 or result.get("status") != "PASSED":
+        if result.get("status") == "PASSED":
+            result["status"] = "FAILED"
+        result["runner_errors"] = errors + ([] if exit_code in {0, None} else ["DRIVER_EXIT_NONZERO"])
+    serialized = json.dumps(result, ensure_ascii=False)
+    leaked = [reference for reference in config.get("secret_env", {}).values() if environ.get(reference) and environ[reference] in serialized]
+    if leaked:
+        result = _base_result(case_id, "FAILED", "Acceptance result contained configured secret material.")
+        result["runner_errors"] = ["RESULT_SECRET_LEAK"]
+    result.update({
+        "started_utc": started_utc,
+        "finished_utc": _utc_now(),
+        "duration_ms": round((time.monotonic() - started) * 1000),
+        "exit_code": exit_code,
+        "log": f"logs/{case_id}.log",
+        "command": [
+            "python", str(definition["driver"]), "--config", "$CONFIG",
+            "--output", f"$REPORT/driver-results/{case_id}.json",
+        ],
+    })
+    return result
+
+
+def _write_junit(path: Path, results: list[dict[str, Any]], elapsed: float) -> None:
+    failures = sum(result["status"] != "PASSED" for result in results)
+    suite = ElementTree.Element("testsuite", {
+        "name": "OpenNexus phase3 production acceptance",
+        "tests": str(len(results)),
+        "failures": str(failures),
+        "errors": "0",
+        "skipped": "0",
+        "time": f"{elapsed:.3f}",
+    })
+    for result in results:
+        case = ElementTree.SubElement(suite, "testcase", {
+            "classname": "phase3.production",
+            "name": result["case_id"],
+            "time": f"{result.get('duration_ms', 0) / 1000:.3f}",
+        })
+        if result["status"] != "PASSED":
+            failure = ElementTree.SubElement(case, "failure", {"type": result["status"]})
+            failure.text = str(result.get("reason") or ",".join(result.get("runner_errors", ())))
+    ElementTree.ElementTree(suite).write(path, encoding="utf-8", xml_declaration=True)
+
+
+def _repository_evidence(config: dict[str, Any]) -> dict[str, Any]:
+    locks = {}
+    for relative in ("backend/uv.lock", "frontend/pnpm-lock.yaml", "frontend/src-tauri/Cargo.lock", "server sync/uv.lock", "community-server/uv.lock"):
+        path = ROOT / relative
+        if path.is_file():
+            locks[relative] = _sha256(path)
+    artifacts = {}
+    for name, raw in config.get("artifacts", {}).items():
+        path = Path(raw)
+        artifacts[name] = _sha256(path) if path.is_file() else None
+    commit = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, check=False).stdout.strip()
+    return {"commit": commit or None, "lock_sha256": locks, "artifact_sha256": artifacts}
+
+
+def execute(args: argparse.Namespace, environ: dict[str, str] | None = None) -> int:
+    environ = os.environ if environ is None else environ
+    if args.list_cases:
+        payload = {suite: list(cases) for suite, cases in CASE_SUITES.items()}
+        print(json.dumps(payload, ensure_ascii=False, indent=2) if args.json else "\n".join(f"{suite}: {', '.join(cases)}" for suite, cases in CASE_SUITES.items()))
+        return 0
+    if not args.suite or not args.config or not args.report_dir:
+        raise AcceptanceError("ARGUMENTS_REQUIRED")
+    selected = select_cases(args.suite, args.case)
+    config_path = Path(args.config).resolve()
+    config = load_config(config_path, environ)
+    if args.suite == "all" and not config.get("platform_profile"):
+        raise AcceptanceError("CONFIG_PLATFORM_PROFILE_REQUIRED")
+    prepare_isolated_root(config)
+    report_root = Path(args.report_dir).resolve(strict=False)
+    if _inside(report_root, VAULT_ROOT) or _inside(VAULT_ROOT, report_root):
+        raise AcceptanceError("REPORT_PERSONAL_VAULT_FORBIDDEN")
+    if report_root.exists() and (report_root.is_symlink() or not report_root.is_dir()):
+        raise AcceptanceError("REPORT_DIRECTORY_INVALID")
+    if report_root.exists() and any(report_root.iterdir()):
+        raise AcceptanceError("REPORT_DIRECTORY_NOT_EMPTY")
+    report_root.mkdir(parents=True, exist_ok=True)
+    started = time.monotonic()
+    results = [run_case(case_id, config_path, config, report_root, environ) for case_id in selected]
+    cases_root = report_root / "cases"
+    cases_root.mkdir()
+    for result in results:
+        (cases_root / f"{result['case_id']}.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    elapsed = time.monotonic() - started
+    _write_junit(report_root / "junit.xml", results, elapsed)
+    manifest = {"schema": SCHEMA, "suites": {name: list(cases) for name, cases in CASE_SUITES.items()}, "all_cases": list(ALL_CASES)}
+    if len(ALL_CASES) != 30 or len(set(ALL_CASES)) != 30:
+        raise AcceptanceError("CASE_MANIFEST_INCOMPLETE")
+    (report_root / "case-manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    passed = sum(result["status"] == "PASSED" for result in results)
+    summary = {
+        "schema": SCHEMA,
+        "run_id": config["run_id"],
+        "suite": args.suite,
+        "platform_profile": config["platform_profile"],
+        "status": "PASSED" if passed == len(results) else "NOT_PASSED",
+        "selected": list(selected),
+        "passed": passed,
+        "failed": len(results) - passed,
+        "duration_ms": round(elapsed * 1000),
+        "generated_utc": _utc_now(),
+        "config_sha256": _sha256(config_path),
+        "contains_credentials": False,
+        **_repository_evidence(config),
+    }
+    (report_root / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps({"status": summary["status"], "report_dir": str(report_root), "passed": passed, "total": len(results)}, ensure_ascii=False))
+    return 0 if summary["status"] == "PASSED" else 1
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser(description=__doc__)
+    result.add_argument("--suite", choices=(*CASE_SUITES, "all"))
+    result.add_argument("--case")
+    result.add_argument("--config")
+    result.add_argument("--report-dir")
+    result.add_argument("--list-cases", action="store_true")
+    result.add_argument("--json", action="store_true", help="Use JSON with --list-cases")
+    return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    try:
+        return execute(parser().parse_args(argv))
+    except AcceptanceError as error:
+        print(str(error), file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())