18 files changed, 2161 insertions, 289 deletions

diff --git a/README.md b/README.md
index 1563303..05ca145 100644
--- a/README.md
+++ b/README.md
@@ -48,10 +48,11 @@ python audiobook.py
 
 A full-screen TUI opens and shows each backend's status in a table — **unavailable** (red, name dimmed: not installed and no server running), **installed** (orange), or **running** (green) with the source(s) in brackets: `[local]` for a server this tool started, `[remote]` for an externally-run server found by probing the backend's remote URL, or `[local, remote]` when both are up. From the menu you can:
 
-- **Convert books…** — set everything on one screen. The first field picks the **Backend**: each backend appears as a managed entry (e.g. `audio.cpp`) when it's installed and configured here, plus a `[remote]` entry (e.g. `audio.cpp [remote]`) when a running server was found at its remote URL. The rest of the options change to what that backend supports: model, voice and instructions for audio.cpp; speaker or clone .wav for qwen; voice for faster — plus output format, speed, whether to combine all chapters into one file, and debug mode. A managed entry reads its local `server.json` / `voices.json`; a `[remote]` entry queries the server itself instead (audio.cpp lists its models and voices over HTTP, faster asks you to type a voice name). Focus starts on **Generate!**, so Enter accepts the defaults. The "combine chapters" option is hidden for `m4b`, which is always one file. For a managed entry whose server isn't running it's started automatically; after the conversion you're asked whether to stop it (remote entries never autostart — that server is not this tool's to manage), or
+- **Convert books…** — set everything on one screen. The first field picks the **Backend**: each backend appears as a managed entry (e.g. `audio.cpp`) when it's installed and configured here, plus a `[remote]` entry (e.g. `audio.cpp [remote]`) when a running server was found at its remote URL. The rest of the options change to what that backend supports: model, voice and instructions for audio.cpp; speaker or clone .wav for qwen; voice for faster — plus output format, speed, whether to combine all chapters into one file, and debug mode. A managed entry reads its local `server.json` / `voices.json`; a `[remote]` entry queries the server itself instead (audio.cpp lists its models and voices over HTTP, faster asks you to type a voice name). Focus starts on **Generate!**, so Enter accepts the defaults. The "combine chapters" option is hidden for `m4b`, which is always one file.
+- After **Generate!**, a full-screen run view takes over instead of dumping you into console output. The top shows the server status — *starting* (a managed server that needed booting is spawned and waited on until it actually answers HTTP, not just accepts TCP connections), *ready*, *processing*, or *error* — and the bottom shows the conversion with a progress bar for the current book's chunks (`Chunk 45/120`) and elapsed time. Esc or `q` first asks whether to cancel processing, then (when this run started the server) whether to shut it down, then returns to the menu. An error — the server exits while booting, stops mid-conversion, or a chunk fails and the book aborts — switches the corresponding state to *error* and waits for a key before returning to the menu, so the failure is never scrolled away (full detail stays in `app/logs/audiobook_<date>.log`), or
 - **Set up a backend…** — clone, build, and configure a backend end-to-end (audio.cpp, qwen, faster), or
 - **Configure a backend…** — regenerate its config (a new `server.json`, rebuild `voices.json`, change ports/speaker), or
-- **Server…** — manually start or stop a configured backend's server (the hub spawns it in the managed venv and polls until it accepts connections).
+- **Server…** — manually start or stop a configured backend's server (the hub spawns it in the managed venv and polls until it answers).
 
 **Configure a backend…** and **Server…** only appear once at least one backend is installed — a merely-running external server unlocks **Convert books…**, but configuring it or starting/stopping its server needs it on this machine.
 
@@ -104,6 +105,16 @@ Transcription affects the output a lot. Whisper does not always give perfect tra
 
 Even tiny amounts of pause between phrases in the sample audio can have a big impact. Try increasing or decreasing them or find a sample with different cadence.
 
+### audio.cpp `model contract spec not found for family '...'`
+
+The audio.cpp server discovers `model_specs/<family>.json` relative to its **process working directory**, so it must be started from the audio.cpp checkout. The hub starts it that way automatically, and the launch hint it prints is prefixed with `cd <checkout> &&`. If you start `audiocpp_server` by hand, run it from the checkout root:
+
+```bash
+cd app/audio.cpp && ./build/<platform>-<backend>-release/bin/audiocpp_server --config server.json
+```
+
+If the error instead mentions a model path that does not exist, the model package was never downloaded — the hub's status table shows `installed (models missing)` for that case. Install it from the checkout (the exact command is in the convert-menu warning), e.g. `python tools/model_manager_v2.py install qwen3_tts_0_6b_base_q8_0`.
+
 ## License
 
 MIT
diff --git a/app/backends/__init__.py b/app/backends/__init__.py
index 9e8bf31..488c36e 100644
--- a/app/backends/__init__.py
+++ b/app/backends/__init__.py
@@ -24,6 +24,7 @@ automatically.
 
 import shlex
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Callable, Dict, List, Optional
 
 
@@ -35,10 +36,24 @@ class ServerSpec:
     on separate ports). ARGV is the exact command line the hub spawns (using
     the managed venv's absolute binaries, so no shell activation is needed);
     URL is the endpoint ``common.server_running`` probes to decide readiness.
+
+    CWD is the working directory the server is spawned in. It matters for
+    servers that discover resources relative to their process working
+    directory (audio.cpp resolves ``model_specs/<family>.json`` by walking up
+    from its cwd), so the hub starts them from their checkout root. None
+    inherits the hub's cwd (today's behavior).
+
+    IDENTITY is the ``backends.probe.IDENTITY_*`` constant the server is
+    expected to answer as once it is truly ready. When set, ``servers.start``
+    waits for the server to answer HTTP with that identity — not merely to
+    accept TCP connections — so "listening but still starting" servers are
+    caught. None keeps the plain TCP-connect readiness check.
     """
     name: str
     url: str
     argv: List[str]
+    cwd: Optional[Path] = None
+    identity: Optional[str] = None
 
 
 @dataclass
@@ -72,6 +87,12 @@ class BackendStatus:
     RUNNING_MODELS names which of a multi-server backend's models answered
     (qwen: "Base" and/or "CustomVoice", local and remote combined), shown in
     parentheses in the hub's status table.
+
+    MODELS_MISSING says the server config references model files that are not
+    on disk (e.g. an audio.cpp server.json entry whose ``path`` was never
+    downloaded); DETAILS then names them. The backend still counts as ready
+    (the hub surfaces the warning), but a conversion would fail until the
+    models are installed.
     """
     key: str
     label: str
@@ -86,6 +107,7 @@ class BackendStatus:
     remote: bool = False
     remote_urls: Dict[str, str] = field(default_factory=dict)
     remote_models: List[str] = field(default_factory=list)
+    models_missing: bool = False
 
     @property
     def ready(self) -> bool:
@@ -94,8 +116,18 @@ class BackendStatus:
 
 
 def format_launch_hint(servers: List[ServerSpec]) -> str:
-    """Join a backend's server argvs into a copy-pasteable launch hint."""
-    return " ; ".join(shlex.join(s.argv) for s in servers)
+    """Join a backend's server argvs into a copy-pasteable launch hint.
+
+    A server with a CWD is prefixed with ``cd <cwd> &&`` so the hint works
+    pasted into a shell (the server relies on that working directory).
+    """
+    parts = []
+    for server in servers:
+        command = shlex.join(server.argv)
+        if server.cwd is not None:
+            command = f"cd {shlex.quote(str(server.cwd))} && {command}"
+        parts.append(command)
+    return " ; ".join(parts)
 
 
 @dataclass
diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py
index ee0ee21..a502c8f 100755
--- a/app/backends/audiocpp.py
+++ b/app/backends/audiocpp.py
@@ -1276,6 +1276,78 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
             }
 
 
