#!/usr/bin/env python3 """Browser CPU guard — notifies when Safari/Chrome/WindowServer CPU spikes. CPU高負荷でブラウザ系が重くなる再発防止。 - Safari / Chrome 全プロセス合計CPU 50%超 30秒継続 → actionable dialog - WindowServer CPU 50%超 60秒継続 → ウィンドウ整理推奨 dialog - 通知クールダウン: 同一カテゴリ10分1回 """ 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, Optional import psutil # ── Thresholds ──────────────────────────────────────────────────────────────── # 2026-06-30 修正: psutil の cpu_percent は「1コア=100%」基準で合計するとコア数倍に # 膨らむ（10コア機なら理論上限 1000%）。旧 50% 単純合計はコア未正規化で慢性誤検知し # 10分おきにダイアログ連射していた。→ コア数で正規化（マシン全体に対する%）して比較。 CORE_COUNT: int = psutil.cpu_count() or 1 # 「マシン全体CPUの何%をブラウザ系が占めたら警告か」。50%＝マシンの半分をブラウザが食う水準。 BROWSER_CPU_THRESHOLD: float = 50.0 # % （正規化後・マシン全体比）ブラウザ合計CPU → 警告 BROWSER_WARN_DURATION: int = 30 # 秒継続で発火 WINDOWSERVER_CPU_THRESHOLD: float = 50.0 # % （正規化後・マシン全体比） WINDOWSERVER_WARN_DURATION: int = 60 COOLDOWN: int = 1800 # 秒（30分）同一カテゴリクールダウン（連射防止強化） LOOP_INTERVAL: int = 5 # ループ間隔（秒） VERSION: str = "1.1.0" # ───────────────────────────────────────────────────────────────────────────── LOG_DIR = Path.home() / "jarvis_safety" LOG_FILE = LOG_DIR / "browser_cpu_log.jsonl" _running: bool = True _cooldown_last: Dict[str, float] = {} # カテゴリ別「高負荷開始時刻」 _browser_high_since: Optional[float] = None _windowserver_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") # ── Cooldown ───────────────────────────────────────────────────────────────── 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 # ── Notification ────────────────────────────────────────────────────────────── def _esc(s: str) -> str: return s.replace("\\", "\\\\").replace('"', '\\"') # 直近に起動したダイアログ Popen ハンドル群（堆積防止のため毎回 reap する） _dialog_procs: list[subprocess.Popen] = [] # ダイアログ自動消滅（秒）。これを過ぎると AppleScript 側で勝手に閉じる。 # 旧コードは閉じるまで osascript が永続ブロック → AppleEvent/プロセス堆積の原因。 DIALOG_GIVE_UP_SEC: int = 300 def _reap_dialogs() -> None: """終了済みダイアログ Popen を回収（ゾンビ/AppleEvent堆積防止）。""" global _dialog_procs alive = [] for proc in _dialog_procs: if proc.poll() is None: alive.append(proc) # まだ表示中（ユーザー操作待ち） # poll() != None なら終了済み＝回収完了（再 append しない） _dialog_procs = alive def _actionable_dialog(title: str, body: str, open_activity_monitor: bool = True) -> None: """「重いタブを閉じて」+「Activity Monitor開く？」の2択ダイアログ。 2026-06-30 修正: `giving up after` で自動消滅させ、osascript が永続ブロックして AppleEvent/プロセスが堆積するのを防ぐ。起動ハンドルは _dialog_procs に積み 毎回 _reap_dialogs() で回収する。 """ _reap_dialogs() if open_activity_monitor: script = f""" tell application "System Events" set theResult to (display alert "{_esc(title)}" ¬ message "{_esc(body)}" ¬ as critical ¬ buttons {{"Activity Monitor を開く", "閉じる"}} ¬ default button 1 ¬ giving up after {DIALOG_GIVE_UP_SEC}) if (gave up of theResult) is false and (button returned of theResult) = "Activity Monitor を開く" then do shell script "open -a 'Activity Monitor'" end if end tell """ else: script = f""" tell application "System Events" display alert "{_esc(title)}" message "{_esc(body)}" as critical ¬ giving up after {DIALOG_GIVE_UP_SEC} end tell """ _dialog_procs.append(subprocess.Popen(["osascript", "-e", script])) def notify_simple(title: str, body: str) -> None: script = f'display notification "{_esc(body)}" with title "{_esc(title)}"' subprocess.run(["osascript", "-e", script], capture_output=True) # ── CPU collection ──────────────────────────────────────────────────────────── def _collect_browser_cpu() -> tuple[float, float]: """(safari_total_cpu, chrome_total_cpu) を返す。値はコア数正規化済（マシン全体比%）。""" safari_cpu = 0.0 chrome_cpu = 0.0 # psutil.process_iter の cpu_percent は 2回目以降のコールで差分計算される。 # interval=None で都度取得（ループ5秒間隔なので十分）。 for p in psutil.process_iter(["name", "cpu_percent"]): try: name = (p.info.get("name") or "").lower() cpu = p.info.get("cpu_percent") or 0.0 if "safari" in name or "webkit" in name: safari_cpu += cpu elif "google chrome" in name or "chrome" in name: chrome_cpu += cpu except (psutil.NoSuchProcess, psutil.AccessDenied): pass # コア数で正規化（1コア=100%基準の合計 → マシン全体に対する%）。誤検知防止の要。 return safari_cpu / CORE_COUNT, chrome_cpu / CORE_COUNT def _collect_windowserver_cpu() -> float: """WindowServer の CPU をコア数正規化（マシン全体比%）で返す。""" for p in psutil.process_iter(["name", "cpu_percent"]): try: name = (p.info.get("name") or "").lower() if name == "windowserver": return (p.info.get("cpu_percent") or 0.0) / CORE_COUNT except (psutil.NoSuchProcess, psutil.AccessDenied): pass return 0.0 # ── Checks ──────────────────────────────────────────────────────────────────── def check_browser_cpu() -> None: global _browser_high_since safari_cpu, chrome_cpu = _collect_browser_cpu() total = safari_cpu + chrome_cpu if total >= BROWSER_CPU_THRESHOLD: if _browser_high_since is None: _browser_high_since = time.time() elif time.time() - _browser_high_since >= BROWSER_WARN_DURATION: if not _in_cooldown("browser_cpu"): parts = [] if safari_cpu > 0: parts.append(f"Safari {safari_cpu:.0f}%") if chrome_cpu > 0: parts.append(f"Chrome {chrome_cpu:.0f}%") detail = " + ".join(parts) if parts else f"合計 {total:.0f}%" title = "🔴 Jarvis Safety: ブラウザCPU高負荷" body = ( f"{detail} が {BROWSER_WARN_DURATION}秒継続中。\n" "重いタブ・拡張機能を閉じてください。" ) _actionable_dialog(title, body, open_activity_monitor=True) log({ "type": "browser_cpu", "level": "warn", "detail": f"safari={safari_cpu:.1f}% chrome={chrome_cpu:.1f}% total={total:.1f}%", "action_taken": "dialog_shown", }) else: _browser_high_since = None def check_windowserver_cpu() -> None: global _windowserver_high_since cpu = _collect_windowserver_cpu() if cpu >= WINDOWSERVER_CPU_THRESHOLD: if _windowserver_high_since is None: _windowserver_high_since = time.time() elif time.time() - _windowserver_high_since >= WINDOWSERVER_WARN_DURATION: if not _in_cooldown("windowserver_cpu"): title = "🟡 Jarvis Safety: WindowServer高負荷" body = ( f"WindowServer CPU {cpu:.0f}% が {WINDOWSERVER_WARN_DURATION}秒継続中。\n" "不要なウィンドウ・デスクトップを整理してください。" ) _actionable_dialog(title, body, open_activity_monitor=True) log({ "type": "windowserver_cpu", "level": "warn", "detail": f"cpu={cpu:.1f}%", "action_taken": "dialog_shown", }) else: _windowserver_high_since = None # ── 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: _actionable_dialog( "✅ Jarvis Safety: ブラウザCPU guard テスト", "browser_cpu_guard は正常に動作しています。このダイアログが出れば成功。", open_activity_monitor=True, ) print("Test dialog sent.") def main() -> None: parser = argparse.ArgumentParser(description="Jarvis Browser CPU Guard") parser.add_argument("--test-notify", action="store_true", help="テスト通知を送って終了") 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}) # psutil の cpu_percent は初回 0.0 を返す仕様 → 1サイクル空回し for p in psutil.process_iter(["name", "cpu_percent"]): pass time.sleep(LOOP_INTERVAL) while _running: for check in (check_browser_cpu, check_windowserver_cpu): try: check() except Exception as e: log({"type": "error", "level": "error", "detail": str(e)}) time.sleep(LOOP_INTERVAL) if __name__ == "__main__": main()