feat(gateway): gateway-owned control socket — identify/status verbs, fleet consumers prefer it over scans (#92091 step 1)
The gateway now creates a local control socket at startup (Unix domain socket at $HERMES_HOME/gateway.sock with a pointer-file fallback for long paths; named pipe on Windows) and answers versioned JSON verbs: - identify: pid, profile, hermes_home, code_sha/code_version (#91283 stamps, now queryable live), self-declared supervisor kind, start_time - status: the live runtime-status payload, answered by the process itself Bound immediately after the PID-file O_EXCL claim (the moment the process becomes the authoritative gateway for its HERMES_HOME), removed on clean shutdown; a successor clears any stale socket on bind. Strictly non-fatal: bind failure only means consumers use the old path. Consumers migrated (observability only, scan layer demoted to fallback, never deleted): - collect_fleet_versions() (post-update fleet matrix): prefers a live identify answer over gateway_state.json; entries carry source=socket - collect_runtime_inventory() (hermes update --plan): prefers the socket, and takes the gateway's own supervisor declaration instead of inferring it from PID scans Old gateways mid-upgrade, crashed processes, and bind failures behave exactly as before. Never a TCP port; filesystem/pipe ACLs are the auth boundary (0600 socket). Part of #91277 (fleet-update reliability). Design: #92091.
This commit is contained in:
Teknium
@@ -0,0 +1,513 @@
|
||||
"""
|
||||
Gateway control socket — the gateway-owned local coordination surface.
|
||||
|
||||
Migration step 1 of the #92091 design: every other process on the machine
|
||||
(the updater, `hermes serve`/dashboard, the Desktop app) currently discovers
|
||||
gateway identity/state by scanning the process table and string-matching argv
|
||||
or by reading ``gateway_state.json`` (which can outlive its writer). This
|
||||
module gives the gateway an OWNED contract instead: a local-only socket the
|
||||
gateway process creates at startup and removes on clean shutdown, answering
|
||||
versioned JSON verbs. A connectable socket with a well-formed ``identify``
|
||||
answer IS liveness — no PID-reuse heuristics.
|
||||
|
||||
v1 verbs (observation only — no behavior change for the gateway):
|
||||
|
||||
- ``identify`` → pid, profile label, hermes_home, code_sha/code_version
|
||||
(the #91283 stamps, now queryable live), supervisor kind, served profiles,
|
||||
start_time, protocol version.
|
||||
- ``status`` → the live runtime-status payload (what ``gateway_state.json``
|
||||
holds today, but answered by the process itself, race-free).
|
||||
|
||||
Transport:
|
||||
|
||||
- POSIX: Unix domain socket at ``$HERMES_HOME/gateway.sock``. When the home
|
||||
path is too long for ``sun_path`` (~104 bytes on macOS/BSD), the socket is
|
||||
bound in the system temp dir and a pointer file
|
||||
``$HERMES_HOME/gateway.sock.path`` records the real location; clients
|
||||
follow the pointer transparently.
|
||||
- Windows: named pipe ``\\\\.\\pipe\\hermes-gateway-<home-hash>`` served via
|
||||
the proactor event loop. Same trust model (per-user namespace).
|
||||
|
||||
Never a TCP port. Filesystem/pipe ACLs are the auth boundary — the same
|
||||
trust model as ``gateway_state.json`` today.
|
||||
|
||||
Consumers (``hermes update --plan`` inventory, the post-update fleet version
|
||||
matrix) PREFER the socket when it answers and fall back to the existing
|
||||
state-file/scan layer when it doesn't — old gateways mid-upgrade and crashed
|
||||
processes keep working exactly as before. The scan layer is demoted, not
|
||||
deleted.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CONTROL_PROTOCOL_VERSION = 1
|
||||
|
||||
_SOCKET_FILENAME = "gateway.sock"
|
||||
_POINTER_FILENAME = "gateway.sock.path"
|
||||
_IS_WINDOWS = sys.platform == "win32"
|
||||
|
||||
# Practical sun_path limit: 104 on macOS/BSD, 108 on Linux. Stay under the
|
||||
# smaller bound with margin for the NUL terminator.
|
||||
_MAX_UNIX_PATH = 100
|
||||
|
||||
# Requests and responses are single JSON lines. Bound them so a misbehaving
|
||||
# peer can't balloon gateway memory.
|
||||
_MAX_REQUEST_BYTES = 64 * 1024
|
||||
_MAX_RESPONSE_BYTES = 512 * 1024
|
||||
|
||||
_DEFAULT_CLIENT_TIMEOUT = 2.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Path resolution (shared by server and client)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _home_hash(home: Path) -> str:
|
||||
canonical = os.path.normcase(str(Path(home).expanduser().resolve(strict=False)))
|
||||
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
def windows_pipe_name(home: Path) -> str:
|
||||
"""Per-HERMES_HOME named pipe path (Windows transport)."""
|
||||
return rf"\\.\pipe\hermes-gateway-{_home_hash(home)}"
|
||||
|
||||
|
||||
def _pointer_path(home: Path) -> Path:
|
||||
return Path(home) / _POINTER_FILENAME
|
||||
|
||||
|
||||
def _default_socket_path(home: Path) -> Path:
|
||||
return Path(home) / _SOCKET_FILENAME
|
||||
|
||||
|
||||
def _fallback_socket_path(home: Path) -> Path:
|
||||
"""Short temp-dir path for homes whose direct socket path exceeds sun_path."""
|
||||
return Path(tempfile.gettempdir()) / f"hermes-gw-{_home_hash(home)}.sock"
|
||||
|
||||
|
||||
def resolve_server_socket_path(home: Path) -> tuple[Path, Optional[Path]]:
|
||||
"""Where the server should bind, plus the pointer file to write (or None).
|
||||
|
||||
Returns ``(bind_path, pointer_file)``. ``pointer_file`` is non-None only
|
||||
when the direct in-home path is too long and the temp-dir fallback is in
|
||||
use — the server must then persist the real location for clients.
|
||||
"""
|
||||
direct = _default_socket_path(home)
|
||||
if len(str(direct).encode("utf-8")) <= _MAX_UNIX_PATH:
|
||||
return direct, None
|
||||
return _fallback_socket_path(home), _pointer_path(home)
|
||||
|
||||
|
||||
def resolve_client_socket_path(home: Path) -> Optional[Path]:
|
||||
"""Where a client should connect for ``home``, or None when nothing exists."""
|
||||
direct = _default_socket_path(home)
|
||||
if direct.exists():
|
||||
return direct
|
||||
pointer = _pointer_path(home)
|
||||
try:
|
||||
if pointer.is_file():
|
||||
target = pointer.read_text(encoding="utf-8").strip()
|
||||
if target:
|
||||
candidate = Path(target)
|
||||
if candidate.exists():
|
||||
return candidate
|
||||
except OSError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Default payload builders (import-light; overridable at wiring time)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _detect_supervisor() -> str:
|
||||
"""Best-effort supervisor kind for THIS process, from its own environment.
|
||||
|
||||
Unlike the outside-in `_detect_supervisor_for_pid` scan, this answers from
|
||||
the process's own launch context — which is exactly the provenance the
|
||||
#92091 design wants declared rather than inferred.
|
||||
"""
|
||||
env = os.environ
|
||||
if env.get("INVOCATION_ID"):
|
||||
return "systemd"
|
||||
if sys.platform == "darwin" and (
|
||||
env.get("XPC_SERVICE_NAME", "").startswith("ai.hermes")
|
||||
or env.get("LAUNCHD_SOCKET")
|
||||
):
|
||||
return "launchd"
|
||||
if env.get("HERMES_DESKTOP_MANAGED"):
|
||||
return "desktop"
|
||||
if "--external-supervisor" in sys.argv:
|
||||
return "external"
|
||||
return "manual"
|
||||
|
||||
|
||||
def build_identify_payload() -> dict[str, Any]:
|
||||
"""Default ``identify`` answer, built from gateway.status primitives."""
|
||||
from gateway.status import (
|
||||
_build_pid_record,
|
||||
_get_code_identity_fields,
|
||||
_profile_label_for_home,
|
||||
read_runtime_status,
|
||||
)
|
||||
|
||||
record = _build_pid_record()
|
||||
payload: dict[str, Any] = {
|
||||
"protocol": CONTROL_PROTOCOL_VERSION,
|
||||
"kind": record.get("kind"),
|
||||
"pid": record.get("pid"),
|
||||
"start_time": record.get("start_time"),
|
||||
"hermes_home": record.get("hermes_home"),
|
||||
"profile": _profile_label_for_home(record.get("hermes_home") or ""),
|
||||
"supervisor": _detect_supervisor(),
|
||||
}
|
||||
payload.update(_get_code_identity_fields())
|
||||
# served_profiles (multiplex mode) is stamped into the runtime status by
|
||||
# the runner; surface it when present so fleet consumers see coverage.
|
||||
try:
|
||||
runtime = read_runtime_status() or {}
|
||||
served = runtime.get("served_profiles")
|
||||
if isinstance(served, list) and served:
|
||||
payload["served_profiles"] = served
|
||||
except Exception:
|
||||
pass
|
||||
return payload
|
||||
|
||||
|
||||
def build_status_payload() -> dict[str, Any]:
|
||||
"""Default ``status`` answer — current runtime status, answered live."""
|
||||
from gateway.status import read_runtime_status
|
||||
|
||||
payload = read_runtime_status() or {}
|
||||
payload = dict(payload)
|
||||
payload["protocol"] = CONTROL_PROTOCOL_VERSION
|
||||
payload["answered_at"] = time.time()
|
||||
payload["answering_pid"] = os.getpid()
|
||||
return payload
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Server
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class GatewayControlServer:
|
||||
"""Gateway-owned control socket server (identify/status, v1).
|
||||
|
||||
Lifecycle is owned by the gateway process: ``start()`` after the PID-file
|
||||
claim (the point where this process becomes the authoritative gateway for
|
||||
its HERMES_HOME), ``stop()`` on shutdown. All failures are non-fatal —
|
||||
the gateway never refuses to serve messaging because its control socket
|
||||
couldn't bind; consumers simply fall back to the scan layer.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
home: Optional[Path] = None,
|
||||
*,
|
||||
verb_handlers: Optional[dict[str, Callable[[], dict[str, Any]]]] = None,
|
||||
) -> None:
|
||||
if home is None:
|
||||
from gateway.status import _get_process_hermes_home
|
||||
|
||||
home = _get_process_hermes_home()
|
||||
self._home = Path(home)
|
||||
self._server: Optional[asyncio.AbstractServer] = None
|
||||
self._pipe_server: Any = None # Windows proactor pipe server
|
||||
self._bind_path: Optional[Path] = None
|
||||
self._pointer_file: Optional[Path] = None
|
||||
self._handlers: dict[str, Callable[[], dict[str, Any]]] = {
|
||||
"identify": build_identify_payload,
|
||||
"status": build_status_payload,
|
||||
}
|
||||
if verb_handlers:
|
||||
self._handlers.update(verb_handlers)
|
||||
|
||||
# -- lifecycle ---------------------------------------------------------
|
||||
|
||||
async def start(self) -> bool:
|
||||
"""Bind and start serving. Returns True on success, False otherwise."""
|
||||
try:
|
||||
if _IS_WINDOWS:
|
||||
return await self._start_windows()
|
||||
return await self._start_posix()
|
||||
except Exception as exc:
|
||||
logger.warning("Gateway control socket failed to start (non-fatal): %s", exc)
|
||||
return False
|
||||
|
||||
async def _start_posix(self) -> bool:
|
||||
bind_path, pointer_file = resolve_server_socket_path(self._home)
|
||||
# Clear a stale socket left by a crashed predecessor. We only get
|
||||
# here after winning the PID-file O_EXCL race, so any existing file
|
||||
# is either stale or a plain collision — never a live sibling for
|
||||
# this HERMES_HOME.
|
||||
with contextlib.suppress(OSError):
|
||||
if bind_path.exists():
|
||||
bind_path.unlink()
|
||||
self._server = await asyncio.start_unix_server(
|
||||
self._handle_connection, path=str(bind_path)
|
||||
)
|
||||
with contextlib.suppress(OSError):
|
||||
os.chmod(bind_path, 0o600)
|
||||
self._bind_path = bind_path
|
||||
if pointer_file is not None:
|
||||
pointer_file.write_text(str(bind_path), encoding="utf-8")
|
||||
self._pointer_file = pointer_file
|
||||
logger.info("Gateway control socket listening at %s", bind_path)
|
||||
return True
|
||||
|
||||
async def _start_windows(self) -> bool:
|
||||
loop = asyncio.get_running_loop()
|
||||
start_serving_pipe = getattr(loop, "start_serving_pipe", None)
|
||||
if start_serving_pipe is None:
|
||||
logger.debug(
|
||||
"Event loop %s has no start_serving_pipe — control socket "
|
||||
"disabled (selector loop on Windows).",
|
||||
type(loop).__name__,
|
||||
)
|
||||
return False
|
||||
pipe_name = windows_pipe_name(self._home)
|
||||
|
||||
def _factory():
|
||||
return _PipeControlProtocol(self)
|
||||
|
||||
servers = await start_serving_pipe(_factory, pipe_name)
|
||||
self._pipe_server = servers[0] if servers else None
|
||||
logger.info("Gateway control pipe listening at %s", pipe_name)
|
||||
return self._pipe_server is not None
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Stop serving and remove the socket/pointer files."""
|
||||
if self._server is not None:
|
||||
self._server.close()
|
||||
with contextlib.suppress(Exception):
|
||||
await self._server.wait_closed()
|
||||
self._server = None
|
||||
if self._pipe_server is not None:
|
||||
with contextlib.suppress(Exception):
|
||||
self._pipe_server.close()
|
||||
self._pipe_server = None
|
||||
self.cleanup_files()
|
||||
|
||||
def cleanup_files(self) -> None:
|
||||
"""Best-effort removal of socket + pointer files (atexit-safe)."""
|
||||
if self._bind_path is not None:
|
||||
with contextlib.suppress(OSError):
|
||||
self._bind_path.unlink(missing_ok=True)
|
||||
if self._pointer_file is not None:
|
||||
with contextlib.suppress(OSError):
|
||||
self._pointer_file.unlink(missing_ok=True)
|
||||
|
||||
# -- request handling ----------------------------------------------------
|
||||
|
||||
def handle_request_line(self, raw: bytes) -> bytes:
|
||||
"""Process one JSON request line, return one JSON response line.
|
||||
|
||||
Shared by the POSIX stream handler and the Windows pipe protocol.
|
||||
Never raises.
|
||||
"""
|
||||
request_id: Any = None
|
||||
try:
|
||||
request = json.loads(raw.decode("utf-8"))
|
||||
if not isinstance(request, dict):
|
||||
raise ValueError("request must be a JSON object")
|
||||
request_id = request.get("id")
|
||||
verb = request.get("verb")
|
||||
handler = self._handlers.get(verb) if isinstance(verb, str) else None
|
||||
if handler is None:
|
||||
response: dict[str, Any] = {
|
||||
"ok": False,
|
||||
"error": f"unknown verb: {verb!r}",
|
||||
"protocol": CONTROL_PROTOCOL_VERSION,
|
||||
"supported_verbs": sorted(self._handlers),
|
||||
}
|
||||
else:
|
||||
response = {
|
||||
"ok": True,
|
||||
"protocol": CONTROL_PROTOCOL_VERSION,
|
||||
"result": handler(),
|
||||
}
|
||||
except Exception as exc:
|
||||
response = {
|
||||
"ok": False,
|
||||
"error": f"{type(exc).__name__}: {exc}",
|
||||
"protocol": CONTROL_PROTOCOL_VERSION,
|
||||
}
|
||||
if request_id is not None:
|
||||
response["id"] = request_id
|
||||
try:
|
||||
encoded = json.dumps(response, default=str).encode("utf-8")
|
||||
except Exception:
|
||||
encoded = b'{"ok": false, "error": "response serialization failed"}'
|
||||
if len(encoded) > _MAX_RESPONSE_BYTES:
|
||||
encoded = b'{"ok": false, "error": "response too large"}'
|
||||
return encoded + b"\n"
|
||||
|
||||
async def _handle_connection(
|
||||
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
|
||||
) -> None:
|
||||
try:
|
||||
raw = await asyncio.wait_for(
|
||||
reader.readline(), timeout=_DEFAULT_CLIENT_TIMEOUT
|
||||
)
|
||||
if not raw or len(raw) > _MAX_REQUEST_BYTES:
|
||||
return
|
||||
writer.write(self.handle_request_line(raw.rstrip(b"\n")))
|
||||
await writer.drain()
|
||||
except (asyncio.TimeoutError, ConnectionError, OSError):
|
||||
pass
|
||||
except Exception:
|
||||
logger.debug("Control socket connection handler error", exc_info=True)
|
||||
finally:
|
||||
with contextlib.suppress(Exception):
|
||||
writer.close()
|
||||
|
||||
|
||||
class _PipeControlProtocol(asyncio.Protocol):
|
||||
"""One-shot request/response protocol for the Windows named pipe."""
|
||||
|
||||
def __init__(self, server: GatewayControlServer) -> None:
|
||||
self._server = server
|
||||
self._transport: Any = None
|
||||
self._buffer = bytearray()
|
||||
|
||||
def connection_made(self, transport) -> None: # pragma: no cover - windows
|
||||
self._transport = transport
|
||||
|
||||
def data_received(self, data: bytes) -> None: # pragma: no cover - windows
|
||||
self._buffer.extend(data)
|
||||
if len(self._buffer) > _MAX_REQUEST_BYTES:
|
||||
self._transport.close()
|
||||
return
|
||||
if b"\n" in self._buffer:
|
||||
line, _, _ = bytes(self._buffer).partition(b"\n")
|
||||
try:
|
||||
self._transport.write(self._server.handle_request_line(line))
|
||||
finally:
|
||||
self._transport.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Client (synchronous — used by CLI/updater consumers)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def query_gateway_control(
|
||||
home: Path,
|
||||
verb: str,
|
||||
*,
|
||||
timeout: float = _DEFAULT_CLIENT_TIMEOUT,
|
||||
) -> Optional[dict[str, Any]]:
|
||||
"""Ask the gateway serving ``home`` a control verb; None when unanswered.
|
||||
|
||||
Returns the verb's ``result`` payload on success. Any failure — no
|
||||
socket, stale socket nobody accepts on, timeout, malformed answer,
|
||||
``ok: false`` — returns None so callers fall back to the scan layer.
|
||||
Never raises.
|
||||
"""
|
||||
request = (
|
||||
json.dumps({"verb": verb, "id": 1, "protocol": CONTROL_PROTOCOL_VERSION})
|
||||
.encode("utf-8")
|
||||
+ b"\n"
|
||||
)
|
||||
try:
|
||||
if _IS_WINDOWS:
|
||||
raw = _query_windows_pipe(Path(home), request, timeout)
|
||||
else:
|
||||
raw = _query_unix_socket(Path(home), request, timeout)
|
||||
except Exception:
|
||||
return None
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
response = json.loads(raw.decode("utf-8"))
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
return None
|
||||
if not isinstance(response, dict) or response.get("ok") is not True:
|
||||
return None
|
||||
result = response.get("result")
|
||||
return result if isinstance(result, dict) else None
|
||||
|
||||
|
||||
def _query_unix_socket(home: Path, request: bytes, timeout: float) -> Optional[bytes]:
|
||||
path = resolve_client_socket_path(home)
|
||||
if path is None:
|
||||
return None
|
||||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(timeout)
|
||||
try:
|
||||
sock.connect(str(path))
|
||||
except (ConnectionRefusedError, FileNotFoundError, OSError):
|
||||
return None
|
||||
sock.sendall(request)
|
||||
chunks: list[bytes] = []
|
||||
deadline = time.monotonic() + timeout
|
||||
while time.monotonic() < deadline:
|
||||
try:
|
||||
chunk = sock.recv(65536)
|
||||
except socket.timeout:
|
||||
return None
|
||||
if not chunk:
|
||||
break
|
||||
chunks.append(chunk)
|
||||
if b"\n" in chunk:
|
||||
break
|
||||
if sum(len(c) for c in chunks) > _MAX_RESPONSE_BYTES:
|
||||
return None
|
||||
data = b"".join(chunks)
|
||||
line, _, _ = data.partition(b"\n")
|
||||
return line or None
|
||||
|
||||
|
||||
def _query_windows_pipe(
|
||||
home: Path, request: bytes, timeout: float
|
||||
) -> Optional[bytes]: # pragma: no cover - exercised on the wine2e lane
|
||||
pipe_name = windows_pipe_name(home)
|
||||
deadline = time.monotonic() + timeout
|
||||
handle = None
|
||||
while handle is None:
|
||||
try:
|
||||
handle = open(pipe_name, "r+b", buffering=0)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
except OSError:
|
||||
# Pipe busy (another client mid-handshake) — brief retry window.
|
||||
if time.monotonic() >= deadline:
|
||||
return None
|
||||
time.sleep(0.05)
|
||||
try:
|
||||
handle.write(request)
|
||||
chunks: list[bytes] = []
|
||||
while time.monotonic() < deadline:
|
||||
chunk = handle.read(65536)
|
||||
if not chunk:
|
||||
break
|
||||
chunks.append(chunk)
|
||||
if b"\n" in chunk:
|
||||
break
|
||||
if sum(len(c) for c in chunks) > _MAX_RESPONSE_BYTES:
|
||||
return None
|
||||
data = b"".join(chunks)
|
||||
line, _, _ = data.partition(b"\n")
|
||||
return line or None
|
||||
finally:
|
||||
with contextlib.suppress(Exception):
|
||||
handle.close()
|
||||
|
||||
|
||||
def identify_gateway(home: Path, *, timeout: float = _DEFAULT_CLIENT_TIMEOUT) -> Optional[dict[str, Any]]:
|
||||
"""Convenience wrapper: ``identify`` the gateway serving ``home``."""
|
||||
return query_gateway_control(home, "identify", timeout=timeout)
|
||||
@@ -30878,6 +30878,26 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
|
||||
atexit.register(remove_pid_file)
|
||||
atexit.register(release_gateway_runtime_lock)
|
||||
|
||||
# Control socket (#92091 step 1) — the gateway-owned identify/status
|
||||
# surface. Started immediately after the PID-file claim: winning that
|
||||
# O_EXCL race is the moment this process becomes the authoritative
|
||||
# gateway for its HERMES_HOME, so from here on "does a socket answer?"
|
||||
# is a truthful liveness/identity query for updater and fleet consumers.
|
||||
# Strictly non-fatal: a bind failure only means consumers fall back to
|
||||
# the process-scan/state-file layer, exactly as before this feature.
|
||||
_control_server = None
|
||||
try:
|
||||
from gateway.control_socket import GatewayControlServer
|
||||
|
||||
_control_server = GatewayControlServer()
|
||||
if not await _control_server.start():
|
||||
_control_server = None
|
||||
else:
|
||||
atexit.register(_control_server.cleanup_files)
|
||||
except Exception as _cs_exc:
|
||||
logger.debug("Control socket startup failed (non-fatal): %s", _cs_exc)
|
||||
_control_server = None
|
||||
|
||||
# Lifecycle ledger (NS-608): report if the previous gateway life died
|
||||
# uncleanly (SIGKILL / OOM / VM death — no exit path ran), then claim
|
||||
# the sentinel for this life. Placed after the PID-file/lock claim so
|
||||
@@ -31075,6 +31095,17 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
|
||||
# Wait for shutdown
|
||||
await runner.wait_for_shutdown()
|
||||
|
||||
# Stop the control socket first: once shutdown begins this process is no
|
||||
# longer a truthful "the gateway is serving here" answer, and a successor
|
||||
# (--replace / supervisor respawn) must be able to bind. Early-exit paths
|
||||
# above don't reach this; their process exit runs the atexit
|
||||
# cleanup_files hook, and a successor clears any stale socket on bind.
|
||||
if _control_server is not None:
|
||||
try:
|
||||
await _control_server.stop()
|
||||
except Exception:
|
||||
logger.debug("Control socket stop failed (non-fatal)", exc_info=True)
|
||||
|
||||
try:
|
||||
from hermes_cli.nous_auth_keepalive import stop_nous_auth_keepalive
|
||||
|
||||
|
||||
@@ -183,6 +183,42 @@ def collect_runtime_inventory() -> UpdatePlan:
|
||||
from gateway.status import _pid_exists, read_runtime_status
|
||||
|
||||
for profile, home in profile_homes:
|
||||
# Prefer the gateway-owned control socket (#92091): identity
|
||||
# declared by the process itself, including its own supervisor
|
||||
# provenance — no argv/PID inference. Scan fallback below.
|
||||
identity = None
|
||||
try:
|
||||
from gateway.control_socket import identify_gateway
|
||||
|
||||
identity = identify_gateway(home)
|
||||
except Exception:
|
||||
identity = None
|
||||
if identity:
|
||||
try:
|
||||
sock_pid = int(identity.get("pid"))
|
||||
except (TypeError, ValueError):
|
||||
sock_pid = None
|
||||
if sock_pid is not None:
|
||||
seen_pids.add(sock_pid)
|
||||
declared = identity.get("supervisor")
|
||||
supervisor = (
|
||||
str(declared)
|
||||
if declared
|
||||
else _detect_supervisor_for_pid(sock_pid, service_pids)
|
||||
)
|
||||
sock_sha = identity.get("code_sha")
|
||||
plan.runtimes.append(
|
||||
RuntimeRecord(
|
||||
kind="gateway",
|
||||
profile=profile,
|
||||
pid=sock_pid,
|
||||
supervisor=supervisor,
|
||||
code_sha=str(sock_sha) if sock_sha else None,
|
||||
code_version=identity.get("code_version"),
|
||||
restart_via=_restart_mechanism(supervisor, profile),
|
||||
)
|
||||
)
|
||||
continue
|
||||
record = read_runtime_status(home / "gateway_state.json")
|
||||
pid: Optional[int] = None
|
||||
code_sha = code_version = None
|
||||
|
||||
@@ -319,6 +319,41 @@ def collect_fleet_versions() -> list[dict[str, Any]]:
|
||||
homes.append((entry.name, entry))
|
||||
|
||||
for profile, home in homes:
|
||||
# Prefer the gateway-owned control socket (#92091): a live
|
||||
# `identify` answer is authoritative — no PID-reuse or stale-file
|
||||
# heuristics. Fall back to gateway_state.json for gateways that
|
||||
# predate the socket or whose socket didn't bind.
|
||||
identity = None
|
||||
try:
|
||||
from gateway.control_socket import identify_gateway
|
||||
|
||||
identity = identify_gateway(home)
|
||||
except Exception:
|
||||
identity = None
|
||||
if identity:
|
||||
try:
|
||||
pid = int(identity.get("pid"))
|
||||
except (TypeError, ValueError):
|
||||
pid = None
|
||||
if pid is not None:
|
||||
code_sha = identity.get("code_sha")
|
||||
if not code_sha or not expected_sha:
|
||||
state = "unknown"
|
||||
elif str(code_sha) == str(expected_sha):
|
||||
state = "current"
|
||||
else:
|
||||
state = "stale"
|
||||
results.append(
|
||||
{
|
||||
"profile": profile,
|
||||
"pid": pid,
|
||||
"code_sha": str(code_sha) if code_sha else None,
|
||||
"code_version": identity.get("code_version"),
|
||||
"state": state,
|
||||
"source": "socket",
|
||||
}
|
||||
)
|
||||
continue
|
||||
status_path = home / "gateway_state.json"
|
||||
record = read_runtime_status(status_path)
|
||||
if not record:
|
||||
|
||||
@@ -0,0 +1,351 @@
|
||||
"""Tests for the gateway control socket (#92091 migration step 1)."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import socket
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.control_socket import (
|
||||
CONTROL_PROTOCOL_VERSION,
|
||||
GatewayControlServer,
|
||||
identify_gateway,
|
||||
query_gateway_control,
|
||||
resolve_client_socket_path,
|
||||
resolve_server_socket_path,
|
||||
windows_pipe_name,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
sys.platform == "win32",
|
||||
reason="Unix-socket transport; the named-pipe half is covered on the wine2e lane",
|
||||
)
|
||||
|
||||
|
||||
def _run(coro):
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def home(tmp_path: Path) -> Path:
|
||||
d = tmp_path / "home" / ".hermes"
|
||||
d.mkdir(parents=True)
|
||||
return d
|
||||
|
||||
|
||||
def _serve(home: Path, handlers=None):
|
||||
"""Context helper: start a server in a fresh loop, yield inside coro."""
|
||||
return GatewayControlServer(home, verb_handlers=handlers)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Path resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_short_home_binds_in_home(home: Path):
|
||||
bind, pointer = resolve_server_socket_path(home)
|
||||
assert bind == home / "gateway.sock"
|
||||
assert pointer is None
|
||||
|
||||
|
||||
def test_long_home_uses_pointer_fallback(tmp_path: Path):
|
||||
deep = tmp_path / ("x" * 120) / ".hermes"
|
||||
deep.mkdir(parents=True)
|
||||
bind, pointer = resolve_server_socket_path(deep)
|
||||
assert bind != deep / "gateway.sock"
|
||||
assert len(str(bind).encode()) <= 100
|
||||
assert pointer == deep / "gateway.sock.path"
|
||||
|
||||
|
||||
def test_client_resolution_prefers_direct_then_pointer(home: Path, tmp_path: Path):
|
||||
assert resolve_client_socket_path(home) is None
|
||||
# pointer file to an existing socket-ish file
|
||||
target = tmp_path / "elsewhere.sock"
|
||||
target.touch()
|
||||
(home / "gateway.sock.path").write_text(str(target))
|
||||
assert resolve_client_socket_path(home) == target
|
||||
# direct file wins over pointer
|
||||
direct = home / "gateway.sock"
|
||||
direct.touch()
|
||||
assert resolve_client_socket_path(home) == direct
|
||||
|
||||
|
||||
def test_windows_pipe_name_is_stable_and_home_scoped(tmp_path: Path):
|
||||
a = windows_pipe_name(tmp_path / "a")
|
||||
b = windows_pipe_name(tmp_path / "b")
|
||||
assert a.startswith(r"\\.\pipe\hermes-gateway-")
|
||||
assert a != b
|
||||
assert a == windows_pipe_name(tmp_path / "a")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Server lifecycle + verbs (real sockets, real event loop)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_server_answers_identify_and_status(home: Path):
|
||||
async def scenario():
|
||||
server = GatewayControlServer(
|
||||
home,
|
||||
verb_handlers={
|
||||
"identify": lambda: {"pid": 4242, "code_sha": "abc123", "protocol": 1},
|
||||
"status": lambda: {"gateway_state": "running"},
|
||||
},
|
||||
)
|
||||
assert await server.start()
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
ident = await loop.run_in_executor(
|
||||
None, lambda: query_gateway_control(home, "identify")
|
||||
)
|
||||
status = await loop.run_in_executor(
|
||||
None, lambda: query_gateway_control(home, "status")
|
||||
)
|
||||
return ident, status
|
||||
finally:
|
||||
await server.stop()
|
||||
|
||||
ident, status = _run(scenario())
|
||||
assert ident == {"pid": 4242, "code_sha": "abc123", "protocol": 1}
|
||||
assert status == {"gateway_state": "running"}
|
||||
|
||||
|
||||
def test_unknown_verb_and_malformed_request(home: Path):
|
||||
async def scenario():
|
||||
server = GatewayControlServer(
|
||||
home, verb_handlers={"identify": lambda: {"pid": 1}}
|
||||
)
|
||||
assert await server.start()
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
unknown = await loop.run_in_executor(
|
||||
None, lambda: query_gateway_control(home, "restart")
|
||||
)
|
||||
|
||||
def raw_garbage():
|
||||
path = resolve_client_socket_path(home)
|
||||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
|
||||
s.settimeout(2)
|
||||
s.connect(str(path))
|
||||
s.sendall(b"this is not json\n")
|
||||
return s.recv(65536)
|
||||
|
||||
garbage_reply = await loop.run_in_executor(None, raw_garbage)
|
||||
return unknown, garbage_reply
|
||||
finally:
|
||||
await server.stop()
|
||||
|
||||
unknown, garbage_reply = _run(scenario())
|
||||
# unknown verb → ok:false → client returns None (fallback signal)
|
||||
assert unknown is None
|
||||
payload = json.loads(garbage_reply.decode())
|
||||
assert payload["ok"] is False
|
||||
assert payload["protocol"] == CONTROL_PROTOCOL_VERSION
|
||||
|
||||
|
||||
def test_stop_removes_socket_and_pointer(home: Path):
|
||||
async def scenario():
|
||||
server = GatewayControlServer(
|
||||
home, verb_handlers={"identify": lambda: {"pid": 1}}
|
||||
)
|
||||
assert await server.start()
|
||||
bind, _ = resolve_server_socket_path(home)
|
||||
assert bind.exists()
|
||||
await server.stop()
|
||||
return bind
|
||||
|
||||
bind = _run(scenario())
|
||||
assert not bind.exists()
|
||||
assert resolve_client_socket_path(home) is None
|
||||
# queries after stop cleanly return None
|
||||
assert query_gateway_control(home, "identify") is None
|
||||
|
||||
|
||||
def test_stale_socket_file_is_replaced_on_bind(home: Path):
|
||||
(home / "gateway.sock").touch() # crashed predecessor's leftover
|
||||
|
||||
async def scenario():
|
||||
server = GatewayControlServer(
|
||||
home, verb_handlers={"identify": lambda: {"pid": 7}}
|
||||
)
|
||||
assert await server.start()
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
return await loop.run_in_executor(None, lambda: identify_gateway(home))
|
||||
finally:
|
||||
await server.stop()
|
||||
|
||||
assert _run(scenario()) == {"pid": 7}
|
||||
|
||||
|
||||
def test_long_home_end_to_end_via_pointer(tmp_path: Path):
|
||||
deep = tmp_path / ("p" * 120) / ".hermes"
|
||||
deep.mkdir(parents=True)
|
||||
|
||||
async def scenario():
|
||||
server = GatewayControlServer(
|
||||
deep, verb_handlers={"identify": lambda: {"pid": 9}}
|
||||
)
|
||||
assert await server.start()
|
||||
try:
|
||||
assert (deep / "gateway.sock.path").is_file()
|
||||
loop = asyncio.get_running_loop()
|
||||
return await loop.run_in_executor(None, lambda: identify_gateway(deep))
|
||||
finally:
|
||||
await server.stop()
|
||||
|
||||
assert _run(scenario()) == {"pid": 9}
|
||||
assert not (deep / "gateway.sock.path").exists()
|
||||
|
||||
|
||||
def test_no_socket_returns_none_fast(home: Path):
|
||||
assert identify_gateway(home) is None
|
||||
assert query_gateway_control(home, "status") is None
|
||||
|
||||
|
||||
def test_default_identify_payload_shape(home: Path, monkeypatch):
|
||||
"""The real identify handler carries the fleet-consumer contract fields."""
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
|
||||
async def scenario():
|
||||
server = GatewayControlServer(home) # default handlers
|
||||
assert await server.start()
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
return await loop.run_in_executor(None, lambda: identify_gateway(home))
|
||||
finally:
|
||||
await server.stop()
|
||||
|
||||
ident = _run(scenario())
|
||||
assert ident is not None
|
||||
assert ident["protocol"] == CONTROL_PROTOCOL_VERSION
|
||||
assert ident["pid"] == __import__("os").getpid()
|
||||
# contract keys exist even when values are None/absent-degradable
|
||||
for key in ("hermes_home", "supervisor", "kind", "start_time"):
|
||||
assert key in ident
|
||||
assert ident["supervisor"] in {
|
||||
"systemd",
|
||||
"launchd",
|
||||
"desktop",
|
||||
"external",
|
||||
"manual",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Consumer integration: fleet matrix + inventory prefer socket, fall back
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _fake_identity(pid: int, sha: str):
|
||||
return {
|
||||
"protocol": 1,
|
||||
"pid": pid,
|
||||
"code_sha": sha,
|
||||
"code_version": "9.9.9",
|
||||
"supervisor": "systemd",
|
||||
"kind": "hermes-gateway",
|
||||
}
|
||||
|
||||
|
||||
def test_collect_fleet_versions_prefers_socket(tmp_path: Path, monkeypatch):
|
||||
import hermes_cli.update_receipt as ur
|
||||
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.build_info.get_code_identity",
|
||||
lambda refresh=False: {"sha": "HEADSHA", "version": "1.0"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.profiles._get_default_hermes_home", lambda: home
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.profiles._get_profiles_root", lambda: tmp_path / "no-profiles"
|
||||
)
|
||||
# stale state file that would report a WRONG pid — socket must win
|
||||
(home / "gateway_state.json").write_text(
|
||||
json.dumps({"pid": 1, "code_sha": "stalefile", "kind": "hermes-gateway"})
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"gateway.control_socket.identify_gateway",
|
||||
lambda h, **kw: _fake_identity(31337, "HEADSHA"),
|
||||
)
|
||||
|
||||
fleet = ur.collect_fleet_versions()
|
||||
assert len(fleet) == 1
|
||||
entry = fleet[0]
|
||||
assert entry["pid"] == 31337
|
||||
assert entry["state"] == "current"
|
||||
assert entry["source"] == "socket"
|
||||
|
||||
|
||||
def test_collect_fleet_versions_falls_back_to_state_file(tmp_path: Path, monkeypatch):
|
||||
import os
|
||||
|
||||
import hermes_cli.update_receipt as ur
|
||||
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.build_info.get_code_identity",
|
||||
lambda refresh=False: {"sha": "HEADSHA", "version": "1.0"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.profiles._get_default_hermes_home", lambda: home
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.profiles._get_profiles_root", lambda: tmp_path / "no-profiles"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"gateway.control_socket.identify_gateway", lambda h, **kw: None
|
||||
)
|
||||
(home / "gateway_state.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"pid": os.getpid(), # a live pid so _pid_exists passes
|
||||
"code_sha": "OLDSHA",
|
||||
"kind": "hermes-gateway",
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
fleet = ur.collect_fleet_versions()
|
||||
assert len(fleet) == 1
|
||||
assert fleet[0]["pid"] == os.getpid()
|
||||
assert fleet[0]["state"] == "stale"
|
||||
assert "source" not in fleet[0]
|
||||
|
||||
|
||||
def test_runtime_inventory_prefers_socket_supervisor(tmp_path: Path, monkeypatch):
|
||||
import hermes_cli.update_inventory as ui
|
||||
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.profiles._get_default_hermes_home", lambda: home
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.profiles._get_profiles_root", lambda: tmp_path / "no-profiles"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.gateway._get_service_pids", lambda all_profiles=False: set()
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.gateway.find_profile_gateway_processes", lambda: []
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"gateway.control_socket.identify_gateway",
|
||||
lambda h, **kw: _fake_identity(555, "SHA555"),
|
||||
)
|
||||
|
||||
plan = ui.collect_runtime_inventory()
|
||||
gws = [r for r in plan.runtimes if r.kind == "gateway"]
|
||||
assert len(gws) == 1
|
||||
assert gws[0].pid == 555
|
||||
# supervisor comes from the gateway's own declaration, not a PID scan
|
||||
assert gws[0].supervisor == "systemd"
|
||||
assert gws[0].code_sha == "SHA555"
|
||||
@@ -93,7 +93,7 @@ The same inventory is embedded in every real update's receipt (`~/.hermes/logs/u
|
||||
|
||||
### Update receipts and the fleet version check
|
||||
|
||||
Every `hermes update` run writes a machine-readable receipt to `~/.hermes/logs/update_receipts/` (last 20 kept, `latest.json` always points at the most recent): the pre-update fleet plan, each step taken, anything skipped and why, the gateway restart outcome, and the final fleet version matrix. After the restart phase the updater compares each live gateway's running code against the freshly updated checkout and prints a per-profile matrix — a gateway still serving pre-update code is reported loudly with the exact restart command, and the update exits non-zero so automation never treats a mixed-version fleet as healthy.
|
||||
Every `hermes update` run writes a machine-readable receipt to `~/.hermes/logs/update_receipts/` (last 20 kept, `latest.json` always points at the most recent): the pre-update fleet plan, each step taken, anything skipped and why, the gateway restart outcome, and the final fleet version matrix. After the restart phase the updater compares each live gateway's running code against the freshly updated checkout and prints a per-profile matrix — a gateway still serving pre-update code is reported loudly with the exact restart command, and the update exits non-zero so automation never treats a mixed-version fleet as healthy. Both `--plan` and the fleet check ask each running gateway directly over its local control socket (`gateway.sock` in the profile's data directory, a named pipe on Windows) when available, so version and supervisor information comes from the gateway itself; gateways from older versions are still discovered through their state files as before.
|
||||
|
||||
### Full pre-update backup: `--backup`
|
||||
|
||||
|
||||
Reference in New Issue
Block a user