Coverage for src/lilbee/cli/engine.py: 100%

1"""Engine lifecycle commands: manage the shared engine without opening the TUI.""" 

2 

3from __future__ import annotations 

4 

5import typer 

6 

7from lilbee.cli.app import console, json_out 

8from lilbee.core.config import cfg 

9from lilbee.providers.fleet.swap_manager import stop_engine 

10from lilbee.runtime.engine_lock import build_lock, machine_engine_dir, private_engine_dir 

11 

12engine_app = typer.Typer( 

13 name="engine", 

14 help="Manage the local inference engine.", 

15 no_args_is_help=True, 

16) 

17 

18_STOPPED = "Stopped the engine and freed its memory." 

19_NOTHING_RUNNING = "No engine is running." 

20 

21 

22@engine_app.command("stop") 

23def stop() -> None: 

24 """Stop the shared engine now, whoever started it.""" 

25 stopped: list[str] = [] 

26 # Machine slot, this root's overflow dir, and the pre-upgrade legacy location 

27 # (base builds recorded state directly under data_dir): a stale legacy engine 

28 # must still be stoppable after an upgrade relocated the engine dirs. 

29 dirs = dict.fromkeys((machine_engine_dir(), private_engine_dir(cfg.data_root), cfg.data_dir)) 

30 for engine_dir in dirs: 

31 # Serialize against a concurrent builder, whose spawn + state write + health 

32 # wait all run under this lock; an unlocked stop landing mid-build would kill 

33 # the just-spawned engine and unlink its fresh record. best_effort so a wedged 

34 # builder cannot make the off switch hang. 

35 with build_lock(engine_dir, best_effort=True): 

36 stopped.extend(stop_engine(engine_dir)) 

37 if cfg.json_mode: 

38 json_out({"command": "engine stop", "stopped": sorted(set(stopped))}) 

39 return 

40 console.print(_STOPPED if stopped else _NOTHING_RUNNING)