fix: implement SQLite partial unique indexes to prevent session_id co…

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v2.3.0
+    rev: v6.0.0
     hooks:
       - id: check-yaml
         args: [--allow-multiple-documents]

CHANGELOG.md

@@ -24,6 +24,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Partial-index repair log is INFO, not WARNING** (`jvspatial/db/sqlite.py`).
+  Dropping a non-partial index so it can be recreated with `WHERE` is expected
+  one-shot migration noise; log at info. Also satisfy ruff SIM110 in
+  `_index_needs_partial_repair` and mypy narrowing for `$eq` string literals in
+  `_sqlite_translate.py`.
+
+- **SQLite connect-time repair of global ``session_id`` unique indexes**
+  (`jvspatial/db/sqlite.py`). Opening a SQLite DB now drops UNIQUE indexes on
+  ``json_extract(data, '$.context.session_id')`` that lack a ``WHERE`` clause,
+  so a process that never re-ran ``ensure_indexes(Conversation)`` after the
+  partial-filter fix still stops wiping Interaction rows. Partial unique
+  ``create_index`` failures now raise instead of logging a warning.
+  Coverage: `tests/db/test_sqlite_partial_index.py`.
+
+- **SQLite partial unique indexes** (`jvspatial/db/sqlite.py`, `_sqlite_translate.py`).
+  `SQLiteDB.create_index` ignored Mongo-style `partialFilterExpression` /
+  `partial_filter_expression` kwargs and created **global** unique indexes on
+  shared `node` collections. That made `INSERT OR REPLACE` wipe `Interaction`
+  rows when they shared `context.session_id` with a `Conversation` (orchestrator
+  history empty on SQLite, fine on JsonDB). SQLite now translates the same
+  small dialect as Postgres into a `WHERE` clause, raises if a unique partial
+  filter cannot be translated, and drops/recreates a pre-existing non-partial
+  index of the same name so a restart self-heals. Coverage in
+  `tests/db/test_sqlite_partial_index.py` and translator unit tests.
 
 - **`SQLiteDB.find` treated `sort=[]` as an untranslatable sort**
   (`jvspatial/db/sqlite.py`). An empty list failed the `sort is None` guard, so
@@ -142,14 +166,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `(sort_value, id)` cursor — the default implementation tracks `id` only, so a
   non-`id` sort drops records that sort late but carry a lower `id`.
 
-
 ## [0.0.11] - 2026-07-02
 
 ### Fixed
 
 - **Single-hop `find_connected_nodes` fast path dropped valid neighbors** (`jvspatial/core/entities/node.py`). Two regressions introduced with the 0.0.10 fast path, both invisible to the existing suite because `JsonDB` exposes no `find_connected_nodes` (so tests fell back to the slow edge scan):
-  - *Limit-before-filter:* `limit` was pushed into the DB scan **before** the Python node-type filter, so `node(node=T)` / `nodes(node=T, limit=n)` could return nothing when a non-matching neighbor sorted first, even though matching neighbors existed. `limit` is now applied to the DB scan only when there is no node filter; otherwise it is applied after filtering.
-  - *Subtype resolution under an entity-name collision:* rows were deserialized with the base `Node` class, so when two `Node` subclasses persisted in one database shared an entity name (e.g. an app `User` and an embedded-agent `User`), `find_subclass_by_name` returned the first global match and `isinstance`-based filtering silently dropped the valid neighbor. The fast path now hydrates using the caller's requested concrete type as the resolution hint.
+  - _Limit-before-filter:_ `limit` was pushed into the DB scan **before** the Python node-type filter, so `node(node=T)` / `nodes(node=T, limit=n)` could return nothing when a non-matching neighbor sorted first, even though matching neighbors existed. `limit` is now applied to the DB scan only when there is no node filter; otherwise it is applied after filtering.
+  - _Subtype resolution under an entity-name collision:_ rows were deserialized with the base `Node` class, so when two `Node` subclasses persisted in one database shared an entity name (e.g. an app `User` and an embedded-agent `User`), `find_subclass_by_name` returned the first global match and `isinstance`-based filtering silently dropped the valid neighbor. The fast path now hydrates using the caller's requested concrete type as the resolution hint.
 - Regression coverage: `tests/core/test_connected_nodes_fast_path.py` installs a `find_connected_nodes` shim so the fast path is exercised (the stock `JsonDB` never did).
 
 ## [0.0.10] - 2026-07-02
