From 4fc5d75fe01258bbc469373eafdcc2d8afb55cfa Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 8 Aug 2026 23:09:30 +0900 Subject: [PATCH 1/2] test(opencode): specify shadow detector verifier pool --- .../opencode-review-shadow-quality-ci.yml | 90 +++++ tests/opencode_review_shadow_test_support.py | 284 ++++++++++++++++ .../test_opencode_review_shadow_execution.py | 267 +++++++++++++++ tests/test_opencode_review_shadow_routing.py | 191 +++++++++++ .../test_opencode_review_shadow_validation.py | 319 ++++++++++++++++++ ...est_opencode_review_shadow_verification.py | 193 +++++++++++ 6 files changed, 1344 insertions(+) create mode 100644 .github/workflows/opencode-review-shadow-quality-ci.yml create mode 100644 tests/opencode_review_shadow_test_support.py create mode 100644 tests/test_opencode_review_shadow_execution.py create mode 100644 tests/test_opencode_review_shadow_routing.py create mode 100644 tests/test_opencode_review_shadow_validation.py create mode 100644 tests/test_opencode_review_shadow_verification.py diff --git a/.github/workflows/opencode-review-shadow-quality-ci.yml b/.github/workflows/opencode-review-shadow-quality-ci.yml new file mode 100644 index 000000000..a97e5cd31 --- /dev/null +++ b/.github/workflows/opencode-review-shadow-quality-ci.yml @@ -0,0 +1,90 @@ +name: OpenCode Review Shadow Quality CI + +on: + pull_request: + branches: + - main + - feat/opencode-review-decision-envelope + paths: + - ".github/workflows/opencode-review-shadow-quality-ci.yml" + - "scripts/ci/opencode_review_shadow.py" + - "scripts/ci/opencode_review_shadow_primitives.py" + - "scripts/ci/opencode_review_verify.py" + - "scripts/ci/run_opencode_semantic_review_pool.sh" + - "tests/opencode_review_shadow_test_support.py" + - "tests/test_opencode_review_shadow_*.py" + - "docs/doctoring/opencode-review-shadow-orchestration.md" + - "CHANGELOG.md" + +permissions: + contents: read + +concurrency: + group: opencode-review-shadow-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + shadow-review-quality: + name: shadow-review-quality + if: github.event_name != 'pull_request' || github.event.action != 'closed' + runs-on: ubuntu-24.04 + timeout-minutes: 15 + steps: + - name: Checkout exact source revision + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + + - name: Install exact hash-verified test runner dependencies + env: + PIP_DISABLE_PIP_VERSION_CHECK: "1" + PIP_NO_INPUT: "1" + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + cat >"${RUNNER_TEMP}/opencode-review-shadow-requirements.txt" <<'REQEOF' + coverage==7.15.2 --hash=sha256:b9a6367e4aff723e8ee8190836836124284e8fcd4265e307c844010cfa074f3f + iniconfig==2.1.0 --hash=sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760 + packaging==26.2 --hash=sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e + pluggy==1.6.0 --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746 + pygments==2.20.0 --hash=sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176 + pytest==9.1.1 --hash=sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c + REQEOF + python -m pip install \ + --only-binary=:all: \ + --require-hashes \ + -r "${RUNNER_TEMP}/opencode-review-shadow-requirements.txt" + + - name: Verify shadow detector-verifier contracts + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha || github.sha }}" + python -m coverage run \ + --branch \ + --source=scripts/ci \ + -m pytest \ + tests/test_opencode_review_shadow_routing.py \ + tests/test_opencode_review_shadow_execution.py \ + tests/test_opencode_review_shadow_verification.py \ + tests/test_opencode_review_shadow_validation.py \ + -q + python -m coverage report \ + --include='scripts/ci/opencode_review_shadow.py,scripts/ci/opencode_review_shadow_primitives.py,scripts/ci/opencode_review_verify.py' \ + --fail-under=100 \ + --show-missing + bash -n scripts/ci/run_opencode_semantic_review_pool.sh + python -m compileall -q \ + scripts/ci/opencode_review_shadow.py \ + scripts/ci/opencode_review_shadow_primitives.py \ + scripts/ci/opencode_review_verify.py \ + tests/opencode_review_shadow_test_support.py \ + tests/test_opencode_review_shadow_routing.py \ + tests/test_opencode_review_shadow_execution.py \ + tests/test_opencode_review_shadow_verification.py \ + tests/test_opencode_review_shadow_validation.py + git diff --exit-code diff --git a/tests/opencode_review_shadow_test_support.py b/tests/opencode_review_shadow_test_support.py new file mode 100644 index 000000000..561080dfd --- /dev/null +++ b/tests/opencode_review_shadow_test_support.py @@ -0,0 +1,284 @@ +"""Shared fixtures for OpenCode shadow detector-verifier tests.""" + +from __future__ import annotations + +import hashlib +import importlib.util +import json +from pathlib import Path +from types import ModuleType +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +SHADOW_PATH = ROOT / "scripts/ci/opencode_review_shadow.py" +VERIFY_PATH = ROOT / "scripts/ci/opencode_review_verify.py" +WRAPPER_PATH = ROOT / "scripts/ci/run_opencode_semantic_review_pool.sh" + + +def load_module(name: str, path: Path) -> ModuleType: + """Load one exact production module without package import side effects.""" + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +shadow = load_module("opencode_review_shadow", SHADOW_PATH) +verify = load_module("opencode_review_verify", VERIFY_PATH) + + +def digest_text(value: str) -> str: + """Return the canonical SHA-256 label used by evidence fixtures.""" + return f"sha256:{hashlib.sha256(value.encode('utf-8')).hexdigest()}" + + +def model( + descriptor_id: str, + model_id: str, + *, + roles: list[str], + efforts: list[str] | None = None, + agent_name: str = "ci-review", + provider_id: str = "nvidia-nim", +) -> dict[str, Any]: + """Build one provider-neutral, credential-free OpenCode model descriptor.""" + return { + "descriptor_id": descriptor_id, + "provider_id": provider_id, + "model_id": model_id, + "agent_name": agent_name, + "role_codes": roles, + "reasoning_efforts": efforts or ["low", "medium", "high"], + "prompt_sha256": digest_text(f"prompt:{descriptor_id}"), + } + + +def changed_file( + path: str = "src/example.py", + *, + language: str = "python", + additions: int = 20, + deletions: int = 5, + risk_tags: list[str] | None = None, +) -> dict[str, Any]: + """Build one exact-head changed-file routing record.""" + return { + "path": path, + "primary_language": language, + "additions": additions, + "deletions": deletions, + "risk_tags": risk_tags or [], + } + + +def request( + *, + files: list[dict[str, Any]] | None = None, + maximum_detector_attempts: int = 5, + maximum_recursive_verification_depth: int = 1, + models: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Build one strict shadow-review request and bounded model policy.""" + default_models = [ + model( + "general_super", + "nvidia/llama-3.3-nemotron-super-49b-v1.5", + roles=["general_detector", "correctness_detector"], + ), + model( + "security_ultra", + "nvidia/nemotron-3-ultra-550b-a55b", + roles=["security_detector", "workflow_detector"], + ), + model( + "numerical_mistral", + "mistralai/mistral-large-2-instruct", + roles=["numerical_detector", "data_model_detector"], + ), + model( + "experience_llama", + "meta/llama-3.3-70b-instruct", + roles=["experience_detector", "documentation_detector"], + ), + model( + "verifier_gemma", + "google/gemma-4-31b-it", + roles=["verifier", "recursive_verifier"], + agent_name="ci-review-fallback", + ), + model( + "verifier_deepseek", + "deepseek-ai/deepseek-v4-pro", + roles=["verifier", "recursive_verifier"], + agent_name="ci-review-fallback", + ), + ] + return { + "schema_version": "1.0", + "review_request_id": "review_request_001", + "repository": "ContextualWisdomLab/example", + "pull_request_number": 42, + "base_sha": "a" * 40, + "head_sha": "b" * 40, + "diff_sha256": digest_text("diff"), + "evidence_sha256": digest_text("evidence"), + "changed_files": files if files is not None else [changed_file()], + "policy": { + "shadow_mode": True, + "publication_enabled": False, + "maximum_detector_attempts": maximum_detector_attempts, + "maximum_recursive_verification_depth": maximum_recursive_verification_depth, + "attempt_timeout_seconds": 7200, + "model_pool": models if models is not None else default_models, + }, + } + + +def source_index() -> list[dict[str, Any]]: + """Build trusted source-line receipts for one candidate and one connected line.""" + return [ + { + "path": "src/example.py", + "line": 12, + "source_line_sha256": digest_text("if identity in seen:"), + "relationship": "changed", + }, + { + "path": "src/helper.py", + "line": 4, + "source_line_sha256": digest_text("return identity"), + "relationship": "connected", + }, + ] + + +def attempt( + attempt_id: str, + *, + phase: str, + role_code: str, + model_id: str, + provider_id: str = "nvidia-nim", + status: str = "complete", +) -> dict[str, Any]: + """Build one exact-head detector or verifier attempt receipt.""" + return { + "attempt_id": attempt_id, + "phase": phase, + "role_code": role_code, + "provider_id": provider_id, + "model_id": model_id, + "reviewed_head_sha": "b" * 40, + "status": status, + "output_sha256": digest_text(f"output:{attempt_id}"), + } + + +def candidate( + candidate_id: str = "candidate_001", + *, + detector_attempt_id: str = "detector_001", + path: str = "src/example.py", + line: int = 12, + source_line_sha256: str | None = None, + infrastructure_only: bool = False, + root_cause: str = "The identity set is not checked before aggregation.", +) -> dict[str, Any]: + """Build one complete normalized detector candidate.""" + return { + "candidate_id": candidate_id, + "detector_attempt_id": detector_attempt_id, + "reviewed_head_sha": "b" * 40, + "infrastructure_only": infrastructure_only, + "path": path, + "line": line, + "source_line_sha256": source_line_sha256 or digest_text("if identity in seen:"), + "defect_class": "correctness", + "severity": "high", + "blocking": True, + "trigger": "The input contains a duplicate exact-head identity.", + "impact": "The benchmark counts one pull request twice.", + "root_cause": root_cause, + "fix_direction": "Reject duplicate repository, PR, and head tuples.", + "regression_target": "Add a duplicate exact-head fixture.", + } + + +def verifier_decision( + candidate_id: str = "candidate_001", + *, + verifier_attempt_id: str = "verifier_001", + outcome: str = "supported", + source_line_sha256: str | None = None, +) -> dict[str, Any]: + """Build one normalized independent verifier decision.""" + return { + "candidate_id": candidate_id, + "verifier_attempt_id": verifier_attempt_id, + "outcome": outcome, + "reason": "Exact source and connected context support the candidate." + if outcome == "supported" + else "The candidate is not supported by the exact source.", + "source_line_sha256": source_line_sha256 or digest_text("if identity in seen:"), + } + + +def verification_input( + *, + candidates: list[dict[str, Any]] | None = None, + decisions: list[dict[str, Any]] | None = None, + minimum_independent_verifiers: int = 1, + require_model_diversity: bool = True, +) -> dict[str, Any]: + """Build one exact-head shadow verification bundle.""" + return { + "schema_version": "1.0", + "verification_id": "verification_001", + "repository": "ContextualWisdomLab/example", + "pull_request_number": 42, + "base_sha": "a" * 40, + "head_sha": "b" * 40, + "evidence_sha256": digest_text("evidence"), + "risk_tier": "high", + "verification_policy": { + "shadow_mode": True, + "publication_enabled": False, + "minimum_independent_verifiers": minimum_independent_verifiers, + "require_model_diversity": require_model_diversity, + }, + "source_index": source_index(), + "detector_attempts": [ + attempt( + "detector_001", + phase="detector", + role_code="general_detector", + model_id="nvidia/llama-3.3-nemotron-super-49b-v1.5", + ) + ], + "verifier_attempts": [ + attempt( + "verifier_001", + phase="verifier", + role_code="verifier", + model_id="google/gemma-4-31b-it", + ), + attempt( + "verifier_002", + phase="verifier", + role_code="recursive_verifier", + model_id="deepseek-ai/deepseek-v4-pro", + ), + ], + "candidates": candidates if candidates is not None else [candidate()], + "verifier_decisions": decisions + if decisions is not None + else [verifier_decision()], + } + + +def write_json(path: Path, value: Any) -> None: + """Write deterministic UTF-8 JSON for CLI and execution fixtures.""" + path.write_text( + json.dumps(value, ensure_ascii=False, sort_keys=True), encoding="utf-8" + ) diff --git a/tests/test_opencode_review_shadow_execution.py b/tests/test_opencode_review_shadow_execution.py new file mode 100644 index 000000000..7fbcffb10 --- /dev/null +++ b/tests/test_opencode_review_shadow_execution.py @@ -0,0 +1,267 @@ +"""Execution tests for the bounded non-publishing OpenCode shadow pool.""" + +from __future__ import annotations + +import json +import os +import stat +import subprocess +import sys +from pathlib import Path + +import pytest + +TEST_DIR = Path(__file__).resolve().parent +if str(TEST_DIR) not in sys.path: + sys.path.insert(0, str(TEST_DIR)) + +from opencode_review_shadow_test_support import ( + WRAPPER_PATH, + changed_file, + request, + shadow, + write_json, +) + + +def fake_opencode(path: Path, *, fail_role: str = "", sleep_role: str = "") -> Path: + """Create a deterministic fake OpenCode CLI that validates credential mapping.""" + path.write_text( + "#!/usr/bin/env python3\n" + "import json, os, sys, time\n" + "args = sys.argv[1:]\n" + "message = args[-1]\n" + "role = message.split('role=', 1)[1].split()[0]\n" + "assert os.environ.get('NVIDIA_API_KEY') == 'nim-secret'\n" + "if role == " + repr(sleep_role) + ": time.sleep(2)\n" + "if role == " + repr(fail_role) + ":\n" + " print('bounded fake failure', file=sys.stderr)\n" + " raise SystemExit(7)\n" + "print(json.dumps({'argv': args, 'role': role, 'secret_exposed': 'nim-secret' in json.dumps(args)}))\n", + encoding="utf-8", + ) + path.chmod(0o700) + return path + + +def run_inputs(tmp_path: Path) -> tuple[dict[str, object], Path, Path]: + """Create one plan, exact evidence file, and working directory.""" + evidence = tmp_path / "evidence.md" + evidence.write_text("evidence", encoding="utf-8") + workdir = tmp_path / "worktree" + workdir.mkdir() + return shadow.build_plan(request()), evidence, workdir + + +def test_execute_plan_invokes_detectors_before_verifiers_without_publication( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The runner uses fixed OpenCode arguments and passes detector output to verifiers.""" + plan, evidence, workdir = run_inputs(tmp_path) + executable = fake_opencode(tmp_path / "opencode") + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + output = tmp_path / "output" + manifest = shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=output, + opencode_binary=executable, + working_directory=workdir, + ) + assert manifest["shadow_mode"] is True + assert manifest["publication_enabled"] is False + assert manifest["plan_sha256"] == plan["plan_sha256"] + assert all(item["status"] == "complete" for item in manifest["attempts"]) + phases = [item["phase"] for item in manifest["attempts"]] + assert phases == ["detector", "verifier"] + + detector_record, verifier_record = manifest["attempts"] + detector_event = json.loads( + (output / detector_record["stdout_file"]).read_text(encoding="utf-8") + ) + verifier_event = json.loads( + (output / verifier_record["stdout_file"]).read_text(encoding="utf-8") + ) + for event, record in ( + (detector_event, detector_record), + (verifier_event, verifier_record), + ): + argv = event["argv"] + assert argv[0] == "run" + assert "--agent" in argv + assert "--model" in argv + assert "--variant" in argv + assert argv[argv.index("--format") + 1] == "json" + assert argv[argv.index("--dir") + 1] == str(workdir) + assert "--share" not in argv + assert "--command" not in argv + assert event["secret_exposed"] is False + assert record["stdout_sha256"].startswith("sha256:") + assert record["stderr_sha256"].startswith("sha256:") + verifier_files = [ + verifier_event["argv"][index + 1] + for index, value in enumerate(verifier_event["argv"]) + if value == "--file" + ] + assert str(evidence) in verifier_files + assert str(output / detector_record["stdout_file"]) in verifier_files + assert manifest["execution_sha256"].startswith("sha256:") + + +def test_runner_records_partial_failure_and_keeps_independent_work_product( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """One detector failure is isolated while a successful detector still feeds verification.""" + value = request( + files=[ + changed_file("src/auth.py", risk_tags=["security"]), + ] + ) + plan = shadow.build_plan(value) + executable = fake_opencode(tmp_path / "opencode", fail_role="security_detector") + evidence = tmp_path / "evidence.md" + evidence.write_text("evidence", encoding="utf-8") + workdir = tmp_path / "worktree" + workdir.mkdir() + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + manifest = shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=workdir, + ) + statuses = {item["role_code"]: item["status"] for item in manifest["attempts"]} + assert statuses["general_detector"] == "complete" + assert statuses["security_detector"] == "failed" + assert statuses["verifier"] == "complete" + assert manifest["completed_attempt_count"] == 2 + assert manifest["failed_attempt_count"] == 1 + + +def test_all_detector_failures_skip_dependent_verifier( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A verifier is not run on an empty detector evidence set.""" + plan, evidence, workdir = run_inputs(tmp_path) + executable = fake_opencode(tmp_path / "opencode", fail_role="general_detector") + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + manifest = shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=workdir, + ) + assert [item["status"] for item in manifest["attempts"]] == [ + "failed", + "dependency_failed", + ] + assert manifest["failed_attempt_count"] == 2 + + +def test_timeout_is_bounded_and_recorded_without_exception_escape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A slow model attempt is terminated and downstream verification is skipped.""" + value = request() + value["policy"]["attempt_timeout_seconds"] = 1 + plan = shadow.build_plan(value) + evidence = tmp_path / "evidence.md" + evidence.write_text("evidence", encoding="utf-8") + workdir = tmp_path / "worktree" + workdir.mkdir() + executable = fake_opencode(tmp_path / "opencode", sleep_role="general_detector") + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + manifest = shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=workdir, + ) + assert [item["status"] for item in manifest["attempts"]] == [ + "timed_out", + "dependency_failed", + ] + + +def test_execution_fails_before_process_start_on_untrusted_boundary( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Credential, evidence, executable, and worktree boundaries fail closed.""" + plan, evidence, workdir = run_inputs(tmp_path) + executable = fake_opencode(tmp_path / "opencode") + monkeypatch.delenv("NVIDIA_NIM_API_KEY", raising=False) + with pytest.raises(shadow.ShadowExecutionError, match="NVIDIA_NIM_API_KEY"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=workdir, + ) + + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + evidence.write_text("changed", encoding="utf-8") + with pytest.raises(shadow.ShadowExecutionError, match="evidence_sha256"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=workdir, + ) + + evidence.write_text("evidence", encoding="utf-8") + executable.chmod(stat.S_IRWXU | stat.S_IWGRP) + with pytest.raises(shadow.ShadowExecutionError, match="writable"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=workdir, + ) + + executable.chmod(0o700) + symlink = tmp_path / "opencode-link" + symlink.symlink_to(executable) + with pytest.raises(shadow.ShadowExecutionError, match="symlink"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=symlink, + working_directory=workdir, + ) + + +def test_shell_wrapper_is_thin_non_publishing_and_functional(tmp_path: Path) -> None: + """The permanent wrapper delegates to Python and has no GitHub mutation path.""" + source = WRAPPER_PATH.read_text(encoding="utf-8") + assert "exec python3" in source + assert "opencode_review_shadow.py" in source + for forbidden in ("gh ", "curl ", "git push", "pulls/", "reviews"): + assert forbidden not in source + subprocess.run(["bash", "-n", str(WRAPPER_PATH)], check=True) + + request_path = tmp_path / "request.json" + output_path = tmp_path / "plan.json" + write_json(request_path, request()) + completed = subprocess.run( + [ + "bash", + str(WRAPPER_PATH), + "plan", + "--input", + str(request_path), + "--output", + str(output_path), + ], + check=False, + text=True, + capture_output=True, + ) + assert completed.returncode == 0, completed.stderr + assert json.loads(output_path.read_text(encoding="utf-8"))["shadow_mode"] is True diff --git a/tests/test_opencode_review_shadow_routing.py b/tests/test_opencode_review_shadow_routing.py new file mode 100644 index 000000000..7d0e2911d --- /dev/null +++ b/tests/test_opencode_review_shadow_routing.py @@ -0,0 +1,191 @@ +"""Routing tests for risk-adaptive OpenCode shadow orchestration.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +TEST_DIR = Path(__file__).resolve().parent +if str(TEST_DIR) not in sys.path: + sys.path.insert(0, str(TEST_DIR)) + +from opencode_review_shadow_test_support import changed_file, request, shadow + + +def roles(plan: dict[str, object]) -> list[str]: + """Return ordered attempt roles from one normalized shadow plan.""" + return [item["role_code"] for item in plan["attempts"]] # type: ignore[index] + + +def test_low_risk_documentation_change_uses_one_detector_and_one_verifier() -> None: + """Small documentation-only changes must avoid unnecessary multi-agent compute.""" + plan = shadow.build_plan( + request( + files=[ + changed_file( + "docs/usage.md", + language="markdown", + additions=12, + deletions=2, + risk_tags=["documentation"], + ) + ] + ) + ) + assert plan["risk_tier"] == "low" + assert plan["diff_size_bucket"] == "small" + assert roles(plan) == ["general_detector", "verifier"] + assert [item["reasoning_effort"] for item in plan["attempts"]] == [ + "low", + "medium", + ] + assert plan["shadow_mode"] is True + assert plan["publication_enabled"] is False + assert plan["maximum_recursive_verification_depth"] == 0 + + +def test_ordinary_source_change_uses_general_detector_and_independent_verifier() -> None: + """Ordinary source changes receive a semantic detector plus a distinct verifier.""" + plan = shadow.build_plan(request()) + assert plan["risk_tier"] == "standard" + assert roles(plan) == ["general_detector", "verifier"] + detector, verifier = plan["attempts"] + assert detector["model_id"] != verifier["model_id"] + assert detector["phase"] == "detector" + assert verifier["phase"] == "verifier" + assert detector["reasoning_effort"] == "medium" + assert verifier["reasoning_effort"] == "medium" + + +def test_security_workflow_and_data_model_changes_add_specialists() -> None: + """Material trust changes allocate diverse specialists and a high-effort verifier.""" + plan = shadow.build_plan( + request( + files=[ + changed_file( + ".github/workflows/release.yml", + language="yaml", + additions=90, + deletions=12, + risk_tags=["security", "workflow", "release"], + ), + changed_file( + "database/migrations/0009_account_policy.sql", + language="sql", + additions=80, + deletions=10, + risk_tags=["data_model", "migration"], + ), + ] + ) + ) + assert plan["risk_tier"] == "critical" + assert plan["diff_size_bucket"] == "medium" + assert roles(plan) == [ + "general_detector", + "security_detector", + "workflow_detector", + "data_model_detector", + "verifier", + "recursive_verifier", + ] + assert len( + { + (item["provider_id"], item["model_id"]) + for item in plan["attempts"] + if item["phase"] == "detector" + } + ) >= 3 + assert plan["maximum_recursive_verification_depth"] == 1 + assert all(item["reasoning_effort"] == "high" for item in plan["attempts"]) + assert set(plan["risk_reasons"]) >= { + "security", + "workflow", + "release", + "data_model", + "migration", + } + + +def test_numerical_and_experience_changes_route_to_role_specific_detectors() -> None: + """Numerical and buyer-facing changes use relevant specialists without fixed topology.""" + plan = shadow.build_plan( + request( + files=[ + changed_file( + "crates/estimator/src/kernel.rs", + language="rust", + additions=310, + deletions=70, + risk_tags=["numerical", "performance"], + ), + changed_file( + "apps/web/src/ReportView.tsx", + language="typescript", + additions=100, + deletions=20, + risk_tags=["experience", "accessibility", "public_api"], + ), + ] + ) + ) + assert plan["risk_tier"] == "high" + assert roles(plan) == [ + "general_detector", + "numerical_detector", + "experience_detector", + "verifier", + ] + assert plan["diff_size_bucket"] == "large" + assert plan["maximum_recursive_verification_depth"] == 0 + + +def test_detector_budget_is_fail_closed_instead_of_silently_dropping_specialists() -> None: + """A detector limit below the required specialist set must reject the plan.""" + value = request( + maximum_detector_attempts=2, + files=[ + changed_file( + ".github/workflows/security.yml", + language="yaml", + risk_tags=["security", "workflow", "release"], + ) + ], + ) + with pytest.raises(shadow.InsufficientPoolError, match="detector attempt budget"): + shadow.build_plan(value) + + +def test_missing_role_or_model_diversity_is_rejected() -> None: + """High-risk review must not degrade to a general model or self-verification.""" + no_security = request() + no_security["changed_files"] = [ + changed_file("src/auth.py", risk_tags=["security"]) + ] + no_security["policy"]["model_pool"] = [ + item + for item in no_security["policy"]["model_pool"] + if "security_detector" not in item["role_codes"] + ] + with pytest.raises(shadow.InsufficientPoolError, match="security_detector"): + shadow.build_plan(no_security) + + no_verifier_diversity = request() + only = no_verifier_diversity["policy"]["model_pool"][0] + only["role_codes"].append("verifier") + no_verifier_diversity["policy"]["model_pool"] = [only] + with pytest.raises(shadow.InsufficientPoolError, match="independent verifier"): + shadow.build_plan(no_verifier_diversity) + + +def test_same_request_and_policy_produce_one_content_addressed_plan() -> None: + """Routing is deterministic and records exact evidence and policy receipts.""" + first = shadow.build_plan(request()) + second = shadow.build_plan(request()) + assert first == second + assert first["input_sha256"].startswith("sha256:") + assert first["plan_sha256"].startswith("sha256:") + assert all(item["prompt_sha256"].startswith("sha256:") for item in first["attempts"]) + assert all("credential" not in key for item in first["attempts"] for key in item) diff --git a/tests/test_opencode_review_shadow_validation.py b/tests/test_opencode_review_shadow_validation.py new file mode 100644 index 000000000..b66f1490f --- /dev/null +++ b/tests/test_opencode_review_shadow_validation.py @@ -0,0 +1,319 @@ +"""Strict validation and CLI tests for shadow routing and verification.""" + +from __future__ import annotations + +import json +import runpy +import sys +from pathlib import Path +from typing import Any + +import pytest + +TEST_DIR = Path(__file__).resolve().parent +if str(TEST_DIR) not in sys.path: + sys.path.insert(0, str(TEST_DIR)) + +from opencode_review_shadow_test_support import ( + SHADOW_PATH, + VERIFY_PATH, + candidate, + request, + shadow, + verification_input, + verifier_decision, + verify, + write_json, +) + + +@pytest.mark.parametrize( + ("mutate", "message"), + [ + (lambda value: value.update({"unexpected": True}), "unknown fields"), + ( + lambda value: value["policy"].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["changed_files"][0].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["policy"]["model_pool"][0].update( + {"unexpected": True} + ), + "unknown fields", + ), + (lambda value: value.update({"schema_version": "2.0"}), "schema_version"), + (lambda value: value.update({"pull_request_number": True}), "integer"), + ( + lambda value: value["policy"].update({"shadow_mode": False}), + "shadow_mode", + ), + ( + lambda value: value["policy"].update({"publication_enabled": True}), + "publication_enabled", + ), + ( + lambda value: value["changed_files"][0].update({"path": "../secret"}), + "relative source path", + ), + ( + lambda value: value["changed_files"][0].update({"additions": True}), + "integer", + ), + ( + lambda value: value["policy"]["model_pool"][0].update( + {"prompt_sha256": "sha256:bad"} + ), + "sha256", + ), + ( + lambda value: value["policy"].update({"attempt_timeout_seconds": 0}), + "timeout", + ), + ], +) +def test_routing_request_rejects_malformed_or_extensible_evidence( + mutate: Any, message: str +) -> None: + """Every request, policy, file, and model layer must fail closed.""" + value = request() + mutate(value) + with pytest.raises(shadow.ShadowValidationError, match=message): + shadow.build_plan(value) + + +def test_routing_rejects_empty_files_duplicate_models_and_invalid_roles() -> None: + """The planner requires material evidence and unique supported model descriptors.""" + empty = request(files=[]) + with pytest.raises(shadow.ShadowValidationError, match="changed_files"): + shadow.build_plan(empty) + + duplicate = request() + duplicate["policy"]["model_pool"].append( + dict(duplicate["policy"]["model_pool"][0]) + ) + with pytest.raises(shadow.ShadowValidationError, match="descriptor_id"): + shadow.build_plan(duplicate) + + invalid_role = request() + invalid_role["policy"]["model_pool"][0]["role_codes"] = ["administrator"] + with pytest.raises(shadow.ShadowValidationError, match="role_codes"): + shadow.build_plan(invalid_role) + + +@pytest.mark.parametrize( + ("mutate", "message"), + [ + (lambda value: value.update({"unexpected": True}), "unknown fields"), + ( + lambda value: value["verification_policy"].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["source_index"][0].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["detector_attempts"][0].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["candidates"][0].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["verifier_decisions"][0].update( + {"unexpected": True} + ), + "unknown fields", + ), + (lambda value: value.update({"head_sha": "main"}), "commit SHA"), + ( + lambda value: value["verification_policy"].update( + {"shadow_mode": False} + ), + "shadow_mode", + ), + ( + lambda value: value["verification_policy"].update( + {"publication_enabled": True} + ), + "publication_enabled", + ), + ( + lambda value: value["verification_policy"].update( + {"minimum_independent_verifiers": True} + ), + "integer", + ), + ( + lambda value: value["detector_attempts"][0].update( + {"reviewed_head_sha": "c" * 40} + ), + "reviewed_head_sha", + ), + ( + lambda value: value["candidates"][0].update( + {"reviewed_head_sha": "c" * 40} + ), + "reviewed_head_sha", + ), + ( + lambda value: value["verifier_decisions"][0].update( + {"outcome": "uncertain"} + ), + "outcome", + ), + ], +) +def test_verification_bundle_rejects_malformed_or_stale_evidence( + mutate: Any, message: str +) -> None: + """Every verification layer must remain strict and exact-head bound.""" + value = verification_input() + mutate(value) + with pytest.raises(verify.VerificationValidationError, match=message): + verify.verify_bundle(value) + + +def test_verification_rejects_duplicate_or_unknown_identity_references() -> None: + """Source, attempt, candidate, and decision identities cannot be duplicated or forged.""" + duplicate_source = verification_input() + duplicate_source["source_index"].append(dict(duplicate_source["source_index"][0])) + with pytest.raises(verify.VerificationValidationError, match="source identity"): + verify.verify_bundle(duplicate_source) + + duplicate_attempt = verification_input() + duplicate_attempt["detector_attempts"].append( + dict(duplicate_attempt["detector_attempts"][0]) + ) + with pytest.raises(verify.VerificationValidationError, match="attempt_id"): + verify.verify_bundle(duplicate_attempt) + + duplicate_candidate = verification_input() + duplicate_candidate["candidates"].append(dict(duplicate_candidate["candidates"][0])) + with pytest.raises(verify.VerificationValidationError, match="candidate_id"): + verify.verify_bundle(duplicate_candidate) + + unknown_candidate = verification_input( + decisions=[verifier_decision("unknown_candidate")] + ) + with pytest.raises(verify.VerificationValidationError, match="unknown candidate"): + verify.verify_bundle(unknown_candidate) + + unknown_attempt = verification_input( + candidates=[candidate(detector_attempt_id="unknown_detector")] + ) + with pytest.raises(verify.VerificationValidationError, match="unknown detector"): + verify.verify_bundle(unknown_attempt) + + +def test_strict_json_loaders_reject_duplicate_keys_and_nonfinite_numbers( + tmp_path: Path, +) -> None: + """Both tools reject ambiguous JSON objects and Python numeric extensions.""" + for module in (shadow, verify): + duplicate = tmp_path / f"duplicate-{module.__name__}.json" + duplicate.write_text('{"schema_version":"1.0","schema_version":"1.0"}') + with pytest.raises(module.validation_error_type(), match="duplicate JSON key"): + module.load_json(duplicate) + + nonfinite = tmp_path / f"nonfinite-{module.__name__}.json" + nonfinite.write_text('{"line": Infinity}') + with pytest.raises(module.validation_error_type(), match="non-finite JSON number"): + module.load_json(nonfinite) + + +def test_plan_and_verification_clis_write_atomic_outputs_with_stable_statuses( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Offline CLIs distinguish success from malformed evidence and leave no temp files.""" + request_path = tmp_path / "request.json" + plan_path = tmp_path / "nested" / "plan.json" + write_json(request_path, request()) + assert ( + shadow.main( + ["plan", "--input", str(request_path), "--output", str(plan_path)] + ) + == 0 + ) + assert json.loads(plan_path.read_text(encoding="utf-8"))["shadow_mode"] is True + assert not plan_path.with_name(f".{plan_path.name}.tmp").exists() + + request_path.write_text("[]", encoding="utf-8") + assert ( + shadow.main( + ["plan", "--input", str(request_path), "--output", str(plan_path)] + ) + == 2 + ) + assert "shadow review request rejected" in capsys.readouterr().err + + bundle_path = tmp_path / "bundle.json" + report_path = tmp_path / "nested" / "verification.json" + write_json(bundle_path, verification_input()) + assert ( + verify.main( + ["--input", str(bundle_path), "--output", str(report_path)] + ) + == 0 + ) + assert json.loads(report_path.read_text(encoding="utf-8"))[ + "publication_enabled" + ] is False + assert not report_path.with_name(f".{report_path.name}.tmp").exists() + + bundle_path.write_text("[]", encoding="utf-8") + assert ( + verify.main( + ["--input", str(bundle_path), "--output", str(report_path)] + ) + == 2 + ) + assert "shadow verification rejected" in capsys.readouterr().err + + +def test_module_entrypoints_and_public_docstrings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Direct execution routes through tested CLIs and every public callable is documented.""" + request_path = tmp_path / "request.json" + plan_path = tmp_path / "plan.json" + write_json(request_path, request()) + monkeypatch.setattr( + "sys.argv", + [ + str(SHADOW_PATH), + "plan", + "--input", + str(request_path), + "--output", + str(plan_path), + ], + ) + with pytest.raises(SystemExit, match="0"): + runpy.run_path(str(SHADOW_PATH), run_name="__main__") + + bundle_path = tmp_path / "bundle.json" + report_path = tmp_path / "report.json" + write_json(bundle_path, verification_input()) + monkeypatch.setattr( + "sys.argv", + [str(VERIFY_PATH), "--input", str(bundle_path), "--output", str(report_path)], + ) + with pytest.raises(SystemExit, match="0"): + runpy.run_path(str(VERIFY_PATH), run_name="__main__") + + for module in (shadow, verify): + missing = [ + name + for name, value in vars(module).items() + if not name.startswith("_") + and (isinstance(value, type) or callable(value)) + and getattr(value, "__module__", None) == module.__name__ + and not getattr(value, "__doc__", None) + ] + assert missing == [] diff --git a/tests/test_opencode_review_shadow_verification.py b/tests/test_opencode_review_shadow_verification.py new file mode 100644 index 000000000..b868071fd --- /dev/null +++ b/tests/test_opencode_review_shadow_verification.py @@ -0,0 +1,193 @@ +"""Verification tests for normalized detector and independent verifier outputs.""" + +from __future__ import annotations + +import copy +import sys +from pathlib import Path + +import pytest + +TEST_DIR = Path(__file__).resolve().parent +if str(TEST_DIR) not in sys.path: + sys.path.insert(0, str(TEST_DIR)) + +from opencode_review_shadow_test_support import ( + candidate, + digest_text, + verifier_decision, + verification_input, + verify, +) + + +def test_supported_source_candidate_becomes_shadow_finding_without_publication() -> None: + """A fully supported current-head candidate is retained only in shadow output.""" + report = verify.verify_bundle(verification_input()) + assert report["shadow_mode"] is True + assert report["publication_enabled"] is False + assert report["published_findings"] == [] + assert len(report["shadow_findings"]) == 1 + finding = report["shadow_findings"][0] + assert finding["path"] == "src/example.py" + assert finding["line"] == 12 + assert finding["detector_attempt_ids"] == ["detector_001"] + assert finding["verifier_attempt_ids"] == ["verifier_001"] + assert finding["finding_fingerprint"].startswith("sha256:") + assert report["metrics"] == { + "candidate_count": 1, + "accepted_finding_count": 1, + "rejected_candidate_count": 0, + "duplicate_candidate_count": 0, + "infrastructure_only_candidate_count": 0, + "unsupported_candidate_count": 0, + "source_contract_failure_count": 0, + "insufficient_verifier_count": 0, + } + assert report["verification_sha256"].startswith("sha256:") + + +def test_infrastructure_only_candidate_is_rejected_without_source_authority() -> None: + """Coverage or check commentary cannot enter the semantic shadow finding set.""" + value = verification_input(candidates=[candidate(infrastructure_only=True)]) + report = verify.verify_bundle(value) + assert report["shadow_findings"] == [] + assert report["metrics"]["infrastructure_only_candidate_count"] == 1 + assert report["rejected_candidates"][0]["reason_code"] == "infrastructure_only" + assert "path" not in report["rejected_candidates"][0] + assert "line" not in report["rejected_candidates"][0] + + +def test_source_receipt_mismatch_is_rejected_not_silently_reanchored() -> None: + """A candidate and verifier decision must match the trusted exact-line receipt.""" + wrong = digest_text("different line") + value = verification_input( + candidates=[candidate(source_line_sha256=wrong)], + decisions=[verifier_decision(source_line_sha256=wrong)], + ) + report = verify.verify_bundle(value) + assert report["shadow_findings"] == [] + assert report["metrics"]["source_contract_failure_count"] == 1 + assert report["rejected_candidates"][0]["reason_code"] == "source_receipt_mismatch" + + +def test_rejected_or_missing_verifier_support_cannot_pass() -> None: + """Detector prose alone is never a publishable or accepted shadow finding.""" + rejected = verify.verify_bundle( + verification_input(decisions=[verifier_decision(outcome="rejected")]) + ) + assert rejected["shadow_findings"] == [] + assert rejected["metrics"]["unsupported_candidate_count"] == 1 + + missing = verify.verify_bundle(verification_input(decisions=[])) + assert missing["shadow_findings"] == [] + assert missing["metrics"]["insufficient_verifier_count"] == 1 + + +def test_high_assurance_policy_requires_two_distinct_verifier_models() -> None: + """Critical findings can require diverse independent verification rather than repetition.""" + value = verification_input( + minimum_independent_verifiers=2, + decisions=[ + verifier_decision(verifier_attempt_id="verifier_001"), + verifier_decision(verifier_attempt_id="verifier_002"), + ], + ) + report = verify.verify_bundle(value) + assert len(report["shadow_findings"]) == 1 + assert report["shadow_findings"][0]["verifier_attempt_ids"] == [ + "verifier_001", + "verifier_002", + ] + + same_model = copy.deepcopy(value) + same_model["verifier_attempts"][1]["model_id"] = same_model["verifier_attempts"][0][ + "model_id" + ] + report = verify.verify_bundle(same_model) + assert report["shadow_findings"] == [] + assert report["metrics"]["insufficient_verifier_count"] == 1 + + +def test_detector_and_verifier_model_must_be_independent_when_policy_requires() -> None: + """A model cannot verify its own finding under the diversity policy.""" + value = verification_input() + value["verifier_attempts"][0]["model_id"] = value["detector_attempts"][0][ + "model_id" + ] + report = verify.verify_bundle(value) + assert report["shadow_findings"] == [] + assert report["metrics"]["insufficient_verifier_count"] == 1 + + +def test_duplicate_candidates_collapse_to_one_finding_with_all_receipts() -> None: + """Equivalent detector findings are deduplicated by source and normalized root cause.""" + second = candidate( + "candidate_002", + detector_attempt_id="detector_002", + root_cause=" The identity set is not checked before aggregation. ", + ) + value = verification_input( + candidates=[candidate(), second], + decisions=[ + verifier_decision("candidate_001"), + verifier_decision("candidate_002"), + ], + ) + value["detector_attempts"].append( + { + **value["detector_attempts"][0], + "attempt_id": "detector_002", + "model_id": "mistralai/mistral-large-2-instruct", + "output_sha256": digest_text("output:detector_002"), + } + ) + report = verify.verify_bundle(value) + assert len(report["shadow_findings"]) == 1 + assert report["shadow_findings"][0]["detector_attempt_ids"] == [ + "detector_001", + "detector_002", + ] + assert report["metrics"]["duplicate_candidate_count"] == 1 + + +def test_failed_detector_or_verifier_attempt_cannot_supply_evidence() -> None: + """Only completed exact-head attempts count toward detector or verifier evidence.""" + failed_detector = verification_input() + failed_detector["detector_attempts"][0]["status"] = "failed" + report = verify.verify_bundle(failed_detector) + assert report["shadow_findings"] == [] + assert report["rejected_candidates"][0]["reason_code"] == "detector_not_complete" + + failed_verifier = verification_input() + failed_verifier["verifier_attempts"][0]["status"] = "failed" + report = verify.verify_bundle(failed_verifier) + assert report["shadow_findings"] == [] + assert report["metrics"]["insufficient_verifier_count"] == 1 + + +def test_equivalent_bundle_produces_deterministic_sorted_output() -> None: + """Candidate order cannot change fingerprints, metrics, or output receipts.""" + c1 = candidate("candidate_b") + c2 = candidate( + "candidate_a", + path="src/helper.py", + line=4, + source_line_sha256=digest_text("return identity"), + root_cause="The helper returns an unsafe identity.", + ) + d1 = verifier_decision("candidate_b") + d2 = verifier_decision( + "candidate_a", source_line_sha256=digest_text("return identity") + ) + first = verify.verify_bundle( + verification_input(candidates=[c1, c2], decisions=[d1, d2]) + ) + second = verify.verify_bundle( + verification_input(candidates=[c2, c1], decisions=[d2, d1]) + ) + assert first == second + assert [item["path"] for item in first["shadow_findings"]] == [ + "src/example.py", + "src/helper.py", + ] From 94a54be6b3d424b59b5a32854099fff24f5794e6 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 22:27:09 +0900 Subject: [PATCH 2/2] feat(opencode): implement bounded shadow review pool --- CHANGELOG.md | 1 + .../opencode-review-shadow-orchestration.md | 116 ++++++ scripts/ci/opencode_review_shadow.py | 348 ++++++++++++++++++ .../ci/opencode_review_shadow_primitives.py | 123 +++++++ scripts/ci/opencode_review_verify.py | 279 ++++++++++++++ .../ci/run_opencode_semantic_review_pool.sh | 5 + .../test_opencode_review_shadow_execution.py | 95 ++++- .../test_opencode_review_shadow_validation.py | 92 +++++ 8 files changed, 1057 insertions(+), 2 deletions(-) create mode 100644 docs/doctoring/opencode-review-shadow-orchestration.md create mode 100644 scripts/ci/opencode_review_shadow.py create mode 100644 scripts/ci/opencode_review_shadow_primitives.py create mode 100644 scripts/ci/opencode_review_verify.py create mode 100755 scripts/ci/run_opencode_semantic_review_pool.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ab3862b7..c4f118df6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ Semantic Versioning where the repository publishes a release. ### Added +- Added a production-independent OpenCode shadow review pool with deterministic risk-adaptive routing, bounded non-shell detector and verifier execution, child-only NVIDIA credential mapping, secret redaction, immutable evidence receipts, strict output-directory boundaries, partial-failure isolation, and non-publishing verified findings backed by 100% owned production statement, branch, and callable-docstring evidence. - Added an exact-head OpenCode decision envelope that keeps semantic source verdicts independent from coverage, checks, approval, and branch-protection merge readiness; emits path-free infrastructure blockers; fails closed on stale or malformed evidence; and preserves 100% production statement, branch, and public-docstring evidence. - Added deterministic exact-head corpus sampling and blinded two-expert-plus-adjudicator gold-freeze tooling, with strict JSON, immutable evidence receipts, hard language/size/risk/defect coverage, atomic outputs, stable failure classes, and permanent 100% production statement/branch/docstring evidence. - Added an empirical OpenCode review-quality benchmark, fail-closed scorer, exact-head quality workflow, and APA 7th doctoring that keep lifecycle-yield evidence separate from head-matched expert-gold precision and recall, require Wilson-bound non-inferiority before any CodeRabbit-parity claim, and preserve 100% production statement/branch/docstring evidence. diff --git a/docs/doctoring/opencode-review-shadow-orchestration.md b/docs/doctoring/opencode-review-shadow-orchestration.md new file mode 100644 index 000000000..12a3e7d14 --- /dev/null +++ b/docs/doctoring/opencode-review-shadow-orchestration.md @@ -0,0 +1,116 @@ +# OpenCode shadow review orchestration + +Status: Active pull-request implementation +Date: 2026-08-11 +Owner: ContextualWisdomLab central review infrastructure + +## Purpose and maturity + +This implementation provides a production-independent detector–verifier pool for exact-head OpenCode review experiments. It allocates a bounded review topology from trusted pull-request metadata, runs detectors before independent verifiers, and emits validated `shadow_findings` for evaluation. + +The capability is `active_pr`, not `implemented_on_protected_main`. It cannot publish a GitHub comment, review, check, approval, merge, branch update, or release. A later integration must preserve that authority separation and earn protected-main operational evidence before the capability is described as deployed. + +## Components + +| Component | Responsibility | +|---|---| +| `opencode_review_shadow_primitives.py` | Strict JSON parsing, schema validation, canonical serialization, digests, safe paths, and atomic output writes. | +| `opencode_review_shadow.py` | Deterministic routing plans and bounded OpenCode attempt execution. | +| `opencode_review_verify.py` | Exact-head receipt validation, independent verification, semantic deduplication, and shadow-only reporting. | +| `run_opencode_semantic_review_pool.sh` | Thin plan-only command wrapper with no GitHub mutation path. | + +The modules are deliberately standalone. They do not import or modify the production dispatch workflow, reviewer identities, merge policy, or release controls. + +## Deterministic routing + +The plan command accepts one strict request document containing the repository identity, pull-request number, `base_sha`, `head_sha`, changed-file metadata, bounded model pools, and the detector budget. Unknown fields, malformed SHA values, duplicate JSON keys, non-finite numbers, unsafe paths, invalid model entries, or an impossible role budget fail closed. + +Routing classifies the change into small, medium, or large diff buckets and low, standard, high, or critical risk. The resulting roles are selected only when material: + +- a general semantic detector for ordinary source changes; +- security, workflow, data-model, numerical, experience, or documentation specialists for corresponding evidence; +- an independent verifier for every executable topology; and +- bounded recursive verification for high-risk disagreement when the supplied budget permits it. + +The canonical plan contains no credential. Its `plan_sha256` binds the complete normalized request and selected attempts, so a caller can prove which exact plan it executed. + +## Execution boundary + +The run command validates the plan, executable, evidence files, worktree, and output location before starting a child process. Each attempt uses a fixed argument vector rather than a shell: + +```text +opencode run --agent --model --variant --format json +``` + +Detectors run before verifiers. A verifier receives the successful detector evidence selected by the plan; it is marked `dependency_failed` when its detector dependency did not produce trusted evidence. Timeouts and non-zero exits are recorded per attempt, allowing one failure to remain isolated without converting infrastructure failure into a semantic source finding. + +The execution environment is intentionally minimal. `NVIDIA_NIM_API_KEY` is read by the trusted parent and mapped only to the child process as `NVIDIA_API_KEY`. It is never placed in the plan or process arguments. Exact secret echoes in child stdout or stderr are replaced with `[REDACTED_NVIDIA_API_KEY]` before evidence is persisted or hashed. + +The output directory must be absent or an empty private directory. Symlinks, non-directories, group/world-writable directories, and non-empty directories are rejected. A newly created output directory uses mode `0700`. Evidence is written atomically through a temporary sibling file and replacement. + +## Evidence and verification + +The verifier accepts strict execution and source-receipt bundles bound to one repository, pull request, base SHA, and head SHA. It rejects stale heads, failed attempts presented as findings, unsupported source paths or lines, infrastructure-only claims rendered as source defects, detector self-verification, insufficient verifier count, and required model-diversity violations. + +Candidate findings must have trusted source-line evidence and a successful independent verifier receipt. Normalization and semantic deduplication are deterministic. The report contains: + +```json +{ + "publication_enabled": false, + "published_findings": [], + "shadow_findings": [] +} +``` + +`shadow_findings` are evaluation evidence only. They are not formal GitHub reviews, qualifying human approvals, merge decisions, or release authorization. + +## Operator procedure + +Generate a content-addressed plan: + +```bash +scripts/ci/run_opencode_semantic_review_pool.sh plan \ + --request request.json \ + --output plan.json +``` + +Run the validated attempts from a trusted worktree: + +```bash +python3 scripts/ci/opencode_review_shadow.py run \ + --plan plan.json \ + --worktree /trusted/exact-head-worktree \ + --evidence-dir /trusted/evidence \ + --output-dir /private/empty/output \ + --opencode /trusted/bin/opencode +``` + +Verify the exact-head attempt bundle without publishing: + +```bash +python3 scripts/ci/opencode_review_verify.py \ + --request verification-request.json \ + --output shadow-report.json +``` + +The trusted caller must independently resolve the live base tip. A pull-request event's base snapshot is historical evidence and must not be substituted for the current protected base. + +## Verification contract + +The permanent tests cover routing determinism, strict schemas, adversarial JSON, unsafe paths, immutable digests, budget exhaustion, model diversity, partial failures, timeouts, dependency ordering, credential minimization and redaction, executable/worktree/output boundaries, source receipt authority, deduplication, and publication denial. + +Acceptance for the owned production modules requires: + +- all focused behavioral and adversarial tests passing; +- exactly 100% production statement and branch coverage; +- exactly 100% production callable docstring coverage; +- successful Python compilation and Bash syntax checks; and +- a clean exact-head diff and hosted Python 3.14 workflow result. + +Local evidence does not replace hosted exact-head evidence. A successful pull-request workflow does not establish protected-main operational acceptance, reviewer independence, or commercial review parity. + +## Recovery and rollback + +Malformed input, a changed executable or evidence digest, an unsafe worktree/output boundary, missing credentials, timeout, model failure, stale head, or verification-policy failure stops only the affected plan or attempt and produces no published finding. Operators should preserve the immutable inputs and attempt receipts, correct the first failing boundary, and generate a new content-addressed plan rather than editing evidence in place. + +Rollback is removal of this standalone pool and its caller integration. Because this slice has no GitHub publication or merge authority and no persistent database, rollback does not require data migration. Evaluation artifacts should be retained only under the repository's scoped evidence-retention policy. diff --git a/scripts/ci/opencode_review_shadow.py b/scripts/ci/opencode_review_shadow.py new file mode 100644 index 000000000..998cc1b63 --- /dev/null +++ b/scripts/ci/opencode_review_shadow.py @@ -0,0 +1,348 @@ +"""Plan and execute a bounded, non-publishing OpenCode shadow review pool.""" + +from __future__ import annotations + +import argparse +import os +import runpy +import stat +import subprocess +import sys +from pathlib import Path +from typing import Any, Sequence + +_PRIMITIVES = runpy.run_path( + str(Path(__file__).with_name("opencode_review_shadow_primitives.py")) +) +atomic_write_json = _PRIMITIVES["atomic_write_json"] +digest_bytes = _PRIMITIVES["digest_bytes"] +digest_json = _PRIMITIVES["digest_json"] +require_commit = _PRIMITIVES["require_commit"] +require_fields = _PRIMITIVES["require_fields"] +require_integer = _PRIMITIVES["require_integer"] +require_object = _PRIMITIVES["require_object"] +require_relative_path = _PRIMITIVES["require_relative_path"] +require_sha256 = _PRIMITIVES["require_sha256"] +require_string = _PRIMITIVES["require_string"] +strict_load_json = _PRIMITIVES["strict_load_json"] + +ROOT_FIELDS = { + "schema_version", "review_request_id", "repository", "pull_request_number", + "base_sha", "head_sha", "diff_sha256", "evidence_sha256", "changed_files", "policy", +} +POLICY_FIELDS = { + "shadow_mode", "publication_enabled", "maximum_detector_attempts", + "maximum_recursive_verification_depth", "attempt_timeout_seconds", "model_pool", +} +FILE_FIELDS = {"path", "primary_language", "additions", "deletions", "risk_tags"} +MODEL_FIELDS = { + "descriptor_id", "provider_id", "model_id", "agent_name", "role_codes", + "reasoning_efforts", "prompt_sha256", +} +ROLES = { + "general_detector", "correctness_detector", "security_detector", "workflow_detector", + "data_model_detector", "numerical_detector", "experience_detector", + "documentation_detector", "verifier", "recursive_verifier", +} +EFFORTS = {"low", "medium", "high"} + + +class ShadowValidationError(ValueError): + """Raised when an untrusted routing request violates its strict contract.""" + + +class InsufficientPoolError(ShadowValidationError): + """Raised when policy cannot allocate every required independent role.""" + + +class ShadowExecutionError(RuntimeError): + """Raised before execution when a credential or filesystem boundary is untrusted.""" + + +def validation_error_type() -> type[ShadowValidationError]: + """Return the public validation error used by strict JSON loading.""" + return ShadowValidationError + + +def load_json(path: Path) -> Any: + """Load one strict JSON input file.""" + return strict_load_json(path, ShadowValidationError) + + +def _string_list(value: Any, label: str, *, allowed: set[str] | None = None) -> list[str]: + """Validate one non-empty, unique string-list field.""" + if not isinstance(value, list) or not value: + raise ShadowValidationError(f"{label} must be a non-empty list") + result = [require_string(item, label, ShadowValidationError) for item in value] + if len(set(result)) != len(result): + raise ShadowValidationError(f"{label} contains duplicates") + if allowed is not None and not set(result) <= allowed: + raise ShadowValidationError(f"{label} contains unsupported values") + return result + + +def _validate_request(raw: Any) -> dict[str, Any]: + """Validate every layer of an untrusted shadow-review request.""" + value = require_object(raw, "shadow review request", ShadowValidationError) + require_fields(value, ROOT_FIELDS, "shadow review request", ShadowValidationError) + if value["schema_version"] != "1.0": + raise ShadowValidationError("unsupported schema_version") + require_string(value["review_request_id"], "review_request_id", ShadowValidationError) + require_string(value["repository"], "repository", ShadowValidationError) + require_integer(value["pull_request_number"], "pull_request_number", ShadowValidationError, minimum=1) + require_commit(value["base_sha"], "base_sha", ShadowValidationError) + require_commit(value["head_sha"], "head_sha", ShadowValidationError) + require_sha256(value["diff_sha256"], "diff_sha256", ShadowValidationError) + require_sha256(value["evidence_sha256"], "evidence_sha256", ShadowValidationError) + files = value["changed_files"] + if not isinstance(files, list) or not files: + raise ShadowValidationError("changed_files must be a non-empty list") + for index, raw_file in enumerate(files): + item = require_object(raw_file, f"changed_files[{index}]", ShadowValidationError) + require_fields(item, FILE_FIELDS, f"changed_files[{index}]", ShadowValidationError) + require_relative_path(item["path"], "relative source path", ShadowValidationError) + require_string(item["primary_language"], "primary_language", ShadowValidationError) + require_integer(item["additions"], "additions integer", ShadowValidationError) + require_integer(item["deletions"], "deletions integer", ShadowValidationError) + tags = item["risk_tags"] + if not isinstance(tags, list) or any(not isinstance(tag, str) or not tag for tag in tags): + raise ShadowValidationError("risk_tags must be a string list") + policy = require_object(value["policy"], "policy", ShadowValidationError) + require_fields(policy, POLICY_FIELDS, "policy", ShadowValidationError) + if policy["shadow_mode"] is not True: + raise ShadowValidationError("shadow_mode must be true") + if policy["publication_enabled"] is not False: + raise ShadowValidationError("publication_enabled must be false") + require_integer(policy["maximum_detector_attempts"], "maximum_detector_attempts", ShadowValidationError, minimum=1) + require_integer(policy["maximum_recursive_verification_depth"], "maximum_recursive_verification_depth", ShadowValidationError) + require_integer(policy["attempt_timeout_seconds"], "attempt timeout", ShadowValidationError, minimum=1) + models = policy["model_pool"] + if not isinstance(models, list) or not models: + raise ShadowValidationError("model_pool must be a non-empty list") + descriptor_ids: set[str] = set() + for index, raw_model in enumerate(models): + model = require_object(raw_model, f"model_pool[{index}]", ShadowValidationError) + require_fields(model, MODEL_FIELDS, f"model_pool[{index}]", ShadowValidationError) + descriptor = require_string(model["descriptor_id"], "descriptor_id", ShadowValidationError) + if descriptor in descriptor_ids: + raise ShadowValidationError("descriptor_id must be unique") + descriptor_ids.add(descriptor) + for field in ("provider_id", "model_id", "agent_name"): + require_string(model[field], field, ShadowValidationError) + _string_list(model["role_codes"], "role_codes", allowed=ROLES) + _string_list(model["reasoning_efforts"], "reasoning_efforts", allowed=EFFORTS) + require_sha256(model["prompt_sha256"], "prompt_sha256", ShadowValidationError) + return value + + +def _risk_profile(value: dict[str, Any]) -> tuple[str, str, list[str], list[str], str, int]: + """Derive deterministic risk, size, role, effort, and recursion policy.""" + tags = sorted({tag for item in value["changed_files"] for tag in item["risk_tags"]}) + total = sum(item["additions"] + item["deletions"] for item in value["changed_files"]) + bucket = "small" if total <= 50 else "medium" if total <= 250 else "large" + specialist_roles: list[str] = [] + mapping = ( + ("security", "security_detector"), ("workflow", "workflow_detector"), + ("data_model", "data_model_detector"), ("numerical", "numerical_detector"), + ("experience", "experience_detector"), + ) + for tag, role in mapping: + if tag in tags: + specialist_roles.append(role) + documentation_only = bool(tags) and set(tags) <= {"documentation"} + critical = ({"security", "workflow", "release"} <= set(tags)) or ( + "migration" in tags and bool({"security", "workflow"} & set(tags)) + ) + tier = "low" if documentation_only else "critical" if critical else "high" if specialist_roles else "standard" + effort = "low" if tier == "low" else "medium" if tier == "standard" else "high" + depth = min(value["policy"]["maximum_recursive_verification_depth"], 1) if tier == "critical" else 0 + return tier, bucket, tags, specialist_roles, effort, depth + + +def _choose_model( + models: list[dict[str, Any]], role: str, effort: str, excluded: set[str] +) -> dict[str, Any]: + """Select the first eligible model outside a prohibited identity set.""" + for model in models: + if role in model["role_codes"] and effort in model["reasoning_efforts"] and model["model_id"] not in excluded: + return model + message = "independent verifier" if role in {"verifier", "recursive_verifier"} else role + raise InsufficientPoolError(f"model pool cannot supply {message}") + + +def build_plan(raw: Any) -> dict[str, Any]: + """Validate a request and build a deterministic, content-addressed shadow plan.""" + value = _validate_request(raw) + tier, bucket, reasons, specialists, effort, depth = _risk_profile(value) + detector_roles = ["general_detector", *specialists] + if len(detector_roles) > value["policy"]["maximum_detector_attempts"]: + raise InsufficientPoolError("detector attempt budget is below required roles") + attempts: list[dict[str, Any]] = [] + detector_models: set[str] = set() + for index, role in enumerate(detector_roles, start=1): + model = _choose_model(value["policy"]["model_pool"], role, effort, set()) + detector_models.add(model["model_id"]) + attempts.append(_attempt(model, role, "detector", effort, f"detector_{index:03d}")) + verifier_effort = "medium" if tier == "low" else effort + verifier = _choose_model(value["policy"]["model_pool"], "verifier", verifier_effort, detector_models) + attempts.append(_attempt(verifier, "verifier", "verifier", verifier_effort, "verifier_001")) + if depth: + recursive = _choose_model( + value["policy"]["model_pool"], "recursive_verifier", effort, + detector_models | {verifier["model_id"]}, + ) + attempts.append(_attempt(recursive, "recursive_verifier", "verifier", effort, "verifier_002")) + plan: dict[str, Any] = { + "schema_version": "1.0", "review_request_id": value["review_request_id"], + "repository": value["repository"], "pull_request_number": value["pull_request_number"], + "base_sha": value["base_sha"], "head_sha": value["head_sha"], + "evidence_sha256": value["evidence_sha256"], "input_sha256": digest_json(value), + "risk_tier": tier, "risk_reasons": reasons, "diff_size_bucket": bucket, + "shadow_mode": True, "publication_enabled": False, + "maximum_recursive_verification_depth": depth, + "attempt_timeout_seconds": value["policy"]["attempt_timeout_seconds"], + "attempts": attempts, + } + plan["plan_sha256"] = digest_json(plan) + return plan + + +def _attempt(model: dict[str, Any], role: str, phase: str, effort: str, attempt_id: str) -> dict[str, Any]: + """Build a credential-free normalized attempt descriptor.""" + return { + "attempt_id": attempt_id, "phase": phase, "role_code": role, + "provider_id": model["provider_id"], "model_id": model["model_id"], + "agent_name": model["agent_name"], "reasoning_effort": effort, + "prompt_sha256": model["prompt_sha256"], + } + + +def _validate_execution_boundary(plan: dict[str, Any], evidence: Path, binary: Path, worktree: Path) -> str: + """Validate secret, evidence, executable, and worktree boundaries.""" + secret = os.environ.get("NVIDIA_NIM_API_KEY") + if not secret: + raise ShadowExecutionError("NVIDIA_NIM_API_KEY is required") + if digest_bytes(evidence.read_bytes()) != plan["evidence_sha256"]: + raise ShadowExecutionError("evidence_sha256 does not match evidence") + if binary.is_symlink(): + raise ShadowExecutionError("OpenCode binary must not be a symlink") + mode = binary.stat().st_mode + if not stat.S_ISREG(mode) or not os.access(binary, os.X_OK): + raise ShadowExecutionError("OpenCode binary must be an executable file") + if mode & (stat.S_IWGRP | stat.S_IWOTH): + raise ShadowExecutionError("OpenCode binary must not be group/world writable") + if not worktree.is_dir() or worktree.is_symlink(): + raise ShadowExecutionError("working directory must be a trusted directory") + return secret + + +def _run_attempt( + attempt: dict[str, Any], plan: dict[str, Any], evidence: Path, output: Path, + binary: Path, worktree: Path, detector_files: list[Path], secret: str, +) -> dict[str, Any]: + """Run one fixed-argument OpenCode attempt and normalize its evidence.""" + stdout_path = output / f"{attempt['attempt_id']}.stdout.json" + stderr_path = output / f"{attempt['attempt_id']}.stderr.txt" + command = [ + str(binary), "run", "--agent", attempt["agent_name"], "--model", attempt["model_id"], + "--variant", attempt["reasoning_effort"], "--format", "json", "--dir", str(worktree), + "--file", str(evidence), + ] + for detector_file in detector_files: + command.extend(("--file", str(detector_file))) + command.append(f"role={attempt['role_code']} head={plan['head_sha']} shadow=true") + environment = {"PATH": os.environ.get("PATH", ""), "NVIDIA_API_KEY": secret} + try: + completed = subprocess.run( + command, check=False, capture_output=True, text=True, + timeout=plan["attempt_timeout_seconds"], env=environment, + ) + stdout, stderr = completed.stdout, completed.stderr + status_value = "complete" if completed.returncode == 0 else "failed" + exit_code: int | None = completed.returncode + except subprocess.TimeoutExpired as error: + stdout = error.stdout.decode() if isinstance(error.stdout, bytes) else (error.stdout or "") + stderr = error.stderr.decode() if isinstance(error.stderr, bytes) else (error.stderr or "") + status_value, exit_code = "timed_out", None + stdout = stdout.replace(secret, "[REDACTED_NVIDIA_API_KEY]") + stderr = stderr.replace(secret, "[REDACTED_NVIDIA_API_KEY]") + stdout_path.write_text(stdout, encoding="utf-8") + stderr_path.write_text(stderr, encoding="utf-8") + return { + "attempt_id": attempt["attempt_id"], "phase": attempt["phase"], + "role_code": attempt["role_code"], "provider_id": attempt["provider_id"], + "model_id": attempt["model_id"], "reviewed_head_sha": plan["head_sha"], + "status": status_value, "exit_code": exit_code, + "stdout_file": stdout_path.relative_to(output).as_posix(), + "stderr_file": stderr_path.relative_to(output).as_posix(), + "stdout_sha256": digest_bytes(stdout.encode("utf-8")), + "stderr_sha256": digest_bytes(stderr.encode("utf-8")), + } + + +def _prepare_output_directory(output: Path) -> None: + """Create a private empty output directory or reject unsafe reuse.""" + if output.is_symlink() or (output.exists() and not output.is_dir()): + raise ShadowExecutionError("output directory must be a real directory") + if output.exists(): + if output.stat().st_mode & (stat.S_IWGRP | stat.S_IWOTH): + raise ShadowExecutionError("output directory must not be group/world writable") + if any(output.iterdir()): + raise ShadowExecutionError("output directory must be empty") + else: + output.mkdir(parents=True, mode=0o700) + + +def execute_plan( + plan: dict[str, Any], *, evidence_path: Path, output_directory: Path, + opencode_binary: Path, working_directory: Path, +) -> dict[str, Any]: + """Execute detectors before verifiers with bounded isolation and no publication path.""" + secret = _validate_execution_boundary(plan, evidence_path, opencode_binary, working_directory) + _prepare_output_directory(output_directory) + records: list[dict[str, Any]] = [] + detector_files: list[Path] = [] + for attempt in plan["attempts"]: + if attempt["phase"] == "verifier" and not detector_files: + records.append({ + "attempt_id": attempt["attempt_id"], "phase": attempt["phase"], + "role_code": attempt["role_code"], "provider_id": attempt["provider_id"], + "model_id": attempt["model_id"], "reviewed_head_sha": plan["head_sha"], + "status": "dependency_failed", + }) + continue + record = _run_attempt( + attempt, plan, evidence_path, output_directory, opencode_binary, + working_directory, detector_files if attempt["phase"] == "verifier" else [], secret, + ) + records.append(record) + if attempt["phase"] == "detector" and record["status"] == "complete": + detector_files.append(output_directory / record["stdout_file"]) + manifest: dict[str, Any] = { + "schema_version": "1.0", "shadow_mode": True, "publication_enabled": False, + "plan_sha256": plan["plan_sha256"], "head_sha": plan["head_sha"], "attempts": records, + "completed_attempt_count": sum(item["status"] == "complete" for item in records), + "failed_attempt_count": sum(item["status"] != "complete" for item in records), + } + manifest["execution_sha256"] = digest_json(manifest) + return manifest + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the offline plan CLI and return a stable process status.""" + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + plan_parser = subparsers.add_parser("plan") + plan_parser.add_argument("--input", required=True, type=Path) + plan_parser.add_argument("--output", required=True, type=Path) + arguments = parser.parse_args(argv) + try: + atomic_write_json(arguments.output, build_plan(load_json(arguments.input))) + except (ShadowValidationError, OSError) as error: + print(f"shadow review request rejected: {error}", file=sys.stderr) + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ci/opencode_review_shadow_primitives.py b/scripts/ci/opencode_review_shadow_primitives.py new file mode 100644 index 000000000..3bed2a13d --- /dev/null +++ b/scripts/ci/opencode_review_shadow_primitives.py @@ -0,0 +1,123 @@ +"""Strict deterministic primitives shared by the OpenCode shadow tools.""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +from pathlib import Path, PurePosixPath +from typing import Any, NoReturn + +SHA256_RE = re.compile(r"sha256:[0-9a-f]{64}\Z") +COMMIT_RE = re.compile(r"[0-9a-f]{40}\Z") + + +def canonical_json(value: Any) -> str: + """Serialize a value to the repository's stable JSON representation.""" + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def digest_bytes(value: bytes) -> str: + """Return a labelled SHA-256 digest for bytes.""" + return f"sha256:{hashlib.sha256(value).hexdigest()}" + + +def digest_json(value: Any) -> str: + """Return a labelled SHA-256 digest for canonical JSON.""" + return digest_bytes(canonical_json(value).encode("utf-8")) + + +def _duplicate_key(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + """Build an object while rejecting ambiguous duplicate JSON keys.""" + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate JSON key: {key}") + result[key] = value + return result + + +def _nonfinite(value: str) -> NoReturn: + """Reject a non-standard non-finite JSON numeric literal.""" + raise ValueError(f"non-finite JSON number: {value}") + + +def strict_load_json(path: Path, error_type: type[Exception]) -> Any: + """Load UTF-8 JSON while rejecting duplicate keys and non-finite numbers.""" + try: + return json.loads( + path.read_text(encoding="utf-8"), + object_pairs_hook=_duplicate_key, + parse_constant=_nonfinite, + ) + except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as error: + raise error_type(str(error)) from error + + +def atomic_write_json(path: Path, value: Any) -> None: + """Atomically replace a UTF-8 JSON output without leaving a stable temp file.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text(canonical_json(value) + "\n", encoding="utf-8") + os.replace(temporary, path) + + +def require_object(value: Any, label: str, error_type: type[Exception]) -> dict[str, Any]: + """Return a JSON object or raise the caller's validation error.""" + if not isinstance(value, dict): + raise error_type(f"{label} must be an object") + return value + + +def require_fields( + value: dict[str, Any], allowed: set[str], label: str, error_type: type[Exception] +) -> None: + """Require an exact, non-extensible object field set.""" + unknown = set(value) - allowed + missing = allowed - set(value) + if unknown: + raise error_type(f"{label} has unknown fields: {sorted(unknown)}") + if missing: + raise error_type(f"{label} is missing fields: {sorted(missing)}") + + +def require_integer(value: Any, label: str, error_type: type[Exception], *, minimum: int = 0) -> int: + """Require a real integer at or above a lower bound; booleans are rejected.""" + if isinstance(value, bool) or not isinstance(value, int): + raise error_type(f"{label} must be an integer") + if value < minimum: + raise error_type(f"{label} must be at least {minimum}") + return value + + +def require_string(value: Any, label: str, error_type: type[Exception]) -> str: + """Require a non-empty string.""" + if not isinstance(value, str) or not value: + raise error_type(f"{label} must be a non-empty string") + return value + + +def require_sha256(value: Any, label: str, error_type: type[Exception]) -> str: + """Require a lowercase labelled SHA-256 digest.""" + text = require_string(value, label, error_type) + if not SHA256_RE.fullmatch(text): + raise error_type(f"{label} must be a sha256 digest") + return text + + +def require_commit(value: Any, label: str, error_type: type[Exception]) -> str: + """Require a full lowercase hexadecimal commit SHA.""" + text = require_string(value, label, error_type) + if not COMMIT_RE.fullmatch(text): + raise error_type(f"{label} must be a full commit SHA") + return text + + +def require_relative_path(value: Any, label: str, error_type: type[Exception]) -> str: + """Require a normalized relative POSIX source path.""" + text = require_string(value, label, error_type) + parsed = PurePosixPath(text) + if parsed.is_absolute() or ".." in parsed.parts or text in {".", ""} or "\\" in text: + raise error_type(f"{label} must be a relative source path") + return text diff --git a/scripts/ci/opencode_review_verify.py b/scripts/ci/opencode_review_verify.py new file mode 100644 index 000000000..8ca3ace82 --- /dev/null +++ b/scripts/ci/opencode_review_verify.py @@ -0,0 +1,279 @@ +"""Validate and normalize non-publishing OpenCode shadow-review evidence.""" + +from __future__ import annotations + +import argparse +import re +import runpy +import sys +from pathlib import Path +from typing import Any, Sequence + +_PRIMITIVES = runpy.run_path( + str(Path(__file__).with_name("opencode_review_shadow_primitives.py")) +) +atomic_write_json = _PRIMITIVES["atomic_write_json"] +digest_json = _PRIMITIVES["digest_json"] +require_commit = _PRIMITIVES["require_commit"] +require_fields = _PRIMITIVES["require_fields"] +require_integer = _PRIMITIVES["require_integer"] +require_object = _PRIMITIVES["require_object"] +require_relative_path = _PRIMITIVES["require_relative_path"] +require_sha256 = _PRIMITIVES["require_sha256"] +require_string = _PRIMITIVES["require_string"] +strict_load_json = _PRIMITIVES["strict_load_json"] + +ROOT_FIELDS = { + "schema_version", "verification_id", "repository", "pull_request_number", "base_sha", + "head_sha", "evidence_sha256", "risk_tier", "verification_policy", "source_index", + "detector_attempts", "verifier_attempts", "candidates", "verifier_decisions", +} +POLICY_FIELDS = {"shadow_mode", "publication_enabled", "minimum_independent_verifiers", "require_model_diversity"} +SOURCE_FIELDS = {"path", "line", "source_line_sha256", "relationship"} +ATTEMPT_FIELDS = {"attempt_id", "phase", "role_code", "provider_id", "model_id", "reviewed_head_sha", "status", "output_sha256"} +CANDIDATE_FIELDS = { + "candidate_id", "detector_attempt_id", "reviewed_head_sha", "infrastructure_only", + "path", "line", "source_line_sha256", "defect_class", "severity", "blocking", + "trigger", "impact", "root_cause", "fix_direction", "regression_target", +} +DECISION_FIELDS = {"candidate_id", "verifier_attempt_id", "outcome", "reason", "source_line_sha256"} + + +class VerificationValidationError(ValueError): + """Raised when a verification bundle violates its strict evidence contract.""" + + +def validation_error_type() -> type[VerificationValidationError]: + """Return the public validation error used by strict JSON loading.""" + return VerificationValidationError + + +def load_json(path: Path) -> Any: + """Load one strict verification JSON file.""" + return strict_load_json(path, VerificationValidationError) + + +def _list(value: Any, label: str) -> list[Any]: + """Validate a JSON array without silently coercing other iterables.""" + if not isinstance(value, list): + raise VerificationValidationError(f"{label} must be a list") + return value + + +def _validate_bundle(raw: Any) -> dict[str, Any]: + """Validate all exact-head identity and evidence references in a bundle.""" + value = require_object(raw, "verification bundle", VerificationValidationError) + require_fields(value, ROOT_FIELDS, "verification bundle", VerificationValidationError) + if value["schema_version"] != "1.0": + raise VerificationValidationError("unsupported schema_version") + require_string(value["verification_id"], "verification_id", VerificationValidationError) + require_string(value["repository"], "repository", VerificationValidationError) + require_integer(value["pull_request_number"], "pull_request_number", VerificationValidationError, minimum=1) + require_commit(value["base_sha"], "base_sha", VerificationValidationError) + head = require_commit(value["head_sha"], "head_sha commit SHA", VerificationValidationError) + require_sha256(value["evidence_sha256"], "evidence_sha256", VerificationValidationError) + if value["risk_tier"] not in {"low", "standard", "high", "critical"}: + raise VerificationValidationError("risk_tier is unsupported") + policy = require_object(value["verification_policy"], "verification_policy", VerificationValidationError) + require_fields(policy, POLICY_FIELDS, "verification_policy", VerificationValidationError) + if policy["shadow_mode"] is not True: + raise VerificationValidationError("shadow_mode must be true") + if policy["publication_enabled"] is not False: + raise VerificationValidationError("publication_enabled must be false") + require_integer(policy["minimum_independent_verifiers"], "minimum_independent_verifiers integer", VerificationValidationError, minimum=1) + if not isinstance(policy["require_model_diversity"], bool): + raise VerificationValidationError("require_model_diversity must be boolean") + + source_identities: set[tuple[str, int]] = set() + for index, raw_source in enumerate(_list(value["source_index"], "source_index")): + source = require_object(raw_source, f"source_index[{index}]", VerificationValidationError) + require_fields(source, SOURCE_FIELDS, f"source_index[{index}]", VerificationValidationError) + path = require_relative_path(source["path"], "source path", VerificationValidationError) + line = require_integer(source["line"], "source line integer", VerificationValidationError, minimum=1) + require_sha256(source["source_line_sha256"], "source_line_sha256", VerificationValidationError) + if source["relationship"] not in {"changed", "connected"}: + raise VerificationValidationError("source relationship is unsupported") + identity = (path, line) + if identity in source_identities: + raise VerificationValidationError("source identity must be unique") + source_identities.add(identity) + + attempt_ids: set[str] = set() + attempts: dict[str, dict[str, Any]] = {} + for collection, expected_phase in (("detector_attempts", "detector"), ("verifier_attempts", "verifier")): + for index, raw_attempt in enumerate(_list(value[collection], collection)): + attempt = require_object(raw_attempt, f"{collection}[{index}]", VerificationValidationError) + require_fields(attempt, ATTEMPT_FIELDS, f"{collection}[{index}]", VerificationValidationError) + attempt_id = require_string(attempt["attempt_id"], "attempt_id", VerificationValidationError) + if attempt_id in attempt_ids: + raise VerificationValidationError("attempt_id must be unique") + attempt_ids.add(attempt_id) + if attempt["phase"] != expected_phase: + raise VerificationValidationError("attempt phase does not match collection") + for field in ("role_code", "provider_id", "model_id"): + require_string(attempt[field], field, VerificationValidationError) + if attempt["reviewed_head_sha"] != head: + raise VerificationValidationError("reviewed_head_sha must match head_sha") + if attempt["status"] not in {"complete", "failed", "timed_out", "dependency_failed"}: + raise VerificationValidationError("attempt status is unsupported") + require_sha256(attempt["output_sha256"], "output_sha256", VerificationValidationError) + attempts[attempt_id] = attempt + + candidate_ids: set[str] = set() + for index, raw_candidate in enumerate(_list(value["candidates"], "candidates")): + candidate_value = require_object(raw_candidate, f"candidates[{index}]", VerificationValidationError) + require_fields(candidate_value, CANDIDATE_FIELDS, f"candidates[{index}]", VerificationValidationError) + candidate_id = require_string(candidate_value["candidate_id"], "candidate_id", VerificationValidationError) + if candidate_id in candidate_ids: + raise VerificationValidationError("candidate_id must be unique") + candidate_ids.add(candidate_id) + detector_id = require_string(candidate_value["detector_attempt_id"], "detector_attempt_id", VerificationValidationError) + if detector_id not in attempts or attempts[detector_id]["phase"] != "detector": + raise VerificationValidationError("unknown detector attempt") + if candidate_value["reviewed_head_sha"] != head: + raise VerificationValidationError("candidate reviewed_head_sha must match head_sha") + if not isinstance(candidate_value["infrastructure_only"], bool) or not isinstance(candidate_value["blocking"], bool): + raise VerificationValidationError("candidate booleans are invalid") + require_relative_path(candidate_value["path"], "candidate path", VerificationValidationError) + require_integer(candidate_value["line"], "candidate line integer", VerificationValidationError, minimum=1) + require_sha256(candidate_value["source_line_sha256"], "candidate source_line_sha256", VerificationValidationError) + for field in ("defect_class", "severity", "trigger", "impact", "root_cause", "fix_direction", "regression_target"): + require_string(candidate_value[field], field, VerificationValidationError) + + decision_ids: set[tuple[str, str]] = set() + for index, raw_decision in enumerate(_list(value["verifier_decisions"], "verifier_decisions")): + decision = require_object(raw_decision, f"verifier_decisions[{index}]", VerificationValidationError) + require_fields(decision, DECISION_FIELDS, f"verifier_decisions[{index}]", VerificationValidationError) + candidate_id = require_string(decision["candidate_id"], "candidate_id", VerificationValidationError) + verifier_id = require_string(decision["verifier_attempt_id"], "verifier_attempt_id", VerificationValidationError) + if candidate_id not in candidate_ids: + raise VerificationValidationError("verifier decision references unknown candidate") + if verifier_id not in attempts or attempts[verifier_id]["phase"] != "verifier": + raise VerificationValidationError("verifier decision references unknown verifier") + identity = (candidate_id, verifier_id) + if identity in decision_ids: + raise VerificationValidationError("verifier decision identity must be unique") + decision_ids.add(identity) + if decision["outcome"] not in {"supported", "rejected"}: + raise VerificationValidationError("verifier decision outcome is unsupported") + require_string(decision["reason"], "verifier reason", VerificationValidationError) + require_sha256(decision["source_line_sha256"], "decision source_line_sha256", VerificationValidationError) + return value + + +def _reject(candidate: dict[str, Any], reason: str) -> dict[str, Any]: + """Return a non-source-bearing rejection receipt.""" + return {"candidate_id": candidate["candidate_id"], "reason_code": reason} + + +def _normal_root(value: str) -> str: + """Normalize semantic whitespace and case for deterministic deduplication.""" + return re.sub(r"\s+", " ", value).strip().casefold() + + +def verify_bundle(raw: Any) -> dict[str, Any]: + """Verify exact-head source authority and return deterministic shadow-only findings.""" + value = _validate_bundle(raw) + sources = {(item["path"], item["line"]): item for item in value["source_index"]} + detectors = {item["attempt_id"]: item for item in value["detector_attempts"]} + verifiers = {item["attempt_id"]: item for item in value["verifier_attempts"]} + decisions: dict[str, list[dict[str, Any]]] = {} + for decision in value["verifier_decisions"]: + decisions.setdefault(decision["candidate_id"], []).append(decision) + metrics = { + "candidate_count": len(value["candidates"]), "accepted_finding_count": 0, + "rejected_candidate_count": 0, "duplicate_candidate_count": 0, + "infrastructure_only_candidate_count": 0, "unsupported_candidate_count": 0, + "source_contract_failure_count": 0, "insufficient_verifier_count": 0, + } + rejected: list[dict[str, Any]] = [] + accepted: list[dict[str, Any]] = [] + for candidate_value in sorted(value["candidates"], key=lambda item: item["candidate_id"]): + if candidate_value["infrastructure_only"]: + reason = "infrastructure_only" + metrics["infrastructure_only_candidate_count"] += 1 + elif detectors[candidate_value["detector_attempt_id"]]["status"] != "complete": + reason = "detector_not_complete" + else: + source = sources.get((candidate_value["path"], candidate_value["line"])) + if source is None or source["source_line_sha256"] != candidate_value["source_line_sha256"]: + reason = "source_receipt_mismatch" + metrics["source_contract_failure_count"] += 1 + else: + candidate_decisions = decisions.get(candidate_value["candidate_id"], []) + supported = [ + item for item in candidate_decisions + if item["outcome"] == "supported" + and item["source_line_sha256"] == candidate_value["source_line_sha256"] + and verifiers[item["verifier_attempt_id"]]["status"] == "complete" + ] + if candidate_decisions and not any(item["outcome"] == "supported" for item in candidate_decisions): + reason = "unsupported" + metrics["unsupported_candidate_count"] += 1 + else: + verifier_models = {verifiers[item["verifier_attempt_id"]]["model_id"] for item in supported} + detector_model = detectors[candidate_value["detector_attempt_id"]]["model_id"] + required = value["verification_policy"]["minimum_independent_verifiers"] + diverse = not value["verification_policy"]["require_model_diversity"] or detector_model not in verifier_models + if len(verifier_models) < required or not diverse: + reason = "insufficient_verifier_evidence" + metrics["insufficient_verifier_count"] += 1 + else: + finding = { + key: candidate_value[key] for key in ( + "path", "line", "source_line_sha256", "defect_class", "severity", + "blocking", "trigger", "impact", "root_cause", "fix_direction", "regression_target", + ) + } + finding["detector_attempt_ids"] = [candidate_value["detector_attempt_id"]] + finding["verifier_attempt_ids"] = sorted(item["verifier_attempt_id"] for item in supported) + finding["finding_fingerprint"] = digest_json({ + "path": finding["path"], "line": finding["line"], + "root_cause": _normal_root(finding["root_cause"]), + }) + accepted.append(finding) + continue + rejected.append(_reject(candidate_value, reason)) + grouped: dict[tuple[str, int, str], dict[str, Any]] = {} + for finding in accepted: + identity = (finding["path"], finding["line"], _normal_root(finding["root_cause"])) + if identity in grouped: + existing = grouped[identity] + existing["detector_attempt_ids"] = sorted(set(existing["detector_attempt_ids"] + finding["detector_attempt_ids"])) + existing["verifier_attempt_ids"] = sorted(set(existing["verifier_attempt_ids"] + finding["verifier_attempt_ids"])) + metrics["duplicate_candidate_count"] += 1 + else: + grouped[identity] = finding + findings = sorted(grouped.values(), key=lambda item: (item["path"], item["line"], item["finding_fingerprint"])) + metrics["accepted_finding_count"] = len(findings) + metrics["rejected_candidate_count"] = len(rejected) + report: dict[str, Any] = { + "schema_version": "1.0", "verification_id": value["verification_id"], + "repository": value["repository"], "pull_request_number": value["pull_request_number"], + "base_sha": value["base_sha"], "head_sha": value["head_sha"], + "evidence_sha256": value["evidence_sha256"], "risk_tier": value["risk_tier"], + "shadow_mode": True, "publication_enabled": False, + "shadow_findings": findings, "published_findings": [], + "rejected_candidates": sorted(rejected, key=lambda item: item["candidate_id"]), + "metrics": metrics, + } + report["verification_sha256"] = digest_json(report) + return report + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the offline verifier CLI and return a stable process status.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + arguments = parser.parse_args(argv) + try: + atomic_write_json(arguments.output, verify_bundle(load_json(arguments.input))) + except (VerificationValidationError, OSError) as error: + print(f"shadow verification rejected: {error}", file=sys.stderr) + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ci/run_opencode_semantic_review_pool.sh b/scripts/ci/run_opencode_semantic_review_pool.sh new file mode 100755 index 000000000..3beaee9eb --- /dev/null +++ b/scripts/ci/run_opencode_semantic_review_pool.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +exec python3 "${SCRIPT_DIR}/opencode_review_shadow.py" "$@" diff --git a/tests/test_opencode_review_shadow_execution.py b/tests/test_opencode_review_shadow_execution.py index 7fbcffb10..0a4413607 100644 --- a/tests/test_opencode_review_shadow_execution.py +++ b/tests/test_opencode_review_shadow_execution.py @@ -24,7 +24,13 @@ ) -def fake_opencode(path: Path, *, fail_role: str = "", sleep_role: str = "") -> Path: +def fake_opencode( + path: Path, + *, + fail_role: str = "", + sleep_role: str = "", + leak_role: str = "", +) -> Path: """Create a deterministic fake OpenCode CLI that validates credential mapping.""" path.write_text( "#!/usr/bin/env python3\n" @@ -37,7 +43,11 @@ def fake_opencode(path: Path, *, fail_role: str = "", sleep_role: str = "") -> P "if role == " + repr(fail_role) + ":\n" " print('bounded fake failure', file=sys.stderr)\n" " raise SystemExit(7)\n" - "print(json.dumps({'argv': args, 'role': role, 'secret_exposed': 'nim-secret' in json.dumps(args)}))\n", + "event = {'argv': args, 'role': role, 'secret_exposed': 'nim-secret' in json.dumps(args)}\n" + "if role == " + repr(leak_role) + ":\n" + " event['untrusted_echo'] = os.environ['NVIDIA_API_KEY']\n" + " print(os.environ['NVIDIA_API_KEY'], file=sys.stderr)\n" + "print(json.dumps(event))\n", encoding="utf-8", ) path.chmod(0o700) @@ -61,6 +71,7 @@ def test_execute_plan_invokes_detectors_before_verifiers_without_publication( executable = fake_opencode(tmp_path / "opencode") monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") output = tmp_path / "output" + output.mkdir() manifest = shadow.execute_plan( plan, evidence_path=evidence, @@ -186,6 +197,33 @@ def test_timeout_is_bounded_and_recorded_without_exception_escape( ] +def test_child_secret_echo_is_redacted_from_all_persisted_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An untrusted child cannot persist its mapped provider secret in evidence.""" + plan, evidence, workdir = run_inputs(tmp_path) + executable = fake_opencode( + tmp_path / "opencode", leak_role="general_detector" + ) + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + output = tmp_path / "output" + manifest = shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=output, + opencode_binary=executable, + working_directory=workdir, + ) + persisted = "\n".join( + (output / record[field]).read_text(encoding="utf-8") + for record in manifest["attempts"] + if record["status"] == "complete" + for field in ("stdout_file", "stderr_file") + ) + assert "nim-secret" not in persisted + assert "[REDACTED_NVIDIA_API_KEY]" in persisted + + def test_execution_fails_before_process_start_on_untrusted_boundary( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -236,6 +274,59 @@ def test_execution_fails_before_process_start_on_untrusted_boundary( working_directory=workdir, ) + non_executable = tmp_path / "not-executable" + non_executable.write_text("not executable", encoding="utf-8") + with pytest.raises(shadow.ShadowExecutionError, match="executable file"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=non_executable, + working_directory=workdir, + ) + + invalid_worktree = tmp_path / "not-a-worktree" + invalid_worktree.write_text("not a directory", encoding="utf-8") + with pytest.raises(shadow.ShadowExecutionError, match="trusted directory"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=tmp_path / "output", + opencode_binary=executable, + working_directory=invalid_worktree, + ) + + +@pytest.mark.parametrize("boundary", ["symlink", "file", "writable", "nonempty"]) +def test_execution_rejects_untrusted_output_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, boundary: str +) -> None: + """Output evidence cannot follow links or overwrite a reusable untrusted path.""" + plan, evidence, workdir = run_inputs(tmp_path) + executable = fake_opencode(tmp_path / "opencode") + output = tmp_path / "output" + if boundary == "symlink": + target = tmp_path / "target" + target.mkdir() + output.symlink_to(target, target_is_directory=True) + elif boundary == "file": + output.write_text("not a directory", encoding="utf-8") + else: + output.mkdir() + if boundary == "writable": + output.chmod(0o770) + else: + (output / "existing.txt").write_text("existing", encoding="utf-8") + monkeypatch.setenv("NVIDIA_NIM_API_KEY", "nim-secret") + with pytest.raises(shadow.ShadowExecutionError, match="output directory"): + shadow.execute_plan( + plan, + evidence_path=evidence, + output_directory=output, + opencode_binary=executable, + working_directory=workdir, + ) + def test_shell_wrapper_is_thin_non_publishing_and_functional(tmp_path: Path) -> None: """The permanent wrapper delegates to Python and has no GitHub mutation path.""" diff --git a/tests/test_opencode_review_shadow_validation.py b/tests/test_opencode_review_shadow_validation.py index b66f1490f..f6973425a 100644 --- a/tests/test_opencode_review_shadow_validation.py +++ b/tests/test_opencode_review_shadow_validation.py @@ -104,6 +104,40 @@ def test_routing_rejects_empty_files_duplicate_models_and_invalid_roles() -> Non shadow.build_plan(invalid_role) +@pytest.mark.parametrize( + ("mutate", "message"), + [ + (lambda value: value.pop("repository"), "missing fields"), + (lambda value: value.update({"repository": ""}), "non-empty string"), + ( + lambda value: value["policy"]["model_pool"][0].update( + {"role_codes": []} + ), + "non-empty list", + ), + ( + lambda value: value["policy"]["model_pool"][0].update( + {"role_codes": ["general_detector", "general_detector"]} + ), + "duplicates", + ), + ( + lambda value: value["changed_files"][0].update({"risk_tags": [""]}), + "risk_tags", + ), + (lambda value: value["policy"].update({"model_pool": []}), "model_pool"), + ], +) +def test_routing_rejects_empty_duplicate_or_incomplete_contract_fields( + mutate: Any, message: str +) -> None: + """Strict routing validation covers missing and structurally empty evidence.""" + value = request() + mutate(value) + with pytest.raises(shadow.ShadowValidationError, match=message): + shadow.build_plan(value) + + @pytest.mark.parametrize( ("mutate", "message"), [ @@ -211,6 +245,64 @@ def test_verification_rejects_duplicate_or_unknown_identity_references() -> None verify.verify_bundle(unknown_attempt) +@pytest.mark.parametrize( + ("mutate", "message"), + [ + (lambda value: value.update({"source_index": {}}), "must be a list"), + (lambda value: value.update({"schema_version": "2.0"}), "schema_version"), + (lambda value: value.update({"risk_tier": "unknown"}), "risk_tier"), + ( + lambda value: value["verification_policy"].update( + {"require_model_diversity": 1} + ), + "require_model_diversity", + ), + ( + lambda value: value["source_index"][0].update( + {"relationship": "untrusted"} + ), + "relationship", + ), + ( + lambda value: value["detector_attempts"][0].update( + {"phase": "verifier"} + ), + "phase", + ), + ( + lambda value: value["detector_attempts"][0].update( + {"status": "queued"} + ), + "status", + ), + ( + lambda value: value["candidates"][0].update({"blocking": 1}), + "booleans", + ), + ( + lambda value: value["verifier_decisions"][0].update( + {"verifier_attempt_id": "unknown_verifier"} + ), + "unknown verifier", + ), + ( + lambda value: value["verifier_decisions"].append( + dict(value["verifier_decisions"][0]) + ), + "decision identity", + ), + ], +) +def test_verification_rejects_additional_closed_contract_failures( + mutate: Any, message: str +) -> None: + """Strict verification validation covers every closed-schema authority boundary.""" + value = verification_input() + mutate(value) + with pytest.raises(verify.VerificationValidationError, match=message): + verify.verify_bundle(value) + + def test_strict_json_loaders_reject_duplicate_keys_and_nonfinite_numbers( tmp_path: Path, ) -> None: