app.py

import os

os.environ.setdefault("HF_HOME", "/tmp/.cache/huggingface")
os.environ.setdefault("HF_MODULES_CACHE", "/tmp/hf_modules")
os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib")
os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
os.environ.setdefault("GRADIO_SSR_MODE", "false")
if not os.environ.get("HF_TOKEN"):
    for _token_key in ("HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_HUB_TOKEN"):
        if os.environ.get(_token_key):
            os.environ["HF_TOKEN"] = os.environ[_token_key]
            break

print("codex-traces startup: env configured", flush=True)

import html
import json
import math
import re
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
from typing import Any

import gradio as gr
from huggingface_hub import hf_hub_download, list_repo_files

print("codex-traces startup: imports complete", flush=True)


APP_TITLE = "Codex Traces"
APP_REVISION = "parser-v2"
DATASET_REPO = "Mike0021/codex-sessions"
REPO_TYPE = "dataset"
FILE_SUFFIX = "_rollout.jsonl"
SESSION_PATH_PREFIX = "sessions/rollout-"
SESSION_PATH_SUFFIX = ".jsonl"
PAGE_SIZE = 120
MAX_OUTPUT_CHARS = 22000
MAX_MESSAGE_CHARS = 30000
KNOWN_SESSIONS = (
    "agents-a1-demo",
    "ai-agent-soccer",
    "anime-soccer-generator",
    "asasr-space",
    "codex-traces-viewer",
    "craft-agents-oss",
    "deepspec-space",
    "edit-anything",
    "fractal-pi-extension",
    "gaussian-splat-demo",
    "hf-history-article",
    "hf-motion-video",
    "horus-hiero-space",
    "locate-anything-space",
    "ltx-3dreal-space",
    "microworld-space",
    "olmoearth-demo",
    "paris-13-landing",
    "pulpie-demo",
    "pulpie-gguf",
    "pulpie-mlx",
    "pulpie-onnx",
    "pulpie-web-demo",
    "qwen3-asr-space",
    "rampart-demo",
    "sync-lora-space",
    "tabfm-arena",
)

ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
TIMESTAMP_KEYS = (
    "timestamp",
    "time",
    "created_at",
    "updated_at",
    "started_at",
    "completed_at",
)


def _hub_token() -> str | None:
    return (
        os.environ.get("HF_TOKEN")
        or os.environ.get("HUGGING_FACE_HUB_TOKEN")
        or os.environ.get("HUGGINGFACE_HUB_TOKEN")
        or None
    )


def _display_name(file_name: str) -> str:
    if file_name.startswith(SESSION_PATH_PREFIX) and file_name.endswith(SESSION_PATH_SUFFIX):
        return file_name[len(SESSION_PATH_PREFIX) : -len(SESSION_PATH_SUFFIX)]
    return file_name[: -len(FILE_SUFFIX)] if file_name.endswith(FILE_SUFFIX) else file_name


def _is_rollout_file(file_name: str) -> bool:
    return (
        file_name.startswith(SESSION_PATH_PREFIX)
        and file_name.endswith(SESSION_PATH_SUFFIX)
    ) or file_name.endswith(FILE_SUFFIX)


def _file_candidates(session_name: str | None) -> list[str]:
    if not session_name:
        return []
    name = str(session_name).strip()
    if name.endswith(".jsonl"):
        return [name]

    candidates = [
        f"{SESSION_PATH_PREFIX}{name}{SESSION_PATH_SUFFIX}",
        f"{name}{FILE_SUFFIX}",
    ]
    return list(dict.fromkeys(candidates))


def _fmt_number(value: int | None) -> str:
    return f"{int(value or 0):,}"


def _fmt_bytes(value: int | None) -> str:
    value = int(value or 0)
    units = ("B", "KB", "MB", "GB")
    size = float(value)
    for unit in units:
        if size < 1024 or unit == units[-1]:
            return f"{size:.1f} {unit}" if unit != "B" else f"{value} B"
        size /= 1024
    return f"{value} B"


def _strip_ansi(text: Any) -> str:
    return ANSI_RE.sub("", str(text or ""))


def _escape(text: Any) -> str:
    return html.escape(str(text or ""), quote=True)


def _limit_text(text: str, limit: int, label: str) -> tuple[str, str]:
    if len(text) <= limit:
        return text, ""
    hidden = len(text) - limit
    note = f"\n\n[{label} truncated by {hidden:,} characters for browser responsiveness.]"
    return text[:limit] + note, f" truncated {hidden:,} chars"


def _parse_time(value: Any) -> datetime | None:
    if value is None:
        return None
    if isinstance(value, (int, float)):
        seconds = float(value) / 1000 if value > 10_000_000_000 else float(value)
        try:
            return datetime.fromtimestamp(seconds, tz=timezone.utc)
        except (OSError, OverflowError, ValueError):
            return None
    if isinstance(value, str):
        raw = value.strip()
        if not raw:
            return None
        try:
            if raw.endswith("Z"):
                raw = raw[:-1] + "+00:00"
            parsed = datetime.fromisoformat(raw)
            return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
        except ValueError:
            return None
    return None


