Technical Specification: SQLite Database Backend for Honcho

Author: Honcho Engineering Date: 2026-03-25 Status: Draft Scope: Add SQLite as an alternative database backend for local development/testing, architected for future production-grade extension.

Problem Statement
Goals and Non-Goals
Design
- 3.1 DatabaseDialect Protocol
- 3.2 Dialect Registry and Auto-Detection
- 3.3 Model Layer Changes
- 3.4 sqlite-vec Integration
- 3.5 FTS5 Full-Text Search
- 3.6 JSON Handling
- 3.7 Locking Strategy
- 3.8 Partial Unique Indexes and Constraints
- 3.9 PostgreSQL-Specific SQL Patterns
- 3.10 Connection and Engine Configuration
- 3.11 Alembic Migration Strategy
Implementation Phases
Files to Modify
Risk Assessment
Verification Plan
Open Questions

1. Problem Statement

Honcho currently requires a PostgreSQL database with pgvector for all operation, including local development and testing. This creates friction for contributors and local experimentation:

Developers must run a PostgreSQL server (often via Docker) just to start Honcho locally.
pgvector requires a PostgreSQL extension that is not always trivially available.
Integration tests require a running PostgreSQL instance.
Quick prototyping and demos are heavyweight.

The codebase is tightly coupled to PostgreSQL through direct use of JSONB, Vector(1536), GIN indexes, HNSW indexes, to_tsvector() / ts_rank(), pg_advisory_xact_lock(), ::jsonb casts, the ~ regex operator in check constraints, postgresql_using / postgresql_where / postgresql_ops / postgresql_with / postgresql_include index parameters, FOR UPDATE SKIP LOCKED, the || JSONB merge operator, and sqlalchemy.dialects.postgresql.insert with on_conflict_do_nothing / on_conflict_do_update.

2. Goals and Non-Goals

Goals

G1: Allow Honcho to start and operate against a SQLite database by setting DB_CONNECTION_URI=sqlite+aiosqlite:///path/to/honcho.db.
G2: Preserve 100% backward compatibility with existing PostgreSQL deployments; the default remains PostgreSQL.
G3: Use sqlite-vec for local vector similarity search (cosine distance on BLOB-stored float32 vectors).
G4: Use FTS5 virtual tables for full-text search on messages.
G5: Replace pg_advisory_xact_lock() with an application-level asyncio.Lock keyed by (workspace_name, session_name).
G6: Conditional Alembic migrations that emit correct DDL per dialect.
G7: Architect the abstraction layer so that future database backends (e.g., MySQL, CockroachDB) can be added by implementing the dialect protocol.
G8: All existing tests should pass against both backends (parameterized by backend in CI).

Non-Goals

NG1: Production-grade SQLite deployment (WAL tuning, backup strategies, replication). This spec targets local dev/testing.
NG2: CLI integration. The CLI spec is separate; this spec covers only the library/server layer.
NG3: Migrating existing PostgreSQL data to SQLite.
NG4: Supporting SQLite-specific features not needed for Honcho (e.g., R-tree indexes, custom collations).
NG5: Horizontal scaling or multi-process write concurrency on SQLite.
NG6: External vector store support (Turbopuffer, LanceDB) when running in SQLite mode. SQLite mode uses sqlite-vec exclusively for vectors.

3. Design

3.1 DatabaseDialect Protocol

A new module src/dialect.py introduces a DatabaseDialect protocol that abstracts all database-engine-specific behavior. Every PostgreSQL-specific operation in the codebase is routed through this protocol.

# src/dialect.py
"""
Database dialect abstraction layer.
 
Provides a protocol and concrete implementations for PostgreSQL and SQLite,
allowing the rest of the codebase to remain database-agnostic.
"""
 
from __future__ import annotations
 
import asyncio
import hashlib
from typing import Any, Protocol, runtime_checkable
 
from sqlalchemy import DDL, Index, Table, Text, event, text
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.types import TypeEngine
 
 
@runtime_checkable
class DatabaseDialect(Protocol):
    """Protocol defining database-engine-specific behavior."""
 
    @property
    def name(self) -> str:
        """Return dialect name: 'postgresql' or 'sqlite'."""
        ...
 
    # --- Column Types ---
 
    def get_vector_column_type(self, dimensions: int) -> TypeEngine:
        """Return the column type for storing embedding vectors."""
        ...
 
    def get_json_column_type(self) -> TypeEngine:
        """Return the column type for storing JSON/JSONB data."""
        ...
 
    def get_text_column_type(self) -> TypeEngine:
        """Return the column type for TEXT columns."""
        ...
 
    # --- Index Creation ---
 
    def create_vector_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        """Return DDL for creating a vector similarity index, or None if not supported."""
        ...
 
    def create_fts_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        """Return DDL for creating a full-text search index, or None if not supported."""
        ...
 
    def create_gin_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        """Return DDL for creating a GIN index on a JSON column, or None if not supported."""
        ...
 
    # --- Locking ---
 
    async def acquire_session_lock(
        self, db: AsyncSession, workspace_name: str, session_name: str
    ) -> None:
        """Acquire a serialization lock for writes to a specific session."""
        ...
 
    async def release_session_lock(
        self, db: AsyncSession, workspace_name: str, session_name: str
    ) -> None:
        """Release a session lock. No-op for transaction-scoped locks (PostgreSQL advisory)."""
        ...
 
    # --- Dialect-Specific SQL Helpers ---
 
    def json_merge_expression(self, column: Any, update_data: Any) -> Any:
        """Return an expression for merging JSON data into a column.
        PostgreSQL: column || :update_data (JSONB merge)
        SQLite: json_patch(column, :update_data)
        """
        ...
 
    def json_contains_expression(self, column: Any, value: Any) -> Any:
        """Return an expression for checking if a JSON column contains a value.
        PostgreSQL: column @> :value (JSONB containment)
        SQLite: json_extract based comparison
        """
        ...
 
    def json_field_accessor(self, column: Any, field_name: str) -> Any:
        """Return an expression for accessing a field within a JSON column.
        PostgreSQL: column->>field_name
        SQLite: json_extract(column, '$.field_name')
        """
        ...
 
    def regex_check_constraint(self, column_name: str, pattern: str) -> str:
        """Return a CHECK constraint expression for regex validation.
        PostgreSQL: column ~ 'pattern'
        SQLite: fallback to length-only or glob-based check
        """
        ...
 
    def cosine_distance_expression(self, column: Any, query_vector: Any) -> Any:
        """Return an ORDER BY expression for cosine distance.
        PostgreSQL: column <=> query_vector (pgvector)
        SQLite: vec_distance_cosine(column, query_vector) (sqlite-vec)
        """
        ...
 
    def fts_search_expression(self, table_name: str, query: str) -> Any:
        """Return a WHERE clause for full-text search.
        PostgreSQL: to_tsvector('english', content) @@ plainto_tsquery('english', query)
        SQLite: FTS5 MATCH via shadow table
        """
        ...
 
    def fts_rank_expression(self, table_name: str, query: str) -> Any:
        """Return an ORDER BY expression for full-text search ranking.
        PostgreSQL: ts_rank(to_tsvector('english', content), plainto_tsquery('english', query))
        SQLite: rank from FTS5 (bm25)
        """
        ...
 
    def supports_partial_unique_index(self) -> bool:
        """Whether the dialect supports partial unique indexes (WHERE clause on CREATE UNIQUE INDEX)."""
        ...
 
    def supports_for_update_skip_locked(self) -> bool:
        """Whether the dialect supports SELECT ... FOR UPDATE SKIP LOCKED."""
        ...
 
    def server_default_json_empty_object(self) -> str:
        """Return the server default expression for an empty JSON object.
        PostgreSQL: '{}'::jsonb
        SQLite: '{}'
        """
        ...
 
    def server_default_json_null(self) -> str:
        """Return the server default expression for a JSON NULL.
        PostgreSQL: NULL
        SQLite: NULL
        """
        ...
 
    def register_engine_events(self, engine: Engine) -> None:
        """Register dialect-specific engine-level events (e.g., loading extensions)."""
        ...
 
    # --- Upsert ---
 
    def upsert_statement(
        self, table: Any, values: list[dict[str, Any]], index_elements: list[str],
        set_: dict[str, Any]
    ) -> Any:
        """Return a dialect-specific INSERT ... ON CONFLICT DO UPDATE statement.
        PostgreSQL: sqlalchemy.dialects.postgresql.insert().on_conflict_do_update()
        SQLite: sqlalchemy.dialects.sqlite.insert().on_conflict_do_update()
        """
        ...
 
    def insert_or_ignore_statement(self, table: Any, values: list[dict[str, Any]]) -> Any:
        """Return a dialect-specific INSERT ... ON CONFLICT DO NOTHING statement.
        PostgreSQL: sqlalchemy.dialects.postgresql.insert().on_conflict_do_nothing()
        SQLite: sqlalchemy.dialects.sqlite.insert().on_conflict_do_nothing()
        """
        ...

