feat(tui): simultaneous build and model download

6 files changed, 923 insertions, 86 deletions

diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py
index 9cb8e24..f95216e 100755
--- a/app/backends/audiocpp.py
+++ b/app/backends/audiocpp.py
@@ -891,7 +891,7 @@ def _decide_download(audiocpp_dir: Path,
         return False
     return confirm(
         "Automatically download the selected models with model_manager_v2.py "
-        "now?", False)
+        "now?", True)
 
 
 def _build_tree_families(catalog: List[dict]) -> List[dict]:
@@ -1981,17 +1981,26 @@ def _build_audiocpp_tui(emit, cancel, argv: List[str], command: str,
     return rc
 
 
-def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None:
+def _print_launch_hint(audiocpp_dir: Path, output_path: Path,
+                       pending_build: bool = False) -> None:
     """Print the exact command to start the server (or build guidance).
 
     The command is prefixed with ``cd <checkout> &&`` because the server
     discovers model_specs/<family>.json relative to its working directory.
+    PENDING_BUILD is True when the server is still building in a parallel
+    lane; instead of "build it first" remediation the hint then names the
+    command to run once that build finishes.
     """
     binary = find_audiocpp_server_bin(audiocpp_dir)
     print()
     if binary is not None:
         print("Start the server with:")
         print(f"  cd {audiocpp_dir} && {binary} --config {output_path}")
+    elif pending_build:
+        print("The server is still building in the other panel — once it "
+              "finishes, start it with:")
+        print(f"  cd {audiocpp_dir} && ./build/<platform>-<backend>-release"
+              f"/bin/audiocpp_server --config {output_path}")
     else:
         print("[INFO] audiocpp_server binary not found. Build it first, e.g.:")
         script = find_build_script(audiocpp_dir)
@@ -2002,23 +2011,30 @@ def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None:
               f"-release/bin/audiocpp_server --config {output_path}")
 
 
