from __future__ import annotations
import argparse
import json
import socket
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 compatibility report artifacts"
)
parser.add_argument("--report-dir", default="artifacts")
parser.add_argument("--report-prefix", default="service_compatibility")
parser.add_argument("--listen", default="127.0.0.1:0")
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(
"--work-dir",
default="/tmp/iridium-service-retrieval-quality",
help="Scratch directory used for the service-backed retrieval evaluation",
)
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 run_command(command: list[str]) -> None:
subprocess.run(
command,
cwd=ROOT,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True,
)
def load_json(path: Path) -> Dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def resolve_listen(listen: str) -> str:
if not listen.endswith(":0"):
return listen
host = listen.rsplit(":", 1)[0]
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind((host, 0))
return f"{host}:{sock.getsockname()[1]}"
def write_markdown(path: Path, payload: Dict[str, Any]) -> None:
md_lines = [
"# Service Compatibility Report",
"",
f"- schema: `{payload['schema']}`",
f"- service_support_band: `{payload['service']['support_band']}`",
f"- profile_id: `{payload['service']['profile_id']}`",
f"- runtime_mode: `{payload['retrieval_evidence']['runtime_mode']}`",
f"- overall_compatible: `{str(payload['compatibility_gate']['overall_compatible']).lower()}`",
"",
"## Service Surface",
f"- listen_address: `{payload['service']['listen_address']}`",
f"- liveness: `{payload['service']['paths']['liveness']}`",
f"- readiness: `{payload['service']['paths']['readiness']}`",
f"- admin_status: `{payload['service']['paths']['admin_status']}`",
"",
"## Compatibility Checks",
]
for name, passed in payload["compatibility_gate"]["checks"].items():
md_lines.append(f"- `{name}`: `{str(passed).lower()}`")
md_lines.extend(
[
"",
"## Artifact Set",
f"- retrieval_report: `{payload['artifact_set']['retrieval_report']}`",
f"- retrieval_markdown: `{payload['artifact_set']['retrieval_markdown']}`",
f"- retrieval_contract: `{payload['artifact_set']['retrieval_contract']}`",
f"- service_report: `{payload['artifact_set']['service_report']}`",
f"- service_candidate_report: `{payload['artifact_set']['service_candidate_report']}`",
f"- service_candidate_markdown: `{payload['artifact_set']['service_candidate_markdown']}`",
f"- service_lifecycle_report: `{payload['artifact_set']['service_lifecycle_report']}`",
f"- service_lifecycle_markdown: `{payload['artifact_set']['service_lifecycle_markdown']}`",
]
)
path.write_text("\n".join(md_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)
listen = resolve_listen(args.listen)
run_command(
[
sys.executable,
str(ROOT / "scripts" / "service_candidate_report.py"),
"--report-dir",
str(report_dir),
"--listen",
listen,
"--telemetry-endpoint",
args.telemetry_endpoint,
"--tls",
args.tls,
"--admin-token",
args.admin_token,
"--binary",
args.binary,
]
)
run_command(
[
sys.executable,
str(ROOT / "scripts" / "service_lifecycle_report.py"),
"--report-dir",
str(report_dir),
"--listen",
listen,
"--telemetry-endpoint",
args.telemetry_endpoint,
"--tls",
args.tls,
"--admin-token",
args.admin_token,
"--binary",
args.binary,
]
)
run_command(
[
sys.executable,
str(ROOT / "scripts" / "retrieval_quality_gate.py"),
"--report-dir",
str(report_dir),
"--report-prefix",
"retrieval_quality_service",
"--runtime-mode",
"service",
"--work-dir",
args.work_dir,
"--listen",
listen,
"--telemetry-endpoint",
args.telemetry_endpoint,
"--tls",
args.tls,
"--admin-token",
args.admin_token,
"--binary",
args.binary,
]
)
service = run_json([args.binary, "service-report", "--listen", listen])
validation = run_json(
[
args.binary,
"service-validate",
"--listen",
listen,
"--telemetry-endpoint",
args.telemetry_endpoint,
"--tls",
args.tls,
"--admin-token",
args.admin_token,
"--max-requests",
"4",
]
)
retrieval_report = load_json(report_dir / "retrieval_quality_service_report.json")
candidate_report = load_json(report_dir / "service_candidate_report.json")
lifecycle_report = load_json(report_dir / "service_lifecycle_report.json")
compatibility_gate = {
"checks": {
"service_validation_compatible": bool(validation["compatible"]),
"retrieval_quality_pass": bool(retrieval_report["quality_gate"]["overall_pass"]),
"service_mode_evidence": retrieval_report["runtime_mode"] == "service",
"admin_status_route_exposed": any(
route["path"] == "/admin/status" for route in service["admin_routes"]
),
"lifecycle_stop_verified": bool(lifecycle_report["lifecycle"]["stop_requested"]),
}
}
compatibility_gate["overall_compatible"] = all(
compatibility_gate["checks"].values()
)
payload = {
"schema": "iridium.service-compatibility-report.v1",
"service": service,
"validation": validation,
"retrieval_evidence": {
"schema": retrieval_report["schema"],
"runtime_mode": retrieval_report["runtime_mode"],
"overall_pass": retrieval_report["quality_gate"]["overall_pass"],
"macro": retrieval_report["macro"],
},
"compatibility_gate": compatibility_gate,
"artifact_set": {
"retrieval_report": "retrieval_quality_service_report.json",
"retrieval_markdown": "retrieval_quality_service_report.md",
"retrieval_contract": "retrieval_quality_service_alloy_contract.json",
"service_report": "retrieval_quality_service_service_report.json",
"service_candidate_report": "service_candidate_report.json",
"service_candidate_markdown": "service_candidate_report.md",
"service_lifecycle_report": "service_lifecycle_report.json",
"service_lifecycle_markdown": "service_lifecycle_report.md",
},
"operator_inputs": candidate_report["operator_inputs"],
}
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")
write_markdown(md_path, payload)
print(f"wrote: {json_path}")
print(f"wrote: {md_path}")
return 0 if compatibility_gate["overall_compatible"] else 1
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))