scripts/iso-certify.py

#!/usr/bin/env python3
"""iso-certify — AC-4 dashboard turn-isolation certify harness.

Certifies Mechanism-B process isolation
(``docs/desktop/2026-07-04-dashboard-process-isolation-PRD.md``, AC-4): under
6 concurrent heavy agent turns, the dashboard's HTTP/ws SERVING plane must stay
responsive (p99 < 1s) with zero event-loop stalls.

What it does
------------
1. Spawns a SCRATCH dashboard (``hermes dashboard``) bound to loopback on a
   free port, with an ISOLATED ``HERMES_HOME`` (temp dir, minimal seeded state).
   It NEVER touches the live :9119 dashboard / ai.hermes.dashboard / live
   state.db. Loopback bind ⇒ no auth gate (web_server.should_require_auth).
2. Arms the synthetic GIL-heavy turn seam (``HERMES_ISO_CERTIFY_SYNTH_TURN=1``,
   see ``tui_gateway/synthetic_turn.py``) so 6 concurrent turns reproduce the
   ``take_gil`` interpreter-contention regime WITHOUT real model calls. A
   network/sleep stub would release the GIL and NOT reproduce the incident, so
   it would be a fake green — the synthetic turn is pure-Python CPU on purpose.
3. Drives 6 concurrent heavy turns over ws (session.create → prompt.submit),
   and CONCURRENTLY probes the serving path — a ws ``session.list`` round-trip
   AND a REST ``GET /api/status`` — every 500ms, timing each.
4. Reports p50/p95/p99 latency for both probes + the count of probes over the
   1s threshold ("serving stalls") + the count of ``event loop stalled`` /
   ``ws write slow`` lines the dashboard logged during the run.

Verdict
-------
The AC-4 verdict comes ONLY from the heavy run: PASS iff serving p99 < 1s AND
zero serving stalls AND zero ``event loop stalled`` log lines over the sustained
window. ``--dry-run`` runs ONE short light turn as a plumbing smoke test and is
explicitly NOT a verdict (a dry-run green is a fake green — the spec says so).

Turn isolation is controlled by the ``dashboard.turn_isolation`` config knob in
the scratch HERMES_HOME; ``--isolation on|off`` sets it. Run BOTH:
    iso-certify --isolation off   # baseline: expect stalls
    iso-certify --isolation on    # the measurement that decides AC-4
"""

from __future__ import annotations

import argparse
import json
import os
import re
import socket
import subprocess
import sys
import tempfile
import threading
import time
import shutil
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any

try:
    from websockets.sync.client import connect as ws_connect
except Exception as exc:  # pragma: no cover - dependency guard
    print(f"iso-certify requires the 'websockets' package: {exc}", file=sys.stderr)
    raise

REPO_ROOT = Path(__file__).resolve().parents[1]
_READY_RE = re.compile(r"HERMES_(?:DASHBOARD|BACKEND)_READY port=(\d+)")
_STALL_LOG_RE = re.compile(r"event loop stalled|ws write slow \(loop stalled")


# ── stats ──────────────────────────────────────────────────────────────
def percentile(values: list[float], pct: float) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    if len(ordered) == 1:
        return ordered[0]
    rank = (len(ordered) - 1) * (pct / 100.0)
    lo = int(rank)
    hi = min(lo + 1, len(ordered) - 1)
    frac = rank - lo
    return ordered[lo] * (1.0 - frac) + ordered[hi] * frac


def summarize(values: list[float]) -> dict[str, float]:
    return {
        "count": len(values),
        "p50_ms": percentile(values, 50),
        "p95_ms": percentile(values, 95),
        "p99_ms": percentile(values, 99),
        "max_ms": max(values) if values else 0.0,
    }


def free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return int(s.getsockname()[1])


