From 4fc5d75fe01258bbc469373eafdcc2d8afb55cfa Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 8 Aug 2026 23:09:30 +0900 Subject: [PATCH] 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", + ]