tui_spike.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/scripts/tui_spike.py
 
"""
Standalone prompt_toolkit TUI feasibility spike (Phase TUI-1).
 
Purpose
-------
Prove — before committing to a full TUI rewrite of OC Chat — that the
risky parts of a prompt_toolkit-based three-pane terminal application
actually work on Windows Terminal + PowerShell, the project's
declared host. The risky parts (Codex review):
 
1. Rich-ANSI rendering for Markdown headings, fenced code with syntax
   highlighting, GitHub-style tables, and wide unicode (CJK / combining
   marks) inside a prompt_toolkit pane.
2. Independent pane scrolling via PgUp/PgDn/Home/End on the
   conversation pane only — header and input stay pinned.
3. SIGINT-during-stream behavior. A worker thread appends fake
   streaming chunks; Ctrl+C must cancel the stream and leave the app
   alive (no terminal corruption, no hang).
4. Resize tolerance. SIGWINCH on POSIX; prompt_toolkit's internal
   terminal-size polling on Windows.
5. Alt-screen vs non-alt-screen exit semantics. We use the default
   (non-alt-screen + full_screen=True style with explicit redraw) and
   verify the terminal restores cleanly after Ctrl+Q.
 
The spike is **deliberately not wired into the OC Chat runtime**. It
is a standalone runnable: ``python -m oc_chat.scripts.tui_spike``. The
companion ``README_TUI_SPIKE.md`` documents how to run it, the exit
criteria that gate Round 2, and a results table for manual smoke.
 
Responsibilities
----------------
- :func:`main` — bootstrap the prompt_toolkit Application and run it.
- :func:`build_application` — assemble the three-pane layout, key
  bindings, and styled controls. Returns a ready-to-run
  :class:`Application`.
- :class:`_StreamSimulator` — background thread that appends fake
  Rich-ANSI streaming chunks to the conversation buffer at a bounded
  rate. Stop event for SIGINT cancellation.
- :data:`SAMPLE_RICH_BLOCKS` — representative Rich-ANSI samples
  (Markdown heading, fenced code, table, long paragraph, wide
  unicode) used to populate the conversation pane on launch.
 
Diagnostics
-----------
Domain: OC_CHAT.TUI_SPIKE
Levels:
    L2 — application lifecycle (start / exit)
    L3 — stream simulation events (start, cancel, chunk rate cap)
    L4 — per-chunk trace
 
Contracts
---------
- The spike must not import any OC Chat runtime modules other than
  :mod:`oc_diagnostics`. It does not touch sessions, providers, the
  agent, or the registry. Deletable later without touching anything
  else.
- Rendering uses ``FormattedTextControl(text=ANSI(...))`` so Rich's
  ANSI output is passed through verbatim — prompt_toolkit does not
  re-parse the escape codes; it just hands them to the terminal.
- Worker-thread → UI handoff goes via
  :meth:`Application.create_background_task` /
  ``app.loop.call_soon_threadsafe``. Direct buffer mutation from the
  worker thread would race with the event loop.
- The simulated chunk rate is capped at 80 chunks/sec, matching the
  observed max from real DeepSeek-v4-pro generations. Above this,
  prompt_toolkit's redraw cadence struggles and the test isn't
  representative.
- SIGINT during streaming sets a ``threading.Event`` that the worker
  polls between chunks; the app stays alive. Ctrl+Q exits via the
  normal key-binding path.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- prompt_toolkit (lazy; 3.x Application, layout and key bindings, assumed installed for any developer running this spike)
- rich (lazy; optional ANSI rendering of the sample blocks, falls back to a plain-text sample when missing)
"""
from __future__ import annotations
 