@@ -290,7 +313,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `jvspatial.utils.stability.emit_experimental_once(name, message)` — public hook for opt-in surfaces that need to emit the experimental warning without going through the `@experimental` decorator (replaces private `_emit_once` calls). (Audit §7.7.)
 - `tests/db/test_sqlite_cross_loop_audit.py`, `tests/utils/test_wave4_polish_audit.py` — 10 new regression cases pinning Wave 4 audit fixes.
 - `jvspatial.core.entities.TraversalSkipped` and `TraversalPaused` exception classes. `Walker.skip()` now raises `TraversalSkipped` (caught via `except TraversalSkipped`); previously relied on substring-matching `"Node skipped"` in the message. (Audit §2.9 / SPEC §6.5.)
-- `PathSanitizer` rejects Windows-reserved filenames (`CON`, `PRN`, `AUX`, `NUL`, `COM1-9`, `LPT1-9`) regardless of host OS. ``CON.txt`` is rejected; ``CONFIG.json`` passes. (Audit §4.18 / SPEC §15.1.)
+- `PathSanitizer` rejects Windows-reserved filenames (`CON`, `PRN`, `AUX`, `NUL`, `COM1-9`, `LPT1-9`) regardless of host OS. `CON.txt` is rejected; `CONFIG.json` passes. (Audit §4.18 / SPEC §15.1.)
 - `tests/storage/test_windows_reserved_audit.py`, `tests/core/test_wave5_walker_audit.py`, `tests/api/test_deferred_invoke_fail_closed_audit.py`, `tests/db/test_sqlite_id_coercion_audit.py` — 23 new regression cases pinning Wave 5 audit fixes.
 
 ### Fixed

jvspatial/db/_sqlite_translate.py

@@ -295,4 +295,80 @@ def translate_sort(sort: Optional[List[Tuple[str, int]]]) -> Optional[str]:
     return ", ".join(parts)
 
 
-__all__ = ["translate_query", "translate_sort"]
+def _sql_string_literal(value: str) -> str:
+    """Quote a Python string as a SQLite string literal (single-quote escape)."""
+    return "'" + value.replace("'", "''") + "'"
+
+
+def translate_partial_filter_expression(
+    pfe: Dict[str, Any],
+) -> Optional[str]:
+    """Translate a Mongo-style ``partialFilterExpression`` to a SQLite WHERE.
+
+    Used by ``SQLiteDB.create_index`` so partial unique indexes are not
+    silently demoted to global unique indexes (which collides when other
+    Node subclasses share indexed field values — e.g. Conversation and
+    Interaction both writing ``context.session_id``).
+
+    Same deliberately small dialect as Postgres:
+
+    - ``{field: scalar}`` → ``json_extract(...) = <literal>``
+    - ``{field: {"$eq": scalar}}`` → same
+    - ``{field: {"$gt": scalar}}`` → ``json_extract(...) > <literal>``
+    - ``{field: {"$exists": True/False}}`` → ``IS NOT NULL`` / ``IS NULL``
+
+    Top-level keys are AND-ed. Values are inlined (CREATE INDEX WHERE
+    cannot use bound parameters). Returns ``None`` for unsupported shapes.
+    """
+    if not isinstance(pfe, dict) or not pfe:
+        return None
+    clauses: List[str] = []
+    for field, spec in pfe.items():
+        if not _safe_field_path(field):
+            return None
+        extract = _json_extract(field)
+        if isinstance(spec, dict):
+            if len(spec) != 1:
+                return None
+            op, val = next(iter(spec.items()))
+            if op == "$eq":
+                if not isinstance(val, (str, int, float, bool)):
+                    return None
+                if isinstance(val, bool):
+                    clauses.append(f"{extract} = {1 if val else 0}")
+                elif isinstance(val, (int, float)) and not isinstance(val, bool):
+                    clauses.append(f"{extract} = {val}")
+                elif isinstance(val, str):
+                    clauses.append(f"{extract} = {_sql_string_literal(val)}")
+                else:
+                    return None
+            elif op == "$gt":
+                if not isinstance(val, (str, int, float)) or isinstance(val, bool):
+                    return None
+                if isinstance(val, (int, float)):
+                    clauses.append(f"{extract} > {val}")
+                else:
+                    clauses.append(f"{extract} > {_sql_string_literal(val)}")
+            elif op == "$exists":
+                if val:
+                    clauses.append(f"{extract} IS NOT NULL")
+                else:
+                    clauses.append(f"{extract} IS NULL")
+            else:
+                return None
+        elif isinstance(spec, bool):
+            clauses.append(f"{extract} = {1 if spec else 0}")
+        elif isinstance(spec, (int, float)):
+            clauses.append(f"{extract} = {spec}")
+        elif isinstance(spec, str):
+            clauses.append(f"{extract} = {_sql_string_literal(spec)}")
+        else:
+            return None
+    return " AND ".join(clauses)
+
+
+__all__ = [
+    "translate_query",
+    "translate_sort",
+    "translate_partial_filter_expression",
+]

jvspatial/db/sqlite.py

@@ -16,7 +16,11 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union
 
