6Plain-ANSI console presenter for OC Chat.
10:class:`ConsolePresenter` implements the
11:class:`~oc_ui.presenter_base.Presenter` contract using simple
12``print`` calls and ANSI escape codes from
13:mod:`oc_chat.core.config`. It is the default presenter and the
14fallback used by CI, headless runs, and any environment without
15optional ``rich`` / ``textual`` dependencies installed.
17The Phase 0 quick wins live here:
19- Horizontal rule + turn number + timestamp at :meth:`start_turn`.
20- :meth:`show_waiting` indicator that is cleared by the first
21 reasoning / content chunk.
22- Compact :meth:`show_turn_summary` (one dim line) — full
23 per-iteration breakdown only when ``verbose`` is set.
24- ``NO_COLOR`` / ``FORCE_COLOR`` honoured via the STYLE_* constants
25 in :mod:`oc_chat.core.config`.
29- :class:`ConsolePresenter` — concrete :class:`Presenter`
30 implementation backed by ``print``.
31- :func:`_color_enabled` — env-var-based ANSI gating.
32- All ``show_*`` event hooks for reasoning / content / tool / usage
33 / turn-summary / compressed / welcome.
34- :meth:`turn` returns a no-op ``contextmanager`` (no live region).
46- Output goes to ``stdout`` exclusively; diagnostics and warnings
47 live on ``stderr`` via ``oc_diagnostics``.
48- ``NO_COLOR`` set (any value) disables every ANSI sequence.
49- ``FORCE_COLOR`` overrides ``NO_COLOR`` and forces ANSI on.
50- ``show_*`` hooks are append-only — the presenter never rewinds.
51- Safe to use under CI / non-TTY: ``shutil.get_terminal_size`` is
52 used with a fallback when the size is undetermined.
56- oc_diagnostics (structured debug logging via _dbg)
57- oc_ui.presenter_base (Presenter ABC contract)
58- oc_ui.terminal.activity (plan_progress_from_result for the plan progress line under planner results)
59- oc_ui.terminal.diffview (mutation argument pairing and plain unified diff rendering for write_file and edit_file)
60- oc_ui.terminal.turn_layout (turn indent, rule labels, answer box markers and line indenters)
61- oc_chat.core.commands (lazy; _fmt_tokens_display for the session list table)
62- oc_chat.core.formatting (lazy; format_duration for the turn summary)
63- oc_chat.core.iteration_stat (lazy; iteration_stat_human_line for the per-iteration stats line)
64- oc_chat.core.session (lazy; ChatSession type hint for show_welcome)
65- oc_ui.terminal.statusline (lazy; StatusBarState type hint for bind_status)
67from __future__
import annotations
71from contextlib
import contextmanager
72from datetime
import datetime
73from typing
import TYPE_CHECKING, Callable, Iterator
75from oc_diagnostics
import _dbg
83 summarize_mutation_call,
107 """Return True iff ANSI styles should be emitted."""
109 if os.getenv(
"FORCE_COLOR"):
111 if os.getenv(
"NO_COLOR")
is not None:
116STYLE_RESET =
"\033[0m" if _COLOR
else ""
117STYLE_DIM =
"\033[2m" if _COLOR
else ""
118STYLE_ITALIC =
"\033[3m" if _COLOR
else ""
119STYLE_BOLD =
"\033[1m" if _COLOR
else ""
120STYLE_CYAN =
"\033[36m" if _COLOR
else ""
121STYLE_YELLOW =
"\033[33m" if _COLOR
else ""
122STYLE_GREEN =
"\033[32m" if _COLOR
else ""
123STYLE_RED =
"\033[31m" if _COLOR
else ""
124REASONING_EFFORT_LEVELS = [
"low",
"medium",
"high"]
128 """Terminal UI presenter using plain ANSI escape codes.
130 Always satisfies the :class:`Presenter` contract with no optional
131 dependencies. Suitable for CI, headless servers, and any environment
132 where Rich/Textual aren't appropriate.
134 *verbose* (default ``False``) controls whether per-iteration
135 :meth:`show_usage` is emitted. When ``False``, only the compact turn
136 summary is printed at the end of each turn.
160 def bind_status(self, provider: Callable[[],
"StatusBarState | None"]) ->
None:
161 """Store the status-snapshot provider (plan 147 D1 / plan 218 T2).
163 The plain console has no live chrome to refresh, so this was a
164 no-op before plan 218. It now keeps the provider because the
165 turn-end rule's running total is read from that snapshot — the
166 plain presenter must be able to show the same two numbers the
167 rich one does, not a degraded rule missing them.
173 """Terminal width, with the module's long-standing 80x24 fallback."""
174 return max(20, shutil.get_terminal_size((80, 24)).columns)
177 """Wall-clock ``HH:MM:SS`` for the rule labels — the one clock read."""
178 return datetime.now().strftime(
"%H:%M:%S")
181 """The turn indent while a turn is open; nothing outside one."""
182 return TURN_INDENT
if self.
_in_turn else ""
185 trailer: str =
"") ->
None:
186 """Print the two-line plain separator: full rule, then the label line.
188 Structure and both style tokens are exactly the pre-218 ones; only
189 the label's CONTENT grows (plan 218 P1 element 4) and the trailer
190 is new. The label line keeps its ``·`` fill.
193 print(f
"\n{STYLE_DIM}{'─' * width}{STYLE_RESET}")
195 f
"{rule_line(label, width, trailer, PLAIN_RULE_FILL)}"
199 """The ``↻ total · turn / session`` trailer, or ``""`` when unknown.
201 Same contract as the rich presenter's: read off the bound status
202 snapshot, add this turn to the session figure (``finish_turn``
203 folds it in only after the summary renders), and render NOTHING
204 rather than a zero when no provider is bound.
206 func =
"_running_total"
211 except Exception
as exc:
214 _dbg(
"OC_UI.TERMINAL", func,
215 f
"EXTERNAL_ERROR: {type(exc).__name__}: {exc}", 2, override=
True)
219 return running_total_text(state.turn_tokens,
220 state.session_tokens + state.turn_tokens)
223 """Emit the degraded box's closing marker (plan 218 P2/W5).
225 Called from every hook that can follow the answer, so the box is
226 closed before anything that logically comes after it prints — and
227 it is closed on the error path too, carrying the failure text
228 inside. Idempotent: a second call with no open box does nothing.
236 print(f
"{TURN_INDENT}{STYLE_RED}{error}{STYLE_RESET}")
237 marker = plain_box_bottom(self.
_width() - len(TURN_INDENT), subtitle,
239 print(f
"{TURN_INDENT}{STYLE_DIM}{marker}{STYLE_RESET}")
244 """Draw the ``start @ <time>`` rule that opens the session (P1.1)."""
248 def turn(self) -> Iterator[None]:
249 """Context manager wrapping the streamed portion of a turn.
251 The plain console has no Live region — every chunk prints as it
252 arrives — but plan 218 P2 still needs the answer box CLOSED on
253 every exit path, so the teardown does that here rather than
254 relying on a later hook that an exception may never reach.
258 except Exception
as exc:
270 """Reset per-turn flags and print the turn separator."""
272 _dbg(
"OC_UI.TERMINAL", func, f
"turn={self._turn_number + 1}", 4)
289 """Finalise the current turn (trailing newline)."""
299 """Print a transient "Thinking…" indicator before the first chunk."""
301 print(f
"{STYLE_DIM}{STYLE_ITALIC}⏳ Thinking…{STYLE_RESET}",
306 """Erase the in-line waiting indicator with carriage-return + spaces."""
308 print(
"\r" +
" " * 20 +
"\r", end=
"", flush=
True)
312 """Print chain-of-thought (reasoning) text as it streams.
314 Plan 218: line-buffered so the turn indent lands at line starts —
315 a chunk that stops mid-sentence waits for its newline rather than
316 splicing three spaces into the middle of a word.
321 print(f
"{self._indent()}{STYLE_CYAN}{REASONING_TITLE}{STYLE_RESET}")
324 print(f
"{STYLE_DIM}{STYLE_ITALIC}{line}{STYLE_RESET}", flush=
True)
327 """Stream the answer inside the degraded (glyph-free) answer box.
329 Plan 218 P2/P4: the opening marker prints with the first chunk,
330 each completed line prints indented as it lands, and the closing
331 marker — carrying the iteration metadata — prints at turn end via
332 :meth:`_close_answer_box`. No box-drawing glyph is used, so a dumb
333 terminal renders the same information faithfully (W5).
339 print(f
"{STYLE_DIM}{STYLE_ITALIC}{line}{STYLE_RESET}")
342 marker = plain_box_top(self.
_width() - len(TURN_INDENT), indent=
"")
343 print(f
"{TURN_INDENT}{STYLE_BOLD}{marker}{STYLE_RESET}")
347 print(line, flush=
True)
352 """Display a tool-call invocation with pretty-printed arguments."""
353 func =
"show_tool_call"
358 args_pretty = json.dumps(json.loads(arguments
or "{}"), ensure_ascii=
False)
359 except (json.JSONDecodeError, ValueError)
as e:
361 _dbg(
"OC_UI.TERMINAL", func, f
"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
362 args_pretty = arguments
367 args_pretty = summarize_mutation_call(name, arguments)
or args_pretty
368 print(f
"\n{self._indent()}{STYLE_YELLOW}"
369 f
"[Tool call → {name}({args_pretty})]{STYLE_RESET}")
372 """No-op for the plain console; rich/textual presenters spin here."""
377 """Display the result of a tool call (truncated if > 400 chars).
379 Plan 188 T1: an applied ``write_file`` / ``edit_file`` renders as a
380 plain unified diff (ANSI-coloured when colour is enabled) instead of
381 the truncated JSON line. Everything else — every other tool, and any
382 mutation result that is an error, a gate block, or an unpaired call —
383 keeps the legacy rendering exactly.
385 If *result* is a ``requires_confirmation`` dict (from Tier-2
386 shell commands), it is displayed as a warning prompt.
390 if isinstance(result, dict)
and result.get(
"requires_confirmation"):
391 msg = result.get(
"message", str(result))
392 print(f
"{indent}{STYLE_YELLOW}"
393 f
"[⏳ Confirmation required — {msg}]{STYLE_RESET}")
396 diff = build_mutation_diff(name, self.
_mutation_args.take(name), result)
398 for line
in render_diff_plain(diff, color=_COLOR):
399 print(f
"{indent}{line}")
402 result_str = json.dumps(result, ensure_ascii=
False)
403 if len(result_str) > 400:
404 result_str = result_str[:400] +
"…"
405 print(f
"{indent}{STYLE_GREEN}[{name} ← {result_str}]{STYLE_RESET}")
409 progress = plan_progress_from_result(name, result)
419 "warn": STYLE_YELLOW,
423 def show_activity(self, label: str, detail: str =
"", *, tone: str =
"info") ->
None:
424 """Print a compact one-line state update for a non-streaming lane."""
427 body = f
" {detail}" if detail
else ""
428 print(f
"{self._indent()}{style}[{label}]{STYLE_RESET}"
429 f
"{STYLE_DIM}{body}{STYLE_RESET}")
435 prompt_cached=
None, reasoning=
None, content=
None) -> str:
436 """Build a one-line usage summary string."""
437 parts = [f
"{label}: prompt={prompt}"]
438 if prompt_cached
is not None:
439 parts.append(f
"(cache hits: {prompt_cached})")
440 parts.append(f
" completion={completion}")
441 if reasoning
is not None and content
is not None:
442 parts.append(f
"(reasoning:{reasoning} content:{content})")
443 parts.append(f
" total={total}")
444 return "".join(parts)
447 """Per-iteration token usage — compact one-line format.
449 Plan 134 C1: render only in ``--verbose`` mode (restores the contract
450 this class's own docstring already promised). Terminal rendering only;
451 structured emission/logging is unaffected.
456 prompt = usage.get(
"prompt_tokens", 0)
457 completion = usage.get(
"completion_tokens", 0)
458 total = usage.get(
"total_tokens", 0)
460 if usage.get(
"prompt_tokens_details"):
461 pd = usage[
"prompt_tokens_details"]
462 if isinstance(pd, dict):
463 prompt_cached = pd.get(
"cached_tokens")
468 cached = f
" · cache:{cached}"
471 f
"{self._indent()}{STYLE_DIM} prompt:{self._fmt_compact_tokens(prompt)}"
472 f
" · completion:{self._fmt_compact_tokens(completion)}"
473 f
" · total:{self._fmt_compact_tokens(total)}"
474 f
"{cached}{STYLE_RESET}"
479 """Render the per-iteration stats — in the answer box's closing marker.
481 Plan 218 P1 element 3, degraded per P4: when this iteration
482 produced an answer, the stat line IS the box's closing marker.
483 An iteration with no answer (a pure tool-call round) has no box to
484 close, so the line renders as before, just carrying the indent.
487 line = iteration_stat_human_line(stat)
491 print(f
"{self._indent()}{STYLE_DIM}{line}{STYLE_RESET}")
495 """Render a token count compactly (``12.4k`` / ``1.6M``)."""
497 return f
"{n / 1_000_000:.2f}M"
499 return f
"{n / 1_000:.1f}k"
503 iterations: int, tool_calls: int,
504 prompt_cached: int |
None =
None, *,
505 cache_hit_pct: float |
None =
None,
506 cache_saved_tokens: int |
None =
None,
507 performance_stats: object =
None,
508 model: str |
None =
None,
509 context_used_pct: float |
None =
None,
510 duration_ms: int |
None =
None,
511 turn_cost: object =
None) ->
None:
512 """Print the end-of-turn rule, with the turn stats absorbed into it.
514 Plan 218 P1 element 4: the stats string that used to occupy its own
515 line under the rule is now part of the rule's label, and the
516 running total rides in the right-hand fill run. Wording and both
517 style tokens are unchanged \u2014 only the position is (P3).
519 Plan 188 T4: when *turn_cost* is present the compact line gains a
520 ``~$`` estimate; when it is ``None`` no cost is rendered at all (an
521 unpriceable turn shows nothing rather than a wrong figure). The
522 price table's capture date is appended in verbose mode, where the
523 provenance is worth the width.
529 if prompt_cached
is not None and prompt_cached > 0:
530 cached_str = f
" \u00b7 cache:{self._fmt_compact_tokens(prompt_cached)}"
533 if cache_hit_pct
is not None:
534 cache_hit_str = f
" \u00b7 cache-hit:{cache_hit_pct:.0f}%"
536 if context_used_pct
is not None:
537 ctx_str = f
" \u00b7 ctx:{context_used_pct:.0f}%"
539 ctx_str += f
" of {model}"
541 if duration_ms
is not None:
543 duration_str = f
" \u00b7 {format_duration(duration_ms)}"
546 f
" this turn: prompt:{self._fmt_compact_tokens(prompt)}"
547 f
" \u00b7 completion:{self._fmt_compact_tokens(completion)}"
548 f
" \u00b7 total:{self._fmt_compact_tokens(total)}"
549 f
"{cached_str}{cache_hit_str}{ctx_str}{duration_str}{cost_str}"
552 absorbed = turn_end_rule_label(self.
_turn_number, ts, stats)
553 if rule_fits(absorbed, self.
_width()):
559 print(f
"{STYLE_DIM}{stats}{STYLE_RESET}")
563 """Format the plan 188 T4 cost segment, or ``""`` when unpriceable.
565 Shared by every terminal presenter (Rich subclasses reuse it) so the
566 honesty markers cannot drift apart: the leading ``~`` says modelled,
567 not billed, and verbose adds the rate table's capture date.
569 total = getattr(turn_cost,
"total_usd",
None)
572 out = f
" \u00b7 ~${total:.4f}"
573 as_of = getattr(turn_cost,
"as_of",
"")
574 if verbose
and as_of:
575 out += f
" (prices as of {as_of})"
579 """Display the welcome banner with model info and available commands."""
580 print(f
"\nChat with DeepSeek — model: {session.model}")
581 print(f
"Reasoning effort: {session.reasoning_effort} "
582 f
"(options: {', '.join(REASONING_EFFORT_LEVELS)})")
586 print(
"Type /help for full command details and session subcommands.")
590 """Print a table of available models with capability indicators."""
591 print(
"\nAvailable models:")
592 for name, caps
in models.items():
593 thinking =
"✓" if caps.get(
"supports_thinking")
else "✗"
594 reasoning =
"✓" if caps.get(
"supports_reasoning")
else "✗"
595 tools =
"✓" if caps.get(
"supports_tools",
True)
else "✗"
596 print(f
" {name:<25} thinking:{thinking} reasoning:{reasoning} tools:{tools}")
600 """Print the saved-sessions table for ``/session list``."""
602 print(
"(no saved sessions)")
607 f
"{'Name':<30} {'Model':<22} {'Msgs':>4} {'Tools':>5} "
608 f
"{'Tokens':>10} Last Modified"
613 marker =
"*" if s[
"name"] == current_name
else " "
615 tokens_str = _fmt_tokens_display(stats)
617 f
"{marker}{s['name']:<29} {s['model']:<22} "
618 f
"{stats.message_count:>4} {stats.tool_call_count:>5} "
619 f
"{tokens_str:>10} {s['last_modified']}"
623 def show_help(self, command_help: dict, tools: list[str] |
None =
None) ->
None:
624 """Print the ``/help`` listing of slash commands and registered tools."""
626 for line
in command_help.values():
629 print(
"\nAgent tools (callable by the model via tool-use):")
635 """Notify the user that context compression just occurred."""
637 print(f
"\n{self._indent()}{STYLE_DIM}"
638 f
"[Context compressed — older messages summarised]{STYLE_RESET}")
641 """Display an error message in red."""
643 print(f
"\n{self._indent()}{STYLE_RED}Error: {msg}{STYLE_RESET}")
646 """Display an informational message."""
648 print(f
"\n{self._indent()}{msg}")
653 """No-op on the plain console — returns a handle with ``status=None``."""
657 """No-op on the plain console."""
None show_activity(self, str label, str detail="", *, str tone="info")
None _close_answer_box(self, str subtitle="", str error="")
None show_compressed(self)
str _fmt_compact_tokens(int n)
None __init__(self, bool verbose=False)
None show_error(self, str msg)
None show_help(self, dict command_help, list[str]|None tools=None)
None show_reasoning(self, str text)
None show_activity(self, str label, str detail="", *, str tone="info")
Iterator[None] turn(self)
None show_turn_summary(self, int prompt, int completion, int total, int iterations, int tool_calls, int|None prompt_cached=None, *, float|None cache_hit_pct=None, int|None cache_saved_tokens=None, object performance_stats=None, str|None model=None, float|None context_used_pct=None, int|None duration_ms=None, object turn_cost=None)
SpinnerHandle spinner_start(self, str label)
None show_available_models(self, dict models)
None show_tool_call(self, str name, str arguments)
None show_info(self, str msg)
str _fmt_usage_line(str label, int prompt, int completion, int total, prompt_cached=None, reasoning=None, content=None)
None show_content(self, str text)
None show_tool_progress(self, str name)
Callable[[],]|None _status_provider
None show_session_start(self)
str _fmt_turn_cost(object turn_cost, *, bool verbose=False)
None spinner_stop(self, SpinnerHandle handle)
None _print_rule(self, "str | list[tuple[str, str]]" label, str trailer="")
None show_iteration_stat(self, "IterationStat" stat)
None show_usage(self, dict usage)
None show_welcome(self, ChatSession session)
None _clear_waiting(self)
None bind_status(self, Callable[[], "StatusBarState | None"] provider)
None show_session_list(self, list sessions, str|None current_name=None)
None show_tool_result(self, str name, object result)