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