""" code_runner.py — Lv2 Jarvis常駐デーモン: Claude Code 呼び出しモジュール claude -p "プロンプト" を subprocess で叩く。 API直叩きしない → Max 20xプラン枠で動く。 """ import subprocess import json from datetime import datetime, timezone, timedelta from pathlib import Path from time import time from typing import Optional DNA_PATH = Path(__file__).parent.parent.parent / "clone_dna.md" ACTION_LOG_PATH = Path(__file__).parent.parent / "action_log.jsonl" JST = timezone(timedelta(hours=9)) SYSTEM_PROMPT_HEADER = """あなたは「ジャービス」と呼ばれて反応した J.A.R.V.I.S（Just A Rather Very Intelligent System）。 ユーザー（りつは・高1・起業家）の能力延長線として動く分配型AI。 応答ルール： - 短く・要点のみ・日本語 - 結論先出し - 必要なら3行以内で補足 - ナレーション禁止（「〜します」NG・体言止めOK） - 確実でない情報は「未確認」と明示 """ def load_dna_core(max_lines: int = 100) -> str: """clone_dna.md の核心部分を読み込む（先頭 max_lines 行）。""" if not DNA_PATH.exists(): return "" try: lines = DNA_PATH.read_text(encoding="utf-8").splitlines() return "\n".join(lines[:max_lines]) except Exception as e: print(f"[code_runner] DNA読込失敗: {e}") return "" def build_prompt(transcript_excerpt: str, recent_history: Optional[list] = None) -> str: """ Claude Code に渡すプロンプトを組み立てる。 構成： 1. SYSTEM_PROMPT_HEADER 2. DNA核心部分（load_dna_core） 3. 直近会話履歴（recent_history があれば） 4. ユーザー発話：transcript_excerpt 5. 「上記の発話に対して短く回答せよ」 目標：合計 5,000 tok 以下（実測キャラ数 / 1.5 で概算）。 """ parts = [SYSTEM_PROMPT_HEADER.strip()] dna = load_dna_core(100) if dna: parts.append("## DNA核心（思考パターン参考）\n" + dna) if recent_history: history_text = "\n".join(f"- {h}" for h in recent_history[-5:]) parts.append("## 直近会話履歴（最新5件）\n" + history_text) parts.append(f"## ユーザー発話\n{transcript_excerpt}") parts.append("上記の発話に対して短く回答せよ。") prompt = "\n\n".join(parts) # 5000 tok 超えたら DNA を削って再構成 if estimate_tokens(prompt) > 5000: parts[1] = "## DNA核心（思考パターン参考）\n" + load_dna_core(30) prompt = "\n\n".join(parts) return prompt def estimate_tokens(text: str) -> int: """雑なtoken見積：日本語混在で chars / 1.5 程度。""" return int(len(text) / 1.5) def run_claude_code(prompt: str, timeout: int = 60) -> dict: """ `claude -p "プロンプト"` を subprocess.run で実行。 戻り値: { "ok": bool, "stdout": str, "stderr": str, "returncode": int, "duration_sec": float, "estimated_input_tokens": int, "estimated_output_tokens": int, } """ # 🔴 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相当)化が必要。") start = time() try: result = subprocess.run( ["claude", "-p", prompt], capture_output=True, text=True, timeout=timeout, ) duration = round(time() - start, 2) stdout = result.stdout or "" stderr = result.stderr or "" return { "ok": result.returncode == 0, "stdout": stdout, "stderr": stderr, "returncode": result.returncode, "duration_sec": duration, "estimated_input_tokens": estimate_tokens(prompt), "estimated_output_tokens": estimate_tokens(stdout), } except subprocess.TimeoutExpired: return { "ok": False, "stdout": "", "stderr": "timeout", "returncode": -1, "duration_sec": round(time() - start, 2), "estimated_input_tokens": estimate_tokens(prompt), "estimated_output_tokens": 0, } except FileNotFoundError: return { "ok": False, "stdout": "", "stderr": "claude_not_found", "returncode": -1, "duration_sec": round(time() - start, 2), "estimated_input_tokens": estimate_tokens(prompt), "estimated_output_tokens": 0, } except Exception as e: print(f"[code_runner] run_claude_code 例外: {e}") return { "ok": False, "stdout": "", "stderr": str(e), "returncode": -1, "duration_sec": round(time() - start, 2), "estimated_input_tokens": estimate_tokens(prompt), "estimated_output_tokens": 0, } def log_action(transcript_excerpt: str, prompt: str, result: dict, mode: str, dry_run: bool = False) -> None: """action_log.jsonl に1行追記（action_log_rules.md 準拠）。""" ts = datetime.now(JST).isoformat() date_str = datetime.now(JST).strftime("%Y-%m-%d") entry = { "ts": ts, "agent": "jarvis_lv2_daemon", "action": "live_response", "target": f"dashboard:live_meeting_{date_str}", "reverse": f"dashboard.delete_card(timestamp={ts})", "reason": transcript_excerpt[:60], "reversibility": "auto", "category": "B", "result": "success" if result.get("ok") else "error", "mode": mode, "dry_run": dry_run, "estimated_input_tokens": result.get("estimated_input_tokens", 0), "estimated_output_tokens": result.get("estimated_output_tokens", 0), "duration_sec": result.get("duration_sec", 0.0), } try: ACTION_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with ACTION_LOG_PATH.open("a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") except Exception as e: print(f"[code_runner] log_action 失敗: {e}") def fire( transcript_excerpt: str, recent_history: Optional[list] = None, mode: str = "WAITING", dry_run: bool = False, ) -> dict: """ 総合実行。 手順: 1. build_prompt でプロンプト組み立て 2. dry_run=True なら subprocess 呼ばず estimate_tokens のみ返す 3. dry_run=False なら run_claude_code 実行 4. log_action で action_log.jsonl に記録 5. 結果を返す """ prompt = build_prompt(transcript_excerpt, recent_history) if dry_run: result = { "ok": True, "stdout": "", "stderr": "", "returncode": 0, "duration_sec": 0.0, "estimated_input_tokens": estimate_tokens(prompt), "estimated_output_tokens": 0, } else: result = run_claude_code(prompt) log_action(transcript_excerpt, prompt, result, mode, dry_run) return {**result, "prompt_used": prompt, "logged": True} # main: 自己テスト if __name__ == "__main__": import sys print("=== code_runner self-test ===") # DNA読込テスト dna = load_dna_core(50) print(f"DNA核心 50行 読込: {len(dna)} chars") # プロンプト組み立てテスト test_transcript = "ジャービス、今日の天気は？" prompt = build_prompt(test_transcript) est_tokens = estimate_tokens(prompt) print(f"プロンプト組み立て: {len(prompt)} chars / 推定 {est_tokens} tok") # dry-runテスト if "--dry-run" in sys.argv: result = fire(test_transcript, mode="WAITING", dry_run=True) print(f"dry-run結果: {result}") # 実行テスト（明示指定時のみ） if "--real" in sys.argv: print("実行中... (subprocess claude -p)") result = fire(test_transcript, mode="WAITING", dry_run=False) print(f"実行結果: ok={result.get('ok')}, duration={result.get('duration_sec')}s") print(f"出力先頭: {result.get('stdout', '')[:200]}")