iridium-db 0.4.0

A high-performance vector-graph hybrid storage and indexing engine
#!/usr/bin/env python3
"""Emit Sprint 8 appliance-profile evidence 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-profile evidence artifacts"
    )
    parser.add_argument("--report-dir", default="artifacts")
    parser.add_argument("--report-prefix", default="appliance_profile")
    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_220_PREFIX",
            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-profile",
        help="Scratch appliance root used for appliance execution 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_appliance_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([(1, 1, [2, 3]), (2, 1, [3]), (3, 1, [])])
    client.ingest_edges_batch([(1, 2, "edge-1"), (2, 2, "edge-2")])
    client.ingest_vector(1, 2, [1.0, 0.0])
    client.finish_ingest()
    rows = json.loads(client.query_json("MATCH (n) RETURN n LIMIT 3"))
    return {
        "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_diagnostics(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 Profile Report",
        "",
        f"- schema: `{payload['schema']}`",
        f"- appliance_profile_id: `{payload['package_identity']['appliance_profile_id']}`",
        f"- support_band: `{payload['package_identity']['support_band']}`",
        f"- release_ready: `{str(payload['appliance_gate']['release_ready']).lower()}`",
        "",
        "## Appliance Checks",
    ]
    for name, passed in payload["appliance_gate"]["checks"].items():
        lines.append(f"- `{name}`: `{str(passed).lower()}`")
    lines.extend(
        [
            "",
            "## Evidence Artifacts",
        ]
    )
    for key, value in payload["artifact_set"].items():
        lines.append(f"- `{key}`: `{value}`")
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def write_handoff_markdown(path: Path, payload: Dict[str, Any]) -> None:
    lines = [
        "# Appliance Profile 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.extend(
            [
                f"### {artifact['artifact_id']}",
                f"- path: `{artifact['path']}`",
                f"- 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

    plexus_env = python_env.copy()
    plexus_env["TABLEGEN_210_PREFIX"] = args.llvm_prefix
    plexus_env["MLIR_SYS_210_PREFIX"] = args.llvm_prefix
    plexus_env["LLVM_SYS_210_PREFIX"] = args.llvm_prefix
    plexus_env["TABLEGEN_220_PREFIX"] = args.llvm_prefix
    plexus_env["MLIR_SYS_220_PREFIX"] = args.llvm_prefix
    plexus_env["LLVM_SYS_220_PREFIX"] = args.llvm_prefix
    plexus_env["PLEXUS_BUILD_DIR"] = args.plexus_build_dir
    plexus_env["LIBRARY_PATH"] = args.library_path

    run_command(
        [
            args.python_bin,
            str(ROOT / "scripts" / "flagship_release_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=plexus_env,
    )

    appliance_root = Path(args.work_dir)
    if appliance_root.exists():
        shutil.rmtree(appliance_root)
    live_root = appliance_root / "live"
    live_data = live_root / "data"
    backup_root = appliance_root / "backup"
    restore_root = appliance_root / "restore"
    restore_data = restore_root / "data"
    live_data.mkdir(parents=True, exist_ok=True)

    live_query = build_appliance_fixture(live_data)
    shutil.copytree(live_data, backup_root / "data")
    shutil.copytree(backup_root / "data", restore_data)

    restored_client = iridium.Client(
        data_dir=str(restore_data),
        scan_start=0,
        scan_end_exclusive=64,
        morsel_size=16,
        parallel_workers=0,
    )
    restored_rows = json.loads(restored_client.query_json("MATCH (n) RETURN n LIMIT 3"))
    restored_query = {
        "row_count": len(restored_rows.get("rows", [])),
        "first_node_id": restored_rows.get("rows", [{}])[0].get("node_id", 0)
        if restored_rows.get("rows")
        else 0,
    }

    diagnostics = collect_diagnostics(appliance_root)
    diagnostics_json = report_dir / "appliance_diagnostics_manifest.json"
    diagnostics_json.write_text(json.dumps(diagnostics, indent=2), encoding="utf-8")

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

    appliance_gate = {
        "checks": {
            "flagship_release_ready": bool(flagship["release_gate"]["release_ready"]),
            "backup_materialized": (backup_root / "data").exists(),
            "restore_query_matches_live": live_query == restored_query,
            "diagnostics_manifest_present": diagnostics["file_count"] > 0,
        }
    }
    appliance_gate["release_ready"] = all(appliance_gate["checks"].values())

    payload = {
        "schema": "iridium.appliance-profile-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",
        },
        "flagship_dependency": {
            "schema": flagship["schema"],
            "release_candidate": flagship["release_candidate"],
            "release_ready": flagship["release_gate"]["release_ready"],
        },
        "appliance_execution": {
            "work_dir": str(appliance_root),
            "live_query": live_query,
            "restored_query": restored_query,
        },
        "lifecycle_evidence": {
            "backup_path": str((backup_root / "data").resolve()),
            "restore_path": str(restore_data.resolve()),
            "diagnostics_manifest": diagnostics_json.name,
        },
        "support_boundary": {
            "iridium": "single-node appliance execution, backup/restore evidence, and diagnostics manifest",
            "rhodium": "object-store acceleration posture remains Rhodium-owned and external to this local appliance execution report",
            "alloy": "shared lifecycle, certification, and support-band packaging remain alloy-owned integration inputs",
        },
        "artifact_set": {
            "flagship_release_report": "flagship_release_report.json",
            "flagship_release_handoff": "flagship_release_handoff.json",
            "appliance_diagnostics_manifest": diagnostics_json.name,
        },
        "appliance_gate": appliance_gate,
    }

    handoff = {
        "schema": "iridium.appliance-profile-handoff.v1",
        "package_target": "alloy-appliance-packaging",
        "handoff_ready": appliance_gate["release_ready"],
        "attached_artifacts": [
            {
                "artifact_id": "iridium-flagship-release-report",
                "path": "artifacts/flagship_release_report.json",
                "role": "execution-baseline",
            },
            {
                "artifact_id": "iridium-flagship-release-handoff",
                "path": "artifacts/flagship_release_handoff.json",
                "role": "flagship-packaging-input",
            },
            {
                "artifact_id": "iridium-appliance-diagnostics-manifest",
                "path": f"artifacts/{diagnostics_json.name}",
                "role": "diagnostics-evidence",
            },
        ],
        "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",
        ],
    }

    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 appliance_gate["release_ready"] else 1


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