-def _execute_steps(settings: dict,
-                   args: argparse.Namespace) -> List[taskview.TaskStep]:
-    """Build the ordered setup steps for the in-TUI task view.
-
-    The same work ``_execute`` runs on the console, split into named steps so
-    the view can show per-step state (build / transcribe / write / download)
-    and progress. Shared results (the transcription mapping) travel through a
-    small closure dict. Each step's ``work(emit, cancel)`` returns its exit
-    code; subprocess steps stream through EMIT and abort on CANCEL, while
-    print()-based steps are captured by the view's stdout redirect.
+def _execute_lanes(settings: dict,
+                   args: argparse.Namespace,
+                   parallel: bool = False) -> List[taskview.TaskLane]:
+    """Build the ordered setup steps for the in-TUI task view, per lane.
+
+    The same work ``_execute`` runs on the console, split into two lanes so
+    the view can run the build in one pane while configuring and downloading
+    models in the other (both progress bars visible at once). The build lane
+    exists only when ``settings["build"]`` is set; the models lane always
+    exists (transcribe → write server.json → download/print commands).
+    Shared results (the transcription mapping) travel through a small closure
+    dict scoped to the models lane. Each step's ``work(emit, cancel)``
+    returns its exit code; subprocess steps stream through EMIT and abort on
+    CANCEL, while print()-based steps are captured by the view's stdout
+    routing. PARALLEL marks the launch hint as concurrent-with-build so it
+    does not claim the binary is missing while the build is still running.
     """
     audiocpp_dir = settings["audiocpp_dir"]
     state: dict = {}
-    steps: List[taskview.TaskStep] = []
+    build = settings.get("build")
+    lanes: List[taskview.TaskLane] = []
 
-    if settings.get("build"):
-        def build(emit, cancel):
+    if build:
+        def build_step(emit, cancel):
             rc = build_audiocpp(audiocpp_dir, settings["backend"],
                                 emit=emit, cancel=cancel)
             if rc != 0:
@@ -2028,8 +2044,11 @@ def _execute_steps(settings: dict,
             else:
                 print("[OK] build complete")
             return rc
-        steps.append(taskview.TaskStep(
-            f"Build audiocpp_server ({settings['backend']})", build))
+        lanes.append(taskview.TaskLane(
+            "Build",
+            [taskview.TaskStep(
+                f"Build audiocpp_server ({settings['backend']})",
+                build_step)]))
 
     def transcribe(emit, cancel):
         args.input_dir = settings["wav_dir"]
@@ -2045,7 +2064,6 @@ def _execute_steps(settings: dict,
         state["transcripts"] = transcripts
         state["write_prompt"] = write_prompt
         return 0
-    steps.append(taskview.TaskStep("Transcribe reference voices", transcribe))
 
     def write(emit, cancel):
         # Port sync (applied now that the terminal is back).
@@ -2074,17 +2092,36 @@ def _execute_steps(settings: dict,
                                         settings["sync_model_ids"])
         print_empty_transcript_warning(state["transcripts"])
         return 0
-    steps.append(taskview.TaskStep("Write server.json & sync config", write))
 
     def install(emit, cancel):
         _install_models(audiocpp_dir, settings["install_guidance"],
                         settings["download"], emit=emit, cancel=cancel)
-        _print_launch_hint(audiocpp_dir, settings["output_path"])
+        _print_launch_hint(audiocpp_dir, settings["output_path"],
+                           pending_build=bool(build and parallel))
         return 0
     install_title = "Download models" if settings.get("download") \
         else "Print model install commands"
-    steps.append(taskview.TaskStep(install_title, install))
 
+    lanes.append(taskview.TaskLane(
+        "Configure & download",
+        [taskview.TaskStep("Transcribe reference voices", transcribe),
+         taskview.TaskStep("Write server.json & sync config", write),
+         taskview.TaskStep(install_title, install)]))
+
+    return lanes
+
+
+def _execute_steps(settings: dict,
+                   args: argparse.Namespace) -> List[taskview.TaskStep]:
+    """The ordered setup steps for the sequential console path.
+
+    The lanes ``_execute_lanes`` builds, flattened into one ordered list
+    (build first, then transcribe → write → download), so the console tail
+    is byte-identical to the pre-lanes behavior.
+    """
+    steps: List[taskview.TaskStep] = []
+    for lane in _execute_lanes(settings, args):
+        steps.extend(lane.steps)
     return steps
 
 
@@ -2104,30 +2141,38 @@ def setup_screen(stdscr) -> int:
 
     The hub drives this as one screen of its own ``tui.Wizard`` stack, so
     Esc on the wizard's first screen simply returns here and the hub pops
-    back to the menu that launched it. The setup tail (build/transcribe/
-    write/download) runs inside the TUI task view on this same screen, so
+    back to the menu that launched it. The setup tail (build, transcribe,
+    write, download) runs inside the TUI task view on this same screen, so
     the hub's curses session stays intact and the user sees per-step status
-    and progress instead of being dropped to the console. Returns 0 on
-    completion, 1 when the user aborted.
+    and progress instead of being dropped to the console. On a fresh install
+    the build and the model setup run as two parallel lanes (a split view),
+    so cloning → configuring → building+downloading is one continuous,
+    one-click flow; the individual "Build" and "Download Missing Models" hub
+    actions remain only as fallbacks when something fails or is interrupted.
+    Returns 0 on completion, 1 when the user aborted.
     """
     parser = build_parser()
     args = parser.parse_args([])
     settings = _wizard(stdscr, args, parser)
     if settings is None:
         return 1
