turn_layout.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/turn_layout.py
 
"""
Turn-scoped layout primitives shared by every terminal presenter (plan 218).
 
Purpose
-------
Plan 218 makes each chat turn read as ONE bounded, indented block:
a ``start @`` rule opens the session, the assistant's answer sits in a
box whose top border carries ``Assistant Answer:`` and whose bottom
border carries that iteration's metadata, the turn-end rule absorbs the
turn stats plus the running total, and every line between the turn
rules is indented by :data:`TURN_INDENT`.
 
None of that geometry is presenter-specific, so it lives here as pure
string logic: :class:`~oc_ui.terminal.rich_presenter.RichPresenter`
maps the composed segments onto ``rich`` renderables and
:class:`~oc_ui.terminal.presenter.ConsolePresenter` maps the same
segments onto ASCII marker lines. One algorithm, two skins — the plain
presenter can never drift out of structural agreement with the rich one.
 
Framework-free by the same reasoning as
:mod:`~oc_ui.terminal.statusline`: no ``rich`` / ``prompt_toolkit``
import, so both presenters and the prompt driver can share it.
 
Responsibilities
----------------
- :data:`TURN_INDENT` / :func:`indent_text` — the three-space turn indent.
- :class:`LineIndenter` — line-buffered indenting for streamed chunks,
  so a presenter that prints as it streams can still prefix whole lines.
- :func:`compose_rule` — the two-cluster rule layout (centred label plus
  a trailer centred in the right-hand fill run) used by the turn-end rule.
- :func:`start_rule_label` / :func:`turn_rule_label` /
  :func:`turn_end_rule_label` / :func:`running_total_text` — the rule
  label strings.
- :func:`plain_box_top` / :func:`plain_box_bottom` — the glyph-free
  degraded box markers for the plain presenter (P4).
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Pure functions: no I/O, no clock reads, no framework imports. Callers
  pass the timestamp/width they want rendered.
- :func:`compose_rule` always returns segments whose joined text is
  EXACTLY *width* columns when the label fits, and never wraps: a label
  wider than the available run is emitted un-truncated with the fill
  runs collapsed to zero rather than being cut (the same "never wrap,
  never lie" discipline :func:`~oc_ui.terminal.statusline.render_status_line`
  applies to the status bar).
- The trailer is DROPPED, not truncated, when it cannot fit beside the
  label — a partial running total would be a wrong number on screen.
- :class:`LineIndenter` is append-only and never re-emits a line it has
  already returned; ``flush`` returns whatever tail is still buffered.
- No colour/style decisions here — segments are TAGGED (``"fill"`` /
  ``"label"`` / ``"trailer"``) and each presenter maps tags onto its own
  existing style tokens, so plan 218's byte-preserved-colour pin (P3)
  stays enforceable at the presenter boundary.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_ui.terminal.statusline (lazy; fmt_tokens_short as the default running total formatter)
"""
from __future__ import annotations
 
from typing import TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:
    from typing import Callable
 
_dbg("OC_UI.TERMINAL", "module", "loaded", 4)
 
 
#: The turn-scoped indent (plan 218 P1 element 5). Every line rendered
#: between the turn rules carries it; the rules themselves and the status
#: bar do not.
TURN_INDENT = "   "
 
#: Title cut into the answer box's top border (plan 218 P1 element 2).
#: Identical string to the pre-218 standalone header line, so a reader
#: (and every existing substring assertion) sees the same words.
ANSWER_BOX_TITLE = "Assistant Answer:"
 
#: Reasoning header, kept verbatim from the pre-218 layout.
REASONING_TITLE = "Assistant CoT:"
 
#: Fill character for the rich rules (unchanged from pre-218).
RULE_FILL = "─"
 
#: Fill character for the plain presenter's label line (unchanged).
PLAIN_RULE_FILL = "·"
 
#: Marker character for the plain presenter's degraded answer box (P4/W5) —
#: deliberately ASCII so a dumb terminal renders it faithfully.
PLAIN_BOX_FILL = "-"
 
 
def indent_text(text: str, indent: str = TURN_INDENT) -> str:
    """Prefix every line of *text* with *indent*.
 
    A trailing newline is preserved without gaining a phantom indented
    empty line, so ``indent_text("a\\n")`` is ``"   a\\n"`` — the caller can
    hand the result straight to ``print(..., end="")``.
    """
    if not text:
        return text
    lines = text.split("\n")
    tail = ""
    if lines and lines[-1] == "":
        lines = lines[:-1]
        tail = "\n"
    return "\n".join(indent + line for line in lines) + tail
 
 
