statusline.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/statusline.py
 
"""
Shared status-bar data model and formatter — the realtime chrome spine.
 
Purpose
-------
Owns the one pure snapshot type (:class:`StatusBarState`) and the one
shared string-formatting logic for the always-visible status bar that
plan 147 renders across all three UI regimes (idle prompt, streaming
Live, tool-dispatch chrome). Framework-free by design — imports
neither ``rich`` nor ``prompt_toolkit`` — so both
:class:`~oc_ui.terminal.input.PromptDriver` (prompt_toolkit) and
:class:`~oc_ui.terminal.rich_presenter.RichPresenter` (rich) render
from the SAME segment/format logic instead of hand-rolling their own
bar strings.
 
Responsibilities
----------------
- :class:`StatusBarState` — frozen, slotted, pure data snapshot. No
  busy/activity fields — that state is presenter-owned (see
  :meth:`~oc_ui.presenter_base.Presenter.bind_status`).
- :func:`fmt_tokens_short` — compact token-count formatting (``1.2K``).
- :func:`render_status_line` — build the ordered segments for a
  snapshot + busy label, then join/drop them to fit a target width.
- :func:`busy_label_with_activity` — compose a busy label with the
  in-flight tool call's elapsed time and last output line (plan 279).
- :func:`busy_width_left` — the columns the bar still has free once it
  has been rendered with a given busy label.
- :func:`busy_hint_text` — the dim Ctrl+C/Ctrl+D hint line shown
  adjacent to the bar.
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Framework-free: no ``rich`` / ``prompt_toolkit`` imports here — both
  presenters format through this module instead of redefining bar
  strings.
- ``render_status_line`` never wraps; it drops segments right-to-left
  in the pinned order (effort -> cache-saved -> cache% -> tools ->
  iters, then -- for a bar that carries an activity tail -- the session
  cluster, then the turn cluster) until the rendered line fits
  ``target_width``, then stops (the remaining core segments may still
  overflow a very narrow width).
- The activity tail OUTRANKS every droppable segment and both clusters
  (operator ruling TUI-06, 2026-09-16; amended 2026-09-17): the bar gives
  up effort, the cache percentages, the saved-token segments, the tool
  and iteration counters and then the session and turn clusters -- in
  that order -- before it will shrink the tail or drop it. The model, the
  session name, the busy label with its elapsed timer, and the standing
  markers (a live grant, the session's spend, the checkpoint latch) are
  never dropped.
- A bar with no in-flight tool call carries no tail: it stops dropping at
  the end of ``_DROPPABLE`` and renders byte-identically to the pre-280
  bar at every width.
- ``StatusBarState`` carries no busy/activity fields — presenters pass
  ``busy_label`` as a separate argument; it is never captured on the
  snapshot itself. Plan 279's tool timer and tail are composed INTO
  that busy label by :func:`busy_label_with_activity`, so the snapshot
  and the segment table are unchanged by it.
- Width comparisons count wide/fullwidth Unicode (CJK) as two terminal
  columns via :func:`unicodedata.east_asian_width`, not raw ``len()``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.formatting (lazy; format_duration for the turn, session and tool call timers)
"""
from __future__ import annotations
 
import unicodedata
from dataclasses import dataclass
 
from oc_diagnostics import _dbg
 
