diff options
Diffstat (limited to 'agent-id.py')
| -rw-r--r-- | agent-id.py | 208 |
1 files changed, 208 insertions, 0 deletions
diff --git a/agent-id.py b/agent-id.py new file mode 100644 index 0000000..00473f7 --- /dev/null +++ b/agent-id.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""agent-id — detect which agent harness called this tool. + +Walks /proc/self/.. up the process tree, inspecting cmdlines, comm names, +and ancestor environment variables to identify Claude Code, Codex CLI, +or Hermes/pi agent — without the caller passing any flags or env vars. + +Outputs JSON: {"agent": "...", "session_id": "...", "call_id": "...", + "pid": N, "ppid": N, "cmdline": "..."} +""" + +import json +import os +import re +import sys + + +def read_proc_file(pid: int, name: str) -> str | None: + """Read /proc/<pid>/<name>, return stripped content or None.""" + try: + with open(f"/proc/{pid}/{name}", "rb") as f: + raw = f.read() + return raw.decode("utf-8", errors="replace").strip("\x00\n ") + except (FileNotFoundError, PermissionError, ProcessLookupError): + return None + + +def read_cmdline(pid: int) -> str | None: + """Read null-joined argv from /proc/<pid>/cmdline.""" + raw = read_proc_file(pid, "cmdline") + if raw is None: + return None + return raw.replace("\x00", " ").strip() + + +def read_comm(pid: int) -> str | None: + """Read process comm name from /proc/<pid>/comm.""" + return read_proc_file(pid, "comm") + + +def read_status_field(pid: int, field: str) -> str | None: + """Read a specific field value from /proc/<pid>/status.""" + raw = read_proc_file(pid, "status") + if raw is None: + return None + for line in raw.splitlines(): + if line.startswith(field + ":"): + return line.split(":", 1)[-1].strip() + return None + + +def read_ppid(pid: int) -> int | None: + """Get parent PID from /proc/<pid>/status.""" + val = read_status_field(pid, "PPid") + if val is None: + return None + try: + return int(val) + except ValueError: + return None + + +def read_environ_dict(pid: int) -> dict[str, str]: + """Parse /proc/<pid>/environ into a dict.""" + raw = read_proc_file(pid, "environ") + if raw is None: + return {} + result = {} + for entry in raw.split("\x00"): + if "=" in entry: + k, _, v = entry.partition("=") + result[k] = v + return result + + +def get_own_environ() -> dict[str, str]: + """Get our own environment via /proc/self/environ (not os.environ).""" + return read_environ_dict(os.getpid()) + + +def walk_process_tree() -> list[dict]: + """Walk up from our PID to init, return ordered list of process info dicts.""" + pid = os.getpid() + tree = [] + visited = set() + while pid is not None and pid > 1 and pid not in visited: + visited.add(pid) + cmdline = read_cmdline(pid) + comm = read_comm(pid) + env = read_environ_dict(pid) + ppid = read_ppid(pid) + tree.append({ + "pid": pid, + "ppid": ppid, + "comm": comm, + "cmdline": cmdline, + "env": env, + }) + pid = ppid + return tree + + +# ── Agent detection patterns ────────────────────────────────────────────── + +PATTERNS: list[tuple[str, re.Pattern, str]] = [ + # (agent_name, cmdline_regex, description) + ("hermes", re.compile(r"python.*hermes_cli", re.I), "python -m hermes_cli.main"), + ("hermes", re.compile(r"hermes[_-]gateway", re.I), "hermes gateway"), + ("claude", re.compile(r"claude", re.I), "claude binary"), + ("claude", re.compile(r"claude[_-]code", re.I), "claude-code"), + ("codex", re.compile(r"codex\b(?!\s+exec)", re.I), "codex binary"), + ("codex", re.compile(r"codex[_-]cli", re.I), "codex-cli"), +] + +SESSION_ENV_VARS: dict[str, list[str]] = { + "hermes": ["HERMES_SESSION_ID"], + "claude": [ + "CLAUDE_CODE_CHILD_SESSION", + "CLAUDE_CODE_SESSION_LOG", + "CLAUDE_SESSION_INGRESS_TOKEN_FILE", + ], + "codex": [ + "CODEX_THREAD_ID", + "CODEX_EVENT_ID", + "CODEX_CALL_ID", + ], +} + +CALL_ENV_VARS: dict[str, list[str]] = { + "hermes": [], + "claude": [], + +"codex": ["CODEX_CALL_ID", "CODEX_EVENT_ID", "CODEX_TURN_ID"], +} + + +def detect_agent(tree: list[dict]) -> dict: + """Detect which agent called us by analyzing the process tree.""" + self_pid = tree[0]["pid"] if tree else os.getpid() + self_cmdline = tree[0]["cmdline"] if tree else None + + result: dict = { + "agent": "unknown", + "session_id": None, + "call_id": None, + "pid": self_pid, + "ppid": tree[0]["ppid"] if len(tree) > 0 else None, + "cmdline": self_cmdline, + "ancestors": [], + } + + for i, node in enumerate(tree): + cmdline = node["cmdline"] or "" + comm = node["comm"] or "" + env = node["env"] + + ancestor_info = { + "pid": node["pid"], + "comm": comm, + "cmdline_preview": cmdline[:120] + ("..." if len(cmdline) > 120 else ""), + "agent_hit": None, + } + + for agent_name, pattern, desc in PATTERNS: + if pattern.search(cmdline) or pattern.search(comm): + if result["agent"] == "unknown": + result["agent"] = agent_name + ancestor_info["agent_hit"] = f"{agent_name} ({desc})" + break + + if i > 0 and comm == "hermes": + if result["agent"] == "unknown": + result["agent"] = "hermes" + ancestor_info["agent_hit"] = "hermes (comm=hermes)" + + for agent_name, var_names in SESSION_ENV_VARS.items(): + for var in var_names: + if var in env and env[var]: + if result["agent"] == "unknown": + result["agent"] = agent_name + if not result["session_id"]: + result["session_id"] = env[var] + ancestor_info["agent_hit"] = ( + ancestor_info["agent_hit"] or "" + ) + f" {agent_name}({var}={env[var][:60]})" + break + + for agent_name, var_names in CALL_ENV_VARS.items(): + for var in var_names: + if var in env and env[var]: + if not result["call_id"]: + result["call_id"] = env[var] + break + + result["ancestors"].append(ancestor_info) + + return result + + +def main(): + tree = walk_process_tree() + result = detect_agent(tree) + output = {k: v for k, v in result.items() if k != "ancestors"} + print(json.dumps(output, indent=2)) + + +if __name__ == "__main__": + main() |