enhanced_presenter.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/enhanced_presenter.py
 
"""
Enhanced terminal presenter with statistics hooks and optional fixed header.
 
Purpose
-------
Subclass of :class:`RichPresenter` that adds a thread-safe
:class:`StatisticsTracker` for turn / session / context-usage counters
and (when the terminal is interactive) installs a fixed-position
ANSI-coloured header at the top of the scroll region.  Selected via
``--ui enhanced``.
 
Responsibilities
----------------
- :class:`EnhancedPresenter` — Rich-styled presenter with stats hooks.
- ``_install_fixed_header`` / ``_redraw_fixed_header`` — manage the
  fixed top region and re-render after each turn or context change.
- Pass-through hooks for agent / sub-loop tracking that delegate to
  :class:`StatisticsTracker`.
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events (init, close)
    L3 — per-turn / context-usage updates
    L4 — deep tracing (per-redraw decisions)
 
Contracts
---------
- All rendering still goes through the parent :class:`RichPresenter`;
  no presenter API surface is added or removed.
- The fixed header is opt-in: skipped when ``stdout`` is not a TTY or
  when ``OC_CHAT_ENHANCED_FIXED_HEADER`` is falsy.
- ``close()`` always restores the terminal's full scroll region.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_ui.terminal.stats (StatisticsTracker)
- oc_ui.terminal.rich_presenter (RichPresenter base class)
- oc_chat.core.config (lazy; get_max_context_tokens for the context gauge window size)
- oc_chat.core.formatting (lazy; format_duration for the fixed header turn and session durations)
- oc_chat.core.session (lazy; ChatSession type hint for bind_session and show_welcome)
"""
 
from __future__ import annotations
 
import os
import shutil
from typing import Any, TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
from oc_ui.terminal.stats import StatisticsTracker
from oc_ui.terminal.rich_presenter import RichPresenter
 
if TYPE_CHECKING:
    from oc_chat.core.session import ChatSession
 
 
