refactor: put shared ui screen code into ui.viewkit

1 files changed, 251 insertions, 0 deletions

diff --git a/app/ui/viewkit.py b/app/ui/viewkit.py
new file mode 100644
index 0000000..301a0af
--- /dev/null
+++ b/app/ui/viewkit.py
@@ -0,0 +1,251 @@
+"""Shared plumbing for full-screen views (the "viewkit").
+
+``ScreenView`` is the base class behind TaskView and RunView: it owns the
+event queue and drain loop, the cancellation event, the timed-redraw main
+loop with its Esc/q cancel flow, and the blocking/non-blocking getch
+switching that lets confirm dialogs own the screen. Subclasses provide
+``handle_event``/``render`` plus small hooks for how a terminal phase
+exits, and everything else — thread start-up, key handling, cleanup — is
+identical across views.
+
+The drawing helpers at the bottom are the shared primitives both views'
+render methods build on.
+"""
+
+import threading
+import time
+from queue import Empty, Queue
+from typing import List, Optional
+
+from ui import tui
+
+# Terminal states: the view's work is over and the screen waits for a key.
+TERMINAL_PHASES = ("done", "error", "cancelled")
+
+# Redraw cadence for the timed getch (milliseconds).
+DRAW_TIMEOUT_MS = 250
+
+
+class ScreenView:
+    """Base class for worker-thread-driven full-screen views.
+
+    Subclasses set ``self._worker`` (a Thread running ``_worker_main``)
+    and implement ``handle_event(event)``, ``render()`` and
+    ``_terminal_result()``. The ``run`` template below drives everything
+    else; its behavior is tuned through the hooks:
+
+    - ``_start_workers``    start threads (default: just the worker)
+    - ``_early_exit``       pre-render exit check (returns a result or None)
+    - ``_after_cancel``     result once the cancel flow completed
+    - ``_on_stop``          finally-block cleanup (cancel + block getch)
+
+    ESC/Q/Ctrl-C asks ``_prompt_cancel`` (overridable); confirming sets
+    ``self.cancelling``/``self._cancel`` and winds the worker down.
+    """
+
+    def __init__(self, scr, clock=time.time):
+        import curses
+        self.curses = curses
+        self.scr = scr
+        self.theme = tui._ensure_theme(curses)
+        self._clock = clock
+        # -- state -----------------------------------------------------
+        self.phase = "running"
+        self.finished_at: Optional[float] = None
+        self.cancelled = False
+        self.cancelling = False
+        # -- threads ---------------------------------------------------
+        self._queue: Queue = Queue()
+        self._cancel = threading.Event()
+        self._worker = None
+
+    def _now(self) -> float:
+        return self._clock()
+
+    def _finish(self, phase: str) -> None:
+        """Enter a terminal phase, freezing the elapsed clock."""
+        self.phase = phase
+        if self.finished_at is None:
+            self.finished_at = self._now()
+
+    # ------------------------------------------------------------------
+    # Event plumbing
+    # ------------------------------------------------------------------
+
+    def handle_event(self, event: dict) -> None:
+        """Fold one queued event into the view state (no drawing)."""
+        raise NotImplementedError
+
+    def _drain(self) -> None:
+        """Fold every queued event into the state."""
+        while True:
+            try:
+                event = self._queue.get_nowait()
+            except Empty:
+                return
+            self.handle_event(event)
+
+    # ------------------------------------------------------------------
+    # Main loop
+    # ------------------------------------------------------------------
+
+    def run(self):
+        """Drive the view until a terminal phase exits the loop."""
+        try:
+            self.scr.timeout(DRAW_TIMEOUT_MS)
+        except Exception:
+            pass
+        self._start_workers()
+        try:
+            while True:
+                self._drain()
+                early = self._early_exit()
+                if early is not None:
+                    return early
+                self.render()
+                key = self._get_key()
+                if key is None:
+                    continue
+                if self.phase in TERMINAL_PHASES:
+                    return self._terminal_result()
+                if key in (27, ord("q"), 3) and not self.cancelling:
+                    if self._prompt_cancel():
+                        return self._after_cancel()
+        finally:
+            self._on_stop()
+
+    def _start_workers(self) -> None:
+        if self._worker is not None:
+            self._worker.start()
+
+    def _early_exit(self):
+        """Optional pre-render exit check; a non-None value ends the view."""
+        return None
+
+    def _terminal_result(self):
+        """The view's return value when the work reached a terminal phase."""
+        raise NotImplementedError
+
+    def _after_cancel(self):
+        """The view's return value after a confirmed cancel flow."""
+        return self._terminal_result()
+
+    def _on_stop(self) -> None:
+        self._cancel.set()
+        # Leave the screen blocking again: the timed redraw getch must
+        # not make later hub dialogs (e.g. tui.flash) dismiss themselves.
+        self._blocking()
+
+    def _get_key(self) -> Optional[int]:
+        """One key from the screen (None on the redraw timeout)."""
+        try:
+            key = self.scr.getch()
+        except KeyboardInterrupt:
+            return 3
+        if key == -1:
+            return None
+        return key
+
+    def _prompt_cancel(self) -> bool:
+        """Esc/q: confirm cancel, then wait for the worker to wind down."""
+        self._blocking()
+        try:
+            answer = tui.confirm(self.scr, "Cancel this step?", default=False,
+                                 cancel_value=False)
+        finally:
+            self._nonblocking()
+        if not answer:
+            return False
+        self.cancelling = True
+        self._cancel.set()
+        self._worker.join(timeout=60)
+        return True
+
+    def _blocking(self) -> None:
+        """Make getch block (used while a confirm dialog owns the screen)."""
+        try:
+            self.scr.timeout(-1)
+        except Exception:
+            pass
+
+    def _nonblocking(self) -> None:
+        """Restore the redraw-cadence getch timeout."""
+        try:
+            self.scr.timeout(DRAW_TIMEOUT_MS)
+        except Exception:
+            pass
+
+
+# ----------------------------------------------------------------------
+# Shared drawing primitives
+# ----------------------------------------------------------------------
+
+def _text(scr, theme, y, x, text, attr) -> None:
+    """addstr wrapper that ignores out-of-bounds errors."""
+    try:
+        scr.addstr(y, x, text, attr)
+    except Exception:
+        pass
+
+
+def _box(scr, curses, theme, height, width) -> None:
+    """Draw the full-screen frame."""
+    border = theme["border"]
+    try:
+        scr.addch(0, 0, curses.ACS_ULCORNER, border)
+        scr.addch(0, width - 1, curses.ACS_URCORNER, border)
+        scr.addch(height - 1, 0, curses.ACS_LLCORNER, border)
+        scr.addch(height - 1, width - 1, curses.ACS_LRCORNER, border)
+        scr.hline(0, 1, curses.ACS_HLINE, width - 2, border)
+        scr.hline(height - 1, 1, curses.ACS_HLINE, width - 2, border)
+        for y in range(1, height - 1):
+            scr.addch(y, 0, curses.ACS_VLINE, border)
+            scr.addch(y, width - 1, curses.ACS_VLINE, border)
+    except Exception:
+        pass
+
+
+def _sep(scr, curses, theme, y, width) -> None:
+    """A horizontal separator line inside the frame."""
+    try:
+        scr.addch(y, 0, curses.ACS_LTEE, theme["border"])
+        scr.addch(y, width - 1, curses.ACS_RTEE, theme["border"])
+        scr.hline(y, 1, curses.ACS_HLINE, width - 2, theme["dim"])
+    except Exception:
+        pass
+
+
+def _fit(text: str, width: int) -> str:
+    """Truncate TEXT to WIDTH columns, appending '~' when cut."""
+    if width < 1:
+        return ""
+    if len(text) <= width:
+        return text
+    return text[: max(0, width - 1)] + "~"
+
+
+def _wrap(text: str, width: int) -> List[str]:
+    """Greedy word wrap (no textwrap dependency on curses chars)."""
+    lines: List[str] = []
+    current = ""
+    for word in text.split():
+        candidate = f"{current} {word}".strip()
+        if len(candidate) <= max(10, width):
+            current = candidate
+        else:
+            if current:
+                lines.append(current)
+            current = word
+    if current:
+        lines.append(current)
+    return lines
+
+
+def _format_elapsed(seconds: float) -> str:
+    """Format a duration as H:MM:SS / M:SS."""
+    seconds = max(0, int(seconds))
+    hours, remainder = divmod(seconds, 3600)
+    minutes, secs = divmod(remainder, 60)
+    if hours:
+        return f"{hours}:{minutes:02d}:{secs:02d}"
+    return f"{minutes}:{secs:02d}"