ClaudeCodePanel / process_manager.py
smlflg's picture
Initial public upload from Projekte/ClaudeCodePanel
bbc65cd verified
Raw History Blame Contribute Delete
20.9 kB
#!/usr/bin/env python3
"""Process Manager — GTK3 widget module for Claude Code Control Panel.
Provides build_processes_tab() returning a Gtk.ScrolledWindow with:
- List of all Claude-relevant processes (opencode-mcp, gemini-mcp, claude CLI, etc.)
- Per-process: PID, name, RAM (MB), CPU%, uptime, start date
- Kill button per process row with confirmation dialog
- "Kill All Ghosts" button (kills processes older than 24h)
- Summary stats at top: total processes, total RAM, ghost count
- Auto-refresh every 30s via GLib timer
Theme: Catppuccin Mocha (dark) / Latte (light) via theme.py
"""
import logging
import os
import signal
import subprocess
import threading
import time
from datetime import datetime
from pathlib import Path
log = logging.getLogger("claude_panel.process_manager")
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import GLib, Gtk, Pango
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
from theme import get_palette, hex_to_pango_rgb
from utils import idle_once, short_name_from_path, HOME, CLK_TCK
GHOST_HOURS = 24 # processes older than this are "ghosts"
# Process name keywords to detect Claude-relevant processes
PROCESS_KEYWORDS = [
"opencode-mcp",
"opencode",
"gemini-mcp",
"server-filesystem",
"server-memory",
"server-github",
"server-postgres",
"claude",
"codex",
"mcp-server",
]
# ---------------------------------------------------------------------------
# Module-level state (kept between refreshes)
# ---------------------------------------------------------------------------
_list_box: Gtk.ListBox | None = None
_stats_label: Gtk.Label | None = None
_all_processes: list[dict] = []
# ---------------------------------------------------------------------------
# Process scanning
# ---------------------------------------------------------------------------
def _scan_processes() -> list[dict]:
"""Run ps aux and filter for Claude-relevant processes.
Returns list of dicts with: pid, name, cpu, ram_mb, start_str, uptime_h, is_ghost
"""
processes = []
try:
result = subprocess.run(
["ps", "aux", "--no-headers"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode != 0:
return processes
now = time.time()
for line in result.stdout.splitlines():
parts = line.split(None, 10)
if len(parts) < 11:
continue
# ps aux columns: USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND
try:
pid = int(parts[1])
cpu = float(parts[2])
rss_kb = int(parts[5])
tty = parts[6] # e.g. "pts/33" or "?"
start_str = parts[8] # e.g. "10:30" or "Feb12"
time_str = parts[9] # cumulative CPU time e.g. "0:00"
command = parts[10]
except (ValueError, IndexError):
continue
# Check if this process matches any keyword
cmd_lower = command.lower()
matched_keyword = None
for kw in PROCESS_KEYWORDS:
if kw in cmd_lower:
matched_keyword = kw
break
if matched_keyword is None:
continue
# Skip self (the ps command or this Python process)
if pid == os.getpid():
continue
# Derive display name from command
name = _derive_name(command, matched_keyword, pid)
# RAM in MB
ram_mb = rss_kb / 1024
# Estimate process age via /proc/<pid>/stat if available
uptime_h = _get_uptime_hours(pid, now)
is_ghost = uptime_h >= GHOST_HOURS
processes.append({
"pid": pid,
"name": name,
"command": command[:80],
"cpu": cpu,
"ram_mb": ram_mb,
"tty": tty if tty != "?" else "",
"start_str": start_str,
"uptime_h": uptime_h,
"is_ghost": is_ghost,
})
except (subprocess.TimeoutExpired, OSError):
pass
# Sort: ghosts first, then by RAM desc
processes.sort(key=lambda p: (-p["is_ghost"], -p["ram_mb"]))
return processes
_FRIENDLY_NAMES = {
"opencode-mcp": "OpenCode MCP",
"opencode": "OpenCode Server",
"gemini-mcp": "Gemini MCP",
"server-filesystem": "Filesystem MCP",
"server-memory": "Memory MCP",
"server-github": "GitHub MCP",
"codex": "Codex CLI",
"mcp-server": "MCP Server",
}
def _get_cwd_project(pid: int) -> str:
"""Try to read the CWD of a process and extract a project name."""
try:
cwd = os.readlink(f"/proc/{pid}/cwd")
name = short_name_from_path(cwd)
return "" if name == "Home" else name
except OSError:
return ""
def _derive_name(command: str, keyword: str, pid: int = 0) -> str:
"""Derive a friendly display name from the full command string."""
# Special handling for claude CLI — try to show project context
if keyword == "claude":
# Check if resumed session (-r flag)
resumed = " -r" in command or "\x00-r" in command
project = _get_cwd_project(pid) if pid else ""
if project:
return f"Claude CLI — {project}"
return "Claude CLI (resumed)" if resumed else "Claude CLI"
# Check friendly name table (normalize mcp- prefix for lookup)
lookup = keyword.removeprefix("mcp-") if keyword not in _FRIENDLY_NAMES else keyword
if lookup in _FRIENDLY_NAMES:
return _FRIENDLY_NAMES[lookup]
# Fallback: executable basename
parts = command.split()
if not parts:
return keyword
exe_base = Path(parts[0]).name
name_map = {
"node": f"node/{keyword}",
"python3": f"python/{keyword}",
"python": f"python/{keyword}",
}
if exe_base in name_map:
return name_map[exe_base]
return exe_base if exe_base else keyword
def _get_uptime_hours(pid: int, now: float) -> float:
"""Estimate process uptime in hours via /proc/<pid>/stat."""
try:
with open(f"/proc/{pid}/stat") as f:
stat_data = f.read().split()
# Field 22 (index 21) is starttime in clock ticks since boot
starttime_ticks = int(stat_data[21])
clk_tck = CLK_TCK
with open("/proc/uptime") as f:
boot_seconds_ago = float(f.read().split()[0])
# Process start time as seconds since epoch
process_start = now - boot_seconds_ago + (starttime_ticks / clk_tck)
uptime_seconds = now - process_start
return max(0.0, uptime_seconds / 3600)
except (OSError, ValueError, IndexError):
return 0.0
# ---------------------------------------------------------------------------
# Summary stats
# ---------------------------------------------------------------------------
def _compute_stats(processes: list[dict]) -> str:
total = len(processes)
total_ram = sum(p["ram_mb"] for p in processes)
ghosts = sum(1 for p in processes if p["is_ghost"])
return (
f"{total} Prozesse | RAM: {total_ram:.0f} MB | "
f"Ghosts (>{GHOST_HOURS}h): {ghosts}"
)
# ---------------------------------------------------------------------------
# Kill helpers
# ---------------------------------------------------------------------------
def _kill_process(pid: int, parent_widget: Gtk.Widget) -> bool:
"""Send SIGTERM to pid. Returns True if successful."""
try:
os.kill(pid, signal.SIGTERM)
return True
except ProcessLookupError:
_show_error(parent_widget, f"Prozess {pid} nicht gefunden (bereits beendet?)")
return False
except PermissionError:
_show_error(parent_widget, f"Keine Berechtigung, Prozess {pid} zu beenden.")
return False
except OSError as e:
_show_error(parent_widget, f"Fehler beim Beenden von {pid}: {e}")
return False
def _show_error(parent: Gtk.Widget, msg: str) -> None:
"""Show a simple error dialog."""
toplevel = parent.get_toplevel()
dialog = Gtk.MessageDialog(
transient_for=toplevel if isinstance(toplevel, Gtk.Window) else None,
flags=0,
message_type=Gtk.MessageType.ERROR,
buttons=Gtk.ButtonsType.OK,
text=msg,
)
dialog.run()
dialog.destroy()
def _confirm_kill(parent: Gtk.Widget, pid: int, name: str) -> bool:
"""Show confirmation dialog. Returns True if user confirmed."""
toplevel = parent.get_toplevel()
dialog = Gtk.MessageDialog(
transient_for=toplevel if isinstance(toplevel, Gtk.Window) else None,
flags=0,
message_type=Gtk.MessageType.WARNING,
buttons=Gtk.ButtonsType.YES_NO,
text=f"Prozess beenden?",
)
dialog.format_secondary_text(f"PID {pid} — {name}\n\nSignal: SIGTERM")
response = dialog.run()
dialog.destroy()
return response == Gtk.ResponseType.YES
def _confirm_kill_ghosts(parent: Gtk.Widget, count: int) -> bool:
"""Show confirmation dialog for killing all ghost processes."""
toplevel = parent.get_toplevel()
dialog = Gtk.MessageDialog(
transient_for=toplevel if isinstance(toplevel, Gtk.Window) else None,
flags=0,
message_type=Gtk.MessageType.WARNING,
buttons=Gtk.ButtonsType.YES_NO,
text=f"Alle Ghosts beenden?",
)
dialog.format_secondary_text(
f"{count} Prozesse aelter als {GHOST_HOURS}h werden per SIGTERM beendet."
)
response = dialog.run()
dialog.destroy()
return response == Gtk.ResponseType.YES
# ---------------------------------------------------------------------------
# Row builder
# ---------------------------------------------------------------------------
def _ram_color(ram_mb: float) -> str:
"""Return theme-aware color based on RAM usage."""
p = get_palette()
if ram_mb < 100:
return p["green"]
elif ram_mb < 300:
return p["peach"]
else:
return p["red"]
def _format_uptime(hours: float) -> str:
if hours < 1:
return f"{int(hours * 60)}m"
elif hours < 24:
return f"{hours:.1f}h"
else:
days = int(hours // 24)
remaining_h = int(hours % 24)
return f"{days}d {remaining_h}h"
def _build_process_row(proc: dict, parent_widget: Gtk.Widget) -> Gtk.ListBoxRow:
"""Build a single styled ListBoxRow for one process."""
row = Gtk.ListBoxRow()
row.set_name("process-row")
# Outer box with padding
outer = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
outer.set_margin_top(6)
outer.set_margin_bottom(6)
outer.set_margin_start(10)
outer.set_margin_end(10)
row.add(outer)
# Left: name + command stacked vertically
info_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
info_box.set_hexpand(True)
outer.pack_start(info_box, True, True, 0)
# Process name — bold, ghost shown in red
name_label = Gtk.Label(label=proc["name"])
name_label.set_halign(Gtk.Align.START)
name_label.set_ellipsize(Pango.EllipsizeMode.END)
name_label.set_max_width_chars(35)
name_attrs = Pango.AttrList()
name_attrs.insert(Pango.attr_weight_new(Pango.Weight.BOLD))
if proc["is_ghost"]:
# Red tint for ghost processes (theme-aware)
r, g, b = hex_to_pango_rgb(get_palette()["red"])
name_attrs.insert(Pango.attr_foreground_new(r, g, b))
name_label.set_attributes(name_attrs)
info_box.pack_start(name_label, False, False, 0)
# Command preview — dimmed, smaller
cmd_label = Gtk.Label(label=proc["command"])
cmd_label.set_halign(Gtk.Align.START)
cmd_label.set_ellipsize(Pango.EllipsizeMode.END)
cmd_label.set_max_width_chars(50)
cmd_label.get_style_context().add_class("stat-label")
cmd_attrs = Pango.AttrList()
cmd_attrs.insert(Pango.attr_scale_new(0.82))
cmd_label.set_attributes(cmd_attrs)
info_box.pack_start(cmd_label, False, False, 0)
# Middle: PID + uptime
pid_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
pid_box.set_valign(Gtk.Align.CENTER)
outer.pack_start(pid_box, False, False, 0)
pid_text = f"PID {proc['pid']}"
if proc.get("tty"):
pid_text += f" {proc['tty']}"
pid_label = Gtk.Label(label=pid_text)
pid_label.get_style_context().add_class("stat-label")
pid_attrs = Pango.AttrList()
pid_attrs.insert(Pango.attr_scale_new(0.85))
pid_label.set_attributes(pid_attrs)
pid_box.pack_start(pid_label, False, False, 0)
uptime_label = Gtk.Label(label=_format_uptime(proc["uptime_h"]))
uptime_label.get_style_context().add_class("stat-label")
uptime_attrs = Pango.AttrList()
uptime_attrs.insert(Pango.attr_scale_new(0.82))
uptime_label.set_attributes(uptime_attrs)
pid_box.pack_start(uptime_label, False, False, 0)
# Right: RAM + CPU + Kill button
meta_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
meta_box.set_valign(Gtk.Align.CENTER)
outer.pack_start(meta_box, False, False, 0)
# RAM + CPU on one line
stats_line = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
meta_box.pack_start(stats_line, False, False, 0)
ram_label = Gtk.Label(label=f"{proc['ram_mb']:.0f} MB")
ram_attrs = Pango.AttrList()
ram_attrs.insert(Pango.attr_scale_new(0.85))
# Color code RAM
r, g, b = hex_to_pango_rgb(_ram_color(proc["ram_mb"]))
ram_attrs.insert(Pango.attr_foreground_new(r, g, b))
ram_label.set_attributes(ram_attrs)
stats_line.pack_start(ram_label, False, False, 0)
cpu_label = Gtk.Label(label=f"CPU {proc['cpu']:.1f}%")
cpu_label.get_style_context().add_class("stat-label")
cpu_attrs = Pango.AttrList()
cpu_attrs.insert(Pango.attr_scale_new(0.85))
cpu_label.set_attributes(cpu_attrs)
stats_line.pack_start(cpu_label, False, False, 0)
# Kill button
kill_btn = Gtk.Button(label="Kill")
kill_btn.get_style_context().add_class("destructive-action")
kill_btn.set_tooltip_text(f"SIGTERM an PID {proc['pid']} senden")
pid = proc["pid"]
name = proc["name"]
def on_kill_clicked(_btn: Gtk.Button, _pid: int = pid, _name: str = name) -> None:
if _confirm_kill(_btn, _pid, _name):
_kill_process(_pid, _btn)
idle_once(refresh_processes)
kill_btn.connect("clicked", on_kill_clicked)
meta_box.pack_start(kill_btn, False, False, 0)
row._process_data = proc # type: ignore[attr-defined]
row.show_all()
return row
# ---------------------------------------------------------------------------
# Populate / refresh the list
# ---------------------------------------------------------------------------
def _populate_list_box(processes: list[dict]) -> bool:
"""Clear and re-populate the ListBox. Returns False (idle_add one-shot)."""
global _all_processes
_all_processes = processes
if _list_box is None:
return False
for child in _list_box.get_children():
_list_box.remove(child)
if not processes:
empty_label = Gtk.Label(label="Keine Claude-relevanten Prozesse gefunden.")
empty_label.get_style_context().add_class("stat-label")
empty_label.set_margin_top(20)
_list_box.add(empty_label)
else:
for proc in processes:
row = _build_process_row(proc, _list_box)
_list_box.add(row)
_list_box.show_all()
if _stats_label is not None:
_stats_label.set_text(_compute_stats(processes))
return False # one-shot for idle_add
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
_refresh_in_flight = False
_refresh_lock = threading.Lock()
def _refresh_processes_thread() -> None:
"""Scan processes in background thread, push result to GTK via idle_add."""
global _refresh_in_flight
try:
processes = _scan_processes()
GLib.idle_add(_populate_list_box, processes)
except Exception:
log.exception("refresh processes (background)")
finally:
with _refresh_lock:
_refresh_in_flight = False
def refresh_processes() -> bool:
"""Re-scan processes in background thread. Returns True to keep GLib timer alive."""
global _refresh_in_flight
with _refresh_lock:
if _refresh_in_flight:
return True
_refresh_in_flight = True
threading.Thread(
target=_refresh_processes_thread,
daemon=True,
name="claude-panel-process-scan",
).start()
return True # keep timer running
def build_processes_tab() -> Gtk.ScrolledWindow:
"""Build and return the Processes tab widget (Gtk.ScrolledWindow)."""
global _list_box, _stats_label
# Root: ScrolledWindow -> main VBox
scrolled = Gtk.ScrolledWindow()
scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
scrolled.set_hexpand(True)
scrolled.set_vexpand(True)
main_vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
main_vbox.set_margin_top(8)
main_vbox.set_margin_bottom(8)
main_vbox.set_margin_start(8)
main_vbox.set_margin_end(8)
scrolled.add(main_vbox)
# Toolbar: title + Kill All Ghosts + refresh button
toolbar = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
toolbar.set_margin_bottom(6)
main_vbox.pack_start(toolbar, False, False, 0)
title_label = Gtk.Label(label="Prozesse")
title_label.get_style_context().add_class("section-title")
title_attrs = Pango.AttrList()
title_attrs.insert(Pango.attr_weight_new(Pango.Weight.BOLD))
title_attrs.insert(Pango.attr_scale_new(1.1))
title_label.set_attributes(title_attrs)
title_label.set_halign(Gtk.Align.START)
toolbar.pack_start(title_label, False, False, 0)
# Spacer
spacer = Gtk.Box()
spacer.set_hexpand(True)
toolbar.pack_start(spacer, True, True, 0)
# Kill All Ghosts button
kill_ghosts_btn = Gtk.Button(label=f"Kill All Ghosts (>{GHOST_HOURS}h)")
kill_ghosts_btn.get_style_context().add_class("destructive-action")
kill_ghosts_btn.set_tooltip_text(f"Alle Prozesse aelter als {GHOST_HOURS}h per SIGTERM beenden")
def on_kill_ghosts_clicked(_btn: Gtk.Button) -> None:
ghosts = [p for p in _all_processes if p["is_ghost"]]
if not ghosts:
return
if _confirm_kill_ghosts(_btn, len(ghosts)):
for p in ghosts:
_kill_process(p["pid"], _btn)
idle_once(refresh_processes)
kill_ghosts_btn.connect("clicked", on_kill_ghosts_clicked)
toolbar.pack_start(kill_ghosts_btn, False, False, 0)
# Refresh button
refresh_btn = Gtk.Button()
refresh_icon = Gtk.Image.new_from_icon_name("view-refresh-symbolic", Gtk.IconSize.BUTTON)
refresh_btn.add(refresh_icon)
refresh_btn.set_tooltip_text("Prozessliste aktualisieren")
refresh_btn.connect("clicked", lambda _b: refresh_processes())
toolbar.pack_start(refresh_btn, False, False, 0)
# Stats bar
stats_frame = Gtk.Frame()
stats_frame.set_margin_bottom(8)
main_vbox.pack_start(stats_frame, False, False, 0)
_stats_label = Gtk.Label(label="Lade Prozesse...")
_stats_label.get_style_context().add_class("stat-label")
_stats_label.set_margin_top(6)
_stats_label.set_margin_bottom(6)
_stats_label.set_margin_start(10)
_stats_label.set_margin_end(10)
_stats_label.set_halign(Gtk.Align.START)
stats_frame.add(_stats_label)
# Column headers
header_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
header_box.set_margin_start(10)
header_box.set_margin_end(10)
header_box.set_margin_bottom(4)
main_vbox.pack_start(header_box, False, False, 0)
for col_label, expand in [("Name / Befehl", True), ("PID / Uptime", False), ("RAM / CPU / Kill", False)]:
lbl = Gtk.Label(label=col_label)
lbl.get_style_context().add_class("stat-label")
col_attrs = Pango.AttrList()
col_attrs.insert(Pango.attr_weight_new(Pango.Weight.BOLD))
col_attrs.insert(Pango.attr_scale_new(0.85))
lbl.set_attributes(col_attrs)
lbl.set_halign(Gtk.Align.START)
header_box.pack_start(lbl, expand, expand, 0)
# ListBox for processes
_list_box = Gtk.ListBox()
_list_box.set_selection_mode(Gtk.SelectionMode.NONE)
_list_box.get_style_context().add_class("view")
main_vbox.pack_start(_list_box, True, True, 0)
idle_once(lambda: _populate_list_box(_scan_processes()))
scrolled.show_all()
return scrolled