class LineIndenter:
    """Line-buffered indenter for streamed chunks (plan 218 P2).
 
    A presenter that prints each streaming chunk as it arrives cannot
    simply prefix the chunk — chunks split mid-line, so the indent would
    land in the middle of a sentence. Feed chunks in; get back only the
    lines that have COMPLETED, already indented and newline-terminated.
    Whatever is still partial stays buffered until :meth:`flush`.
 
    Parameters
    ----------
    indent:
        The prefix applied to each completed line.
    """
 
    def __init__(self, indent: str = TURN_INDENT) -> None:
        self._indent = indent
        self._buf = ""
 
    def feed(self, text: str) -> list[str]:
        """Consume *text*; return the indented lines it completed."""
        self._buf += text
        out: list[str] = []
        while "\n" in self._buf:
            line, _, self._buf = self._buf.partition("\n")
            out.append(self._indent + line)
        return out
 
    def flush(self) -> list[str]:
        """Return the buffered tail (if any) as one indented line and reset."""
        if not self._buf:
            return []
        line = self._indent + self._buf
        self._buf = ""
        return [line]
 
    @property
    def pending(self) -> bool:
        """``True`` while a partial line is still buffered."""
        return bool(self._buf)
 
 
# ── Rule labels ──────────────────────────────────────────────────────────────
 
 
def start_rule_label(ts: str) -> str:
    """Label for the session-opening rule: ``" start @ 02:13:32 "``."""
    return f" start @ {ts} "
 
 
def turn_rule_label(turn_number: int, ts: str) -> str:
    """Label for the turn-opening rule: ``" turn 1 @ 02:13:37 "``."""
    return f" turn {turn_number} @ {ts} "
 
 
def turn_end_rule_label(turn_number: int, ts: str, stats: str = "") -> str:
    """Label for the turn-closing rule, with the turn stats absorbed.
 
    Plan 218 P1 element 4: the stats that used to occupy their own line
    under the rule move INTO the rule text. *stats* is the presenter's
    existing stats string, passed through unchanged (P3 — positions
    change, styles and wording do not); an empty *stats* yields the
    pre-218 label.
    """
    body = f" turn {turn_number} end @ {ts}"
    if stats:
        body += f" - {stats.strip()}"
    return body + " "
 
 
def turn_end_rule_segments(turn_number: int, ts: str,
                           stats: str = "") -> list[tuple[str, str]]:
    """:func:`turn_end_rule_label` split so the stats keep their own style.
 
    Returns ``[(rule-label, "label"), (stats, "stats")]``. The split
    exists for exactly one reason (plan 218 P3): the rule title was dim
    cyan and the stats line was dim, and moving the stats INTO the rule
    must not repaint them in the rule's colour. Presenters map the two
    tags onto the two style tokens they already used.
    """
    head = f" turn {turn_number} end @ {ts}"
    if not stats:
        return [(head + " ", "label")]
    return [(head + " - ", "label"), (stats.strip() + " ", "stats")]
 
 
def running_total_text(turn_tokens: int, session_tokens: int,
                       fmt: "Callable[[int], str] | None" = None) -> str:
    """The running-total trailer: ``"↻ total · 7.2K / 7.2K"``.
 
    Reuses the ``↻`` marker and the compact token formatting the
    between-turns rprompt already shows
    (:meth:`~oc_ui.terminal.input.TurnStats.rprompt_text`), so the same
    two numbers keep the same shape wherever they appear. *fmt* defaults
    to :func:`~oc_ui.terminal.statusline.fmt_tokens_short`; it is a
    parameter only so this module keeps its no-sibling-import purity in
    tests.
    """
    if fmt is None:
        from oc_ui.terminal.statusline import fmt_tokens_short
        fmt = fmt_tokens_short
    return f"↻ total · {fmt(turn_tokens)} / {fmt(session_tokens)}"
 
 
