logging_presenter.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/logging_presenter.py
 
"""
Decorator presenter that tees every event to a JSON-lines file.
 
Purpose
-------
Wraps any concrete :class:`~oc_ui.presenter_base.Presenter` and emits
one JSON object per call to *<path>*. The wrapped presenter still
renders normally — :class:`LoggingPresenter` is a strict pass-through
that just records.
 
Use cases:
 
* **Replay** — drive a different presenter through a recorded session
  to compare rendering changes without re-running the model.
* **CI diffing** — capture the event stream for ``ConsolePresenter``
  and ``RichPresenter`` against the same scripted input and assert
  they emit the same logical events.
* **Debugging** — when a user reports a UI glitch, asking them to
  re-run with ``--log-events sessions/<name>/events.jsonl`` produces
  a reproducible trace.
 
The format is intentionally minimal — one ``{event, args, ts}``
object per line, using :func:`json.dumps` with ``ensure_ascii=False``.
The file is opened in append mode so multiple turns concatenate into
the same log; ``flush=True`` after each event keeps the log usable
from ``tail -f`` for live monitoring.
 
Responsibilities
----------------
- :class:`LoggingPresenter` — implement every :class:`Presenter` hook
  by writing a JSON line to the configured path *and* delegating to
  the wrapped presenter.
- Provide a ``close`` method so callers can flush the log
  deterministically at session shutdown.
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- The decorator is *strict pass-through*: every event reaches the
  inner presenter, in order, with unmodified arguments.
- Each line is a complete JSON object terminated by ``\\n``; partial
  writes are not permitted.
- The log file is append-mode — never truncated.
- Failures while writing the log are surfaced as exceptions; the
  decorator does not silently drop events.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_ui.presenter_base (Presenter ABC contract)
- oc_chat.core.iteration_stat (lazy; show_iteration_stat imports its JSON and human-line formatters, currently unused)
- oc_chat.core.session (lazy; ChatSession type hint for show_welcome and bind_session)
- oc_ui.terminal.statusline (lazy; StatusBarState type hint for bind_status)
"""
from __future__ import annotations
 
import json
import time
from contextlib import AbstractContextManager
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable
 
from oc_diagnostics import _dbg
 
from oc_ui.presenter_base import Presenter, SpinnerHandle
 
if TYPE_CHECKING:
    from oc_chat.core.session import ChatSession
    from oc_ui.terminal.statusline import StatusBarState
 
 
class _TeeContext(AbstractContextManager):
    """Wrap an inner contextmanager + record entry/exit events."""
 
    def __init__(self, inner: AbstractContextManager,
                 emit, kind: str = "turn") -> None:
        self._inner = inner
        self._emit = emit
        self._kind = kind
 
    def __enter__(self):
        self._emit(f"{self._kind}.enter", {})
        return self._inner.__enter__()
 
    def __exit__(self, exc_type, exc, tb):
        self._emit(f"{self._kind}.exit", {
            "exc_type": exc_type.__name__ if exc_type else None,
        })
        return self._inner.__exit__(exc_type, exc, tb)
 
 
