Skip to content

Status and diagnostics API

locallore.status

RuntimeStatus

Bases: TypedDict

daemon_version instance-attribute

uptime_seconds instance-attribute

refresh_state instance-attribute

last_refresh_started_at instance-attribute

last_refresh_completed_at instance-attribute

last_successful_refresh_at instance-attribute

last_refresh_duration_seconds instance-attribute

refresh_queued instance-attribute

last_refresh_files_added instance-attribute

last_refresh_files_removed instance-attribute

last_refresh_messages_added instance-attribute

last_refresh_messages_removed instance-attribute

last_background_error instance-attribute

Status

Bases: RuntimeStatus

schema_version instance-attribute

last_refresh instance-attribute

sessions instance-attribute

messages instance-attribute

embedded_messages instance-attribute

embedding_model_id instance-attribute

import_errors instance-attribute

get_status(database_path=None, runtime_status=None)

Source code in src/locallore/status.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_status(
    database_path: Path | None = None,
    runtime_status: RuntimeStatus | None = None,
) -> Status:
    sessions = messages = embedded_messages = 0
    embedding_model_id = None
    errors: list[str] = []
    last_refresh = None
    if database_path is not None and database_path.exists():
        connection = sqlite3.connect(database_path)
        sessions = connection.execute("SELECT count(*) FROM sessions").fetchone()[0]
        messages = connection.execute("SELECT count(*) FROM messages").fetchone()[0]
        embedded_messages = connection.execute(
            "SELECT count(*) FROM embeddings"
        ).fetchone()[0]
        model = connection.execute(
            "SELECT model_id FROM embeddings GROUP BY model_id ORDER BY count(*) DESC LIMIT 1"
        ).fetchone()
        embedding_model_id = model[0] if model else None
        errors = [
            row[0]
            for row in connection.execute(
                "SELECT path FROM import_files WHERE last_error IS NOT NULL"
            )
        ]
        last_refresh = connection.execute(
            "SELECT max(updated_at) FROM import_files"
        ).fetchone()[0]
        connection.close()
    background = runtime_status or RuntimeStatus(
        daemon_version=__version__,
        uptime_seconds=None,
        refresh_state="idle",
        last_refresh_started_at=None,
        last_refresh_completed_at=None,
        last_successful_refresh_at=last_refresh,
        last_refresh_duration_seconds=None,
        refresh_queued=False,
        last_refresh_files_added=0,
        last_refresh_files_removed=0,
        last_refresh_messages_added=0,
        last_refresh_messages_removed=0,
        last_background_error=None,
    )
    return {
        "schema_version": SCHEMA_VERSION,
        "last_refresh": last_refresh,
        "sessions": sessions,
        "messages": messages,
        "embedded_messages": embedded_messages,
        "embedding_model_id": embedding_model_id,
        "import_errors": errors,
        **background,
    }

locallore.doctor

DoctorError

Bases: RuntimeError

Raised when a runtime prerequisite is unavailable or unsafe.

DoctorReport(checks) dataclass

checks instance-attribute

run_doctor(settings)

Source code in src/locallore/doctor.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def run_doctor(settings: Settings) -> DoctorReport:
    checks: list[str] = []
    _require_directory(settings.sessions_path, "session")
    checks.append("session directory is readable")

    database_parent = settings.database_path.parent
    _require_directory(database_parent, "database", readable=False)
    try:
        with tempfile.NamedTemporaryFile(dir=database_parent):
            pass
    except OSError as exc:
        raise DoctorError(
            f"database directory is not writable: {database_parent}: {exc}"
        ) from exc
    checks.append("database directory is writable")

    connection: sqlite3.Connection | None = None
    try:
        connection = connect(settings.database_path)
        migrate(connection)
        if connection.execute("PRAGMA foreign_keys").fetchone()[0] != 1:
            raise DoctorError("SQLite foreign-key enforcement is disabled")
        connection.execute("SELECT count(*) FROM messages_fts").fetchone()
    except sqlite3.Error as exc:
        raise DoctorError(f"database or FTS5 check failed: {exc}") from exc
    finally:
        if connection is not None:
            connection.close()
    checks.append("database, migrations, foreign keys, and FTS5 are ready")

    _require_directory(settings.model_path, "embedding model")
    try:
        embedder = FastEmbedder(
            settings.embedding_model,
            settings.model_path,
            settings.embedding_dimension,
        )
        vector = embedder.encode_query("LocalLore offline diagnostic")
    except Exception as exc:
        raise DoctorError(f"embedding model check failed: {exc}") from exc
    if vector.shape != (settings.embedding_dimension,):
        raise DoctorError(f"embedding model returned unexpected shape {vector.shape}")
    checks.append("local embedding assets and inference are ready")

    return DoctorReport(tuple(checks))