| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from pathlib import Path |
| import os |
| import signal |
| import subprocess |
| import time |
|
|
|
|
| PROCESS_KEYWORDS = ( |
| "claudecodepanel", |
| "claude-code", |
| "claude code", |
| "claude", |
| "codex", |
| "opencode", |
| "mcp-server", |
| "server-filesystem", |
| "server-memory", |
| "server-github", |
| "gemini-mcp", |
| "gemini", |
| ) |
|
|
| STALE_SECONDS = 24 * 3600 |
| CLK_TCK = os.sysconf(os.sysconf_names.get("SC_CLK_TCK", "SC_CLK_TCK")) |
|
|
|
|
| @dataclass(frozen=True) |
| class ProcessInfo: |
| pid: int |
| name: str |
| command: str |
| cpu: float |
| ram_mb: float |
| tty: str |
| cwd: str |
| uptime_seconds: float |
| stale: bool |
| family: str = "Agent" |
|
|
|
|
| def _match_keyword(command: str) -> str | None: |
| lower = command.lower() |
| if "codex_dashboard" in lower or "codex-dashboard" in lower: |
| return None |
| for keyword in PROCESS_KEYWORDS: |
| if keyword in lower: |
| return keyword |
| return None |
|
|
|
|
| def _friendly_name(command: str, keyword: str) -> str: |
| first = Path(command.split()[0]).name if command.split() else keyword |
| if keyword == "claudecodepanel": |
| return "Claude Panel" |
| if keyword in {"claude", "claude-code", "claude code"}: |
| return "Claude Code" |
| if keyword == "codex": |
| return "Codex CLI" |
| if keyword == "opencode": |
| return "OpenCode" |
| if keyword in {"gemini", "gemini-mcp"}: |
| return "Gemini" |
| if keyword.startswith("server-"): |
| return keyword.replace("server-", "").title() + " MCP" |
| if keyword == "mcp-server": |
| return "MCP Server" |
| if first in {"node", "python", "python3"}: |
| return f"{first}/{keyword}" |
| return first or keyword |
|
|
|
|
| def _family(keyword: str) -> str: |
| if keyword in {"claude", "claude-code", "claude code", "claudecodepanel"}: |
| return "Claude" |
| if keyword == "codex": |
| return "Codex" |
| if keyword == "opencode": |
| return "OpenCode" |
| if "mcp" in keyword or keyword.startswith("server-"): |
| return "MCP" |
| if keyword.startswith("gemini"): |
| return "Gemini" |
| return "Agent" |
|
|
|
|
| def _read_cwd(pid: int) -> str: |
| try: |
| return os.readlink(f"/proc/{pid}/cwd") |
| except OSError: |
| return "" |
|
|
|
|
| def _read_uptime_seconds(pid: int, now: float | None = None) -> float: |
| now = time.time() if now is None else now |
| try: |
| stat_data = Path(f"/proc/{pid}/stat").read_text().split() |
| starttime_ticks = int(stat_data[21]) |
| uptime = float(Path("/proc/uptime").read_text().split()[0]) |
| except (OSError, ValueError, IndexError): |
| return 0.0 |
| process_start = now - uptime + (starttime_ticks / CLK_TCK) |
| return max(0.0, now - process_start) |
|
|
|
|
| def parse_ps_aux_line( |
| line: str, *, now: float | None = None, read_proc_metadata: bool = True |
| ) -> ProcessInfo | None: |
| parts = line.split(None, 10) |
| if len(parts) < 11: |
| return None |
| try: |
| pid = int(parts[1]) |
| cpu = float(parts[2]) |
| rss_kb = int(parts[5]) |
| except ValueError: |
| return None |
| command = parts[10] |
| keyword = _match_keyword(command) |
| if keyword is None or pid == os.getpid(): |
| return None |
| uptime_seconds = _read_uptime_seconds(pid, now) if read_proc_metadata else 0.0 |
| return ProcessInfo( |
| pid=pid, |
| name=_friendly_name(command, keyword), |
| command=command, |
| cpu=cpu, |
| ram_mb=rss_kb / 1024, |
| tty="" if parts[6] == "?" else parts[6], |
| cwd=_read_cwd(pid) if read_proc_metadata else "", |
| uptime_seconds=uptime_seconds, |
| stale=uptime_seconds >= STALE_SECONDS, |
| family=_family(keyword), |
| ) |
|
|
|
|
| def scan_processes(*, ps_output: str | None = None) -> list[ProcessInfo]: |
| read_proc_metadata = ps_output is None |
| if ps_output is None: |
| try: |
| result = subprocess.run( |
| ["ps", "aux", "--no-headers"], |
| capture_output=True, |
| text=True, |
| timeout=5, |
| ) |
| except (OSError, subprocess.TimeoutExpired): |
| return [] |
| if result.returncode != 0: |
| return [] |
| ps_output = result.stdout |
|
|
| processes = [ |
| process |
| for line in ps_output.splitlines() |
| if (process := parse_ps_aux_line( |
| line, read_proc_metadata=read_proc_metadata |
| )) is not None |
| ] |
| processes.sort(key=lambda item: (-int(item.stale), -item.ram_mb, item.name)) |
| return processes |
|
|
|
|
| def terminate_process(pid: int) -> None: |
| os.kill(pid, signal.SIGTERM) |
|
|
|
|
| def format_uptime(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" |
|
|
