hermes-agent/scripts/mcp_zombie_cleanup.py

#!/usr/bin/env python3
"""
MCP zombie process cleanup — kills orphaned MCP server processes.

Problem: MCP servers (especially morrowind) spawn stdio subprocesses that
survive restarts. Over time, 80+ zombie processes accumulate.

Fix: Scan for processes matching known MCP server patterns, kill older
instances, keep only the latest N.

Usage:
    python3 scripts/mcp_zombie_cleanup.py [--dry-run] [--keep 3] [--max-age 3600]
    python3 scripts/mcp_zombie_cleanup.py --kill-all  # nuclear option
"""

import argparse
import os
import re
import signal
import subprocess
import sys
import time
from typing import List, Tuple


# Patterns that identify MCP server processes
MCP_PROCESS_PATTERNS = [
    re.compile(r"morrowind[/\]mcp_server", re.IGNORECASE),
    re.compile(r"mcp_server\.py", re.IGNORECASE),
    re.compile(r"mcp[-_]server", re.IGNORECASE),
    re.compile(r"hermes.*mcp.*stdio", re.IGNORECASE),
]


def find_mcp_processes() -> List[Tuple[int, float, str]]:
    """Find MCP server processes.

    Returns list of (pid, start_time_epoch, command_line).
    """
    my_pid = os.getpid()
    results = []

    try:
        # Use ps to get all processes with start time and command
        ps_out = subprocess.check_output(
            ["ps", "-eo", "pid,lstart,command"],
            text=True, stderr=subprocess.DEVNULL
        )
    except (subprocess.CalledProcessError, FileNotFoundError):
        # Fallback: macOS ps format
        try:
            ps_out = subprocess.check_output(
                ["ps", "-eo", "pid,lstart,args"],
                text=True, stderr=subprocess.DEVNULL
            )
        except Exception:
            return results

    for line in ps_out.strip().splitlines()[1:]:  # Skip header
        parts = line.strip().split(None, 6)
        if len(parts) < 7:
            continue

        try:
            pid = int(parts[0])
        except ValueError:
            continue

        if pid == my_pid:
            continue

        # Parse lstart: "pid Mon Apr 14 16:02:03 2026 command..."
        # parts[1:5] = month, day, time, year
        cmd = parts[6] if len(parts) > 6 else ""

        # Check if command matches MCP patterns
        is_mcp = any(p.search(cmd) for p in MCP_PROCESS_PATTERNS)
        if not is_mcp:
            continue

        # Parse start time
        try:
            start_str = " ".join(parts[1:5])
            start_struct = time.strptime(start_str, "%b %d %H:%M:%S %Y")
            start_epoch = time.mktime(start_struct)
        except (ValueError, OverflowError):
            start_epoch = 0

        results.append((pid, start_epoch, cmd))

    return results


def cleanup_zombies(
    keep: int = 3,
    max_age_seconds: int = 3600,
    dry_run: bool = False,
    kill_all: bool = False,
) -> dict:
    """Clean up zombie MCP processes.

    Args:
        keep: Number of newest processes to keep alive
        max_age_seconds: Kill processes older than this (even if under keep count)
        dry_run: If True, don't actually kill anything
        kill_all: If True, kill ALL MCP processes regardless of age/count

    Returns:
        Dict with counts: found, killed, kept
    """
    processes = find_mcp_processes()

    if not processes:
        return {"found": 0, "killed": 0, "kept": 0}

    # Sort by start time, newest first
    processes.sort(key=lambda x: x[1], reverse=True)

    now = time.time()
    killed = 0
    kept = 0
    kill_pids = []

    for pid, start_time, cmd in processes:
        age = now - start_time if start_time > 0 else float('inf')

        if kill_all:
            kill_pids.append((pid, age, cmd))
        elif kept < keep and age < max_age_seconds:
            # Keep this one (new enough and under keep count)
            kept += 1
        else:
            # Too old or over keep limit
            kill_pids.append((pid, age, cmd))

    for pid, age, cmd in kill_pids:
        if dry_run:
            print(f"  [DRY RUN] Would kill PID {pid} (age={age:.0f}s): {cmd[:80]}")
            killed += 1
        else:
            try:
                os.kill(pid, signal.SIGTERM)
                print(f"  Killed PID {pid} (age={age:.0f}s): {cmd[:80]}")
                killed += 1
            except ProcessLookupError:
                print(f"  PID {pid} already exited")
            except PermissionError:
                print(f"  No permission to kill PID {pid}")
                try:
                    os.kill(pid, signal.SIGKILL)
                    print(f"  Force-killed PID {pid}")
                    killed += 1
                except Exception:
                    pass

    return {"found": len(processes), "killed": killed, "kept": kept}


def main(argv=None):
    parser = argparse.ArgumentParser(description="Clean up zombie MCP processes")
    parser.add_argument("--dry-run", action="store_true", help="Don't kill, just show")
    parser.add_argument("--keep", type=int, default=3, help="Keep N newest processes (default: 3)")
    parser.add_argument("--max-age", type=int, default=3600, help="Kill processes older than N seconds (default: 3600)")
    parser.add_argument("--kill-all", action="store_true", help="Kill ALL MCP processes")
    args = parser.parse_args(argv)

    processes = find_mcp_processes()
    print(f"Found {len(processes)} MCP processes")

    if processes and not args.dry_run:
        processes.sort(key=lambda x: x[1], reverse=True)
        print(f"Newest: PID {processes[0][0]} ({time.time() - processes[0][1]:.0f}s ago)")
        print(f"Oldest: PID {processes[-1][0]} ({time.time() - processes[-1][1]:.0f}s ago)")

    result = cleanup_zombies(
        keep=args.keep,
        max_age_seconds=args.max_age,
        dry_run=args.dry_run,
        kill_all=args.kill_all,
    )

    print(f"\nResult: found={result['found']}, killed={result['killed']}, kept={result['kept']}")

    # Verify cleanup
    remaining = find_mcp_processes()
    print(f"Remaining MCP processes: {len(remaining)}")

    if len(remaining) > 5:
        print(f"WARNING: Still {len(remaining)} MCP processes (threshold: 5)")

    return 0 if len(remaining) <= 5 else 1


if __name__ == "__main__":
    sys.exit(main())