def _duration_label(timestamps: list[datetime]) -> str:
    if len(timestamps) < 2:
        return "not recorded"
    delta = max(timestamps) - min(timestamps)
    seconds = max(0, int(delta.total_seconds()))
    if seconds < 60:
        return f"{seconds}s"
    minutes, sec = divmod(seconds, 60)
    if minutes < 60:
        return f"{minutes}m {sec}s"
    hours, minutes = divmod(minutes, 60)
    return f"{hours}h {minutes}m"


@lru_cache(maxsize=1)
def _list_sessions_cached() -> tuple[tuple[str, ...], str]:
    try:
        files = list_repo_files(DATASET_REPO, repo_type=REPO_TYPE, token=_hub_token())
    except Exception as exc:
        return (), f"{type(exc).__name__}: {exc}"
    sessions = tuple(sorted({_display_name(f) for f in files if _is_rollout_file(f)}))
    if not sessions:
        return (), "No *_rollout.jsonl files were found in the dataset."
    return sessions, ""


def _session_choices(live: bool = True) -> list[str]:
    if not live:
        return list(KNOWN_SESSIONS)
    sessions, _ = _list_sessions_cached()
    return list(sessions or KNOWN_SESSIONS)


def _default_session() -> str | None:
    choices = _session_choices(live=False)
    return choices[0] if choices else None


def _download_session(file_name: str) -> Path:
    return Path(
        hf_hub_download(
            repo_id=DATASET_REPO,
            filename=file_name,
            repo_type=REPO_TYPE,
            token=_hub_token(),
        )
    )


def _command_status(item: dict[str, Any]) -> tuple[str, str]:
    status = str(item.get("status") or "").lower()
    exit_code = item.get("exit_code")
    if status in {"failed", "error"} or (exit_code not in (None, 0)):
        return "failed", "fail"
    if status in {"completed", "succeeded", "success"}:
        return "success", "success"
    if status in {"in_progress", "running"}:
        return "running", "running"
    return status or "unknown", "neutral"


def _usage_total(usage: dict[str, Any]) -> int:
    return int(usage.get("input_tokens") or 0) + int(usage.get("output_tokens") or 0)


def _collect_timestamps(obj: dict[str, Any], timestamps: list[datetime]) -> None:
    for key in TIMESTAMP_KEYS:
        parsed = _parse_time(obj.get(key))
        if parsed:
            timestamps.append(parsed)
    item = obj.get("item")
    if isinstance(item, dict):
        for key in TIMESTAMP_KEYS:
            parsed = _parse_time(item.get(key))
            if parsed:
                timestamps.append(parsed)


def _payload_command(payload: dict[str, Any]) -> str:
    name = str(payload.get("name") or "function_call")
    arguments = payload.get("arguments")
    if isinstance(arguments, dict):
        command = arguments.get("command")
        if command:
            return str(command)
        return f"{name}({json.dumps(arguments, ensure_ascii=True)})"
    if arguments:
        return f"{name}({arguments})"
    return name


