fix: shorten reconciliation cycle + fix 'IN' equality check

.env.template

@@ -232,13 +232,6 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here
 # =============================================================================
 # Vector Store Settings
 # =============================================================================
-# VECTOR_STORE_TYPE="lancedb"
-# VECTOR_STORE_NAMESPACE="honcho"
-# VECTOR_STORE_TURBOPUFFER_API_KEY=
-# VECTOR_STORE_TURBOPUFFER_REGION=
-# VECTOR_STORE_LANCEDB_PATH="./lancedb_data"
-
-# Vector store settings
 # Vector store type: "pgvector", "turbopuffer", or "lancedb"
 VECTOR_STORE_TYPE=pgvector
 
@@ -249,14 +242,14 @@ VECTOR_STORE_MIGRATED=false
 # Namespaces follow the pattern:
 # - Documents: {NAMESPACE}.{workspace}.{observer}.{observed}
 # - Messages: {NAMESPACE}.{workspace}.messages
-VECTOR_STORE_NAMESPACE=honcho
+# VECTOR_STORE_NAMESPACE=honcho  # Inherits from NAMESPACE if not set
 
 # Embedding dimensions (default: 1536 for OpenAI text-embedding-3-small)
-VECTOR_STORE_DIMENSIONS=1536
+# VECTOR_STORE_DIMENSIONS=1536
 
 # Turbopuffer-specific settings (required if TYPE is "turbopuffer")
 # VECTOR_STORE_TURBOPUFFER_API_KEY=your-turbopuffer-api-key
-# VECTOR_STORE_TURBOPUFFER_REGION=us-east-1
+# VECTOR_STORE_TURBOPUFFER_REGION=gcp-us-east4
 
 # LanceDB-specific settings (local embedded mode)
-VECTOR_STORE_LANCEDB_PATH=./lancedb_data
+# VECTOR_STORE_LANCEDB_PATH=./lancedb_data

src/deriver/queue_manager.py

@@ -51,7 +51,7 @@ class WorkerOwnership(NamedTuple):
 
 
 QUEUE_CLEANUP_INTERVAL_SECONDS = 43200  # 12 hours
-RECONCILIATION_INTERVAL_SECONDS = 900  # 15 minutes
+RECONCILIATION_INTERVAL_SECONDS = 300  # 5 minutes
 
 
 class QueueManager:

src/deriver/vector_reconciliation.py

@@ -479,7 +479,6 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics:
 
     from src.crud.document import cleanup_soft_deleted_documents
 
-    print("Running vector reconciliation cycle")
     async with tracked_db("reconciliation") as db:
         # If no external vector store (pgvector mode), only clean up soft-deleted documents
         if external_vector_store is None:

src/vector_store/lancedb.py

@@ -6,6 +6,7 @@ This module provides a LanceDB-based implementation of the VectorStore interface
 
 import asyncio
 import logging
+from collections.abc import Sequence
 from typing import Any, cast
 
 import lancedb
