""" output_dispatcher.py — MARK VII 統合出力ハブ Python 3.9 互換 Jarvis 介入を全方向に出力： ① dashboard カード書き込み（live_meeting_YYYY-MM-DD.md） ② 音声読み上げ（voice_output 経由） ③ intervention_log.jsonl 記録 ④ mirror_to_dashboard.sh 実行（キャッシュ即時刷新） 並行出力・部分失敗耐性・ログ・キャッシュ刷新 """ import json import os import subprocess import sys import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Dict # ────────────────────────────────────────────── # パス定義 # ────────────────────────────────────────────── # このファイルの場所: live_session/v3/ V3_DIR = Path(__file__).parent LIVE_SESSION_DIR = V3_DIR.parent DAEMON_DIR = LIVE_SESSION_DIR / "daemon" # daemon/notifier.py を import できるようにパス追加 sys.path.insert(0, str(DAEMON_DIR)) # intervention_log はデフォルト v3/ 配下 INTERVENTION_LOG_PATH = V3_DIR / "intervention_log.jsonl" # mirror_to_dashboard.sh の候補パス（存在すれば使う） MIRROR_SCRIPT_CANDIDATES = [ LIVE_SESSION_DIR / "mirror_to_dashboard.sh", V3_DIR / "mirror_to_dashboard.sh", Path("/Users/rk/jarvis-dashboard/start.sh"), # fallback: dashboard refresh ] # ────────────────────────────────────────────── # notifier.py import（部分失敗耐性） # ────────────────────────────────────────────── try: from notifier import write_dashboard_card # type: ignore _NOTIFIER_AVAILABLE = True except ImportError as _e: _NOTIFIER_AVAILABLE = False _NOTIFIER_ERROR = str(_e) # ────────────────────────────────────────────── # メイン dispatch 関数 # ────────────────────────────────────────────── def dispatch( transcript_excerpt: str, jarvis_response: str, mode: str = "OBSERVING", trigger_reason: str = "", tokens_used: int = 0, duration_sec: float = 0.0, voice_enabled: bool = True, log_path: Path = None, ) -> Dict: """ 統合出力。全部成功で {"ok": True}。 部分失敗（音声NG など）でも継続して残りは実行。 戻り値: { "ok": True/False, # 全ステップ成功かどうか "dashboard": True/False, "voice": True/False, "log": True/False, "mirror": True/False, } """ results = { "ok": True, "dashboard": False, "voice": False, "log": False, "mirror": False, } # ① dashboard カード書き込み try: if not _NOTIFIER_AVAILABLE: raise ImportError(f"notifier not available: {_NOTIFIER_ERROR}") write_dashboard_card( transcript_excerpt=transcript_excerpt, jarvis_response=jarvis_response, mode=mode, ) results["dashboard"] = True except Exception as e: print(f"[output_dispatcher] WARN dashboard: {e}", file=sys.stderr) results["ok"] = False # ② 音声出力（voice_output 経由・非同期） try: if voice_enabled and os.environ.get("VOICE_ENABLED", "1") != "0": # voice_output は v3/ 配下にある voice_module_path = V3_DIR / "voice_output.py" if voice_module_path.exists(): # subprocess で呼ぶことで import 依存を避ける # （voice_output 自体が self-contained なので直 import でも OK だが # sys.path 汚染を避けるため subprocess 方式を選択） proc = subprocess.Popen( [ sys.executable, "-c", ( "import sys; sys.path.insert(0, repr(str('')));" f"import importlib.util;" f"spec = importlib.util.spec_from_file_location('voice_output', '{voice_module_path}');" f"m = importlib.util.module_from_spec(spec);" f"spec.loader.exec_module(m);" f"m.speak({repr(jarvis_response)}, blocking=False)" ), ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) results["voice"] = proc.pid is not None else: # voice_output が同ディレクトリになければ直 import 試行 _vpath = str(V3_DIR) if _vpath not in sys.path: sys.path.insert(0, _vpath) import importlib.util as _ilu _spec = _ilu.spec_from_file_location("voice_output", voice_module_path) _vo = _ilu.module_from_spec(_spec) _spec.loader.exec_module(_vo) pid = _vo.speak(jarvis_response, blocking=False) results["voice"] = pid is not None else: # 音声無効 → スキップ（これは失敗ではない） results["voice"] = True # intentional skip = success except Exception as e: print(f"[output_dispatcher] WARN voice: {e}", file=sys.stderr) results["ok"] = False # ③ intervention_log.jsonl に append try: _log_path = log_path or INTERVENTION_LOG_PATH Path(_log_path).parent.mkdir(parents=True, exist_ok=True) entry = { "ts": datetime.now(timezone.utc).isoformat(), "mode": mode, "trigger": trigger_reason, "transcript": transcript_excerpt, "response": jarvis_response, "tokens": tokens_used, "duration_sec": duration_sec, } with open(_log_path, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") results["log"] = True except Exception as e: print(f"[output_dispatcher] WARN log: {e}", file=sys.stderr) results["ok"] = False # ④ mirror_to_dashboard.sh 実行（2026-05-06 非同期化） # 旧実装：subprocess.run timeout=10s → 体感「死んだ」遅延の元凶 # 新実装：Popen 投げっぱなし。dispatch は即返る。mirror成否は dispatch を遅延させない try: mirror_script = None for candidate in MIRROR_SCRIPT_CANDIDATES: if candidate.exists(): mirror_script = candidate break if mirror_script: # Popen で起動して即手放す。stdout/stderr は破棄（ログ汚染防止） # start_new_session=True で親プロセス終了時もmirrorを巻き込まない subprocess.Popen( ["bash", str(mirror_script)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) results["mirror"] = True # 起動成功（完了は待たない） else: print( "[output_dispatcher] INFO mirror: mirror_to_dashboard.sh 未存在・スキップ", file=sys.stderr, ) results["mirror"] = True # intentional skip = success except Exception as e: print(f"[output_dispatcher] WARN mirror: {e}", file=sys.stderr) # mirror失敗はdispatch全体の ok を倒さない（重要度低） return results # ────────────────────────────────────────────── # self-test（--dry-run） # ────────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="output_dispatcher dry-run test") parser.add_argument("--dry-run", action="store_true", help="セルフテストモード") args = parser.parse_args() if not args.dry_run: print("使い方: python output_dispatcher.py --dry-run") sys.exit(0) print("=== output_dispatcher.py --dry-run self-test ===\n") all_pass = True # 音声を VOICE_ENABLED=0 で強制 OFF os.environ["VOICE_ENABLED"] = "0" # log は tmp に向ける with tempfile.TemporaryDirectory() as tmpdir: tmp_log = Path(tmpdir) / "intervention_log_test.jsonl" print("[1] dispatch() 呼び出し（dashboard・音声OFF・tmp log・mirror）") result = dispatch( transcript_excerpt="テスト発話：今日の会議どうする？", jarvis_response="確認します。スケジュールを整理してから提案できます。", mode="OBSERVING", trigger_reason="dry_run_test", tokens_used=1234, duration_sec=2.5, voice_enabled=False, log_path=tmp_log, ) print(f" 戻り値: {result}") # dashboard チェック（notifier が import できない環境では skip） if _NOTIFIER_AVAILABLE: if result["dashboard"]: print(" [dashboard] OK: カード書き込み成功") else: print(" [dashboard] FAIL: カード書き込み失敗") all_pass = False else: print(f" [dashboard] SKIP: notifier import 不可（{_NOTIFIER_ERROR}）") # voice チェック（VOICE_ENABLED=0 なので intentional skip = True） if result["voice"]: print(" [voice] OK: VOICE_ENABLED=0 でスキップ（正常）") else: print(" [voice] FAIL: 予期しない失敗") all_pass = False # log チェック if tmp_log.exists(): lines = tmp_log.read_text(encoding="utf-8").strip().splitlines() entry = json.loads(lines[-1]) assert entry["mode"] == "OBSERVING", f"mode mismatch: {entry}" assert entry["tokens"] == 1234, f"tokens mismatch: {entry}" print(f" [log] OK: {tmp_log} に {len(lines)} エントリ書き込み済") else: print(" [log] FAIL: ログファイルが作成されなかった") all_pass = False # mirror チェック（存在しないスクリプトは skip 扱い） if result["mirror"]: print(" [mirror] OK: 成功 or 意図的スキップ") else: print(" [mirror] WARN: mirror 実行失敗（non-critical）") # mirror は ok フラグを倒さないので all_pass には影響させない print() if all_pass: print("=== ALL PASS ===") sys.exit(0) else: print("=== SOME TESTS FAILED ===") sys.exit(1)