_dbg("OC_UI.TERMINAL", "module", "loaded", 4)
 
 
@dataclass(frozen=True, slots=True)
class StatusBarState:
    """Pure snapshot of status-bar data for one render.
 
    Busy/activity state is deliberately NOT here — it is presenter-owned
    (driven by ``show_waiting`` / ``spinner_start`` / ``spinner_stop``)
    and passed to :func:`render_status_line` as a separate ``busy_label``
    argument, never captured on the snapshot. All duration fields are
    integer milliseconds; all cache-hit fields are a 0-100 percentage or
    ``None`` when no cache data is available yet.
    """
 
    model: str
    """Active model id."""
    session_name: str
    """Session display name."""
    effort: str
    """Active reasoning-effort level (``low`` / ``medium`` / ``high``)."""
 
    turn_tokens: int
    """Total tokens (prompt + completion) for the in-progress or last turn."""
    turn_iterations: int
    """ReAct iteration count for the in-progress or last turn."""
    turn_tools: int
    """Tool-call count for the in-progress or last turn."""
    turn_cache_hit_pct: float | None
    """Cache-hit ratio (0-100) for the in-progress or last turn; ``None`` = no cache data."""
    turn_cache_saved_tokens: int
    """Prompt tokens saved by cache hits, in-progress or last turn."""
    turn_elapsed_ms: int
    """Milliseconds elapsed in the in-progress turn, or the last completed turn's total duration when idle."""
 
    session_tokens: int
    """Total tokens (prompt + completion) accumulated this session."""
    session_tools: int
    """Total tool calls accumulated this session."""
    session_cache_hit_pct: float | None
    """Cache-hit ratio (0-100) across the session; ``None`` = no cache data."""
    session_cache_saved_tokens: int
    """Prompt tokens saved by cache hits across the session."""
    session_elapsed_ms: int
    """Milliseconds since the active session became current in this process. Resets whenever the active
    session is replaced or (re)loaded — see :func:`oc_chat.core.commands.refresh_presenter_bindings`."""
 
    session_grant: str | None = None
    """Short name of an active session Tier-2 python grant (plan 235 / SHELL-02), or ``None``.
 
    The only optional field on this snapshot: a status bar that has never seen a
    grant renders byte-identically to the pre-235 one. When set (the granted
    directory's name, e.g. ``_work``), it marks that Tier-2 python runs in that
    directory currently skip the human prompt — a standing authority the operator
    must be able to see without asking for it."""
 
    session_cost_usd: float | None = None
    """Priced session spend in USD (plan 237 / CAPS-01), or ``None`` when
    nothing has been priced yet. ``None`` renders nothing at all, so a status
    bar that has never seen a priced sample is byte-identical to the pre-237 one."""
    unpriced_samples: int = 0
    """Usage samples that could NOT be priced this session. Rendered beside the
    dollar figure when non-zero, because that figure is a LOWER BOUND and a
    reader who cannot see how many samples are missing would take it for the
    whole story — the residual the hard cap cannot cover."""
    checkpoint_auto: bool = False
    """True while the tool-call checkpoint is in "don't ask again this session"
    (the operator's ``a``). Like ``session_grant`` this is a standing decision
    to stop asking, so it stays visible for as long as it holds."""
 
 
def fmt_tokens_short(n: int) -> str:
    """Format a token count compactly: ``1.2K``, ``430K``, ``1.5M``."""
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    return str(n)
 
 
def busy_hint_text(busy: bool) -> str:
    """Return the dim hint line shown adjacent to the status bar.
 
    *busy* — ``True`` while a turn is streaming or tools are dispatching
    (the wording is verified against the two-stage SIGINT handler); Ctrl+D
    is never advertised while busy since it does nothing there. ``False``
    at the idle prompt.
    """
    if busy:
        return "Ctrl+C stop current action (press twice to force-stop)"
    return "Ctrl+D exit · Ctrl+P palette"
 
 
def _display_width(text: str) -> int:
    """Approximate terminal column width — wide/fullwidth CJK counts as 2 columns."""
    return sum(2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 for ch in text)
 
 
#: What separates the activity tail from the label+duration it follows.
_TAIL_SEPARATOR = " · "
 
#: Columns below which a tail is not worth showing at all. A two- or
#: three-character stub plus an ellipsis tells the reader nothing and costs the
#: bar the same space the session cluster could have used, so the tail is
#: dropped whole instead.
_MIN_TAIL_COLUMNS = 8
 
 
def _truncate_to_width(text: str, columns: int) -> str:
    """The longest prefix of *text* that fits in *columns* terminal columns."""
    if columns <= 0:
        return ""
    kept: list[str] = []
    used = 0
    for ch in text:
        width = 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
        if used + width > columns:
            break
        kept.append(ch)
        used += width
    return "".join(kept)
 
 
def _sanitized_tail(last_line: "str | None") -> str:
    """*last_line* with control characters removed, ready for a Live region.
 
    A child's output is arbitrary bytes: a carriage return would rewind the
    cursor mid-bar and an ANSI escape would restyle everything after it, so the
    C0/C1 ranges are dropped rather than passed through to the terminal.
    """
    if not last_line:
        return ""
    cleaned = "".join(
        " " if (ch in "\t\r\n" or unicodedata.category(ch) == "Cc") else ch
        for ch in last_line
    )
    return " ".join(cleaned.split())
 
 
def busy_label_with_activity(
    label: str,
    elapsed_ms: int,
    last_line: "str | None",
    width_left: int,
) -> str:
    """Compose the busy segment for an in-flight tool call (plan 279 P4).
 
    Returns ``"<label> <duration>"``, plus ``" · <last line>"` when the bar has
    room for it. *width_left* is the number of columns the bar still has free
    once it has been rendered with the label and duration alone
    (:func:`busy_width_left` computes it), so a tail added here can never push
    a state segment off the line: it either fits in the space nothing else
    wanted, is truncated with a visible ``…`` to fit it, or is dropped whole.
 
    Pure and framework-free, so the truncate-then-drop behaviour is testable at
    any width without a terminal. The elapsed time is rendered by
    :func:`~oc_chat.core.formatting.format_duration`, the same formatter the
    turn and session timers use, so all three durations on the bar read alike.
    """
    from oc_chat.core.formatting import format_duration
 
    base = f"{label} {format_duration(elapsed_ms)}" if label else format_duration(elapsed_ms)
    tail = _sanitized_tail(last_line)
    if not tail:
        return base
    budget = width_left - _display_width(_TAIL_SEPARATOR)
    if budget < _MIN_TAIL_COLUMNS:
        return base
    if _display_width(tail) > budget:
        tail = _truncate_to_width(tail, budget - 1) + "…"
    return base + _TAIL_SEPARATOR + tail
 
 
