#!/usr/bin/env python3 """Mac resource monitor daemon — prevents crash by watching memory/CPU/whisper-cli.""" from __future__ import annotations import sys import json import time import signal import argparse import subprocess from pathlib import Path from datetime import datetime, timezone from typing import Dict, List, Optional import psutil # ── Thresholds (edit here, then reload launchd) ────────────────────────────── MEM_WARN: int = 85 # % memory usage → notify MEM_KILL: int = 92 # % memory usage → notify (no auto-kill, user decides) CPU_WARN_THRESHOLD: int = 90 # % cpu usage considered high CPU_WARN_DURATION: int = 25 # seconds of sustained high CPU before warning (5 × 5s) WHISPER_WARN_COUNT: int = 3 # simultaneous whisper-cli processes → warn WHISPER_KILL_COUNT: int = 5 # simultaneous whisper-cli processes → kill oldest WHISPER_LONG_RUN_SEC: int = 60 # whisper-cli running longer than this → kill # whisper-stream は常駐前提だが、止め忘れ事故が再発（2026-05-09: 1日9時間放置） # したため別閾値で長寿命検知。3時間超えたら確実に止め忘れ。 WHISPER_STREAM_LONG_RUN_SEC: int = 3 * 60 * 60 # whisper-stream 3時間超で kill # 監視対象のプロセス名（厳密一致） # 2026-05-04 暴走事故: whisper-cli 26並列 → 多重起動&短時間長寿命検知で防ぐ # 2026-05-09 暴走事故: whisper-stream 1日9時間放置 → 別の長時間閾値で防ぐ WHISPER_CLI_NAMES = ("whisper-cli",) WHISPER_STREAM_NAMES = ("whisper-stream",) WHISPER_TARGET_NAMES = WHISPER_CLI_NAMES + WHISPER_STREAM_NAMES # 多重起動カウント対象に含める cmdline 部分一致トークン（subprocess/モジュール起動対策） # 2026-06-30 修正: `python -m mlx_whisper` は name=Python で厳密一致を貫通→監視漏れ。 # cmdline 全体に下記トークンを含むプロセスも whisper 多重カウントに含める。 WHISPER_CMDLINE_TOKENS = ("mlx_whisper",) LOOP_INTERVAL: int = 5 # main loop cadence (seconds) COOLDOWN: int = 60 # minimum seconds between same-type events VERSION: str = "1.2.0" # ───────────────────────────────────────────────────────────────────────────── LOG_DIR = Path.home() / "jarvis_safety" LOG_FILE = LOG_DIR / "resource_log.jsonl" _running: bool = True _cooldown_last: Dict[str, float] = {} _cpu_high_since: Optional[float] = None # ── Logging ─────────────────────────────────────────────────────────────────── def _ts() -> str: return datetime.now(timezone.utc).isoformat() def log(entry: dict) -> None: entry.setdefault("ts", _ts()) LOG_FILE.parent.mkdir(parents=True, exist_ok=True) with LOG_FILE.open("a") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") # ── Notification ────────────────────────────────────────────────────────────── TERMINAL_NOTIFIER = "/opt/homebrew/bin/terminal-notifier" def _esc(s: str) -> str: return s.replace("\\", "\\\\").replace('"', '\\"') def notify(title: str, body: str, critical: bool = False) -> None: if critical: # display alertダイアログ。OK押すまで残るので絶対気づける script = ( f'tell application "System Events" to ' f'display alert "{_esc(title)}" message "{_esc(body)}" as critical' ) subprocess.Popen(["osascript", "-e", script]) return if Path(TERMINAL_NOTIFIER).exists(): subprocess.run( [TERMINAL_NOTIFIER, "-title", title, "-message", body, "-sound", "default"], capture_output=True, ) else: script = f'display notification "{_esc(body)}" with title "{_esc(title)}"' subprocess.run(["osascript", "-e", script], capture_output=True) def _in_cooldown(key: str) -> bool: last = _cooldown_last.get(key, 0.0) if time.time() - last < COOLDOWN: return True _cooldown_last[key] = time.time() return False # ── Kill helpers ────────────────────────────────────────────────────────────── def _kill_process(proc: psutil.Process, reason: str) -> str: """SIGTERM → wait 2s → SIGKILL. Returns action string.""" try: proc.terminate() time.sleep(2) if proc.is_running(): proc.kill() return f"killed pid={proc.pid} ({reason})" except (psutil.NoSuchProcess, psutil.AccessDenied) as e: return f"kill failed pid={proc.pid}: {e}" # ── Checks ──────────────────────────────────────────────────────────────────── def check_memory() -> None: pct = psutil.virtual_memory().percent if pct >= MEM_KILL: if not _in_cooldown("mem_kill"): msg = f"メモリ {pct:.1f}% — 危険域。手動でアプリを閉じてください" notify("🔴 Jarvis Safety: メモリ危険", msg, critical=True) log({"type": "memory", "level": "kill", "detail": f"{pct:.1f}%", "action_taken": "notified_only"}) elif pct >= MEM_WARN: if not _in_cooldown("mem_warn"): msg = f"メモリ {pct:.1f}% — このまま行くと落ちます" notify("🟡 Jarvis Safety: メモリ警告", msg) log({"type": "memory", "level": "warn", "detail": f"{pct:.1f}%", "action_taken": "notified"}) def check_cpu() -> None: global _cpu_high_since pct = psutil.cpu_percent(interval=1) if pct >= CPU_WARN_THRESHOLD: if _cpu_high_since is None: _cpu_high_since = time.time() elif time.time() - _cpu_high_since >= CPU_WARN_DURATION: if not _in_cooldown("cpu_warn"): msg = f"CPU {pct:.1f}% が {CPU_WARN_DURATION}秒継続中" notify("🟡 Jarvis Safety: CPU高負荷", msg) log({"type": "cpu", "level": "warn", "detail": f"{pct:.1f}%", "action_taken": "notified"}) else: _cpu_high_since = None def _proc_name_match(p: psutil.Process, names: tuple) -> bool: """name または cmdline 先頭basename が names に厳密一致したら True。""" try: n = (p.info.get("name") or "").strip().lower() if n in names: return True cmdline = p.info.get("cmdline") or [] if cmdline: first = Path(cmdline[0]).name.strip().lower() if first in names: return True except (psutil.NoSuchProcess, psutil.AccessDenied): pass return False def _procs_by_names(names: tuple) -> List[psutil.Process]: procs = [] for p in psutil.process_iter(["name", "pid", "create_time", "cmdline"]): try: if _proc_name_match(p, names): procs.append(p) except (psutil.NoSuchProcess, psutil.AccessDenied): pass return sorted(procs, key=lambda p: p.info.get("create_time", 0)) def _is_whisper_target(p: psutil.Process) -> bool: """whisper 多重カウント対象なら True（多重起動カウント専用・kill 判定とは別）。 1) name または cmdline 先頭 basename が WHISPER_TARGET_NAMES に厳密一致 2) cmdline 全体のどこかに WHISPER_CMDLINE_TOKENS（例: mlx_whisper）を含む → `python -m mlx_whisper` は name=Python で(1)を貫通するため(2)で救う。 注意: 本関数は「多重起動カウント」用。長寿命 kill は名前別ループ (_procs_by_names) で whisper-cli / whisper-stream を厳密一致で扱う。 """ try: name = (p.info.get("name") or "").strip().lower() if name in WHISPER_TARGET_NAMES: return True # cmdline 先頭の basename もチェック（subprocess 経由対策） cmdline = p.info.get("cmdline") or [] if cmdline: first = Path(cmdline[0]).name.strip().lower() if first in WHISPER_TARGET_NAMES: return True # cmdline 全体に mlx_whisper 等のトークンを含むか（モジュール起動対策） joined = " ".join(cmdline).lower() if any(tok in joined for tok in WHISPER_CMDLINE_TOKENS): return True except (psutil.NoSuchProcess, psutil.AccessDenied): pass return False def _whisper_procs() -> List[psutil.Process]: procs = [] for p in psutil.process_iter(["name", "pid", "create_time", "cmdline"]): try: if _is_whisper_target(p): procs.append(p) except (psutil.NoSuchProcess, psutil.AccessDenied): pass return sorted(procs, key=lambda p: p.info.get("create_time", 0)) def check_whisper() -> None: procs = _whisper_procs() count = len(procs) # multi-instance kill if count >= WHISPER_KILL_COUNT: if not _in_cooldown("whisper_kill"): # kill oldest first, keep newest 2 targets = procs[: count - 2] actions = [_kill_process(p, "多重起動") for p in targets] msg = f"whisper-cli {count}プロセス検出 → 古いものを強制終了" notify("🔴 Jarvis Safety: whisper多重起動Kill", msg, critical=True) log({"type": "whisper_count", "level": "kill", "detail": f"count={count}", "action_taken": actions}) elif count >= WHISPER_WARN_COUNT: if not _in_cooldown("whisper_warn"): notify("🟡 Jarvis Safety: whisper多重起動", f"whisper-cli {count}プロセス同時稼働中") log({"type": "whisper_count", "level": "warn", "detail": f"count={count}", "action_taken": "notified"}) # long-running detection（60秒閾値）は whisper-cli のみ対象。 # 2026-06-30 修正: 旧コードは procs（cli+stream+mlx_whisper 全部）を回し # age>=60s で kill していたため、常駐前提の whisper-stream が 60秒で誤kill され # 3時間免除（下の専用ループ）が一度も効かなかった。cli 専用に分離して解消。 now = time.time() cli_procs = _procs_by_names(WHISPER_CLI_NAMES) for p in cli_procs: try: age = now - p.create_time() if age >= WHISPER_LONG_RUN_SEC: key = f"whisper_long_{p.pid}" if not _in_cooldown(key): action = _kill_process(p, f"長時間稼働 {age:.0f}s") notify("🔴 Jarvis Safety: whisper-cli長時間稼働", f"pid {p.pid} が {age:.0f}秒稼働 → 終了", critical=True) log({"type": "whisper_long", "level": "kill", "detail": f"pid={p.pid} age={age:.0f}s", "action_taken": action}) except (psutil.NoSuchProcess, psutil.AccessDenied): pass # whisper-stream: 3時間超で止め忘れ判定 → kill（常駐前提なので 60秒では絶対killしない） # 2026-05-09 追加。1日9時間放置事故対策。MARK Ⅶ正規運用は3時間以内想定。 stream_procs = _procs_by_names(WHISPER_STREAM_NAMES) for p in stream_procs: try: age = now - p.create_time() if age >= WHISPER_STREAM_LONG_RUN_SEC: key = f"whisper_stream_long_{p.pid}" if not _in_cooldown(key): hrs = age / 3600 action = _kill_process(p, f"whisper-stream 止め忘れ {hrs:.1f}h") notify( "🔴 Jarvis Safety: whisper-stream 止め忘れ", f"pid {p.pid} が {hrs:.1f}時間稼働 → 自動終了", critical=True, ) log({ "type": "whisper_stream_long", "level": "kill", "detail": f"pid={p.pid} age={age:.0f}s ({hrs:.1f}h)", "action_taken": action, }) except (psutil.NoSuchProcess, psutil.AccessDenied): pass # ── Signal handling ─────────────────────────────────────────────────────────── def _shutdown(signum, frame) -> None: global _running _running = False log({"type": "shutdown", "level": "info", "detail": f"signal={signum}"}) # ── Entry point ─────────────────────────────────────────────────────────────── def test_notify() -> None: notify("✅ Jarvis Safety: テスト通知", "resource_guard は正常に動作しています", critical=True) print("Test notification sent (critical dialog).") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--test-notify", action="store_true") args = parser.parse_args() if args.test_notify: test_notify() return LOG_DIR.mkdir(parents=True, exist_ok=True) signal.signal(signal.SIGINT, _shutdown) signal.signal(signal.SIGTERM, _shutdown) log({"type": "startup", "level": "info", "pid": psutil.Process().pid, "version": VERSION}) while _running: for check in (check_memory, check_cpu, check_whisper): try: check() except Exception as e: log({"type": "error", "level": "error", "detail": str(e)}) time.sleep(LOOP_INTERVAL) if __name__ == "__main__": main()