Skip to content

core.runtime

Runtime modules: startup lifecycle, top-level application container, configuration, and asyncio primitives singleton.

Shared State

core.runtime.shared_state

Shared runtime state — replaces cross-module asyncio globals.

Holds asyncio.Event and asyncio.Lock instances that are accessed from multiple modules (agent_classes, audio_processing, image_processing). Events and locks are created lazily on first access so they bind to the running event loop rather than a stale module-level loop.

SharedState

Container for cross-module async primitives.

All asyncio.Event and asyncio.Lock instances are created lazily on first access, ensuring they bind to the running loop.

agent_generating: bool property writable

True when the main agent is actively streaming tokens from Ollama.

Background tasks check this flag and back off to avoid concurrent Ollama pressure that can trigger WDDM TDR or llama-server crashes.

bg_task_notifications: queue.Queue[str] property

Thread-safe queue of background task completion messages.

The background task runner pushes messages here. The main loop drains them before each iteration.

generation_cancel_event: asyncio.Event property

Signal to cancel an active agent generation turn.

Sets a structured cancellation point that the Agent's streaming generator checks at every loop iteration — alongside tts_interrupt_event so TTS playback and LLM generation stop together.

stop_recording_event: asyncio.Event property

Signal to stop active recording (set by desktop text message).

text_only_flag: bool property writable

When True, desktop text message should bypass audio transcription.

append_screenshot(text: str) -> None

Append a sentence to the screenshot description (thread-safe).

clear_screenshot() -> None

Clear the screenshot description.

reset() -> None

Clear interrupt and reset can_speak.

Does NOT recreate the Event objects (they stay bound to the same loop).

get_shared_state() -> SharedState

Return the global SharedState instance.

Raises RuntimeError if called before initialize_shared_state().

Lifecycle Manager

core.runtime.lifecycle_manager

Lifecycle manager for the Vector Companion runtime.

Encapsulates all initialization and shutdown phases that currently live at the top level of main.py. Each phase is a public method so that tests, the integration harness, and future refactors can call them independently.

Phases (called in order by initialize()): 1. setup_environment() — OLLAMA_HOST env var, faulthandler 2. setup_logging() — loguru stderr + file handlers 3. detect_device() — torch device selection 4. setup_torch() — cudnn flags, cache management 5. initialize_shared_state() — asyncio events/locks singleton 6. load_audio_model() — Qwen3-ASR transcription model 7. load_tts_model() — ChatterboxTTS + T3 dtype cast 8. load_reranker() — CrossEncoder reranker 9. build_agents() — lazy agent factory 10. load_history() — conversation history JSON 11. load_tools() — tool schemas from tools/schemas.py 12. setup_telemetry() — MetricsCollector singleton

LifecycleManager

Orchestrates startup initialization and shutdown cleanup.

Parameters

config : RuntimeConfig All configuration values needed for initialization.

audio_playback_lock: Any property

Return the audio playback lock.

can_speak_event: Any property

Return the can_speak asyncio event.

tts_interrupt_event: Any property

Return the TTS interrupt event.

build_agents(*, tool_host) -> None

Phase 9: Build agents via the factory.

Parameters

tool_host : ToolHost The ToolHost instance providing the 8-step security pipeline. This parameter is mandatory — legacy bypass mode has been removed.

detect_device() -> None

Phase 3: Detect and store the compute device.

initialize() -> None

Run all initialization phases in order.

Note: Phase 9 (build_agents) is NOT included here because it requires a ToolHost which is created after lifecycle init in the production flow (main.py). Call build_agents(tool_host=...) explicitly after constructing the ToolHost via ToolHostAdapter.

initialize_shared_state() -> None

Phase 5: Initialize the SharedState singleton.

load_audio_model() -> None

Phase 6: Load STT backend based on config.stt_backend.

Builds an ordered fallback chain of available backends, wraps in STTBackendAdapter for Qwen3ASRModel-compatible interface so existing call sites (audio_processing.py::transcribe_audio) require no changes.

