refactor: extract cli entrypoint orchestration helpers

TODO.md

@@ -1,4 +1,4 @@
-# Jarvis TODO — Wave 74/106 Runtime Decomposition
+# Jarvis TODO — Wave 74/107 Runtime Decomposition
 
 Last updated: 2026-02-27
 
@@ -8,8 +8,8 @@ Last updated: 2026-02-27
 - `[x]` Completed
 
 ## Completion summary
-- Total items: 279
-- Completed: 279
+- Total items: 287
+- Completed: 287
 - Remaining: 0
 
 ---
@@ -402,6 +402,17 @@ Last updated: 2026-02-27
 - [x] `W106-AJ07` Run full quality/security/readiness gates.
 - [x] `W106-AJ08` Update TODO + tranche snapshot.
 
+## AK) Wave 107 (completed): main entrypoint orchestration extraction
+
+- [x] `W107-AK01` Profile top-level `main()` orchestration in `src/jarvis/__main__.py`.
+- [x] `W107-AK02` Create `runtime_entrypoint.py` for backup/restore handling and event-loop bootstrap helpers.
+- [x] `W107-AK03` Rewire `main()` to runtime entrypoint helpers while preserving behavior.
+- [x] `W107-AK04` Preserve backup/restore JSON error semantics and signal-driven shutdown behavior.
+- [x] `W107-AK05` Add focused unit coverage in `tests/test_runtime_entrypoint.py`.
+- [x] `W107-AK06` Extend import-boundary coverage for `runtime_entrypoint`.
+- [x] `W107-AK07` Run full quality/security/readiness gates.
+- [x] `W107-AK08` Update TODO + tranche snapshot.
+
 ---
 
 ## Outcome snapshot (latest completed tranche)
@@ -433,6 +444,7 @@ Last updated: 2026-02-27
   - `runtime_voice_status.py`
   - `runtime_observability_snapshot.py`
   - `runtime_watchdog.py`
+  - `runtime_entrypoint.py`
 - Compatibility preservation:
   - `services.py` now re-exports defaults/constants via `_services_defaults` alias.
   - Mutable bootstrap defaults (`_proactive_state`, `_privacy_posture`, `_motion_safety_envelope`, `_release_channel_state`) now initialize from factory helpers.
@@ -472,6 +484,7 @@ Last updated: 2026-02-27
   - Persona-style update + memory summary upsert moved from `Jarvis._set_persona_style` into `runtime_preferences.set_persona_style`.
   - Observability status publication moved from `Jarvis._publish_observability_status` into `runtime_observability_status.publish_observability_status`.
   - Telemetry tool-error counter refresh moved from `Jarvis._refresh_tool_error_counters` into `runtime_telemetry.refresh_tool_error_counters`.
+  - CLI backup/restore and asyncio event-loop orchestration moved from `main()` body into `runtime_entrypoint` helpers.
 - Validation (Wave 75D/E):
   - `uv run pytest -q tests/test_import_boundaries.py`: `177 passed`.
   - `uv run pytest -q tests/test_runtime_operator_status.py tests/test_main_lifecycle.py -k "operator_auth or startup_summary_lines_include_core_status"`: `8 passed`.
@@ -515,7 +528,9 @@ Last updated: 2026-02-27
   - `uv run pytest -q tests/test_runtime_watchdog.py tests/test_runtime_observability_snapshot.py tests/test_runtime_voice_status.py tests/test_runtime_turn.py tests/test_main_audio.py`: `21 passed`.
   - `uv run pytest -q tests/test_runtime_telemetry.py tests/test_runtime_observability_status.py tests/test_main_lifecycle.py tests/test_import_boundaries.py -k "runtime_telemetry or runtime_observability_status or publish_observability_status or refresh_tool_error_counters or import_boundary"`: `208 passed, 35 deselected`.
   - `uv run pytest -q tests/test_runtime_watchdog.py tests/test_runtime_observability_snapshot.py tests/test_runtime_voice_status.py tests/test_runtime_preferences.py tests/test_runtime_multimodal.py`: `19 passed`.
+  - `uv run pytest -q tests/test_runtime_entrypoint.py tests/test_import_boundaries.py tests/test_main_lifecycle.py -k "runtime_entrypoint or import_boundary or backup or restore or main"`: `242 passed`.
+  - `uv run pytest -q tests/test_runtime_observability_status.py tests/test_runtime_telemetry.py tests/test_runtime_watchdog.py tests/test_runtime_observability_snapshot.py tests/test_runtime_voice_status.py`: `14 passed`.
   - `./scripts/test_soak_reliability.sh 2`: accepted; `cycles_completed=2/2`, `phase_count=8`, `failed_count=0`, artifact `.artifacts/quality/soak-profile-fast-repeat2.json`.
