monitor session file

scripts/monitor_session.py

@@ -0,0 +1,214 @@
+"""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()

scripts/review_pr.py

@@ -24,6 +24,8 @@ import time
 import urllib.error
 import urllib.request
 
+from monitor_session import SessionMonitor
+
 MAX_DIFF_CHARS = 200_000  # ~50k words, safe for Claude context
 
 
@@ -573,12 +575,17 @@ def main():
     else:
         print("[review_pr] No --repo-dir provided — skipping global impact analysis")
 
+    mon = None
     prompt_file = os.path.join(diff_dir, f"{prefix}_prompt.txt")
     try:
         # ── 5. Build prompt & run Claude ─────────────────────────────────
         prompt = build_prompt(meta, files, diff_path, agent_md, impact_report_path)
         print("[review_pr] Invoking Claude for analysis...")
+
+        mon = SessionMonitor(project_root)
+        mon.start()
         stdout = run_claude(claude_bin, project_root, prompt, extra_dir=repo_dir, prompt_file=prompt_file)
+        mon.stop()
 
         # ── 6. Parse Claude output ───────────────────────────────────────
         review = extract_json_from_output(stdout)
@@ -593,6 +600,8 @@ def main():
 
         print("[review_pr] Done.")
     finally:
+        if mon:
+            mon.stop()
         # Clean up temp files
         if os.path.exists(diff_path):
             os.unlink(diff_path)