Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246"""Owned process groups: identify one, probe it, end it, report what happened.
The kernel's job supervision and the supervisor's checked signal helper both runthis ladder, so the two layers agree on what "terminated" means. A group is liveonly while it holds a member we can still signal; a reaped leader or a group ofnon-running (zombie) members is not work."""
from __future__ import annotations
from collections.abc import Callable, Sequencefrom dataclasses import dataclassfrom typing import Anyimport asyncioimport osimport signalimport subprocessimport sys
TERM_GRACE = 0.25 # seconds a group gets to honour TERMKILL_GRACE = 2.0 # seconds a group gets to die after KILLPROBE_INTERVAL = 0.02
def reap_stopped_safely(loop: asyncio.AbstractEventLoop) -> None: """Let `loop` run children that get paused with SIGSTOP.
Python 3.14's threaded child watcher waits with waitid(WEXITED | WNOWAIT), which macOS also answers for a stopped child, and then reaps with a blocking waitpid on the event loop: a job paused for a heavy slot froze the kernel. The replacement reaps in its thread with a waitpid that ignores stops, as 3.13 did. Linux uses pidfds, which only wake on exit, and needs nothing. """ threaded: Any = getattr( getattr(asyncio, "unix_events", None), "_ThreadedChildWatcher", None ) if ( sys.platform != "darwin" or threaded is None or not isinstance(getattr(loop, "_watcher", None), threaded) ): return
class ExitWatcher(threaded): def _do_waitpid( self, loop: asyncio.AbstractEventLoop, expected_pid: int, callback: Callable[..., object], args: tuple[object, ...], ) -> None: try: returncode = os.waitstatus_to_exitcode(os.waitpid(expected_pid, 0)[1]) except ChildProcessError: returncode = 255 try: loop.call_soon_threadsafe(callback, expected_pid, returncode, *args) except RuntimeError: # the loop closed first pass self._threads.pop(expected_pid, None)
setattr(loop, "_watcher", ExitWatcher())
@dataclass(frozen=True)class Group: """A target to end: a whole process group, or one process that leads none.
`leader` identifies the leader where /proc exists, so a reused id is refused. """
pgid: int leader: str | None = None whole: bool = True
@dataclass(frozen=True)class Termination: """What one termination attempt established about one process group."""
pgid: int | None signals: tuple[str, ...] gone: bool failures: tuple[str, ...] = () note: str = ""
def report(self) -> str: where = ( "no process group" if self.pgid is None else f"process group {self.pgid}" ) outcome = "terminated" if self.gone else "SURVIVED" sent = "+".join(self.signals) if self.signals else "no signal" detail = ( "; ".join((*self.failures, self.note)) if self.note else "; ".join(self.failures) ) return f"{where} {outcome} after {sent}" + (f": {detail}" if detail else "")
def as_json(self) -> dict[str, object]: return { "pgid": self.pgid, "signals": list(self.signals), "gone": self.gone, "failures": list(self.failures), "note": self.note, }
def start_token(stat: bytes) -> str | None: """Start time of one /proc/<pid>/stat line (field 22, after the command name).""" try: return stat.rsplit(b")", 1)[1].split()[19].decode() except (IndexError, UnicodeDecodeError): return None
def leader_token(pid: int) -> str | None: """Identity token for a live pid, where the platform exposes /proc.""" try: with open(f"/proc/{pid}/stat", "rb") as stat: return start_token(stat.read()) except OSError: return None
def live_members(pgid: int) -> list[int] | None: """Pids in a group that are not zombies; None where /proc is unavailable.""" try: entries = os.listdir("/proc") except OSError: if sys.platform != "darwin": return None # Darwin has no /proc. Ask its process table; EPERM alone never means dead. try: result = subprocess.run( ["/bin/ps", "-A", "-o", "pid=", "-o", "pgid=", "-o", "stat="], capture_output=True, text=True, timeout=0.5, check=True, ) rows = [line.split() for line in result.stdout.splitlines()] return [ int(pid) for pid, group, state in rows if int(group) == pgid and not state.startswith("Z") ] except (OSError, subprocess.SubprocessError, ValueError): return None members = [] for entry in entries: if not entry.isdigit(): continue try: with open(f"/proc/{entry}/stat", "rb") as stat: fields = stat.read().rsplit(b")", 1)[1].split() except FileNotFoundError: continue # process exited while enumerating except (OSError, IndexError): return None # an incomplete view cannot prove the group empty if fields[0] == b"Z" or fields[2] != str(pgid).encode(): continue members.append(int(entry)) return members
def deliver(group: Group, sig: int) -> None: """Signal the target: its group, or the single process when it leads none.""" (os.killpg if group.whole else os.kill)(group.pgid, sig)
def alive(group: Group) -> bool: """True while the target exists and still looks like the one we created.""" try: deliver(group, 0) except ProcessLookupError: return False except PermissionError: return True # existence is known; inability to signal is a cleanup failure if group.leader is None: return True token = leader_token(group.pgid) return token is None or token == group.leader
def current(group: Group) -> bool: """alive/1 refined by membership: zombies awaiting reaping are not work.""" if not alive(group): return False if not group.whole: return True members = live_members(group.pgid) return True if members is None else bool(members)
def send_signal(group: Group, sig: int) -> str | None: """Send one signal to a target; returns the failure text when it did not land.""" try: deliver(group, sig) return None except ProcessLookupError: return None # already gone; the probe decides except OSError as error: return f"{signal.Signals(sig).name} to {group.pgid}: {error.strerror or error}"
async def settled(live: dict[int, Group], window: float) -> dict[int, Group]: """The groups still alive after one shared wait, so batches share a deadline.""" deadline = asyncio.get_running_loop().time() + window while True: remaining = {pgid: group for pgid, group in live.items() if alive(group)} if not remaining or asyncio.get_running_loop().time() >= deadline: return remaining await asyncio.sleep(PROBE_INTERVAL)
async def terminate( groups: Sequence[Group], term: float = TERM_GRACE, kill: float = KILL_GRACE) -> list[Termination]: """End every group: TERM, then KILL, one shared deadline per step.""" live = {group.pgid: group for group in groups if alive(group)} sent: dict[int, list[str]] = {pgid: [] for pgid in live} failures: dict[int, list[str]] = {pgid: [] for pgid in live} for sig, window in ((signal.SIGTERM, term), (signal.SIGKILL, kill)): for pgid, group in live.items(): failure = send_signal(group, sig) sent[pgid].append(signal.Signals(sig).name) if failure is not None: failures[pgid].append(failure) live = await settled(live, window) if not live: break endings = [] for group in groups: pgid, signals = group.pgid, sent.get(group.pgid, []) surviving, note = pgid in live, "" if surviving and not current(group): surviving, note = False, "only non-running members remained" endings.append( Termination( pgid, tuple(signals), not surviving, tuple(failures.get(pgid, [])), note ) ) return endings