-    return taskview.run_steps(stdscr, "Setting up audio.cpp",
-                              _execute_steps(settings, args))
+    return taskview.run_lanes(stdscr, "Setting up audio.cpp",
+                              _execute_lanes(settings, args, parallel=True))
 
 
 def build_screen(stdscr) -> int:
     """Build audiocpp_server from the hub when the checkout has no binary.
 
     Asks which backend to build for (pre-selecting the backend an existing
-    server.json records, else cuda), runs the build inside the TUI task view,
-    then updates server.json's ``backend`` field to match. Returns 0 on
-    success, non-zero when the user backed out, cancelled, or the build
-    failed. This is the hub's "Build audio.cpp server" action, so a checkout
-    that was cloned but never built is always buildable from the TUI.
+    server.json records, else cuda), runs the build inside the TUI task view
+    — alongside a download of any missing models when server.json is already
+    configured and those models map to an install command (the split view),
+    or just the build otherwise — then updates server.json's ``backend``
+    field to match. Returns 0 on success, non-zero when the user backed out,
+    cancelled, or the build failed. This is the hub's "Build audio.cpp
+    server" action, so a checkout that was cloned but never built is always
+    buildable from the TUI; the standalone "Download Missing Models" action
+    stays as the fallback when the download fails or is interrupted.
     """
     checkout = find_local_checkout()
     if checkout is None:
@@ -2148,11 +2193,32 @@ def build_screen(stdscr) -> int:
         "for?", options, default_index=default, back_value=_GO_BACK)
     if backend is _GO_BACK:
         return 1
-    rc = taskview.run_steps(stdscr, "Build audiocpp_server", [
-        taskview.TaskStep(
-            f"Build audiocpp_server ({backend})",
-            lambda emit, cancel: build_audiocpp(
-                checkout, backend, emit=emit, cancel=cancel))])
+
+    def build_step(emit, cancel):
+        return build_audiocpp(checkout, backend, emit=emit, cancel=cancel)
+
+    lanes = [taskview.TaskLane(
+        "Build", [taskview.TaskStep(
+            f"Build audiocpp_server ({backend})", build_step)])]
+
+    # Missing models this build can also fetch, so a configured backend that
+    # lost its binary is restored to "installed" in one step.
+    server_json = checkout / "server.json"
+    missing = missing_model_entries(server_json) if server_json.exists() else []
+    guidance = missing_model_install_guidance(checkout, missing) \
+        if missing else []
+
+    if guidance:
+        def download_step(emit, cancel):
+            install_models(checkout, guidance, emit=emit, cancel=cancel)
+            return 0
+        lanes.append(taskview.TaskLane(
+            "Download models",
+            [taskview.TaskStep("Download missing models", download_step)]))
+
+    title = "Build & download models" if len(lanes) == 2 \
+        else "Build audiocpp_server"
+    rc = taskview.run_lanes(stdscr, title, lanes)
     if rc != 0:
         return rc
     if update_server_backend(backend):
@@ -2161,6 +2227,10 @@ def build_screen(stdscr) -> int:
         tui.flash(stdscr, f"audiocpp_server built for {backend}. (Could not "
                   "update server.json's backend field — reconfigure audio.cpp "
                   "if it was already configured.)", "warn")
+    # Models that can't be mapped to an install command still need hand
+    # installation; say so now rather than leaving the user in the dark.
+    if missing and not guidance:
+        tui.flash(stdscr, hand_install_guidance(checkout, missing), "err")
     return 0
 
 
