replay.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/scripts/replay.py
 
"""
CLI shell for offline session replay over the event NDJSON — M2.7.
 
Purpose
-------
A thin ``argparse`` front-end over :mod:`oc_chat.core.replay`. Resolves an event
directory, reads + reconstructs the session tree, and prints it as text or
``replay/v0`` JSON. Three subcommands: ``session`` (full tree), ``turn`` (one
turn by id or ≥4-char prefix), ``summary`` (per-kind counts only). Never calls a
provider; never mutates the log.
 
Responsibilities
----------------
- :func:`main` — parse args, dispatch, map outcomes to exit codes (0 ok,
  1 unexpected error, 2 user error) with a one-line ``error:`` on stderr.
- :func:`_resolve_turn` — turn-id prefix matching with ambiguity handling.
 
Diagnostics
-----------
Domain: OC_CHAT.REPLAY
Levels:
    L2 — CLI invocation + chosen subcommand.
    L3 — resolution detail (event count, matched turn).
    L4 — module-load coverage marker.
 
Contracts
---------
- Exit 0 on success, 2 on user error (missing dir / bad prefix / bad args),
  1 on unexpected exception; all errors print ``error: …`` to stderr first.
- Read-only: no provider calls, no mutation of the event log.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.replay (read_events_from_dir, build_replay_tree, renderers)
"""
from __future__ import annotations
 
import argparse
import sys
from pathlib import Path
from typing import Optional, Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.core.replay import (
    ReplayTree,
    build_replay_tree,
    read_events_from_dir,
    render_tree_json,
    render_tree_text,
)
 
_dbg("OC_CHAT.REPLAY", "replay_cli", "module", 4)
 
_MIN_PREFIX = 4
 
 
class _UserError(Exception):
    """Raised for user-facing errors that map to exit code 2."""
 
 
def _load_tree(event_dir: str) -> ReplayTree:
    """Read *event_dir* and build the replay tree; raise :class:`_UserError` if missing."""
    base = Path(event_dir)
    if not base.exists():
        raise _UserError(f"event directory not found: {event_dir}")
    events, skipped = read_events_from_dir(base)
    tree = build_replay_tree(events, skipped=skipped)
    tree.source_dir = str(base)
    return tree
 
 
def _resolve_turn(tree: ReplayTree, prefix: str) -> str:
    """Return the single ``turn_id`` matching *prefix* (≥4 chars); raise on miss/ambiguity."""
    if len(prefix) < _MIN_PREFIX:
        raise _UserError(f"turn-id prefix must be at least {_MIN_PREFIX} characters")
    matches = [t.turn_id for t in tree.turns if t.turn_id.startswith(prefix)]
    if not matches:
        raise _UserError(f"no turn matches prefix '{prefix}'")
    if len(matches) > 1:
        raise _UserError(f"prefix '{prefix}' matches {len(matches)} turns; please narrow to a longer prefix")
    return matches[0]
 
 
def _build_parser() -> argparse.ArgumentParser:
    """Build the ``replay`` argument parser with its three subcommands."""
    parser = argparse.ArgumentParser(
        prog="python -m oc_chat.scripts.replay",
        description="Offline replay of an OC Chat session from its event NDJSON (no provider calls).",
    )
    sub = parser.add_subparsers(dest="cmd", required=True)
 
    p_session = sub.add_parser("session", help="render the full session tree")
    p_session.add_argument("event_dir")
    p_session.add_argument("--json", action="store_true", help="emit canonical replay/v0 JSON")
    p_session.add_argument("--limit", type=int, default=None, help="render only the last N turns")
 
    p_turn = sub.add_parser("turn", help="render one turn by id or >=4-char prefix")
    p_turn.add_argument("event_dir")
    p_turn.add_argument("turn_id")
    p_turn.add_argument("--json", action="store_true", help="emit canonical replay/v0 JSON")
 
    p_summary = sub.add_parser("summary", help="print per-kind counts + turn count only")
    p_summary.add_argument("event_dir")
    p_summary.add_argument("--json", action="store_true", help="emit canonical replay/v0 JSON")
    return parser
 
 
def _run_summary(tree: ReplayTree, as_json: bool) -> None:
    """Print the cheap per-kind count view for *tree*."""
    if as_json:
        print(render_tree_json(ReplayTree(
            session_id=tree.session_id, turn_count=tree.turn_count,
            event_count=tree.event_count, skipped_count=tree.skipped_count,
            source_dir=tree.source_dir, event_counts=tree.event_counts, turns=[],
        )))
        return
    print(f"── Replay summary: session {(tree.session_id or '-')[:8]} ──")
    print(f"  events: {tree.event_count}  turns: {tree.turn_count}  skipped: {tree.skipped_count}")
    for kind in sorted(tree.event_counts):
        print(f"  {kind:<24} {tree.event_counts[kind]}")
 
 
def main(argv: Optional[Sequence[str]] = None) -> int:
    """Entry point. Returns the process exit code (0 ok, 1 error, 2 user error)."""
    func = "main"
    for _stream in (sys.stdout, sys.stderr):
        try:
            _stream.reconfigure(errors="replace")
        except (AttributeError, ValueError):
            pass  # non-reconfigurable stream (tests, pipes wrapped by harnesses)
    parser = _build_parser()
    args = parser.parse_args(argv)
    try:
        _dbg("OC_CHAT.REPLAY", func, f"cmd={args.cmd} dir={args.event_dir}", 2)
        tree = _load_tree(args.event_dir)
        if args.cmd == "summary":
            _run_summary(tree, args.json)
        elif args.cmd == "turn":
            turn_id = _resolve_turn(tree, args.turn_id)
            print(render_tree_json(tree, single_turn_id=turn_id) if args.json
                  else render_tree_text(tree, single_turn_id=turn_id))
        else:  # session
            if args.limit is not None and args.limit >= 0:
                tree.turns = tree.turns[-args.limit:]
            print(render_tree_json(tree) if args.json else render_tree_text(tree))
        return 0
    except _UserError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    except Exception as exc:  # noqa: BLE001 — CLI boundary: no stack trace to the user
        _dbg("OC_CHAT.REPLAY", func, f"SYSTEM_ERROR: {type(exc).__name__}: {exc}", 1, override=True)
        print(f"error: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1
 
 
if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())