iridium-db 0.3.0

A high-performance vector-graph hybrid storage and indexing engine
#!/usr/bin/env python3
"""Emit Sprint 8 appliance lifecycle and support-boundary artifacts for Iridium."""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any, Dict, Sequence

import iridium


ROOT = Path(__file__).resolve().parents[1]


def parse_args(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Write the Iridium appliance lifecycle evidence artifacts"
    )
    parser.add_argument("--report-dir", default="artifacts")
    parser.add_argument("--report-prefix", default="appliance_lifecycle")
    parser.add_argument(
        "--python-bin",
        default=os.environ.get("PYTHON_BIN", sys.executable),
        help="Python interpreter to use for Python-backed reporting stages",
    )
    parser.add_argument(
        "--llvm-prefix",
        default=os.environ.get("MLIR_SYS_210_PREFIX", "/opt/homebrew/opt/llvm"),
        help="LLVM/MLIR prefix used by Plexus-backed cargo commands",
    )
    parser.add_argument(
        "--plexus-build-dir",
        default=os.environ.get(
            "PLEXUS_BUILD_DIR",
            str((ROOT.parent / "plexus" / "build").resolve()),
        ),
        help="Path to the built Plexus C++/MLIR artifacts",
    )
    parser.add_argument(
        "--library-path",
        default=os.environ.get(
            "LIBRARY_PATH", "/opt/homebrew/lib:/opt/homebrew/opt/llvm/lib"
        ),
        help="Library path used by Plexus-backed cargo link steps",
    )
    parser.add_argument(
        "--work-dir",
        default="/tmp/iridium-appliance-lifecycle",
        help="Scratch appliance lifecycle root used for upgrade and diagnostics evidence",
    )
    return parser.parse_args(argv)


def run_command(command: list[str], env: dict[str, str] | None = None) -> None:
    subprocess.run(
        command,
        cwd=ROOT,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=True,
        env=env,
    )


def load_json(path: Path) -> Dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def build_fixture(data_dir: Path) -> Dict[str, Any]:
    client = iridium.Client(
        data_dir=str(data_dir),
        scan_start=0,
        scan_end_exclusive=64,
        morsel_size=16,
        parallel_workers=0,
    )
    client.begin_ingest()
    client.ingest_nodes_batch([(10, 1, [20, 30]), (20, 1, [30]), (30, 1, [])])
    client.ingest_edges_batch([(10, 2, "upgrade-1"), (20, 2, "upgrade-2")])
    client.ingest_vector(10, 2, [0.75, 0.25])
    client.finish_ingest()
    del client
    return query_fixture(data_dir)


def query_fixture(data_dir: Path) -> Dict[str, Any]:
    client = iridium.Client(
        data_dir=str(data_dir),
        scan_start=0,
        scan_end_exclusive=64,
        morsel_size=16,
        parallel_workers=0,
    )
    rows = json.loads(client.query_json("MATCH (n) RETURN n LIMIT 3"))
    del client
    return {
        "query": "MATCH (n) RETURN n LIMIT 3",
        "row_count": len(rows.get("rows", [])),
        "first_node_id": rows.get("rows", [{}])[0].get("node_id", 0)
        if rows.get("rows")
        else 0,
    }


def collect_file_manifest(root: Path) -> Dict[str, Any]:
    entries = []
    for path in sorted(root.rglob("*")):
        if path.is_file():
            entries.append(
                {
                    "path": str(path.relative_to(root)),
                    "bytes": path.stat().st_size,
                }
            )
    return {"file_count": len(entries), "entries": entries}