PostgreSQL Implementation

# src/dialect_postgresql.py
"""PostgreSQL dialect implementation."""
 
from __future__ import annotations
 
import hashlib
from typing import Any
 
from pgvector.sqlalchemy import Vector
from sqlalchemy import DDL, Index, func, text
from sqlalchemy.dialects.postgresql import JSONB, TEXT, insert as pg_insert
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.types import TypeEngine
 
 
class PostgreSQLDialect:
    """PostgreSQL-specific database operations."""
 
    @property
    def name(self) -> str:
        return "postgresql"
 
    def get_vector_column_type(self, dimensions: int) -> TypeEngine:
        return Vector(dimensions)
 
    def get_json_column_type(self) -> TypeEngine:
        return JSONB
 
    def get_text_column_type(self) -> TypeEngine:
        return TEXT()
 
    def create_vector_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        return DDL(
            f"CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} "
            f"USING hnsw ({column_name} vector_cosine_ops) "
            f"WITH (m = 16, ef_construction = 64)"
        )
 
    def create_fts_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        return DDL(
            f"CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} "
            f"USING gin (to_tsvector('english', {column_name}))"
        )
 
    def create_gin_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        return DDL(
            f"CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} "
            f"USING gin ({column_name})"
        )
 
    async def acquire_session_lock(
        self, db: AsyncSession, workspace_name: str, session_name: str
    ) -> None:
        await db.execute(text("SET LOCAL lock_timeout = '5s'"))
        await db.execute(
            text(
                "SELECT pg_advisory_xact_lock(hashtext(:workspace_name), hashtext(:session_name))"
            ),
            {"workspace_name": workspace_name, "session_name": session_name},
        )
 
    async def release_session_lock(
        self, db: AsyncSession, workspace_name: str, session_name: str
    ) -> None:
        # Advisory transaction locks are released automatically on commit/rollback
        pass
 
    def json_merge_expression(self, column: Any, update_data: Any) -> Any:
        return column.op("||")(update_data)
 
    def json_contains_expression(self, column: Any, value: Any) -> Any:
        return column.contains(value)
 
    def json_field_accessor(self, column: Any, field_name: str) -> Any:
        return column[field_name].astext
 
    def regex_check_constraint(self, column_name: str, pattern: str) -> str:
        return f"{column_name} ~ '{pattern}'"
 
    def cosine_distance_expression(self, column: Any, query_vector: Any) -> Any:
        return column.cosine_distance(query_vector)
 
    def fts_search_expression(self, table_name: str, query: str) -> Any:
        from src import models
        content_col = getattr(models, table_name.capitalize(), models.Message).content
        return func.to_tsvector("english", content_col).op("@@")(
            func.plainto_tsquery("english", query)
        )
 
    def fts_rank_expression(self, table_name: str, query: str) -> Any:
        from src import models
        content_col = getattr(models, table_name.capitalize(), models.Message).content
        return func.ts_rank(
            func.to_tsvector("english", content_col),
            func.plainto_tsquery("english", query),
        )
 
    def supports_partial_unique_index(self) -> bool:
        return True
 
    def supports_for_update_skip_locked(self) -> bool:
        return True
 
    def server_default_json_empty_object(self) -> str:
        return "'{}'::jsonb"
 
    def server_default_json_null(self) -> str:
        return "NULL"
 
    def register_engine_events(self, engine: Engine) -> None:
        # No special events needed for PostgreSQL; pgvector extension is loaded via migrations
        pass
 
    def upsert_statement(
        self, table: Any, values: list[dict[str, Any]], index_elements: list[str],
        set_: dict[str, Any]
    ) -> Any:
        stmt = pg_insert(table).values(values)
        return stmt.on_conflict_do_update(index_elements=index_elements, set_=set_)
 
    def insert_or_ignore_statement(self, table: Any, values: list[dict[str, Any]]) -> Any:
        stmt = pg_insert(table).values(values)
        return stmt.on_conflict_do_nothing()

SQLite Implementation

# src/dialect_sqlite.py
"""SQLite dialect implementation."""
 
from __future__ import annotations
 
import asyncio
import hashlib
import json
from typing import Any
 
from sqlalchemy import DDL, Text, event, func, literal_column, text
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.types import LargeBinary, TypeDecorator, TypeEngine
 
 
class VectorBLOB(TypeDecorator):
    """Custom type for storing float32 vectors as BLOBs for sqlite-vec.
 
    sqlite-vec expects vectors as raw bytes (little-endian float32 arrays).
    This type handles Python list[float] <-> bytes conversion transparently.
    """
    impl = LargeBinary
    cache_ok = True
 
    def __init__(self, dimensions: int):
        super().__init__()
        self.dimensions = dimensions
 
    def process_bind_param(self, value: list[float] | bytes | None, dialect: Any) -> bytes | None:
        if value is None:
            return None
        if isinstance(value, bytes):
            return value
        import struct
        return struct.pack(f"<{len(value)}f", *value)
 
    def process_result_value(self, value: bytes | None, dialect: Any) -> list[float] | None:
        if value is None:
            return None
        import struct
        count = len(value) // 4
        return list(struct.unpack(f"<{count}f", value))
 
 
# Application-level lock registry for session serialization.
# Keyed by (workspace_name, session_name) to match pg_advisory_xact_lock semantics.
_session_locks: dict[tuple[str, str], asyncio.Lock] = {}
_session_locks_guard = asyncio.Lock()
 
 
async def _get_session_lock(workspace_name: str, session_name: str) -> asyncio.Lock:
    """Get or create an asyncio.Lock for a (workspace, session) pair."""
    key = (workspace_name, session_name)
    if key not in _session_locks:
        async with _session_locks_guard:
            if key not in _session_locks:
                _session_locks[key] = asyncio.Lock()
    return _session_locks[key]
 
 