-from ._sqlite_translate import translate_query, translate_sort
+from ._sqlite_translate import (
+    translate_partial_filter_expression,
+    translate_query,
+    translate_sort,
+)
 from .database import Database, finalize_find_results
 from .query import QueryEngine
 
@@ -196,6 +200,10 @@ async def _get_connection(self) -> "Connection":
                 """
             )
             await self._connection.commit()
+            # Drop legacy global unique indexes on session_id before any
+            # writes — CREATE INDEX IF NOT EXISTS would leave them in place
+            # and Interaction rows sharing a Conversation session_id get wiped.
+            await self._repair_non_partial_unique_indexes(self._connection)
             self._initialized = True
 
         return self._connection
@@ -212,6 +220,116 @@ def _json_path(self, field_path: str) -> str:
         # Convert "context.user_id" to "$.context.user_id" for JSON extraction
         return f"$.{field_path}"
 
+    @staticmethod
+    def _is_non_partial_session_id_unique_index(sql: Optional[str]) -> bool:
+        """True for UNIQUE indexes on context.session_id with no WHERE clause."""
+        if not sql:
+            return False
+        lower = sql.lower()
+        if "unique" not in lower:
+            return False
+        if "where" in lower:
+            return False
+        return "json_extract(data, '$.context.session_id')" in lower
+
+    async def _repair_non_partial_unique_indexes(
+        self, connection: "Connection"
+    ) -> None:
+        """Drop global unique indexes on ``context.session_id`` that lack WHERE.
+
+        Pre-partial-filter SQLite builds created a blanket UNIQUE on
+        ``session_id`` for the whole ``node`` collection. Conversation and
+        Interaction share that field; ``INSERT OR REPLACE`` then deleted
+        Interaction rows whenever Conversation was saved. Dropping here
+        lets the next ``ensure_indexes(Conversation)`` recreate a partial
+        unique index.
+        """
+        cursor = await connection.execute(
+            "SELECT name, sql FROM sqlite_master WHERE type='index' AND sql IS NOT NULL"
+        )
+        rows = await cursor.fetchall()
+        dropped = 0
+        for row in rows:
+            name = row[0]
+            sql = row[1]
+            if not self._is_non_partial_session_id_unique_index(sql):
+                continue
+            # Only drop indexes we (or prior jvspatial) named — avoid
+            # touching unrelated user indexes with similar SQL text.
+            if not str(name).startswith("idx_"):
+                continue
+            logger.warning(
+                "Dropping non-partial unique index '%s' on connect "
+                "(session_id uniqueness must be partial so Interaction "
+                "rows are not wiped). It will be recreated with WHERE on "
+                "the next Conversation ensure_indexes.",
+                name,
+            )
+            await connection.execute(f'DROP INDEX IF EXISTS "{name}"')
+            dropped += 1
+            for names in self._created_indexes.values():
+                names.discard(name)
+        if dropped:
+            await connection.commit()
+
+    @staticmethod
+    def _resolve_index_where(**kwargs: Any) -> Optional[str]:
+        """Resolve a CREATE INDEX WHERE clause from kwargs.
+
+        Honors an explicit ``where=`` string, otherwise translates a
+        Mongo-style partial filter under any of the names
+        ``get_indexes()`` / annotations may emit. Returns ``None`` when
+        no partial filter was requested. Raises ``ValueError`` when a
+        partial filter is present but cannot be translated — never
+        silently demote a partial unique index to a global unique.
+        """
+        explicit = kwargs.get("where")
+        if isinstance(explicit, str) and explicit.strip():
+            return explicit.strip()
+
+        pfe = (
+            kwargs.get("index_partial_filter_expression")
+            or kwargs.get("partial_filter_expression")
+            or kwargs.get("partialFilterExpression")
+        )
+        if not pfe:
+            return None
+        translated = translate_partial_filter_expression(pfe)
+        if translated is None:
+            raise ValueError(
+                "SQLiteDB.create_index: cannot translate "
+                f"index_partial_filter_expression {pfe!r} to a "
+                "SQLite WHERE clause. Pass an explicit ``where=`` "
+                "argument or use a supported filter shape "
+                "(equality / $gt / $exists on safe field paths "
+                "with scalar values)."
+            )
+        return translated
+
+    @staticmethod
+    def _index_needs_partial_repair(
+        existing_sql: Optional[str], where_sql: Optional[str]
+    ) -> bool:
+        """Return True when an on-disk index must be dropped and rebuilt.
+
+        A global unique index created before partial-filter support will
+        lack a WHERE clause; ``CREATE INDEX IF NOT EXISTS`` would leave
+        it in place and keep wiping colliding rows (Conversation vs
+        Interaction ``session_id``).
+        """
+        if not where_sql:
+            return False
+        if not existing_sql:
+            return False
+        lower = existing_sql.lower()
+        if "where" not in lower:
+            return True
+        # Require each AND-ed predicate from the desired WHERE to appear
+        # (SQLite stores CREATE INDEX SQL as we wrote it).
+        return any(
+            part.strip() not in existing_sql for part in where_sql.split(" AND ")
+        )
+
     async def create_index(
         self,
         collection: str,
@@ -225,11 +343,16 @@ async def create_index(
             collection: Collection name
             field_or_fields: Single field name (str) or list of (field_name, direction) tuples for compound indexes
             unique: Whether the index should enforce uniqueness
-            **kwargs: Additional options (ignored for SQLite)
+            **kwargs: Partial-filter options (``partialFilterExpression``,
+                ``partial_filter_expression``,
+                ``index_partial_filter_expression``, or explicit ``where=``).
+                Same Mongo dialect as PostgresDB.create_index.
 
         Note:
             SQLite indexes on nested JSON fields use json_extract() function.
             Direction parameter is ignored for SQLite (always ascending).
+            When a partial filter is required and an older non-partial
+            index of the same name exists, it is dropped and recreated.
         """
         connection = await self._get_connection()
 