@@ -250,6 +251,10 @@ class LanceDBVectorStore(VectorStore):
         """
         Convert a filter dict to SQL WHERE clause syntax.
 
+        Supports filter formats:
+        - {"key": "value"} -> key = 'value'
+        - {"key": {"in": ["a", "b"]}} -> key IN ('a', 'b')
+
         Args:
             filters: Dictionary of attribute -> value filters
 
@@ -261,8 +266,20 @@ class LanceDBVectorStore(VectorStore):
 
         conditions: list[str] = []
         for key, value in filters.items():
+            # Check if value is a dict with "in" operator
+            if isinstance(value, dict) and "in" in value:
+                # IN clause for list membership
+                in_values = cast(Sequence[Any], value["in"])
+                if in_values:
+                    escaped_values = [
+                        f"'{str(v).replace(chr(39), chr(39) + chr(39))}'"
+                        if isinstance(v, str)
+                        else str(v)
+                        for v in in_values
+                    ]
+                    conditions.append(f"{key} IN ({', '.join(escaped_values)})")
             # Handle string values with proper quoting
-            if isinstance(value, str):
+            elif isinstance(value, str):
                 # Escape single quotes in the value
                 escaped_value = value.replace("'", "''")
                 conditions.append(f"{key} = '{escaped_value}'")

src/vector_store/turbopuffer.py

@@ -6,7 +6,7 @@ This module provides a Turbopuffer-based implementation of the VectorStore inter
 
 import logging
 from collections.abc import Sequence
-from typing import Any, Literal
+from typing import Any, Literal, cast
 
 from turbopuffer import AsyncTurbopuffer, NotFoundError
 from turbopuffer.lib.namespace import AsyncNamespace
@@ -18,8 +18,9 @@ from . import VectorQueryResult, VectorRecord, VectorStore, VectorUpsertResult
 
 logger = logging.getLogger(__name__)
 
-# Type alias for Turbopuffer's equality filter format
+# Type aliases for Turbopuffer's filter formats
 EqFilter = tuple[str, Literal["Eq"], Any]
+InFilter = tuple[str, Literal["In"], Sequence[Any]]
 AndFilter = tuple[Literal["And"], Sequence[Filter]]
 
 DISTANCE_METRIC = "cosine_distance"
@@ -192,8 +193,13 @@ class TurbopufferVectorStore(VectorStore):
         Convert a filter dict to Turbopuffer filter format.
 
         Turbopuffer uses tuples like (attribute, "Eq", value) for filters,
+        (attribute, "In", [values]) for membership filters,
         and ("And", [filters]) for combining multiple filters.
 
+        Supports filter formats:
+        - {"key": "value"} -> ("key", "Eq", "value")
+        - {"key": {"in": ["a", "b"]}} -> ("key", "In", ["a", "b"])
+
         Args:
             filters: Dictionary of attribute -> value filters
 
@@ -203,11 +209,16 @@ class TurbopufferVectorStore(VectorStore):
         if not filters:
             return None
 
-        filter_list: list[EqFilter] = []
+        filter_list: list[EqFilter | InFilter] = []
         for key, value in filters.items():
-            # Simple equality filter using "Eq" operator
-            eq_filter: EqFilter = (key, "Eq", value)
-            filter_list.append(eq_filter)
+            # Check if value is a dict with "in" operator
+            if isinstance(value, dict) and "in" in value:
+                # Membership filter using "In" operator
+                in_values = cast(Sequence[Any], value["in"])
+                filter_list.append((key, "In", in_values))
+            else:
+                # Simple equality filter using "Eq" operator
+                filter_list.append((key, "Eq", cast(Any, value)))
 
         if not filter_list:
             return None

tests/integration/test_message_embeddings.py

@@ -12,12 +12,26 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src import models
+from src.config import settings
 from src.crud import create_messages
 from src.models import Peer, Workspace
 from src.schemas import MessageCreate
 from src.utils.search import search
 
 
+def _stores_embeddings_in_postgres() -> bool:
+    """Check if current config stores MessageEmbedding rows in postgres."""
+    return settings.VECTOR_STORE.TYPE == "pgvector" or not settings.VECTOR_STORE.MIGRATED
+
+
+# Skip tests that depend on MessageEmbedding rows when using external store in migrated mode
+requires_postgres_embeddings = pytest.mark.skipif(
+    not _stores_embeddings_in_postgres(),
+    reason="MessageEmbedding rows not created when TYPE != 'pgvector' and MIGRATED=true",
+)
+
+
+@requires_postgres_embeddings
 @pytest.mark.asyncio
 async def test_message_embedding_created_when_setting_enabled(
     db_session: AsyncSession,
@@ -123,6 +137,7 @@ async def test_message_embedding_not_created_when_setting_disabled(
     assert embedding_record is None
 
 
+@requires_postgres_embeddings
 @pytest.mark.asyncio
 async def test_multiple_message_embeddings_created_when_setting_enabled(
     db_session: AsyncSession,
@@ -182,6 +197,7 @@ async def test_multiple_message_embeddings_created_when_setting_enabled(
         assert embedding_record.peer_name == test_peer.name
 
 
+@requires_postgres_embeddings
 @pytest.mark.asyncio
 async def test_semantic_search_when_embeddings_enabled(
     db_session: AsyncSession,
@@ -257,6 +273,7 @@ async def test_semantic_search_when_embeddings_enabled(
     assert created_message.public_id in found_message_ids
 
 
+@requires_postgres_embeddings
 @pytest.mark.asyncio
 async def test_message_chunking_creates_multiple_embeddings(
     db_session: AsyncSession,

tests/utils/test_agent_tools.py

@@ -312,9 +312,14 @@ class TestSearchMemory:
     """Tests for _handle_search_memory."""
 
     async def test_returns_matching_observations(
-        self, make_tool_context: Callable[..., ToolContext]
+        self,
+        make_tool_context: Callable[..., ToolContext],
+        monkeypatch: pytest.MonkeyPatch,
     ):
         """Returns observations matching semantic query."""
+        # Force pgvector queries since test documents are created directly in postgres
+        monkeypatch.setattr("src.config.settings.VECTOR_STORE.MIGRATED", False)
+
         ctx = make_tool_context()
 
         result = await _handle_search_memory(ctx, {"query": "coffee preferences"})