#!/usr/bin/env python3 """Build the integrated Working Leo DB packet manifest. This manifest composes the individually clone-proven packets into the only safe dependency order for a full disposable-clone rehearsal. It is still a packet builder, not an executor. Production apply requires a separate explicit authorization boundary. """ from __future__ import annotations import argparse import json from pathlib import Path from typing import Any REPO_REPORT_DIR = Path("docs/reports/leo-working-state-20260709") PACKETS = [ { "id": "mapped-rich-base", "apply_sql": "production-apply-authorized-commit.sql", "rollback_sql": "production-delete-rollback.sql", "packet_json": "production-apply-packet.json", "reason": "Creates the base claim/source/evidence/claim-edge rows required by cross-surface, Rio, and governance/concept packets.", }, { "id": "cross-surface-anchor", "apply_sql": "cross-surface-apply-authorized-commit.sql", "rollback_sql": "cross-surface-delete-rollback.sql", "packet_json": "cross-surface-packet.json", "reason": "Moves the active Leo policy anchor from the retired compressed internet-finance claim to Claim D.", }, { "id": "rio-strategy-context", "apply_sql": "rio-strategy-apply-authorized-commit.sql", "rollback_sql": "rio-strategy-delete-rollback.sql", "packet_json": "rio-strategy-packet.json", "reason": "Creates Rio strategy nodes and anchors that depend on mapped base claims A/C/D.", }, { "id": "governance-concept-schema", "apply_sql": "governance-concept-apply-authorized-commit.sql", "rollback_sql": "governance-concept-delete-rollback.sql", "packet_json": "governance-concept-packet.json", "reason": "Creates minimal concept-map and governance-link schema/rows that depend on mapped base claims.", }, { "id": "helmer-7powers", "apply_sql": "helmer-7powers-production-apply-authorized-commit.sql", "rollback_sql": "helmer-7powers-production-delete-rollback.sql", "packet_json": "helmer-7powers-production-apply-packet.json", "reason": "Applies the independent full-coverage Helmer 7 Powers proposal and ledger flip.", }, ] def _read_json(report_dir: Path, filename: str) -> dict[str, Any]: path = report_dir / filename if not path.exists(): raise FileNotFoundError(path) return json.loads(path.read_text(encoding="utf-8")) def _assert_packet_files_exist(report_dir: Path) -> None: missing: list[str] = [] for packet in PACKETS: for key in ("apply_sql", "rollback_sql", "packet_json"): path = report_dir / packet[key] if not path.exists(): missing.append(str(path)) if missing: raise FileNotFoundError("missing integrated packet files: " + ", ".join(missing)) def build_manifest(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]: _assert_packet_files_exist(report_dir) base = _read_json(report_dir, "production-apply-packet.json") helmer = _read_json(report_dir, "helmer-7powers-production-apply-packet.json") rio = _read_json(report_dir, "rio-strategy-packet.json") governance = _read_json(report_dir, "governance-concept-packet.json") cross_surface = _read_json(report_dir, "cross-surface-packet.json") base_counts = base["row_counts"] helmer_counts = helmer["row_counts"] return { "manifest_version": 1, "purpose": "Integrated production-order rehearsal manifest for Working Leo DB packets.", "production_apply_status": "not_executed", "report_dir": str(report_dir), "apply_order": [packet["id"] for packet in PACKETS], "rollback_order": [packet["id"] for packet in reversed(PACKETS)], "packets": PACKETS, "expected_after_apply_counts": { "claims": base_counts.get("claims", 0) + helmer_counts.get("claims", 0), "sources": base_counts.get("sources", 0) + helmer_counts.get("sources", 0), "claim_edges": base_counts.get("claim_edges", 0) + helmer_counts.get("claim_edges", 0), "claim_evidence": base_counts.get("claim_evidence", 0) + helmer_counts.get("claim_evidence", 0), "reasoning_tools": helmer_counts.get("reasoning_tools", 0), "cross_surface_anchor_updates": len(cross_surface.get("safe_writes", [])), "rio_strategy_nodes": rio["generated_counts"]["strategy_nodes"], "rio_strategy_anchors": rio["generated_counts"]["strategy_node_anchors"], "concept_maps": governance["generated_counts"]["concept_maps"], "claim_concept_map_links": governance["generated_counts"]["claim_concept_map_links"], "claim_governance_gate_links": governance["generated_counts"]["claim_governance_gate_links"], "helmer_ledger_rows_marked_applied": len(helmer.get("proposal_ids", [])), }, "dependency_notes": [ "mapped-rich-base must precede cross-surface-anchor, rio-strategy-context, and governance-concept-schema.", "cross-surface-anchor must roll back before mapped-rich-base so Claim D still exists while the anchor is restored.", "rio-strategy-context and governance-concept-schema must roll back before mapped-rich-base because their rows reference mapped base claims.", "helmer-7powers is independent but included in the integrated rehearsal because it is an approved full-coverage KB packet.", ], "not_done_conditions": [ "Production canonical rows/schema remain unapplied until explicit production apply authorization.", "GCP parity remains unproven until account auth allows read-only GCP service/DB readback.", "Live Telegram canaries remain regression proof for current VPS behavior and should be rerun after any production apply.", ], } def write_manifest_files(manifest: dict[str, Any], output_dir: Path) -> None: output_dir.mkdir(parents=True, exist_ok=True) (output_dir / "working-leo-integrated-packet.json").write_text( json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) lines = [ "# Working Leo Integrated Packet", "", "Production apply status: `not_executed`.", "", "## Apply Order", ] lines.extend(f"{idx}. `{packet_id}`" for idx, packet_id in enumerate(manifest["apply_order"], start=1)) lines.extend(["", "## Rollback Order"]) lines.extend(f"{idx}. `{packet_id}`" for idx, packet_id in enumerate(manifest["rollback_order"], start=1)) lines.extend(["", "## Expected Counts After Apply", ""]) lines.extend( f"- `{key}`: `{value}`" for key, value in sorted(manifest["expected_after_apply_counts"].items()) ) lines.extend(["", "## Dependency Notes", ""]) lines.extend(f"- {note}" for note in manifest["dependency_notes"]) lines.extend(["", "## Not Done", ""]) lines.extend(f"- {item}" for item in manifest["not_done_conditions"]) (output_dir / "working-leo-integrated-packet.md").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("--output-dir", type=Path) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) manifest = build_manifest(args.report_dir) if args.output_dir: write_manifest_files(manifest, args.output_dir) print(json.dumps(manifest, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())