def busy_width_left(
    state: StatusBarState, busy_label: "str | None", target_width: int,
) -> int:
    """Columns *target_width* still has free after the bar is rendered without a tail.
 
    The counterpart to :func:`busy_label_with_activity`: render once with the
    label alone to learn what the state segments actually took, then spend only
    what is left on the tail. Never negative — a bar whose always-on core already
    overflows a very narrow terminal has nothing to spare.
 
    The tail outranks every droppable segment and both clusters (TUI-06), so a
    bar with less than the tail's minimum room gives those up, one rank at a
    time and in the pinned order, until the tail has its place — or until
    nothing droppable is left, in which case the room returned stays below the
    caller's floor and :func:`busy_label_with_activity` drops the tail.
    """
    floor = _MIN_TAIL_COLUMNS + _display_width(_TAIL_SEPARATOR)
    _segments, line, dropped = _fit(state, busy_label, target_width, _DROPPABLE)
    room = max(0, target_width - _display_width(line))
    if room >= floor:
        return room
    for field in _BEHIND_THE_TAIL:
        dropped.add(field)
        _segments, line, _dropped = _fit(
            state, busy_label, target_width, _DROPPABLE, frozenset(dropped))
        room = max(0, target_width - _display_width(line))
        if room >= floor:
            return room
    return room
 
 
# Droppable segment fields, in the pinned right-to-left drop priority
# (first entry is dropped first under a narrow target width).
_DROPPABLE = ("effort", "cache_saved", "cache_pct", "tools", "iters")
 
#: The ranks that yield to the activity tail, and only to it (TUI-06): a bar
#: carrying a tail gives up the session cluster and then the turn cluster —
#: after every field in :data:`_DROPPABLE` is already gone — rather than shrink
#: the tail or lose it. A bar with no tail never drops either cluster.
_BEHIND_THE_TAIL = ("session", "turn")
 
 
def _tail_of(busy_label: "str | None") -> str:
    """The activity tail carried on *busy_label*, or ``""`` when there is none.
 
    :func:`busy_label_with_activity` is the only thing that appends
    ``_TAIL_SEPARATOR`` to a busy label, so the separator IS the tail's marker
    (no production label contains it: ``waiting for response…`` and
    ``running <tool>…`` are the two the tree builds).
    """
    if busy_label and _TAIL_SEPARATOR in busy_label:
        return busy_label.partition(_TAIL_SEPARATOR)[2]
    return ""
 
 
def _drop_order(busy_label: "str | None") -> tuple[str, ...]:
    """The ranks this bar may give up, in the pinned order.
 
    A bar with no tail stops at :data:`_DROPPABLE`, which is what keeps the
    no-activity bar byte-identical to the pre-280 one at every width.
    """
    if _tail_of(busy_label):
        return _DROPPABLE + _BEHIND_THE_TAIL
    return _DROPPABLE
 
 
def _fit(
    state: StatusBarState,
    busy_label: "str | None",
    target_width: int,
    order: tuple[str, ...],
    given_up: frozenset[str] = frozenset(),
) -> "tuple[list[str], str, set[str]]":
    """Build the segments, then drop *order* rank by rank until the line fits.
 
    *given_up* are ranks already conceded before the loop starts (the tail's
    callers hand it the prefix it has decided to yield). Returns the kept
    segments, the line, and the full set of dropped ranks.
    """
    dropped = set(given_up)
    segments = _build_segments(state, busy_label, frozenset(dropped))
    line = " │ ".join(segments)
    for field in order:
        if _display_width(line) <= target_width:
            break
        if field in dropped:
            continue
        dropped.add(field)
        segments = _build_segments(state, busy_label, frozenset(dropped))
        line = " │ ".join(segments)
    return segments, line, dropped
 
 
