teleo-infrastructure/scripts/validate_working_leo_production_receipt.py

#!/usr/bin/env python3
"""Validate a Working Leo production-apply receipt.

This validator is intentionally offline. It reads the retained authorized apply
runbook plus a receipt JSON file and checks that the receipt proves the
post-apply state expected by the runbook. It never opens a database connection,
never runs SQL, and never posts to Telegram.
"""

from __future__ import annotations

import argparse
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

REPORT_DIR = Path("docs/reports/leo-working-state-20260709")
DEFAULT_RUNBOOK = REPORT_DIR / "working-leo-authorized-apply-runbook-current.json"
DEFAULT_RECEIPT = REPORT_DIR / "working-leo-production-apply-receipt-current.json"
DEFAULT_OUT = REPORT_DIR / "working-leo-production-apply-receipt-validation-current.json"
DEFAULT_MARKDOWN_OUT = REPORT_DIR / "working-leo-production-apply-receipt-validation-current.md"

EMPTY_PLACEHOLDERS = {"", "todo", "tbd", "n/a", "none", "null", "placeholder"}


def _load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def _is_nonempty(value: Any) -> bool:
    if value is None:
        return False
    if isinstance(value, str):
        return value.strip().lower() not in EMPTY_PLACEHOLDERS
    if isinstance(value, dict | list):
        return bool(value)
    return True


def _as_list(value: Any) -> list[Any]:
    if isinstance(value, list):
        return value
    if value is None:
        return []
    return [value]


def _resolve(path: str, *, repo_root: Path) -> Path:
    candidate = Path(path)
    return candidate if candidate.is_absolute() else repo_root / candidate


def _paths_exist(receipt: dict[str, Any], field: str, *, repo_root: Path) -> tuple[bool, list[str]]:
    values = _as_list(receipt.get(field))
    missing = [str(value) for value in values if not _resolve(str(value), repo_root=repo_root).exists()]
    return bool(values) and not missing, missing


def _counts_match(expected: dict[str, Any], actual: Any) -> bool:
    if not isinstance(actual, dict):
        return False
    try:
        normalized_actual = {str(key): int(value) for key, value in actual.items()}
        normalized_expected = {str(key): int(value) for key, value in expected.items()}
    except (TypeError, ValueError):
        return False
    return normalized_actual == normalized_expected


def _direct_claim_score_passes(path_value: Any, *, repo_root: Path) -> tuple[bool, str]:
    values = _as_list(path_value)
    if not values:
        return False, "missing direct_claim_score_path"
    path = _resolve(str(values[0]), repo_root=repo_root)
    if not path.exists():
        return False, f"missing direct claim score artifact: {path}"
    text = path.read_text(encoding="utf-8")
    if path.suffix == ".json":
        data = json.loads(text)
        score = data.get("score", {})
        if score.get("pass") is True and score.get("passes") == score.get("expected_prompt_count") == 6:
            return True, "json score pass"
        return False, "json score is not 6/6 pass"
    if "Overall pass for scored coverage: `True`" in text and "Prompts scored: `6/6`" in text:
        return True, "markdown score pass"
    return False, "score artifact does not show 6/6 pass"


def _service_readback_ok(value: Any) -> bool:
    if not isinstance(value, dict):
        return False
    return (
        value.get("ActiveState") == "active"
        and value.get("SubState") == "running"
        and _is_nonempty(value.get("MainPID"))
        and _is_nonempty(value.get("NRestarts"))
    )


def _cleanup_readback_ok(value: Any) -> bool:
    if not isinstance(value, dict):
        return False
    checks = value.get("checks") if isinstance(value.get("checks"), dict) else value
    leftovers = checks.get("leftover_rehearsal_clone_or_staging_databases", [])
    return checks.get("cleanup_ok") is True and leftovers == []