class SQLiteDialect:
    """SQLite-specific database operations."""
 
    @property
    def name(self) -> str:
        return "sqlite"
 
    def get_vector_column_type(self, dimensions: int) -> TypeEngine:
        return VectorBLOB(dimensions)
 
    def get_json_column_type(self) -> TypeEngine:
        # SQLite stores JSON as TEXT; json1 extension provides json_extract(), etc.
        return Text()
 
    def get_text_column_type(self) -> TypeEngine:
        return Text()
 
    def create_vector_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        # sqlite-vec uses virtual tables for indexing, not standard CREATE INDEX.
        # Vector search is done via the vec0 virtual table (see Section 3.4).
        # Return None; virtual table creation is handled at init_db() time.
        return None
 
    def create_fts_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        # FTS5 virtual table creation is handled at init_db() time (see Section 3.5).
        return None
 
    def create_gin_index(
        self, table_name: str, column_name: str, index_name: str
    ) -> DDL | None:
        # SQLite has no GIN indexes. For source_ids containment queries,
        # we use json_each() at query time. No index needed for dev workloads.
        return None
 
    async def acquire_session_lock(
        self, db: AsyncSession, workspace_name: str, session_name: str
    ) -> None:
        lock = await _get_session_lock(workspace_name, session_name)
        await lock.acquire()
 
    async def release_session_lock(
        self, db: AsyncSession, workspace_name: str, session_name: str
    ) -> None:
        lock = await _get_session_lock(workspace_name, session_name)
        if lock.locked():
            lock.release()
 
    def json_merge_expression(self, column: Any, update_data: Any) -> Any:
        # json_patch(base, patch) merges top-level keys
        return func.json_patch(column, update_data)
 
    def json_contains_expression(self, column: Any, value: Any) -> Any:
        # For simple key-value containment: json_extract(column, '$.key') = value
        # This is handled at the filter layer, not as a single expression.
        # For JSONB @> operator equivalent, we decompose at the call site.
        raise NotImplementedError(
            "json_contains_expression must be decomposed into json_extract calls "
            "at the call site for SQLite. See _build_sqlite_json_contains()."
        )
 
    def json_field_accessor(self, column: Any, field_name: str) -> Any:
        return func.json_extract(column, f"$.{field_name}")
 
    def regex_check_constraint(self, column_name: str, pattern: str) -> str:
        # SQLite has no native regex. For nanoid format validation
        # ('^[A-Za-z0-9_-]+$'), use GLOB which supports character classes.
        # Convert regex '^[A-Za-z0-9_-]+$' to GLOB equivalent.
        if pattern == "^[A-Za-z0-9_-]+$":
            # GLOB is case-sensitive and supports [chars] ranges.
            # '*' matches any sequence. There is no '+' equivalent, but
            # length checks are already separate constraints.
            return f"({column_name} GLOB '*[A-Za-z0-9_-]*' AND length({column_name}) > 0)"
        # Fallback: skip regex constraint on SQLite (validation happens in Python layer)
        return "1=1"
 
    def cosine_distance_expression(self, column: Any, query_vector: Any) -> Any:
        # sqlite-vec: vec_distance_cosine(vector_blob, query_blob)
        return func.vec_distance_cosine(column, query_vector)
 
    def fts_search_expression(self, table_name: str, query: str) -> Any:
        # Use the FTS5 shadow table for matching.
        # Returns a subquery: rowid IN (SELECT rowid FROM messages_fts WHERE messages_fts MATCH :query)
        # This is constructed at the call site using the FTS5 table name.
        raise NotImplementedError(
            "FTS search expressions for SQLite must be constructed at the call site "
            "using the FTS5 shadow table. See Section 3.5."
        )
 
    def fts_rank_expression(self, table_name: str, query: str) -> Any:
        # FTS5 rank is accessed via the rank column of the FTS5 table.
        raise NotImplementedError(
            "FTS rank expressions for SQLite must be constructed at the call site. "
            "See Section 3.5."
        )
 
    def supports_partial_unique_index(self) -> bool:
        # SQLite supports partial indexes via CREATE UNIQUE INDEX ... WHERE ...
        # since version 3.8.0 (2013). However, SQLAlchemy does not emit
        # sqlite_where on Index(). We create these via raw DDL.
        return True  # Supported, but needs raw DDL
 
    def supports_for_update_skip_locked(self) -> bool:
        return False
 
    def server_default_json_empty_object(self) -> str:
        return "'{}'"
 
    def server_default_json_null(self) -> str:
        return "NULL"
 
    def register_engine_events(self, engine: Engine) -> None:
        """Load sqlite-vec and enable WAL mode on every new connection."""
 
        @event.listens_for(engine.sync_engine, "connect")
        def _on_connect(dbapi_conn: Any, connection_record: Any) -> None:
            # Enable WAL mode for better concurrent read performance
            dbapi_conn.execute("PRAGMA journal_mode=WAL")
            dbapi_conn.execute("PRAGMA foreign_keys=ON")
            dbapi_conn.execute("PRAGMA busy_timeout=5000")
 
            # Load sqlite-vec extension
            dbapi_conn.enable_load_extension(True)
            import sqlite_vec
            sqlite_vec.load(dbapi_conn)
            dbapi_conn.enable_load_extension(False)
 
    def upsert_statement(
        self, table: Any, values: list[dict[str, Any]], index_elements: list[str],
        set_: dict[str, Any]
    ) -> Any:
        stmt = sqlite_insert(table).values(values)
        return stmt.on_conflict_do_update(index_elements=index_elements, set_=set_)
 
    def insert_or_ignore_statement(self, table: Any, values: list[dict[str, Any]]) -> Any:
        stmt = sqlite_insert(table).values(values)
        return stmt.on_conflict_do_nothing()

3.2 Dialect Registry and Auto-Detection

A singleton registry in src/dialect.py selects the dialect based on the DB_CONNECTION_URI scheme:

# Appended to src/dialect.py
 
from functools import cache
from src.config import settings
 
 
@cache
def get_dialect() -> DatabaseDialect:
    """
    Return the appropriate DatabaseDialect based on DB_CONNECTION_URI.
 
    Detection rules:
    - URI starts with "sqlite" -> SQLiteDialect
    - URI starts with "postgresql" -> PostgreSQLDialect
    - Otherwise -> raise ValueError
 
    This function is cached (singleton); the dialect never changes at runtime.
    """
    uri = settings.DB.CONNECTION_URI
    if uri.startswith("sqlite"):
        from src.dialect_sqlite import SQLiteDialect
        return SQLiteDialect()
    elif uri.startswith("postgresql"):
        from src.dialect_postgresql import PostgreSQLDialect
        return PostgreSQLDialect()
    else:
        raise ValueError(
            f"Unsupported database URI scheme: {uri.split('://')[0]}. "
            "Honcho supports 'postgresql+psycopg://...' and 'sqlite+aiosqlite:///...'."
        )
 
 
def is_sqlite() -> bool:
    """Convenience check for conditional logic."""
    return get_dialect().name == "sqlite"
 
 
def is_postgresql() -> bool:
    """Convenience check for conditional logic."""
    return get_dialect().name == "postgresql"

3.3 Model Layer Changes

The current src/models.py imports JSONB, TEXT, and Vector directly from PostgreSQL-specific modules. These must be replaced with dialect-aware types.

Strategy: Conditional Column Types via Module-Level Helpers

Rather than making models dynamically generated (which would break type checking and IDE support), we use a thin indirection layer at module import time:

# At the top of src/models.py, replace:
#   from pgvector.sqlalchemy import Vector
#   from sqlalchemy.dialects.postgresql import JSONB, TEXT
# with:
 
from src.dialect import get_dialect
 
_dialect = get_dialect()
 
# Column type aliases resolved at import time
_JSONB = _dialect.get_json_column_type()    # JSONB on PG, Text on SQLite
_TEXT = _dialect.get_text_column_type()       # TEXT on both
_VECTOR = lambda dims: _dialect.get_vector_column_type(dims)  # Vector(n) on PG, VectorBLOB(n) on SQLite

Then throughout models.py, replace:

JSONB with _JSONB
TEXT with _TEXT
Vector(1536) with _VECTOR(1536)
server_default=text("'{}'::jsonb") with server_default=text(_dialect.server_default_json_empty_object())

Check Constraints with Regex

The PostgreSQL ~ regex operator is used in 6 check constraints for nanoid format validation. Replace with dialect-aware expressions:

# Before:
CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'", name="id_format")
 
# After:
CheckConstraint(
    _dialect.regex_check_constraint("id", "^[A-Za-z0-9_-]+$"),
    name="id_format",
)

On SQLite, this emits a GLOB-based check that validates the character set without regex.

PostgreSQL-Specific Index Parameters

Indexes using postgresql_using, postgresql_with, postgresql_ops, postgresql_where, and postgresql_include are silently ignored by SQLAlchemy when running against a non-PostgreSQL dialect. This means the existing Index(...) declarations in __table_args__ are safe as-is for SQLite — SQLAlchemy will create a standard B-tree index or skip unsupported parameters. However, this means:

HNSW vector indexes — No equivalent on SQLite. Vector search uses sqlite-vec virtual tables instead (see Section 3.4).
GIN indexes on JSONB — No equivalent on SQLite. JSON queries use json_extract() scans, which is acceptable for dev workloads.
GIN FTS index — Replaced by FTS5 virtual table (see Section 3.5).
Partial unique indexes — SQLite supports CREATE UNIQUE INDEX ... WHERE ... natively since 3.8.0, but SQLAlchemy does not emit the sqlite_where parameter. These must be created via raw DDL in migrations or init_db().
postgresql_include (covering index) — Silently ignored on SQLite. No functional impact.

Action: Keep the existing Index(...) declarations in __table_args__ unchanged. They are PostgreSQL-only by definition (the postgresql_* kwargs are ignored on other dialects). Supplementary SQLite indexes/virtual tables are created separately.

3.4 sqlite-vec Integration

sqlite-vec provides vector similarity search as a SQLite extension. It stores vectors in virtual tables and supports cosine distance queries.

Extension Loading

The extension is loaded on every connection via the SQLiteDialect.register_engine_events() method (see Section 3.1). The sqlite-vec Python package (pip install sqlite-vec) provides a load() function that handles finding and loading the shared library.

Virtual Table Schema

For each table with vector columns (documents, message_embeddings), a companion vec0 virtual table is created:

-- Shadow table for document embeddings
CREATE VIRTUAL TABLE IF NOT EXISTS documents_vec USING vec0(
    id TEXT PRIMARY KEY,
    embedding float[1536]
);
 
-- Shadow table for message embeddings
CREATE VIRTUAL TABLE IF NOT EXISTS message_embeddings_vec USING vec0(
    id TEXT PRIMARY KEY,
    embedding float[1536]
);

These are created during init_db() for SQLite mode, not via Alembic migrations (virtual tables are not well-supported by Alembic’s autogenerate).

Sync Between Main Table and Virtual Table

When a document or message embedding is created/updated:

The main table row is inserted with the embedding column stored as a BLOB (via VectorBLOB type).
After commit, a separate INSERT/REPLACE into the *_vec virtual table is performed.

This mirrors the existing pattern where pgvector stores embeddings in the main table column and the HNSW index is automatically maintained.

Helper function:

# src/dialect_sqlite.py (additional helper)
 
async def sync_vector_to_virtual_table(
    db: AsyncSession,
    virtual_table: str,  # "documents_vec" or "message_embeddings_vec"
    record_id: str | int,
    embedding: list[float],
) -> None:
    """Insert or replace a vector in the sqlite-vec virtual table."""
    import struct
    blob = struct.pack(f"<{len(embedding)}f", *embedding)
    await db.execute(
        text(f"INSERT OR REPLACE INTO {virtual_table}(id, embedding) VALUES (:id, :embedding)"),
        {"id": str(record_id), "embedding": blob},
    )

Query Pattern

# Cosine similarity search for documents (SQLite mode)
# In src/crud/document.py::query_documents(), the pgvector path:
#   .order_by(models.Document.embedding.cosine_distance(embedding))
# becomes (for SQLite):
 
from sqlalchemy import literal_column
 
vec_blob = struct.pack(f"<{len(embedding)}f", *embedding)
 
# Subquery: get IDs ordered by distance from the vec0 virtual table
vec_subquery = (
    select(
        literal_column("id").label("vec_id"),
        literal_column("distance").label("vec_distance"),
    )
    .select_from(text("documents_vec"))
    .where(text("embedding MATCH :query_vec"))
    .params(query_vec=vec_blob)
    .order_by(literal_column("distance"))
    .limit(top_k)
    .subquery()
)
 
# Join with main documents table
stmt = (
    select(models.Document)
    .join(vec_subquery, models.Document.id == vec_subquery.c.vec_id)
    .where(models.Document.deleted_at.is_(None))
    .where(models.Document.workspace_name == workspace_name)
    .order_by(vec_subquery.c.vec_distance)
)

Implementation note: The sqlite-vec MATCH operator accepts a raw float32 BLOB. The VectorBLOB type handles serialization.

3.5 FTS5 Full-Text Search

PostgreSQL uses to_tsvector('english', content) with GIN indexes and ts_rank() for full-text search. SQLite replaces this with FTS5.

FTS5 Virtual Table

-- Created during init_db() for SQLite mode
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
    content,
    content_rowid='id',  -- links to messages.id (BIGINT primary key)
    tokenize='porter unicode61'
);

The content column of messages_fts mirrors messages.content. Since messages.id is a BIGINT Identity() primary key, it serves as the content_rowid.

Keeping FTS5 in Sync

FTS5 content tables can be configured as “external content” tables that reference the main table, but this requires triggers. For simplicity and reliability, we use explicit sync:

-- Triggers created during init_db() for SQLite mode
CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages
BEGIN
    INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
 
CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages
BEGIN
    INSERT INTO messages_fts(messages_fts, rowid, content) VALUES ('delete', old.id, old.content);
END;
 
CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE OF content ON messages
BEGIN
    INSERT INTO messages_fts(messages_fts, rowid, content) VALUES ('delete', old.id, old.content);
    INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;

Query Pattern

The existing _fulltext_search() function in src/utils/search.py currently uses:

fts_condition = func.to_tsvector("english", models.Message.content).op("@@")(
    func.plainto_tsquery("english", query)
)

For SQLite, this becomes:

if is_sqlite():
    # FTS5 MATCH via a subquery on the virtual table
    fts_subquery = (
        select(literal_column("rowid").label("fts_rowid"))
        .select_from(text("messages_fts"))
        .where(text("messages_fts MATCH :fts_query"))
        .params(fts_query=query)
        .subquery()
    )
    fts_condition = models.Message.id.in_(
        select(fts_subquery.c.fts_rowid)
    )
    # Ranking: join with messages_fts to get bm25 rank
    # FTS5 provides a built-in rank column when queried directly
else:
    # Existing PostgreSQL FTS logic (unchanged)
    ...

For ranking on SQLite, FTS5 provides a rank column when you query the FTS table directly. The implementation should join messages_fts and order by rank.

3.6 JSON Handling

Column Type

PostgreSQL JSONB becomes SQLite TEXT. SQLite’s json1 extension (built-in since 3.38.0, always available in modern Python sqlite3) provides json_extract(), json_patch(), json_each(), json_type(), etc.

Server Defaults

PostgreSQL	SQLite
`server_default=text("'{}'::jsonb")`	`server_default=text("'{}'")`
`server_default=text("NULL")`	`server_default=text("NULL")`

JSONB Containment (`@>`)

The @> operator is used extensively in src/utils/filter.py via column.contains(value). On SQLite, this must be decomposed:

# src/utils/filter.py adaptation for SQLite
# For column.contains({"key": "value"}):
 
if is_sqlite():
    conditions = []
    for k, v in value.items():
        conditions.append(
            func.json_extract(column, f"$.{k}") == json.dumps(v) if isinstance(v, (dict, list))
            else func.json_extract(column, f"$.{k}") == v
        )
    return and_(*conditions)
else:
    return column.contains(value)

JSONB Array Containment (`source_ids @> [parent_id]`)

Used in src/crud/document.py::get_child_observations():

models.Document.source_ids.contains([parent_id])

On SQLite, replace with a json_each() subquery:

if is_sqlite():
    # Check if parent_id exists in the JSON array source_ids
    json_each_subq = (
        select(literal_column("value"))
        .select_from(text(f"json_each({models.Document.__tablename__}.source_ids)"))
        .where(literal_column("value") == parent_id)
        .exists()
    )
    stmt = stmt.where(json_each_subq)
else:
    stmt = stmt.where(models.Document.source_ids.contains([parent_id]))

JSONB Merge (`||` operator)

Used in src/utils/summarizer.py, src/crud/collection.py, and src/crud/peer_card.py:

column.op("||")(update_data)

On SQLite, replace with json_patch():

dialect = get_dialect()
merged = dialect.json_merge_expression(column, func.json(json.dumps(update_data)))

The json_patch(base, patch) function performs RFC 7396 JSON Merge Patch, which matches the || shallow-merge semantics for the top-level keys used in Honcho.

JSONB Field Access in Filters (`column->>field_name`)

The filter system accesses JSONB fields via column[field_name].astext. On SQLite:

if is_sqlite():
    field_accessor = func.json_extract(column, f"$.{field_name}")
else:
    field_accessor = column[field_name].astext

3.7 Locking Strategy

Problem

Honcho uses two PostgreSQL-specific locking mechanisms:

pg_advisory_xact_lock() in src/crud/message.py::create_messages() to serialize message writes per session.
SELECT ... FOR UPDATE SKIP LOCKED in src/deriver/queue_manager.py, src/reconciler/sync_vectors.py, and src/crud/document.py::cleanup_soft_deleted_documents() for concurrent queue/reconciliation processing.

SQLite does not support either mechanism.

Solution: Advisory Locks → asyncio.Lock

For pg_advisory_xact_lock(), the SQLite dialect uses an in-process asyncio.Lock keyed by (workspace_name, session_name):

# In src/crud/message.py::create_messages()
 
dialect = get_dialect()
await dialect.acquire_session_lock(db, workspace_name, session_name)
try:
    # ... create messages, compute seq_in_session ...
    await db.commit()
finally:
    await dialect.release_session_lock(db, workspace_name, session_name)

Key differences from PostgreSQL advisory locks:

Property	PostgreSQL Advisory Lock	asyncio.Lock
Scope	Transaction (auto-released on commit/rollback)	Explicit acquire/release
Cross-process	Yes (database-level)	No (single process only)
Deadlock detection	Yes (PostgreSQL detects)	No (must be careful with ordering)
Performance	Minimal overhead	Minimal overhead

Acceptable tradeoff: For local dev/testing (single process), asyncio.Lock provides equivalent serialization. Production SQLite deployments (a non-goal) would need a different approach.

Solution: FOR UPDATE SKIP LOCKED → Sequential Processing or Application-Level Locking

For SELECT ... FOR UPDATE SKIP LOCKED, the SQLite dialect uses two strategies depending on context:

Strategy A: Skip the clause entirely (queue manager, reconciler)

Since SQLite mode targets single-process dev environments, there is no concurrent worker contention. The with_for_update(skip_locked=True) calls are conditionally skipped:

stmt = select(models.Document).where(...)
 
if dialect.supports_for_update_skip_locked():
    stmt = stmt.with_for_update(skip_locked=True)
# On SQLite, just run without FOR UPDATE -- single process means no contention

Strategy B: Application-level deduplication (document cleanup)

The cleanup_soft_deleted_documents() function uses FOR UPDATE SKIP LOCKED to prevent multiple deriver instances from processing the same documents. In SQLite mode (single process), this is unnecessary. The conditional skip is sufficient.

3.8 Partial Unique Indexes and Constraints

The queue table uses two partial unique indexes for task deduplication:

Index(
    "uq_queue_reconciler_pending_work_unit_key",
    "work_unit_key",
    unique=True,
    postgresql_where=text("task_type = 'reconciler' AND processed = false"),
)
Index(
    "uq_queue_dream_pending_work_unit_key",
    "work_unit_key",
    unique=True,
    postgresql_where=text("task_type = 'dream' AND processed = false"),
)

SQLite supports partial indexes with WHERE clauses, but SQLAlchemy only passes postgresql_where to the PostgreSQL dialect. For SQLite, these indexes must be created via raw DDL.

Implementation: In init_db() for SQLite mode, after Base.metadata.create_all():

CREATE UNIQUE INDEX IF NOT EXISTS uq_queue_reconciler_pending_work_unit_key
ON queue(work_unit_key) WHERE task_type = 'reconciler' AND processed = 0;
 
CREATE UNIQUE INDEX IF NOT EXISTS uq_queue_dream_pending_work_unit_key
ON queue(work_unit_key) WHERE task_type = 'dream' AND processed = 0;
 
CREATE INDEX IF NOT EXISTS ix_queue_message_id_not_null
ON queue(message_id) WHERE message_id IS NOT NULL;

Note: SQLite uses 0/1 for boolean values, not true/false.

3.9 PostgreSQL-Specific SQL Patterns

`set_config('application_name', ...)` in dependencies.py

Used for request tracing. On SQLite, this is a no-op:

# In src/dependencies.py
if is_postgresql() and settings.DB.TRACING:
    await db.execute(
        text("SELECT set_config('application_name', :name, false)"),
        {"name": context},
    )

`on_conflict_do_nothing()` / `on_conflict_do_update()` in queue_manager.py and session.py

Currently imported from sqlalchemy.dialects.postgresql.insert. Both PostgreSQL and SQLite support INSERT ... ON CONFLICT syntax, but through different dialect-specific insert() functions.

Solution: Route through the dialect:

# In src/deriver/queue_manager.py::claim_work_units()
dialect = get_dialect()
stmt = dialect.insert_or_ignore_statement(
    models.ActiveQueueSession, values
).returning(
    models.ActiveQueueSession.work_unit_key, models.ActiveQueueSession.id
)
 
# In src/crud/session.py (upsert_session_peers)
dialect = get_dialect()
stmt = dialect.upsert_statement(
    models.SessionPeer.__table__,
    values=[...],
    index_elements=["session_name", "peer_name", "workspace_name"],
    set_={...},
)

Note on RETURNING: SQLite supports RETURNING since version 3.35.0 (2021-03-12). Python 3.11+ ships with SQLite 3.39+, so this is safe. The aiosqlite driver supports it.

`hashtext()` in advisory lock calls

Used only within pg_advisory_xact_lock(hashtext(...), hashtext(...)). Since the SQLite dialect replaces the entire locking mechanism with asyncio.Lock, hashtext() is never called on SQLite.

Window Functions (`func.sum(...).over(...)`)

Used in src/crud/message.py::_apply_token_limit() and src/deriver/queue_manager.py::get_queue_item_batch(). SQLite supports window functions since 3.25.0 (2018). No changes needed.

`BigInteger Identity()` Primary Keys

SQLite uses INTEGER PRIMARY KEY AUTOINCREMENT for auto-increment. SQLAlchemy’s BigInteger with Identity() maps correctly to SQLite’s INTEGER PRIMARY KEY (SQLite has only one integer type). The Identity() construct is PostgreSQL-specific but SQLAlchemy handles the translation. If any issues arise, conditionally use autoincrement=True without Identity() on SQLite.

3.10 Connection and Engine Configuration

The current src/db.py is PostgreSQL-centric:

connect_args = {"prepare_threshold": None}  # psycopg-specific
engine_kwargs = {...}  # pool configuration
engine = create_async_engine(settings.DB.CONNECTION_URI, connect_args=connect_args, ...)

Changes:

# src/db.py
 
from src.dialect import get_dialect, is_sqlite
 
dialect = get_dialect()
 
if is_sqlite():
    connect_args = {}
    engine_kwargs = {
        # SQLite does not support connection pooling in the traditional sense.
        # Use StaticPool for aiosqlite to reuse a single connection (required for
        # in-memory databases; also works well for file-based).
        "poolclass": StaticPool if ":memory:" in settings.DB.CONNECTION_URI else NullPool,
    }
else:
    # Existing PostgreSQL configuration
    connect_args = {"prepare_threshold": None}
    engine_kwargs = {
        # ... existing pool settings ...
    }
 
engine = create_async_engine(
    settings.DB.CONNECTION_URI,
    connect_args=connect_args,
    echo=settings.DB.SQL_DEBUG,
    **engine_kwargs,
)
 
# Register dialect-specific engine events (e.g., loading sqlite-vec)
dialect.register_engine_events(engine)

