path: root/app/ui/taskview.py

#!/usr/bin/env python3
"""A full-screen task runner for long setup steps that stay in the TUI.

Long backend-setup steps (git clone, audiocpp_server build, model downloads,
pip installs, whisper transcription) used to run under ``tui.suspend``, which
dumped the user into plain console output. This widget keeps them inside the
hub's curses session: a worker thread runs an ordered list of ``TaskStep``s
while the main thread redraws a DOS-style frame showing each step's state
(pending / running with a spinner and elapsed clock / [OK] / [FAIL]), an
optional progress bar for the current step, and a dim log tail of the step's
output filling the remaining screen height. Every line the view shows is also
mirrored to the ``tui_YYYYMMDD.log`` day stream under app/logs (see
``_ConsoleLog``), so console output survives the curses session even when the
step itself keeps no log.

Steps stream their output by calling ``emit(line)`` (or simply printing to
stdout/stderr, which the view captures). The view turns output into progress
three ways, best-effort:

  * ``AUDIOCPP_PROGRESS downloaded=N total=M``  (audio.cpp model downloads,
    hidden from the log) — an exact bytes bar;
  * ``NN%`` (git ``Receiving objects: 45%``, cmake/make ``[ 45%]``, tqdm) —
    a percent bar;
  * ``[done/total]`` (ninja build output) — a count bar.

A ``threading.Event`` passed to every step is set when the user confirms
cancel (Esc/q); subprocess runners kill their child process group, and
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
from datetime import datetime
from queue import Empty, Queue
from typing import Callable, List, Optional, Tuple

import logging_kit

from ui import tui
from ui.viewkit import (TERMINAL_PHASES as _TERMINAL,
                        DRAW_TIMEOUT_MS as _DRAW_TIMEOUT_MS,
                        ScreenView, _box, _fit, _format_elapsed, _sep,
                        _text)

# How many recent output lines the tail keeps in memory. The on-screen tail
# draws as many as fit (see render); the full run is also mirrored to the
# ``tui_`` day stream under app/logs (see _ConsoleLog).
_LOG_KEEP = 1000

# Progress-line matchers, in order of precedence.
_PROGRESS_BYTES = re.compile(r"AUDIOCPP_PROGRESS downloaded=(\d+) total=(\d+)")
_PROGRESS_PERCENT = re.compile(r"(\d{1,3})%")
_PROGRESS_COUNT = re.compile(r"\[(\d+)/(\d+)\]")

# A spinner frame set for the running step marker.
_SPINNER = ("|", "/", "-", "\\")

# How long a running step may stay silent (no output lines) before the
# view starts saying so next to its elapsed clock — an early "this looks
# wedged" cue for the no-output watchdog that eventually kills the step.
_SILENCE_CUE_SECS = 60


@dataclass
class TaskStep:
    """One step of a task view run.

    WORK is ``work(emit, cancel) -> int``: it streams output lines through
    EMIT and returns its exit code (0 = success). CANCEL is a
    ``threading.Event`` the view sets when the user confirms cancel; WORK
    should stop promptly and may return any code (the view reports the run
    as "cancelled" regardless).
    """
    title: str
    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 _silent_secs(last_line_at: Optional[float], now: float
                 ) -> Optional[float]:
    """Seconds the running step has been silent, or None (no cue).

    None when nothing is tracked (no output yet is not tracked here — the
    caller decides), or when the silence is still below the cue threshold.
    Module-level for testability.
    """
    if last_line_at is None:
        return None
    silent = now - last_line_at
    if silent < _SILENCE_CUE_SECS:
        return None
    return silent


def _silence_cue_text(silent: float) -> str:
    """The dim text shown next to the elapsed clock for SILENT seconds."""
    return f"(no output {int(silent // 60)}m)"


class _ConsoleLog:
    """Mirrors a task view's console output into the ``tui_`` day stream.

    Every line the view shows (minus machine-readable progress lines) is
    appended to ``app/logs/tui_YYYYMMDD.log``, with a separator header per
    run and step start/finish markers, so no in-TUI console output is lost.
    The file is opened lazily on the first line — a run with no output
    creates nothing — and every write is best-effort: an unwritable
    app/logs simply disables the mirror. One instance per view run; both
    lanes of a LanesView share theirs (ingestion runs on the main thread).
    """

    def __init__(self, title: str):
        self._title = title
        self._handle = None
        self._started = False

    def line(self, text: str) -> None:
        """Append TEXT (and, once, the run's separator header)."""
        if not self._started:
            self._started = True
            self._handle = logging_kit.day_stream("tui")
            logging_kit.write_line(self._handle, "")
            logging_kit.write_line(
                self._handle, f"=== {self._title} — "
                              f"{datetime.now():%Y-%m-%d %H:%M:%S} ===")
        logging_kit.write_line(self._handle, text)

    def close(self) -> None:
        if self._handle is not None:
            try:
                self._handle.close()
            except OSError:
                pass
            self._handle = 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],
              wait_on_finish: bool = True) -> int:
    """Run STEPS in order inside the curses screen; return the first bad rc.

    Returns 0 when every step succeeded, otherwise the first non-zero exit
    code (a cancelled run returns a non-zero code too). With WAIT_ON_FINISH
    False the view returns to the caller as soon as the run reaches a
    terminal phase instead of waiting for a key press (used by the hub's
    start/stop actions, which land straight back on the menu).
    """
    view = TaskView(scr, title, steps, wait_on_finish=wait_on_finish)
    return view.run()