+def _model_path_present(path: Path) -> bool:
+    """True when a server.json model path holds actual model files.
+
+    A present path is either a file (a single-model package) or a non-empty
+    directory (the usual GGUF package target directory; an empty one means a
+    download that never ran or was cleaned up halfway).
+    """
+    try:
+        if path.is_file():
+            return True
+        if path.is_dir():
+            return any(path.iterdir())
+    except OSError:
+        return False
+    return False
+
+
+def missing_model_entries(server_json: Path) -> List[dict]:
+    """Return the server.json model entries whose files are not on disk.
+
+    Paths resolve exactly like audiocpp_server resolves them (relative paths
+    against the server.json's directory). Each returned entry carries the
+    entry ``id`` and ``rel`` (the configured path string); used by ``detect``
+    to warn that a conversion would fail until the models are installed.
+    """
+    try:
+        data = json.loads(server_json.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return []
+    if not isinstance(data, dict):
+        return []
+    base = server_json.parent
+    missing: List[dict] = []
+    for entry in data.get("models") or []:
+        if not isinstance(entry, dict):
+            continue
+        rel = entry.get("path")
+        if not isinstance(rel, str) or not rel:
+            continue
+        path = Path(rel) if Path(rel).is_absolute() else base / rel
+        if _model_path_present(path):
+            continue
+        missing.append({"id": str(entry.get("id") or rel), "rel": rel})
+    return missing
+
+
+def model_install_hints(audiocpp_dir: Path,
+                        missing: List[dict]) -> List[str]:
+    """Remediation lines for MISSING model entries (see missing_model_entries).
+
+    Maps each entry's configured path back to the catalog package that
+    installs it (``models/<target_directory>`` -> install id) so the line
+    carries the exact ``model_manager_v2.py install`` command; entries whose
+    directory matches no catalog package just name the path.
+    """
+    by_path: Dict[str, str] = {}
+    try:
+        for entry in load_model_catalog(audiocpp_dir):
+            by_path[entry["default_path"]] = entry["install_id"]
+    except (NotADirectoryError, OSError):
+        pass
+    hints: List[str] = []
+    for item in missing:
+        install_id = by_path.get(item["rel"])
+        hint = f"model not downloaded: {item['id']} ({item['rel']})"
+        if install_id:
+            hint += (f" — install with: python tools/model_manager_v2.py "
+                     f"install {install_id}")
+        hints.append(hint)
+    return hints
+
+
 def find_local_checkout() -> Optional[Path]:
     """Best-effort location of an audio.cpp checkout with model_specs.
 
@@ -1421,20 +1493,24 @@ def build_audiocpp(audiocpp_dir: Path, backend: str) -> int:
 
 
 def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None:
-    """Print the exact command to start the server (or build guidance)."""
+    """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.
+    """
     binary = find_audiocpp_server_bin(audiocpp_dir)
     print()
     if binary is not None:
         print("Start the server with:")
-        print(f"  {binary} --config {output_path}")
+        print(f"  cd {audiocpp_dir} && {binary} --config {output_path}")
     else:
         print("[INFO] audiocpp_server binary not found. Build it first, e.g.:")
         script = find_build_script(audiocpp_dir)
         if script is not None:
             print(f"  sh {script} --backend <cuda|vulkan|hip|cpu> "
                   "--target audiocpp_server")
-        print(f"  then run: ./build/<platform>-<backend>-release/bin/"
-              f"audiocpp_server --config {output_path}")
+        print(f"  then run: cd {audiocpp_dir} && ./build/<platform>-<backend>"
+              f"-release/bin/audiocpp_server --config {output_path}")
 
 
 def _execute(settings: dict, args: argparse.Namespace) -> int:
@@ -1767,15 +1843,23 @@ def detect() -> BackendStatus:
     server_json = checkout / "server.json"
     configured = server_json.exists()
     specs: List[ServerSpec] = []
+    missing = missing_model_entries(server_json) if configured else []
     if configured:
         details.append(f"config: {server_json}")
+        if missing:
+            # The config references model files that are not on disk; a
+            # conversion would fail at model-load time, so say so now.
+            details.extend(model_install_hints(checkout, missing))
         if built:
+            # Spawned from the checkout: audiocpp_server discovers
+            # model_specs/<family>.json relative to its working directory.
             specs = [ServerSpec(
                 "audiocpp", config.AUDIOCPP_API_URL,
-                [str(binary), "--config", str(server_json)])]
+                [str(binary), "--config", str(server_json)],
+                cwd=checkout, identity=probe.IDENTITY_AUDIOCPP)]
         else:
-            launch = (f"./build/<platform>-<backend>-release/bin/"
-                      f"audiocpp_server --config {server_json}")
+            launch = (f"cd {checkout} && ./build/<platform>-<backend>-release"
+                      f"/bin/audiocpp_server --config {server_json}")
     else:
         details.append("no server.json — run setup to configure models")
     if specs:
@@ -1787,7 +1871,8 @@ def detect() -> BackendStatus:
                          running=managed or remote_running,
                          details=details, launch_hint=launch,
                          servers=specs, managed=managed,
-                         remote=remote_running, remote_urls=remote_urls)
+                         remote=remote_running, remote_urls=remote_urls,
+                         models_missing=bool(missing))
 
 
 def _detect_remote(managed: bool = False) -> Tuple[bool, dict]:
diff --git a/app/backends/faster.py b/app/backends/faster.py
index 3dc30dd..e1249ca 100755
--- a/app/backends/faster.py
+++ b/app/backends/faster.py
@@ -362,7 +362,10 @@ def detect() -> BackendStatus:
         argv = [str(envs.env_python()),
                 str(_checkout() / "examples" / "openai_server.py"),
                 "--voices", str(voices_json), "--port", str(_config_port())]
-        specs = [ServerSpec("faster", config.FASTER_API_URL, argv)]
+        # identity: /health must report model_loaded before the server is
+        # really usable (the model loads after the port opens).
+        specs = [ServerSpec("faster", config.FASTER_API_URL, argv,
+                            identity=probe.IDENTITY_FASTER)]
         launch = format_launch_hint(specs)
     managed = servers.manages(specs)
     remote_running, remote_urls = _detect_remote(managed)
diff --git a/app/backends/probe.py b/app/backends/probe.py
index ebef86e..efa2963 100644
--- a/app/backends/probe.py
+++ b/app/backends/probe.py
@@ -126,3 +126,26 @@ def same_endpoint(url_a: str, url_b: str) -> bool:
     port_a = a.port or (443 if (a.scheme or "http") == "https" else 80)
     port_b = b.port or (443 if (b.scheme or "http") == "https" else 80)
     return host_a == host_b and port_a == port_b
+
+
+def health_payload(url: str, timeout: float = DEFAULT_TIMEOUT) -> Optional[dict]:
+    """Return the server's ``/health`` JSON document, or None.
+
+    A cheaper, raw check than ``identify_server``: used by the run view's
+    background poll to tell "process alive" from "server answering" without
+    probing every identity endpoint.
+    """
+    if not url:
+        return None
+    return _get_json(f"{url.rstrip('/')}/health", timeout)
+
+
+def faster_model_loaded(url: str, timeout: float = DEFAULT_TIMEOUT) -> bool:
+    """True when a faster server at URL reports its model loaded.
+
+    faster's ``/health`` answers with a ``model_loaded`` flag only after the
+    weights are resident, so this is the "truly ready" signal used while
+    waiting for a started server to become usable.
+    """
+    payload = health_payload(url, timeout)
+    return bool(payload and payload.get("model_loaded"))
diff --git a/app/backends/qwen.py b/app/backends/qwen.py
index 160c0f1..7f821fa 100644
--- a/app/backends/qwen.py
+++ b/app/backends/qwen.py
@@ -227,10 +227,12 @@ def detect() -> BackendStatus:
     specs = [
         ServerSpec("qwen-custom", config.QWEN_API_URL,
                    [demo, QWEN_CUSTOMVOICE_MODEL, "--ip", "127.0.0.1",
-                    "--port", str(custom_port)]),
+                    "--port", str(custom_port)],
+                   identity=probe.IDENTITY_QWEN_CUSTOM),
         ServerSpec("qwen-clone", config.CLONE_API_URL,
                    [demo, QWEN_BASE_MODEL, "--ip", "127.0.0.1",
-                    "--port", str(clone_port)]),
+                    "--port", str(clone_port)],
+                   identity=probe.IDENTITY_QWEN_CLONE),
     ]
     managed = servers.manages(specs)
     # Which local servers this tool started (pid alive) name the running
diff --git a/app/backends/servers.py b/app/backends/servers.py
index 912c09d..62eadb5 100644
--- a/app/backends/servers.py
+++ b/app/backends/servers.py
@@ -1,16 +1,25 @@
 """Start and stop TTS backend servers from the TUI hub.
 
 Each backend's ``detect()`` returns a list of ``ServerSpec`` — the exact argv
-(absolute binaries in the managed venv, no shell activation needed) and the
-URL to probe for readiness. This module turns those specs into running
-processes: ``start`` spawns the server, streams its output to
-``app/logs/<name>-server.log``, records its pid, and polls the URL until it
-accepts connections (model loads are slow, so the timeout is generous);
-``stop`` terminates the process group the hub started.
-
-Everything here runs in the plain console tail after the curses TUI returns
-(matching the wizards' build/pip streaming), so progress and log tails appear
-normally. Pid/log files live under ``app/logs/`` which is already gitignored.
+(absolute binaries in the managed venv, no shell activation needed), the
+working directory to spawn it in, and the URL to probe for readiness. This
+module turns those specs into running processes: ``start`` spawns the server
+(streaming its output to ``app/logs/<name>-server.log``, in the spec's cwd
+when it has one — audio.cpp resolves model_specs/ relative to its process
+working directory), records its pid, and polls the URL until it accepts
+connections (model loads are slow, so the timeout is generous). With a spec
+``identity`` the poll also verifies the server answers HTTP as that backend,
+so readiness means "serving", not just "listening".
+
+Progress reporting goes through an optional ``progress`` callback (see
+``_console_progress`` for the event shapes); the default callback prints the
+same lines as before, so the plain-console flow is unchanged. ``stop``
+terminates the process group the hub started.
+
+Everything here can run in the plain console tail after the curses TUI
+returns (matching the wizards' build/pip streaming) or behind the run view's
+boot screen, which renders the same events. Pid/log files live under
+``app/logs/`` which is already gitignored.
 """
 
 import os
@@ -19,9 +28,9 @@ import subprocess
 import sys
 import time
 from pathlib import Path
-from typing import List
+from typing import Callable, List, Optional
 
-from backends import common
+from backends import common, probe
 from backends.common import APP_DIR
 
 LOG_DIR = APP_DIR / "logs"
@@ -34,6 +43,48 @@ SERVER_START_TIMEOUT = 600
 # Grace period after SIGTERM before escalating to SIGKILL (POSIX).
 STOP_GRACE_SECONDS = 10
 
+# How often the start poll re-checks readiness (seconds).
+POLL_INTERVAL = 1
+
+# Progress callback: called with an event dict. KIND is one of:
+#   "starting"  {name, argv, cwd, log_path, pid}  spawned, waiting for boot
+#   "elapsed"   {name, seconds}                   heartbeat while waiting
+#   "ready"     {name, url}                       server is up and answering
+#   "exited"    {name, returncode, log_tail}      process exited while booting
+#   "timeout"   {name, seconds, log_tail}         readiness deadline elapsed
+#   "running"   {name, url}                       already up (no spawn)
+#   "cancelled" {name}                            boot aborted via cancel
+#   "error"     {message}                         could not spawn the executable
+ProgressCallback = Optional[Callable[[dict], None]]
+
+
+def _console_progress(event: dict) -> None:
+    """Print PROGRESS events as the plain-console output (the old behavior)."""
+    kind = event.get("kind")
+    if kind == "starting":
+        print(f"[INFO] starting {event['name']} server: {event['argv']}")
+        if event.get("cwd"):
+            print(f"[INFO] working directory: {event['cwd']}")
+        print(f"[INFO] pid {event['pid']}; logs: {event['log_path']}")
+    elif kind == "elapsed":
+        print(f"[INFO] still waiting for the server ({int(event['seconds'])}s)...")
+    elif kind == "ready":
+        print(f"[OK] {event['name']} server is up on {event['url']}")
+    elif kind == "running":
+        print(f"[INFO] {event['name']} server already running on {event['url']}")
+    elif kind == "cancelled":
+        print(f"[INFO] {event['name']} server start cancelled")
+    elif kind == "exited":
+        print(f"[ERROR] {event['name']} server exited with code "
+              f"{event['returncode']}")
+        _print_tail(event.get("log_tail"))
+    elif kind == "timeout":
+        print(f"[ERROR] {event['name']} server did not start within "
+              f"{int(event['seconds'])}s")
+        _print_tail(event.get("log_tail"))
+    elif kind == "error":
+        print(f"[ERROR] {event['message']}")
+
 
 def _log_path(name: str) -> Path:
     return LOG_DIR / f"{name}-server.log"
@@ -43,17 +94,20 @@ def _pid_path(name: str) -> Path:
     return LOG_DIR / f"{name}-server.pid"
 
 
-def _tail_log(name: str, lines: int = 20) -> None:
-    """Print the last LINES of the server's log (best-effort)."""
-    path = _log_path(name)
+def _read_log_tail(name: str, lines: int = 20) -> List[str]:
+    """Return the last LINES of the server's log (best-effort)."""
     try:
-        text = path.read_text(encoding="utf-8", errors="replace")
+        text = _log_path(name).read_text(encoding="utf-8", errors="replace")
     except OSError:
-        return
-    tail = "\n".join(text.splitlines()[-lines:])
+        return []
+    return text.splitlines()[-lines:]
+
+
+def _print_tail(tail: List[str]) -> None:
+    """Print a log-tail event payload (used by the console callback)."""
     if tail:
-        print(f"--- last {lines} lines of {path} ---")
-        print(tail)
+        print(f"--- last {len(tail)} lines of the server log ---")
+        print("\n".join(tail))
         print("---")
 
 
@@ -123,24 +177,53 @@ def _kill_pid(pid: int) -> bool:
     return True
 
 
-def start(spec) -> bool:
+def _server_ready(spec) -> bool:
+    """True when the server described by SPEC is usable, not just listening.
+
+    Without an IDENTITY this is the plain TCP-connect check. With one, the
+    server must also answer HTTP as that backend (``probe.identify_server``);
+    for the faster backend (whose model loads after the port opens) the
+    ``/health`` model_loaded flag must additionally be true.
+    """
+    if not common.server_running(spec.url):
+        return False
+    identity = getattr(spec, "identity", None)
+    if identity is None:
+        return True
+    if probe.identify_server(spec.url) != identity:
+        return False
+    if identity == probe.IDENTITY_FASTER:
+        return probe.faster_model_loaded(spec.url)
+    return True
+
+
+def start(spec, progress: ProgressCallback = None,
+          cancel=None) -> bool:
     """Start the server described by SPEC (a ``backends.ServerSpec``).
 
-    Spawns its argv with stdout/stderr to ``logs/<name>-server.log``, records
-    the pid, and polls ``common.server_running(spec.url)`` until it accepts
-    connections or ``SERVER_START_TIMEOUT`` elapses. Returns True when the
-    server is up; on timeout or early exit, prints the log tail and returns
+    Spawns its argv with stdout/stderr to ``logs/<name>-server.log``, in the
+    spec's CWD when it has one (audio.cpp discovers model_specs/ from its
+    process working directory), records the pid, and polls readiness —
+    ``_server_ready``, so an IDENTITY spec must actually answer HTTP — until
+    it is up or ``SERVER_START_TIMEOUT`` elapses. Returns True when the
+    server is up; on timeout or early exit reports the log tail and returns
     False. A no-op (True) when the server is already running.
+
+    PROGRESS, when given, receives each boot event (see ProgressCallback);
+    the default ``_console_progress`` prints them, preserving the old
+    console output. CANCEL (a threading.Event) aborts the boot: the spawned
+    process is terminated and False is reported (event kind "cancelled").
     """
+    report = progress if progress is not None else _console_progress
     argv: List[str] = list(spec.argv)
     exe = Path(argv[0])
     if not exe.exists():
-        print(f"[ERROR] server executable not found: {exe}")
-        print("        run 'Set up a backend' for "
-              f"{spec.name!r} first.")
+        report({"kind": "error",
+                "message": f"server executable not found: {exe}. Run 'Set "
+                           "up a backend' first."})
         return False
-    if common.server_running(spec.url):
-        print(f"[INFO] {spec.name} server already running on {spec.url}")
+    if _server_ready(spec):
+        report({"kind": "running", "name": spec.name, "url": spec.url})
         return True
 
     LOG_DIR.mkdir(parents=True, exist_ok=True)
@@ -151,10 +234,11 @@ def start(spec) -> bool:
         except OSError:
             pass
 
-    print(f"[INFO] starting {spec.name} server: "
-          + " ".join(str(a) for a in argv))
+    cwd = getattr(spec, "cwd", None)
     log_handle = _log_path(spec.name).open("w", encoding="utf-8")
     popen_kwargs = {"stdout": log_handle, "stderr": subprocess.STDOUT}
+    if cwd is not None:
+        popen_kwargs["cwd"] = str(cwd)
     if sys.platform == "win32":
         popen_kwargs["creationflags"] = \
             subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore[attr-defined]
@@ -163,31 +247,51 @@ def start(spec) -> bool:
     try:
         proc = subprocess.Popen(argv, **popen_kwargs)
     except OSError as exc:
-        print(f"[ERROR] could not start server: {exc}")
+        report({"kind": "error",
+                "message": f"could not start server: {exc}"})
         log_handle.close()
         return False
 
     pid_file.write_text(str(proc.pid), encoding="utf-8")
-    print(f"[INFO] pid {proc.pid}; logs: {_log_path(spec.name)}")
-
-    deadline = time.time() + SERVER_START_TIMEOUT
+    report({"kind": "starting", "name": spec.name,
+            "argv": " ".join(str(a) for a in argv),
+            "cwd": str(cwd) if cwd is not None else None,
+            "log_path": str(_log_path(spec.name)), "pid": proc.pid})
+
+    started = time.time()
+    next_heartbeat = started + 15
+    deadline = started + SERVER_START_TIMEOUT
     while time.time() < deadline:
+        if cancel is not None and cancel.is_set():
+            # User cancelled while booting: kill what we spawned (the
+            # server we started is not left loading in the background).
+            _kill_pid(proc.pid)
+            try:
+                pid_file.unlink()
+            except OSError:
+                pass
+            report({"kind": "cancelled", "name": spec.name})
+            return False
         if proc.poll() is not None:
-            print(f"[ERROR] {spec.name} server exited with code "
-                  f"{proc.returncode}")
-            _tail_log(spec.name)
+            report({"kind": "exited", "name": spec.name,
+                    "returncode": proc.returncode,
+                    "log_tail": _read_log_tail(spec.name)})
             try:
                 pid_file.unlink()
             except OSError:
                 pass
             return False
-        if common.server_running(spec.url):
-            print(f"[OK] {spec.name} server is up on {spec.url}")
+        if _server_ready(spec):
+            report({"kind": "ready", "name": spec.name, "url": spec.url})
             return True
-        time.sleep(1)
-    print(f"[ERROR] {spec.name} server did not start within "
-          f"{SERVER_START_TIMEOUT}s")
-    _tail_log(spec.name)
+        if time.time() >= next_heartbeat:
+            report({"kind": "elapsed",
+                    "seconds": time.time() - started})
+            next_heartbeat += 15
+        time.sleep(POLL_INTERVAL)
+    report({"kind": "timeout", "name": spec.name,
+            "seconds": SERVER_START_TIMEOUT,
+            "log_tail": _read_log_tail(spec.name)})
     # Leave the pid file in place so stop() can kill it (it may still load).
     return False
 
diff --git a/app/converter/converter.py b/app/converter/converter.py
index 5f67f4a..b2c2923 100644
--- a/app/converter/converter.py
+++ b/app/converter/converter.py
@@ -5,12 +5,13 @@ import logging
 import re
 import shutil
 import sys
+import threading
 import time
 import traceback
 from collections import Counter
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple
+from typing import Callable, Dict, List, Optional, Tuple
 
 from . import audio, chunking, config, cover, extractors
 from .audio import TrackMeta
@@ -19,6 +20,7 @@ from .tts import (
     BACKEND_AUDIOCPP,
     BACKEND_FASTER,
     BACKEND_QWEN,
+    ConversionCancelled,
     MODEL_SIZE,
     VOICE_MODE_CLONE,
     VOICE_MODE_CUSTOM,
@@ -54,13 +56,15 @@ def _console_log_filter(record: logging.LogRecord) -> bool:
     return not record.name.startswith(("httpx", "httpcore"))
 
 
-def setup_logging(debug: bool = False) -> None:
-    """Configure logging to a dated file and the console.
+def setup_logging(debug: bool = False, console: bool = True) -> None:
+    """Configure logging to a dated file and (optionally) the console.
 
     The file keeps the full record (DEBUG with --debug), including httpx
     request logs. The console handler only surfaces warnings and errors
     (DEBUG with --debug) so progress prints are never mirrored as
     timestamped log lines; httpx/httpcore request logs stay file-only.
+    CONSOLE=False (the TUI run view owns the screen) keeps every record
+    in the file only.
     """
     LOGS_FOLDER.mkdir(parents=True, exist_ok=True)
     file_handler = logging.FileHandler(
@@ -68,13 +72,16 @@ def setup_logging(debug: bool = False) -> None:
         encoding="utf-8",
     )
     file_handler.setLevel(logging.DEBUG if debug else logging.INFO)
-    console_handler = logging.StreamHandler(sys.stdout)
-    console_handler.setLevel(logging.DEBUG if debug else logging.WARNING)
-    console_handler.addFilter(_console_log_filter)
+    handlers = [file_handler]
+    if console:
+        console_handler = logging.StreamHandler(sys.stdout)
+        console_handler.setLevel(logging.DEBUG if debug else logging.WARNING)
+        console_handler.addFilter(_console_log_filter)
+        handlers.append(console_handler)
     logging.basicConfig(
         level=logging.INFO,
         format="%(asctime)s - %(levelname)s - %(message)s",
-        handlers=[file_handler, console_handler],
+        handlers=handlers,
     )
     if debug:
         logging.getLogger("converter").setLevel(logging.DEBUG)
@@ -87,6 +94,23 @@ def setup_directories() -> None:
         Path(directory).mkdir(parents=True, exist_ok=True)
 
 
+def voice_mode_for(backend: str, voice: Optional[str] = None,
+                   clone: Optional[str] = None) -> str:
+    """The voice mode a run with these options would use.
+
+    Mirrors the choice ``audiobook.convert`` makes from the same inputs
+    (faster always clones; audiocpp clones through a server-side voice;
+    qwen clones only with a reference .wav), so the hub can run the
+    pre-flight overwrite checks against exactly the output names the
+    conversion will produce.
+    """
+    if backend == BACKEND_FASTER:
+        return VOICE_MODE_CLONE
+    if backend == BACKEND_AUDIOCPP:
+        return VOICE_MODE_CLONE if voice else VOICE_MODE_CUSTOM
+    return VOICE_MODE_CLONE if clone else VOICE_MODE_CUSTOM
+
+
 def find_existing_outputs(output_name: str, output_format: str) -> List[Path]:
     """Return existing output files that a conversion would overwrite.
 
@@ -104,19 +128,32 @@ def find_existing_outputs(output_name: str, output_format: str) -> List[Path]:
     return existing
 
 
-def prompt_overwrite(existing: List[Path], output_name: str) -> bool:
+def _overwrite_message(existing: List[Path], output_name: str) -> str:
+    """The overwrite question for the files in EXISTING."""
+    if len(existing) == 1:
+        return (f"{existing[0].name} already exists. Convert anyway "
+                "and overwrite it?")
+    return (f"{len(existing)} output files for '{output_name}' already exist "
+            f"(e.g. {existing[0].name}). Convert anyway and overwrite them?")
+
+
+def prompt_overwrite(existing: List[Path], output_name: str,
+                     confirm: Optional[Callable[[str, bool], bool]] = None) -> bool:
     """Ask whether to reconvert a book whose output files already exist.
 
     All overwrite questions are asked before any conversion starts so the
     rest of the run is unattended. Pressing Enter defaults to yes (so a
     user can just hit Enter through the prompts), but a closed stdin
     (non-interactive run) declines and keeps existing files safe.
+
+    CONFIRM, when given, replaces the console ``input()`` prompt: it is
+    called once with (message, default) and must return the answer — the
+    hub passes a TUI yes/no dialog so the questions are asked inside the
+    menu instead of the console.
     """
-    if len(existing) == 1:
-        message = f"{existing[0].name} already exists. Convert anyway and overwrite it?"
-    else:
-        message = (f"{len(existing)} output files for '{output_name}' already exist "
-                   f"(e.g. {existing[0].name}). Convert anyway and overwrite them?")
+    message = _overwrite_message(existing, output_name)
+    if confirm is not None:
+        return confirm(message, True)
     while True:
         try:
             answer = input(f"{message} [Y/n]: ").strip().lower()
@@ -135,6 +172,10 @@ def prompt_overwrite(existing: List[Path], output_name: str) -> bool:
 class AudiobookConverter:
     """Audiobook converter using a local TTS API."""
 
+    # Class-level default so a partially-constructed instance (tests build
+    # these with __new__) behaves like a plain console run.
+    _progress = None
+
     def __init__(self, voice_mode: str = VOICE_MODE_CUSTOM, voice_clone_ref_audio: Optional[str] = None,
                  voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False,
                  speed: float = 1.0, single_file: bool = False, output_format: str = config.AUDIO_FORMAT,
@@ -143,7 +184,9 @@ class AudiobookConverter:
                  model_id: Optional[str] = None,
                  instructions: Optional[str] = None,
                  request_options: Optional[Dict[str, str]] = None,
-                 api_url: Optional[str] = None):
+                 api_url: Optional[str] = None,
+                 progress: Optional[Callable[[dict], None]] = None,
+                 cancel=None):
         if speed <= 0:
             raise ValueError(f"Speed must be a positive number, got {speed}")
         if output_format not in AUDIO_FORMATS:
@@ -193,6 +236,29 @@ class AudiobookConverter:
                 language=self.language,
                 api_url=api_url,
             )
+        # Interactive reporting/cancellation (the TUI run view): PROGRESS
+        # receives an event dict per state change and turns the console
+        # prints off (the view owns the screen); CANCEL (a
+        # threading.Event) stops the run between requests.
+        self._progress = progress
+        self.tts.cancel = cancel
+        self.tts.quiet = progress is not None
+
+    def _emit(self, event: dict) -> None:
+        """Send one progress event (a no-op without a progress callback)."""
+        if self._progress is not None:
+            self._progress(event)
+
+    def _say(self, message: str) -> None:
+        """Print a console progress line unless the run view owns the screen."""
+        if self._progress is None:
+            print(message)
+
+    def _check_cancelled(self) -> None:
+        """Raise ConversionCancelled when the run's cancel event is set."""
+        cancel = getattr(getattr(self, "tts", None), "cancel", None)
+        if isinstance(cancel, threading.Event) and cancel.is_set():
+            raise ConversionCancelled("Cancelled by user")
 
     def _validate_configuration(self) -> None:
         """Validate configuration settings."""
@@ -307,7 +373,10 @@ class AudiobookConverter:
         return book_debug_dir / f"{index:02d}_{AudiobookConverter._sanitize_filename(title)}"
 
     def convert_book(self, file_path: Path, output_name: Optional[str] = None) -> bool:
-        """Convert a single book to one or more audiobook files."""
+        """Convert a single book to one or more audiobook files.
+
+        Raises ConversionCancelled when the run's cancel event is set.
+        """
         logger.info("Converting: %s", file_path.name)
         start_time = time.time()
 
@@ -334,7 +403,7 @@ class AudiobookConverter:
             cover_path = cover.generate_cover(
                 book.title, CHUNKS_FOLDER / "chunk_cover.png")
             if cover_path:
-                print(f"[INFO] Generated cover art for '{book.title}'")
+                self._say(f"[INFO] Generated cover art for '{book.title}'")
             meta = TrackMeta(title=book.title, artist=book.author, album=book.title)
 
             # m4b is always a single file; multi-chapter books get embedded
@@ -356,6 +425,9 @@ class AudiobookConverter:
 
             success = True
             for index, section in enumerate(sections, 1):
+                self._check_cancelled()
+                self._emit({"kind": "chapter", "index": index,
+                            "total": len(sections)})
                 chapter_name = f"{stem}_{index:02d}_{self._sanitize_filename(section.title)}"
                 output_path = AUDIOBOOKS_FOLDER / f"{chapter_name}.{self.output_format}"
                 track_meta = meta._replace(
@@ -368,6 +440,8 @@ class AudiobookConverter:
                 ) and success
             return success
 
+        except ConversionCancelled:
+            raise
         except Exception as exc:
             logger.error("Conversion failed: %s", exc)
             logger.error(traceback.format_exc())
@@ -392,11 +466,15 @@ class AudiobookConverter:
         titles = []
         total_chapters = len(sections)
         for index, section in enumerate(sections, 1):
+            self._check_cancelled()
+            self._emit({"kind": "chapter", "index": index,
+                        "total": total_chapters})
             chapter_path = CHUNKS_FOLDER / f"chapter_{index:04d}.wav"
             title = (section.title or "").strip() or f"Chapter {index}"
-            print(f"\n{'=' * 50}")
-            print(f"CHAPTER {index}/{total_chapters}: {title}")
-            print(f"{'=' * 50}")
+            if self._progress is None:
+                print(f"\n{'=' * 50}")
+                print(f"CHAPTER {index}/{total_chapters}: {title}")
+                print(f"{'=' * 50}")
             logger.info("Converting chapter %d/%d: %s", index, total_chapters, title)
             if not self._convert_text(section.text, chapter_path, time.time(),
                                       speed=1.0, output_format="wav",
@@ -430,15 +508,18 @@ class AudiobookConverter:
         chunk: a partial audiobook is never assembled, so the remaining
         chunks are not requested. When ``debug_dir`` is given (--debug),
         each chunk's request text and returned audio are also dumped there,
-        and every request/response is logged.
+        and every request/response is logged. Raises ConversionCancelled
+        when the run's cancel event is set (between chunks).
         """
         total_chunks = len(chunks)
-        print(f"\n{'=' * 50}")
-        print(f"PROCESSING {total_chunks} CHUNKS")
-        print(f"{'=' * 50}")
+        if self._progress is None:
+            print(f"\n{'=' * 50}")
+            print(f"PROCESSING {total_chunks} CHUNKS")
+            print(f"{'=' * 50}")
 
         results: Dict[int, Optional[Path]] = {}
         for chunk_num, chunk_text in enumerate(chunks, 1):
+            self._check_cancelled()
             if debug_dir is not None:
                 # Written before the request so the exact text survives a
                 # crash mid-generation; failed chunks keep their dumps.
@@ -456,24 +537,34 @@ class AudiobookConverter:
                         destination = f" -> {copied.name}" if copied else ""
                         logger.debug("Chunk %d/%d response in %.1fs%s",
                                      chunk_num, total_chunks, elapsed, destination)
-                    print(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed")
+                    self._say(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed")
                     logger.info("+ Chunk %d/%d completed", chunk_num, total_chunks)
+                    self._emit({"kind": "chunk_done", "chunk": chunk_num,
+                                "total": total_chunks,
+                                "seconds": time.time() - request_start})
                 else:
                     logger.error("Chunk %d/%d failed; aborting the remaining chunks",
                                  chunk_num, total_chunks)
+                    self._emit({"kind": "chunk_failed", "chunk": chunk_num,
+                                "total": total_chunks})
                     break
 
+            except ConversionCancelled:
+                raise
             except Exception as exc:
                 results[chunk_num] = None
                 logger.error("Chunk %d/%d error: %s; aborting the remaining chunks",
                              chunk_num, total_chunks, exc)
+                self._emit({"kind": "chunk_failed", "chunk": chunk_num,
+                            "total": total_chunks, "error": str(exc)})
                 break
 
         successful_chunks = sum(1 for path in results.values() if path)
-        print(f"\n{'=' * 50}")
-        print("CHUNK PROCESSING COMPLETE")
-        print(f"Successful: {successful_chunks}/{total_chunks}")
-        print(f"{'=' * 50}")
+        if self._progress is None:
+            print(f"\n{'=' * 50}")
+            print("CHUNK PROCESSING COMPLETE")
+            print(f"Successful: {successful_chunks}/{total_chunks}")
+            print(f"{'=' * 50}")
         logger.info("Chunk processing completed: %d/%d chunks", successful_chunks, total_chunks)
         return results
 
@@ -522,7 +613,8 @@ class AudiobookConverter:
                 BACKEND_AUDIOCPP: "audio.cpp server",
             }
             backend = backend_labels.get(self.backend, "Qwen API")
-            print(f"[INFO] Processing {total_chunks} chunks via {backend}...")
+            self._say(f"[INFO] Processing {total_chunks} chunks via {backend}...")
+            self._emit({"kind": "chunks", "total": total_chunks})
 
             results = self._synthesize_chunks(chunks, debug_dir=debug_dir)
             successful_chunks = sum(1 for path in results.values() if path)
@@ -546,8 +638,8 @@ class AudiobookConverter:
                     logger.info("Chapter %d/%d converted in %dm %ds (%d/%d chunks)",
                                 chapter[0], chapter[1], minutes, seconds,
                                 successful_chunks, total_chunks)
-                    print(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted "
-                          f"({successful_chunks}/{total_chunks} chunks)")
+                    self._say(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted "
+                              f"({successful_chunks}/{total_chunks} chunks)")
                 else:
                     logger.info("Conversion completed in %dm %ds: %s", minutes, seconds, output_path)
             else:
@@ -555,6 +647,8 @@ class AudiobookConverter:
 
             return success
 
+        except ConversionCancelled:
+            raise
         except Exception as exc:
             logger.error("Conversion failed: %s", exc)
             logger.error(traceback.format_exc())
@@ -562,53 +656,53 @@ class AudiobookConverter:
 
     def _print_banner(self) -> None:
         """Print the startup summary for the selected backend."""
-        print("=" * 70)
-        print("TTS AUDIOBOOK GENERATOR")
-        print("=" * 70)
-        print(f"Books folder: {BOOKS_FOLDER}")
-        print(f"Output folder: {AUDIOBOOKS_FOLDER}")
+        self._say("=" * 70)
+        self._say("TTS AUDIOBOOK GENERATOR")
+        self._say("=" * 70)
+        self._say(f"Books folder: {BOOKS_FOLDER}")
+        self._say(f"Output folder: {AUDIOBOOKS_FOLDER}")
         if self.backend == BACKEND_FASTER:
-            print(f"Faster TTS endpoint: {config.FASTER_API_URL}")
-            print("Backend: faster (voice cloning, reference configured on server)")
-            print(f"Voice: {self.voice or config.FASTER_VOICE}")
+            self._say(f"Faster TTS endpoint: {config.FASTER_API_URL}")
+            self._say("Backend: faster (voice cloning, reference configured on server)")
+            self._say(f"Voice: {self.voice or config.FASTER_VOICE}")
         elif self.backend == BACKEND_AUDIOCPP:
-            print(f"audio.cpp endpoint: {config.AUDIOCPP_API_URL}")
-            print(f"Model id: {self.tts.model_id}")
-            print(f"Model family: {getattr(self.tts, 'family', 'unknown')}")
+            self._say(f"audio.cpp endpoint: {config.AUDIOCPP_API_URL}")
+            self._say(f"Model id: {self.tts.model_id}")
+            self._say(f"Model family: {getattr(self.tts, 'family', 'unknown')}")
             if self.voice:
-                print("Backend: audio.cpp (voice cloning, reference configured on server)")
-                print(f"Voice: {self.voice}")
+                self._say("Backend: audio.cpp (voice cloning, reference configured on server)")
+                self._say(f"Voice: {self.voice}")
             elif self.instructions:
-                print("Backend: audio.cpp (voice from --instructions description)")
-                print(f"Instruction: {self.instructions}")
+                self._say("Backend: audio.cpp (voice from --instructions description)")
+                self._say(f"Instruction: {self.instructions}")
             else:
-                print("Backend: audio.cpp (custom voice, built-in speaker)")
-                print(f"Speaker: {config.SPEAKER}")
+                self._say("Backend: audio.cpp (custom voice, built-in speaker)")
+                self._say(f"Speaker: {config.SPEAKER}")
             if self.request_options:
-                print(f"Request options: {self.request_options}")
-            print(f"Language: {self.language}")
+                self._say(f"Request options: {self.request_options}")
+            self._say(f"Language: {self.language}")
         else:
             tts_client = getattr(self, "tts", None)
             api_url = (getattr(tts_client, "api_url", None)
                        or (config.CLONE_API_URL
                            if self.voice_mode == VOICE_MODE_CLONE
                            else config.QWEN_API_URL))
-            print(f"Qwen API endpoint: {api_url}")
-            print(f"Voice mode: {self.voice_mode}")
-            print(f"Model size: {MODEL_SIZE} (always)")
+            self._say(f"Qwen API endpoint: {api_url}")
+            self._say(f"Voice mode: {self.voice_mode}")
+            self._say(f"Model size: {MODEL_SIZE} (always)")
             if self.voice_mode == VOICE_MODE_CUSTOM:
-                print(f"Speaker: {config.SPEAKER}")
-                print(f"Language: {self.language}")
+                self._say(f"Speaker: {config.SPEAKER}")
+                self._say(f"Language: {self.language}")
             elif self.voice_mode == VOICE_MODE_CLONE:
-                print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}")
-                print(f"Language: {self.language}")
-        print(f"Output format: {self.output_format}")
+                self._say(f"Reference audio: {Path(self.voice_clone_ref_audio).name}")
+                self._say(f"Language: {self.language}")
+        self._say(f"Output format: {self.output_format}")
         if self.single_file and self.output_format != "m4b":
-            print("Chapter mode: single file (--single-file)")
+            self._say("Chapter mode: single file (--single-file)")
         if abs(self.speed - 1.0) >= 1e-6:
-            print(f"Playback speed: {self.speed:g}x")
+            self._say(f"Playback speed: {self.speed:g}x")
         if self.debug:
-            print(f"Debug dumps (per-chunk text + raw audio): {DEBUG_FOLDER}")
+            self._say(f"Debug dumps (per-chunk text + raw audio): {DEBUG_FOLDER}")
         print("=" * 70)
 
     # ------------------------------------------------------------------
@@ -620,7 +714,8 @@ class AudiobookConverter:
                              voice_mode: str,
                              voice_clone_ref_audio: Optional[str],
                              output_format: str,
-                             instructions: Optional[str] = None
+                             instructions: Optional[str] = None,
+                             confirm: Optional[Callable[[str, bool], bool]] = None,
                              ) -> Tuple[List[Path], List[Tuple[Path, str]]]:
         """Discover books and ask every overwrite question up front.
 
@@ -632,6 +727,8 @@ class AudiobookConverter:
 
         Asking before connecting means a user who declines a prompt (or has
         nothing to convert) never waits on a slow server handshake.
+        CONFIRM replaces the console ``input()`` prompt (the hub passes a
+        TUI yes/no dialog).
         """
         book_files = sorted(
             f for f in BOOKS_FOLDER.iterdir()
@@ -656,7 +753,8 @@ class AudiobookConverter:
                 output_name = f"{book_file.stem}_{book_file.suffix.lstrip('.')}"
             output_name = f"{output_name}_{narrator_tag}"
             existing = find_existing_outputs(output_name, output_format)
-            if existing and not prompt_overwrite(existing, output_name):
+            if existing and not prompt_overwrite(existing, output_name,
+                                                 confirm=confirm):
                 print(f"[INFO] Skipping {book_file.name} (existing output kept)")
                 continue
             planned.append((book_file, output_name))
@@ -667,7 +765,10 @@ class AudiobookConverter:
     # ------------------------------------------------------------------
 
     def run(self) -> bool:
-        """Main conversion process. Returns True if all books converted."""
+        """Main conversion process. Returns True if all books converted.
+
+        Raises ConversionCancelled when the run's cancel event is set.
+        """
         run_start = time.time()
         self._print_banner()
 
@@ -684,37 +785,57 @@ class AudiobookConverter:
                 self.instructions)
 
         if not book_files:
-            print(f"[INFO] No supported files found in {BOOKS_FOLDER}")
-            print(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}")
-            print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file "
-                  f"to {BOOKS_FOLDER} and run again.")
+            self._say(f"[INFO] No supported files found in {BOOKS_FOLDER}")
+            self._say(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}")
+            self._say("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file "
+                      f"to {BOOKS_FOLDER} and run again.")
+            self._emit({"kind": "done", "ok": 0, "total": 0})
             return True
 
         if not planned:
-            print("[INFO] Nothing to convert (all books skipped)")
+            self._say("[INFO] Nothing to convert (all books skipped)")
+            self._emit({"kind": "done", "ok": 0, "total": 0})
             return True
 
-        print(f"[INFO] Converting {len(planned)} of {len(book_files)} book(s)")
+        self._say(f"[INFO] Converting {len(planned)} of {len(book_files)} book(s)")
 
         results = {}
-        for book_file, output_name in planned:
+        cancelled = False
+        for index, (book_file, output_name) in enumerate(planned, 1):
+            self._check_cancelled()
+            self._emit({"kind": "book", "index": index, "total": len(planned),
+                        "name": book_file.name})
             try:
                 success = self.convert_book(book_file, output_name=output_name)
                 results[book_file.name] = success
+                self._emit({"kind": "book_done", "name": book_file.name,
+                            "ok": bool(success)})
+            except ConversionCancelled:
+                self._emit({"kind": "cancelled"})
+                logger.info("Conversion cancelled by user at %s", book_file.name)
+                cancelled = True
+                break
             except KeyboardInterrupt:
-                print("\n[WARNING] Conversion interrupted by user")
+                self._say("\n[WARNING] Conversion interrupted by user")
                 results[book_file.name] = False
                 break
             except Exception as exc:
                 logger.error("Unexpected error: %s", exc)
                 results[book_file.name] = False
-            if not results[book_file.name]:
+                self._emit({"kind": "book_failed", "name": book_file.name,
+                            "error": str(exc)})
+            if not results.get(book_file.name):
                 logger.error("Conversion of %s failed; aborting the remaining books",
                              book_file.name)
                 break
 
         successful = sum(results.values())
         total = len(results)
+        self._emit({"kind": "done", "ok": successful,
+                    "total": total or len(planned), "cancelled": cancelled})
+
+        if self._progress is not None:
+            return not cancelled and total > 0 and successful == total
 
         print("\n" + "=" * 70)
         print("CONVERSION SUMMARY")
diff --git a/app/converter/tts.py b/app/converter/tts.py
index a83896d..0667ec3 100644
--- a/app/converter/tts.py
+++ b/app/converter/tts.py
@@ -34,6 +34,17 @@ from .chunking import split_into_chunks
 
 logger = logging.getLogger(__name__)
 
+
+class ConversionCancelled(Exception):
+    """Raised inside a conversion whose cancel event was set.
+
+    The TUI run view sets a ``threading.Event`` on the TTS client (and the
+    converter checks it between chunks/chapters/books); the retry loops
+    raise this so the cancellation propagates out of a sleeping or retrying
+    request promptly instead of finishing the retry ladder.
+    """
+
+
 # Voice modes (re-exported for the CLI and the converter orchestrator).
 VOICE_MODE_CUSTOM = "custom_voice"
 VOICE_MODE_CLONE = "voice_clone"
@@ -281,10 +292,35 @@ def whisper_backend_available() -> Optional[str]:
 class _BaseTTSClient:
     """Shared chunk retry logic, heartbeat, and chunk file bookkeeping."""
 
+    # Set by the converter when the run is cancellable (the TUI run view):
+    # a threading.Event that, once set, aborts the run between requests
+    # (and interrupts retry back-off sleeps). ``quiet`` silences console
+    # prints (the run view owns the screen).
+    cancel = None
+    quiet = False
+
     def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]:
         """Generate one audio chunk; returns its path in the chunks folder."""
         raise NotImplementedError
 
+    def _cancel_requested(self) -> bool:
+        """True when the run's cancel event has been set (if any)."""
+        return isinstance(self.cancel, threading.Event) \
+            and self.cancel.is_set()
+
+    def _check_cancelled(self) -> None:
+        """Raise ConversionCancelled when the cancel event is set."""
+        if self._cancel_requested():
+            raise ConversionCancelled("Cancelled by user")
+
+    def _sleep(self, seconds: float) -> None:
+        """Sleep SECONDS, cut short (raising) when the cancel event sets."""
+        if isinstance(self.cancel, threading.Event):
+            if self.cancel.wait(seconds):
+                raise ConversionCancelled("Cancelled by user")
+        else:
+            time.sleep(seconds)
+
     def _chunk_path(self, chunk_num: int, suffix: str) -> Path:
         """Resolve the target path for a chunk, removing stale files first.
 
@@ -302,28 +338,31 @@ class _BaseTTSClient:
         """Process a chunk with retry logic.
 
         Returns the generated chunk file's path, or None when all attempts
-        failed.
+        failed. Raises ConversionCancelled when the run was cancelled.
         """
         for attempt in range(config.MAX_RETRIES):
+            self._check_cancelled()
             try:
                 result = self.generate_chunk(text, chunk_num)
                 if result and Path(result).exists():
                     return Path(result)
                 logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1)
+            except ConversionCancelled:
+                raise
             except Exception as exc:
                 logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc)
 
             if attempt < config.MAX_RETRIES - 1:
                 sleep_time = 5 + (2 ** attempt)
                 logger.info("Waiting %ds before retry...", sleep_time)
-                time.sleep(sleep_time)
+                self._sleep(sleep_time)
 
         logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES)
         return None
 
     @contextlib.contextmanager
     def _chunk_heartbeat(self, chunk_num: int):
-        """Print a periodic "still working" message while a request generates."""
+        """Log a periodic "still working" record while a request generates."""
         stop = threading.Event()
         subject = f"Chunk {chunk_num}"
 
@@ -331,8 +370,13 @@ class _BaseTTSClient:
             start = time.time()
             while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS):
                 elapsed = time.time() - start
-                print(f"[...] {subject} still generating — "
-                      f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", flush=True)
+                if self.quiet:
+                    logger.info("%s still generating — %dm %ds elapsed",
+                                subject, int(elapsed // 60), int(elapsed % 60))
+                else:
+                    print(f"[...] {subject} still generating — "
+                          f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed",
+                          flush=True)
 
         thread = threading.Thread(target=_beat, daemon=True)
         thread.start()
@@ -524,6 +568,8 @@ class QwenTTSClient(_BaseTTSClient):
                          chunk_num, len(sub_texts))
             return str(output_path)
 
+        except ConversionCancelled:
+            raise
         except Exception as exc:
             logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc)
             return None
@@ -706,13 +752,16 @@ class FasterTTSClient(_BaseTTSClient):
                                 sub_total: int) -> bytes:
         """Request one sub-chunk, retrying transient failures."""
         for attempt in range(config.MAX_RETRIES):
+            self._check_cancelled()
             try:
                 return self._request_pcm(text)
+            except ConversionCancelled:
+                raise
             except Exception as exc:
                 logger.warning("Chunk %d sub-chunk %d/%d attempt %d failed: %s",
                                chunk_num, sub_num, sub_total, attempt + 1, exc)
                 if attempt < config.MAX_RETRIES - 1:
-                    time.sleep(2 + 2 * attempt)
+                    self._sleep(2 + 2 * attempt)
         raise RuntimeError(f"Sub-chunk {sub_num}/{sub_total} failed after "
                            f"{config.MAX_RETRIES} attempts")
 
@@ -744,6 +793,8 @@ class FasterTTSClient(_BaseTTSClient):
             logger.debug("Chunk %d generated (%d sub-chunks)", chunk_num, len(sub_chunks))
             return str(output_path)
 
+        except ConversionCancelled:
+            raise
         except Exception as exc:
             logger.error("Faster chunk processing failed for chunk %d: %s", chunk_num, exc)
             return None
@@ -1254,13 +1305,16 @@ class AudioCppTTSClient(_BaseTTSClient):
                                 sub_total: int) -> bytes:
         """Request one sub-chunk, retrying transient failures."""
         for attempt in range(config.MAX_RETRIES):
+            self._check_cancelled()
             try:
                 return self._request_wav(text)
+            except ConversionCancelled:
+                raise
             except Exception as exc:
                 logger.warning("Chunk %d sub-chunk %d/%d attempt %d failed: %s",
                                chunk_num, sub_num, sub_total, attempt + 1, exc)
                 if attempt < config.MAX_RETRIES - 1:
-                    time.sleep(2 + 2 * attempt)
+                    self._sleep(2 + 2 * attempt)
         raise RuntimeError(f"Sub-chunk {sub_num}/{sub_total} failed after "
                            f"{config.MAX_RETRIES} attempts")
 
@@ -1301,6 +1355,8 @@ class AudioCppTTSClient(_BaseTTSClient):
                          chunk_num, len(sub_texts))
             return str(output_path)
 
+        except ConversionCancelled:
+            raise
         except Exception as exc:
             logger.error("audio.cpp chunk processing failed for chunk %d: %s",
                          chunk_num, exc)
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py
index 2cb5a95..acee6b6 100644
--- a/app/tests/test_backends.py
+++ b/app/tests/test_backends.py
@@ -5,7 +5,18 @@ import unittest
 from pathlib import Path
 from unittest.mock import patch
 
-from backends import REGISTRY, detect_all, get
+from backends import REGISTRY, ServerSpec, detect_all, format_launch_hint, get
+
+
+class FormatLaunchHintTests(unittest.TestCase):
+    def test_plain_specs_join_argv(self):
+        specs = [ServerSpec("a", "http://x", ["cmd", "--flag"])]
+        self.assertEqual(format_launch_hint(specs), "cmd --flag")
+
+    def test_cwd_prefixes_the_command(self):
+        specs = [ServerSpec("a", "http://x", ["cmd"], cwd=Path("/opt/audio.cpp"))]
+        self.assertEqual(format_launch_hint(specs),
+                         "cd /opt/audio.cpp && cmd")
 
 
 class RegistryTests(unittest.TestCase):
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index e2b09d0..563ed78 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -1148,5 +1148,127 @@ class FetchServerEndpointsTests(unittest.TestCase):
                 make_server.fetch_server_voices("http://h", "qwen"))
 
 
+class MissingModelEntriesTests(unittest.TestCase):
+    """missing_model_entries: server.json paths vs. files on disk."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _server_json(self, models):
+        path = self.dir / "server.json"
+        path.write_text(json.dumps({"models": models}), encoding="utf-8")
+        return path
+
+    def test_relative_path_resolves_against_config_dir(self):
+        (self.dir / "models" / "present").mkdir(parents=True)
+        (self.dir / "models" / "present" / "m.gguf").write_bytes(b"x")
+        path = self._server_json([
+            {"id": "a", "path": "models/present"},
+            {"id": "b", "path": "models/absent"},
+        ])
+        missing = make_server.missing_model_entries(path)
+        self.assertEqual([m["id"] for m in missing], ["b"])
+
+    def test_empty_directory_counts_as_missing(self):
+        (self.dir / "models" / "empty").mkdir(parents=True)
+        path = self._server_json([{"id": "a", "path": "models/empty"}])
+        self.assertEqual(len(make_server.missing_model_entries(path)), 1)
+
+    def test_absolute_paths_honored(self):
+        target = self.dir / "absolute"
+        target.mkdir()
+        (target / "m.gguf").write_bytes(b"x")
+        path = self._server_json([{"id": "a", "path": str(target)}])
+        self.assertEqual(make_server.missing_model_entries(path), [])
+
+    def test_unreadable_json_returns_empty(self):
+        path = self.dir / "server.json"
+        path.write_text("not json", encoding="utf-8")
+        self.assertEqual(make_server.missing_model_entries(path), [])
+
+    def test_no_models_returns_empty(self):
+        path = self._server_json([])
+        self.assertEqual(make_server.missing_model_entries(path), [])
+
+
+class ModelInstallHintsTests(unittest.TestCase):
+    """model_install_hints: maps missing paths to the install command."""
+
+    def test_maps_path_to_install_id_via_catalog(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            checkout = Path(td)
+            specs = checkout / "model_specs"
+            specs.mkdir()
+            (specs / "qwen3_tts.json").write_text(json.dumps({
+                "family": "qwen3_tts", "category": "tts",
+                "tasks": ["tts"],
+                "packages": [{
+                    "id": "qwen3_tts_0_6b_base_q8_0", "format": "gguf",
+                    "target_directory": "Qwen3-TTS-12Hz-0.6B-Base-GGUF",
+                }],
+            }), encoding="utf-8")
+            missing = [{"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}]
+            hints = make_server.model_install_hints(checkout, missing)
+        self.assertEqual(len(hints), 1)
+        self.assertIn("qwen3_tts_0_6b_base_q8_0", hints[0])
+
+    def test_unmapped_path_names_the_path(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            checkout = Path(td)
+            (checkout / "model_specs").mkdir()
+            hints = make_server.model_install_hints(
+                checkout, [{"id": "x", "rel": "models/nope"}])
+        self.assertIn("models/nope", hints[0])
+        self.assertNotIn("install", hints[0])
+
+
+class DetectServerSpecTests(unittest.TestCase):
+    """detect(): the server spec carries the checkout cwd + identity."""
+
+    def _checkout(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        checkout = Path(tmp.name)
+        (checkout / "model_specs").mkdir()
+        build = checkout / "build" / "linux-cuda-release" / "bin"
+        build.mkdir(parents=True)
+        (build / "audiocpp_server").write_bytes(b"x")
+        (checkout / "server.json").write_text(json.dumps({
+            "models": [{"id": "qwen", "family": "qwen3_tts",
+                        "path": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}],
+        }), encoding="utf-8")
+        return checkout
+
+    def test_spec_has_cwd_and_identity(self):
+        checkout = self._checkout()
+        with patch.object(make_server, "find_local_checkout",
+                          return_value=checkout), \
+                patch.object(make_server, "_detect_remote",
+                             return_value=(False, {})):
+            status = make_server.detect()
+        self.assertEqual(len(status.servers), 1)
+        spec = status.servers[0]
+        self.assertEqual(spec.cwd, checkout)
+        self.assertEqual(spec.identity, "audiocpp")
+        self.assertIn("--config", spec.argv)
+
+    def test_models_missing_flag_and_details(self):
+        checkout = self._checkout()
+        with patch.object(make_server, "find_local_checkout",
+                          return_value=checkout), \
+                patch.object(make_server, "_detect_remote",
+                             return_value=(False, {})):
+            status = make_server.detect()
+        self.assertTrue(status.models_missing)
+        self.assertTrue(any("not downloaded" in line
+                            for line in status.details))
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index 987ff24..201d5b6 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -78,6 +78,108 @@ class StartTests(unittest.TestCase):
             ok = servers.start(self.spec)
         self.assertFalse(ok)
 
+    def _boot_proc(self):
+        proc = MagicMock()
+        proc.pid = 4242
+        proc.poll.return_value = None
+        return proc
+
+    def test_cwd_passed_to_popen(self):
+        """A spec with a cwd spawns the server in that working directory.
+
+        audio.cpp resolves model_specs/<family>.json relative to its
+        process working directory, so the hub must start it from the
+        checkout.
+        """
+        spec = ServerSpec("test", "http://127.0.0.1:9999",
+                          [str(self.exe)], cwd=Path("/opt/audio.cpp"))
+        proc = self._boot_proc()
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc) as mk, \
+                patch("backends.common.server_running",
+                      side_effect=[False, True]), \
+                patch("time.sleep"):
+            ok = servers.start(spec)
+        self.assertTrue(ok)
+        kwargs = mk.call_args.kwargs
+        self.assertEqual(kwargs.get("cwd"), "/opt/audio.cpp")
+
+    def test_identity_spec_waits_for_http_identity(self):
+        """Readiness needs the server to answer HTTP as its identity.
+
+        A TCP-accepting but still-booting server (lazy model load, slow
+        listen-before-serve) must not count as ready.
+        """
+        spec = ServerSpec("test", "http://127.0.0.1:9999",
+                          [str(self.exe)], identity="audiocpp")
+        proc = self._boot_proc()
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc), \
+                patch("backends.common.server_running", return_value=True), \
+                patch.object(servers.probe, "identify_server",
+                             side_effect=[None, None, "audiocpp"]), \
+                patch("time.sleep"):
+            ok = servers.start(spec)
+        self.assertTrue(ok)
+
+    def test_faster_identity_requires_model_loaded(self):
+        """The faster identity additionally waits for /health model_loaded."""
+        spec = ServerSpec("test", "http://127.0.0.1:9999",
+                          [str(self.exe)], identity="faster")
+        proc = self._boot_proc()
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc), \
+                patch("backends.common.server_running", return_value=True), \
+                patch.object(servers.probe, "identify_server",
+                             return_value="faster"), \
+                patch.object(servers.probe, "faster_model_loaded",
+                             side_effect=[False, True]), \
+                patch("time.sleep"):
+            ok = servers.start(spec)
+        self.assertTrue(ok)
+
+    def test_progress_receives_boot_events(self):
+        events = []
+        proc = self._boot_proc()
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc), \
+                patch("backends.common.server_running",
+                      side_effect=[False, True]), \
+                patch("time.sleep"):
+            ok = servers.start(self.spec, progress=events.append)
+        self.assertTrue(ok)
+        kinds = [event["kind"] for event in events]
+        self.assertEqual(kinds, ["starting", "ready"])
+        self.assertEqual(events[0]["pid"], 4242)
+        self.assertIn("--port", events[0]["argv"])
+
+    def test_cancel_aborts_boot_kills_process_and_reports(self):
+        import threading
+        cancel = threading.Event()
+        cancel.set()
+        proc = self._boot_proc()
+        with patch.object(servers, "LOG_DIR", self.dir), \
+                patch("subprocess.Popen", return_value=proc), \
+                patch("backends.common.server_running", return_value=False), \
+                patch.object(servers, "_kill_pid", return_value=True) as mk, \
+                patch("time.sleep"):
+            ok = servers.start(self.spec, cancel=cancel)
+        self.assertFalse(ok)
+        mk.assert_called_once_with(4242)
+        self.assertFalse((self.dir / "test-server.pid").exists())
+
+    def test_console_progress_prints_events(self):
+        import io
+        from contextlib import redirect_stdout
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            servers._console_progress({"kind": "running", "name": "test",
+                                       "url": "http://127.0.0.1:9999"})
+            servers._console_progress({"kind": "error", "message": "boom"})
+        out = buf.getvalue()
+        self.assertIn("already running", out)
+        self.assertIn("boom", out)
+
 
 class StopTests(unittest.TestCase):
     def setUp(self):
diff --git a/app/tests/test_converter_progress.py b/app/tests/test_converter_progress.py
new file mode 100644
index 0000000..1041173
--- /dev/null
+++ b/app/tests/test_converter_progress.py
@@ -0,0 +1,183 @@
+"""Tests for the converter's progress-event and cancellation plumbing.
+
+These exercise the wiring the TUI run view relies on: a ``progress``
+callback receiving book/chunk/done events, a ``cancel`` (threading.Event)
+aborting the run between chunks (raising ConversionCancelled), and the
+injectable ``confirm`` hook on the overwrite prompt.
+"""
+
+import io
+import tempfile
+import threading
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from converter import config, tts
+from converter import converter as converter_mod
+from converter.converter import (
+    AudiobookConverter,
+    ConversionCancelled,
+    prompt_overwrite,
+    voice_mode_for,
+)
+
+
+class VoiceModeForTests(unittest.TestCase):
+    def test_faster_always_clones(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_FASTER),
+                         tts.VOICE_MODE_CLONE)
+
+    def test_audiocpp_voice_clones(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP, voice="narrator"),
+                         tts.VOICE_MODE_CLONE)
+
+    def test_audiocpp_no_voice_is_custom(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP),
+                         tts.VOICE_MODE_CUSTOM)
+
+    def test_qwen_clone_wav_clones(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_QWEN, clone="x.wav"),
+                         tts.VOICE_MODE_CLONE)
+
+    def test_qwen_no_clone_is_custom(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_QWEN),
+                         tts.VOICE_MODE_CUSTOM)
+
+
+class PromptOverwriteConfirmTests(unittest.TestCase):
+    def test_confirm_callback_receives_message_and_default(self):
+        calls = []
+        result = prompt_overwrite([Path("out.mp3")], "out",
+                                  confirm=lambda m, d: calls.append((m, d)) or False)
+        self.assertFalse(result)
+        self.assertEqual(len(calls), 1)
+        self.assertTrue(calls[0][1])  # default yes
+        self.assertIn("out.mp3", calls[0][0])
+
+
+class _ConvertFixture:
+    """A real AudiobookConverter whose TTS client is stubbed."""
+
+    def __init__(self, test_case):
+        self.test = test_case
+        self._books_tmp = tempfile.TemporaryDirectory()
+        self._output_tmp = tempfile.TemporaryDirectory()
+        self._orig = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER)
+        converter_mod.BOOKS_FOLDER = Path(self._books_tmp.name)
+        converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name)
+        (converter_mod.BOOKS_FOLDER / "book.txt").write_text(
+            "one two three four five", encoding="utf-8")
+        # The stub returns a path that does not exist on disk, so the
+        # final assembly (and cover art) is patched out of the run() path.
+        self._patchers = [
+            patch.object(converter_mod.audio, "combine_chunks",
+                         return_value=True),
+            patch.object(converter_mod.audio, "combine_chapters_to_m4b",
+                         return_value=True),
+            patch.object(converter_mod.cover, "generate_cover",
+                         return_value=None),
+        ]
+        for patcher in self._patchers:
+            patcher.start()
+        test_case.addCleanup(self.cleanup)
+
+    def cleanup(self):
+        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._orig
+        for patcher in self._patchers:
+            patcher.stop()
+        self._books_tmp.cleanup()
+        self._output_tmp.cleanup()
+
+    def build(self, progress=None, cancel=None):
+        # Patch the TTS client construction so the real constructor runs
+        # (exercising the progress/cancel wiring) without dialing a server.
+        with patch.object(converter_mod, "QwenTTSClient",
+                          return_value=MagicMock()):
+            converter = AudiobookConverter(
+                voice_mode=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_QWEN,
+                output_format="mp3", language="English",
+                progress=progress, cancel=cancel)
+        converter.tts.process_chunk_with_retry.return_value = "chunk_0001.wav"
+        converter._book_files = [converter_mod.BOOKS_FOLDER / "book.txt"]
+        converter._planned = [(converter_mod.BOOKS_FOLDER / "book.txt",
+                               "book_Vivian")]
+        return converter
+
+
+class ProgressEventTests(unittest.TestCase):
+    def setUp(self):
+        self.fixture = _ConvertFixture(self)
+
+    def test_run_emits_book_chunks_done(self):
+        events = []
+        converter = self.fixture.build(progress=events.append)
+        ok = converter.run()
+        self.assertTrue(ok)
+        kinds = [event["kind"] for event in events]
+        self.assertEqual(kinds, ["book", "chunks", "chunk_done",
+                                 "book_done", "done"])
+        self.assertEqual(events[0]["name"], "book.txt")
+        self.assertEqual(events[-1]["ok"], 1)
+
+    def test_run_suppresses_console_prints_when_progress_set(self):
+        buf = io.StringIO()
+        converter = self.fixture.build(progress=lambda e: None)
+        with redirect_stdout(buf):
+            converter.run()
+        # The banner/summary/chunk prints are replaced by events.
+        out = buf.getvalue()
+        self.assertNotIn("CONVERSION SUMMARY", out)
+        self.assertNotIn("PROCESSING", out)
+        self.assertNotIn("completed", out)
+
+    def test_chunk_failed_sets_error_state(self):
+        events = []
+        converter = self.fixture.build(progress=events.append)
+        converter.tts.process_chunk_with_retry.return_value = None
+        converter.run()
+        self.assertIn("chunk_failed",
+                      [event["kind"] for event in events])
+        self.assertEqual(events[-1]["kind"], "done")
+        self.assertEqual(events[-1]["ok"], 0)
+
+
+class CancelTests(unittest.TestCase):
+    def setUp(self):
+        self.fixture = _ConvertFixture(self)
+
+    def test_cancel_between_chunks_aborts_and_emits_cancelled(self):
+        events = []
+        cancel = threading.Event()
+        converter = self.fixture.build(progress=events.append, cancel=cancel)
+        # Cancel as the first chunk completes; the next chunk's pre-check
+        # must raise ConversionCancelled before requesting it.
+        def generate(chunk_num, text):
+            cancel.set()
+            return "chunk_0001.wav"
+
+        converter.tts.process_chunk_with_retry.side_effect = generate
+        with patch.object(converter_mod, "chunking") as mk_chunking:
+            mk_chunking.split_into_chunks.return_value = [
+                "one two", "three four", "five"]
+            converter.run()
+        kinds = [event["kind"] for event in events]
+        self.assertIn("cancelled", kinds)
+        self.assertEqual(events[-1]["kind"], "done")
+        self.assertTrue(events[-1]["cancelled"])
+
+    def test_check_cancelled_raises_when_event_set(self):
+        cancel = threading.Event()
+        cancel.set()
+        converter = self.fixture.build(cancel=cancel)
+        with self.assertRaises(ConversionCancelled):
+            converter._check_cancelled()
+
+    def test_check_cancelled_silent_when_not_set(self):
+        converter = self.fixture.build(cancel=threading.Event())
+        converter._check_cancelled()  # no raise
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py
index a0545cc..5f91a61 100644
--- a/app/tests/test_hub.py
+++ b/app/tests/test_hub.py
@@ -917,92 +917,125 @@ class SelectSpecTests(unittest.TestCase):
         self.assertIsNone(hub._select_spec(st, {}))
 
 
