feat(tui): simultaneous build and model download

1 files changed, 481 insertions, 37 deletions

diff --git a/app/ui/taskview.py b/app/ui/taskview.py
index 65237af..7f8134b 100644
--- a/app/ui/taskview.py
+++ b/app/ui/taskview.py
@@ -26,10 +26,20 @@ in-process steps are expected to check it between units of work. When all
 steps finish (or are cancelled) the view shows a summary and waits for a key
 press, so a failure is never scrolled away. ``run_steps`` returns the first
 non-zero step exit code (0 when every step succeeded).
+
+Steps can also be grouped into ``TaskLane``s and run through ``run_lanes``:
+two lanes each get their own worker thread, step list, progress bar, and log
+tail, drawn side by side (or stacked on a narrow terminal) so independent
+work — the audio.cpp build in one lane, model downloads in the other — runs
+simultaneously. Because ``redirect_stdout`` is process-global, the multi-lane
+view installs a thread-routing stdout/stderr proxy for the run's duration, so
+each lane's ``print()`` output lands in its own log. A single lane renders
+exactly like ``run_steps``.
 """
 
 import contextlib
 import re
+import sys
 import threading
 import time
 from dataclasses import dataclass
@@ -70,6 +80,69 @@ class TaskStep:
     work: Callable[[Callable[[str], None], threading.Event], int]
 
 
+@dataclass
+class TaskLane:
+    """One column of a (possibly parallel) task view.
+
+    A lane is a titled, ordered list of steps that run in its own worker
+    thread. ``run_lanes`` draws a single lane full-width exactly like
+    ``run_steps``, and splits the screen in half when two lanes are given so
+    their steps (e.g. build and model download) run simultaneously.
+    """
+
+    title: str
+    steps: List[TaskStep]
+
+
+def _progress_match(text: str) -> Optional[Tuple[float, float, str]]:
+    """Parse a progress line into ``(done, total, kind)``, else None.
+
+    KIND is one of ``"bytes"`` (``AUDIOCPP_PROGRESS``), ``"percent"``
+    (``NN%``), or ``"count"`` (``[done/total]``), with the same guards the
+    single-lane view applies (percents capped at 100, counts bounded by
+    their total).
+    """
+    match = _PROGRESS_BYTES.search(text)
+    if match:
+        return (int(match.group(1)), int(match.group(2)), "bytes")
+    match = _PROGRESS_PERCENT.search(text)
+    if match:
+        percent = int(match.group(1))
+        if percent <= 100:
+            return (percent, 100, "percent")
+    match = _PROGRESS_COUNT.search(text)
+    if match:
+        done = int(match.group(1))
+        total = int(match.group(2))
+        if total > 0 and done <= total:
+            return (done, total, "count")
+    return None
+
+
+def _lane_step_mark(current: Optional[int],
+                    results: List[Optional[int]],
+                    cancelled_step: Optional[int],
+                    index: int, now: float, terminal: bool
+                    ) -> Tuple[str, str]:
+    """The (mark, kind) for step INDEX of one lane; see TaskView._step_mark."""
+    if terminal:
+        if index == cancelled_step:
+            return "[x]", "warn"
+        if results[index] == 0:
+            return "[OK]", "ok"
+        if results[index] is not None:
+            return "[FAIL]", "err"
+        return "[  ]", "dim"
+    if index == current:
+        frame = _SPINNER[int(now * 4) % len(_SPINNER)]
+        return f"[{frame}]", "warn"
+    if results[index] == 0:
+        return "[OK]", "ok"
+    if results[index] is not None:
+        return "[FAIL]", "err"
+    return "[  ]", "dim"
+
+
 def run_steps(scr, title: str, steps: List[TaskStep]) -> int:
     """Run STEPS in order inside the curses screen; return the first bad rc.
 
@@ -96,6 +169,25 @@ def run_steps_inline(steps: List[TaskStep], emit=None, cancel=None) -> int:
     return first
 
 