# ── scratch HERMES_HOME ─────────────────────────────────────────────────
def seed_scratch_home(home: Path, *, isolation: str, heartbeat_secs: int, respawn_max: int) -> None:
    """Write a minimal config.yaml with the isolation knob set."""
    home.mkdir(parents=True, exist_ok=True)
    (home / "state").mkdir(parents=True, exist_ok=True)
    (home / "logs").mkdir(parents=True, exist_ok=True)
    cfg = {
        # Pin the model/provider to what SyntheticHeavyAgent reports so the
        # per-turn _sync_agent_model_with_config sees a match and no-ops — a
        # mismatch would try (and fail) a real model switch, erroring the turn
        # before its heavy loop runs (which would be a FALSE green: serving
        # stays responsive because no heavy compute happened). The synthetic
        # seam means no real API call is ever made regardless.
        "provider": "synthetic",
        "model": "synthetic-heavy",
        "dashboard": {
            "turn_isolation": (isolation == "on"),
            "compute_host_heartbeat_secs": heartbeat_secs,
            "compute_host_respawn_max": respawn_max,
        },
        # Keep memory/mem0/skills side-machinery from reaching out.
        "memory": {"enabled": False},
    }
    # config.yaml is the canonical config; write it directly.
    import yaml  # provided by the runtime venv

    (home / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=True), encoding="utf-8")
    # A stub .env so credential resolution doesn't spelunk the real home.
    (home / ".env").write_text("OPENAI_API_KEY=sk-synthetic-not-used\n", encoding="utf-8")


# ── dashboard process ───────────────────────────────────────────────────
class ScratchDashboard:
    def __init__(
        self,
        *,
        home: Path,
        port: int,
        isolation: str,
        env_extra: dict[str, str],
    ) -> None:
        self.home = home
        self.port = port
        self.isolation = isolation
        self.env_extra = env_extra
        self.proc: subprocess.Popen[str] | None = None
        self.actual_port = port
        self.log_lines: list[str] = []
        self._log_lock = threading.Lock()
        self._ready = threading.Event()

    @property
    def stall_log_count(self) -> int:
        with self._log_lock:
            return sum(1 for ln in self.log_lines if _STALL_LOG_RE.search(ln))

    def _drain(self, stream: Any) -> None:
        for raw in stream:
            line = raw.rstrip("\n")
            with self._log_lock:
                self.log_lines.append(line)
            m = _READY_RE.search(line)
            if m:
                self.actual_port = int(m.group(1))
                self._ready.set()

    def __enter__(self) -> "ScratchDashboard":
        venv_py = REPO_ROOT / "venv" / "bin" / "python"
        python = str(venv_py) if venv_py.exists() else sys.executable
        env = dict(os.environ)
        env.update(self.env_extra)
        env["HERMES_HOME"] = str(self.home)
        env["HOME"] = str(self.home.parent) if str(self.home.parent) else env.get("HOME", "")
        env["HERMES_HOME"] = str(self.home)
        env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
        env["HERMES_ISO_CERTIFY_SYNTH_TURN"] = "1"
        cmd = [
            python, "-m", "hermes_cli.main", "dashboard",
            "--no-open", "--host", "127.0.0.1", "--port", str(self.port),
        ]
        self.proc = subprocess.Popen(
            cmd, cwd=str(REPO_ROOT), env=env,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1,
        )
        threading.Thread(target=self._drain, args=(self.proc.stdout,), name="dash-log", daemon=True).start()
        if not self._ready.wait(timeout=90.0):
            self._dump_tail()
            raise RuntimeError("scratch dashboard did not become ready within 90s")
        # Give uvicorn a beat to actually bind the ws route.
        time.sleep(1.0)
        return self

    def __exit__(self, *_exc: object) -> None:
        if self.proc is None:
            return
        try:
            self.proc.terminate()
            self.proc.wait(timeout=10)
        except Exception:
            try:
                self.proc.kill()
            except Exception:
                pass

    def _dump_tail(self, n: int = 40) -> None:
        with self._log_lock:
            tail = self.log_lines[-n:]
        sys.stderr.write("---- scratch dashboard log tail ----\n")
        for ln in tail:
            sys.stderr.write(ln + "\n")
        sys.stderr.write("------------------------------------\n")


