23b1c87ac0
Some checks are pending
CI / lint-and-test (push) Waiting to run
Generate ready-not-executed operator runbook for the integrated apply packet set. Extend Cory-style benchmark with no-context direct-claim prompts and expected follow-ups. Refresh retained reports and tests for the 20-prompt mixed-evidence score.
226 lines
9.6 KiB
Python
226 lines
9.6 KiB
Python
#!/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",
|
|
},
|
|
}
|
|
|
|
|
|
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
|
|
),
|
|
}
|
|
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.",
|
|
"Rerun live Telegram regression canaries after production postflight.",
|
|
"Record production apply receipt, row counts, service MainPID/NRestarts, and cleanup readback.",
|
|
],
|
|
"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(["", "## 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())
|