Add processes service and state wiring

main.py

@@ -19,6 +19,7 @@ import handlers.system
 import handlers.help
 import handlers.callbacks
 import handlers.arcane
+import handlers.processes
 
 
 async def notify_start():

services/processes.py

@@ -0,0 +1,88 @@
+import time
+from typing import Any
+
+import psutil
+
+
+def _safe_name(info: dict[str, Any]) -> str:
+    name = info.get("name") or "unknown"
+    return str(name)
+
+
+def get_top_processes(limit: int = 5, interval: float = 0.2) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    procs = []
+    for p in psutil.process_iter(attrs=["pid", "name"]):
+        procs.append(p)
+
+    for p in procs:
+        try:
+            p.cpu_percent(None)
+        except Exception:
+            continue
+
+    time.sleep(interval)
+
+    items = []
+    for p in procs:
+        try:
+            cpu = p.cpu_percent(None)
+            mem = p.memory_percent()
+            info = p.info
+            items.append({
+                "pid": info.get("pid"),
+                "name": _safe_name(info),
+                "cpu": cpu,
+                "mem": mem,
+            })
+        except Exception:
+            continue
+
+    top_cpu = sorted(items, key=lambda x: x["cpu"], reverse=True)[:limit]
+    top_mem = sorted(items, key=lambda x: x["mem"], reverse=True)[:limit]
+    return top_cpu, top_mem
+
+
+def search_processes(query: str, limit: int = 10) -> list[dict[str, Any]]:
+    needle = query.lower().strip()
+    if not needle:
+        return []
+
+    results = []
+    for p in psutil.process_iter(attrs=["pid", "name", "cmdline"]):
+        try:
+            info = p.info
+            name = _safe_name(info)
+            cmdline = " ".join(info.get("cmdline") or [])
+            hay = f"{name} {cmdline}".lower()
+            if needle not in hay:
+                continue
+            results.append({
+                "pid": info.get("pid"),
+                "name": name,
+                "cmdline": cmdline,
+            })
+        except Exception:
+            continue
+
+    return results[:limit]
+
+
+def terminate_process(pid: int, timeout: float = 5.0) -> str:
+    try:
+        proc = psutil.Process(pid)
+    except Exception:
+        return f"Process {pid} not found"
+
+    try:
+        proc.terminate()
+        proc.wait(timeout=timeout)
+        return f"Process {pid} terminated"
+    except psutil.TimeoutExpired:
+        try:
+            proc.kill()
+            proc.wait(timeout=timeout)
+            return f"Process {pid} killed"
+        except Exception as e:
+            return f"Kill failed for {pid}: {e}"
+    except Exception as e:
+        return f"Terminate failed for {pid}: {e}"

state.py

@@ -7,3 +7,5 @@ ARCANE_CACHE: Dict[int, dict] = {}
 REBOOT_PENDING: Dict[int, dict] = {}
 METRICS_STORE = None
 NPMPLUS_TOKEN: Dict[str, object] = {}
+PROC_SEARCH_PENDING: Dict[int, dict] = {}
+PROC_KILL_PENDING: Dict[int, dict] = {}