Schema handling: PostgreSQL uses settings.DB.SCHEMA (e.g., "public"). SQLite does not have schemas. Conditionally set:

if is_sqlite():
    meta = MetaData(naming_convention=convention)
    # No schema for SQLite
else:
    meta = MetaData(naming_convention=convention)
    meta.schema = settings.DB.SCHEMA

init_db() changes:

async def init_db():
    """Initialize the database."""
    if is_sqlite():
        # For SQLite: create all tables directly, then create virtual tables
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
 
            # Create FTS5 virtual table for message search
            await conn.execute(text("""
                CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
                    content,
                    content_rowid='id',
                    tokenize='porter unicode61'
                )
            """))
 
            # Create FTS5 sync triggers
            await conn.execute(text("""
                CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages
                BEGIN
                    INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
                END
            """))
            await conn.execute(text("""
                CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages
                BEGIN
                    INSERT INTO messages_fts(messages_fts, rowid, content)
                    VALUES ('delete', old.id, old.content);
                END
            """))
            await conn.execute(text("""
                CREATE TRIGGER IF NOT EXISTS messages_fts_update
                AFTER UPDATE OF content ON messages
                BEGIN
                    INSERT INTO messages_fts(messages_fts, rowid, content)
                    VALUES ('delete', old.id, old.content);
                    INSERT INTO messages_fts(rowid, content)
                    VALUES (new.id, new.content);
                END
            """))
 
            # Create sqlite-vec virtual tables for vector search
            await conn.execute(text("""
                CREATE VIRTUAL TABLE IF NOT EXISTS documents_vec USING vec0(
                    id TEXT PRIMARY KEY,
                    embedding float[1536]
                )
            """))
            await conn.execute(text("""
                CREATE VIRTUAL TABLE IF NOT EXISTS message_embeddings_vec USING vec0(
                    id TEXT PRIMARY KEY,
                    embedding float[1536]
                )
            """))
 
            # Create partial unique indexes (SQLAlchemy doesn't emit sqlite_where)
            await conn.execute(text("""
                CREATE UNIQUE INDEX IF NOT EXISTS uq_queue_reconciler_pending_work_unit_key
                ON queue(work_unit_key) WHERE task_type = 'reconciler' AND processed = 0
            """))
            await conn.execute(text("""
                CREATE UNIQUE INDEX IF NOT EXISTS uq_queue_dream_pending_work_unit_key
                ON queue(work_unit_key) WHERE task_type = 'dream' AND processed = 0
            """))
            await conn.execute(text("""
                CREATE INDEX IF NOT EXISTS ix_queue_message_id_not_null
                ON queue(message_id) WHERE message_id IS NOT NULL
            """))
 
    else:
        # Existing PostgreSQL init via Alembic
        from alembic import command
        from alembic.config import Config
 
        async with engine.connect() as connection:
            await connection.execute(
                text(f'CREATE SCHEMA IF NOT EXISTS "{settings.DB.SCHEMA}"')
            )
            await connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
            await connection.commit()
 
        alembic_cfg = Config("alembic.ini")
        command.upgrade(alembic_cfg, "head")

3.11 Alembic Migration Strategy

Alembic migrations are only run for PostgreSQL. For SQLite, Base.metadata.create_all() is used instead (appropriate for local dev/testing where the database is ephemeral).

However, if future production SQLite support is desired, migrations should be made dialect-aware. The following pattern is used in existing and future migration files:

Migration Helper Module

# migrations/dialects.py
"""
Dialect-aware migration helpers.
 
Usage in migration files:
    from migrations.dialects import is_postgresql, is_sqlite, skip_if_not_postgresql
 
    def upgrade():
        if is_postgresql():
            op.execute("CREATE EXTENSION IF NOT EXISTS vector")
 
        # Common DDL works on both
        op.create_table(...)
 
        if is_postgresql():
            op.create_index(..., postgresql_using="gin", ...)
        elif is_sqlite():
            op.execute("CREATE VIRTUAL TABLE IF NOT EXISTS ...")
"""
 
from alembic import op
 
 
def get_dialect_name() -> str:
    """Get the current migration dialect name."""
    return op.get_context().dialect.name
 
 
def is_postgresql() -> bool:
    return get_dialect_name() == "postgresql"
 
 
def is_sqlite() -> bool:
    return get_dialect_name() == "sqlite"

Migration File Pattern

For any migration that includes PostgreSQL-specific DDL:

# Example: migrations/versions/xxxx_add_vector_index.py
 
from migrations.dialects import is_postgresql, is_sqlite
 
def upgrade():
    # Common table creation (works on both dialects)
    op.add_column("documents", sa.Column("embedding", ...))
 
    if is_postgresql():
        # HNSW index for pgvector
        op.execute(
            "CREATE INDEX ix_documents_embedding_hnsw ON documents "
            "USING hnsw (embedding vector_cosine_ops) "
            "WITH (m = 16, ef_construction = 64)"
        )
    # SQLite: vec0 virtual table is created in init_db(), not in migrations

`migrations/env.py` Changes

The existing env.py has PostgreSQL-specific logic (schema creation, pgvector extension, search_path, prepare_threshold, port conversion). These must be guarded:

# In run_migrations_online():
dialect_name = op.get_context().dialect.name if context.config else "postgresql"
 
if dialect_name == "postgresql":
    # Existing PostgreSQL-specific setup:
    # - CREATE SCHEMA IF NOT EXISTS
    # - CREATE EXTENSION IF NOT EXISTS vector
    # - SET search_path
    ...
elif dialect_name == "sqlite":
    # SQLite-specific setup:
    # - Enable WAL mode
    # - Load sqlite-vec extension
    # - PRAGMA foreign_keys = ON
    ...

4. Implementation Phases

Phase 1: Foundation (Dialect Abstraction Layer)

Goal: Introduce the dialect protocol and registry without changing any runtime behavior. All existing PostgreSQL paths remain the default.

Tasks:

Create src/dialect.py with DatabaseDialect protocol and get_dialect() / is_sqlite() / is_postgresql() functions.
Create src/dialect_postgresql.py implementing PostgreSQLDialect — wrapping all existing PostgreSQL-specific operations.
Create src/dialect_sqlite.py with SQLiteDialect stub (methods raise NotImplementedError initially).
Add sqlite-vec, aiosqlite to pyproject.toml optional dependencies under a [sqlite] extra.
Create migrations/dialects.py helper module.

Verification: All existing tests pass with no behavior change. get_dialect() returns PostgreSQLDialect by default.

Phase 2: Model Layer Decoupling

Goal: Remove direct PostgreSQL imports from src/models.py and route through dialect.

Tasks:

Replace from pgvector.sqlalchemy import Vector and from sqlalchemy.dialects.postgresql import JSONB, TEXT with dialect-resolved types.
Replace all server_default=text("'{}'::jsonb") with server_default=text(get_dialect().server_default_json_empty_object()).
Replace regex check constraints with _dialect.regex_check_constraint(...).
Verify that postgresql_* Index kwargs are silently ignored on SQLite (they are, by SQLAlchemy design).

Verification: All existing PostgreSQL tests still pass. Model can be imported when DB_CONNECTION_URI points to SQLite (even if operations fail).

Phase 3: Engine and Connection Configuration

Goal: src/db.py correctly initializes for both backends.

Tasks:

Modify src/db.py to conditionally set connect_args, engine_kwargs, and schema.
Implement SQLiteDialect.register_engine_events() to load sqlite-vec and set PRAGMAs.
Implement init_db() SQLite path: create_all(), FTS5 virtual tables + triggers, sqlite-vec virtual tables, partial indexes.
Guard PostgreSQL-specific code in src/dependencies.py (set_config calls).

Verification: init_db() succeeds against a file-based SQLite database. Tables are created. sqlite-vec and FTS5 virtual tables exist.

Phase 4: CRUD Layer Adaptation

Goal: All CRUD operations work on SQLite.

Tasks (by file):

