company-haness/.claude/hooks/verify_run.py

#!/usr/bin/env python3
"""Run one verifier and append a typed, context-bound evidence receipt.

Claude Code's generic PostToolUse response does not always expose a process exit
code.  This sanctioned runner owns the subprocess, so exit status, assertion
status, command argv, output hashes, workflow/session/agent context, and subject
are recorded together.  It never invokes a shell.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import time
import uuid
from datetime import datetime, timezone

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import _workspace as W  # noqa: E402

CATEGORIES = (
    "acceptance-criteria", "test", "security", "privacy", "data-quality",
    "reliability", "release-readiness",
)
TRIVIAL_EXECUTABLES = {"true", "false", "echo", "printf", "ls", "cat", "grep", "pwd"}


def _now():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _sha(value):
    return hashlib.sha256(value).hexdigest()


def _safe_cwd(raw):
    root = os.path.realpath(W.work_root())
    candidate = raw if os.path.isabs(raw) else os.path.join(root, raw)
    candidate = os.path.realpath(candidate)
    if os.path.commonpath([root, candidate]) != root or not os.path.isdir(candidate):
        raise ValueError(f"--cwd must be an existing directory inside workspace: {raw}")
    return candidate


def _append_receipt(receipt):
    evidence_dir = W.evidence_dir()
    os.makedirs(evidence_dir, exist_ok=True)
    path = os.path.join(evidence_dir, "ledger.jsonl")
    with open(path, "a", encoding="utf-8") as fh:
        try:
            import fcntl
            fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
        except Exception:
            pass
        fh.write(json.dumps(receipt, ensure_ascii=False) + "\n")
        fh.flush()
        os.fsync(fh.fileno())
        try:
            import fcntl
            fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
        except Exception:
            pass


def build_parser():
    parser = argparse.ArgumentParser(description="execute a verifier and mint a typed receipt")
    parser.add_argument("--workflow", required=True)
    parser.add_argument("--agent", required=True)
    parser.add_argument("--session", required=True)
    parser.add_argument("--category", choices=CATEGORIES, required=True)
    parser.add_argument("--subject", required=True,
                        help="specific criterion/component being verified")
    parser.add_argument("--source-revision-sha256")
    parser.add_argument("--cwd", default=".")
    parser.add_argument("command", nargs=argparse.REMAINDER)
    return parser


def main(argv=None):
    args = build_parser().parse_args(argv)
    command = list(args.command)
    if command and command[0] == "--":
        command = command[1:]
    if not command:
        sys.stderr.write("[verify_run] verifier command required after --\n")
        return 2
    executable = os.path.basename(command[0]).lower()
    if executable in TRIVIAL_EXECUTABLES:
        sys.stderr.write(f"[verify_run] trivial command cannot prove verification: {executable}\n")
        return 2
    if executable in {"sh", "bash", "zsh", "fish"}:
        sys.stderr.write("[verify_run] shell interpreters are forbidden; pass verifier argv directly\n")
        return 2
    if args.source_revision_sha256 and (
            len(args.source_revision_sha256) != 64
            or any(ch not in "0123456789abcdef" for ch in args.source_revision_sha256.lower())):
        sys.stderr.write("[verify_run] --source-revision-sha256 must be 64-hex\n")
        return 2
    try:
        cwd = _safe_cwd(args.cwd)
    except Exception as exc:
        sys.stderr.write(f"[verify_run] {exc}\n")
        return 2

    started_at = _now()
    started = time.monotonic()
    try:
        completed = subprocess.run(command, cwd=cwd, capture_output=True, check=False)
        exit_code = completed.returncode
        stdout = completed.stdout or b""
        stderr = completed.stderr or b""
    except OSError as exc:
        exit_code, stdout, stderr = 127, b"", str(exc).encode("utf-8", "replace")
    finished_at = _now()
    receipt_id = f"vr-{int(time.time())}-{uuid.uuid4().hex[:12]}"
    receipt = {
        "receipt_id": receipt_id,
        "tool_use_id": receipt_id,
        "receipt_type": "verification-run",
        "tool_name": "VerifyRun",
        "workflow_id": args.workflow,
        "session_id": args.session,
        "agent_id": args.agent,
        "verification_category": args.category,
        "verification_subject": args.subject,
        "assertion_status": "passed" if exit_code == 0 else "failed",
        "exit_code": exit_code,
        "command_argv": command,
        "command_argv_sha256": _sha(
            json.dumps(command, ensure_ascii=False, separators=(",", ":")).encode("utf-8")),
        "cwd": cwd,
        "started_at": started_at,
        "ts": finished_at,
        "duration_ms": round((time.monotonic() - started) * 1000),
        "stdout_sha256": _sha(stdout),
        "stderr_sha256": _sha(stderr),
    }
    if args.source_revision_sha256:
        receipt["source_revision_sha256"] = args.source_revision_sha256.lower()
    try:
        _append_receipt(receipt)
    except Exception as exc:
        sys.stderr.write(f"[verify_run] receipt append failed: {exc}\n")
        return 125

    if stdout:
        sys.stdout.buffer.write(stdout)
        if not stdout.endswith(b"\n"):
            sys.stdout.buffer.write(b"\n")
    if stderr:
        sys.stderr.buffer.write(stderr)
        if not stderr.endswith(b"\n"):
            sys.stderr.buffer.write(b"\n")
    sys.stderr.write(
        f"[verify_run] receipt-id={receipt_id} status={receipt['assertion_status']} "
        f"exit={exit_code} category={args.category} subject={args.subject}\n")
    return exit_code if 0 <= exit_code <= 124 else 124


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