fix(gateway): arm the loop-scheduling witness on Windows via TCP loopback

`asyncio.start_unix_server` does not exist on Windows (no AF_UNIX event-loop
support in asyncio), so arming the loop-tick witness in
`loop_heartbeat_forever` raised AttributeError on every native-Windows
gateway start. The broad except swallowed it and recorded
`loop_tick_socket=False`, so every stale-heartbeat probe classified the
gateway as UNKNOWN — never WEDGED, never ALIVE-with-stalled-write. The
two-witness interlock from a1c83ef9 (issue #90502 follow-up) has been
effectively disabled on Windows since it landed: a wedged native-Windows
gateway could never be detected, and an alive one could never be
distinguished from a stalled heartbeat write.

On non-POSIX platforms the witness now arms over a TCP loopback server on
127.0.0.1 (OS-assigned dynamic port) instead:

- same protocol — connect, read one byte "1"
- same semantics — pure in-memory, zero disk I/O, answered only while the
  loop is dispatching, armed by the loop task itself (an awaited
  `asyncio.start_server` is structurally loop-owned exactly like the Unix
  variant, so a wedged loop cannot keep answering pings)
- the assigned port is published in the heartbeat payload as
  `loop_tick_tcp_port`, and `probe_gateway_loop_liveness` prefers the TCP
  witness when the producer published a port, falling back to the AF_UNIX
  socket for POSIX/legacy producers

POSIX behavior is unchanged: the AF_UNIX arm (including the stale-node
sweep) stays gated behind `os.name == "posix"` so the missing attribute can
never raise on Windows again. Legacy heartbeats without `loop_tick_tcp_port`
keep the existing socket-node contract untouched.

Tested end-to-end on native Windows: witness arms, port is published,
`_probe_loop_tick_tcp` answers from an external thread while the loop
dispatches, and the existing loop-liveness suite passes unchanged (the
AF_UNIX structural test still passes — the Unix arm text is preserved
inside the POSIX branch).

Adds two tests pinning the new behavior: an E2E test that arms the TCP
witness and probes it (skipped on POSIX, where the Unix arm is the real
witness), and a structural test that the TCP arm stays awaited on the loop
task and the AF_UNIX arm stays POSIX-gated.

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

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:<port> 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

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"
+    )