-class RunConversionTests(unittest.TestCase):
-    """_run_conversion: autostart, hint-when-manual, and stop-after."""
-
-    def test_autostart_starts_server_then_converts(self):
-        spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"])
-        status = BackendStatus("qwen", "qwen-tts", installed=True,
-                               configured=True, running=False,
-                               servers=[spec])
-        kwargs = {"autostart": "qwen-custom"}
-        with patch.object(hub, "detect_all", return_value=[status]), \
-                patch.object(hub, "_find_spec", return_value=spec), \
-                patch.object(hub.servers, "start", return_value=True) as mk_start, \
-                patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \
-                patch("builtins.input", return_value="n") as mk_input, \
-                patch.object(hub.servers, "stop") as mk_stop:
-            hub._run_conversion("qwen", kwargs)
-        mk_start.assert_called_once_with(spec)
-        mk_conv.assert_called_once()
-        # User declined stopping → stop not called.
-        mk_stop.assert_not_called()
-
-    def test_autostart_stop_when_user_says_yes(self):
-        spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"])
-        status = BackendStatus("qwen", "qwen-tts", installed=True,
-                               configured=True, running=False,
-                               servers=[spec])
-        kwargs = {"autostart": "qwen-custom"}
-        with patch.object(hub, "detect_all", return_value=[status]), \
-                patch.object(hub, "_find_spec", return_value=spec), \
-                patch.object(hub.servers, "start", return_value=True), \
-                patch.object(hub.audiobook, "convert", return_value=0), \
-                patch("builtins.input", return_value="y"), \
-                patch.object(hub.servers, "stop") as mk_stop:
-            hub._run_conversion("qwen", kwargs)
-        mk_stop.assert_called_once_with("qwen-custom")
-
-    def test_autostart_aborts_when_server_fails(self):
-        spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"])
-        status = BackendStatus("qwen", "qwen-tts", installed=True,
-                               configured=True, running=False,
-                               launch_hint="hint cmd", servers=[spec])
-        kwargs = {"autostart": "qwen-custom"}
-        with patch.object(hub, "detect_all", return_value=[status]), \
-                patch.object(hub, "_find_spec", return_value=spec), \
-                patch.object(hub.servers, "start", return_value=False), \
-                patch.object(hub.audiobook, "convert") as mk_conv, \
-                patch.object(hub.servers, "stop") as mk_stop:
-            hub._run_conversion("qwen", kwargs)
-        mk_conv.assert_not_called()
-        mk_stop.assert_not_called()
-
-    def test_no_autostart_prints_hint_when_not_running(self):
-        status = BackendStatus("qwen", "qwen-tts", installed=True,
-                               configured=True, running=False,
-                               launch_hint="the-hint")
-        with patch.object(hub, "detect_all", return_value=[status]), \
-                patch.object(hub.audiobook, "convert", return_value=0) as mk_conv:
-            hub._run_conversion("qwen", {})
-        mk_conv.assert_called_once()
-
-    def test_remote_conversion_skips_setup_checks(self):
-        # A remote conversion targets an external server: no autostart, no
-        # "not fully set up" warning, no launch hint — just convert.
-        with patch.object(hub, "detect_all", return_value=[]) as mk_detect, \
-                patch.object(hub.audiobook, "convert", return_value=0) as mk_conv:
-            hub._run_conversion("audiocpp", {"api_url": "http://10.0.0.5:8080"})
-        mk_conv.assert_called_once_with(
-            backend="audiocpp", api_url="http://10.0.0.5:8080")
+class PrepareRunConfigTests(unittest.TestCase):
+    """_prepare_run_config: the run view's inputs from the accepted form."""
+
+    def _spec(self, name="qwen-custom", url="http://127.0.0.1:7860"):
+        return ServerSpec(name, url, ["x"])
+
+    def test_remote_targets_the_api_url(self):
+        with patch.object(hub, "detect_all", return_value=[]) as mk_detect:
+            cfg = hub._prepare_run_config(
+                "audiocpp", {"api_url": "http://10.0.0.5:8080"})
+        self.assertEqual(cfg.server_url, "http://10.0.0.5:8080")
+        self.assertIsNone(cfg.autostart_spec)
+        self.assertEqual(cfg.server_identity, "audiocpp")
+        self.assertIn("remote", cfg.backend_label)
         # The remote path never re-detects or touches managed-instance state.
         mk_detect.assert_not_called()
 