src/crud/message.py:
- Replace pg_advisory_xact_lock in create_messages() with dialect.acquire_session_lock() / release_session_lock().
- Replace cosine_distance() calls in search_messages() and search_messages_temporal() with sqlite-vec virtual table queries when on SQLite.
- Add vector sync to message_embeddings_vec virtual table after creating MessageEmbedding rows.
src/utils/search.py:
- Replace _fulltext_search() FTS logic with FTS5 MATCH queries on SQLite.
- Replace _semantic_search() pgvector queries with sqlite-vec virtual table queries on SQLite.
- Replace ts_rank() with FTS5 rank column on SQLite.
src/utils/filter.py:
- Replace column.contains(value) for JSONB with json_extract() decomposition on SQLite.
- Replace column[field_name].astext with func.json_extract(column, f"$.{field_name}") on SQLite.
src/crud/document.py:
- Replace cosine_distance() in query_documents() with sqlite-vec virtual table queries.
- Replace source_ids.contains([parent_id]) with json_each() subquery on SQLite.
- Remove with_for_update(skip_locked=True) on SQLite in cleanup_soft_deleted_documents().
- Add vector sync to documents_vec virtual table after creating/updating documents.
src/crud/session.py:
- Replace from sqlalchemy.dialects.postgresql import insert as pg_insert with dialect-routed upsert.
src/crud/collection.py:
- Replace column.op("||")(update_data) with dialect.json_merge_expression().
src/crud/peer_card.py:
- Replace column.op("||")(update_data) with dialect.json_merge_expression().
src/utils/summarizer.py:
- Replace column.op("||")(update_data) with dialect.json_merge_expression().
src/deriver/queue_manager.py:
- Replace from sqlalchemy.dialects.postgresql import insert with dialect-routed insert.
- Remove with_for_update(skip_locked=True) on SQLite in cleanup_stale_work_units().
src/reconciler/sync_vectors.py:
- Remove with_for_update(skip_locked=True) on SQLite.
- In SQLite mode, reconciliation syncs to sqlite-vec virtual tables instead of external vector stores.

Phase 5: Migration Compatibility

Goal: Alembic migrations are dialect-aware.

Tasks:

Update migrations/env.py to detect dialect and conditionally run PostgreSQL-specific setup.
Add migrations/dialects.py with is_postgresql() / is_sqlite() helpers.
Audit all existing migration files in migrations/versions/ and add guards around:
- from pgvector.sqlalchemy import Vector
- from sqlalchemy.dialects import postgresql
- postgresql.JSONB usage
- op.execute() calls with PostgreSQL-specific SQL
- postgresql_using, postgresql_where etc. in index creation

Note: For SQLite, the recommended path is init_db() with create_all(), not running through the full migration chain. Migrations are only needed if/when SQLite is used in persistent scenarios requiring schema evolution.

Phase 6: Testing

Goal: Full test coverage on both backends.

Tasks:

Create a conftest.py fixture that parameterizes tests by backend:

@pytest.fixture(params=["postgresql", "sqlite"])
async def db_backend(request, tmp_path):
    if request.param == "sqlite":
        db_path = tmp_path / "test.db"
        os.environ["DB_CONNECTION_URI"] = f"sqlite+aiosqlite:///{db_path}"
    else:
        os.environ["DB_CONNECTION_URI"] = POSTGRESQL_TEST_URI
    # Re-initialize engine, dialect, etc.
    ...

Add SQLite-specific test cases for:
- Vector search via sqlite-vec (cosine distance ordering)
- FTS5 search (MATCH queries, ranking)
- JSON filtering (json_extract decomposition)
- Session locking (asyncio.Lock serialization)
- Partial unique index enforcement
CI matrix: Run test suite against both PostgreSQL and SQLite.

5. Files to Modify

New Files

File	Purpose
`src/dialect.py`	`DatabaseDialect` protocol, `get_dialect()`, `is_sqlite()`, `is_postgresql()`
`src/dialect_postgresql.py`	`PostgreSQLDialect` implementation
`src/dialect_sqlite.py`	`SQLiteDialect` implementation, `VectorBLOB` type, `_session_locks` registry
`migrations/dialects.py`	`is_postgresql()` / `is_sqlite()` migration helpers

Modified Files

