Files
hermes-agent/gateway/cgroup_cleanup.py
ethernet 2be53ecd7b fix(encoding): read kernel pseudo-files as plain utf-8
utf-8-sig exists to tolerate BOMs that Windows tooling adds to files
users edit. /proc and /sys files are generated by the Linux kernel, never
BOM'd and absent on Windows, so -sig there only muddies the read/write
policy. Switch every literal /proc/ and /sys/ read to utf-8 and teach
the footgun read rule that string literals starting with /proc/ or
/sys/ are exempt (user-edited files keep utf-8-sig).
2026-09-24 11:50:30 -04:00

67 lines
2.0 KiB
Python

"""SIGKILL any process left in this systemd unit's cgroup.
Runs as ``ExecStopPost=`` after the gateway's main process has exited: the
safety net for long-lived helpers the gateway doesn't track (``adb``, platform
bridges) that would otherwise be orphaned in the cgroup and block
``Restart=always``. Per-PID SIGKILLs over ``cgroup.procs`` are used instead of
writing ``1`` to ``cgroup.kill``: the kernel has returned ``EINVAL`` on the
cgroup-wide kill while per-PID signal delivery still works.
"""
from __future__ import annotations
import contextlib
import os
import re
import signal
import sys
from pathlib import Path
def _own_cgroup_path() -> str | None:
"""Return the cgroup v2 path for the calling process, or None."""
try:
text = Path("/proc/self/cgroup").read_text(encoding="utf-8")
except OSError:
return None
match = re.search(r"^0::(.+)$", text, re.MULTILINE)
return match.group(1).strip() if match else None
def _read_cgroup_pids(cgroup_path: str) -> list[int]:
try:
raw = Path(f"/sys/fs/cgroup{cgroup_path}/cgroup.procs").read_text(encoding="utf-8")
except OSError:
return []
pids: list[int] = []
for line in raw.splitlines():
with contextlib.suppress(ValueError):
pids.append(int(line.strip()))
return pids
def reap_cgroup(cgroup_path: str | None = None) -> int:
"""SIGKILL every PID in the cgroup other than the caller. Returns the count killed."""
cgroup_path = _own_cgroup_path() if cgroup_path is None else cgroup_path
if not cgroup_path:
return 0
killed = 0
for pid in _read_cgroup_pids(cgroup_path):
if pid == os.getpid():
continue
try:
os.kill(pid, signal.SIGKILL) # windows-footgun: ok — Linux-only (reads /proc, /sys/fs/cgroup; runs from a systemd unit)
killed += 1
except (ProcessLookupError, PermissionError):
continue
return killed
def main() -> int:
reap_cgroup()
return 0
if __name__ == "__main__":
sys.exit(main())