Default: Parakeet-TDT via Transformers AutoModel (lazy-loaded). Fallback: FallbackSTTBackend (silent) if Parakeet unavailable. Qwen3-ASR is incompatible with transformers v5 and treated as unavailable. Controlled by config.stt_backend: "auto" or "parakeet_tdt_v3".

load_history() -> None

Phase 10: Load conversation history from disk (if present).

load_reranker() -> None

Phase 8: Reranker is now lazy-loaded via LazyLoader.

The CrossEncoder reranker (~250MB on GPU) is only loaded when memory_mode activates folder-scoped RAG, not at startup. Lifecycle phase is preserved for ordering but defers actual load.

load_tools() -> None

Phase 11: Load tool schemas.

load_tts_model() -> None

Phase 7: Load ChatterboxTTS and cast T3 to bfloat16.

setup_environment() -> None

Phase 1: Set environment variables and enable faulthandler.

setup_logging() -> None

Phase 2: Configure loguru handlers (stderr + file) and StructuredLogger.

setup_telemetry() -> None

Phase 12: Initialize the MetricsCollector singleton.

setup_torch() -> None

Phase 4: Configure PyTorch flags and clear cache.

shutdown() -> None

Run shutdown cleanup.

Runtime App

core.runtime.runtime_app

Runtime application container for Vector Companion.

RuntimeApp is the top-level container that brings together:

  • RuntimeConfig — all configuration values
  • ServiceRegistry — DI container for service implementations
  • LifecycleManager — phased startup/shutdown orchestration

It provides a single run() method that initialises all phases, launches the main event loop (or a supplied hook), and shuts down cleanly.

Usage

app = RuntimeApp() app.initialize()

wire services into agent/tool dependencies ...

app.shutdown()

RuntimeApp

Application container for the Vector Companion runtime.

Parameters

config : RuntimeConfig, optional Configuration values. If omitted a default RuntimeConfig is created via from_config_module(). services : ServiceRegistry, optional Pre-populated DI container (useful for tests). When omitted a fresh ServiceRegistry is created. lifecycle : LifecycleManager, optional Pre-built lifecycle manager. When omitted one is created from config.

agents: list[Any] property

Return the built agents.

device: str property

Return the detected compute device.

messages: list[dict] property

Return the conversation history messages.

run_id: str property

Return the lifecycle run identifier (empty until setup_logging()).

shared_state: Any property

Return the shared state singleton.

get_registered_count() -> int

Return the number of registered services.

initialize(skip_heavy: bool = False) -> None

Run all initialization phases.

Parameters

skip_heavy : bool When True skip the heavyweight model-loading phases (audio, TTS). Useful for CI and integration tests.

register_service(service_type: type, instance: Any) -> None

Bind instance to service_type in the service registry.

register_services_if_absent(mapping: dict[type, Any]) -> dict[type, bool]

Register each entry only if absent. Returns per-type result map.

report_health(name: str, healthy: bool, detail: str = '', latency_ms: float | None = None) -> None

Forward a health report to the degraded-mode manager (if present).

run(main_loop: Callable[..., Any], **kwargs: Any) -> None async

Initialize, run main_loop, then shut down.

Parameters

main_loop : Callable Async coroutine (the event-loop body from main.py). kwargs : Any Additional keyword arguments forwarded to main_loop.

shutdown() -> None

Cancel background tasks and run shutdown cleanup.

Runtime Config

core.runtime.runtime_config

Runtime configuration dataclass for Vector Companion.

Collects all module-level globals from main.py into a single immutable-ish container so that lifecycle services can be constructed without reaching into the global namespace.

This is a data-only container — no instantiation side effects.

RuntimeConfig dataclass

Holds all configuration values needed to bootstrap the runtime.

Derived from config.config module attributes plus a handful of hardcoded defaults that currently live at the top of main.py.

from_config_module() -> RuntimeConfig classmethod

Populate defaults from config.config module attributes.

Falls back to class defaults for any attribute not present in the module — this avoids import-time breakage if config.py is trimmed.

to_dict() -> dict[str, Any]

Serialize to a plain dict (useful for logging/debugging).