diff --git a/app/docs/backend-audiocpp.md b/app/docs/backend-audiocpp.md
index 96c6644..c9ef883 100644
--- a/app/docs/backend-audiocpp.md
+++ b/app/docs/backend-audiocpp.md
@@ -4,7 +4,7 @@
 
 The easiest way is the TUI: run `python audiobook.py`, choose **Configure backends… → Install Backend → audio.cpp**, and it clones `audio.cpp` into `app/audio.cpp` (or reuses an existing checkout), builds `audiocpp_server`, lets you pick model families/packages from an expandable checkbox tree (reading the checkout's `model_specs/`), transcribes `.wav` voices with `whisper`, writes `server.json` into the checkout, syncs `app/converter/config.py`, and prints the launch command (the hub can also start the server for you via the **Start/Stop Backend Servers** menu or automatically when converting). The clone, build, transcription and model downloads all run inside the TUI — each shows a status (and, where the tool can measure it, a progress bar), and can be cancelled — instead of dropping to console output. Run it directly with `python app/backends/audiocpp.py` (flags like `--wavs`, `--families`, `--build-backend`, `--clone` skip the corresponding screens for scripting). The TUI runs in the managed `app/envs/tts` venv, which includes `whisper` via `requirements.txt`; for a manual setup, make sure `whisper` (or `faster_whisper`) is installed in the environment you run the wizard from. The Qwen3-TTS model tree also offers hosting the VoiceDesign package as a `vdes` entry.
 
-The hub's backend status table distinguishes how far audio.cpp is set up: `unavailable` (nothing present), `downloaded (not built)` (checkout cloned, `audiocpp_server` not built), `built (not configured)` (binary built, no `server.json`), `installed` (ready; or `installed (models missing)` when the config references undownloaded models), and `running` once its server answers. Whenever the checkout exists but `audiocpp_server` is missing, **Configure backends… → Build audio.cpp server** builds it from the TUI (the wizard offers the build during setup too), so a backend whose build you skipped is never stuck as "unavailable". The setup steps run in order — build, then configure, then download models — so **Build audio.cpp server** and **Download Missing Models (audio.cpp)** are never offered at the same time; the model download appears only once the server binary is built.
+The hub's backend status table distinguishes how far audio.cpp is set up: `unavailable` (nothing present), `downloaded (not built)` (checkout cloned, `audiocpp_server` not built), `built (not configured)` (binary built, no `server.json`), `installed` (ready; or `installed (models missing)` when the config references undownloaded models), and `running` once its server answers. Whenever the checkout exists but `audiocpp_server` is missing, **Configure backends… → Build audio.cpp server** builds it from the TUI (the wizard offers the build during setup too), so a backend whose build you skipped is never stuck as "unavailable". On a fresh install the setup is one continuous flow: clone → configure → and then the build and the model downloads run **simultaneously** in a split view (half building, half downloading). The setup steps are therefore ordered build > configure > download, and **Build audio.cpp server** and **Download Missing Models (audio.cpp)** are never offered at the same time; **Build audio.cpp server** downloads any missing models alongside the build, and **Download Missing Models (audio.cpp)** remains only as a fallback for when a download fails or is interrupted.
 
 If you prefer to install the backend yourself (in your own environment, not the managed venv), the manual steps are below. Either way the hub detects a running server by its port, so a manually-installed backend works once its server is up.
 
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 8f79085..2f41131 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -649,6 +649,11 @@ class InstallModelsTests(unittest.TestCase):
         self.assertTrue(make_server._decide_download(self.checkout, confirm))
         confirm.assert_called_once()
 
+    def test_decide_download_defaults_to_yes(self):
+        confirm = MagicMock(return_value=True)
+        make_server._decide_download(self.checkout, confirm)
+        self.assertIs(confirm.call_args[0][1], True)
+
 
 class TranscribeWavDirTests(unittest.TestCase):
     def setUp(self):
@@ -1793,27 +1798,132 @@ if __name__ == "__main__":
 
 class SetupScreenTests(unittest.TestCase):
     """setup_screen: the wizard run on the hub's screen, setup tail via the
-    in-TUI task view."""
+    in-TUI task view (two parallel lanes on a fresh install)."""
 
     def test_abort_returns_one_without_executing(self):
         with patch.object(make_server, "_wizard", return_value=None) as mk_wizard, \
-                patch.object(make_server, "_execute_steps") as mk_steps:
+                patch.object(make_server, "_execute_lanes") as mk_lanes:
             rc = make_server.setup_screen(None)
         self.assertEqual(rc, 1)
         mk_wizard.assert_called_once()
-        mk_steps.assert_not_called()
+        mk_lanes.assert_not_called()
 
     def test_success_runs_the_tail_in_the_task_view(self):
         settings = {"audiocpp_dir": Path("/x")}
-        steps = [make_server.taskview.TaskStep("t", lambda emit, cancel: 0)]
+        lanes = [make_server.taskview.TaskLane(
+            "Build", [make_server.taskview.TaskStep("t", lambda emit, cancel: 0)])]
         with patch.object(make_server, "_wizard", return_value=settings), \
-                patch.object(make_server, "_execute_steps",
-                             return_value=steps) as mk_steps, \
-                patch.object(make_server.taskview, "run_steps",
+                patch.object(make_server, "_execute_lanes",
+                             return_value=lanes) as mk_lanes, \
+                patch.object(make_server.taskview, "run_lanes",
                              return_value=0) as mk_run:
             rc = make_server.setup_screen(None)
         self.assertEqual(rc, 0)
-        mk_steps.assert_called_once()
-        self.assertIs(mk_steps.call_args[0][0], settings)
+        mk_lanes.assert_called_once()
+        self.assertIs(mk_lanes.call_args[0][0], settings)
+        self.assertTrue(mk_lanes.call_args[1]["parallel"])
         mk_run.assert_called_once()
-        self.assertEqual(mk_run.call_args[0][2], steps)
+        self.assertEqual(mk_run.call_args[0][2], lanes)
+
+
+class ExecuteLanesTests(unittest.TestCase):
+    """_execute_lanes: two lanes (build + configure/download) and the
+    flattened console order."""
+
+    def _settings(self, **overrides):
+        settings = {
+            "audiocpp_dir": Path("/x"),
+            "backend": "cuda",
+            "build": True,
+            "download": True,
+            "include_clone": False,
+            "wav_dir": None,
+            "plan": None,
+            "sync_port": None,
+            "sync_model_ids": None,
+            "delete_unused": False,
+            "unused_entries": [],
+            "model_entries": [],
+            "entry_ids": [],
+            "install_guidance": [],
+            "output_path": Path("/x/server.json"),
+            "host": "127.0.0.1",
+            "port": 8080,
+            "lazy_load": True,
+        }
+        settings.update(overrides)
+        return settings
+
+    def test_two_lanes_when_building(self):
+        args = make_server.build_parser().parse_args([])
+        lanes = make_server._execute_lanes(self._settings(), args)
+        self.assertEqual([lane.title for lane in lanes],
+                         ["Build", "Configure & download"])
+        self.assertEqual([s.title for s in lanes[0].steps],
+                         ["Build audiocpp_server (cuda)"])
+        self.assertEqual([s.title for s in lanes[1].steps],
+                         ["Transcribe reference voices",
+                          "Write server.json & sync config",
+                          "Download models"])
+
+    def test_single_lane_when_not_building(self):
+        args = make_server.build_parser().parse_args([])
+        lanes = make_server._execute_lanes(
+            self._settings(build=False), args)
+        self.assertEqual([lane.title for lane in lanes],
+                         ["Configure & download"])
+
+    def test_flattened_console_steps_keep_the_build_first(self):
+        args = make_server.build_parser().parse_args([])
+        steps = make_server._execute_steps(self._settings(), args)
+        self.assertEqual([s.title for s in steps],
+                         ["Build audiocpp_server (cuda)",
+                          "Transcribe reference voices",
+                          "Write server.json & sync config",
+                          "Download models"])
+
+    def test_download_step_prints_the_parallel_launch_hint(self):
+        args = make_server.build_parser().parse_args([])
+        lanes = make_server._execute_lanes(self._settings(), args,
+                                           parallel=True)
+        install_step = lanes[1].steps[2]
+        with patch.object(make_server, "_install_models") as mk_install, \
+                patch.object(make_server, "_print_launch_hint") as mk_hint:
+            install_step.work(lambda line: None, threading.Event())
+        mk_hint.assert_called_once()
+        self.assertTrue(mk_hint.call_args[1]["pending_build"])
+
+    def test_download_step_console_hint_is_not_pending(self):
+        args = make_server.build_parser().parse_args([])
+        lanes = make_server._execute_lanes(self._settings(), args)
+        install_step = lanes[1].steps[2]
+        with patch.object(make_server, "_install_models") as mk_install, \
+                patch.object(make_server, "_print_launch_hint") as mk_hint:
+            install_step.work(lambda line: None, threading.Event())
+        mk_hint.assert_called_once()
+        self.assertFalse(mk_hint.call_args[1]["pending_build"])
+
+
+class LaunchHintTests(unittest.TestCase):
+    """_print_launch_hint: exact command vs. the pending-build message."""
+
+    def _capture(self, audiocpp_dir, output_path, pending_build=False):
+        buf = io.StringIO()
+        with redirect_stdout(buf), \
+                patch.object(make_server, "find_audiocpp_server_bin",
+                             return_value=None):
+            make_server._print_launch_hint(audiocpp_dir, output_path,
+                                           pending_build=pending_build)
+        return buf.getvalue()
+
+    def test_pending_build_names_the_post_build_command(self):
+        out = self._capture(Path("/tmp/acpp"), Path("/tmp/acpp/server.json"),
+                            pending_build=True)
+        self.assertIn("still building", out)
+        self.assertNotIn("Build it first", out)
+        self.assertIn("audiocpp_server --config /tmp/acpp/server.json", out)
+
+    def test_missing_binary_gives_build_remediation(self):
+        out = self._capture(Path("/tmp/acpp"), Path("/tmp/acpp/server.json"))
+        self.assertIn("Build it first", out)
+        self.assertNotIn("still building", out)
diff --git a/app/tests/test_taskview.py b/app/tests/test_taskview.py
index 15fc501..d1bc658 100644
--- a/app/tests/test_taskview.py
+++ b/app/tests/test_taskview.py
@@ -7,8 +7,11 @@ the progress-line parsing through ``TaskView._ingest_line``, and state
 transitions through ``handle_event`` + ``_step_mark`` + ``_result_rc``.
 """
 
+import io
 import sys
+import threading
 import unittest
+from queue import Empty
 from unittest.mock import patch
 
 from tests.test_tui import FakeCurses, FakeScreen
@@ -227,5 +230,212 @@ class LabelTests(unittest.TestCase):
                          "45%")
 
 
