iridium-db 0.3.0

A high-performance vector-graph hybrid storage and indexing engine
#!/usr/bin/env python3
"""Emit Sprint 7 flagship release evidence and handoff artifacts for Iridium."""

from __future__ import annotations

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


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


def parse_args(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Write the Iridium flagship release evidence and handoff artifacts"
    )
    parser.add_argument("--report-dir", default="artifacts")
    parser.add_argument("--report-prefix", default="flagship_release")
    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(
        "--binary",
        default=str(ROOT / "target" / "debug" / "ir"),
        help="Path to the Iridium CLI binary",
    )
    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 run_and_capture(
    command: list[str], log_path: Path, env: dict[str, str] | None = None
) -> int:
    completed = subprocess.run(
        command,
        cwd=ROOT,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        check=False,
        env=env,
    )
    log_path.write_text(completed.stdout, encoding="utf-8")
    return completed.returncode


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


def write_markdown(path: Path, payload: Dict[str, Any]) -> None:
    lines = [
        "# Flagship Release Report",
        "",
        f"- schema: `{payload['schema']}`",
        f"- release_candidate: `{payload['release_candidate']}`",
        f"- release_ready: `{str(payload['release_gate']['release_ready']).lower()}`",
        f"- topology: `{payload['topology']}`",
        "",
        "## Release Checks",
    ]
    for name, passed in payload["release_gate"]["checks"].items():
        lines.append(f"- `{name}`: `{str(passed).lower()}`")
    lines.extend(
        [
            "",
            "## Artifact Set",
        ]
    )
    for key, value in payload["artifact_set"].items():
        lines.append(f"- `{key}`: `{value}`")
    lines.extend(
        [
            "",
            "## Dependency Boundaries",
        ]
    )
    for key, value in payload["dependency_boundaries"].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 = [
        "# Flagship Release Handoff",
        "",
        f"- schema: `{payload['schema']}`",
        f"- package_target: `{payload['package_target']}`",
        f"- handoff_ready: `{str(payload['handoff_ready']).lower()}`",
        "",
        "## Attached Artifacts",
    ]
    for artifact in payload["attached_artifacts"]:
        lines.extend(
            [
                f"### {artifact['artifact_id']}",
                f"- path: `{artifact['path']}`",
                f"- role: `{artifact['role']}`",
                f"- producer: `{artifact['producer']}`",
                "",
            ]
        )
    lines.extend(
        [
            "## Reproduction Commands",
        ]
    )
    for command in payload["reproduction_commands"]:
        lines.append(f"- `{command}`")
    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["PLEXUS_BUILD_DIR"] = args.plexus_build_dir
    plexus_env["LIBRARY_PATH"] = args.library_path

    run_command(
        [
            args.python_bin,
            str(ROOT / "scripts" / "integrated_harness_report.py"),
            "--python-bin",
            args.python_bin,
            "--report-dir",
            str(report_dir),
        ],
        env=python_env,
    )
    run_command(
        [
            args.python_bin,
            str(ROOT / "scripts" / "service_install_report.py"),
            "--report-dir",
            str(report_dir),
            "--binary",
            args.binary,
        ],
        env=python_env,
    )

    compiled_plan_log = report_dir / f"{args.report_prefix}_compiled_plan_refresh.log"
    compiled_plan_status = run_and_capture(
        [
            "cargo",
            "test",
            "--test",
            "rhodium_compiled_plan_reference",
            "--features",
            "rhodium-backend",
            "--",
            "--nocapture",
        ],
        compiled_plan_log,
        env=plexus_env,
    )

    integrated = load_json(report_dir / "integrated_harness_report.json")
    service_install = load_json(report_dir / "service_install_report.json")
    service_compatibility = load_json(report_dir / "service_compatibility_report.json")
    recovery_gate = load_json(report_dir / "core_recovery_gate_report.json")
    cache_report_path = report_dir / "rhodium_compiled_plan_reference_report.json"
    if compiled_plan_status != 0 and not cache_report_path.exists():
        raise SystemExit(
            "compiled-plan cache refresh failed and no existing reference artifact is available"
        )
    cache_report = load_json(cache_report_path)
    cache_refresh_status = (
        "refreshed" if compiled_plan_status == 0 else "reused-existing-artifact"
    )

    release_gate = {
        "checks": {
            "integrated_harness_pass": bool(integrated["overall_pass"]),
            "service_package_compatible": bool(
                service_compatibility["compatibility_gate"]["overall_compatible"]
            ),
            "recovery_gate_pass": bool(recovery_gate["passed"]),
            "compiled_plan_cache_ready": cache_report["disposition"]
            == "cache-miss-rehydrated",
        }
    }
    release_gate["release_ready"] = all(release_gate["checks"].values())

    payload = {
        "schema": "iridium.flagship-release-report.v1",
        "release_candidate": "graphrag-reference-stack-v0",
        "topology": "flagship-graphrag-reference-v0",
        "release_gate": release_gate,
        "integrated_execution": {
            "schema": integrated["schema"],
            "runtime_contract": integrated["runtime_contract"],
            "stages": integrated["stages"],
        },
        "recovery_baseline": recovery_gate["metrics"],
        "service_package": {
            "schema": service_install["schema"],
            "package_identity": service_install["package_identity"],
            "compatibility_gate": service_install["compatibility_gate"],
        },
        "compiled_plan_cache": {
            "schema": cache_report["schema"],
            "refresh_status": cache_refresh_status,
            "cache_namespace": cache_report["cache_namespace"],
            "cache_key": cache_report["cache_key"],
            "hydrated_tier": cache_report["hydrated_tier"],
            "retrieval_artifact_key": cache_report["retrieval_artifact_key"],
            "query_paths": cache_report["query_paths"],
            "artifact_paths": cache_report["artifact_paths"],
        },
        "artifact_set": {
            "integrated_harness": "integrated_harness_report.json",
            "integrated_harness_markdown": "integrated_harness_report.md",
            "service_install": "service_install_report.json",
            "service_install_markdown": "service_install_report.md",
            "service_compatibility": "service_compatibility_report.json",
            "recovery_gate": "core_recovery_gate_report.json",
            "compiled_plan_cache": "rhodium_compiled_plan_reference_report.json",
            "compiled_plan_cache_markdown": "rhodium_compiled_plan_reference_report.md",
            "compiled_plan_refresh_log": compiled_plan_log.name,
        },
        "dependency_boundaries": {
            "plexus": "query/profile assumptions and compiled-plan cache contract remain Plexus-owned inputs",
            "rhodium": "compiled-plan and retrieval artifact cache storage remain Rhodium-owned producer seams",
            "strontium": "fault and replay publication remain Strontium-owned packaging inputs",
            "palladium": "bounded-cluster control-plane integration remains a Palladium-owned boundary",
            "alloy": "shared flagship release packaging remains alloy-owned aggregation work",
        },
    }

    handoff = {
        "schema": "iridium.flagship-release-handoff.v1",
        "package_target": "alloy-flagship-packaging",
        "handoff_ready": payload["release_gate"]["release_ready"],
        "release_candidate": payload["release_candidate"],
        "attached_artifacts": [
            {
                "artifact_id": "iridium-integrated-harness",
                "path": "artifacts/integrated_harness_report.json",
                "role": "execution-evidence",
                "producer": "iridium",
            },
            {
                "artifact_id": "iridium-service-install",
                "path": "artifacts/service_install_report.json",
                "role": "package-baseline",
                "producer": "iridium",
            },
            {
                "artifact_id": "iridium-service-compatibility",
                "path": "artifacts/service_compatibility_report.json",
                "role": "compatibility-baseline",
                "producer": "iridium",
            },
            {
                "artifact_id": "iridium-core-recovery-gate",
                "path": "artifacts/core_recovery_gate_report.json",
                "role": "recovery-baseline",
                "producer": "iridium",
            },
            {
                "artifact_id": "rhodium-compiled-plan-cache-reference",
                "path": "artifacts/rhodium_compiled_plan_reference_report.json",
                "role": "cache-baseline",
                "producer": "iridium-consumer-of-rhodium",
            },
        ],
        "reproduction_commands": [
            f"{args.python_bin} scripts/integrated_harness_report.py --python-bin {args.python_bin} --report-dir artifacts",
            f"{args.python_bin} scripts/service_install_report.py --report-dir artifacts",
            "cargo test --test rhodium_compiled_plan_reference --features rhodium-backend -- --nocapture",
        ],
        "dependency_notes": payload["dependency_boundaries"],
    }

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


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