def _turn_cluster(state: StatusBarState, drop: frozenset[str]) -> str:
    from oc_chat.core.formatting import format_duration
 
    parts: list[str] = []
    if state.turn_tokens > 0:
        parts.append(fmt_tokens_short(state.turn_tokens))
    if state.turn_iterations > 0 and "iters" not in drop:
        n = state.turn_iterations
        parts.append(f"{n} iter{'s' if n != 1 else ''}")
    if state.turn_tools > 0 and "tools" not in drop:
        n = state.turn_tools
        parts.append(f"{n} tool{'s' if n != 1 else ''}")
    if state.turn_cache_hit_pct is not None and "cache_pct" not in drop:
        parts.append(f"{state.turn_cache_hit_pct:.0f}% cache")
    if state.turn_cache_hit_pct is not None and "cache_saved" not in drop:
        parts.append(f"saved ~{fmt_tokens_short(state.turn_cache_saved_tokens)} tok")
    if state.turn_elapsed_ms > 0:
        parts.append(format_duration(state.turn_elapsed_ms))
    return "this turn: " + " · ".join(parts) if parts else ""
 
 
def _standing_markers(state: StatusBarState) -> list[str]:
    """The session cluster's never-droppable markers, in render order.
 
    Plan 235 (SHELL-02): a live Tier-2 python grant is a standing authority.
    Plan 237 (CAPS-01): money, and the honesty that has to travel with it.
    Deliberately NOT in ``_DROPPABLE`` for the same reason: a narrow terminal
    may lose a cache percentage, never what the session is spending, the fact
    that a checkpoint has stopped asking, or the fact that a confirmation gate
    is down. Plan 280 keeps them when the session cluster itself yields to the
    activity tail — the cluster is droppable, these are not.
    """
    parts: list[str] = []
    if state.session_grant:
        parts.append(f"grant:{state.session_grant}")
    if state.session_cost_usd is not None:
        cost = f"${state.session_cost_usd:.2f}"
        if state.unpriced_samples > 0:
            cost += f" +{state.unpriced_samples} unpriced"
        parts.append(cost)
    if state.checkpoint_auto:
        parts.append("auto")
    return parts
 
 
def _session_cluster(state: StatusBarState, drop: frozenset[str]) -> str:
    from oc_chat.core.formatting import format_duration
 
    parts: list[str] = []
    if state.session_tokens > 0:
        parts.append(fmt_tokens_short(state.session_tokens))
    if state.session_tools > 0 and "tools" not in drop:
        n = state.session_tools
        parts.append(f"{n} tool{'s' if n != 1 else ''}")
    if state.session_cache_hit_pct is not None and "cache_pct" not in drop:
        parts.append(f"{state.session_cache_hit_pct:.0f}% cache")
    if state.session_cache_hit_pct is not None and "cache_saved" not in drop:
        parts.append(f"saved ~{fmt_tokens_short(state.session_cache_saved_tokens)} tok")
    # The session timer is the always-visible realtime element (D3) — it
    # renders even at zero tokens, unlike the turn cluster above.
    parts.append(format_duration(state.session_elapsed_ms))
    parts.extend(_standing_markers(state))
    return "session: " + " · ".join(parts)
 
 
def _build_segments(
    state: StatusBarState, busy_label: str | None, drop: frozenset[str],
) -> list[str]:
    """The bar's segments, in fixed order, with every dropped rank left out.
 
    ``drop`` holds droppable field names (:data:`_DROPPABLE`) and, for a bar
    carrying an activity tail, the cluster ranks (plan 280). The model, the
    session name, the busy label and the standing markers are never in it.
    """
    segments: list[str] = [f"model: {state.model}", f"name: {state.session_name}"]
    if busy_label:
        segments.append(busy_label)
    if "turn" not in drop:
        turn_seg = _turn_cluster(state, drop)
        if turn_seg:
            segments.append(turn_seg)
    if "session" in drop:
        # The cluster yields to the tail; its standing markers do not.
        markers = _standing_markers(state)
        if markers:
            segments.append("session: " + " · ".join(markers))
    else:
        segments.append(_session_cluster(state, drop))
    if "effort" not in drop:
        segments.append(f"effort: {state.effort}")
    return segments
 
 
def render_status_line(
    state: StatusBarState,
    busy_label: str | None,
    target_width: int,
) -> tuple[list[str], str]:
    """Render the status bar, dropping segments right-to-left to fit *target_width*.
 
    Returns ``(kept_segments, rendered_line)``. Never wraps: once every
    droppable field (:data:`_DROPPABLE`) is gone — and, for a bar carrying an
    activity tail, both clusters as well — the line is returned as-is even if
    the always-on core (model/name/busy/tokens/durations) still overflows
    *target_width*.
 
    The drop order is the pinned one: effort, cache-saved, cache%, tools,
    iters, and — only for the bar that carries plan 279's activity tail — the
    session cluster and then the turn cluster. The tail outranks all of them
    (TUI-06 ruling, 2026-09-16, amended 2026-09-17).
    """
    segments, line, _dropped = _fit(
        state, busy_label, target_width, _drop_order(busy_label))
    return segments, line