diff --git a/CLAUDE.md b/CLAUDE.md index fb09e20..d739553 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,7 +32,8 @@ pixi run setup # One-time Pi setup: udev + wifi-powersave + systemd (ne pixi run udev # Install udev rules + dialout group (needs sudo) pixi run wifi-powersave # Disable WiFi power save via NetworkManager (needs sudo) pixi run setup-ids # Guided servo ID assignment tool -pixi run kill # Kill stale ROS processes and reset daemon +pixi run kill # Kill this checkout's ROS processes and reset daemon +pixi run sweep # Report ROS processes leaked by dead agent jobs (--kill to reap) pixi run identity # Fleet identity CLI: show / id / set --id --name --site pixi run tailnet # Join this machine to the Tailscale overlay (needs sudo) pixi run provision # Render cloud-init user-data for a clean Pi @@ -200,6 +201,8 @@ The C++ that runs *inside* other people's processes: a Nav2 plugin and two compo ### `mote_bringup` (Python/ament) Launch files, config, udev rules, NetworkManager drop-ins, systemd services, and the fleet foundation: `mote_home.py` (per-robot state root), `identity.py` (`identity` console script), `provision.py` + `provisioning/user-data.template` (`provision`), `dds_participants.py` (`dds_participants`), `twist_relay.py` (`twist_relay`, the Foxglove teleop seam), `explore.py` (`explore`, `pixi run explore` — autonomous mapping coverage: left-wall following with a Nav2 frontier-seek escape and a stuck detector for lidar-invisible obstacles like rug edges; run beside `pixi run mapping` **on the Pi**, so a wifi drop cannot end the mission; `--sim-time` gates it on /clock for the sim, and `--cruise`/`--obstacle`/`--desired-left`/`--follow-band` tighten the corridor-scale defaults for domestic layouts), the `foxglove/` layout, and `tailscale/install.sh`, plus `bundle.py` — the site bundle's *content*: a ROS-free reader/validator/packer for a map revision, shared with the fleet server's registry (M4) so both ends check a revision with the same code. See Fleet above. +**Stray ROS processes** are `sweep_orphans.py`'s subject, from both ends (`mote_bringup/README.md` "Clearing stray ROS processes"). *Cure* is `pixi run sweep`: agent worktrees leave nodes behind that reparent to init and outlive the job by days — 44 of them, 371 MB, were found on the dev box — and they are the exact process names a benchmark measures, against system-wide counters no `overhead.py`-style scoping can isolate. **Matching is on identity in /proc, never on the command line**, which is not fastidiousness: `pixi run kill`'s old `pkill -9 -f ''` matched the shell running the task (the names are in its own command line), SIGKILLed it, and so **never reached the `ros2 daemon` reset that followed it** — reproduced, exit 137 — besides matching every other checkout on the box. Both modes now require a ROS environment, a path under the directory in question, and absence from the sweeper's own ancestry; `kill` keeps that scope (`--scope .`) and drops the orphan/age tests, since a wedged stack is live by definition. The sweep adds them, because a `setsid`-detached run — the sim smoke test's — is indistinguishable from a leak by ancestry alone, so `--min-age` (30 min) is what separates them and reporting is the default. *Prevention* is `spawn_reapable`/`reap_group` in the same module: **`ros2 run` is a wrapper that `Popen`s the real executable and handles no SIGTERM** (only `KeyboardInterrupt`, assuming a terminal Ctrl-C reached the whole group), so `proc.terminate()` on it kills the wrapper and hands the node to init — once per run, *on the success path*. Measured on `test_twist_mux_arbitration.py`: 6 tests pass, 1 `twist_mux` survives; 0 after. Every `ros2 run` spawn goes through it (`test_twist_mux_arbitration.py`, `test_foxglove_teleop.py`, `tools/icp_gate_replay.py`; `bench.py`/`replay.py`/`camera_layer_decay.py` already did it by hand). The other half — a job killed outright, taking pytest with it before any teardown — no fixture can fix, and is what the sweep exists for. + **On-robot reliability** (see `mote_bringup/README.md`): `pixi run robot`/`mapping` include the health monitor, so a manual run publishes `/health` too; the systemd units are installed by `pixi run setup` but **not enabled** (autostart would drain the battery on a desk — opt in with `systemctl enable --now mote-bringup mote-health`). the systemd services restart with backoff and never permanently give up (`Restart=always`, `RestartSec`/`RestartSteps`/`RestartMaxDelaySec`, `StartLimitIntervalSec=0`), order after the udev-tagged `dev-mote_*.device` units, and bound the journal. A pre-flight self-check (`self_check.py`, run as `mote-bringup`'s `ExecStartPre`; `pixi run self-check`) gates bringup on servo ping + lidar/camera/disk/clock/config and keeps the robot idle with a clear reason on failure. A health monitor (`health_monitor.py`, `mote-health.service` with a `Type=notify` watchdog; `pixi run health`) publishes per-subsystem `diagnostic_msgs/DiagnosticArray` on `/diagnostics_agg` and a single OK/DEGRADED/FAULT summary on `/health`. **Wheel slip needs no IMU**: kinematic_icp takes wheel odometry as its prior and corrects it against the scan, so the correction already measures how wrong the wheels were — `slip_monitor.py` (started by `mote_launch.py` beside `system_monitor`) compares the two over a 1 s sliding window and publishes `slip` / `stuck` / `icp_fault` as the `slip` status on `/diagnostics`, plus the raw residual on `slip/residual`. All three are DEGRADED, never FAULT: each is a reason to stop and re-plan, not to refuse to drive, and a monitor that can halt the robot on a threshold is a worse failure than the slip. The maths is ROS-free in `odom_residual.py` and **shared with `tools/slip_replay.py`**, which is what set `config/slip.yaml`'s thresholds from the residual distribution over the real mapping bags — a threshold calibrated offline only means something if the robot computes the same number. Two things are load-bearing. **Only translation is thresholded**: the yaw residual is published but its p99 reaches the yaw rate itself, so no threshold survives a hard turn — and a lag sweep puts the two streams within ±10 ms, so this is scan-match jitter, *not* the stamp skew task 165 suspected. **A stalled lidar must not read as slip**: the window would freeze at the last pose while the wheels keep turning, growing without bound, so a source older than `max_lag` yields no verdict at all. `health_monitor` lifts named statuses off the shared `/diagnostics` by exact name, listed in `health.yaml`'s `diagnostic_statuses`. Six real events (two stuck robots, four scan-match excursions) were found in the existing "known-good" bags; the derivation and the sim demonstration are in `docs/tuning/2026-07-28-slip-detection.md`. Driver and nav2 nodes are `respawn=True` for per-node recovery under the whole-service systemd restart. Battery voltage is **not** software-measurable (the power bank exposes no telemetry); `system_monitor` reports the Pi's `get_throttled` flags as the only power signal — read via **`vcgencmd`**, since the Pi 4 sysfs node does not exist on a Pi 5, alongside the Active Cooler's `fan_rpm` from the `pwmfan` hwmon. **Launch hierarchy:** the two mission launches (`mapping_launch.py`, `robot_launch.py`) each take a `base` arg (default true) that includes the hardware base, and a `use_sim_time` arg they forward to everything they include. The sim runs these *same* files with `base:=false`, supplying a Gazebo base in place of the drivers — so the missions are defined once and the sim exercises the real launch files. diff --git a/mote_bringup/README.md b/mote_bringup/README.md index dfd48e5..4f8b2cd 100644 --- a/mote_bringup/README.md +++ b/mote_bringup/README.md @@ -330,6 +330,54 @@ The monitor is also the systemd watchdog feeder: it sends `READY=1` once up and pets the watchdog on every publish (`sd_notify.py`, a dependency-free `$NOTIFY_SOCKET` client that no-ops outside systemd). +## Clearing stray ROS processes — `sweep_orphans.py` + +Two different messes, one module, and the difference is worth knowing before +reaching for either. + +`pixi run kill` clears **this checkout's** ROS processes and resets the daemon — +the "my stack is wedged" reset. `pixi run sweep` clears **other jobs'** leftovers: +processes an agent worktree started and never reaped, which reparent to init and +run until the box is rebooted. It reports by default and only acts with `--kill`: + +```bash +pixi run sweep # what would go, grouped by the job that left it +pixi run sweep -- --kill # reap them +pixi run sweep -- --json # for a script +``` + +That second mess is not just untidiness. Leftovers are the exact process names a +benchmark measures, and the system-wide counters a benchmark sits in — context +switches, interrupts, memory pressure, CPU contention — cannot be scoped the way +`overhead.py` scopes its own match. A drifting background makes every +measurement on the box a little less comparable than it looks. + +**Matching is on process identity, never on the command line.** The `pkill -9 -f +''` that `kill` used to run matched the shell running the task +itself — those names are in its own command line — SIGKILLed it, and so never +reached the `ros2 daemon` reset that followed; it also matched every other +checkout and worktree on the machine. Both modes now read `/proc` and require a +ROS environment, a path under the directory in question, and absence from the +sweeper's own ancestry. The sweep additionally requires that the process be +orphaned and older than `--min-age` (30 min), because a deliberately +session-detached run — the sim smoke test `setsid`s its launch — is +indistinguishable from a leak by ancestry alone. + +### Why they escape + +`ros2 run` is a wrapper: it `Popen`s the real executable and installs no SIGTERM +handler, tolerating only `KeyboardInterrupt` on the assumption that the signal +reached the whole process group — true of a Ctrl-C at a terminal, false of a +`proc.terminate()` from a test fixture. Terminating the wrapper therefore kills +the wrapper and hands the node to init, **once per run, on the path where the run +succeeded**. Measured on `test_twist_mux_arbitration.py`: six tests pass and one +`twist_mux` survives. + +So anything spawning `ros2 run` uses `spawn_reapable` / `reap_group` from +`sweep_orphans`, which put the child in its own session and signal the group. +The other half — a job killed outright, taking pytest with it before any teardown +runs — no fixture can fix, and that is what the sweep is for. + ## Known gap: battery voltage The USB-C power bank exposes **no state-of-charge or voltage telemetry**, so the diff --git a/mote_bringup/mote_bringup/sweep_orphans.py b/mote_bringup/mote_bringup/sweep_orphans.py new file mode 100644 index 0000000..83fe7b6 --- /dev/null +++ b/mote_bringup/mote_bringup/sweep_orphans.py @@ -0,0 +1,579 @@ +"""Find and kill ROS processes left behind by agent worktrees that are gone. + +Every agent job runs in its own worktree under ``~/.claude/jobs//`` or +``/.claude/worktrees//``. When a job ends, its worktree goes +away — but a process it started does not, if nothing reaped it. Those survivors +reparent to init and run until the box is rebooted: two whole Nav2 stacks, a +dozen ``twist_mux`` and a handful of brokers were found here 2-4 days after the +jobs that spawned them had finished. + +They matter beyond the wasted core and RAM. They are the exact process names a +benchmark measures, and while ``overhead.py`` scopes its own match to the +current checkout, the system-wide counters a benchmark sits in — context +switches, interrupts, memory pressure, CPU contention — cannot be scoped. Every +measurement on the box is taken against a drifting background until they go. + + ros2 run mote_bringup sweep_orphans # pixi run sweep: report only + ros2 run mote_bringup sweep_orphans --kill # actually reap them + ros2 run mote_bringup sweep_orphans --json + +**Matching is on identity, never on the command line.** ``pkill -f`` matches the +sweeper's own shell (whose command line contains the pattern it was handed) and +every other checkout's live run alike; it is how a cleanup turns into an outage. +So a process is only a candidate when four independent things hold: + +* it carries a **ROS environment** (``AMENT_PREFIX_PATH`` and friends), which is + what separates a node from the file manager that happens to have a job path in + its argv; +* something about it — its executable, its working directory, its command line + or its ament/colcon/pixi prefixes — **lives under an agent root**, which is + what scopes the sweep to agent leakage rather than to the user's own work; +* it is **orphaned**: walking its ancestry reaches init without passing a live + process that is not itself a candidate, so anything still owned by a shell, a + pytest run or an agent session is left alone; +* it is **older than ``--min-age``** (default 30 min). A deliberately + session-detached run — the sim smoke test ``setsid``s its launch — is + indistinguishable from an orphan by ancestry alone, and age is what tells them + apart. The real leaks were days old; nothing legitimate is that old and idle. + +A small never-sweep set (shells, editors, ``claude`` itself) is the belt to that +brace: those inherit a ROS environment from a ``pixi shell`` like anything else, +and killing the operator's terminal is not a cleanup. + +Signalling re-checks each pid's start time against the value read during the +scan, so a pid recycled between scanning and killing is skipped rather than +shot — the window is small but the consequence is somebody else's process. +""" + +import argparse +import json +import os +import signal +import subprocess +import sys +import time +from dataclasses import dataclass +from pathlib import Path + +CLK_TCK = os.sysconf("SC_CLK_TCK") +PAGE_SIZE = os.sysconf("SC_PAGE_SIZE") + +#: Environment variables that mark a process as belonging to a ROS graph. +ROS_ENV_MARKERS = ("AMENT_PREFIX_PATH", "ROS_DISTRO", "RMW_IMPLEMENTATION") + +#: Environment variables naming a prefix that implicates a process in a worktree. +PREFIX_ENV_VARS = ("AMENT_PREFIX_PATH", "COLCON_PREFIX_PATH", "PIXI_PROJECT_ROOT") + +#: Never swept, whatever else matches: a `pixi shell` in a worktree gives these +#: a ROS environment and agent provenance, and they are the operator's, not ours. +NEVER_SWEEP = frozenset( + { + "bash", + "sh", + "zsh", + "fish", + "dash", + "tcsh", + "csh", + "claude", + "code", + "cursor", + "nvim", + "vim", + "vi", + "emacs", + "nano", + "tmux", + "screen", + "sshd", + "ssh", + "mosh-server", + "systemd", + "init", + "dbus-daemon", + "gnome-terminal-server", + "git", + "pixi", + "docker", + "containerd", + "dockerd", + } +) + +DEFAULT_MIN_AGE = 1800.0 # seconds + + +def agent_roots(checkout: Path | None = None) -> list[Path]: + """Directories under which an agent job's worktree lives.""" + roots = [Path.home() / ".claude" / "jobs"] + if checkout is not None: + roots.append(checkout / ".claude" / "worktrees") + return roots + + +def _read(path: str) -> bytes | None: + try: + with open(path, "rb") as handle: + return handle.read() + except OSError: + return None + + +def _readlink(path: str) -> str | None: + try: + return os.readlink(path) + except OSError: + return None + + +def _uptime() -> float: + with open("/proc/uptime") as handle: + return float(handle.read().split()[0]) + + +@dataclass +class Proc: + """One process as the sweep sees it. + + Everything the sweep decides on is captured at scan time, including + ``starttime`` — the value that makes a later signal safe against pid reuse. + Reading it out of /proc is ``read_proc``'s job, so every rule below is a + plain function of plain data and can be tested without a process to match. + """ + + pid: int + ppid: int + starttime: int + age: float + argv: list[str] + env: dict[str, str] + exe: str | None = None + cwd: str | None = None + rss: int = 0 + + @property + def cmdline(self) -> str: + return " ".join(self.argv) + + @property + def name(self) -> str: + """The program this process *is*, as a bare name. + + A node run through a console script is ``python3.12 /.../lib/pkg/node``, + so the interpreter's own name says nothing; the script path does. + """ + exe = Path(self.exe.removesuffix(" (deleted)")).name if self.exe else "" + if exe.startswith("python") and len(self.argv) > 1: + for arg in self.argv[1:]: + if arg.startswith("-"): + continue + return Path(arg).name + return exe or Path(self.argv[0]).name + + def provenance(self) -> list[str]: + """Every path that ties this process to a place on disk.""" + paths = [] + for value in (self.exe, self.cwd): + if value: + paths.append(value.removesuffix(" (deleted)")) + paths.extend(arg for arg in self.argv if arg.startswith("/")) + for var in PREFIX_ENV_VARS: + paths.extend(p for p in self.env.get(var, "").split(":") if p) + return paths + + def is_ros(self) -> bool: + return any(marker in self.env for marker in ROS_ENV_MARKERS) + + +def read_proc(pid: int, uptime: float) -> Proc | None: + """Read one process out of /proc, or None if it is not one we can judge. + + Kernel threads have an empty command line and a process that exits mid-read + leaves partial files; neither is something to sweep, so both come back None + rather than raising into the scan loop. + """ + raw_cmdline = _read(f"/proc/{pid}/cmdline") or b"" + argv = [a for a in raw_cmdline.decode("utf8", "replace").split("\0") if a] + stat = _read(f"/proc/{pid}/stat") + if stat is None or not argv: + return None + try: + # The comm field is parenthesised and may itself contain spaces, so the + # numeric fields start after the *last* ')'. + fields = stat.decode("utf8", "replace").rsplit(")", 1)[1].split() + ppid = int(fields[1]) + starttime = int(fields[19]) # field 22, 1-indexed, of proc(5) + except (IndexError, ValueError): + return None + + environ = _read(f"/proc/{pid}/environ") or b"" + statm = _read(f"/proc/{pid}/statm") + return Proc( + pid=pid, + ppid=ppid, + starttime=starttime, + age=max(0.0, uptime - starttime / CLK_TCK), + argv=argv, + env=dict( + entry.split("=", 1) + for entry in environ.decode("utf8", "replace").split("\0") + if "=" in entry + ), + exe=_readlink(f"/proc/{pid}/exe"), + cwd=_readlink(f"/proc/{pid}/cwd"), + rss=int(statm.split()[1]) * PAGE_SIZE if statm else 0, + ) + + +def read_all() -> dict[int, Proc]: + """Every process on the host that the sweep can read, keyed by pid.""" + uptime = _uptime() + procs = {} + for entry in os.listdir("/proc"): + if not entry.isdigit(): + continue + proc = read_proc(int(entry), uptime) + if proc is not None: + procs[proc.pid] = proc + return procs + + +def _under(path: str, root: Path) -> bool: + try: + Path(path).relative_to(root) + except ValueError: + return False + return True + + +def owning_job(proc: Proc, roots: list[Path]) -> str | None: + """The agent job this process came out of, or None. + + Answered as the first path component below an agent root — a job id under + ``~/.claude/jobs``, a worktree name under ``.claude/worktrees`` — so every + process from one job groups under one heading, whichever of its paths + happened to implicate it. Jobs vary in where they put their worktree + (``/wt-foo`` and ``/tmp/wt-foo`` both occur), and the job is the unit + that died in either case. + """ + for path in proc.provenance(): + for root in roots: + if not _under(path, root): + continue + parts = Path(path).relative_to(root).parts + if not parts: + continue + return str(root / parts[0]) + return None + + +def _orphaned(proc: Proc, by_pid: dict[int, Proc], candidates: set[int]) -> bool: + """True when nothing outside the candidate set still owns this process. + + Walks up the ancestry. Reaching init having only passed candidates means the + whole tree was reparented — the job that owned it is gone. Meeting any live + process that is not a candidate (a shell, a pytest run, an agent session) + means the run is still owned, and is left alone. + """ + seen = set() + pid = proc.ppid + while pid > 1 and pid not in seen: + seen.add(pid) + parent = by_pid.get(pid) + if parent is None: # exited between reads; treat the chain as broken + return True + if pid not in candidates: + return False + pid = parent.ppid + return True + + +def select( + by_pid: dict[int, Proc], + roots: list[Path], + min_age: float = DEFAULT_MIN_AGE, + self_pid: int | None = None, +) -> list[Proc]: + """Every orphaned ROS process in ``by_pid`` traceable to a job under ``roots``. + + A pure function of the process table, so the whole rule — including the + self-match guard that ``pkill -f`` lacks — is testable without a process to + match against. + """ + ours = _ancestry(os.getpid() if self_pid is None else self_pid, by_pid) + candidates = { + pid + for pid, proc in by_pid.items() + if proc.is_ros() + and proc.name not in NEVER_SWEEP + and owning_job(proc, roots) is not None + and pid not in ours + } + return sorted( + ( + by_pid[pid] + for pid in candidates + if by_pid[pid].age >= min_age and _orphaned(by_pid[pid], by_pid, candidates) + ), + key=lambda p: p.pid, + ) + + +def scan(roots: list[Path], min_age: float = DEFAULT_MIN_AGE) -> list[Proc]: + """Every orphaned ROS process on this host traceable to a job under ``roots``.""" + return select(read_all(), roots, min_age=min_age) + + +def owned_by( + by_pid: dict[int, Proc], scope: Path, self_pid: int | None = None +) -> list[Proc]: + """Every ROS process running out of ``scope``, orphaned or not. + + What ``pixi run kill`` wants: a stuck stack in this checkout is not orphaned + — its launch is right there — so the sweep's ancestry and age rules would + spare exactly the processes it exists to clear. The self-ancestry and + never-sweep guards still apply, because the shell running the reset is + itself a process of this checkout. + """ + scope = scope.resolve() + ours = _ancestry(os.getpid() if self_pid is None else self_pid, by_pid) + return sorted( + ( + proc + for pid, proc in by_pid.items() + if proc.is_ros() + and proc.name not in NEVER_SWEEP + and pid not in ours + and any(_under(path, scope) for path in proc.provenance()) + ), + key=lambda p: p.pid, + ) + + +def _ancestry(pid: int, by_pid: dict[int, Proc]) -> set[int]: + """A pid and all of its ancestors — the sweeper never signals its own line.""" + chain = {pid} + while (proc := by_pid.get(pid)) is not None and proc.ppid > 1: + pid = proc.ppid + if pid in chain: + break + chain.add(pid) + return chain + + +def _still_is(proc: Proc) -> bool: + """True when this pid is still the process that was scanned. + + A pid freed and reissued between the scan and the signal would otherwise be + killed in place of the orphan. + """ + stat = _read(f"/proc/{proc.pid}/stat") + if stat is None: + return False + try: + return ( + int(stat.decode("utf8", "replace").rsplit(")", 1)[1].split()[19]) + == proc.starttime + ) + except (IndexError, ValueError): + return False + + +def reap(procs: list[Proc], grace: float = 5.0) -> tuple[list[int], list[int]]: + """SIGTERM, then SIGKILL whatever is left. Returns (terminated, killed).""" + terminated, killed = [], [] + for proc in procs: + if not _still_is(proc): + continue + try: + os.kill(proc.pid, signal.SIGTERM) + terminated.append(proc.pid) + except (ProcessLookupError, PermissionError): + continue + + deadline = time.monotonic() + grace + remaining = [p for p in procs if p.pid in terminated] + while remaining and time.monotonic() < deadline: + time.sleep(0.2) + remaining = [p for p in remaining if _still_is(p)] + + for proc in remaining: + if not _still_is(proc): + continue + try: + os.kill(proc.pid, signal.SIGKILL) + killed.append(proc.pid) + except (ProcessLookupError, PermissionError): + continue + return terminated, killed + + +# --- not leaking in the first place ---------------------------------------- + + +def spawn_reapable(command: list[str], **kwargs) -> subprocess.Popen: + """Start a process that can actually be stopped again. + + ``ros2 run`` is a Python wrapper that ``Popen``s the real executable and + installs no SIGTERM handler — it only tolerates ``KeyboardInterrupt``, on the + assumption (true only of a Ctrl-C at a terminal) that the signal reached the + whole process group. So ``proc.terminate()`` on the wrapper kills the wrapper + alone and hands the node to init. That is where most of this repo's leaked + nodes came from: one per run, on the path where the run *succeeded*. + + Putting the child in its own session makes the wrapper and the node + addressable together, which is what ``reap_group`` then signals. + """ + return subprocess.Popen(command, start_new_session=True, **kwargs) + + +def reap_group(proc: subprocess.Popen | None, grace: float = 10.0) -> None: + """SIGTERM a ``spawn_reapable`` process group, then SIGKILL what remains.""" + if proc is None or proc.poll() is not None: + return + for sig in (signal.SIGTERM, signal.SIGKILL): + try: + os.killpg(os.getpgid(proc.pid), sig) + except (ProcessLookupError, PermissionError): + return + try: + proc.wait(timeout=grace) + return + except subprocess.TimeoutExpired: + continue + + +def _age(seconds: float) -> str: + days, rest = divmod(int(seconds), 86400) + hours, rest = divmod(rest, 3600) + minutes = rest // 60 + if days: + return f"{days}d{hours:02d}h" + if hours: + return f"{hours}h{minutes:02d}m" + return f"{minutes}m" + + +def report( + procs: list[Proc], roots: list[Path], scope: Path | None = None, stream=sys.stdout +) -> None: + if not procs: + where = f"running out of {scope}" if scope else "from dead agent worktrees" + print(f"no ROS processes {where}", file=stream) + return + + groups: dict[str, list[Proc]] = {} + for proc in procs: + key = str(scope) if scope else (owning_job(proc, roots) or "?") + groups.setdefault(key, []).append(proc) + + total_rss = sum(p.rss for p in procs) + what = ( + f"{len(procs)} ROS process(es) in this checkout" + if scope + else f"{len(procs)} orphaned ROS process(es) from {len(groups)} dead job(s)" + ) + print(f"{what}, {total_rss / 1e6:.0f} MB resident", file=stream) + home = str(Path.home()) + for owner, members in sorted(groups.items()): + print(f"\n {owner.replace(home, '~')}", file=stream) + for proc in members: + # An executable whose file is unlinked is the clearest sign the + # worktree it came from has already been removed underneath it. + gone = " (exe deleted)" if (proc.exe or "").endswith(" (deleted)") else "" + print( + f" {proc.pid:>8} {_age(proc.age):>6} {proc.rss / 1e6:>6.1f} MB " + f"{proc.name}{gone}", + file=stream, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument( + "--kill", + action="store_true", + help="actually reap them; without this the sweep only reports", + ) + parser.add_argument( + "--min-age", + type=float, + default=DEFAULT_MIN_AGE, + metavar="SECONDS", + help="ignore processes younger than this (default: %(default)s). A " + "session-detached run that is still live looks orphaned by ancestry; " + "age is what separates it from a leak.", + ) + parser.add_argument( + "--root", + action="append", + type=Path, + metavar="DIR", + help="additional directory under which agent worktrees live " + "(repeatable); defaults to ~/.claude/jobs and /.claude/worktrees", + ) + parser.add_argument( + "--checkout", + type=Path, + default=Path.cwd(), + help="checkout whose .claude/worktrees to include (default: cwd)", + ) + parser.add_argument( + "--scope", + type=Path, + metavar="DIR", + help="instead of sweeping dead jobs, take every ROS process running out " + "of DIR whether orphaned or not — what `pixi run kill` needs to clear a " + "stuck stack, which is live by definition", + ) + parser.add_argument("--json", action="store_true", help="machine-readable output") + args = parser.parse_args(argv) + + roots = agent_roots(args.checkout) + list(args.root or []) + if args.scope is not None: + procs = owned_by(read_all(), args.scope) + else: + procs = scan(roots, min_age=args.min_age) + + if args.kill: + terminated, killed = reap(procs) + else: + terminated, killed = [], [] + + if args.json: + json.dump( + { + "processes": [ + { + "pid": p.pid, + "ppid": p.ppid, + "name": p.name, + "age_s": round(p.age, 1), + "rss_bytes": p.rss, + "worktree": owning_job(p, roots), + "cmdline": p.cmdline, + } + for p in procs + ], + "terminated": terminated, + "killed": killed, + "swept": args.kill, + }, + sys.stdout, + indent=2, + ) + print() + else: + report(procs, roots, scope=args.scope) + if args.kill and procs: + print( + f"\nreaped {len(terminated)} (SIGKILL needed for {len(killed)})", + file=sys.stdout, + ) + elif procs: + print("\nnothing killed — re-run with --kill to reap them", file=sys.stdout) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/mote_bringup/setup.py b/mote_bringup/setup.py index 27d30e3..40c6d4b 100644 --- a/mote_bringup/setup.py +++ b/mote_bringup/setup.py @@ -39,6 +39,7 @@ "identity = mote_bringup.identity:main", "provision = mote_bringup.provision:main", "dds_participants = mote_bringup.dds_participants:main", + "sweep_orphans = mote_bringup.sweep_orphans:main", "explore = mote_bringup.explore:main", ], }, diff --git a/mote_bringup/test/test_foxglove_teleop.py b/mote_bringup/test/test_foxglove_teleop.py index 89460b4..93213f2 100644 --- a/mote_bringup/test/test_foxglove_teleop.py +++ b/mote_bringup/test/test_foxglove_teleop.py @@ -31,6 +31,8 @@ import pytest +from mote_bringup.sweep_orphans import reap_group, spawn_reapable + try: import websockets except ImportError: # the robot environment carries no WebSocket client @@ -98,8 +100,10 @@ def teleop_stack(): os.environ["ROS_DOMAIN_ID"] = str(random.randint(80, 160)) port = _free_port() + # spawn_reapable, not Popen: `ros2 run` forwards no signal to the node it + # spawned, so terminating the wrapper alone leaks the bridge and the relay. procs = [ - subprocess.Popen( + spawn_reapable( [ "ros2", "run", @@ -114,7 +118,7 @@ def teleop_stack(): stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT, ), - subprocess.Popen( + spawn_reapable( [ "ros2", "run", @@ -139,7 +143,7 @@ def teleop_stack(): time.sleep(0.5) else: for p in procs: - p.kill() + reap_group(p) pytest.fail("foxglove_bridge never opened its port") time.sleep(3) # let the relay finish discovery @@ -161,11 +165,7 @@ def _spin(): sink.destroy_node() rclpy.try_shutdown() for p in procs: - p.terminate() - try: - p.wait(timeout=10) - except subprocess.TimeoutExpired: - p.kill() + reap_group(p) async def _drive(port: int, schema_name: str, count: int = 20) -> None: diff --git a/mote_bringup/test/test_sweep_orphans.py b/mote_bringup/test/test_sweep_orphans.py new file mode 100644 index 0000000..d295bb2 --- /dev/null +++ b/mote_bringup/test/test_sweep_orphans.py @@ -0,0 +1,386 @@ +"""The sweep decides what to kill, so what it *declines* to kill is the test. + +The rule it replaces is ``pkill -9 -f ``, whose failure modes are all of +the shape "matched something that was never ours": the sweeper's own shell, +another checkout's live run, a file manager with the pattern in its argv. Most +of these cases are a process table away from being unreproducible, so the +selection rule is a pure function of one and the fabricated tables below are the +real subject. One end-to-end case spawns an actual orphan and reaps it, to keep +the /proc reading honest. +""" + +import os +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from mote_bringup import sweep_orphans as sweep + +JOBS = Path("/home/u/.claude/jobs") +WORKTREES = Path("/home/u/checkout/.claude/worktrees") +ROOTS = [JOBS, WORKTREES] + +ROS_ENV = {"AMENT_PREFIX_PATH": "/home/u/checkout/install/mote_bringup"} +OLD = sweep.DEFAULT_MIN_AGE * 2 + + +def proc(pid, *, ppid=1, argv=None, env=None, exe=None, cwd=None, age=OLD, **kw): + return sweep.Proc( + pid=pid, + ppid=ppid, + starttime=pid * 100, + age=age, + argv=argv or ["/usr/bin/thing"], + env=dict(ROS_ENV if env is None else env), + exe=exe, + cwd=cwd, + **kw, + ) + + +def table(*procs): + return {p.pid: p for p in procs} + + +# A pid that is not in any fabricated table, so the sweeper's own ancestry never +# collides with the processes under test. +SELF = 99999 + + +def select(*procs, min_age=sweep.DEFAULT_MIN_AGE, self_pid=SELF): + return sweep.select(table(*procs), ROOTS, min_age=min_age, self_pid=self_pid) + + +# --- what gets swept ------------------------------------------------------- + + +def test_orphaned_node_from_a_dead_job_is_swept(): + node = proc(10, exe=str(JOBS / "abc/wt-x/install/mote_nav/lib/icp_odom_gate")) + assert [p.pid for p in select(node)] == [10] + + +def test_a_chain_of_orphans_is_swept_whole(): + """The launch died with the job; its children are orphans of an orphan.""" + launch = proc(10, exe=str(JOBS / "abc/wt-x/.pixi/bin/ros2")) + child = proc(11, ppid=10, exe=str(JOBS / "abc/wt-x/lib/foxglove_bridge")) + assert [p.pid for p in select(launch, child)] == [10, 11] + + +def test_provenance_can_come_from_the_environment_alone(): + """A worktree run against the main checkout's env has a main-checkout exe. + + ``pixi run`` from a worktree uses the primary checkout's environment, so the + interpreter and the binaries live outside the job directory entirely and + only the overlay prefixes implicate it. Matching on the executable alone — + as the obvious version of this sweep would — misses these. + """ + node = proc( + 10, + exe="/home/u/checkout/.pixi/envs/default/bin/python3.12", + cwd="/home/u/checkout", + argv=["python3.12", "-u", "-c", "pass"], + env={"AMENT_PREFIX_PATH": str(JOBS / "abc/tmp/overlay/mote_bringup")}, + ) + assert [p.pid for p in select(node)] == [10] + + +def test_a_deleted_executable_still_implicates_its_job(): + """An unlinked exe reads back with a ' (deleted)' suffix glued on.""" + node = proc(10, exe=str(JOBS / "abc/wt-x/lib/mosquitto") + " (deleted)") + assert [p.pid for p in select(node)] == [10] + assert node.name == "mosquitto" + + +def test_worktrees_under_the_checkout_are_swept_too(): + node = proc(10, cwd=str(WORKTREES / "sim-sites")) + assert [p.pid for p in select(node)] == [10] + + +# --- what is left alone ---------------------------------------------------- + + +def test_a_run_still_owned_by_a_shell_is_left_alone(): + """Another agent's live run: its ancestry reaches a shell, not init.""" + shell = proc(10, argv=["/usr/bin/zsh"], exe="/usr/bin/zsh", env={}) + launch = proc(11, ppid=10, exe=str(JOBS / "abc/wt-x/.pixi/bin/ros2")) + node = proc(12, ppid=11, exe=str(JOBS / "abc/wt-x/lib/twist_mux")) + assert select(shell, launch, node) == [] + + +def test_the_sweeper_never_matches_its_own_ancestry(): + """The ``pkill -f`` foot-gun: the sweep must not be able to kill itself. + + Both the sweeping process and everything that spawned it are excluded, so a + sweep run from inside a worktree — the normal case — cannot reap the shell + it was typed into, nor the job it belongs to. + """ + shell = proc(10, exe=str(JOBS / "abc/wt-x/.pixi/bin/python3.12")) + me = proc(11, ppid=10, exe=str(JOBS / "abc/wt-x/.pixi/bin/python3.12")) + stranger = proc(12, exe=str(JOBS / "def/wt-y/lib/twist_mux")) + assert [p.pid for p in select(shell, me, stranger, self_pid=11)] == [12] + + +def test_a_non_ros_process_holding_a_job_path_is_left_alone(): + """A file manager opened on a job directory has the path but not the graph.""" + files = proc( + 10, + argv=["/usr/bin/cosmic-files", str(JOBS / "abc/tmp/")], + exe="/usr/bin/cosmic-files", + env={}, + ) + assert select(files) == [] + + +def test_never_swept_programs_are_left_alone(): + """A `pixi shell` in a worktree gives a shell every marker a node has.""" + for name in ("zsh", "claude", "nvim", "git"): + shell = proc( + 10, + argv=[f"/usr/bin/{name}"], + exe=f"/usr/bin/{name}", + cwd=str(JOBS / "abc/wt-x"), + ) + assert select(shell) == [], name + + +def test_a_young_process_is_left_alone(): + """A session-detached live run looks orphaned; only its age says otherwise. + + ``run_sim_smoke.sh`` ``setsid``s its launch, so a perfectly healthy smoke + test has ppid 1 and no live non-candidate ancestor. Age is the only thing + separating it from a leak, which is why the sweep has a floor. + """ + node = proc(10, exe=str(JOBS / "abc/wt-x/lib/twist_mux"), age=20.0) + assert select(node) == [] + assert [p.pid for p in select(node, min_age=0)] == [10] + + +def test_a_process_outside_every_agent_root_is_left_alone(): + """The user's own checkout is not agent leakage, however orphaned it is.""" + node = proc( + 10, + exe="/home/u/checkout/install/mote_nav/lib/icp_odom_gate", + cwd="/home/u/checkout", + ) + assert select(node) == [] + + +# --- identity and grouping ------------------------------------------------- + + +def test_name_resolves_through_the_interpreter(): + """`python3.12 /.../lib/mote_bringup/twist_relay` is a twist_relay.""" + node = proc( + 10, + exe="/home/u/checkout/.pixi/envs/default/bin/python3.12", + argv=[ + "/home/u/.../python3.12", + str(JOBS / "a/install/mote_bringup/lib/twist_relay"), + "--ros-args", + ], + ) + assert node.name == "twist_relay" + + +def test_name_skips_interpreter_flags(): + node = proc( + 10, exe="/usr/bin/python3", argv=["python3", "-u", "/opt/thing/server.py"] + ) + assert node.name == "server.py" + + +def test_name_falls_back_to_the_executable(): + node = proc( + 10, exe="/opt/ros/lib/twist_mux/twist_mux", argv=["twist_mux", "--ros-args"] + ) + assert node.name == "twist_mux" + + +@pytest.mark.parametrize( + "path", + [ + JOBS / "abc/wt-x/install/lib/node", + JOBS / "abc/tmp/wt-x/install/lib/node", + JOBS / "abc/tmp/overlay/lib/node", + ], +) +def test_owning_job_groups_by_job_whatever_the_layout(path): + """Jobs disagree about where the worktree goes; the job is what died.""" + assert sweep.owning_job(proc(10, exe=str(path)), ROOTS) == str(JOBS / "abc") + + +def test_owning_job_names_the_worktree_for_checkout_worktrees(): + node = proc(10, cwd=str(WORKTREES / "sim-sites/mote_nav")) + assert sweep.owning_job(node, ROOTS) == str(WORKTREES / "sim-sites") + + +def test_owning_job_is_none_outside_the_roots(): + assert sweep.owning_job(proc(10, exe="/usr/bin/thing"), ROOTS) is None + + +# --- the scoped mode behind `pixi run kill` -------------------------------- + +CHECKOUT = Path("/home/u/checkout") + + +def test_scope_takes_a_live_stack_not_just_orphans(): + """A stuck stack has its launch right there; the orphan rule would spare it.""" + shell = proc(10, argv=["/usr/bin/zsh"], exe="/usr/bin/zsh", env={}) + launch = proc(11, ppid=10, exe=str(CHECKOUT / ".pixi/envs/default/bin/ros2")) + node = proc(12, ppid=11, exe=str(CHECKOUT / "install/mote_nav/lib/icp_odom_gate")) + found = sweep.owned_by(table(shell, launch, node), CHECKOUT, self_pid=SELF) + assert [p.pid for p in found] == [11, 12] + + +def test_scope_is_ageless(): + """A stack started a second ago is exactly what a reset is aimed at.""" + node = proc(10, exe=str(CHECKOUT / "install/lib/node"), age=0.0) + assert [p.pid for p in sweep.owned_by(table(node), CHECKOUT, self_pid=SELF)] == [10] + + +def test_scope_spares_the_shell_running_it(): + """The regression that made `pixi run kill` truncate itself. + + ``pkill -9 -f ''`` matched the task's own shell — those names + are in its command line — killed it, and so never ran the daemon reset that + followed. Nothing in the sweeper's own ancestry may be selected, and a shell + is never selected regardless. + """ + shell = proc( + 10, + argv=["sh", "-c", "sweep --scope . --kill; ros2 daemon stop"], + exe="/usr/bin/sh", + ) + me = proc( + 11, + ppid=10, + exe=str(CHECKOUT / ".pixi/envs/default/bin/python3.12"), + argv=["python", "-m", "mote_bringup.sweep_orphans"], + ) + node = proc(12, exe=str(CHECKOUT / "install/mote_nav/lib/icp_odom_gate")) + found = sweep.owned_by(table(shell, me, node), CHECKOUT, self_pid=11) + assert [p.pid for p in found] == [12] + + +def test_scope_ignores_another_checkout(): + node = proc( + 10, + exe="/home/u/other-checkout/install/lib/node", + env={"AMENT_PREFIX_PATH": "/home/u/other-checkout/install/mote_nav"}, + ) + assert sweep.owned_by(table(node), CHECKOUT, self_pid=SELF) == [] + + +# --- reaping --------------------------------------------------------------- + + +def test_pid_reuse_is_not_killed(): + """A pid recycled between the scan and the signal must be skipped. + + ``_still_is`` compares the start time recorded at scan against the live one, + so a stale record — here, our own pid with somebody else's start time — + fails to match and is never signalled. + """ + live = sweep.read_proc(os.getpid(), sweep._uptime()) + assert sweep._still_is(live) + + stale = proc(os.getpid()) + stale.starttime = live.starttime + 1 + assert not sweep._still_is(stale) + + +def test_reap_ignores_a_process_that_has_already_gone(): + gone = proc(2**22 - 1) # above /proc/sys/kernel/pid_max on any sane host + assert sweep.reap([gone], grace=0.1) == ([], []) + + +def _alive(pid): + return Path(f"/proc/{pid}").exists() + + +def _ppid(pid): + proc = sweep.read_proc(pid, sweep._uptime()) + return proc.ppid if proc else None + + +@pytest.fixture +def orphan(tmp_path): + """A real orphaned process wearing a ROS environment, under a fake job root. + + The leak this sweep exists for is a *grandchild*: a fixture terminated the + `ros2 run` wrapper it held, and the node the wrapper had spawned reparented + to init. So the process here is spawned through a launcher that exits + immediately, which is the only way to get a genuine ppid of 1 — + ``start_new_session`` detaches the session but leaves the parent in place. + """ + root = tmp_path / "jobs" + workdir = root / "job1" / "wt-x" + workdir.mkdir(parents=True) + pidfile = tmp_path / "pid" + + launcher = ( + "import os, subprocess, sys\n" + f"p = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(300)'],\n" + f" cwd={str(workdir)!r},\n" + f" env={{**os.environ, 'AMENT_PREFIX_PATH': {str(workdir / 'install/mote_bringup')!r}}},\n" + " stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)\n" + f"open({str(pidfile)!r}, 'w').write(str(p.pid))\n" + ) + subprocess.run([sys.executable, "-c", launcher], check=True, timeout=60) + pid = int(pidfile.read_text()) + + deadline = time.monotonic() + 10 + while _ppid(pid) not in (1, None) and time.monotonic() < deadline: + time.sleep(0.1) + if _ppid(pid) != 1: + pytest.skip("orphan was not reparented to init (a subreaper claimed it)") + + try: + yield pid, root + finally: + if _alive(pid): + os.kill(pid, 9) + + +def test_finds_and_reaps_a_real_orphan(orphan): + """End to end against a real process, without needing ROS to be running. + + An orphan with a ROS-looking environment and a working directory under an + agent root is exactly the shape of the leak, so this exercises the /proc + reading, the ancestry test and the kill together. + """ + pid, root = orphan + + found = sweep.select(sweep.read_all(), [root], min_age=0) + assert pid in [p.pid for p in found], "the orphan was not found" + + target = next(p for p in found if p.pid == pid) + assert sweep.owning_job(target, [root]) == str(root / "job1") + + terminated, _ = sweep.reap([target], grace=5.0) + assert terminated == [pid] + + deadline = time.monotonic() + 10 + while _alive(pid) and time.monotonic() < deadline: + time.sleep(0.1) + assert not _alive(pid), "the orphan survived the sweep" + + +def test_a_real_orphan_is_spared_while_it_is_too_young(orphan): + """The age floor is what keeps a live session-detached run out of the sweep.""" + pid, root = orphan + found = sweep.select(sweep.read_all(), [root]) # default min age + assert pid not in [p.pid for p in found] + assert _alive(pid) + + +def test_the_sweep_ignores_a_real_orphan_outside_its_roots(orphan, tmp_path): + """Same process, a root that does not contain it: not the sweep's business.""" + pid, _ = orphan + elsewhere = tmp_path / "not-jobs" + elsewhere.mkdir() + found = sweep.select(sweep.read_all(), [elsewhere], min_age=0) + assert pid not in [p.pid for p in found] diff --git a/mote_bringup/test/test_twist_mux_arbitration.py b/mote_bringup/test/test_twist_mux_arbitration.py index 98ffb91..27c1a44 100644 --- a/mote_bringup/test/test_twist_mux_arbitration.py +++ b/mote_bringup/test/test_twist_mux_arbitration.py @@ -34,6 +34,8 @@ import pytest import yaml +from mote_bringup.sweep_orphans import reap_group, spawn_reapable + REPO = pathlib.Path(__file__).resolve().parents[2] CONFIG = REPO / "mote_bringup" / "config" / "twist_mux.yaml" CONTROLLERS = REPO / "mote_bringup" / "config" / "controllers.yaml" @@ -141,7 +143,9 @@ def mux(): # is the one a robot on the bench drives on. os.environ["ROS_DOMAIN_ID"] = str(random.randint(80, 160)) - proc = subprocess.Popen( + # spawn_reapable, not Popen: `ros2 run` forwards no signal to the node it + # spawned, so terminating the wrapper alone leaks a twist_mux per run. + proc = spawn_reapable( [ "ros2", "run", @@ -174,15 +178,14 @@ def mux(): time.sleep(0.1) node.drive_nav(False) if not node.out: - proc.terminate() + reap_group(proc) executor.shutdown() rclpy.try_shutdown() pytest.fail("no message reached the drive topic within 30 s") yield node - proc.terminate() - proc.wait(timeout=10) + reap_group(proc) executor.shutdown() node.destroy_node() rclpy.try_shutdown() diff --git a/mote_bringup/tools/icp_gate_replay.py b/mote_bringup/tools/icp_gate_replay.py index 48e131e..5a0b7cc 100755 --- a/mote_bringup/tools/icp_gate_replay.py +++ b/mote_bringup/tools/icp_gate_replay.py @@ -21,7 +21,6 @@ from __future__ import annotations import argparse -import subprocess import time from pathlib import Path @@ -29,6 +28,7 @@ import rosbag2_py import yaml from ament_index_python.packages import get_package_share_directory +from mote_bringup.sweep_orphans import reap_group, spawn_reapable from geometry_msgs.msg import TransformStamped from nav_msgs.msg import Odometry from rclpy.node import Node @@ -168,7 +168,7 @@ def main(): raise SystemExit("bag carries neither edge; nothing to replay") robot = robot_config() - gate = subprocess.Popen( + gate = spawn_reapable( [ "ros2", "run", @@ -234,8 +234,7 @@ def main(): node.tick(icp[-1][0]) rclpy.spin_once(node, timeout_sec=0.01) finally: - gate.terminate() - gate.wait(timeout=10) + reap_group(gate) print(f"gate published {len(node.gated)} of {len(icp)} poses") if len(node.gated) < len(icp): diff --git a/pixi.toml b/pixi.toml index 623ade2..9b2d0c1 100644 --- a/pixi.toml +++ b/pixi.toml @@ -57,7 +57,15 @@ inference-deploy = "bash mote_perception/deploy/inference-deploy.sh" # GPU box deploy-test = "bash mote_perception/deploy/test/drill.sh" # blue/green drill, no GPU # Dev tasks -kill = "pkill -9 -f 'robot_state_publisher|ros2_control_node|sllidar|v4l2_camera|spawner'; ros2 daemon stop; rm -f /dev/shm/fast*; ros2 daemon start; true" +# Matching is on each process' identity in /proc, not on its command line: the +# `pkill -9 -f ''` this replaces matched the shell running the task +# itself (the names are in its own command line), SIGKILLed it, and so never +# reached the daemon reset below -- and it matched every other checkout and +# worktree on the box besides. See mote_bringup/sweep_orphans.py. +kill = "PYTHONPATH=mote_bringup python -m mote_bringup.sweep_orphans --scope . --kill; ros2 daemon stop; rm -f /dev/shm/fast*; ros2 daemon start; true" +# Reap ROS processes left behind by agent worktrees whose job has ended. Reports +# only; add --kill to act. +sweep = "PYTHONPATH=mote_bringup python -m mote_bringup.sweep_orphans" # '.git' has no trailing slash on purpose: a submodule's .git is a *file* # holding a path into this checkout's .git/modules, and '.git/' (directory-only) # would ship it to the robot and break the robot's submodule refs.