| from __future__ import annotations |
|
|
| from pathlib import Path |
|
|
| from .models import Classification, ProcessSample |
|
|
|
|
| RUNTIME_NAMES = {"node", "python", "python3", "uv", "npx", "npm", "pnpm", "bun", "mise"} |
| SHELL_NAMES = {"bash", "zsh", "fish", "sh"} |
| TERMINAL_NAMES = {"gnome-terminal", "kgx", "kitty", "alacritty", "wezterm", "cosmic-term"} |
| EDITOR_NAMES = {"code", "codium", "zed", "vim", "nvim", "emacs"} |
| BROWSER_NAMES = {"chrome", "google-chrome", "chromium", "firefox"} |
|
|
|
|
| def _argv_text(sample: ProcessSample) -> str: |
| return " ".join(sample.argv).lower() |
|
|
|
|
| def _base_names(sample: ProcessSample) -> set[str]: |
| names = {sample.comm.lower()} |
| if sample.argv: |
| names.add(Path(sample.argv[0]).name.lower()) |
| if sample.exe_path: |
| names.add(Path(sample.exe_path).name.lower()) |
| return {item for item in names if item} |
|
|
|
|
| def classify_process(sample: ProcessSample) -> Classification: |
| text = _argv_text(sample) |
| names = _base_names(sample) |
|
|
| bridge_markers = { |
| "gemini_acp_mcp_bridge.mjs": "mcp.bridge.gemini_acp", |
| "opencode_mcp_bridge.py": "mcp.bridge.opencode", |
| } |
| for marker, known_class in bridge_markers.items(): |
| if marker in text: |
| return Classification( |
| known_class=known_class, |
| confidence=0.98, |
| reasons=(f"argv contains {marker}",), |
| is_agent_related=True, |
| is_bridge=True, |
| is_mcp=True, |
| ) |
|
|
| if "opencode-mcp" in text: |
| return Classification( |
| "mcp.server.opencode", |
| 0.95, |
| ("argv contains opencode-mcp",), |
| True, |
| True, |
| True, |
| ) |
| if "gemini-mcp" in text: |
| return Classification( |
| "mcp.server.gemini", |
| 0.95, |
| ("argv contains gemini-mcp",), |
| True, |
| True, |
| True, |
| ) |
| if "codex mcp-server" in text or "mcp-server" in text or "modelcontextprotocol" in text: |
| return Classification( |
| "mcp.server", |
| 0.88, |
| ("argv contains MCP server marker",), |
| True, |
| False, |
| True, |
| ) |
|
|
| agent_rules = ( |
| ("agent.claude", {"claude", "claude-code"}, ("@anthropic-ai/claude-code", ".claude")), |
| ("agent.codex", {"codex"}, ("@openai/codex", "openai-codex", ".codex")), |
| ("agent.hermes", {"hermes", "hermes-agent"}, (".hermes", "hermes-agent")), |
| ("agent.opencode", {"opencode"}, (".opencode", "sst/opencode")), |
| ("agent.gemini", {"gemini"}, ("@google/gemini-cli", "gemini-cli")), |
| ) |
| for known_class, exact_names, markers in agent_rules: |
| if names & exact_names: |
| return Classification( |
| known_class, |
| 0.93, |
| (f"basename matches {sorted(names & exact_names)[0]}",), |
| True, |
| False, |
| False, |
| ) |
| matched = next((marker for marker in markers if marker in text), None) |
| if matched: |
| return Classification( |
| known_class, |
| 0.80, |
| (f"argv contains {matched}",), |
| True, |
| False, |
| False, |
| ) |
|
|
| if names & RUNTIME_NAMES: |
| return Classification("wrapper.runtime", 0.55, ("runtime wrapper",)) |
| if names & SHELL_NAMES: |
| return Classification("wrapper.shell", 0.50, ("shell wrapper",)) |
| if names & TERMINAL_NAMES: |
| return Classification("terminal", 0.45, ("terminal process",)) |
| if names & EDITOR_NAMES: |
| return Classification("editor", 0.45, ("editor process",)) |
| if names & BROWSER_NAMES: |
| return Classification("browser", 0.45, ("browser process",)) |
| if sample.pid <= 2 or not sample.argv: |
| return Classification("system", 0.30, ("system or kernel-like process",)) |
|
|
| if " mcp " in f" {text} ": |
| return Classification("possible.mcp", 0.40, ("low-confidence mcp token",)) |
|
|
| return Classification("unknown", 0.0, ()) |
|
|
|
|
| def classify_all(samples: list[ProcessSample]) -> dict[str, Classification]: |
| return {sample.process_key: classify_process(sample) for sample in samples} |
|
|
|
|
