diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS new file mode 100644 index 0000000..08e2a20 --- /dev/null +++ b/.github/CODEOWNERS @@ -0,0 +1,9 @@ +# Changes to what the gate checks require human review (ADR 0002, decision 6). +# schemas/ is included beyond the ADR's literal list: the contract schemas +# define what doctor accepts, so they are the same enforcement surface. +# Everything else in this repo stays autonomous. +/loop/ @SollanSystems +/schemas/ @SollanSystems +/action.yml @SollanSystems +/.github/workflows/ @SollanSystems +/.github/CODEOWNERS @SollanSystems diff --git a/.github/workflows/attest.yml b/.github/workflows/attest.yml new file mode 100644 index 0000000..7b3de20 --- /dev/null +++ b/.github/workflows/attest.yml @@ -0,0 +1,109 @@ +name: attest + +on: + push: + branches: [main] + +permissions: + contents: read + +jobs: + verdict: + # Push-to-default-branch only, by ADR 0002 decision 5: attesting on a PR + # would mint a signed verdict under the repository's identity before review. + runs-on: ubuntu-latest + permissions: + contents: read + id-token: write + attestations: write + steps: + - uses: actions/checkout@v7 + + - uses: actions/setup-python@v7 + with: + python-version: "3.12" + + - name: Install seed dependencies + run: python -m pip install --upgrade pip pyyaml jsonschema + + - name: Seed a terminated, chained workspace + id: seed + # Through the REAL writer path — the runner's dispatch + auto-terminal — + # never a hand-written terminal (doctor flags those as desynced) and + # never a tracked example (no examples/* contract ships an events.db, so + # the attest step would skip and this job would pass having attested + # nothing: unfalsifiable, the exact false-completion shape this project + # refuses). Two dispatches: the first executes the one task and binds + # its evidence into the chain, the second fires the auto-terminal. + run: | + workspace="${RUNNER_TEMP}/verdict-ws" + head="$(python -B - "$workspace" <<'PY' + import json + import pathlib + import sys + + sys.path.insert(0, ".") + from loop import emit + from loop.contract import doctor_report + from loop.events import SQLiteEventStore + from loop.runner import dispatch_once + + ws = pathlib.Path(sys.argv[1]) + emit.open_contract(ws) + task = {"id": "T-1", "title": "T-1", "status": "pending", + "criterion_ref": "T-1", "verify": "./scripts/verify-fast.sh", + "depends_on": [], "attempts": 0, "evidence": None} + (ws / "TASKS.json").write_text( + json.dumps({"schema": "loop-engineer/tasks@1", "tasks": [task]}), + encoding="utf-8") + script = ws / "scripts" / "verify-fast.sh" + script.parent.mkdir(parents=True, exist_ok=True) + script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + script.chmod(0o755) + store = SQLiteEventStore(ws / ".loop" / "events.db") + store.append("run-1", "contract_opened", {"workspace": ws.name}, actor="ci") + for state in ("plan", "critique-plan", "queue-tasks", "execute-task"): + store.append("run-1", "iteration_appended", + {"iteration_id": 0, "outcome": "replanned", "state": state}, + actor="ci") + dispatch_once(ws) + dispatch_once(ws) + report = doctor_report(ws) + if not report["ok"]: + raise SystemExit( + f"seeded workspace is not doctor-clean: " + f"{[issue['code'] for issue in report['issues']]}") + head = (((report["event_store"].get("chain") or {}).get("head") or {}) + .get("event_hash")) or "" + if not head: + raise SystemExit("seeded workspace has no chain head") + print(head) + PY + )" + echo "head=$head" >> "$GITHUB_OUTPUT" + echo "workspace=$workspace" >> "$GITHUB_OUTPUT" + + - id: gate + uses: ./ + with: + path: ${{ steps.seed.outputs.workspace }} + attest: "true" + + - name: assert an attestation was actually created + # Without this the job passes when the attest step skips, which is the + # failure mode this whole task exists to avoid. + env: + SEEDED_HEAD: ${{ steps.seed.outputs.head }} + OBSERVED_HEAD: ${{ steps.gate.outputs.chain-head }} + ATTESTATION: ${{ steps.gate.outputs.attestation-url }} + run: | + if [ -z "$ATTESTATION" ]; then + echo "::error::attest was requested but no attestation URL was produced" + exit 1 + fi + if [ "$OBSERVED_HEAD" != "$SEEDED_HEAD" ]; then + echo "::error::gate observed head '$OBSERVED_HEAD', seed produced '$SEEDED_HEAD'" + exit 1 + fi + echo "chain head: $OBSERVED_HEAD" >> "$GITHUB_STEP_SUMMARY" + echo "attestation: $ATTESTATION" >> "$GITHUB_STEP_SUMMARY" diff --git a/CHANGELOG.md b/CHANGELOG.md index de191c3..ff3e67d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,38 @@ All notable changes to `loop-engineer` are documented here. `WORKFLOW.md` and `README.md` are reworded to describe the mechanism; the 0.3.4 history is left intact. +## Unreleased + +**A verdict you can hand to a signer (slice 4a of tamper-evident provenance).** +The kernel gains `loop verdict `: a pure projection of a finished +run — doctor verdict, chain head, terminal outcome, and the chain-bound +evidence digests that pass the strict verified-evidence bar — into one +canonical `loop-engineer/verdict@1` predicate body (`schemas/verdict.schema.json`, +normative in `reference/repo-os-contract.md` §23). Digests, enums, and issue +codes only; `run_id` is the single operator-controlled string; the field set is +an allowlist held by test. The kernel never signs, never builds an in-toto +Statement, and never reads an environment variable — `scripts/test_verdict_purity.py` +makes each boundary mechanical. + +The composite action gains an opt-in `attest` input (default false): it writes +the predicate to the runner temp dir and hands it to `actions/attest` with +`subject-name: loop-chain-head` / `subject-digest: sha256:`, +exposing `attestation-url`/`attestation-id` outputs; a legible permission +precheck replaces the raw OIDC 403, and an empty chain head skips with a +warning rather than shipping a malformed subject. `.github/workflows/attest.yml` +mints a real attestation on every push to main over a workspace seeded through +the runner's own dispatch + auto-terminal path, and fails loud if no +attestation URL is produced or the observed head differs from the seeded one. + +What this does not buy: the signature attests context — repo, workflow, +trigger, time — never correctness, so a signed verdict over a weakened gate is +just a signed weakened gate. An agent with ordinary merge rights can loosen +`loop/**`/`schemas/**`/`action.yml`/the workflow and then mint a perfectly +genuine attestation for the result — the control is code-owner review on those +paths, which is in force only once the repository ruleset requires it — and an +unattested chain rewrite is detected at best one run late. Verification (`--compare`, anchor auto-resolution, signer-trust policy) +is slice 4b and does not ship here. + ## 0.11.0 — 2026-07-26 **Verifier identity, and evidence that is load-bearing.** Two slices of the diff --git a/action.yml b/action.yml index 78a6779..de4a30d 100644 --- a/action.yml +++ b/action.yml @@ -1,5 +1,5 @@ name: "loop-engineer gate" -description: "Proof-of-done gate for agent-loop contracts: hard-fails on doctor, scores with inspect (warn-only by default)." +description: "Proof-of-done gate for agent-loop contracts: hard-fails on doctor, scores with inspect (warn-only by default), and can keylessly attest the run's verdict@1 predicate (opt-in)." branding: icon: "check-circle" color: "green" @@ -32,11 +32,26 @@ inputs: detection — the gate then only records the head for a later comparison. required: false default: "" + attest: + description: >- + Emit a loop-engineer/verdict@1 predicate and attest it keylessly via + GitHub OIDC. Requires the CALLING JOB to declare `id-token: write` and + `attestations: write` — a composite action cannot declare its own + permissions. Default false. The predicate is PUBLIC and permanent for + public repositories. + required: false + default: "false" outputs: chain-head: description: "Chain head event_hash observed by this gate run ('' when the store has no chained events)." value: ${{ steps.chain-head.outputs.chain-head }} + attestation-url: + description: "URL of the attestation created by this run ('' when attest is false)." + value: ${{ steps.attest.outputs.attestation-url }} + attestation-id: + description: "ID of the attestation created by this run ('' when attest is false)." + value: ${{ steps.attest.outputs.attestation-id }} runs: using: "composite" @@ -106,6 +121,45 @@ runs: open(sys.argv[3], "a").write(f"chain-head={value}\n") PY + - name: verdict predicate + id: verdict + if: ${{ inputs.attest == 'true' }} + shell: bash + env: + LOOP_PATH: "${{ inputs.path }}" + run: | + # Fail with a legible message rather than a raw OIDC 403 from the attest + # step: a composite action cannot declare permissions, so the CALLING + # job must. ACTIONS_ID_TOKEN_REQUEST_URL is present only when the job + # declared id-token: write. + if [ -z "${ACTIONS_ID_TOKEN_REQUEST_URL:-}" ]; then + echo "::error::attest: true requires the calling job to declare" \ + "'permissions: { id-token: write, attestations: write }'." \ + "A composite action cannot declare its own permissions." + exit 1 + fi + loop verdict "$LOOP_PATH" > "${RUNNER_TEMP}/verdict.json" + echo "predicate-path=${RUNNER_TEMP}/verdict.json" >> "$GITHUB_OUTPUT" + + - name: attest verdict + id: attest + if: ${{ inputs.attest == 'true' && steps.chain-head.outputs.chain-head != '' }} + uses: actions/attest@v4 + with: + subject-name: loop-chain-head + subject-digest: sha256:${{ steps.chain-head.outputs.chain-head }} + predicate-type: urn:loop-engineer:verdict:1 + predicate-path: ${{ steps.verdict.outputs.predicate-path }} + push-to-registry: false + # Only effective when push-to-registry is true, but pinned false + # explicitly per ADR 0002 open item 5: no storage record, ever. + create-storage-record: false + + - name: attest skipped (no chained events) + if: ${{ inputs.attest == 'true' && steps.chain-head.outputs.chain-head == '' }} + shell: bash + run: echo "::warning::attest requested but the store has no chained events; nothing to attest." + - name: loop inspect (scorecard) shell: bash env: diff --git a/docs/superpowers/plans/2026-07-25-slice4a-verdict-emission.md b/docs/superpowers/plans/2026-07-25-slice4a-verdict-emission.md index 270d334..44d1b59 100644 --- a/docs/superpowers/plans/2026-07-25-slice4a-verdict-emission.md +++ b/docs/superpowers/plans/2026-07-25-slice4a-verdict-emission.md @@ -311,10 +311,10 @@ import pytest def _scaffold_terminal(tmp_path, name, state="Succeeded", policy="all_required"): """A doctor-clean scaffold advanced to a terminal record.""" - from loop.contract import scaffold_contract # existing scaffold entry point + from loop.scaffold import scaffold target = tmp_path / name - scaffold_contract(target) + scaffold(target) terminal = { "schema": "loop-engineer/terminal@1", "state": state, @@ -357,11 +357,11 @@ def test_issue_codes_are_sorted_deduplicated_and_carry_no_detail(tmp_path): def test_build_verdict_refuses_a_workspace_with_no_terminal_record(tmp_path): - from loop.contract import scaffold_contract + from loop.scaffold import scaffold from loop.verdict import VerdictError, build_verdict target = tmp_path / "no-terminal" - scaffold_contract(target) + scaffold(target) with pytest.raises(VerdictError, match="no terminal record"): build_verdict(target) @@ -379,7 +379,7 @@ def test_build_verdict_refuses_a_nonexistent_target(tmp_path): Run: `uv run --with pyyaml --with jsonschema --with pytest python3 -B -m pytest -q -p no:cacheprovider scripts/test_verdict.py -v` Expected: FAIL with `ImportError: cannot import name 'build_verdict'` -> If `scaffold_contract` is not the exported scaffold name at HEAD, run `uv run --with pyyaml python3 -B -c "import loop.contract as c; print([n for n in dir(c) if 'scaffold' in n])"` and use the real one in the helper. Do not invent an API. +> Verified at HEAD: the scaffold entry point is `loop.scaffold.scaffold(target)` (`loop/scaffold.py:104`). There is no `scaffold_contract`. `LoopPaths` also exposes `.terminal` directly (`loop/paths.py:24`), so prefer `paths.terminal` over `paths.loop_dir / "terminal_state.json"`. - [ ] **Step 3: Implement the projection** @@ -1033,10 +1033,19 @@ git commit -m "feat(action): opt-in keyless attestation of the verdict predicate - Create: `.github/workflows/attest.yml` **Interfaces:** -- Consumes: the composite action from Task 6. -- Produces: a real attestation on every push to the default branch, over a **tracked** `examples/*` contract. +- Consumes: the composite action from Task 6; the existing `scripts/ci_anchor_probe.py`. +- Produces: a real attestation on every push to the default branch, over a **seeded chained workspace**. -Never point this at the live gitignored `.loop/` — CI runs on a fresh checkout where it does not exist. +**Pre-flight correction.** An earlier draft pointed this job at `examples/flaky-test-triage`. That is +wrong: **no tracked `examples/*` contract ships an `events.db`** (verified — event stores are runtime +artifacts and `.loop/` is gitignored). With no store the chain head is empty, Task 6 Step 4's guard +skips the attest step, and the job goes green having attested nothing — an unfalsifiable CI job, which +is the exact false-completion shape this project exists to refuse. + +Reuse the pattern the repo already proved. `ci.yml`'s `chain anchor (live end-to-end)` job seeds a +chained workspace via `python -B scripts/ci_anchor_probe.py "$workspace"`, which prints the resulting +head; its own comment states the rationale verbatim — *"action-dogfood gates a store-free example, so +the anchor surface has no live cover there."* Never point this at the live gitignored `.loop/`. - [ ] **Step 1: Write the workflow** @@ -1060,25 +1069,59 @@ jobs: id-token: write attestations: write steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v7 + + - uses: actions/setup-python@v7 + with: + python-version: "3.12" + + - name: Install probe dependencies + run: python -m pip install --upgrade pip pyyaml jsonschema + + - name: Seed a chained workspace + id: seed + # Same probe the chain-anchor job uses. A store-free example would make + # the attest step skip and this job unfalsifiable. + run: | + workspace="${RUNNER_TEMP}/verdict-ws" + head="$(python -B scripts/ci_anchor_probe.py "$workspace")" + echo "head=$head" >> "$GITHUB_OUTPUT" + echo "workspace=$workspace" >> "$GITHUB_OUTPUT" + - id: gate uses: ./ with: - path: examples/flaky-test-triage + path: ${{ steps.seed.outputs.workspace }} attest: "true" - - name: record + + - name: assert an attestation was actually created + # Without this the job passes when the attest step skips, which is the + # failure mode this whole task exists to avoid. + env: + SEEDED_HEAD: ${{ steps.seed.outputs.head }} + OBSERVED_HEAD: ${{ steps.gate.outputs.chain-head }} + ATTESTATION: ${{ steps.gate.outputs.attestation-url }} run: | - echo "chain head: ${{ steps.gate.outputs.chain-head }}" >> "$GITHUB_STEP_SUMMARY" - echo "attestation: ${{ steps.gate.outputs.attestation-url }}" >> "$GITHUB_STEP_SUMMARY" + if [ -z "$ATTESTATION" ]; then + echo "::error::attest was requested but no attestation URL was produced" + exit 1 + fi + if [ "$OBSERVED_HEAD" != "$SEEDED_HEAD" ]; then + echo "::error::gate observed head '$OBSERVED_HEAD', seed produced '$SEEDED_HEAD'" + exit 1 + fi + echo "chain head: $OBSERVED_HEAD" >> "$GITHUB_STEP_SUMMARY" + echo "attestation: $ATTESTATION" >> "$GITHUB_STEP_SUMMARY" ``` -- [ ] **Step 2: Confirm the checkout action major matches the repo's other workflows** +- [ ] **Step 2: Confirm the action majors match the repo's other workflows** ```bash -grep -rn "actions/checkout@" .github/workflows/ +grep -rn "actions/checkout@\|actions/setup-python@" .github/workflows/ ``` -Use whatever major `ci.yml` uses. A mismatched pin is a dependabot PR waiting to happen. +Pre-flight observed `actions/checkout@v7` and `actions/setup-python@v7` in `ci.yml`; confirm before +committing. A mismatched pin is a dependabot PR waiting to happen. - [ ] **Step 3: Commit** @@ -1227,6 +1270,6 @@ Carried from ADR 0002. None blocks starting; each blocks the task that touches i 1. **Subject digest algorithm** (Task 6). The chain head is not a hash of retrievable bytes, so the `sha256` DigestSet key licenses a false inference. A namespaced key is correct in in-toto terms but `actions/attest`'s `subject-digest` may accept only `sha256:`. Resolve by experiment. If the input constrains us, §23 must carry the disambiguation instead. 2. **`create-storage-record` / `push-to-registry` defaults** (Task 6). Pass both explicitly so the two-permission claim is true by construction. -3. **`scaffold_contract` name** (Task 2 Step 2). Confirm the real scaffold entry point before writing the test helper. -4. **`event_store` key names** (Task 2 Step 5). Confirm `run_id`, `chain.head.event_hash`, `chain.head.sequence`, `chain.unchained_prefix` against real doctor output. +3. ~~`scaffold_contract` name~~ — **RESOLVED in pre-flight.** The entry point is `loop.scaffold.scaffold`; the plan now uses it. +4. **`event_store` key names** (Task 2 Step 5). Pre-flight confirmed the store-absent shape is exactly `{"present": false}` and that `doctor_report` also returns `paths` (absolute filesystem paths — deliberately excluded from the predicate) and `requested_mode`. `validation_mode` is the single token `"jsonschema"`, so Task 5's no-whitespace assertion is safe. The populated `chain.head.*` key names still need confirming against a seeded store. 5. **The chain-bound evidence map producer** (Task 3 Step 4). Reuse `loop/contract.py`'s, never re-derive. diff --git a/loop/__main__.py b/loop/__main__.py index b0b3dce..2f5e1a8 100644 --- a/loop/__main__.py +++ b/loop/__main__.py @@ -12,13 +12,13 @@ _PROG = "python3 -m loop" -_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "migrate", "architect") +_COMMANDS = ("scaffold", "doctor", "validate", "verify", "verdict", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "migrate", "architect") # Read commands operate on an EXISTING contract dir; scaffold CREATES one, so it # is exempt from the "target must exist" guard. -_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "migrate") +_READ_COMMANDS = ("doctor", "validate", "verify", "verdict", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "migrate") -_USAGE = f"usage: {_PROG} " +_USAGE = f"usage: {_PROG} " _HELP = f"""{_PROG} — validate, inspect, and measure a portable repo-OS loop contract. @@ -26,6 +26,7 @@ {_PROG} metrics [--baseline] {_PROG} doctor|validate|verify [--mode basic|strict|release] [--expect-chain-head SHA256] + {_PROG} verdict [--mode basic|strict|release] {_PROG} status [--mode basic|strict|release] {_PROG} replay [--mode basic|strict|release] {_PROG} simulate [--mode basic|strict|release] @@ -42,6 +43,9 @@ doctor Validate the contract objects; --mode selects validation strength. validate Alias for doctor. verify Alias for doctor — check the contract's state. + verdict Emit the predicate body only; the signer (actions/attest) constructs + the in-toto Statement. Never signs and never verifies a signature. + (schema loop-engineer/verdict@1: doctor status, chain head, terminal outcome, verified-evidence digests) inspect Score an existing loop against the prime-directive checklist (emits a weak/strong verdict and a gap report). metrics Derive false-completion-rate + repair-productivity from the loop's @@ -70,7 +74,7 @@ options: --mode {{basic,strict,release}} - (doctor/validate/verify/plan-lint/status/replay/simulate/run) basic forces structural + (doctor/validate/verify/verdict/plan-lint/status/replay/simulate/run) basic forces structural checks; strict/release require jsonschema. Default: auto-detect. --expect-chain-head SHA256 (doctor/validate/verify) fail unless the event store's chain head @@ -241,7 +245,7 @@ def main(argv: list[str] | None = None) -> int: return 2 mode = None - if command in {"doctor", "validate", "verify", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel"}: + if command in {"doctor", "validate", "verify", "verdict", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel"}: try: mode, argv = _extract_mode_flag(argv) except ValueError as exc: @@ -367,6 +371,17 @@ def main(argv: list[str] | None = None) -> int: print(f"{command}: {exc}", file=sys.stderr) return 2 + if command == "verdict": + from .verdict import VerdictError, build_verdict + from .chain import ChainHashError, canonical_json + + try: + print(canonical_json(build_verdict(target, mode=mode))) + return 0 + except (VerdictError, ChainHashError) as exc: + print(f"verdict: {exc}", file=sys.stderr) + return 2 + if command == "plan-lint": try: return _print_json(validate_plan(target, mode=mode)) diff --git a/loop/verdict.py b/loop/verdict.py new file mode 100644 index 0000000..c5cb62c --- /dev/null +++ b/loop/verdict.py @@ -0,0 +1,171 @@ +"""Project a loop run into a loop-engineer/verdict@1 predicate body. + +This module builds a document. It NEVER signs one, never verifies a signature, +never constructs an in-toto Statement, and never reads an environment variable. +The signer lane (action.yml -> actions/attest) owns the envelope claims and every +cryptographic operation. See docs/adr/0002-ci-attested-verdict.md. +""" + +from __future__ import annotations + +import hashlib +import json +from importlib import metadata +from pathlib import Path +from typing import Any + +from ._resources import schemas_dir +from .contract import _strict_evidence_failure, doctor_report +from .paths import LoopPaths, resolve_loop_paths +from .runtime import RuntimeStoreError, bound_artifact_digests + +VERDICT_SCHEMA_ID = "loop-engineer/verdict@1" +PREDICATE_TYPE = "urn:loop-engineer:verdict:1" + + +class VerdictError(ValueError): + """A verdict cannot be projected from this workspace.""" + + +def _load_verdict_schema() -> dict[str, Any]: + return json.loads((schemas_dir() / "verdict.schema.json").read_text(encoding="utf-8")) + + +def _tool_version() -> str | None: + try: + return metadata.version("loop-engineer") + except metadata.PackageNotFoundError: + return None + + +def _terminal_record(paths: LoopPaths) -> dict[str, Any]: + path = paths.loop_dir / "terminal_state.json" + if not path.is_file(): + raise VerdictError( + "no terminal record: a verdict projects a finished run " + f"({path.name} is absent)" + ) + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise VerdictError(f"terminal record is unreadable: {exc}") from exc + if not isinstance(data, dict): + raise VerdictError("terminal record is not an object") + if "false_completion" not in data: + raise VerdictError("terminal record is missing required false_completion") + if not isinstance(data["false_completion"], bool): + raise VerdictError("terminal record false_completion must be a boolean") + return data + + +def _evidence_digests(entry: object, paths: LoopPaths) -> dict[str, str | None] | None: + """Return the chain-committed record digest and verifier digests for an entry.""" + if not isinstance(entry, str): + return None + try: + record_bytes = (paths.workspace / entry).read_bytes() + record = json.loads(record_bytes.decode("utf-8")) + except (OSError, ValueError, UnicodeDecodeError, json.JSONDecodeError): + return None + if not isinstance(record, dict): + return None + verified_by = record.get("verified_by") + return { + "digest": hashlib.sha256(record_bytes).hexdigest(), + "code_digest": verified_by.get("code_digest") if isinstance(verified_by, dict) else None, + "policy_digest": verified_by.get("policy_digest") if isinstance(verified_by, dict) else None, + } + + +def _bound_evidence(paths: LoopPaths) -> dict[str, tuple[str, ...]] | None: + """Read evidence record digests committed by the event chain.""" + return bound_artifact_digests(paths.workspace) + + +def _verified_evidence( + terminal: dict[str, Any], paths: LoopPaths, bound: dict[str, tuple[str, ...]] | None +) -> list[dict[str, str | None]]: + """Project only terminal evidence that clears the shared strict bar.""" + entries = terminal.get("evidence") + if not isinstance(entries, list): + return [] + projected = set() + for entry in entries: + if _strict_evidence_failure(entry, paths, bound) is not None: + continue + digests = _evidence_digests(entry, paths) + if digests is None: + continue + if bound is not None and (digests["digest"],) != bound.get(entry): + # The bar validated its own read of the record; this projection read + # hashed differently, so the bytes moved between the two reads. A + # digest the chain never committed must not enter a signed document. + continue + projected.add((digests["digest"], digests["code_digest"], digests["policy_digest"])) + return [ + {"digest": digest, "code_digest": code_digest, "policy_digest": policy_digest} + for digest, code_digest, policy_digest in sorted( + projected, key=lambda item: (item[0], item[1] or "", item[2] or "") + ) + ] + + +def build_verdict(target: str | Path, *, mode: str | None = None) -> dict[str, Any]: + """Project local run state into a ``verdict@1`` predicate body. + + Pure over the workspace: no environment, network, signing, or verification. + """ + try: + paths = resolve_loop_paths(target) + except (OSError, ValueError, RuntimeError) as exc: + # RuntimeError is pathlib's symlink-loop signal on Python <= 3.12. + raise VerdictError(f"cannot resolve a loop workspace at {target}: {exc}") from exc + + # Separate from resolution so an invalid mode= is not reported as a path failure; + # ValidationModeError is a RuntimeError subclass and would otherwise land above. + try: + report = doctor_report(paths.workspace, mode=mode) + except (OSError, ValueError, RuntimeError) as exc: + raise VerdictError(f"cannot read the contract at {paths.workspace}: {exc}") from exc + + terminal = _terminal_record(paths) + try: + bound = _bound_evidence(paths) + except RuntimeStoreError: + evidence = [] + else: + evidence = _verified_evidence(terminal, paths, bound) + store = report.get("event_store") or {} + chain = store.get("chain") or {} + head = chain.get("head") or {} + policy = terminal.get("completion_policy") + policy_mode = policy.get("mode") if isinstance(policy, dict) else None + + return { + "schema": VERDICT_SCHEMA_ID, + "run_id": str(store.get("run_id") or paths.workspace.name), + "tool": {"name": "loop-engineer", "version": _tool_version()}, + "doctor": { + "ok": bool(report.get("ok")), + "validation_mode": str(report.get("validation_mode") or "unknown"), + "issue_codes": sorted({ + str(issue.get("code")) + for issue in report.get("issues", []) + if isinstance(issue, dict) and issue.get("code") + }), + "schemas_checked": sorted( + str(schema) for schema in report.get("schemas_checked", []) + ), + }, + "chain": { + "head": head.get("event_hash"), + "sequence": head.get("sequence"), + "unchained_prefix": int(chain.get("unchained_prefix") or 0), + }, + "terminal": { + "state": terminal.get("state"), + "completion_policy": policy_mode, + "false_completion": terminal["false_completion"], + }, + "evidence": evidence, + } diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index b818e83..3a55b17 100644 --- a/reference/repo-os-contract.md +++ b/reference/repo-os-contract.md @@ -1507,3 +1507,136 @@ match, the mismatch itself surfaces in doctor's issue list as Always pass an anchor in CI. A bare `loop doctor` treats a fully deleted store — no database, no sidecars — as a valid never-ran contract, so "delete the evidence" is a passing run without one. + + +## 23. `loop-engineer/verdict@1` — the CI-attested verdict predicate + +`loop verdict ` projects a **finished** run into one canonical JSON +document: the doctor verdict, the chain head, the terminal outcome, and the +chain-bound evidence digests that pass §17's strict verified-evidence bar. The +kernel emits the **predicate body only**. It never signs, never verifies a +signature, never constructs an in-toto Statement (`_type`, `subject`, +`predicateType`, and `predicate` keys are forbidden in its output), and never +reads an environment variable — every one of those boundaries is a mechanical +test, not an intention. The signer lane (`action.yml`'s opt-in `attest` input +→ `actions/attest`) owns the envelope, the subject, and all cryptography: +**the kernel disposes on contents; the CI lane notarizes context; neither is +the other.** + +Serialization is §16's canonical JSON — sorted keys, compact separators, +`ensure_ascii: false`, `allow_nan: false` — so the same run always projects +the same bytes, and a value canonical JSON cannot carry (a NaN smuggled into a +terminal record) is a typed refusal at the CLI, never a crash. + +**Conformance vector (machine-pinned).** `null` is legal for `chain.head`, +`chain.sequence`, `tool.version`, and `terminal.completion_policy`; every +other key is always present. The field set is an allowlist enforced by test — +everything in this document is public, append-only, and permanent, so adding a +field is a one-way door. + +```json +{ + "schema": "loop-engineer/verdict@1", + "run_id": "coverage-repair", + "tool": { "name": "loop-engineer", "version": "0.11.0" }, + "doctor": { + "ok": true, + "validation_mode": "jsonschema", + "issue_codes": [], + "schemas_checked": ["loop-engineer/manifest@1", "loop-engineer/state@1"] + }, + "chain": { + "head": "9f2c…64hex", + "sequence": 41, + "unchained_prefix": 0 + }, + "terminal": { + "state": "Succeeded", + "completion_policy": "all_required_verified_evidence", + "false_completion": false + }, + "evidence": [ + { + "digest": "a1b2…64hex", + "code_digest": "c3d4…64hex", + "policy_digest": "e5f6…64hex" + } + ] +} +``` + +Field semantics: + +- `run_id` — the **one operator-controlled string** in the document, + allowlisted deliberately so a verdict can be correlated to a run. It lands + in a permanent public log: run ids must not embed sensitive text. Every + other string is a digest, an enum, a snake_case issue code, or a schema id — + a whitespace-bearing value anywhere else is a conformance failure. +- `doctor.issue_codes` — **codes only, sorted, de-duplicated.** Never + `message`, never `path`. Free-text detail strings and workspace paths do + not leave the machine. +- `doctor.validation_mode` — which strength the contract was validated at + (`jsonschema` or `structural-fallback`); without it, `ok: true` from a + fallback environment would read as the stronger claim. +- `chain.head` / `chain.sequence` — the store's chained head (§22's `chain` + block). `null` for a store-less workspace **and** for a store whose chain + never established a head; `unchained_prefix` carries §16's honest count of + events the chain does not cover. +- `terminal.completion_policy` — the policy `mode` string, or `null` for a + legacy record that never declared one. Without it a reader cannot tell what + `Succeeded` meant (`all_required` vs `all_required_verified_evidence`). +- `evidence[]` — one entry per **chain-bound** terminal-evidence record that + passes §17's strict bar, digests only. `digest` is the SHA-256 of the + evidence@1 **record file bytes** — the exact digest the event chain + committed — never the record's own `sha256` field, which hashes the cited + artifact instead. `code_digest`/`policy_digest` lift from the record's + `verified_by`. Entries are de-duplicated and sorted by + `(digest, code_digest, policy_digest)` with `null` ordered as the empty + string. No URIs: a URI is a workspace path, and this document is public. + +**Refusals and degradations.** A workspace with no terminal record refuses, +typed — a verdict projects a finished run. A terminal record whose +`false_completion` is absent or non-boolean refuses: projecting `false` for an +unknown safety flag would trade an alarming truth for a reassuring lie. A +store-less workspace **projects** (`chain.head: null`, evidence under §17's +documented store-dependent degradation) — the projection is honest about what +it cannot prove rather than refusing to say anything. An **unreadable** store +is not that degradation: chain-boundness is then unestablished, so `evidence` +projects empty — an errored check fails, it never skips — while the store +failure itself surfaces in `doctor.issue_codes`. + +**Predicate identity.** `predicateType` is +`urn:loop-engineer:verdict:1` — the schema `$id` transliterated +(`/` and `@` → `:`), an equality asserted by test so the two names cannot +drift. It names no vendor host, no organization, and no repository: a +predicateType is written immutably into a public log, and a rename must not be +able to orphan it. + +**The subject seam — read this before verifying anything.** The signer binds +`subject-name: loop-chain-head` with +`subject-digest: sha256:`. That `sha256:` key satisfies the +signer's required `algorithm:hex_digest` form, but the chain head is a SHA-256 +over a **synthesized event preimage** (§16), not over any retrievable +artifact's bytes. A consumer must never conclude "fetch the bytes, re-hash, +compare" — there are no bytes to fetch. The only meaningful comparison is +equality against a chain head recomputed locally from the store (`loop +doctor`'s `chain` block, or the anchor check via `--expect-chain-head`). +Attested-vs-local agreement is a 4b (`--compare`) concern; authenticity +(`gh attestation verify`) and agreement are separate checks, in that order, +and neither implies the other. + +**What an attestation buys — and does not.** The signature attests *context*: +which repository, which workflow, which trigger, at what time. It never +attests correctness — a signed verdict over a weak gate is a signed weak +gate. A worker with ordinary merge rights can loosen the gate +(`loop/**`, `schemas/**` — the contract schemas define what doctor accepts — +`action.yml`, or the workflow) and then mint a perfectly genuine attestation +for the loosened gate; the control is human review on those paths (ADR 0002 +decision 6), not signing — and CODEOWNERS is that control **only while the +repository's ruleset requires code-owner review**. A CODEOWNERS file the +ruleset does not enforce is documentation, not a control. The chain proves order and non-tampering +relative to an anchor, not that the events happened when claimed — a history +fabricated wholesale at authoring time is byte-valid. Detection of an +unattested rewrite is at best one run late. And an attestation nothing +verifies is decoration: until a consumer checks it, this section describes a +publication surface, not a gate. diff --git a/schemas/verdict.schema.json b/schemas/verdict.schema.json new file mode 100644 index 0000000..70ad538 --- /dev/null +++ b/schemas/verdict.schema.json @@ -0,0 +1,75 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "loop-engineer/verdict@1", + "title": "Loop Engineer Verdict @1", + "description": "A CI-attestable projection of a loop run's doctor verdict, chain head, terminal record, and verified-evidence digests. Emitted as an in-toto predicate body; the signer constructs the Statement.", + "type": "object", + "additionalProperties": false, + "required": ["schema", "run_id", "tool", "doctor", "chain", "terminal", "evidence"], + "properties": { + "schema": { "const": "loop-engineer/verdict@1" }, + "run_id": { "type": "string", "minLength": 1, "maxLength": 256 }, + "tool": { + "type": "object", + "additionalProperties": false, + "required": ["name", "version"], + "properties": { + "name": { "const": "loop-engineer" }, + "version": { "type": ["string", "null"], "maxLength": 64 } + } + }, + "doctor": { + "type": "object", + "additionalProperties": false, + "required": ["ok", "validation_mode", "issue_codes", "schemas_checked"], + "properties": { + "ok": { "type": "boolean" }, + "validation_mode": { "type": "string", "maxLength": 64 }, + "issue_codes": { + "type": "array", + "items": { "type": "string", "pattern": "^[a-z0-9_]{1,64}$", "maxLength": 64 } + }, + "schemas_checked": { + "type": "array", + "items": { "type": "string", "maxLength": 128 } + } + } + }, + "chain": { + "type": "object", + "additionalProperties": false, + "required": ["head", "sequence", "unchained_prefix"], + "properties": { + "head": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$", "maxLength": 64 }, + "sequence": { "type": ["integer", "null"], "minimum": 0 }, + "unchained_prefix": { "type": "integer", "minimum": 0 } + } + }, + "terminal": { + "type": "object", + "additionalProperties": false, + "required": ["state", "completion_policy", "false_completion"], + "properties": { + "state": { + "enum": ["Succeeded", "FailedUnverifiable", "FailedBlocked", "FailedBudget", + "FailedSafety", "FailedSpecGap", "AbortedByHuman"] + }, + "completion_policy": { "type": ["string", "null"], "maxLength": 64 }, + "false_completion": { "type": "boolean" } + } + }, + "evidence": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["digest", "code_digest", "policy_digest"], + "properties": { + "digest": { "type": "string", "pattern": "^[0-9a-f]{64}$", "maxLength": 64 }, + "code_digest": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$", "maxLength": 64 }, + "policy_digest": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$", "maxLength": 64 } + } + } + } + } +} diff --git a/scripts/test_verdict.py b/scripts/test_verdict.py new file mode 100644 index 0000000..c869955 --- /dev/null +++ b/scripts/test_verdict.py @@ -0,0 +1,467 @@ +import hashlib +import json +import re + +import pytest + +from loop import emit +from loop.completion import VERIFIED_EVIDENCE_MODE +from loop.events import SQLiteEventStore +from loop.runner import dispatch_once +from loop.scaffold import scaffold + + +def _workspace_with_terminal(tmp_path, name="workspace", *, completion_policy=None): + target = tmp_path / name + scaffold(target) + (target / ".loop" / "terminal_state.json").write_text( + json.dumps({ + "schema": "loop-engineer/terminal@1", + "state": "Succeeded", + "criteria_met": {"gate": True}, + "evidence": [], + "false_completion": False, + "completion_policy": {"mode": "all_required"} if completion_policy is None else completion_policy, + }), + encoding="utf-8", + ) + return target + + +def _ready(tmp_path): + """A real task workspace positioned for its first dispatched iteration.""" + workspace = tmp_path / "dispatched" + emit.open_contract(workspace) + task = { + "id": "T-1", "title": "T-1", "status": "pending", "criterion_ref": "T-1", + "verify": "./scripts/verify-fast.sh", "depends_on": [], "attempts": 0, + "evidence": None, + } + (workspace / "TASKS.json").write_text( + json.dumps({"schema": "loop-engineer/tasks@1", "tasks": [task]}), encoding="utf-8" + ) + verifier = workspace / "scripts" / "verify-fast.sh" + verifier.parent.mkdir(parents=True, exist_ok=True) + verifier.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + verifier.chmod(0o755) + store = SQLiteEventStore(workspace / ".loop" / "events.db") + store.append("run-1", "contract_opened", {"workspace": workspace.name}, actor="test") + for state in ("plan", "critique-plan", "queue-tasks", "execute-task"): + store.append("run-1", "iteration_appended", { + "iteration_id": 0, "outcome": "replanned", "state": state, + }, actor="test") + return workspace + + +def _dispatched_workspace(tmp_path): + workspace = _ready(tmp_path) + dispatch_once(workspace) + return workspace + + +def _handwritten_record(workspace, *, name="evidence-handwritten.json"): + """Create self-consistent evidence that is only chain-bound in storeless workspaces.""" + bundle = workspace / ".loop" / "artifacts" / "verify-handwritten.json" + bundle.parent.mkdir(parents=True, exist_ok=True) + bundle_bytes = b'{"outcome": "PASS", "passed": true}' + bundle.write_bytes(bundle_bytes) + record = { + "schema": "loop-engineer/evidence@1", "id": "hand:1:verify", + "kind": "verify-bundle", "uri": ".loop/artifacts/verify-handwritten.json", + "sha256": hashlib.sha256(bundle_bytes).hexdigest(), "media_type": "application/json", + "produced_by": {"run_id": "run-1", "task_id": None, "attempt": 1, + "executor": "worker-a"}, + "created_at": "2026-07-25T00:00:00+00:00", + "verified_by": {"by": "ci", "at": "2026-07-25T00:00:00+00:00"}, + } + path = workspace / ".loop" / "evidence" / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(record, sort_keys=True) + "\n", encoding="utf-8") + return f".loop/evidence/{name}" + + +def _rewrite_terminal_evidence(workspace, evidence): + path = workspace / ".loop" / "terminal_state.json" + terminal = json.loads(path.read_text(encoding="utf-8")) + terminal["evidence"] = evidence + path.write_text(json.dumps(terminal, sort_keys=True) + "\n", encoding="utf-8") + + +def test_schema_id_and_predicate_type_are_pinned(): + from loop.verdict import PREDICATE_TYPE, VERDICT_SCHEMA_ID + + assert VERDICT_SCHEMA_ID == "loop-engineer/verdict@1" + assert PREDICATE_TYPE == "urn:loop-engineer:verdict:1" + + +def test_predicate_type_is_derived_from_the_schema_id(): + """The two constants are one identity in two encodings, not two names. + + ADR 0002 chose a URN matching the schema $id. A URN cannot idiomatically + carry '/' or '@', so the mapping is a transliteration -- which means + nothing stops the two from silently drifting apart unless it is asserted. + """ + from loop.verdict import PREDICATE_TYPE, VERDICT_SCHEMA_ID + + assert PREDICATE_TYPE == "urn:" + VERDICT_SCHEMA_ID.replace("/", ":").replace("@", ":") + + +def test_schema_file_declares_the_matching_id(): + from loop._resources import schemas_dir + from loop.verdict import VERDICT_SCHEMA_ID + + schema = json.loads((schemas_dir() / "verdict.schema.json").read_text(encoding="utf-8")) + assert schema["$id"] == VERDICT_SCHEMA_ID + assert schema["$schema"] == "https://json-schema.org/draft/2020-12/schema" + + +def test_verdict_schema_is_not_a_contract_artifact(): + # SCHEMA_IDS is the contract-object tuple (manifest/state/tasks/terminal). + # verdict@1 is a projection, not a contract object, and must stay out of it. + from loop.contract import SCHEMA_IDS + from loop.verdict import VERDICT_SCHEMA_ID + + assert VERDICT_SCHEMA_ID not in SCHEMA_IDS + + +def test_build_verdict_has_the_required_top_level_shape(tmp_path): + from loop.verdict import build_verdict + + verdict = build_verdict(_workspace_with_terminal(tmp_path)) + + assert set(verdict) == {"schema", "run_id", "tool", "doctor", "chain", "terminal", "evidence"} + assert verdict["schema"] == "loop-engineer/verdict@1" + assert verdict["tool"]["name"] == "loop-engineer" + assert verdict["evidence"] == [] + + +def test_build_verdict_projects_nonempty_normalized_doctor_issue_codes(tmp_path): + from loop.verdict import build_verdict + + target = _workspace_with_terminal(tmp_path) + (target / "scripts" / "verify-fast").unlink() + (target / "scripts" / "verify-full").unlink() + + doctor = build_verdict(target)["doctor"] + + assert set(doctor) == {"ok", "validation_mode", "issue_codes", "schemas_checked"} + assert doctor["issue_codes"] + assert doctor["issue_codes"] == sorted(set(doctor["issue_codes"])) + assert "unresolved_task_verify" in doctor["issue_codes"] + assert all(" " not in code and "/" not in code for code in doctor["issue_codes"]) + + +def test_build_verdict_projects_terminal_and_requires_terminal_record(tmp_path): + from loop.verdict import VerdictError, build_verdict + + target = _workspace_with_terminal(tmp_path) + terminal = build_verdict(target)["terminal"] + + assert set(terminal) == {"state", "completion_policy", "false_completion"} + assert terminal["completion_policy"] == "all_required" + + no_policy = _workspace_with_terminal(tmp_path, "no-policy", completion_policy=None) + (no_policy / ".loop" / "terminal_state.json").write_text( + json.dumps({"state": "Succeeded", "false_completion": False}), encoding="utf-8" + ) + assert build_verdict(no_policy)["terminal"]["completion_policy"] is None + + non_object_policy = _workspace_with_terminal(tmp_path, "non-object-policy", completion_policy="all_required") + assert build_verdict(non_object_policy)["terminal"]["completion_policy"] is None + + plain = tmp_path / "plain" + scaffold(plain) + with pytest.raises(VerdictError, match="no terminal record"): + build_verdict(plain) + with pytest.raises(VerdictError): + build_verdict(tmp_path / "missing") + + +def test_build_verdict_projects_chain_head_from_real_store(tmp_path): + from loop.verdict import build_verdict + + target = _workspace_with_terminal(tmp_path) + head = SQLiteEventStore(target / ".loop" / "events.db").append( + "run-1", "contract_opened", {"workspace": target.name}, actor="test" + ) + + chain = build_verdict(target)["chain"] + + assert chain["head"] is not None + assert re.fullmatch(r"[0-9a-f]{64}", chain["head"]) + assert chain["sequence"] == head["sequence"] + + +def test_build_verdict_handles_an_absent_event_store(tmp_path): + from loop.verdict import build_verdict + + verdict = build_verdict(_workspace_with_terminal(tmp_path)) + + assert verdict["chain"] == {"head": None, "sequence": None, "unchained_prefix": 0} + + +def test_build_verdict_degrades_for_an_unreadable_event_store(tmp_path): + from loop.verdict import build_verdict + + target = _workspace_with_terminal(tmp_path) + store_path = target / ".loop" / "events.db" + SQLiteEventStore(store_path).append( + "run-1", "contract_opened", {"workspace": target.name}, actor="test" + ) + store_path.write_bytes(b"not a SQLite database") + + verdict = build_verdict(target) + + assert verdict["chain"] == {"head": None, "sequence": None, "unchained_prefix": 0} + assert verdict["doctor"]["issue_codes"] + + +def test_build_verdict_rejects_terminal_without_false_completion(tmp_path): + from loop.verdict import VerdictError, build_verdict + + target = _workspace_with_terminal(tmp_path) + terminal_path = target / ".loop" / "terminal_state.json" + terminal = json.loads(terminal_path.read_text(encoding="utf-8")) + del terminal["false_completion"] + terminal_path.write_text(json.dumps(terminal), encoding="utf-8") + + with pytest.raises(VerdictError, match="false_completion"): + build_verdict(target) + + +# None is the original defect value: bool(None) is False, so the fail-open +# projection this guard replaced would have claimed "not a false completion" +# for a null flag. JSON null round-trips to None, so the key IS present and +# only the isinstance check stands between it and a signed false claim. +@pytest.mark.parametrize("value", ["false", 1, None]) +def test_build_verdict_rejects_non_boolean_false_completion(tmp_path, value): + from loop.verdict import VerdictError, build_verdict + + target = _workspace_with_terminal(tmp_path) + terminal_path = target / ".loop" / "terminal_state.json" + terminal = json.loads(terminal_path.read_text(encoding="utf-8")) + terminal["false_completion"] = value + terminal_path.write_text(json.dumps(terminal), encoding="utf-8") + + with pytest.raises(VerdictError, match="false_completion"): + build_verdict(target) + + +def test_build_verdict_reports_an_invalid_mode_as_a_contract_read_failure(tmp_path): + """An invalid mode= is a contract-read failure, not a path-resolution one. + + ValidationModeError subclasses RuntimeError, so a single guard around both + resolution and doctor_report would label it "cannot resolve a loop + workspace" -- a misleading frame for an argument error. + """ + from loop.verdict import VerdictError, build_verdict + + target = _workspace_with_terminal(tmp_path) + + with pytest.raises(VerdictError, match="cannot read the contract"): + build_verdict(target, mode="not-a-validation-mode") + + +def test_build_verdict_rejects_non_object_terminal_record(tmp_path): + from loop.verdict import VerdictError, build_verdict + + target = _workspace_with_terminal(tmp_path) + (target / ".loop" / "terminal_state.json").write_text("[]", encoding="utf-8") + + with pytest.raises(VerdictError, match="not an object"): + build_verdict(target) + + +def test_build_verdict_wraps_path_resolution_runtime_errors_as_verdict_error( + tmp_path, monkeypatch +): + import loop.verdict as verdict + from loop.verdict import VerdictError, build_verdict + + def raise_symlink_loop(_target): + raise RuntimeError("Symlink loop from 'x'") + + monkeypatch.setattr(verdict, "resolve_loop_paths", raise_symlink_loop) + + with pytest.raises(VerdictError): + build_verdict(tmp_path / "anything") + + +def test_build_verdict_validates_against_its_loaded_schema(tmp_path): + jsonschema = pytest.importorskip("jsonschema") + from loop.verdict import _load_verdict_schema, build_verdict + + schema = _load_verdict_schema() + verdict = build_verdict(_workspace_with_terminal(tmp_path)) + invalid_verdict = {**verdict, "chain": {**verdict["chain"], "head": "not-a-hash"}} + + assert schema["$id"] == "loop-engineer/verdict@1" + jsonschema.validate(verdict, schema) + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate(invalid_verdict, schema) + + +def test_evidence_carries_only_entries_that_pass_the_strict_bar(tmp_path, monkeypatch): + import loop.verdict as verdict + + monkeypatch.setattr(verdict, "_strict_evidence_failure", + lambda entry, _paths, _bound: None if entry == "good" else "bad") + monkeypatch.setattr(verdict, "_evidence_digests", lambda entry, _paths: { + "digest": "a" * 64, "code_digest": None, "policy_digest": None, + } if entry == "good" else {"digest": "b" * 64, "code_digest": None, + "policy_digest": None}) + target = _workspace_with_terminal(tmp_path) + _rewrite_terminal_evidence(target, ["good", "bad"]) + + assert verdict.build_verdict(target)["evidence"] == [ + {"digest": "a" * 64, "code_digest": None, "policy_digest": None} + ] + + +def test_evidence_entries_carry_digests_only(tmp_path): + from loop.verdict import build_verdict + + workspace = _dispatched_workspace(tmp_path) + emit.terminate(workspace, state="Succeeded", criteria_met={"C-1": True}, + evidence=[".loop/evidence/evidence-iter1.json"], + completion_policy=VERIFIED_EVIDENCE_MODE) + + evidence = build_verdict(workspace)["evidence"] + assert evidence + assert all(set(entry) == {"digest", "code_digest", "policy_digest"} for entry in evidence) + + +def test_evidence_is_sorted_by_digest(tmp_path, monkeypatch): + import loop.verdict as verdict + + digests = { + "later": {"digest": "f" * 64, "code_digest": None, "policy_digest": None}, + "first": {"digest": "0" * 64, "code_digest": None, "policy_digest": None}, + } + monkeypatch.setattr(verdict, "_strict_evidence_failure", lambda *_args: None) + monkeypatch.setattr(verdict, "_evidence_digests", lambda entry, _paths: digests[entry]) + target = _workspace_with_terminal(tmp_path) + _rewrite_terminal_evidence(target, ["later", "first"]) + + assert [entry["digest"] for entry in verdict.build_verdict(target)["evidence"]] == [ + "0" * 64, "f" * 64, + ] + + +def test_evidence_digest_is_the_chain_committed_record_digest(tmp_path): + from loop.runtime import bound_artifact_digests + from loop.verdict import build_verdict + + workspace = _dispatched_workspace(tmp_path) + entry = ".loop/evidence/evidence-iter1.json" + emit.terminate(workspace, state="Succeeded", criteria_met={"C-1": True}, evidence=[entry], + completion_policy=VERIFIED_EVIDENCE_MODE) + record_path = workspace / entry + record = json.loads(record_path.read_text(encoding="utf-8")) + + projected = build_verdict(workspace)["evidence"] + expected = hashlib.sha256(record_path.read_bytes()).hexdigest() + assert projected == [{"digest": expected, + "code_digest": record["verified_by"]["code_digest"], + "policy_digest": record["verified_by"]["policy_digest"]}] + assert expected == bound_artifact_digests(workspace)[entry][0] + assert expected != record["sha256"] + + +def test_unbound_record_is_excluded_when_a_store_exists(tmp_path): + from loop.verdict import build_verdict + + workspace = _dispatched_workspace(tmp_path) + bound_entry = ".loop/evidence/evidence-iter1.json" + emit.terminate(workspace, state="Succeeded", criteria_met={"C-1": True}, + evidence=[bound_entry], completion_policy=VERIFIED_EVIDENCE_MODE) + unbound_entry = _handwritten_record(workspace) + _rewrite_terminal_evidence(workspace, [bound_entry, unbound_entry]) + + evidence = build_verdict(workspace)["evidence"] + assert [entry["digest"] for entry in evidence] == [ + hashlib.sha256((workspace / bound_entry).read_bytes()).hexdigest() + ] + + +def test_absent_store_projects_evidence_under_the_documented_degradation(tmp_path): + from loop.verdict import build_verdict + + workspace = _workspace_with_terminal(tmp_path) + entry = _handwritten_record(workspace) + _rewrite_terminal_evidence(workspace, [entry]) + + assert build_verdict(workspace)["evidence"] == [{ + "digest": hashlib.sha256((workspace / entry).read_bytes()).hexdigest(), + "code_digest": None, "policy_digest": None, + }] + + +def test_unreadable_store_projects_no_evidence_and_does_not_raise(tmp_path): + from loop.verdict import build_verdict + + workspace = _workspace_with_terminal(tmp_path) + entry = _handwritten_record(workspace) + _rewrite_terminal_evidence(workspace, [entry]) + (workspace / ".loop" / "events.db").write_bytes(b"not a SQLite database") + + assert build_verdict(workspace)["evidence"] == [] + + +def test_identical_evidence_entries_project_once(tmp_path): + from loop.verdict import build_verdict + + workspace = _workspace_with_terminal(tmp_path) + entry = _handwritten_record(workspace) + _rewrite_terminal_evidence(workspace, [entry, entry]) + + assert build_verdict(workspace)["evidence"] == [{ + "digest": hashlib.sha256((workspace / entry).read_bytes()).hexdigest(), + "code_digest": None, "policy_digest": None, + }] + + +def test_projection_digest_must_match_the_chain_committed_digest(tmp_path, monkeypatch): + """The bar validates its own read of the record; the projection read must + re-match the chain-committed digest, or the entry is dropped (TOCTOU).""" + import loop.verdict as verdict + + workspace = _dispatched_workspace(tmp_path) + entry = ".loop/evidence/evidence-iter1.json" + emit.terminate(workspace, state="Succeeded", criteria_met={"C-1": True}, evidence=[entry], + completion_policy=VERIFIED_EVIDENCE_MODE) + monkeypatch.setattr(verdict, "_evidence_digests", lambda _entry, _paths: { + "digest": "f" * 64, "code_digest": None, "policy_digest": None, + }) + + assert verdict.build_verdict(workspace)["evidence"] == [] + + +def test_populated_evidence_validates_against_the_schema(tmp_path): + """A verdict carrying a non-empty evidence list satisfies verdict.schema.json.""" + jsonschema = pytest.importorskip("jsonschema") + from loop.verdict import _load_verdict_schema, build_verdict + + workspace = _dispatched_workspace(tmp_path) + emit.terminate(workspace, state="Succeeded", criteria_met={"C-1": True}, + evidence=[".loop/evidence/evidence-iter1.json"], + completion_policy=VERIFIED_EVIDENCE_MODE) + + verdict = build_verdict(workspace) + assert verdict["evidence"] + jsonschema.validate(verdict, _load_verdict_schema()) + + +def test_terminal_record_with_invalid_utf8_raises_verdict_error(tmp_path): + """An undecodable terminal file must surface as VerdictError, never a raw + UnicodeDecodeError. Today the doctor_report wrapper converts it (doctor + raises UnicodeDecodeError, a ValueError); if projection order ever changes, + _terminal_record's own guard becomes the conversion site - either way the + typed contract holds, so no message is pinned.""" + from loop.verdict import VerdictError, build_verdict + + target = _workspace_with_terminal(tmp_path) + (target / ".loop" / "terminal_state.json").write_bytes(b"\xff\xfe{}") + + with pytest.raises(VerdictError): + build_verdict(target) diff --git a/scripts/test_verdict_cli.py b/scripts/test_verdict_cli.py new file mode 100644 index 0000000..14940ac --- /dev/null +++ b/scripts/test_verdict_cli.py @@ -0,0 +1,129 @@ +"""CLI contract tests for the read-only ``loop verdict`` projection.""" + +import importlib.util +import json +import pathlib +import subprocess +import sys + +import pytest + + +@pytest.fixture +def repo_root() -> pathlib.Path: + return pathlib.Path(__file__).resolve().parent.parent + + +def _run(repo_root: pathlib.Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "-B", "-m", "loop", *args], + capture_output=True, + text=True, + cwd=repo_root, + ) + + +def test_verdict_emits_canonical_json_and_exits_zero(repo_root: pathlib.Path): + result = _run(repo_root, "verdict", "examples/flaky-test-triage") + assert result.returncode == 0, result.stderr + doc = json.loads(result.stdout) + assert doc["schema"] == "loop-engineer/verdict@1" + + +def test_verdict_output_is_byte_stable(repo_root: pathlib.Path): + first = _run(repo_root, "verdict", "examples/flaky-test-triage") + second = _run(repo_root, "verdict", "examples/flaky-test-triage") + assert first.returncode == second.returncode == 0 + assert first.stdout == second.stdout + + +def test_verdict_emits_no_statement_envelope(repo_root: pathlib.Path): + result = _run(repo_root, "verdict", "examples/flaky-test-triage") + assert result.returncode == 0, result.stderr + doc = json.loads(result.stdout) + assert not {"_type", "subject", "predicateType", "predicate"} & doc.keys() + + +def test_verdict_fails_loud_without_a_terminal_record(repo_root: pathlib.Path, tmp_path: pathlib.Path): + workspace = tmp_path / "workspace" + scaffolded = _run(repo_root, "scaffold", str(workspace)) + assert scaffolded.returncode == 0, scaffolded.stderr + result = _run(repo_root, "verdict", str(workspace)) + assert result.returncode == 2 + assert result.stdout == "" + assert "Traceback" not in result.stderr + assert "no terminal record" in result.stderr + assert result.stderr.startswith("verdict:") + + +def test_verdict_reports_a_typed_error_when_the_document_cannot_be_canonicalized( + repo_root: pathlib.Path, tmp_path: pathlib.Path +): + workspace = tmp_path / "workspace" + scaffolded = _run(repo_root, "scaffold", str(workspace)) + assert scaffolded.returncode == 0, scaffolded.stderr + terminal = json.dumps({"state": "Succeeded", "false_completion": False}) + (workspace / ".loop" / "terminal_state.json").write_text( + terminal.replace('"Succeeded"', "NaN"), encoding="utf-8" + ) + result = _run(repo_root, "verdict", str(workspace)) + assert result.returncode == 2 + assert result.stdout == "" + assert "Traceback" not in result.stderr + assert result.stderr.startswith("verdict:") + + +def test_verdict_appears_in_usage_and_help(repo_root: pathlib.Path): + help_result = _run(repo_root, "--help") + assert "verdict" in help_result.stdout + assert "Never signs and never verifies" in help_result.stdout + missing_target = _run(repo_root, "verdict") + assert missing_target.returncode == 2 + assert "verdict" in missing_target.stderr + + +def test_verdict_stdout_is_the_canonical_json_of_build_verdict(repo_root: pathlib.Path): + from loop.chain import canonical_json + from loop.verdict import build_verdict + + result = _run(repo_root, "verdict", "examples/flaky-test-triage") + assert result.returncode == 0, result.stderr + assert result.stdout == canonical_json(build_verdict("examples/flaky-test-triage")) + "\n" + + +def test_verdict_missing_target_exits_2_with_the_read_command_hint(repo_root: pathlib.Path, tmp_path: pathlib.Path): + result = _run(repo_root, "verdict", str(tmp_path / "missing")) + assert result.returncode == 2 + assert "target path does not exist" in result.stderr + + +def test_verdict_rejects_an_invalid_mode_value(repo_root: pathlib.Path): + result = _run(repo_root, "verdict", "--mode", "bogus", "examples/flaky-test-triage") + assert result.returncode == 2 + assert "invalid --mode value" in result.stderr + + +def test_verdict_mode_flag_reaches_the_projection(repo_root: pathlib.Path): + basic = _run(repo_root, "verdict", "--mode", "basic", "examples/flaky-test-triage") + assert basic.returncode == 0, basic.stderr + basic_doc = json.loads(basic.stdout) + assert basic_doc["doctor"]["validation_mode"] == "structural-fallback" + + default = _run(repo_root, "verdict", "examples/flaky-test-triage") + assert default.returncode == 0, default.stderr + default_doc = json.loads(default.stdout) + has_jsonschema = importlib.util.find_spec("jsonschema") is not None + if has_jsonschema: + assert basic_doc["doctor"]["validation_mode"] != default_doc["doctor"]["validation_mode"] + + +def test_verdict_rejects_expect_chain_head(repo_root: pathlib.Path): + result = _run( + repo_root, + "verdict", + "--expect-chain-head", + "a" * 64, + "examples/flaky-test-triage", + ) + assert result.returncode == 2 + assert "only valid for doctor/validate/verify" in result.stderr diff --git a/scripts/test_verdict_purity.py b/scripts/test_verdict_purity.py new file mode 100644 index 0000000..7012c6f --- /dev/null +++ b/scripts/test_verdict_purity.py @@ -0,0 +1,111 @@ +"""The ADR 0002 boundary, made mechanical. + +These tests exist so the kernel/signer boundary cannot decay into an intention: +the kernel may hash, it must never sign, never read the environment, and never +emit anything but the allowlisted, prose-free predicate body. +""" +import ast +import json +import pathlib +import subprocess +import sys + +import pytest + +REPO = pathlib.Path(__file__).resolve().parent.parent +LOOP = REPO / "loop" + +_SIGNING_TOKENS = ("sigstore", "cosign", "fulcio", "rekor", "dsse", + "private_key", "PRIVATE KEY", "ACTIONS_ID_TOKEN", + "id_token", "oidc") + + +def test_kernel_never_references_a_signing_stack(): + """ADR 0002: the kernel may hash; it must never sign.""" + offenders = [] + for path in LOOP.rglob("*.py"): + text = path.read_text(encoding="utf-8").lower() + for token in _SIGNING_TOKENS: + if token.lower() in text: + offenders.append(f"{path.relative_to(REPO)}:{token}") + assert offenders == [], offenders + + +def test_kernel_reads_no_environment_variable(): + """Zero matches at HEAD 0025acc; this keeps it that way. A verdict that + read GITHUB_SHA would put a vendor identifier inside the portable layer.""" + offenders = [] + for path in LOOP.rglob("*.py"): + text = path.read_text(encoding="utf-8") + if "os.environ" in text or "getenv" in text: + offenders.append(str(path.relative_to(REPO))) + assert offenders == [], offenders + + +def test_verdict_module_imports_only_stdlib_and_loop(): + tree = ast.parse((LOOP / "verdict.py").read_text(encoding="utf-8")) + third_party = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + if node.level == 0 and node.module and node.module.split(".")[0] not in sys.stdlib_module_names: + third_party.append(node.module) + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name.split(".")[0] not in sys.stdlib_module_names: + third_party.append(alias.name) + assert third_party == [], third_party + + +def _predicate(): + proc = subprocess.run( + [sys.executable, "-B", "-m", "loop", "verdict", "examples/flaky-test-triage"], + capture_output=True, text=True, cwd=REPO) + assert proc.returncode == 0, proc.stderr + return json.loads(proc.stdout) + + +def test_predicate_field_allowlist_holds(): + """Everything here is public, append-only, and permanent. Adding a field is + a one-way door — this test is the door.""" + doc = _predicate() + assert set(doc) == {"schema", "run_id", "tool", "doctor", "chain", "terminal", "evidence"} + assert set(doc["doctor"]) == {"ok", "validation_mode", "issue_codes", "schemas_checked"} + assert set(doc["chain"]) == {"head", "sequence", "unchained_prefix"} + assert set(doc["terminal"]) == {"state", "completion_policy", "false_completion"} + for entry in doc["evidence"]: + assert set(entry) == {"digest", "code_digest", "policy_digest"} + + +def test_predicate_carries_no_free_text_but_run_id(): + """run_id is the ONE operator-controlled string, allowlisted deliberately. + Any other prose in the document is a leak into a permanent public log.""" + doc = _predicate() + strings = [] + + def walk(node, path): + if isinstance(node, dict): + for k, v in node.items(): + walk(v, f"{path}.{k}") + elif isinstance(node, list): + for i, v in enumerate(node): + walk(v, f"{path}[{i}]") + elif isinstance(node, str): + strings.append((path, node)) + + walk(doc, "") + for path, value in strings: + if path == ".run_id": + continue + # Whitespace is the proxy for prose. Every other string in the document + # is a digest, an enum, a snake_case issue code, or a slash-separated + # schema id — none of which contain a space. + assert " " not in value, f"free text at {path}: {value!r}" + + +def test_predicate_validates_against_its_own_schema(): + # importorskip, not bare __import__: in the structural-fallback environment + # this must skip honestly rather than error. + jsonschema = pytest.importorskip("jsonschema") + from loop.verdict import _load_verdict_schema + + jsonschema.validate(_predicate(), _load_verdict_schema())