src/agent_process_dashboard/app.py

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()