def write_markdown(path: Path, payload: Dict[str, Any]) -> None:
    lines = [
        "# Appliance Lifecycle Report",
        "",
        f"- schema: `{payload['schema']}`",
        f"- appliance_profile_id: `{payload['package_identity']['appliance_profile_id']}`",
        f"- support_band: `{payload['package_identity']['support_band']}`",
        f"- lifecycle_ready: `{str(payload['lifecycle_gate']['release_ready']).lower()}`",
        "",
        "## Lifecycle Checks",
    ]
    for name, passed in payload["lifecycle_gate"]["checks"].items():
        lines.append(f"- `{name}`: `{str(passed).lower()}`")
    lines.extend(
        [
            "",
            "## Upgrade Evidence",
            f"- pre_upgrade_row_count: `{payload['upgrade_evidence']['pre_upgrade_query']['row_count']}`",
            f"- post_upgrade_row_count: `{payload['upgrade_evidence']['post_upgrade_query']['row_count']}`",
            f"- upgrade_mode: `{payload['upgrade_evidence']['upgrade_mode']}`",
            "",
            "## Support Boundary Review",
        ]
    )
    for owner, description in payload["support_boundary_review"]["ownership"].items():
        lines.append(f"- `{owner}`: {description}")
    lines.extend(
        [
            "",
            "## Lifecycle Packet References",
        ]
    )
    for name, artifact in payload["artifact_set"].items():
        lines.append(f"- `{name}`: `{artifact}`")
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def write_handoff_markdown(path: Path, payload: Dict[str, Any]) -> None:
    lines = [
        "# Appliance Lifecycle Handoff",
        "",
        f"- schema: `{payload['schema']}`",
        f"- handoff_ready: `{str(payload['handoff_ready']).lower()}`",
        f"- package_target: `{payload['package_target']}`",
        "",
        "## Attached Artifacts",
    ]
    for artifact in payload["attached_artifacts"]:
        lines.append(
            f"- `{artifact['artifact_id']}`: path=`{artifact['path']}` role=`{artifact['role']}`"
        )
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main(argv: Sequence[str]) -> int:
    args = parse_args(argv)
    report_dir = Path(args.report_dir)
    report_dir.mkdir(parents=True, exist_ok=True)

    python_env = os.environ.copy()
    python_env["PYTHON_BIN"] = args.python_bin
    python_env["TABLEGEN_210_PREFIX"] = args.llvm_prefix
    python_env["MLIR_SYS_210_PREFIX"] = args.llvm_prefix
    python_env["LLVM_SYS_210_PREFIX"] = args.llvm_prefix
    python_env["PLEXUS_BUILD_DIR"] = args.plexus_build_dir
    python_env["LIBRARY_PATH"] = args.library_path

    run_command(
        [
            args.python_bin,
            str(ROOT / "scripts" / "appliance_profile_report.py"),
            "--python-bin",
            args.python_bin,
            "--llvm-prefix",
            args.llvm_prefix,
            "--plexus-build-dir",
            args.plexus_build_dir,
            "--library-path",
            args.library_path,
            "--report-dir",
            str(report_dir),
        ],
        env=python_env,
    )

    lifecycle_root = Path(args.work_dir)
    if lifecycle_root.exists():
        shutil.rmtree(lifecycle_root)

    pre_upgrade_data = lifecycle_root / "pre_upgrade" / "data"
    upgraded_data = lifecycle_root / "upgraded" / "data"
    support_snapshot = lifecycle_root / "support-snapshot"
    pre_upgrade_data.mkdir(parents=True, exist_ok=True)
    support_snapshot.mkdir(parents=True, exist_ok=True)

    pre_upgrade_query = build_fixture(pre_upgrade_data)
    shutil.copytree(pre_upgrade_data, upgraded_data)
    post_upgrade_query = query_fixture(upgraded_data)

    support_snapshot_manifest = collect_file_manifest(lifecycle_root)
    support_snapshot_json = report_dir / "appliance_support_snapshot.json"
    support_snapshot_json.write_text(
        json.dumps(support_snapshot_manifest, indent=2),
        encoding="utf-8",
    )

    appliance = load_json(report_dir / "appliance_profile_report.json")
    flagship = load_json(report_dir / "flagship_release_report.json")

    checks = {
        "flagship_release_ready": bool(flagship["release_gate"]["release_ready"]),
        "appliance_profile_ready": bool(appliance["appliance_gate"]["release_ready"]),
        "upgrade_query_matches_baseline": pre_upgrade_query == post_upgrade_query,
        "support_snapshot_present": support_snapshot_manifest["file_count"] > 0,
        "support_boundary_declared": True,
    }
    lifecycle_gate = {
        "checks": checks,
        "release_ready": all(checks.values()),
    }

    payload = {
        "schema": "iridium.appliance-lifecycle-report.v1",
        "package_identity": {
            "name": "iridium-appliance-candidate",
            "version": "0.1.0",
            "appliance_profile_id": "iridium-appliance-execution-v0-1",
            "maturity_tier": "appliance-candidate",
            "support_band": "candidate",
        },
        "upgrade_evidence": {
            "upgrade_mode": "current-release reopens staged appliance dataset without format migration",
            "pre_upgrade_query": pre_upgrade_query,
            "post_upgrade_query": post_upgrade_query,
        },
        "backup_restore_reference": appliance["lifecycle_evidence"],
        "diagnostics_reference": {
            "appliance_diagnostics_manifest": appliance["lifecycle_evidence"][
                "diagnostics_manifest"
            ],
            "support_snapshot": support_snapshot_json.name,
        },
        "support_boundary_review": {
            "review_reference": "../alloy/docs/support-boundary.md",
            "packaging_workflow_reference": "../alloy/docs/appliance-packaging-workflow.md",
            "ownership": {
                "iridium": "appliance execution evidence, upgrade continuity proof, local backup/restore proof, and diagnostics/support snapshots",
                "rhodium": "acceleration posture and object-store behavior remain Rhodium-owned external seams",
                "strontium": "certification and replay publication remain Strontium-owned workflows",
                "alloy": "shared lifecycle metadata, support-band interpretation, and packaging slots remain Alloy-owned integration inputs",
                "palladium": "bounded-cluster control-plane and certification review remain external to this appliance candidate",
            },
        },
        "artifact_set": {
            "flagship_release_report": "flagship_release_report.json",
            "appliance_profile_report": "appliance_profile_report.json",
            "appliance_profile_handoff": "appliance_profile_handoff.json",
            "appliance_diagnostics_manifest": appliance["lifecycle_evidence"][
                "diagnostics_manifest"
            ],
            "appliance_support_snapshot": support_snapshot_json.name,
        },
        "lifecycle_gate": lifecycle_gate,
    }

    handoff = {
        "schema": "iridium.appliance-lifecycle-handoff.v1",
        "package_target": "alloy-appliance-lifecycle-slot",
        "handoff_ready": lifecycle_gate["release_ready"],
        "attached_artifacts": [
            {
                "artifact_id": "iridium-appliance-profile-report",
                "path": "artifacts/appliance_profile_report.json",
                "role": "appliance-execution-evidence",
            },
            {
                "artifact_id": "iridium-appliance-diagnostics-manifest",
                "path": f"artifacts/{appliance['lifecycle_evidence']['diagnostics_manifest']}",
                "role": "diagnostics-evidence",
            },
            {
                "artifact_id": "iridium-appliance-support-snapshot",
                "path": f"artifacts/{support_snapshot_json.name}",
                "role": "support-snapshot",
            },
            {
                "artifact_id": "iridium-flagship-release-report",
                "path": "artifacts/flagship_release_report.json",
                "role": "upstream-release-baseline",
            },
        ],
        "reproduction_commands": [
            f"{args.python_bin} scripts/flagship_release_report.py --python-bin {args.python_bin} --report-dir artifacts",
            f"{args.python_bin} scripts/appliance_profile_report.py --python-bin {args.python_bin} --report-dir artifacts",
            f"{args.python_bin} scripts/appliance_lifecycle_report.py --python-bin {args.python_bin} --report-dir artifacts",
        ],
    }

    json_path = report_dir / f"{args.report_prefix}_report.json"
    md_path = report_dir / f"{args.report_prefix}_report.md"
    handoff_json = report_dir / f"{args.report_prefix}_handoff.json"
    handoff_md = report_dir / f"{args.report_prefix}_handoff.md"
    json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    write_markdown(md_path, payload)
    handoff_json.write_text(json.dumps(handoff, indent=2), encoding="utf-8")
    write_handoff_markdown(handoff_md, handoff)
    print(f"wrote: {json_path}")
    print(f"wrote: {md_path}")
    print(f"wrote: {handoff_json}")
    print(f"wrote: {handoff_md}")
    return 0 if lifecycle_gate["release_ready"] else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))