-    def test_managed_conversion_warns_when_port_occupied_by_other_server(self):
+    def test_autostart_sets_the_spec_and_pops_the_flag(self):
+        spec = self._spec()
+        kwargs = {"autostart": "qwen-custom"}
+        with patch.object(hub, "detect_all", return_value=[]), \
+                patch.object(hub, "_find_spec", return_value=spec):
+            cfg = hub._prepare_run_config("qwen", kwargs)
+        self.assertIs(cfg.autostart_spec, spec)
+        self.assertEqual(cfg.server_name, "qwen-custom")
+        self.assertNotIn("autostart", kwargs)
+
+    def test_autostart_with_missing_spec_continues_with_notice(self):
+        kwargs = {"autostart": "gone"}
+        with patch.object(hub, "detect_all", return_value=[]), \
+                patch.object(hub, "_find_spec", return_value=None):
+            cfg = hub._prepare_run_config("qwen", kwargs)
+        self.assertIsNone(cfg.autostart_spec)
+        self.assertIn("gone", cfg.notice)
+
+    def test_managed_conversion_flags_foreign_server_on_port(self):
         spec = ServerSpec("audiocpp", "http://127.0.0.1:8080", ["x"])
         status = BackendStatus("audiocpp", "audio.cpp", installed=True,
-                               configured=True, running=False,
-                               servers=[spec])
+                               configured=True, servers=[spec])
         with patch.object(hub, "detect_all", return_value=[status]), \