@lru_cache(maxsize=6)
def _load_session(file_name: str) -> dict[str, Any]:
    path = _download_session(file_name)
    events: list[dict[str, Any]] = []
    timestamps: list[datetime] = []
    pending_commands: list[str] = []
    command_ids: set[str] = set()
    failed_commands = 0
    turn_completed = 0
    turn_started = 0
    parse_errors = 0
    usage_summary = {
        "input_tokens": 0,
        "cached_input_tokens": 0,
        "output_tokens": 0,
        "reasoning_output_tokens": 0,
    }

    with path.open("r", encoding="utf-8", errors="replace") as handle:
        for line_no, raw_line in enumerate(handle, start=1):
            line = raw_line.strip()
            if not line.startswith("{"):
                parse_errors += 1
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                parse_errors += 1
                continue

            obj_type = obj.get("type")
            payload = obj.get("payload") if isinstance(obj.get("payload"), dict) else {}
            payload_type = payload.get("type")
            item = obj.get("item") if isinstance(obj.get("item"), dict) else {}
            item_type = item.get("type")
            _collect_timestamps(obj, timestamps)

            if obj_type == "turn_context":
                turn_started += 1
                continue

            if obj_type == "event_msg":
                if payload_type in {"agent_message", "user_message"}:
                    text = str(payload.get("message") or "")
                    is_user = payload_type == "user_message"
                    events.append(
                        {
                            "kind": "user" if is_user else "agent",
                            "role": "USER" if is_user else "AGENT",
                            "line": line_no,
                            "id": f"line-{line_no}",
                            "text": text,
                            "search": text,
                        }
                    )
                    continue

                if payload_type == "token_count":
                    turn_completed += 1
                    usage = {key: int(payload.get(key) or 0) for key in usage_summary}
                    for key, value in usage.items():
                        usage_summary[key] += value
                    events.append(
                        {
                            "kind": "usage",
                            "line": line_no,
                            "usage": usage,
                            "search": " ".join(
                                [
                                    "turn completed usage tokens",
                                    " ".join(f"{k} {v}" for k, v in usage.items()),
                                ]
                            ),
                        }
                    )
                    continue

                if payload_type == "item_completed" and payload.get("item_type") == "todo_list":
                    todos = payload.get("todos") or payload.get("items") or []
                    if isinstance(todos, list):
                        text = " ".join(str(todo.get("text") or "") for todo in todos if isinstance(todo, dict))
                        events.append(
                            {
                                "kind": "todo",
                                "line": line_no,
                                "id": f"line-{line_no}",
                                "items": todos,
                                "search": text,
                            }
                        )
                    continue

                if payload_type == "patch_apply_end":
                    path_value = str(payload.get("path") or "patch")
                    events.append(
                        {
                            "kind": "files",
                            "line": line_no,
                            "id": f"line-{line_no}",
                            "changes": [{"kind": "patch", "path": path_value}],
                            "status": payload.get("status") or "",
                            "search": f"patch {path_value} {payload.get('status') or ''}",
                        }
                    )
                    continue

            if obj_type == "response_item":
                if payload_type == "function_call":
                    pending_commands.append(_payload_command(payload))
                    continue

                if payload_type == "function_call_output":
                    command = pending_commands.pop(0) if pending_commands else "function_call_output"
                    output = _strip_ansi(payload.get("output") or "")
                    exit_code = payload.get("exit_code")
                    status = "failed" if exit_code not in (None, 0) else "completed"
                    label, status_class = _command_status({"status": status, "exit_code": exit_code})
                    if status_class == "fail":
                        failed_commands += 1
                    item_id = f"line-{line_no}"
                    command_ids.add(item_id)
                    events.append(
                        {
                            "kind": "command",
                            "line": line_no,
                            "id": item_id,
                            "command": command,
                            "output": output,
                            "exit_code": exit_code,
                            "status": label,
                            "status_class": status_class,
                            "search": " ".join([command, output, str(exit_code), label]),
                        }
                    )
                    continue

            if obj_type == "turn.started":
                turn_started += 1
                continue

            if obj_type == "turn.completed":
                turn_completed += 1
                usage = obj.get("usage") or {}
                if isinstance(usage, dict):
                    for key in usage_summary:
                        usage_summary[key] += int(usage.get(key) or 0)
                    events.append(
                        {
                            "kind": "usage",
                            "line": line_no,
                            "usage": usage,
                            "search": " ".join(
                                [
                                    "turn completed usage tokens",
                                    " ".join(f"{k} {v}" for k, v in usage.items()),
                                ]
                            ),
                        }
                    )
                continue

            if obj_type not in {"item.completed", "item.updated"}:
                continue

            if item_type == "agent_message":
                text = str(item.get("text") or "")
                events.append(
                    {
                        "kind": "agent",
                        "line": line_no,
                        "id": item.get("id") or f"line-{line_no}",
                        "text": text,
                        "search": text,
                    }
                )
                continue

            if item_type == "command_execution":
                item_id = str(item.get("id") or f"line-{line_no}")
                if item_id not in command_ids:
                    command_ids.add(item_id)
                label, status_class = _command_status(item)
                if status_class == "fail":
                    failed_commands += 1
                command = str(item.get("command") or "")
                output = _strip_ansi(item.get("aggregated_output") or "")
                exit_code = item.get("exit_code")
                events.append(
                    {
                        "kind": "command",
                        "line": line_no,
                        "id": item_id,
                        "command": command,
                        "output": output,
                        "exit_code": exit_code,
                        "status": label,
                        "status_class": status_class,
                        "search": " ".join([command, output, str(exit_code), label]),
                    }
                )
                continue

            if item_type == "todo_list":
                todos = item.get("items") or []
                if isinstance(todos, list):
                    text = " ".join(str(todo.get("text") or "") for todo in todos if isinstance(todo, dict))
                    events.append(
                        {
                            "kind": "todo",
                            "line": line_no,
                            "id": item.get("id") or f"line-{line_no}",
                            "items": todos,
                            "search": text,
                        }
                    )
                continue

            if item_type == "file_change":
                changes = item.get("changes") or []
                if isinstance(changes, list):
                    text = " ".join(
                        f"{change.get('kind', '')} {change.get('path', '')}"
                        for change in changes
                        if isinstance(change, dict)
                    )
                    events.append(
                        {
                            "kind": "files",
                            "line": line_no,
                            "id": item.get("id") or f"line-{line_no}",
                            "changes": changes,
                            "status": item.get("status") or "",
                            "search": text,
                        }
                    )

    total_turns = turn_completed or turn_started
    return {
        "file_name": file_name,
        "path": str(path),
        "bytes": path.stat().st_size,
        "events": events,
        "summary": {
            "turns": total_turns,
            "commands": len(command_ids),
            "failed_commands": failed_commands,
            "duration": _duration_label(timestamps),
            "parse_errors": parse_errors,
            **usage_summary,
            "total_tokens": usage_summary["input_tokens"] + usage_summary["output_tokens"],
        },
    }


def _filter_events(events: list[dict[str, Any]], query: str | None) -> list[dict[str, Any]]:
    needle = (query or "").strip().lower()
    if not needle:
        return events
    terms = [part for part in needle.split() if part]
    return [
        event
        for event in events
        if all(term in str(event.get("search", "")).lower() for term in terms)
    ]


