fix(runtime): recover orphaned gateway clients

This commit is contained in:
Xubin Ren
2026-08-17 20:56:10 +08:00
parent 92727c99a8
commit d6847b0107
3 changed files with 145 additions and 15 deletions
+7 -3
View File
@@ -433,10 +433,14 @@ class GatewayClientLease:
record = cast(dict[str, object], value)
pid = record.get("pid")
identity = record.get("identity")
if not isinstance(pid, int) or not self._process_is_running(pid):
stale.append(token)
continue
current_identity = self._process_identity(pid)
if (
not isinstance(pid, int)
or not self._process_is_running(pid)
or (identity is not None and identity != self._process_identity(pid))
identity is not None
and current_identity is not None
and identity != current_identity
):
stale.append(token)
for token in stale:
+60 -11
View File
@@ -15,7 +15,7 @@ from contextlib import suppress
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Generic, TypeVar, cast
from typing import Any, Generic, Literal, TypeVar, cast
from filelock import FileLock
@@ -162,7 +162,14 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
return ProcessResult(False, self._message("not_running"), status)
state = self._read_state()
if not self._record_matches_process(state, status.pid):
identity_match = self._process_identity_match(state, status.pid)
if identity_match == "unknown":
return ProcessResult(
False,
self._message("identity_unavailable"),
status,
)
if identity_match == "mismatch":
self._clear_state()
return ProcessResult(
False,
@@ -205,7 +212,8 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
)
assert state is not None
if not self._is_pid_running(pid) or not self._record_matches_process(state, pid):
identity_match = self._process_identity_match(state, pid)
if not self._is_pid_running(pid) or identity_match == "mismatch":
self._clear_state()
return ProcessStatus(
running=False,
@@ -224,7 +232,9 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
started_at=_as_str(state.get("started_at")),
port=_as_int(state.get("port")),
command=tuple(cast(list[str], command)) if isinstance(command, list) else (),
reason=reason or "running",
reason=reason or (
"identity_unavailable" if identity_match == "unknown" else "running"
),
)
def read_log_tail(self, *, tail: int = 200) -> list[str]:
@@ -389,18 +399,31 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
return None
def _record_matches_process(self, state: dict[str, Any] | None, pid: int) -> bool:
return self._process_identity_match(state, pid) == "match"
def _process_identity_match(
self,
state: dict[str, Any] | None,
pid: int,
) -> Literal["match", "mismatch", "unknown"]:
if not state:
return False
return "mismatch"
recorded = state.get("identity")
if recorded is None:
return True
return "match"
current = self._process_identity(pid)
if current is None:
return "unknown"
if recorded == current:
return True
return "match"
# Older POSIX state files stored only the process group id.
return isinstance(recorded, int) and isinstance(current, str) and current.startswith(
f"{recorded}:"
)
if (
isinstance(recorded, int)
and isinstance(current, str)
and current.startswith(f"{recorded}:")
):
return "match"
return "mismatch"
def _read_state(self) -> dict[str, Any] | None:
try:
@@ -457,7 +480,33 @@ def process_is_running(pid: int, *, platform_name: str | None = None) -> bool:
return True
except OSError:
return False
return True
return _posix_process_state(pid, platform_name=host_platform) != "Z"
def _posix_process_state(pid: int, *, platform_name: str) -> str | None:
"""Return the host process state when available; zombies are not live clients."""
if platform_name == "Linux":
try:
stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
except OSError:
return None
closing_paren = stat.rfind(")")
fields = stat[closing_paren + 2 :].split() if closing_paren >= 0 else []
return fields[0] if fields else None
if platform_name == "Darwin":
try:
result = subprocess.run(
["ps", "-o", "stat=", "-p", str(pid)],
check=False,
capture_output=True,
text=True,
timeout=1,
)
except (OSError, subprocess.SubprocessError):
return None
value = getattr(result, "stdout", "").strip()
return value[:1].upper() or None
return None
def _utc_now() -> str:
+78 -1
View File
@@ -486,6 +486,26 @@ def test_lease_snapshot_prunes_a_reused_client_pid(tmp_path, monkeypatch):
}
def test_lease_snapshot_keeps_a_client_when_identity_probe_is_unavailable(
tmp_path,
monkeypatch,
):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
identity: str | None = "created-at"
monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
monkeypatch.setattr(runtime, "_process_identity", lambda _pid: identity)
client = GatewayClientLease(runtime, kind="tui", pid=12345, token="client")
client.acquire()
client.mark_ephemeral()
identity = None
snapshot = client.snapshot()
assert snapshot.auto_stop is True
assert snapshot.clients == 1
async def test_client_monitor_stops_an_orphaned_on_demand_gateway(tmp_path):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
lease = GatewayClientLease(runtime, kind="gateway-monitor")
@@ -588,6 +608,17 @@ def test_windows_host_probe_stays_safe_when_target_platform_is_posix(monkeypatch
assert process_is_running(54321, platform_name="Darwin") is False
def test_posix_process_probe_treats_a_zombie_as_stopped(monkeypatch):
monkeypatch.setattr("nanobot.process_runtime._platform_name", lambda: "Darwin")
monkeypatch.setattr("nanobot.process_runtime.os.kill", lambda *_args: None)
monkeypatch.setattr(
"nanobot.process_runtime.subprocess.run",
lambda *_args, **_kwargs: SimpleNamespace(stdout="Z+"),
)
assert process_is_running(12345, platform_name="Darwin") is False
def test_windows_host_identity_stays_safe_when_target_platform_is_posix(tmp_path, monkeypatch):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
monkeypatch.setattr("nanobot.process_runtime._platform_name", lambda: "Windows")
@@ -631,6 +662,52 @@ def test_status_clears_state_when_pid_identity_changes(tmp_path, monkeypatch):
assert not runtime.paths.state_path.exists()
def test_status_keeps_live_state_when_identity_probe_is_temporarily_unavailable(
tmp_path,
monkeypatch,
):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
runtime.paths.run_dir.mkdir(parents=True)
runtime.paths.state_path.write_text(
'{"pid": 12345, "identity": "created-at"}',
encoding="utf-8",
)
monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
monkeypatch.setattr(runtime, "_process_identity", lambda _pid: None)
status = runtime.status()
assert status.running is True
assert status.reason == "identity_unavailable"
assert runtime.paths.state_path.exists()
def test_stop_refuses_to_signal_a_process_when_identity_cannot_be_verified(
tmp_path,
monkeypatch,
):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
runtime.paths.run_dir.mkdir(parents=True)
runtime.paths.state_path.write_text(
'{"pid": 12345, "identity": "created-at"}',
encoding="utf-8",
)
monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
monkeypatch.setattr(runtime, "_process_identity", lambda _pid: None)
monkeypatch.setattr(
runtime,
"_terminate",
lambda *_args, **_kwargs: pytest.fail("an unverified PID must not be signalled"),
)
result = runtime.stop()
assert result.ok is False
assert result.message == "gateway_identity_unavailable"
assert result.status.running is True
assert runtime.paths.state_path.exists()
def test_posix_process_identity_includes_start_time_and_accepts_legacy_state(
tmp_path,
monkeypatch,
@@ -700,7 +777,7 @@ def test_stop_succeeds_when_process_exits_at_timeout_boundary(tmp_path, monkeypa
statuses = iter([running, stopped])
monkeypatch.setattr(runtime, "status", lambda **_kwargs: next(statuses))
monkeypatch.setattr(runtime, "_read_state", lambda: {"pid": 12345, "identity": 12345})
monkeypatch.setattr(runtime, "_record_matches_process", lambda *_args: True)
monkeypatch.setattr(runtime, "_process_identity_match", lambda *_args: "match")
monkeypatch.setattr(runtime, "_terminate", lambda *_args, **_kwargs: False)
result = runtime.stop(timeout_s=0)