teleo-infrastructure/scripts/verify_working_leo_apply_readiness.py

#!/usr/bin/env python3
"""Verify the Working Leo integrated apply packet is ready but not executed.

This is a read-only retained-artifact verifier. It does not connect to
production and does not execute SQL. The purpose is to make the final
production-apply boundary explicit: all packet files, manifest ordering, clone
proof markers, cleanup markers, and "not production-applied" claim ceilings must
line up before an operator treats the packet set as ready for separately
authorized execution.
"""

from __future__ import annotations

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

import working_leo_integrated_packet as integrated_packet

REPO_REPORT_DIR = Path("docs/reports/leo-working-state-20260709")
EXPECTED_COUNTS = {
    "claim_concept_map_links": 1,
    "claim_edges": 18,
    "claim_evidence": 34,
    "claim_governance_gate_links": 2,
    "claims": 12,
    "concept_maps": 1,
    "cross_surface_anchor_updates": 1,
    "helmer_ledger_rows_marked_applied": 1,
    "reasoning_tools": 1,
    "rio_strategy_anchors": 5,
    "rio_strategy_nodes": 2,
    "sources": 28,
}
CLONE_LOG = "working-leo-integrated-clone-rehearsal-current.log"
CLONE_REPORT = "working-leo-integrated-packet-current.md"


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _ordered_markers_present(text: str, markers: list[str]) -> tuple[bool, list[str]]:
    missing: list[str] = []
    cursor = 0
    for marker in markers:
        index = text.find(marker, cursor)
        if index < 0:
            missing.append(marker)
        else:
            cursor = index + len(marker)
    return not missing, missing


def _read_manifest(report_dir: Path) -> dict[str, Any]:
    path = report_dir / "working-leo-integrated-packet.json"
    if not path.exists():
        raise FileNotFoundError(path)
    return json.loads(path.read_text(encoding="utf-8"))


def _file_check(path: Path, *, kind: str) -> dict[str, Any]:
    exists = path.exists()
    result: dict[str, Any] = {
        "path": str(path),
        "exists": exists,
        "kind": kind,
        "ok": False,
    }
    if not exists:
        result["error"] = "missing"
        return result

    data = path.read_bytes()
    result.update(
        {
            "bytes": len(data),
            "sha256": hashlib.sha256(data).hexdigest(),
            "non_empty": len(data) > 0,
        }
    )
    if kind == "json":
        try:
            json.loads(data.decode("utf-8"))
            result["json_valid"] = True
        except json.JSONDecodeError as exc:
            result["json_valid"] = False
            result["error"] = str(exc)
        result["ok"] = bool(result["non_empty"] and result.get("json_valid"))
        return result

    text = data.decode("utf-8")
    result.update(
        {
            "starts_transaction": bool(re.search(r"(^|\n)\s*(\\set\s+ON_ERROR_STOP\s+on\s*)?begin;", text, re.I)),
            "ends_commit": bool(re.search(r"commit;\s*$", text, re.I)),
        }
    )
    result["ok"] = bool(result["non_empty"] and result["starts_transaction"] and result["ends_commit"])
    return result


