fix(core): reject unreachable tcp endpoints during discovery

Author: Your Name

src/kagan/core/ipc/discovery.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import socket
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
@@ -82,6 +83,14 @@ def _is_process_alive(pid: int) -> bool:
     return pid_exists(pid)
 
 
+def _is_tcp_endpoint_reachable(address: str, port: int) -> bool:
+    try:
+        with socket.create_connection((address, port), timeout=0.25):
+            return True
+    except OSError:
+        return False
+
+
 def discover_core_endpoint() -> CoreEndpoint | None:
     """Discover a running Kagan core by reading runtime files.
 
@@ -113,6 +122,18 @@ def discover_core_endpoint() -> CoreEndpoint | None:
     if pid is not None and not _is_process_alive(pid):
         logger.info("Core process (PID %d) is no longer running; stale endpoint", pid)
         return None
+    if normalized_transport == "tcp":
+        raw_port = data.get("port")
+        if not isinstance(raw_port, int) or raw_port <= 0:
+            logger.warning("Malformed TCP endpoint file at %s", endpoint_path)
+            return None
+        if not _is_tcp_endpoint_reachable(str(address), raw_port):
+            logger.info(
+                "Core endpoint tcp://%s:%s is unreachable; treating runtime metadata as stale",
+                address,
+                raw_port,
+            )
+            return None
 
     token = _read_text(get_core_token_path())
 

tests/core/unit/test_core_ipc_discovery.py

@@ -11,17 +11,19 @@
     from pathlib import Path
 
 
-def _write_runtime_files(runtime_dir: Path, *, owner_pid: int) -> None:
+def _write_runtime_files(
+    runtime_dir: Path,
+    *,
+    owner_pid: int,
+    transport: str = "socket",
+    address: str = "/tmp/kagan-core.sock",
+    port: int | None = None,
+) -> None:
     runtime_dir.mkdir(parents=True, exist_ok=True)
-    (runtime_dir / "endpoint.json").write_text(
-        json.dumps(
-            {
-                "transport": "socket",
-                "address": "/tmp/kagan-core.sock",
-            }
-        ),
-        encoding="utf-8",
-    )
+    endpoint_payload: dict[str, object] = {"transport": transport, "address": address}
+    if port is not None:
+        endpoint_payload["port"] = port
+    (runtime_dir / "endpoint.json").write_text(json.dumps(endpoint_payload), encoding="utf-8")
     (runtime_dir / "token").write_text("lease-token", encoding="utf-8")
     (runtime_dir / "core.lease.json").write_text(
         json.dumps(
@@ -67,3 +69,22 @@ def test_discover_core_endpoint_rejects_stale_dead_lease(monkeypatch, tmp_path:
     endpoint = discover_core_endpoint()
 
     assert endpoint is None
+
+
+def test_discover_core_endpoint_rejects_unreachable_tcp_endpoint(
+    monkeypatch, tmp_path: Path
+) -> None:
+    runtime_dir = tmp_path / "core-runtime"
+    _write_runtime_files(
+        runtime_dir,
+        owner_pid=os.getpid(),
+        transport="tcp",
+        address="127.0.0.1",
+        port=54321,
+    )
+    monkeypatch.setenv("KAGAN_CORE_RUNTIME_DIR", str(runtime_dir))
+    monkeypatch.setattr("kagan.core.ipc.discovery._is_tcp_endpoint_reachable", lambda *_: False)
+
+    endpoint = discover_core_endpoint()
+
+    assert endpoint is None