@@ -248,9 +371,37 @@ async def create_index(
             field_names = "_".join(field.replace(".", "_") for field, _ in fields)
             index_name = f"idx_{collection}_{field_names}"
 
-        # Check if index already exists
+        where_sql = self._resolve_index_where(**kwargs)
+
+        # Inspect on-disk definition so a prior global unique index can
+        # be repaired after upgrading to partial-filter support.
+        cursor = await connection.execute(
+            "SELECT sql FROM sqlite_master WHERE type='index' AND name=?",
+            (index_name,),
+        )
+        row = await cursor.fetchone()
+        existing_sql = row[0] if row else None
+
+        if self._index_needs_partial_repair(existing_sql, where_sql):
+            logger.info(
+                "Dropping non-partial index '%s' so it can be recreated "
+                "with WHERE %s",
+                index_name,
+                where_sql,
+            )
+            await connection.execute(f"DROP INDEX IF EXISTS {index_name}")
+            await connection.commit()
+            self._created_indexes[collection].discard(index_name)
+            existing_sql = None
+
+        # Check if index already exists (in-process or on disk, correct form)
         if index_name in self._created_indexes[collection]:
-            return  # Index already created
+            return
+        if existing_sql is not None and not self._index_needs_partial_repair(
+            existing_sql, where_sql
+        ):
+            self._created_indexes[collection].add(index_name)
+            return
 
         # Build SQLite index creation statement
         # For nested fields, use json_extract() to extract values from JSON
@@ -261,15 +412,15 @@ async def create_index(
 
         index_columns = ", ".join(index_expressions)
         unique_clause = "UNIQUE" if unique else ""
+        where_clause = f" WHERE {where_sql}" if where_sql else ""
 
         try:
-            # Create index on the records table
-            # Include collection in the index to support efficient filtering
-            # SQLite doesn't support parameterized WHERE clauses in CREATE INDEX,
-            # so we include collection as the first column
+            # Include collection in the index to support efficient filtering.
+            # Partial filters become a SQLite partial index WHERE clause so
+            # uniqueness is scoped the same way as Mongo/Postgres.
             sql = f"""
             CREATE {unique_clause} INDEX IF NOT EXISTS {index_name}
-            ON records (collection, {index_columns})
+            ON records (collection, {index_columns}){where_clause}
             """
 
             await connection.execute(sql)
@@ -280,10 +431,18 @@ async def create_index(
 
             logger.debug(
                 f"Created index '{index_name}' on collection '{collection}' "
-                f"(unique={unique}, fields={[f[0] for f in fields]})"
+                f"(unique={unique}, fields={[f[0] for f in fields]}"
+                f"{', partial' if where_sql else ''})"
             )
 
         except Exception as e:
+            # Partial unique indexes protect Conversation/Interaction
+            # coexistence — never swallow create failures for those.
+            if unique and where_sql:
+                raise RuntimeError(
+                    f"Failed to create partial unique index '{index_name}' "
+                    f"on collection '{collection}': {e}"
+                ) from e
             logger.warning(
                 f"Failed to create index '{index_name}' on collection '{collection}': {e}"
             )

jvspatial/storage/interfaces/local.py

@@ -438,8 +438,11 @@ async def save_file(
                     "Skipping MIME allowlist for internal marker: %s", file_path
                 )
             else:
+                hint_mime = (
+                    (metadata or {}).get("mime") if isinstance(metadata, dict) else None
+                )
                 validation = self.validator.validate_file(
-                    content=content, filename=filename
+                    content=content, filename=filename, hint_mime=hint_mime
                 )
                 logger.debug(f"File validation passed: {validation}")