manual_slop/scripts/cleanup_orphaned_processes.py

"""Identify and clean up orphaned manual_slop processes left over from prior test runs.

Categorizes running python/uv processes into:
  - sloppy.py:        manual_slop GUI app spawned by live_gui tests; KILL (these are the orphans)
  - mcp_server.py:    manual_slop's MCP server; preserve (legitimate, used by opencode)
  - minimax-coding-plan-mcp: opencode's MCP server; preserve (legitimate, used by opencode)
  - test subprocess:  long-running pytest/test helpers; flag, kill if --kill-tests

Default mode is DRY-RUN (lists orphans, prints what would be killed, kills nothing).
Pass --kill to actually kill sloppy.py orphans.
Pass --kill-tests to also kill the "test subprocess" category.
Pass --kill-mcp to kill the MCP servers (off by default; usually you don't want to).
Pass --json for machine-readable output.

Usage:
  uv run python scripts/cleanup_orphaned_processes.py             # dry-run
  uv run python scripts/cleanup_orphaned_processes.py --kill      # kill sloppy.py orphans
  uv run python scripts/cleanup_orphaned_processes.py --kill --kill-tests
  uv run python scripts/cleanup_orphaned_processes.py --json
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path


@dataclass(frozen=True)
class ProcInfo:
 pid: int
 name: str
 command_line: str
 age_seconds: float
 cpu_seconds: float
 category: str

 def render(self) -> str:
  cmd_short = self.command_line[:120].replace("\n", " ").replace("\r", " ")
  return (f"  PID {self.pid:>6}  {self.name:<10}  age={self.age_seconds:>7.1f}s  "
   f"cpu={self.cpu_seconds:>7.1f}s  {self.category}\n"
   f"           cmd: {cmd_short}")


def _enum_processes() -> list[ProcInfo]:
 r = subprocess.run(
  ["powershell.exe", "-NoProfile", "-Command",
   "Get-CimInstance Win32_Process -Filter \"Name='python.exe' OR Name='uv.exe'\" "
   "| Where-Object { $_.CommandLine } "
   "| Select-Object ProcessId, Name, CommandLine, CreationDate, KernelModeTime, UserModeTime "
   "| ConvertTo-Json -Depth 2 -Compress"],
  capture_output=True, text=True, timeout=15,
 )
 if r.returncode != 0:
  print(f"ERROR: powershell enumeration failed: {r.stderr}", file=sys.stderr)
  return []
 out = r.stdout.strip()
 if not out or out == "null":
  return []
 raw = json.loads(out)
 if isinstance(raw, dict):
  raw = [raw]
 procs: list[ProcInfo] = []
 for item in raw:
  pid = int(item.get("ProcessId", 0))
  name = str(item.get("Name", ""))
  cl = str(item.get("CommandLine", "") or "")
  if not cl:
   continue
  creation = item.get("CreationDate", "")
  try:
   from datetime import datetime, timezone
   dt = datetime.fromisoformat(creation.replace("Z", "+00:00"))
   age = (datetime.now(timezone.utc) - dt).total_seconds()
  except Exception:
   age = 0.0
  kernel = float(item.get("KernelModeTime", 0) or 0) / 1e7
  user = float(item.get("UserModeTime", 0) or 0) / 1e7
  cpu = kernel + user
  cat = _categorize(cl)
  if cat is None:
   continue
  procs.append(ProcInfo(pid=pid, name=name, command_line=cl, age_seconds=age, cpu_seconds=cpu, category=cat))
 return procs


def _categorize(cl: str) -> str | None:
 if "sloppy.py" in cl and "--enable-test-hooks" in cl:
  return "sloppy.py (test leftover)"
 if "scripts/mcp_server.py" in cl or "manual_slop" in cl and "mcp" in cl.lower():
  return "manual_slop MCP server (preserve)"
 if "minimax-coding-plan-mcp" in cl:
  return "minimax MCP server (preserve)"
 if "pytest" in cl and "conftest" not in cl:
  return "pytest runner"
 if "src.gui_2" in cl and "App()" in cl:
  return "stuck test subprocess (src.gui_2 + App)"
 return None


def _kill(pid: int) -> bool:
 r = subprocess.run(["powershell.exe", "-NoProfile", "-Command", f"Stop-Process -Id {pid} -Force -ErrorAction SilentlyContinue; if (Get-Process -Id {pid} -ErrorAction SilentlyContinue) {{ 'STILL_ALIVE' }} else {{ 'KILLED' }}"],
  capture_output=True, text=True, timeout=10)
 return "KILLED" in r.stdout


def main() -> int:
 ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
 ap.add_argument("--kill", action="store_true", help="Actually kill sloppy.py orphans (default: dry-run)")
 ap.add_argument("--kill-tests", action="store_true", help="Also kill stuck test subprocesses")
 ap.add_argument("--kill-mcp", action="store_true", help="Also kill MCP servers (usually DON'T want this)")
 ap.add_argument("--json", action="store_true", help="Machine-readable JSON output")
 args = ap.parse_args()

 procs = _enum_processes()
 by_cat: dict[str, list[ProcInfo]] = {}
 for p in procs:
  by_cat.setdefault(p.category, []).append(p)

 if args.json:
  out = {
   "dry_run": not args.kill,
   "categories": {cat: [asdict(p) for p in items] for cat, items in by_cat.items()},
  }
  print(json.dumps(out, indent=2))
  return 0

 print("=" * 80)
 print(f"manual_slop process cleanup  ({'DRY-RUN' if not args.kill else 'KILL MODE'})")
 print("=" * 80)

 kill_targets: list[ProcInfo] = []
 preserve_targets: list[ProcInfo] = []

 for cat, items in by_cat.items():
  is_orphan = cat.startswith("sloppy.py")
  is_test = cat.startswith("pytest") or cat.startswith("stuck test")
  is_mcp = "MCP server" in cat
  if is_orphan:
   kill_targets.extend(items)
   header = f"[KILL: sloppy.py orphans]   ({len(items)} found)"
  elif is_test and args.kill_tests:
   kill_targets.extend(items)
   header = f"[KILL: test subprocesses]  ({len(items)} found)"
  elif is_mcp and args.kill_mcp:
   kill_targets.extend(items)
   header = f"[KILL: MCP servers]        ({len(items)} found)"
  else:
   preserve_targets.extend(items)
   header = f"[PRESERVE]                 ({len(items)} found)  {cat}"
  print()
  print(header)
  for p in items:
   print(p.render())

 print()
 print("=" * 80)
 print(f"Totals: {len(kill_targets)} kill candidates, {len(preserve_targets)} preserved")
 print("=" * 80)

 if not kill_targets:
  print("Nothing to kill.")
  return 0

 if not args.kill:
  print()
  print("DRY-RUN: pass --kill to actually terminate the above kill candidates.")
  return 0

 print()
 print(f"Killing {len(kill_targets)} process(es)...")
 killed = 0
 failed = []
 for p in kill_targets:
  if _kill(p.pid):
   killed += 1
   print(f"  killed PID {p.pid:>6}  ({p.category})")
  else:
   failed.append(p.pid)
   print(f"  FAILED PID {p.pid:>6}  ({p.category})")
 print()
 print(f"Result: {killed} killed, {len(failed)} failed")
 return 0 if not failed else 1


if __name__ == "__main__":
 raise SystemExit(main())