scripts.start_test_services

Reference › scripts › start_test_services

scripts/start_test_services.py is the parallel infrastructure readiness check for orchestrata-ultimate: it checks (and unless told not to, starts) the services the test suite needs, in parallel, and reports each one. run_tests calls it before a run, project_status reads its --status --json output, and a human can run it directly; three test modules (test_start_test_services, test_start_test_services_coverage, test_start_test_services_sweep_once) cover it. It uses orchestrata_ultimate.core.env_bootstrap, orchestrata_ultimate.core.graph_factory and the CXP codec and transport in orchestrata_ultimate.cxp.

Each service has a check function (for example check_redis), built from helpers such as _port_open, _wait_for_port and _run_bg; the gateway, kernel, graph and the gateway-to-kernel link are among the services. _load_dotenv loads .env and then overlays the per-environment endpoint profile. _is_local_host decides whether a service is host-local and safe to spawn here, and processes the script started are killed on exit only when CLEANUP_KERNEL=1.

Options: --status (check only, start nothing), --json, --skip (a comma-separated service list such as kernel,gateway,graph,link; neo4j is a deprecated alias for graph; skipping the gateway or kernel also skips the link check) and --ensure-worker-dbs N, which creates the per-worker databases orchkg_w0 to orchkg_w<N-1> after the graph check when the engine is Memgraph. When it starts services and the graph is ready, it also runs a test-pollution sweep through sweep_test_pollution, skipped when the outer runner has already swept (ADR-0087). --status always exits 0 and carries readiness in its output; the start path exits 1 when a required service cannot be brought up, so the runner aborts.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ServiceResult (dataclass)

Decorators: dataclass · Line 171

InfraReport (dataclass)

Decorators: dataclass · Line 191

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/start_test_services.py

#!/usr/bin/env python3
"""Parallel infrastructure readiness check for orchestrata-ultimate.

Checks all four services (Redis, Graph, Gateway, Kernel) in parallel.  Starts
any that are down, verifies each with a functional test, and prints a
colour-coded report.  Idempotent — safe to run any number of times.

A fifth, non-parallel check (Link) runs after the four above have settled: it
asks the gateway, over its real WebSocket transport, whether it is actually
wired to the kernel over gRPC (``gateway.info``'s "kernel" block) — because
"the gateway answers gateway.ping" and "the kernel accepts a connection" can
both be true while the gateway is silently running against an in-process
fallback with zero real traffic reaching the kernel. See ``check_kernel_link``.

Usage:
    python scripts/start_test_services.py            # Check + start + verify
    python scripts/start_test_services.py --status   # Check only (no start)
    python scripts/start_test_services.py --json     # JSON report
    python scripts/start_test_services.py --skip KERNEL,GATEWAY  # Skip services
    python scripts/start_test_services.py --skip neo4j  # deprecated alias for
                                                          # graph (stderr note)
"""

from __future__ import annotations

import argparse
import atexit
import importlib.util
import json
import os
import signal
import socket
import subprocess
import sys
import time
from collections.abc import Callable, Mapping, MutableMapping
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, LiteralString, cast

if TYPE_CHECKING:
    from neo4j import Driver

# ── .env auto-load ──────────────────────────────────────────────────────────


def _parse_dotenv(path: Path) -> dict[str, str]:
    out: dict[str, str] = {}
    if not path.exists():
        return out
    with open(path) as fh:
        for raw in fh:
            line = raw.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, _, value = line.partition("=")
            key = key.strip()
            if key:
                out[key] = value.strip().strip('"').strip("'")
    return out


# This module and `core.env_bootstrap.load_env_contract` are the only two
# loaders that overlay the env profile over a running process. Both consult the
# SAME ownership marker (`ORCHESTRATA_ENV_OWNED_KEYS`, claimed via
# `core.env_bootstrap.declare_owned`), so a key a process computed for itself
# survives the overlay while every other profile key stays authoritative.
#
# Until 2026-08-07 they disagreed: this module protected three keys by a
# hardcoded name list, that one protected nothing. Measured cost of the gap:
#
#   * ~169 interactive-only KG failures (2026-08-04). `tests/conftest.py`
#     computes `KG_DATABASE=orchkg_w<N>` from `ORCHESTRATA_WORKER_SLOT` at
#     import; `tests/plugins/services.py` imports this module mid-session and
#     re-ran the overlay, resetting it to the profile's bare `orchkg`. Read-back
#     queries keyed off the frozen `tests.infrastructure.KG_DATABASE` looked in
#     `orchkg_w<N>` and saw nothing — failures shaped as "0 rows" naming nothing.
#   * `MEMGRAPH_URIS` / `MEMGRAPH_HA_CONTROLLER_URL` (2026-08-05). conftest
#     BLANKS both so the standalone test node cannot HA-discover its way to the
#     prod coordinators; the overlay re-planted the profile value over the blank.
#   * Four authoritative gate runs misdiagnosed as Memgraph contention
#     (2026-08-07): `KG_DATABASE` orchkg_w9 -> orchkg, `KG_GRPC_PORT` 50161 ->
#     50052, triggered by any of 31 modules that reach an entrypoint.
#
# A marker rather than a name list, because a list in one loader cannot protect
# the other, and because the same `KG_DATABASE` may arrive from conftest
# (authoritative) or from a stale shell export (must lose) — only the process
# that computed the value can tell those apart. `scripts/validate_ci.py` gate 13
# holds both loaders to it.
try:
    from orchestrata_ultimate.core.env_bootstrap import owned_keys as _owned_keys
except ImportError:  # pragma: no cover - standalone harness without the package
    # Keep the VAR NAME in one place even on this path; `scripts/validate_ci.py`
    # asserts the literal here matches env_bootstrap.ENV_OWNED_KEYS_VAR.
    _ENV_OWNED_KEYS_VAR = "ORCHESTRATA_ENV_OWNED_KEYS"

    def _owned_keys(
        environ: MutableMapping[str, str] | None = None,
    ) -> frozenset[str]:
        target = os.environ if environ is None else environ
        raw = target.get(_ENV_OWNED_KEYS_VAR, "")
        return frozenset(p.strip() for p in raw.split(",") if p.strip())


def _load_dotenv() -> int:
    """Load ``.env`` (secrets/base), then OVERLAY the per-environment endpoint
    profile ``env/<ORCHESTRATA_ENV_PROFILE>.env`` with OVERRIDE.

    This makes the WHOLE harness (kg-sweep, infra preflight, kernel/gateway
    probe) target the selected fleet -- not just pytest via conftest. Without
    the overlay the harness read ``.env`` (prod) and probed prod infra + the
    prod-pinned host kernel while the tests targeted the test fleet: a
    split-brain. Default profile is ``test`` (the isolated fleet + its
    containerized app services); the prod-subnet guard in conftest still backs
    this up. ``.env`` values are applied only when a key is unset (so an
    explicit caller env wins); the profile is applied with override (endpoints
    are authoritative) EXCEPT for keys the process claimed via
    ``core.env_bootstrap.declare_owned`` — see that function.
    """
    repo_root = Path(__file__).resolve().parent.parent
    loaded = 0
    for key, value in _parse_dotenv(repo_root / ".env").items():
        if key not in os.environ:
            # This IS the env loader, in the process that owns the environment;
            # it cannot import core.env_bootstrap because the harness must also
            # gate-ok: run standalone — see the ImportError fallback above
            os.environ[key] = value
            loaded += 1
    profile = os.environ.get("ORCHESTRATA_ENV_PROFILE", "test")
    profile_path = repo_root / "env" / f"{profile}.env"
    if not profile_path.exists():
        raise RuntimeError(
            f"env profile not found: {profile_path}; the harness must load a "
            "pinned endpoint profile so it never targets the wrong fleet."
        )
    for key, value in _parse_dotenv(profile_path).items():
        if key in _owned_keys() and key in os.environ:
            continue
        # gate-ok: same standalone loader, phase 2
        os.environ[key] = value  # override — the profile endpoints win
    return loaded


