From 31ddb8d49cfc113c90b606095c046ddeb82dc595 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 08:53:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(acceptance):=20=E6=B7=BB=E5=8A=A0=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E5=85=B3=E9=97=AD=E7=9A=84=E7=AC=AC=E4=B8=89=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=E9=AA=8C=E6=94=B6=E8=BF=90=E8=A1=8C=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/ci.yml | 1 + .../tests/test_phase3_acceptance_runner.py | 133 +++++ .../第三阶段生产化工程规划与验收目标.md | 2 +- .../OpenNexus生产化实施进度-2026-09-08.md | 9 + docs/development/OpenNexus生产验收Runner.md | 27 + scripts/phase3-acceptance-config.example.json | 15 + scripts/phase3-production-acceptance.py | 7 + scripts/phase3_acceptance.py | 498 ++++++++++++++++++ 8 files changed, 691 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_phase3_acceptance_runner.py create mode 100644 docs/development/OpenNexus生产验收Runner.md create mode 100644 scripts/phase3-acceptance-config.example.json create mode 100644 scripts/phase3-production-acceptance.py create mode 100644 scripts/phase3_acceptance.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 458de1c..62a2a57 100644 --- a/.gitea/workflows/ci.yml +++ b/.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 diff --git a/backend/tests/test_phase3_acceptance_runner.py b/backend/tests/test_phase3_acceptance_runner.py new file mode 100644 index 0000000..1e4bea9 --- /dev/null +++ b/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 '' 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 diff --git a/docs/architecture/第三阶段生产化工程规划与验收目标.md b/docs/architecture/第三阶段生产化工程规划与验收目标.md index 943f2e5..1b8ba10 100644 --- a/docs/architecture/第三阶段生产化工程规划与验收目标.md +++ b/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 --config --report-dir `；suite取`sidecar/credentials/sandbox/extensions/sync-client/sync-service/e2e/all`，支持`--case `单例复跑。配置由环境模板产生，含临时数据根、服务URL、安装包路径、平台profile、seed及秘密的环境引用，不能含生产凭据。runner必须拒绝仓库个人Vault及未标记隔离环境，缺依赖/安装包/权限时非零退出；必测项skip也非零，不能只打印成功。 +验收入口已经建立：仓库根执行`python scripts/phase3-production-acceptance.py --suite --config --report-dir `；suite取`sidecar/credentials/sandbox/extensions/sync-client/sync-service/e2e/all`，支持`--case `单例复跑。配置由[隔离模板](../../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清单并校验报告完整性。 diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 4cb5903..40bc018 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/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 因本增量变为通过。后续逐项接入时仍须满足规划表的全部断言、实机/生产依赖和量化指标，完整生产化目标保持未完成。 diff --git a/docs/development/OpenNexus生产验收Runner.md b/docs/development/OpenNexus生产验收Runner.md new file mode 100644 index 0000000..f0160fa --- /dev/null +++ b/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/.json` 和脱敏的 `logs/.log`。摘要记录 commit、各锁文件 SHA-256、配置摘要与已提供安装产物摘要。日志将仓库、数据根、报告根、用户主目录和配置声明的秘密值替换为占位符，并限制为 10 MiB。报告目录必须为空，避免单例复跑覆盖原始证据。 + +当前 runner 与失败闭合行为已实现，各生产验收 ID 的完整 driver 尚未登记。因此现在运行任一 ID 会生成 `NOT_IMPLEMENTED` 证据并退出 1；这用于阻止误报，不是验收通过。 diff --git a/scripts/phase3-acceptance-config.example.json b/scripts/phase3-acceptance-config.example.json new file mode 100644 index 0000000..d127654 --- /dev/null +++ b/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": {} +} diff --git a/scripts/phase3-production-acceptance.py b/scripts/phase3-production-acceptance.py new file mode 100644 index 0000000..42d9fbf --- /dev/null +++ b/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()) diff --git a/scripts/phase3_acceptance.py b/scripts/phase3_acceptance.py new file mode 100644 index 0000000..840ac6c --- /dev/null +++ b/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())