-                patch.object(hub, "_select_spec", return_value=spec), \
                 patch("backends.common.server_running", return_value=True), \
-                patch.object(hub.servers, "alive", return_value=False), \
-                patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \
-                patch("builtins.print") as mk_print:
-            hub._run_conversion("audiocpp", {})
-        mk_conv.assert_called_once()
-        printed = " ".join(str(call.args[0]) for call in mk_print.call_args_list)
-        self.assertIn("did not start", printed)
+                patch.object(hub.servers, "alive", return_value=False):
+            cfg = hub._prepare_run_config("audiocpp", {})
+        self.assertEqual(cfg.server_url, spec.url)
+        self.assertIsNone(cfg.autostart_spec)
+        self.assertIn("did not start", cfg.notice)
+
+    def test_managed_not_running_sets_url_without_autostart(self):
+        spec = self._spec()
+        status = BackendStatus("qwen", "qwen-tts", installed=True,
+                               configured=True, servers=[spec])
+        with patch.object(hub, "detect_all", return_value=[status]), \
+                patch("backends.common.server_running", return_value=False):
+            cfg = hub._prepare_run_config("qwen", {"clone": None})
+        self.assertEqual(cfg.server_url, spec.url)
+        self.assertIsNone(cfg.autostart_spec)
+
+
+class PreflightTests(unittest.TestCase):
+    """_preflight: overwrite prompts run in the TUI, plan stashed in kwargs."""
+
+    def _cmd(self):
+        return ("convert", "qwen", {"clone": None, "output_format": "mp3"})
+
+    def test_books_stashed_on_kwargs(self):
+        stdscr = object()
+        cmd = self._cmd()
+        with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+                          return_value=(["book.txt"], [("book.txt", "x")])) \
+                as mk_pre:
+            self.assertTrue(hub._preflight(stdscr, cmd))
+        self.assertEqual(cmd[2]["book_files"], ["book.txt"])
+        self.assertEqual(cmd[2]["planned"], [("book.txt", "x")])
+        # The confirm callback passed to preflight is a TUI yes/no.
+        confirm = mk_pre.call_args.kwargs["confirm"]
+        with patch.object(hub.tui, "confirm", return_value=True) as mk_confirm:
+            self.assertTrue(confirm("overwrite?", True))
+        mk_confirm.assert_called_once()
+
+    def test_nothing_to_convert_flashes_and_returns_false(self):
+        stdscr = object()
+        with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+                          return_value=([], [])), \
+                patch.object(hub.tui, "flash") as mk_flash:
+            self.assertFalse(hub._preflight(stdscr, self._cmd()))
+        mk_flash.assert_called_once()
+
+    def test_all_skipped_flashes_and_returns_false(self):
+        stdscr = object()
+        with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+                          return_value=(["book.txt"], [])), \
+                patch.object(hub.tui, "flash") as mk_flash:
+            self.assertFalse(hub._preflight(stdscr, self._cmd()))
+        mk_flash.assert_called_once()
+
+
+class DispatchConversionTests(unittest.TestCase):
+    """_dispatch_conversion: builds the config and runs the run view."""
+
+    def test_runs_run_view_inside_curses(self):
+        from tests.test_tui import FakeScreen
+
+        class FakeView:
+            def __init__(self, scr, config):
+                self.config = config
+            def run(self):
+                pass
+
+        made = []
+        with patch.object(hub, "_prepare_run_config",
+                          return_value=hub.runview.RunConfig(
+                              backend="qwen", backend_label="qwen-tts",
+                              kwargs={}, book_files=[], planned=[])) as mk_cfg, \
+                patch("curses.wrapper",
+                      side_effect=lambda cb: cb(FakeScreen())) as mk_wrapper, \
+                patch.object(hub.runview, "RunView", FakeView):
+            hub._dispatch_conversion("qwen", {})
+        mk_cfg.assert_called_once()
+        mk_wrapper.assert_called_once()
 
 
 class AddAutostartTests(unittest.TestCase):
diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py
new file mode 100644
index 0000000..73cae5f
--- /dev/null
+++ b/app/tests/test_runview.py
@@ -0,0 +1,210 @@
+"""Tests for the run view (ui/runview.py) — the conversion status screen.
+
+The view is driven the same way as the other TUI widgets: the fake curses
+module and recording screen from test_tui stand in for a terminal, the
+worker/monitor threads are stubbed, and events are fed through the view's
+own queue to exercise state transitions, rendering, and the Esc/q
+cancel → stop-server flow.
+"""
+
+import sys
+import unittest
+from unittest.mock import patch
+
+from tests.test_tui import FakeCurses, FakeScreen
+from ui import runview
+
+
+def _config(**overrides):
+    kwargs = dict(backend="audiocpp", backend_label="audio.cpp",
+                  kwargs={}, book_files=["book.txt"], planned=["book.txt"],
+                  server_name="audiocpp",
+                  server_url="http://127.0.0.1:8080",
+                  server_identity="audiocpp")
+    kwargs.update(overrides)
+    return runview.RunConfig(**kwargs)
+
+
+class _FakeTui:
+    """Stand-in for the curses module (installed into sys.modules)."""
+
+    def setUp(self):
+        self.curses = FakeCurses()
+        patcher = patch.dict(sys.modules, {"curses": self.curses})
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        runview.tui._THEME.clear()
+        self.addCleanup(runview.tui._THEME.clear)
+
+    def make_view(self, keys=(), width=80, height=24, **cfg):
+        screen = FakeScreen(keys=keys, width=width, height=height)
+        # Patch the thread targets at the class level BEFORE construction so
+        # __init__'s Thread(target=self._worker_main) binds the stub.
+        with patch.object(runview.RunView, "_worker_main", lambda self: None), \
+                patch.object(runview.RunView, "_monitor_main",
+                             lambda self: None):
+            view = runview.RunView(screen, _config(**cfg),
+                                   clock=lambda: 1000.0)
+        return view, screen
+
+
+class FormatTests(_FakeTui, unittest.TestCase):
+    def test_format_elapsed(self):
+        self.assertEqual(runview._format_elapsed(0), "0:00")
+        self.assertEqual(runview._format_elapsed(65), "1:05")
+        self.assertEqual(runview._format_elapsed(3661), "1:01:01")
+
+    def test_fit_truncates_with_tilde(self):
+        self.assertEqual(runview._fit("hello", 3), "he~")
+        self.assertEqual(runview._fit("hi", 10), "hi")
+
+    def test_wrap_wraps_on_word_boundaries(self):
+        self.assertEqual(runview._wrap("aaaa bbbb cccc dddd", 12),
+                         ["aaaa bbbb", "cccc dddd"])
+
+
+class StateTransitionTests(_FakeTui, unittest.TestCase):
+    def test_boot_flow_starting_to_ready(self):
+        view, _ = self.make_view()
+        view.handle_event({"kind": "starting", "name": "audiocpp",
+                           "pid": 1, "log_path": "/tmp/x.log"})
+        self.assertEqual(view.phase, "boot")
+        self.assertEqual(view.server, "starting")
+        self.assertTrue(view.started_server)
+        view.handle_event({"kind": "ready", "name": "audiocpp",
+                           "url": "http://x"})
+        self.assertEqual(view.server, "ready")
+
+    def test_chunk_progress_updates(self):
+        view, _ = self.make_view()
+        view.handle_event({"kind": "book", "index": 1, "total": 2,
+                           "name": "book.txt"})
+        self.assertEqual(view.phase, "convert")
+        view.handle_event({"kind": "chunks", "total": 10})
+        view.handle_event({"kind": "chunk_done", "chunk": 4, "total": 10})
+        self.assertEqual(view.chunk_done, 4)
+        self.assertEqual(view.chunk_total, 10)
+
+    def test_done_all_books_is_terminal(self):
+        view, _ = self.make_view()
+        view.handle_event({"kind": "book", "index": 1, "total": 1,
+                           "name": "b"})
+        view.handle_event({"kind": "book_done", "name": "b", "ok": True})
+        view.handle_event({"kind": "done", "ok": 1, "total": 1})
+        self.assertEqual(view.phase, "done")
+
+    def test_chunk_failure_leads_to_error(self):
+        view, _ = self.make_view()
+        view.handle_event({"kind": "book", "index": 1, "total": 1,
+                           "name": "b"})
+        view.handle_event({"kind": "chunk_failed", "chunk": 3, "total": 5})
+        view.handle_event({"kind": "book_done", "name": "b", "ok": False})
+        view.handle_event({"kind": "done", "ok": 0, "total": 1})
+        self.assertEqual(view.phase, "error")
+        self.assertTrue(view.error_message)
+
+    def test_server_exit_during_boot_is_error(self):
+        view, _ = self.make_view()
+        view.handle_event({"kind": "starting", "name": "audiocpp"})
+        view.handle_event({"kind": "exited", "name": "audiocpp",
+                           "returncode": 1, "log_tail": ["boom"]})
+        self.assertEqual(view.phase, "error")
+        self.assertEqual(view.server, "error")
+        self.assertEqual(view.log_tail, ["boom"])
+
+    def test_server_down_during_convert(self):
+        view, _ = self.make_view()
+        view.handle_event({"kind": "book", "index": 1, "total": 1,
+                           "name": "b"})
+        view.handle_event({"kind": "server_down"})
+        self.assertEqual(view.server, "down")
+
+
+class RenderTests(_FakeTui, unittest.TestCase):
+    def _strings(self, screen):
+        return " ".join(text for _, _, text, _ in screen.strings)
+
+    def test_boot_screen_shows_server_and_status(self):
+        view, screen = self.make_view()
+        view.handle_event({"kind": "starting", "name": "audiocpp",
+                           "pid": 1, "log_path": "/tmp/x.log"})
+        view.render()
+        text = self._strings(screen)
+        self.assertIn("Server", text)
+        self.assertIn("audio.cpp", text)
+        self.assertIn("Status", text)
+        self.assertIn("starting", text)
+        self.assertIn("Esc or q: cancel", text)
+
+    def test_summary_screen_after_done(self):
+        view, screen = self.make_view()
+        view.handle_event({"kind": "book", "index": 1, "total": 1,
+                           "name": "book.txt"})
+        view.handle_event({"kind": "book_done", "name": "book.txt",
+                           "ok": True})
+        view.handle_event({"kind": "done", "ok": 1, "total": 1})
+        view.render()
+        text = self._strings(screen)
+        self.assertIn("completed", text)
+        self.assertIn("book.txt", text)
+        self.assertIn("press any key", text)
+
+    def test_error_screen_shows_detail_and_log(self):
+        view, screen = self.make_view(log_path="/tmp/audiobook.log")
+        view.handle_event({"kind": "book", "index": 1, "total": 1,
+                           "name": "b"})
+        view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 3})
+        view.handle_event({"kind": "book_done", "name": "b", "ok": False})
+        view.handle_event({"kind": "done", "ok": 0, "total": 1})
+        view.render()
+        text = self._strings(screen)
+        self.assertIn("failed", text)
+        self.assertIn("/tmp/audiobook.log", text)
+
+
+class RunLoopTests(_FakeTui, unittest.TestCase):
+    def test_terminal_screen_key_returns(self):
+        view, screen = self.make_view(keys=[ord("x")])
+        view._queue.put({"kind": "done", "ok": 1, "total": 1})
+        view.run()
+        # Returned to the menu without touching the server stop prompt
+        # (started_server is False).
+        self.assertEqual(view.phase, "done")
+
+    def test_esc_cancels_and_confirms_stop_server(self):
+        confirm_answers = [True, True]  # cancel? yes; stop server? yes
+        with patch.object(runview.tui, "confirm",
+                          side_effect=confirm_answers), \
+                patch.object(runview.servers, "alive", return_value=True), \
+                patch.object(runview.servers, "stop") as mk_stop:
+            view, screen = self.make_view(keys=[27, ord("x")],
+                                          autostart_spec="SPEC")
+            view.started_server = True
+            view.run()
+        mk_stop.assert_called_once_with("audiocpp")
+
+    def test_esc_decline_cancel_keeps_running(self):
+        # First Esc: "cancel?" answered No → the run continues; a second
+        # key then exits via a terminal state the test feeds.
+        with patch.object(runview.tui, "confirm",
+                          side_effect=[False]):
+            view, screen = self.make_view(keys=[27])
+            # Feeds a done event after the declined cancel so run() can exit.
+            view._queue.put({"kind": "done", "ok": 1, "total": 1})
+            screen.keys.append(ord("x"))
+            view.run()
+        self.assertEqual(view.phase, "done")
+
+    def test_stop_server_not_asked_when_dead(self):
+        with patch.object(runview.tui, "confirm", return_value=True), \
+                patch.object(runview.servers, "alive", return_value=False), \
+                patch.object(runview.servers, "stop") as mk_stop:
+            view, screen = self.make_view(keys=[27, ord("x")],
+                                          autostart_spec="SPEC")
+            view.started_server = True
+            view.run()
+        mk_stop.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/app/ui/hub.py b/app/ui/hub.py
index 77d0796..95ac06a 100644
--- a/app/ui/hub.py
+++ b/app/ui/hub.py
@@ -12,10 +12,13 @@ Esc on the main menu quits the hub ('q' mirrors Esc on every screen).
 Esc inside a sub-menu falls back to the main menu.
 """
 
+import contextlib
+import io
 import json
 import re
 import shutil
 import urllib.parse
+from datetime import datetime
 from pathlib import Path
 from typing import Callable, Optional, Tuple
 
@@ -31,9 +34,15 @@ from backends import (
 )
 from backends import audiocpp as audiocpp_backend
 from backends import faster as faster_backend
+from backends import probe as backend_probe
 from backends import qwen as qwen_backend
 from converter import config
-from converter.converter import AUDIO_FORMATS
+from converter.converter import (
+    AUDIO_FORMATS,
+    AudiobookConverter,
+    LOGS_FOLDER,
+    voice_mode_for,
+)
 from converter.tts import (
     AUDIOCPP_FAMILY_QWEN3_TTS,
     BACKEND_AUDIOCPP,
@@ -41,7 +50,7 @@ from converter.tts import (
     BACKEND_QWEN,
     normalize_language,
 )
-from ui import tui
+from ui import runview, tui
 
 _GO_BACK = object()
 
@@ -70,7 +79,7 @@ def run() -> int:
             if info is not None and command[2] < len(info.configure_actions):
                 info.configure_actions[command[2]].run()
         elif kind == "convert":
-            _run_conversion(command[1], command[2])
+            _dispatch_conversion(command[1], command[2])
         elif kind == "server":
             _run_server_action(command[1], command[2])
 
@@ -188,6 +197,8 @@ def _status_mark(status: Optional[BackendStatus]) -> Tuple[str, str, str]:
             text += " (" + ", ".join(status.running_models) + ")"
         return (text, "ok", "body")
     if status is not None and status.installed:
+        if status.models_missing and not status.running:
+            return ("installed (models missing)", "warn", "body")
         return ("installed", "warn", "body")
     return ("unavailable", "err", "dim")
 
@@ -287,9 +298,50 @@ def _convert_menu(stdscr, statuses) -> Optional[tuple]:
     if cmd is None:
         return None
     _add_autostart(cmd, statuses)
+    if not _preflight(stdscr, cmd):
+        return None
     return cmd
 
 
+def _preflight(stdscr, cmd: tuple) -> bool:
+    """Run the overwrite checks in the TUI; stash the plan on the command.
+
+    Asks every "output exists — overwrite?" question now (tui.confirm
+    instead of the console input()) so the run view itself is unattended,
+    and records the discovered books / accepted plan in the command's
+    kwargs (``book_files``/``planned``) for ``audiobook.convert``. Returns
+    False when nothing would be converted (a flash explains why), so the
+    user stays in the menu instead of entering an empty run.
+    """
+    _kind, backend, kwargs = cmd
+    voice_mode = voice_mode_for(backend, kwargs.get("voice"),
+                                kwargs.get("clone"))
+
+    def confirm(message: str, default: bool) -> bool:
+        return tui.confirm(stdscr, message, default=default,
+                           cancel_value=False)
+
+    with contextlib.redirect_stdout(io.StringIO()):
+        book_files, planned = AudiobookConverter.preflight_overwrites(
+            backend=backend, voice=kwargs.get("voice"),
+            voice_mode=voice_mode,
+            voice_clone_ref_audio=kwargs.get("clone"),
+            output_format=kwargs.get("output_format") or config.AUDIO_FORMAT,
+            instructions=kwargs.get("instructions"),
+            confirm=confirm)
+    if not book_files:
+        tui.flash(stdscr, "No books to convert. Add a .txt, .pdf or .epub "
+                  "file to the input folder first.")
+        return False
+    if not planned:
+        tui.flash(stdscr, "Nothing to convert — every existing output was "
+                  "kept.")
+        return False
+    kwargs["book_files"] = book_files
+    kwargs["planned"] = planned
+    return True
+
+
 def _gate_backend(field: dict, key: str) -> Callable:
     """A visible() that shows FIELD only when the Backend field is KEY.
 
@@ -429,6 +481,24 @@ def _audiocpp_fields(stdscr, api_url: Optional[str] = None) -> Optional[tuple]:
             # converter's default: unknown family means qwen3_tts.
             entry["family"] = AUDIOCPP_FAMILY_QWEN3_TTS
 
+    if local:
+        # Only offer entries whose model files are actually on disk: a
+        # server.json can reference a package that was never downloaded,
+        # and picking it would fail the whole run at model-load time.
+        missing = audiocpp_backend.missing_model_entries(server_json)
+        if missing:
+            missing_ids = {item["id"] for item in missing}
+            models = [entry for entry in models
+                      if entry.get("id") not in missing_ids]
+            if not models:
+                hints = audiocpp_backend.model_install_hints(checkout,
+                                                             missing)
+                message = hints[0] if hints \
+                    else "Download the models first."
+                tui.flash(stdscr, "No model files are downloaded for "
+                          f"audio.cpp. {message}")
+                return None
+
     local_voices = _list_voices(data.get("voice_dir")) \
         if data.get("voice_dir") else []
     voice_cache: dict = {}  # model id -> voices (local: shared list)
@@ -871,76 +941,122 @@ def _write_config(updates: dict) -> None:
     path.write_text(text, encoding="utf-8")
 
 
-def _run_conversion(backend: str, kwargs: dict) -> None:
-    """Run a conversion in the plain console (after the TUI returns).
+def _dispatch_conversion(backend: str, kwargs: dict) -> None:
+    """Run a conversion in the full-screen run view (its own curses session).
 
