From d6847b01070160aa74b468f75fdb972a81faeced Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:18:34 +0800 Subject: [PATCH] fix(runtime): recover orphaned gateway clients --- nanobot/gateway/runtime.py | 10 +++-- nanobot/process_runtime.py | 71 ++++++++++++++++++++++++++----- tests/gateway/test_runtime.py | 79 ++++++++++++++++++++++++++++++++++- 3 files changed, 145 insertions(+), 15 deletions(-) diff --git a/nanobot/gateway/runtime.py b/nanobot/gateway/runtime.py index f376c7d36..528d49da7 100644 --- a/nanobot/gateway/runtime.py +++ b/nanobot/gateway/runtime.py @@ -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: diff --git a/nanobot/process_runtime.py b/nanobot/process_runtime.py index b05514b12..d981eea65 100644 --- a/nanobot/process_runtime.py +++ b/nanobot/process_runtime.py @@ -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: diff --git a/tests/gateway/test_runtime.py b/tests/gateway/test_runtime.py index 847206ef8..127d0e63e 100644 --- a/tests/gateway/test_runtime.py +++ b/tests/gateway/test_runtime.py @@ -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)