+def run_lanes(scr, title: str, lanes: List[TaskLane]) -> int:
+    """Run LANES inside the curses screen; return the first bad rc.
+
+    Each lane is an ordered list of steps that run in its own worker thread.
+    A single lane renders full-width exactly like ``run_steps``; two lanes
+    are drawn side by side (or stacked on a narrow terminal) so their steps
+    run simultaneously — the audio.cpp one-click setup builds the server in
+    one lane while configuring and downloading models in the other. Empty
+    lanes are dropped, so callers can build a lane list conditionally and
+    always end up with "just build", "just download", or both.
+    """
+    lanes = [lane for lane in lanes if lane.steps]
+    if not lanes:
+        return 0
+    if len(lanes) == 1:
+        return run_steps(scr, title, lanes[0].steps)
+    return LanesView(scr, title, lanes).run()
+
+
 class TaskView:
     """Draws and drives one list of setup steps; see the module docstring."""
 
@@ -201,29 +293,15 @@ class TaskView:
         line = text.rstrip("\r\n")
         if not line:
             return
-        match = _PROGRESS_BYTES.search(line)
+        match = _progress_match(line)
         if match:
-            total = int(match.group(2))
-            done = int(match.group(1))
+            done, total, kind = match
             self._progress = (done, total)
-            self._progress_kind = "bytes"
-            return  # machine-readable progress is not part of the log
-        match = _PROGRESS_PERCENT.search(line)
-        if match:
-            percent = int(match.group(1))
-            if percent <= 100:
-                self._progress = (percent, 100)
-                self._progress_kind = "percent"
-                # Fall through: keep the line in the log (the tail already
-                # collapses rapid \r updates to the last full line).
-        else:
-            match = _PROGRESS_COUNT.search(line)
-            if match:
-                done = int(match.group(1))
-                total = int(match.group(2))
-                if total > 0 and done <= total:
-                    self._progress = (done, total)
-                    self._progress_kind = "count"
+            self._progress_kind = kind
+            if kind == "bytes":
+                return  # machine-readable progress is not part of the log
+            # Percent/count lines stay in the log (the tail already
+            # collapses rapid \r updates to the last full line).
         self.log_tail.append(line)
         if len(self.log_tail) > _LOG_TAIL:
             del self.log_tail[: len(self.log_tail) - _LOG_TAIL]
@@ -401,22 +479,9 @@ class TaskView:
 
     def _step_mark(self, index: int) -> Tuple[str, str]:
         """The (mark, kind) for step INDEX."""
-        if self.phase in _TERMINAL:
-            if index == self.cancelled_step:
-                return "[x]", "warn"
-            if self.results[index] == 0:
-                return "[OK]", "ok"
-            if self.results[index] is not None:
-                return "[FAIL]", "err"
-            return "[  ]", "dim"
-        if index == self.current:
-            frame = _SPINNER[int(self._now() * 4) % len(_SPINNER)]
-            return f"[{frame}]", "warn"
-        if self.results[index] == 0:
-            return "[OK]", "ok"
-        if self.results[index] is not None:
-            return "[FAIL]", "err"
-        return "[  ]", "dim"
+        return _lane_step_mark(self.current, self.results,
+                               self.cancelled_step, index,
+                               self._now(), self.phase in _TERMINAL)
 
 
 # ---------------------------------------------------------------------------
@@ -535,3 +600,382 @@ def _fmt_bytes(size: float) -> str:
             return f"{value:.1f}{unit}"
         value /= 1024
     return f"{value:.1f}GB"