# ── ws client (one connection = one lane) ───────────────────────────────
class WSClient:
    def __init__(self, port: int, token: str) -> None:
        self.url = f"ws://127.0.0.1:{port}/api/ws?token={token}"
        self.ws = ws_connect(
            self.url,
            open_timeout=15,
            max_size=None,
            additional_headers={"Origin": f"http://127.0.0.1:{port}"},
        )
        self._id = 0
        self._lock = threading.Lock()
        # Drain the gateway.ready event.
        self._recv_until(lambda o: o.get("method") == "event", timeout=10)

    def _next_id(self) -> str:
        with self._lock:
            self._id += 1
            return f"r{self._id}"

    def _recv_until(self, pred, timeout: float) -> dict:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                raw = self.ws.recv(timeout=max(0.05, deadline - time.monotonic()))
            except TimeoutError:
                continue
            try:
                obj = json.loads(raw)
            except Exception:
                continue
            if pred(obj):
                return obj
        raise TimeoutError("ws recv predicate timed out")

    def rpc(self, method: str, params: dict, timeout: float = 30.0) -> dict:
        rid = self._next_id()
        self.ws.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}))
        return self._recv_until(lambda o: o.get("id") == rid, timeout=timeout)

    def send_only(self, method: str, params: dict) -> str:
        rid = self._next_id()
        self.ws.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}))
        return rid

    def close(self) -> None:
        try:
            self.ws.close()
        except Exception:
            pass


# ── heavy-turn lane ─────────────────────────────────────────────────────
def drive_heavy_turn(port: int, token: str, turn_spec: dict, stop_at: float, results: list[dict]) -> None:
    """One lane: create a session, submit heavy turns until the deadline."""
    lane: dict[str, Any] = {"turns": 0, "errors": [], "turn_durations_s": [], "min_deltas": None}
    try:
        cli = WSClient(port, token)
    except Exception as exc:
        lane["errors"].append(f"connect: {exc}")
        results.append(lane)
        return
    try:
        resp = cli.rpc("session.create", {"cols": 80, "source": "iso-certify"}, timeout=30)
        sid = ((resp.get("result") or {}).get("session_id")) or ((resp.get("result") or {}).get("id"))
        if not sid:
            lane["errors"].append(f"session.create bad resp: {resp}")
            results.append(lane)
            return
        lane["sid"] = sid
        spec_text = json.dumps(turn_spec)
        while time.monotonic() < stop_at:
            # prompt.submit returns immediately ("streaming"); the turn runs async.
            r = cli.rpc("prompt.submit", {"session_id": sid, "text": spec_text}, timeout=30)
            if r.get("error"):
                lane["errors"].append(f"prompt.submit: {r['error']}")
                break
            # Wait for the REAL turn boundary. The isolated path emits
            # session.info MID-turn (metadata mirror), so waiting on it would
            # false-complete a turn in <1s and inflate the count while NO heavy
            # compute ran — the acceptance-gate "proxy not effect" trap. The
            # turn is done only on message.complete. Count deltas + duration so
            # a fast-erroring turn (e.g. a failed model switch) is caught, not
            # masked as sustained load.
            per_turn_budget = turn_spec.get("duration_s", 8.0) + 30.0
            deadline = time.monotonic() + per_turn_budget
            turn_start = time.monotonic()
            deltas = 0
            started = False
            done = False
            errored = False
            while time.monotonic() < deadline:
                try:
                    o = cli._recv_until(
                        lambda o: o.get("method") == "event"
                        and (o.get("params") or {}).get("type")
                        in {"message.start", "message.delta", "message.complete", "error"},
                        timeout=max(0.5, deadline - time.monotonic()),
                    )
                except TimeoutError:
                    break
                ptype = (o.get("params") or {}).get("type")
                if ptype == "message.start":
                    started = True
                    turn_start = time.monotonic()
                    deltas = 0
                    continue
                # prompt.submit's RPC response may race with a duplicate terminal
                # event from the previous turn. Only events after this turn's
                # message.start can satisfy or score its load boundary.
                if not started:
                    continue
                if ptype == "message.delta":
                    deltas += 1
                    continue
                if ptype == "error":
                    msg = str(((o.get("params") or {}).get("payload") or {}).get("message") or "")
                    lane["errors"].append(f"turn error: {msg[:160]}")
                    errored = True
                    break
                if ptype == "message.complete":
                    done = True
                    break
            turn_dur = time.monotonic() - turn_start
            if done:
                lane["turns"] += 1
                lane["turn_durations_s"].append(round(turn_dur, 2))
                lane["min_deltas"] = deltas if lane["min_deltas"] is None else min(lane["min_deltas"], deltas)
            if errored:
                break
    except Exception as exc:
        lane["errors"].append(f"lane: {exc}")
    finally:
        cli.close()
        results.append(lane)