class LoggingPresenter(Presenter):
    """A :class:`Presenter` that tees every event to a JSONL log.
 
    All :class:`Presenter` methods are pass-through to *inner*; in
    addition, each call appends a JSON line to *path* of the form::
 
        {"ts": 1714809000.123, "event": "show_content", "args": {"text": "..."}}
 
    *truncate* opens the log fresh (overwriting any previous content);
    the default is to append.
    """
 
    def __init__(self, inner: Presenter, path: str, truncate: bool = False) -> None:
        func = "__init__"
        self._inner = inner
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)
        mode = "w" if truncate else "a"
        # Line-buffered, utf-8 — survives tail -f and never holds a write.
        self._file = self._path.open(mode, encoding="utf-8", buffering=1)
        _dbg("OC_UI.TERMINAL", func, f"LoggingPresenter path={path} truncate={truncate}", 2)
 
    # ── Internal ──────────────────────────────────────────────────────────
 
    def _emit(self, event: str, args: dict[str, Any]) -> None:
        record = {
            "ts": round(time.time(), 3),
            "event": event,
            "args": args,
        }
        try:
            line = json.dumps(record, ensure_ascii=False)
        except (TypeError, ValueError):
            # Fall back to a repr-based serialisation if a non-JSONable
            # object slips through.  We never want logging to crash the
            # presenter.
            safe_args = {k: repr(v) for k, v in args.items()}
            line = json.dumps({"ts": record["ts"], "event": event,
                               "args": safe_args, "_repr_fallback": True},
                              ensure_ascii=False)
        self._file.write(line + "\n")
 
    def close(self) -> None:
        """Close the underlying log file.  Safe to call multiple times."""
        if not self._file.closed:
            self._file.close()
 
    # ── Pass-through with tee ─────────────────────────────────────────────
 
    def turn(self) -> AbstractContextManager:
        return _TeeContext(self._inner.turn(), self._emit, kind="turn")
 
    def start_turn(self) -> None:
        self._emit("start_turn", {})
        self._inner.start_turn()
 
    def end_turn(self) -> None:
        self._emit("end_turn", {})
        self._inner.end_turn()
 
    def show_waiting(self) -> None:
        self._emit("show_waiting", {})
        self._inner.show_waiting()
 
    def show_reasoning(self, text: str) -> None:
        self._emit("show_reasoning", {"text": text})
        self._inner.show_reasoning(text)
 
    def show_content(self, text: str) -> None:
        self._emit("show_content", {"text": text})
        self._inner.show_content(text)
 
    def show_tool_call(self, name: str, arguments: str) -> None:
        self._emit("show_tool_call", {"name": name, "arguments": arguments})
        self._inner.show_tool_call(name, arguments)
 
    def show_tool_progress(self, name: str) -> None:
        self._emit("show_tool_progress", {"name": name})
        self._inner.show_tool_progress(name)
 
    def show_tool_result(self, name: str, result: object) -> None:
        self._emit("show_tool_result", {"name": name, "result": result})
        self._inner.show_tool_result(name, result)
 
    def show_usage(self, usage: dict) -> None:
        self._emit("show_usage", {"usage": usage})
        self._inner.show_usage(usage)
 
    def show_activity(self, label: str, detail: str = "", *, tone: str = "info") -> None:
        """Record the plan 188 T2 activity line and delegate.
 
        Strict pass-through like every other hook — the decorator records
        the logical event and the wrapped presenter renders it.
        """
        self._emit("show_activity", {"label": label, "detail": detail, "tone": tone})
        self._inner.show_activity(label, detail, tone=tone)
 
    def show_session_start(self) -> None:
        """Record the plan 218 session-opening rule and delegate.
 
        Strict pass-through like every other hook: the decorator records
        the logical event and the wrapped presenter draws the rule.
        """
        self._emit("show_session_start", {})
        self._inner.show_session_start()
 
    def show_interrupt_notice(self, stage: str) -> None:
        """Record the plan 188 T5 Ctrl+C stage and delegate.
 
        Signal-handler context: ``_emit`` is one buffered line write and the
        inner presenter owns the terminal write, so this stays within the
        one-short-write budget the hook's contract sets.
        """
        self._emit("show_interrupt_notice", {"stage": stage})
        self._inner.show_interrupt_notice(stage)
 
 
    def show_iteration_stat(self, stat: "IterationStat") -> None:
        """Log and delegate iteration stat."""
        from oc_chat.core.iteration_stat import iteration_stat_to_json, iteration_stat_human_line
        self._emit("show_iteration_stat", {"stat": stat})
        self._inner.show_iteration_stat(stat)
    def show_turn_summary(self, prompt: int, completion: int, total: int,
                          iterations: int, tool_calls: int,
                          prompt_cached: int | None = None, *,
                          cache_hit_pct: float | None = None,
                          cache_saved_tokens: int | None = None,
                          performance_stats: object = None,
                          model: str | None = None,
                          context_used_pct: float | None = None,
                          duration_ms: int | None = None,
                          turn_cost: object = None) -> None:
        # Plan 188 T4: record the cost as flat JSON-able fields (the dataclass
        # itself is not JSON-serialisable) so a replay can audit both the
        # figure AND the rate-table provenance behind it. ``None`` stays None —
        # an unpriceable turn must read as absent, not as zero.
        _cost = None
        if turn_cost is not None:
            _cost = {
                "total_usd": getattr(turn_cost, "total_usd", None),
                "model": getattr(turn_cost, "model", None),
                "provider": getattr(turn_cost, "provider", None),
                "as_of": getattr(turn_cost, "as_of", None),
                "source": getattr(turn_cost, "source", None),
            }
        self._emit("show_turn_summary", {
            "prompt": prompt, "completion": completion, "total": total,
            "iterations": iterations, "tool_calls": tool_calls,
            "prompt_cached": prompt_cached,
            "cache_hit_pct": cache_hit_pct,
            "cache_saved_tokens": cache_saved_tokens,
            # Plan 134 C2: record the honest context gauge + its model too.
            "model": model,
            "context_used_pct": context_used_pct,
            # Plan 147 D1: the real wall-clock turn duration.
            "duration_ms": duration_ms,
            # Plan 188 T4: the modelled USD estimate + its price provenance.
            "turn_cost": _cost,
        })
        self._inner.show_turn_summary(
            prompt, completion, total, iterations, tool_calls, prompt_cached,
            cache_hit_pct=cache_hit_pct, cache_saved_tokens=cache_saved_tokens,
            performance_stats=performance_stats,
            model=model, context_used_pct=context_used_pct,
            duration_ms=duration_ms, turn_cost=turn_cost,
        )
 
    def show_welcome(self, session: ChatSession) -> None:
        self._emit("show_welcome", {"name": getattr(session, "name", "?"),
                                    "model": getattr(session, "model", "?")})
        self._inner.show_welcome(session)
 
    def show_available_models(self, models: dict) -> None:
        self._emit("show_available_models", {"count": len(models)})
        self._inner.show_available_models(models)
 
    def show_session_list(self, sessions: list, current_name: str | None = None) -> None:
        self._emit("show_session_list",
                   {"count": len(sessions), "current": current_name})
        self._inner.show_session_list(sessions, current_name)
 
    def show_help(self, command_help: dict, tools: list[str] | None = None) -> None:
        self._emit("show_help",
                   {"command_count": len(command_help),
                    "tool_count": len(tools or [])})
        self._inner.show_help(command_help, tools)
 
    def show_compressed(self) -> None:
        self._emit("show_compressed", {})
        self._inner.show_compressed()
 
    def show_error(self, msg: str) -> None:
        self._emit("show_error", {"msg": msg})
        self._inner.show_error(msg)
 
    def show_info(self, msg: str) -> None:
        self._emit("show_info", {"msg": msg})
        self._inner.show_info(msg)
 
    # ── Progress spinner (Phase H.4) ──────────────────────────────────────
 
    def spinner_start(self, label: str) -> SpinnerHandle:
        """Record the spinner_start event and delegate to the wrapped presenter."""
        self._emit("spinner_start", {"label": label})
        return self._inner.spinner_start(label)
 
    def spinner_stop(self, handle: SpinnerHandle) -> None:
        """Record the spinner_stop event and delegate to the wrapped presenter."""
        self._emit("spinner_stop", {})
        self._inner.spinner_stop(handle)
 
    # ── Pass-through for optional bind_session (Rich-only) ────────────────
 
    def bind_session(self, session: ChatSession) -> None:
        """Forward to the wrapped presenter if it implements it.
 
        Not part of the abstract contract — only Rich/Textual implement
        this — but :func:`main._select_presenter` checks via ``hasattr``.
        """
        self._emit("bind_session", {"name": getattr(session, "name", "?")})
        if hasattr(self._inner, "bind_session"):
            self._inner.bind_session(session)
 
    def bind_status(self, provider: Callable[[], StatusBarState | None]) -> None:
        """Record the bind_status event and delegate to the wrapped presenter.
 
        Plan 147 D1: pass-through only — a ``LoggingPresenter`` records
        logical events and cannot see rendering, so it never calls
        *provider* itself; it just hands it to ``self._inner``.
        """
        self._emit("bind_status", {})
        self._inner.bind_status(provider)