core.data¶
Data modules: SQLite WAL-mode persistence, ChromaDB embedding, CrossEncoder reranking, and conversation management.
SQLite Store¶
core.data.sqlite_store
¶
Canonical SQLite-backed local persistence store.
Consolidates conversations, messages, settings, memories, attachments, audit events, tool grants, and background tasks into a single WAL-mode SQLite database. ChromaDB remains the semantic index adapter; embeddings are rebuildable from the canonical store.
Follows the SqliteAuditLog pattern: per-call connections, WAL mode,
asyncio lock, schema auto-creation, JSON serialization, frozen dataclasses.
AttachmentRecord
dataclass
¶
Immutable attachment record.
BackgroundTaskRecord
dataclass
¶
Immutable background task record.
ConversationRecord
dataclass
¶
Immutable conversation record.
EmbeddingIndexRecord
dataclass
¶
Tracks which messages have been embedded in ChromaDB.
FolderRecord
dataclass
¶
Immutable folder record.
MemoryRecord
dataclass
¶
Immutable memory record.
MessageRecord
dataclass
¶
Immutable message record.
SQLiteStore
¶
Canonical SQLite-backed local persistence store.
Satisfies the LocalStoreProtocol. Uses WAL mode for concurrent
reads/writes. All CRUD methods are async and lock-protected.
Parameters¶
db_path :
Path to the SQLite database file.
wal_mode :
Enable Write-Ahead Logging (default True).
busy_timeout :
SQLite busy timeout in milliseconds (default 5000).
add_attachment(*, message_id: str, atype: str, data_path: str = '', data_bytes: bytes | None = None, metadata: dict | None = None) -> AttachmentRecord
async
¶
Add an attachment to a message.
add_message(*, conversation_id: str, role: str, content: str, tool_call_id: str = '', metadata: dict | None = None, tool_name: str = '') -> MessageRecord
async
¶
Add a message to a conversation.
backup(dest_path: str | Path) -> None
async
¶
Create an online backup via the SQLite backup API.
close() -> None
async
¶
Close any lingering connections (no-op for per-call pattern).
complete_task(task_id: str, *, result: dict | None = None) -> bool
async
¶
Mark a task as completed.
conversation_count() -> int
async
¶
Return the total number of conversations.
create_conversation(*, agent: str, mode_flags: dict | None = None, metadata: dict | None = None, title: str = '', folder_id: str | None = None) -> ConversationRecord
async
¶
Create a new conversation and return its record.
create_folder(*, name: str) -> FolderRecord
async
¶
Create a new folder and return its record.
create_task(*, task_id: str, description: str) -> BackgroundTaskRecord
async
¶
Create a new background task record.
db_path() -> str
¶
Return the database file path.
delete_attachment(attachment_id: str) -> bool
async
¶
Delete an attachment. Returns True if it existed.
delete_completed_tasks(older_than_days: int = 30) -> int
async
¶
Delete completed tasks older than a threshold.
delete_conversation(conversation_id: str) -> int
async
¶
Delete a conversation and its messages (cascade). Returns rows deleted.
delete_embeddings_for_conversation(conversation_id: str) -> int
async
¶
Delete embedding index entries for a conversation.
delete_embeddings_for_folder(folder_id: str) -> int
async
¶
Delete embedding index entries for a folder.
delete_folder(folder_id: str) -> int
async
¶
Delete a folder (conversations become standalone via ON DELETE SET NULL).
delete_memories(*, collection: str, doc_ids: list[str] | None = None) -> int
async
¶
Delete memories by collection, optionally filtered by IDs.
delete_memory(doc_id: str) -> bool
async
¶
Delete a memory. Returns True if it existed.
delete_messages_for_conversation(conversation_id: str) -> int
async
¶
Delete all messages for a conversation.
delete_setting(key: str) -> bool
async
¶
Delete a setting. Returns True if it existed.
execute(sql: str, params: tuple | None = None) -> list[tuple]
¶
Execute a raw SQL query and return all rows.
Provides the duck-typed _StoreLike.execute contract expected by
ConversationSearch. Read-only — callers should pass parameterized
queries to avoid SQL injection.
export_settings() -> dict[str, Any]
async
¶
Alias for list_settings.
fail_task(task_id: str, *, error: str) -> bool
async
¶
Mark a task as failed.
get_attachment(attachment_id: str) -> AttachmentRecord | None
async
¶
Fetch an attachment by ID.
get_collections() -> list[str]
async
¶
Return distinct collection names.
get_conversation(conversation_id: str) -> ConversationRecord | None
async
¶
Fetch a conversation by ID.
get_embedding_index(message_id: str) -> EmbeddingIndexRecord | None
async
¶
Fetch embedding index entry by message_id.
get_embedding_index_by_conversation(conversation_id: str) -> list[EmbeddingIndexRecord]
async
¶
Fetch all embedding index entries for a conversation.
get_embedding_index_by_folder(folder_id: str) -> list[EmbeddingIndexRecord]
async
¶
Fetch all embedding index entries for a folder.
get_folder(folder_id: str) -> FolderRecord | None
async
¶
Fetch a folder by ID.
get_memory(doc_id: str) -> MemoryRecord | None
async
¶
Fetch a memory by doc_id.
get_message(message_id: str) -> MessageRecord | None
async
¶
Fetch a single message by ID.
get_message_hashes(conversation_id: str) -> set[str]
async
¶
Return set of content hashes for persistable messages in a conversation.
Used by the persist loop to detect genuinely new messages regardless of positional indexing — avoids drift when cleanup_message_history() applies stricter filtering than historical runs.
Whitespace is normalized (' '.join(s.split())) to match the output of _clean_user_content(), ensuring hash consistency even when stored content has different internal whitespace than cleaned in-memory content.
get_messages(conversation_id: str, *, limit: int | None = 100, offset: int = 0) -> list[MessageRecord]
async
¶
Get messages for a conversation, newest first.
When limit is None, returns all messages (no LIMIT clause).
get_setting(key: str, default: Any = None) -> Any
async
¶
Get a setting by key.
get_task(task_id: str) -> BackgroundTaskRecord | None
async
¶
Fetch a task by ID.
get_tool_grant(tool_name: str) -> ToolGrantRecord | None
async
¶
Fetch the most recent grant/denial for a tool.
grant_tool(*, tool_name: str, decision: str, persistent: bool = False) -> ToolGrantRecord
async
¶
Record a tool grant or denial.
has_persistent_deny(tool_name: str) -> bool
async
¶
Check if a tool has a persistent denial.
has_persistent_grant(tool_name: str) -> bool
async
¶
Check if a tool has a persistent grant.
import_settings(data: dict[str, Any]) -> None
async
¶
Import settings from a dict (overwrites existing keys).
list_all_with_folders() -> dict[str, Any]
async
¶
Return folders and conversations together for sidebar hydration.
list_attachments(message_id: str) -> list[AttachmentRecord]
async
¶
List attachments for a message.
list_conversations(*, agent: str | None = None, limit: int | None = 50) -> list[ConversationRecord]
async
¶
List conversations, optionally filtered by agent.
When limit is None, returns all conversations (no LIMIT clause).
list_folders() -> list[FolderRecord]
async
¶
Return all folders ordered by name.
list_memories(*, collection: str, limit: int = 100) -> list[MemoryRecord]
async
¶
List memories in a collection.
list_settings() -> dict[str, Any]
async
¶
Return all settings as a dict.
list_tasks(*, status: str | None = None, limit: int = 50) -> list[BackgroundTaskRecord]
async
¶
List tasks, optionally filtered by status.
list_tool_grants() -> list[ToolGrantRecord]
async
¶
List all tool grants.
log_audit(*, event_type: str, source: str, correlation_id: str = '', data: dict | None = None, level: str = 'info') -> None
async
¶
Log an audit event.
memory_count(*, collection: str | None = None) -> int
async
¶
Count memories, optionally filtered by collection.
message_count(conversation_id: str) -> int
async
¶
Count messages in a conversation.
Counts only persistable roles (user/assistant/tool) with non-empty content to stay in sync with the persist loop's filtering in main.py. This prevents count drift when cleanup_message_history strips empty messages.
query_audit(*, event_type: str | None = None, source: str | None = None, correlation_id: str | None = None, level: str | None = None, limit: int = 100) -> list[dict[str, Any]]
async
¶
Query audit events with filters, most recent first.
reset() -> None
async
¶
Drop all tables and recreate the schema (for tests).
revoke_tool_grant(tool_name: str) -> bool
async
¶
Revoke all grants/denials for a tool.
schema_version() -> int
¶
Return the current schema version.
set_setting(key: str, value: Any) -> None
async
¶
Set a setting (upsert).
update_conversation(conversation_id: str, *, ended_at: str | None | Any = None, metadata: dict | None = None, title: str | None | Any = None, folder_id: str | None | Any = _UNSET) -> bool
async
¶
Update conversation fields. Returns True if a row was changed.
Pass folder_id=... (including None to unset) to change the
folder assignment. Omit the parameter entirely to leave it unchanged.
update_embedding_folder(*, conversation_id: str, old_folder_id: str | None, new_folder_id: str | None) -> int
async
¶
update_folder(folder_id: str, *, name: str) -> bool
async
¶
Rename a folder. Returns True if a row was changed.
update_message(message_id: str, *, content: str | None = None, metadata: dict | None = None) -> bool
async
¶
Update message fields. Returns True if a row was changed.
update_task_status(task_id: str, *, status: str, result: dict | None = None, error: str | None = None) -> bool
async
¶
Update task status and optional fields.
upsert_embedding_index(*, message_id: str, chroma_doc_id: str, collection: str, folder_id: str | None, conversation_id: str) -> EmbeddingIndexRecord
async
¶
Record or update an embedding index entry.
upsert_memory(*, collection: str, document: str, embedding_ref: str = '', metadata: dict | None = None, doc_id: str | None = None) -> MemoryRecord
async
¶
Insert or update a memory record.
SchemaMigration
dataclass
¶
A single schema migration step.
Parameters¶
version : Monotonically increasing version number. sql : SQL statements to apply for this migration. description : Human-readable description of the change.
SettingRecord
dataclass
¶
Immutable setting record.
ToolGrantRecord
dataclass
¶
Immutable tool grant record.
Embedding¶
core.data.embedding
¶
delete_documents_by_ids(doc_ids: list[str], db_client, collection: str = 'chat_messages') -> int
¶
delete_folder_embeddings(folder_id: str, db_client, collection: str = 'chat_messages') -> int
¶
query_collection(query, db_client=None, collection='web_search', embedding_model='qwen3-embedding:0.6b', n_results=100, include=None, where=None)
¶
Query a ChromaDB collection by semantic similarity.
Parameters¶
where :
Optional ChromaDB metadata filter dict, e.g.
{"folder_id": {"$eq": "abc123"}}. Passed directly to
collection_ref.query(where=where).
query_folder_messages(query_text, folder_id, db_client, *, exclude_conversation_id=None, collection='chat_messages', embedding_model='qwen3-embedding:0.6b', n_results=20)
¶
Query the chat_messages collection for a folder.
Retrieves semantically similar passages from all conversations in folder_id, optionally excluding results from the active conversation.
Parameters¶
exclude_conversation_id : If provided, post-filter results to drop passages from this conversation (so the agent doesn't echo the current turn back).
Returns¶
dict
ChromaDB query result with documents, metadatas,
distances, and ids keys (post-filtered if requested).
set_embedding_backend(backend)
¶
Set an optional ModelBackend for embedding operations.
When set, all ollama.embed() calls route through the backend so that switching LLM providers (ollama -> llama.cpp) doesn't require code changes.
update_folder_for_conversation(conversation_id: str, new_folder_id: str | None, db_client, collection: str = 'chat_messages') -> int
¶
Update the folder_id metadata for all documents in a conversation.
Retrieves existing documents from ChromaDB, updates their
folder_id metadata, and re-upserts them with the same
embeddings.
Parameters¶
conversation_id :
The conversation whose documents should be updated.
new_folder_id :
New folder ID (None becomes "_none_").
db_client :
ChromaDB client instance.
collection :
ChromaDB collection name.
Returns¶
int Number of documents updated.
upsert_message_turn(db_client, message_id, conversation_id, folder_id, role, content, collection='chat_messages', embedding_model='qwen3-embedding:0.6b')
¶
Embed a single message turn into the chat_messages collection.
Splits content into sentence-level chunks and upserts each with folder / conversation / message metadata so that folder-scoped RAG queries can retrieve relevant historical passages.
Sentences are embedded in batches of 64 to stay within Ollama's embedding model context window (default ~8192 tokens). A per-sentence length warning is emitted for sentences exceeding 1000 characters.
Returns¶
int Number of sentence chunks upserted (0 if content is empty).
Reranker¶
core.data.reranker
¶
Reranker
¶
Seq-cls CrossEncoder reranker for: tomaarsen/Qwen3-Reranker-0.6B-seq-cls Model card provides the required formatting template. (See HF model card "Updated Usage".)
chunk_document(queries, document)
¶
Chunks text into 32-line chunks or less. Returns a list of query/document pairs, where each document is a 32-line long piece of text or less.
Conversation Backup¶
core.data.conversation_backup
¶
Conversation backup, export, and import (Item #45 / Phase 7).
Provides a portable JSON format for exporting conversations, folders, settings, memories, attachments, tool grants, and audit summaries from the SQLite store. Supports optional content redaction for privacy and versioned imports with schema validation.
Design principles¶
- Local-first — exports are self-contained JSON files readable without the application.
- Versioned — each backup carries a schema version for forward compatibility.
- Redactable — sensitive content (message text, memory documents) can be replaced with placeholders before export.
- Verifiable — imports validate structure before writing anything.
Usage¶
from core.conversation_backup import BackupExporter, BackupImporter from core.sqlite_store import SQLiteStore store = SQLiteStore(path) exporter = BackupExporter(store) backup = await exporter.export(redact_content=True) backup.to_file("backup.json")
later ...¶
importer = BackupImporter(store) report = await importer.from_file("backup.json")
BackupExporter
¶
Reads from a SQLiteStore and produces a ConversationBackup.
Parameters¶
store : An initialized SQLiteStore instance (imported via duck typing). redact_content : If True, replace message.content and memory.document with placeholders. Structure is preserved; text is removed. include_memories : If False, skip vector memory records entirely. include_audit : If False, skip audit event summary (reduces file size).
export() -> ConversationBackup
async
¶
Gather all data from the store and return a backup.
BackupImporter
¶
Validates and imports a ConversationBackup into a SQLiteStore.
Parameters¶
store : An initialized SQLiteStore instance (imported via duck typing). skip_errors : If True, continue importing other sections on individual record errors. If False, abort the entire import on first error.
BackupMetadata
dataclass
¶
Immutable header for every backup file.
ConversationBackup
dataclass
¶
ImportErrorInfo
dataclass
¶
Single validation error during import.
ImportReport
dataclass
¶
Summary of an import operation.
create_backup_exporter(store: Any, *, redact_content: bool = False, include_memories: bool = True, include_audit: bool = True) -> BackupExporter
¶
Convenience factory for BackupExporter.
create_backup_importer(store: Any, *, skip_errors: bool = True) -> BackupImporter
¶
Convenience factory for BackupImporter.
Conversation Search¶
core.data.conversation_search
¶
Conversation search — Full-text search across SQLite conversations/messages (Item #47).
Provides keyword and phrase search over persisted conversation content using
SQLite LIKE queries (no FTS5 extension required). The module is a standalone
search engine that reads from the SQLiteStore duck-typed interface without
importing it directly, keeping the search layer decoupled.
Design principles¶
- Read-only — never modifies data; queries against existing messages/conversations.
- Duck-typed — accepts any store with the expected query methods.
- Structured output — frozen dataclasses for deterministic consumption.
- Extension-free — uses LIKE-based matching for portability across SQLite builds.
Usage¶
from core.conversation_search import ConversationSearch, create_conversation_search searcher = create_conversation_search(store) results = await searcher.search("how do I fix CUDA error") for r in results: ... print(f"[{r.score:.2f}] {r.content_preview} (conv={r.conversation_id})")
Term scoring: messages matching more query terms receive higher scores.
ConversationSearch
¶
Full-text conversation search using SQLite LIKE queries.
Parameters¶
store : Duck-typed persistence store (SQLiteStore or FakeLocalStore). db_path : Path to the SQLite database file (reserved for future FTS5 support).
document_count() -> int
async
¶
Return the total number of messages in the store.
search(query: str, *, filters: SearchFilters | None = None) -> list[SearchResult]
async
¶
Search conversation messages by keyword/phrase.
Parameters¶
query : Free-text search string. Each word is matched with a separate LIKE clause. Quoted phrases are matched as exact substrings. filters : Optional role/conversation/folder/time filters.
Returns¶
list[SearchResult] Results sorted by relevance (highest score first).
search_conversations(query: str, *, filters: SearchFilters | None = None) -> list[SearchResult]
async
¶
Search conversation titles and metadata.
Returns aggregated results per conversation (first matching message).
SearchFilters
dataclass
¶
Optional filters applied to a search query.
Parameters¶
role :
Only search messages with this role (user, assistant, etc.).
None searches all roles.
conversation_id :
Restrict results to a single conversation.
folder_id :
Restrict results to conversations within a folder.
since :
ISO-8601 lower bound for timestamps (inclusive).
until :
ISO-8601 upper bound for timestamps (exclusive).
limit :
Maximum number of results to return (default 50).
SearchResult
dataclass
¶
A single search hit.
Parameters¶
conversation_id :
The conversation this message belongs to.
message_id :
Unique identifier of the matching message.
role :
Message role (user, assistant, system, tool).
content_preview :
Snippet around the matching term (up to 200 chars).
full_content :
Full message text.
score :
Relevance score (higher = more relevant; based on term match count).
timestamp :
ISO-8601 timestamp of the message.
conversation_title :
Title of the conversation, if available.
folder_id :
Folder identifier, if set.
Memory Lifecycle¶
core.data.memory_lifecycle
¶
Memory lifecycle policies.
Defines how memories are created, updated, summarized, forgotten, tagged, searched, audited, exported, and deleted. Includes user-visible controls for memory scopes and per-agent/persona usage.
ForgetResult
dataclass
¶
Outcome of a forget operation across both stores.
Parameters¶
sqlite_deleted :
Number of SQLite rows removed (messages + embedding index).
chroma_deleted :
Number of ChromaDB documents removed.
success :
True if at least one store reports deletions or both
are empty for the requested resource.
MemoryAction
¶
Bases: Enum
Memory lifecycle actions.
MemoryEntry
dataclass
¶
A single memory entry.
Parameters¶
id : Unique identifier content : Memory content scope : Memory scope tags : Tags for organization created_at : ISO-8601 creation timestamp updated_at : ISO-8601 last update timestamp consent_given : Whether user consented to this memory metadata : Arbitrary additional context
MemoryLifecycleManager
¶
Orchestrates dual-store delete across SQLite and ChromaDB.
When a user "forgets" something it must be removed from both stores so it cannot be retrieved again. This manager coordinates the two deletion paths and reports how many rows/documents were affected.
Parameters¶
store : SQLiteStore instance (messages, embedding index). db_client : ChromaDB persistent client (vector embeddings).
forget_conversation(conversation_id: str) -> ForgetResult
async
¶
Delete all messages and embeddings for a conversation.
Steps: 1. Collect all message IDs and their content (for ChromaDB ID reconstruction). 2. Delete ChromaDB documents by metadata filter (conversation_id) and/or by IDs. 3. Cascade-delete messages, embedding index, and the conversation from SQLite.
Parameters¶
conversation_id : Conversation to forget entirely.
Returns¶
ForgetResult Counts of deleted resources.
forget_message(conversation_id: str, message_id: str) -> ForgetResult
async
¶
Delete one message from both SQLite and ChromaDB.
Steps: 1. Fetch the message row to obtain content (for ChromaDB ID reconstruction). 2. Delete embedding-index entries for this message. 3. Delete corresponding ChromaDB sentence-chunk documents. 4. Delete the message row from SQLite.
Parameters¶
conversation_id : Parent conversation. message_id : Message to forget.
Returns¶
ForgetResult Counts of deleted rows/documents.
MemoryManager
¶
Manages memory lifecycle.
Parameters¶
policy : Memory policy
add(entry: MemoryEntry) -> MemoryEntry
¶
Add a memory entry.
count() -> int
¶
Count total entries.
delete(entry_id: str) -> bool
¶
Delete a memory entry.
get(entry_id: str) -> MemoryEntry | None
¶
Get a memory entry.
list_entries(scope: MemoryScope | None = None, tags: list[str] | None = None) -> list[MemoryEntry]
¶
List memory entries with optional filtering.
MemoryPolicy
dataclass
¶
Policy for memory lifecycle.
Parameters¶
max_age_days : Maximum age before auto-summary max_entries : Maximum number of entries per scope auto_summarize : Whether to auto-summarize old memories require_consent : Whether user consent is required for creation allow_global : Whether global-scope memories are allowed retention_days : Days to retain memories before cleanup
MemoryScope
¶
Bases: Enum
Memory scoping levels.