fix(gateway): stabilize process identities

This commit is contained in:
Xubin Ren
2026-08-18 03:21:00 +08:00
parent 2b4a04fb71
commit 19ad1adfe7
3 changed files with 240 additions and 35 deletions
+18 -6
View File
@@ -535,12 +535,7 @@ class GatewayClientLease:
if not isinstance(pid, int) or not self._process_is_running(pid): if not isinstance(pid, int) or not self._process_is_running(pid):
stale.append(token) stale.append(token)
continue continue
current_identity = self._process_identity(pid) if self._process_identity_match(identity, pid) == "mismatch":
if (
identity is not None
and current_identity is not None
and identity != current_identity
):
stale.append(token) stale.append(token)
for token in stale: for token in stale:
clients.pop(token, None) clients.pop(token, None)
@@ -551,6 +546,23 @@ class GatewayClientLease:
value = resolver(pid) if callable(resolver) else None value = resolver(pid) if callable(resolver) else None
return value if isinstance(value, (str, int)) else None return value if isinstance(value, (str, int)) else None
def _process_identity_match(
self,
recorded: object,
pid: int,
) -> Literal["match", "mismatch", "unknown"]:
matcher = getattr(self.runtime, "process_identity_match", None)
if callable(matcher):
result = matcher(recorded, pid)
if result in {"match", "mismatch", "unknown"}:
return cast(Literal["match", "mismatch", "unknown"], result)
if recorded is None:
return "match"
current = self._process_identity(pid)
if current is None:
return "unknown"
return "match" if recorded == current else "mismatch"
def _process_is_running(self, pid: int) -> bool: def _process_is_running(self, pid: int) -> bool:
checker = getattr(self.runtime, "process_is_running", None) checker = getattr(self.runtime, "process_is_running", None)
return bool(checker(pid)) if callable(checker) else process_is_running(pid) return bool(checker(pid)) if callable(checker) else process_is_running(pid)
+152 -29
View File
@@ -5,7 +5,9 @@ from __future__ import annotations
import ctypes import ctypes
import json import json
import os import os
import re
import signal import signal
import struct
import subprocess import subprocess
import sys import sys
import tempfile import tempfile
@@ -15,6 +17,7 @@ from contextlib import suppress
from ctypes import wintypes from ctypes import wintypes
from dataclasses import dataclass from dataclasses import dataclass
from datetime import UTC, datetime from datetime import UTC, datetime
from functools import lru_cache
from pathlib import Path from pathlib import Path
from typing import Any, Generic, Literal, TypeVar, cast from typing import Any, Generic, Literal, TypeVar, cast
@@ -270,6 +273,33 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
"""Return an identity that changes when an operating-system PID is reused.""" """Return an identity that changes when an operating-system PID is reused."""
return self._process_identity(pid) return self._process_identity(pid)
def process_identity_match(
self,
recorded: object,
pid: int,
) -> Literal["match", "mismatch", "unknown"]:
"""Compare a recorded identity with the current process safely."""
if recorded is None:
return "match"
current = self._process_identity(pid)
if current is None:
return "unknown"
if recorded == current:
return "match"
# Older POSIX state files stored only the process group id.
if (
isinstance(recorded, int)
and isinstance(current, str)
and (
current.startswith(f"{recorded}:")
or current.startswith(f"darwin:{recorded}:")
)
):
return "match"
if self.platform_name == "Darwin":
return _darwin_identity_match(recorded, current)
return "mismatch"
def process_is_running(self, pid: int) -> bool: def process_is_running(self, pid: int) -> bool:
"""Return whether the recorded operating-system process is still live.""" """Return whether the recorded operating-system process is still live."""
return self._is_pid_running(pid) return self._is_pid_running(pid)
@@ -365,8 +395,18 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
# Process inspection must follow the host API even when tests inject a # Process inspection must follow the host API even when tests inject a
# target platform. On Windows, falling through to POSIX calls is not # target platform. On Windows, falling through to POSIX calls is not
# merely unsupported: ``os.kill(pid, 0)`` broadcasts CTRL_C_EVENT. # merely unsupported: ``os.kill(pid, 0)`` broadcasts CTRL_C_EVENT.
if _platform_name() == "Windows" or self.platform_name == "Windows": host_platform = _platform_name()
if host_platform == "Windows" or self.platform_name == "Windows":
return _windows_process_identity(pid) return _windows_process_identity(pid)
if self.platform_name == "Darwin":
birth = _darwin_process_birth(pid)
if birth is None:
return None
process_group, started_at_seconds, started_at_microseconds = birth
return (
f"darwin:{process_group}:{started_at_seconds}:"
f"{started_at_microseconds}"
)
try: try:
process_group = os.getpgid(pid) process_group = os.getpgid(pid)
except OSError: except OSError:
@@ -384,19 +424,6 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
fields = stat[closing_paren + 2 :].split() if closing_paren >= 0 else [] fields = stat[closing_paren + 2 :].split() if closing_paren >= 0 else []
# /proc/<pid>/stat fields after comm begin at field 3; starttime is field 22. # /proc/<pid>/stat fields after comm begin at field 3; starttime is field 22.
return fields[19] if len(fields) > 19 else None return fields[19] if len(fields) > 19 else None
if self.platform_name == "Darwin":
try:
result = self._subprocess_run(
["ps", "-o", "lstart=", "-p", str(pid)],
check=False,
capture_output=True,
text=True,
timeout=1,
)
except (OSError, subprocess.SubprocessError):
return None
started_at = getattr(result, "stdout", "").strip()
return started_at or None
return None return None
def _record_matches_process(self, state: dict[str, Any] | None, pid: int) -> bool: def _record_matches_process(self, state: dict[str, Any] | None, pid: int) -> bool:
@@ -410,21 +437,7 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
if not state: if not state:
return "mismatch" return "mismatch"
recorded = state.get("identity") recorded = state.get("identity")
if recorded is None: return self.process_identity_match(recorded, pid)
return "match"
current = self._process_identity(pid)
if current is None:
return "unknown"
if recorded == current:
return "match"
# Older POSIX state files stored only the process group id.
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: def _read_state(self) -> dict[str, Any] | None:
try: try:
@@ -529,6 +542,116 @@ def _as_str(value: object) -> str | None:
return value if isinstance(value, str) else None return value if isinstance(value, str) else None
def _darwin_identity_match(
recorded: object,
current: object,
) -> Literal["match", "mismatch", "unknown"]:
"""Compare the new numeric identity with a pre-upgrade ``ps`` identity."""
if not isinstance(recorded, str) or not isinstance(current, str):
return "mismatch"
current_match = re.fullmatch(r"darwin:(\d+):(\d+):(\d+)", current)
if current_match is None:
return "mismatch"
recorded_group, separator, recorded_started_at = recorded.partition(":")
if not separator or not recorded_group.isdigit():
return "mismatch"
if int(recorded_group) != int(current_match.group(1)):
return "mismatch"
legacy_epoch = _legacy_darwin_started_at(recorded_started_at)
if legacy_epoch is None:
# The PID is alive and its process group still matches, but an older
# locale produced a date we cannot safely parse. Keep the record until
# the owning client exits instead of killing a live gateway.
return "unknown"
return "match" if legacy_epoch == int(current_match.group(2)) else "mismatch"
def _legacy_darwin_started_at(value: str) -> int | None:
"""Parse the English and numeric macOS ``ps lstart`` formats we released."""
english = re.fullmatch(
r"[A-Za-z]{3}\s+([A-Za-z]{3})\s+(\d{1,2})\s+"
r"(\d{2}):(\d{2}):(\d{2})\s+(\d{4})",
value.strip(),
)
months = {
"Jan": 1,
"Feb": 2,
"Mar": 3,
"Apr": 4,
"May": 5,
"Jun": 6,
"Jul": 7,
"Aug": 8,
"Sep": 9,
"Oct": 10,
"Nov": 11,
"Dec": 12,
}
if english is not None:
month = months.get(english.group(1))
if month is None:
return None
day, hour, minute, second, year = map(int, english.groups()[1:])
else:
numeric = re.fullmatch(
r"\S+\s+(\d{1,2})/(\d{1,2})\s+"
r"(\d{2}):(\d{2}):(\d{2})\s+(\d{4})",
value.strip(),
)
if numeric is None:
return None
month, day, hour, minute, second, year = map(int, numeric.groups())
try:
return int(time.mktime((year, month, day, hour, minute, second, -1, -1, -1)))
except (OverflowError, ValueError):
return None
@lru_cache(maxsize=1)
def _darwin_proc_pidinfo() -> Any | None:
if sys.platform != "darwin":
return None
try:
proc_pidinfo = ctypes.CDLL(
"/usr/lib/libproc.dylib",
use_errno=True,
).proc_pidinfo
except (AttributeError, OSError):
return None
proc_pidinfo.argtypes = [
ctypes.c_int,
ctypes.c_int,
ctypes.c_uint64,
ctypes.c_void_p,
ctypes.c_int,
]
proc_pidinfo.restype = ctypes.c_int
return proc_pidinfo
def _darwin_process_birth(pid: int) -> tuple[int, int, int] | None:
"""Read PGID and microsecond process birth time from ``proc_bsdinfo``."""
proc_pidinfo = _darwin_proc_pidinfo()
if proc_pidinfo is None:
return None
# ``proc_bsdinfo`` is 136 bytes on supported macOS versions. These stable
# field offsets come from ``sys/proc_info.h``: pid=12, pgid=100,
# start_tvsec=120, and start_tvusec=128.
buffer = ctypes.create_string_buffer(136)
try:
written = proc_pidinfo(pid, 3, 0, buffer, len(buffer))
except (OSError, ValueError):
return None
if written != len(buffer) or struct.unpack_from("=I", buffer, 12)[0] != pid:
return None
process_group = struct.unpack_from("=I", buffer, 100)[0]
started_at_seconds = struct.unpack_from("=Q", buffer, 120)[0]
started_at_microseconds = struct.unpack_from("=Q", buffer, 128)[0]
if started_at_seconds <= 0:
return None
return process_group, started_at_seconds, started_at_microseconds
def _windows_process_identity(pid: int) -> str | None: def _windows_process_identity(pid: int) -> str | None:
if os.name != "nt": if os.name != "nt":
return None return None
+70
View File
@@ -642,6 +642,24 @@ def test_lease_snapshot_prunes_a_reused_client_pid(tmp_path, monkeypatch):
} }
def test_lease_snapshot_keeps_a_legacy_localized_darwin_client(tmp_path, monkeypatch):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
started_at = int(time.mktime((2026, 8, 18, 2, 17, 54, -1, -1, -1)))
identity = "42:二 8/18 02:17:54 2026"
monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
monkeypatch.setattr(runtime, "_process_identity", lambda _pid: identity)
client = GatewayClientLease(runtime, kind="webui", pid=12345, token="client")
client.acquire()
client.mark_ephemeral()
identity = f"darwin:42:{started_at}:123456"
snapshot = client.snapshot()
assert snapshot.auto_stop is True
assert snapshot.clients == 1
def test_lease_snapshot_keeps_a_client_when_identity_probe_is_unavailable( def test_lease_snapshot_keeps_a_client_when_identity_probe_is_unavailable(
tmp_path, tmp_path,
monkeypatch, monkeypatch,
@@ -791,6 +809,26 @@ def test_windows_host_identity_stays_safe_when_target_platform_is_posix(tmp_path
assert runtime.process_identity(12345) == "created-at" assert runtime.process_identity(12345) == "created-at"
def test_windows_lease_prunes_a_reused_pid_by_creation_time(tmp_path, monkeypatch):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Windows")
identity = "filetime:first-process"
monkeypatch.setattr("nanobot.process_runtime._platform_name", lambda: "Windows")
monkeypatch.setattr(
"nanobot.process_runtime._windows_process_identity",
lambda _pid: identity,
)
client = GatewayClientLease(runtime, kind="tui", pid=12345, token="client")
client.acquire()
client.mark_ephemeral()
identity = "filetime:replacement-process"
snapshot = client.snapshot()
assert snapshot.auto_stop is True
assert snapshot.clients == 0
def test_status_clears_stale_state(tmp_path, monkeypatch): def test_status_clears_stale_state(tmp_path, monkeypatch):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux") runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
runtime.paths.run_dir.mkdir(parents=True) runtime.paths.run_dir.mkdir(parents=True)
@@ -877,6 +915,38 @@ def test_posix_process_identity_includes_start_time_and_accepts_legacy_state(
assert runtime._record_matches_process({"identity": 42}, 12345) is True assert runtime._record_matches_process({"identity": 42}, 12345) is True
def test_darwin_process_identity_is_locale_independent(tmp_path, monkeypatch):
runtime = GatewayRuntime(
paths=_paths(tmp_path),
platform_name="Darwin",
subprocess_run=lambda *_args, **_kwargs: pytest.fail(
"Darwin identities must not depend on localized subprocess output"
),
)
monkeypatch.setenv("LANG", "zh_CN.UTF-8")
monkeypatch.setenv("LC_ALL", "zh_CN.UTF-8")
started_at = int(time.mktime((2026, 8, 18, 2, 17, 54, -1, -1, -1)))
monkeypatch.setattr(
"nanobot.process_runtime._darwin_process_birth",
lambda _pid: (42, started_at, 123456),
)
assert runtime.process_identity(12345) == f"darwin:42:{started_at}:123456"
assert runtime._record_matches_process({"identity": 42}, 12345) is True
@pytest.mark.skipif(sys.platform != "darwin", reason="requires macOS proc_pidinfo")
def test_darwin_live_process_identity_is_stable(tmp_path):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
first = runtime.process_identity(os.getpid())
second = runtime.process_identity(os.getpid())
assert isinstance(first, str)
assert first.startswith("darwin:")
assert second == first
def test_stop_terminates_recorded_process(tmp_path, monkeypatch): def test_stop_terminates_recorded_process(tmp_path, monkeypatch):
runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux") runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
runtime.paths.run_dir.mkdir(parents=True) runtime.paths.run_dir.mkdir(parents=True)