code-review-agent/scripts/monitor_session.py

"""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":
            for block in msg.get("message", {}).get("content", []):
                if block.get("type") == "text":
                    text = block.get("text", "")
                    if text:
                        print(f"[{ts}] {text}", file=self._output)
                    return
                if block.get("type") == "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
            print(f"[{ts}] (empty)", file=self._output)

        elif msg_type == "user":
            for block in msg.get("message", {}).get("content", []):
                if block.get("type") == "tool_result":
                    print(f"[{ts}] tool: result", file=self._output)
                    return
            print(f"[{ts}] user", file=self._output)

        elif msg_type == "system":
            subtype = msg.get("subtype", "?")
            print(f"[{ts}] system: {subtype}", file=self._output)

        else:
            print(f"[{ts}] {msg_type}", 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()