def _metric(label: str, value: str, tone: str = "") -> str:
    return (
        f'<div class="metric {tone}">'
        f'<span class="metric-label">{_escape(label)}</span>'
        f'<span class="metric-value">{_escape(value)}</span>'
        "</div>"
    )


def _render_summary(data: dict[str, Any], matches: int, query: str) -> str:
    summary = data["summary"]
    file_label = _display_name(data["file_name"])
    query_label = f"Filtered: {matches:,} matches" if query.strip() else f"{matches:,} visible events"
    failed = int(summary["failed_commands"])
    return (
        '<section class="summary-panel">'
        '<div class="summary-topline">'
        f'<div><h1>{_escape(file_label)}</h1>'
        f'<p>{_escape(data["file_name"])} · {_fmt_bytes(data["bytes"])} · {query_label}</p></div>'
        f'<div class="repo-pill">private dataset · {_escape(APP_REVISION)}</div>'
        "</div>"
        '<div class="metrics-grid">'
        f'{_metric("Turns", _fmt_number(summary["turns"]))}'
        f'{_metric("Total tokens", _fmt_number(summary["total_tokens"]))}'
        f'{_metric("Input", _fmt_number(summary["input_tokens"]))}'
        f'{_metric("Output", _fmt_number(summary["output_tokens"]))}'
        f'{_metric("Reasoning", _fmt_number(summary["reasoning_output_tokens"]))}'
        f'{_metric("Duration", str(summary["duration"]))}'
        f'{_metric("Commands", _fmt_number(summary["commands"]))}'
        f'{_metric("Failed", _fmt_number(failed), "danger" if failed else "ok")}'
        "</div>"
        "</section>"
    )


def _render_agent(event: dict[str, Any]) -> str:
    text, note = _limit_text(str(event.get("text") or ""), MAX_MESSAGE_CHARS, "message")
    role = str(event.get("role") or "AGENT").upper()
    role_class = "user" if role == "USER" else "agent"
    return (
        f'<article class="event {role_class}-event">'
        f'<div class="event-rail {role_class}-rail">{_escape(role)}</div>'
        f'<div class="bubble {role_class}-bubble">'
        f'<div class="event-meta">line {_fmt_number(event["line"])}{_escape(note)}</div>'
        f'<div class="message-text">{_escape(text)}</div>'
        "</div>"
        "</article>"
    )


def _render_command(event: dict[str, Any]) -> str:
    output = str(event.get("output") or "")
    output_limited, note = _limit_text(output, MAX_OUTPUT_CHARS, "output")
    exit_code = event.get("exit_code")
    exit_label = "pending" if exit_code is None else str(exit_code)
    status = str(event.get("status") or "unknown")
    status_class = str(event.get("status_class") or "neutral")
    output_count = f"{len(output):,} chars"
    return (
        '<article class="event command-event">'
        '<div class="event-rail command-rail">CMD</div>'
        '<div class="command-card">'
        '<div class="command-head">'
        f'<span class="status-badge {status_class}">{_escape(status)}</span>'
        f'<span class="exit-badge">exit {_escape(exit_label)}</span>'
        f'<span class="event-meta">line {_fmt_number(event["line"])}</span>'
        "</div>"
        f'<pre class="code command-code"><code>{_escape(event.get("command") or "")}</code></pre>'
        f'<details class="output-drawer"><summary>Command output · {output_count}{_escape(note)}</summary>'
        f'<pre class="code output-code"><code>{_escape(output_limited) if output_limited else "No output."}</code></pre>'
        "</details>"
        "</div>"
        "</article>"
    )


def _render_usage(event: dict[str, Any]) -> str:
    usage = event.get("usage") or {}
    input_tokens = int(usage.get("input_tokens") or 0)
    cached = int(usage.get("cached_input_tokens") or 0)
    output = int(usage.get("output_tokens") or 0)
    reasoning = int(usage.get("reasoning_output_tokens") or 0)
    total = input_tokens + output
    return (
        '<article class="event usage-event">'
        '<div class="event-rail usage-rail">TURN</div>'
        '<div class="usage-card">'
        f'<span>Total {_fmt_number(total)}</span>'
        f'<span>Input {_fmt_number(input_tokens)}</span>'
        f'<span>Cached {_fmt_number(cached)}</span>'
        f'<span>Output {_fmt_number(output)}</span>'
        f'<span>Reasoning {_fmt_number(reasoning)}</span>'
        f'<span class="event-meta">line {_fmt_number(event["line"])}</span>'
        "</div>"
        "</article>"
    )


def _render_todo(event: dict[str, Any]) -> str:
    items = event.get("items") or []
    rows = []
    for todo in items:
        if not isinstance(todo, dict):
            continue
        done = bool(todo.get("completed"))
        rows.append(
            '<li class="todo-row">'
            f'<span class="todo-check {"done" if done else "open"}">{"&#10003;" if done else "&#9711;"}</span>'
            f'<span>{_escape(todo.get("text") or "")}</span>'
            "</li>"
        )
    return (
        '<article class="event todo-event">'
        '<div class="event-rail todo-rail">TODO</div>'
        '<div class="todo-card">'
        f'<div class="event-meta">line {_fmt_number(event["line"])}</div>'
        f'<ul>{"".join(rows) if rows else "<li>No todo items.</li>"}</ul>'
        "</div>"
        "</article>"
    )


