feat: cli auto-starts and stops locally-managed servers if no --api-url is passed

3 files changed, 415 insertions, 5 deletions

diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py
index d947c5d..f13cf9e 100644
--- a/app/tests/test_audiobook_cli.py
+++ b/app/tests/test_audiobook_cli.py
@@ -1,12 +1,13 @@
-"""Tests for the audiobook.py CLI — single-book flags and arg validation.
+"""Tests for the audiobook.py CLI — single-book flags, arg validation, and
+the managed-server wiring.
 
 audiobook.py lives at the repo root (one level above app/), so the tests
 bootstrap the root onto sys.path to import it. main() runs with the
 managed-environment bootstrap stubbed (it would otherwise re-exec the
 process into envs/tts) and convert() mocked, asserting only argparse
-behavior and what reaches convert(); convert()'s single-book wiring and
-the pre-flight overrides are tested against the real functions with
-temporary directories.
+behavior and what reaches convert(); convert()'s single-book wiring, the
+pre-flight overrides, and the manage_server lifecycle are tested against
+the real functions with temporary directories.
 """
 
 import contextlib
@@ -250,6 +251,23 @@ class MainHappyPathTests(MainTestCase):
         self.assertIsNone(kwargs["input_file"])
         self.assertIsNone(kwargs["output_file"])
 
+    def test_manage_server_requested_without_api_url(self):
+        # Without --api-url the CLI asks convert() to manage the server
+        # lifecycle (convert() performs the actual boot/stop).
+        code, _, convert = self.run_main([])
+        self.assertEqual(code, 0)
+        self.assertIs(convert.call_args.kwargs["manage_server"], True)
+
+    def test_api_url_run_still_carries_the_manage_flag(self):
+        # The flag travels too; convert() itself skips management when an
+        # explicit api_url targets an external server.
+        code, _, convert = self.run_main(
+            ["--api-url", "10.20.30.40:8080"])
+        self.assertEqual(code, 0)
+        self.assertEqual(convert.call_args.kwargs["api_url"],
+                         "http://10.20.30.40:8080")
+        self.assertIs(convert.call_args.kwargs["manage_server"], True)
+
 
 class ConvertWiringTests(unittest.TestCase):
     """convert() turns the single-book flags into the pre-flight overrides."""
@@ -314,6 +332,115 @@ class ConvertWiringTests(unittest.TestCase):
             self._convert(output_file=self.tmp / "dune.mp3")
 
 
+class ManagedServerWiringTests(unittest.TestCase):
+    """convert(manage_server=True) boots and stops the server around the run.
+
+    The lifecycle decisions live in backends.managed (tested there); these
+    pin convert()'s wiring: when the boot happens relative to pre-flight
+    and the conversion, that shutdown runs even on failure or Ctrl-C, and
+    that an external api_url or the hub's progress-callback path never
+    touch the server.
+    """
+
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_managed_"))
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+        self.book = _make_book(self.tmp)
+        self._old_folders = (converter_mod.BOOKS_FOLDER,
+                             converter_mod.AUDIOBOOKS_FOLDER)
+        self.addCleanup(self._restore_folders)
+
+    def _restore_folders(self):
+        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = \
+            self._old_folders
+
+    def _convert(self, *, server_ok=True, run_result=True, run_raises=None,
+                 preflight_result=None, **kwargs):
+        """Run convert() with the managed-server and converter mocked.
+
+        Returns (code, ensure mock, server mock, events, preflight mock);
+        EVENTS records the order of ensure_running / run / shutdown.
+        """
+        kwargs.setdefault("backend", "audiocpp")
+        kwargs.setdefault("manage_server", True)
+        if preflight_result is None:
+            preflight_result = ([self.book], [(self.book, "dune")])
+        preflight = MagicMock(return_value=preflight_result)
+        fake_instance = MagicMock()
+        events = []
+
+        def _run():
+            events.append("run")
+            if run_raises is not None:
+                raise run_raises
+            return run_result
+        fake_instance.run.side_effect = _run
+        fake_class = MagicMock(return_value=fake_instance)
+        fake_class.preflight_overwrites = preflight
+        server = MagicMock()
+        server.ok = server_ok
+        server.shutdown.side_effect = lambda: events.append("shutdown")
+        ensure = MagicMock(return_value=server)
+
+        def _ensure(backend, voice_mode):
+            events.append(("ensure", backend, voice_mode))
+            return server
+        ensure.side_effect = _ensure
+        with patch.object(audiobook, "setup_logging"), \
+                patch.object(audiobook, "setup_directories"), \
+                patch.object(audiobook, "AudiobookConverter", fake_class), \
+                patch("backends.managed.ensure_running", ensure):
+            code = audiobook.convert(**kwargs)
+        return code, ensure, server, events, preflight
+
+    def test_server_boots_before_the_run_and_stops_after(self):
+        code, ensure, _, events, _ = self._convert()
+        self.assertEqual(code, 0)
+        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
+                                  "run", "shutdown"])
+
+    def test_qwen_run_needs_its_voice_mode_model(self):
+        _, ensure, _, events, _ = self._convert(
+            backend="qwen", clone="ref.wav")
+        self.assertEqual(events[0], ("ensure", "qwen", "voice_clone"))
+
+    def test_not_ok_boot_stops_before_converting(self):
+        code, ensure, server, events, _ = self._convert(server_ok=False)
+        self.assertEqual(code, 1)
+        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
+                                  "shutdown"])
+
+    def test_shutdown_runs_when_the_conversion_fails(self):
+        code, _, _, events, _ = self._convert(
+            run_raises=RuntimeError("server unreachable"))
+        self.assertEqual(code, 1)
+        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
+                                  "run", "shutdown"])
+
+    def test_shutdown_runs_on_ctrl_c(self):
+        code, _, _, events, _ = self._convert(run_raises=KeyboardInterrupt)
+        self.assertEqual(code, 130)
+        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
+                                  "run", "shutdown"])
+
+    def test_no_management_for_an_explicit_api_url(self):
+        code, ensure, _, _, _ = self._convert(
+            api_url="http://10.20.30.40:8080")
+        self.assertEqual(code, 0)
+        ensure.assert_not_called()
+
+    def test_no_management_for_the_hub_path(self):
+        # The run view boots/stops the server itself: manage_server False.
+        _, ensure, _, _, _ = self._convert(manage_server=False)
+        ensure.assert_not_called()
+
+    def test_no_server_boot_when_nothing_to_convert(self):
+        code, ensure, _, _, preflight = self._convert(
+            preflight_result=([], []))
+        self.assertEqual(code, 0)
+        ensure.assert_not_called()
+
+
 class PreflightOverrideTests(unittest.TestCase):
     """preflight_overwrites honors the explicit book list and output name."""
 
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py
index 64fd483..7a39880 100644
--- a/app/tests/test_backends.py
+++ b/app/tests/test_backends.py
@@ -169,7 +169,7 @@ class DetectAllTests(unittest.TestCase):
         for model, wanted in (("Base", IDENTITY_QWEN_CLONE),
                               ("VoiceDesign", IDENTITY_QWEN_DESIGN)):
             with self.subTest(model=model):
-                spec = qwen._build_spec(model)
+                spec = qwen.build_spec(model)
                 self.assertEqual(spec.identity, wanted)
                 self.assertIn(qwen.MODEL_REPOS[model], spec.argv)
 
diff --git a/app/tests/test_backends_managed.py b/app/tests/test_backends_managed.py
new file mode 100644
index 0000000..cd1f03a
--- /dev/null
+++ b/app/tests/test_backends_managed.py
@@ -0,0 +1,283 @@
+"""Tests for the CLI's managed-server bootstrap (backends/managed.py).
+
+The CLI (without --api-url) calls ``ensure_running`` before converting and
+``ManagedServer.shutdown`` after; both are tested against mocked
+``servers``/``probe``/registry-detect so no process is ever spawned. The
+qwen paths exercise the real ``build_spec``/``model_for_voice_mode``
+mapping (pure config reads) to pin the model-per-voice-mode contract.
+"""
+
+import io
+import unittest
+from contextlib import redirect_stdout
+from types import SimpleNamespace
+from unittest.mock import ANY, patch
+
+from backends import ServerSpec, managed, servers
+from backends.managed import ManagedServer, ensure_running
+from backends.probe import (IDENTITY_AUDIOCPP, IDENTITY_QWEN_CLONE,
+                            IDENTITY_QWEN_CUSTOM, IDENTITY_QWEN_DESIGN)
+from converter.clients import (BACKEND_AUDIOCPP, BACKEND_QWEN,
+                               VOICE_MODE_CLONE, VOICE_MODE_CUSTOM,
+                               VOICE_MODE_DESIGN)
+
+
+def _spec(name="audiocpp", url="http://127.0.0.1:8080", identity=None):
+    return ServerSpec(name, url, ["/bin/fake_server"], identity=identity)
+
+
+def _status(specs, installed=True, label="audio.cpp"):
+    """A minimal stand-in for the registry's BackendStatus."""
+    return SimpleNamespace(servers=specs, installed=installed, label=label)
+
+
+class ShutdownTests(unittest.TestCase):
+    """ManagedServer.shutdown stops only what the run started."""
+
+    def test_noop_when_not_started(self):
+        with patch.object(servers, "stop") as mk_stop:
+            ManagedServer(_spec()).shutdown()
+        mk_stop.assert_not_called()
+
+    def test_stops_the_spec_when_started(self):
+        spec = _spec()
+        with patch.object(servers, "stop") as mk_stop:
+            ManagedServer(spec, started=True).shutdown()
+        mk_stop.assert_called_once_with(spec.name)
+
+
+class EnsureRunningTests(unittest.TestCase):
+    """ensure_running resolves the spec and boots or reuses the server."""
+
+    def setUp(self):
+        self.spec = _spec(identity=IDENTITY_AUDIOCPP)
+
+    def _run(self, backend=BACKEND_AUDIOCPP, voice_mode="custom_voice"):
+        out = io.StringIO()
+        with redirect_stdout(out):
+            result = ensure_running(backend, voice_mode)
+        return result, out.getvalue()
+
+    def test_not_installed_returns_none_with_warning(self):
+        with patch("backends.detect",
+                   return_value=_status([self.spec], installed=False)), \
+                patch.object(servers, "start") as mk_start:
+            result, output = self._run()
+        self.assertIsNone(result)
+        self.assertIn("not installed", output)
+        mk_start.assert_not_called()
+
+    def test_unknown_backend_returns_none_with_warning(self):
+        with patch("backends.detect", return_value=None), \
+                patch.object(servers, "start") as mk_start:
+            result, output = self._run(backend="gone")
+        self.assertIsNone(result)
+        self.assertIn("not installed", output)
+        mk_start.assert_not_called()
+
+    def test_no_specs_returns_none_with_warning(self):
+        with patch("backends.detect",
+                   return_value=_status([], installed=True)), \
+                patch.object(servers, "start") as mk_start:
+            result, _ = self._run()
+        self.assertIsNone(result)
+        mk_start.assert_not_called()
+
+    def test_starts_when_port_free(self):
+        with patch("backends.detect",
+                   return_value=_status([self.spec])), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start", return_value=True) as mk_start:
+            result, _ = self._run()
+        mk_start.assert_called_once_with(self.spec, progress=ANY)
+        self.assertTrue(result.ok)
+        self.assertTrue(result.started)
+        self.assertEqual(result.spec, self.spec)
+
+    def test_start_failure_reports_not_ok_and_never_stops(self):
+        with patch("backends.detect",
+                   return_value=_status([self.spec])), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start", return_value=False), \
+                patch.object(servers, "stop") as mk_stop:
+            result, _ = self._run()
+            result.shutdown()
+        self.assertFalse(result.ok)
+        self.assertFalse(result.started)
+        mk_stop.assert_not_called()
+
+    def test_running_server_is_reused_and_left_running(self):
+        with patch("backends.detect",
+                   return_value=_status([self.spec])), \
+                patch("backends.common.server_running",
+                      return_value=True), \
+                patch.object(servers, "start") as mk_start:
+            result, _ = self._run()
+            result.shutdown()
+        mk_start.assert_not_called()
+        self.assertTrue(result.ok)
+        self.assertFalse(result.started)
+
+    def test_running_event_marks_the_run_as_a_reuser(self):
+        """A server appearing under us between the port check and the spawn.
+
+        ``servers.start`` reports "running" instead of spawning in that
+        race; the run must then not stop the server at shutdown.
+        """
+        seen = []
+
+        def fake_start(spec, progress=None, cancel=None):
+            event = {"kind": "running", "name": spec.name, "url": spec.url}
+            seen.append(event)
+            if progress is not None:
+                progress(event)
+            return True
+
+        with patch("backends.detect",
+                   return_value=_status([self.spec])), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start", side_effect=fake_start), \
+                patch.object(servers, "_console_progress") as mk_print, \
+                patch.object(servers, "stop") as mk_stop:
+            result, _ = self._run()
+            result.shutdown()
+        self.assertTrue(result.ok)
+        self.assertFalse(result.started)
+        # The console still hears the event (delegated printer).
+        mk_print.assert_called_once_with(seen[0])
+        mk_stop.assert_not_called()
+
+    def test_keyboard_interrupt_stops_the_spawned_server(self):
+        with patch("backends.detect",
+                   return_value=_status([self.spec])), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start",
+                             side_effect=KeyboardInterrupt), \
+                patch.object(servers, "pid_for", return_value=4242), \
+                patch.object(servers, "stop") as mk_stop, \
+                self.assertRaises(KeyboardInterrupt):
+            self._run()
+        mk_stop.assert_called_once_with(self.spec.name)
+
+    def test_keyboard_interrupt_without_spawn_skips_the_stop(self):
+        with patch("backends.detect",
+                   return_value=_status([self.spec])), \
+                patch("backends.common.server_running",
+                      return_value=False), \
+                patch.object(servers, "start",
+                             side_effect=KeyboardInterrupt), \
+                patch.object(servers, "pid_for", return_value=None), \
+                patch.object(servers, "stop") as mk_stop, \
+                self.assertRaises(KeyboardInterrupt):
+            self._run()
+        mk_stop.assert_not_called()
+
+
+class QwenEnsureRunningTests(unittest.TestCase):
+    """qwen hosts one model per server: the running-model check is aware."""
+
+    def _detect_qwen(self):
+        from backends import qwen
+        return _status([qwen.build_spec("CustomVoice")],
+                       installed=True, label="qwen-tts")
+
+    def _run(self, voice_mode, backend=BACKEND_QWEN):
+        out = io.StringIO()
+        with redirect_stdout(out):
+            result = ensure_running(backend, voice_mode)
+        return result, out.getvalue()
+
+    def test_spec_aims_at_the_model_the_voice_mode_needs(self):
+        from backends import qwen
+        cases = [(VOICE_MODE_CLONE, "Base", IDENTITY_QWEN_CLONE),
+                 (VOICE_MODE_CUSTOM, "CustomVoice", IDENTITY_QWEN_CUSTOM),
+                 (VOICE_MODE_DESIGN, "VoiceDesign", IDENTITY_QWEN_DESIGN)]
+        for voice_mode, model, identity in cases:
+            with self.subTest(voice_mode=voice_mode):
+                with patch("backends.detect", return_value=self._detect_qwen()), \
+                        patch("backends.common.server_running",
+                              return_value=False), \
+                        patch.object(servers, "start",
+                                     return_value=True) as mk_start:
+                    result, _ = self._run(voice_mode)
+                spec = mk_start.call_args.args[0]
+                self.assertEqual(spec.identity, identity)
+                self.assertIn(qwen.MODEL_REPOS[model], spec.argv)
+                self.assertTrue(result.started)
+
+    def test_running_server_hosting_the_wanted_model_is_reused(self):
+        with patch("backends.detect", return_value=self._detect_qwen()), \
+                patch("backends.common.server_running", return_value=True), \
+                patch("backends.probe.identify_server",
+                      return_value=IDENTITY_QWEN_CLONE), \
+                patch.object(servers, "start") as mk_start, \
+                patch.object(servers, "stop") as mk_stop:
+            result, _ = self._run(VOICE_MODE_CLONE)
+            result.shutdown()
+        mk_start.assert_not_called()
+        mk_stop.assert_not_called()
+        self.assertTrue(result.ok)
+        self.assertFalse(result.started)
+
+    def test_managed_server_hosting_another_model_is_rebooted(self):
+        with patch("backends.detect", return_value=self._detect_qwen()), \
+                patch("backends.common.server_running", return_value=True), \
+                patch("backends.probe.identify_server",
+                      return_value=IDENTITY_QWEN_CUSTOM), \
+                patch.object(servers, "alive", return_value=True), \
+                patch.object(servers, "start", return_value=True) as mk_start, \
+                patch.object(servers, "stop") as mk_stop:
+            result, output = self._run(VOICE_MODE_CLONE)
+        self.assertIn("restarting", output)
+        mk_stop.assert_called_once_with("qwen")
+        spec = mk_start.call_args.args[0]
+        self.assertEqual(spec.identity, IDENTITY_QWEN_CLONE)
+        self.assertTrue(result.started)
+        # The rebooted server is this run's: shutdown stops it again.
+        with patch.object(servers, "stop") as mk_stop:
+            result.shutdown()
+        mk_stop.assert_called_once_with("qwen")
+
+    def test_foreign_server_hosting_another_model_refuses_the_run(self):
+        with patch("backends.detect", return_value=self._detect_qwen()), \
+                patch("backends.common.server_running", return_value=True), \
+                patch("backends.probe.identify_server",
+                      return_value=IDENTITY_QWEN_CUSTOM), \
+                patch.object(servers, "alive", return_value=False), \
+                patch.object(servers, "start") as mk_start, \
+                patch.object(servers, "stop") as mk_stop:
+            result, output = self._run(VOICE_MODE_CLONE)
+        self.assertFalse(result.ok)
+        self.assertIn("this run needs Base", output)
+        mk_start.assert_not_called()
+        mk_stop.assert_not_called()
+
+    def test_unidentified_running_server_counts_as_unknown(self):
+        """A port squatting service that probes as nothing known refuses."""
+        with patch("backends.detect", return_value=self._detect_qwen()), \
+                patch("backends.common.server_running", return_value=True), \
+                patch("backends.probe.identify_server", return_value=None), \
+                patch.object(servers, "alive", return_value=False), \
+                patch.object(servers, "start") as mk_start:
+            result, output = self._run(VOICE_MODE_CLONE)
+        self.assertFalse(result.ok)
+        self.assertIn("an unknown", output)
+        mk_start.assert_not_called()
+
+
+class ManagedModuleSmokeTests(unittest.TestCase):
+    """Import-surface sanity for the module the CLI wires in."""
+
+    def test_managed_server_defaults(self):
+        server = ManagedServer(_spec())
+        self.assertFalse(server.started)
+        self.assertTrue(server.ok)
+
+    def test_module_uses_the_servers_module_singletons(self):
+        # managed delegates to the same servers module the TUI uses, so
+        # pid/log files and console output stay identical.
+        self.assertIs(managed.servers, servers)