#!/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())

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):
        # This retained pack copy is deeper than the original skills tree.
        # Both tests must validate against the repository's canonical schema.
        schema_path = next(
            ancestor / "schemas" / "verdict.v2.schema.json"
            for ancestor in Path(__file__).resolve().parents
            if (ancestor / "schemas" / "verdict.v2.schema.json").is_file()
        )
        schema = json.loads(schema_path.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_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")


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("PASS cannot contain not_checked items")
    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 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):
        allowed = {"id", "result", "evidence_refs", "reason"}
        if not isinstance(criterion, dict) or not {"id", "result", "evidence_refs"}.issubset(criterion) or not set(criterion).issubset(allowed):
            raise ContractError(f"verdict.v2 criteria[{index}] has invalid fields")
        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):
        if not isinstance(finding, dict) or set(finding) != {"id", "summary", "evidence_refs"}:
            raise ContractError(f"verdict.v2 findings[{index}] has invalid fields")
        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("verdict.v2 PASS cannot contain not_checked items")


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 `verdict.v2` writer when persistence is requested' "$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/scripts/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/scripts" -p 'test_validate.py'
echo 'validate skill contract: PASS'