_loaded = _load_dotenv()
if _loaded:
    _profile = os.environ.get("ORCHESTRATA_ENV_PROFILE", "test")
    print(
        f"[env] Loaded .env + overlaid env/{_profile}.env "
        f"(REDIS_HOST={os.environ.get('REDIS_HOST')}, "
        f"CXP_GRPC_HOST={os.environ.get('CXP_GRPC_HOST')})",
        flush=True,
    )

# This script is local dev/test infrastructure tooling (it starts a gateway
# instance purely to exercise the test suite) -- not a production deploy
# path. It runs as its own subprocess (no PYTEST_CURRENT_TEST), so it must
# explicitly opt into the "dev-secret" JWT default (SPEC-099 fail-closed
# guard in gateway/config.py) rather than relying on a real JWT_SECRET.
os.environ.setdefault("ALLOW_DEV_JWT_SECRET", "1")


def _env(key: str, default: str = "") -> str:
    return os.environ.get(key, default)


# ── Data model ──────────────────────────────────────────────────────────────


@dataclass
class ServiceResult:
    name: str
    status: str = "pending"  # pending | running | started | failed | skipped
    port: int = 0
    host: str = "localhost"
    detail: str = ""
    duration_ms: float = 0.0
    pid: int = 0

    def emoji(self) -> str:
        return {
            "running": "✅",
            "started": "🔄",
            "failed": "❌",
            "skipped": "⏭️",
            "pending": "⬜",
        }.get(self.status, "❓")


@dataclass
class InfraReport:
    services: list[ServiceResult] = field(default_factory=list)
    all_required_ok: bool = False
    total_ms: float = 0.0

    def summary(self) -> str:
        lines = ["=" * 60, "INFRASTRUCTURE REPORT", "=" * 60, ""]
        for s in self.services:
            lines.append(
                f"  {s.emoji()} {s.name:<10} port={s.port:<6} {s.status:<10} {s.detail}"
            )
        lines.append("")
        lines.append("=" * 60)
        if self.all_required_ok:
            lines.append("✅ ALL SERVICES READY — you can run tests now")
        else:
            lines.append("❌ SOME SERVICES NOT READY")
            for s in self.services:
                if s.status == "failed":
                    lines.append(f"   - {s.name}: {s.detail}")
        lines.append("=" * 60)
        return "\n".join(lines)


# ── Process tracking (for atexit cleanup) ───────────────────────────────────

_started_procs: dict[str, subprocess.Popen] = {}


def _cleanup() -> None:
    """Kill processes we started, but only when CLEANUP_KERNEL=1."""
    if not os.environ.get("CLEANUP_KERNEL"):
        return
    drain_seconds = int(os.environ.get("ORCHESTRATA_SIGTERM_DRAIN_SECONDS", "10"))
    for name, proc in list(_started_procs.items()):
        if proc.poll() is None:
            print(f"🧹 Stopping {name} (atexit cleanup)...")
            proc.send_signal(signal.SIGTERM)
            try:
                proc.wait(timeout=drain_seconds)
            except subprocess.TimeoutExpired:
                proc.kill()


atexit.register(_cleanup)


# ── Helpers ─────────────────────────────────────────────────────────────────


def _port_open(host: str, port: int, timeout: float = 2.0) -> bool:
    """Check if a TCP port is accepting connections."""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except Exception:
        return False


def _wait_for_port(host: str, port: int, max_wait: float | None = None) -> bool:
    """Block until *port* is open or *max_wait* seconds elapse.

    ``max_wait`` defaults to ``ORCHESTRATA_PORT_WAIT_MAX`` env (or 30s),
    overridable so tests can exercise the timeout branch without paying the
    full production wait.
    """
    if max_wait is None:
        max_wait = float(os.environ.get("ORCHESTRATA_PORT_WAIT_MAX", "30"))
    deadline = time.monotonic() + max_wait
    while time.monotonic() < deadline:
        if _port_open(host, port):
            return True
        time.sleep(min(0.5, max(0.0, deadline - time.monotonic())))
    return False


