Coverage for src/lilbee/app/services.py: 100%
305 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Typed service container: single point of access for all singletons.
3All runtime dependencies (provider, store, embedder, reranker, concepts,
4clusterer, searcher, worker pool) are created lazily on first call to
5``get_services()`` and cached for the process lifetime. Tests call
6``reset_services()`` between runs.
8``build_services(config)`` is the construction seam: it builds a full container
9against an arbitrary Config without touching the process-global singleton. The
10library API (:class:`lilbee.Lilbee`) builds one per instance and installs it for
11the duration of each call via :func:`services_scope`, so ingest code reaching for
12``get_services()`` resolves the caller's container. The override is a ContextVar,
13so ``reset_services`` / ``set_services`` / ``peek_services`` (which operate only
14on the global singleton) never see it.
15"""
17from __future__ import annotations
19import asyncio
20import atexit
21import logging
22import os
23import signal
24import sys
25import threading
26import time
27import weakref
28from collections.abc import Callable
29from contextlib import contextmanager, suppress
30from contextvars import ContextVar
31from dataclasses import dataclass, field
32from typing import TYPE_CHECKING
34if TYPE_CHECKING:
35 from collections.abc import Iterator
37 from lilbee.catalog.hf_client import HfClient
38 from lilbee.core.config import Config
39 from lilbee.core.profile_files import ProfileStore
40 from lilbee.data.store import Store
41 from lilbee.modelhub.model_manager import ModelManager
42 from lilbee.modelhub.model_manager.discovery import KnownModelCache
43 from lilbee.modelhub.registry import ModelRegistry
44 from lilbee.providers.base import LLMProvider
45 from lilbee.providers.roles import WorkerRole
46 from lilbee.retrieval.clustering import Clusterer
47 from lilbee.retrieval.concepts import ConceptGraph
48 from lilbee.retrieval.embedder import Embedder
49 from lilbee.retrieval.query import Searcher
50 from lilbee.retrieval.reranker import Reranker
51 from lilbee.runtime.ingest_lock import IngestLockRegistry
52 from lilbee.runtime.progress import DetailedProgressCallback, EventType, ProgressEvent
53 from lilbee.sessions import SessionStore
56log = logging.getLogger(__name__)
58_SIGNAL_EXIT_BASE = 128
60_HARD_EXIT_THREAD_NAME = "hard-exit-teardown"
63def _default_session_store() -> SessionStore:
64 """Build the file-backed session store, importing it lazily.
66 ``lilbee.sessions`` pulls in the config/catalog import chain, so importing it
67 at this module's top would form a cycle during CLI config load.
68 """
69 from lilbee.sessions import SessionStore
71 return SessionStore()
74def _default_profile_store() -> ProfileStore:
75 """Build the profile store, importing it lazily."""
76 # circular: core.config -> catalog (model-ref validator) -> app.services -> core.profile_files
77 from lilbee.core.profile_files import ProfileStore
79 return ProfileStore()
82@dataclass
83class CrawlerSyncState:
84 """Process-wide sync coordination state (lock + last-run timestamp)."""
86 lock: threading.Lock = field(default_factory=threading.Lock)
87 last_run: float = 0.0
90@dataclass(frozen=True)
91class Services:
92 """Holds all runtime service instances.
94 Inference lifecycle (cancel, per-role reload, spawn notifications) is owned
95 by the provider, which manages the llama-server fleet. Services exposes thin
96 pass-throughs so callers (Ctrl+C, the chat-stream cancel action, the settings
97 and model-bar pickers, the TUI task bar) need not reach into the provider's
98 API. ``cancel_inference()`` is the canonical cancel entry point.
99 """
101 provider: LLMProvider
102 store: Store
103 embedder: Embedder
104 reranker: Reranker
105 concepts: ConceptGraph
106 clusterer: Clusterer
107 searcher: Searcher
108 registry: ModelRegistry
109 hf_client: HfClient
110 ingest_lock_registry: IngestLockRegistry
111 model_manager: ModelManager
112 crawler_semaphore: asyncio.Semaphore | None
113 crawler_sync_state: CrawlerSyncState
114 known_models: KnownModelCache
115 session_store: SessionStore = field(default_factory=_default_session_store)
116 profile_store: ProfileStore = field(default_factory=_default_profile_store)
118 def cancel_inference(self) -> None:
119 """Interrupt any in-flight generation. Idempotent.
121 The fleet engine severs its live chat streams (llama-server stops
122 generating when the connection drops); providers with nothing in
123 flight treat this as a no-op.
124 """
125 self.provider.cancel_inference()
127 def reload_role(self, role_name: WorkerRole, *, wait: bool = False) -> None:
128 """Respawn only *role_name*'s model server so it picks up changed cfg.
130 Other roles' servers and any in-flight stream they own are untouched. Use
131 when one role-bound model setting changed (e.g. embedding_model). The
132 respawn runs off the caller's thread, so this returns immediately, unless
133 ``wait=True`` (the caller is already off the event loop and wants to block
134 until the new model has loaded).
135 """
136 self.provider.reload_role(role_name, wait=wait)
138 def add_pool_listener(
139 self,
140 *,
141 on_spawning: Callable[[WorkerRole], None] | None = None,
142 on_spawned: Callable[[WorkerRole], None] | None = None,
143 ) -> None:
144 """Subscribe to server spawn lifecycle events.
146 Forwards to :meth:`LLMProvider.add_spawn_listener`. The TUI uses this to
147 surface "Starting <role>..." / "<role> ready" notifications when a role's
148 server (re)spawns (cold start after a non-eager boot, or a reload).
149 """
150 self.provider.add_spawn_listener(on_spawning=on_spawning, on_spawned=on_spawned)
153class _ServicesState:
154 """The cached process-global singleton plus the per-task scoped override.
156 ``singleton`` is set on first ``get_services()`` call. Concurrency
157 contract: creation is serialized by ``_singleton_create_lock`` (several
158 worker threads can first-touch services at once, and a duplicate build
159 would collide in xberg's process-global backend registry), and the
160 Services dataclass is logically immutable post-construction, so concurrent
161 reads are safe without a lock. Tests that need a custom container call
162 ``set_services(make_mock_services(...))``; ``peek_services()`` is the
163 read-only inspector for cleanup fixtures.
165 ``override`` shadows the singleton for the entering task: set by
166 :func:`services_scope` (the library API's per-call binding), read by
167 :func:`get_services`, and invisible to ``reset_services`` /
168 ``set_services`` / ``peek_services``, which only touch the singleton.
169 """
171 def __init__(self) -> None:
172 self.singleton: Services | None = None
173 self.override: ContextVar[Services | None] = ContextVar(
174 "lilbee_services_override", default=None
175 )
176 # Whether this process is an interactive session (the TUI). Recorded by
177 # the interactive entry point before anything builds the container, and
178 # read once at build so the provider it creates holds its fleet resident
179 # for the session. Build-time intent only; the state that matters after
180 # that lives on the provider itself.
181 self.interactive: bool = False
182 # Where this process shows a library upgrade; each surface sets its own.
183 self.on_upgrade: DetailedProgressCallback | None = None
184 # How this process takes config.toml now; set by each surface that follows the file.
185 self.config_refresh: Callable[[], None] | None = None
186 # Every store this process built, so the exit can wait for any library open
187 # (the first, the one /reset makes, a lazy one) without each path telling it.
188 self.stores: weakref.WeakSet[Store] = weakref.WeakSet()
189 self.stores_lock = threading.Lock()
191 def report_upgrade(self, event_type: EventType, event: ProgressEvent) -> None:
192 """Pass a library upgrade event to the reporter the surface set last, if any."""
193 if self.on_upgrade is not None:
194 self.on_upgrade(event_type, event)
197_state = _ServicesState()
200def _new_store(config: Config) -> Store:
201 """Build a Store and record it, so ``wait_for_opens`` sees its open; the one way to make one."""
202 from lilbee.data.store import Store
204 store = Store(config, on_upgrade=_state.report_upgrade)
205 with _state.stores_lock:
206 _state.stores.add(store)
207 return store
210def wait_for_opens(timeout: float) -> bool:
211 """Wait up to *timeout* seconds in all, for every library open still running.
213 Return False when an open is still running at the deadline.
214 """
215 deadline = time.monotonic() + timeout
216 with _state.stores_lock:
217 stores = list(_state.stores)
218 done = True
219 for store in stores:
220 if not store.wait_for_open(max(0.0, deadline - time.monotonic())):
221 done = False
222 return done
225def build_services(
226 config: Config,
227 *,
228 provider: LLMProvider | None = None,
229 registry: ModelRegistry | None = None,
230 interactive: bool = False,
231) -> Services:
232 """Build a full Services container bound to *config*, without caching it.
234 ``get_services()`` calls this with the process-global cfg; the library API
235 calls it per instance. Service modules are imported inside the function to
236 keep CLI startup fast (they transitively pull in lancedb / xberg). Pass
237 *provider* to reuse a caller-supplied one; otherwise it is built from
238 *config* via the provider factory. Pass *registry* to reuse one already built
239 (get_services builds it for embedding-dim reconciliation). Embedding-dim
240 reconciliation is a global-cfg concern owned by :func:`get_services`, not
241 done here.
243 Side effect: binds *provider* into xberg's process-global OCR and embedding
244 backends (the chunker binds the tokenizer on demand). The registry is one
245 process-wide slot, so every container (singleton or per-instance library)
246 binds its own, not only get_services.
247 """
248 from lilbee.catalog.hf_client import HfClient
249 from lilbee.modelhub.model_manager import ModelManager
250 from lilbee.modelhub.model_manager.discovery import KnownModelCache
251 from lilbee.modelhub.registry import ModelRegistry
252 from lilbee.modelhub.role_validator import warn_unregistered_role_refs
253 from lilbee.providers.factory import create_provider
254 from lilbee.retrieval.clustering import Clusterer
255 from lilbee.retrieval.concepts import ConceptGraph
256 from lilbee.retrieval.embedder import Embedder
257 from lilbee.retrieval.query import Searcher
258 from lilbee.retrieval.reranker import Reranker
259 from lilbee.runtime.ingest_lock import IngestLockRegistry
261 provider = provider or create_provider(
262 config, hold_warm=interactive, config_refresh=refresh_config
263 )
264 from lilbee.data.extract.backends import sync_xberg_backends
266 sync_xberg_backends(provider)
267 registry = registry or ModelRegistry(config.models_dir)
268 # The first point that sees env vars, --model and config.toml together.
269 warn_unregistered_role_refs(config, registry)
270 store = _new_store(config)
271 embedder = Embedder(config, provider)
272 reranker = Reranker(config)
273 concepts = ConceptGraph(config, store)
274 clusterer = Clusterer(config, store)
275 searcher = Searcher(config, provider, store, embedder, reranker, concepts)
276 hf_client = HfClient()
277 ingest_lock_registry = IngestLockRegistry()
278 model_manager = ModelManager(config.models_dir)
279 crawler_semaphore = (
280 asyncio.Semaphore(config.crawl_max_concurrent) if config.crawl_max_concurrent > 0 else None
281 )
282 crawler_sync_state = CrawlerSyncState()
283 known_models = KnownModelCache()
284 return Services(
285 provider=provider,
286 store=store,
287 embedder=embedder,
288 reranker=reranker,
289 concepts=concepts,
290 clusterer=clusterer,
291 searcher=searcher,
292 registry=registry,
293 hf_client=hf_client,
294 ingest_lock_registry=ingest_lock_registry,
295 model_manager=model_manager,
296 crawler_semaphore=crawler_semaphore,
297 crawler_sync_state=crawler_sync_state,
298 known_models=known_models,
299 )
302# Serializes first-touch singleton creation: several worker threads (e.g.
303# concurrent downloads) can call get_services() before the singleton exists,
304# and a duplicate build re-registers xberg's process-global backends mid-flight,
305# raising 'already registered' in the losing thread.
306_singleton_create_lock = threading.Lock()
309def get_services() -> Services:
310 """Return the active container: a scoped override if set, else the cached singleton.
312 Creates the singleton on first call (against the process-global cfg). A
313 config-file embedding_model with no embedding_dim would otherwise build the
314 store at the stale 768 default, so the width is pinned to the embedder before
315 the store is built.
316 """
317 override = _state.override.get()
318 if override is not None:
319 return override
320 if _state.singleton is not None:
321 return _state.singleton
323 from lilbee.core.config import cfg
325 with _singleton_create_lock:
326 if _state.singleton is not None:
327 return _state.singleton
329 from lilbee.app.settings import reconcile_embedding_dim
330 from lilbee.modelhub.registry import ModelRegistry
332 registry = ModelRegistry(cfg.models_dir)
333 # Pin the store width to the embedder before Store(); pass the registry so
334 # resolution doesn't re-enter this half-built get_services.
335 reconcile_embedding_dim(registry)
336 services = build_services(cfg, registry=registry, interactive=_state.interactive)
337 _state.singleton = services
338 # Eager start is the default: pay the spawn cost per role server at TUI mount
339 # so the first user action lands on a warm fleet. Roles whose model is unset
340 # are skipped, so a setup with only chat + embed never spawns rerank or
341 # vision. Set ``cfg.worker_pool_eager_start = false`` for headless scripts
342 # where mount time matters more than first-call latency.
343 if cfg.worker_pool_eager_start:
344 from contextlib import suppress
346 with suppress(Exception):
347 services.provider.warm_up_pool()
348 return services
351@contextmanager
352def services_scope(services: Services) -> Iterator[None]:
353 """Bind *services* as the container ``get_services()`` returns for this block.
355 Isolated to the entering task via a ContextVar (it propagates into
356 ``to_ingest_thread`` workers), and never affects the global singleton, so
357 ``reset_services`` is unnecessary and unused around a scoped call.
358 """
359 token = _state.override.set(services)
360 try:
361 yield
362 finally:
363 _state.override.reset(token)
366def mark_interactive_session() -> None:
367 """Record that this process is an interactive session before services build.
369 The TUI owns the process for its whole lifetime, so an engine bound to it
370 keeps its weights resident rather than idle-unloading under a user who is
371 still in the app; closing lilbee releases it. An engine that outlives the
372 session (shared with its other users, or kept warm) keeps its idle window
373 instead. Called by the interactive entry point before anything touches
374 ``get_services``, so the provider is constructed with that intent; a
375 one-shot CLI or the MCP server never calls it.
376 """
377 _state.interactive = True
380def set_config_refresh(hook: Callable[[], None] | None) -> None:
381 """Register how this process takes config.toml changes now (None clears it).
383 serve, the stdio MCP server and the TUI register the check they already run, so
384 the local engine can take an owner's new settings before it retries a call.
385 """
386 _state.config_refresh = hook
389def refresh_config() -> None:
390 """Take config.toml changes now through the registered check; nothing without one."""
391 hook = _state.config_refresh
392 if hook is not None:
393 hook()
396def report_library_upgrade(callback: DetailedProgressCallback | None) -> None:
397 """Show this process's library upgrades through *callback* (None: nowhere), for every store."""
398 _state.on_upgrade = callback
401def set_services(services: Services | None) -> None:
402 """Replace the cached Services singleton (for testing)."""
403 _state.singleton = services
406def chunks_in_store() -> bool:
407 """True when the index holds a chunk, asked of the store alone.
409 A bound container (scoped or cached) answers with its own store. Without
410 one, a short-lived store answers and no provider is built, so the check
411 starts no model warm-up.
412 """
413 from lilbee.core.config import cfg
415 bound = _state.override.get() or _state.singleton
416 if bound is not None:
417 return bound.store.has_chunks()
418 store = _new_store(cfg)
419 try:
420 return store.has_chunks()
421 finally:
422 store.close()
425def peek_services() -> Services | None:
426 """Return the cached Services container, or None if not yet initialized.
428 Public read-only accessor for test cleanup helpers that need to
429 inspect the singleton without forcing initialization.
430 """
431 return _state.singleton
434# Serializes the singleton swap in reset_services: a signal's teardown thread
435# and an exiting caller's can race, and both tearing down the same container
436# would double-close the store.
437_reset_swap_lock = threading.Lock()
440def reset_services() -> None:
441 """Shut down and discard all cached instances.
443 Swap the module reference to ``None`` *before* tearing the old instances
444 down, so a new caller never observes a half-closed container. The swap is
445 locked so concurrent callers (a signal's teardown thread plus an exiting
446 one) tear the container down exactly once. On the shared HTTP daemon every
447 entry point that would call this mid-flight is refused, so it only ever
448 runs single-client (CLI, TUI, stdio MCP).
449 """
450 with _reset_swap_lock:
451 old = _state.singleton
452 _state.singleton = None
453 if old is not None:
454 old.provider.shutdown()
455 old.store.close()
458def reset_store() -> None:
459 """Close and rebuild only the Store and its dependents; keep providers loaded.
461 Used after a data-dir wipe (``/reset``) where the LanceDB handle is invalid
462 but the running provider/embedder/reranker are still good. Avoids the
463 multi-second reload cost of ``reset_services()``.
464 """
465 svc = _state.singleton
466 if svc is None:
467 return
468 from dataclasses import replace
470 from lilbee.core.config import cfg
471 from lilbee.retrieval.clustering import Clusterer
472 from lilbee.retrieval.concepts import ConceptGraph
473 from lilbee.retrieval.query import Searcher
475 # Build the replacement, swap the reference, then close the old store last so
476 # a new caller never observes a closed handle mid-swap.
477 old_store = svc.store
478 store = _new_store(cfg)
479 concepts = ConceptGraph(cfg, store)
480 clusterer = Clusterer(cfg, store)
481 searcher = Searcher(cfg, svc.provider, store, svc.embedder, svc.reranker, concepts)
482 _state.singleton = replace(
483 svc,
484 store=store,
485 concepts=concepts,
486 clusterer=clusterer,
487 searcher=searcher,
488 )
489 old_store.close()
492class _ServerExit:
493 """Holds the running server's graceful-exit callback while it serves."""
495 def __init__(self) -> None:
496 self._hook: Callable[[int], None] | None = None
498 def set(self, hook: Callable[[int], None] | None) -> None:
499 self._hook = hook
501 def request(self, status: int) -> bool:
502 """Run the hook with the exit *status*; False when no server registered one."""
503 if self._hook is None:
504 return False
505 self._hook(status)
506 return True
509_server_exit = _ServerExit()
512def set_server_exit_hook(hook: Callable[[int], None] | None) -> None:
513 """Register the running server's graceful-exit callback (None clears it)."""
514 _server_exit.set(hook)
517def request_server_exit(status: int = 0) -> bool:
518 """Ask the running server to exit with *status*; False when none registered.
520 A nonzero *status* is a hard exit: the server does not wait for responses in flight.
521 """
522 return _server_exit.request(status)
525class _EngineLifecycle:
526 """Owns the hard-exit hooks that stop the engine fleet."""
528 def __init__(self) -> None:
529 self._installed = False
531 @staticmethod
532 def _hard_exit_signals() -> tuple[signal.Signals, ...]: # pragma: no cover - platform split
533 """Signals whose default disposition kills us without running atexit."""
534 if sys.platform == "win32": # Windows has no SIGHUP
535 return (signal.SIGTERM,)
536 return (signal.SIGTERM, signal.SIGHUP)
538 def install(self) -> None:
539 """Route hard-exit signals through teardown. Idempotent; no-op off the main thread."""
540 if self._installed:
541 return
542 try:
543 for sig in self._hard_exit_signals():
544 signal.signal(sig, self._on_hard_exit)
545 except ValueError:
546 return
547 self._installed = True
549 def route_in_loop(self, loop: asyncio.AbstractEventLoop) -> None:
550 """Deliver Ctrl-C and the hard-exit signals as loop callbacks.
552 A plain handler runs only when the main thread returns to bytecode, and the
553 kernel may hand the signal to another thread while the loop sits in select.
554 The loop's wakeup descriptor makes the delivery certain.
555 """
556 for sig in (signal.SIGINT, *self._hard_exit_signals()):
557 with suppress(NotImplementedError, RuntimeError): # Windows, off main thread
558 loop.add_signal_handler(sig, self._on_hard_exit, sig, None)
560 def reset(self) -> None:
561 """Forget that handlers were installed."""
562 self._installed = False
564 def _on_hard_exit(self, signum: int, frame: object) -> None:
565 """Stop the fleet on its own thread, then exit with the signal status.
567 A serving loop is asked to stop and exits with the status itself. A
568 SystemExit raised into the loop cancels the ASGI lifespan with it,
569 which logs a traceback, or dies inside a request's ASGI wrapper. With
570 no serving loop the SystemExit unwinds the main thread. Signal handlers
571 all run on the main thread, and a second signal can interrupt this one
572 mid-teardown: the kernel pairs SIGCONT with SIGHUP for an orphaned
573 process group, and Textual's SIGCONT handler raises once the event loop
574 is gone, which aborted the reap half-done and orphaned a loaded fleet.
575 A dedicated non-daemon thread cannot be interrupted by signals, and the
576 interpreter waits for it even as the SystemExit unwinds the main thread.
577 """
578 del frame
579 status = _SIGNAL_EXIT_BASE + signum
580 serving = request_server_exit(status)
581 threading.Thread(
582 target=_teardown_for_signal, args=(signum,), name=_HARD_EXIT_THREAD_NAME
583 ).start()
584 if not serving:
585 raise SystemExit(status)
588_FORCE_QUIT_EXIT_CODE = 130
590_STOPPING_NOTE = "lilbee: stopping the engine. Press Ctrl-C again to force quit.\n"
592_STOPPING_NOTE_GRACE_S = 1.0
593"""Seconds a teardown may run silently; a fast exit stays as quiet as ever."""
595_FORCE_QUIT_NOTE = "lilbee: force quit. The engine stops with it.\n"
597_STRAGGLER_BUDGET_S = 5.0
598"""Seconds every leftover non-daemon thread gets, together, before the exit."""
600_STRAGGLER_NOTE = "lilbee: a background task would not stop; exiting anyway.\n"
602_OPEN_EXIT_NOTE = "lilbee: the library is still opening; exiting now. The next start finishes it.\n"
605def _write_exit_note(note: str) -> None:
606 """Print an exit-path status line; stderr may already be gone at exit."""
607 try:
608 sys.stderr.write(note)
609 sys.stderr.flush()
610 except (OSError, ValueError):
611 pass
614def _flush_logs() -> None:
615 """Flush every log handler and stdio; ``os._exit`` drops what a buffer still holds."""
616 # loggerDict also holds PlaceHolder entries for loggers that have no object yet.
617 named = [e for e in logging.Logger.manager.loggerDict.values() if isinstance(e, logging.Logger)]
618 loggers = [logging.getLogger(), *named]
619 for logger in loggers:
620 for handler in logger.handlers:
621 with suppress(OSError, ValueError):
622 handler.flush()
623 for stream in (sys.stdout, sys.stderr):
624 with suppress(AttributeError, OSError, ValueError):
625 stream.flush()
628def _exit_now(status: int, note: str) -> None:
629 """Write *note*, flush the logs, then end the process with *status*, skipping atexit."""
630 _write_exit_note(note)
631 _flush_logs()
632 os._exit(status)
635def _all_threads() -> list[threading.Thread]:
636 """Seam over threading.enumerate, so tests never patch the stdlib global."""
637 return threading.enumerate()
640def wait_for_hard_exit_teardown() -> None:
641 """Block until any teardown thread (signal-driven or exit-driven) finishes.
643 Lets ``serve`` hold its OS locks through the fleet stop, so a successor
644 cannot acquire them while this server's models still occupy memory.
646 An interrupt while waiting force-quits the process: waiting is the only
647 thing left, the interpreter would otherwise still join the non-daemon
648 teardown thread, and the engine reaper (or, without kernel file locks, the
649 child guard) stops the engine once this was its last user.
650 """
651 try:
652 for thread in _all_threads():
653 if thread.name != _HARD_EXIT_THREAD_NAME:
654 continue
655 thread.join(_STOPPING_NOTE_GRACE_S)
656 if thread.is_alive():
657 _write_exit_note(_STOPPING_NOTE)
658 thread.join()
659 except KeyboardInterrupt:
660 _exit_now(_FORCE_QUIT_EXIT_CODE, _FORCE_QUIT_NOTE)
663def reset_services_on_exit() -> None:
664 """Tear the container down on a thread no signal reaches, and wait for it.
666 Engine release waits on the fleet build lock before it releases anything, so
667 a Ctrl-C on the main thread skips the release, and atexit cannot retry: the
668 singleton is already cleared. The teardown thread is non-daemon and takes
669 no signals; a wait that outlives the grace names what is happening.
670 """
671 if peek_services() is None:
672 return
673 threading.Thread(target=reset_services, name=_HARD_EXIT_THREAD_NAME).start()
674 wait_for_hard_exit_teardown()
677def _straggler_threads() -> list[threading.Thread]:
678 """Non-daemon threads the interpreter would join at shutdown, besides us."""
679 return [
680 thread
681 for thread in _all_threads()
682 if thread.is_alive()
683 and not thread.daemon
684 and thread is not threading.main_thread()
685 and thread is not threading.current_thread()
686 ]
689def exit_when_stragglers_would_hang() -> None:
690 """Exit rather than let interpreter shutdown join a wedged thread forever.
692 threading joins every non-daemon thread after atexit, so one worker
693 blocked in an unbounded network read hangs the process after all real
694 work is done. Stragglers share a short budget; whatever remains cannot
695 finish, and the child guard reaps the engine when the process dies.
696 Called from the TUI's own exit path and inert elsewhere: a test process
697 or embedding host owns its threads, and an exit would take them with it.
698 """
699 if not _state.interactive:
700 return
701 _exit_if_stragglers(0)
704def _exit_if_stragglers(status: int) -> None:
705 """Join leftover non-daemon threads within the budget, else exit with *status*."""
706 try:
707 deadline = time.monotonic() + _STRAGGLER_BUDGET_S
708 for thread in _straggler_threads():
709 thread.join(max(0.0, deadline - time.monotonic()))
710 if _straggler_threads():
711 _exit_now(status, _STRAGGLER_NOTE)
712 except KeyboardInterrupt:
713 _exit_now(_FORCE_QUIT_EXIT_CODE, _FORCE_QUIT_NOTE)
716def exit_if_opens_remain(status: int) -> None:
717 """End the process with *status* when a library open still runs; else return."""
718 # A daemon thread inside LanceDB at finalization aborts Python 3.11 and 3.12
719 # (CPython gh-87135). The exit skips atexit, so the teardown runs first.
720 if wait_for_opens(0.0):
721 return
722 reset_services_on_exit()
723 _exit_now(status, _OPEN_EXIT_NOTE)
726def exit_on_signals_in_loop(loop: asyncio.AbstractEventLoop) -> None:
727 """Make Ctrl-C and terminate signals end a process whose loop has no UI to catch them."""
728 _lifecycle.route_in_loop(loop)
731def exit_after_signal(status: int) -> None:
732 """Finish the signal-driven teardown, then exit with *status* even past a wedged thread.
734 The stdio MCP server reads stdin on a non-daemon thread that no signal
735 wakes, so the interpreter would wait on it forever after the main thread
736 unwound.
737 """
738 reset_services_on_exit()
739 _exit_if_stragglers(status)
742def _teardown_for_signal(signum: int) -> None:
743 """Log the fatal signal, then stop services; runs off the signal handler's thread."""
744 log.info(
745 "Received signal %s; stopping the engine fleet before exit", signal.Signals(signum).name
746 )
747 reset_services()
750_lifecycle = _EngineLifecycle()
753def install_engine_lifecycle_hooks() -> None:
754 """Make a terminal close or ``kill`` stop the engine fleet instead of orphaning it."""
755 _lifecycle.install()
758atexit.register(reset_services_on_exit)