+class RunLanesTests(_FakeTui, unittest.TestCase):
+    """run_lanes: one lane falls back to run_steps, two use the split view."""
+
+    def test_single_lane_delegates_to_run_steps(self):
+        lane = taskview.TaskLane("Build", [_step("one")])
+        with patch.object(taskview, "run_steps", return_value=0) as mk_run:
+            rc = taskview.run_lanes(None, "Setup", [lane])
+        self.assertEqual(rc, 0)
+        mk_run.assert_called_once()
+        self.assertEqual(mk_run.call_args[0][2], lane.steps)
+
+    def test_empty_lanes_are_dropped(self):
+        lane = taskview.TaskLane("Build", [_step("one")])
+        with patch.object(taskview, "run_steps", return_value=0) as mk_run:
+            rc = taskview.run_lanes(
+                None, "Setup",
+                [taskview.TaskLane("Empty", []), lane])
+        self.assertEqual(rc, 0)
+        self.assertEqual(mk_run.call_args[0][2], lane.steps)
+
+    def test_no_lanes_returns_zero_without_running(self):
+        with patch.object(taskview, "run_steps") as mk_run, \
+                patch.object(taskview, "LanesView") as mk_view:
+            rc = taskview.run_lanes(None, "Setup", [])
+        self.assertEqual(rc, 0)
+        mk_run.assert_not_called()
+        mk_view.assert_not_called()
+
+    def test_two_lanes_uses_the_split_view(self):
+        lanes = [taskview.TaskLane("Build", [_step("one")]),
+                 taskview.TaskLane("Download", [_step("two")])]
+        with patch.object(taskview, "LanesView") as mk_view:
+            mk_view.return_value.run.return_value = 0
+            rc = taskview.run_lanes(None, "Setup", lanes)
+        self.assertEqual(rc, 0)
+        mk_view.assert_called_once_with(None, "Setup", lanes)
+        mk_view.return_value.run.assert_called_once_with()
+
+
+class ThreadRouterTests(unittest.TestCase):
+    def test_routes_to_registered_thread_and_falls_back(self):
+        fallback = io.StringIO()
+        router = taskview._ThreadRouter(fallback)
+        captured = []
+        writer = taskview._LineWriter(captured.append)
+        with router.for_thread(writer):
+            router.write("hello\n")
+        self.assertEqual(captured, ["hello"])
+        # Unregistered thread falls through to the original stream.
+        router.write("fallback\n")
+        self.assertEqual(fallback.getvalue(), "fallback\n")
+
+    def test_concurrent_prints_land_in_their_own_writer(self):
+        registry = {}
+        router = taskview._ThreadRouter(io.StringIO(), registry)
+        lines_a, lines_b = [], []
+        wa = taskview._LineWriter(lines_a.append)
+        wb = taskview._LineWriter(lines_b.append)
+        ready_a, ready_b = threading.Event(), threading.Event()
+        go = threading.Event()
+
+        def body(writer, tag, ready):
+            with router.for_thread(writer):
+                ready.set()
+                go.wait()
+                for i in range(50):
+                    print(f"{tag}-{i}")
+
+        with patch.object(sys, "stdout", router):
+            threads = [threading.Thread(target=body,
+                                        args=(wa, "A", ready_a)),
+                       threading.Thread(target=body,
+                                        args=(wb, "B", ready_b))]
+            for t in threads:
+                t.start()
+            ready_a.wait()
+            ready_b.wait()
+            go.set()
+            for t in threads:
+                t.join()
+        self.assertTrue(lines_a)
+        self.assertTrue(lines_b)
+        self.assertTrue(all(line.startswith("A-") for line in lines_a))
+        self.assertTrue(all(line.startswith("B-") for line in lines_b))
+
+
+class LanesViewTests(_FakeTui, unittest.TestCase):
+    """LanesView: parallel lanes, isolated logs/progress, split rendering."""
+
+    def make_view(self, lanes, width=80, height=24):
+        screen = FakeScreen(width=width, height=height)
+        view = taskview.LanesView(screen, "Setup", lanes,
+                                  clock=lambda: 1000.0)
+        return view, screen
+
+    def _two_lanes(self):
+        return [taskview.TaskLane("Build", [_step("one"), _step("two")]),
+                taskview.TaskLane("Download models", [_step("dl")])]
+
+    def test_two_lane_workers_run_both_steps(self):
+        ran = []
+
+        def work(name):
+            def _w(emit, cancel):
+                ran.append(name)
+                return 0
+            return _w
+
+        lanes = [taskview.TaskLane("A", [taskview.TaskStep("a", work("A"))]),
+                 taskview.TaskLane("B", [taskview.TaskStep("b", work("B"))])]
+        view, _ = self.make_view(lanes)
+        registry = {}
+        router = taskview._ThreadRouter(sys.stdout, registry)
+        threads = [threading.Thread(target=view._lane_worker,
+                                    args=(lane_state, router, view._cancel))
+                   for lane_state in view._lanes]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        view._drain()
+        self.assertCountEqual(ran, ["A", "B"])
+        self.assertTrue(all(lane.finished for lane in view._lanes))
+
+    def test_lane_worker_routes_prints_to_its_lane(self):
+        def work(emit, cancel):
+            print("lane-A-log")
+            return 0
+
+        lanes = [taskview.TaskLane("A", [taskview.TaskStep("a", work)])]
+        view, _ = self.make_view(lanes)
+        registry = {}
+        router = taskview._ThreadRouter(io.StringIO(), registry)
+        lane_state = view._lanes[0]
+        with patch.object(sys, "stdout", router):
+            view._lane_worker(lane_state, router, view._cancel)
+        events = []
+        while True:
+            try:
+                events.append(lane_state.queue.get_nowait())
+            except Empty:
+                break
+        lines = [e["text"] for e in events if e.get("kind") == "line"]
+        self.assertIn("lane-A-log", lines)
+        self.assertTrue(any(e.get("kind") == "lane_finish" for e in events))
+
+    def test_progress_is_isolated_per_lane(self):
+        view, _ = self.make_view(self._two_lanes())
+        view._ingest_lane_line(view._lanes[0],
+                               "AUDIOCPP_PROGRESS downloaded=512 total=2048")
+        self.assertEqual(view._lanes[0].progress, (512, 2048))
+        self.assertIsNone(view._lanes[1].progress)
+        view._ingest_lane_line(view._lanes[1], "[ 45%] building")
+        self.assertEqual(view._lanes[1].progress, (45, 100))
+        self.assertEqual(view._lanes[0].progress_kind, "bytes")
+
+    def test_result_rc_returns_first_nonzero_across_lanes(self):
+        view, _ = self.make_view(self._two_lanes())
+        # Lane 0 step 0 succeeds, lane 0 step 1 fails, lane 1 succeeds.
+        view._lanes[0].results = [0, 7]
+        view._lanes[1].results = [0]
+        self.assertEqual(view._result_rc(), 7)
+
+    def test_all_lanes_finished_flips_phase_to_done(self):
+        view, _ = self.make_view(self._two_lanes())
+        for lane in view._lanes:
+            lane.finished = True
+            lane.rc = 0
+        view._drain()
+        self.assertEqual(view.phase, "done")
+        self.assertEqual(view._result_rc(), 0)
+
+    def test_error_phase_when_a_lane_reports_failure(self):
+        view, _ = self.make_view(self._two_lanes())
+        view._lanes[0].finished = True
+        view._lanes[0].rc = 3
+        view._lanes[1].finished = True
+        view._lanes[1].rc = 0
+        view._drain()
+        self.assertEqual(view.phase, "error")
+
+    def test_cancelled_run_returns_nonzero(self):
+        view, _ = self.make_view(self._two_lanes())
+        view._cancel.set()
+        for lane in view._lanes:
+            lane.finished = True
+        view._drain()
+        self.assertEqual(view.phase, "cancelled")
+        self.assertTrue(view.cancelled)
+        self.assertEqual(view._result_rc(), 1)
+
+    def test_split_render_draws_both_lane_titles(self):
+        view, screen = self.make_view(self._two_lanes())
+        view.render()
+        text = " ".join(t for _, _, t, _ in screen.strings)
+        self.assertIn("Build", text)
+        self.assertIn("Download models", text)
+        self.assertIn("Esc or q: cancel", text)
+
+    def test_split_render_stacks_on_narrow_terminal(self):
+        view, screen = self.make_view(self._two_lanes(), width=60)
+        view.render()
+        text = " ".join(t for _, _, t, _ in screen.strings)
+        self.assertIn("Build", text)
+        self.assertIn("Download models", text)
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/app/ui/hub.py b/app/ui/hub.py
index 247dd49..6862b0a 100644
--- a/app/ui/hub.py
+++ b/app/ui/hub.py
@@ -150,7 +150,10 @@ class _Hub:
         populated from the detected statuses: configure each installed
         backend, install (backends with nothing on disk), and uninstall.
         Selecting one pushes the next screen; Esc pops back to the main
-        menu.
+        menu. The Build action downloads any missing models alongside the
+        build (a split view), so it heals a configured-but-unbuilt backend
+        in one step; Download Missing Models stays as the fallback for when
+        a download fails or is interrupted.
         """
         while True:
             statuses = detect_all()
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