import threading
import time
from typing import Optional
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.TUI_SPIKE", "module", "loaded", 4)
 
 
# ── Constants ─────────────────────────────────────────────────────────────
 
 
#: Maximum chunks-per-second the worker thread will emit. Matches the
#: observed DeepSeek-v4-pro real-world max; higher rates expose
#: prompt_toolkit redraw scheduling issues that aren't representative
#: of production load.
MAX_CHUNKS_PER_SECOND: int = 80
 
 
# ── Rich-ANSI sample blocks ───────────────────────────────────────────────
 
 
def _build_sample_blocks() -> list[str]:
    """Return a list of Rich-rendered ANSI strings.
 
    Tests the four risky rendering paths: Markdown heading, fenced
    code with syntax highlighting, table, long paragraph, and wide
    unicode. Falls back to plain text if ``rich`` is missing so the
    spike still loads on a bare environment.
    """
    func = "_build_sample_blocks"
    try:
        from io import StringIO
        from rich.console import Console
        from rich.markdown import Markdown
        from rich.table import Table
    except ImportError as exc:
        _dbg("OC_CHAT.TUI_SPIKE", func, f"rich missing — plain fallback: {exc}", 3)
        return [
            "# Heading (plain)\n",
            "```py\nprint('hello')\n```\n",
            "| a | b |\n|---|---|\n| 1 | 2 |\n",
            (
                "Long paragraph: "
                "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
                * 30
            ),
            # OCT-LINT: disable=no_hardcoded_secrets reason="plain-fallback TUI sample text with wide Unicode characters; no credential material (operator-ruled 2026-09-28)"
            "Wide unicode: 你好 こんにちは 안녕하세요 — combining: á é\n",
        ]
 
    def _render(renderable: object) -> str:
        buf = StringIO()
        console = Console(file=buf, force_terminal=True, color_system="truecolor", width=100)
        console.print(renderable)
        return buf.getvalue()
 
    blocks: list[str] = []
    blocks.append(_render(Markdown("# Markdown heading — render sanity")))
    blocks.append(_render(Markdown(
        "```python\n"
        "def hello(name: str) -> str:\n"
        "    return f'Hello, {name}!'\n"
        "```\n"
    )))
    table = Table(title="Table sanity (Rich → ANSI → prompt_toolkit)")
    table.add_column("Phase", style="cyan")
    table.add_column("Status", style="green")
    table.add_column("Notes", style="white")
    table.add_row("TUI-1 spike", "in flight", "this run")
    table.add_row("TUI-1 last-pager", "in flight", "Agent A")
    table.add_row("TUI-1 runtime-prep", "in flight", "Agent B (this)")
    blocks.append(_render(table))
    blocks.append(_render(
        "Long paragraph: "
        + (
            "Lorem ipsum dolor sit amet, consectetur adipiscing elit, "
            "sed do eiusmod tempor incididunt ut labore et dolore magna "
            "aliqua. Ut enim ad minim veniam, quis nostrud exercitation. "
        ) * 20
    ))
    blocks.append(_render(
        # OCT-LINT: disable=no_hardcoded_secrets reason="TUI render-sanity sample text with wide Unicode and emoji; no credential material (operator-ruled 2026-09-28)"
        "Wide unicode sanity: "
        "中文 日本語 한국어 — combining marks: café, naïve, résumé. "
        "Emoji: \U0001F600 \U0001F525 \U0001F4DD."
    ))
    return blocks
 
 
# ── Stream simulator ──────────────────────────────────────────────────────
 
 
class _StreamSimulator:
    """Background thread that appends fake streaming chunks.
 
    Polls a :class:`threading.Event` between chunks so SIGINT (or any
    explicit ``.stop()`` call) cancels mid-stream cleanly.
    """
 
    __slots__ = ("_stop", "_thread", "_rate", "_on_chunk")
 
    def __init__(self, on_chunk: "callable[[str], None]", rate_per_sec: int) -> None:
        rate = min(max(int(rate_per_sec), 1), MAX_CHUNKS_PER_SECOND)
        self._rate = rate
        self._on_chunk = on_chunk
        self._stop = threading.Event()
        self._thread: Optional[threading.Thread] = None
 
    def start(self) -> None:
        func = "_StreamSimulator.start"
        if self._thread is not None and self._thread.is_alive():
            _dbg("OC_CHAT.TUI_SPIKE", func, "already running", 3)
            return
        self._stop.clear()
        self._thread = threading.Thread(target=self._loop, daemon=True, name="tui-spike-stream")
        self._thread.start()
        _dbg("OC_CHAT.TUI_SPIKE", func, f"started @ {self._rate} chunks/s", 2)
 
    def stop(self) -> None:
        func = "_StreamSimulator.stop"
        self._stop.set()
        _dbg("OC_CHAT.TUI_SPIKE", func, "stop requested", 3)
 
    def is_running(self) -> bool:
        return self._thread is not None and self._thread.is_alive()
 
    def _loop(self) -> None:
        func = "_StreamSimulator._loop"
        interval = 1.0 / self._rate
        idx = 0
        words = (
            "The quick brown fox jumps over the lazy dog. ".split()
            + "Spike chunk".split()
        )
        while not self._stop.is_set():
            chunk = words[idx % len(words)] + " "
            idx += 1
            try:
                self._on_chunk(chunk)
            except Exception as exc:
                _dbg("OC_CHAT.TUI_SPIKE", func,
                     f"SYSTEM_ERROR: on_chunk: {type(exc).__name__}: {exc}",
                     1, override=True)
                break
            time.sleep(interval)
        _dbg("OC_CHAT.TUI_SPIKE", func, f"loop ended after {idx} chunks", 2)
 
 