def _run_bg(cmd: list[str], name: str, cwd: str | None = None) -> subprocess.Popen:
    """Spawn a background subprocess and track it for cleanup."""
    proc = subprocess.Popen(
        cmd,
        cwd=cwd,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    _started_procs[name] = proc
    return proc


# Hosts we can legitimately spawn a service ON. A configured endpoint outside
# this set (e.g. a containerized systemd unit at 10.53.2.13) is EXTERNALLY
# managed: we probe + verify it but must never fork a host-local process for it.
# Spawning locally would (a) bind the host's interfaces, never the remote IP, so
# the readiness probe could not pass, and (b) `source .env` (prod endpoints) —
# reintroducing the very split-brain the env profiles exist to prevent.
_LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1", "0.0.0.0", "::1", ""})


def _is_local_host(host: str) -> bool:
    """True iff a service on ``host`` is host-local and safe to spawn here."""
    return host in _LOCAL_HOSTS


# ── Service: Redis ──────────────────────────────────────────────────────────


def _redis_config() -> tuple[str, int]:
    host = _env("REDIS_HOST", "localhost")
    port = int(_env("REDIS_PORT", "6379"))
    return host, port


def check_redis() -> ServiceResult:
    host, port = _redis_config()
    t0 = time.monotonic()
    ok = _port_open(host, port)
    if ok:
        return ServiceResult(
            name="Redis",
            status="running",
            host=host,
            port=port,
            detail=f"{host}:{port}",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Redis",
        status="failed",
        host=host,
        port=port,
        detail=f"not listening on {host}:{port}",
    )


def start_redis() -> ServiceResult:
    host, port = _redis_config()
    t0 = time.monotonic()

    # Idempotent: if already healthy, don't attempt to (re)spawn one. This
    # mirrors start_gateway()/start_kernel()'s own check_fn() short-circuit —
    # required here because start_redis() (unlike start_kernel()) is also
    # called directly, not only via _run_one() after check_redis() already
    # ran. A previous version skipped this and relied solely on a local
    # "incus exec redis-queue -- redis-cli ping" probe to detect "already
    # up"; against the shared test fleet (container named "redis" in the
    # "orchestrata-test" Incus project, not "redis-queue" in the default
    # project) that probe always failed, silently fell through to the
    # spawn-then-wait path, and reported "started" for an instance that was
    # already running the whole time.
    existing = check_redis()
    if existing.status == "running":
        existing.detail = f"{host}:{port} (already up)"
        return existing

    if not _is_local_host(host):
        # Externally managed (containerized systemd unit in orchestrata-test).
        # check_redis() already found it DOWN — fail loudly rather than spawn
        # a prod-.env host redis that could never bind the container IP
        # anyway (see the _LOCAL_HOSTS module comment).
        return ServiceResult(
            name="Redis",
            status="failed",
            host=host,
            port=port,
            detail=(
                f"not reachable at {host}:{port} — redis is externally managed "
                f"(start the app-container service; host spawn refused)"
            ),
        )

    # Redis runs in a local Incus container — try redis-cli ping, then redis-server
    try:
        result = subprocess.run(
            ["incus", "exec", "redis-queue", "--", "redis-cli", "ping"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if "PONG" in result.stdout:
            return ServiceResult(
                name="Redis",
                status="running",
                host=host,
                port=port,
                detail=f"{host}:{port} (already up)",
            )

        incus_timeout = int(os.environ.get("ORCHESTRATA_INCUS_TIMEOUT", "10"))
        subprocess.run(
            [
                "incus",
                "exec",
                "redis-queue",
                "--",
                "redis-server",
                "--daemonize",
                "yes",
            ],
            capture_output=True,
            text=True,
            timeout=incus_timeout,
        )
    except FileNotFoundError:
        return ServiceResult(
            name="Redis",
            status="failed",
            host=host,
            port=port,
            detail="incus not found — cannot manage Redis",
        )
    except Exception as e:
        return ServiceResult(
            name="Redis", status="failed", host=host, port=port, detail=str(e)
        )

    if _wait_for_port(host, port):
        return ServiceResult(
            name="Redis",
            status="started",
            host=host,
            port=port,
            detail=f"{host}:{port}",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Redis",
        status="failed",
        host=host,
        port=port,
        detail="did not start within port-wait window",
    )


def verify_redis() -> ServiceResult:
    host, port = _redis_config()
    try:
        import redis as redis_lib

        client = redis_lib.Redis(host=host, port=port, socket_connect_timeout=3)
        ok = client.ping()
        if ok:
            return ServiceResult(
                name="Redis", status="running", host=host, port=port, detail="PONG"
            )
        return ServiceResult(
            name="Redis",
            status="failed",
            host=host,
            port=port,
            detail="PING returned False",
        )
    except ImportError:
        return ServiceResult(
            name="Redis",
            status="running",
            host=host,
            port=port,
            detail="redis-py not installed (skipped query check)",
        )
    except Exception as e:
        return ServiceResult(
            name="Redis", status="failed", host=host, port=port, detail=str(e)
        )


# ── Service: Graph ──────────────────────────────────────────────────────────


def _kg_uri() -> str:
    """Resolve the target graph-engine URI, honoring ``KG_ENGINE``.

    ``KG_ENGINE=memgraph``: ``KG_ROUTING_URI`` -> ``MEMGRAPH_URI`` ->
    ``KG_ROUTING_URI``/``KG_URI``. ``KG_ENGINE=neo4j`` (default):
    ``KG_ROUTING_URI`` -> ``KG_URI`` — unaffected, this script does its own
    env resolution (standalone tool, no ``orchestrata_ultimate`` import at
    module scope) mirroring ``core.graph_factory._resolve_uri``'s
    priority order for consistency with the rest of the migration's env
    seam (plans/neo4j-to-orchgraph-migration-plan.md).
    """
    if _kg_engine() == "memgraph":
        return _env("KG_ROUTING_URI") or _env("MEMGRAPH_URI") or _env("KG_URI", "")
    return _env("KG_ROUTING_URI", "") or _env("KG_URI", "")


def _graph_config() -> tuple[str, int]:
    """Resolve the graph host/port from the configured URI.

    Returns ``("", 0)`` when the URI is unset, or has no explicit
    ``host:port`` segment, or a non-numeric port — rather than letting
    ``str.rsplit``/``int()`` raise. An unset ``KG_ROUTING_URI``/``KG_URI``
    (or a malformed one) must surface as a clean "failed"/"skipped"
    preflight row, never a bare traceback that kills the whole run before
    a single service has been checked (F5).
    """
    uri = _kg_uri()
    if not uri:
        return "", 0
    host_port = uri.replace("bolt://", "").replace("neo4j://", "")
    if "/" in host_port:
        host_port = host_port.split("/")[0]
    if ":" not in host_port:
        return host_port, 0
    host, port_str = host_port.rsplit(":", 1)
    try:
        return host, int(port_str)
    except ValueError:
        return host, 0


def check_graph() -> ServiceResult:
    host, port = _graph_config()
    t0 = time.monotonic()
    if not host or not port:
        return ServiceResult(
            name="Graph",
            status="failed",
            host=host,
            port=port,
            detail="KG_ROUTING_URI/KG_URI unset (or missing a host:port)",
        )
    ok = _port_open(host, port)
    if ok:
        return ServiceResult(
            name="Graph",
            status="running",
            host=host,
            port=port,
            detail=f"{host}:{port}",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Graph",
        status="failed",
        host=host,
        port=port,
        detail=f"not listening on {host}:{port}",
    )


def _report_orchgraph(host: str, port: int, t0: float) -> ServiceResult:
    """Report the externally-managed orchgraph cluster's actual state.

    The cluster is a shared HA fleet (plans/neo4j-to-orchgraph-migration-
    plan.md), not a single Incus `neo4j-db` container this script can
    `incus exec ... neo4j start`. There is nothing to start, so this reports
    what is really there: "running" when the cluster answers, "failed" when
    it does not — a real outage for a human to fix, never something this
    script silently papers over.
    """
    if _port_open(host, port):
        return ServiceResult(
            name="Graph",
            status="running",
            host=host,
            port=port,
            detail=f"{host}:{port} (orchgraph cluster; externally managed, "
            "not started by this script)",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Graph",
        status="failed",
        host=host,
        port=port,
        detail="orchgraph cluster unreachable (KG_ENGINE=memgraph; "
        "externally managed, not started by this script)",
    )


def start_graph() -> ServiceResult:
    host, port = _graph_config()
    t0 = time.monotonic()
    if _kg_engine() == "memgraph":
        return _report_orchgraph(host, port, t0)
    try:
        incus_timeout = int(os.environ.get("ORCHESTRATA_INCUS_TIMEOUT", "30"))
        result = subprocess.run(
            ["incus", "exec", "neo4j-db", "--", "neo4j", "start"],
            capture_output=True,
            text=True,
            timeout=incus_timeout,
        )
        if result.returncode == 0 or "already running" in result.stderr.lower():
            if _wait_for_port(host, port):
                return ServiceResult(
                    name="Graph",
                    status="started",
                    host=host,
                    port=port,
                    detail=f"{host}:{port}",
                    duration_ms=(time.monotonic() - t0) * 1000,
                )
        return ServiceResult(
            name="Graph",
            status="failed",
            host=host,
            port=port,
            detail=result.stderr.strip() or "unknown error",
        )
    except FileNotFoundError:
        return ServiceResult(
            name="Graph",
            status="failed",
            host=host,
            port=port,
            detail="incus not found",
        )
    except Exception as e:
        return ServiceResult(
            name="Graph", status="failed", host=host, port=port, detail=str(e)
        )


def _kg_engine() -> str:
    """ "neo4j" (default) or "memgraph" — the Neo4j -> orchgraph dual-fleet
    switch (plans/neo4j-to-orchgraph-migration-plan.md Phase 2 Step 2.3),
    mirroring the ``KG_ENGINE`` read in tests/conftest.py."""
    return _env("KG_ENGINE", "neo4j")


def _ensure_kg_database(driver: Driver, database: str) -> None:
    """Idempotently ``CREATE DATABASE <database>`` on the orchgraph cluster.

    Measured live 2026-07-26: this fork's ``CREATE DATABASE`` is already
    idempotent (a second call for an existing database is a silent no-op,
    not an error) — no ``IF NOT EXISTS`` clause needed (and none exists:
    the parser rejects it). Still defensive against a future/older build
    that DOES error on a duplicate, by swallowing an "already exists"-
    shaped message.
    """
    # No database= binding: this statement CREATES the database, so it must
    # run on a session that is not already bound to it.
    with driver.session() as session:
        try:
            # The cast is load-bearing for pyright (the gated checker):
            # session.run() takes LiteralString, and an f-string is str.
            # mypy disagrees and calls it redundant — silenced there, not here.
            session.run(
                cast(  # type: ignore[redundant-cast]
                    LiteralString, f"CREATE DATABASE {database}"
                )
            ).consume()
        except Exception as exc:
            if "already exists" not in str(exc).lower():
                raise


def verify_graph() -> ServiceResult:
    host, port = _graph_config()
    uri = _kg_uri()
    user = ""
    pwd = ""
    engine = _kg_engine()
    try:
        from neo4j import GraphDatabase

        from orchestrata_ultimate.core.graph_factory import resolve_kg_database

        driver = GraphDatabase.driver(uri, auth=(user, pwd))
        if engine == "memgraph":
            # Multi-tenant orchgraph cluster: ensure THIS worker's database
            # exists (idempotent-guarded, plan Step 2.3) before probing it
            # — a fresh cluster/worker-slot combination has never created
            # it yet. auth=None on the live cluster (Step 0.2 ruling); the
            # (user, pwd) tuple above is accepted-but-ignored by the fork
            # when auth is disabled server-side.
            database = resolve_kg_database()
            _ensure_kg_database(driver, database)
        else:
            database = None
        session_cm = driver.session(database=database) if database else driver.session()
        with session_cm as session:
            record = session.run("RETURN 1 AS test").single()
            driver.close()
            if record and record["test"] == 1:
                return ServiceResult(
                    name="Graph",
                    status="running",
                    host=host,
                    port=port,
                    detail=f"RETURN 1 (engine={engine}, database={database or 'default'})",
                )
        return ServiceResult(
            name="Graph", status="failed", host=host, port=port, detail="query failed"
        )
    except ImportError:
        return ServiceResult(
            name="Graph",
            status="running",
            host=host,
            port=port,
            detail="neo4j package not installed (skipped query check)",
        )
    except Exception as e:
        return ServiceResult(
            name="Graph", status="failed", host=host, port=port, detail=str(e)
        )


# ── Service: Gateway ────────────────────────────────────────────────────────
#
# Professional long-running service management:
#   1. PID file         — /tmp/orchestrata-gateway.pid
#   2. Health check     — gateway.ping RPC (authenticated with demo token)
#   3. Graceful shutdown — SIGTERM → wait for PID removal → SIGKILL
#   4. Idempotent start — PID + port + health check before starting
#   5. Readiness probe  — wait for gateway.ping to respond, not just port
#   6. Log file         — /tmp/orchestrata-gateway.log (rotated by logrotate)

GATEWAY_PID_FILE = "/tmp/orchestrata-gateway.pid"
GATEWAY_LOG_FILE = "/tmp/orchestrata-gateway.log"
# Demo token generated at script runtime by _refresh_demo_token()
_DEMO_TOKEN: str = ""


def _refresh_demo_token() -> str:
    """Generate a fresh demo JWT token valid for 30 days."""
    import time as _time

    import jwt

    from orchestrata_ultimate.gateway.auth import TokenAuthenticator

    secret = TokenAuthenticator._get_secret()
    return jwt.encode(
        {
            "sub": "demo-user",
            "iat": int(_time.time()),
            "exp": int(_time.time()) + 86400 * 30,
        },
        secret,
        algorithm="HS256",
    )


_DEMO_TOKEN = _refresh_demo_token()


def _gateway_config() -> tuple[str, int]:
    # Harness PROBE target for the gateway. It reads a DEDICATED var
    # (ORCHESTRATA_GATEWAY_PROBE_HOST) rather than GATEWAY_HOST on purpose:
    # GATEWAY_HOST is what the gateway SERVER binds (gateway/config.py), and the
    # suite's in-process gateway tests must keep binding localhost. Pointing the
    # harness at the containerized gateway (10.53.2.13) via this var therefore
    # never perturbs what those servers bind. Falls back to GATEWAY_HOST then
    # localhost for host-local runs. Port is GATEWAY_PORT — the key the gateway
    # server itself uses (its default 8765 is what the container unit listens on).
    host = (
        _env("ORCHESTRATA_GATEWAY_PROBE_HOST", "")
        or _env("GATEWAY_HOST", "localhost")
        or "localhost"
    )
    port = int(
        _env("ORCHESTRATA_GATEWAY_PROBE_PORT", "") or _env("GATEWAY_PORT", "8765")
    )
    return host, port


def _gateway_pid() -> int | None:
    """Read the gateway PID from the PID file.  Returns None if stale."""
    try:
        with open(GATEWAY_PID_FILE) as fh:
            pid = int(fh.read().strip())
        os.kill(pid, 0)  # verify process exists
        return pid
    except FileNotFoundError, ValueError, ProcessLookupError, PermissionError:
        return None


def _gateway_health(host: str, port: int, timeout: float = 5.0) -> bool:
    """Check if gateway responds to gateway.ping (authenticated)."""
    try:
        import asyncio as _aio

        import websockets as _ws

        async def _ping() -> bool:
            # open_timeout too: the websockets default (10s) applies to the
            # HTTP upgrade handshake, so an endpoint that accepts TCP but
            # never answers (exactly what health-failure tests simulate) used
            # to burn 10s where the caller asked for `timeout` (measured
            # 2026-08-11: four failure tests at exactly 10.01s each).
            async with _ws.connect(
                f"ws://{host}:{port}", close_timeout=timeout, open_timeout=timeout
            ) as ws:
                await ws.send(
                    json.dumps(
                        {
                            "jsonrpc": "2.0",
                            "method": "auth.authenticate",
                            "params": {"token": _DEMO_TOKEN},
                            "id": 0,
                        }
                    )
                )
                auth_resp = await _aio.wait_for(ws.recv(), timeout=timeout)
                if "error" in json.loads(auth_resp):
                    return False
                await ws.send(
                    json.dumps({"jsonrpc": "2.0", "method": "gateway.ping", "id": 1})
                )
                resp = await _aio.wait_for(ws.recv(), timeout=timeout)
                # bool(): json.loads is Any, and Any == str is Any, not bool.
                return bool(json.loads(resp).get("result", {}).get("status") == "pong")

        return _aio.run(_ping())
    except Exception:
        return False


def _gateway_info(host: str, port: int, timeout: float = 5.0) -> dict[str, Any] | None:
    """Authenticate to the gateway and call gateway.info over its real
    WebSocket transport — the exact protocol ``_gateway_health`` speaks,
    reusing the same auth handshake, only reading ``gateway.info`` instead
    of ``gateway.ping``.

    Returns the RPC's ``result`` object, or ``None`` when the round trip
    itself could not be completed (connection refused, auth rejected,
    timeout, malformed JSON, a response carrying "error", or a "result"
    that isn't even a JSON object). ``None`` means "could not ask" — it is
    the caller's job to tell that apart from "asked, and the answer says
    something we don't like" (see ``check_kernel_link``).
    """
    try:
        import asyncio as _aio

        import websockets as _ws

        async def _info() -> dict[str, Any] | None:
            async with _ws.connect(
                f"ws://{host}:{port}", close_timeout=timeout, open_timeout=timeout
            ) as ws:
                await ws.send(
                    json.dumps(
                        {
                            "jsonrpc": "2.0",
                            "method": "auth.authenticate",
                            "params": {"token": _DEMO_TOKEN},
                            "id": 0,
                        }
                    )
                )
                auth_resp = json.loads(await _aio.wait_for(ws.recv(), timeout=timeout))
                if "error" in auth_resp:
                    return None
                await ws.send(
                    json.dumps({"jsonrpc": "2.0", "method": "gateway.info", "id": 1})
                )
                resp = json.loads(await _aio.wait_for(ws.recv(), timeout=timeout))
                if "error" in resp:
                    return None
                result = resp.get("result")
                return result if isinstance(result, dict) else None

        return _aio.run(_info())
    except Exception:
        return None


def _stop_gateway() -> ServiceResult:
    """Gracefully stop a running gateway.  SIGTERM → wait → SIGKILL."""
    pid = _gateway_pid()
    if pid is None:
        return ServiceResult(
            name="Gateway", status="failed", port=0, detail="not running"
        )
    try:
        os.kill(pid, signal.SIGTERM)
        drain_seconds = int(os.environ.get("ORCHESTRATA_SIGTERM_DRAIN_SECONDS", "10"))
        deadline = time.monotonic() + drain_seconds
        while time.monotonic() < deadline:
            try:
                os.kill(pid, 0)
                time.sleep(0.5)
            except ProcessLookupError:
                break
        else:
            try:
                os.kill(pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
    except ProcessLookupError:
        pass
    try:
        os.remove(GATEWAY_PID_FILE)
    except FileNotFoundError:
        pass
    return ServiceResult(
        name="Gateway",
        status="running",
        port=_gateway_config()[1],
        detail="stopped gracefully",
    )


def check_gateway() -> ServiceResult:
    """Report whether a HEALTHY gateway is answering, not whether we own it.

    Provenance is not health. This used to require ``GATEWAY_PID_FILE``, which
    only the launcher below writes, so a gateway started any other way was
    reported "port open but no PID — foreign process?" and the whole run
    aborted. That is exactly what happened when the gateway moved to
    ``orchestrata-gateway.service`` (2026-07-27): systemd owns the process and
    writes no such file, so a perfectly healthy gateway read as a failure.
    Same reasoning as ``_report_orchgraph`` — an externally-managed service
    that answers its health check IS running.

    The PID file is still consulted, but only to enrich the detail line and to
    clean itself up when it names a process that no longer holds the port.
    """
    host, port = _gateway_config()
    t0 = time.monotonic()
    pid = _gateway_pid()
    port_ok = _port_open(host, port)

    if port_ok:
        if _gateway_health(host, port, timeout=3):
            owner = f"pid={pid}" if pid is not None else "externally managed"
            return ServiceResult(
                name="Gateway",
                status="running",
                host=host,
                port=port,
                detail=f"{owner}, gateway.ping OK",
                # 0 is this dataclass's "no pid known" default — an
                # externally-managed gateway has no PID file to read.
                pid=pid if pid is not None else 0,
                duration_ms=(time.monotonic() - t0) * 1000,
            )
        owner = f"pid={pid}, " if pid is not None else ""
        return ServiceResult(
            name="Gateway",
            status="failed",
            host=host,
            port=port,
            detail=f"{owner}port open but health check failed",
        )

    if pid is not None:
        try:
            os.remove(GATEWAY_PID_FILE)
        except FileNotFoundError:
            pass
        return ServiceResult(
            name="Gateway",
            status="failed",
            host=host,
            port=port,
            detail="stale PID file cleaned up",
        )
    return ServiceResult(
        name="Gateway",
        status="failed",
        host=host,
        port=port,
        detail=f"not running on {host}:{port}",
    )


def start_gateway() -> ServiceResult:
    host, port = _gateway_config()
    t0 = time.monotonic()

    # Idempotent: if already healthy, don't restart
    existing = check_gateway()
    if existing.status == "running":
        existing.detail = f"already running (pid={existing.pid})"
        return existing

    if not _is_local_host(host):
        # Externally managed (containerized systemd unit in orchestrata-test).
        # check_gateway found it unhealthy — fail loudly rather than spawn a
        # prod-.env host gateway that could never bind the container IP.
        return ServiceResult(
            name="Gateway",
            status="failed",
            host=host,
            port=port,
            detail=(
                f"not reachable at {host}:{port} — gateway is externally managed "
                f"(start the app-container service; host spawn refused)"
            ),
        )

    # Refuse if port is in use by something else
    if _port_open(host, port) and _gateway_pid() is None:
        return ServiceResult(
            name="Gateway",
            status="failed",
            host=host,
            port=port,
            detail=f"port {port} in use by foreign process",
        )

    repo_root = Path(__file__).resolve().parent.parent
    subprocess.Popen(
        [
            "bash",
            "-c",
            "set -a; source .env; set +a; "
            f"echo $$ > {GATEWAY_PID_FILE}; "
            f"exec uv run python -m orchestrata_ultimate.gateway.cli serve "
            f">> {GATEWAY_LOG_FILE} 2>&1",
        ],
        cwd=str(repo_root),
        start_new_session=True,
    )

    # Readiness: wait for health check to pass
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline:
        if _gateway_pid() and _gateway_health(host, port, timeout=3):
            pid = _gateway_pid()
            return ServiceResult(
                name="Gateway",
                status="started",
                host=host,
                port=port,
                detail=f"pid={pid}, gateway.ping OK",
                pid=pid or 0,
                duration_ms=(time.monotonic() - t0) * 1000,
            )
        time.sleep(0.5)

    _stop_gateway()
    return ServiceResult(
        name="Gateway",
        status="failed",
        host=host,
        port=port,
        detail=f"did not become healthy within 30s (log: {GATEWAY_LOG_FILE})",
    )


def verify_gateway() -> ServiceResult:
    host, port = _gateway_config()
    if _gateway_health(host, port):
        return ServiceResult(
            name="Gateway",
            status="running",
            host=host,
            port=port,
            detail="gateway.ping → pong",
        )
    return ServiceResult(
        name="Gateway",
        status="failed",
        host=host,
        port=port,
        detail="gateway.ping failed",
    )


# ── Service: Kernel ─────────────────────────────────────────────────────────


def _kernel_config() -> tuple[str, int, bool]:
    raw_port = _env("CXP_GRPC_PORT", "")
    configured = bool(raw_port)
    # An explicitly-set-but-empty CXP_GRPC_PORT (e.g. `env CXP_GRPC_PORT=`) must
    # be treated the same as "unset" for the port value too — falling through
    # to int(_env(..., "50053")) would re-read the same empty string (the
    # default only applies when the key is ABSENT from os.environ, not when
    # it's present-but-empty) and crash with ValueError instead of reporting
    # "not configured".
    port = int(raw_port) if raw_port else 50053
    # Honor CXP_GRPC_HOST from the active env profile. This USED to be hardcoded
    # "localhost", which made the harness probe/verify a host-local kernel while
    # pytest (via conftest + env/<profile>.env) targeted a different fleet — the
    # split-brain that let the standing prod-pinned host kernel masquerade as the
    # test kernel. The kernel now lives as a containerized systemd unit in the
    # orchestrata-test project (10.53.2.13), so the harness must probe THERE.
    host = _env("CXP_GRPC_HOST", "localhost") or "localhost"
    return host, port, configured


def check_kernel() -> ServiceResult:
    host, port, configured = _kernel_config()
    t0 = time.monotonic()
    if not configured:
        return ServiceResult(
            name="Kernel",
            status="skipped",
            host=host,
            port=port,
            detail="CXP_GRPC_PORT not configured",
        )
    ok = _port_open(host, port)
    if ok:
        return ServiceResult(
            name="Kernel",
            status="running",
            host=host,
            port=port,
            detail=f"grpc://{host}:{port}",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Kernel",
        status="failed",
        host=host,
        port=port,
        detail=f"not listening on {host}:{port}",
    )


def start_kernel() -> ServiceResult:
    host, port, configured = _kernel_config()
    if not configured:
        return ServiceResult(
            name="Kernel",
            status="skipped",
            host=host,
            port=port,
            detail="not configured",
        )
    t0 = time.monotonic()
    if not _is_local_host(host):
        # Externally managed (containerized systemd unit in orchestrata-test).
        # check_kernel already found it DOWN — fail loudly rather than spawn a
        # prod-.env host kernel that could never bind the container IP anyway.
        return ServiceResult(
            name="Kernel",
            status="failed",
            host=host,
            port=port,
            detail=(
                f"not reachable at {host}:{port} — kernel is externally managed "
                f"(start the app-container service; host spawn refused)"
            ),
        )
    repo_root = Path(__file__).resolve().parent.parent

    _run_bg(
        [
            "bash",
            "-c",
            "set -a; source .env; set +a; "
            "uv run python -m orchestrata_ultimate.agents.runtime_main",
        ],
        name="kernel",
        cwd=str(repo_root),
    )

    start_timeout = int(os.environ.get("ORCHESTRATA_KERNEL_START_TIMEOUT", "30"))
    if _wait_for_port(host, port, max_wait=start_timeout):
        return ServiceResult(
            name="Kernel",
            status="started",
            host=host,
            port=port,
            detail=f"grpc://{host}:{port}",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Kernel",
        status="failed",
        host=host,
        port=port,
        detail=f"did not start within {start_timeout}s",
    )


def verify_kernel() -> ServiceResult:
    host, port, configured = _kernel_config()
    if not configured:
        return ServiceResult(
            name="Kernel",
            status="skipped",
            host=host,
            port=port,
            detail="not configured",
        )
    try:
        import asyncio as _aio

        from orchestrata_ultimate.cxp.proto_codec import CXPClientConfig
        from orchestrata_ultimate.cxp.transport import create_cxp_client

        async def _test() -> bool:
            client = create_cxp_client(
                config=CXPClientConfig(host=host, port=port, timeout_seconds=5)
            )
            await client.connect()
            await client.disconnect()
            return True

        ok = _aio.run(_test())
        if ok:
            return ServiceResult(
                name="Kernel",
                status="running",
                host=host,
                port=port,
                detail="CXP transport OK",
            )
        return ServiceResult(
            name="Kernel",
            status="failed",
            host=host,
            port=port,
            detail="gRPC connect failed",
        )
    except ImportError:
        return ServiceResult(
            name="Kernel",
            status="running",
            host=host,
            port=port,
            detail="package not installed (skipped)",
        )
    except Exception as e:
        return ServiceResult(
            name="Kernel", status="failed", host=host, port=port, detail=str(e)[:80]
        )


# ── Service: Link (Gateway ↔ Kernel) ────────────────────────────────────────
#
# check_gateway/verify_gateway and check_kernel/verify_kernel above each probe
# ONE process in isolation. THE BUG this check closes: the deployed gateway
# probed the kernel's gRPC port before the kernel bound it, fell back to the
# in-process PipelineKernelClient, and never retried — yet both independent
# probes still reported healthy, because "the gateway answers gateway.ping"
# and "the kernel accepts a CXP connection" are each true on their own even
# when the gateway is not actually talking to that kernel. This check is the
# only one that asks the gateway what it thinks it is wired to.
#
# Contract (fixed elsewhere — this check only reads it):
#   gateway.info returns a "kernel" block: {"mode": "grpc"|"in-process"|
#   "unavailable", "endpoint": "host:port"|None, "wired": bool}. PASS only
#   when mode == "grpc". A MISSING "kernel" block (e.g. a gateway build that
#   hasn't shipped the contract yet) is UNKNOWN, not OK — it must FAIL, the
#   same as a bad mode, so "0 problems found" is never confused with
#   "the reporter couldn't even ask the question".


def check_kernel_link() -> ServiceResult:
    """Verify the gateway is actually talking to the kernel it reports,
    not merely that gateway.ping and the kernel's own transport each answer
    independently. See the module comment above for the bug this closes.
    """
    host, port = _gateway_config()
    t0 = time.monotonic()
    info = _gateway_info(host, port, timeout=5)
    if info is None:
        return ServiceResult(
            name="Link",
            status="failed",
            host=host,
            port=port,
            detail="gateway.info unreachable — cannot verify the gateway↔kernel link",
        )
    kernel = info.get("kernel")
    if not isinstance(kernel, dict) or "mode" not in kernel:
        return ServiceResult(
            name="Link",
            status="failed",
            host=host,
            port=port,
            detail=(
                "gateway.info has no 'kernel' block — wiring is UNKNOWN "
                "(older gateway build?), not verified OK"
            ),
        )
    mode = kernel.get("mode")
    endpoint = kernel.get("endpoint") or "?"
    if mode == "grpc":
        # `mode` is the STARTUP outcome; `wired` folds in a liveness probe
        # taken when gateway.info was answered. Trusting mode alone would
        # pass for a gateway whose kernel died after boot -- a stale green,
        # which is the exact failure class this check exists to catch. Both
        # keys must agree, and a build that omits `wired` is UNKNOWN, not OK.
        if "wired" not in kernel:
            return ServiceResult(
                name="Link",
                status="failed",
                host=host,
                port=port,
                detail=(
                    "gateway.info kernel block has no 'wired' field — "
                    "liveness is UNKNOWN, not verified OK"
                ),
            )
        if kernel.get("wired") is not True:
            return ServiceResult(
                name="Link",
                status="failed",
                host=host,
                port=port,
                detail=(
                    f"gateway was wired to the kernel at {endpoint} at boot "
                    "but it does NOT answer now (mode=grpc, wired=false) — "
                    "traffic is not reaching the Agent Runtime kernel"
                ),
            )
        return ServiceResult(
            name="Link",
            status="running",
            host=host,
            port=port,
            detail=f"gateway wired to kernel via grpc at {endpoint} (live)",
            duration_ms=(time.monotonic() - t0) * 1000,
        )
    return ServiceResult(
        name="Link",
        status="failed",
        host=host,
        port=port,
        detail=(
            f"gateway↔kernel link mode={mode!r} — the gateway is running "
            "KERNEL-LESS (no real gRPC wiring to the Agent Runtime kernel)"
        ),
    )


# ── Orchestration ───────────────────────────────────────────────────────────

# Each service: (check_fn, start_fn, verify_fn, required: bool)
SERVICES: list[tuple[str, Callable, Callable, Callable, bool]] = [
    ("redis", check_redis, start_redis, verify_redis, True),
    ("graph", check_graph, start_graph, verify_graph, True),
    ("gateway", check_gateway, start_gateway, verify_gateway, True),
    # The Agent Runtime kernel is REQUIRED, not optional. It is the beating
    # heart of the system: a suite run that does not start AND verify it is not
    # a valid run — kernel-dependent paths would go silently untested and the
    # result would be false-green. Flipping this from False (which let
    # _run_one skip the kernel outright) to True makes the kernel started +
    # gRPC-verified every run, and makes its failure fatal (all_required_ok).
    # A rare environment that genuinely cannot run it must opt out LOUDLY and
    # explicitly via `--skip kernel`, which marks the run non-authoritative.
    ("kernel", check_kernel, start_kernel, verify_kernel, True),
]


def _run_one(
    check_fn: Callable[[], ServiceResult],
    start_fn: Callable[[], ServiceResult],
    verify_fn: Callable[[], ServiceResult],
    required: bool,
    start_services: bool,
) -> ServiceResult:
    """Check → start (if down) → verify a single service.

    The service name is not a parameter: each check/start/verify function stamps
    its own canonical name onto the ``ServiceResult`` it returns, and the caller
    tracks the lowercase key separately in the futures map.
    """
    # 1. Check
    result = check_fn()
    if result.status == "running":
        # 2. Verify — confirm it's actually functional
        v = verify_fn()
        if v.status == "running":
            v.duration_ms = result.duration_ms
            return v
        # Port open but query failed — mark as failed
        result.status = "failed"
        result.detail = f"port open but query failed: {v.detail}"
        return result

    # Service is down
    if not start_services:
        # Status-only mode — report as-is
        return result

    if not required:
        return ServiceResult(
            name=result.name,
            status="skipped",
            port=result.port,
            detail="optional service not started",
        )

    # 3. Start
    started = start_fn()
    if started.status != "started":
        return started  # failed to start

    # 4. Verify
    v = verify_fn()
    if v.status == "running":
        v.status = "started"  # distinguish from already-running
        v.duration_ms = started.duration_ms
        return v
    v.status = "failed"
    v.detail = f"started but query failed: {v.detail}"
    return v


_SKIP_PORT_LOOKUP: dict[str, Callable[[], int]] = {
    "redis": lambda: _redis_config()[1],
    "graph": lambda: _graph_config()[1],
    "gateway": lambda: _gateway_config()[1],
    "kernel": lambda: _kernel_config()[1],
    "link": lambda: _gateway_config()[1],
}


def _skip_result(name: str) -> ServiceResult:
    """Build the "skipped" row for a ``--skip``'d service.

    The port lookup runs INSIDE the submitted callable (executed by the
    thread pool, not as an eagerly-evaluated default-argument expression at
    loop-construction time) and is wrapped so an unconfigured/malformed env
    for the skipped service itself (e.g. ``--skip graph`` with
    ``KG_ROUTING_URI`` unset) can never crash the whole preflight before a
    single service has even been checked (F5).
    """
    try:
        port = _SKIP_PORT_LOOKUP.get(name, lambda: 0)()
    except Exception:
        port = 0
    return ServiceResult(
        name=name, status="skipped", port=port, detail="excluded via --skip"
    )


def run_all(start_services: bool = True, skip: set[str] | None = None) -> InfraReport:
    """Run check/start/verify for all services in parallel, then verify the
    LINK between the gateway and the kernel (see ``check_kernel_link``).

    The link check runs strictly AFTER the ``ThreadPoolExecutor`` block below
    has joined — i.e. only once gateway and kernel have each fully settled
    their own check/start/verify cycle — so it can never race their startup
    and report a false "kernel-less" failure for a gateway that simply
    hadn't finished connecting yet.
    """
    skip = skip or set()
    report = InfraReport()
    t0 = time.monotonic()

    with ThreadPoolExecutor(max_workers=4) as pool:
        futures = {}
        for name, check_fn, start_fn, verify_fn, required in SERVICES:
            if name in skip:
                futures[pool.submit(_skip_result, name)] = name
            else:
                futures[
                    pool.submit(
                        _run_one,
                        check_fn,
                        start_fn,
                        verify_fn,
                        required,
                        start_services,
                    )
                ] = name

        for future in as_completed(futures):
            result = future.result()
            report.services.append(result)

    # Sort for consistent output
    report.services.sort(
        key=lambda s: ["redis", "graph", "gateway", "kernel"].index(s.name.lower())
    )

    # ── Link (appended last — see check_kernel_link's module comment) ──────
    # There is nothing wired to verify when gateway and/or kernel were not
    # themselves checked this run, so an explicit `--skip gateway`/`--skip
    # kernel` cascades to skip the link too, same as an explicit `--skip
    # link`. This is what keeps `--skip kernel` (a legitimate, already-
    # supported opt-out — see the SERVICES comment above) from being
    # silently upgraded into a hard failure by a check that has no gateway
    # or kernel run of its own to reason about.
    link_cascaded = bool({"gateway", "kernel"} & skip)
    if "link" in skip:
        link_result = _skip_result("link")
    elif link_cascaded:
        link_result = ServiceResult(
            name="link",
            status="skipped",
            port=_gateway_config()[1],
            detail="gateway and/or kernel excluded via --skip — link not verified",
        )
    else:
        link_result = check_kernel_link()
    report.services.append(link_result)

    # Check required services are all running/started
    required_names = {n for n, _, _, _, r in SERVICES if r} - skip
    if "link" not in skip and not link_cascaded:
        required_names = required_names | {"link"}
    ok_count = sum(
        1
        for s in report.services
        if s.name.lower() in required_names and s.status in ("running", "started")
    )
    report.all_required_ok = ok_count == len(required_names)

    report.total_ms = (time.monotonic() - t0) * 1000
    return report


# ── CLI ─────────────────────────────────────────────────────────────────────


def _validate_env_or_exit() -> None:
    """Preflight: fail fast on missing REQUIRED env keys, before any infra work.

    Loads scripts/validate_env.py by file path (not a package import — this
    script is invoked directly via ``python scripts/start_test_services.py``,
    where ``scripts`` may not be resolvable as a parent package) and calls
    its ``validate_required()`` seam. Reproducibility plan Phase R2.
    """
    repo_root = Path(__file__).resolve().parent.parent
    spec = importlib.util.spec_from_file_location(
        "validate_env", repo_root / "scripts" / "validate_env.py"
    )
    if spec is None or spec.loader is None:  # pragma: no cover - defensive
        return
    validate_env = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(validate_env)
    missing = validate_env.validate_required()
    if missing:
        print(
            f"[env] FAIL: {len(missing)} REQUIRED env key(s) missing/empty: "
            f"{', '.join(missing)}",
            file=sys.stderr,
        )
        print(
            "[env] Set them in .env (see .env.example), or run "
            "`python scripts/validate_env.py` for the full report.",
            file=sys.stderr,
        )
        sys.exit(1)


def _suite_already_swept(env: Mapping[str, str] | None = None) -> bool:
    """True when an outer runner already verified the infra and swept.

    The marker is the EXISTING ``PYTEST_SKIP_SERVICE_START`` that
    ``run_tests._check_infra`` exports only AFTER this script succeeded for
    it (so after the outermost run's sweep). Nested ``run_tests.py`` /
    ``start_test_services.py`` runs inherit it and skip the deep sweep —
    the 5-6 minute scan of the 725,636-node test primary that every nested
    test run used to repeat (ADR-0087 item 3, precedent ``f5b937feb``:
    duplicated work removed, verification kept).
    """
    env = env if env is not None else os.environ
    return bool(env.get("PYTEST_SKIP_SERVICE_START"))


def main() -> int:
    _validate_env_or_exit()
    parser = argparse.ArgumentParser(
        description="Parallel infrastructure readiness check"
    )
    parser.add_argument(
        "--status", action="store_true", help="Check only — do not start any services"
    )
    parser.add_argument("--json", action="store_true", help="Output report as JSON")
    parser.add_argument(
        "--skip",
        default="",
        help="Comma-separated service names to skip (e.g. kernel,gateway,graph,link). "
        "'neo4j' is accepted as a deprecated alias for 'graph'. Skipping "
        "'gateway' or 'kernel' also skips 'link' — the gateway<->kernel "
        "wiring check has nothing to verify when either side wasn't run.",
    )
    parser.add_argument(
        "--ensure-worker-dbs",
        type=int,
        default=0,
        metavar="N",
        help="Idempotently CREATE the per-worker band databases orchkg_w0.."
        "orchkg_w<N-1> after the graph check. The batched gate runner passes "
        "its worker count here: gate children run with "
        "PYTEST_SKIP_SERVICE_START=1 and never provision their own band, so "
        "band existence was historical sediment — a purged/fresh cluster "
        "failed every worker whose band no other run had ever created "
        "(iteration-13, 2026-08-05: 29 failures on orchkg_w14/w15).",
    )
    args = parser.parse_args()

    skip_set = {s.strip().lower() for s in args.skip.split(",") if s.strip()}
    if "neo4j" in skip_set:
        # Deprecated alias (graph-rename plan §2): "neo4j" was the service
        # key before the rename to "graph". Map it forward instead of
        # silently skipping nothing, so old muscle memory (`--skip neo4j`)
        # still skips the graph service.
        print(
            "[skip] 'neo4j' is a deprecated alias for the 'graph' service — "
            "update your --skip value; mapping it to 'graph' for this run.",
            file=sys.stderr,
        )
        skip_set.discard("neo4j")
        skip_set.add("graph")
    start_services = not args.status

    report = run_all(start_services=start_services, skip=skip_set)

    # Test-pollution sweep (backlog #124): crashed/SIGKILL'd runs leak their
    # seeded ``_test_``-marked nodes into the shared KG because fixture
    # teardown never ran. Sweep them at suite start — before any worker has
    # seeded anything — so pollution cannot accumulate. Best-effort: a dead
    # KG is already reported by the Graph readiness gate.
    if (
        start_services
        and "graph" not in skip_set
        and report.all_required_ok
        and _suite_already_swept()
    ):
        # ADR-0087 item 3: the OUTERMOST runner already verified the
        # infrastructure and swept (it sets PYTEST_SKIP_SERVICE_START only
        # after this very script succeeded). A nested runner's readiness
        # checks above still ran; only the deep sweep is not repeated.
        # stderr: stdout may carry a --json document.
        print(
            "[kg-sweep] skipped: the outer runner already swept this suite "
            "(PYTEST_SKIP_SERVICE_START is set)",
            file=sys.stderr,
        )
    elif start_services and "graph" not in skip_set and report.all_required_ok:
        # Resolve the sibling module even when this file is imported from
        # outside scripts/ (tests load it via importlib; scripts/ is then
        # not on sys.path and the bare import raises ModuleNotFoundError).
        _scripts_dir = str(Path(__file__).resolve().parent)
        if _scripts_dir not in sys.path:
            sys.path.insert(0, _scripts_dir)
        from sweep_test_pollution import sweep_if_polluted

        # deep=True: nothing of this run has seeded anything yet at this
        # point (see sweep_test_pollution's module docstring, "fourth,
        # DEEP-MARKER criterion" paragraph) — the one safe moment to also
        # sweep the DEEP-MARKER pollution criterion against the primary.
        _, sweep_detail = sweep_if_polluted(deep=True)
        if sweep_detail != "clean":
            print(f"[kg-sweep] {sweep_detail}")

    if (
        args.ensure_worker_dbs > 0
        and start_services
        and "graph" not in skip_set
        and report.all_required_ok
        and _kg_engine() == "memgraph"
    ):
        from neo4j import GraphDatabase

        driver = GraphDatabase.driver(_kg_uri(), auth=None)
        try:
            for slot in range(args.ensure_worker_dbs):
                _ensure_kg_database(driver, f"orchkg_w{slot}")
        finally:
            driver.close()
        print(f"[worker-dbs] ensured orchkg_w0..orchkg_w{args.ensure_worker_dbs - 1}")

    if args.json:
        out = {
            "all_required_ok": report.all_required_ok,
            "total_ms": report.total_ms,
            "services": [
                {"name": s.name, "status": s.status, "port": s.port, "detail": s.detail}
                for s in report.services
            ],
        }
        print(json.dumps(out, indent=2))
    else:
        print(report.summary())

    # --status is a pure report: it exits 0 as long as the check ran, with the
    # readiness state carried in the output (text/JSON). Only the start path
    # gates on readiness (exit 1) so run_tests.py aborts when a required
    # service can't be brought up. Exiting 0 here also lets infra_check.py read
    # the per-service breakdown — it only parses the report when returncode==0.
    if args.status:
        return 0
    return 0 if report.all_required_ok else 1


if __name__ == "__main__":
    sys.exit(main())