""" safety_guard.py — Lv2 Jarvis常駐デーモン 安全装置モジュール トークン消費上限・モード管理・dry-run・whisper-cli監視 """ from __future__ import annotations from enum import Enum from dataclasses import dataclass from datetime import datetime, timedelta from pathlib import Path from typing import Optional import json import subprocess import time class Mode(Enum): WAITING = "🟢" # 待機モード MEETING = "🟡" # 会議モード LEARNING = "🔴" # 学習モード # モード別1日上限（tokens） DAILY_LIMITS = { Mode.WAITING: 30_000, Mode.MEETING: 200_000, Mode.LEARNING: 500_000, } # 1fire 上限警告しきい値 FIRE_LIMIT_TOKENS = 8_000 # モード自動降格時間 AUTO_DEMOTE_HOURS = 3 # whisper-cli プロセス数の上限（過去事故：26並列で爆死） WHISPER_PROCESS_LIMIT = 5 TOKENS_LOG_PATH = Path(__file__).parent.parent / "tokens.jsonl" MODE_STATE_PATH = Path(__file__).parent.parent / "mode_state.json" @dataclass class ModeState: current: Mode started_at: str # ISO8601 last_activity_at: str def to_dict(self) -> dict: return { "current": self.current.name, "started_at": self.started_at, "last_activity_at": self.last_activity_at, } @classmethod def from_dict(cls, d: dict) -> "ModeState": return cls( current=Mode[d["current"]], started_at=d["started_at"], last_activity_at=d["last_activity_at"], ) def _now_iso() -> str: return datetime.now().astimezone().isoformat() def load_mode() -> ModeState: """mode_state.json から現在のモードを読む。存在しなければ WAITING で初期化""" if not MODE_STATE_PATH.exists(): state = ModeState( current=Mode.WAITING, started_at=_now_iso(), last_activity_at=_now_iso(), ) save_mode(state) return state try: data = json.loads(MODE_STATE_PATH.read_text(encoding="utf-8")) return ModeState.from_dict(data) except Exception as e: print(f"[safety_guard] load_mode 失敗、WAITING にフォールバック: {e}") return ModeState( current=Mode.WAITING, started_at=_now_iso(), last_activity_at=_now_iso(), ) def save_mode(state: ModeState) -> None: """mode_state.json に書き込む""" try: MODE_STATE_PATH.parent.mkdir(parents=True, exist_ok=True) MODE_STATE_PATH.write_text( json.dumps(state.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8", ) except Exception as e: print(f"[safety_guard] save_mode 失敗: {e}") def set_mode(new_mode: Mode) -> ModeState: """モード切替。started_at を現在時刻に。""" now = _now_iso() state = ModeState( current=new_mode, started_at=now, last_activity_at=now, ) save_mode(state) return state def check_auto_demote() -> tuple[bool, ModeState]: """ 現在モードが MEETING/LEARNING で、started_at から AUTO_DEMOTE_HOURS 経過してたら WAITING に降格。 戻り値: (降格したか, 現在のModeState) """ state = load_mode() if state.current == Mode.WAITING: return False, state try: started = datetime.fromisoformat(state.started_at) except Exception as e: print(f"[safety_guard] check_auto_demote: started_at パース失敗: {e}") return False, state elapsed = datetime.now().astimezone() - started if elapsed >= timedelta(hours=AUTO_DEMOTE_HOURS): print( f"[safety_guard] 自動降格: {state.current.value} → {Mode.WAITING.value} " f"({elapsed.seconds // 3600}h{(elapsed.seconds % 3600) // 60}m 経過)" ) new_state = set_mode(Mode.WAITING) return True, new_state return False, state def get_today_tokens() -> int: """tokens.jsonl から今日（ローカルタイム）の合計消費 tokens を集計""" if not TOKENS_LOG_PATH.exists(): return 0 today = datetime.now().date() total = 0 try: lines = TOKENS_LOG_PATH.read_text(encoding="utf-8").splitlines() except Exception as e: print(f"[safety_guard] get_today_tokens: ファイル読込失敗: {e}") return 0 for line in lines: line = line.strip() if not line: continue try: entry = json.loads(line) ts = datetime.fromisoformat(entry["ts"]) if ts.date() == today: total += int(entry.get("tokens", 0)) except Exception as e: print(f"[safety_guard] トークンログ行スキップ: {e} | {line[:80]}") return total def can_fire(estimated_tokens: int = 5000) -> tuple[bool, str]: """ 発火可否判定。 判定： 1. 現在モード読込 2. 自動降格チェック 3. 今日の消費 + estimated_tokens がモード上限を超えるか 4. 1fire見積が FIRE_LIMIT_TOKENS 超えてないか 戻り値: (発火OKか, 理由文) """ _, state = check_auto_demote() mode = state.current limit = DAILY_LIMITS[mode] today = get_today_tokens() # 1fire 単体サイズ警告 if estimated_tokens > FIRE_LIMIT_TOKENS: return False, ( f"1fire見積 {estimated_tokens:,} tok が上限 {FIRE_LIMIT_TOKENS:,} tok 超過" ) # 今日の累計チェック projected = today + estimated_tokens if projected > limit: return False, ( f"{mode.value} 1日上限 {limit:,} tok 超過見込み " f"(今日 {today:,} + 見積 {estimated_tokens:,} = {projected:,})" ) # 80% 警告（通過はするが警告文をログに出す） warning = warn_threshold(today, mode) if warning: print(f"[safety_guard] {warning}") return True, ( f"OK ({mode.value} 今日 {today:,}/{limit:,} tok, 見積 +{estimated_tokens:,})" ) def warn_threshold(today_tokens: int, mode: Mode) -> Optional[str]: """ 80% 到達警告。 今日の消費がモード上限の80%以上なら警告文を返す（None なら警告なし）。 """ limit = DAILY_LIMITS[mode] ratio = today_tokens / limit if limit > 0 else 0 if ratio >= 0.8: return ( f"⚠️ {mode.value} 残量警告: {today_tokens:,}/{limit:,} tok " f"({ratio*100:.0f}% 消費済み)" ) return None def log_fire(actual_tokens: int, dry_run: bool = False) -> None: """ 発火を tokens.jsonl に追記。 形式: {"ts": ISO8601, "mode": "WAITING", "tokens": 4500, "dry_run": false} """ state = load_mode() entry = { "ts": _now_iso(), "mode": state.current.name, "tokens": actual_tokens, "dry_run": dry_run, } try: TOKENS_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with TOKENS_LOG_PATH.open("a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") except Exception as e: print(f"[safety_guard] log_fire 失敗: {e}") def check_whisper_processes() -> tuple[int, bool]: """ whisper-cli プロセス数を ps で数える。 上限超えたら True を返す（後段で警告 or 強制停止判断）。 戻り値: (現在数, 上限超えてるか) """ try: result = subprocess.run( ["ps", "-ax", "-o", "pid,comm"], capture_output=True, text=True, timeout=5, ) lines = result.stdout.splitlines() count = sum(1 for line in lines if "whisper-cli" in line) over = count > WHISPER_PROCESS_LIMIT return count, over except Exception as e: print(f"[safety_guard] check_whisper_processes 失敗: {e}") return 0, False class DryRun: """dry-runコンテキストマネージャ。with DryRun(): で囲んだ間は actual fire しない""" _enabled = False def __enter__(self): DryRun._enabled = True return self def __exit__(self, *args): DryRun._enabled = False @classmethod def is_active(cls) -> bool: return cls._enabled # main: 自己テスト if __name__ == "__main__": print(f"現在モード: {load_mode().current.value}") print(f"今日の消費: {get_today_tokens()} tok") ok, reason = can_fire(5000) print(f"発火可否: {ok} ({reason})") count, over = check_whisper_processes() print(f"whisper-cli プロセス: {count}個 {'⚠️ 上限超過' if over else '✅ OK'}") # dry-run テスト with DryRun(): print(f"dry-run モード: {DryRun.is_active()}") log_fire(actual_tokens=1234, dry_run=DryRun.is_active()) print(f"dry-run 終了後: {DryRun.is_active()}") # 自動降格テスト表示 demoted, state = check_auto_demote() print(f"自動降格: {demoted} / 現在: {state.current.value} (開始: {state.started_at})")