# ── Rule composition ─────────────────────────────────────────────────────────
 
 
def compose_rule(label: str | list[tuple[str, str]], width: int,
                 trailer: str = "", fill: str = RULE_FILL) -> list[tuple[str, str]]:
    """Compose a rule line as tagged segments.
 
    Geometry (plan 218 P1 element 4, read off the operator's mock): the
    *label* block is centred over the full *width* exactly as the pre-218
    ``rich.rule.Rule`` centred its title, and the *trailer* block is then
    centred within the right-hand fill run that remains. Each block is
    padded with two spaces either side, matching the two-space gutter
    ``Rule`` already produced for a label that itself carried one space.
 
    *label* is either a plain string or a list of ``(text, tag)`` pairs
    (see :func:`turn_end_rule_segments`) when parts of the label must
    keep distinct styles.
 
    Returns a list of ``(text, tag)`` pairs where *tag* is ``"fill"``,
    ``"trailer"``, or whichever tag the caller supplied for a label part
    (``"label"`` for a plain-string label). Callers map tags onto their
    own style tokens; this function makes no colour decision.
    """
    label_parts: list[tuple[str, str]]
    if isinstance(label, str):
        label_parts = [(label, "label")] if label else []
    else:
        label_parts = [part for part in label if part[0]]
    if label_parts:
        # One space either side of the whole label block, on the outermost
        # parts, so a multi-part label still reads as one centred title.
        label_parts = list(label_parts)
        label_parts[0] = (" " + label_parts[0][0], label_parts[0][1])
        label_parts[-1] = (label_parts[-1][0] + " ", label_parts[-1][1])
    label_len = sum(len(text) for text, _tag in label_parts)
    trailer_block = f"  {trailer}  " if trailer else ""
 
    if label_len >= width:
        # Never wrap, never truncate: emit the label alone and let the
        # terminal decide. The trailer is dropped rather than half-shown.
        return label_parts
 
    left = (width - label_len) // 2
    right_run = width - left - label_len
 
    segments: list[tuple[str, str]] = []
    if left:
        segments.append((fill * left, "fill"))
    segments.extend(label_parts)
 
    # The trailer needs its block plus at least one fill cell either side,
    # otherwise it would collide with the label or the line end.
    if trailer_block and right_run >= len(trailer_block) + 2:
        gap = (right_run - len(trailer_block)) // 2
        tail = right_run - gap - len(trailer_block)
        segments.append((fill * gap, "fill"))
        segments.append((trailer_block, "trailer"))
        if tail:
            segments.append((fill * tail, "fill"))
    elif right_run:
        segments.append((fill * right_run, "fill"))
    return segments
 
 
def rule_line(label: str | list[tuple[str, str]], width: int, trailer: str = "",
              fill: str = RULE_FILL) -> str:
    """Return :func:`compose_rule`'s segments joined into one plain string."""
    return "".join(text for text, _tag in compose_rule(label, width, trailer, fill))
 
 
#: Minimum fill cells either side of a label before it counts as "in a rule"
#: rather than "a line that happens to be a label".
_MIN_RULE_FILL = 4
 
 
def rule_fits(label: str | list[tuple[str, str]], width: int) -> bool:
    """Can *label* be absorbed into a *width*-column rule and still read as one?
 
    Plan 218's turn-end rule swallows the whole turn-stats string, which on
    a narrow terminal is longer than the terminal itself. Callers use this
    to choose between the absorbed one-line rule and the pre-218 two-line
    form (rule, then stats beneath): degrading to the OLD layout keeps
    every figure legible, whereas letting an oversized label wrap would
    split a rule across lines and cut a ``ctx:NN% of <model>`` gauge in
    half. Never truncate, never wrap — degrade.
    """
    if isinstance(label, str):
        label_len = len(label)
    else:
        label_len = sum(len(text) for text, _tag in label)
    return label_len + 2 + _MIN_RULE_FILL <= width
 
 
# ── Degraded (glyph-free) answer box — plain presenter, plan 218 P4/W5 ───────
 
 
def plain_box_top(width: int, title: str = ANSWER_BOX_TITLE,
                  indent: str = TURN_INDENT) -> str:
    """ASCII opening marker for the degraded answer box.
 
    Renders ``"   --- Assistant Answer: -----"`` — the same information
    the rich box carries in its top border, using only characters a dumb
    terminal is guaranteed to draw (W5).
    """
    inner = max(0, width - len(indent))
    head = f"{PLAIN_BOX_FILL * 3} {title} "
    pad = max(0, inner - len(head))
    return indent + head + PLAIN_BOX_FILL * pad
 
 
def plain_box_bottom(width: int, subtitle: str = "",
                     indent: str = TURN_INDENT) -> str:
    """ASCII closing marker for the degraded answer box.
 
    Carries the iteration metadata in the same place the rich box's
    bottom border does. Always emitted — the plain presenter closes its
    box on every path too (W2).
    """
    inner = max(0, width - len(indent))
    head = PLAIN_BOX_FILL * 3
    if subtitle:
        head += f" {subtitle} "
    else:
        head += " "
    pad = max(0, inner - len(head))
    return indent + head + PLAIN_BOX_FILL * pad