-    A remote conversion (``api_url`` in the kwargs) targets an externally-run
-    server, so no autostart is attempted and the managed instance's setup
-    state is irrelevant. Otherwise, when the convert menu recorded an
-    ``autostart`` server (the user opted to have the hub start it), spawn it
-    now and abort the conversion if it does not come up; a managed server
-    whose port is already occupied by a server this tool did not start is
-    left alone but warned about. After the conversion, offer to stop a
-    server we started.
+    ``_prepare_run_config`` turns the accepted form (plus the autostart
+    decision the convert menu recorded) into everything the run view needs;
+    the view then boots the server when required, runs the conversion with
+    progress events, and asks the cancel/stop-server questions itself. A
+    crash inside the view cancels the worker and returns to the menu
+    instead of taking the whole hub down.
     """
-    autostart = kwargs.pop("autostart", None)
-    api_url = kwargs.get("api_url")
-    if api_url:
-        print(f"[INFO] Converting against remote server at {api_url}")
-    else:
-        status = next((s for s in detect_all() if s.key == backend), None)
-        if status is not None and not status.ready and not status.running:
-            print(f"[WARNING] {status.label} is not fully set up.")
-        if autostart:
-            spec = _find_spec(autostart)
-            if spec is None:
-                print(f"[WARNING] no server named '{autostart}'; continuing")
-            elif not servers.start(spec):
-                print("[ERROR] could not start the server; aborting conversion.")
-                if status is not None and status.launch_hint:
-                    print("Start it manually and run the conversion again:")
-                    print(f"  {status.launch_hint}")
-                return
-        elif status is not None and status.servers:
-            # The managed server's port may be held by a server we did not
-            # start (its pid file is absent); the conversion would silently
-            # talk to that server, so call it out.
-            spec = _select_spec(status, kwargs)
-            if spec is not None and common.server_running(spec.url) \
-                    and not servers.alive(spec.name):
-                print(f"[WARNING] A server this tool did not start is already "
-                      f"running at {spec.url}; the conversion will talk to it. "
-                      f"Stop it (or change the port) to use the managed "
-                      f"{status.label} instance.")
-            elif status.launch_hint:
-                print("[INFO] Make sure the server is running. Start it with:")
-                print(f"  {status.launch_hint}")
-    try:
-        audiobook.convert(backend=backend, **kwargs)
-    finally:
-        if autostart:
-            _maybe_stop_server(autostart)
+    import curses
+    run_config = _prepare_run_config(backend, kwargs)
+    if run_config is None:
+        return
+    holder: dict = {}
 
+    def main(stdscr) -> None:
+        view = runview.RunView(stdscr, run_config)
+        holder["view"] = view
+        view.run()
 
-def _maybe_stop_server(name: str) -> None:
-    """Ask (in the plain console) whether to stop a server we auto-started."""
     try:
-        ans = input(f"\n[?] Stop the '{name}' server now? [y/N] ").strip().lower()
-    except EOFError:
-        return
-    if ans in ("y", "yes"):
-        servers.stop(name)
+        curses.wrapper(main)
+    except tui.WizardCancelled:
+        pass
+    except KeyboardInterrupt:
+        pass
+    except Exception as exc:  # noqa: BLE001 - keep the hub alive
+        view = holder.get("view")
+        if view is not None:
+            view._cancel.set()
+            view._worker.join(timeout=30)
+        print(f"[ERROR] The run view failed: {exc}")
+    finally:
+        try:
+            curses.curs_set(1)  # restore the text cursor hidden by the TUI
+        except Exception:
+            pass
+
+
+def _prepare_run_config(backend: str, kwargs: dict
+                        ) -> Optional[runview.RunConfig]:
+    """Build the run view's config from the accepted conversion kwargs.
+
+    A remote conversion (``api_url``) targets an externally-run server, so
+    no autostart is attempted and the managed instance's setup state is
+    irrelevant. Otherwise, when the convert menu recorded an ``autostart``
+    server (the server was not running), the run view boots it first; a
+    managed server whose port is already occupied by a server this tool
+    did not start is left alone but flagged with a notice. Returns None
+    when the backend disappeared between the menu and the dispatch.
+    """
+    label = backend
+    info = get(backend)
+    if info is not None:
+        label = info.label
+    log_path = str(LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log")
+    autostart = kwargs.pop("autostart", None)
+    api_url = kwargs.get("api_url")
+
+    if api_url:
+        identity = _remote_identity(backend, kwargs)
+        return runview.RunConfig(
+            backend=backend, backend_label=f"{label} [remote]",
+            kwargs=kwargs, book_files=kwargs.get("book_files") or [],
+            planned=kwargs.get("planned") or [],
+            server_url=api_url, server_identity=identity,
+            log_path=log_path)
+
+    status = next((s for s in detect_all() if s.key == backend), None)
+    notice = ""
+    spec: Optional[ServerSpec] = None
+    if autostart:
+        spec = _find_spec(autostart)
+    elif status is not None:
+        spec = _select_spec(status, kwargs)
+        if spec is not None and common.server_running(spec.url) \
+                and not servers.alive(spec.name):
+            notice = (f"a server this tool did not start is running at "
+                      f"{spec.url} — the conversion will talk to it")
+    if autostart and spec is None:
+        # The recorded server vanished (backend reconfigured meanwhile):
+        # converting without it is still meaningful, so continue.
+        notice = (f"no server named '{autostart}' — starting it was skipped")
+    return runview.RunConfig(
+        backend=backend, backend_label=label, kwargs=kwargs,
+        book_files=kwargs.get("book_files") or [],
+        planned=kwargs.get("planned") or [],
+        server_name=spec.name if spec is not None else None,
+        server_url=spec.url if spec is not None else None,
+        server_identity=spec.identity if spec is not None else None,
+        autostart_spec=spec if autostart else None,
+        log_path=log_path, notice=notice)
+
+
+def _remote_identity(backend: str, kwargs: dict) -> Optional[str]:
+    """The probe identity of the remote server a conversion targets."""
+    if backend == BACKEND_AUDIOCPP:
+        return backend_probe.IDENTITY_AUDIOCPP
+    if backend == BACKEND_QWEN:
+        return backend_probe.IDENTITY_QWEN_CLONE if kwargs.get("clone") \
+            else backend_probe.IDENTITY_QWEN_CUSTOM
+    if backend == BACKEND_FASTER:
+        return backend_probe.IDENTITY_FASTER
+    return None
 
 
 def _add_autostart(cmd: tuple, statuses) -> None:
     """Auto-start the conversion's target server when it isn't running.
 
     Records the chosen server spec name as ``kwargs['autostart']`` for