def build_readiness(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]:
    manifest = _read_manifest(report_dir)
    expected_apply_order = [packet["id"] for packet in integrated_packet.PACKETS]
    expected_rollback_order = list(reversed(expected_apply_order))
    clone_log_path = report_dir / CLONE_LOG
    clone_report_path = report_dir / CLONE_REPORT
    clone_log = clone_log_path.read_text(encoding="utf-8") if clone_log_path.exists() else ""
    clone_report = clone_report_path.read_text(encoding="utf-8") if clone_report_path.exists() else ""

    file_checks: list[dict[str, Any]] = []
    for packet in manifest.get("packets", []):
        file_checks.append(_file_check(report_dir / packet["apply_sql"], kind="sql"))
        file_checks.append(_file_check(report_dir / packet["rollback_sql"], kind="sql"))
        file_checks.append(_file_check(report_dir / packet["packet_json"], kind="json"))

    apply_markers = [
        f"=== clone apply {index} {packet_id} ==="
        for index, packet_id in enumerate(expected_apply_order, start=1)
    ]
    rollback_markers = [
        f"=== clone rollback {index} {packet_id} ==="
        for index, packet_id in enumerate(expected_rollback_order, start=1)
    ]
    apply_markers_ok, missing_apply_markers = _ordered_markers_present(clone_log, apply_markers)
    rollback_markers_ok, missing_rollback_markers = _ordered_markers_present(clone_log, rollback_markers)
    cleanup_db_dropped = bool(re.search(r"disposable_db_exists\s*-+\s*\n\s*0\s*\n", clone_log))
    service_stable = all(marker in clone_log for marker in ["MainPID=3252143", "NRestarts=0", "ActiveState=active"])
    report_says_not_modified = all(
        phrase in clone_report
        for phrase in [
            "Production was not modified.",
            "no production apply SQL was executed",
            "Production apply still requires explicit authorization",
        ]
    )

    checks = {
        "manifest_status_not_executed": manifest.get("production_apply_status") == "not_executed",
        "apply_order_matches_manifest": manifest.get("apply_order") == expected_apply_order,
        "rollback_order_matches_manifest": manifest.get("rollback_order") == expected_rollback_order,
        "expected_counts_match": manifest.get("expected_after_apply_counts") == EXPECTED_COUNTS,
        "all_packet_files_ok": all(item["ok"] for item in file_checks),
        "clone_apply_markers_ordered": apply_markers_ok,
        "clone_rollback_markers_ordered": rollback_markers_ok,
        "clone_cleanup_db_dropped": cleanup_db_dropped,
        "service_stable_after_rehearsal": service_stable,
        "clone_report_says_production_not_modified": report_says_not_modified,
    }
    ready = all(checks.values())
    return {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "mode": "retained_artifact_readiness_verify",
        "report_dir": str(report_dir),
        "ready_for_authorized_production_apply_packet": ready,
        "production_apply_executed": False,
        "production_apply_authorization_present": False,
        "claim_ceiling": (
            "Ready means the retained packet set is complete, ordered, clone-proven, and cleanly rolled back. "
            "It does not authorize or prove production DB mutation."
        ),
        "checks": checks,
        "missing_apply_markers": missing_apply_markers,
        "missing_rollback_markers": missing_rollback_markers,
        "file_checks": file_checks,
        "clone_log": {
            "path": str(clone_log_path),
            "exists": clone_log_path.exists(),
            "sha256": _sha256(clone_log_path) if clone_log_path.exists() else None,
        },
        "clone_report": {
            "path": str(clone_report_path),
            "exists": clone_report_path.exists(),
            "sha256": _sha256(clone_report_path) if clone_report_path.exists() else None,
        },
        "not_done": [
            "Production apply still requires explicit operator authorization.",
            "Live Telegram regression must be rerun after any production apply.",
            "GCP parity remains separate from this VPS packet-readiness gate.",
        ],
    }


def write_markdown_report(path: Path, readiness: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "# Working Leo Apply Readiness",
        "",
        f"Generated UTC: `{readiness['generated_at_utc']}`",
        f"Mode: `{readiness['mode']}`",
        f"Ready for separately authorized production apply packet: `{readiness['ready_for_authorized_production_apply_packet']}`",
        f"Production apply executed: `{readiness['production_apply_executed']}`",
        f"Production apply authorization present: `{readiness['production_apply_authorization_present']}`",
        "",
        "## Checks",
        "",
    ]
    lines.extend(f"- `{key}`: `{value}`" for key, value in readiness["checks"].items())
    lines.extend(["", "## Packet Files", ""])
    for item in readiness["file_checks"]:
        lines.append(
            f"- `{Path(item['path']).name}`: `ok={item['ok']}`, `bytes={item.get('bytes')}`, "
            f"`sha256={item.get('sha256')}`"
        )
    lines.extend(
        [
            "",
            "## Evidence",
            "",
            f"- Clone log: `{readiness['clone_log']['path']}`",
            f"- Clone report: `{readiness['clone_report']['path']}`",
            "",
            "## Claim Ceiling",
            "",
            readiness["claim_ceiling"],
            "",
            "## Not Done",
            "",
        ]
    )
    lines.extend(f"- {item}" for item in readiness["not_done"])
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--report-dir", type=Path, default=REPO_REPORT_DIR)
    parser.add_argument(
        "--out",
        type=Path,
        default=REPO_REPORT_DIR / "working-leo-apply-readiness-current.json",
    )
    parser.add_argument(
        "--markdown-out",
        type=Path,
        default=REPO_REPORT_DIR / "working-leo-apply-readiness-current.md",
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    readiness = build_readiness(args.report_dir)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(readiness, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    write_markdown_report(args.markdown_out, readiness)
    print(
        json.dumps(
            {
                "out": str(args.out),
                "markdown_out": str(args.markdown_out),
                "ready": readiness["ready_for_authorized_production_apply_packet"],
                "production_apply_executed": readiness["production_apply_executed"],
            },
            indent=2,
            sort_keys=True,
        )
    )
    return 0 if readiness["ready_for_authorized_production_apply_packet"] else 1


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