def validate_receipt(receipt_path: Path, runbook_path: Path = DEFAULT_RUNBOOK, *, repo_root: Path = Path(".")) -> dict[str, Any]:
    repo_root = repo_root.resolve()
    runbook = _load_json(runbook_path)
    required_fields = list(runbook["production_apply_receipt_required_fields"])
    expected_counts = dict(runbook["expected_after_apply_counts"])

    if not receipt_path.exists():
        return {
            "generated_at_utc": datetime.now(timezone.utc).isoformat(),
            "status": "missing_receipt",
            "valid": False,
            "receipt_path": str(receipt_path),
            "runbook_path": str(runbook_path),
            "missing_fields": required_fields,
            "errors": [f"receipt file does not exist: {receipt_path}"],
            "claim_ceiling": "No production apply receipt was validated.",
        }

    receipt = _load_json(receipt_path)
    missing_fields = [field for field in required_fields if field not in receipt]
    errors: list[str] = []
    warnings: list[str] = []
    checks: dict[str, bool] = {}

    checks["all_required_fields_present"] = not missing_fields
    if missing_fields:
        errors.append(f"missing required fields: {', '.join(missing_fields)}")

    for field in ("explicit_authorization_record", "operator_identity", "git_commit_sha"):
        checks[f"{field}_present"] = _is_nonempty(receipt.get(field))
        if not checks[f"{field}_present"]:
            errors.append(f"{field} is missing or placeholder")

    git_sha = str(receipt.get("git_commit_sha", ""))
    checks["git_commit_sha_shape"] = bool(re.fullmatch(r"[0-9a-f]{7,40}", git_sha))
    if not checks["git_commit_sha_shape"]:
        errors.append("git_commit_sha must be a 7-40 char lowercase hex SHA")

    authorization = receipt.get("explicit_authorization_record")
    checks["authorization_record_not_false"] = not (
        isinstance(authorization, dict) and authorization.get("authorized") is False
    )
    if not checks["authorization_record_not_false"]:
        errors.append("explicit_authorization_record explicitly says authorized=false")

    for field in (
        "preflight_output_paths",
        "apply_output_paths",
        "postflight_output_paths",
        "telegram_or_handler_regression_paths",
        "rollback_packet_paths",
    ):
        ok, missing = _paths_exist(receipt, field, repo_root=repo_root)
        checks[f"{field}_exist"] = ok
        if missing:
            errors.append(f"{field} contains missing path(s): {', '.join(missing)}")
        elif not _as_list(receipt.get(field)):
            errors.append(f"{field} must be a non-empty path list")

    checks["expected_counts_match_runbook"] = _counts_match(expected_counts, receipt.get("expected_after_apply_counts"))
    if not checks["expected_counts_match_runbook"]:
        errors.append("expected_after_apply_counts does not match the authorized runbook")

    checks["actual_counts_match_expected"] = _counts_match(expected_counts, receipt.get("actual_after_apply_counts"))
    if not checks["actual_counts_match_expected"]:
        errors.append("actual_after_apply_counts does not match the authorized runbook expected counts")

    score_ok, score_note = _direct_claim_score_passes(receipt.get("direct_claim_score_path"), repo_root=repo_root)
    checks["direct_claim_score_passes"] = score_ok
    if not score_ok:
        errors.append(score_note)

    checks["service_readback_ok"] = _service_readback_ok(receipt.get("service_readback"))
    if not checks["service_readback_ok"]:
        errors.append("service_readback must show active/running service with MainPID and NRestarts")

    checks["cleanup_readback_ok"] = _cleanup_readback_ok(receipt.get("cleanup_readback"))
    if not checks["cleanup_readback_ok"]:
        errors.append("cleanup_readback must show cleanup_ok=true and no rehearsal/clone/staging DB leftovers")

    checks["known_residual_risks_list"] = isinstance(receipt.get("known_residual_risks"), list)
    if not checks["known_residual_risks_list"]:
        errors.append("known_residual_risks must be a list")
    elif not receipt["known_residual_risks"]:
        warnings.append("known_residual_risks is empty; verify that this is intentional")

    valid = all(checks.values()) and not errors
    return {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "status": "valid" if valid else "invalid",
        "valid": valid,
        "receipt_path": str(receipt_path),
        "runbook_path": str(runbook_path),
        "checks": checks,
        "missing_fields": missing_fields,
        "errors": errors,
        "warnings": warnings,
        "claim_ceiling": (
            "This validates a retained production-apply receipt against the authorized runbook. It does not run SQL, "
            "does not mutate production, and does not create authorization."
        ),
    }


def write_markdown(path: Path, result: dict[str, Any]) -> None:
    lines = [
        "# Working Leo Production Apply Receipt Validation",
        "",
        f"Generated UTC: `{result['generated_at_utc']}`",
        f"Status: `{result['status']}`",
        f"Valid: `{result['valid']}`",
        f"Receipt: `{result['receipt_path']}`",
        f"Runbook: `{result['runbook_path']}`",
        "",
        "## Checks",
        "",
    ]
    for key, value in sorted(result.get("checks", {}).items()):
        lines.append(f"- `{key}`: `{value}`")
    lines.extend(["", "## Errors", ""])
    errors = result.get("errors", [])
    lines.extend(f"- {item}" for item in errors) if errors else lines.append("- none")
    lines.extend(["", "## Warnings", ""])
    warnings = result.get("warnings", [])
    lines.extend(f"- {item}" for item in warnings) if warnings else lines.append("- none")
    lines.extend(["", "## Claim Ceiling", "", result["claim_ceiling"], ""])
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines), encoding="utf-8")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--receipt", type=Path, default=DEFAULT_RECEIPT)
    parser.add_argument("--runbook", type=Path, default=DEFAULT_RUNBOOK)
    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
    parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT)
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    result = validate_receipt(args.receipt, args.runbook)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    write_markdown(args.markdown_out, result)
    print(
        json.dumps(
            {
                "out": str(args.out),
                "markdown_out": str(args.markdown_out),
                "status": result["status"],
                "valid": result["valid"],
            },
            indent=2,
            sort_keys=True,
        )
    )
    return 0 if result["valid"] else 1


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