-    ``_run_conversion`` to act on. The user already accepted the run on the
-    Generate! screen, so no start-server prompt is asked here — the server
-    is simply started. Mode-aware for qwen (custom vs clone). Remote
-    conversions (a ``api_url`` in the kwargs) never autostart: the server
-    is external to this tool.
+    ``_prepare_run_config`` to act on. The user already accepted the run on
+    the Generate! screen, so no start-server prompt is asked here — the
+    server is simply started. Mode-aware for qwen (custom vs clone).
+    Remote conversions (an ``api_url`` in the kwargs) never autostart: the
+    server is external to this tool.
     """
     _, key, kwargs = cmd
     if kwargs.get("api_url"):
diff --git a/app/ui/runview.py b/app/ui/runview.py
new file mode 100644
index 0000000..d65978d
--- /dev/null
+++ b/app/ui/runview.py
@@ -0,0 +1,643 @@
+#!/usr/bin/env python3
+"""The full-screen run view: server boot + conversion on one screen.
+
+Replaces the old plain-console drop after "Generate!": instead of dumping
+the user into scrolling log output, this widget keeps them in the TUI and
+shows the two processes that matter — the TTS server (top) and the
+conversion (bottom, with a chunk progress bar and elapsed time).
+
+The screen is fed by two threads the widget spawns:
+
+  * the worker runs the same code the console path runs —
+    ``backends.servers.start`` (when the conversion needs to boot a managed
+    server; its progress events stream in as they happen) followed by
+    ``audiobook.convert`` with a ``progress`` callback — so behavior is
+    identical to the CLI, only the presentation differs;
+  * a monitor polls the server URL while the conversion runs and reports
+    when it stops answering.
+
+Esc and 'q' do the same thing everywhere: a confirmation to cancel
+processing, then (when this run started the server) a confirmation to shut
+it down, then back to the hub menu. Errors (the server exits while
+booting, the server stops mid-conversion, a chunk fails and the book
+aborts) put the corresponding state into error and wait for a key press
+before returning to the menu, so the failure is never scrolled away.
+"""
+
+import contextlib
+import io
+import threading
+import time
+from dataclasses import dataclass, field
+from queue import Empty, Queue
+from typing import Callable, List, Optional
+
+from backends import common, servers
+from ui import tui
+
+# Terminal states: the run is over and the screen waits for a key.
+_TERMINAL = ("done", "error", "cancelled")
+
+# Server panel states -> (text, theme kind) with the elapsed clock added
+# while booting.
+_SERVER_STATES = {
+    "starting": ("starting", "warn"),
+    "ready": ("ready", "ok"),
+    "processing": ("processing", "ok"),
+    "down": ("not responding", "err"),
+    "error": ("error", "err"),
+    "stopped": ("stopped", "info"),
+}
+
+# Redraw cadence / poll cadence (milliseconds / seconds).
+_DRAW_TIMEOUT_MS = 250
+_MONITOR_INTERVAL = 2.0
+
+
+@dataclass
+class RunConfig:
+    """Everything the run view needs to execute one conversion.
+
+    BACKEND/BACKEND_LABEL identify the chosen backend (label for display);
+    KWARGS are the converter keyword arguments the hub collected (voice,
+    clone, output format, api_url, ...); BOOK_FILES/PLANNED carry the
+    pre-flight overwrite result so the questions are not asked again.
+
+    SERVER_NAME/SERVER_URL/SERVER_IDENTITY describe the TTS server the
+    conversion talks to (the name is the backends.ServerSpec name; the URL
+    is what the monitor polls). AUTOSTART_SPEC, when not None, is the
+    ServerSpec the worker boots first (the hub only sets it when the
+    server is not already running). LOG_PATH names the converter's log
+    file for the error screen's "details" hint. NOTICE is an optional
+    warning line shown under the progress panel (e.g. a foreign server
+    holding the managed port).
+    """
+    backend: str
+    backend_label: str
+    kwargs: dict
+    book_files: list
+    planned: list
+    server_name: Optional[str] = None
+    server_url: Optional[str] = None
+    server_identity: Optional[str] = None
+    autostart_spec: object = None
+    log_path: str = ""
+    notice: str = ""
+
+
+class RunView:
+    """Draws and drives one conversion run; see the module docstring."""
+
+    def __init__(self, scr, config: RunConfig,
+                 clock: Callable[[], float] = time.time):
+        import curses
+        self.curses = curses
+        self.scr = scr
+        self.config = config
+        self.theme = tui._ensure_theme(curses)
+        self._clock = clock
+        # -- state -----------------------------------------------------
+        self.phase = "boot"     # boot | convert | done | error | cancelled
+        self.server = "starting"
+        self.server_message = ""
+        self.log_tail: List[str] = []
+        self.book: Optional[tuple] = None      # (index, total, name)
+        self.chapter: Optional[tuple] = None   # (index, total)
+        self.chunk_done = 0
+        self.chunk_total = 0
+        self.book_results: List[tuple] = []    # (name, ok)
+        self.error_message = ""
+        self.cancelled = False
+        self.cancelling = False
+        self.started_server = False
+        self.finished_at: Optional[float] = None
+        self.boot_started: Optional[float] = None
+        self.convert_started: Optional[float] = None
+        self.server_log_path = ""
+        # -- threads ---------------------------------------------------
+        self._queue: Queue = Queue()
+        self._cancel = threading.Event()
+        self._monitor_stop = threading.Event()
+        self._worker = threading.Thread(target=self._worker_main,
+                                        daemon=True)
+
+    # ------------------------------------------------------------------
+    # Event handling (pure state transitions; no drawing)
+    # ------------------------------------------------------------------
+
+    def handle_event(self, event: dict) -> None:
+        """Fold one worker/monitor event into the view state."""
+        kind = event.get("kind")
+        if kind == "starting":
+            self.phase = "boot"
+            self.server = "starting"
+            self.boot_started = self._now()
+            self.started_server = True
+            self.server_log_path = event.get("log_path") or ""
+        elif kind == "running":
+            self.server = "ready"
+            self.boot_started = self.boot_started or self._now()
+        elif kind == "ready":
+            self.server = "ready"
+        elif kind in ("exited", "timeout"):
+            self.server = "error"
+            self.server_message = {
+                "exited": f"server exited with code "
+                          f"{event.get('returncode')}",
+                "timeout": "server did not become ready in time",
+            }[kind]
+            self.log_tail = list(event.get("log_tail") or [])
+            self._finish("error")
+        elif kind == "cancelled":
+            self.cancelled = True
+            if self.server in ("starting", "ready", "processing"):
+                self.server = "stopped"
+            self._finish("cancelled")
+        elif kind == "server_down":
+            if self.phase == "convert":
+                self.server = "down"
+        elif kind == "book":
+            self.phase = "convert"
+            self.book = (event.get("index"), event.get("total"),
+                         event.get("name") or "")
+            self.chapter = None
+            self.chunk_done = 0
+            self.chunk_total = 0
+            self.convert_started = self.convert_started or self._now()
+            if self.server == "ready":
+                self.server = "processing"
+        elif kind == "chapter":
+            self.chapter = (event.get("index"), event.get("total"))
+            self.chunk_done = 0
+            self.chunk_total = 0
+        elif kind == "chunks":
+            self.chunk_total = event.get("total") or 0
+            self.chunk_done = 0
+        elif kind == "chunk_done":
+            self.chunk_done = event.get("chunk") or self.chunk_done
+            self.chunk_total = event.get("total") or self.chunk_total
+            if self.server in ("ready", "processing"):
+                self.server = "processing"
+        elif kind == "chunk_failed":
+            self.error_message = (f"chunk {event.get('chunk')}/"
+                                  f"{event.get('total')} failed")
+            if self.server in ("ready", "processing"):
+                self.server = "ready"
+        elif kind == "book_done":
+            self.book_results.append((event.get("name") or "?",
+                                      bool(event.get("ok"))))
+        elif kind == "book_failed":
+            self.book_results.append((event.get("name") or "?", False))
+            self.error_message = self.error_message or \
+                (event.get("error") or "conversion failed")
+        elif kind == "done":
+            ok = event.get("ok") or 0
+            total = event.get("total") or 0
+            if event.get("cancelled"):
+                self.cancelled = True
+                self._finish("cancelled")
+            elif total and ok >= total and not self.error_message:
+                self._finish("done")
+            else:
+                self.error_message = self.error_message or \
+                    f"{total - ok} of {total} book(s) failed"
+                self._finish("error")
+        elif kind == "error":
+            self.error_message = str(event.get("message") or "error")
+            self._finish("error")
+        elif kind == "worker_exit":
+            if self.phase not in _TERMINAL:
+                self.error_message = self.error_message or \
+                    "the conversion ended unexpectedly"
+                self._finish("error")
+
+    def _finish(self, phase: str) -> None:
+        """Enter a terminal phase, freezing the elapsed clock."""
+        self.phase = phase
+        if self.finished_at is None:
+            self.finished_at = self._now()
+
+    def _now(self) -> float:
+        return self._clock()
+
+    # ------------------------------------------------------------------
+    # Threads
+    # ------------------------------------------------------------------
+
+    def _worker_main(self) -> None:
+        """Boot the server (when asked) and run the conversion."""
+        import audiobook
+        config = self.config
+        try:
+            with contextlib.redirect_stdout(io.StringIO()):
+                if config.autostart_spec is not None:
+                    ok = servers.start(config.autostart_spec,
+                                       progress=self._queue.put,
+                                       cancel=self._cancel)
+                    if not ok:
+                        if self._cancel.is_set() and self.phase != "error":
+                            self._queue.put({"kind": "cancelled"})
+                        return
+                if self._cancel.is_set():
+                    self._queue.put({"kind": "cancelled"})
+                    return
+                audiobook.convert(backend=config.backend,
+                                  progress=self._queue.put,
+                                  cancel=self._cancel,
+                                  book_files=config.book_files,
+                                  planned=config.planned,
+                                  **config.kwargs)
+        except Exception as exc:  # noqa: BLE001 - reported to the view
+            self._queue.put({"kind": "error", "message": f"{exc}"})
+        finally:
+            self._queue.put({"kind": "worker_exit"})
+
+    def _monitor_main(self) -> None:
+        """Watch the server URL while converting; report when it drops."""
+        url = self.config.server_url
+        if not url:
+            return
+        # Give a booting server the full start window before judging it.
+        while not self._monitor_stop.wait(_MONITOR_INTERVAL):
+            if self.phase in _TERMINAL:
+                return
+            if self.phase != "convert":
+                continue
+            if not common.server_running(url):
+                self._queue.put({"kind": "server_down"})
+                return
+
+    # ------------------------------------------------------------------
+    # Main loop
+    # ------------------------------------------------------------------
+
+    def run(self) -> None:
+        """Run the view until the user leaves the terminal screen."""
+        scr = self.scr
+        try:
+            self.scr.timeout(_DRAW_TIMEOUT_MS)
+        except Exception:
+            pass
+        self._worker.start()
+        monitor = threading.Thread(target=self._monitor_main, daemon=True)
+        monitor.start()
+        try:
+            while True:
+                self._drain()
+                self.render()
+                key = self._get_key()
+                if key is None:
+                    continue
+                if self.phase in _TERMINAL:
+                    self._confirm_stop_server()
+                    return
+                if key in (27, ord("q"), 3) and not self.cancelling:
+                    if self._prompt_cancel():
+                        return
+        finally:
+            self._monitor_stop.set()
+            self._cancel.set()
+
+    def _get_key(self) -> Optional[int]:
+        """One key from the screen (None on the redraw timeout)."""
+        try:
+            key = self.scr.getch()
+        except KeyboardInterrupt:
+            return 3
+        if key == -1:
+            return None
+        return key
+
+    def _drain(self) -> None:
+        """Fold every queued event into the state."""
+        while True:
+            try:
+                event = self._queue.get_nowait()
+            except Empty:
+                return
+            self.handle_event(event)
+
+    def _prompt_cancel(self) -> bool:
+        """The Esc/q flow: confirm cancel, then confirm stopping the server.
+
+        Returns True when the run view should return to the menu (the run
+        is over); False when the user changed their mind and the run keeps
+        going.
+        """
+        self._blocking()
+        answer = tui.confirm(self.scr, "Cancel processing?", default=False,
+                             cancel_value=False)
+        if not answer:
+            self._nonblocking()
+            return False
+        self.cancelling = True
+        self._cancel.set()
+        # When this run booted the server, offer to shut it down too (the
+        # boot path kills it itself when cancelled before ready).
+        self._confirm_stop_server()
+        # Wait for the worker to wind down so the hub menu shows the real
+        # backend state (and the summary screen is drawn at least once).
+        self._worker.join(timeout=60)
+        self._drain()
+        self.render()
+        # One more key press acknowledges the final screen.
+        self._blocking()
+        try:
+            self.scr.getch()
+        except KeyboardInterrupt:
+            pass
+        return True
+
+    def _confirm_stop_server(self) -> None:
+        """Ask whether to stop the server this run started (once)."""
+        if not self.started_server or self._server_stopped_confirmed:
+            return
+        self._server_stopped_confirmed = True
+        name = self.config.server_name
+        if not name or not servers.alive(name):
+            return
+        self._blocking()
+        answer = tui.confirm(self.scr,
+                             f"Stop the '{name}' server now?", default=True,
+                             cancel_value=False)
+        if answer:
+            with contextlib.redirect_stdout(io.StringIO()):
+                servers.stop(name)
+            self.server = "stopped"
+
+    def _blocking(self) -> None:
+        """Make getch block (used while a confirm dialog owns the screen)."""
+        try:
+            self.scr.timeout(-1)
+        except Exception:
+            pass
+
+    def _nonblocking(self) -> None:
+        """Restore the redraw-cadence getch timeout."""
+        try:
+            self.scr.timeout(_DRAW_TIMEOUT_MS)
+        except Exception:
+            pass
+
+    _server_stopped_confirmed = False
+
+    # ------------------------------------------------------------------
+    # Drawing
+    # ------------------------------------------------------------------
+
+    def render(self) -> None:
+        """Repaint the whole screen from the current state."""
+        curses, theme = self.curses, self.theme
+        scr = self.scr
+        scr.erase()
+        height, width = scr.getmaxyx()
+        if height < 14 or width < 46:
+            _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, " Converting audiobooks ", theme["title"])
+
+        inner_x = 3
+        label_w = 9          # "Server", "Status", "Chunk", "Elapsed"
+        value_x = inner_x + label_w + 1
+        value_w = width - value_x - 3
+
+        # -- server panel ------------------------------------------------
+        y = 2
+        url = self.config.server_url or "not managed"
+        _text(scr, theme, y, inner_x, "Server".ljust(label_w), theme["dim"])
+        _text(scr, theme, y, value_x,
+              _fit(f"{self.config.backend_label} @ {url}", value_w),
+              theme["body"])
+        y += 1
+        state_text, state_kind = _SERVER_STATES.get(
+            self.server, (self.server, "info"))
+        if self.server == "starting" and self.boot_started is not None:
+            state_text += f" ({int(self._now() - self.boot_started)}s)"
+        if self.config.autostart_spec is None and self.server == "ready":
+            state_text += " (external)"
+        _text(scr, theme, y, inner_x, "Status".ljust(label_w), theme["dim"])
+        _text(scr, theme, y, value_x, _fit(state_text, value_w),
+              theme.get(state_kind, theme["body"]))
+        y += 2
+
+        # -- separator ---------------------------------------------------
+        _sep(scr, curses, theme, y, width)
+        y += 2
+
+        if self.phase in _TERMINAL:
+            y = self._draw_summary(scr, theme, y, inner_x, label_w,
+                                   value_x, value_w, width)
+        else:
+            y = self._draw_progress(scr, theme, y, inner_x, label_w,
+                                    value_x, value_w, width)
+
+        # -- footer ------------------------------------------------------
+        if self.cancelling and self.phase not in _TERMINAL:
+            footer = "cancelling..."
+        elif self.phase in _TERMINAL:
+            footer = "press any key to return to the menu"
+        else:
+            footer = "Esc or q: cancel"
+        _text(scr, theme, height - 2, 2, _fit(footer, width - 4),
+              theme["dim"])
+        scr.refresh()
+
+    def _draw_progress(self, scr, theme, y, inner_x, label_w, value_x,
+                       value_w, width) -> int:
+        """The live panel: book, chapter, chunk bar, elapsed, message."""
+        # Book line
+        if self.book is not None:
+            index, total, name = self.book
+            book_text = f"{index}/{total}  {name}"
+        else:
+            book_text = "waiting..." if self.phase == "convert" else "-"
+        _text(scr, theme, y, inner_x, "Book".ljust(label_w), theme["dim"])
+        _text(scr, theme, y, value_x, _fit(book_text, value_w), theme["body"])
+        y += 1
+        # Chapter line (only while a multi-chapter book is converting)
+        if self.chapter is not None:
+            _text(scr, theme, y, inner_x, "Chapter".ljust(label_w),
+                  theme["dim"])
+            _text(scr, theme, y, value_x,
+                  _fit(f"{self.chapter[0]}/{self.chapter[1]}", value_w),
+                  theme["body"])
+            y += 1
+        # Chunk bar
+        bar_label = "Chunk".ljust(label_w)
+        _text(scr, theme, y, inner_x, bar_label, theme["dim"])
+        bar_x = value_x
+        bar_room = max(10, value_w - 12)
+        filled = 0
+        if self.chunk_total:
+            filled = round(bar_room * self.chunk_done / self.chunk_total)
+            filled = max(0, min(bar_room, filled))
+        try:
+            scr.addstr(y, bar_x, " " * filled, theme["bar"])
+        except Exception:
+            pass
+        _text(scr, theme, y, bar_x + bar_room + 1,
+              f"{self.chunk_done}/{self.chunk_total or '?'}",
+              theme["accent"])
+        y += 1
+        # Elapsed
+        started = self.convert_started or self.boot_started or self._now()
+        _text(scr, theme, y, inner_x, "Elapsed".ljust(label_w), theme["dim"])
+        _text(scr, theme, y, value_x, _format_elapsed(self._now() - started),
+              theme["body"])
+        y += 2
+        # Message line (last error / current activity)
+        if self.error_message:
+            _text(scr, theme, y, inner_x,
+                  _fit(self.error_message, width - inner_x - 3),
+                  theme["err"])
+            y += 1
+        elif self.server == "down":
+            _text(scr, theme, y, inner_x,
+                  _fit("the server stopped responding; the conversion "
+                       "will fail", width - inner_x - 3), theme["err"])
+            y += 1
+        elif self.config.notice:
+            _text(scr, theme, y, inner_x,
+                  _fit(self.config.notice, width - inner_x - 3),
+                  theme["warn"])
+            y += 1
+        elif self.server_log_path and self.phase == "boot":
+            _text(scr, theme, y, inner_x,
+                  _fit(f"loading the model can take a while — log: "
+                       f"{self.server_log_path}", width - inner_x - 3),
+                  theme["dim"])
+            y += 1
+        return y
+
+    def _draw_summary(self, scr, theme, y, inner_x, label_w, value_x,
+                      value_w, width) -> int:
+        """The terminal panel: result, per-book lines, error detail."""
+        if self.phase == "done":
+            result, kind = "completed", "ok"
+        elif self.phase == "cancelled":
+            result, kind = "cancelled", "warn"
+        else:
+            result, kind = "failed", "err"
+        _text(scr, theme, y, inner_x, "Result".ljust(label_w), theme["dim"])
+        _text(scr, theme, y, value_x, _fit(result, value_w),
+              theme.get(kind, theme["body"]))
+        y += 1
+        for name, ok in self.book_results[:5]:
+            mark = "[OK] " if ok else "[FAIL]"
+            _text(scr, theme, y, value_x,
+                  _fit(f"{mark} {name}", value_w),
+                  theme["ok"] if ok else theme["err"])
+            y += 1
+        if len(self.book_results) > 5:
+            _text(scr, theme, y, value_x,
+                  _fit(f"... and {len(self.book_results) - 5} more",
+                       value_w), theme["dim"])
+            y += 1
+        if self.phase == "error":
+            detail = self.error_message or self.server_message
+            if detail:
+                for line in _wrap(detail, width - inner_x - 3)[:2]:
+                    _text(scr, theme, y, inner_x, line, theme["err"])
+                    y += 1
+            if self.log_tail:
+                for line in self.log_tail[:3]:
+                    _text(scr, theme, y, inner_x,
+                          _fit(line.strip() or " ", width - inner_x - 3),
+                          theme["dim"])
+                    y += 1
+            if self.config.log_path:
+                _text(scr, theme, y, inner_x,
+                      _fit(f"details: {self.config.log_path}",
+                           width - inner_x - 3), theme["dim"])
+                y += 1
+        elif self.phase == "cancelled":
+            _text(scr, theme, y, inner_x,
+                  "no audiobook was produced for the cancelled book",
+                  theme["dim"])
+            y += 1
+        return y
+
+
+# ---------------------------------------------------------------------------
+# Small drawing/formatting helpers (module-level for testability)
+# ---------------------------------------------------------------------------
+
+def _text(scr, theme, y, x, text, attr) -> None:
+    """addstr wrapper that ignores out-of-bounds errors."""
+    try:
+        scr.addstr(y, x, text, attr)
+    except Exception:
+        pass
+
+
+def _box(scr, curses, theme, height, width) -> None:
+    """Draw the full-screen frame."""
+    border = theme["border"]
+    try:
+        scr.addch(0, 0, curses.ACS_ULCORNER, border)
+        scr.addch(0, width - 1, curses.ACS_URCORNER, border)
+        scr.addch(height - 1, 0, curses.ACS_LLCORNER, border)
+        scr.addch(height - 1, width - 1, curses.ACS_LRCORNER, border)
+        scr.hline(0, 1, curses.ACS_HLINE, width - 2, border)
+        scr.hline(height - 1, 1, curses.ACS_HLINE, width - 2, border)
+        for y in range(1, height - 1):
+            scr.addch(y, 0, curses.ACS_VLINE, border)
+            scr.addch(y, width - 1, curses.ACS_VLINE, border)
+    except Exception:
+        pass
+
+
+def _sep(scr, curses, theme, y, width) -> None:
+    """A horizontal separator line inside the frame."""
+    try:
+        scr.addch(y, 0, curses.ACS_LTEE, theme["border"])
+        scr.addch(y, width - 1, curses.ACS_RTEE, theme["border"])
+        scr.hline(y, 1, curses.ACS_HLINE, width - 2, theme["dim"])
+    except Exception:
+        pass
+
+
+def _fit(text: str, width: int) -> str:
+    """Truncate TEXT to WIDTH columns, appending '~' when cut."""
+    if width < 1:
+        return ""
+    if len(text) <= width:
+        return text
+    return text[: max(0, width - 1)] + "~"
+
+
+def _wrap(text: str, width: int) -> List[str]:
+    """Greedy word wrap (no textwrap dependency on curses chars)."""
+    lines: List[str] = []
+    current = ""
+    for word in text.split():
+        candidate = f"{current} {word}".strip()
+        if len(candidate) <= max(10, width):
+            current = candidate
+        else:
+            if current:
+                lines.append(current)
+            current = word
+    if current:
+        lines.append(current)
+    return lines
+
+
+def _format_elapsed(seconds: float) -> str:
+    """Format a duration as H:MM:SS / M:SS."""
+    seconds = max(0, int(seconds))
+    hours, remainder = divmod(seconds, 3600)
+    minutes, secs = divmod(remainder, 60)
+    if hours:
+        return f"{hours}:{minutes:02d}:{secs:02d}"
+    return f"{minutes}:{secs:02d}"
+
+
+def run(scr, config: RunConfig) -> None:
+    """Enter the run view (called inside curses.wrapper by the hub)."""
+    view = RunView(scr, config)
+    view.run()
diff --git a/audiobook.py b/audiobook.py
index 7a9cbda..9a0da29 100755
--- a/audiobook.py
+++ b/audiobook.py
@@ -58,7 +58,9 @@ def convert(backend: str = None, voice: str = None, clone: str = None,
             output_format: str = None, debug: bool = False,
             model_id: str = None, instructions: str = None,
             request_options: dict = None, input_dir: Path = None,
-            output_dir: Path = None, api_url: str = None) -> int:
+            output_dir: Path = None, api_url: str = None,
+            progress=None, cancel=None, confirm=None,
+            book_files=None, planned=None) -> int:
     """Run one conversion pass with explicit options (used by the CLI and hub).
 
     Returns the process exit code (0 on success, 1 on failure, 130 on
@@ -68,6 +70,13 @@ def convert(backend: str = None, voice: str = None, clone: str = None,
     when given. API_URL, when given, overrides the configured server URL
     for the selected backend (used by the hub's "[remote]" entries and
     --api-url).
+
+    PROGRESS (a callback taking an event dict), CANCEL (a
+    threading.Event the caller sets to stop between requests), CONFIRM
+    (a (message, default) -> bool callback replacing the console
+    overwrite prompts) and BOOK_FILES/PLANNED (a pre-flight result, so
+    the overwrite prompts are not asked again) wire the conversion into
+    the TUI run view; without them everything behaves like the CLI.
     """
     backend = backend or config.BACKEND
     output_format = output_format or config.AUDIO_FORMAT
@@ -76,7 +85,9 @@ def convert(backend: str = None, voice: str = None, clone: str = None,
         _converter_mod.BOOKS_FOLDER = Path(input_dir)
     if output_dir is not None:
         _converter_mod.AUDIOBOOKS_FOLDER = Path(output_dir)
-    setup_logging(debug=debug)
+    # With a progress callback the run view owns the screen: keep log
+    # records in the file only and let the events carry the state.
+    setup_logging(debug=debug, console=progress is None)
     setup_directories()
 
     if backend == BACKEND_FASTER:
@@ -86,11 +97,12 @@ def convert(backend: str = None, voice: str = None, clone: str = None,
     else:
         voice_mode = VOICE_MODE_CLONE if clone else VOICE_MODE_CUSTOM
 
-    book_files, planned = AudiobookConverter.preflight_overwrites(
-        backend=backend, voice=voice, voice_mode=voice_mode,
-        voice_clone_ref_audio=clone, output_format=output_format,
-        instructions=instructions,
-    )
+    if book_files is None or planned is None:
+        book_files, planned = AudiobookConverter.preflight_overwrites(
+            backend=backend, voice=voice, voice_mode=voice_mode,
+            voice_clone_ref_audio=clone, output_format=output_format,
+            instructions=instructions, confirm=confirm,
+        )
     if not book_files:
         print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file "
               "to the input folder and run again.")
@@ -108,6 +120,7 @@ def convert(backend: str = None, voice: str = None, clone: str = None,
             language=language, backend=backend, voice=voice, debug=debug,
             model_id=model_id, instructions=instructions,
             request_options=request_options, api_url=api_url,
+            progress=progress, cancel=cancel,
         )
         converter._book_files = book_files
         converter._planned = planned
@@ -118,6 +131,8 @@ def convert(backend: str = None, voice: str = None, clone: str = None,
     except Exception as exc:
         print(f"[FATAL] Fatal error: {exc}")
         traceback.print_exc()
+        if progress is not None:
+            progress({"kind": "error", "message": str(exc)})
         return 1
     return 0 if ok else 1