# Validate mechanics

Loaded by `SKILL.md` at the manifest step (helper commands), at cross-family
dispatch (adapters), and at scope disclosure (the homes table). `$SKILL_DIR`
is the directory containing `SKILL.md`: `skills/validate/` in a repository
checkout, `.agents/skills/validate/` in an installed runtime.

## Helper commands

Installed mechanics run through `ao provenance` (evidence helper version 1),
using `subject-manifest.v1` and `verdict.v2` unchanged. A fresh Validate agent
supplies the semantic result; Go computes identities, verifies structure and
stores the supplied result. No Python interpreter runs on this path.

| Command | Required | Optional |
|---|---|---|
| `manifest` | `--root <dir>`, `--include <path>` (repeatable) | `--exclude <path-or-glob>` (repeatable), `--base-manifest <file>`, `--git-metadata-json <json>`, `--out <relative-file>` with `--evidence-root <dir>` |
| `verify-manifest` | `--root <dir>`, `--manifest <file>` | `--base-manifest <file>` |
| `snapshot-intent` | `--source <file>` (`-` reads stdin), `--evidence-root <dir>` | none |
| `digest` | `<json-file>` positional | `--json` |
| `store-verdict` | `--root`, `--evidence-root`, `--draft`, `--intent-source`, `--subject-manifest`, `--author-context-id`, `--validator-context-id`, `--freshness-source <runtime\|caller>`, `--freshness-attester-id`, `--scope-result <PASS\|FAIL\|NOT_PROVEN>` | `--base-manifest <file>` |
| `verify-verdict` | `--verdict <digest.json>` | none |
| `verify-subject` | `--root <dir>`, `--manifest <file>`, `--verdict <digest.json>`, `--intent <file>` | `--base-manifest <file>` |

Every leaf accepts `--helper-version 1`; an incompatible version fails before
mutation. Evidence operations emit JSON by default except `digest`, which prints
the digest; `--json` requests JSON explicitly and global `--output`/`-o` selects
JSON or YAML formatting. Conflicting explicit formats fail before any write.
Exit 0 means the mechanical operation completed. Invalid input, failed
verification or filesystem errors exit 1. A stored `FAIL` or `NOT_PROVEN` may
complete storage successfully; that exit status never means semantic PASS.
`--dry-run` rejects evidence writes before mutation. `ao capabilities` carries
the actual family and leaf argument, output, effect and exit contracts.

```sh
ao provenance manifest --root . --include skills/validate \
  --exclude '**/*.log' --evidence-root "$EVIDENCE_ROOT" --out manifest.json
```

`manifest` uses only filesystem content. Symlinks bind target bytes without
following directory symlinks; executable bits and deletions bind identity.
Optional Git metadata is descriptive and excluded from identity. Unknown fields,
duplicate JSON keys, malformed paths and canonical digest mismatches fail closed.
`verify-manifest` recomputes identity and requires the matching base for deletions.
Version 1 retains the reference's asymmetric root rule: live files and symlinks
use literal include roots, while deletion selection and structural membership
use the historical filename-pattern match against the base. Verification still
requires that exact base and recomputes the complete manifest.

`store-verdict` verifies a nonempty manifest against the current subject before
storage, binds exact intent bytes and explicit runtime identities/freshness/scope,
and validates the resulting artifact using the same strict reader as `ao status`.
Author/judge collision, missing runtime facts, incomplete scope or a PASS with
unverified acceptance cannot persist an admitted PASS. Proven scope failure
forces FAIL; integrity gaps retain NOT_PROVEN with `validate.integrity` findings.
The caller still owns deriving complete changed-path coverage and freshness;
these helper inputs are attestations, not independently discovered runtime facts.

The artifact digest is SHA-256 over canonical JSON with `artifact_digest`
omitted. A synced private temporary file is atomically published without
replacing an existing address; directory durability uses the shared storage
barrier. Identical existing bytes are idempotent. Conflicting verdict bytes
remain intact and produce a separate NOT_PROVEN integrity artifact. Intent
snapshot collisions fail. Explicit manifest outputs likewise never overwrite
different existing bytes. Existing standalone proof is preserved by its owner.

## Compatibility mapping and explicit evidence routing

The old `validate.py` commands map to the same names under `ao provenance`.
`manifest`, `verify-manifest` and `digest` keep their identity contracts.
Python's manifest `--output` maps to Go's `--out`, a relative file within explicit
`--evidence-root`. AO's existing global `--output`/`-o` remains the output format;
it never names a destination file.
`snapshot-intent` replaces `--workspace`/`--intent-dir` defaults with a required
`--evidence-root`. `store-verdict` replaces `--workspace`/`--verdict-dir` with that
same explicit root and adds required `--root` to verify current subject bytes.
Its other runtime-fact flags retain their meanings. Unsupported legacy flags
fail before storage; no wrapper silently uses the old workspace default.

The evidence root must already exist outside ordinary, bare and linked Git
repositories, including symlink aliases. All output branches are checked before
any directory or temporary-file write. Intents go under
`<root>/intents/sha256/<digest>.intent`, verdicts under
`<root>/verdicts/sha256/<digest>.json`, and explicit manifest outputs stay under
that same root. Missing or invalid roots fail with no workspace fallback.
The guard also rejects split common storage exposing `objects` and `refs` even
when `HEAD` lives elsewhere. Before writes it resolves active `GIT_DIR`,
`GIT_COMMON_DIR`, `GIT_OBJECT_DIRECTORY`, `GIT_ALTERNATE_OBJECT_DIRECTORIES`,
`GIT_WORK_TREE`, and the parent of `GIT_INDEX_FILE`. `GIT_DIR` must resolve to a
directory; its optional `commondir` pointer is followed when `GIT_COMMON_DIR` is
not supplied. Known common-directory `objects`, `refs` and `logs` symlinks are
resolved too. Relative environment paths are relative to the invocation's
working directory; a relative `commondir` pointer is relative to `GIT_DIR`.
Alternate environment paths support Git's C-quoted path-list syntax.

Every storage caller (`snapshot-intent`, `manifest --out`, `store-verdict`)
accepts repeatable `--exclude-git-root <existing-dir>` for additional caller-known
Git storage. It is passed through every preflight and publication check. A root
that contains or is contained by a declared boundary is rejected, including
canonical aliases. Missing, malformed or denied required bindings/exclusions
fail before any write; the helper never initializes a replacement directory.
A not-yet-created `GIT_INDEX_FILE` requires an existing resolvable parent, which
is excluded as a directory. Fixed Git path bindings are environment inputs, not
AO configuration resolution, and require no Git executable.

An unmarked directory referenced by an unrelated repository cannot prove the
absence of Git storage through ancestry alone. There is no universal reverse
lookup of repository configuration or alternates files: callers must supply
known external storage roots not represented by the active bindings. Missing
knowledge remains a caller boundary, not a claim that all possible external Git
references were discovered. The guard does not establish runtime authorization.

Destination descendants cannot be symlinks. The guard is a filesystem check;
native access controls still own confidentiality and hostile concurrent writers.

For CDLC knowledge/disclosure review, the caller resolves the protected external
`context.evidence_root` and passes it explicitly. These generic helpers do not
read configuration; T11 owns routing through T05. Drafts, manifests, receipts
and diagnostics also belong in that protected destination by caller policy.
Standalone product-proof placement remains explicitly caller-selected.

`verify-subject` compares current subject identity and the supplied verdict to
an independently supplied immutable `--intent`. Use distinct expected acceptance
for factual-support and destination-disclosure review; require every selected
leg to bind both identities. Pin any required profile version and policy in
those immutable bytes. No format-specific `--profile` validator is advertised;
structural validity, semantic factual support, destination permission and later
usefulness remain separate questions. The candidate cannot choose its own
expected policy. Evidence references remain declared strings, not verified
citations. Read permission does not authorize model transmission or Git ingestion.

## Developer-only reference checks

`tests/validate.py` retains the independent Python reference mechanics;
`tests/test_validate.py` and `tests/check_contract_corpus.py` keep their schema
and cross-language coverage. Run `bash skills/validate/tests/validate.sh` and
`bash scripts/check-verdict-contract-corpus.sh` in the development environment.
The installed `scripts/validate.sh` only checks the skill's contract text.
`tests/test_evidence_cli.py`, with an explicit source-built `AO_BIN`, exercises
candidate evidence operations with an empty runtime PATH. RPI/swarm Python
modules remain developer references; native skill execution does not invoke them.

## Proportionate fresh checks

Apply the owning skill's fresh same-family default. Risk sizes evidence depth;
only caller selection requires a different model family. Preserve an explicitly
requested leg until the caller changes it. Every mode retains exact subject,
full acceptance, evidence for every criterion, and the empty-`not_checked` bar.