# ── serving-path probes ─────────────────────────────────────────────────
def warmup_serving(port: int, token: str, rounds: int = 6) -> None:
    """Prime serving-path caches before the measured window.

    The first ``/api/status`` on a freshly-booted process pays one-time
    cold-start cost (config-version check, gateway-health probe, DB connect) that
    a real dashboard — warm for hours before the incident regime — never pays
    during a stall. AC-4 measures sustained-load responsiveness, not cold boot,
    so we hit both serving endpoints a few times UNMEASURED first. This is not a
    green-washing shortcut: the measured window still runs the full 6-lane heavy
    load; warmup only removes a one-time boot artifact from the p99.
    """
    rest_url = f"http://127.0.0.1:{port}/api/status"
    warm_ws: WSClient | None = None
    try:
        warm_ws = WSClient(port, token)
    except Exception:
        warm_ws = None
    for _ in range(rounds):
        try:
            with urllib.request.urlopen(rest_url, timeout=30) as fh:
                fh.read()
        except Exception:
            pass
        if warm_ws is not None:
            try:
                warm_ws.rpc("session.list", {"limit": 20}, timeout=30)
            except Exception:
                pass
        time.sleep(0.2)
    if warm_ws is not None:
        warm_ws.close()


def probe_loop(port: int, token: str, stop_at: float, cadence_s: float, ws_samples: list[float], rest_samples: list[float]) -> None:
    """Probe ws session.list + REST /api/status every ``cadence_s`` seconds."""
    probe_ws: WSClient | None = None
    try:
        probe_ws = WSClient(port, token)
    except Exception:
        probe_ws = None
    rest_url = f"http://127.0.0.1:{port}/api/status"
    next_tick = time.monotonic()
    while time.monotonic() < stop_at:
        # ws round-trip (session.list — the serving-plane DB read AC-4 protects).
        if probe_ws is not None:
            t0 = time.perf_counter()
            try:
                probe_ws.rpc("session.list", {"limit": 20}, timeout=30)
                ws_samples.append((time.perf_counter() - t0) * 1000.0)
            except Exception:
                # A failed/timed-out probe is itself a serving stall; record the
                # elapsed as the sample so it counts against p99.
                ws_samples.append((time.perf_counter() - t0) * 1000.0)
                try:
                    probe_ws.close()
                    probe_ws = WSClient(port, token)
                except Exception:
                    probe_ws = None
        # REST round-trip.
        t1 = time.perf_counter()
        try:
            with urllib.request.urlopen(rest_url, timeout=30) as fh:
                fh.read()
            rest_samples.append((time.perf_counter() - t1) * 1000.0)
        except Exception:
            rest_samples.append((time.perf_counter() - t1) * 1000.0)
        next_tick += cadence_s
        sleep_for = next_tick - time.monotonic()
        if sleep_for > 0:
            time.sleep(sleep_for)
        else:
            next_tick = time.monotonic()
    if probe_ws is not None:
        probe_ws.close()


