diff --git a/.github/workflows/central-review.yml b/.github/workflows/central-review.yml index a96cbff40..e3ebe944f 100644 --- a/.github/workflows/central-review.yml +++ b/.github/workflows/central-review.yml @@ -231,9 +231,14 @@ jobs: run: | set -euo pipefail python - <<'PY' + import hashlib import os + from datetime import datetime, timedelta, timezone from pathlib import Path + from noema_reviewer.claim_evidence_runtime import ( + produce_current_head_source_manifest, + ) from noema_reviewer.github_io import fetch_manifest from noema_reviewer.sandbox import DockerCodeGraphRunner @@ -248,6 +253,26 @@ jobs: ) output = Path(os.environ["RUNNER_TEMP"]) / "noema-manifest.json" output.write_text(manifest.model_dump_json(indent=2), encoding="utf-8") + issued_at = datetime.now(timezone.utc) + claim_manifest = produce_current_head_source_manifest( + manifest, + source_root=Path(source_root), + workflow_ref=( + "ContextualWisdomLab/noema/.github/workflows/central-review.yml@" + + os.environ["GITHUB_SHA"] + ), + run_id=int(os.environ["GITHUB_RUN_ID"]), + run_attempt=int(os.environ["GITHUB_RUN_ATTEMPT"]), + issued_at=issued_at, + expires_at=issued_at + timedelta(hours=6), + ) + claim_output = Path(os.environ["RUNNER_TEMP"]) / "noema-claim-evidence-manifest.json" + claim_output.write_bytes(claim_manifest) + claim_digest = hashlib.sha256(claim_manifest).hexdigest() + (Path(os.environ["RUNNER_TEMP"]) / "noema-claim-evidence-manifest.sha256").write_text( + f"{claim_digest} noema-claim-evidence-manifest.json\n", + encoding="ascii", + ) print( f"Collected bounded manifest for {manifest.repo}#{manifest.pr_number} " f"head={manifest.head_sha} checks={len(manifest.check_conclusions)} " @@ -256,6 +281,7 @@ jobs: PY cd "$RUNNER_TEMP" sha256sum noema-manifest.json >noema-manifest.sha256 + sha256sum --check noema-claim-evidence-manifest.sha256 - name: Upload short-lived bounded review manifest uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 @@ -264,6 +290,8 @@ jobs: path: | ${{ runner.temp }}/noema-manifest.json ${{ runner.temp }}/noema-manifest.sha256 + ${{ runner.temp }}/noema-claim-evidence-manifest.json + ${{ runner.temp }}/noema-claim-evidence-manifest.sha256 if-no-files-found: error retention-days: 1 @@ -293,6 +321,7 @@ jobs: set -euo pipefail cd "$RUNNER_TEMP/noema-attestation-input" sha256sum --check noema-manifest.sha256 + sha256sum --check noema-claim-evidence-manifest.sha256 jq -e \ --arg repo "$TARGET_REPOSITORY" \ --argjson pr "$PR_NUMBER" \ @@ -310,20 +339,34 @@ jobs: create-storage-record: false show-summary: false - - name: Stage signed manifest attestation + - name: Attest claim evidence manifest provenance + id: attest_claim_evidence + uses: actions/attest@59d89421af93a897026c735860bf21b6eb4f7b26 # v4.1.0 + with: + subject-path: ${{ runner.temp }}/noema-attestation-input/noema-claim-evidence-manifest.json + create-storage-record: false + show-summary: false + + - name: Stage signed manifest attestations env: - ATTESTATION_BUNDLE_PATH: ${{ steps.attest.outputs.bundle-path }} + REVIEW_ATTESTATION_BUNDLE_PATH: ${{ steps.attest.outputs.bundle-path }} + CLAIM_ATTESTATION_BUNDLE_PATH: ${{ steps.attest_claim_evidence.outputs.bundle-path }} run: | set -euo pipefail - test -f "$ATTESTATION_BUNDLE_PATH" - install -m 0600 "$ATTESTATION_BUNDLE_PATH" \ + test -f "$REVIEW_ATTESTATION_BUNDLE_PATH" + test -f "$CLAIM_ATTESTATION_BUNDLE_PATH" + install -m 0600 "$REVIEW_ATTESTATION_BUNDLE_PATH" \ "$RUNNER_TEMP/noema-manifest-attestation.json" + install -m 0600 "$CLAIM_ATTESTATION_BUNDLE_PATH" \ + "$RUNNER_TEMP/noema-claim-evidence-manifest-attestation.json" - name: Upload short-lived signed manifest attestation uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: noema-manifest-attestation-${{ needs.collect_evidence.outputs.head_sha }} - path: ${{ runner.temp }}/noema-manifest-attestation.json + path: | + ${{ runner.temp }}/noema-manifest-attestation.json + ${{ runner.temp }}/noema-claim-evidence-manifest-attestation.json if-no-files-found: error retention-days: 1 @@ -383,6 +426,9 @@ jobs: install -m 0600 \ "$RUNNER_TEMP/noema-attestation/noema-manifest-attestation.json" \ noema-manifest-attestation.json + install -m 0600 \ + "$RUNNER_TEMP/noema-attestation/noema-claim-evidence-manifest-attestation.json" \ + noema-claim-evidence-manifest-attestation.json gh attestation verify noema-manifest.json \ --bundle noema-manifest-attestation.json \ --repo ContextualWisdomLab/noema \ @@ -392,7 +438,17 @@ jobs: --source-ref refs/heads/main \ --cert-oidc-issuer https://token.actions.githubusercontent.com \ --deny-self-hosted-runners + gh attestation verify noema-claim-evidence-manifest.json \ + --bundle noema-claim-evidence-manifest-attestation.json \ + --repo ContextualWisdomLab/noema \ + --signer-workflow ContextualWisdomLab/noema/.github/workflows/central-review.yml \ + --signer-digest "$GITHUB_SHA" \ + --source-digest "$GITHUB_SHA" \ + --source-ref refs/heads/main \ + --cert-oidc-issuer https://token.actions.githubusercontent.com \ + --deny-self-hosted-runners sha256sum --check noema-manifest.sha256 + sha256sum --check noema-claim-evidence-manifest.sha256 live_json="$(gh api "repos/${TARGET_REPOSITORY}/pulls/${PR_NUMBER}")" state="$(jq -r '.state // empty' <<<"$live_json")" live_head="$(jq -r '.head.sha // empty' <<<"$live_json")" @@ -459,6 +515,11 @@ jobs: set +e python -m noema_reviewer \ --manifest-file "$RUNNER_TEMP/noema-evidence/noema-manifest.json" \ + --claim-evidence-manifest-file "$RUNNER_TEMP/noema-evidence/noema-claim-evidence-manifest.json" \ + --claim-evidence-manifest-sha256 "$(cut -d ' ' -f1 "$RUNNER_TEMP/noema-evidence/noema-claim-evidence-manifest.sha256")" \ + --claim-evidence-workflow-ref "ContextualWisdomLab/noema/.github/workflows/central-review.yml@${GITHUB_SHA}" \ + --claim-evidence-run-id "$GITHUB_RUN_ID" \ + --claim-evidence-run-attempt "$GITHUB_RUN_ATTEMPT" \ --strict \ --publish \ --token-source "$NOEMA_REVIEW_TOKEN_SOURCE" \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 5574ceeb3..9ccb042e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ # Changelog +- Add a Noema-owned exact-claim evidence receipt contract whose execution and research producers serialize one canonical artifact that binds every receipt semantic field, including command/result/isolation/network or source revision/excerpt/retrieval policy. Admission accepts only a receipt ID from untrusted model output. The owner API first verifies the exact authenticated OpenCode-handoff manifest digest, canonical envelope bytes, reviewed producer-to-kind policy, and repository/head/workflow/run/attempt identity before it can construct an immutable typed index; admission then reconstructs each canonical artifact and verifies time/claim/artifact identity. The version-2 manifest now binds a separate producer-authenticated `ClaimEvidenceRequirement` containing the exact claim, independently required evidence kind, and `context` or `finding` publication authority. Raw current-head source lines are context only: they are withheld from finding-reference prompts and cannot publish a finding or `request_changes`; an explicitly producer-authorized source finding remains usable and retains exact path/line checks. Finding-free model `request_changes` and `blocked` verdicts cannot bypass receipt admission to publish a vacuous blocking review. Requirement/receipt kind mismatch, fixed-artifact semantic substitution, caller-supplied receipt dictionaries, model self-classification, stale identities, cross-kind receipts, marker-only sandbox output, noncanonical artifact bytes, and expired receipts fail closed before the GitHub publisher. This remains the owner prerequisite for ContextualWisdomLab/.github#1641 and issue #555. The reviewed `sandboxed_verify` adapter exists in owner source, but its actual central stdout/stderr/marker-to-manifest wiring and the trusted research producer are not yet integrated; exact-head hosted GREEN, immutable release, and the verified central consumer bump remain required. + ## Unreleased - `noema-core` provider-neutral Shared Kernel을 추가하여 이미 해석된 PydanticAI `Model`과 역할별 prompt/schema만 받아 Agent를 구성한다. 문자열 model identifier와 provider discovery·credential·routing·retry·failover는 Shared Kernel 밖에 두고 `Agent(..., retries=0)`으로 repository-local model-attempt authority를 만들지 않는다. Reviewer wheel·sdist·editable 설치는 canonical `packages/noema-core` source를 포함하거나 참조하며 별도 100% coverage·docstring과 clean install smoke로 검증한다. 외부 소비는 immutable versioned publication·exact source identity·SBOM/provenance·licensing/NOTICE·compatibility/rollback evidence 전에는 허용하지 않는다. - `writeAcquisitionPrivateFile`의 기존 대상 사전-교체 검증 read(`existingDescriptor` open)에 `O_NONBLOCK`을 추가해 fail-closed를 강화한다. 이 open은 이미 필수 filesystem capability로 `O_NONBLOCK`을 검증했지만 실제로는 사용하지 않아, 로컬 권한을 가진 행위자가 사전 `lstatSync` 정규 파일 확인과 이 open 사이에 대상 경로를 FIFO로 교체하면 writer가 나타날 때까지 무한정 블로킹해 writer lease를 계속 점유할 수 있었다. `O_NONBLOCK`은 정규 파일에는 영향이 없고, FIFO에서는 open이 즉시 반환되어 이어지는 descriptor 타입 검증이 그대로 fail-closed로 거부한다. 회귀 테스트(`test/acquisition-private-output-existing-target-nonblocking.test.ts`)와 기존 open-flags 계약 테스트 갱신으로 고정했다. diff --git a/reviewer/README.md b/reviewer/README.md index 1af6e874a..b36849fb0 100644 --- a/reviewer/README.md +++ b/reviewer/README.md @@ -52,6 +52,30 @@ carries structured actionability rather than relying on free-form prose: } ``` +### Claim-evidence publication authority + +An authenticated claim-evidence manifest entry binds two separate objects: a +producer receipt and a `ClaimEvidenceRequirement`. The requirement owns the +exact claim, independently required evidence kind, and either `context` or +`finding` publication authority. The verifier rejects a requirement/receipt +kind mismatch; the model supplies neither field and can cite only a receipt ID +for a producer-authorized exact finding claim. + +The bounded current-head line collector marks raw source lines as `context`. +Those receipts remain authenticated source context, but are withheld from the +model's finding-reference list and cannot authorize a published finding or +`request_changes`. A trusted source producer may explicitly issue `source` + +`finding` authority for an exact source-only defect; exact path and line binding +still applies. Execution or research claims require their independently bound +kind, so an authentic source line cannot be reused as proof of runtime or +external behavior. Admission fails before deterministic gates and the GitHub +publisher instead of rewriting an unsupported claim into a blocking verdict. + +This contract does not by itself claim that the central `sandboxed_verify` +result or a trusted research retrieval has populated the production manifest. +Those producer adapters, protected exact-head verification, immutable release, +and the released `.github` consumer remain separate completion conditions. + `check_name` is optional for ordinary source, SARIF, dependency, and review-thread findings. A finding offered as the RCA for a failed current-head check must bind to that exact check name. The deterministic gate requires each ordinary failed diff --git a/reviewer/noema_reviewer/__init__.py b/reviewer/noema_reviewer/__init__.py index d61d6fa8e..fc1e1c7f2 100644 --- a/reviewer/noema_reviewer/__init__.py +++ b/reviewer/noema_reviewer/__init__.py @@ -17,6 +17,38 @@ from typing import Any +from .claim_evidence import ( + ClaimEvidenceRequirement, + ClaimEvidenceReceipt, + ClaimPublicationAuthority, + EvidenceKind, + ExecutionClaimReceipt, + ProducedClaimEvidence, + ResearchClaimReceipt, + SourceClaimReceipt, + VerifiedClaimEvidenceIndex, + admit_claim_evidence, + index_claim_evidence_receipts, + produce_claim_evidence_manifest, + produce_execution_claim_receipt, + produce_research_claim_receipt, + sha256_text, + verify_claim_evidence_manifest, +) +from .claim_evidence_reference import ( + admit_claim_evidence_reference, + parse_claim_evidence_reference, +) +from .claim_evidence_runtime import ( + admit_review_verdict_evidence, + produce_current_head_source_manifest, + prompt_claim_evidence_references, + verify_claim_evidence_file, +) +from .sandboxed_verify_claim_evidence import ( + produce_sandboxed_verify_execution_claim_receipt, +) +from .source_claim_evidence import produce_source_claim_receipt from .manifest import ReviewManifest from .models import Confidence, EvidenceType, Finding, Priority, ReviewVerdict, Severity, Verdict from .patch_image_validation import ( @@ -50,6 +82,12 @@ def __getattr__(name: str) -> Any: __all__ = [ + "ClaimEvidenceRequirement", + "ClaimEvidenceReceipt", + "ClaimPublicationAuthority", + "EvidenceKind", + "ExecutionClaimReceipt", + "ProducedClaimEvidence", "Confidence", "DockerPatchValidationRunner", "DockerPatchValidatorImageRunner", @@ -64,13 +102,31 @@ def __getattr__(name: str) -> Any: "PatchValidatorImageResult", "PatchValidatorImageStatus", "PydanticAIReviewAgent", + "ResearchClaimReceipt", "Priority", "ReviewAgent", "ReviewManifest", "ReviewVerdict", "Severity", + "SourceClaimReceipt", + "VerifiedClaimEvidenceIndex", "Verdict", + "admit_claim_evidence", + "admit_claim_evidence_reference", + "admit_review_verdict_evidence", "build_agent", "inspect_patch_bytes", + "index_claim_evidence_receipts", "inspect_patch_for_image", + "parse_claim_evidence_reference", + "produce_claim_evidence_manifest", + "produce_current_head_source_manifest", + "produce_execution_claim_receipt", + "produce_research_claim_receipt", + "produce_sandboxed_verify_execution_claim_receipt", + "produce_source_claim_receipt", + "prompt_claim_evidence_references", + "sha256_text", + "verify_claim_evidence_file", + "verify_claim_evidence_manifest", ] diff --git a/reviewer/noema_reviewer/agent.py b/reviewer/noema_reviewer/agent.py index ac95b8e66..3dbced7cd 100644 --- a/reviewer/noema_reviewer/agent.py +++ b/reviewer/noema_reviewer/agent.py @@ -10,6 +10,8 @@ from __future__ import annotations +from collections.abc import Callable +from datetime import datetime, timezone from typing import Protocol, runtime_checkable from noema_core import NOEMA_PERSONA @@ -17,6 +19,11 @@ from pydantic_ai import Agent, ModelSettings from pydantic_ai.models import Model +from .claim_evidence import VerifiedClaimEvidenceIndex +from .claim_evidence_runtime import ( + admit_review_verdict_evidence, + prompt_claim_evidence_references, +) from .config import ReviewerConfig, resolve_config, resolve_model from .gating import apply_gates from .manifest import ReviewManifest @@ -85,8 +92,11 @@ def _dependency_lines(manifest: ReviewManifest) -> list[str]: return lines -def build_prompt(manifest: ReviewManifest) -> str: - """Build the bounded user prompt handed to the model for one review.""" +def build_prompt( + manifest: ReviewManifest, + trusted_index: VerifiedClaimEvidenceIndex | None = None, +) -> str: + """Build the bounded prompt plus producer-authenticated receipt references.""" sections: list[str] = [ f"Repository: {manifest.repo}", f"PR: #{manifest.pr_number}", @@ -120,7 +130,13 @@ def build_prompt(manifest: ReviewManifest) -> str: files = [f"### {changed.path}\n{changed.content}" for changed in manifest.changed_files] if files: sections.append("Changed-file context:\n" + "\n\n".join(files)) - + claim_references = prompt_claim_evidence_references(trusted_index) + if claim_references: + sections.append( + "Trusted claim evidence references (copy an exact full line into Finding.evidence; " + "do not alter the claim or receipt ID):\n- " + + "\n- ".join(claim_references) + ) sections.append("Diff:\n" + (manifest.diff or "(no diff provided)")) return "\n\n".join(sections) @@ -140,6 +156,8 @@ def __init__( model: Model, *, model_settings: ModelSettings | None = None, + claim_evidence_index: VerifiedClaimEvidenceIndex | None = None, + admitted_at: Callable[[], datetime] | None = None, ) -> None: """Build the agent around an already resolved real or test model.""" if isinstance(model, str): @@ -147,6 +165,8 @@ def __init__( "PydanticAIReviewAgent requires a pre-resolved Model; " "provider/model routing belongs to contextual-orchestrator" ) + self._claim_evidence_index = claim_evidence_index + self._admitted_at = admitted_at or (lambda: datetime.now(timezone.utc)) self._agent: Agent[None, ReviewVerdict] = build_core_agent( model, output_type=ReviewVerdict, @@ -154,18 +174,48 @@ def __init__( ) self._model_settings = model_settings + def bind_claim_evidence( + self, + trusted_index: VerifiedClaimEvidenceIndex, + *, + admitted_at: Callable[[], datetime] | None = None, + ) -> "PydanticAIReviewAgent": + """Bind one verified workflow index before model execution and publication.""" + if self._claim_evidence_index is not None: + raise ValueError("claim evidence index is already bound") + self._claim_evidence_index = trusted_index + if admitted_at is not None: + self._admitted_at = admitted_at + return self + + def prompt_for(self, manifest: ReviewManifest) -> str: + """Return the exact prompt including only verified receipt references.""" + return build_prompt(manifest, self._claim_evidence_index) + def review(self, manifest: ReviewManifest, *, strict: bool = False) -> ReviewVerdict: - """Run the model over the manifest and apply the deterministic gates.""" - prompt = build_prompt(manifest) - result = self._agent.run_sync(prompt, model_settings=self._model_settings) - return apply_gates(manifest, result.output, strict=strict) + """Admit model evidence before deterministic gates can add trusted findings.""" + result = self._agent.run_sync( + self.prompt_for(manifest), + model_settings=self._model_settings, + ) + admitted = admit_review_verdict_evidence( + result.output, + trusted_index=self._claim_evidence_index, + admitted_at=self._admitted_at(), + ) + return apply_gates(manifest, admitted, strict=strict) -def build_agent(config: ReviewerConfig | None = None) -> PydanticAIReviewAgent: - """Build a production review agent from one validated gateway configuration.""" +def build_agent( + config: ReviewerConfig | None = None, + *, + claim_evidence_index: VerifiedClaimEvidenceIndex | None = None, +) -> PydanticAIReviewAgent: + """Build a production reviewer with an optional verified evidence index.""" resolved = config or resolve_config() model = resolve_model(resolved) return PydanticAIReviewAgent( model, model_settings=model_settings_for_config(resolved), + claim_evidence_index=claim_evidence_index, ) diff --git a/reviewer/noema_reviewer/claim_evidence.py b/reviewer/noema_reviewer/claim_evidence.py new file mode 100644 index 000000000..487a9b83c --- /dev/null +++ b/reviewer/noema_reviewer/claim_evidence.py @@ -0,0 +1,606 @@ +"""Produce and admit exact-claim evidence at the Noema reviewer boundary. + +Trusted workflow producers create canonical artifacts and frozen receipts. An +authenticated manifest requirement independently owns the exact claim, required +evidence kind, and publication authority. The untrusted model may cite only a +receipt ID; it never supplies those policy fields, the authoritative receipt +payload, producer identity, or artifact path. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from base64 import b64decode, b64encode +from binascii import Error as Base64Error +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from enum import Enum +from types import MappingProxyType +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, model_validator + + +_SHA256_PATTERN = r"^[0-9a-f]{64}$" +_HEAD_SHA_PATTERN = r"^[0-9a-f]{40}$" +_REPOSITORY_PATTERN = r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$" +_WORKFLOW_REF_PATTERN = ( + r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+/\.github/workflows/" + r"[A-Za-z0-9_.-]+\.ya?ml@[0-9a-f]{40}$" +) +_SLUG_PATTERN = r"^[a-z0-9][a-z0-9._-]{0,79}$" +_NonEmptyText = Annotated[str, Field(min_length=1)] + + +class EvidenceKind(str, Enum): + """Trusted producer classes; model prose cannot select admission authority.""" + + SOURCE = "source" + EXECUTION = "execution" + RESEARCH = "research" + + +class ClaimPublicationAuthority(str, Enum): + """Producer-owned authority for using a claim in a published finding.""" + + CONTEXT = "context" + FINDING = "finding" + + +class ClaimEvidenceRequirement(BaseModel): + """Authenticated claim policy independent from the cited receipt payload.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + claim: str = Field(min_length=1) + required_evidence_kind: EvidenceKind + publication_authority: ClaimPublicationAuthority + + +class _ReceiptIdentity(BaseModel): + """Exact workflow and artifact identity shared by every receipt class.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal[1] + receipt_id: str = Field(pattern=_SLUG_PATTERN) + repository: str = Field(pattern=_REPOSITORY_PATTERN) + head_sha: str = Field(pattern=_HEAD_SHA_PATTERN) + workflow_ref: str = Field(pattern=_WORKFLOW_REF_PATTERN) + run_id: int = Field(gt=0) + run_attempt: int = Field(gt=0) + claim_sha256: str = Field(pattern=_SHA256_PATTERN) + artifact_sha256: str = Field(pattern=_SHA256_PATTERN) + artifact_size: int = Field(gt=0) + producer_id: str = Field(pattern=_SLUG_PATTERN) + producer_version: str = Field(pattern=_SLUG_PATTERN) + policy_version: str = Field(pattern=_SLUG_PATTERN) + issued_at: datetime + expires_at: datetime + + @model_validator(mode="after") + def require_bounded_aware_validity(self) -> "_ReceiptIdentity": + """Reject ambiguous or unbounded receipt validity windows.""" + if self.issued_at.utcoffset() is None or self.expires_at.utcoffset() is None: + raise ValueError("claim evidence receipt timestamps must be timezone-aware") + if self.expires_at <= self.issued_at: + raise ValueError("claim evidence receipt expiry must follow issue time") + return self + + +class SourceClaimReceipt(_ReceiptIdentity): + """Receipt proving exact current-head source bytes, not external behavior.""" + + evidence_kind: Literal[EvidenceKind.SOURCE] + source_path: str = Field(min_length=1) + source_line: int = Field(gt=0) + source_line_sha256: str = Field(pattern=_SHA256_PATTERN) + + +class ExecutionClaimReceipt(_ReceiptIdentity): + """Receipt proving one sandboxed execution result and its bounded output.""" + + evidence_kind: Literal[EvidenceKind.EXECUTION] + argv: tuple[_NonEmptyText, ...] = Field(min_length=1) + tool_identity: str = Field(min_length=1) + tool_version: str = Field(min_length=1) + exit_code: int + stdout_sha256: str = Field(pattern=_SHA256_PATTERN) + stderr_sha256: str = Field(pattern=_SHA256_PATTERN) + isolation_policy: str = Field(min_length=1) + network_policy: str = Field(min_length=1) + + +class ResearchClaimReceipt(_ReceiptIdentity): + """Receipt proving one immutable external-source retrieval and excerpt.""" + + evidence_kind: Literal[EvidenceKind.RESEARCH] + source_uri: str = Field(min_length=1) + source_revision: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + excerpt_sha256: str = Field(pattern=_SHA256_PATTERN) + retrieval_policy: str = Field(min_length=1) + + +ClaimEvidenceReceipt = Annotated[ + SourceClaimReceipt | ExecutionClaimReceipt | ResearchClaimReceipt, + Field(discriminator="evidence_kind"), +] +_RECEIPT_ADAPTER = TypeAdapter(ClaimEvidenceReceipt) + + +@dataclass(frozen=True) +class ProducedClaimEvidence: + """One producer-owned receipt plus its canonical upload artifact bytes.""" + + receipt: ClaimEvidenceReceipt + artifact: bytes + + +class VerifiedClaimEvidenceIndex: + """Immutable receipt index constructible only by manifest verification.""" + + __slots__ = ("artifacts", "claims", "receipts", "requirements") + + def __init__(self) -> None: + """Reject direct construction that would bypass manifest verification.""" + raise TypeError("use verify_claim_evidence_manifest") + + @classmethod + def _from_verified( + cls, + *, + receipts: Mapping[str, ClaimEvidenceReceipt], + artifacts: Mapping[str, bytes], + claims: Mapping[str, str], + requirements: Mapping[str, ClaimEvidenceRequirement], + ) -> "VerifiedClaimEvidenceIndex": + """Build one immutable index after all envelope checks pass.""" + instance = object.__new__(cls) + instance.receipts = MappingProxyType(dict(receipts)) + instance.artifacts = MappingProxyType(dict(artifacts)) + instance.claims = MappingProxyType(dict(claims)) + instance.requirements = MappingProxyType(dict(requirements)) + return instance + + +def _sha256_bytes(value: bytes) -> str: + """Return the lowercase SHA-256 digest of exact bytes.""" + return hashlib.sha256(value).hexdigest() + + +def sha256_text(value: str) -> str: + """Return the lowercase SHA-256 digest of exact UTF-8 claim bytes.""" + return _sha256_bytes(value.encode("utf-8")) + + +def _utc_text(value: datetime) -> str: + """Return one canonical UTC timestamp for a producer artifact.""" + if value.utcoffset() is None: + raise ValueError("claim evidence artifact timestamps must be timezone-aware") + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _canonical_artifact(payload: Mapping[str, object]) -> bytes: + """Serialize one semantic receipt payload with deterministic exact bytes.""" + return ( + json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + "\n" + ).encode("utf-8") + + +def _common_artifact_payload( + *, + receipt_id: str, + evidence_kind: EvidenceKind, + repository: str, + head_sha: str, + workflow_ref: str, + run_id: int, + run_attempt: int, + claim: str, + producer_id: str, + producer_version: str, + policy_version: str, + issued_at: datetime, + expires_at: datetime, +) -> dict[str, object]: + """Return shared semantic fields sealed into every evidence artifact.""" + return { + "schema_version": 1, + "receipt_id": receipt_id, + "evidence_kind": evidence_kind.value, + "repository": repository, + "head_sha": head_sha, + "workflow_ref": workflow_ref, + "run_id": run_id, + "run_attempt": run_attempt, + "claim_sha256": sha256_text(claim), + "producer_id": producer_id, + "producer_version": producer_version, + "policy_version": policy_version, + "issued_at": _utc_text(issued_at), + "expires_at": _utc_text(expires_at), + } + + +def _receipt_identity( + payload: Mapping[str, object], artifact: bytes +) -> dict[str, object]: + """Add exact canonical artifact identity to shared receipt fields.""" + return { + key: payload[key] + for key in ( + "schema_version", + "receipt_id", + "repository", + "head_sha", + "workflow_ref", + "run_id", + "run_attempt", + "claim_sha256", + "producer_id", + "producer_version", + "policy_version", + "issued_at", + "expires_at", + ) + } | { + "artifact_sha256": _sha256_bytes(artifact), + "artifact_size": len(artifact), + } + + +def produce_execution_claim_receipt( + *, + receipt_id: str, + repository: str, + head_sha: str, + workflow_ref: str, + run_id: int, + run_attempt: int, + claim: str, + producer_id: str, + producer_version: str, + policy_version: str, + issued_at: datetime, + expires_at: datetime, + argv: Sequence[str], + tool_identity: str, + tool_version: str, + exit_code: int, + stdout: bytes, + stderr: bytes, + isolation_policy: str, + network_policy: str, +) -> ProducedClaimEvidence: + """Seal execution semantics and output digests into one canonical artifact.""" + stdout_sha256 = _sha256_bytes(stdout) + stderr_sha256 = _sha256_bytes(stderr) + payload = _common_artifact_payload( + receipt_id=receipt_id, + evidence_kind=EvidenceKind.EXECUTION, + repository=repository, + head_sha=head_sha, + workflow_ref=workflow_ref, + run_id=run_id, + run_attempt=run_attempt, + claim=claim, + producer_id=producer_id, + producer_version=producer_version, + policy_version=policy_version, + issued_at=issued_at, + expires_at=expires_at, + ) | { + "argv": list(argv), + "tool_identity": tool_identity, + "tool_version": tool_version, + "exit_code": exit_code, + "stdout_sha256": stdout_sha256, + "stderr_sha256": stderr_sha256, + "isolation_policy": isolation_policy, + "network_policy": network_policy, + } + artifact = _canonical_artifact(payload) + receipt = ExecutionClaimReceipt( + **_receipt_identity(payload, artifact), + evidence_kind=EvidenceKind.EXECUTION, + argv=tuple(argv), + tool_identity=tool_identity, + tool_version=tool_version, + exit_code=exit_code, + stdout_sha256=stdout_sha256, + stderr_sha256=stderr_sha256, + isolation_policy=isolation_policy, + network_policy=network_policy, + ) + return ProducedClaimEvidence(receipt=receipt, artifact=artifact) + + +def produce_research_claim_receipt( + *, + receipt_id: str, + repository: str, + head_sha: str, + workflow_ref: str, + run_id: int, + run_attempt: int, + claim: str, + producer_id: str, + producer_version: str, + policy_version: str, + issued_at: datetime, + expires_at: datetime, + source_uri: str, + retrieved_content: bytes, + excerpt: bytes, + retrieval_policy: str, +) -> ProducedClaimEvidence: + """Seal content-addressed retrieval semantics into one canonical artifact.""" + source_revision = f"sha256:{_sha256_bytes(retrieved_content)}" + excerpt_sha256 = _sha256_bytes(excerpt) + payload = _common_artifact_payload( + receipt_id=receipt_id, + evidence_kind=EvidenceKind.RESEARCH, + repository=repository, + head_sha=head_sha, + workflow_ref=workflow_ref, + run_id=run_id, + run_attempt=run_attempt, + claim=claim, + producer_id=producer_id, + producer_version=producer_version, + policy_version=policy_version, + issued_at=issued_at, + expires_at=expires_at, + ) | { + "source_uri": source_uri, + "source_revision": source_revision, + "excerpt_sha256": excerpt_sha256, + "retrieval_policy": retrieval_policy, + } + artifact = _canonical_artifact(payload) + receipt = ResearchClaimReceipt( + **_receipt_identity(payload, artifact), + evidence_kind=EvidenceKind.RESEARCH, + source_uri=source_uri, + source_revision=source_revision, + excerpt_sha256=excerpt_sha256, + retrieval_policy=retrieval_policy, + ) + return ProducedClaimEvidence(receipt=receipt, artifact=artifact) + + +def index_claim_evidence_receipts( + receipts: Sequence[ClaimEvidenceReceipt], +) -> dict[str, ClaimEvidenceReceipt]: + """Index unique producer objects from an authenticated out-of-band manifest.""" + indexed: dict[str, ClaimEvidenceReceipt] = {} + for receipt in receipts: + if not isinstance( + receipt, + (SourceClaimReceipt, ExecutionClaimReceipt, ResearchClaimReceipt), + ): + raise TypeError("trusted claim evidence index requires receipt objects") + if receipt.receipt_id in indexed: + raise ValueError("duplicate claim evidence receipt ID") + indexed[receipt.receipt_id] = receipt + return indexed + + +def produce_claim_evidence_manifest( + entries: Sequence[tuple[ClaimEvidenceRequirement, ProducedClaimEvidence]], +) -> bytes: + """Bind producer results to independent claim policy in one canonical body.""" + payload_entries: list[dict[str, object]] = [] + receipt_ids: set[str] = set() + for requirement, produced in entries: + if not isinstance(requirement, ClaimEvidenceRequirement): + raise TypeError("claim evidence manifest requires authenticated claim policy") + receipt = produced.receipt + if receipt.receipt_id in receipt_ids: + raise ValueError("duplicate claim evidence receipt ID") + if receipt.claim_sha256 != sha256_text(requirement.claim): + raise ValueError("claim evidence receipt claim digest mismatch") + if receipt.evidence_kind is not requirement.required_evidence_kind: + raise ValueError("claim evidence requirement kind mismatch") + receipt_ids.add(receipt.receipt_id) + payload_entries.append( + { + "artifact_base64": b64encode(produced.artifact).decode("ascii"), + "claim_requirement": requirement.model_dump(mode="json"), + "receipt": receipt.model_dump(mode="json"), + } + ) + return _canonical_artifact( + {"schema_version": 2, "entries": payload_entries} + ) + + +def verify_claim_evidence_manifest( + manifest: bytes, + *, + expected_manifest_sha256: str, + expected_repository: str, + expected_head_sha: str, + expected_workflow_ref: str, + expected_run_id: int, + expected_run_attempt: int, + expected_producers: Mapping[str, EvidenceKind], +) -> VerifiedClaimEvidenceIndex: + """Verify an authenticated manifest and return its immutable receipt index. + + The expected manifest digest comes from the already authenticated OpenCode + artifact handoff. Producer identity and allowed evidence kind come from + reviewed caller policy, never from model output or the manifest itself. + + Raises: + ValueError: If manifest bytes, shape, identity, producer policy, claim, + artifact, or canonical serialization do not match exactly. + """ + if not isinstance(expected_manifest_sha256, str) or not re.fullmatch( + _SHA256_PATTERN, expected_manifest_sha256 + ): + raise ValueError("claim evidence manifest expected digest is invalid") + if _sha256_bytes(manifest) != expected_manifest_sha256: + raise ValueError("claim evidence manifest digest mismatch") + try: + payload = json.loads(manifest) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError("claim evidence manifest is not valid JSON") from exc + if ( + not isinstance(payload, dict) + or set(payload) != {"schema_version", "entries"} + or payload.get("schema_version") != 2 + or not isinstance(payload.get("entries"), list) + ): + raise ValueError("claim evidence manifest shape is invalid") + if manifest != _canonical_artifact(payload): + raise ValueError("claim evidence manifest is not canonical") + + receipts: list[ClaimEvidenceReceipt] = [] + artifacts: dict[str, bytes] = {} + claims: dict[str, str] = {} + requirements: dict[str, ClaimEvidenceRequirement] = {} + for entry in payload["entries"]: + if not isinstance(entry, dict) or set(entry) != { + "artifact_base64", + "claim_requirement", + "receipt", + }: + raise ValueError("claim evidence manifest entry shape is invalid") + try: + requirement = ClaimEvidenceRequirement.model_validate( + entry["claim_requirement"] + ) + except (TypeError, ValueError) as exc: + raise ValueError("claim evidence requirement is invalid") from exc + claim = requirement.claim + encoded_artifact = entry["artifact_base64"] + if not isinstance(claim, str) or not isinstance(encoded_artifact, str): + raise ValueError("claim evidence manifest entry type is invalid") + try: + artifact = b64decode(encoded_artifact, validate=True) + except (Base64Error, ValueError) as exc: + raise ValueError("claim evidence manifest artifact is invalid base64") from exc + if b64encode(artifact).decode("ascii") != encoded_artifact: + raise ValueError("claim evidence manifest artifact base64 is not canonical") + receipt = _RECEIPT_ADAPTER.validate_python(entry["receipt"]) + observed_identity = ( + receipt.repository, + receipt.head_sha, + receipt.workflow_ref, + receipt.run_id, + receipt.run_attempt, + ) + expected_identity = ( + expected_repository, + expected_head_sha, + expected_workflow_ref, + expected_run_id, + expected_run_attempt, + ) + if observed_identity != expected_identity: + raise ValueError("claim evidence receipt identity mismatch") + if expected_producers.get(receipt.producer_id) != receipt.evidence_kind: + raise ValueError("claim evidence receipt producer policy mismatch") + if receipt.evidence_kind is not requirement.required_evidence_kind: + raise ValueError("claim evidence requirement kind mismatch") + if receipt.claim_sha256 != sha256_text(claim): + raise ValueError("claim evidence receipt claim digest mismatch") + if receipt.artifact_size != len(artifact): + raise ValueError("claim evidence receipt artifact size mismatch") + if receipt.artifact_sha256 != _sha256_bytes(artifact): + raise ValueError("claim evidence receipt artifact digest mismatch") + if artifact != _canonical_artifact(_artifact_payload(receipt)): + raise ValueError("claim evidence receipt semantic artifact mismatch") + receipts.append(receipt) + artifacts[receipt.receipt_id] = artifact + claims[receipt.receipt_id] = claim + requirements[receipt.receipt_id] = requirement + indexed = index_claim_evidence_receipts(receipts) + return VerifiedClaimEvidenceIndex._from_verified( + receipts=indexed, + artifacts=artifacts, + claims=claims, + requirements=requirements, + ) + + +def _artifact_payload(receipt: ClaimEvidenceReceipt) -> dict[str, object]: + """Reconstruct the exact semantic artifact covered by a trusted receipt.""" + common = { + "schema_version": receipt.schema_version, + "receipt_id": receipt.receipt_id, + "evidence_kind": receipt.evidence_kind.value, + "repository": receipt.repository, + "head_sha": receipt.head_sha, + "workflow_ref": receipt.workflow_ref, + "run_id": receipt.run_id, + "run_attempt": receipt.run_attempt, + "claim_sha256": receipt.claim_sha256, + "producer_id": receipt.producer_id, + "producer_version": receipt.producer_version, + "policy_version": receipt.policy_version, + "issued_at": _utc_text(receipt.issued_at), + "expires_at": _utc_text(receipt.expires_at), + } + if isinstance(receipt, ExecutionClaimReceipt): + return common | { + "argv": list(receipt.argv), + "tool_identity": receipt.tool_identity, + "tool_version": receipt.tool_version, + "exit_code": receipt.exit_code, + "stdout_sha256": receipt.stdout_sha256, + "stderr_sha256": receipt.stderr_sha256, + "isolation_policy": receipt.isolation_policy, + "network_policy": receipt.network_policy, + } + if isinstance(receipt, ResearchClaimReceipt): + return common | { + "source_uri": receipt.source_uri, + "source_revision": receipt.source_revision, + "excerpt_sha256": receipt.excerpt_sha256, + "retrieval_policy": receipt.retrieval_policy, + } + return common | { + "source_path": receipt.source_path, + "source_line": receipt.source_line, + "source_line_sha256": receipt.source_line_sha256, + } + + +def admit_claim_evidence( + receipt_id: str, + *, + trusted_index: VerifiedClaimEvidenceIndex, + claim: str, + admitted_at: datetime, + required_kind: EvidenceKind, +) -> ClaimEvidenceReceipt: + """Resolve a model citation from trusted receipts and admit exact evidence. + + ``trusted_index`` can only be created by exact manifest verification. The + model controls only ``receipt_id`` and exact claim text; it cannot submit or + relabel a receipt payload, artifact, producer identity, or evidence kind. + + Raises: + ValueError: If the cited receipt is absent or any exact identity, + validity, kind, claim, semantic artifact, or artifact bytes differ. + """ + if not isinstance(trusted_index, VerifiedClaimEvidenceIndex): + raise TypeError("claim evidence admission requires a verified manifest index") + receipt = trusted_index.receipts.get(receipt_id) + if receipt is None: + raise ValueError("claim evidence receipt is missing from trusted manifest") + if admitted_at.utcoffset() is None: + raise ValueError("claim evidence admission time must be timezone-aware") + if admitted_at < receipt.issued_at or admitted_at >= receipt.expires_at: + raise ValueError("claim evidence receipt is not valid at admission time") + if receipt.evidence_kind != required_kind: + raise ValueError("claim evidence receipt kind mismatch") + if trusted_index.claims[receipt_id] != claim or receipt.claim_sha256 != sha256_text(claim): + raise ValueError("claim evidence receipt claim digest mismatch") + return receipt diff --git a/reviewer/noema_reviewer/claim_evidence_reference.py b/reviewer/noema_reviewer/claim_evidence_reference.py new file mode 100644 index 000000000..a9cc0011f --- /dev/null +++ b/reviewer/noema_reviewer/claim_evidence_reference.py @@ -0,0 +1,71 @@ +"""Bind model-visible receipt citations to the trusted claim-evidence admission port. + +The model controls only one canonical text reference ending in ``[receipt:]``. +The caller still owns the required evidence kind and the verified out-of-band +manifest index; this module performs no provider, tool, or semantic inference. +""" + +from __future__ import annotations + +import re +from datetime import datetime + +from .claim_evidence import ( + ClaimEvidenceReceipt, + EvidenceKind, + VerifiedClaimEvidenceIndex, + admit_claim_evidence, +) + + +_RECEIPT_REFERENCE_RE = re.compile( + r"^(?P\S(?:[^\r\n]*\S)?) \[receipt:(?P[a-z0-9][a-z0-9._-]{0,79})\]$" +) + + +def parse_claim_evidence_reference(reference: str) -> tuple[str, str]: + """Return exact claim text and receipt ID from one canonical citation. + + The syntax deliberately admits one single-line claim, one ASCII separator, + and one lowercase trailing receipt marker. It does not normalize whitespace, + case, synonyms, tool names, or evidence kinds because those transformations + would change the claim bytes authenticated by the trusted producer. + + Raises: + TypeError: If the model output is not a string. + ValueError: If the reference is missing, ambiguous, or noncanonical. + """ + if not isinstance(reference, str): + raise TypeError("claim evidence reference must be a string") + match = _RECEIPT_REFERENCE_RE.fullmatch(reference) + if match is None: + raise ValueError( + "claim evidence reference must use canonical ' [receipt:]' syntax" + ) + claim = match.group("claim") + if "[receipt:" in claim: + raise ValueError("claim evidence reference must contain exactly one receipt marker") + return claim, match.group("receipt_id") + + +def admit_claim_evidence_reference( + reference: str, + *, + trusted_index: VerifiedClaimEvidenceIndex, + admitted_at: datetime, + required_kind: EvidenceKind, +) -> ClaimEvidenceReceipt: + """Admit one canonical model citation through caller-owned evidence authority. + + ``required_kind`` remains caller policy. The model-visible reference carries + no authoritative evidence-kind field and cannot construct or modify the + verified manifest index. + """ + claim, receipt_id = parse_claim_evidence_reference(reference) + return admit_claim_evidence( + receipt_id, + trusted_index=trusted_index, + claim=claim, + admitted_at=admitted_at, + required_kind=required_kind, + ) diff --git a/reviewer/noema_reviewer/claim_evidence_runtime.py b/reviewer/noema_reviewer/claim_evidence_runtime.py new file mode 100644 index 000000000..05a045f44 --- /dev/null +++ b/reviewer/noema_reviewer/claim_evidence_runtime.py @@ -0,0 +1,252 @@ +"""Connect trusted claim-evidence producers to model admission and publication.""" + +from __future__ import annotations + +import hashlib +from datetime import datetime +from pathlib import Path + +from .claim_evidence import ( + ClaimEvidenceRequirement, + ClaimPublicationAuthority, + EvidenceKind, + ProducedClaimEvidence, + SourceClaimReceipt, + VerifiedClaimEvidenceIndex, + produce_claim_evidence_manifest, + verify_claim_evidence_manifest, +) +from .claim_evidence_reference import admit_claim_evidence_reference +from .manifest import ReviewManifest +from .models import ReviewVerdict +from .source_claim_evidence import produce_source_claim_receipt + + +MAX_SOURCE_RECEIPTS = 40 +MAX_SOURCE_FILE_BYTES = 1_000_000 +MAX_SOURCE_CLAIM_CHARS = 300 +TRUSTED_CLAIM_EVIDENCE_PRODUCERS = { + "changed-source-map": EvidenceKind.SOURCE, + "sandboxed-verify": EvidenceKind.EXECUTION, + "trusted-research-retrieval": EvidenceKind.RESEARCH, +} +ADMITTED_RECOMMENDATION_BY_KIND = { + EvidenceKind.SOURCE: ( + "Review the cited current-head source and apply a source-backed correction." + ), + EvidenceKind.EXECUTION: "Address the cited producer-observed execution result.", + EvidenceKind.RESEARCH: "Address the cited immutable research evidence.", +} + + +def _safe_source_file(source_root: Path, relative_path: str) -> Path | None: + """Return one exact regular checkout file without following symlinks.""" + if ( + not relative_path + or Path(relative_path).is_absolute() + or any(part in {"", ".", ".."} for part in relative_path.split("/")) + ): + return None + root = source_root.resolve(strict=True) + candidate = root.joinpath(*relative_path.split("/")) + try: + if candidate.is_symlink() or not candidate.is_file(): + return None + resolved = candidate.resolve(strict=True) + resolved.relative_to(root) + if resolved.stat().st_size > MAX_SOURCE_FILE_BYTES: + return None + except (OSError, ValueError): + return None + return resolved + + +def produce_current_head_source_manifest( + manifest: ReviewManifest, + *, + source_root: Path, + workflow_ref: str, + run_id: int, + run_attempt: int, + issued_at: datetime, + expires_at: datetime, + max_receipts: int = MAX_SOURCE_RECEIPTS, +) -> bytes: + """Produce bounded exact-line receipts from the already verified head checkout.""" + if max_receipts < 1 or max_receipts > MAX_SOURCE_RECEIPTS: + raise ValueError("source receipt limit must be within the reviewed bound") + entries: list[tuple[ClaimEvidenceRequirement, ProducedClaimEvidence]] = [] + seen_receipts: set[str] = set() + for changed in manifest.changed_files: + source_file = _safe_source_file(source_root, changed.path) + if source_file is None: + continue + try: + source_lines = source_file.read_bytes().splitlines(keepends=True) + except OSError: + continue + for line_number, source_line in enumerate(source_lines, start=1): + try: + claim = source_line.rstrip(b"\r\n").decode("utf-8") + except UnicodeDecodeError: + continue + if ( + not claim.strip() + or len(claim) > MAX_SOURCE_CLAIM_CHARS + or not claim.isprintable() + ): + continue + receipt_hash = hashlib.sha256( + changed.path.encode("utf-8") + + b"\0" + + str(line_number).encode("ascii") + + b"\0" + + source_line + ).hexdigest()[:24] + receipt_id = f"source-{receipt_hash}" + if receipt_id in seen_receipts: + continue + produced = produce_source_claim_receipt( + receipt_id=receipt_id, + repository=manifest.repo, + head_sha=manifest.head_sha, + workflow_ref=workflow_ref, + run_id=run_id, + run_attempt=run_attempt, + claim=claim, + producer_id="changed-source-map", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=issued_at, + expires_at=expires_at, + source_path=changed.path, + source_line=line_number, + source_line_bytes=source_line, + ) + entries.append( + ( + ClaimEvidenceRequirement( + claim=claim, + required_evidence_kind=EvidenceKind.SOURCE, + publication_authority=ClaimPublicationAuthority.CONTEXT, + ), + produced, + ) + ) + seen_receipts.add(receipt_id) + if len(entries) == max_receipts: + return produce_claim_evidence_manifest(entries) + return produce_claim_evidence_manifest(entries) + + +def verify_claim_evidence_file( + path: Path, + *, + expected_manifest_sha256: str, + expected_repository: str, + expected_head_sha: str, + expected_workflow_ref: str, + expected_run_id: int, + expected_run_attempt: int, +) -> VerifiedClaimEvidenceIndex: + """Read and verify one workflow-authenticated manifest using reviewed policy.""" + if path.is_symlink() or not path.is_file(): + raise ValueError("claim evidence manifest path is not a regular file") + return verify_claim_evidence_manifest( + path.read_bytes(), + expected_manifest_sha256=expected_manifest_sha256, + expected_repository=expected_repository, + expected_head_sha=expected_head_sha, + expected_workflow_ref=expected_workflow_ref, + expected_run_id=expected_run_id, + expected_run_attempt=expected_run_attempt, + expected_producers=TRUSTED_CLAIM_EVIDENCE_PRODUCERS, + ) + + +def prompt_claim_evidence_references( + trusted_index: VerifiedClaimEvidenceIndex | None, +) -> list[str]: + """Return only producer-authorized finding claims for model citation.""" + if trusted_index is None: + return [] + return [ + f"{trusted_index.claims[receipt_id]} [receipt:{receipt_id}]" + for receipt_id in sorted(trusted_index.receipts) + if trusted_index.requirements[receipt_id].publication_authority + is ClaimPublicationAuthority.FINDING + ] + + +def admit_review_verdict_evidence( + verdict: ReviewVerdict, + *, + trusted_index: VerifiedClaimEvidenceIndex | None, + admitted_at: datetime, +) -> ReviewVerdict: + """Admit and project model findings before gates or publication. + + The authenticated claim requirement, not the cited receipt or model prose, + owns the required kind and publication authority. Context-only receipts are + rejected before publication. Model summary and recommendation text is then + replaced by producer-kind-specific, non-authoritative action text. + """ + if not verdict.findings and not verdict.is_approval(): + raise ValueError( + "model non-approval requires producer-authenticated findings" + ) + if not verdict.findings: + return verdict + if trusted_index is None: + raise ValueError("model finding evidence requires a verified receipt manifest") + admitted_findings = [] + for finding in verdict.findings: + _, receipt_id = _parse_reference(finding.evidence) + receipt = trusted_index.receipts.get(receipt_id) + if receipt is None: + raise ValueError("claim evidence receipt is missing from trusted manifest") + requirement = trusted_index.requirements[receipt_id] + if ( + requirement.publication_authority + is not ClaimPublicationAuthority.FINDING + ): + raise ValueError( + "claim evidence receipt does not authorize a publishable finding" + ) + if isinstance(receipt, SourceClaimReceipt) and ( + finding.path != receipt.source_path or finding.line != receipt.source_line + ): + raise ValueError("source claim evidence finding coordinate mismatch") + admitted = admit_claim_evidence_reference( + finding.evidence, + trusted_index=trusted_index, + admitted_at=admitted_at, + required_kind=requirement.required_evidence_kind, + ) + admitted_findings.append( + finding.model_copy( + update={"recommendation": _admitted_recommendation(admitted.evidence_kind)} + ) + ) + finding_word = "finding" if len(admitted_findings) == 1 else "findings" + return verdict.model_copy( + update={ + "summary": ( + f"Noema identified {len(admitted_findings)} model {finding_word} backed " + "by producer-authenticated claim evidence." + ), + "findings": admitted_findings, + } + ) + + +def _admitted_recommendation(evidence_kind: EvidenceKind) -> str: + """Return non-authoritative action text for one producer-owned claim kind.""" + return ADMITTED_RECOMMENDATION_BY_KIND[evidence_kind] + + +def _parse_reference(reference: str) -> tuple[str, str]: + """Parse through the canonical reference port without duplicating its grammar.""" + from .claim_evidence_reference import parse_claim_evidence_reference + + return parse_claim_evidence_reference(reference) diff --git a/reviewer/noema_reviewer/cli.py b/reviewer/noema_reviewer/cli.py index e642c6336..b0fc16537 100644 --- a/reviewer/noema_reviewer/cli.py +++ b/reviewer/noema_reviewer/cli.py @@ -14,8 +14,11 @@ import stat import sys from collections.abc import Callable, Sequence +from pathlib import Path from .agent import ReviewAgent, build_agent +from .claim_evidence import VerifiedClaimEvidenceIndex +from .claim_evidence_runtime import verify_claim_evidence_file from .github_io import default_codegraph_runner, fetch_manifest, publish_verdict from .manifest import ReviewManifest from .models import ReviewVerdict, Verdict @@ -287,6 +290,15 @@ def parse_args(argv: list[str]) -> argparse.Namespace: parser.add_argument("--repo", default="", help="Target repository in owner/name form.") parser.add_argument("--pr-number", type=int, default=0, help="Pull request number.") parser.add_argument("--manifest-file", default="", help="Path to a prepared manifest JSON (skips GitHub fetch).") + parser.add_argument( + "--claim-evidence-manifest-file", + default="", + help="Path to the workflow-authenticated claim-evidence manifest.", + ) + parser.add_argument("--claim-evidence-manifest-sha256", default="") + parser.add_argument("--claim-evidence-workflow-ref", default="") + parser.add_argument("--claim-evidence-run-id", type=int, default=0) + parser.add_argument("--claim-evidence-run-attempt", type=int, default=0) parser.add_argument( "--source-root", default="", @@ -303,6 +315,33 @@ def parse_args(argv: list[str]) -> argparse.Namespace: return parser.parse_args(argv) +def _load_claim_evidence_index( + args: argparse.Namespace, + manifest: ReviewManifest, +) -> VerifiedClaimEvidenceIndex | None: + """Load one all-or-nothing authenticated producer handoff for this review.""" + values = ( + args.claim_evidence_manifest_file, + args.claim_evidence_manifest_sha256, + args.claim_evidence_workflow_ref, + args.claim_evidence_run_id, + args.claim_evidence_run_attempt, + ) + if not any(values): + return None + if not all(values): + raise ValueError("claim evidence handoff identity must be complete") + return verify_claim_evidence_file( + Path(args.claim_evidence_manifest_file), + expected_manifest_sha256=args.claim_evidence_manifest_sha256, + expected_repository=manifest.repo, + expected_head_sha=manifest.head_sha, + expected_workflow_ref=args.claim_evidence_workflow_ref, + expected_run_id=args.claim_evidence_run_id, + expected_run_attempt=args.claim_evidence_run_attempt, + ) + + def run_review( args: argparse.Namespace, *, @@ -331,7 +370,11 @@ def run_review( f"checks={len(manifest.check_conclusions)} comments={len(manifest.review_comments)}", file=sys.stderr, ) - agent = resolved_factory() + trusted_index = _load_claim_evidence_index(args, manifest) + if agent_factory is None and trusted_index is not None: + agent = build_agent(claim_evidence_index=trusted_index) + else: + agent = resolved_factory() verdict = agent.review(manifest, strict=args.strict) serialized = verdict.model_dump_json(indent=2) diff --git a/reviewer/noema_reviewer/sandboxed_verify_claim_evidence.py b/reviewer/noema_reviewer/sandboxed_verify_claim_evidence.py new file mode 100644 index 000000000..0c5ce141e --- /dev/null +++ b/reviewer/noema_reviewer/sandboxed_verify_claim_evidence.py @@ -0,0 +1,126 @@ +"""Adapt the reviewed central ``sandboxed_verify`` result into execution evidence. + +The central helper is a foreign-owner execution producer. Noema does not copy or +reimplement it; this adapter accepts only a reviewed immutable helper revision, +its machine-readable result marker, and separately captured command stdout and +stderr. The legacy marker by itself therefore remains insufficient authority. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field, StrictInt, model_validator + +from .claim_evidence import ProducedClaimEvidence, produce_execution_claim_receipt + + +_RESULT_PREFIX = "SANDBOXED_VERIFY_RESULT " +_TOOL_IDENTITY = "ContextualWisdomLab/.github:scripts/ci/sandboxed_verify.py" +_NonEmptyText = Annotated[str, Field(min_length=1)] + + +@dataclass(frozen=True) +class _SandboxedVerifyPolicy: + """Reviewed semantics for one immutable central helper revision.""" + + isolation_policy: str + + +_SUPPORTED_POLICIES = { + "c9052e607e5f3cc76e73207e7786b21500721b79": _SandboxedVerifyPolicy( + isolation_policy="workspace-copy+scrubbed-env;os-process-isolation=none" + ) +} + + +class _SandboxedVerifyMarker(BaseModel): + """Exact machine-readable marker shape emitted by the reviewed helper.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + allowed_env: tuple[str, ...] + command: tuple[_NonEmptyText, ...] = Field(min_length=1) + cwd: _NonEmptyText + elapsed_seconds: float = Field(ge=0, allow_inf_nan=False) + evidence_note: str + exit_code: StrictInt + network: Literal["default", "required", "not-required"] + sandbox: _NonEmptyText + sandboxed: Literal[True] + + @model_validator(mode="after") + def require_fully_receipted_capabilities(self) -> "_SandboxedVerifyMarker": + """Reject environment capabilities until their influence is receipt-bound.""" + if self.allowed_env: + raise ValueError( + "sandboxed_verify allowed_env capabilities are not yet receipt-bound" + ) + return self + + +def _parse_sandboxed_verify_marker(marker: str) -> _SandboxedVerifyMarker: + """Parse one exact current helper marker and reject non-marker text.""" + if not marker.startswith(_RESULT_PREFIX): + raise ValueError("sandboxed_verify result marker is missing") + return _SandboxedVerifyMarker.model_validate_json(marker[len(_RESULT_PREFIX) :]) + + +def produce_sandboxed_verify_execution_claim_receipt( + *, + receipt_id: str, + repository: str, + head_sha: str, + workflow_ref: str, + run_id: int, + run_attempt: int, + claim: str, + policy_version: str, + issued_at: datetime, + expires_at: datetime, + marker: str, + command_stdout: bytes | None, + command_stderr: bytes | None, + tool_version: str, +) -> ProducedClaimEvidence: + """Issue one execution receipt from a reviewed helper result and exact streams. + + ``tool_version`` is the immutable commit of the central ``.github`` helper. + The current reviewed helper copies the repository into a temporary workspace + and scrubs its environment but launches the command as an ordinary host + subprocess; its ``--network`` value is explicitly metadata rather than an + enforced network control. Those limitations are recorded in the receipt + instead of being promoted to stronger isolation claims. + """ + if command_stdout is None: + raise ValueError("sandboxed_verify execution evidence requires stdout capture") + if command_stderr is None: + raise ValueError("sandboxed_verify execution evidence requires stderr capture") + policy = _SUPPORTED_POLICIES.get(tool_version) + if policy is None: + raise ValueError("unsupported sandboxed_verify version requires policy review") + parsed = _parse_sandboxed_verify_marker(marker) + return produce_execution_claim_receipt( + receipt_id=receipt_id, + repository=repository, + head_sha=head_sha, + workflow_ref=workflow_ref, + run_id=run_id, + run_attempt=run_attempt, + claim=claim, + producer_id="sandboxed-verify", + producer_version=tool_version, + policy_version=policy_version, + issued_at=issued_at, + expires_at=expires_at, + argv=parsed.command, + tool_identity=_TOOL_IDENTITY, + tool_version=tool_version, + exit_code=parsed.exit_code, + stdout=command_stdout, + stderr=command_stderr, + isolation_policy=policy.isolation_policy, + network_policy=f"declared:{parsed.network};enforced=false", + ) diff --git a/reviewer/noema_reviewer/source_claim_evidence.py b/reviewer/noema_reviewer/source_claim_evidence.py new file mode 100644 index 000000000..fcd02b8d1 --- /dev/null +++ b/reviewer/noema_reviewer/source_claim_evidence.py @@ -0,0 +1,83 @@ +"""Canonical producer for exact current-head source evidence.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime + +from .claim_evidence import ( + EvidenceKind, + ProducedClaimEvidence, + SourceClaimReceipt, + _canonical_artifact, + _common_artifact_payload, + _receipt_identity, + _sha256_bytes, +) + + +def _exact_source_line_claim(source_line_bytes: bytes) -> str: + """Decode exactly one source line while preserving its semantic bytes.""" + if source_line_bytes.endswith(b"\r\n"): + line_body = source_line_bytes[:-2] + elif source_line_bytes.endswith((b"\n", b"\r")): + line_body = source_line_bytes[:-1] + else: + line_body = source_line_bytes + if b"\n" in line_body or b"\r" in line_body: + raise ValueError("source claim evidence requires exactly one source line") + try: + return line_body.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError("source claim evidence line must be valid UTF-8") from exc + + +def produce_source_claim_receipt( + *, + receipt_id: str, + repository: str, + head_sha: str, + workflow_ref: str, + run_id: int, + run_attempt: int, + claim: str, + producer_id: str, + producer_version: str, + policy_version: str, + issued_at: datetime, + expires_at: datetime, + source_path: str, + source_line: int, + source_line_bytes: bytes, +) -> ProducedClaimEvidence: + """Seal exact current-head source-line bytes into one canonical receipt artifact.""" + if _exact_source_line_claim(source_line_bytes) != claim: + raise ValueError("source claim must equal exact source line") + payload: Mapping[str, object] = _common_artifact_payload( + receipt_id=receipt_id, + evidence_kind=EvidenceKind.SOURCE, + repository=repository, + head_sha=head_sha, + workflow_ref=workflow_ref, + run_id=run_id, + run_attempt=run_attempt, + claim=claim, + producer_id=producer_id, + producer_version=producer_version, + policy_version=policy_version, + issued_at=issued_at, + expires_at=expires_at, + ) | { + "source_path": source_path, + "source_line": source_line, + "source_line_sha256": _sha256_bytes(source_line_bytes), + } + artifact = _canonical_artifact(payload) + receipt = SourceClaimReceipt( + **_receipt_identity(payload, artifact), + evidence_kind=EvidenceKind.SOURCE, + source_path=source_path, + source_line=source_line, + source_line_sha256=payload["source_line_sha256"], + ) + return ProducedClaimEvidence(receipt=receipt, artifact=artifact) diff --git a/reviewer/tests/test_claim_evidence_publication_boundary.py b/reviewer/tests/test_claim_evidence_publication_boundary.py new file mode 100644 index 000000000..21b425cb8 --- /dev/null +++ b/reviewer/tests/test_claim_evidence_publication_boundary.py @@ -0,0 +1,797 @@ +"""Integration RED for trusted claim evidence at the real review boundary.""" + +from __future__ import annotations + +import hashlib +import io +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest +from pydantic_ai.models.test import TestModel + +from noema_reviewer import cli +from noema_reviewer.agent import PydanticAIReviewAgent +from noema_reviewer.claim_evidence import ( + ClaimEvidenceRequirement, + ClaimPublicationAuthority, + EvidenceKind, + produce_claim_evidence_manifest, + produce_execution_claim_receipt, + verify_claim_evidence_manifest, +) +from noema_reviewer.claim_evidence_runtime import ( + MAX_SOURCE_CLAIM_CHARS, + MAX_SOURCE_FILE_BYTES, + MAX_SOURCE_RECEIPTS, + admit_review_verdict_evidence, + prompt_claim_evidence_references, + produce_current_head_source_manifest, + verify_claim_evidence_file, +) +from noema_reviewer.manifest import ChangedFile, CheckConclusion, ReviewManifest +from noema_reviewer.models import ( + EvidenceType, + Finding, + Priority, + ReviewVerdict, + Severity, + Verdict, +) +from noema_reviewer.source_claim_evidence import produce_source_claim_receipt + + +CLAIM = ( + "--locked is not a valid invocation: --locked is not accepted by the " + "generate-lockfile subcommand. This will always fail, breaking the workflow." +) +HEAD = "a" * 40 +WORKFLOW = "ContextualWisdomLab/noema/.github/workflows/central-review.yml@" + "b" * 40 +ISSUED = datetime(2026, 9, 7, tzinfo=timezone.utc) +EXPIRES = ISSUED + timedelta(days=1) + + +def _finding( + evidence: str, + *, + path: str = ".github/workflows/ci.yml", + line: int | None = 1, + recommendation: str = "Remove the unsupported flag.", +) -> Finding: + """Return one actionable finding compatible with the inherited wire contract.""" + return Finding( + severity=Severity.HIGH, + priority=Priority.P1, + path=path, + line=line, + evidence=evidence, + evidence_type=EvidenceType.NEARBY_IMPLEMENTATION, + observable_impact="The review would block a valid workflow command.", + trigger="Review the changed workflow command.", + recommendation=recommendation, + regression_command=( + "pytest -q reviewer/tests/test_claim_evidence_publication_boundary.py" + ), + ) + + +def _review_manifest() -> ReviewManifest: + """Return a complete bounded manifest for the actual review path.""" + return ReviewManifest( + repo="ContextualWisdomLab/ConceptWeave", + pr_number=35, + head_sha=HEAD, + diff="diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml", + changed_files=[ + ChangedFile( + path=".github/workflows/ci.yml", + content="run: cargo generate-lockfile --locked\n", + ) + ], + check_conclusions=[CheckConclusion(name="ci", conclusion="success")], + codegraph_status="## codegraph explore\nci workflow command", + ) + + +def _execution_manifest(tmp_path: Path) -> tuple[Path, str]: + """Write one canonical execution receipt manifest and return its digest.""" + produced = produce_execution_claim_receipt( + receipt_id="execution-1", + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=CLAIM, + producer_id="sandboxed-verify", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + argv=["cargo", "generate-lockfile", "--locked"], + tool_identity="cargo", + tool_version="1.90.0", + exit_code=1, + stdout=b"", + stderr=b"unsupported\n", + isolation_policy="sandboxed-verify-v1", + network_policy="disabled", + ) + manifest = produce_claim_evidence_manifest( + [ + ( + ClaimEvidenceRequirement( + claim=CLAIM, + required_evidence_kind=EvidenceKind.EXECUTION, + publication_authority=ClaimPublicationAuthority.FINDING, + ), + produced, + ) + ] + ) + path = tmp_path / "claim-evidence.json" + path.write_bytes(manifest) + return path, hashlib.sha256(manifest).hexdigest() + + +def _source_manifest(authority: ClaimPublicationAuthority) -> bytes: + """Return one source receipt with caller-owned publication authority.""" + claim = "run: cargo generate-lockfile --locked" + produced = produce_source_claim_receipt( + receipt_id="source-finding-1", + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=claim, + producer_id="changed-source-map", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + source_path=".github/workflows/ci.yml", + source_line=1, + source_line_bytes=b"run: cargo generate-lockfile --locked\n", + ) + return produce_claim_evidence_manifest( + [ + ( + ClaimEvidenceRequirement( + claim=claim, + required_evidence_kind=EvidenceKind.SOURCE, + publication_authority=authority, + ), + produced, + ) + ] + ) + + +def _args(manifest_path: Path, **extra: object): + """Return parsed production CLI arguments.""" + argv = ["--manifest-file", str(manifest_path)] + for key, value in extra.items(): + flag = "--" + key.replace("_", "-") + if value is True: + argv.append(flag) + elif value is not False: + argv.extend([flag, str(value)]) + return cli.parse_args(argv) + + +def _model_agent(evidence: str) -> PydanticAIReviewAgent: + """Return the production driver around a deterministic offline model.""" + finding = _finding(evidence) + return PydanticAIReviewAgent( + TestModel( + custom_output_args=ReviewVerdict( + verdict=Verdict.REQUEST_CHANGES, + summary="The workflow command is invalid.", + findings=[finding], + ).model_dump(mode="json") + ) + ) + + +def test_trusted_manifest_reaches_agent_gate_before_publication( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A producer receipt is verified, shown to the model, admitted, then published.""" + review_manifest = _review_manifest() + review_path = tmp_path / "review.json" + review_path.write_text(review_manifest.model_dump_json(), encoding="utf-8") + evidence_path, digest = _execution_manifest(tmp_path) + published: list[ReviewVerdict] = [] + + def factory(*, claim_evidence_index): + agent = _model_agent(f"{CLAIM} [receipt:execution-1]") + agent.bind_claim_evidence(claim_evidence_index, admitted_at=lambda: ISSUED) + return agent + + monkeypatch.setattr(cli, "build_agent", factory) + code = cli.run_review( + _args( + review_path, + claim_evidence_manifest_file=evidence_path, + claim_evidence_manifest_sha256=digest, + claim_evidence_workflow_ref=WORKFLOW, + claim_evidence_run_id=12, + claim_evidence_run_attempt=1, + ), + publisher=lambda _repo, _pr, verdict, _head, _source: ( + published.append(verdict) or "REQUEST_CHANGES" + ), + out=io.StringIO(), + ) + assert code == 2 + assert len(published) == 0 + + cli.run_review( + _args( + review_path, + publish=True, + claim_evidence_manifest_file=evidence_path, + claim_evidence_manifest_sha256=digest, + claim_evidence_workflow_ref=WORKFLOW, + claim_evidence_run_id=12, + claim_evidence_run_attempt=1, + ), + publisher=lambda _repo, _pr, verdict, _head, _source: ( + published.append(verdict) or "REQUEST_CHANGES" + ), + out=io.StringIO(), + ) + assert published and published[-1].findings[0].evidence.endswith( + "[receipt:execution-1]" + ) + + +def test_free_text_model_evidence_never_reaches_publisher( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Free prose cannot become execution evidence without a verified receipt ID.""" + review_path = tmp_path / "review.json" + review_path.write_text(_review_manifest().model_dump_json(), encoding="utf-8") + evidence_path, digest = _execution_manifest(tmp_path) + monkeypatch.setattr( + cli, + "build_agent", + lambda *, claim_evidence_index: _model_agent( + "Runtime behavior confirms the command is unsupported." + ).bind_claim_evidence(claim_evidence_index, admitted_at=lambda: ISSUED), + ) + with pytest.raises(ValueError, match="receipt"): + cli.run_review( + _args( + review_path, + publish=True, + claim_evidence_manifest_file=evidence_path, + claim_evidence_manifest_sha256=digest, + claim_evidence_workflow_ref=WORKFLOW, + claim_evidence_run_id=12, + claim_evidence_run_attempt=1, + ), + publisher=lambda *_args: pytest.fail("unadmitted evidence must not publish"), + out=io.StringIO(), + ) + + +@pytest.mark.parametrize("verdict", [Verdict.REQUEST_CHANGES, Verdict.BLOCKED]) +def test_empty_nonapproval_cannot_bypass_receipt_admission( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + verdict: Verdict, +) -> None: + """A finding-free model verdict cannot publish a vacuous blocking review.""" + review_path = tmp_path / "review.json" + review_path.write_text(_review_manifest().model_dump_json(), encoding="utf-8") + evidence_path, digest = _execution_manifest(tmp_path) + + def factory(*, claim_evidence_index): + agent = PydanticAIReviewAgent( + TestModel( + custom_output_args=ReviewVerdict( + verdict=verdict, + summary="The change must be rejected.", + blocked_reasons=["The command is invalid."] + if verdict is Verdict.BLOCKED + else [], + ).model_dump(mode="json") + ) + ) + return agent.bind_claim_evidence( + claim_evidence_index, + admitted_at=lambda: ISSUED, + ) + + monkeypatch.setattr(cli, "build_agent", factory) + with pytest.raises(ValueError, match="requires producer-authenticated findings"): + cli.run_review( + _args( + review_path, + publish=True, + claim_evidence_manifest_file=evidence_path, + claim_evidence_manifest_sha256=digest, + claim_evidence_workflow_ref=WORKFLOW, + claim_evidence_run_id=12, + claim_evidence_run_attempt=1, + ), + publisher=lambda *_args: pytest.fail( + "finding-free non-approval must not publish" + ), + out=io.StringIO(), + ) + + +def test_current_head_source_producer_populates_verified_prompt_receipts( + tmp_path: Path, +) -> None: + """Raw source context is verified but withheld from finding-authority prompts.""" + source = tmp_path / ".github/workflows/ci.yml" + source.parent.mkdir(parents=True) + source.write_text("run: cargo generate-lockfile --locked\n", encoding="utf-8") + manifest = produce_current_head_source_manifest( + _review_manifest(), + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + index = verify_claim_evidence_manifest( + manifest, + expected_manifest_sha256=hashlib.sha256(manifest).hexdigest(), + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + expected_producers={"changed-source-map": EvidenceKind.SOURCE}, + ) + agent = _model_agent("run: cargo generate-lockfile --locked [receipt:missing]") + agent.bind_claim_evidence(index, admitted_at=lambda: ISSUED) + assert "[receipt:source-" not in agent.prompt_for(_review_manifest()) + requirement = next(iter(index.requirements.values())) + assert requirement.publication_authority is ClaimPublicationAuthority.CONTEXT + + receipt_id = next(iter(index.receipts)) + mismatched = ReviewVerdict( + verdict=Verdict.REQUEST_CHANGES, + summary="wrong coordinate", + findings=[ + _finding( + f"run: cargo generate-lockfile --locked [receipt:{receipt_id}]", + path="another-file", + recommendation="fix", + ) + ], + ) + with pytest.raises(ValueError, match="does not authorize"): + admit_review_verdict_evidence( + mismatched, + trusted_index=index, + admitted_at=ISSUED, + ) + + +def test_producer_authorized_source_finding_reaches_real_publisher( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A trusted source finding remains publishable without blanket suppression.""" + review_manifest = _review_manifest() + review_path = tmp_path / "review.json" + review_path.write_text(review_manifest.model_dump_json(), encoding="utf-8") + claim = "run: cargo generate-lockfile --locked" + manifest = _source_manifest(ClaimPublicationAuthority.FINDING) + evidence_path = tmp_path / "source-finding-evidence.json" + evidence_path.write_bytes(manifest) + published: list[ReviewVerdict] = [] + + def factory(*, claim_evidence_index): + agent = _model_agent(f"{claim} [receipt:source-finding-1]") + return agent.bind_claim_evidence( + claim_evidence_index, + admitted_at=lambda: ISSUED, + ) + + monkeypatch.setattr(cli, "build_agent", factory) + code = cli.run_review( + _args( + review_path, + publish=True, + claim_evidence_manifest_file=evidence_path, + claim_evidence_manifest_sha256=hashlib.sha256(manifest).hexdigest(), + claim_evidence_workflow_ref=WORKFLOW, + claim_evidence_run_id=12, + claim_evidence_run_attempt=1, + ), + publisher=lambda _repo, _pr, verdict, _head, _source: ( + published.append(verdict) or "REQUEST_CHANGES" + ), + out=io.StringIO(), + ) + + assert code == 2 + assert len(published) == 1 + assert published[0].findings[0].evidence.endswith( + "[receipt:source-finding-1]" + ) + + +@pytest.mark.parametrize( + ("path", "line"), + [("other.yml", 1), (".github/workflows/ci.yml", 2)], +) +def test_authorized_source_finding_requires_exact_coordinates( + path: str, + line: int, +) -> None: + """Finding authority never weakens source path/line identity.""" + manifest = _source_manifest(ClaimPublicationAuthority.FINDING) + index = verify_claim_evidence_manifest( + manifest, + expected_manifest_sha256=hashlib.sha256(manifest).hexdigest(), + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + expected_producers={"changed-source-map": EvidenceKind.SOURCE}, + ) + verdict = ReviewVerdict( + verdict=Verdict.REQUEST_CHANGES, + summary="wrong coordinate", + findings=[ + _finding( + "run: cargo generate-lockfile --locked " + "[receipt:source-finding-1]", + path=path, + line=line, + ) + ], + ) + with pytest.raises(ValueError, match="coordinate mismatch"): + admit_review_verdict_evidence( + verdict, + trusted_index=index, + admitted_at=ISSUED, + ) + + +def test_source_receipt_cannot_publish_unreceipted_runtime_claims( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Source existence alone cannot publish a blocking runtime finding.""" + review_manifest = _review_manifest() + review_path = tmp_path / "review.json" + review_path.write_text(review_manifest.model_dump_json(), encoding="utf-8") + source_root = tmp_path / "source" + source = source_root / ".github/workflows/ci.yml" + source.parent.mkdir(parents=True) + source.write_text("run: cargo generate-lockfile --locked\n", encoding="utf-8") + manifest = produce_current_head_source_manifest( + review_manifest, + source_root=source_root, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + evidence_path = tmp_path / "source-evidence.json" + evidence_path.write_bytes(manifest) + published: list[ReviewVerdict] = [] + + def factory(*, claim_evidence_index): + receipt_id = next(iter(claim_evidence_index.receipts)) + finding = _finding( + ( + "run: cargo generate-lockfile --locked " + f"[receipt:{receipt_id}]" + ), + recommendation="Remove the unsupported flag because the command fails.", + ) + agent = PydanticAIReviewAgent( + TestModel( + custom_output_args=ReviewVerdict( + verdict=Verdict.REQUEST_CHANGES, + summary="The generate-lockfile command rejects --locked.", + findings=[finding], + ).model_dump(mode="json") + ) + ) + return agent.bind_claim_evidence( + claim_evidence_index, + admitted_at=lambda: ISSUED, + ) + + monkeypatch.setattr(cli, "build_agent", factory) + with pytest.raises(ValueError, match="does not authorize a publishable finding"): + cli.run_review( + _args( + review_path, + publish=True, + claim_evidence_manifest_file=evidence_path, + claim_evidence_manifest_sha256=hashlib.sha256(manifest).hexdigest(), + claim_evidence_workflow_ref=WORKFLOW, + claim_evidence_run_id=12, + claim_evidence_run_attempt=1, + ), + publisher=lambda _repo, _pr, verdict, _head, _source: ( + published.append(verdict) or "REQUEST_CHANGES" + ), + out=io.StringIO(), + ) + + assert published == [] + + +def test_source_manifest_bounds_and_unsafe_paths_fail_closed(tmp_path: Path) -> None: + """The source producer skips unsafe files and rejects unreviewed cardinality.""" + unsafe = _review_manifest().model_copy( + update={ + "changed_files": [ + ChangedFile(path="", content=""), + ChangedFile(path="/etc/passwd", content=""), + ChangedFile(path="../escape", content=""), + ChangedFile(path="missing", content=""), + ] + } + ) + for limit in (0, MAX_SOURCE_RECEIPTS + 1): + with pytest.raises(ValueError, match="limit"): + produce_current_head_source_manifest( + unsafe, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + max_receipts=limit, + ) + + manifest = produce_current_head_source_manifest( + unsafe, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + assert b'"entries":[]' in manifest + + +def test_source_manifest_skips_unsafe_bytes_and_stops_at_bound( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Only bounded printable UTF-8 source lines become producer receipts.""" + source = tmp_path / "source.txt" + source.write_bytes( + b"\n" + + b"x" * (MAX_SOURCE_CLAIM_CHARS + 1) + + b"\ninvalid-utf8:\xff\nnot-printable:\x00\nfirst\nsecond\n" + ) + manifest_model = _review_manifest().model_copy( + update={ + "changed_files": [ + ChangedFile(path="source.txt", content=""), + ChangedFile(path="source.txt", content=""), + ] + } + ) + bounded = produce_current_head_source_manifest( + manifest_model, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + max_receipts=1, + ) + assert bounded.count(b'"receipt_id"') == 1 + + original_read_bytes = Path.read_bytes + + def failed_read(path: Path) -> bytes: + if path.name == "source.txt": + raise OSError("read failed") + return original_read_bytes(path) + + monkeypatch.setattr(Path, "read_bytes", failed_read) + skipped = produce_current_head_source_manifest( + manifest_model, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + assert b'"entries":[]' in skipped + + +def test_source_manifest_deduplicates_and_survives_resolution_race( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Duplicate paths emit once and a disappearing file is skipped safely.""" + source = tmp_path / "source.txt" + source.write_text("one\n", encoding="utf-8") + model = _review_manifest().model_copy( + update={ + "changed_files": [ + ChangedFile(path="source.txt", content=""), + ChangedFile(path="source.txt", content=""), + ] + } + ) + deduplicated = produce_current_head_source_manifest( + model, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + assert deduplicated.count(b'"receipt_id"') == 1 + + original_resolve = Path.resolve + + def failed_candidate_resolve(path: Path, *, strict: bool = False) -> Path: + if path.name == "source.txt": + raise OSError("file disappeared") + return original_resolve(path, strict=strict) + + monkeypatch.setattr(Path, "resolve", failed_candidate_resolve) + skipped = produce_current_head_source_manifest( + model, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + assert b'"entries":[]' in skipped + + +def test_source_manifest_rejects_symlink_and_oversized_file(tmp_path: Path) -> None: + """Symlinks and oversized files never enter the trusted source manifest.""" + real = tmp_path / "real.txt" + real.write_text("line\n", encoding="utf-8") + linked = tmp_path / "linked.txt" + linked.symlink_to(real) + oversized = tmp_path / "large.txt" + oversized.write_bytes(b"x" * (MAX_SOURCE_FILE_BYTES + 1)) + model = _review_manifest().model_copy( + update={ + "changed_files": [ + ChangedFile(path="linked.txt", content=""), + ChangedFile(path="large.txt", content=""), + ] + } + ) + produced = produce_current_head_source_manifest( + model, + source_root=tmp_path, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + issued_at=ISSUED, + expires_at=EXPIRES, + ) + assert b'"entries":[]' in produced + + +def test_runtime_admission_rejects_missing_index_and_receipt(tmp_path: Path) -> None: + """No finding or unknown ID can borrow authority from free-form evidence.""" + empty = ReviewVerdict(verdict=Verdict.APPROVE, summary="ok") + assert admit_review_verdict_evidence( + empty, + trusted_index=None, + admitted_at=ISSUED, + ) == empty + finding = ReviewVerdict( + verdict=Verdict.REQUEST_CHANGES, + summary="blocked", + findings=[ + _finding( + f"{CLAIM} [receipt:missing]", + path="x", + line=None, + recommendation="fix", + ) + ], + ) + with pytest.raises(ValueError, match="verified receipt"): + admit_review_verdict_evidence( + finding, + trusted_index=None, + admitted_at=ISSUED, + ) + evidence_path, digest = _execution_manifest(tmp_path) + index = verify_claim_evidence_file( + evidence_path, + expected_manifest_sha256=digest, + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + ) + with pytest.raises(ValueError, match="missing from trusted manifest"): + admit_review_verdict_evidence( + finding, + trusted_index=index, + admitted_at=ISSUED, + ) + assert prompt_claim_evidence_references(None) == [] + + +def test_agent_evidence_binding_is_single_assignment(tmp_path: Path) -> None: + """A caller cannot swap the verified index after the review agent is bound.""" + evidence_path, digest = _execution_manifest(tmp_path) + index = verify_claim_evidence_file( + evidence_path, + expected_manifest_sha256=digest, + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + ) + agent = _model_agent(f"{CLAIM} [receipt:execution-1]") + assert agent.bind_claim_evidence(index) is agent + with pytest.raises(ValueError, match="already bound"): + agent.bind_claim_evidence(index) + + +def test_cli_rejects_partial_claim_evidence_identity(tmp_path: Path) -> None: + """A manifest path without its authenticated run identity is never loaded.""" + review_path = tmp_path / "review.json" + review_path.write_text(_review_manifest().model_dump_json(), encoding="utf-8") + with pytest.raises(ValueError, match="identity must be complete"): + cli.run_review( + _args(review_path, claim_evidence_manifest_file="partial.json"), + agent_factory=lambda: pytest.fail("partial handoff must fail before model"), + out=io.StringIO(), + ) + + +def test_manifest_file_requires_regular_non_symlink(tmp_path: Path) -> None: + """The workflow loader does not follow a model-controlled manifest alias.""" + missing = tmp_path / "missing.json" + with pytest.raises(ValueError, match="regular file"): + verify_claim_evidence_file( + missing, + expected_manifest_sha256="0" * 64, + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + ) + target, _ = _execution_manifest(tmp_path) + alias = tmp_path / "alias.json" + alias.symlink_to(target) + with pytest.raises(ValueError, match="regular file"): + verify_claim_evidence_file( + alias, + expected_manifest_sha256="0" * 64, + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + ) diff --git a/reviewer/tests/test_claim_evidence_receipt.py b/reviewer/tests/test_claim_evidence_receipt.py new file mode 100644 index 000000000..59e128150 --- /dev/null +++ b/reviewer/tests/test_claim_evidence_receipt.py @@ -0,0 +1,572 @@ +"""Tests for authenticated exact-claim receipt production and admission.""" + +from __future__ import annotations + +import hashlib +import json +from base64 import b64encode +from datetime import datetime, timedelta, timezone + +import pytest +from pydantic import ValidationError + +from noema_reviewer.claim_evidence import ( + ClaimEvidenceRequirement, + ClaimPublicationAuthority, + EvidenceKind, + ExecutionClaimReceipt, + ProducedClaimEvidence, + ResearchClaimReceipt, + SourceClaimReceipt, + VerifiedClaimEvidenceIndex, + admit_claim_evidence, + index_claim_evidence_receipts, + produce_claim_evidence_manifest, + produce_execution_claim_receipt, + produce_research_claim_receipt, + sha256_text, + verify_claim_evidence_manifest, +) + + +CLAIM = ( + "--locked is not a valid invocation: --locked is not accepted by the " + "generate-lockfile subcommand. This will always fail, breaking the workflow." +) +HEAD = "a" * 40 +WORKFLOW = "ContextualWisdomLab/noema/.github/workflows/central-review.yml@" + "b" * 40 +ISSUED = datetime(2026, 9, 7, tzinfo=timezone.utc) +EXPIRES = ISSUED + timedelta(hours=1) + + +def _requirement( + claim: str, + bundle: ProducedClaimEvidence, + *, + authority: ClaimPublicationAuthority = ClaimPublicationAuthority.FINDING, +) -> ClaimEvidenceRequirement: + """Return caller-owned claim policy independent from one producer receipt.""" + return ClaimEvidenceRequirement( + claim=claim, + required_evidence_kind=bundle.receipt.evidence_kind, + publication_authority=authority, + ) + + +def _execution_bundle() -> ProducedClaimEvidence: + """Produce one canonical execution artifact and receipt.""" + return produce_execution_claim_receipt( + receipt_id="execution-1", + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=CLAIM, + producer_id="sandboxed-verify", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + argv=["cargo", "generate-lockfile", "--locked"], + tool_identity="cargo", + tool_version="1.90.0", + exit_code=1, + stdout=b"", + stderr=b"unsupported\n", + isolation_policy="sandboxed-verify-v1", + network_policy="disabled", + ) + + +def _research_bundle() -> ProducedClaimEvidence: + """Produce one content-addressed research artifact and receipt.""" + return produce_research_claim_receipt( + receipt_id="research-1", + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=CLAIM, + producer_id="pinned-research", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + source_uri="https://doc.rust-lang.org/cargo/commands/cargo-generate-lockfile.html", + retrieved_content=b"immutable documentation bytes\n", + excerpt=b"generate-lockfile accepts --offline", + retrieval_policy="originweave-pinned-v1", + ) + + +def _source_bundle() -> ProducedClaimEvidence: + """Return a canonical source receipt fixture for the third union branch.""" + payload = { + "schema_version": 1, + "receipt_id": "source-1", + "evidence_kind": "source", + "repository": "ContextualWisdomLab/ConceptWeave", + "head_sha": HEAD, + "workflow_ref": WORKFLOW, + "run_id": 12, + "run_attempt": 1, + "claim_sha256": sha256_text(CLAIM), + "producer_id": "changed-source-map", + "producer_version": "v1", + "policy_version": "review-evidence-v1", + "issued_at": "2026-09-07T00:00:00Z", + "expires_at": "2026-09-07T01:00:00Z", + "source_path": ".github/workflows/ci.yml", + "source_line": 42, + "source_line_sha256": "3" * 64, + } + artifact = (json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n").encode() + receipt = SourceClaimReceipt( + **payload, + artifact_sha256=hashlib.sha256(artifact).hexdigest(), + artifact_size=len(artifact), + ) + return ProducedClaimEvidence(receipt=receipt, artifact=artifact) + + +def _producer_policy(bundle: ProducedClaimEvidence) -> dict[str, EvidenceKind]: + """Return reviewed producer policy for one focused fixture.""" + return {bundle.receipt.producer_id: bundle.receipt.evidence_kind} + + +def _verify( + bundle: ProducedClaimEvidence | None = None, + *, + manifest: bytes | None = None, + **overrides: object, +) -> VerifiedClaimEvidenceIndex: + """Verify one producer manifest against caller-owned exact identity.""" + selected = bundle or _execution_bundle() + body = manifest or produce_claim_evidence_manifest( + [(_requirement(CLAIM, selected), selected)] + ) + values = { + "manifest": body, + "expected_manifest_sha256": hashlib.sha256(body).hexdigest(), + "expected_repository": "ContextualWisdomLab/ConceptWeave", + "expected_head_sha": HEAD, + "expected_workflow_ref": WORKFLOW, + "expected_run_id": 12, + "expected_run_attempt": 1, + "expected_producers": _producer_policy(selected), + } + values.update(overrides) + return verify_claim_evidence_manifest(**values) + + +def _admit( + bundle: ProducedClaimEvidence | None = None, + **overrides: object, +) -> ExecutionClaimReceipt | ResearchClaimReceipt | SourceClaimReceipt: + """Admit one model citation using only an authenticated manifest index.""" + selected = bundle or _execution_bundle() + values = { + "receipt_id": selected.receipt.receipt_id, + "trusted_index": _verify(selected), + "claim": CLAIM, + "admitted_at": ISSUED + timedelta(minutes=1), + "required_kind": selected.receipt.evidence_kind, + } + values.update(overrides) + return admit_claim_evidence(**values) + + +def _manifest_with_receipt_mutation( + bundle: ProducedClaimEvidence, + mutation: dict[str, object], +) -> bytes: + """Return canonical envelope bytes with only receipt fields substituted.""" + payload = json.loads( + produce_claim_evidence_manifest([(_requirement(CLAIM, bundle), bundle)]) + ) + payload["entries"][0]["receipt"].update(mutation) + return (json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n").encode() + + +def test_execution_producer_seals_every_semantic_field() -> None: + """Canonical execution artifacts cover identity, command, result, and policy.""" + bundle = _execution_bundle() + payload = json.loads(bundle.artifact) + assert isinstance(bundle.receipt, ExecutionClaimReceipt) + assert payload["claim_sha256"] == sha256_text(CLAIM) + assert payload["argv"] == ["cargo", "generate-lockfile", "--locked"] + assert payload["exit_code"] == 1 + assert payload["stderr_sha256"] == hashlib.sha256(b"unsupported\n").hexdigest() + assert bundle.receipt.artifact_sha256 == hashlib.sha256(bundle.artifact).hexdigest() + + +def test_research_producer_content_addresses_retrieval_and_excerpt() -> None: + """Research semantics bind fetched content and the bounded cited excerpt.""" + bundle = _research_bundle() + payload = json.loads(bundle.artifact) + assert isinstance(bundle.receipt, ResearchClaimReceipt) + assert bundle.receipt.source_revision == "sha256:" + hashlib.sha256( + b"immutable documentation bytes\n" + ).hexdigest() + assert payload["excerpt_sha256"] == hashlib.sha256( + b"generate-lockfile accepts --offline" + ).hexdigest() + + +def test_manifest_producer_and_verifier_create_immutable_id_index() -> None: + """Only an exact authenticated envelope can create the admission index.""" + bundle = _execution_bundle() + index = _verify(bundle) + assert index.receipts[bundle.receipt.receipt_id] == bundle.receipt + assert index.artifacts[bundle.receipt.receipt_id] == bundle.artifact + assert index.claims[bundle.receipt.receipt_id] == CLAIM + with pytest.raises(TypeError, match="verify_claim_evidence_manifest"): + VerifiedClaimEvidenceIndex() + with pytest.raises(TypeError): + index.receipts["other"] = bundle.receipt # type: ignore[index] + with pytest.raises(TypeError): + index.requirements["other"] = _requirement( # type: ignore[index] + CLAIM, + bundle, + ) + with pytest.raises(TypeError, match="authenticated claim policy"): + produce_claim_evidence_manifest( # type: ignore[list-item] + [(CLAIM, bundle)] + ) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("argv", ["cargo", "other"]), + ("tool_identity", "not-cargo"), + ("tool_version", "other"), + ("exit_code", 0), + ("stdout_sha256", "1" * 64), + ("stderr_sha256", "2" * 64), + ("isolation_policy", "other"), + ("network_policy", "enabled"), + ], +) +def test_execution_semantic_field_substitution_fails_closed( + field: str, + value: object, +) -> None: + """A fixed artifact cannot authorize substituted execution semantics.""" + bundle = _execution_bundle() + manifest = _manifest_with_receipt_mutation(bundle, {field: value}) + with pytest.raises(ValueError, match="semantic artifact"): + _verify(bundle, manifest=manifest) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("source_uri", "https://example.invalid/other"), + ("source_revision", "sha256:" + "1" * 64), + ("excerpt_sha256", "2" * 64), + ("retrieval_policy", "other"), + ], +) +def test_research_semantic_field_substitution_fails_closed( + field: str, + value: object, +) -> None: + """A fixed artifact cannot authorize substituted research semantics.""" + bundle = _research_bundle() + manifest = _manifest_with_receipt_mutation(bundle, {field: value}) + with pytest.raises(ValueError, match="semantic artifact"): + _verify(bundle, manifest=manifest) + + +def test_model_dict_cannot_enter_index_or_admission() -> None: + """The model cannot submit authoritative receipt or manifest dictionaries.""" + bundle = _execution_bundle() + with pytest.raises(TypeError, match="receipt objects"): + index_claim_evidence_receipts( # type: ignore[list-item] + [bundle.receipt.model_dump(mode="json")] + ) + with pytest.raises(TypeError, match="verified manifest index"): + admit_claim_evidence( # type: ignore[arg-type] + bundle.receipt.receipt_id, + trusted_index={bundle.receipt.receipt_id: bundle.receipt}, + claim=CLAIM, + admitted_at=ISSUED, + required_kind=EvidenceKind.EXECUTION, + ) + + +def test_source_receipt_branch_is_bound_and_cannot_authorize_research() -> None: + """Source semantics bind to their artifact and remain a distinct authority.""" + bundle = _source_bundle() + assert isinstance(_admit(bundle), SourceClaimReceipt) + manifest = _manifest_with_receipt_mutation(bundle, {"source_line": 43}) + with pytest.raises(ValueError, match="semantic artifact"): + _verify(bundle, manifest=manifest) + with pytest.raises(ValueError, match="kind mismatch"): + _admit(bundle, required_kind=EvidenceKind.RESEARCH) + + +def test_missing_or_duplicate_trusted_receipt_fails_closed() -> None: + """A citation must resolve uniquely in the authenticated manifest index.""" + bundle = _execution_bundle() + with pytest.raises(ValueError, match="missing"): + _admit(bundle, receipt_id="absent") + with pytest.raises(ValueError, match="duplicate"): + index_claim_evidence_receipts([bundle.receipt, bundle.receipt]) + with pytest.raises(ValueError, match="duplicate"): + produce_claim_evidence_manifest( + [ + (_requirement(CLAIM, bundle), bundle), + (_requirement(CLAIM, bundle), bundle), + ] + ) + + +@pytest.mark.parametrize( + ("override", "value", "message"), + [ + ("expected_repository", "ContextualWisdomLab/other", "identity"), + ("expected_head_sha", "e" * 40, "identity"), + ( + "expected_workflow_ref", + "ContextualWisdomLab/noema/.github/workflows/ci.yml@" + "f" * 40, + "identity", + ), + ("expected_run_id", 9, "identity"), + ("expected_run_attempt", 2, "identity"), + ("expected_producers", {"other": EvidenceKind.EXECUTION}, "producer policy"), + ], +) +def test_caller_owned_manifest_identity_and_policy_fail_closed( + override: str, + value: object, + message: str, +) -> None: + """Manifest bytes cannot alter caller-owned workflow or producer policy.""" + with pytest.raises(ValueError, match=message): + _verify(**{override: value}) + + +def test_model_claim_mismatch_fails_closed_after_manifest_verification() -> None: + """A receipt ID cannot authorize model text other than its exact claim.""" + with pytest.raises(ValueError, match="claim digest"): + _admit(claim=CLAIM + " altered") + + +@pytest.mark.parametrize("when", [ISSUED - timedelta(seconds=1), EXPIRES]) +def test_receipt_outside_bounded_validity_fails_closed(when: datetime) -> None: + """Not-yet-issued and expired receipts cannot authorize a current review.""" + with pytest.raises(ValueError, match="not valid"): + _admit(admitted_at=when) + + +def test_ambiguous_or_inverted_validity_fails_closed() -> None: + """Producer, manifest, and admission timestamps require aware forward time.""" + with pytest.raises(ValueError, match="artifact timestamps"): + produce_execution_claim_receipt( + **{ + key: value + for key, value in _execution_producer_arguments().items() + if key != "issued_at" + }, + issued_at=datetime(2026, 9, 7), + ) + bundle = _execution_bundle() + for mutation in ( + {"issued_at": "2026-09-07T00:00:00"}, + {"expires_at": "2026-09-08T00:00:00"}, + {"expires_at": "2026-09-07T00:00:00Z"}, + ): + manifest = _manifest_with_receipt_mutation(bundle, mutation) + with pytest.raises(ValidationError): + _verify(bundle, manifest=manifest) + with pytest.raises(ValueError, match="admission time"): + _admit(admitted_at=datetime(2026, 9, 7)) + + +def _execution_producer_arguments() -> dict[str, object]: + """Expose canonical producer inputs for one timestamp failure case.""" + return { + "receipt_id": "execution-1", + "repository": "ContextualWisdomLab/ConceptWeave", + "head_sha": HEAD, + "workflow_ref": WORKFLOW, + "run_id": 12, + "run_attempt": 1, + "claim": CLAIM, + "producer_id": "sandboxed-verify", + "producer_version": "v1", + "policy_version": "review-evidence-v1", + "issued_at": ISSUED, + "expires_at": EXPIRES, + "argv": ["cargo"], + "tool_identity": "cargo", + "tool_version": "1.90.0", + "exit_code": 1, + "stdout": b"", + "stderr": b"unsupported\n", + "isolation_policy": "sandboxed-verify-v1", + "network_policy": "disabled", + } + + +def test_artifact_size_digest_and_semantic_bytes_each_fail_closed() -> None: + """Manifest verification distinguishes truncation, substitution, and encoding.""" + bundle = _execution_bundle() + assert _admit(bundle) == bundle.receipt + for artifact, message in ( + (bundle.artifact + b"x", "artifact size"), + (b"x" * len(bundle.artifact), "artifact digest"), + ): + altered = ProducedClaimEvidence(receipt=bundle.receipt, artifact=artifact) + with pytest.raises(ValueError, match=message): + _verify(altered) + semantically_equal = json.dumps(json.loads(bundle.artifact), indent=2).encode() + mutated_receipt = bundle.receipt.model_copy( + update={ + "artifact_sha256": hashlib.sha256(semantically_equal).hexdigest(), + "artifact_size": len(semantically_equal), + } + ) + altered = ProducedClaimEvidence(receipt=mutated_receipt, artifact=semantically_equal) + with pytest.raises(ValueError, match="semantic artifact"): + _verify(altered) + + +@pytest.mark.parametrize( + "mutation", + [ + {"claim_type": "research"}, + {"head_sha": "A" * 40}, + {"artifact_sha256": "not-a-digest"}, + {"run_attempt": 0}, + {"argv": [""]}, + {"artifact_path": "/tmp/model-selected"}, + ], +) +def test_malformed_or_model_authored_authority_fails_schema( + mutation: dict[str, object], +) -> None: + """Unknown authority and noncanonical identity never enter a verified index.""" + bundle = _execution_bundle() + manifest = _manifest_with_receipt_mutation(bundle, mutation) + with pytest.raises(ValidationError): + _verify(bundle, manifest=manifest) + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + (b"SANDBOXED_VERIFY_RESULT status=passed", "valid JSON"), + (b'{"schema_version":1,"entries":{}}\n', "shape"), + (b'{"entries":[],"schema_version":3}\n', "shape"), + (b'{"schema_version":1,"entries":[],"extra":1}\n', "shape"), + ], +) +def test_marker_only_or_invalid_manifest_shape_fails_closed( + payload: bytes, + message: str, +) -> None: + """A marker or malformed envelope cannot become a verified producer index.""" + with pytest.raises(ValueError, match=message): + _verify(manifest=payload) + + +def test_manifest_digest_and_canonical_bytes_fail_closed() -> None: + """The OpenCode handoff digest and one canonical serialization are required.""" + bundle = _execution_bundle() + manifest = produce_claim_evidence_manifest( + [(_requirement(CLAIM, bundle), bundle)] + ) + with pytest.raises(ValueError, match="expected digest"): + _verify(manifest=manifest, expected_manifest_sha256="INVALID") + with pytest.raises(ValueError, match="digest mismatch"): + _verify(manifest=manifest, expected_manifest_sha256="0" * 64) + pretty = json.dumps(json.loads(manifest), indent=2).encode() + with pytest.raises(ValueError, match="not canonical"): + _verify(manifest=pretty) + + +def test_manifest_entry_shape_type_and_base64_fail_closed() -> None: + """Manifest records require exact fields, text claims, and canonical base64.""" + bundle = _execution_bundle() + base = json.loads( + produce_claim_evidence_manifest([(_requirement(CLAIM, bundle), bundle)]) + ) + variants = [] + extra = json.loads(json.dumps(base)) + extra["entries"][0]["extra"] = 1 + variants.append((extra, "entry shape")) + bad_claim = json.loads(json.dumps(base)) + bad_claim["entries"][0]["claim_requirement"]["claim"] = 1 + variants.append((bad_claim, "requirement")) + bad_base64 = json.loads(json.dumps(base)) + bad_base64["entries"][0]["artifact_base64"] = "***" + variants.append((bad_base64, "invalid base64")) + bad_artifact_type = json.loads(json.dumps(base)) + bad_artifact_type["entries"][0]["artifact_base64"] = 1 + variants.append((bad_artifact_type, "entry type")) + noncanonical_base64 = json.loads(json.dumps(base)) + encoded = b64encode(bundle.artifact).decode("ascii") + alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" + padding = len(encoded) - len(encoded.rstrip("=")) + final_index = len(encoded) - padding - 1 + low_bit_mask = 0b1111 if padding == 2 else 0b0011 + replacement = alphabet[alphabet.index(encoded[final_index]) ^ low_bit_mask] + noncanonical_base64["entries"][0]["artifact_base64"] = ( + encoded[:final_index] + replacement + encoded[final_index + 1 :] + ) + variants.append((noncanonical_base64, "not canonical")) + for payload, message in variants: + manifest = ( + json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n" + ).encode() + with pytest.raises(ValueError, match=message): + _verify(bundle, manifest=manifest) + + +def test_manifest_claim_and_producer_contract_fail_closed() -> None: + """Producer output cannot relabel its exact claim or evidence authority.""" + bundle = _execution_bundle() + wrong_claim = CLAIM + " altered" + with pytest.raises(ValueError, match="claim digest"): + produce_claim_evidence_manifest( + [(_requirement(wrong_claim, bundle), bundle)] + ) + manifest = json.loads( + produce_claim_evidence_manifest([(_requirement(CLAIM, bundle), bundle)]) + ) + manifest["entries"][0]["claim_requirement"]["claim"] = wrong_claim + body = (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode() + with pytest.raises(ValueError, match="claim digest"): + _verify(bundle, manifest=body) + + +def test_manifest_requirement_kind_is_independent_from_receipt_kind() -> None: + """Authenticated claim policy cannot borrow its required kind from a receipt.""" + bundle = _execution_bundle() + wrong_requirement = ClaimEvidenceRequirement( + claim=CLAIM, + required_evidence_kind=EvidenceKind.SOURCE, + publication_authority=ClaimPublicationAuthority.FINDING, + ) + with pytest.raises(ValueError, match="requirement kind mismatch"): + produce_claim_evidence_manifest([(wrong_requirement, bundle)]) + + manifest = json.loads( + produce_claim_evidence_manifest([(_requirement(CLAIM, bundle), bundle)]) + ) + manifest["entries"][0]["claim_requirement"][ + "required_evidence_kind" + ] = EvidenceKind.SOURCE.value + body = (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode() + with pytest.raises(ValueError, match="requirement kind mismatch"): + _verify(bundle, manifest=body) + + +def test_sha256_text_binds_exact_utf8_claim_bytes() -> None: + """Claim wording changes produce a different exact digest.""" + assert sha256_text(CLAIM) != sha256_text(CLAIM.casefold()) diff --git a/reviewer/tests/test_claim_evidence_reference.py b/reviewer/tests/test_claim_evidence_reference.py new file mode 100644 index 000000000..2c1c9e023 --- /dev/null +++ b/reviewer/tests/test_claim_evidence_reference.py @@ -0,0 +1,209 @@ +"""Tests for canonical model-visible claim-evidence receipt references.""" + +from __future__ import annotations + +import hashlib +from datetime import datetime, timedelta, timezone + +import noema_reviewer +import pytest + +from noema_reviewer.claim_evidence import ( + ClaimEvidenceRequirement, + ClaimPublicationAuthority, + EvidenceKind, + ProducedClaimEvidence, + VerifiedClaimEvidenceIndex, + produce_claim_evidence_manifest, + produce_execution_claim_receipt, + verify_claim_evidence_manifest, +) +from noema_reviewer.claim_evidence_reference import ( + admit_claim_evidence_reference, + parse_claim_evidence_reference, +) +from noema_reviewer.source_claim_evidence import produce_source_claim_receipt + + +CLAIM = ( + "--locked is not a valid invocation: --locked is not accepted by the " + "generate-lockfile subcommand. This will always fail, breaking the workflow." +) +SOURCE_CLAIM = "cargo generate-lockfile --locked" +SYNONYM = "The generate-lockfile subcommand rejects --locked, so this workflow cannot succeed." +HEAD = "a" * 40 +WORKFLOW = "ContextualWisdomLab/noema/.github/workflows/central-review.yml@" + "b" * 40 +ISSUED = datetime(2026, 9, 7, tzinfo=timezone.utc) +EXPIRES = ISSUED + timedelta(hours=1) + + +def _execution_bundle(claim: str = CLAIM, *, receipt_id: str = "execution-1") -> ProducedClaimEvidence: + """Return one producer-issued execution receipt for an exact claim.""" + return produce_execution_claim_receipt( + receipt_id=receipt_id, + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=claim, + producer_id="sandboxed-verify", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + argv=["cargo", "generate-lockfile", "--locked"], + tool_identity="cargo", + tool_version="1.90.0", + exit_code=1, + stdout=b"", + stderr=b"unsupported\n", + isolation_policy="sandboxed-verify-v1", + network_policy="disabled", + ) + + +def _source_bundle() -> ProducedClaimEvidence: + """Return one producer-issued current-head source receipt.""" + return produce_source_claim_receipt( + receipt_id="source-1", + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=SOURCE_CLAIM, + producer_id="changed-source-map", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + source_path=".github/workflows/ci.yml", + source_line=42, + source_line_bytes=b"cargo generate-lockfile --locked\n", + ) + + +def _verified_index( + claim: str, + bundle: ProducedClaimEvidence, +) -> VerifiedClaimEvidenceIndex: + """Authenticate one producer manifest against caller-owned workflow identity.""" + manifest = produce_claim_evidence_manifest( + [ + ( + ClaimEvidenceRequirement( + claim=claim, + required_evidence_kind=bundle.receipt.evidence_kind, + publication_authority=ClaimPublicationAuthority.FINDING, + ), + bundle, + ) + ] + ) + return verify_claim_evidence_manifest( + manifest, + expected_manifest_sha256=hashlib.sha256(manifest).hexdigest(), + expected_repository="ContextualWisdomLab/ConceptWeave", + expected_head_sha=HEAD, + expected_workflow_ref=WORKFLOW, + expected_run_id=12, + expected_run_attempt=1, + expected_producers={bundle.receipt.producer_id: bundle.receipt.evidence_kind}, + ) + + +def test_reference_adapter_is_part_of_the_public_reviewer_port() -> None: + """Consumers can use one public Noema port instead of importing a private submodule.""" + assert noema_reviewer.parse_claim_evidence_reference is parse_claim_evidence_reference + assert noema_reviewer.admit_claim_evidence_reference is admit_claim_evidence_reference + + +def test_exact_receipt_marker_admits_the_original_external_behavior_claim() -> None: + """The original consumer RED becomes admissible only through its exact receipt.""" + bundle = _execution_bundle() + receipt = admit_claim_evidence_reference( + f"{CLAIM} [receipt:{bundle.receipt.receipt_id}]", + trusted_index=_verified_index(CLAIM, bundle), + admitted_at=ISSUED + timedelta(minutes=1), + required_kind=EvidenceKind.EXECUTION, + ) + assert receipt == bundle.receipt + + +def test_synonym_cannot_reuse_a_receipt_for_different_claim_bytes() -> None: + """Semantic similarity cannot substitute for exact producer-authenticated claim bytes.""" + bundle = _execution_bundle() + with pytest.raises(ValueError, match="claim digest"): + admit_claim_evidence_reference( + f"{SYNONYM} [receipt:{bundle.receipt.receipt_id}]", + trusted_index=_verified_index(CLAIM, bundle), + admitted_at=ISSUED + timedelta(minutes=1), + required_kind=EvidenceKind.EXECUTION, + ) + + synonym_bundle = _execution_bundle(SYNONYM, receipt_id="execution-synonym") + assert ( + admit_claim_evidence_reference( + f"{SYNONYM} [receipt:{synonym_bundle.receipt.receipt_id}]", + trusted_index=_verified_index(SYNONYM, synonym_bundle), + admitted_at=ISSUED + timedelta(minutes=1), + required_kind=EvidenceKind.EXECUTION, + ) + == synonym_bundle.receipt + ) + + +def test_caller_owned_kind_prevents_source_receipt_from_authorizing_execution() -> None: + """A model-visible source citation cannot self-promote into execution authority.""" + bundle = _source_bundle() + with pytest.raises(ValueError, match="kind mismatch"): + admit_claim_evidence_reference( + f"{SOURCE_CLAIM} [receipt:{bundle.receipt.receipt_id}]", + trusted_index=_verified_index(SOURCE_CLAIM, bundle), + admitted_at=ISSUED + timedelta(minutes=1), + required_kind=EvidenceKind.EXECUTION, + ) + + +def test_marker_only_sandbox_result_is_not_a_trusted_receipt() -> None: + """A sandboxed_verify-style text marker cannot manufacture producer evidence.""" + bundle = _execution_bundle() + with pytest.raises(ValueError, match="missing from trusted manifest"): + admit_claim_evidence_reference( + "sandboxed_verify: success [receipt:sandboxed-verify-result]", + trusted_index=_verified_index(CLAIM, bundle), + admitted_at=ISSUED + timedelta(minutes=1), + required_kind=EvidenceKind.EXECUTION, + ) + + +@pytest.mark.parametrize( + "reference", + [ + CLAIM, + f"{CLAIM}[receipt:execution-1]", + f"{CLAIM} [receipt:execution-1]", + f"{CLAIM} [RECEIPT:execution-1]", + f"{CLAIM}\n[receipt:execution-1]", + f"{CLAIM} [receipt:execution-1] trailing", + ], +) +def test_noncanonical_or_missing_receipt_marker_fails_closed(reference: str) -> None: + """Only one exact trailing receipt marker is admitted as model-visible syntax.""" + with pytest.raises(ValueError, match="canonical"): + parse_claim_evidence_reference(reference) + + +def test_multiple_receipt_markers_fail_closed() -> None: + """A model cannot offer alternate receipt IDs and leave authority selection ambiguous.""" + with pytest.raises(ValueError, match="exactly one"): + parse_claim_evidence_reference( + f"{CLAIM} [receipt:first] [receipt:second]" + ) + + +def test_non_string_reference_is_rejected_before_parsing() -> None: + """Structured model objects cannot bypass the canonical text reference port.""" + with pytest.raises(TypeError, match="string"): + parse_claim_evidence_reference({"receipt": "execution-1"}) # type: ignore[arg-type] diff --git a/reviewer/tests/test_sandboxed_verify_claim_evidence.py b/reviewer/tests/test_sandboxed_verify_claim_evidence.py new file mode 100644 index 000000000..f43b34614 --- /dev/null +++ b/reviewer/tests/test_sandboxed_verify_claim_evidence.py @@ -0,0 +1,127 @@ +"""Tests for adapting the trusted central sandboxed_verify result into receipts.""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timedelta, timezone + +import pytest +from pydantic import ValidationError + +from noema_reviewer import ( + ExecutionClaimReceipt, + produce_sandboxed_verify_execution_claim_receipt, +) + + +CENTRAL_SHA = "c9052e607e5f3cc76e73207e7786b21500721b79" +HEAD = "a" * 40 +WORKFLOW = "ContextualWisdomLab/noema/.github/workflows/central-review.yml@" + "b" * 40 +ISSUED = datetime(2026, 9, 7, tzinfo=timezone.utc) +EXPIRES = ISSUED + timedelta(hours=1) +CLAIM = "cargo generate-lockfile exited with an unsupported --locked argument." +COMMAND_STDOUT = b"" +COMMAND_STDERR = b"error: unexpected argument '--locked' found\n" + + +def _marker(**overrides: object) -> str: + """Return one current central sandboxed_verify marker with selected overrides.""" + payload: dict[str, object] = { + "allowed_env": [], + "command": ["cargo", "generate-lockfile", "--locked"], + "cwd": "/tmp/sandboxed-verify/repo", + "elapsed_seconds": 0.125, + "evidence_note": "", + "exit_code": 1, + "network": "not-required", + "sandbox": "(removed)", + "sandboxed": True, + } + payload.update(overrides) + return "SANDBOXED_VERIFY_RESULT " + json.dumps(payload, sort_keys=True) + + +def _produce(**overrides: object) -> ExecutionClaimReceipt: + """Produce one execution receipt through the public trusted adapter.""" + values: dict[str, object] = { + "receipt_id": "execution-sandboxed-verify-1", + "repository": "ContextualWisdomLab/ConceptWeave", + "head_sha": HEAD, + "workflow_ref": WORKFLOW, + "run_id": 12, + "run_attempt": 1, + "claim": CLAIM, + "policy_version": "review-evidence-v1", + "issued_at": ISSUED, + "expires_at": EXPIRES, + "marker": _marker(), + "command_stdout": COMMAND_STDOUT, + "command_stderr": COMMAND_STDERR, + "tool_version": CENTRAL_SHA, + } + values.update(overrides) + produced = produce_sandboxed_verify_execution_claim_receipt(**values) + assert isinstance(produced.receipt, ExecutionClaimReceipt) + return produced.receipt + + +def test_current_central_marker_becomes_exact_execution_receipt() -> None: + """The adapter binds command/result/transcripts and conservative runtime policy.""" + receipt = _produce() + assert receipt.argv == ("cargo", "generate-lockfile", "--locked") + assert receipt.tool_identity == "ContextualWisdomLab/.github:scripts/ci/sandboxed_verify.py" + assert receipt.tool_version == CENTRAL_SHA + assert receipt.producer_id == "sandboxed-verify" + assert receipt.producer_version == CENTRAL_SHA + assert receipt.exit_code == 1 + assert receipt.stdout_sha256 == hashlib.sha256(COMMAND_STDOUT).hexdigest() + assert receipt.stderr_sha256 == hashlib.sha256(COMMAND_STDERR).hexdigest() + assert receipt.isolation_policy == ( + "workspace-copy+scrubbed-env;os-process-isolation=none" + ) + assert receipt.network_policy == "declared:not-required;enforced=false" + + +def test_marker_only_without_exact_command_transcripts_fails_closed() -> None: + """The legacy marker alone cannot become authenticated execution evidence.""" + with pytest.raises(ValueError, match="stdout capture"): + _produce(command_stdout=None) + with pytest.raises(ValueError, match="stderr capture"): + _produce(command_stderr=None) + + +def test_unreviewed_sandboxed_verify_version_fails_closed() -> None: + """Changed helper semantics require a reviewed Noema adapter-policy bump.""" + with pytest.raises(ValueError, match="unsupported sandboxed_verify version"): + _produce(tool_version="d" * 40) + + +def test_nonempty_environment_capability_fails_closed_until_receipted() -> None: + """Environment capabilities cannot silently influence an execution receipt.""" + with pytest.raises(ValueError, match="allowed_env"): + _produce(marker=_marker(allowed_env=["PRIVATE_INDEX_TOKEN"])) + + +@pytest.mark.parametrize( + "marker", + [ + "not-a-result", + "SANDBOXED_VERIFY_RESULT []", + _marker(sandboxed=False), + _marker(command=[]), + _marker(network="disabled"), + _marker(exit_code=True), + _marker(elapsed_seconds=float("inf")), + ], +) +def test_malformed_or_overclaiming_marker_fails_closed(marker: str) -> None: + """Only the reviewed current central marker shape can issue execution authority.""" + with pytest.raises((ValueError, ValidationError)): + _produce(marker=marker) + + +def test_extra_marker_field_fails_closed() -> None: + """Central marker expansion is a versioned contract change, not implicit authority.""" + with pytest.raises(ValidationError): + _produce(marker=_marker(new_semantic_field="unreviewed")) diff --git a/reviewer/tests/test_source_claim_evidence_producer.py b/reviewer/tests/test_source_claim_evidence_producer.py new file mode 100644 index 000000000..fca06856d --- /dev/null +++ b/reviewer/tests/test_source_claim_evidence_producer.py @@ -0,0 +1,99 @@ +"""Regression coverage for canonical source-claim evidence production.""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timedelta, timezone + +import pytest + +from noema_reviewer import EvidenceKind, ProducedClaimEvidence, produce_source_claim_receipt + + +ISSUED = datetime(2026, 9, 7, tzinfo=timezone.utc) +EXPIRES = ISSUED + timedelta(hours=1) +HEAD = "a" * 40 +WORKFLOW = "ContextualWisdomLab/noema/.github/workflows/central-review.yml@" + "b" * 40 +LINE = b"run: cargo generate-lockfile --locked\n" +CLAIM = "run: cargo generate-lockfile --locked" + + +def _produce_source_claim( + *, + claim: str = CLAIM, + source_line_bytes: bytes = LINE, +) -> ProducedClaimEvidence: + """Produce one current-head source receipt from the exact fixture line.""" + return produce_source_claim_receipt( + receipt_id="source-1", + repository="ContextualWisdomLab/ConceptWeave", + head_sha=HEAD, + workflow_ref=WORKFLOW, + run_id=12, + run_attempt=1, + claim=claim, + producer_id="changed-source-map", + producer_version="v1", + policy_version="review-evidence-v1", + issued_at=ISSUED, + expires_at=EXPIRES, + source_path=".github/workflows/ci.yml", + source_line=42, + source_line_bytes=source_line_bytes, + ) + + +def test_source_producer_seals_exact_line_bytes_into_canonical_artifact() -> None: + """Source evidence binds the exact decoded line and its original line bytes.""" + produced = _produce_source_claim() + + assert isinstance(produced, ProducedClaimEvidence) + assert produced.receipt.evidence_kind is EvidenceKind.SOURCE + assert produced.receipt.source_line_sha256 == hashlib.sha256(LINE).hexdigest() + payload = json.loads(produced.artifact) + assert payload["source_path"] == ".github/workflows/ci.yml" + assert payload["source_line"] == 42 + assert payload["source_line_sha256"] == hashlib.sha256(LINE).hexdigest() + assert produced.receipt.artifact_sha256 == hashlib.sha256(produced.artifact).hexdigest() + + +@pytest.mark.parametrize( + "source_line_bytes", + [ + b"run: cargo generate-lockfile --locked\r\n", + b"run: cargo generate-lockfile --locked", + ], +) +def test_source_producer_accepts_exact_claim_across_canonical_line_endings( + source_line_bytes: bytes, +) -> None: + """CRLF and unterminated checkout lines retain the same exact claim text.""" + produced = _produce_source_claim(source_line_bytes=source_line_bytes) + assert produced.receipt.source_line_sha256 == hashlib.sha256( + source_line_bytes + ).hexdigest() + + +def test_source_producer_rejects_claim_not_derived_from_source_line() -> None: + """A path/line receipt cannot authorize model prose absent from that exact line.""" + with pytest.raises(ValueError, match="claim must equal exact source line"): + _produce_source_claim( + claim="the workflow invokes cargo generate-lockfile --locked" + ) + + +@pytest.mark.parametrize( + ("source_line_bytes", "message"), + [ + (b"first\nsecond\n", "exactly one source line"), + (b"\xff\n", "valid UTF-8"), + ], +) +def test_source_producer_rejects_ambiguous_or_non_utf8_line_bytes( + source_line_bytes: bytes, + message: str, +) -> None: + """A source receipt cannot collapse multiple lines or undecodable bytes into a claim.""" + with pytest.raises(ValueError, match=message): + _produce_source_claim(claim="irrelevant", source_line_bytes=source_line_bytes)