-  - `make check`: `831 passed`.
-  - `make security-gate`: `831 passed`; fault subset `3 passed`.
+  - `make check`: `836 passed`.
+  - `make security-gate`: `836 passed`; fault subset `3 passed`.
   - `./scripts/jarvis_readiness.sh fast`: pass; strict eval `159/159`.

src/jarvis/__main__.py

@@ -15,9 +15,7 @@ from __future__ import annotations
 
 import argparse
 import asyncio
-import json
 import logging
-import signal
 import time
 import threading
 from pathlib import Path
@@ -161,6 +159,10 @@ from jarvis.runtime_multimodal import (
     multimodal_grounding_snapshot_for_runtime as _runtime_multimodal_grounding_snapshot_for_runtime,
 )
 from jarvis.runtime_watchdog import watchdog_loop as _runtime_watchdog_loop
+from jarvis.runtime_entrypoint import (
+    maybe_run_backup_or_restore as _runtime_maybe_run_backup_or_restore,
+    run_jarvis_event_loop as _runtime_run_jarvis_event_loop,
+)
 from jarvis.runtime_constants import (
     ATTENTION_RECENCY_SEC,
     CONVERSATION_TRACE_MAXLEN,
@@ -988,40 +990,16 @@ def main():
         datefmt="%H:%M:%S",
     )
 
-    if args.backup or args.restore:
-        config = Config()
-        try:
-            if args.backup:
-                result = create_backup_bundle(config, args.backup)
-            else:
-                result = restore_backup_bundle(config, args.restore, overwrite=bool(args.force))
-        except Exception as exc:
-            print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
-            raise SystemExit(1) from exc
-        print(json.dumps(result, indent=2))
+    if _runtime_maybe_run_backup_or_restore(
+        args,
+        config_class=Config,
+        create_backup_bundle_fn=create_backup_bundle,
+        restore_backup_bundle_fn=restore_backup_bundle,
+    ):
         return
 
     jarvis = Jarvis(args)
-
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    task = loop.create_task(jarvis.run())
-
-    def shutdown(sig, frame):
-        task.cancel()
-
-    signal.signal(signal.SIGINT, shutdown)
-    if hasattr(signal, "SIGTERM"):
-        signal.signal(signal.SIGTERM, shutdown)
-
-    try:
-        loop.run_until_complete(task)
-    finally:
-        with suppress(Exception):
-            loop.run_until_complete(loop.shutdown_asyncgens())
-        with suppress(Exception):
-            loop.run_until_complete(loop.shutdown_default_executor())
-        loop.close()
+    _runtime_run_jarvis_event_loop(jarvis)
 
 
 if __name__ == "__main__":

src/jarvis/runtime_entrypoint.py

