teleo-infrastructure/scripts/working_leo_integrated_packet.py

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