#!/usr/bin/env python3 """ Long-lived actionable-notification daemon for agent-reaper WARN candidates. agent-reaper.py (the 60s timer) only WRITES flagged candidates to the state file; this daemon watches that state and renders ONE persistent, actionable desktop notification per NEWLY-flagged pid (dedup by pid; re-notifies only if the pid drops out of the flagged set and later reappears). """ import os import shutil import subprocess import sys sys.dont_write_bytecode = True sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import _agent_reaper_lib as lib resolve_session_mod = None def _load_resolve_session(): global resolve_session_mod if resolve_session_mod is None: spec_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "resolve-session.py") import importlib.util spec = importlib.util.spec_from_file_location("resolve_session_mod", spec_path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) resolve_session_mod = mod return resolve_session_mod POLL_SECONDS = 5 GATE_PY = os.environ.get("NOTIF_GATE_PY", "/usr/local/lib/notif-gate/notif_gate.py") EMITTER = os.path.realpath(__file__) def gate_allows(summary, body): """Default-deny: only an owner-approved source may reach the session bus. Anything other than a clean approval — gate absent, crashed, or slow — denies and leaves the daemon otherwise untouched. """ try: proc = subprocess.run( ["python3", GATE_PY, "check", EMITTER, summary, body, "dbus-gated"], capture_output=True, text=True, timeout=5, check=False, ) except (OSError, subprocess.SubprocessError) as exc: print(f"reaper-notifier: gate unavailable ({exc}); suppressing", file=sys.stderr) return False return proc.returncode == 0 def should_desktop_notify(candidate): return candidate.get("kind") != "leaked-test-orphan" def build_test_orphan_payload(pid, candidate): age_m = (candidate.get("age_s") or 0) / 60 comm = candidate.get("comm", "?") reason = candidate.get("reason", "?") cmdline = (candidate.get("cmdline") or "").strip() summary = f"🧟 leaked test process? PID {pid}" body = (f"orphaned {age_m:.0f}m ago · {comm}\n" f"{reason}\n" f"{cmdline[:80]}") actions = [("reap", "reap tree"), ("dismiss", "dismiss")] return {"summary": summary, "body": body, "actions": actions, "transcript_path": None} def build_notification_payload(pid, candidate, session): if candidate.get("kind") == "leaked-test-orphan": return build_test_orphan_payload(pid, candidate) age_m = (candidate.get("transcript_age_s") or 0) / 60 cpu = candidate.get("pcpu_now") or 0 comm = candidate.get("comm", "?") summary = f"⚠ runaway agent? PID {pid}" body = (f"{session['title']}\n" f"cwd: {session['cwd'] or '?'}\n" f"stale {age_m:.0f}m · {cpu:.0f}% · {comm}") actions = [("kill", "kill"), ("log", "log")] return {"summary": summary, "body": body, "actions": actions, "transcript_path": session.get("transcript_path")} def _run_reaper_ctl(subcommand, pid): reaper_ctl = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reaper-ctl") from gi.repository import Gio proc = Gio.Subprocess.new([reaper_ctl, subcommand, str(pid)], Gio.SubprocessFlags.NONE) proc.wait_async(None, lambda p, r: p.wait_finish(r)) def run_kill_action(pid): _run_reaper_ctl("kill", pid) def run_reap_action(pid): _run_reaper_ctl("reap", pid) def run_log_action(transcript_path): from gi.repository import Gio if not transcript_path: return opener = "gedit" if shutil.which("gedit") else ("xdg-open" if shutil.which("xdg-open") else None) if opener is None: return proc = Gio.Subprocess.new([opener, transcript_path], Gio.SubprocessFlags.NONE) proc.wait_async(None, lambda p, r: p.wait_finish(r)) class Notifier: def __init__(self): import gi gi.require_version("Notify", "0.7") from gi.repository import Notify, GLib self.Notify = Notify self.GLib = GLib Notify.init("agent-reaper") self.active = {} def poll(self): state = lib.load_state() flagged = state.get("flagged", {}) for pid_key in list(self.active.keys()): if pid_key not in flagged: notif = self.active.pop(pid_key) try: notif.close() except Exception: pass for pid_key, candidate in flagged.items(): if pid_key in self.active: continue if not should_desktop_notify(candidate): continue if candidate.get("kind") == "leaked-test-orphan": session = None # orphans have no agent session/transcript else: session = _load_resolve_session().resolve_session(int(pid_key)) payload = build_notification_payload(pid_key, candidate, session) self._raise(pid_key, payload) return True def _raise(self, pid_key, payload): notif = self.Notify.Notification.new(payload["summary"], payload["body"]) notif.set_urgency(self.Notify.Urgency.CRITICAL) notif.set_timeout(0) for action_id, label in payload["actions"]: notif.add_action(action_id, label, self._on_action, payload) self.active[pid_key] = notif if not gate_allows(payload["summary"], payload["body"]): return try: notif.show() except Exception as e: print(f"reaper-notifier: failed to show notification for {pid_key}: {e}", file=sys.stderr) def _on_action(self, notif, action_id, payload): pid_key = None for k, v in list(self.active.items()): if v is notif: pid_key = k break if action_id == "kill" and pid_key is not None: run_kill_action(pid_key) notif.close() elif action_id == "reap" and pid_key is not None: run_reap_action(pid_key) notif.close() elif action_id == "dismiss" and pid_key is not None: notif.close() elif action_id == "log": run_log_action(payload.get("transcript_path")) def run(self): self.GLib.timeout_add_seconds(POLL_SECONDS, self.poll) self.poll() loop = self.GLib.MainLoop() loop.run() def self_test(): import json import shutil as _shutil import tempfile tmpdir = tempfile.mkdtemp(prefix="reaper-notifier-selftest-") try: transcript_path = os.path.join(tmpdir, "fake.jsonl") with open(transcript_path, "w") as f: f.write(json.dumps({"type": "summary", "summary": "Investigate flaky deploy job"}) + "\n") session = {"transcript_path": transcript_path, "title": "Investigate flaky deploy job", "cwd": "/home/user/Projects/thing"} candidate = {"transcript_age_s": 50 * 60, "pcpu_now": 87.0, "comm": "claude", "ppid": 1, "tty_nr": 0, "verdict": "ENFORCE_ELIGIBLE"} payload = build_notification_payload("4242", candidate, session) assert payload["summary"] == "⚠ runaway agent? PID 4242", payload["summary"] assert "Investigate flaky deploy job" in payload["body"] assert "cwd: /home/user/Projects/thing" in payload["body"] assert "stale 50m" in payload["body"] assert "87%" in payload["body"] assert "claude" in payload["body"] assert payload["actions"] == [("kill", "kill"), ("log", "log")] assert payload["transcript_path"] == transcript_path rs = _load_resolve_session() with open(transcript_path) as f: pass resolved_title = rs.title_from_transcript(transcript_path) assert resolved_title == "Investigate flaky deploy job", resolved_title no_session = {"transcript_path": None, "title": "thing", "cwd": None} no_candidate = {"transcript_age_s": None, "pcpu_now": None, "comm": "claude"} payload2 = build_notification_payload("1", no_candidate, no_session) assert "cwd: ?" in payload2["body"] # leaked-test-orphan payload: distinct summary + reap action, no session needed orphan = {"kind": "leaked-test-orphan", "age_s": 22 * 60, "comm": "bash", "reason": "ci-integration-http-signal-harness", "cmdline": "bash .wt-plan-int/apps/web/scripts/ci-integration-http-signal-harness.sh int"} po = build_notification_payload("777", orphan, None) assert po["summary"] == "🧟 leaked test process? PID 777", po["summary"] assert "orphaned 22m ago" in po["body"] assert "ci-integration-http-signal-harness" in po["body"] assert po["actions"] == [("reap", "reap tree"), ("dismiss", "dismiss")] assert po["transcript_path"] is None assert not should_desktop_notify(orphan), "idle test remnants must stay off desktop" assert should_desktop_notify(candidate), "runaway agents must remain actionable" class ClosedNotification: def __init__(self): self.closed = False def close(self): self.closed = True notifier = Notifier.__new__(Notifier) dismissed = ClosedNotification() notifier.active = {"777": dismissed} notifier._on_action(dismissed, "dismiss", {}) assert dismissed.closed assert "777" in notifier.active, "dismissed candidate must stay deduped until scanner clears it" print("reaper-notifier self-test OK (no notification raised, no process spawned)", file=sys.stderr) return True finally: _shutil.rmtree(tmpdir, ignore_errors=True) if __name__ == "__main__": if "--self-test" in sys.argv: sys.exit(0 if self_test() else 1) Notifier().run()