def run_steps_inline(steps: List[TaskStep], emit=None, cancel=None) -> int:
    """Run STEPS in order without the curses view; return the first bad rc.

    The console/CLI counterpart of ``run_steps``: each step's work is called
    directly (EMIT None keeps the current plain-console subprocess behavior),
    and every step runs even when an earlier one failed — matching how the
    wizards warn-and-continue today.
    """
    first = 0
    for step in steps:
        rc = step.work(emit, cancel)
        if rc and not first:
            first = rc
    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(ScreenView):
    """Draws and drives one list of setup steps; see the module docstring."""

    def __init__(self, scr, title: str, steps: List[TaskStep],
                 clock: Callable[[], float] = time.time,
                 wait_on_finish: bool = True):
        super().__init__(scr, clock=clock)
        self.title = title
        self.steps = steps
        self.wait_on_finish = wait_on_finish
        # -- state -----------------------------------------------------
        self.current: Optional[int] = None   # index of the running step
        self.results: List[Optional[int]] = [None] * len(steps)
        self.cancelled_step: Optional[int] = None
        self.log_tail: List[str] = []
        self._progress: Optional[Tuple[float, float]] = None  # (done, total)
        self._progress_kind = ""   # "bytes" | "percent" | "count" | ""
        self.step_started: List[Optional[float]] = [None] * len(steps)
        self.last_line_at: Optional[float] = None  # silence cue (see _SILENCE_CUE_SECS)
        self.finished_at: Optional[float] = None
        self.cancelled = False
        self.cancelling = False
        # -- console mirror (app/logs/tui_YYYYMMDD.log) ----------------
        self._console_log = _ConsoleLog(title)
        # -- threads ---------------------------------------------------
        self._queue: Queue = Queue()
        self._cancel = threading.Event()
        self._worker = threading.Thread(target=self._worker_main, daemon=True)

    # ------------------------------------------------------------------
    # Worker
    # ------------------------------------------------------------------

    def _worker_main(self) -> None:
        first_failure = 0
        for index, step in enumerate(self.steps):
            if self._cancel.is_set():
                break
            self._queue.put({"kind": "step_start", "index": index,
                             "title": step.title})
            try:
                with contextlib.redirect_stdout(_LineWriter(self._emit)), \
                        contextlib.redirect_stderr(_LineWriter(self._emit)):
                    rc = step.work(self._emit, self._cancel)
            except Exception as exc:  # noqa: BLE001 - reported to the view
                self._queue.put({"kind": "line",
                                 "text": f"[ERROR] {exc}"})
                rc = 1
            if self._cancel.is_set():
                self._queue.put({"kind": "step_cancelled", "index": index})
                break
            self._queue.put({"kind": "step_done", "index": index, "rc": rc})
            if rc != 0:
                first_failure = first_failure or rc
                # Keep going where the console path would only warn; the
                # failing step stays marked [FAIL].
        if self._cancel.is_set():
            self._queue.put({"kind": "finish", "phase": "cancelled",
                             "rc": first_failure or 1})
        elif first_failure:
            self._queue.put({"kind": "finish", "phase": "error",
                             "rc": first_failure})
        else:
            self._queue.put({"kind": "finish", "phase": "done", "rc": 0})

    def _emit(self, line: str) -> None:
        """Forward one output line to the view queue (progress-aware)."""
        self._queue.put({"kind": "line", "text": line})

    # ------------------------------------------------------------------
    # Event handling
    # ------------------------------------------------------------------

    def handle_event(self, event: dict) -> None:
        kind = event.get("kind")
        if kind == "step_start":
            self.current = event["index"]
            self.step_started[self.current] = self._now()
            self.last_line_at = self._now()
            self._progress = None
            self._progress_kind = ""
            self._console_log.line(f"--- {event.get('title') or ''} ---")
        elif kind == "line":
            text = event.get("text") or ""
            self._ingest_line(text)
        elif kind == "step_done":
            index = event["index"]
            rc = event.get("rc") or 0
            self.results[index] = rc
            self.current = None
            self.last_line_at = None
            self._progress = None
            self._progress_kind = ""
            self._console_log.line(
                f"[{'OK' if rc == 0 else 'FAIL'}] "
                f"{self.steps[index].title} (exit {rc})")
        elif kind == "step_cancelled":
            index = event["index"]
            self.cancelled_step = index
            self.current = None
            self.last_line_at = None
            self._progress = None
            self._progress_kind = ""
            self._console_log.line(
                f"[x] {self.steps[index].title} (cancelled)")
        elif kind == "finish":
            self.phase = event.get("phase") or "done"
            self.cancelled = self.phase == "cancelled"
            self.finished_at = self._now()
            self.current = None
            self._console_log.line(
                f"=== {self.phase} (exit {event.get('rc') or 0}) ===")

    def _ingest_line(self, text: str) -> None:
        """Fold one output line into the log tail and progress bar."""
        line = text.rstrip("\r\n")
        self.last_line_at = self._now()
        if not line:
            return
        match = _progress_match(line)
        if match:
            done, total, kind = match
            self._progress = (done, total)
            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_KEEP:
            del self.log_tail[: len(self.log_tail) - _LOG_KEEP]
        self._console_log.line(line)

    # ScreenView hooks ------------------------------------------------

    def _early_exit(self):
        """A no-wait run returns as soon as the steps are over."""
        if self.phase in _TERMINAL and not self.wait_on_finish:
            return self._result_rc()
        return None

    def _terminal_result(self) -> int:
        return self._result_rc()

    def _after_cancel(self) -> int:
        # Fold the worker's final events in so the result reflects them.
        self._drain()
        return self._result_rc()

    def _result_rc(self) -> int:
        """The exit code for the whole run (cancelled counts as failure)."""
        if self.cancelled:
            return 1
        return next((rc for rc in self.results if rc), 0)

    def _on_stop(self) -> None:
        self._console_log.close()
        super()._on_stop()

    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

    # ------------------------------------------------------------------
    # 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"])

        inner_x = 2
        y = 2
        # -- step list -------------------------------------------------
        for index, step in enumerate(self.steps):
            mark, kind = self._step_mark(index)
            label = _fit(f" {step.title} ", max(8, width - inner_x - 14))
            _text(scr, theme, y, inner_x, mark, theme.get(kind, theme["body"]))
            _text(scr, theme, y, inner_x + 5, label, theme["body"])
            if index == self.current and self.phase not in _TERMINAL:
                started = self.step_started[index] or self._now()
                elapsed_text = f" {_format_elapsed(self._now() - started)}"
                _text(scr, theme, y, inner_x + 5 + len(label) + 1,
                      elapsed_text, theme["dim"])
                silent = _silent_secs(self.last_line_at, self._now())
                if silent is not None:
                    _text(scr, theme, y,
                          inner_x + 5 + len(label) + 1 + len(elapsed_text),
                          f" {_silence_cue_text(silent)}", theme["dim"])
            y += 1

        y += 1
        _sep(scr, curses, theme, y, width)
        y += 1

        # -- progress bar ----------------------------------------------
        if self._progress is not None and self.phase not in _TERMINAL:
            done, total = self._progress
            label = _progress_label(self._progress, self._progress_kind)
            bar_x = inner_x + 10
            # Reserve a space plus the label inside the right border so even
            # long byte counts never clip against the screen edge.
            bar_room = max(10, width - bar_x - len(label) - 2)
            filled = 0
            if total:
                filled = round(bar_room * min(done, total) / total)
                filled = max(0, min(bar_room, filled))
            _text(scr, theme, y, inner_x, "Progress".ljust(9), theme["dim"])
            try:
                scr.addstr(y, bar_x, " " * filled, theme["bar"])
            except Exception:
                pass
            _text(scr, theme, y, bar_x + bar_room + 1,
                  _fit(label, max(1, (width - 2) - (bar_x + bar_room))),
                  theme["accent"])
            y += 1

        # -- log tail --------------------------------------------------
        # Every recent line that fits between here and the footer; the
        # full run lives in the tui_ day stream (see _ConsoleLog).
        room = (height - 3) - y
        if room > 0:
            for line in self.log_tail[-room:]:
                _text(scr, theme, y, inner_x, _fit(line, width - inner_x - 2),
                      theme["dim"])
                y += 1

        # -- footer ----------------------------------------------------
        suffix = "" if not self.wait_on_finish else " — press any key to return"
        if self.phase == "done":
            footer = "completed" + suffix
            kind = "ok"
        elif self.phase == "cancelled":
            footer = "cancelled" + suffix
            kind = "warn"
        elif self.phase == "error":
            footer = "finished with errors" + suffix
            kind = "err"
        elif self.cancelling:
            footer = "cancelling..."
            kind = "warn"
        else:
            footer = "Esc or q: cancel"
            kind = "dim"
        _text(scr, theme, height - 2, 2, _fit(footer, width - 4),
              theme[kind])
        scr.refresh()

    def _step_mark(self, index: int) -> Tuple[str, str]:
        """The (mark, kind) for step INDEX."""
        return _lane_step_mark(self.current, self.results,
                               self.cancelled_step, index,
                               self._now(), self.phase in _TERMINAL)