+
+
+def _rect_box(scr, curses, theme, x: int, y: int, w: int, h: int) -> None:
+    """Draw a box around the rectangle ``(x, y, w, h)``."""
+    border = theme["border"]
+    try:
+        scr.addch(y, x, curses.ACS_ULCORNER, border)
+        scr.addch(y, x + w - 1, curses.ACS_URCORNER, border)
+        scr.addch(y + h - 1, x, curses.ACS_LLCORNER, border)
+        scr.addch(y + h - 1, x + w - 1, curses.ACS_LRCORNER, border)
+        scr.hline(y, x + 1, curses.ACS_HLINE, w - 2, border)
+        scr.hline(y + h - 1, x + 1, curses.ACS_HLINE, w - 2, border)
+        for yy in range(y + 1, y + h - 1):
+            scr.addch(yy, x, curses.ACS_VLINE, border)
+            scr.addch(yy, x + w - 1, curses.ACS_VLINE, border)
+    except Exception:
+        pass
+
+
+class _ThreadRouter:
+    """A file-like object that routes writes to a per-thread writer.
+
+    ``contextlib.redirect_stdout`` is process-global, so two lanes running in
+    parallel would interleave their ``print()`` output. Instead, one router is
+    installed on ``sys.stdout``/``sys.stderr`` for the whole view run and each
+    lane's worker registers its ``_LineWriter`` while a step runs; writes from
+    an unregistered thread fall through to the original stream.
+    """
+
+    def __init__(self, fallback, registry: dict = None):
+        self._fallback = fallback
+        self._registry = registry if registry is not None else {}
+        self._lock = threading.Lock()
+
+    @contextlib.contextmanager
+    def for_thread(self, writer):
+        ident = threading.get_ident()
+        with self._lock:
+            self._registry[ident] = writer
+        try:
+            yield
+        finally:
+            with self._lock:
+                self._registry.pop(ident, None)
+
+    def write(self, text):
+        writer = self._registry.get(threading.get_ident())
+        if writer is not None:
+            return writer.write(text)
+        return self._fallback.write(text)
+
+    def flush(self):
+        writer = self._registry.get(threading.get_ident())
+        if writer is not None:
+            writer.flush()
+        else:
+            self._fallback.flush()
+
+    def isatty(self) -> bool:
+        return False
+
+
+class _LaneState:
+    """Mutable state for one lane of a ``LanesView`` (see TaskView fields)."""
+
+    def __init__(self, title: str, steps: List[TaskStep]):
+        self.title = title
+        self.steps = list(steps)
+        self.queue: Queue = Queue()
+        self.worker = None
+        self.current: Optional[int] = None
+        self.results: List[Optional[int]] = [None] * len(self.steps)
+        self.log_tail: List[str] = []
+        self.progress: Optional[Tuple[float, float]] = None
+        self.progress_kind = ""
+        self.step_started: List[Optional[float]] = [None] * len(self.steps)
+        self.cancelled_step: Optional[int] = None
+        self.rc = 0
+        self.finished = False
+
+
+class LanesView:
+    """A full-screen task view that runs two step lists in parallel.
+
+    The two-lane counterpart of ``TaskView``: each lane gets its own worker
+    thread, event queue, and state (step marks, progress bar, log tail), and
+    the screen is split into two panes so both lanes' progress is visible at
+    once. One shared cancel event stops both lanes. The run reaches its
+    terminal phase only once every lane has finished; the returned rc is the
+    first non-zero step rc across the lanes, in lane order.
+    """
+
+    def __init__(self, scr, title: str, lanes: List[TaskLane],
+                 clock: Callable[[], float] = time.time):
+        import curses
+        self.curses = curses
+        self.scr = scr
+        self.title = title
+        self.theme = tui._ensure_theme(curses)
+        self._clock = clock
+        self._lanes = [_LaneState(lane.title, lane.steps) for lane in lanes]
+        self.phase = "running"   # running | done | error | cancelled
+        self.cancelled = False
+        self.cancelling = False
+        self.finished_at: Optional[float] = None
+        self._cancel = threading.Event()
+
+    # -- worker ------------------------------------------------------
+
+    def _lane_worker(self, lane: _LaneState, router: _ThreadRouter,
+                     cancel: threading.Event) -> None:
+        first_failure = 0
+
+        def emit(line: str) -> None:
+            lane.queue.put({"kind": "line", "text": line})
+
+        for index, step in enumerate(lane.steps):
+            if cancel.is_set():
+                break
+            lane.queue.put({"kind": "step_start", "index": index,
+                            "title": step.title})
+            try:
+                with router.for_thread(_LineWriter(emit)):
+                    rc = step.work(emit, cancel)
+            except Exception as exc:  # noqa: BLE001 - reported to the view
+                lane.queue.put({"kind": "line",
+                                "text": f"[ERROR] {exc}"})
+                rc = 1
+            if cancel.is_set():
+                lane.queue.put({"kind": "step_cancelled", "index": index})
+                break
+            lane.queue.put({"kind": "step_done", "index": index, "rc": rc})
+            if rc != 0:
+                first_failure = first_failure or rc
+        lane.queue.put({"kind": "lane_finish", "rc": first_failure})
+
+    # -- event handling ----------------------------------------------
+
+    def _handle_lane_event(self, lane: _LaneState, event: dict) -> None:
+        kind = event.get("kind")
+        if kind == "step_start":
+            lane.current = event["index"]
+            lane.step_started[lane.current] = self._now()
+            lane.progress = None
+            lane.progress_kind = ""
+        elif kind == "line":
+            self._ingest_lane_line(lane, event.get("text") or "")
+        elif kind == "step_done":
+            lane.results[event["index"]] = event.get("rc") or 0
+            lane.current = None
+            lane.progress = None
+            lane.progress_kind = ""
+        elif kind == "step_cancelled":
+            lane.cancelled_step = event["index"]
+            lane.current = None
+            lane.progress = None
+            lane.progress_kind = ""
+        elif kind == "lane_finish":
+            lane.rc = event.get("rc") or 0
+            lane.finished = True
+
+    def _ingest_lane_line(self, lane: _LaneState, text: str) -> None:
+        """Fold one output line into LANE's log tail and progress bar."""
+        line = text.rstrip("\r\n")
+        if not line:
+            return
+        match = _progress_match(line)
+        if match:
+            done, total, kind = match
+            lane.progress = (done, total)
+            lane.progress_kind = kind
+            if kind == "bytes":
+                return
+        lane.log_tail.append(line)
+        if len(lane.log_tail) > _LOG_TAIL:
+            del lane.log_tail[: len(lane.log_tail) - _LOG_TAIL]
+
+    def _drain(self) -> None:
+        for lane in self._lanes:
+            while True:
+                try:
+                    event = lane.queue.get_nowait()
+                except Empty:
+                    break
+                self._handle_lane_event(lane, event)
+        if self.phase == "running" and all(lane.finished
+                                           for lane in self._lanes):
+            self._finish()
+
+    def _finish(self) -> None:
+        if self._cancel.is_set():
+            self.phase = "cancelled"
+            self.cancelled = True
+        else:
+            self.phase = "done"
+            for lane in self._lanes:
+                if lane.rc:
+                    self.phase = "error"
+                    break
+        self.finished_at = self._now()
+
+    def _now(self) -> float:
+        return self._clock()
+
+    def _result_rc(self) -> int:
+        """The exit code for the whole run (cancelled counts as failure)."""
+        if self.cancelled:
+            return 1
+        for lane in self._lanes:
+            for rc in lane.results:
+                if rc:
+                    return rc
+        return 0
+
+    # -- main loop ---------------------------------------------------
+
+    def run(self) -> int:
+        scr = self.scr
+        try:
+            scr.timeout(_DRAW_TIMEOUT_MS)
+        except Exception:
+            pass
+        registry = {}
+        router_out = _ThreadRouter(sys.stdout, registry)
+        router_err = _ThreadRouter(sys.stderr, registry)
+        saved_out, saved_err = sys.stdout, sys.stderr
+        sys.stdout, sys.stderr = router_out, router_err
+        try:
+            for lane in self._lanes:
+                lane.worker = threading.Thread(
+                    target=self._lane_worker,
+                    args=(lane, router_out, self._cancel), daemon=True)
+                lane.worker.start()
+            try:
+                while True:
+                    self._drain()
+                    self.render()
+                    key = self._get_key()
+                    if key is None:
+                        continue
+                    if self.phase in _TERMINAL:
+                        return self._result_rc()
+                    if key in (27, ord("q"), 3) and not self.cancelling:
+                        if self._prompt_cancel():
+                            self._drain()
+                            return self._result_rc()
+            finally:
+                self._cancel.set()
+        finally:
+            sys.stdout, sys.stderr = saved_out, saved_err
+
+    def _get_key(self) -> Optional[int]:
+        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 both workers 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()
+        for lane in self._lanes:
+            if lane.worker is not None:
+                lane.worker.join(timeout=60)
+        return True
+
+    def _blocking(self) -> None:
+        try:
+            self.scr.timeout(-1)
+        except Exception:
+            pass
+
+    def _nonblocking(self) -> None:
+        try:
+            self.scr.timeout(_DRAW_TIMEOUT_MS)
+        except Exception:
+            pass
+
+    # -- drawing -----------------------------------------------------
+
+    def render(self) -> None:
+        curses, theme = self.curses, self.theme
+        scr = self.scr
+        scr.erase()
+        height, width = scr.getmaxyx()
+        if height < 12 or width < 40:
+            _text(scr, theme, height // 2, 2, "Terminal too small",
+                  curses.A_BOLD)
+            scr.refresh()
+            return
+
+        _box(scr, curses, theme, height, width)
+        _text(scr, theme, 0, 2, _fit(f" {self.title} ", width - 4),
+              theme["title"])
+
+        terminal = self.phase in _TERMINAL
+        inner_h = height - 3
+        if width >= 76:
+            pane_w = (width - 3) // 2
+            rects = [(1, 1, pane_w, inner_h),
+                     (1 + pane_w + 1, 1, (width - 3) - pane_w, inner_h)]
+        else:
+            top_h = (inner_h - 1) // 2
+            rects = [(1, 1, width - 2, top_h),
+                     (1, 2 + top_h, width - 2, inner_h - top_h - 1)]
+
+        for lane, (x, y, w, h) in zip(self._lanes, rects):
+            self._draw_pane(curses, theme, x, y, w, h, lane, terminal)
+
+        if self.phase == "done":
+            footer, kind = "completed — press any key to return", "ok"
+        elif self.phase == "cancelled":
+            footer, kind = "cancelled — press any key to return", "warn"
+        elif self.phase == "error":
+            footer, kind = "finished with errors — press any key to return", "err"
+        elif self.cancelling:
+            footer, kind = "cancelling...", "warn"
+        else:
+            footer, kind = "Esc or q: cancel", "dim"
+        _text(scr, theme, height - 2, 2, _fit(footer, width - 4), theme[kind])
+        scr.refresh()
+
+    def _draw_pane(self, curses, theme, x: int, y: int, w: int, h: int,
+                   lane: _LaneState, terminal: bool) -> None:
+        scr = self.scr
+        _rect_box(scr, curses, theme, x, y, w, h)
+        _text(scr, theme, y, x + 1, _fit(f" {lane.title} ", w - 2),
+              theme["title"])
+
+        row = y + 1
+        for index, step in enumerate(lane.steps):
+            mark, kind = _lane_step_mark(lane.current, lane.results,
+                                         lane.cancelled_step, index,
+                                         self._now(), terminal)
+            label = _fit(f" {step.title} ", max(6, w - 8))
+            _text(scr, theme, row, x + 1, mark, theme.get(kind, theme["body"]))
+            _text(scr, theme, row, x + 6, label, theme["body"])
+            if index == lane.current and not terminal:
+                started = lane.step_started[index] or self._now()
+                _text(scr, theme, row, x + 6 + len(label) + 1,
+                      f" {_format_elapsed(self._now() - started)}",
+                      theme["dim"])
+            row += 1
+
+        row += 1
+        if lane.progress is not None and not terminal:
+            done, total = lane.progress
+            bar_x = x + 10
+            bar_room = max(6, w - 12)
+            filled = 0
+            if total:
+                filled = round(bar_room * min(done, total) / total)
+                filled = max(0, min(bar_room, filled))
+            _text(scr, theme, row, x + 1, "Progress".ljust(9), theme["dim"])
+            try:
+                scr.addstr(row, bar_x, " " * filled, theme["bar"])
+            except Exception:
+                pass
+            _text(scr, theme, row, bar_x + bar_room + 1,
+                  _progress_label(lane.progress, lane.progress_kind),
+                  theme["accent"])
+            row += 1
+
+        for line in lane.log_tail[-_LOG_TAIL:]:
+            if row >= y + h - 1:
+                break
+            _text(scr, theme, row, x + 1, _fit(line, w - 3), theme["dim"])
+            row += 1