Server runtime API

locallore.server.runtime

logger = logging.getLogger(__name__) module-attribute

Snapshot = tuple[tuple[str, str, int, int], ...] module-attribute

RefreshStats(files_added=0, files_removed=0, messages_added=0, messages_removed=0, messages_embedded=0) dataclass

files_added = 0 class-attribute instance-attribute

files_removed = 0 class-attribute instance-attribute

messages_added = 0 class-attribute instance-attribute

messages_removed = 0 class-attribute instance-attribute

messages_embedded = 0 class-attribute instance-attribute

LocalLoreRuntime(settings)

Own daemon state, one lazy model, and one background refresh worker.

Source code in src/locallore/server/runtime.py

def __init__(self, settings: Settings) -> None:
    self.settings = settings
    self.inference_lock = threading.Lock()
    self._model_lock = threading.Lock()
    self._embedder: FastEmbedder | None = None
    self._locked_embedder = _LockedEmbedder(self)
    self._refresh_event = asyncio.Event()
    self._tasks: list[asyncio.Task[None]] = []
    self._stopping = False
    self._snapshot: Snapshot | None = None
    self._model_id: str | None = None
    self._started_monotonic = time.monotonic()
    self.refresh_state = "starting"
    self.last_refresh_started_at: str | None = None
    self.last_refresh_completed_at: str | None = None
    self.last_successful_refresh_at: str | None = None
    self.last_refresh_duration_seconds: float | None = None
    self.last_background_error: str | None = None
    self.last_stats = RefreshStats()

settings = settings instance-attribute

inference_lock = threading.Lock() instance-attribute

refresh_state = 'starting' instance-attribute

last_refresh_started_at = None instance-attribute

last_refresh_completed_at = None instance-attribute

last_successful_refresh_at = None instance-attribute

last_refresh_duration_seconds = None instance-attribute

last_background_error = None instance-attribute

last_stats = RefreshStats() instance-attribute

model_id property

search_embedder property

start() async

Source code in src/locallore/server/runtime.py

async def start(self) -> None:
    if not self.settings.sessions_path.is_dir():
        raise FileNotFoundError(
            f"session directory does not exist: {self.settings.sessions_path}"
        )
    self._model_id = embedding_model_id(
        self.settings.embedding_model, self.settings.model_path
    )
    connection = connect(self.settings.database_path)
    try:
        migrate(connection)
    finally:
        connection.close()
    self._tasks = [
        asyncio.create_task(self._refresh_worker(), name="locallore-indexer"),
        asyncio.create_task(self._watch_sources(), name="locallore-watcher"),
    ]
    self.request_refresh()

stop() async

Source code in src/locallore/server/runtime.py

async def stop(self) -> None:
    self._stopping = True
    for task in self._tasks:
        task.cancel()
    await asyncio.gather(*self._tasks, return_exceptions=True)
    self._tasks.clear()

request_refresh()

Source code in src/locallore/server/runtime.py

def request_refresh(self) -> None:
    self._refresh_event.set()

wait_until_ready(timeout=300.0) async

Source code in src/locallore/server/runtime.py

async def wait_until_ready(self, timeout: float = 300.0) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if (
            self.refresh_state == "idle"
            and self.last_successful_refresh_at is not None
        ):
            return True
        await asyncio.sleep(0.05)
    return False

status()

Source code in src/locallore/server/runtime.py

def status(self) -> RuntimeStatus:
    return {
        "daemon_version": __version__,
        "uptime_seconds": round(time.monotonic() - self._started_monotonic, 3),
        "refresh_state": self.refresh_state,
        "last_refresh_started_at": self.last_refresh_started_at,
        "last_refresh_completed_at": self.last_refresh_completed_at,
        "last_successful_refresh_at": self.last_successful_refresh_at,
        "last_refresh_duration_seconds": self.last_refresh_duration_seconds,
        "refresh_queued": self._refresh_event.is_set(),
        "last_refresh_files_added": self.last_stats.files_added,
        "last_refresh_files_removed": self.last_stats.files_removed,
        "last_refresh_messages_added": self.last_stats.messages_added,
        "last_refresh_messages_removed": self.last_stats.messages_removed,
        "last_background_error": self.last_background_error,
    }

source_snapshot(root)

Return a stable, content-free snapshot of all JSONL sources.

Source code in src/locallore/server/runtime.py

def source_snapshot(root: Path) -> Snapshot:
    """Return a stable, content-free snapshot of all JSONL sources."""
    return tuple(
        (
            source.relative_path,
            source.identity,
            source.size_bytes,
            source.mtime_ns,
        )
        for source in discover(root)
    )