iridium-db 0.2.0

A high-performance vector-graph hybrid storage and indexing engine
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

#!/usr/bin/env python3
"""Emit service candidate operator artifacts for the Sprint 4 path."""

from __future__ import annotations

import argparse
import json
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 service candidate report artifacts"
    )
    parser.add_argument("--report-dir", default="artifacts")
    parser.add_argument("--report-prefix", default="service_candidate")
    parser.add_argument("--listen", default="127.0.0.1:7001")
    parser.add_argument("--telemetry-endpoint", default="stdout")
    parser.add_argument("--tls", default="operator-optional")
    parser.add_argument("--admin-token", default="local-dev")
    parser.add_argument(
        "--binary",
        default=str(ROOT / "target" / "debug" / "ir"),
        help="Path to the Iridium CLI binary",
    )
    return parser.parse_args(argv)


def run_json(command: list[str]) -> Dict[str, Any]:
    completed = subprocess.run(
        command,
        cwd=ROOT,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=True,
    )
    return json.loads(completed.stdout)


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

    service_report = run_json(
        [
            args.binary,
            "service-report",
            "--listen",
            args.listen,
        ]
    )
    validation_report = run_json(
        [
            args.binary,
            "service-validate",
            "--listen",
            args.listen,
            "--telemetry-endpoint",
            args.telemetry_endpoint,
            "--tls",
            args.tls,
            "--admin-token",
            args.admin_token,
            "--max-requests",
            "4",
        ]
    )

    payload = {
        "schema": "iridium.service-candidate-report.v1",
        "service": service_report,
        "validation": validation_report,
        "operator_inputs": {
            "listen": args.listen,
            "telemetry_endpoint": args.telemetry_endpoint,
            "tls": args.tls,
            "admin_token_configured": bool(args.admin_token),
        },
    }

    json_path = report_dir / f"{args.report_prefix}_report.json"
    md_path = report_dir / f"{args.report_prefix}_report.md"
    json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")

    md_lines = [
        "# Service Candidate Report",
        "",
        f"- schema: `{payload['schema']}`",
        f"- service_scope: `{service_report['service_scope']}`",
        f"- deploy_scope: `{service_report['deploy_scope']}`",
        f"- telemetry_scope: `{service_report['telemetry_scope']}`",
        f"- profile_id: `{service_report['profile_id']}`",
        f"- listen_address: `{service_report['listen_address']}`",
        f"- support_band: `{service_report['support_band']}`",
        f"- validation_compatible: `{str(validation_report['compatible']).lower()}`",
        "",
        "## Operator Inputs",
        f"- telemetry_endpoint: `{args.telemetry_endpoint}`",
        f"- tls: `{args.tls}`",
        f"- admin_token_configured: `{str(bool(args.admin_token)).lower()}`",
        "",
        "## Endpoints",
        f"- liveness: `{service_report['paths']['liveness']}`",
        f"- readiness: `{service_report['paths']['readiness']}`",
        f"- admin_status: `{service_report['paths']['admin_status']}`",
        "",
        "## Validation Warnings",
    ]
    if validation_report["warnings"]:
        for warning in validation_report["warnings"]:
            md_lines.append(f"- {warning}")
    else:
        md_lines.append("- none")

    md_path.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
    print(f"wrote: {json_path}")
    print(f"wrote: {md_path}")
    return 0


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