Reuse existing digest-bound check receipts when their subject, inputs, tool
identity, and claimed criterion still match. Rerun the fast discriminating
check for a changed or uncertain criterion; rerun broader checks when the change
invalidates their receipts or acceptance explicitly requires them. A new receipt
label, changed digest, reduced finding count, or repeated review is not useful
progress without evidence that a named acceptance gap closed. Reuse the current
findings/evidence fields for causal comparisons; create no progress ledger.

## Cross-family adapters

Use the single [agent-native model-dispatch recipe](../../agent-native/references/model-dispatch.md)
for caller selection, host authorization and bounded invocation. The fresh
same-family leg and any explicitly selected cross-family leg receive independent
initial inputs. Time bounds come from the caller or native deadline, with no
fixed ten-minute cap. A judge reads and judges; it never mutates the subject. Record
actual author/judge model and context identities in protected evidence refs
and freshness attestation notes; the `verdict.v2` schema is unchanged.
Transport, output, exit and process completion are facts, not semantic PASS.

## Where each scope limit lives inside a PASS

| Scope limit | Home | Example |
|---|---|---|
| A criterion proven by a bounded check | `criteria[].reason` on that criterion | "proven by the unit suite; the full integration matrix was not replayed" |
| A declared non-goal or out-of-scope area | the intent source's non-goals, optionally restated as an evidence-backed boundary criterion in `criteria` | "`cli/**` is a declared non-goal; the diff proves it untouched" |
| Residual risk or judgment caveat | the caller-facing report | "the migration path is untested against pre-3.0 stores" |
| Acceptance that genuinely went unverified | `not_checked`, and the result is `NOT_PROVEN` rather than PASS | "criterion 3 needs hardware this context cannot reach" |

#!/bin/sh
# Installed contract checks. Python/schema/reference tests live in ../tests.
set -eu
skill_dir=$(CDPATH='' cd "$(dirname "$0")/.." && pwd)
grep -q '^name: validate$' "$skill_dir/SKILL.md"
grep -Fq 'PASS`, `FAIL`, or `NOT_PROVEN`' "$skill_dir/SKILL.md"
grep -Fq 'sole semantic author of `verdict.v2`' "$skill_dir/SKILL.md"
grep -Fq 'Only when the caller requests machine-readable evidence' "$skill_dir/SKILL.md"
grep -Fq 'nonempty implementation candidate' "$skill_dir/SKILL.md"
echo 'validate installed skill contract: PASS'

#!/usr/bin/env python3
"""Run the shared verdict-contract golden corpus through the Python validator
and (when jsonschema is available) the canonical JSON schema.

The same cases run through the Go reader (cli/internal/verdictcheck
TestGoldenCorpus). Any disagreement between the three implementations is a
contract fork and must fail CI.

Exit 0: every case matches its expected outcome.
Exit 1: at least one implementation disagrees with the corpus.
"""
from __future__ import annotations

import importlib.util
import json
import os
import pathlib
import sys

ROOT = pathlib.Path(__file__).resolve().parents[3]
CASES = ROOT / "tests" / "fixtures" / "verdict-contract" / "cases"
SCHEMA = ROOT / "schemas" / "verdict.v2.schema.json"