# ---------------------------------------------------------------------------
# Small helpers (module-level for testability)
# ---------------------------------------------------------------------------

class _LineWriter:
    """A file-like object that forwards writes to a per-line callback.

    Handles carriage-return progress updates (git/tqdm) by treating ``\r``
    as a line terminator too, so the last full line always reflects the
    latest progress.
    """

    def __init__(self, emit: Callable[[str], None]):
        self._emit = emit
        self._buffer = ""

    def write(self, text: str) -> int:
        if not text:
            return 0
        self._buffer += text
        while True:
            cut = _find_line_end(self._buffer)
            if cut < 0:
                break
            line, self._buffer = self._buffer[:cut], self._buffer[cut + 1:]
            if line:
                self._emit(line)
        return len(text)

    def flush(self) -> None:
        if self._buffer:
            self._emit(self._buffer)
            self._buffer = ""

    def isatty(self) -> bool:
        return False


def _find_line_end(text: str) -> int:
    """Index of the earliest ``\n`` or ``\r`` in TEXT, else -1."""
    newline = text.find("\n")
    carriage = text.find("\r")
    if newline < 0:
        return carriage
    if carriage < 0:
        return newline
    return min(newline, carriage)


def _progress_label(progress: Tuple[float, float], kind: str) -> str:
    done, total = progress
    if kind == "bytes":
        return f"{_fmt_bytes(done)} / {_fmt_bytes(total)}"
    if kind == "count":
        return f"{int(done)}/{int(total)}"
    return f"{int(done)}%"


