diff --git a/.github/workflows/opencode-review-decision-quality-ci.yml b/.github/workflows/opencode-review-decision-quality-ci.yml new file mode 100644 index 000000000..1e6d2ccc3 --- /dev/null +++ b/.github/workflows/opencode-review-decision-quality-ci.yml @@ -0,0 +1,86 @@ +name: OpenCode Review Decision Quality CI + +on: + pull_request: + branches: + - main + - feat/opencode-review-gold-corpus + paths: + - ".github/workflows/opencode-review-decision-quality-ci.yml" + - "scripts/ci/opencode_review_decision.py" + - "scripts/ci/opencode_review_decision_primitives.py" + - "scripts/ci/opencode_review_decision_validation.py" + - "tests/opencode_review_decision_test_support.py" + - "tests/test_opencode_review_decision_*.py" + - "docs/doctoring/opencode-review-decision-envelope.md" + - "CHANGELOG.md" + +permissions: + contents: read + +concurrency: + group: opencode-review-decision-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + decision-envelope-quality: + name: decision-envelope-quality + if: github.event_name != 'pull_request' || github.event.action != 'closed' + runs-on: ubuntu-24.04 + timeout-minutes: 10 + 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-decision-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-decision-requirements.txt" + + - name: Verify independent decision channels + 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_decision_channels.py \ + tests/test_opencode_review_decision_validation.py \ + tests/test_opencode_review_decision_cli.py \ + -q + python -m coverage report \ + --include='scripts/ci/opencode_review_decision.py,scripts/ci/opencode_review_decision_primitives.py,scripts/ci/opencode_review_decision_validation.py' \ + --fail-under=100 \ + --show-missing + python -m compileall -q \ + scripts/ci/opencode_review_decision.py \ + scripts/ci/opencode_review_decision_primitives.py \ + scripts/ci/opencode_review_decision_validation.py \ + tests/opencode_review_decision_test_support.py \ + tests/test_opencode_review_decision_channels.py \ + tests/test_opencode_review_decision_validation.py \ + tests/test_opencode_review_decision_cli.py + git diff --exit-code 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/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 000000000..64e2bddeb --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,87 @@ +# Architecture — ContextualWisdomLab `.github` + +This repository is the organization control plane. It is not naruon and it +does not own product data. Sibling products remain standalone modules; this +repo publishes org profile assets, reusable required workflows, and the +review/merge schedulers those products consume. + +## System context + +```mermaid +flowchart LR + Buyer["Commercial buyer / reviewer"] + Agents["Agents on AGENTS.md"] + Project["GitHub Project #1"] + Hub["This repo: org .github"] + Products["Owned products
naruon · orchestrator · engines"] + Runner["Required workflows in each repo context"] + + Buyer --> Hub + Agents --> Project + Agents --> Hub + Project --> Hub + Hub --> Runner + Runner --> Products + Products -->|"standalone or as module"| Buyer +``` + +## Shadow review pool + +```mermaid +flowchart TD + Meta["Trusted PR metadata"] + Plan["Deterministic risk-adaptive plan"] + Detect["Bounded detector attempts"] + Verify["Independent verifier"] + Receipt["Immutable shadow findings"] + Block["No comment, review, check, approval, or merge"] + + Meta --> Plan + Plan --> Detect + Detect --> Verify + Verify --> Receipt + Receipt --> Block +``` + +CWE-345: shadow findings are evaluation evidence, not authenticity +evidence for merge. CWE-841: semantic source judgment and merge +readiness stay independent. Reviewers stay `edit: deny`. Bind +`NVIDIA_NIM_API_KEY`, never `COPILOT_GITHUB_TOKEN`. + +## Control-plane data flow + +```mermaid +sequenceDiagram + participant PR as Pull request + participant RW as Required workflows + participant OC as OpenCode reviewer + participant SV as sandboxed_verify / web E2E + participant MS as Merge scheduler + + PR->>RW: pull_request_target on trusted base + RW->>OC: bounded evidence + NVIDIA NIM / OpenCode + OC->>SV: PoC command in isolated copy + SV-->>OC: redacted stdout/stderr + command metadata + OC-->>PR: APPROVE or request changes + MS->>PR: merge only on current-head approval + green checks +``` + +## Trust boundaries + +- Required review workflows execute **base-branch** scripts. +- Reviewer agents stay `edit: deny`. +- Logs redact credential shapes. They do not mask operational PII. +- LLM and scheduled agents bind `NVIDIA_NIM_API_KEY`. They never use + `COPILOT_GITHUB_TOKEN`. +- Rust remains the psychometric arithmetic owner. + +## Quality gates + +`scripts/ci/` ships with 100% statement/branch coverage and 100% +docstrings. + +## Related durable documents + +- [`docs/CWL-MASTER-CONTEXT.md`](docs/CWL-MASTER-CONTEXT.md) +- [`PR_GOVERNANCE_AUDIT.md`](PR_GOVERNANCE_AUDIT.md) +- [`docs/doctoring/opencode-review-shadow-orchestration.md`](docs/doctoring/opencode-review-shadow-orchestration.md) diff --git a/CHANGELOG.md b/CHANGELOG.md index 675ecb8d7..42439c263 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ 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. The decision record now cites CWE-345 so shadow findings cannot be treated as authenticity evidence for merge. +- Recorded the org control-plane architecture, including the shadow review pool, so agents reconstruct the evaluation-versus-merge trust boundary from the repo instead of private memory. +- 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. The decision record now cites CWE-841 so a coverage or policy failure cannot be converted into a source defect. - 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. - Added a trusted pull-request comment router for `@cwl-noema-review` and review-only `@opencode-agent` dispatches, with an organization sweep, exact-head receipts, repository allowlisting, fixed runners, immutable checkout pins, and a permanent 100% statement/branch/docstring quality gate. diff --git a/CLAUDE.md b/CLAUDE.md index 1c7bdb2f6..eeddd2c77 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -64,7 +64,9 @@ Details: `README.md` and `PR_GOVERNANCE_AUDIT.md`. - `fuzz/` + `.clusterfuzzlite/` — Atheris fuzz targets for the review-output normalizer and the ClusterFuzzLite discovery marker. - `docs/` — master context, Project protocol, `org-required-workflow-rollout.md`, - `scorecard-governance.md`, SBOM inventory. + `scorecard-governance.md`, SBOM inventory. Doctoring records live under + `docs/doctoring/`. [`ARCHITECTURE.md`](ARCHITECTURE.md) is the control-plane + diagram for the shadow review pool and merge trust boundaries. - `.jules/` — recorded performance (`bolt.md`) and security (`sentinel.md`) learnings from past work on `scripts/ci/`; worth scanning before optimizing or hardening those scripts. diff --git a/docs/doctoring/opencode-review-decision-envelope.md b/docs/doctoring/opencode-review-decision-envelope.md new file mode 100644 index 000000000..b3768c88e --- /dev/null +++ b/docs/doctoring/opencode-review-decision-envelope.md @@ -0,0 +1,298 @@ +# OpenCode review semantic and merge-readiness decision envelope + +Status: Proposed operational architecture +Date: 2026-08-08 +Owner: ContextualWisdomLab central review infrastructure + +## Decision summary + +OpenCode Review must represent two independent decisions: + +1. **Semantic review verdict** — whether exact-head source and connected context contain a substantiated defect. +2. **Merge readiness** — whether exact-head checks, coverage, independent approval, branch protection, and repository policy permit integration. + +Infrastructure or policy failure may block integration. It must never be converted into a source finding, severity, path, or line number. A failed coverage collector is evidence that coverage has not been proven, not evidence that the pull-request implementation contains a high-severity defect. + +CWE-841 forbids collapsing two required behaviors into one action +(MITRE, 2026). Semantic source judgment and merge-readiness checks are +therefore independent outputs. + +This change introduces an offline, deterministic decision-composition module. It does not yet modify the production OpenCode dispatch because other active branches own that large workflow. Production integration requires a later test-first slice after the writer lease clears. + +## Problem + +The current central review workflow can construct a synthetic `REQUEST_CHANGES` finding when coverage evidence acquisition fails. That behavior combines two different questions: + +```text +Does the changed source contain a defect? + ≠ +May this exact head merge under repository policy? +``` + +The conflation creates several operational failure modes: + +- repeated infrastructure-only reviews that provide no semantic source value; +- fabricated source anchors such as a workflow line that did not cause the product defect; +- inability to distinguish reviewer abstention from a negative code judgment; +- false defect counts in quality evaluation; +- developer confusion about whether to repair code or review infrastructure; and +- stale-head evidence accidentally appearing authoritative after a new commit. + +The empirical lifecycle pilot in `benchmarks/opencode_review/pilot_baseline_v1.json` directly observed eight completed OpenCode attempts that produced only infrastructure review output and no source findings. That pilot is not a head-matched precision or recall study, but it establishes that the decision-channel conflation is operationally material. + +## Versioned input contract + +The decision module accepts one strict JSON object: + +```json +{ + "schema_version": "1.0", + "decision_id": "decision_001", + "quality_policy_version": "opencode-review-quality-v1", + "repository": "ContextualWisdomLab/example", + "pull_request_number": 42, + "base_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "semantic_review": { + "status": "complete", + "reviewed_head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "findings": [] + }, + "merge_evidence": { + "evidence_head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "coverage_state": "success", + "independent_approval_state": "success", + "branch_protection_state": "success", + "required_checks": [] + } +} +``` + +### Semantic review input + +`semantic_review.status` is one of: + +- `complete` — semantic review finished on the exact current head; +- `unavailable` — no trusted semantic review result was available; or +- `failed` — the semantic review process failed before producing a valid result. + +A complete review must bind `reviewed_head_sha` to the decision's exact head. An unavailable or failed review must use a null reviewed head and must contain no findings. This prevents a partial or failed process from publishing synthetic source claims. + +Each semantic finding must contain: + +- stable finding identifier; +- defect class and calibrated severity; +- explicit blocking classification; +- repository-relative changed or connected source path; +- positive source line; +- trigger condition; +- observable impact; +- source-backed root cause; +- minimal fix direction; and +- exact regression target. + +### Merge-evidence input + +The merge channel carries only policy evidence: + +- coverage state; +- independent-approval state; +- branch-protection state; and +- named required or advisory checks. + +Every record must bind to the same exact head. A trusted composition root determines which checks are required according to repository policy; the pure decision module does not infer branch protection or fetch GitHub state itself. + +The accepted evidence states are: + +```text +success +failure +pending +queued +absent +cancelled +skipped +neutral +``` + +This vocabulary deliberately distinguishes hard negative evidence, latency, and absence. It does not reinterpret GitHub conclusions as product defects. + +## Versioned output contract + +The output preserves both decisions and both evidence classes: + +```json +{ + "schema_version": "1.0", + "review_verdict": "APPROVE", + "merge_readiness": "BLOCKED", + "semantic_status": "complete", + "findings": [], + "infrastructure_blockers": [ + { + "blocker_code": "coverage_not_successful", + "evidence_name": "coverage", + "state": "failure", + "check_name": null + } + ], + "evidence_manifest": {}, + "decision_sha256": "sha256:..." +} +``` + +Infrastructure blockers have no `path`, `line`, `severity`, `trigger`, `root_cause`, or fix authority. They identify the evidence surface and state only. + +## Semantic verdict rules + +| Semantic status and findings | `review_verdict` | +|---|---| +| Complete, no finding | `APPROVE` | +| Complete, only non-blocking findings | `COMMENT` | +| Complete, one or more blocking findings | `REQUEST_CHANGES` | +| Unavailable or failed | `ABSTAIN` | + +The semantic verdict never reads coverage, checks, approval, or branch-protection state. + +`APPROVE` in this envelope means that the semantic channel found no blocking source defect. It is not a formal GitHub approval and must not be submitted as a review by the change author or any non-independent identity. + +## Merge-readiness rules + +| Evidence | `merge_readiness` | +|---|---| +| Semantic review complete without blocking findings; all required policy evidence successful | `READY` | +| Blocking semantic finding | `BLOCKED` | +| Required evidence `failure`, `cancelled`, `skipped`, or `neutral` | `BLOCKED` | +| Required evidence `pending`, `queued`, or `absent` | `UNKNOWN` | +| Semantic review unavailable or failed, with no hard policy failure | `UNKNOWN` | + +Advisory-check failure is recorded in the evidence manifest but does not block unless repository policy marks that check required. + +A hard policy failure takes precedence over latency. For example, one failed required check and one pending check produce `BLOCKED`, not `UNKNOWN`. + +## Exact-head and evidence-integrity rules + +The module rejects: + +- a semantic review for another head; +- coverage, approval, protection, or check evidence for another head; +- duplicate case-insensitive finding identifiers; +- duplicate case-insensitive check names; +- unsafe or parent-traversing source paths; +- zero or negative line numbers; +- Boolean values passed as integers; +- unknown fields at every governed schema layer; +- duplicate JSON member names; +- Python's non-standard `NaN`, `Infinity`, and `-Infinity` JSON extensions; and +- incomplete semantic reviews that claim findings or an exact reviewed head. + +Canonical strict JSON produces an input SHA-256 receipt. The normalized decision, excluding its own digest, produces a separate decision SHA-256 receipt. JSON and Markdown outputs use temporary sibling files followed by atomic replacement. + +## Markdown presentation + +The human-readable output has physically separate sections: + +```text +Semantic findings + +Infrastructure and policy blockers +``` + +Only semantic findings may render `path:line`. Infrastructure blockers display the evidence name, state, stable blocker code, and optional check name. This presentation rule prevents a later renderer from reintroducing a synthetic source defect even when the underlying JSON remains separated. + +## Security and privacy boundary + +The decision module: + +- runs offline; +- does not execute repository code, model output, commands, or patches; +- does not call GitHub or another network service; +- does not read a secret, cookie, token, environment credential, or model credential; +- accepts only policy-normalized evidence from a trusted caller; +- records no source body or personal data beyond bounded finding text and repository identity; and +- introduces no `COPILOT_GITHUB_TOKEN` use. + +Scheduled OpenCode model execution, when used elsewhere, continues to use the existing `NVIDIA_NIM_API_KEY` credential boundary. This pure module neither selects nor invokes a model. + +## Production integration boundary + +This pull request does not edit `.github/workflows/opencode-review-dispatch.yml`. At the time of implementation, active central branches `#789`, `#812`, `#816`, and `#827` modify that workflow. A competing edit would violate the one-writer lease and could discard exact-head coverage and toolchain repairs. + +After those branches integrate or relinquish the file, a separate implementation must begin with a failing production contract that requires the dispatch to: + +1. continue bounded semantic review whenever safe exact-head source evidence exists; +2. emit `ABSTAIN` rather than a source finding when semantic review cannot complete; +3. place coverage and check failures only in merge-readiness evidence; +4. pass the versioned decision envelope to publication; +5. publish source comments only from validated semantic findings; +6. expose infrastructure blockers through check summary or a non-source status surface; +7. preserve reviewer identity and credential chains; and +8. keep merge, auto-merge, branch update, and release authority separately controlled. + +No predecessor-head result from this pure module authorizes that later production integration. + +## Verification + +The exact-head quality workflow runs on Python 3.14 with immutable action pins, read-only repository permission, persisted checkout credentials disabled, and hash-verified test dependencies. It requires: + +- behavior and adversarial schema tests; +- exact-head checkout verification; +- production statement coverage 100%; +- production branch coverage 100%; +- public production callable docstrings 100%; +- `compileall`; and +- a clean Git worktree. + +Regression cases cover the exact operational failure: coverage failure with a complete defect-free semantic review yields `APPROVE` plus `BLOCKED`, with no source finding. + +## Monitoring after integration + +When the envelope reaches the production dispatch, monitor at least: + +- semantic completion, failure, and abstention rates; +- infrastructure-only blocker rate; +- source findings per completed semantic review; +- current-head duplicate publication rate; +- stale-head evidence rejection count; +- blocked versus unknown readiness rates; +- time to first useful semantic comment; +- developer dismissal and resolution rates; and +- any occurrence of a path or line in an infrastructure blocker. + +The last metric must remain zero. + +## Rollback + +The pure decision module can be removed from a caller without changing reviewer credentials, model selection, or GitHub branch protection. Rollback must not restore the old synthetic source-finding path. If the envelope cannot be consumed safely, the caller should fail closed with: + +- semantic verdict `ABSTAIN`; and +- merge readiness `UNKNOWN` or `BLOCKED` according to independently available policy evidence. + +## Limitations + +- The module does not discover which GitHub checks are required; a trusted policy collector must supply that classification. +- It does not prove that a formal independent approval is valid; it validates only normalized approval state and exact-head identity supplied by a trusted caller. +- It does not itself collect coverage or branch-protection evidence. +- It does not calibrate semantic severity or verify model findings; detector-verifier orchestration and expert-gold evaluation remain separate work. +- A `READY` output is a deterministic policy composition result, not authority to bypass GitHub rulesets or merge administratively. + +## References + +MITRE. (2026). *CWE-841: Improper enforcement of behavioral workflow*. +https://cwe.mitre.org/data/definitions/841.html + +Booth, H., Souppaya, M., Vassilev, A., Ogata, M., Stanley, M., Scarfone, K., & Dodson, D. (2024). *Secure software development practices for generative AI and dual-use foundation models: An SSDF community profile* (NIST Special Publication 800-218A). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-218A + +GitHub. (n.d.). *About protected branches*. GitHub Docs. Retrieved August 8, 2026, from https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/managing-protected-branches/about-protected-branches + +GitHub. (n.d.). *Approving a pull request with required reviews*. GitHub Docs. Retrieved August 8, 2026, from https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/approving-a-pull-request-with-required-reviews + +GitHub. (n.d.). *Status checks*. GitHub Docs. Retrieved August 8, 2026, from https://docs.github.com/en/pull-requests/committing-changes-to-your-project/troubleshooting-commits/status-checks + +GitHub. (n.d.). *Troubleshooting required status checks*. GitHub Docs. Retrieved August 8, 2026, from https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/managing-protected-branches/troubleshooting-required-status-checks + +SLSA Community. (2025). *SLSA specification, version 1.2*. https://slsa.dev/spec/v1.2/ + +Souppaya, M., Scarfone, K., & Dodson, D. (2022). *Secure Software Development Framework (SSDF) version 1.1: Recommendations for mitigating the risk of software vulnerabilities* (NIST Special Publication 800-218). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-218 + +Sun, T., Xu, J., Li, Y., Yan, Z., Zhang, G., Xie, L., Geng, L., Wang, Z., Chen, Y., Lin, Q., Duan, W., & Sui, K. (2025). BitsAI-CR: Automated code review via LLM in practice. In *Proceedings of the 33rd ACM International Conference on the Foundations of Software Engineering*. https://doi.org/10.1145/3696630.3728552 diff --git a/docs/doctoring/opencode-review-shadow-orchestration.md b/docs/doctoring/opencode-review-shadow-orchestration.md new file mode 100644 index 000000000..979034dcc --- /dev/null +++ b/docs/doctoring/opencode-review-shadow-orchestration.md @@ -0,0 +1,121 @@ +# 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. CWE-345 forbids treating unverified or evaluation-only output as authenticity evidence (MITRE, 2026). Shadow findings therefore cannot authorize merge. 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. + +## References + +MITRE. (2026). *CWE-345: Insufficient verification of data authenticity*. +https://cwe.mitre.org/data/definitions/345.html diff --git a/scripts/ci/opencode_review_decision.py b/scripts/ci/opencode_review_decision.py new file mode 100644 index 000000000..2d0763b9d --- /dev/null +++ b/scripts/ci/opencode_review_decision.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +"""Build independent semantic-review and merge-readiness decisions.""" + +from __future__ import annotations + +import argparse +import json +import sys +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any + +MODULE_DIR = Path(__file__).resolve().parent +if str(MODULE_DIR) not in sys.path: + sys.path.insert(0, str(MODULE_DIR)) + +from opencode_review_decision_primitives import ( # noqa: E402 + HARD_BLOCKING_STATES, + UNKNOWN_STATES, + DecisionValidationError, + array_value, + bool_value, + commit_sha_value, + content_digest, + load_json, + reject_constant, + strict_pairs, + text_value, + write_text, +) +from opencode_review_decision_validation import validate_decision_input # noqa: E402 + + +def blocker( + blocker_code: str, evidence_name: str, state: str, check_name: str | None = None +) -> dict[str, Any]: + """Build one path-free infrastructure or policy blocker.""" + return { + "blocker_code": blocker_code, + "evidence_name": evidence_name, + "state": state, + "check_name": check_name, + } + + +def classify_review_verdict(semantic_review: Mapping[str, Any]) -> str: + """Choose a semantic verdict without consulting merge-readiness evidence.""" + if semantic_review["status"] != "complete": + return "ABSTAIN" + findings = semantic_review["findings"] + if any(item["blocking"] for item in findings): + return "REQUEST_CHANGES" + return "COMMENT" if findings else "APPROVE" + + +def collect_blockers(merge_evidence: Mapping[str, Any]) -> list[dict[str, Any]]: + """Collect non-successful policy evidence without source-location authority.""" + blockers: list[dict[str, Any]] = [] + policy_surfaces = ( + ( + "coverage_state", + "coverage_not_successful", + "coverage", + ), + ( + "independent_approval_state", + "independent_approval_not_successful", + "independent_approval", + ), + ( + "branch_protection_state", + "branch_protection_not_successful", + "branch_protection", + ), + ) + for field, code, evidence_name in policy_surfaces: + state = merge_evidence[field] + if state != "success": + blockers.append(blocker(code, evidence_name, state)) + for check in merge_evidence["required_checks"]: + if check["required"] and check["state"] != "success": + blockers.append( + blocker( + "required_check_not_successful", + "required_check", + check["state"], + check["name"], + ) + ) + return blockers + + +def classify_merge_readiness( + review_verdict: str, blockers: Sequence[Mapping[str, Any]] +) -> str: + """Classify merge readiness using fail-closed policy evidence and latency states.""" + if review_verdict == "REQUEST_CHANGES": + return "BLOCKED" + blocker_states = {item["state"] for item in blockers} + if blocker_states & HARD_BLOCKING_STATES: + return "BLOCKED" + if review_verdict == "ABSTAIN" or blocker_states & UNKNOWN_STATES: + return "UNKNOWN" + return "READY" + + +def build_decision(raw_value: Any) -> dict[str, Any]: + """Build one deterministic exact-head decision with independent channels.""" + value = validate_decision_input(raw_value) + semantic_review = value["semantic_review"] + merge_evidence = value["merge_evidence"] + review_verdict = classify_review_verdict(semantic_review) + blockers = collect_blockers(merge_evidence) + required_checks = [item for item in merge_evidence["required_checks"] if item["required"]] + advisory_checks = [item for item in merge_evidence["required_checks"] if not item["required"]] + report_without_digest = { + "schema_version": "1.0", + "decision_id": value["decision_id"], + "quality_policy_version": value["quality_policy_version"], + "repository": value["repository"], + "pull_request_number": value["pull_request_number"], + "base_sha": value["base_sha"], + "head_sha": value["head_sha"], + "semantic_status": semantic_review["status"], + "review_verdict": review_verdict, + "merge_readiness": classify_merge_readiness(review_verdict, blockers), + "findings": semantic_review["findings"], + "infrastructure_blockers": blockers, + "evidence_manifest": { + "input_sha256": content_digest(value), + "semantic_reviewed_head_sha": semantic_review["reviewed_head_sha"], + "merge_evidence_head_sha": merge_evidence["evidence_head_sha"], + "coverage_state": merge_evidence["coverage_state"], + "independent_approval_state": merge_evidence[ + "independent_approval_state" + ], + "branch_protection_state": merge_evidence["branch_protection_state"], + "required_check_count": len(required_checks), + "successful_required_check_count": sum( + item["state"] == "success" for item in required_checks + ), + "advisory_check_count": len(advisory_checks), + "checks": merge_evidence["required_checks"], + }, + } + return { + **report_without_digest, + "decision_sha256": content_digest(report_without_digest), + } + + +def render_markdown(report: Mapping[str, Any]) -> str: + """Render a human-readable decision without turning blockers into source defects.""" + lines = [ + "# OpenCode review decision", + "", + f"Review verdict: **{report['review_verdict']}** ", + f"Merge readiness: **{report['merge_readiness']}** ", + f"Semantic status: **{report['semantic_status']}** ", + f"Exact head: `{report['head_sha']}`", + "", + "## Semantic findings", + "", + ] + if report["findings"]: + for finding in report["findings"]: + lines.extend( + [ + f"- **{finding['severity'].upper()}** `{finding['path']}:{finding['line']}` — {finding['trigger']}", + f" - Impact: {finding['impact']}", + f" - Root cause: {finding['root_cause']}", + f" - Fix direction: {finding['fix_direction']}", + f" - Regression target: {finding['regression_target']}", + ] + ) + else: + lines.append("- None.") + lines.extend(["", "## Infrastructure and policy blockers", ""]) + if report["infrastructure_blockers"]: + for item in report["infrastructure_blockers"]: + suffix = f" / check `{item['check_name']}`" if item["check_name"] else "" + lines.append( + f"- `{item['evidence_name']}` — `{item['state']}` ({item['blocker_code']}){suffix}" + ) + else: + lines.append("- None.") + lines.extend( + [ + "", + "## Evidence receipt", + "", + f"- Input: `{report['evidence_manifest']['input_sha256']}`", + f"- Decision: `{report['decision_sha256']}`", + "", + ] + ) + return "\n".join(lines) + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the decision CLI and return a stable validation status.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", type=Path, required=True) + parser.add_argument("--json-output", type=Path, required=True) + parser.add_argument("--markdown-output", type=Path, required=True) + arguments = parser.parse_args(argv) + try: + report = build_decision(load_json(arguments.input)) + except DecisionValidationError as error: + print(f"decision evidence rejected: {error}", file=sys.stderr) + return 2 + write_text( + arguments.json_output, + json.dumps( + report, + ensure_ascii=False, + indent=2, + sort_keys=True, + allow_nan=False, + ) + + "\n", + ) + write_text(arguments.markdown_output, render_markdown(report)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ci/opencode_review_decision_primitives.py b/scripts/ci/opencode_review_decision_primitives.py new file mode 100644 index 000000000..170d82f00 --- /dev/null +++ b/scripts/ci/opencode_review_decision_primitives.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Strict primitives for independent OpenCode review decisions.""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping +from pathlib import Path, PurePosixPath +from typing import Any + +REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +COMMIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +VALID_SEMANTIC_STATUSES = {"complete", "unavailable", "failed"} +VALID_SEVERITIES = {"critical", "high", "medium", "low"} +VALID_EVIDENCE_STATES = { + "success", + "failure", + "pending", + "queued", + "absent", + "cancelled", + "skipped", + "neutral", +} +HARD_BLOCKING_STATES = {"failure", "cancelled", "skipped", "neutral"} +UNKNOWN_STATES = {"pending", "queued", "absent"} + + +class DecisionValidationError(ValueError): + """Signal malformed or internally inconsistent decision evidence.""" + + +def reject(message: str) -> None: + """Raise one stable decision validation error.""" + raise DecisionValidationError(message) + + +def object_value(value: Any, path: str) -> Mapping[str, Any]: + """Return a JSON object or reject its shape.""" + if not isinstance(value, Mapping): + reject(f"{path} must be an object") + return value + + +def array_value(value: Any, path: str) -> list[Any]: + """Return a JSON array or reject its shape.""" + if not isinstance(value, list): + reject(f"{path} must be an array") + return value + + +def require_exact_fields( + value: Mapping[str, Any], path: str, allowed_fields: set[str] +) -> None: + """Reject unreviewed extension fields at one governed schema layer.""" + unknown = sorted(set(value) - allowed_fields) + if unknown: + reject(f"{path} has unknown fields: {', '.join(unknown)}") + + +def text_value(value: Any, path: str) -> str: + """Return stripped non-empty text or reject it.""" + if not isinstance(value, str) or not value.strip(): + reject(f"{path} must be non-empty text") + return value.strip() + + +def bool_value(value: Any, path: str) -> bool: + """Return an actual Boolean rather than an integer lookalike.""" + if not isinstance(value, bool): + reject(f"{path} must be boolean") + return value + + +def positive_int_value(value: Any, path: str) -> int: + """Return a strictly positive integer without Boolean coercion.""" + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + reject(f"{path} must be a positive integer") + return value + + +def commit_sha_value(value: Any, path: str) -> str: + """Return one full lowercase commit SHA.""" + result = text_value(value, path) + if not COMMIT_SHA_RE.fullmatch(result): + reject(f"{path} must be a 40-character lowercase commit SHA") + return result + + +def optional_commit_sha_value(value: Any, path: str) -> str | None: + """Return ``None`` or one full lowercase commit SHA.""" + return None if value is None else commit_sha_value(value, path) + + +def enum_value(value: Any, path: str, allowed: set[str]) -> str: + """Return a normalized enumerated value or reject it.""" + result = text_value(value, path).casefold() + if result not in allowed: + reject(f"{path} is invalid: {result!r}") + return result + + +def source_path_value(value: Any, path: str) -> str: + """Return a safe repository-relative POSIX source path.""" + result = text_value(value, path) + pure = PurePosixPath(result) + if ( + pure.is_absolute() + or "\\" in result + or any(part in {"", ".", ".."} for part in pure.parts) + ): + reject(f"{path} must be a safe relative source path") + return pure.as_posix() + + +def canonical_json(value: Any) -> str: + """Serialize JSON deterministically for content-addressed receipts.""" + return json.dumps( + value, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + allow_nan=False, + ) + + +def content_digest(value: Any) -> str: + """Return the canonical SHA-256 digest for a JSON-compatible value.""" + encoded = canonical_json(value).encode("utf-8") + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + +def strict_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + """Build one JSON object while rejecting duplicate member names.""" + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + reject(f"duplicate JSON key: {key}") + result[key] = value + return result + + +def reject_constant(value: str) -> None: + """Reject non-finite constants accepted by Python's permissive JSON parser.""" + reject(f"non-finite JSON number: {value}") + + +def load_json(path: Path) -> Any: + """Load strict UTF-8 JSON with bounded stable validation errors.""" + try: + return json.loads( + path.read_text(encoding="utf-8"), + object_pairs_hook=strict_pairs, + parse_constant=reject_constant, + ) + except (OSError, json.JSONDecodeError) as error: + reject(f"cannot load decision evidence: {error}") + + +def write_text(path: Path, content: str) -> None: + """Atomically replace one UTF-8 output after creating its parent directory.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text(content, encoding="utf-8") + temporary.replace(path) diff --git a/scripts/ci/opencode_review_decision_validation.py b/scripts/ci/opencode_review_decision_validation.py new file mode 100644 index 000000000..30c8c44ed --- /dev/null +++ b/scripts/ci/opencode_review_decision_validation.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +"""Validate exact-head semantic and merge-policy evidence.""" + +from __future__ import annotations + +from typing import Any + +from opencode_review_decision_primitives import ( + REPOSITORY_RE, + VALID_EVIDENCE_STATES, + VALID_SEMANTIC_STATUSES, + VALID_SEVERITIES, + array_value, + bool_value, + commit_sha_value, + enum_value, + object_value, + optional_commit_sha_value, + positive_int_value, + reject, + require_exact_fields, + source_path_value, + text_value, +) + + +def validate_finding(raw_value: Any, path: str) -> dict[str, Any]: + """Validate one complete semantic source finding.""" + value = object_value(raw_value, path) + require_exact_fields( + value, + path, + { + "finding_id", + "defect_class", + "severity", + "blocking", + "path", + "line", + "trigger", + "impact", + "root_cause", + "fix_direction", + "regression_target", + }, + ) + return { + "finding_id": text_value(value.get("finding_id"), f"{path}.finding_id"), + "defect_class": text_value( + value.get("defect_class"), f"{path}.defect_class" + ).casefold(), + "severity": enum_value( + value.get("severity"), f"{path}.severity", VALID_SEVERITIES + ), + "blocking": bool_value(value.get("blocking"), f"{path}.blocking"), + "path": source_path_value(value.get("path"), f"{path}.path"), + "line": positive_int_value(value.get("line"), f"{path}.line"), + "trigger": text_value(value.get("trigger"), f"{path}.trigger"), + "impact": text_value(value.get("impact"), f"{path}.impact"), + "root_cause": text_value(value.get("root_cause"), f"{path}.root_cause"), + "fix_direction": text_value( + value.get("fix_direction"), f"{path}.fix_direction" + ), + "regression_target": text_value( + value.get("regression_target"), f"{path}.regression_target" + ), + } + + +def validate_semantic_review( + raw_value: Any, expected_head_sha: str +) -> dict[str, Any]: + """Validate semantic review evidence independently from merge policy evidence.""" + value = object_value(raw_value, "semantic_review") + require_exact_fields( + value, + "semantic_review", + {"status", "reviewed_head_sha", "findings"}, + ) + status = enum_value( + value.get("status"), "semantic_review.status", VALID_SEMANTIC_STATUSES + ) + reviewed_head_sha = optional_commit_sha_value( + value.get("reviewed_head_sha"), "semantic_review.reviewed_head_sha" + ) + raw_findings = array_value(value.get("findings"), "semantic_review.findings") + if status != "complete": + if reviewed_head_sha is not None: + reject( + "semantic_review.reviewed_head_sha must be null when semantic review is incomplete" + ) + if raw_findings: + reject("incomplete semantic review must not contain findings") + return { + "status": status, + "reviewed_head_sha": None, + "findings": [], + } + if reviewed_head_sha != expected_head_sha: + reject( + "semantic_review.reviewed_head_sha must equal the exact decision head_sha" + ) + findings: list[dict[str, Any]] = [] + seen_ids: set[str] = set() + for index, raw_finding in enumerate(raw_findings): + finding = validate_finding( + raw_finding, f"semantic_review.findings[{index}]" + ) + identity = finding["finding_id"].casefold() + if identity in seen_ids: + reject( + f"semantic_review.findings[{index}].finding_id duplicates {finding['finding_id']!r}" + ) + seen_ids.add(identity) + findings.append(finding) + findings.sort(key=lambda item: item["finding_id"].casefold()) + return { + "status": status, + "reviewed_head_sha": reviewed_head_sha, + "findings": findings, + } + + +def validate_check(raw_value: Any, path: str, expected_head_sha: str) -> dict[str, Any]: + """Validate one exact-head required or advisory check record.""" + value = object_value(raw_value, path) + require_exact_fields(value, path, {"name", "state", "required", "head_sha"}) + head_sha = commit_sha_value(value.get("head_sha"), f"{path}.head_sha") + if head_sha != expected_head_sha: + reject(f"{path}.head_sha must equal the exact decision head_sha") + return { + "name": text_value(value.get("name"), f"{path}.name"), + "state": enum_value( + value.get("state"), f"{path}.state", VALID_EVIDENCE_STATES + ), + "required": bool_value(value.get("required"), f"{path}.required"), + "head_sha": head_sha, + } + + +def validate_merge_evidence( + raw_value: Any, expected_head_sha: str +) -> dict[str, Any]: + """Validate exact-head coverage, approval, protection, and check evidence.""" + value = object_value(raw_value, "merge_evidence") + require_exact_fields( + value, + "merge_evidence", + { + "evidence_head_sha", + "coverage_state", + "independent_approval_state", + "branch_protection_state", + "required_checks", + }, + ) + evidence_head_sha = commit_sha_value( + value.get("evidence_head_sha"), "merge_evidence.evidence_head_sha" + ) + if evidence_head_sha != expected_head_sha: + reject("merge_evidence.evidence_head_sha must equal the exact decision head_sha") + checks: list[dict[str, Any]] = [] + seen_names: set[str] = set() + for index, raw_check in enumerate( + array_value(value.get("required_checks"), "merge_evidence.required_checks") + ): + path = f"merge_evidence.required_checks[{index}]" + check = validate_check(raw_check, path, expected_head_sha) + normalized_name = check["name"].casefold() + if normalized_name in seen_names: + reject(f"{path}.name duplicates check name {check['name']!r}") + seen_names.add(normalized_name) + checks.append(check) + checks.sort(key=lambda item: item["name"].casefold()) + return { + "evidence_head_sha": evidence_head_sha, + "coverage_state": enum_value( + value.get("coverage_state"), + "merge_evidence.coverage_state", + VALID_EVIDENCE_STATES, + ), + "independent_approval_state": enum_value( + value.get("independent_approval_state"), + "merge_evidence.independent_approval_state", + VALID_EVIDENCE_STATES, + ), + "branch_protection_state": enum_value( + value.get("branch_protection_state"), + "merge_evidence.branch_protection_state", + VALID_EVIDENCE_STATES, + ), + "required_checks": checks, + } + + +def validate_decision_input(raw_value: Any) -> dict[str, Any]: + """Validate and normalize one complete exact-head decision input.""" + value = object_value(raw_value, "decision") + require_exact_fields( + value, + "decision", + { + "schema_version", + "decision_id", + "quality_policy_version", + "repository", + "pull_request_number", + "base_sha", + "head_sha", + "semantic_review", + "merge_evidence", + }, + ) + if value.get("schema_version") != "1.0": + reject("decision.schema_version must equal '1.0'") + repository = text_value(value.get("repository"), "decision.repository") + if not REPOSITORY_RE.fullmatch(repository): + reject("decision.repository must use owner/name") + head_sha = commit_sha_value(value.get("head_sha"), "decision.head_sha") + normalized = { + "schema_version": "1.0", + "decision_id": text_value(value.get("decision_id"), "decision.decision_id"), + "quality_policy_version": text_value( + value.get("quality_policy_version"), "decision.quality_policy_version" + ), + "repository": repository, + "pull_request_number": positive_int_value( + value.get("pull_request_number"), "decision.pull_request_number" + ), + "base_sha": commit_sha_value(value.get("base_sha"), "decision.base_sha"), + "head_sha": head_sha, + "semantic_review": validate_semantic_review( + value.get("semantic_review"), head_sha + ), + "merge_evidence": validate_merge_evidence( + value.get("merge_evidence"), head_sha + ), + } + return normalized 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/opencode_review_decision_test_support.py b/tests/opencode_review_decision_test_support.py new file mode 100644 index 000000000..e805c253c --- /dev/null +++ b/tests/opencode_review_decision_test_support.py @@ -0,0 +1,95 @@ +"""Shared fixtures for OpenCode decision-envelope tests.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import ModuleType +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts/ci/opencode_review_decision.py" + + +def load_module() -> ModuleType: + """Load the exact decision module without package import side effects.""" + spec = importlib.util.spec_from_file_location("opencode_review_decision", MODULE_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 + + +decision = load_module() + + +def finding( + identifier: str = "finding_001", + *, + severity: str = "high", + blocking: bool = True, +) -> dict[str, Any]: + """Build one complete semantic source finding.""" + return { + "finding_id": identifier, + "defect_class": "correctness", + "severity": severity, + "blocking": blocking, + "path": "scripts/ci/example.py", + "line": 12, + "trigger": "The input contains a duplicate exact-head identity.", + "impact": "The benchmark counts one pull request twice.", + "root_cause": "The identity set is not checked before aggregation.", + "fix_direction": "Reject duplicate repository, PR, and head tuples.", + "regression_target": "Add a duplicate exact-head fixture.", + } + + +def check( + name: str = "CI", + *, + state: str = "success", + required: bool = True, + head_sha: str | None = None, +) -> dict[str, Any]: + """Build one exact-head check evidence record.""" + return { + "name": name, + "state": state, + "required": required, + "head_sha": head_sha or "b" * 40, + } + + +def envelope( + *, + semantic_status: str = "complete", + findings: list[dict[str, Any]] | None = None, + coverage_state: str = "success", + approval_state: str = "success", + protection_state: str = "success", + checks: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Build one decision input with all evidence bound to one immutable head.""" + complete = semantic_status == "complete" + return { + "schema_version": "1.0", + "decision_id": "decision_001", + "quality_policy_version": "opencode-review-quality-v1", + "repository": "ContextualWisdomLab/example", + "pull_request_number": 42, + "base_sha": "a" * 40, + "head_sha": "b" * 40, + "semantic_review": { + "status": semantic_status, + "reviewed_head_sha": "b" * 40 if complete else None, + "findings": findings if findings is not None else [], + }, + "merge_evidence": { + "evidence_head_sha": "b" * 40, + "coverage_state": coverage_state, + "independent_approval_state": approval_state, + "branch_protection_state": protection_state, + "required_checks": checks if checks is not None else [check()], + }, + } 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_materialize_base_python_requirements.py b/tests/test_materialize_base_python_requirements.py index 8a383f0c2..10f682b3e 100644 --- a/tests/test_materialize_base_python_requirements.py +++ b/tests/test_materialize_base_python_requirements.py @@ -30,6 +30,13 @@ def _created_tool_directory(path: Path) -> str: return str(path) +def _force_linux_x86_64_installer(monkeypatch: pytest.MonkeyPatch) -> None: + """Exercise the installer path that GitHub-hosted linux x86_64 runners use.""" + monkeypatch.setattr(materializer.sys, "platform", "linux") + monkeypatch.setattr(materializer.platform, "machine", lambda: "x86_64") + materializer._install_trusted_uv.cache_clear() + + def test_materializes_only_regular_hash_locks_from_exact_base(tmp_path: Path) -> None: """A PR-modified lock cannot enter the networked coverage image build context.""" repo = tmp_path / "repo" @@ -644,6 +651,7 @@ def test_install_trusted_uv_verifies_version_and_caches_path( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The installer writes one executable, verifies its version, and caches it.""" + _force_linux_x86_64_installer(monkeypatch) tool_dir = tmp_path / "uv" monkeypatch.setattr( materializer.tempfile, @@ -690,6 +698,7 @@ def test_install_trusted_uv_rejects_version_process_failures( failure: OSError | subprocess.TimeoutExpired, ) -> None: """A missing or hung downloaded executable is removed and rejected.""" + _force_linux_x86_64_installer(monkeypatch) tool_dir = tmp_path / "uv" monkeypatch.setattr( materializer.tempfile, @@ -721,6 +730,7 @@ def test_install_trusted_uv_rejects_wrong_version_or_exit_status( completed: subprocess.CompletedProcess[bytes], ) -> None: """Unexpected version output or a nonzero status cannot satisfy the pin.""" + _force_linux_x86_64_installer(monkeypatch) tool_dir = tmp_path / f"uv-{completed.returncode}-{len(completed.stdout)}" monkeypatch.setattr( materializer.tempfile, diff --git a/tests/test_opencode_review_decision_channels.py b/tests/test_opencode_review_decision_channels.py new file mode 100644 index 000000000..acf1632dd --- /dev/null +++ b/tests/test_opencode_review_decision_channels.py @@ -0,0 +1,149 @@ +"""Behavior tests for independent semantic and merge-readiness channels.""" + +from __future__ import annotations + +import copy +import sys +from pathlib import Path + +TEST_DIR = Path(__file__).resolve().parent +if str(TEST_DIR) not in sys.path: + sys.path.insert(0, str(TEST_DIR)) + +from opencode_review_decision_test_support import check, decision, envelope, finding + + +def test_coverage_failure_cannot_create_a_source_finding() -> None: + """Coverage failure must block readiness without becoming a line-level defect.""" + report = decision.build_decision(envelope(coverage_state="failure")) + assert report["review_verdict"] == "APPROVE" + assert report["merge_readiness"] == "BLOCKED" + assert report["findings"] == [] + assert report["semantic_status"] == "complete" + assert report["infrastructure_blockers"] == [ + { + "blocker_code": "coverage_not_successful", + "evidence_name": "coverage", + "state": "failure", + "check_name": None, + } + ] + assert all( + "path" not in blocker and "line" not in blocker + for blocker in report["infrastructure_blockers"] + ) + + +def test_semantic_finding_survives_independent_coverage_failure() -> None: + """A real source defect and infrastructure blocker remain separate channels.""" + report = decision.build_decision( + envelope(findings=[finding()], coverage_state="failure") + ) + assert report["review_verdict"] == "REQUEST_CHANGES" + assert report["merge_readiness"] == "BLOCKED" + assert [item["finding_id"] for item in report["findings"]] == ["finding_001"] + assert report["infrastructure_blockers"][0]["evidence_name"] == "coverage" + + +def test_semantic_verdict_matrix_is_independent_of_merge_evidence() -> None: + """Complete semantic review alone chooses approve, comment, or request changes.""" + assert decision.build_decision(envelope())["review_verdict"] == "APPROVE" + assert ( + decision.build_decision(envelope(findings=[finding(blocking=False)]))[ + "review_verdict" + ] + == "COMMENT" + ) + assert ( + decision.build_decision(envelope(findings=[finding(blocking=True)]))[ + "review_verdict" + ] + == "REQUEST_CHANGES" + ) + for status in ("unavailable", "failed"): + report = decision.build_decision(envelope(semantic_status=status)) + assert report["review_verdict"] == "ABSTAIN" + assert report["findings"] == [] + + +def test_merge_readiness_ready_blocked_and_unknown_states() -> None: + """Readiness distinguishes hard failure from latency or absent evidence.""" + assert decision.build_decision(envelope())["merge_readiness"] == "READY" + assert ( + decision.build_decision(envelope(findings=[finding()]))["merge_readiness"] + == "BLOCKED" + ) + assert ( + decision.build_decision(envelope(checks=[check(state="cancelled")]))[ + "merge_readiness" + ] + == "BLOCKED" + ) + for state in ("pending", "queued", "absent"): + assert ( + decision.build_decision(envelope(checks=[check(state=state)]))[ + "merge_readiness" + ] + == "UNKNOWN" + ) + assert ( + decision.build_decision(envelope(semantic_status="unavailable"))[ + "merge_readiness" + ] + == "UNKNOWN" + ) + + +def test_required_and_advisory_checks_are_classified_separately() -> None: + """Only required checks block readiness, while advisory evidence is recorded.""" + report = decision.build_decision( + envelope( + checks=[ + check("required-ci", state="success", required=True), + check("advisory-lint", state="failure", required=False), + ] + ) + ) + assert report["merge_readiness"] == "READY" + assert report["infrastructure_blockers"] == [] + manifest = report["evidence_manifest"] + assert manifest["required_check_count"] == 1 + assert manifest["successful_required_check_count"] == 1 + assert manifest["advisory_check_count"] == 1 + + +def test_every_non_successful_policy_surface_produces_non_source_blockers() -> None: + """Coverage, approval, protection, and required checks report stable blockers.""" + report = decision.build_decision( + envelope( + coverage_state="neutral", + approval_state="absent", + protection_state="pending", + checks=[check("unit", state="failure"), check("security", state="skipped")], + ) + ) + assert report["merge_readiness"] == "BLOCKED" + assert { + (item["evidence_name"], item["state"], item["check_name"]) + for item in report["infrastructure_blockers"] + } == { + ("coverage", "neutral", None), + ("independent_approval", "absent", None), + ("branch_protection", "pending", None), + ("required_check", "failure", "unit"), + ("required_check", "skipped", "security"), + } + assert all( + "path" not in item and "line" not in item + for item in report["infrastructure_blockers"] + ) + + +def test_output_is_deterministic_and_receipt_bound() -> None: + """Equivalent exact-head input produces one stable content-addressed decision.""" + value = envelope(findings=[finding(blocking=False)]) + first = decision.build_decision(copy.deepcopy(value)) + second = decision.build_decision(copy.deepcopy(value)) + assert first == second + assert first["decision_sha256"].startswith("sha256:") + assert first["evidence_manifest"]["input_sha256"].startswith("sha256:") diff --git a/tests/test_opencode_review_decision_cli.py b/tests/test_opencode_review_decision_cli.py new file mode 100644 index 000000000..3d8c44e63 --- /dev/null +++ b/tests/test_opencode_review_decision_cli.py @@ -0,0 +1,152 @@ +"""Serialization and CLI tests for OpenCode decision envelopes.""" + +from __future__ import annotations + +import importlib.util +import json +import runpy +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_decision_test_support import MODULE_PATH, decision, envelope, finding + + +def test_direct_module_load_registers_its_support_directory( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A path-based invocation must make sibling decision modules importable.""" + module_dir = str(MODULE_PATH.parent) + monkeypatch.setattr(sys, "path", [entry for entry in sys.path if entry != module_dir]) + spec = importlib.util.spec_from_file_location( + "opencode_review_decision_direct", MODULE_PATH + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + + spec.loader.exec_module(module) + + assert sys.path[0] == module_dir + + +def test_markdown_keeps_findings_and_infrastructure_blockers_separate() -> None: + """Human summaries never render infrastructure failure as a source line.""" + report = decision.build_decision( + envelope(findings=[finding()], coverage_state="failure") + ) + markdown = decision.render_markdown(report) + assert "## Semantic findings" in markdown + assert "scripts/ci/example.py:12" in markdown + assert "## Infrastructure and policy blockers" in markdown + blocker_section = markdown.split("## Infrastructure and policy blockers", 1)[1] + assert "coverage" in blocker_section + assert ".github/workflows/opencode-review.yml:1" not in blocker_section + + +def test_strict_json_rejects_duplicate_keys_and_nonfinite_numbers(tmp_path: Path) -> None: + """Decision evidence rejects ambiguous JSON and numeric extensions.""" + duplicate = tmp_path / "duplicate.json" + duplicate.write_text('{"schema_version":"1.0","schema_version":"1.0"}') + with pytest.raises(decision.DecisionValidationError, match="duplicate JSON key"): + decision.load_json(duplicate) + + nonfinite = tmp_path / "nonfinite.json" + nonfinite.write_text('{"line": NaN}') + with pytest.raises(decision.DecisionValidationError, match="non-finite JSON number"): + decision.load_json(nonfinite) + + +def test_cli_writes_atomic_json_and_markdown_with_stable_errors( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """The CLI publishes both decision views atomically or rejects the input.""" + source = tmp_path / "input.json" + json_output = tmp_path / "nested" / "decision.json" + markdown_output = tmp_path / "nested" / "decision.md" + source.write_text(json.dumps(envelope()), encoding="utf-8") + assert ( + decision.main( + [ + "--input", + str(source), + "--json-output", + str(json_output), + "--markdown-output", + str(markdown_output), + ] + ) + == 0 + ) + assert json.loads(json_output.read_text(encoding="utf-8"))["merge_readiness"] == "READY" + assert "Review verdict: **APPROVE**" in markdown_output.read_text(encoding="utf-8") + assert not json_output.with_name(f".{json_output.name}.tmp").exists() + assert not markdown_output.with_name(f".{markdown_output.name}.tmp").exists() + + source.write_text("[]", encoding="utf-8") + assert ( + decision.main( + [ + "--input", + str(source), + "--json-output", + str(json_output), + "--markdown-output", + str(markdown_output), + ] + ) + == 2 + ) + assert "decision evidence rejected" in capsys.readouterr().err + + +def test_module_entrypoint_routes_through_main( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Direct script execution uses the same tested CLI boundary.""" + source = tmp_path / "input.json" + json_output = tmp_path / "decision.json" + markdown_output = tmp_path / "decision.md" + source.write_text(json.dumps(envelope()), encoding="utf-8") + monkeypatch.setattr( + "sys.argv", + [ + str(MODULE_PATH), + "--input", + str(source), + "--json-output", + str(json_output), + "--markdown-output", + str(markdown_output), + ], + ) + with pytest.raises(SystemExit, match="0"): + runpy.run_path(str(MODULE_PATH), run_name="__main__") + assert json_output.exists() and markdown_output.exists() + + +def test_public_production_callables_have_docstrings() -> None: + """Every production class and function remains beginner-readable.""" + missing = [ + name + for name, value in vars(decision).items() + if not name.startswith("_") + and (isinstance(value, type) or callable(value)) + and getattr(value, "__module__", None) == decision.__name__ + and not getattr(value, "__doc__", None) + ] + assert missing == [] + + +def test_load_json_wraps_syntax_and_filesystem_errors(tmp_path: Path) -> None: + """Malformed or unavailable evidence files must produce bounded stable errors.""" + malformed = tmp_path / "malformed.json" + malformed.write_text("{", encoding="utf-8") + with pytest.raises(decision.DecisionValidationError, match="cannot load"): + decision.load_json(malformed) + with pytest.raises(decision.DecisionValidationError, match="cannot load"): + decision.load_json(tmp_path / "absent.json") diff --git a/tests/test_opencode_review_decision_validation.py b/tests/test_opencode_review_decision_validation.py new file mode 100644 index 000000000..7efae66f2 --- /dev/null +++ b/tests/test_opencode_review_decision_validation.py @@ -0,0 +1,149 @@ +"""Validation tests for exact-head OpenCode decision evidence.""" + +from __future__ import annotations + +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_decision_test_support import check, decision, envelope, finding + + +def test_exact_head_binding_rejects_stale_semantic_and_merge_evidence() -> None: + """No semantic, check, or policy evidence may transfer from another head.""" + stale_semantic = envelope() + stale_semantic["semantic_review"]["reviewed_head_sha"] = "c" * 40 + with pytest.raises(decision.DecisionValidationError, match="reviewed_head_sha"): + decision.build_decision(stale_semantic) + + stale_merge = envelope() + stale_merge["merge_evidence"]["evidence_head_sha"] = "c" * 40 + with pytest.raises(decision.DecisionValidationError, match="evidence_head_sha"): + decision.build_decision(stale_merge) + + stale_check = envelope() + stale_check["merge_evidence"]["required_checks"][0]["head_sha"] = "c" * 40 + with pytest.raises( + decision.DecisionValidationError, match=r"required_checks\[0\].head_sha" + ): + decision.build_decision(stale_check) + + +def test_incomplete_semantic_review_cannot_carry_findings_or_head_claim() -> None: + """Unavailable or failed reviews abstain without synthetic source evidence.""" + for status in ("unavailable", "failed"): + with_findings = envelope(semantic_status=status, findings=[finding()]) + with pytest.raises( + decision.DecisionValidationError, match="must not contain findings" + ): + decision.build_decision(with_findings) + + with_head = envelope(semantic_status=status) + with_head["semantic_review"]["reviewed_head_sha"] = "b" * 40 + with pytest.raises(decision.DecisionValidationError, match="must be null"): + decision.build_decision(with_head) + + +def test_complete_semantic_review_requires_exact_head() -> None: + """A completed semantic verdict without an exact reviewed head is invalid.""" + value = envelope() + value["semantic_review"]["reviewed_head_sha"] = None + with pytest.raises(decision.DecisionValidationError, match="reviewed_head_sha"): + decision.build_decision(value) + + +@pytest.mark.parametrize( + ("mutate", "message"), + [ + (lambda value: value.update({"unexpected": True}), "unknown fields"), + ( + lambda value: value["semantic_review"].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["merge_evidence"].update({"unexpected": True}), + "unknown fields", + ), + ( + lambda value: value["semantic_review"]["findings"][0].update( + {"unexpected": True} + ), + "unknown fields", + ), + ( + lambda value: value["merge_evidence"]["required_checks"][0].update( + {"unexpected": True} + ), + "unknown fields", + ), + (lambda value: value.update({"pull_request_number": True}), "integer"), + ( + lambda value: value["semantic_review"].update({"status": "running"}), + "semantic_review.status", + ), + ( + lambda value: value["merge_evidence"].update( + {"coverage_state": "green"} + ), + "coverage_state", + ), + ( + lambda value: value["semantic_review"]["findings"][0].update( + {"path": "../secret"} + ), + "relative source path", + ), + ( + lambda value: value["semantic_review"]["findings"][0].update({"line": 0}), + "positive integer", + ), + ], +) +def test_strict_schema_rejects_unknown_fields_and_scalar_confusion( + mutate: Any, message: str +) -> None: + """Every evidence layer fails closed on malformed control data.""" + value = envelope(findings=[finding()]) + mutate(value) + with pytest.raises(decision.DecisionValidationError, match=message): + decision.build_decision(value) + + +def test_duplicate_finding_and_check_names_are_rejected() -> None: + """Duplicate semantic or check identities cannot inflate evidence counts.""" + with pytest.raises(decision.DecisionValidationError, match="finding_id"): + decision.build_decision(envelope(findings=[finding(), finding()])) + + with pytest.raises(decision.DecisionValidationError, match="check name"): + decision.build_decision(envelope(checks=[check("CI"), check("ci")])) + + +def test_validation_helpers_reject_remaining_invalid_shapes_and_scalars() -> None: + """Primitive schema helpers must reject unsupported JSON shapes and scalar values.""" + with pytest.raises(decision.DecisionValidationError, match="must be an array"): + decision.array_value({}, "array") + with pytest.raises(decision.DecisionValidationError, match="non-empty text"): + decision.text_value(" ", "text") + with pytest.raises(decision.DecisionValidationError, match="must be boolean"): + decision.bool_value(1, "flag") + with pytest.raises(decision.DecisionValidationError, match="commit SHA"): + decision.commit_sha_value("main", "head") + + +def test_top_level_schema_and_repository_coordinates_are_strict() -> None: + """Decision identity must use the exact schema version and owner/name repository form.""" + wrong_version = envelope() + wrong_version["schema_version"] = "2.0" + with pytest.raises(decision.DecisionValidationError, match="schema_version"): + decision.build_decision(wrong_version) + + invalid_repository = envelope() + invalid_repository["repository"] = "missing-slash" + with pytest.raises(decision.DecisionValidationError, match="owner/name"): + decision.build_decision(invalid_repository) diff --git a/tests/test_opencode_review_shadow_execution.py b/tests/test_opencode_review_shadow_execution.py new file mode 100644 index 000000000..0a4413607 --- /dev/null +++ b/tests/test_opencode_review_shadow_execution.py @@ -0,0 +1,358 @@ +"""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 = "", + leak_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" + "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) + 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" + output.mkdir() + 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_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: + """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, + ) + + 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.""" + 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..f6973425a --- /dev/null +++ b/tests/test_opencode_review_shadow_validation.py @@ -0,0 +1,411 @@ +"""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.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"), + [ + (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) + + +@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: + """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", + ]