"""Real-time monitor for a Claude Code session JSONL file. Can be used standalone:: python scripts/monitor_session.py --project-path /tmp/code-review-agent-1 Or imported and used as a background thread from review_pr.py:: mon = SessionMonitor("/tmp/code-review-agent-1") mon.start() # background thread starts tailing ... # run claude ... mon.stop() # kill the thread """ import json import os import sys import threading import time class SessionMonitor: """Tails the Claude Code session JSONL file in a background thread.""" def __init__(self, project_path, timeout=120, output=sys.stderr): encoded = os.path.abspath(project_path).replace("/", "-") self._sessions_dir = os.path.join( os.path.expanduser("~"), ".claude", "projects", encoded ) self._timeout = timeout self._output = output self._thread = None self._stop = threading.Event() self._session_file = None self._known_files = set() # files that existed before start() # ── public API ──────────────────────────────────────────────────── def start(self): """Snapshot existing files, then launch the monitoring thread.""" try: self._known_files = set( f for f in os.listdir(self._sessions_dir) if f.endswith(".jsonl") ) except FileNotFoundError: self._known_files = set() self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() def stop(self): """Signal the monitor to stop and wait for the thread.""" self._stop.set() if self._thread and self._thread.is_alive(): self._thread.join(timeout=5) @property def session_file(self): """The .jsonl file being tailed, or None if not yet found.""" return self._session_file # ── internals ───────────────────────────────────────────────────── def _run(self): print(f"[monitor] Watching {self._sessions_dir}", file=self._output) self._session_file = self._find_new_session_file() if not self._session_file: print( f"[monitor] No new .jsonl appeared in {self._sessions_dir} " f"within {self._timeout}s", file=self._output, ) return print(f"[monitor] session_id={os.path.basename(self._session_file)}", file=self._output) with open(self._session_file, "r", encoding="utf-8") as f: # read from the very first line while not self._stop.is_set(): line = f.readline() if line: self._pprint_line(line) else: time.sleep(0.2) def _find_new_session_file(self): """Wait for a .jsonl file that did NOT exist at start() time.""" deadline = time.time() + self._timeout while time.time() < deadline and not self._stop.is_set(): try: current = set( f for f in os.listdir(self._sessions_dir) if f.endswith(".jsonl") ) new = current - self._known_files if new: # Pick the newest among new files (by mtime) best = max( new, key=lambda f: os.path.getmtime( os.path.join(self._sessions_dir, f) ), ) path = os.path.join(self._sessions_dir, best) if os.path.getsize(path) > 0: return path except FileNotFoundError: pass time.sleep(1) return None def _pprint_line(self, line): try: msg = json.loads(line) except json.JSONDecodeError: print(f"[session] {line.rstrip()}", file=self._output) return msg_type = msg.get("type", "?") ts = msg.get("timestamp", "")[:19] if msg_type == "assistant": self._handle_assistant(ts, msg) elif msg_type == "user": self._handle_user(ts, msg) elif msg_type == "queue-operation": op = msg.get("operation", "?") print(f"[{ts}] queue: {op}", file=self._output) elif msg_type == "system": subtype = msg.get("subtype", "?") print(f"[{ts}] system: {subtype}", file=self._output) elif msg_type in ("attachment", "last-prompt"): pass # skip noise else: print(f"[{ts}] {msg_type}", file=self._output) def _handle_assistant(self, ts, msg): content = msg.get("message", {}).get("content", []) if isinstance(content, str): if content: print(f"[{ts}] {content}", file=self._output) return # Prefer text/tool_use over thinking; show thinking only if nothing else thinking = None for block in content: if not isinstance(block, dict): continue btype = block.get("type", "") if btype == "text": text = block.get("text", "") if text: print(f"[{ts}] {text}", file=self._output) return if btype == "tool_use": name = block.get("name", "?") inp = json.dumps(block.get("input", {}), ensure_ascii=False) print(f"[{ts}] tool: {name}({inp})", file=self._output) return if btype == "thinking": thinking = block.get("thinking", "") if thinking: for t in thinking.splitlines(): t = t.strip() if len(t) > 20: print(f"[{ts}] think: {t[:200]}{'...' if len(t) > 200 else ''}", file=self._output) return print(f"[{ts}] thinking...", file=self._output) else: print(f"[{ts}] (empty)", file=self._output) def _handle_user(self, ts, msg): content = msg.get("message", {}).get("content", []) # user content can be a plain string (original prompt) or a list of tool_result blocks if isinstance(content, str): return # skip the prompt echo for block in content: if not isinstance(block, dict): continue if block.get("type") == "tool_result": if block.get("is_error"): print(f"[{ts}] tool: ERROR", file=self._output) else: print(f"[{ts}] tool: result", file=self._output) return print(f"[{ts}] user", file=self._output) # ── standalone CLI ──────────────────────────────────────────────────────── def main(): import argparse parser = argparse.ArgumentParser( description="Tail a Claude Code session JSONL file" ) parser.add_argument( "--project-path", required=True, help="Project working directory used by claude (the cwd)", ) parser.add_argument( "--timeout", type=int, default=60, help="Seconds to wait for a session file to appear", ) args = parser.parse_args() mon = SessionMonitor(args.project_path, timeout=args.timeout) mon.start() try: while mon._thread and mon._thread.is_alive(): mon._thread.join(1) except KeyboardInterrupt: mon.stop() if __name__ == "__main__": main()