def _fmt_bytes(size: float) -> str:
    value = float(size)
    for unit in ("B", "KB", "MB", "GB"):
        if value < 1024 or unit == "GB":
            if unit == "B":
                return f"{int(value)}{unit}"
            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.last_line_at: Optional[float] = None  # silence cue
        self.cancelled_step: Optional[int] = None
        self.rc = 0
        self.finished = False


class _GetchModes:
    """The blocking/non-blocking getch switching shared by all views."""

    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


class LanesView(_GetchModes):
    """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._console_log = _ConsoleLog(title)   # shared by both 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.last_line_at = self._now()
            lane.progress = None
            lane.progress_kind = ""
            self._console_log.line(
                f"--- [{lane.title}] {event.get('title') or ''} ---")
        elif kind == "line":
            self._ingest_lane_line(lane, event.get("text") or "")
        elif kind == "step_done":
            index = event["index"]
            rc = event.get("rc") or 0
            lane.results[index] = rc
            lane.current = None
            lane.last_line_at = None
            lane.progress = None
            lane.progress_kind = ""
            self._console_log.line(
                f"[{'OK' if rc == 0 else 'FAIL'}] [{lane.title}] "
                f"{lane.steps[index].title} (exit {rc})")
        elif kind == "step_cancelled":
            index = event["index"]
            lane.cancelled_step = index
            lane.current = None
            lane.last_line_at = None
            lane.progress = None
            lane.progress_kind = ""
            self._console_log.line(
                f"[x] [{lane.title}] {lane.steps[index].title} (cancelled)")
        elif kind == "lane_finish":
            lane.rc = event.get("rc") or 0
            lane.finished = True
            self._console_log.line(
                f"=== [{lane.title}] finished (exit {lane.rc}) ===")

    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")
        lane.last_line_at = self._now()
        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_KEEP:
            del lane.log_tail[: len(lane.log_tail) - _LOG_KEEP]
        self._console_log.line(line)

    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()
        self._console_log.line(f"=== {self.phase} ===")

    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()
                # Leave the screen blocking again: the timed redraw getch
                # must not make later hub dialogs dismiss themselves.
                self._blocking()
        finally:
            self._console_log.close()
            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

    # -- 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, strict=False):
            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()
                elapsed_text = f" {_format_elapsed(self._now() - started)}"
                _text(scr, theme, row, x + 6 + len(label) + 1,
                      elapsed_text, theme["dim"])
                silent = _silent_secs(lane.last_line_at, self._now())
                if silent is not None:
                    _text(scr, theme, row,
                          x + 6 + len(label) + 1 + len(elapsed_text),
                          f" {_silence_cue_text(silent)}", theme["dim"])
            row += 1

        row += 1
        if lane.progress is not None and not terminal:
            done, total = lane.progress
            label = _progress_label(lane.progress, lane.progress_kind)
            bar_x = x + 10
            # Reserve a space plus the label inside the right border so the
            # percentage is never clipped by (or painted onto) the pane edge.
            bar_room = max(6, w - 12 - len(label))
            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,
                  _fit(label, max(1, (x + w - 2) - (bar_x + bar_room))),
                  theme["accent"])
            row += 1

        for line in lane.log_tail:
            if row >= y + h - 1:
                break
            _text(scr, theme, row, x + 1, _fit(line, w - 3), theme["dim"])
            row += 1