from __future__ import annotations import threading import time import gi gi.require_version("Gtk", "3.0") from gi.repository import GLib, Gtk, Pango from .actions import ( ProcessActionRequest, same_process_is_running, send_signal_to_candidate, ) from .models import ( DashboardSnapshot, DryRunCandidate, ProcessGroup, ProcessNode, ProcessSample, ) from .snapshot import build_snapshot, samples_by_key CSS = b""" window { background: #111827; color: #e5e7eb; } .topbar { background: #0f172a; border-bottom: 1px solid #263244; padding: 10px; } .title { color: #f8fafc; font-size: 18px; font-weight: 700; } .muted { color: #9ca3af; } .panel { background: #111827; padding: 10px; } .stat { background: #182235; border: 1px solid #2f3b4f; border-radius: 6px; padding: 8px; } .stat-value { color: #f8fafc; font-size: 20px; font-weight: 700; } .risk-ok { color: #86efac; } .risk-info { color: #7dd3fc; } .risk-warn { color: #fbbf24; } .risk-critical { color: #f87171; } button.danger { background: #7f1d1d; color: #fee2e2; border-color: #991b1b; } treeview, textview, list { background: #101827; color: #e5e7eb; } """ INTRO_TEXT = ( "Dieses Dashboard liest Prozessdaten aus /proc, zeigt Agent-, MCP- und " "Bridge-Prozesse sowie alte oder auffällige Kandidaten. Stop/Kill-Aktionen " "sind nur für aktuelle Dry-run-Kandidaten verfügbar und erfordern eine " "Bestätigung mit frischem PID-Recheck." ) SAFETY_NOTE = "Signale werden nur nach Bestätigung an aktuelle Dry-run-Kandidaten gesendet." HELP_TEXT = """Was macht das Dashboard? Es zeigt lokale Prozessbäume für den aktuellen Benutzer. Der Fokus liegt auf Agent-Prozessen, MCP-Servern und Bridge-Prozessen. Zusätzlich markiert es alte oder auffällige Kandidaten. Verändert es etwas? Beim Scannen verändert das Dashboard nichts. Wenn du in der Kandidatenliste bewusst Stop oder Kill erzwingen bestätigst, sendet es ein Signal an genau diese PID. Direkt vor dem Signal wird erneut geprüft, dass PID und Prozess-Schlüssel noch zu einem aktuellen Dry-run-Kandidaten gehören. Es startet nichts neu und ändert keine Konfigurationen. Begriffe: • MCP: Model Context Protocol. Solche Prozesse stellen Tools oder Datenquellen für Agenten bereit. • Bridge: Ein Verbindungsprozess zwischen Agent, Tool, MCP oder Shell. • Fanout: Anzahl der Kindprozesse unter einem Prozess. Hoher Fanout heißt: sehr viele Unterprozesse. • Stale: Wirkt alt oder nicht mehr aktiv. Das ist nur ein Hinweis, kein Beweis. • Critical: Auffällig hohe Risikostufe nach den Heuristiken. Bitte manuell prüfen. • PID: Prozess-ID. PPID: Prozess-ID des Elternprozesses. • RSS: aktuell belegter Arbeitsspeicher. CWD: Arbeitsordner. Command: Startbefehl. • Risk: zusammengefasste Auffälligkeit. Stale: Einschätzung, ob ein Prozess alt wirkt. Was tun bei vielen Kandidaten? Nicht blind beenden. Prüfe zuerst PID, Command und CWD. Stop kann laufende Agenten- oder Tool-Arbeit abbrechen; Kill erzwingen kann nicht abgefangen werden und erhöht das Risiko für verlorene Zwischenstände.""" KPI_TOOLTIPS = { "agent_roots": "Anzahl der gefundenen Agent-Wurzelprozesse. Das sind Startpunkte der angezeigten Agent-Prozessbäume.", "bridge_count": "Anzahl der erkannten Bridge-Prozesse. Bridges verbinden oft Agent, Tool, MCP oder Shell.", "mcp_count": "Anzahl der erkannten MCP-Prozesse. MCP stellt Agenten Tools oder Datenquellen bereit.", "max_fanout": "Größte Anzahl von Kindprozessen in einem Prozessbaum. Ein hoher Wert kann auf viel Aktivität oder Hänger hinweisen.", "stale_candidates": "Anzahl der Prozesse, die alt oder inaktiv wirken. Das ist nur ein Prüfhilfe-Hinweis.", "cpu_hot": "Anzahl der Prozesse mit auffälliger CPU-Aktivität im letzten Messintervall.", } FILTER_TOOLTIPS = { "agent": "Zeigt nur Gruppen, deren Wurzel als Agent erkannt wurde.", "mcp": "Zeigt nur Gruppen, in denen mindestens ein MCP-Prozess vorkommt.", "bridge": "Zeigt nur Gruppen, in denen mindestens ein Bridge-Prozess vorkommt.", "stale": "Zeigt nur Gruppen mit Prozessen, die alt oder inaktiv wirken. Es wird nichts beendet.", "fanout": "Zeigt nur Gruppen mit vielen Kindprozessen. Das hilft beim Finden großer Prozessbäume.", } DETAILS_HINT = ( "PID=Prozess-ID, PPID=Elternprozess, RSS=Arbeitsspeicher, " "CWD=Arbeitsordner. Risk und Stale sind nur Hinweise." ) def _label( text: str = "", *, css: str | None = None, xalign: float = 0.0, wrap: bool = False ) -> Gtk.Label: label = Gtk.Label(label=text, xalign=xalign) if wrap: label.set_line_wrap(True) label.set_line_wrap_mode(Pango.WrapMode.WORD_CHAR) else: label.set_ellipsize(Pango.EllipsizeMode.END) if css: label.get_style_context().add_class(css) return label def _set_tooltip(widget: Gtk.Widget, text: str) -> None: widget.set_tooltip_text(text) def _join_or_none(items: list[str] | tuple[str, ...]) -> str: return ", ".join(items) or "keine" def _format_age(seconds: float) -> str: if seconds < 60: return "<1m" minutes = int(seconds // 60) if minutes < 60: return f"{minutes}m" hours = minutes // 60 if hours < 48: return f"{hours}h" return f"{hours // 24}d" class AgentProcessDashboard(Gtk.Window): def __init__(self) -> None: super().__init__(title="Agent Process Dashboard") self.set_default_size(1180, 760) self.set_wmclass("agent-process-dashboard", "AgentProcessDashboard") self._snapshot: DashboardSnapshot | None = None self._previous: dict[str, ProcessSample] = {} self._scan_running = False self._last_change_ts = time.time() self._stat_labels: dict[str, Gtk.Label] = {} self._group_rows: dict[str, Gtk.ListBoxRow] = {} self._selected_group_id: str | None = None self._node_by_tree_path: dict[str, ProcessNode] = {} self._force_kill_keys: set[str] = set() self._filter_agent = Gtk.CheckButton(label="Agent") self._filter_mcp = Gtk.CheckButton(label="MCP") self._filter_bridge = Gtk.CheckButton(label="Bridge") self._filter_stale = Gtk.CheckButton(label="Alt/Stale") self._filter_fanout = Gtk.CheckButton(label="Viele Kindprozesse") _set_tooltip(self._filter_agent, FILTER_TOOLTIPS["agent"]) _set_tooltip(self._filter_mcp, FILTER_TOOLTIPS["mcp"]) _set_tooltip(self._filter_bridge, FILTER_TOOLTIPS["bridge"]) _set_tooltip(self._filter_stale, FILTER_TOOLTIPS["stale"]) _set_tooltip(self._filter_fanout, FILTER_TOOLTIPS["fanout"]) for button in ( self._filter_agent, self._filter_mcp, self._filter_bridge, self._filter_stale, self._filter_fanout, ): button.connect("toggled", lambda *_: self._refresh_views()) self._group_list = Gtk.ListBox() self._group_list.connect("row-selected", self._on_group_selected) self._tree_store = Gtk.TreeStore(str, int, str, str, str, str, str, str) self._tree_view = Gtk.TreeView(model=self._tree_store) self._tree_view.get_selection().connect("changed", self._on_tree_selected) self._setup_tree_columns() self._details = Gtk.TextView() self._details.set_editable(False) self._details.set_wrap_mode(Gtk.WrapMode.WORD_CHAR) self._dryrun_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) self._dryrun_box.set_margin_top(8) self._dryrun_box.set_margin_bottom(8) self._dryrun_box.set_margin_start(8) self._dryrun_box.set_margin_end(8) self._action_status = _label("", css="muted", wrap=True) self._status = _label("Starting", css="muted") self._setup_css() self._build_ui() self.connect("destroy", Gtk.main_quit) GLib.timeout_add_seconds(2, self._tick) self._start_scan() def _setup_css(self) -> None: provider = Gtk.CssProvider() provider.load_from_data(CSS) screen = Gtk.Window.get_screen(self) Gtk.StyleContext.add_provider_for_screen( screen, provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION ) def _setup_tree_columns(self) -> None: columns = [ ("Process", 0, 280), ("PID", 1, 70), ("Class", 2, 150), ("Age", 3, 70), ("RSS", 4, 80), ("CPU", 5, 70), ("Risk", 6, 90), ] for title, index, width in columns: renderer = Gtk.CellRendererText() if index == 0: renderer.set_property("ellipsize", Pango.EllipsizeMode.END) column = Gtk.TreeViewColumn(title, renderer, text=index) column.set_resizable(True) column.set_min_width(width) self._tree_view.append_column(column) def _build_ui(self) -> None: root = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) self.add(root) topbar = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10) topbar.get_style_context().add_class("topbar") root.pack_start(topbar, False, False, 0) topbar.pack_start(_label("Agent Process Dashboard", css="title"), False, False, 0) topbar.pack_start(self._status, True, True, 0) refresh = Gtk.Button(label="Refresh") _set_tooltip(refresh, "Liest /proc neu ein. Es wird nichts beendet oder verändert.") refresh.connect("clicked", lambda *_: self._start_scan()) help_button = Gtk.Button(label="Hilfe") _set_tooltip(help_button, "Erklärt die wichtigsten Begriffe und das Sicherheitsmodell.") help_button.connect("clicked", self._show_help) topbar.pack_end(refresh, False, False, 0) topbar.pack_end(help_button, False, False, 0) intro = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) intro.get_style_context().add_class("panel") intro_label = _label(INTRO_TEXT, css="muted", wrap=True) _set_tooltip(intro_label, INTRO_TEXT) intro.pack_start(intro_label, False, False, 0) root.pack_start(intro, False, False, 0) stats = Gtk.Grid(column_spacing=8, row_spacing=8) stats.get_style_context().add_class("panel") root.pack_start(stats, False, False, 0) stat_names = [ ("agent_roots", "Agent Roots"), ("bridge_count", "Bridges"), ("mcp_count", "MCP"), ("max_fanout", "Max Fanout"), ("stale_candidates", "Stale"), ("cpu_hot", "CPU Hot"), ] for col, (key, title) in enumerate(stat_names): box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=3) box.get_style_context().add_class("stat") _set_tooltip(box, KPI_TOOLTIPS[key]) value = _label("0", css="stat-value") title_label = _label(title, css="muted") _set_tooltip(value, KPI_TOOLTIPS[key]) _set_tooltip(title_label, KPI_TOOLTIPS[key]) box.pack_start(value, False, False, 0) box.pack_start(title_label, False, False, 0) stats.attach(box, col, 0, 1, 1) self._stat_labels[key] = value filters = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=12) filters.get_style_context().add_class("panel") for button in ( self._filter_agent, self._filter_mcp, self._filter_bridge, self._filter_stale, self._filter_fanout, ): filters.pack_start(button, False, False, 0) root.pack_start(filters, False, False, 0) main = Gtk.Paned(orientation=Gtk.Orientation.HORIZONTAL) root.pack_start(main, True, True, 0) left_scroller = Gtk.ScrolledWindow() left_scroller.set_min_content_width(260) left_scroller.add(self._group_list) main.pack1(left_scroller, resize=False, shrink=False) right = Gtk.Paned(orientation=Gtk.Orientation.HORIZONTAL) main.pack2(right, resize=True, shrink=False) tree_scroller = Gtk.ScrolledWindow() tree_scroller.add(self._tree_view) right.pack1(tree_scroller, resize=True, shrink=False) side = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) side.set_size_request(340, -1) details_scroller = Gtk.ScrolledWindow() details_scroller.add(self._details) dry_scroller = Gtk.ScrolledWindow() dry_scroller.add_with_viewport(self._dryrun_box) side.pack_start(_label("Details", css="title"), False, False, 8) side.pack_start(_label(DETAILS_HINT, css="muted", wrap=True), False, False, 0) side.pack_start(details_scroller, True, True, 0) side.pack_start(_label("Möglicherweise alte Prozesse", css="title"), False, False, 8) side.pack_start(_label(SAFETY_NOTE, css="muted", wrap=True), False, False, 0) side.pack_start(self._action_status, False, False, 0) side.pack_start(dry_scroller, True, True, 0) right.pack2(side, resize=False, shrink=False) def _show_help(self, *_args: object) -> None: dialog = Gtk.MessageDialog( transient_for=self, flags=Gtk.DialogFlags.MODAL, message_type=Gtk.MessageType.INFO, buttons=Gtk.ButtonsType.CLOSE, text="Hilfe zum Agent Process Dashboard", ) dialog.format_secondary_text(HELP_TEXT) dialog.connect("response", lambda widget, *_: widget.destroy()) dialog.show_all() def _tick(self) -> bool: idle_seconds = time.time() - self._last_change_ts if idle_seconds < 30: self._start_scan() return True if int(idle_seconds) % 6 == 0: self._start_scan() return True def _start_scan(self) -> None: if self._scan_running: return self._scan_running = True self._status.set_text("Scanning /proc") def worker() -> None: try: snapshot = build_snapshot(previous=self._previous) GLib.idle_add(self._finish_scan, snapshot, None) except Exception as exc: # pragma: no cover - UI safety net GLib.idle_add(self._finish_scan, None, exc) threading.Thread(target=worker, daemon=True).start() def _finish_scan(self, snapshot: DashboardSnapshot | None, error: Exception | None) -> bool: self._scan_running = False if error is not None: self._status.set_text(f"Scan failed: {error}") return False if snapshot is None: return False old_count = len(self._snapshot.processes) if self._snapshot else -1 self._snapshot = snapshot self._previous = samples_by_key(snapshot) if len(snapshot.processes) != old_count: self._last_change_ts = time.time() self._status.set_text(f"Last scan: {time.strftime('%H:%M:%S')}") self._refresh_views() return False def _refresh_views(self) -> None: snapshot = self._snapshot if snapshot is None: return for key, label in self._stat_labels.items(): label.set_text(str(snapshot.stats.get(key, 0))) self._populate_groups(snapshot.groups) self._populate_tree() self._populate_dryrun(snapshot) def _populate_groups(self, groups: list[ProcessGroup]) -> None: for child in self._group_list.get_children(): self._group_list.remove(child) self._group_rows.clear() for group in groups: if not self._group_matches_filters(group): continue row = Gtk.ListBoxRow() row.group_id = group.group_id # type: ignore[attr-defined] box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) box.set_margin_top(8) box.set_margin_bottom(8) box.set_margin_start(8) box.set_margin_end(8) box.pack_start(_label(group.label), False, False, 0) box.pack_start( _label( f"{group.risk_level.upper()} - {group.child_count} Kindprozesse - " f"{group.bridge_count} Bridges - {_format_age(group.max_age_seconds)} alt", css=f"risk-{group.risk_level}", ), False, False, 0, ) row.add(box) self._group_list.add(row) self._group_rows[group.group_id] = row self._group_list.show_all() def _group_matches_filters(self, group: ProcessGroup) -> bool: active = any( button.get_active() for button in ( self._filter_agent, self._filter_mcp, self._filter_bridge, self._filter_stale, self._filter_fanout, ) ) if not active: return True return ( (self._filter_agent.get_active() and group.root.classification.is_agent) or (self._filter_mcp.get_active() and group.mcp_count > 0) or (self._filter_bridge.get_active() and group.bridge_count > 0) or (self._filter_stale.get_active() and bool(group.stale_candidates)) or (self._filter_fanout.get_active() and group.child_count >= 20) ) def _on_group_selected(self, _listbox: Gtk.ListBox, row: Gtk.ListBoxRow | None) -> None: self._selected_group_id = getattr(row, "group_id", None) if row else None self._populate_tree() def _populate_tree(self) -> None: snapshot = self._snapshot if snapshot is None: return self._tree_store.clear() self._node_by_tree_path.clear() groups = [ group for group in snapshot.groups if not self._selected_group_id or group.group_id == self._selected_group_id ] if self._selected_group_id is None: groups = [group for group in groups if self._group_matches_filters(group)] for group in groups: self._append_node(group.root, None) if groups: self._tree_view.expand_to_path(Gtk.TreePath.new_first()) def _append_node(self, node: ProcessNode, parent_iter) -> None: sample = node.sample iterator = self._tree_store.append( parent_iter, [ sample.comm, sample.pid, node.classification.known_class, _format_age(sample.age_seconds), f"{sample.rss_mb:.1f}M", f"{sample.cpu_percent_interval:.1f}%", node.risk_level, sample.process_key, ], ) path = self._tree_store.get_path(iterator).to_string() self._node_by_tree_path[path] = node for child in node.children: self._append_node(child, iterator) def _on_tree_selected(self, selection: Gtk.TreeSelection) -> None: model, iterator = selection.get_selected() if iterator is None: return path = model.get_path(iterator).to_string() node = self._node_by_tree_path.get(path) if node: self._show_details(node) def _show_details(self, node: ProcessNode) -> None: sample = node.sample lines = [ f"PID (Prozess-ID): {sample.pid}", f"PPID (Elternprozess-ID): {sample.ppid}", f"Prozess-Schlüssel: {sample.process_key}", f"Einordnung: {node.classification.known_class} ({node.classification.confidence:.2f})", f"Warum diese Einordnung: {_join_or_none(node.classification.reasons)}", f"Status: {sample.state}", f"Alter: {_format_age(sample.age_seconds)}", f"RSS (Arbeitsspeicher): {sample.rss_mb:.1f} MB", f"CPU im letzten Intervall: {sample.cpu_percent_interval:.1f}%", f"Threads: {sample.num_threads}", f"CWD (Arbeitsordner): {sample.cwd or '(nicht lesbar)'}", f"Command (Startbefehl): {sample.argv_display}", f"Verdeckte Argumente: {sample.argv_masked_count}", f"Risk (Risiko-Hinweis): {node.risk_level} - {_join_or_none(node.risk_reasons)}", f"Stale (wirkt alt/inaktiv): {node.stale_state} - {_join_or_none(node.stale_reasons)}", ] self._details.get_buffer().set_text("\n".join(lines)) def _populate_dryrun(self, snapshot: DashboardSnapshot) -> None: report = snapshot.dry_run for child in self._dryrun_box.get_children(): self._dryrun_box.remove(child) candidate_keys = {candidate.process_key for candidate in report.candidates} self._force_kill_keys.intersection_update(candidate_keys) if not report.candidates: self._dryrun_box.pack_start( _label("Aktuell keine alten MCP-/Bridge-Kandidaten gefunden.", wrap=True), False, False, 0, ) else: self._dryrun_box.pack_start( _label( "Prüfe PID, Command und CWD vor jeder Aktion. Stop sendet SIGTERM; " "Kill erzwingen erscheint nur, wenn derselbe Prozess danach noch läuft.", css="muted", wrap=True, ), False, False, 0, ) for candidate in report.candidates[:40]: self._dryrun_box.pack_start( self._candidate_widget(candidate), False, False, 0, ) if len(report.candidates) > 40: self._dryrun_box.pack_start( _label( f"{len(report.candidates) - 40} weitere Kandidaten ausgeblendet.", css="muted", wrap=True, ), False, False, 0, ) if report.exclusions: self._dryrun_box.pack_start( _label("Ausgelassen: " + " | ".join(report.exclusions[:8]), css="muted", wrap=True), False, False, 0, ) self._dryrun_box.show_all() def _candidate_widget(self, candidate: DryRunCandidate) -> Gtk.Widget: box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) box.get_style_context().add_class("stat") header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) title = _label( f"PID {candidate.pid} - {candidate.known_class} - " f"{_format_age(candidate.age_seconds)} alt - " f"{candidate.rss_bytes / 1024 / 1024:.1f}M RSS", wrap=True, ) header.pack_start(title, True, True, 0) stop = Gtk.Button(label="Stop") stop.get_style_context().add_class("danger") _set_tooltip(stop, "Sendet nach Bestätigung SIGTERM an diesen aktuellen Kandidaten.") stop.connect( "clicked", self._confirm_process_signal, candidate.pid, candidate.process_key, "TERM", ) header.pack_end(stop, False, False, 0) if candidate.process_key in self._force_kill_keys: kill = Gtk.Button(label="Kill erzwingen") kill.get_style_context().add_class("danger") _set_tooltip(kill, "Sendet nach Bestätigung SIGKILL, wenn der Kandidat noch aktuell ist.") kill.connect( "clicked", self._confirm_process_signal, candidate.pid, candidate.process_key, "KILL", ) header.pack_end(kill, False, False, 0) box.pack_start(header, False, False, 0) box.pack_start(_label(f"Schlüssel: {candidate.process_key}", css="muted", wrap=True), False, False, 0) box.pack_start( _label(f"Hinweise: {_join_or_none(candidate.reasons)}", css="muted", wrap=True), False, False, 0, ) box.pack_start(_label(f"Befehl: {candidate.command}", css="muted", wrap=True), False, False, 0) return box def _confirm_process_signal( self, _button: Gtk.Button, pid: int, process_key: str, signal_name: str, ) -> None: hard_kill = signal_name == "KILL" title = f"{'SIGKILL' if hard_kill else 'SIGTERM'} an PID {pid} senden?" action_label = "Kill erzwingen" if hard_kill else "Stop senden" consequence = ( "SIGKILL kann nicht abgefangen werden. Der Prozess endet sofort, " "wodurch laufende Agenten- oder Tool-Arbeit abbrechen und Zwischenstände " "verloren gehen können." if hard_kill else "SIGTERM bittet den Prozess, sauber zu beenden. Laufende Agenten- " "oder Tool-Arbeit kann trotzdem abbrechen; Kindprozesse können je nach " "Programm weiterlaufen." ) dialog = Gtk.MessageDialog( transient_for=self, flags=Gtk.DialogFlags.MODAL, message_type=Gtk.MessageType.WARNING, buttons=Gtk.ButtonsType.CANCEL, text=title, ) dialog.add_button(action_label, Gtk.ResponseType.OK) dialog.format_secondary_text( "Direkt vor dem Signal prüft das Dashboard erneut, dass PID und " "Prozess-Schlüssel noch zu einem aktuellen Dry-run-Kandidaten gehören.\n\n" f"Prozess-Schlüssel: {process_key}\n\n" f"Mögliche Folgen: {consequence}" ) response = dialog.run() dialog.destroy() if response == Gtk.ResponseType.OK: self._run_process_action(ProcessActionRequest(pid, process_key, signal_name)) def _run_process_action(self, request: ProcessActionRequest) -> None: result = send_signal_to_candidate(request) self._action_status.set_text(result.message) if result.ok and request.signal_name == "TERM": self._force_kill_keys.discard(request.process_key) GLib.timeout_add_seconds(3, self._check_after_term, request.pid, request.process_key) elif result.ok and request.signal_name == "KILL": self._force_kill_keys.discard(request.process_key) self._start_scan() def _check_after_term(self, pid: int, process_key: str) -> bool: if same_process_is_running(pid, process_key): self._force_kill_keys.add(process_key) self._action_status.set_text( f"PID {pid} läuft nach SIGTERM noch. Kill erzwingen ist jetzt verfügbar." ) else: self._force_kill_keys.discard(process_key) self._action_status.set_text(f"PID {pid} läuft nicht mehr.") self._refresh_views() self._start_scan() return False def run() -> None: window = AgentProcessDashboard() window.show_all() Gtk.main()