def load_validate_module():
    path = pathlib.Path(__file__).with_name("validate.py")
    spec = importlib.util.spec_from_file_location("validate_corpus_subject", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
    """object_pairs_hook that fails closed on a duplicate key at any depth.

    Python's default json decode is last-wins (like Go's map decode), so a
    duplicated key would silently hide the real value and let a payload bind a
    digest its bytes never canonicalize to. The Go reader
    (cli/internal/verdictcheck) rejects the same class; this keeps the Python
    leg of the cross-language corpus in agreement.
    """
    seen: set[str] = set()
    for key, _ in pairs:
        if key in seen:
            raise ValueError(f"duplicate key: {key}")
        seen.add(key)
    return dict(pairs)


def python_verdict(module, case) -> tuple[bool, str]:
    raw = case.get("raw")
    if raw is not None:
        # The Python storage layer parses exactly one JSON document; simulate
        # its read of a payload with trailing data, and fail closed on any
        # duplicate key (mirrors the Go reader).
        try:
            decoder = json.JSONDecoder(object_pairs_hook=_reject_duplicate_keys)
            value, end = decoder.raw_decode(raw)
            if raw[end:].strip():
                return False, "trailing data"
            artifact = value
        except json.JSONDecodeError as exc:
            return False, f"parse: {exc}"
        except ValueError as exc:
            return False, str(exc)
    else:
        artifact = case["artifact"]
    try:
        module.validate_verdict_v2(artifact)
    except Exception as exc:  # ContractError or shape errors
        return False, str(exc)
    # Filename binding: stored artifacts are addressed by artifact_digest.
    if artifact.get("artifact_digest") != case["filename_digest"]:
        return False, "artifact_digest does not match filename"
    return True, ""


def schema_verdict(validator, case) -> tuple[bool, str]:
    raw = case.get("raw")
    if raw is not None:
        try:
            decoder = json.JSONDecoder()
            value, end = decoder.raw_decode(raw)
            if raw[end:].strip():
                return False, "trailing data"
        except json.JSONDecodeError as exc:
            return False, f"parse: {exc}"
        artifact = value
    else:
        artifact = case["artifact"]
    errors = sorted(validator.iter_errors(artifact), key=lambda e: e.json_path)
    if errors:
        return False, errors[0].message
    return True, ""


def main() -> int:
    module = load_validate_module()

    require_schema = os.environ.get("CONTRACT_CORPUS_REQUIRE_SCHEMA") == "1"
    validator = None
    try:
        import jsonschema

        schema = json.loads(SCHEMA.read_text())
        validator = jsonschema.Draft202012Validator(schema)
    except ImportError:
        if require_schema:
            print("check-contract-corpus: FAIL — jsonschema unavailable but the "
                  "schema leg is required (CONTRACT_CORPUS_REQUIRE_SCHEMA=1)", file=sys.stderr)
            return 1
        print("check-contract-corpus: jsonschema unavailable — schema leg skipped", file=sys.stderr)

    failures = []
    cases = sorted(CASES.glob("*.json"))
    if len(cases) < 10:
        print(f"check-contract-corpus: FAIL — suspiciously small corpus ({len(cases)} cases)")
        return 1
    for path in cases:
        case = json.loads(path.read_text())
        expected_valid = case["expected"] == "valid"

        ok, reason = python_verdict(module, case)
        if ok != expected_valid:
            failures.append(f"{case['name']}: python validator said {'valid' if ok else 'invalid'} "
                            f"({reason or 'no error'}), corpus expects {case['expected']}")

        if validator is not None:
            ok, reason = schema_verdict(validator, case)
            if expected_valid and not ok:
                failures.append(f"{case['name']}: schema rejected a valid case: {reason}")
            if not expected_valid and ok and not case.get("schema_lenient"):
                failures.append(f"{case['name']}: schema accepted an invalid case "
                                f"(mark schema_lenient only when JSON Schema cannot express the rule)")

    if failures:
        print("check-contract-corpus: FAIL — contract implementations disagree:")
        for failure in failures:
            print(f"  {failure}")
        return 1
    legs = "python+schema" if validator is not None else "python"
    print(f"check-contract-corpus: PASS ({len(cases)} cases, {legs})")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Developer-only installed-binary evidence test; AO_BIN selects the candidate.

The driver uses Python, but every candidate process runs with an empty PATH.
No reference helper or Python executable is available on that runtime path.
"""
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest


class EvidenceCLI(unittest.TestCase):
    def test_git_storage_boundaries_before_writes(self):
        candidate = Path(os.environ["AO_BIN"]).resolve(strict=True)
        with tempfile.TemporaryDirectory() as temporary:
            base = Path(temporary)
            subject = base / "subject"
            inputs = base / "inputs"
            empty_path = base / "empty-path"
            for directory in (subject, inputs, empty_path):
                directory.mkdir()
            (subject / "value").write_text("candidate\n")
            (inputs / "intent").write_text("independent acceptance\n")
            draft = {"verdict": "PASS", "criteria": [{"id": "c", "result": "PASS", "evidence_refs": ["receipt"]}],
                     "findings": [], "evidence_refs": ["receipt"], "checked": ["value"], "not_checked": [], "validated_at": "2026-07-14T00:00:00Z"}
            (inputs / "draft.json").write_text(json.dumps(draft))
            env = dict(os.environ, PATH=str(empty_path))
            for key in ("GIT_DIR", "GIT_COMMON_DIR", "GIT_OBJECT_DIRECTORY", "GIT_ALTERNATE_OBJECT_DIRECTORIES", "GIT_WORK_TREE", "GIT_INDEX_FILE"):
                env.pop(key, None)

            def invoke(args, active_env):
                return subprocess.run([str(candidate), "provenance", *map(str, args)], cwd=subject, env=active_env, text=True, capture_output=True, timeout=30)

            manifest = invoke(["manifest", "--root", subject, "--include", "value"], env)
            self.assertEqual(manifest.returncode, 0, manifest.stderr)
            (inputs / "manifest.json").write_text(manifest.stdout)

            def fingerprint(root):
                result = {}
                for p in [root, *root.rglob("*")]:
                    mode = p.lstat().st_mode
                    content = os.readlink(p) if p.is_symlink() else p.read_bytes() if p.is_file() else None
                    result[str(p.relative_to(root))] = (mode, content)
                return result

            checked = 0
            for kind in ("objects-env", "split-common-env", "commondir-pointer", "common-markers", "declared", "missing-binding", "missing-exclusion"):
                area = base / kind
                area.mkdir()
                pool = area / "pool"
                pool.mkdir()
                active = dict(env)
                flags = []
                if kind == "objects-env":
                    active["GIT_OBJECT_DIRECTORY"] = str(pool)
                elif kind in ("split-common-env", "commondir-pointer", "common-markers"):
                    (pool / "objects").mkdir()
                    (pool / "refs").mkdir()
                    (pool / "config").write_text("[core]\nrepositoryformatversion = 0\n")
                    if kind != "common-markers":
                        admin = area / "admin"
                        admin.mkdir()
                        (admin / "HEAD").write_text("ref: refs/heads/main\n")
                        active["GIT_DIR"] = str(admin)
                        if kind == "split-common-env":
                            active["GIT_COMMON_DIR"] = str(pool)
                        else:
                            (admin / "commondir").write_text("../pool\n")
                elif kind == "declared":
                    flags = ["--exclude-git-root", pool]
                elif kind == "missing-binding":
                    active["GIT_OBJECT_DIRECTORY"] = str(area / "missing")
                else:
                    flags = ["--exclude-git-root", area / "missing"]
                (pool / "child").mkdir()
                alias = area / "alias"
                alias.symlink_to(pool, target_is_directory=True)
                for destination in (pool, pool / "child", alias, alias / "child"):
                    calls = [
                        ["snapshot-intent", "--source", inputs / "intent"],
                        ["manifest", "--root", subject, "--include", "value", "--out", "new/man.json"],
                        ["store-verdict", "--root", subject, "--subject-manifest", inputs / "manifest.json", "--draft", inputs / "draft.json",
                         "--intent-source", inputs / "intent", "--author-context-id", "author", "--validator-context-id", "judge",
                         "--freshness-source", "runtime", "--freshness-attester-id", "test", "--scope-result", "PASS"],
                    ]
                    for args in calls:
                        before = fingerprint(area)
                        completed = invoke([*args, "--evidence-root", destination, *flags], active)
                        self.assertNotEqual(completed.returncode, 0, (kind, args, completed.stdout))
                        expected_error = {"missing-binding": "resolve GIT_OBJECT_DIRECTORY", "missing-exclusion": "resolve --exclude-git-root"}.get(kind, "Git storage")
                        self.assertIn(expected_error, completed.stderr)
                        self.assertEqual(before, fingerprint(area), (kind, args, "mutated before rejection"))
                        checked += 1
            # A declared unrelated pool does not block a valid external root.
            external = base / "external"
            external.mkdir()
            completed = invoke(["snapshot-intent", "--source", inputs / "intent", "--evidence-root", external,
                                "--exclude-git-root", base / "declared" / "pool"], env)
            self.assertEqual(completed.returncode, 0, completed.stderr)
            print(f"Git storage boundaries: {checked} installed no-write rejections; aliases/descendants; empty PATH; unrelated root admitted")

    def test_python_reference_identity_parity(self):
        candidate = Path(os.environ["AO_BIN"]).resolve(strict=True)
        spec = importlib.util.spec_from_file_location("evidence_reference", Path(__file__).with_name("validate.py"))
        reference = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(reference)
        with tempfile.TemporaryDirectory() as temporary:
            base = Path(temporary)
            root = base / "subject"
            root.mkdir()
            (root / "nested").mkdir()
            (root / "nested" / "value").write_text("canonical \u2028 separator\n")
            (root / "nested" / "value").chmod(0o700)
            (root / "nested" / "skip.log").write_text("excluded")
            (root / "link").symlink_to("nested/value")
            empty = base / "empty-path"
            empty.mkdir()
            env = dict(os.environ, PATH=str(empty))

            def run(*args):
                completed = subprocess.run([str(candidate), "provenance", *map(str, args), "--json"], cwd=root, env=env, text=True, capture_output=True, timeout=30)
                self.assertEqual(completed.returncode, 0, completed.stderr)
                return json.loads(completed.stdout)

            expected = reference.build_manifest(root, ["."], ["**/*.log"], git_metadata={"commit": "descriptive"})
            actual = run("manifest", "--root", root, "--include", ".", "--exclude", "**/*.log", "--git-metadata-json", '{"commit":"descriptive"}')
            self.assertEqual(actual, expected)
            manifest_file = base / "base.json"
            manifest_file.write_text(json.dumps(actual))
            (root / "nested" / "value").unlink()
            expected_deletion = reference.build_manifest(root, ["."], ["**/*.log"], expected)
            actual_deletion = run("manifest", "--root", root, "--include", ".", "--exclude", "**/*.log", "--base-manifest", manifest_file)
            self.assertEqual(actual_deletion, expected_deletion)
            value_file = base / "value.json"
            # Preserve raw numeric spellings so both decoders canonicalize them.
            value_file.write_text('{"a":1e2,"b":1e-5,"c":1e16,"d":-0.0,"e":-0,"f":123456789012345678901234567890,"separator":"\u2028"}')
            actual_digest = run("digest", value_file)["digest"]
            self.assertEqual(actual_digest, reference.digest_value(json.loads(value_file.read_text())))
            print("Python/Go parity: manifest, symlink/executable bits, exclusions, deletions, metadata independence, numeric/Unicode canonical digest")

    def test_installed_evidence_without_python(self):
        candidate = Path(os.environ["AO_BIN"]).resolve(strict=True)
        with tempfile.TemporaryDirectory() as temporary:
            base = Path(temporary)
            consumer = base / "consumer"
            protected = base / "protected"
            empty_path = base / "empty-path"
            for directory in (consumer, protected, empty_path):
                directory.mkdir()
            # Synthetic Git fixture: the helper must leave files, index and
            # object bytes untouched. It never needs to invoke Git.
            (consumer / ".git" / "objects").mkdir(parents=True)
            (consumer / ".git" / "refs").mkdir()
            (consumer / ".git" / "HEAD").write_text("ref: refs/heads/main\n")
            (consumer / ".git" / "index").write_bytes(b"synthetic index sentinel")
            (consumer / ".git" / "objects" / "sentinel").write_bytes(b"synthetic object sentinel")
            (consumer / "value").write_bytes(b"candidate\n")
            before = {str(p.relative_to(consumer)): p.read_bytes() for p in consumer.rglob("*") if p.is_file()}
            env = dict(os.environ, PATH=str(empty_path))
            commands = []

            def run(*args, ok=True):
                commands.append([str(candidate), "provenance", *map(str, args)])
                completed = subprocess.run(commands[-1], cwd=consumer, env=env, capture_output=True, text=True, timeout=30)
                self.assertEqual(completed.returncode == 0, ok, completed.stderr)
                return json.loads(completed.stdout) if ok else completed.stderr

            manifest = run("manifest", "--root", consumer, "--include", "value", "--evidence-root", protected, "--out", "manifest.json", "--json")
            run("verify-manifest", "--root", consumer, "--manifest", protected / "manifest.json", "--json")
            draft = {
                "verdict": "PASS",
                "criteria": [{"id": "criterion", "result": "PASS", "evidence_refs": ["synthetic:receipt"]}],
                "findings": [], "evidence_refs": ["synthetic:receipt"],
                "checked": ["value"], "not_checked": [], "validated_at": "2026-07-14T00:00:00Z",
            }
            (protected / "draft.json").write_text(json.dumps(draft))
            verdicts = []
            for purpose in ("factual-support", "destination-disclosure"):
                intent = protected / (purpose + ".intent")
                payload = (purpose + ": independent immutable acceptance\n").encode()
                intent.write_bytes(payload)
                snap = run("snapshot-intent", "--source", intent, "--evidence-root", protected, "--json")
                self.assertEqual(snap["acceptance_digest"], hashlib.sha256(payload).hexdigest())
                self.assertEqual(Path(snap["intent_ref"]).read_bytes(), payload)
                args = ["store-verdict", "--root", consumer, "--evidence-root", protected,
                        "--draft", protected / "draft.json", "--subject-manifest", protected / "manifest.json",
                        "--intent-source", intent, "--author-context-id", "author", "--validator-context-id", "judge",
                        "--freshness-source", "runtime", "--freshness-attester-id", "test-runtime", "--scope-result", "PASS", "--json"]
                stored = run(*args)
                self.assertEqual(stored["verdict"], "PASS")
                self.assertTrue(run(*args)["idempotent"])
                run("verify-verdict", "--verdict", stored["path"], "--json")
                run("verify-subject", "--root", consumer, "--manifest", protected / "manifest.json", "--verdict", stored["path"], "--intent", intent, "--json")
                verdicts.append(stored)
            self.assertNotEqual(verdicts[0]["acceptance_digest"], verdicts[1]["acceptance_digest"])
            run("verify-subject", "--root", consumer, "--manifest", protected / "manifest.json", "--verdict", verdicts[0]["path"], "--intent", protected / "destination-disclosure.intent", ok=False)
            for destination in (consumer, consumer / "missing", base / "missing"):
                run("snapshot-intent", "--source", protected / "factual-support.intent", "--evidence-root", destination, ok=False)
            run("snapshot-intent", "--source", protected / "factual-support.intent", ok=False)
            run("snapshot-intent", "--source", protected / "factual-support.intent", "--evidence-root", protected, "--helper-version", "unsupported", ok=False)
            subject = consumer / "value"
            original_mode = subject.stat().st_mode & 0o777
            verify = ["verify-subject", "--root", consumer, "--manifest", protected / "manifest.json", "--verdict", verdicts[0]["path"], "--intent", protected / "factual-support.intent"]
            subject.chmod(original_mode ^ 0o100)
            run(*verify, ok=False)
            subject.chmod(original_mode)
            subject.write_bytes(b"changed bytes")
            run(*verify, ok=False)
            subject.unlink()
            subject.symlink_to(protected / "factual-support.intent")
            run(*verify, ok=False)
            subject.unlink()
            subject.write_bytes(before["value"])
            subject.chmod(original_mode)
            after = {str(p.relative_to(consumer)): p.read_bytes() for p in consumer.rglob("*") if p.is_file()}
            self.assertEqual(before, after)
            self.assertFalse((consumer / ".agents").exists())
            self.assertFalse((base / "missing").exists())
            print(f"candidate evidence: {len(commands)} operations; empty PATH; external intent/manifest/verdict bytes; consumer/index/objects unchanged")


if __name__ == "__main__":
    unittest.main()

from __future__ import annotations

import importlib.util
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

import jsonschema


SPEC = importlib.util.spec_from_file_location("validate_tool", Path(__file__).with_name("validate.py"))
tool = importlib.util.module_from_spec(SPEC)
assert SPEC.loader
SPEC.loader.exec_module(tool)


class ValidateV2Tests(unittest.TestCase):
    def draft(self):
        return {
            "acceptance_digest": "a" * 64,
            "subject_manifest_digest": "b" * 64,
            "author_context_id": "author",
            "validator_context_id": "validator",
            "freshness_attestation": {"source": "runtime", "attester_identity": "runtime-1"},
            "verdict": "PASS",
            "criteria": [{"id": "c1", "result": "PASS", "evidence_refs": ["e1"]}],
            "findings": [],
            "evidence_refs": ["e1"],
            "checked": ["c1"],
            "not_checked": [],
            "validated_at": "2026-07-14T00:00:00Z",
        }

    def assert_schema_valid(self, artifact):
        schema = json.loads((Path(__file__).parents[3] / "schemas" / "verdict.v2.schema.json").read_text())
        jsonschema.Draft202012Validator(schema).validate(artifact)

    def runtime_facts(self):
        manifest = {
            "schema_version": "subject-manifest.v1",
            "declared_roots": ["src"],
            "exclusions": [],
            # One real file entry: build_manifest never emits an entry-less
            # manifest for an implementation subject, and the store-verdict
            # CLI refuses one outright.
            "entries": [
                {
                    "path": "src/app.py",
                    "kind": "file",
                    "executable": False,
                    "digest": "0" * 64,
                }
            ],
        }
        manifest["canonical_manifest_digest"] = tool.digest_value(tool.manifest_identity(manifest))
        return b"bead:agentops-test\nacceptance: works\n", manifest

    def store_bound(
        self,
        draft,
        destination,
        *,
        scope="PASS",
        author="author",
        validator="validator",
        freshness_source="runtime",
        freshness_attester="validator",
    ):
        intent, manifest = self.runtime_facts()
        return tool.store_verdict(
            draft,
            destination,
            intent,
            manifest,
            author,
            scope,
            validator,
            freshness_source,
            freshness_attester,
        )

    def test_manifest_is_content_addressed_and_detects_mutation(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            (root / "bin").mkdir()
            subject = root / "bin" / "tool"
            subject.write_text("one", encoding="utf-8")
            subject.chmod(0o755)
            manifest = tool.build_manifest(root, ["bin"], [])
            self.assertTrue(tool.verify_manifest(manifest, root, None)[0])
            subject.write_text("two", encoding="utf-8")
            self.assertFalse(tool.verify_manifest(manifest, root, None)[0])

    def test_git_metadata_is_not_identity_bearing(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            (root / "value").write_text("same", encoding="utf-8")
            first = tool.build_manifest(root, ["."], [], git_metadata={"commit": "one"})
            second = tool.build_manifest(root, ["."], [], git_metadata={"commit": "two"})
            self.assertEqual(first["canonical_manifest_digest"], second["canonical_manifest_digest"])
            self.assertNotEqual(first["git_metadata"], second["git_metadata"])
            self.assertTrue(tool.verify_manifest(first, root, None)[0])
            self.assertTrue(tool.verify_manifest(second, root, None)[0])

    def test_symlink_and_deletion_identity(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            (root / "target").write_text("x", encoding="utf-8")
            (root / "link").symlink_to("target")
            base = tool.build_manifest(root, ["."], [])
            (root / "target").unlink()
            current = tool.build_manifest(root, ["."], [], base)
            kinds = {entry["path"]: entry["kind"] for entry in current["entries"]}
            self.assertEqual(kinds["link"], "symlink")
            self.assertEqual(kinds["target"], "deletion")

    def test_verdict_identity_floor_and_idempotence(self):
        with tempfile.TemporaryDirectory() as raw:
            draft = self.draft()
            draft["author_context_id"] = "same"
            draft["validator_context_id"] = "same"
            first, path, existed = self.store_bound(draft, Path(raw), author="same", validator="same")
            self.assertEqual(first["verdict"], "NOT_PROVEN")
            self.assert_schema_valid(first)
            self.assertFalse(existed)
            second, second_path, existed = self.store_bound(draft, Path(raw), author="same", validator="same")
            self.assertTrue(existed)
            self.assertEqual(path, second_path)
            self.assertEqual(json.loads(path.read_text())["artifact_digest"], first["artifact_digest"])

    def test_runtime_identity_and_attestation_replace_missing_model_fields(self):
        for missing in ("author_context_id", "validator_context_id", "freshness_attestation"):
            with self.subTest(missing=missing), tempfile.TemporaryDirectory() as raw:
                draft = self.draft()
                draft.pop(missing)
                artifact, _path, _existed = self.store_bound(draft, Path(raw))
                self.assertEqual(artifact["verdict"], "PASS")
                self.assert_schema_valid(artifact)

    def test_runtime_validator_and_freshness_override_model_claims(self):
        with tempfile.TemporaryDirectory() as raw:
            draft = self.draft()
            draft["validator_context_id"] = "model-claimed-validator"
            draft["freshness_attestation"] = {"source": "caller", "attester_identity": "model-claimed-attester"}
            artifact, _path, _existed = self.store_bound(draft, Path(raw))
            self.assertEqual(artifact["validator_context_id"], "validator")
            self.assertEqual(
                artifact["freshness_attestation"],
                {"source": "runtime", "attester_identity": "validator"},
            )
            self.assertEqual(artifact["verdict"], "PASS")

    def test_pass_with_failed_criterion_is_downgraded(self):
        with tempfile.TemporaryDirectory() as raw:
            draft = self.draft()
            draft["criteria"][0]["result"] = "FAIL"
            artifact, _path, _existed = self.store_bound(draft, Path(raw))
            self.assertEqual(artifact["verdict"], "NOT_PROVEN")
            self.assert_schema_valid(artifact)

    def test_pass_without_evidence_is_downgraded(self):
        mutations = (
            lambda draft: draft.__setitem__("evidence_refs", []),
            lambda draft: draft.__setitem__("checked", []),
            lambda draft: draft["criteria"][0].__setitem__("evidence_refs", []),
        )
        for mutate in mutations:
            with self.subTest(mutate=mutate), tempfile.TemporaryDirectory() as raw:
                draft = self.draft()
                mutate(draft)
                artifact, _path, _existed = self.store_bound(draft, Path(raw))
                self.assertEqual(artifact["verdict"], "NOT_PROVEN")
                self.assertIn("PASS requires evidence", artifact["findings"][-1]["summary"])
                self.assert_schema_valid(artifact)

    def test_intent_snapshot_is_content_addressed_and_idempotent(self):
        with tempfile.TemporaryDirectory() as raw:
            destination = Path(raw)
            payload = b"caller intent\nacceptance: works\n"
            first, existed = tool.snapshot_intent(payload, destination)
            self.assertFalse(existed)
            self.assertEqual(first.name, f"{tool.hashlib.sha256(payload).hexdigest()}.intent")
            self.assertEqual(first.read_bytes(), payload)
            second, existed = tool.snapshot_intent(payload, destination)
            self.assertTrue(existed)
            self.assertEqual(first, second)

    def test_store_verdict_cli_snapshots_intent_before_persistence(self):
        with tempfile.TemporaryDirectory() as raw:
            workspace = Path(raw)
            intent, manifest = self.runtime_facts()
            intent_path = workspace / "intent.txt"
            manifest_path = workspace / "manifest.json"
            draft_path = workspace / "draft.json"
            intent_path.write_bytes(intent)
            manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
            draft_path.write_text(json.dumps(self.draft()), encoding="utf-8")

            result = subprocess.run(
                [
                    sys.executable,
                    str(Path(__file__).with_name("validate.py")),
                    "store-verdict",
                    "--draft",
                    str(draft_path),
                    "--intent-source",
                    str(intent_path),
                    "--subject-manifest",
                    str(manifest_path),
                    "--author-context-id",
                    "author",
                    "--validator-context-id",
                    "validator",
                    "--freshness-source",
                    "runtime",
                    "--freshness-attester-id",
                    "validator",
                    "--scope-result",
                    "PASS",
                    "--workspace",
                    str(workspace),
                ],
                check=False,
                capture_output=True,
                text=True,
            )
            self.assertEqual(result.returncode, 0, result.stderr)
            response = json.loads(result.stdout)
            snapshot = Path(response["intent_ref"])
            self.assertEqual(snapshot.read_bytes(), intent)
            self.assertEqual(response["acceptance_digest"], tool.hashlib.sha256(intent).hexdigest())

    def test_corrupt_existing_digest_yields_new_not_proven_artifact(self):
        with tempfile.TemporaryDirectory() as raw:
            destination = Path(raw)
            draft = self.draft()
            artifact, path, _ = self.store_bound(draft, destination)
            path.write_text("corrupt\n", encoding="utf-8")
            replacement, replacement_path, existed = self.store_bound(draft, destination)
            self.assertEqual(replacement["verdict"], "NOT_PROVEN")
            self.assertNotEqual(replacement["artifact_digest"], artifact["artifact_digest"])
            self.assertNotEqual(replacement_path, path)
            self.assertFalse(existed)
            self.assert_schema_valid(replacement)

    def test_incomplete_draft_is_rejected_without_writing(self):
        with tempfile.TemporaryDirectory() as raw:
            with self.assertRaisesRegex(tool.ContractError, "missing required fields"):
                tool.store_verdict({"verdict": "FAIL"}, Path(raw))
            self.assertEqual(list(Path(raw).iterdir()), [])

    def test_unknown_field_is_rejected_without_writing(self):
        with tempfile.TemporaryDirectory() as raw:
            draft = self.draft()
            draft["next_action"] = "repair"
            with self.assertRaisesRegex(tool.ContractError, "unknown fields"):
                self.store_bound(draft, Path(raw))
            self.assertEqual(list(Path(raw).iterdir()), [])

    def test_pass_without_runtime_facts_is_not_proven(self):
        with tempfile.TemporaryDirectory() as raw:
            artifact, _path, _existed = tool.store_verdict(self.draft(), Path(raw))
            self.assertEqual(artifact["verdict"], "NOT_PROVEN")
            self.assertIn("runtime intent source is missing", artifact["findings"][-1]["summary"])
            self.assert_schema_valid(artifact)

    def test_runtime_facts_override_model_authored_digests(self):
        with tempfile.TemporaryDirectory() as raw:
            draft = self.draft()
            draft["acceptance_digest"] = "c" * 64
            draft["subject_manifest_digest"] = "d" * 64
            artifact, _path, _existed = self.store_bound(draft, Path(raw))
            intent, manifest = self.runtime_facts()
            self.assertEqual(artifact["acceptance_digest"], tool.hashlib.sha256(intent).hexdigest())
            self.assertEqual(artifact["subject_manifest_digest"], manifest["canonical_manifest_digest"])
            self.assertEqual(artifact["verdict"], "PASS")

    def test_honest_scoped_pass_round_trips_through_documented_homes(self):
        """An honest draft with declared non-goals is representable as PASS.

        Both drafts below carry the same honest content. Draft A parks the
        declared non-goals in ``not_checked``, which is reserved for unverified
        in-scope acceptance: the result is NOT_PROVEN and the finding names
        where each caveat belongs. Draft B moves the same caveats into the
        documented homes and stores PASS with every caveat still readable in
        the persisted artifact. Nothing is deleted to earn the PASS.
        """
        bounded = "proven by the unit suite; the full integration matrix was not replayed"
        boundary = "declared non-goal; the diff proves cli/** untouched"

        with tempfile.TemporaryDirectory() as raw:
            draft_a = self.draft()
            draft_a["not_checked"] = [
                "cli/** (declared non-goal)",
                "Windows runners (declared non-goal)",
            ]
            artifact_a, _path, _existed = self.store_bound(draft_a, Path(raw))
            self.assertEqual(artifact_a["verdict"], "NOT_PROVEN")
            summary = artifact_a["findings"][-1]["summary"]
            self.assertIn("PASS cannot contain not_checked items", summary)
            for home in ("criteria[].reason", "non-goal", "report"):
                self.assertIn(home, summary)
            self.assert_schema_valid(artifact_a)

        with tempfile.TemporaryDirectory() as raw:
            draft_b = self.draft()
            draft_b["criteria"][0]["reason"] = bounded
            draft_b["criteria"].append(
                {
                    "id": "non-goal:cli-untouched",
                    "result": "PASS",
                    "evidence_refs": ["git-diff:cli"],
                    "reason": boundary,
                }
            )
            draft_b["evidence_refs"] = ["e1", "git-diff:cli"]
            draft_b["not_checked"] = []
            artifact_b, path, _existed = self.store_bound(draft_b, Path(raw))
            self.assertEqual(artifact_b["verdict"], "PASS")
            self.assert_schema_valid(artifact_b)
            # Round-trip: the caveats survive in the persisted PASS artifact.
            stored = json.loads(path.read_text(encoding="utf-8"))
            reasons = [criterion.get("reason") for criterion in stored["criteria"]]
            self.assertIn(bounded, reasons)
            self.assertIn(boundary, reasons)
            self.assertEqual(stored["not_checked"], [])
            self.assertEqual(stored["verdict"], "PASS")

    def test_criteria_field_error_names_the_allowed_set(self):
        with tempfile.TemporaryDirectory() as raw:
            draft = self.draft()
            draft["criteria"][0]["confidence"] = "high"
            with self.assertRaisesRegex(
                tool.ContractError,
                r"unknown confidence.*allowed fields are \{id, result, evidence_refs, reason\}",
            ):
                self.store_bound(draft, Path(raw))
            self.assertEqual(list(Path(raw).iterdir()), [])

    def test_runtime_scope_failure_forces_fail(self):
        with tempfile.TemporaryDirectory() as raw:
            artifact, _path, _existed = self.store_bound(self.draft(), Path(raw), scope="FAIL")
            self.assertEqual(artifact["verdict"], "FAIL")
            self.assertEqual(artifact["findings"][-1]["id"], "validate.scope")
            self.assert_schema_valid(artifact)


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3
"""Pure subject identity, scope, and verdict.v2 persistence helpers.

The module intentionally has no Git, tracker, queue, network, release, or
delivery integration. It operates only on explicit files and directories.
"""

from __future__ import annotations

import argparse
from datetime import datetime
import fnmatch
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import stat
import sys
import tempfile
from typing import Any, Iterable


HEX64 = set("0123456789abcdef")

# ``not_checked`` names the *in-scope acceptance surface a validator did not
# verify*.  PASS asserts that the whole declared acceptance surface was
# verified, so a PASS carries no ``not_checked`` entries by construction.
#
# That rule only pays for honest disclosure if every kind of scope limit has a
# home that survives inside a PASS.  Each does, so nothing is ever deleted to
# earn a PASS:
#
#   * a bounded proof of a criterion  -> ``criteria[].reason``
#   * a declared non-goal             -> the intent source's non-goals, and
#                                        optionally an evidence-backed boundary
#                                        criterion in ``criteria``
#   * residual risk                   -> the caller-facing report
#
# ``not_checked`` stays reserved for its one meaning: acceptance that genuinely
# went unverified, which is NOT_PROVEN and not PASS.
NOT_CHECKED_HOMES = (
    "not_checked lists unverified in-scope acceptance surface, so a PASS has none by "
    "construction; record a bounded proof of a criterion in criteria[].reason, a declared "
    "non-goal in the intent source's non-goals (optionally as an evidence-backed boundary "
    "criterion), and residual risk in the report; keep a not_checked entry only when "
    "acceptance genuinely went unverified, which is NOT_PROVEN"
)

CRITERION_KEYS = ("id", "result", "evidence_refs", "reason")
CRITERION_REQUIRED = ("id", "result", "evidence_refs")


class ContractError(ValueError):
    pass


def canonical_bytes(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def digest_value(value: Any) -> str:
    return hashlib.sha256(canonical_bytes(value)).hexdigest()


def normalize_rel(raw: str) -> str:
    raw = raw.replace("\\", "/")
    path = PurePosixPath(raw)
    if path.is_absolute() or ".." in path.parts:
        raise ContractError(f"path escapes subject root: {raw}")
    normalized = path.as_posix()
    if normalized in ("", "."):
        return "."
    return normalized.removeprefix("./")


def path_matches(path: str, pattern: str) -> bool:
    pattern = normalize_rel(pattern)
    if pattern == ".":
        return True
    if any(ch in pattern for ch in "*?["):
        return fnmatch.fnmatchcase(path, pattern)
    return path == pattern or path.startswith(pattern.rstrip("/") + "/")


def is_excluded(path: str, exclusions: Iterable[str]) -> bool:
    return any(path_matches(path, pattern) for pattern in exclusions)


def entry_for(root: Path, rel: str) -> dict[str, Any]:
    full = root if rel == "." else root / rel
    info = full.lstat()
    executable = bool(info.st_mode & (stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
    if full.is_symlink():
        target = os.readlink(full).encode("utf-8")
        return {"path": rel, "kind": "symlink", "executable": executable, "digest": hashlib.sha256(target).hexdigest()}
    if full.is_file():
        return {"path": rel, "kind": "file", "executable": executable, "digest": hashlib.sha256(full.read_bytes()).hexdigest()}
    raise ContractError(f"unsupported subject kind: {rel}")


def walk_declared(root: Path, declared: str, exclusions: list[str]) -> list[dict[str, Any]]:
    full = root if declared == "." else root / declared
    if not full.exists() and not full.is_symlink():
        return []
    if full.is_file() or full.is_symlink():
        return [] if is_excluded(declared, exclusions) else [entry_for(root, declared)]
    entries: list[dict[str, Any]] = []
    for dirpath, dirnames, filenames in os.walk(full, followlinks=False):
        current = Path(dirpath)
        kept_dirs: list[str] = []
        for name in sorted(dirnames):
            child = current / name
            rel = normalize_rel(child.relative_to(root).as_posix())
            if is_excluded(rel, exclusions):
                continue
            if child.is_symlink():
                entries.append(entry_for(root, rel))
            else:
                kept_dirs.append(name)
        dirnames[:] = kept_dirs
        for name in sorted(filenames):
            rel = normalize_rel((current / name).relative_to(root).as_posix())
            if not is_excluded(rel, exclusions):
                entries.append(entry_for(root, rel))
    return entries


def load_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ContractError(f"expected JSON object: {path}")
    return value


def build_manifest(
    root: Path,
    declared_roots: list[str],
    exclusions: list[str],
    base_manifest: dict[str, Any] | None = None,
    git_metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    root = root.resolve()
    if not root.is_dir():
        raise ContractError(f"subject root is not a directory: {root}")
    declared = sorted(set(normalize_rel(item) for item in declared_roots))
    if not declared:
        raise ContractError("at least one declared root is required")
    excluded = sorted(set(normalize_rel(item) for item in exclusions))
    by_path: dict[str, dict[str, Any]] = {}
    for item in declared:
        for entry in walk_declared(root, item, excluded):
            by_path[entry["path"]] = entry

    manifest: dict[str, Any] = {
        "schema_version": "subject-manifest.v1",
        "declared_roots": declared,
        "exclusions": excluded,
        "entries": sorted(by_path.values(), key=lambda item: item["path"]),
    }
    if base_manifest is not None:
        base_digest = base_manifest.get("canonical_manifest_digest")
        if not valid_digest(base_digest):
            raise ContractError("base manifest has no valid canonical_manifest_digest")
        manifest["base_manifest_digest"] = base_digest
        current = set(by_path)
        deletions = []
        for prior in base_manifest.get("entries", []):
            path = normalize_rel(str(prior.get("path", "")))
            declared_here = any(path_matches(path, item) for item in declared)
            if declared_here and path not in current and not is_excluded(path, excluded):
                deletions.append({"path": path, "kind": "deletion", "executable": bool(prior.get("executable", False))})
        manifest["entries"] = sorted(manifest["entries"] + deletions, key=lambda item: item["path"])
    if git_metadata:
        manifest["git_metadata"] = git_metadata
    manifest["canonical_manifest_digest"] = digest_value(manifest_identity(manifest))
    return manifest


def valid_digest(value: Any) -> bool:
    return isinstance(value, str) and len(value) == 64 and all(ch in HEX64 for ch in value)


def manifest_identity(manifest: dict[str, Any]) -> dict[str, Any]:
    """Return only the fields that identify subject content.

    ``git_metadata`` is intentionally descriptive.  Supplying or changing it
    must never change the identity of otherwise identical content.
    """
    return {
        key: value
        for key, value in manifest.items()
        if key not in {"canonical_manifest_digest", "git_metadata"}
    }


def verify_manifest(manifest: dict[str, Any], root: Path, base_manifest: dict[str, Any] | None) -> tuple[bool, str]:
    claimed = manifest.get("canonical_manifest_digest")
    if not valid_digest(claimed) or digest_value(manifest_identity(manifest)) != claimed:
        return False, "manifest canonical digest is invalid"
    rebuilt = build_manifest(
        root,
        list(manifest.get("declared_roots", [])),
        list(manifest.get("exclusions", [])),
        base_manifest,
        manifest.get("git_metadata"),
    )
    if canonical_bytes(rebuilt) != canonical_bytes(manifest):
        return False, "subject content no longer matches manifest"
    return True, "manifest matches subject"


def add_integrity_finding(draft: dict[str, Any], summary: str) -> dict[str, Any]:
    changed = dict(draft)
    changed["verdict"] = "NOT_PROVEN"
    findings = list(changed.get("findings") or [])
    findings.append({"id": "validate.integrity", "summary": summary, "evidence_refs": ["verdict-store"]})
    changed["findings"] = findings
    return changed


def bind_runtime_facts(
    draft: dict[str, Any],
    intent_bytes: bytes | None,
    manifest: dict[str, Any] | None,
    author_context_id: str | None,
    scope_status: str | None,
    validator_context_id: str | None,
    freshness_source: str | None,
    freshness_attester_id: str | None,
) -> dict[str, Any]:
    """Inject runtime-owned identity, freshness, intent, subject, and scope facts."""
    changed = dict(draft)
    problems: list[str] = []
    if intent_bytes is None:
        problems.append("runtime intent source is missing")
    else:
        changed["acceptance_digest"] = hashlib.sha256(intent_bytes).hexdigest()
    if not isinstance(manifest, dict):
        problems.append("runtime subject manifest is missing")
    else:
        claimed = manifest.get("canonical_manifest_digest")
        if not valid_digest(claimed) or digest_value(manifest_identity(manifest)) != claimed:
            problems.append("runtime subject manifest digest is invalid")
        else:
            changed["subject_manifest_digest"] = claimed
    if not isinstance(author_context_id, str) or not author_context_id.strip():
        problems.append("runtime author context ID is missing")
    else:
        changed["author_context_id"] = author_context_id
    if not isinstance(validator_context_id, str) or not validator_context_id.strip():
        problems.append("runtime validator context ID is missing")
    else:
        changed["validator_context_id"] = validator_context_id
    if freshness_source not in {"runtime", "caller"}:
        problems.append("runtime freshness source is missing or invalid")
    elif not isinstance(freshness_attester_id, str) or not freshness_attester_id.strip():
        problems.append("runtime freshness attester identity is missing")
    else:
        changed["freshness_attestation"] = {
            "source": freshness_source,
            "attester_identity": freshness_attester_id,
        }
    if scope_status == "FAIL":
        changed["verdict"] = "FAIL"
        findings = list(changed.get("findings") or [])
        findings.append({"id": "validate.scope", "summary": "runtime-derived changed paths are outside intent scope", "evidence_refs": ["runtime-scope"]})
        changed["findings"] = findings
    elif scope_status != "PASS":
        problems.append("runtime changed-path scope is not proven")
    if problems:
        return add_integrity_finding(changed, "; ".join(problems))
    return changed


def enforce_identity(draft: dict[str, Any]) -> dict[str, Any]:
    draft = dict(draft)
    draft.setdefault("author_context_id", None)
    draft.setdefault("validator_context_id", None)
    draft.setdefault("freshness_attestation", None)
    author = draft.get("author_context_id")
    validator = draft.get("validator_context_id")
    freshness = draft.get("freshness_attestation")
    problems = []
    if not isinstance(author, str) or not author.strip():
        problems.append("author context ID is missing")
    if not isinstance(validator, str) or not validator.strip():
        problems.append("validator context ID is missing")
    if author and validator and author == validator:
        problems.append("author and validator context IDs collide")
    if not isinstance(freshness, dict) or freshness.get("source") not in ("runtime", "caller") or not freshness.get("attester_identity"):
        problems.append("freshness attestation is missing or invalid")
    if draft.get("verdict") == "PASS" and (draft.get("not_checked") or []):
        problems.append(f"PASS cannot contain not_checked items: {NOT_CHECKED_HOMES}")
    criteria = draft.get("criteria")
    if draft.get("verdict") == "PASS" and (
        not isinstance(criteria, list)
        or not criteria
        or any(not isinstance(item, dict) or item.get("result") != "PASS" for item in criteria)
    ):
        problems.append("PASS requires at least one criterion and every criterion must PASS")
    if draft.get("verdict") == "PASS" and (
        any(
            not isinstance(item, dict)
            or not isinstance(item.get("evidence_refs"), list)
            or not item["evidence_refs"]
            for item in criteria or []
        )
        or not draft.get("evidence_refs")
        or not draft.get("checked")
    ):
        problems.append("PASS requires evidence for every criterion plus nonempty evidence_refs and checked")
    if problems:
        return add_integrity_finding(draft, "; ".join(problems))
    return draft


VERDICT_KEYS = {
    "schema_version",
    "acceptance_digest",
    "subject_manifest_digest",
    "author_context_id",
    "validator_context_id",
    "freshness_attestation",
    "verdict",
    "criteria",
    "findings",
    "evidence_refs",
    "checked",
    "not_checked",
    "validated_at",
    "artifact_digest",
}


def require_string_list(value: Any, field: str, *, nonempty: bool = False) -> None:
    if not isinstance(value, list) or (nonempty and not value):
        raise ContractError(f"verdict.v2 {field} must be a{' nonempty' if nonempty else ''} array")
    if any(not isinstance(item, str) or not item for item in value):
        raise ContractError(f"verdict.v2 {field} entries must be nonempty strings")


def criterion_fields_error(index: int, *, missing: list[str], unknown: list[str]) -> str:
    """Return an actionable criteria-shape message naming the allowed field set."""
    detail: list[str] = []
    if missing:
        detail.append(f"missing {', '.join(missing)}")
    if unknown:
        detail.append(f"unknown {', '.join(unknown)}")
    problem = "; ".join(detail) if detail else "not an object"
    return (
        f"verdict.v2 criteria[{index}] has invalid fields ({problem}); allowed fields are "
        f"{{{', '.join(CRITERION_KEYS)}}}, of which {', '.join(CRITERION_REQUIRED)} are required"
    )


def validate_verdict_v2(artifact: dict[str, Any]) -> None:
    """Enforce the complete bundled verdict.v2 contract before persistence."""
    missing = sorted(VERDICT_KEYS - artifact.keys())
    extra = sorted(artifact.keys() - VERDICT_KEYS)
    if missing:
        raise ContractError(f"verdict.v2 missing required fields: {', '.join(missing)}")
    if extra:
        raise ContractError(f"verdict.v2 contains unknown fields: {', '.join(extra)}")
    if artifact["schema_version"] != "verdict.v2":
        raise ContractError("verdict.v2 schema_version must be verdict.v2")
    for field in ("acceptance_digest", "subject_manifest_digest", "artifact_digest"):
        if not valid_digest(artifact[field]):
            raise ContractError(f"verdict.v2 {field} must be a lowercase SHA-256 digest")
    expected_digest = digest_value({key: value for key, value in artifact.items() if key != "artifact_digest"})
    if artifact["artifact_digest"] != expected_digest:
        raise ContractError("verdict.v2 artifact_digest does not match canonical JSON")
    for field in ("author_context_id", "validator_context_id"):
        if artifact[field] is not None and (not isinstance(artifact[field], str) or not artifact[field]):
            raise ContractError(f"verdict.v2 {field} must be null or a nonempty string")
    freshness = artifact["freshness_attestation"]
    if freshness is not None:
        if not isinstance(freshness, dict) or set(freshness) != {"source", "attester_identity"}:
            raise ContractError("verdict.v2 freshness_attestation has invalid fields")
        if freshness["source"] not in {"runtime", "caller"}:
            raise ContractError("verdict.v2 freshness source must be runtime or caller")
        if not isinstance(freshness["attester_identity"], str) or not freshness["attester_identity"]:
            raise ContractError("verdict.v2 freshness attester_identity must be nonempty")
    if artifact["verdict"] not in {"PASS", "FAIL", "NOT_PROVEN"}:
        raise ContractError("verdict.v2 verdict must be PASS, FAIL, or NOT_PROVEN")
    criteria = artifact["criteria"]
    if not isinstance(criteria, list) or not criteria:
        raise ContractError("verdict.v2 criteria must be a nonempty array")
    for index, criterion in enumerate(criteria):
        if not isinstance(criterion, dict):
            raise ContractError(criterion_fields_error(index, missing=[], unknown=[]))
        missing_keys = [key for key in CRITERION_REQUIRED if key not in criterion]
        unknown_keys = sorted(set(criterion) - set(CRITERION_KEYS))
        if missing_keys or unknown_keys:
            raise ContractError(
                criterion_fields_error(index, missing=missing_keys, unknown=unknown_keys)
            )
        if not isinstance(criterion["id"], str) or not criterion["id"]:
            raise ContractError(f"verdict.v2 criteria[{index}].id must be nonempty")
        if criterion["result"] not in {"PASS", "FAIL", "NOT_PROVEN"}:
            raise ContractError(f"verdict.v2 criteria[{index}].result is invalid")
        require_string_list(criterion["evidence_refs"], f"criteria[{index}].evidence_refs")
        if "reason" in criterion and not isinstance(criterion["reason"], str):
            raise ContractError(f"verdict.v2 criteria[{index}].reason must be a string")
    findings = artifact["findings"]
    if not isinstance(findings, list):
        raise ContractError("verdict.v2 findings must be an array")
    for index, finding in enumerate(findings):
        # `class` is the convergence law's second key (ADR-0017) and is
        # OPTIONAL: it names the KIND of defect so a repair phase that mints a
        # fresh id for the same kind every round stays visible. A finding that
        # belongs to no nameable kind simply omits it — but present-and-blank is
        # malformed, never the same as absent, on every leg of the contract.
        if not isinstance(finding, dict) or not {"id", "summary", "evidence_refs"} <= set(finding) or not set(
            finding
        ) <= {"id", "class", "summary", "evidence_refs"}:
            raise ContractError(f"verdict.v2 findings[{index}] has invalid fields")
        if "class" in finding and (
            not isinstance(finding["class"], str) or not finding["class"].strip()
        ):
            raise ContractError(f"verdict.v2 findings[{index}].class must be a nonempty string")
        if not isinstance(finding["id"], str) or not finding["id"]:
            raise ContractError(f"verdict.v2 findings[{index}].id must be nonempty")
        if not isinstance(finding["summary"], str) or not finding["summary"]:
            raise ContractError(f"verdict.v2 findings[{index}].summary must be nonempty")
        require_string_list(finding["evidence_refs"], f"findings[{index}].evidence_refs", nonempty=True)
    for field in ("evidence_refs", "checked", "not_checked"):
        require_string_list(artifact[field], field)
    if not isinstance(artifact["validated_at"], str):
        raise ContractError("verdict.v2 validated_at must be an RFC3339 date-time")
    try:
        timestamp = datetime.fromisoformat(artifact["validated_at"].replace("Z", "+00:00"))
    except ValueError as exc:
        raise ContractError("verdict.v2 validated_at must be an RFC3339 date-time") from exc
    if timestamp.tzinfo is None:
        raise ContractError("verdict.v2 validated_at must include a timezone")
    if artifact["verdict"] == "PASS":
        author = artifact["author_context_id"]
        validator = artifact["validator_context_id"]
        if not author or not validator or author == validator or freshness is None:
            raise ContractError("verdict.v2 PASS requires distinct identities and freshness attestation")
        if any(criterion["result"] != "PASS" for criterion in criteria):
            raise ContractError("verdict.v2 PASS requires every criterion to PASS")
        if any(not criterion["evidence_refs"] for criterion in criteria) or not artifact["evidence_refs"] or not artifact["checked"]:
            raise ContractError("verdict.v2 PASS requires criterion evidence plus nonempty evidence_refs and checked")
        if artifact["not_checked"]:
            raise ContractError(
                f"verdict.v2 PASS cannot contain not_checked items: {NOT_CHECKED_HOMES}"
            )


def artifact_bytes(draft: dict[str, Any]) -> tuple[dict[str, Any], bytes]:
    unsigned = {key: value for key, value in draft.items() if key != "artifact_digest"}
    digest = digest_value(unsigned)
    artifact = dict(unsigned)
    artifact["artifact_digest"] = digest
    return artifact, canonical_bytes(artifact) + b"\n"


def atomic_store(artifact: dict[str, Any], payload: bytes, destination: Path) -> tuple[Path, bool]:
    destination.mkdir(parents=True, exist_ok=True)
    target = destination / f"{artifact['artifact_digest']}.json"
    if target.exists():
        if target.read_bytes() == payload:
            return target, True
        raise ContractError(f"integrity collision at {target}")
    fd, temporary = tempfile.mkstemp(prefix=".verdict-", suffix=".tmp", dir=destination)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, target)
        dir_fd = os.open(destination, os.O_RDONLY)
        try:
            os.fsync(dir_fd)
        finally:
            os.close(dir_fd)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)
    return target, False


def snapshot_intent(payload: bytes, destination: Path) -> tuple[Path, bool]:
    """Persist exact resolved intent bytes under their SHA-256 identity."""
    destination.mkdir(parents=True, exist_ok=True)
    digest = hashlib.sha256(payload).hexdigest()
    target = destination / f"{digest}.intent"
    if target.exists():
        if target.read_bytes() == payload:
            return target, True
        raise ContractError(f"intent snapshot integrity collision at {target}")
    fd, temporary = tempfile.mkstemp(prefix=".intent-", suffix=".tmp", dir=destination)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, target)
        dir_fd = os.open(destination, os.O_RDONLY)
        try:
            os.fsync(dir_fd)
        finally:
            os.close(dir_fd)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)
    return target, False


def store_verdict(
    draft: dict[str, Any],
    destination: Path,
    intent_bytes: bytes | None = None,
    manifest: dict[str, Any] | None = None,
    author_context_id: str | None = None,
    scope_status: str | None = None,
    validator_context_id: str | None = None,
    freshness_source: str | None = None,
    freshness_attester_id: str | None = None,
) -> tuple[dict[str, Any], Path, bool]:
    draft = bind_runtime_facts(
        draft,
        intent_bytes,
        manifest,
        author_context_id,
        scope_status,
        validator_context_id,
        freshness_source,
        freshness_attester_id,
    )
    draft = enforce_identity(draft)
    draft["schema_version"] = "verdict.v2"
    artifact, payload = artifact_bytes(draft)
    validate_verdict_v2(artifact)
    try:
        path, existed = atomic_store(artifact, payload, destination)
    except ContractError as exc:
        artifact, payload = artifact_bytes(add_integrity_finding(draft, str(exc)))
        validate_verdict_v2(artifact)
        path, existed = atomic_store(artifact, payload, destination)
    return artifact, path, existed


def write_json(value: dict[str, Any], output: str | None) -> None:
    payload = json.dumps(value, sort_keys=True, indent=2, ensure_ascii=False) + "\n"
    if output:
        Path(output).write_text(payload, encoding="utf-8")
    else:
        sys.stdout.write(payload)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    manifest = sub.add_parser("manifest", help="compute subject-manifest.v1 without Git")
    manifest.add_argument("--root", required=True)
    manifest.add_argument("--include", action="append", required=True)
    manifest.add_argument("--exclude", action="append", default=[])
    manifest.add_argument("--base-manifest")
    manifest.add_argument("--git-metadata-json")
    manifest.add_argument("--output")
    verify = sub.add_parser("verify-manifest", help="recompute and compare a manifest")
    verify.add_argument("--root", required=True)
    verify.add_argument("--manifest", required=True)
    verify.add_argument("--base-manifest")
    snapshot = sub.add_parser("snapshot-intent", help="persist exact intent bytes under their SHA-256 identity")
    snapshot.add_argument("--source", required=True, help="intent file path, or - for stdin")
    snapshot.add_argument("--workspace", default=".")
    snapshot.add_argument("--intent-dir")
    digest = sub.add_parser("digest", help="print a canonical JSON digest")
    digest.add_argument("json_file")
    store = sub.add_parser("store-verdict", help="atomically persist verdict.v2")
    store.add_argument("--draft", required=True)
    store.add_argument("--intent-source", required=True)
    store.add_argument("--subject-manifest", required=True)
    store.add_argument("--author-context-id", required=True)
    store.add_argument("--validator-context-id", required=True)
    store.add_argument("--freshness-source", required=True, choices=("runtime", "caller"))
    store.add_argument("--freshness-attester-id", required=True)
    store.add_argument("--scope-result", required=True, choices=("PASS", "FAIL", "NOT_PROVEN"))
    store.add_argument("--workspace", default=".")
    store.add_argument("--verdict-dir")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        if args.command == "manifest":
            base = load_json(Path(args.base_manifest)) if args.base_manifest else None
            metadata = json.loads(args.git_metadata_json) if args.git_metadata_json else None
            write_json(build_manifest(Path(args.root), args.include, args.exclude, base, metadata), args.output)
        elif args.command == "verify-manifest":
            manifest = load_json(Path(args.manifest))
            base = load_json(Path(args.base_manifest)) if args.base_manifest else None
            ok, reason = verify_manifest(manifest, Path(args.root), base)
            write_json({"result": "PASS" if ok else "NOT_PROVEN", "reason": reason}, None)
            return 0 if ok else 1
        elif args.command == "snapshot-intent":
            intent_bytes = sys.stdin.buffer.read() if args.source == "-" else Path(args.source).read_bytes()
            destination = Path(args.intent_dir) if args.intent_dir else Path(args.workspace) / ".agents" / "ao" / "intents" / "sha256"
            intent_path, existed = snapshot_intent(intent_bytes, destination)
            write_json({
                "acceptance_digest": hashlib.sha256(intent_bytes).hexdigest(),
                "idempotent": existed,
                "intent_ref": str(intent_path),
            }, None)
        elif args.command == "digest":
            print(digest_value(load_json(Path(args.json_file))))
        elif args.command == "store-verdict":
            destination = Path(args.verdict_dir) if args.verdict_dir else Path(args.workspace) / ".agents" / "ao" / "verdicts" / "sha256"
            intent_bytes = Path(args.intent_source).read_bytes()
            intent_path, intent_existed = snapshot_intent(
                intent_bytes,
                Path(args.workspace) / ".agents" / "ao" / "intents" / "sha256",
            )
            subject_manifest = load_json(Path(args.subject_manifest))
            if not subject_manifest.get("entries"):
                raise ContractError(
                    "subject manifest has no entries; Validate needs a nonempty "
                    "implementation candidate, not a report or plan document"
                )
            artifact, path, existed = store_verdict(
                load_json(Path(args.draft)),
                destination,
                intent_bytes,
                subject_manifest,
                args.author_context_id,
                args.scope_result,
                args.validator_context_id,
                args.freshness_source,
                args.freshness_attester_id,
            )
            write_json({
                "acceptance_digest": hashlib.sha256(intent_bytes).hexdigest(),
                "artifact_digest": artifact["artifact_digest"],
                "idempotent": existed,
                "intent_ref": str(intent_path),
                "intent_snapshot_idempotent": intent_existed,
                "path": str(path),
                "verdict": artifact["verdict"],
            }, None)
        return 0
    except (ContractError, OSError, json.JSONDecodeError) as exc:
        print(f"validate: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env bash
set -euo pipefail

skill_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
repo_root="$(cd "$skill_dir/../.." && pwd)"

grep -q '^name: validate$' "$skill_dir/SKILL.md"
grep -Fq 'PASS`, `FAIL`, or `NOT_PROVEN`' "$skill_dir/SKILL.md"
grep -Fq 'sole semantic author of `verdict.v2`' "$skill_dir/SKILL.md"
grep -Fq 'Only when the caller requests machine-readable evidence' "$skill_dir/SKILL.md"
grep -Fq 'nonempty implementation candidate' "$skill_dir/SKILL.md"

python3 "$skill_dir/tests/validate.py" --help >/dev/null
python3 - "$repo_root" <<'PY'
import json
import sys
from pathlib import Path
from jsonschema import Draft202012Validator

root = Path(sys.argv[1])
names = (
    "subject-manifest.v1.schema.json",
    "verdict.v2.schema.json",
)
for name in names:
    Draft202012Validator.check_schema(json.loads((root / "schemas" / name).read_text(encoding="utf-8")))
PY

PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s "$skill_dir/tests" -p 'test_validate.py'
echo 'validate skill contract: PASS'