def _render_files(event: dict[str, Any]) -> str:
    changes = event.get("changes") or []
    rows = []
    for change in changes:
        if not isinstance(change, dict):
            continue
        rows.append(
            '<li>'
            f'<span class="file-kind">{_escape(change.get("kind") or "change")}</span>'
            f'<code>{_escape(change.get("path") or "")}</code>'
            "</li>"
        )
    return (
        '<article class="event file-event">'
        '<div class="event-rail file-rail">FILE</div>'
        '<div class="file-card">'
        f'<div class="event-meta">line {_fmt_number(event["line"])} · {_escape(event.get("status") or "")}</div>'
        f'<ul>{"".join(rows) if rows else "<li>No file paths recorded.</li>"}</ul>'
        "</div>"
        "</article>"
    )


def _render_event(event: dict[str, Any]) -> str:
    kind = event.get("kind")
    if kind in {"agent", "user"}:
        return _render_agent(event)
    if kind == "command":
        return _render_command(event)
    if kind == "usage":
        return _render_usage(event)
    if kind == "todo":
        return _render_todo(event)
    if kind == "files":
        return _render_files(event)
    return ""


def _render_transcript(events: list[dict[str, Any]], page: int, total_pages: int) -> str:
    if not events:
        return (
            '<section class="empty-state">'
            "<h2>No matching transcript events</h2>"
            "<p>Try a broader search term or choose another session.</p>"
            "</section>"
        )
    start = (page - 1) * PAGE_SIZE
    visible = events[start : start + PAGE_SIZE]
    rendered = "".join(_render_event(event) for event in visible)
    return (
        '<section class="transcript">'
        f'<div class="page-marker">Page {page:,} of {total_pages:,}</div>'
        f"{rendered}"
        "</section>"
    )


def _render_error(title: str, details: str) -> tuple[str, str, str, int]:
    return (
        '<section class="summary-panel error-panel">'
        f"<h1>{_escape(title)}</h1>"
        f"<p>{_escape(details)}</p>"
        "</section>",
        '<section class="empty-state"><h2>Session unavailable</h2></section>',
        "No session loaded.",
        1,
    )


def _initial_view(choices: list[str]) -> tuple[str, str, str, int]:
    choices_label = f"{len(choices):,} session files available" if choices else "No session files available"
    summary = (
        '<section class="summary-panel">'
        '<div class="summary-topline">'
        "<div><h1>Select a session</h1>"
        f"<p>{_escape(choices_label)} from {DATASET_REPO}</p></div>"
        f'<div class="repo-pill">private dataset · {_escape(APP_REVISION)}</div>'
        "</div>"
        '<div class="metrics-grid">'
        f'{_metric("Startup mode", "metadata only")}'
        f'{_metric("Files listed", _fmt_number(len(choices)))}'
        f'{_metric("Downloads", "on demand")}'
        f'{_metric("Page size", _fmt_number(PAGE_SIZE))}'
        "</div>"
        "</section>"
    )
    transcript = (
        '<section class="empty-state">'
        "<h2>No session loaded yet</h2>"
        "<p>Choose a session in the sidebar and select Load session. Files are downloaded lazily.</p>"
        "</section>"
    )
    return summary, transcript, "No session file downloaded yet.", 1


def render_session(session_name: str | None, query: str | None = "", page: int | float | None = 1) -> tuple[str, str, str, int]:
    choices = KNOWN_SESSIONS
    list_error = ""
    if not choices:
        return _render_error("Dataset file list unavailable", list_error or "HF_TOKEN is not configured.")

    session_name = session_name or choices[0]
    if session_name not in choices and _display_name(str(session_name)) in choices:
        session_name = _display_name(str(session_name))
    candidates = _file_candidates(session_name)
    if not candidates:
        return _render_error("No session selected", "Choose a session from the dropdown.")

    last_error: Exception | None = None
    data = None
    for file_name in candidates:
        try:
            data = _load_session(file_name)
            break
        except Exception as exc:
            last_error = exc
    if data is None:
        details = f"{type(last_error).__name__}: {last_error}" if last_error else "No file candidates were available."
        return _render_error("Could not load session", details)

    query = query or ""
    filtered = _filter_events(data["events"], query)
    total_pages = max(1, math.ceil(len(filtered) / PAGE_SIZE))
    try:
        page_int = int(page or 1)
    except (TypeError, ValueError):
        page_int = 1
    page_int = min(max(1, page_int), total_pages)
    start = (page_int - 1) * PAGE_SIZE + 1 if filtered else 0
    end = min(page_int * PAGE_SIZE, len(filtered))
    page_info = (
        f"Showing {start:,}-{end:,} of {len(filtered):,} matching events "
        f"({len(data['events']):,} total parsed)."
    )
    if data["summary"]["parse_errors"]:
        page_info += f" Skipped {data['summary']['parse_errors']:,} non-JSON lines."
    summary = _render_summary(data, len(filtered), query)
    transcript = _render_transcript(filtered, page_int, total_pages)
    return summary, transcript, page_info, page_int


def load_session(session_name: str | None) -> tuple[str, str, str, int, str]:
    summary, transcript, page_info, page = render_session(session_name, "", 1)
    return summary, transcript, page_info, page, ""


