#!/usr/bin/env python3 """Build the Working Leo authorized production-apply runbook. This is a runbook generator, not an executor. It reads the retained integrated packet manifest and readiness proof, then emits the exact preflight/apply/ postflight/rollback order an operator can use after explicit production apply authorization. It never opens a database connection and never runs SQL. """ from __future__ import annotations import argparse import hashlib import json from datetime import datetime, timezone from pathlib import Path from typing import Any import verify_working_leo_apply_readiness as readiness import working_leo_integrated_packet as integrated_packet REPO_REPORT_DIR = Path("docs/reports/leo-working-state-20260709") DEFAULT_OUT = REPO_REPORT_DIR / "working-leo-authorized-apply-runbook-current.json" DEFAULT_MARKDOWN_OUT = REPO_REPORT_DIR / "working-leo-authorized-apply-runbook-current.md" PHASE_FILES: dict[str, dict[str, str]] = { "mapped-rich-base": { "preflight": "production-preflight.sql", "apply": "production-apply-authorized-commit.sql", "postflight": "production-postflight.sql", "rollback": "production-delete-rollback.sql", }, "cross-surface-anchor": { "preflight": "cross-surface-preflight.sql", "apply": "cross-surface-apply-authorized-commit.sql", "postflight": "cross-surface-postflight.sql", "rollback": "cross-surface-delete-rollback.sql", }, "rio-strategy-context": { "preflight": "rio-strategy-preflight.sql", "apply": "rio-strategy-apply-authorized-commit.sql", "postflight": "rio-strategy-postflight.sql", "rollback": "rio-strategy-delete-rollback.sql", }, "governance-concept-schema": { "preflight": "governance-concept-preflight.sql", "apply": "governance-concept-apply-authorized-commit.sql", "postflight": "governance-concept-postflight.sql", "rollback": "governance-concept-delete-rollback.sql", }, "helmer-7powers": { "preflight": "helmer-7powers-production-preflight.sql", "apply": "helmer-7powers-production-apply-authorized-commit.sql", "postflight": "helmer-7powers-production-postflight.sql", "rollback": "helmer-7powers-production-delete-rollback.sql", }, } POST_APPLY_REGRESSION_COMMANDS: list[dict[str, Any]] = [ { "id": "direct_claim_handler_suite", "description": "Re-run the non-posting VPS GatewayRunner direct-claim suite after production postflight.", "command_template": ".venv/bin/python scripts/run_leo_direct_claim_handler_suite.py", "expected_artifact": str(REPO_REPORT_DIR / "telegram-handler-direct-claim-suite-current.json"), "mutates_production_db": False, "posts_to_telegram": False, }, { "id": "direct_claim_strict_score", "description": "Strict-score the retained direct-claim replies; expected score is 6/6.", "command_template": ( ".venv/bin/python scripts/working_leo_open_ended_benchmark.py " "--results-json docs/reports/leo-working-state-20260709/telegram-handler-direct-claim-suite-current.json " "--include-direct-claim-followups " "--out docs/reports/leo-working-state-20260709/telegram-handler-direct-claim-suite-score-current.json " "--markdown-out docs/reports/leo-working-state-20260709/telegram-handler-direct-claim-suite-score-current.md" ), "expected_artifact": str(REPO_REPORT_DIR / "telegram-handler-direct-claim-suite-score-current.md"), "mutates_production_db": False, "posts_to_telegram": False, }, { "id": "focused_local_regression_tests", "description": "Verify benchmark scoring, bridge skill surfaces, and runbook invariants after artifact refresh.", "command_template": ( ".venv/bin/python -m pytest " "tests/test_working_leo_open_ended_benchmark.py " "tests/test_hermes_leoclean_skill_surfaces.py " "tests/test_build_working_leo_authorized_apply_runbook.py -q" ), "expected_artifact": "terminal pytest output", "mutates_production_db": False, "posts_to_telegram": False, }, { "id": "gateway_service_readback", "description": "Read back the VPS Leo gateway service state after postflight and regressions.", "command_template": ( "systemctl show leoclean-gateway.service " "-p ActiveState -p SubState -p MainPID -p NRestarts -p ExecMainStartTimestamp --no-pager" ), "expected_artifact": "retained service readback with ActiveState=active and NRestarts reviewed", "mutates_production_db": False, "posts_to_telegram": False, }, { "id": "production_receipt_validator", "description": "Validate the retained production-apply receipt against the authorized runbook requirements.", "command_template": ( ".venv/bin/python scripts/validate_working_leo_production_receipt.py " "--receipt docs/reports/leo-working-state-20260709/working-leo-production-apply-receipt-current.json" ), "expected_artifact": str(REPO_REPORT_DIR / "working-leo-production-apply-receipt-validation-current.md"), "mutates_production_db": False, "posts_to_telegram": False, }, ] PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS: list[str] = [ "explicit_authorization_record", "operator_identity", "git_commit_sha", "preflight_output_paths", "apply_output_paths", "postflight_output_paths", "expected_after_apply_counts", "actual_after_apply_counts", "telegram_or_handler_regression_paths", "direct_claim_score_path", "service_readback", "cleanup_readback", "rollback_packet_paths", "known_residual_risks", ] def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _file_entry(report_dir: Path, packet_id: str, phase: str) -> dict[str, Any]: filename = PHASE_FILES[packet_id][phase] path = report_dir / filename exists = path.exists() entry: dict[str, Any] = { "packet_id": packet_id, "phase": phase, "file": filename, "path": str(path), "exists": exists, "ok": False, } if not exists: entry["error"] = "missing" return entry text = path.read_text(encoding="utf-8") entry.update( { "bytes": len(text.encode("utf-8")), "sha256": _sha256(path), "has_transaction_boundary": "begin;" in text.lower() and "commit;" in text.lower(), "command_template": f'psql "$TELEO_DATABASE_URL" -v ON_ERROR_STOP=1 -f {path}', } ) entry["ok"] = bool(entry["bytes"] > 0) return entry def _phase_entries(report_dir: Path, packet_ids: list[str], phase: str) -> list[dict[str, Any]]: return [_file_entry(report_dir, packet_id, phase) for packet_id in packet_ids] def build_runbook(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]: manifest = integrated_packet.build_manifest(report_dir) readiness_report = readiness.build_readiness(report_dir) apply_order = manifest["apply_order"] rollback_order = manifest["rollback_order"] preflight = _phase_entries(report_dir, apply_order, "preflight") apply = _phase_entries(report_dir, apply_order, "apply") postflight = _phase_entries(report_dir, apply_order, "postflight") rollback = _phase_entries(report_dir, rollback_order, "rollback") all_files = preflight + apply + postflight + rollback checks = { "readiness_passes": readiness_report["ready_for_authorized_production_apply_packet"] is True, "readiness_says_not_executed": readiness_report["production_apply_executed"] is False, "readiness_says_not_authorized": readiness_report["production_apply_authorization_present"] is False, "apply_order_matches_manifest": apply_order == [packet["id"] for packet in integrated_packet.PACKETS], "rollback_order_matches_manifest": rollback_order == list(reversed(apply_order)), "all_phase_files_exist": all(item["exists"] for item in all_files), "all_phase_files_non_empty": all(item["ok"] for item in all_files), "all_apply_and_rollback_sql_have_transactions": all( item.get("has_transaction_boundary") is True for item in apply + rollback ), "post_apply_regression_commands_present": all( item.get("command_template") and item.get("expected_artifact") for item in POST_APPLY_REGRESSION_COMMANDS ), "production_apply_receipt_schema_present": all(PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS), } ready = all(checks.values()) return { "generated_at_utc": datetime.now(timezone.utc).isoformat(), "mode": "operator_runbook_generation_only", "report_dir": str(report_dir), "status": "ready_not_executed" if ready else "not_ready", "ready_for_authorized_operator": ready, "production_apply_executed": readiness_report["production_apply_executed"], "production_apply_authorization_present": readiness_report["production_apply_authorization_present"], "required_authorization_boundary": ( "Do not run apply commands unless the operator explicitly authorizes production DB apply for the integrated " "Working Leo packet set in dependency order." ), "operator_environment": { "required_env": ["TELEO_DATABASE_URL"], "runbook_generation_runs_sql": False, "operator_commands_run_sql_if_executed": True, "runbook_only": True, }, "checks": checks, "expected_after_apply_counts": manifest["expected_after_apply_counts"], "phases": { "preflight_order": preflight, "apply_order": apply, "postflight_order": postflight, "rollback_order": rollback, }, "post_apply_required_actions": [ "Run every postflight SQL file and retain output.", "Run the post-apply regression command bundle and retain every artifact.", "Record production apply receipt, row counts, service MainPID/NRestarts, and cleanup readback.", ], "post_apply_regression_commands": POST_APPLY_REGRESSION_COMMANDS, "production_apply_receipt_required_fields": PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS, "claim_ceiling": ( "This runbook is proof that the authorized apply sequence is explicit and ordered. It does not execute SQL, " "does not mutate production, and does not replace explicit production apply authorization." ), } def write_markdown(path: Path, runbook: dict[str, Any]) -> None: lines = [ "# Working Leo Authorized Apply Runbook", "", f"Generated UTC: `{runbook['generated_at_utc']}`", f"Mode: `{runbook['mode']}`", f"Status: `{runbook['status']}`", f"Ready for authorized operator: `{runbook['ready_for_authorized_operator']}`", f"Production apply executed: `{runbook['production_apply_executed']}`", f"Production apply authorization present: `{runbook['production_apply_authorization_present']}`", "", "## Authorization Boundary", "", runbook["required_authorization_boundary"], "", "## Checks", "", ] lines.extend(f"- `{key}`: `{value}`" for key, value in runbook["checks"].items()) for title, key in [ ("Preflight Order", "preflight_order"), ("Apply Order", "apply_order"), ("Postflight Order", "postflight_order"), ("Rollback Order", "rollback_order"), ]: lines.extend(["", f"## {title}", ""]) for index, entry in enumerate(runbook["phases"][key], start=1): lines.append(f"{index}. `{entry['packet_id']}` -> `{entry['file']}`") lines.append(f" - `{entry['command_template']}`") lines.extend(["", "## Expected Counts After Apply", ""]) lines.extend(f"- `{key}`: `{value}`" for key, value in sorted(runbook["expected_after_apply_counts"].items())) lines.extend(["", "## Post-Apply Required Actions", ""]) lines.extend(f"- {item}" for item in runbook["post_apply_required_actions"]) lines.extend(["", "## Post-Apply Regression Commands", ""]) for item in runbook["post_apply_regression_commands"]: lines.append(f"- `{item['id']}`: {item['description']}") lines.append(f" - command: `{item['command_template']}`") lines.append(f" - artifact: `{item['expected_artifact']}`") lines.append(f" - mutates production DB: `{item['mutates_production_db']}`") lines.append(f" - posts to Telegram: `{item['posts_to_telegram']}`") lines.extend(["", "## Production Apply Receipt Required Fields", ""]) lines.extend(f"- `{item}`" for item in runbook["production_apply_receipt_required_fields"]) lines.extend(["", "## Claim Ceiling", "", runbook["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("--report-dir", type=Path, default=REPO_REPORT_DIR) 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) runbook = build_runbook(args.report_dir) args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(runbook, indent=2, sort_keys=True) + "\n", encoding="utf-8") write_markdown(args.markdown_out, runbook) print( json.dumps( { "out": str(args.out), "markdown_out": str(args.markdown_out), "status": runbook["status"], "ready_for_authorized_operator": runbook["ready_for_authorized_operator"], "production_apply_executed": runbook["production_apply_executed"], }, indent=2, sort_keys=True, ) ) return 0 if runbook["ready_for_authorized_operator"] else 1 if __name__ == "__main__": raise SystemExit(main())