@@ -0,0 +1,53 @@
+"""Entrypoint orchestration helpers for Jarvis CLI runtime."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import signal
+from contextlib import suppress
+from typing import Any, Callable
+
+
+def maybe_run_backup_or_restore(
+    args: Any,
+    *,
+    config_class: Callable[[], Any],
+    create_backup_bundle_fn: Callable[[Any, str], dict[str, Any]],
+    restore_backup_bundle_fn: Callable[[Any, str], dict[str, Any]],
+) -> bool:
+    if not (args.backup or args.restore):
+        return False
+    config = config_class()
+    try:
+        if args.backup:
+            result = create_backup_bundle_fn(config, args.backup)
+        else:
+            result = restore_backup_bundle_fn(config, args.restore, overwrite=bool(args.force))
+    except Exception as exc:
+        print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
+        raise SystemExit(1) from exc
+    print(json.dumps(result, indent=2))
+    return True
+
+
+def run_jarvis_event_loop(jarvis: Any) -> None:
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    task = loop.create_task(jarvis.run())
+
+    def shutdown(sig, frame):
+        task.cancel()
+
+    signal.signal(signal.SIGINT, shutdown)
+    if hasattr(signal, "SIGTERM"):
+        signal.signal(signal.SIGTERM, shutdown)
+
+    try:
+        loop.run_until_complete(task)
+    finally:
+        with suppress(Exception):
+            loop.run_until_complete(loop.shutdown_asyncgens())
+        with suppress(Exception):
+            loop.run_until_complete(loop.shutdown_default_executor())
+        loop.close()

tests/test_import_boundaries.py

@@ -19,6 +19,7 @@ import pytest
         ("jarvis.runtime_operator_server", "jarvis.__main__"),
         ("jarvis.runtime_audio_output", "jarvis.__main__"),
         ("jarvis.runtime_lifecycle", "jarvis.__main__"),
+        ("jarvis.runtime_entrypoint", "jarvis.__main__"),
         ("jarvis.runtime_observability_snapshot", "jarvis.__main__"),
         ("jarvis.runtime_observability_status", "jarvis.__main__"),
         ("jarvis.runtime_memory_correction", "jarvis.__main__"),

tests/test_runtime_entrypoint.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from jarvis.runtime_entrypoint import (
+    maybe_run_backup_or_restore,
+    run_jarvis_event_loop,
+)
+
+
+def test_maybe_run_backup_or_restore_returns_false_when_no_maintenance_args() -> None:
+    args = SimpleNamespace(backup=None, restore=None, force=False)
+    ran = maybe_run_backup_or_restore(
+        args,
+        config_class=lambda: object(),
+        create_backup_bundle_fn=lambda config, path: {},
+        restore_backup_bundle_fn=lambda config, path, overwrite=False: {},
+    )
+    assert ran is False
+
+
+def test_maybe_run_backup_or_restore_backup_prints_result(capsys) -> None:
+    args = SimpleNamespace(backup="/tmp/backup.zip", restore=None, force=False)
+    ran = maybe_run_backup_or_restore(
+        args,
+        config_class=lambda: object(),
+        create_backup_bundle_fn=lambda config, path: {"ok": True, "path": path},
+        restore_backup_bundle_fn=lambda config, path, overwrite=False: {},
+    )
+    captured = capsys.readouterr()
+    assert ran is True
+    assert '"ok": true' in captured.out.lower()
+    assert "/tmp/backup.zip" in captured.out
+
+
+def test_maybe_run_backup_or_restore_failure_raises_system_exit(capsys) -> None:
+    args = SimpleNamespace(backup="/tmp/backup.zip", restore=None, force=False)
+
+    with pytest.raises(SystemExit) as exc_info:
+        maybe_run_backup_or_restore(
+            args,
+            config_class=lambda: object(),
+            create_backup_bundle_fn=lambda config, path: (_ for _ in ()).throw(RuntimeError("boom")),
+            restore_backup_bundle_fn=lambda config, path, overwrite=False: {},
+        )
+
+    captured = capsys.readouterr()
+    assert exc_info.value.code == 1
+    assert '"ok": false' in captured.out.lower()
+    assert "boom" in captured.out
+
+
+def test_run_jarvis_event_loop_bootstraps_and_cleans_up() -> None:
+    class _FakeTask:
+        def __init__(self) -> None:
+            self.cancelled = False
+
+        def cancel(self) -> None:
+            self.cancelled = True
+
+    class _FakeLoop:
+        def __init__(self) -> None:
+            self.calls: list[object] = []
+            self.task = _FakeTask()
+            self.closed = False
+
+        def create_task(self, coro):
+            self.calls.append(("create_task", coro))
+            return self.task
+
+        def run_until_complete(self, awaitable):
+            self.calls.append(("run_until_complete", awaitable))
+            return None
+
+        def shutdown_asyncgens(self):
+            return "shutdown_asyncgens"
+
+        def shutdown_default_executor(self):
+            return "shutdown_default_executor"
+
+        def close(self):
+            self.closed = True
+
+    fake_loop = _FakeLoop()
+    jarvis = SimpleNamespace(run=lambda: "run-coro")
+    signal_calls: list[tuple[object, object]] = []
+
+    with patch("jarvis.runtime_entrypoint.asyncio.new_event_loop", return_value=fake_loop), \
+         patch("jarvis.runtime_entrypoint.asyncio.set_event_loop") as set_event_loop, \
+         patch("jarvis.runtime_entrypoint.signal.signal", side_effect=lambda sig, handler: signal_calls.append((sig, handler))):
+        run_jarvis_event_loop(jarvis)
+
+    set_event_loop.assert_called_once_with(fake_loop)
+    assert fake_loop.closed is True
+    assert any(call[0] == "create_task" for call in fake_loop.calls)
+    assert len(signal_calls) >= 1