def filter_session(session_name: str | None, query: str | None) -> tuple[str, str, str, int]:
    return render_session(session_name, query, 1)


def next_page(session_name: str | None, query: str | None, page: int | float | None) -> tuple[str, str, str, int]:
    return render_session(session_name, query, int(page or 1) + 1)


def previous_page(session_name: str | None, query: str | None, page: int | float | None) -> tuple[str, str, str, int]:
    return render_session(session_name, query, int(page or 1) - 1)


def refresh_sessions() -> tuple[Any, str, str, str, int, str]:
    _list_sessions_cached.cache_clear()
    choices = _session_choices()
    value = None
    summary, transcript, page_info, page = _initial_view(choices)
    return gr.update(choices=choices, value=value), summary, transcript, page_info, page, ""


CUSTOM_CSS = """
:root {
  --ct-bg: #0b0d10;
  --ct-panel: #11151a;
  --ct-panel-2: #151a21;
  --ct-line: #30363d;
  --ct-text: #e6edf3;
  --ct-muted: #8b949e;
  --ct-green: #3fb950;
  --ct-red: #f85149;
  --ct-yellow: #d29922;
  --ct-blue: #58a6ff;
  --ct-cyan: #39c5cf;
  --ct-mono: "SFMono-Regular", Consolas, "Liberation Mono", Menlo, monospace;
  --ct-sans: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
}

html, body, .gradio-container {
  min-height: 100%;
  background: var(--ct-bg) !important;
  color: var(--ct-text) !important;
  font-family: var(--ct-sans) !important;
}

.gradio-container,
.gradio-container .contain,
.gradio-container main {
  max-width: none !important;
  width: 100% !important;
  padding: 0 !important;
}

#ct-app {
  min-height: 100vh;
  background:
    linear-gradient(180deg, rgba(88, 166, 255, 0.07), transparent 210px),
    var(--ct-bg);
}

.ct-shell {
  display: grid;
  grid-template-columns: minmax(260px, 330px) minmax(0, 1fr);
  gap: 0;
  min-height: 100vh;
}

.ct-sidebar {
  background: #0f1318;
  border-right: 1px solid var(--ct-line);
  padding: 22px 18px;
  position: sticky;
  top: 0;
  height: 100vh;
  overflow: auto;
}

.ct-brand {
  margin-bottom: 22px;
}

.ct-brand h1 {
  font-family: var(--ct-mono);
  font-size: 21px;
  line-height: 1.2;
  margin: 0 0 6px;
  color: var(--ct-text);
  letter-spacing: 0;
}

.ct-brand p {
  margin: 0;
  color: var(--ct-muted);
  font-size: 13px;
  line-height: 1.45;
}

.ct-sidebar .gradio-dropdown,
.ct-sidebar .gradio-textbox,
.ct-sidebar .gradio-button {
  font-family: var(--ct-mono) !important;
}

.ct-sidebar .block,
.ct-sidebar .form,
.ct-sidebar .wrap,
.ct-sidebar .container,
.ct-sidebar .secondary-wrap,
.ct-sidebar .input-container,
.ct-sidebar .block > div {
  background: transparent !important;
  border-color: var(--ct-line) !important;
  box-shadow: none !important;
}

.ct-sidebar .block {
  background: var(--ct-panel) !important;
  border: 1px solid var(--ct-line) !important;
  border-radius: 8px !important;
}

.ct-sidebar label,
.ct-sidebar .block span {
  color: var(--ct-muted) !important;
}

.ct-sidebar input,
.ct-sidebar textarea,
.ct-sidebar .wrap,
.ct-sidebar .container {
  background: var(--ct-panel) !important;
  color: var(--ct-text) !important;
  border-color: var(--ct-line) !important;
}

.ct-sidebar button {
  background: var(--ct-panel-2) !important;
  border: 1px solid var(--ct-line) !important;
  border-radius: 8px !important;
  color: var(--ct-text) !important;
  font-family: var(--ct-mono) !important;
  min-height: 42px;
}

.ct-sidebar button:hover {
  border-color: var(--ct-blue) !important;
  color: #ffffff !important;
}

.ct-sidebar button.primary,
.ct-sidebar .primary button {
  background: #1f6feb !important;
  border-color: #388bfd !important;
  color: #ffffff !important;
}

.ct-main {
  padding: 22px;
  min-width: 0;
}

.summary-panel {
  background: var(--ct-panel);
  border: 1px solid var(--ct-line);
  border-radius: 8px;
  margin-bottom: 18px;
  padding: 18px;
}

.summary-topline {
  display: flex;
  align-items: flex-start;
  justify-content: space-between;
  gap: 16px;
  margin-bottom: 16px;
}

.summary-topline h1 {
  font-family: var(--ct-mono);
  font-size: 22px;
  margin: 0 0 6px;
  color: var(--ct-text);
  letter-spacing: 0;
}

.summary-topline p {
  margin: 0;
  color: var(--ct-muted);
  font-family: var(--ct-mono);
  font-size: 13px;
}

.repo-pill {
  border: 1px solid rgba(57, 197, 207, 0.45);
  color: var(--ct-cyan);
  border-radius: 999px;
  padding: 5px 10px;
  font-family: var(--ct-mono);
  font-size: 12px;
  white-space: nowrap;
}

.metrics-grid {
  display: grid;
  grid-template-columns: repeat(auto-fit, minmax(135px, 1fr));
  gap: 10px;
}

.metric {
  background: var(--ct-panel-2);
  border: 1px solid var(--ct-line);
  border-radius: 8px;
  padding: 10px 12px;
  min-height: 62px;
}

.metric-label {
  display: block;
  color: var(--ct-muted);
  font-size: 12px;
  margin-bottom: 6px;
}

.metric-value {
  display: block;
  color: var(--ct-text);
  font-family: var(--ct-mono);
  font-size: 17px;
  overflow-wrap: anywhere;
}

.metric.ok .metric-value {
  color: var(--ct-green);
}

.metric.danger .metric-value {
  color: var(--ct-red);
}

.transcript {
  display: flex;
  flex-direction: column;
  gap: 14px;
}

.page-marker {
  color: var(--ct-muted);
  font-family: var(--ct-mono);
  font-size: 12px;
  padding: 0 2px;
}

.event {
  display: grid;
  grid-template-columns: 58px minmax(0, 1fr);
  gap: 12px;
  align-items: start;
}

.event-rail {
  position: sticky;
  top: 18px;
  border: 1px solid var(--ct-line);
  border-radius: 8px;
  padding: 8px 0;
  text-align: center;
  font-family: var(--ct-mono);
  font-size: 11px;
  color: var(--ct-muted);
  background: #0f1318;
}

.agent-rail { color: var(--ct-blue); }
.user-rail { color: var(--ct-cyan); }
.command-rail { color: var(--ct-yellow); }
.usage-rail { color: var(--ct-green); }
.todo-rail { color: var(--ct-cyan); }
.file-rail { color: #c9d1d9; }

.bubble,
.command-card,
.usage-card,
.todo-card,
.file-card,
.empty-state {
  border: 1px solid var(--ct-line);
  border-radius: 8px;
  background: var(--ct-panel);
  min-width: 0;
}

.agent-bubble {
  padding: 14px 15px;
  border-left: 3px solid var(--ct-blue);
}

.user-bubble {
  padding: 14px 15px;
  border-left: 3px solid var(--ct-cyan);
  background: #101820;
}

.event-meta {
  color: var(--ct-muted);
  font-family: var(--ct-mono);
  font-size: 12px;
  margin-bottom: 8px;
}

.message-text {
  white-space: pre-wrap;
  overflow-wrap: anywhere;
  color: var(--ct-text);
  line-height: 1.55;
  font-size: 14px;
}

.command-card {
  overflow: hidden;
  border-left: 3px solid var(--ct-yellow);
}

.command-head {
  display: flex;
  align-items: center;
  flex-wrap: wrap;
  gap: 8px;
  padding: 12px 12px 0;
}

.status-badge,
.exit-badge {
  font-family: var(--ct-mono);
  font-size: 12px;
  border-radius: 999px;
  border: 1px solid var(--ct-line);
  padding: 3px 8px;
}

.status-badge.success { color: var(--ct-green); border-color: rgba(63, 185, 80, .5); }
.status-badge.fail { color: var(--ct-red); border-color: rgba(248, 81, 73, .55); }
.status-badge.running { color: var(--ct-yellow); border-color: rgba(210, 153, 34, .55); }
.status-badge.neutral { color: var(--ct-muted); }

.exit-badge {
  color: var(--ct-muted);
}

.code {
  margin: 0;
  overflow: auto;
  font-family: var(--ct-mono);
  font-size: 12.5px;
  line-height: 1.55;
  color: #dbeafe;
  background: #070a0f;
}

.command-code {
  margin: 12px;
  padding: 12px;
  border: 1px solid #1f2937;
  border-radius: 8px;
}

.output-drawer {
  border-top: 1px solid var(--ct-line);
}

.output-drawer summary {
  cursor: pointer;
  color: var(--ct-muted);
  font-family: var(--ct-mono);
  font-size: 12px;
  padding: 10px 12px;
}

.output-code {
  max-height: 520px;
  padding: 12px;
  border-top: 1px solid #1f2937;
}

.usage-card {
  display: flex;
  flex-wrap: wrap;
  gap: 8px;
  padding: 12px;
  border-left: 3px solid var(--ct-green);
}

.usage-card span {
  font-family: var(--ct-mono);
  font-size: 12px;
  color: var(--ct-text);
  background: #0b1117;
  border: 1px solid var(--ct-line);
  border-radius: 999px;
  padding: 4px 9px;
}

.todo-card,
.file-card {
  padding: 12px;
}

.todo-card {
  border-left: 3px solid var(--ct-cyan);
}

.file-card {
  border-left: 3px solid #c9d1d9;
}

.todo-card ul,
.file-card ul {
  list-style: none;
  padding: 0;
  margin: 0;
  display: grid;
  gap: 8px;
}

.todo-row {
  display: grid;
  grid-template-columns: 24px minmax(0, 1fr);
  gap: 8px;
  color: var(--ct-text);
  line-height: 1.45;
}

.todo-check {
  font-family: var(--ct-mono);
  color: var(--ct-muted);
}

.todo-check.done {
  color: var(--ct-green);
}

.file-card li {
  display: flex;
  gap: 10px;
  align-items: baseline;
  min-width: 0;
}

.file-kind {
  color: var(--ct-yellow);
  font-family: var(--ct-mono);
  font-size: 12px;
  min-width: 46px;
}

.file-card code {
  color: var(--ct-text);
  font-family: var(--ct-mono);
  overflow-wrap: anywhere;
}

.empty-state {
  padding: 32px;
  color: var(--ct-muted);
}

.empty-state h2 {
  color: var(--ct-text);
  margin: 0 0 8px;
  font-family: var(--ct-mono);
  font-size: 18px;
  letter-spacing: 0;
}

.error-panel {
  border-color: rgba(248, 81, 73, .6);
}

.ct-page-info {
  color: var(--ct-muted);
  font-family: var(--ct-mono);
  font-size: 12px;
  margin: 8px 0 12px;
}

.ct-page-info,
.ct-page-info * {
  color: var(--ct-muted) !important;
}

.ct-pager {
  display: grid;
  grid-template-columns: 1fr 1fr;
  gap: 8px;
}

@media (max-width: 860px) {
  .ct-shell {
    grid-template-columns: 1fr;
  }
  .ct-sidebar {
    position: relative;
    height: auto;
    border-right: 0;
    border-bottom: 1px solid var(--ct-line);
  }
  .ct-main {
    padding: 14px;
  }
  .event {
    grid-template-columns: 1fr;
  }
  .event-rail {
    position: relative;
    top: 0;
    width: 58px;
  }
  .summary-topline {
    flex-direction: column;
  }
}
"""


