from __future__ import annotations
import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence, Tuple
ROOT = Path(__file__).resolve().parents[1]
SCENARIOS: Sequence[Dict[str, str]] = (
{
"id": "restart-recovery",
"description": "Storage WAL recovery succeeds after reopen.",
"command": "cargo test --release --test storage_paths integration_wal_recovery_path -- --exact",
"log": "durability_restart_recovery.log",
},
{
"id": "restart-requery-consistency",
"description": "Embedded Plexus execution returns the same rows before and after restart.",
"command": "cargo test --release --test storage_paths integration_embedded_driver_plexus_restart_round_trip -- --exact",
"log": "durability_restart_requery.log",
},
{
"id": "capability-rejection",
"description": "Unsupported Plexus plan capability is rejected deterministically.",
"command": "cargo test features::runtime::api::tests::plexus_paths::execute_serialized_plan_rejects_unsupported_op_capability -- --exact",
"log": "durability_capability_rejection.log",
},
)
def run_command(command: str, log_path: Path) -> int:
completed = subprocess.run(
command,
shell=True,
cwd=ROOT,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
check=False,
)
log_path.write_text(completed.stdout, encoding="utf-8")
return completed.returncode
def load_contract(report_dir: Path, prefix: str, binary: str) -> Tuple[str, Dict[str, Any]]:
contract_path = report_dir / f"{prefix}_alloy_contract.json"
completed = subprocess.run(
[binary, "contract-report", "acceptance"],
cwd=ROOT,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True,
)
contract_path.write_text(completed.stdout, encoding="utf-8")
return contract_path.name, json.loads(completed.stdout)
def load_typed_evidence(
report_dir: Path,
prefix: str,
binary: str,
) -> Tuple[str, Dict[str, Any]]:
evidence_path = report_dir / f"{prefix}_restart_requery_evidence.json"
completed = subprocess.run(
[binary, "evidence-report", "restart-requery"],
cwd=ROOT,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True,
)
evidence = json.loads(completed.stdout)
evidence_path.write_text(json.dumps(evidence, indent=2), encoding="utf-8")
return evidence_path.name, evidence
def write_scenario_pack(
report_dir: Path,
prefix: str,
report: Dict[str, Any],
typed_evidence: Dict[str, Any],
) -> str:
scenario_pack_path = report_dir / f"{prefix}_scenario_pack.json"
payload = {
"release_gate": "embedded-release-candidate",
"schema": typed_evidence["schema"],
"scenario_family": "embedded",
"scenario_count": len(report["scenarios"]),
"overall_pass": report["overall_pass"],
"workload_identity": typed_evidence["workload_identity"],
"replay_payload": typed_evidence["replay_payload"],
"scenarios": [
{
"id": scenario["id"],
"description": scenario["description"],
"status": scenario["status"],
"exit_code": scenario["exit_code"],
"reproduction": scenario["command"],
"attachment_log": scenario["log"],
}
for scenario in report["scenarios"]
],
}
scenario_pack_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(f"wrote: {scenario_pack_path}")
return scenario_pack_path.name
def write_reproduction_steps(report_dir: Path, prefix: str, report: Dict[str, Any]) -> str:
reproduction_path = report_dir / f"{prefix}_reproduction.md"
lines: List[str] = [
"# Durability Reproduction Steps",
"",
"Run the named scenarios from the repository root in this order:",
"",
]
for index, scenario in enumerate(report["scenarios"], start=1):
lines.extend(
[
f"{index}. `{scenario['command']}`",
f" Expected status: `{scenario['status']}`",
f" Captured log: `{scenario['log']}`",
]
)
reproduction_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
print(f"wrote: {reproduction_path}")
return reproduction_path.name
def write_trace_summary(
report_dir: Path,
prefix: str,
report: Dict[str, Any],
typed_evidence: Dict[str, Any],
) -> str:
trace_summary_path = report_dir / f"{prefix}_trace_summary.json"
payload = {
"operation": typed_evidence["replay_payload"]["trace_operation"],
"trace_id": typed_evidence["replay_payload"]["trace_id"],
"span_id": typed_evidence["replay_payload"]["span_id"],
"parent_span_id": "iridium-root",
"correlation_id": typed_evidence["replay_payload"]["correlation_id"],
"baseline_metric_names": [
"alloy.build.info",
"alloy.operation.duration_ms",
"alloy.cache.requests",
],
"scenario_status": {
scenario["id"]: {
"status": scenario["status"],
"exit_code": scenario["exit_code"],
"log": scenario["log"],
}
for scenario in report["scenarios"]
},
}
trace_summary_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(f"wrote: {trace_summary_path}")
return trace_summary_path.name
def write_reports(report_dir: Path, prefix: str, report: Dict[str, Any]) -> None:
report_dir.mkdir(parents=True, exist_ok=True)
json_path = report_dir / f"{prefix}_report.json"
md_path = report_dir / f"{prefix}_report.md"
json_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
lines: List[str] = [
"# Durability Verification Report",
"",
f"- commit_sha: {report['commit_sha']}",
f"- alloy_contract_path: {report['alloy_contract_path']}",
f"- restart_requery_evidence_path: {report['restart_requery_evidence_path']}",
f"- scenario_pack_path: {report['scenario_pack_path']}",
f"- reproduction_path: {report['reproduction_path']}",
f"- trace_summary_path: {report['trace_summary_path']}",
f"- overall_pass: {str(report['overall_pass']).lower()}",
"",
"## Scenarios",
]
for scenario in report["scenarios"]:
lines.extend(
[
f"### {scenario['id']}",
f"- description: {scenario['description']}",
f"- status: {scenario['status']}",
f"- reproduction: `{scenario['command']}`",
f"- log: {scenario['log']}",
"",
]
)
md_path.write_text("\n".join(lines), encoding="utf-8")
print(f"wrote: {json_path}")
print(f"wrote: {md_path}")
def parse_args(argv: Sequence[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run the embedded durability verification report flow")
parser.add_argument("--report-dir", default="artifacts")
parser.add_argument("--report-prefix", default="durability_verification")
parser.add_argument("--binary", default=str(ROOT / "target" / "debug" / "ir"))
return parser.parse_args(argv)
def main(argv: Sequence[str]) -> int:
args = parse_args(argv)
report_dir = Path(args.report_dir)
report_dir.mkdir(parents=True, exist_ok=True)
scenarios: List[Dict[str, Any]] = []
overall_pass = True
for scenario in SCENARIOS:
log_path = report_dir / scenario["log"]
status = run_command(scenario["command"], log_path)
passed = status == 0
overall_pass = overall_pass and passed
scenarios.append(
{
**scenario,
"status": "pass" if passed else "fail",
"exit_code": status,
}
)
commit_sha = (
subprocess.run(
["git", "rev-parse", "--short", "HEAD"],
cwd=ROOT,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
).stdout.strip()
or "unknown"
)
alloy_contract_path, contract = load_contract(report_dir, args.report_prefix, args.binary)
typed_evidence_path, typed_evidence = load_typed_evidence(
report_dir, args.report_prefix, args.binary
)
assertion_status = {scenario["id"]: scenario for scenario in scenarios}
for assertion in typed_evidence["assertions"]:
scenario = assertion_status[assertion["assertion_id"]]
assertion["outcome"] = scenario["status"]
assertion["artifact_name"] = scenario["log"]
for artifact in typed_evidence["attached_artifacts"]:
if artifact["role"] == "durability-report":
artifact["artifact_name"] = f"{args.report_prefix}_report.json"
elif artifact["role"] == "durability-summary":
artifact["artifact_name"] = f"{args.report_prefix}_report.md"
typed_evidence["permits_release"] = overall_pass
typed_evidence["replay_payload"]["reproduction_commands"] = [
scenario["command"] for scenario in scenarios
]
report = {
"commit_sha": commit_sha,
"alloy_contract_path": alloy_contract_path,
"overall_pass": overall_pass,
"scenarios": scenarios,
}
typed_evidence["attached_artifacts"].extend(
[
{
"artifact_name": alloy_contract_path,
"media_type": "application/json",
"role": "alloy-contract",
},
]
)
(report_dir / typed_evidence_path).write_text(
json.dumps(typed_evidence, indent=2), encoding="utf-8"
)
print(f"wrote: {report_dir / typed_evidence_path}")
report["restart_requery_evidence_path"] = typed_evidence_path
report["scenario_pack_path"] = write_scenario_pack(
report_dir, args.report_prefix, report, typed_evidence
)
report["reproduction_path"] = write_reproduction_steps(
report_dir, args.report_prefix, report
)
report["trace_summary_path"] = write_trace_summary(
report_dir, args.report_prefix, report, typed_evidence
)
write_reports(report_dir, args.report_prefix, report)
return 0 if overall_pass else 1
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))