codex log monitor example

codex_monitor.py

#!/usr/bin/python
import argparse
import json
import os
import re
import subprocess
import time
from pathlib import Path
from typing import Any, Optional

# Mode switch:
# - "DEMO": rich event-to-state window renaming (original behavior)
# - "CREASE": only turn-active/idle with :! / :z suffixes
MODE = "CREASE"


def run_capture(args: list[str]) -> Optional[str]:
    try:
        return subprocess.check_output(args, stderr=subprocess.DEVNULL, text=True).strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        return None


def run_no_fail(args: list[str]) -> None:
    try:
        subprocess.run(args, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except FileNotFoundError:
        pass


def process_exists(pid: int) -> bool:
    return os.path.exists(f"/proc/{pid}")


class TmuxWindowNamer:
    def __init__(self, pane: str, prefix: str, restore: bool) -> None:
        self.pane = pane
        self.prefix = prefix
        self.restore = restore
        self.window_id = run_capture(["tmux", "display-message", "-p", "-t", pane, "#{window_id}"])
        self.original_name = run_capture(["tmux", "display-message", "-p", "-t", pane, "#{window_name}"])
        self.last_name: Optional[str] = None

    def current_name(self) -> Optional[str]:
        if not self.window_id:
            return None
        return run_capture(["tmux", "display-message", "-p", "-t", self.window_id, "#{window_name}"])

    def set_state(self, state: str) -> None:
        if not self.window_id:
            return
        name = f"{self.prefix}:{state}"
        if self.last_name == name:
            return
        run_no_fail(["tmux", "rename-window", "-t", self.window_id, name])
        self.last_name = name

    def close(self) -> None:
        if not self.restore:
            return
        if not self.window_id or not self.original_name:
            return
        run_no_fail(["tmux", "rename-window", "-t", self.window_id, self.original_name])

    def set_crease_status(self, suffix: str) -> None:
        if not self.window_id:
            return
        current = self.current_name() or self.original_name or ""
        base = re.sub(r"(?::[!z])+$", "", current)
        name = f"{base}:{suffix}"
        if self.last_name == name:
            return
        run_no_fail(["tmux", "rename-window", "-t", self.window_id, name])
        self.last_name = name


def map_event_to_state_demo(event: dict[str, Any]) -> Optional[str]:
    kind = event.get("kind")
    if kind == "session_start":
        return "boot"

    if kind != "codex_event":
        return None

    msg = event.get("payload", {}).get("msg", {})
    msg_type = msg.get("type")

    if msg_type == "session_configured":
        return "ready"
    if msg_type == "mcp_startup_update":
        state = msg.get("status", {}).get("state")
        if isinstance(state, str):
            return f"mcp-{state}"
        return "mcp"
    if msg_type == "task_started":
        return "turn"
    if msg_type == "task_complete":
        return "idle"
    if msg_type == "item_started":
        item = msg.get("item", {})
        item_type = item.get("type")
        if item_type == "UserMessage":
            return "user-msg"
        if item_type == "Reasoning":
            return "thinking"
        if item_type == "AgentMessage":
            phase = item.get("phase")
            if phase == "commentary":
                return "commentary"
            if phase == "final_answer":
                return "final"
            return "agent"
    if msg_type == "exec_command_begin":
        return "tool"
    if msg_type == "exec_command_end":
        exit_code = msg.get("exit_code")
        if isinstance(exit_code, int) and exit_code != 0:
            return "tool-fail"
        return "tool-done"
    return None


def handle_event_crease(event: dict[str, Any], namer: TmuxWindowNamer) -> None:
    if event.get("kind") != "codex_event":
        return

    msg = event.get("payload", {}).get("msg", {})
    msg_type = msg.get("type")
    if msg_type == "task_started":
        namer.set_crease_status("!")
    elif msg_type == "task_complete":
        namer.set_crease_status("z")


def monitor_file(log_path: Path, namer: TmuxWindowNamer, parent_pid: int, poll_interval: float) -> None:
    offset = 0
    buffer = ""

    while True:
        if not process_exists(parent_pid):
            return

        if not log_path.exists():
            time.sleep(poll_interval)
            continue

        try:
            size = log_path.stat().st_size
        except OSError:
            time.sleep(poll_interval)
            continue

        if size < offset:
            offset = 0
            buffer = ""

        if size == offset:
            time.sleep(poll_interval)
            continue

        try:
            with log_path.open("r", encoding="utf-8") as f:
                f.seek(offset)
                chunk = f.read()
                offset = f.tell()
        except OSError:
            time.sleep(poll_interval)
            continue

        if not chunk:
            time.sleep(poll_interval)
            continue

        buffer += chunk
        while True:
            newline = buffer.find("\n")
            if newline < 0:
                break
            line = buffer[:newline]
            buffer = buffer[newline + 1 :]
            if not line.strip():
                continue
            try:
                event = json.loads(line)
            except json.JSONDecodeError:
                continue
            if MODE == "CREASE":
                handle_event_crease(event, namer)
            else:
                state = map_event_to_state_demo(event)
                if state:
                    namer.set_state(state)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Monitor Codex JSONL session logs and rename tmux windows by event")
    parser.add_argument("--log-path", required=True)
    parser.add_argument("--tmux-pane", required=True)
    parser.add_argument("--parent-pid", required=True, type=int)
    parser.add_argument("--prefix", default="cdx")
    parser.add_argument("--poll-interval", type=float, default=0.10)
    parser.add_argument("--no-restore", action="store_true")
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    if not run_capture(["tmux", "display-message", "-p", "-t", args.tmux_pane, "#{window_id}"]):
        return 0

    namer = TmuxWindowNamer(
        pane=args.tmux_pane,
        prefix=args.prefix,
        restore=not args.no_restore,
    )
    try:
        if MODE != "CREASE":
            namer.set_state("init")
        monitor_file(Path(args.log_path), namer, args.parent_pid, args.poll_interval)
    finally:
        namer.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())