# ── Application factory ───────────────────────────────────────────────────
 
 
def build_application(
    *,
    sample_blocks: list[str] | None = None,
    # OCT-LINT: disable=no_hardcoded_secrets reason="default status-line help text naming the key bindings; no credential material (operator-ruled 2026-09-28)"
    initial_status: str = "Ready. Ctrl+S start stream, Ctrl+Q quit, PgUp/PgDn scroll.",
    input: object | None = None,
    output: object | None = None,
) -> "tuple[object, _StreamSimulator]":
    """Construct the prompt_toolkit Application + its stream simulator.
 
    Returns ``(application, stream_simulator)``. The application is not
    yet running — call ``application.run()`` to drive it. Splitting the
    factory from :func:`main` keeps the wiring inspectable from a
    headless test (see :mod:`oc_chat.tests.unit.test_tui_spike`).
 
    Headless test pattern (no real terminal needed)::
 
        from prompt_toolkit.input import create_pipe_input
        from prompt_toolkit.output import DummyOutput
 
        with create_pipe_input() as pipe_input:
            app, _ = build_application(input=pipe_input, output=DummyOutput())
            # ... drive app, inspect state ...
    """
    func = "build_application"
    # Lazy import so module import doesn't require prompt_toolkit when
    # we're only testing the simulator class.
    from prompt_toolkit.application import Application
    from prompt_toolkit.formatted_text import ANSI
    from prompt_toolkit.key_binding import KeyBindings
    from prompt_toolkit.layout import HSplit, Layout, Window
    from prompt_toolkit.layout.controls import FormattedTextControl
    from prompt_toolkit.layout.dimension import Dimension as D
 
    blocks = sample_blocks if sample_blocks is not None else _build_sample_blocks()
    conversation_buffer: list[str] = list(blocks)
    status_buffer = [initial_status]
 
    def _conversation_text() -> "ANSI":
        return ANSI("\n".join(conversation_buffer))
 
    def _header_text() -> "ANSI":
        return ANSI(
            # OCT-LINT: disable=no_hardcoded_secrets reason="ANSI-styled TUI header text naming the key bindings; no credential material (operator-ruled 2026-09-28)"
            "\x1b[1m[OC Chat TUI feasibility spike]\x1b[0m  "
            f"chunks={sum(1 for _ in conversation_buffer)}  "
            "Ctrl+S: stream  Ctrl+C: cancel  Ctrl+Q: quit"
        )
 
    def _input_text() -> "ANSI":
        return ANSI(status_buffer[-1])
 
    conversation_control = FormattedTextControl(text=_conversation_text, focusable=True)
    header_control = FormattedTextControl(text=_header_text, focusable=False)
    input_control = FormattedTextControl(text=_input_text, focusable=False)
 
    # Scrollable middle pane via prompt_toolkit's built-in vertical
    # scroll on Window — we provide our own PgUp/PgDn bindings further
    # down to drive the scroll position deterministically.
    conversation_window = Window(
        content=conversation_control,
        wrap_lines=True,
        always_hide_cursor=True,
        dont_extend_height=False,
    )
 
    layout = Layout(
        HSplit([
            Window(content=header_control, height=D.exact(1)),
            conversation_window,
            Window(content=input_control, height=D.exact(1)),
        ]),
        focused_element=conversation_window,
    )
 
    kb = KeyBindings()
    simulator_ref: list["_StreamSimulator | None"] = [None]
 
    def _append_chunk(chunk: str) -> None:
        """Append a chunk from the worker thread (event-loop safe)."""
        if not conversation_buffer or not conversation_buffer[-1].endswith("\n"):
            if conversation_buffer:
                conversation_buffer[-1] = conversation_buffer[-1] + chunk
            else:
                conversation_buffer.append(chunk)
        else:
            conversation_buffer.append(chunk)
        # Trigger a redraw from the worker thread safely.
        app = application
        if app is not None and app.is_running:
            app.invalidate()
 
    simulator = _StreamSimulator(on_chunk=_append_chunk, rate_per_sec=40)
    simulator_ref[0] = simulator
 
    @kb.add("c-q")
    def _exit(event: object) -> None:
        func = "build_application._exit"
        _dbg("OC_CHAT.TUI_SPIKE", func, "Ctrl+Q pressed — exiting", 2)
        simulator.stop()
        event.app.exit()  # type: ignore[attr-defined]
 
    @kb.add("c-s")
    def _start_stream(event: object) -> None:
        func = "build_application._start_stream"
        _dbg("OC_CHAT.TUI_SPIKE", func, "Ctrl+S pressed — starting stream", 3)
        status_buffer[-1] = "Streaming… Ctrl+C cancels."
        event.app.invalidate()  # type: ignore[attr-defined]
        simulator.start()
 
    @kb.add("c-c")
    def _cancel_stream(event: object) -> None:
        # Cooperative cancel — does NOT exit the app.
        func = "build_application._cancel_stream"
        _dbg("OC_CHAT.TUI_SPIKE", func, "Ctrl+C pressed — cancelling stream", 3)
        if simulator.is_running():
            simulator.stop()
            status_buffer[-1] = "Stream cancelled. Ctrl+S to restart, Ctrl+Q to quit."
        else:
            status_buffer[-1] = "No active stream. Ctrl+S to start, Ctrl+Q to quit."
        event.app.invalidate()  # type: ignore[attr-defined]
 
    @kb.add("pageup")
    def _scroll_up(event: object) -> None:
        # Reset to top is fine for spike; real impl would track offset.
        info = conversation_window.render_info
        if info is not None:
            conversation_window.vertical_scroll = max(
                0, conversation_window.vertical_scroll - info.window_height + 1
            )
        event.app.invalidate()  # type: ignore[attr-defined]
 
    @kb.add("pagedown")
    def _scroll_down(event: object) -> None:
        info = conversation_window.render_info
        if info is not None:
            conversation_window.vertical_scroll += info.window_height - 1
        event.app.invalidate()  # type: ignore[attr-defined]
 
    @kb.add("home")
    def _scroll_home(event: object) -> None:
        conversation_window.vertical_scroll = 0
        event.app.invalidate()  # type: ignore[attr-defined]
 
    @kb.add("end")
    def _scroll_end(event: object) -> None:
        conversation_window.vertical_scroll = 10_000  # clamped by prompt_toolkit
        event.app.invalidate()  # type: ignore[attr-defined]
 
    application_kwargs: dict[str, object] = {
        "layout": layout,
        "key_bindings": kb,
        "full_screen": True,
        "mouse_support": False,
    }
    if input is not None:
        application_kwargs["input"] = input
    if output is not None:
        application_kwargs["output"] = output
    application = Application(**application_kwargs)  # type: ignore[arg-type]
 
    _dbg("OC_CHAT.TUI_SPIKE", func, "Application built", 2)
    return application, simulator
 
 
# ── Entry point ───────────────────────────────────────────────────────────
 
 
def main() -> int:
    """Run the spike. Returns a process exit code (0 on clean exit)."""
    func = "main"
    _dbg("OC_CHAT.TUI_SPIKE", func, "spike starting", 2)
    application, simulator = build_application()
    try:
        application.run()
    except KeyboardInterrupt:
        # Outer Ctrl+C while NOT focused on the cancel binding — exit
        # cleanly. The simulator may still be running; tear it down.
        _dbg("OC_CHAT.TUI_SPIKE", func, "outer KeyboardInterrupt — exiting", 2, override=True)
    finally:
        simulator.stop()
    _dbg("OC_CHAT.TUI_SPIKE", func, "spike exited cleanly", 2)
    return 0
 
 
if __name__ == "__main__":
    import sys
    sys.exit(main())