# ── run ─────────────────────────────────────────────────────────────────
def run_certify(args: argparse.Namespace) -> dict[str, Any]:
    port = free_port()
    import secrets
    token = secrets.token_urlsafe(24)
    parent_tmp = Path(tempfile.mkdtemp(prefix="iso-certify-"))
    home = parent_tmp / "hermes-home"
    seed_scratch_home(
        home,
        isolation=args.isolation,
        heartbeat_secs=args.heartbeat_secs,
        respawn_max=args.respawn_max,
    )

    concurrency = 1 if args.dry_run else args.concurrency
    duration_s = 3.0 if args.dry_run else args.duration_s
    turn_duration = 0.5 if args.dry_run else args.turn_duration_s
    threshold_ms = args.threshold_ms

    turn_spec = {
        "duration_s": turn_duration,
        "delta_interval_s": args.delta_interval_s,
        "tokens_per_delta": args.tokens_per_delta,
        "chunk": args.chunk,
    }

    result: dict[str, Any] = {
        "mode": "dry-run" if args.dry_run else "heavy",
        "isolation": args.isolation,
        "concurrency": concurrency,
        "run_duration_s": duration_s,
        "turn_spec": turn_spec,
        "threshold_ms": threshold_ms,
        "scratch_home": str(home),
        "port": port,
    }

    try:
        with ScratchDashboard(
            home=home, port=port, isolation=args.isolation,
            env_extra={"HERMES_DASHBOARD_SESSION_TOKEN": token},
        ) as dash:
            actual_port = dash.actual_port
            result["port"] = actual_port
            ws_samples: list[float] = []
            rest_samples: list[float] = []
            lane_results: list[dict] = []
            stall_before = dash.stall_log_count

            # Prime serving-path caches so a one-time cold-start artifact does
            # not count as a serving stall (only for the real heavy run; a
            # dry-run stays a raw plumbing smoke).
            if not args.dry_run:
                warmup_serving(actual_port, token)
                stall_before = dash.stall_log_count

            stop_at = time.monotonic() + duration_s
            probe_thread = threading.Thread(
                target=probe_loop,
                args=(actual_port, token, stop_at, args.cadence_s, ws_samples, rest_samples),
                name="probe", daemon=True,
            )
            probe_thread.start()

            with ThreadPoolExecutor(max_workers=concurrency) as pool:
                for _ in range(concurrency):
                    pool.submit(drive_heavy_turn, actual_port, token, turn_spec, stop_at, lane_results)
                # Pool context waits for all lanes.
            probe_thread.join(timeout=30)

            stall_after = dash.stall_log_count
            total_turns = sum(l.get("turns", 0) for l in lane_results)
            lane_errors = [e for l in lane_results for e in l.get("errors", [])]
            all_durations = [d for l in lane_results for d in l.get("turn_durations_s", [])]
            min_deltas = [l["min_deltas"] for l in lane_results if l.get("min_deltas") is not None]
            lanes_with_turn = sum(1 for l in lane_results if l.get("turns", 0) > 0)

            ws_stat = summarize(ws_samples)
            rest_stat = summarize(rest_samples)
            ws_over = sum(1 for v in ws_samples if v > threshold_ms)
            rest_over = sum(1 for v in rest_samples if v > threshold_ms)
            serving_stalls = ws_over + rest_over
            log_stalls = stall_after - stall_before

            # Load validity — the run only certifies anything if the offered load
            # was REAL: every lane completed ≥1 turn, and the completed turns
            # actually held ~the requested heavy duration and streamed deltas.
            # A fast-erroring/short turn is NOT sustained GIL load, so a green off
            # it would be a proxy (serving stays fast because nothing burned).
            median_turn_dur = percentile(all_durations, 50) if all_durations else 0.0
            min_turn_dur = min(all_durations) if all_durations else 0.0
            worst_min_deltas = min(min_deltas) if min_deltas else 0
            expected_turn_s = float(turn_spec.get("duration_s", 8.0))
            load_valid = (
                lanes_with_turn >= concurrency
                and total_turns >= concurrency
                and min_turn_dur >= 0.7 * expected_turn_s
                and worst_min_deltas >= 5
            )

            result.update({
                "ws_probe": ws_stat,
                "rest_probe": rest_stat,
                "ws_probes_over_threshold": ws_over,
                "rest_probes_over_threshold": rest_over,
                "serving_stalls": serving_stalls,
                "event_loop_stall_log_lines": log_stalls,
                "heavy_turns_completed": total_turns,
                "lanes_with_turn": lanes_with_turn,
                "median_turn_duration_s": round(median_turn_dur, 2),
                "min_turn_duration_s": round(min_turn_dur, 2),
                "worst_lane_min_deltas": worst_min_deltas,
                "load_valid": load_valid,
                "lane_errors": lane_errors[:20],
            })

            # AC-4 verdict — heavy run only. A dry-run reports but never PASSes.
            serving_p99 = max(ws_stat["p99_ms"], rest_stat["p99_ms"])
            result["serving_p99_ms"] = serving_p99
            if args.dry_run:
                result["verdict"] = "SMOKE-OK" if (total_turns > 0 and not lane_errors) else "SMOKE-FAIL"
                result["is_verdict"] = False
            else:
                serving_ok = (
                    serving_p99 < threshold_ms
                    and serving_stalls == 0
                    and log_stalls == 0
                    and probe_thread_samples_ok(ws_samples, rest_samples)
                )
                if not load_valid:
                    # Cannot certify: the offered load was not the incident
                    # regime. Never a PASS; report INCONCLUSIVE, not FAIL, so it
                    # is not read as "isolation broke serving".
                    result["verdict"] = "INCONCLUSIVE"
                    result.setdefault("notes", []).append(
                        "load invalid: lanes/turn-duration/deltas below the sustained-heavy-load floor — "
                        "not the AC-4 incident regime, verdict cannot certify"
                    )
                else:
                    result["verdict"] = "PASS" if serving_ok else "FAIL"
                result["is_verdict"] = True
    finally:
        if not args.keep_home:
            shutil.rmtree(parent_tmp, ignore_errors=True)
        else:
            result["scratch_home_kept"] = str(parent_tmp)

    return result