def build_app() -> gr.Blocks:
    print("codex-traces startup: building app", flush=True)
    choices = _session_choices(live=False)
    default = None
    initial_summary, initial_transcript, initial_page_info, initial_page = _initial_view(choices)
    theme = gr.themes.Base(
        primary_hue="blue",
        secondary_hue="cyan",
        neutral_hue="slate",
        radius_size="sm",
    )

    with gr.Blocks(title=APP_TITLE, theme=theme, css=CUSTOM_CSS) as demo:
        with gr.Row(elem_id="ct-app", elem_classes=["ct-shell"]):
            with gr.Column(elem_classes=["ct-sidebar"], scale=0):
                gr.HTML(
                    '<div class="ct-brand">'
                    "<h1>Codex Traces</h1>"
                    "<p>Private rollout.jsonl session viewer</p>"
                    "</div>"
                )
                session = gr.Dropdown(
                    choices=choices,
                    value=default,
                    label="Session file",
                    interactive=True,
                    allow_custom_value=False,
                )
                search = gr.Textbox(
                    label="Search transcript",
                    placeholder="Filter by keyword, command, output, token...",
                    lines=1,
                    max_lines=1,
                )
                filter_button = gr.Button("Filter", variant="primary")
                load_button = gr.Button("Load session")
                refresh_button = gr.Button("Refresh file list")
                page_state = gr.State(initial_page)
                page_info = gr.Markdown(initial_page_info, elem_classes=["ct-page-info"])
                with gr.Row(elem_classes=["ct-pager"]):
                    prev_button = gr.Button("Previous")
                    next_button = gr.Button("Next")

            with gr.Column(elem_classes=["ct-main"], scale=1):
                summary = gr.HTML(initial_summary)
                transcript = gr.HTML(initial_transcript)

        session.change(
            fn=load_session,
            inputs=session,
            outputs=[summary, transcript, page_info, page_state, search],
            api_name="load_session",
            show_progress="minimal",
        )
        load_button.click(
            fn=load_session,
            inputs=session,
            outputs=[summary, transcript, page_info, page_state, search],
            api_name=False,
            show_progress="minimal",
        )
        filter_button.click(
            fn=filter_session,
            inputs=[session, search],
            outputs=[summary, transcript, page_info, page_state],
            api_name="filter_session",
            show_progress="minimal",
        )
        search.submit(
            fn=filter_session,
            inputs=[session, search],
            outputs=[summary, transcript, page_info, page_state],
            api_name=False,
            show_progress="minimal",
        )
        prev_button.click(
            fn=previous_page,
            inputs=[session, search, page_state],
            outputs=[summary, transcript, page_info, page_state],
            api_name="previous_page",
            show_progress="minimal",
        )
        next_button.click(
            fn=next_page,
            inputs=[session, search, page_state],
            outputs=[summary, transcript, page_info, page_state],
            api_name="next_page",
            show_progress="minimal",
        )
        refresh_button.click(
            fn=refresh_sessions,
            inputs=None,
            outputs=[session, summary, transcript, page_info, page_state, search],
            api_name="refresh_sessions",
            show_progress="minimal",
        )

    print("codex-traces startup: app built", flush=True)
    return demo


demo = build_app()
print("codex-traces startup: demo ready", flush=True)

print("codex-traces startup: launching", flush=True)
demo.launch(
    server_name="0.0.0.0",
    server_port=int(os.environ.get("PORT", "7860")),
    show_error=True,
)