File size: 4,236 Bytes
06357e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | 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}
|