""" Jarvis Lv2 常駐デーモン メインループ 中期目標 Phase 1：呼名検出型 Jarvis（映画通り） - ずっと聞いてる、呼ばれた時だけ反応する - 誤発火しても気にしない、学習素材として歓迎 - Claude Code 経由で動く（追加課金ゼロ） 仕様書：../SPEC.md """ import sys import time import json import signal from datetime import datetime, timedelta from pathlib import Path from collections import deque from typing import Optional # --------------------------------------------------------------------------- # モジュール import（未完成モジュールは警告のみ・全体は止めない） # --------------------------------------------------------------------------- sys.path.insert(0, str(Path(__file__).parent)) _import_errors: list[str] = [] try: from wake_word_detector import evaluate, detect_false_positive_marker except ImportError as e: _import_errors.append(f"wake_word_detector: {e}") def evaluate(transcript): # type: ignore[misc] return {"should_fire": False, "wake_word": None, "after_text": None, "ng_matched": False, "ng_pattern": None, "is_command": False, "reason": "module_unavailable"} def detect_false_positive_marker(text): # type: ignore[misc] return False try: from safety_guard import ( Mode, load_mode, set_mode, check_auto_demote, get_today_tokens, can_fire, warn_threshold, log_fire, check_whisper_processes, DryRun, DAILY_LIMITS ) except ImportError as e: _import_errors.append(f"safety_guard: {e}") class Mode: # type: ignore[no-redef] WAITING = "WAITING" MEETING = "MEETING" LEARNING = "LEARNING" def load_mode(): # type: ignore[misc] class _S: current = Mode.WAITING return _S() def set_mode(m): pass # type: ignore[misc] def check_auto_demote(): # type: ignore[misc] return False, load_mode() def get_today_tokens(): # type: ignore[misc] return 0 def can_fire(estimated_tokens=0): # type: ignore[misc] return True, "ok" def warn_threshold(today_tokens, limit): # type: ignore[misc] return False def log_fire(excerpt, tokens_used=0): pass # type: ignore[misc] def check_whisper_processes(): # type: ignore[misc] return [] class DryRun: # type: ignore[no-redef] @staticmethod def is_active(): return False def __enter__(self): return self def __exit__(self, *a): pass DAILY_LIMITS = {Mode.WAITING: 100000, Mode.MEETING: 200000, Mode.LEARNING: 50000} try: from code_runner import fire as run_claude_fire except ImportError as e: _import_errors.append(f"code_runner: {e}") def run_claude_fire(transcript_excerpt, recent_history=None, mode="WAITING", dry_run=False): # type: ignore[misc] return {"ok": False, "stdout": "", "stderr": "module_unavailable"} try: from notifier import ( write_dashboard_card, mark_false_positive, write_notification, notify_mode_switch_suggestion, notify_mode_demote_warning, notify_false_positive_burst, notify_token_threshold, notify_token_exhausted ) except ImportError as e: _import_errors.append(f"notifier: {e}") def write_dashboard_card(**kwargs): pass # type: ignore[misc] def mark_false_positive(**kwargs): pass # type: ignore[misc] def write_notification(icon, msg): print(f"[notify] {icon} {msg}") # type: ignore[misc] def notify_mode_switch_suggestion(count): print(f"[notify] モード切替提案: {count}回") # type: ignore[misc] def notify_mode_demote_warning(**kwargs): pass # type: ignore[misc] def notify_false_positive_burst(count): print(f"[notify] 誤発火多発: {count}回") # type: ignore[misc] def notify_token_threshold(used, limit): print(f"[notify] トークン80%: {used}/{limit}") # type: ignore[misc] def notify_token_exhausted(used, limit): print(f"[notify] トークン上限: {used}/{limit}") # type: ignore[misc] # --------------------------------------------------------------------------- # 設定 # --------------------------------------------------------------------------- TRANSCRIPTS_DIR = Path(__file__).parent.parent / "v2" / "transcripts" STATE_PATH = Path(__file__).parent.parent / "daemon_state.json" FALSE_POSITIVES_PATH = Path(__file__).parent.parent / "wake_word_learning" / "false_positives.jsonl" POLL_INTERVAL_SEC = 2 WHISPER_CHECK_INTERVAL_SEC = 3600 # 1時間ごとに whisper プロセス監視 # 統計バッファ（直近1時間/30分の発火・誤発火カウント） recent_fires: deque = deque() # (datetime, transcript_excerpt) recent_false_positives: deque = deque() # (datetime, transcript_excerpt) # --------------------------------------------------------------------------- # グローバル状態 # --------------------------------------------------------------------------- shutdown_requested = False _last_whisper_check: datetime = datetime.min # --------------------------------------------------------------------------- # シグナルハンドラ # --------------------------------------------------------------------------- def handle_shutdown(signum: int, frame) -> None: global shutdown_requested print(f"\n[shutdown] signal {signum} received. Saving state...") shutdown_requested = True # --------------------------------------------------------------------------- # transcript パス # --------------------------------------------------------------------------- def get_today_transcript_path() -> Path: today = datetime.now().strftime("%Y-%m-%d") return TRANSCRIPTS_DIR / f"{today}.jsonl" # --------------------------------------------------------------------------- # デーモン状態永続化 # --------------------------------------------------------------------------- def load_daemon_state() -> dict: """daemon_state.json から最終読込位置などを読む""" try: return json.loads(STATE_PATH.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): return {} def save_daemon_state(state: dict) -> None: STATE_PATH.parent.mkdir(parents=True, exist_ok=True) STATE_PATH.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") # --------------------------------------------------------------------------- # transcript ファイル読み取り（tail-f 風） # --------------------------------------------------------------------------- def read_new_lines(path: Path, last_offset: int) -> tuple[list[dict], int]: """ transcript ファイルから last_offset バイト以降の新しい行を読む。 戻り値: (新エントリリスト, 新しい byte offset) """ entries: list[dict] = [] try: size = path.stat().st_size if size <= last_offset: return [], last_offset with path.open("r", encoding="utf-8") as f: f.seek(last_offset) for raw in f: raw = raw.strip() if not raw: continue try: entries.append(json.loads(raw)) except json.JSONDecodeError: # 書き込み途中の行はスキップ pass new_offset = f.tell() return entries, new_offset except (FileNotFoundError, OSError): return [], last_offset # --------------------------------------------------------------------------- # モード切替検出 # --------------------------------------------------------------------------- def detect_mode_switch(text: str) -> Optional[object]: """transcript 内のモード切替コマンドを検出""" if "ジャービス会議モード" in text or "ジャービス会議" in text: return Mode.MEETING if "ジャービス学習モード" in text: return Mode.LEARNING if "ジャービス待機モード" in text or "ジャービス通常モード" in text: return Mode.WAITING return None # --------------------------------------------------------------------------- # 統計バッファ管理 # --------------------------------------------------------------------------- def cleanup_recent_buffers() -> None: """直近1時間/30分より古いエントリを削除""" cutoff_1h = datetime.now() - timedelta(hours=1) cutoff_30min = datetime.now() - timedelta(minutes=30) while recent_fires and recent_fires[0][0] < cutoff_1h: recent_fires.popleft() while recent_false_positives and recent_false_positives[0][0] < cutoff_30min: recent_false_positives.popleft() # --------------------------------------------------------------------------- # 通知トリガー判定（毎ループ末尾） # --------------------------------------------------------------------------- def check_periodic_alerts(mode, today_tokens: int) -> bool: """ 各種通知条件を判定。 戻り値: False = 上限到達につき停止すべき """ cleanup_recent_buffers() # 1時間に5回超 → 会議モード提案 if getattr(mode, "name", mode) == "WAITING" and len(recent_fires) >= 5: notify_mode_switch_suggestion(len(recent_fires)) # 30分に3回超 → 学習モード提案 if getattr(mode, "name", mode) != "LEARNING" and len(recent_false_positives) >= 3: notify_false_positive_burst(len(recent_false_positives)) # トークン80% / 上限 limit = DAILY_LIMITS.get(mode, DAILY_LIMITS.get(getattr(mode, "WAITING", "WAITING"), 100000)) if today_tokens >= limit: notify_token_exhausted(today_tokens, limit) return False # 停止シグナル if today_tokens >= limit * 0.8: notify_token_threshold(today_tokens, limit) return True # --------------------------------------------------------------------------- # 誤発火ログ記録 # --------------------------------------------------------------------------- def log_false_positive_entry(entry: dict) -> None: """false_positives.jsonl に記録（誤発火学習用）""" FALSE_POSITIVES_PATH.parent.mkdir(parents=True, exist_ok=True) record = { "ts": entry.get("ts", datetime.now().isoformat()), "text": entry.get("text", ""), "marker_detected": True, } with FALSE_POSITIVES_PATH.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") # --------------------------------------------------------------------------- # 1エントリ処理 # --------------------------------------------------------------------------- _PROCESSED_CHUNKS_PATH = Path(__file__).parent.parent / "processed_chunks.json" _PROCESSED_CHUNKS_MAX = 200 # 古いhash累積を防ぐためのサイズ上限 def _load_processed_chunks() -> set: try: data = json.loads(_PROCESSED_CHUNKS_PATH.read_text(encoding="utf-8")) # サイズ上限超えてたら古い半分を切り捨て（FIFO） if isinstance(data, list) and len(data) > _PROCESSED_CHUNKS_MAX: data = data[-_PROCESSED_CHUNKS_MAX:] return set(data) except Exception: return set() def _save_processed_chunks(s: set) -> None: try: _PROCESSED_CHUNKS_PATH.write_text(json.dumps(list(s), ensure_ascii=False), encoding="utf-8") except Exception as e: print(f"[warn] processed_chunks 保存失敗: {e}") _processed_chunks: set = _load_processed_chunks() def process_entry(entry: dict, mode, dry_run: bool = False) -> None: """transcript の1エントリを処理""" text = entry.get("text", "") if not text: return # L/R 重複排除：同じ src_chunk + 同じ text なら1回だけ処理（daemon再起動跨ぎ） src_chunk = entry.get("src_chunk", "") dedup_key = f"{src_chunk}:{hash(text)}" if dedup_key in _processed_chunks: return _processed_chunks.add(dedup_key) _save_processed_chunks(_processed_chunks) # ① モード切替検出（最優先） new_mode = detect_mode_switch(text) if new_mode is not None: set_mode(new_mode) mode_label = getattr(new_mode, "name", str(new_mode)) write_notification("ℹ️", f"モード切替: {mode_label}") return # ② 「違う」誤発火マーカー検出 if detect_false_positive_marker(text): recent_false_positives.append((datetime.now(), text)) mark_false_positive() write_notification("⚠️", "誤発火マーク追加") log_false_positive_entry(entry) return # ③ 呼名検出 + フィルタ result = evaluate(text) if not result.get("should_fire"): return # 静かに無視 # ④ 安全装置チェック ok, reason = can_fire(estimated_tokens=5000) if not ok: write_notification("⚠️", f"発火スキップ: {reason}") return # ⑤ 発火実行 mode_name = getattr(mode, "name", str(mode)) print(f"[fire] {datetime.now().strftime('%H:%M:%S')} mode={mode_name} text={text[:60]}") # ⑤-a 即時 placeholder カード書き込み（「考え中...」表示・dashboardで進行可視化） placeholder_id = None if not (dry_run or DryRun.is_active()): try: from notifier import write_dashboard_card as _wdc placeholder_id = datetime.now().strftime("%H:%M:%S") _wdc( transcript_excerpt=text, jarvis_response=f"💭 考え中...（約13秒・id={placeholder_id}）", mode=mode_name, ) # mirrorも即実行（CloudStorageキャッシュ即時反映） try: import subprocess as _sp _sp.run(["bash", "/Users/rk/.local/jarvis/mirror_to_dashboard.sh"], capture_output=True, timeout=8) except Exception: pass except Exception as e: print(f"[warn] placeholder書込失敗: {e}") if dry_run or DryRun.is_active(): result_fire = run_claude_fire(text, mode=mode_name, dry_run=True) else: result_fire = run_claude_fire(text, mode=mode_name, dry_run=False) recent_fires.append((datetime.now(), text)) actual_tokens = result_fire.get("estimated_input_tokens", 0) + result_fire.get("estimated_output_tokens", 0) log_fire(actual_tokens=actual_tokens, dry_run=(dry_run or DryRun.is_active())) # ⑥ 実回答書き込み（placeholder の後に追記される形） if not (dry_run or DryRun.is_active()): if result_fire.get("ok"): try: from notifier import write_dashboard_card as _wdc _wdc( transcript_excerpt=text, jarvis_response=result_fire.get("stdout", "(no output)"), mode=mode_name, ) # 実回答書き込み後にmirrorを再実行（最新反映） try: import subprocess as _sp _sp.run(["bash", "/Users/rk/.local/jarvis/mirror_to_dashboard.sh"], capture_output=True, timeout=8) except Exception: pass except Exception as e: print(f"[error] dashboard書込失敗: {e}") else: err = result_fire.get("stderr", "")[:100] write_notification("⚠️", f"Claude実行失敗: {err}") # --------------------------------------------------------------------------- # whisper-cli プロセス監視 # --------------------------------------------------------------------------- def maybe_check_whisper() -> None: """起動時 + 1時間ごとに check_whisper_processes を実行""" global _last_whisper_check now = datetime.now() if (now - _last_whisper_check).total_seconds() < WHISPER_CHECK_INTERVAL_SEC: return _last_whisper_check = now procs = check_whisper_processes() if procs: msg = f"whisper-cli プロセス検出: {len(procs)}件 — 過去事故再発防止チェック" print(f"[whisper] {msg}") write_notification("⚠️", msg) # --------------------------------------------------------------------------- # メインループ # --------------------------------------------------------------------------- def main_loop(dry_run: bool = False) -> None: print("=== Jarvis Lv2 Daemon 起動 ===") if _import_errors: for err in _import_errors: print(f"[warn] import 失敗（フォールバック動作）: {err}") print(f"dry_run: {dry_run}") print(f"transcripts: {TRANSCRIPTS_DIR}") print(f"state: {STATE_PATH}") # 起動時 offset 決定ロジック（優先順）： # 1. daemon_state.json があり、かつそれが現在の jsonl サイズ以下 → 続きから読む # 2. それ以外（state無し or サイズ超過の異常値）→ 現在の末尾から読む（過去再fire防止） today_path_init = get_today_transcript_path() current_size = today_path_init.stat().st_size if today_path_init.exists() else 0 saved_state = load_daemon_state() saved_offset = saved_state.get("last_offset", 0) saved_path = saved_state.get("last_path", "") # path不一致でも、offsetが現サイズ以下なら使う（path変更は無視） if 0 <= saved_offset <= current_size and saved_offset > 0: last_offset = saved_offset last_path = str(today_path_init) print(f"[init] daemon_state継続: offset={last_offset}/{current_size}") else: last_offset = current_size last_path = str(today_path_init) print(f"[init] 末尾から開始: offset={last_offset} (state無し or 異常値)") # 初回 whisper チェック maybe_check_whisper() while not shutdown_requested: try: # 自動降格チェック demoted, mode_state = check_auto_demote() if demoted: write_notification("ℹ️", "モード自動降格: 待機モード") mode = getattr(mode_state, "current", mode_state) today_tokens = get_today_tokens() # 通知トリガー判定 if not check_periodic_alerts(mode, today_tokens): print("[stop] トークン上限到達。停止します。") break # 定期 whisper 監視 maybe_check_whisper() # transcript 読み取り today_path = get_today_transcript_path() if str(today_path) != last_path: # 日付変わり → offset リセット last_offset = 0 last_path = str(today_path) if today_path.exists(): new_entries, new_offset = read_new_lines(today_path, last_offset) last_offset = new_offset for entry in new_entries: if shutdown_requested: break try: process_entry(entry, mode, dry_run=dry_run) except Exception as e: print(f"[error] process_entry: {e}") # 状態保存 save_daemon_state({"last_offset": last_offset, "last_path": last_path}) time.sleep(POLL_INTERVAL_SEC) except Exception as e: print(f"[error] main loop: {e}") time.sleep(POLL_INTERVAL_SEC * 2) # シャットダウン処理 save_daemon_state({"last_offset": last_offset, "last_path": last_path}) print("=== Jarvis Lv2 Daemon 停止 ===") # --------------------------------------------------------------------------- # エントリポイント # --------------------------------------------------------------------------- if __name__ == "__main__": signal.signal(signal.SIGINT, handle_shutdown) signal.signal(signal.SIGTERM, handle_shutdown) dry_run = "--dry-run" in sys.argv main_loop(dry_run=dry_run)