verify_release.py

"""Hermes Console release verification runner.

Usage:
    python verify_release.py          # all implemented release gates
    python verify_release.py --full   # same gates plus pending/experimental contracts

The gateway smoke/tests do not mutate Hermes configuration or select a model.
The runner does stop a running HermesConsole.exe and rebuild that executable.
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent


def run(name: str, command: list[str], timeout: int = 180) -> dict:
    started = time.monotonic()
    try:
        proc = subprocess.run(
            command,
            cwd=ROOT,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            timeout=timeout,
            check=False,
        )
        return {
            "name": name,
            "command": command,
            "exit": proc.returncode,
            "seconds": round(time.monotonic() - started, 3),
            "output": proc.stdout,
        }
    except subprocess.TimeoutExpired as exc:
        output = exc.stdout or ""
        if isinstance(output, bytes):
            output = output.decode(errors="replace")
        return {
            "name": name,
            "command": command,
            "exit": 124,
            "seconds": round(time.monotonic() - started, 3),
            "output": output + f"\nTIMEOUT after {timeout}s\n",
        }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--full", action="store_true", help="include pending parity contracts")
    parser.add_argument("--json", action="store_true", help="emit machine-readable summary")
    args = parser.parse_args()

    # The canonical linker cannot replace a running Windows executable. Stop
    # only Hermes Console itself — never broad-kill Python/Hermes processes.
    subprocess.run(
        ["taskkill", "/F", "/IM", "HermesConsole.exe"],
        cwd=ROOT,
        text=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )

    py = sys.executable
    approval_exe = ROOT / "approval_contract_test.exe"
    interactive_exe = ROOT / "interactive_prompt_contract_test.exe"
    modern_exe = ROOT / "modern_event_behavior_test.exe"
    policy_exe = ROOT / "policy_behavior_test.exe"
    module_exe = ROOT / "module_mutation_behavior_test.exe"
    stream_exe = ROOT / "stream_session_behavior_test.exe"
    gates: list[tuple[str, list[str], int]] = [
        ("interface contracts", [py, "interface_contract_test.py"], 180),
        ("gateway smoke", [py, "gateway_probe.py"], 180),
        (
            "approval contract compile",
            ["gcc", "-std=c99", "-I", "src", "approval_contract_test.c", "-o", str(approval_exe)],
            120,
        ),
        ("approval contracts", [str(approval_exe)], 60),
    ]
    if ROOT.joinpath("interactive_prompt_contract_test.c").exists():
        gates.extend([
            (
                "interactive prompt contract compile",
                ["gcc", "-std=c99", "-I", "src", "interactive_prompt_contract_test.c", "-o", str(interactive_exe)],
                120,
            ),
            ("interactive prompt contracts", [str(interactive_exe)], 60),
        ])
    gates.append(("modern gateway event contracts", [py, "modern_event_contract_test.py"], 120))
    gates.extend([
            ("modern gateway event behavior compile",
             ["gcc", "-std=c99", "-I", "src", "modern_event_behavior_test.c", "-o", str(modern_exe)], 120),
            ("modern gateway event behavior", [str(modern_exe)], 60),
    ])
    gates.append(("gateway policy contracts", [py, "policy_contract_test.py"], 120))
    gates.append(("gateway policy race contracts", [py, "policy_race_contract_test.py"], 120))
    gates.extend([
            ("gateway policy behavior compile",
             ["gcc", "-std=c99", "-I", "src", "policy_behavior_test.c", "-o", str(policy_exe)], 120),
            ("gateway policy behavior", [str(policy_exe)], 60),
    ])
    gates.append(("module mutation contracts", [py, "module_mutation_contract_test.py"], 120))
    gates.extend([
            ("module mutation behavior compile",
             ["gcc", "-std=c99", "-I", "src", "module_mutation_behavior_test.c", "-o", str(module_exe)], 120),
            ("module mutation behavior", [str(module_exe)], 60),
    ])
    gates.append(("stream session contracts", [py, "stream_session_contract_test.py"], 120))
    gates.extend([
        ("stream session behavior compile",
         ["gcc", "-std=c99", "-I", "src", "stream_session_behavior_test.c", "-o", str(stream_exe)], 120),
        ("stream session behavior", [str(stream_exe)], 60),
    ])
    gates.append(("canonical build", ["cmd", "/c", "build.bat"], 180))

    results = []
    try:
        compile_ok = True
        for name, command, timeout in gates:
            if name == "interactive prompt contracts" and not compile_ok:
                results.append({
                    "name": name, "command": command, "exit": 125,
                    "seconds": 0.0, "output": "SKIPPED: compile gate failed\n",
                })
                if not args.json:
                    print(f"[FAIL] {name} (skipped: compile gate failed)")
                continue
            result = run(name, command, timeout)
            results.append(result)
            if name == "interactive prompt contract compile":
                compile_ok = result["exit"] == 0
            if not args.json:
                state = "PASS" if result["exit"] == 0 else "FAIL"
                print(f"[{state}] {name} ({result['seconds']:.3f}s)")
                if result["exit"] != 0:
                    print(result["output"][-6000:])
    finally:
        for artifact in (approval_exe, interactive_exe, modern_exe, policy_exe, module_exe, stream_exe):
            try:
                artifact.unlink()
            except FileNotFoundError:
                pass

    passed = sum(result["exit"] == 0 for result in results)
    summary = {
        "mode": "full" if args.full else "core",
        "passed": passed,
        "total": len(results),
        "ok": passed == len(results),
        "results": [
            {key: value for key, value in result.items() if key != "output"}
            for result in results
        ],
    }
    if args.json:
        print(json.dumps(summary, indent=2))
    else:
        print(f"\nRELEASE GATES: {passed}/{len(results)} passed")
        if args.full and not summary["ok"]:
            print("Full mode exercises every implemented parity contract and behavior test.")
    return 0 if summary["ok"] else 1


if __name__ == "__main__":
    raise SystemExit(main())