def probe_thread_samples_ok(ws_samples: list[float], rest_samples: list[float]) -> bool:
    """Guard against a blind gate: require the probes actually ran.

    A run that produced no probe samples saw NOTHING — it must not PASS. This is
    the tri-state INCONCLUSIVE floor (an empty timeline is never a green).
    """
    return len(ws_samples) >= 3 and len(rest_samples) >= 3


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="AC-4 dashboard turn-isolation certify harness")
    p.add_argument("--isolation", choices=["on", "off"], default="on",
                   help="set dashboard.turn_isolation in the scratch HERMES_HOME")
    p.add_argument("--dry-run", action="store_true",
                   help="1 short light turn plumbing smoke — NOT an AC-4 verdict")
    p.add_argument("--concurrency", type=int, default=6, help="concurrent heavy-turn lanes (AC-4: 6)")
    p.add_argument("--duration-s", type=float, default=600.0, dest="duration_s",
                   help="sustained run window seconds (AC-4: ~600 = 10 min)")
    p.add_argument("--turn-duration-s", type=float, default=12.0, dest="turn_duration_s",
                   help="wall seconds of GIL-holding compute per heavy turn")
    p.add_argument("--delta-interval-s", type=float, default=0.05, dest="delta_interval_s",
                   help="streamed-delta cadence per heavy turn")
    p.add_argument("--tokens-per-delta", type=int, default=512, dest="tokens_per_delta")
    p.add_argument("--chunk", type=int, default=20000, help="pure-Python ops per interrupt-check chunk")
    p.add_argument("--cadence-s", type=float, default=0.5, dest="cadence_s",
                   help="serving-path probe cadence (AC-4: 500ms)")
    p.add_argument("--threshold-ms", type=float, default=1000.0, dest="threshold_ms",
                   help="serving p99 threshold (AC-4: <1s)")
    p.add_argument("--heartbeat-secs", type=int, default=15, dest="heartbeat_secs")
    p.add_argument("--respawn-max", type=int, default=3, dest="respawn_max")
    p.add_argument("--keep-home", action="store_true", help="do not delete the scratch HERMES_HOME on exit")
    p.add_argument("--json-out", type=Path, help="write JSON metrics to this path")
    args = p.parse_args(argv)

    result = run_certify(args)
    text = json.dumps(result, indent=2, sort_keys=True)
    print(text)
    if args.json_out:
        args.json_out.parent.mkdir(parents=True, exist_ok=True)
        args.json_out.write_text(text + "\n", encoding="utf-8")

    # Exit code: 0 only on a real PASS (or a clean dry-run smoke); non-zero
    # otherwise. A dry-run is never treated as a verdict for automation.
    if result.get("is_verdict"):
        return 0 if result.get("verdict") == "PASS" else 1
    return 0 if result.get("verdict") == "SMOKE-OK" else 1


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