class EnhancedPresenter(RichPresenter):
    """Rich-styled presenter with enhanced statistics hooks."""
 
    def __init__(
        self,
        model: str | None = "deepseek-v4-pro",
        max_context_size: int | None = None,
        *,
        verbose: bool = False,
        **rich_kwargs: Any,
    ) -> None:
        func = "EnhancedPresenter.__init__"
        super().__init__(verbose=verbose, **rich_kwargs)
        self._model = model or ""
        # Plan 134 C2/N5: the context gauge must divide by the REAL model
        # window, not a hardcoded 128k default. None → auto-derive from the
        # model via get_max_context_tokens (and refresh at bind_session + on
        # model switch); an explicit value is an override that always wins.
        self._auto_context_size = max_context_size is None
        if max_context_size is None:
            from oc_chat.core.config import get_max_context_tokens
            max_context_size = get_max_context_tokens(self._model)
        self._max_context_size = max_context_size
        self._stats = StatisticsTracker()
        self._stats.set_context_window_size(max_context_size)
        self._header_session = None
        self._fixed_header_active = False
        self._fixed_header_height = 8
        self._fixed_header_enabled = (
            self._console.is_terminal
            and os.getenv("OC_CHAT_ENHANCED_FIXED_HEADER", "1").lower()
            not in {"0", "false", "no", "off"}
        )
        _dbg("OC_UI.TERMINAL", func, "initialized", 2)
 
    def bind_session(self, session: "ChatSession") -> None:
        """Bind the active session and keep the display model + context window current.
 
        Plan 134 N5: ``session.model`` is only known here (the presenter is
        built before the session resolves), and it changes on /model /pro
        /flash. When the window size is auto (no explicit --max-context-size),
        refresh the gauge denominator from the real model so it agrees with the
        turn-summary ctx:% gauge instead of diverging on a 128k default.
        """
        super().bind_session(session)
        self._model = getattr(session, "model", self._model) or self._model
        self._header_session = session
        if self._auto_context_size:
            from oc_chat.core.config import get_max_context_tokens
            win = get_max_context_tokens(self._model)
            self._max_context_size = win
            self._stats.set_context_window_size(win)
        self._redraw_fixed_header()
 
    def show_welcome(self, session: "ChatSession") -> None:
        """Display the welcome panel or install the fixed enhanced header."""
        self._model = getattr(session, "model", self._model) or self._model
        self._header_session = session
        if self._fixed_header_enabled:
            self._install_fixed_header()
            return
        super().show_welcome(session)
 
    def _install_fixed_header(self) -> None:
        """Reserve a fixed top terminal region and draw the header there."""
        columns, lines = shutil.get_terminal_size((100, 30))
        if lines <= self._fixed_header_height + 4:
            super().show_welcome(self._header_session)
            return
 
        self._fixed_header_active = True
        self._console.file.write("\x1b[2J\x1b[H")
        self._console.file.write(f"\x1b[{self._fixed_header_height + 1};{lines}r")
        self._console.file.write(f"\x1b[{self._fixed_header_height + 1};1H")
        self._console.file.flush()
        self._redraw_fixed_header()
 
    def _redraw_fixed_header(self) -> None:
        """Redraw the fixed header without moving the scroll-region cursor."""
        if not self._fixed_header_active or self._header_session is None:
            return
 
        columns, lines = shutil.get_terminal_size((100, 30))
        if lines <= self._fixed_header_height + 4:
            return
 
        header = self._fixed_header_lines(columns)
        self._console.file.write("\x1b[s")
        self._console.file.write("\x1b[1;1H")
        for idx in range(self._fixed_header_height):
            self._console.file.write("\x1b[2K")
            if idx < len(header):
                self._console.file.write(header[idx])
            if idx < self._fixed_header_height - 1:
                self._console.file.write("\n")
        self._console.file.write("\x1b[u")
        self._console.file.flush()
 
    def _fixed_header_lines(self, columns: int) -> list[str]:
        """Build ANSI-coloured header lines sized to the terminal width."""
        session = self._header_session
        stats = self._stats.get_stats_snapshot()
        width = max(50, columns)
        inner = width - 2
        blue = "\033[36m"
        cyan = "\033[96m"
        yellow = "\033[33m"
        dim = "\033[2m"
        reset = "\033[0m"
 
        model = getattr(session, "model", self._model) or self._model
        effort = getattr(session, "reasoning_effort", "")
        name = getattr(session, "name", "session")
        title = " OC Chat "
 
        turn_total = stats["last_turn"]["total"]
        turn_tools = stats["last_turn"]["tool_calls"]
        turn_cache = stats["last_turn"]["cache_hit_rate"]
        sess_total = stats["session"]["tokens"]["total"]
        sess_tools = stats["session"]["timing"]["tool_calls"]
        sess_cache = stats["session"]["tokens"]["cache_hit_rate"]
        context_pct = stats["context"]["usage_percent"]
 
        # Plan 147 D1/D2: real turn/session durations from the bound status
        # provider — blank until the first refresh_presenter_bindings call
        # (e.g. before startup finishes binding).
        turn_duration = ""
        session_duration = ""
        status = self._status_provider() if self._status_provider is not None else None
        if status is not None:
            from oc_chat.core.formatting import format_duration
            turn_duration = f" · {format_duration(status.turn_elapsed_ms)}"
            session_duration = f" · {format_duration(status.session_elapsed_ms)}"
 
        top_fill = max(0, inner - len(title))
        top_left = top_fill // 2
        top_right = top_fill - top_left
 
        def cell(text: str = "") -> str:
            plain = text[:inner]
            return f"{blue}│{reset}{plain.ljust(inner)}{blue}│{reset}"
 
        return [
            f"{blue}╭{'─' * top_left}{title}{'─' * top_right}╮{reset}",
            cell(),
            cell(f"  Model: {cyan}{model}{reset}"),
            cell(f"  Reasoning effort: {cyan}{effort}{reset}  {dim}(name: {name}){reset}"),
            cell(),
            cell(
                "  "
                f"{yellow}this turn:{reset} {self._fmt_compact_tokens(turn_total)} · "
                f"{turn_tools} tools · {turn_cache:.0f}% cache · ctx {context_pct:.0f}%{turn_duration}    "
                f"{yellow}session:{reset} {self._fmt_compact_tokens(sess_total)} · "
                f"{sess_tools} tools · {sess_cache:.0f}% cache{session_duration}"
            ),
            cell(f"  Type {yellow}/help{reset} for full command details and session subcommands."),
            f"{blue}╰{'─' * inner}╯{reset}",
        ]
 
    def update_turn_stats(
        self,
        prompt: int,
        completion: int,
        cache_hits: int = 0,
        duration_ms: int = 0,
        tool_calls: int = 0,
    ) -> None:
        """Update enhanced statistics after a turn or instrumentation event."""
        self._stats.update_turn_stats(
            prompt,
            completion,
            cache_hits,
            duration_ms,
            tool_calls,
        )
        self._redraw_fixed_header()
 
        func = "update_turn_stats"
        _dbg(
            "OC_UI.TERMINAL",
            func,
            f"updated: prompt={prompt}, completion={completion}, "
            f"cache_hits={cache_hits}, duration={duration_ms}ms",
            3,
        )
 
    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:
        """Render the rich summary and mirror it into enhanced stats.
 
        Plan 188 T4: *turn_cost* is pure pass-through — the stats tracker
        counts tokens, and the ``~$`` estimate is rendered by the Rich base.
        """
        self._stats.update_turn_stats(
            prompt,
            completion,
            cache_saved_tokens or prompt_cached or 0,
            duration_ms or 0,
            tool_calls,
        )
        self._redraw_fixed_header()
        super().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 update_context_usage(self, used_tokens: int) -> None:
        """Update context-window usage in the enhanced tracker."""
        self._stats.update_context_usage(used_tokens)
        self._redraw_fixed_header()
 
        func = "update_context_usage"
        _dbg(
            "OC_UI.TERMINAL",
            func,
            f"context usage: {used_tokens}/{self._max_context_size}",
            3,
        )
 
    def add_agent(self) -> None:
        """Add a new agent to tracking."""
        self._stats.add_agent()
        self._redraw_fixed_header()
 
    def finish_agent(self) -> None:
        """Mark an agent as finished."""
        self._stats.finish_agent()
        self._redraw_fixed_header()
 
    def add_sub_loop(self) -> None:
        """Add a new sub-loop to tracking."""
        self._stats.add_sub_loop()
        self._redraw_fixed_header()
 
    def finish_sub_loop(self) -> None:
        """Mark a sub-loop as finished."""
        self._stats.finish_sub_loop()
        self._redraw_fixed_header()
 
    def get_stats_tracker(self) -> StatisticsTracker:
        """Return the enhanced statistics tracker."""
        return self._stats
 
    def show_sub_loop_trigger(self, model: str, task: str) -> None:
        """Display that a sub-loop was triggered using rich styling."""
        self._console.print(f"[cyan]Sub-loop:[/] {model}")
 
    def close(self) -> None:
        """Reset the terminal scroll region on shutdown."""
        if self._fixed_header_active:
            self._console.file.write("\x1b[r")
            self._console.file.write(f"\x1b[{self._fixed_header_height + 1};1H")
            self._console.file.flush()
            self._fixed_header_active = False