diff --git a/gateway/shutdown_watchdog.py b/gateway/shutdown_watchdog.py index d83c5ec039..3dbeff7cf0 100644 --- a/gateway/shutdown_watchdog.py +++ b/gateway/shutdown_watchdog.py @@ -526,33 +526,34 @@ async def loop_heartbeat_forever( # disables the witness, and the payload flag tells probes that staleness is # no longer sufficient authority to escalate. # - # Windows (non-POSIX generally): the server creation below is explicitly - # gated to POSIX — asyncio AF_UNIX support is POSIX-only, and an ungated - # call would raise AttributeError on every native-Windows gateway start. - # The witness is therefore DELIBERATELY left absent there (debug log only, - # no warning): the payload records loop_tick_socket=False and every - # stale-file probe classifies UNKNOWN, never WEDGED. That is deliberate - # fail-safe: a wedged native Windows gateway keeps the graceful-drain - # backstop instead of an escalation verdict built on a witness that cannot - # exist. (WSL2 — the #90502 incident environment — is Linux and arms the - # socket normally.) + # Windows (non-POSIX generally): asyncio AF_UNIX support is POSIX-only, so + # the AF_UNIX arm below is gated to POSIX — an ungated call raised + # AttributeError on every native-Windows gateway start (#96956). Instead of + # leaving the witness permanently absent there, the non-POSIX arm binds a + # TCP loopback server on 127.0.0.1 with an OS-assigned port and publishes + # the port in the heartbeat payload (``loop_tick_tcp_port``) so probes know + # where to connect. Same protocol, same loop-owned semantics. If that bind + # fails, the payload records loop_tick_socket=False and probes classify + # UNKNOWN, never WEDGED — the graceful-drain backstop stays in place. (WSL2 + # — the #90502 incident environment — is Linux and arms the socket.) tick_server = None tick_socket_path = None + tick_tcp_port = None try: - tick_socket_path = get_loop_tick_socket_path(home) - tick_socket_path.parent.mkdir(parents=True, exist_ok=True) - # Re-bind over a leftover node from a dead process (os._exit(75) / - # SIGKILL skip the finally-unlink; PID reuse re-lands on this - # PID-suffixed path) is handled by asyncio itself: - # create_unix_server os.remove()s an existing socket node before - # binding — guarded by test_producer_rebinds_over_stale_socket_node. - # What asyncio does NOT do is clean up SIBLING nodes from other - # dead PIDs, so sweep those to keep state/ from accumulating - # gateway.loop-tick.*.sock nodes across crash-restart cycles. - # POSIX-only: os.kill(pid, 0) is a liveness probe here, but on - # Windows os.kill calls TerminateProcess for non-CTRL signals — - # and AF_UNIX server nodes are never created there anyway. if os.name == "posix": + tick_socket_path = get_loop_tick_socket_path(home) + tick_socket_path.parent.mkdir(parents=True, exist_ok=True) + # Re-bind over a leftover node from a dead process (os._exit(75) / + # SIGKILL skip the finally-unlink; PID reuse re-lands on this + # PID-suffixed path) is handled by asyncio itself: + # create_unix_server os.remove()s an existing socket node before + # binding — guarded by test_producer_rebinds_over_stale_socket_node. + # What asyncio does NOT do is clean up SIBLING nodes from other + # dead PIDs, so sweep those to keep state/ from accumulating + # gateway.loop-tick.*.sock nodes across crash-restart cycles. + # POSIX-only: os.kill(pid, 0) is a liveness probe here, but on + # Windows os.kill calls TerminateProcess for non-CTRL signals — + # and AF_UNIX server nodes are never created there anyway. try: for _stale in tick_socket_path.parent.glob( "gateway.loop-tick.*.sock" @@ -576,14 +577,29 @@ async def loop_heartbeat_forever( _tick_socket_handler, path=str(tick_socket_path) ) else: - logger.debug( - "loop-tick witness intentionally not armed on non-POSIX " - "platform (os.name=%r): asyncio AF_UNIX support is " - "POSIX-only; heartbeat payload records loop_tick_socket=False", - os.name, + # Windows / non-POSIX: no AF_UNIX support, so use a TCP loopback + # server on 127.0.0.1 as the loop-scheduling witness instead. + # Same protocol (connect → read one byte "1"), same semantics + # (pure in-memory, zero disk I/O, answered only when the loop + # is dispatching). Port is dynamic (assigned by the OS) and + # published via the heartbeat payload so external probes know + # where to connect. + tick_server = await asyncio.start_server( + _tick_socket_handler, host="127.0.0.1", port=0 ) + # Get the actual port assigned by the OS + _sock_addrs = tick_server.sockets if hasattr(tick_server, "sockets") else [] + for _s in _sock_addrs: + try: + _sname = _s.getsockname() + if isinstance(_sname, tuple) and len(_sname) >= 2: + tick_tcp_port = int(_sname[1]) + break + except Exception: + pass except Exception: tick_server = None + tick_tcp_port = None logger.warning( "Loop tick socket unavailable — liveness probes will have no " "loop-scheduling witness and will not escalate on a stale heartbeat", @@ -598,7 +614,10 @@ async def loop_heartbeat_forever( write_loop_heartbeat, start_time=start_time, home=home, - extra={"loop_tick_socket": tick_server is not None}, + extra={ + "loop_tick_socket": tick_server is not None, + "loop_tick_tcp_port": tick_tcp_port, + }, ) except asyncio.CancelledError: raise diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 59a7756bd7..4eff026b76 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -481,6 +481,46 @@ def _probe_loop_tick_socket( pass +def _probe_loop_tick_tcp( + port: int, + timeout: float = 1.0, +) -> bool | None: + """Ping the loop-scheduling witness via TCP loopback (Windows). + + Same protocol and semantics as the Unix socket variant: connect to + 127.0.0.1: and expect one byte "1" as proof the loop is + dispatching. Used on Windows / non-POSIX systems where AF_UNIX is not + available in asyncio. + + Returns: + True — the loop answered. + False — the port was reachable but did not answer, or refused. + None — invalid port / could not connect for unrelated reasons. + """ + try: + port_num = int(port) + if port_num <= 0 or port_num > 65535: + return None + except (TypeError, ValueError): + return None + sock = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(max(float(timeout), 0.0)) + sock.connect(("127.0.0.1", port_num)) + return sock.recv(1) == b"1" + except Exception: + # Connection refused, timeout, transient errors: witness exists + # but is silent (or the process is dead and the port is closed). + return False + finally: + if sock is not None: + try: + sock.close() + except Exception: + pass + + def _probe_loop_tick_socket_sustained( pid: int, home: Path | None, @@ -488,6 +528,7 @@ def _probe_loop_tick_socket_sustained( timeout: float = 1.0, strikes: int = 3, gap_s: float = 0.2, + tcp_port: int | None = None, ) -> bool | None: """Probe the tick socket until a reply or the sustained-miss budget. @@ -509,7 +550,10 @@ def _probe_loop_tick_socket_sustained( """ total = max(int(strikes), 0) for attempt in range(total): - result = _probe_loop_tick_socket(pid, home, timeout=timeout) + if tcp_port is not None: + result = _probe_loop_tick_tcp(tcp_port, timeout=timeout) + else: + result = _probe_loop_tick_socket(pid, home, timeout=timeout) if result is True: return True if result is None: @@ -579,14 +623,26 @@ def probe_gateway_loop_liveness( # up, or a stale file from a previous PID. Not evidence of a wedge. return GATEWAY_LOOP_UNKNOWN - witness = _probe_loop_tick_socket(pid, home, timeout=tick_timeout) + # Pick the right witness probe: TCP loopback (Windows / non-POSIX) + # takes priority if the producer published a port, otherwise fall back + # to the AF_UNIX socket (POSIX / legacy). + tcp_port = payload.get("loop_tick_tcp_port") + try: + tcp_port_int = int(tcp_port) if tcp_port is not None else None + except (TypeError, ValueError): + tcp_port_int = None + + if tcp_port_int is not None and tcp_port_int > 0: + witness = _probe_loop_tick_tcp(tcp_port_int, timeout=tick_timeout) + tick_armed = True + else: + witness = _probe_loop_tick_socket(pid, home, timeout=tick_timeout) + tick_armed = payload.get("loop_tick_socket", _LOOP_TICK_ABSENT) if witness is True: # The loop answered a ping — it is dispatching right now. A stale # heartbeat file is a stalled write or a saturated executor, not a # wedge (#90502). return GATEWAY_LOOP_ALIVE - - tick_armed = payload.get("loop_tick_socket", _LOOP_TICK_ABSENT) age = time.time() - mtime if age <= stale_budget: if witness is False: @@ -620,6 +676,7 @@ def probe_gateway_loop_liveness( timeout=tick_timeout, strikes=tick_strikes - 1, gap_s=tick_gap_s, + tcp_port=tcp_port_int, ) if sustained is False: # Both witnesses agree, sustained: the loop did not schedule for diff --git a/tests/gateway/test_loop_liveness_watchdog.py b/tests/gateway/test_loop_liveness_watchdog.py index ae07106b27..d763fbc461 100644 --- a/tests/gateway/test_loop_liveness_watchdog.py +++ b/tests/gateway/test_loop_liveness_watchdog.py @@ -3,8 +3,11 @@ from __future__ import annotations import asyncio +import json +import os import pathlib import inspect +import tempfile import threading import time from unittest.mock import MagicMock, patch @@ -403,3 +406,100 @@ def test_loop_scheduling_witness_is_served_by_the_loop_itself(): assert "await asyncio.start_unix_server(" in body, ( "the loop-scheduling witness socket is not armed by the loop task" ) + + +def test_windows_tcp_witness_arms_and_publishes_port(): + """On non-POSIX platforms the witness must arm over TCP loopback. + + ``asyncio.start_unix_server`` does not exist on Windows (no AF_UNIX + event-loop support), so the producer arm fell into the broad except and + recorded ``loop_tick_socket=False`` — every stale-file probe then + classified UNKNOWN forever, disabling the wedge interlock on Windows + entirely. The TCP loopback witness restores the same contract: armed by + the loop task (an awaited ``asyncio.start_server`` is structurally + loop-owned exactly like the Unix variant), answered only while the loop + dispatches, port published in the heartbeat payload. + """ + if os.name == "posix": + pytest.skip("TCP loopback witness is the non-POSIX arm") + + async def scenario() -> tuple[dict, bool]: + task = asyncio.create_task( + loop_heartbeat_forever(interval_s=1.0, home=tmp_home) + ) + try: + deadline = time.monotonic() + 5.0 + payload = None + while time.monotonic() < deadline: + hb = tmp_home.joinpath(*("state", "gateway.heartbeat")) + if hb.exists(): + try: + payload = json.loads(hb.read_text(encoding="utf-8")) + except Exception: + payload = None + if payload and payload.get("loop_tick_tcp_port"): + break + await asyncio.sleep(0.02) + assert payload is not None, "heartbeat never appeared" + assert payload.get("loop_tick_socket") is True, ( + "witness reported unarmed on a platform where the TCP arm " + "must work" + ) + port = int(payload["loop_tick_tcp_port"]) + assert 0 < port <= 65535, "published port out of range" + + # Probe from a worker thread so the blocking connect/recv never + # stalls the very loop we are witnessing (an external process + # probes from its own loop/thread — reproduce that shape). + from hermes_cli.gateway import _probe_loop_tick_tcp + + result_box: dict[str, object] = {} + + def _probe() -> None: + result_box["r"] = _probe_loop_tick_tcp(port, timeout=2.0) + + worker = threading.Thread(target=_probe) + worker.start() + while worker.is_alive(): + await asyncio.sleep(0.05) + worker.join() + return payload, bool(result_box.get("r") is True) + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + with tempfile.TemporaryDirectory(prefix="lw-tcp-") as raw: + tmp_home = pathlib.Path(raw) + payload, answered = asyncio.run(scenario()) + assert answered, ( + "the loop-tick TCP witness did not answer a probe while the loop " + "was dispatching — the two-witness interlock would misclassify " + "this gateway as UNKNOWN" + ) + + +def test_windows_tcp_witness_arms_on_loop_task_source_shape(): + """The TCP arm must be awaited by the loop task, never thread-owned. + + Structural companion to ``test_loop_scheduling_witness_is_served_by_the_ + loop_itself``: the same property that makes the Unix socket an honest + witness (a coroutine cannot run inside a thread) must hold for the TCP + loopback arm, or a wedged loop could keep answering pings and the + interlock would be void on Windows. + """ + src = pathlib.Path( + inspect.getsourcefile(loop_heartbeat_forever) or "" + ).read_text() + body = src[src.index("async def loop_heartbeat_forever("):] + body = body[: body.index("\ndef ") if "\ndef " in body else len(body)] + assert "await asyncio.start_server(" in body, ( + "the TCP loop-scheduling witness is not armed by the loop task" + ) + # The Unix arm must stay gated to POSIX-only code paths so the missing + # attribute can never raise on Windows again. + assert 'os.name == "posix"' in body, ( + "the AF_UNIX witness arm is not gated to POSIX platforms" + )