File	Changes
`src/models.py`	Replace `JSONB`/`TEXT`/`Vector` imports with dialect-resolved types. Replace `::jsonb` server defaults. Replace regex check constraints with dialect-aware expressions.
`src/db.py`	Conditional `connect_args`, `engine_kwargs`, schema, `StaticPool`/`NullPool` selection. Bifurcated `init_db()`. Engine event registration.
`src/config.py`	No structural changes. `DB_CONNECTION_URI` default remains PostgreSQL. Add `DB_CONNECTION_URI` documentation for SQLite format.
`src/dependencies.py`	Guard `set_config` calls with `is_postgresql()`.
`src/crud/message.py`	Replace advisory lock with `dialect.acquire_session_lock()`/`release_session_lock()`. Replace `cosine_distance()` with dialect-aware vector search. Sync embeddings to sqlite-vec virtual table on SQLite.
`src/crud/document.py`	Replace `cosine_distance()` with dialect-aware vector search. Replace `source_ids.contains()` with `json_each()` on SQLite. Conditionally skip `with_for_update(skip_locked=True)`. Sync vectors to sqlite-vec virtual table on SQLite.
`src/crud/session.py`	Replace `pg_insert` with dialect-routed upsert via `dialect.upsert_statement()`.
`src/crud/collection.py`	Replace `column.op("
`src/crud/peer_card.py`	Replace `column.op("
`src/utils/search.py`	Replace `to_tsvector`/`ts_rank`/`plainto_tsquery` with FTS5 MATCH on SQLite. Replace pgvector semantic search with sqlite-vec on SQLite.
`src/utils/filter.py`	Replace `column.contains()` with `json_extract()` decomposition on SQLite. Replace `column[field_name].astext` with `func.json_extract()` on SQLite.
`src/utils/summarizer.py`	Replace `column.op("
`src/deriver/queue_manager.py`	Replace `sqlalchemy.dialects.postgresql.insert` with dialect-routed insert. Conditionally skip `with_for_update(skip_locked=True)`.
`src/deriver/enqueue.py`	No direct PostgreSQL usage, but verify `insert(QueueItem).returning(QueueItem)` works on SQLite (it does since SQLite 3.35+).
`src/reconciler/sync_vectors.py`	Conditionally skip `with_for_update(skip_locked=True)`. Route vector sync to sqlite-vec on SQLite instead of external vector store.
`src/vector_store/__init__.py`	`get_external_vector_store()` returns `None` in SQLite mode (vectors are handled via sqlite-vec in-database, not externally).
`migrations/env.py`	Guard PostgreSQL-specific setup (schema creation, pgvector extension, search_path, connect_args). Add SQLite path with PRAGMA setup and extension loading.
`migrations/versions/*.py`	Add `from migrations.dialects import is_postgresql` guards around PostgreSQL-specific DDL in each migration file.
`pyproject.toml`	Add `aiosqlite` and `sqlite-vec` to optional dependencies: `[project.optional-dependencies] sqlite = ["aiosqlite>=0.20.0", "sqlite-vec>=0.1.6"]`.
`tests/conftest.py`	Add backend parameterization fixture.

6. Risk Assessment

High Risk

Risk	Mitigation
Performance regression on PostgreSQL path. Adding `if is_sqlite()` checks on every query adds a function call overhead.	`get_dialect()` is `@cache`-decorated (singleton). `is_sqlite()` is a single attribute check on a cached object. Overhead is negligible (<1 microsecond). Profile before/after to confirm.
SQLite concurrent write contention. SQLite uses a single-writer model. Under load, write operations may block.	Acceptable for dev/testing (non-goal for production). WAL mode + `busy_timeout=5000` mitigates most contention.
sqlite-vec availability. The `sqlite-vec` extension must be compiled for the target platform.	The `sqlite-vec` PyPI package provides pre-built wheels for Linux (x86_64, aarch64), macOS (x86_64, arm64), and Windows. CI tests verify availability.

Medium Risk

Risk	Mitigation
Subtle query behavior differences. SQLite’s type system is more relaxed; implicit type coercion may produce different results.	Comprehensive test suite running on both backends. Explicit type casting in queries where needed.
FTS5 tokenization differs from PostgreSQL tsquery. English stemming rules differ between `porter` (FTS5) and PostgreSQL’s `english` dictionary.	Acceptable for dev/testing. Search results may vary slightly but will be functionally equivalent.
JSON handling edge cases. `json_patch()` has different merge semantics than `
asyncio.Lock not released on crash. If the process crashes while holding a session lock, the lock is lost (which is fine — it dies with the process). But if an exception is raised mid-operation, the lock must be released.	Use `try/finally` around all lock acquisitions.

Low Risk

Risk	Mitigation
Alembic migration compatibility. Some migrations may use PostgreSQL syntax that fails on SQLite.	SQLite mode uses `create_all()`, not migrations. Migration dialect guards are defensive for future use.
`RETURNING` clause on SQLite. Some `INSERT ... RETURNING` patterns may not work on older SQLite versions.	Require SQLite >= 3.35.0. Python 3.11+ bundles SQLite 3.39+. Document minimum version requirement.

7. Verification Plan

Unit Tests

Dialect protocol compliance: Test that both PostgreSQLDialect and SQLiteDialect satisfy the DatabaseDialect protocol (runtime_checkable).
VectorBLOB type: Test round-trip serialization of list[float] ←> bytes for vectors of dimension 1536.
JSON merge expression: Test json_patch() produces equivalent results to || for Honcho’s metadata patterns.
Regex constraint fallback: Test that GLOB-based constraint accepts valid nanoid characters and rejects invalid ones.

Integration Tests (SQLite)

Full lifecycle test: Create workspace → Create peer → Create session → Create messages → Search messages (FTS + semantic) → Query documents.
Advisory lock replacement: Two concurrent create_messages() calls on the same session should serialize correctly (no duplicate seq_in_session).
Vector search accuracy: Insert 100 documents with embeddings, query with a known vector, verify top-k results are ordered by cosine distance.
FTS5 search: Insert messages, search with a keyword, verify results include expected messages.
JSON filtering: Create entities with varied metadata, apply filter queries, verify correct results.
Partial unique index: Attempt to insert duplicate reconciler queue items, verify uniqueness violation.
Upsert (session peers): Test rejoin scenario — peer leaves and rejoins session, verify configuration update.
Queue manager: Test full deriver pipeline: enqueue → claim → process → mark processed.

Regression Tests (PostgreSQL)

Run the full existing test suite against PostgreSQL to verify no regressions.
Specifically verify: vector search, FTS, advisory locks, JSONB operations, partial indexes, FOR UPDATE SKIP LOCKED.

CI Matrix

strategy:
  matrix:
    database: [postgresql, sqlite]

Both backends run the same test suite. Backend-specific tests are marked with @pytest.mark.postgresql or @pytest.mark.sqlite.

8. Open Questions

Should VECTOR_STORE.TYPE auto-switch to "sqlite-vec" when DB_CONNECTION_URI is SQLite? Currently, VECTOR_STORE.TYPE can be "pgvector", "turbopuffer", or "lancedb". A SQLite user should not need to configure an external vector store. Proposed: auto-detect and force TYPE="sqlite-vec" when the dialect is SQLite, ignoring any explicit setting.
Should we support sqlite+aiosqlite:///:memory: for testing? In-memory SQLite databases are useful for fast tests but require StaticPool and special handling (the DB is lost when the connection closes). Proposed: yes, with StaticPool in db.py.
What is the minimum SQLite version to require? Features used: FTS5 (3.9.0), partial indexes (3.8.0), RETURNING (3.35.0), window functions (3.25.0), json_patch (3.38.0), WAL mode (3.7.0). The binding constraint is json_patch at 3.38.0. Python 3.11 bundles SQLite 3.39.4. Proposed minimum: SQLite 3.38.0 (Python >= 3.11).
Should the Dreamer and Dialectic agents work in SQLite mode? These agents require LLM API access (external to the database). The database operations they perform (document CRUD, message queries) will work if the CRUD layer is adapted. However, they also depend on embedding generation. Proposed: they work in SQLite mode as long as embedding API keys are configured. The sqlite-vec virtual tables handle the vector storage.
Lock cleanup for long-lived processes. The _session_locks dictionary grows unboundedly as new (workspace, session) pairs are seen. For local dev this is fine. If SQLite is ever used in production, a TTL-based cleanup or weak reference approach would be needed. Proposed: defer to future production spec.
Should we keep the embedding column in the main table on SQLite, or store it only in the vec0 virtual table? Storing in both is redundant but simplifies the code (same ORM model structure). The BLOB in the main table is ~6KB per row (1536 * 4 bytes). For dev workloads this is fine. Proposed: keep in both for now; optimize later if needed.

Plastic Labs Internal Wiki

Explorer

Recently Updated

Company

Operating Model — Decisions Log

Operating Model

Security & Compliance

ideal-reasoned-representations

sqlite-support

Technical Specification: SQLite Database Backend for Honcho

Table of Contents

1. Problem Statement

2. Goals and Non-Goals

Goals

Non-Goals

3. Design

3.1 DatabaseDialect Protocol

PostgreSQL Implementation

SQLite Implementation

3.2 Dialect Registry and Auto-Detection

3.3 Model Layer Changes

Strategy: Conditional Column Types via Module-Level Helpers

Check Constraints with Regex

PostgreSQL-Specific Index Parameters

3.4 sqlite-vec Integration

Extension Loading

Virtual Table Schema

Sync Between Main Table and Virtual Table

Query Pattern

3.5 FTS5 Full-Text Search

FTS5 Virtual Table

Keeping FTS5 in Sync

Query Pattern

3.6 JSON Handling

Column Type

Server Defaults

JSONB Containment (@>)

JSONB Array Containment (source_ids @> [parent_id])

JSONB Merge (|| operator)

JSONB Field Access in Filters (column->>field_name)

3.7 Locking Strategy

Problem

Solution: Advisory Locks → asyncio.Lock

Solution: FOR UPDATE SKIP LOCKED → Sequential Processing or Application-Level Locking

3.8 Partial Unique Indexes and Constraints

3.9 PostgreSQL-Specific SQL Patterns

set_config('application_name', ...) in dependencies.py

on_conflict_do_nothing() / on_conflict_do_update() in queue_manager.py and session.py

hashtext() in advisory lock calls

Window Functions (func.sum(...).over(...))

BigInteger Identity() Primary Keys

3.10 Connection and Engine Configuration

3.11 Alembic Migration Strategy

Migration Helper Module

Migration File Pattern

migrations/env.py Changes

4. Implementation Phases

Phase 1: Foundation (Dialect Abstraction Layer)

Phase 2: Model Layer Decoupling

Phase 3: Engine and Connection Configuration

Phase 4: CRUD Layer Adaptation

Phase 5: Migration Compatibility

Phase 6: Testing

5. Files to Modify

New Files

Modified Files

6. Risk Assessment

High Risk

Medium Risk

Low Risk

7. Verification Plan

Unit Tests

Integration Tests (SQLite)

Regression Tests (PostgreSQL)

CI Matrix

8. Open Questions

Graph View

Table of Contents

Backlinks

JSONB Containment (`@>`)

JSONB Array Containment (`source_ids @> [parent_id]`)

JSONB Merge (`||` operator)

JSONB Field Access in Filters (`column->>field_name`)

`set_config('application_name', ...)` in dependencies.py

`on_conflict_do_nothing()` / `on_conflict_do_update()` in queue_manager.py and session.py

`hashtext()` in advisory lock calls

Window Functions (`func.sum(...).over(...)`)

`BigInteger Identity()` Primary Keys

`migrations/env.py` Changes