""" intervention_judge.py ===================== MARK Ⅶ 介入判定 — 本格判定（claude -p ラッパー） 軽量フィルタ（intervention_filter.py）を通過した場合のみ呼ばれる。 subprocess で `claude -p` を起動し、介入すべきか／介入文を生成させる。 Python 3.9 互換（match 文不使用） """ from __future__ import annotations import json import subprocess import sys import time from pathlib import Path from typing import Optional # --------------------------------------------------------------------------- # パス定数 # --------------------------------------------------------------------------- _V3_DIR = Path(__file__).parent _DNA_PATH = _V3_DIR.parent / "clone_dna.md" _INTERVENTION_LOG = _V3_DIR / "intervention_log.jsonl" # --------------------------------------------------------------------------- # 定数 # --------------------------------------------------------------------------- _CLAUDE_TIMEOUT_SEC = 60 _DNA_HEAD_LINES = 100 _RECENT_INTERVENTIONS_COUNT = 5 # --------------------------------------------------------------------------- # DNA 核心（先頭100行）を読み込む # --------------------------------------------------------------------------- def _load_dna_head() -> str: """clone_dna.md の先頭100行を返す。ファイルなければ空文字。""" if not _DNA_PATH.exists(): return "" lines = _DNA_PATH.read_text(encoding="utf-8").splitlines() return "\n".join(lines[:_DNA_HEAD_LINES]) # --------------------------------------------------------------------------- # 過去介入ログ（末尾N件）を読み込む # --------------------------------------------------------------------------- def _load_recent_interventions(n: int = _RECENT_INTERVENTIONS_COUNT) -> list: """intervention_log.jsonl の末尾 n 件を返す。""" if not _INTERVENTION_LOG.exists(): return [] try: lines = _INTERVENTION_LOG.read_text(encoding="utf-8").splitlines() recent = [] for line in reversed(lines): line = line.strip() if not line: continue try: recent.append(json.loads(line)) except json.JSONDecodeError: continue if len(recent) >= n: break return list(reversed(recent)) except Exception: return [] # --------------------------------------------------------------------------- # プロンプト組み立て # --------------------------------------------------------------------------- def _build_prompt( transcript_60sec: str, trigger_reason: str, matched_text: str, recent_interventions: Optional[list] = None, ) -> str: """claude -p に渡すプロンプトを組み立てる。""" dna = _load_dna_head() if recent_interventions is None: recent_interventions = _load_recent_interventions() recent_str = "" if recent_interventions: items = [] for entry in recent_interventions: ts = entry.get("timestamp", "") comment = entry.get("comment", "") reason = entry.get("trigger_reason", "") items.append(f"- [{ts}] ({reason}) {comment}") recent_str = "\n".join(items) else: recent_str = "（介入履歴なし）" prompt = f"""# J.A.R.V.I.S 介入判定タスク ## あなたの役割（DNA核心・先頭100行） {dna} --- ## 直近60秒のトランスクリプト {transcript_60sec if transcript_60sec else "（無音・沈黙）"} --- ## 軽量フィルタが検出したトリガー - trigger_reason: {trigger_reason} - matched_text: {matched_text} --- ## 過去Jarvis介入履歴（直近{_RECENT_INTERVENTIONS_COUNT}件） {recent_str} --- ## タスク指示 上記のトランスクリプトと文脈を踏まえ、Jarvisとして介入すべきか判定せよ。 介入すべきなら、60文字以内の介入コメントを生成せよ。 ## 出力形式（JSONのみ・他の文章不要） {{"intervene": "yes" or "no", "reason": "判定理由（1文）", "comment": "介入コメント（60字以内・intervene=noなら空文字）"}} """ return prompt # --------------------------------------------------------------------------- # JSON パース（救済ロジック込み） # --------------------------------------------------------------------------- def _parse_response(stdout: str) -> dict: """ claude -p のレスポンスを JSON パースする。 JSON が見つからなければ stdout 全体を comment として intervene=True で返す（救済）。 """ # JSON ブロックを探す（```json...``` or {...} を抽出） text = stdout.strip() # コードブロック除去 if "```" in text: import re code_block = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL) if code_block: text = code_block.group(1) # 単純な {...} 抽出 brace_start = text.find("{") brace_end = text.rfind("}") if brace_start != -1 and brace_end != -1 and brace_end > brace_start: candidate = text[brace_start:brace_end + 1] try: parsed = json.loads(candidate) intervene_val = parsed.get("intervene", "no") if isinstance(intervene_val, str): intervene_bool = intervene_val.lower() == "yes" else: intervene_bool = bool(intervene_val) return { "intervene": intervene_bool, "reason": str(parsed.get("reason", "")), "comment": str(parsed.get("comment", "")), "_parsed_ok": True, } except (json.JSONDecodeError, ValueError): pass # JSON パース失敗 → 救済: stdout 全体を comment 扱いで intervene=True return { "intervene": True, "reason": "json_parse_failed_fallback", "comment": stdout[:60].strip(), # 60字で打ち切り "_parsed_ok": False, } # --------------------------------------------------------------------------- # メイン判定関数 # --------------------------------------------------------------------------- def judge_and_generate( transcript_60sec: str, trigger_reason: str, matched_text: str, recent_interventions: Optional[list] = None, ) -> dict: """ claude -p を呼び出して介入判定＋介入文生成を行う。 Parameters ---------- transcript_60sec : str 直近60秒のトランスクリプト。 trigger_reason : str 軽量フィルタのトリガー理由。 matched_text : str 軽量フィルタのマッチテキスト。 recent_interventions : list | None 過去介入ログ（省略時は intervention_log.jsonl から自動取得）。 Returns ------- dict { "intervene": bool, "reason": str, "comment": str, # intervene=True 時のみ・60字以内 "tokens_estimated": int, "duration_sec": float, "raw_stdout": str, # デバッグ用 } """ # 🔴 2026-06-06 claude -p禁止ガード（6/15課金分割・REMINDER #55）— MARK Ⅶ再起動前にTUI化 raise RuntimeError("claude -p は6/15 Agent SDK課金分割で禁止（REMINDER #55）。MARK Ⅶ live_session 再起動時は TUI(spawn_interactive_claude相当)化が必要。") prompt = _build_prompt( transcript_60sec=transcript_60sec, trigger_reason=trigger_reason, matched_text=matched_text, recent_interventions=recent_interventions, ) tokens_estimated = len(prompt.split()) * 2 # 粗い推定 start = time.monotonic() # claude コマンドの存在確認 try: which_result = subprocess.run( ["which", "claude"], capture_output=True, text=True, timeout=5, ) if which_result.returncode != 0: return { "intervene": False, "reason": "claude_not_found", "comment": "", "tokens_estimated": tokens_estimated, "duration_sec": 0.0, "raw_stdout": "", } except Exception: return { "intervene": False, "reason": "claude_not_found", "comment": "", "tokens_estimated": tokens_estimated, "duration_sec": 0.0, "raw_stdout": "", } # claude -p 呼び出し try: result = subprocess.run( ["claude", "-p", prompt], capture_output=True, text=True, timeout=_CLAUDE_TIMEOUT_SEC, ) duration = time.monotonic() - start raw_stdout = result.stdout parsed = _parse_response(raw_stdout) return { "intervene": parsed["intervene"], "reason": parsed["reason"], "comment": parsed["comment"], "tokens_estimated": tokens_estimated, "duration_sec": round(duration, 2), "raw_stdout": raw_stdout, } except subprocess.TimeoutExpired: duration = time.monotonic() - start return { "intervene": False, "reason": "timeout", "comment": "", "tokens_estimated": tokens_estimated, "duration_sec": round(duration, 2), "raw_stdout": "", } except FileNotFoundError: return { "intervene": False, "reason": "claude_not_found", "comment": "", "tokens_estimated": tokens_estimated, "duration_sec": 0.0, "raw_stdout": "", } except Exception as e: duration = time.monotonic() - start return { "intervene": False, "reason": f"error: {e}", "comment": "", "tokens_estimated": tokens_estimated, "duration_sec": round(duration, 2), "raw_stdout": "", } # --------------------------------------------------------------------------- # セルフテスト（--dry-run） # --------------------------------------------------------------------------- def _test_build_prompt() -> None: """プロンプト組み立てのテスト。""" prompt = _build_prompt( transcript_60sec="これどうしようかな、迷うわ", trigger_reason="question", matched_text="どうしよう", recent_interventions=[ { "timestamp": "2026-05-06T12:00:00", "trigger_reason": "wake_word", "comment": "ジャービスです。何かお手伝いしますか？", } ], ) assert "J.A.R.V.I.S" in prompt, "プロンプトにタスク説明がない" assert "迷うわ" in prompt, "transcriptがプロンプトに含まれていない" assert "question" in prompt, "trigger_reasonがプロンプトに含まれていない" assert "2026-05-06" in prompt, "過去介入履歴がプロンプトに含まれていない" print("[OK] _build_prompt: プロンプト組み立て正常") def _test_parse_response() -> None: """JSON パースのテスト。""" # 正常ケース raw = '{"intervene": "yes", "reason": "迷いシグナル検出", "comment": "どちらも良い選択肢ですが、コストで判断するのが良いと思います"}' parsed = _parse_response(raw) assert parsed["intervene"] is True, f"intervene=True 期待: {parsed}" assert "迷いシグナル" in parsed["reason"], f"reason 期待: {parsed}" print("[OK] _parse_response: yes ケース正常") # no ケース raw_no = '{"intervene": "no", "reason": "介入不要", "comment": ""}' parsed_no = _parse_response(raw_no) assert parsed_no["intervene"] is False, f"intervene=False 期待: {parsed_no}" print("[OK] _parse_response: no ケース正常") # コードブロック付き raw_block = '```json\n{"intervene": "yes", "reason": "test", "comment": "テスト介入文"}\n```' parsed_block = _parse_response(raw_block) assert parsed_block["intervene"] is True, f"コードブロック解析失敗: {parsed_block}" print("[OK] _parse_response: コードブロック付き正常") # JSON 解析失敗 → 救済 raw_bad = "これは介入すべき重要なメッセージです。しっかり聞いてください。" parsed_bad = _parse_response(raw_bad) assert parsed_bad["intervene"] is True, f"救済失敗: {parsed_bad}" assert parsed_bad["reason"] == "json_parse_failed_fallback", f"救済理由期待: {parsed_bad}" print("[OK] _parse_response: JSON失敗→救済 正常") # comment 60字制限確認 assert len(parsed_bad["comment"]) <= 60, f"comment 60字超過: {parsed_bad['comment']}" print("[OK] _parse_response: comment 60字制限 正常") if __name__ == "__main__": dry_run = "--dry-run" in sys.argv if dry_run: print("=== intervention_judge dry-run テスト ===") _test_build_prompt() _test_parse_response() print("\n結果: 全テストpass") sys.exit(0) else: # 実走テストは E2E 段階で実施 print("実走テストは E2E 段階で実施。dry-run には --dry-run を指定してください。") print(" python intervention_judge.py --dry-run") sys.exit(0)