Merge pull request 'Install Yarn in the frontend CI job instead of relying on setup-node's cache' (#89) from fix/ci-frontend-yarn into main

setup-node's `cache: yarn` needs yarn already on PATH, and the Node it
installs does not ship one, so the frontend job failed at setup on every
run. Install the lockfile's Yarn 1.22.22 explicitly instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -195,6 +195,10 @@ OBSIDIAN_E2E_PASSWORD=
 OBSIDIAN_DEVICE_NAME=simbarag
 # Set to true to run continuous sync in the background
 OBSIDIAN_CONTINUOUS_SYNC=false
+# bidirectional: notes and journal tasks Simba writes sync back to your devices.
+# pull-only: download only, Simba's writes stay in the container's copy.
+# mirror-remote: download only and revert local changes.
+OBSIDIAN_SYNC_MODE=bidirectional
 OBSIDIAN_SYNC_INTERVAL=60
 # Local path to Obsidian vault (where files are synced)
 OBSIDIAN_VAULT_PATH=/app/data/obsidian
@@ -204,6 +208,9 @@ OBSIDIAN_VAULT_NAME=
 # LangGraph checkpoint store (SQLite). Must be on a persistent volume.
 # Both blue and green containers share this volume via docker-compose.
 CHECKPOINT_DB_PATH=/app/checkpoints/checkpoints.db
+# Checkpoints keep a full message-list snapshot per turn; this bounds how many
+# are kept per conversation thread (older ones are pruned after each turn).
+CHECKPOINT_KEEP_PER_THREAD=20
 
 # Beli restaurant recommendations
 # Beli account credentials for each profile are stored encrypted in the database.

.gitea/workflows/ci.yml

@@ -34,11 +34,13 @@ jobs:
         working-directory: raggr-frontend
     steps:
       - uses: actions/checkout@v4.2.2
+      # No `cache: yarn` here: setup-node requires yarn to already be on PATH
+      # for that, and the Node it installs doesn't ship one.
       - uses: actions/setup-node@v4.4.0
         with:
           node-version: '22.14.0'
-          cache: yarn
-          cache-dependency-path: raggr-frontend/yarn.lock
+      - name: Install Yarn (lockfile v1)
+        run: npm install --global yarn@1.22.22
       - name: Install locked frontend dependencies
         run: yarn install --frozen-lockfile
       - name: Renderer tests

blueprints/conversation/__init__.py

@@ -32,6 +32,7 @@ from utils.s3_client import get_image as s3_get_image
 from utils.s3_client import upload_file as s3_upload_file
 from utils.s3_client import upload_image as s3_upload_image
 
+from .checkpoints import delete_thread_checkpoints
 from .logic import (
     add_message_to_conversation,
     persist_agent_tool_messages,
@@ -137,9 +138,24 @@ conversation_blueprint = Blueprint(
 
 async def _get_owned_conversation(conversation_id, user_id) -> Conversation | None:
     """Fetch a conversation only if it belongs to the given user."""
+    try:
+        conversation_id = uuid.UUID(str(conversation_id))
+    except ValueError:
+        return None
     return await Conversation.get_or_none(id=conversation_id, user_id=user_id)
 
 
+async def _owns_storage_key(key: str, user_id) -> bool:
+    """True if `key` is stored under a conversation the user owns.
+
+    Every uploaded object lives under conversations/<conversation id>/.
+    """
+    parts = key.split("/")
+    if len(parts) < 3 or parts[0] != "conversations" or ".." in parts:
+        return False
+    return await _get_owned_conversation(parts[1], user_id) is not None
+
+
 def _build_user_message(
     query_text: str,
     image_block: dict | None = None,
@@ -220,6 +236,8 @@ async def upload_image():
 
     if not file or not conversation_id:
         return jsonify({"error": "file and conversation_id are required"}), 400
+    if await _get_owned_conversation(conversation_id, current_user_uuid) is None:
+        return jsonify({"error": "conversation not found"}), 404
 
     file_bytes = file.read()
     content_type = file.content_type or "image/jpeg"
@@ -325,6 +343,8 @@ async def upload_attachment():
 @conversation_blueprint.get("/image/<path:image_key>")
 @jwt_access_required
 async def serve_image(image_key: str):
+    if not await _owns_storage_key(image_key, get_jwt_identity()):
+        return jsonify({"error": "image not found"}), 404
     url = await s3_presigned_url(image_key)
     return jsonify({"url": url})
 
@@ -758,5 +778,11 @@ async def delete_conversation(conversation_id: str):
     store = getattr(current_app, "conversation_summary_store", None)
     if store is not None:
         await store.delete(str(user_uuid), str(conversation.id))
+    thread_id = str(conversation.id)
     await conversation.delete()
+    checkpointer = getattr(
+        getattr(current_app, "main_agent", None), "checkpointer", None
+    )
+    if checkpointer is not None:
+        await delete_thread_checkpoints(checkpointer, thread_id)
     return jsonify({"success": True})

blueprints/conversation/agents.py

@@ -23,6 +23,7 @@ from blueprints.conversation.artifacts_util import (
     note_title,
     obsidian_source_ref,
     search_documents_artifact,
+    search_results_model_context,
     truncate,
     unique,
 )
@@ -162,7 +163,10 @@ async def simba_search(
         )
     except ValueError as exc:
         return str(exc), None
-    return serialized, search_documents_artifact(query, "paperless", docs)
+    artifact = search_documents_artifact(query, "paperless", docs)
+    if artifact is not None:
+        artifact["model_context"] = search_results_model_context(docs, "paperless")
+    return serialized, artifact
 
 
 MEALIE_NOT_CONFIGURED = "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
@@ -488,7 +492,10 @@ async def obsidian_search_notes(
         serialized, docs = await query_vector_store(
             query=query, source="obsidian", **kwargs
         )
-        return serialized, search_documents_artifact(query, "obsidian", docs)
+        artifact = search_documents_artifact(query, "obsidian", docs)
+        if artifact is not None:
+            artifact["model_context"] = search_results_model_context(docs, "obsidian")
+        return serialized, artifact
 
     except Exception as e:
         return f"Error searching Obsidian notes: {str(e)}", None

blueprints/conversation/artifacts_util.py

@@ -143,43 +143,64 @@ def vault_relative_path(filepath: str) -> str:
         return filepath
 
 
+def _result_identity(doc, source: str) -> dict:
+    """Identifying fields shared by the UI snippet and the model's full-text view."""
+    from blueprints.rag.retrieval import citation_url
+
+    meta = doc.metadata or {}
+    result = {}
+    if url := citation_url(doc):
+        result["url"] = url
+        result["source_id"] = str(doc.id)
+    if headings := [meta[k] for k in ("h1", "h2", "h3") if meta.get(k)]:
+        result["headings"] = headings
+    if source == "obsidian":
+        filepath = meta.get("filepath") or ""
+        path = vault_relative_path(filepath) if filepath else ""
+        result["title"] = note_title(path, meta) if path else "Untitled note"
+        if path:
+            result["path"] = path
+        created = meta.get("created_at")
+        if isinstance(created, str) and created:
+            result["date"] = created[:10]
+    else:
+        filename = meta.get("filename") or ""
+        result["title"] = Path(filename).stem or filename or "Document"
+        if filename:
+            result["filename"] = filename
+        if meta.get("document_type"):
+            result["document_type"] = str(meta["document_type"])
+        created = epoch_to_date(meta.get("created_date"))
+        if created:
+            result["date"] = created
+    return result
+
+
 def search_documents_artifact(query: str, source: str, docs) -> dict | None:
     """Build a `search.documents` artifact from vector-store Documents."""
     if not docs:
         return None
-    results = []
-    from blueprints.rag.retrieval import citation_url
-
-    for doc in docs:
-        meta = doc.metadata or {}
-        result = {"snippet": snippet(doc.page_content)}
-        if url := citation_url(doc):
-            result["url"] = url
-            result["source_id"] = str(doc.id)
-        if headings := [meta[k] for k in ("h1", "h2", "h3") if meta.get(k)]:
-            result["headings"] = headings
-        if source == "obsidian":
-            filepath = meta.get("filepath") or ""
-            path = vault_relative_path(filepath) if filepath else ""
-            result["title"] = note_title(path, meta) if path else "Untitled note"
-            if path:
-                result["path"] = path
-            created = meta.get("created_at")
-            if isinstance(created, str) and created:
-                result["date"] = created[:10]
-        else:
-            filename = meta.get("filename") or ""
-            result["title"] = Path(filename).stem or filename or "Document"
-            if filename:
-                result["filename"] = filename
-            if meta.get("document_type"):
-                result["document_type"] = str(meta["document_type"])
-            created = epoch_to_date(meta.get("created_date"))
-            if created:
-                result["date"] = created
-        results.append(result)
+    results = [
+        {"snippet": snippet(doc.page_content), **_result_identity(doc, source)}
+        for doc in docs
+    ]
     return make_artifact(
         "search.documents",
         f"Search: {query}",
         {"query": query, "source": source, "results": results},
     )
+
+
+def search_results_model_context(docs, source: str) -> dict:
+    """Full chunk text for the model, keyed like `search_documents_artifact`'s results.
+
+    Set as a tool's `artifact["model_context"]` so that under compaction (see
+    compact_tool_result in blueprints/conversation/context.py) the model still
+    sees the full chunk text it retrieved, not the UI's 400-char snippets.
+    """
+    return {
+        "results": [
+            {**_result_identity(doc, source), "content": doc.page_content}
+            for doc in docs
+        ]
+    }

blueprints/conversation/checkpoints.py

@@ -0,0 +1,87 @@
+"""Keeps the LangGraph checkpoint DB from growing without bound.
+
+Every `AsyncSqliteSaver` row stores a full snapshot of the graph's channel
+values (the complete message list so far), not a delta — so an unpruned
+thread's checkpoints grow roughly linearly with turn count.
+"""
+
+import logging
+import os
+
+DEFAULT_KEEP_PER_THREAD = 20
+
+
+def keep_per_thread() -> int:
+    raw = os.getenv("CHECKPOINT_KEEP_PER_THREAD", str(DEFAULT_KEEP_PER_THREAD))
+    try:
+        value = int(raw)
+    except ValueError:
+        value = -1
+    if value < 1:
+        logging.warning(
+            "CHECKPOINT_KEEP_PER_THREAD=%r is invalid; falling back to %d",
+            raw,
+            DEFAULT_KEEP_PER_THREAD,
+        )
+        return DEFAULT_KEEP_PER_THREAD
+    return value
+
+
+async def prune_thread_checkpoints(
+    checkpointer, thread_id: str, keep: int | None = None
+) -> None:
+    """Keep only the newest `keep` checkpoints (and their writes) for a thread.
+
+    Safe to call after every turn: each checkpoint row is a self-contained
+    channel snapshot rather than a delta, and `aget_tuple`/`aget_state` only
+    ever read the single newest row for a (thread_id, checkpoint_ns) pair, so
+    dropping older rows never affects state reads. Reuses the saver's own
+    lock so this never races its aput/aget_tuple calls.
+    """
+    keep = keep_per_thread() if keep is None else keep
+    thread_id = str(thread_id)
+    conn = checkpointer.conn
+    async with checkpointer.lock, conn.cursor() as cur:
+        await cur.execute(
+            "SELECT DISTINCT checkpoint_ns FROM checkpoints WHERE thread_id = ?",
+            (thread_id,),
+        )
+        namespaces = [row[0] for row in await cur.fetchall()]
+        for ns in namespaces:
+            await cur.execute(
+                """
+                DELETE FROM checkpoints
+                WHERE thread_id = ? AND checkpoint_ns = ?
+                AND checkpoint_id NOT IN (
+                    SELECT checkpoint_id FROM checkpoints
+                    WHERE thread_id = ? AND checkpoint_ns = ?
+                    ORDER BY checkpoint_id DESC
+                    LIMIT ?
+                )
+                """,
+                (thread_id, ns, thread_id, ns, keep),
+            )
+            await cur.execute(
+                """
+                DELETE FROM writes
+                WHERE thread_id = ? AND checkpoint_ns = ?
+                AND checkpoint_id NOT IN (
+                    SELECT checkpoint_id FROM checkpoints
+                    WHERE thread_id = ? AND checkpoint_ns = ?
+                )
+                """,
+                (thread_id, ns, thread_id, ns),
+            )
+        await conn.commit()
+
+
+async def delete_thread_checkpoints(checkpointer, thread_id: str) -> None:
+    """Delete a conversation's checkpoint thread; never raises.
+
+    Called from conversation delete, which must still succeed even if the
+    checkpoint store is unreachable or the thread is already gone.
+    """
+    try:
+        await checkpointer.adelete_thread(str(thread_id))
+    except Exception:
+        logging.exception("Failed to delete checkpoint thread %s", thread_id)

blueprints/conversation/runtime.py

@@ -13,6 +13,7 @@ from langchain_core.messages import convert_to_messages
 from quart import current_app
 
 from metrics import AGENT_REQUESTS_TOTAL, AGENT_RESPONSE_SECONDS
+from .checkpoints import prune_thread_checkpoints
 from .logic import get_agent_for
 from .memory import build_system_prompt_with_memories
 
@@ -119,6 +120,18 @@ def _record(turn: PreparedTurn, started: float, result: str) -> None:
     )
 
 
+async def _prune_after_turn(turn: PreparedTurn) -> None:
+    checkpointer = getattr(turn.agent, "checkpointer", None)
+    if checkpointer is None:
+        return
+    try:
+        await prune_thread_checkpoints(
+            checkpointer, turn.config["configurable"]["thread_id"]
+        )
+    except Exception:
+        logging.exception("Failed to prune checkpoints")
+
+
 async def invoke_prepared(turn: PreparedTurn) -> dict:
     started = time.monotonic()
     result = "success"
@@ -132,6 +145,7 @@ async def invoke_prepared(turn: PreparedTurn) -> dict:
         raise
     finally:
         _record(turn, started, result)
+        await _prune_after_turn(turn)
 
 
 async def invoke_agent_for_user(
@@ -169,3 +183,4 @@ async def stream_prepared(turn: PreparedTurn):
         raise
     finally:
         _record(turn, started, result)
+        await _prune_after_turn(turn)

blueprints/rag/logic.py

@@ -1,4 +1,6 @@
+import asyncio
 import datetime
+import json
 import logging
 import os
 import re
@@ -16,7 +18,7 @@ from langchain_text_splitters import (
 from sqlalchemy import create_engine, text
 
 from .fetchers import PaperlessNGXService
-from .retrieval import build_filter, citation_url, document_dates
+from .retrieval import build_filter, citation_url, document_date, document_dates
 from utils.obsidian_service import ObsidianService
 
 # Load environment variables
@@ -325,6 +327,57 @@ def _load_obsidian_index_cache() -> dict[str, float]:
         return {row[0]: row[1] for row in result if row[0] is not None}
 
 
+def _walk_vault_mtimes(obsidian_service: ObsidianService) -> dict[str, float]:
+    """Walk the vault and stat every file. Blocking; run via asyncio.to_thread."""
+    return {
+        str(md_path): md_path.stat().st_mtime
+        for md_path in obsidian_service.walk_vault()
+    }
+
+
+def _read_obsidian_document(
+    obsidian_service: ObsidianService, filepath: str, now: float
+) -> Document | None:
+    """Read and parse one vault file into a Document, or None on error."""
+    try:
+        with open(filepath, "r", encoding="utf-8") as f:
+            content = f.read()
+        parsed = obsidian_service.parse_markdown(content, filepath)
+        metadata = {
+            "source": "obsidian",
+            "filepath": parsed["filepath"],
+            "folder": str(Path(parsed["filepath"]).parent)
+            if parsed["filepath"]
+            else "",
+            "tags": parsed["tags"],
+            "created_at": parsed["metadata"].get("created_at"),
+            "indexed_at": now,
+            **{
+                k: v
+                for k, v in parsed["metadata"].items()
+                if k not in ["created_at", "created_by"]
+            },
+        }
+        return Document(
+            page_content=parsed["content"], metadata=_make_serializable(metadata)
+        )
+    except Exception as e:
+        logger.warning(f"Error reading {filepath}: {e}")
+        return None
+
+
+def _read_obsidian_documents(
+    obsidian_service: ObsidianService, filepaths: list[str], now: float
+) -> list[Document]:
+    """Read and parse a batch of vault files. Blocking; run via asyncio.to_thread."""
+    documents = []
+    for filepath in filepaths:
+        document = _read_obsidian_document(obsidian_service, filepath, now)
+        if document is not None:
+            documents.append(document)
+    return documents
+
+
 async def sync_obsidian_documents() -> dict[str, int]:
     """Incrementally sync Obsidian documents to pgvector.
 
@@ -340,69 +393,52 @@ async def sync_obsidian_documents() -> dict[str, int]:
 
     # Load cache from DB on first run
     if not _obsidian_index_cache:
-        _obsidian_index_cache = _load_obsidian_index_cache()
+        _obsidian_index_cache = await asyncio.to_thread(_load_obsidian_index_cache)
 
     # Build map of current vault files -> mtime
-    vault_files: dict[str, float] = {}
-    for md_path in obsidian_service.walk_vault():
-        vault_files[str(md_path)] = md_path.stat().st_mtime
+    vault_files = await asyncio.to_thread(_walk_vault_mtimes, obsidian_service)
 
     added = 0
     updated = 0
-    deleted = 0
 
     # Find files to add or update
     files_to_index: list[str] = []
+    stale_filepaths: list[str] = []
     for filepath, mtime in vault_files.items():
         indexed_at = _obsidian_index_cache.get(filepath)
         if indexed_at is None:
             files_to_index.append(filepath)
             added += 1
         elif mtime > indexed_at:
-            # Delete old chunks first
-            delete_documents_by_metadata("filepath", filepath)
+            stale_filepaths.append(filepath)
             files_to_index.append(filepath)
             updated += 1
 
-    # Find deleted files (in cache but not on disk)
-    for filepath in list(_obsidian_index_cache):
-        if filepath not in vault_files:
-            delete_documents_by_metadata("filepath", filepath)
-            del _obsidian_index_cache[filepath]
-            deleted += 1
+    # Find deleted files (in cache but not on disk). Don't drop them from the
+    # cache yet: if the delete below raises, we want the next sync to still
+    # know about them and retry, rather than leaving their chunks searchable
+    # forever with no record that they need cleaning up.
+    removed_filepaths = [fp for fp in _obsidian_index_cache if fp not in vault_files]
+    stale_filepaths.extend(removed_filepaths)
+    deleted = len(removed_filepaths)
+
+    # Delete old chunks for updated/removed files before re-indexing, in one
+    # transaction instead of a round-trip per file. Left uncaught: a failure
+    # here must stop the sync so the cache isn't marked clean for work that
+    # didn't happen.
+    if stale_filepaths:
+        await asyncio.to_thread(delete_documents_by_filepaths, stale_filepaths)
+
+    # Only now that the delete has succeeded is it safe to forget these files.
+    for filepath in removed_filepaths:
+        del _obsidian_index_cache[filepath]
 
     # Index new/changed files
     if files_to_index:
         now = time.time()
-        documents = []
-        for filepath in files_to_index:
-            try:
-                with open(filepath, "r", encoding="utf-8") as f:
-                    content = f.read()
-                parsed = obsidian_service.parse_markdown(content, filepath)
-                metadata = {
-                    "source": "obsidian",
-                    "filepath": parsed["filepath"],
-                    "folder": str(Path(parsed["filepath"]).parent)
-                    if parsed["filepath"]
-                    else "",
-                    "tags": parsed["tags"],
-                    "created_at": parsed["metadata"].get("created_at"),
-                    "indexed_at": now,
-                    **{
-                        k: v
-                        for k, v in parsed["metadata"].items()
-                        if k not in ["created_at", "created_by"]
-                    },
-                }
-                document = Document(
-                    page_content=parsed["content"],
-                    metadata=_make_serializable(metadata),
-                )
-                documents.append(document)
-            except Exception as e:
-                logger.warning(f"Error reading {filepath}: {e}")
-                continue
+        documents = await asyncio.to_thread(
+            _read_obsidian_documents, obsidian_service, files_to_index, now
+        )
 
         if documents:
             splits = []
@@ -462,7 +498,7 @@ async def query_vector_store(
         ]
     serialized = "\n\n".join(
         (
-            f"Source: {doc.metadata}\n"
+            f"{_chunk_header(doc)}\n"
             + (f"Citation: {citation_url(doc)}\n" if citation_url(doc) else "")
             + f"Content: {doc.page_content}"
         )
@@ -471,6 +507,28 @@ async def query_vector_store(
     return serialized, retrieved_docs
 
 
+def _chunk_title(doc: Document) -> str:
+    from blueprints.conversation.artifacts_util import note_title, vault_relative_path
+
+    meta = doc.metadata or {}
+    if meta.get("source") == "obsidian":
+        path = vault_relative_path(meta.get("filepath") or "")
+        return note_title(path, meta) if path else "Untitled note"
+    filename = meta.get("filename") or ""
+    return Path(filename).stem or filename or "Document"
+
+
+def _chunk_header(doc: Document) -> str:
+    """Compact per-chunk header (title, date, source) replacing the raw metadata dict."""
+    meta = doc.metadata or {}
+    parts = [f"Title: {_chunk_title(doc)}"]
+    if date := document_date(meta):
+        parts.append(f"Date: {date}")
+    if source := meta.get("source"):
+        parts.append(f"Source: {source}")
+    return " | ".join(parts)
+
+
 def delete_all_documents():
     """Delete all documents from the vector store collection."""
     collection_id = _get_collection_id()
@@ -485,6 +543,27 @@ def delete_all_documents():
         conn.commit()
 
 
+def _delete_by_containment(conn, collection_id, filter_dict: dict) -> None:
+    """Delete rows whose cmetadata contains filter_dict.
+
+    Uses JSONB containment (`@>`) rather than `cmetadata->>key = value` so the
+    delete can use the GIN index langchain_postgres creates on cmetadata
+    (ix_cmetadata_gin, jsonb_path_ops); `->>` text extraction can't use it.
+
+    Unlike `->>`, which coerces to text before comparing, `@>` matches on the
+    stored JSON type: the string "123" does not match the number 123. Callers
+    must pass values with the same type as what was stored (today's only
+    caller, filepath/source strings, is fine).
+    """
+    conn.execute(
+        text(
+            "DELETE FROM langchain_pg_embedding "
+            "WHERE collection_id = :cid AND cmetadata @> CAST(:filter AS jsonb)"
+        ),
+        {"cid": collection_id, "filter": json.dumps(filter_dict)},
+    )
+
+
 def delete_documents_by_metadata(key: str, value: str):
     """Delete documents matching a metadata key/value pair."""
     collection_id = _get_collection_id()
@@ -492,13 +571,25 @@ def delete_documents_by_metadata(key: str, value: str):
         return
     engine = _get_engine()
     with engine.connect() as conn:
-        conn.execute(
-            text(
-                "DELETE FROM langchain_pg_embedding "
-                "WHERE collection_id = :cid AND cmetadata->>:key = :value"
-            ),
-            {"cid": collection_id, "key": key, "value": value},
-        )
+        _delete_by_containment(conn, collection_id, {key: value})
+        conn.commit()
+
+
+def delete_documents_by_filepaths(filepaths: list[str]) -> None:
+    """Delete all chunks for several obsidian filepaths in one transaction.
+
+    One containment delete per path, committed together, so a sync pass with
+    many stale files doesn't pay a commit round-trip per file.
+    """
+    if not filepaths:
+        return
+    collection_id = _get_collection_id()
+    if not collection_id:
+        return
+    engine = _get_engine()
+    with engine.connect() as conn:
+        for filepath in filepaths:
+            _delete_by_containment(conn, collection_id, {"filepath": filepath})
         conn.commit()
 
 

config/validation.py

@@ -225,6 +225,14 @@ def validate_configuration(env: Mapping[str, str], *, production: bool = True) -
                 problem(flag, "signature validation must remain enabled in production")
     for name in ("METRICS_TOKEN", "CRON_SECRET", "EMAIL_HMAC_SECRET"):
         secret(name, minimum=32)
+    if env.get("OBSIDIAN_SYNC_MODE", "bidirectional") not in {
+        "bidirectional",
+        "pull-only",
+        "mirror-remote",
+    }:
+        problem(
+            "OBSIDIAN_SYNC_MODE", "must be bidirectional, pull-only or mirror-remote"
+        )
     if env.get("OBSIDIAN_CONTINUOUS_SYNC") == "true":
         for name in ("OBSIDIAN_EMAIL", "OBSIDIAN_PASSWORD", "OBSIDIAN_VAULT_ID"):
             if not value(name):

docker-compose.yml

@@ -11,6 +11,7 @@ x-raggr-common: &raggr-common
     - PROMPT_BUDGET_TOKENS=${PROMPT_BUDGET_TOKENS:-16000}
     - RESPONSE_RESERVE_TOKENS=${RESPONSE_RESERVE_TOKENS:-4096}
     - CHECKPOINT_DB_PATH=${CHECKPOINT_DB_PATH:-/app/checkpoints/checkpoints.db}
+    - CHECKPOINT_KEEP_PER_THREAD=${CHECKPOINT_KEEP_PER_THREAD:-20}
     - CONVERSATION_SUMMARY_DB_PATH=${CONVERSATION_SUMMARY_DB_PATH:-${CHECKPOINT_DB_PATH:-/app/checkpoints/checkpoints.db}.summaries.db}
     - CHAT_RUN_LEASE_SECONDS=${CHAT_RUN_LEASE_SECONDS:-30}
     - CHAT_RUN_MAX_SECONDS=${CHAT_RUN_MAX_SECONDS:-1800}
@@ -60,6 +61,7 @@ x-raggr-common: &raggr-common
     - OBSIDIAN_E2E_PASSWORD=${OBSIDIAN_E2E_PASSWORD}
     - OBSIDIAN_DEVICE_NAME=${OBSIDIAN_DEVICE_NAME}
     - OBSIDIAN_CONTINUOUS_SYNC=${OBSIDIAN_CONTINUOUS_SYNC:-false}
+    - OBSIDIAN_SYNC_MODE=${OBSIDIAN_SYNC_MODE:-bidirectional}
     - OBSIDIAN_VAULT_PATH=${OBSIDIAN_VAULT_PATH:-/app/data/obsidian}
     - OBSIDIAN_VAULT_NAME=${OBSIDIAN_VAULT_NAME:-}
     - S3_ENDPOINT_URL=${S3_ENDPOINT_URL}

docs/deployment.md

@@ -39,7 +39,7 @@ make logs
 make migrate-history
 ```
 
-Startup validates configuration, runs `aerich upgrade`, optionally sets up Obsidian pull-only sync, and replaces the shell with the Python process. **A migration failure stops startup; the application is never launched.** Obsidian login/setup failures still log a warning and allow the application to continue.
+Startup validates configuration, runs `aerich upgrade`, optionally starts Obsidian sync in the background (`OBSIDIAN_SYNC_MODE`, default `bidirectional` so notes and tasks Simba writes reach your devices; containers take turns via a lock file in the shared vault), and replaces the shell with the Python process. **A migration failure stops startup; the application is never launched.** Obsidian login/setup failures still log a warning and allow the application to continue.
 
 An empty database without Aerich metadata may need explicit initialization. Inspect its state first. Run the migration tool in a one-off container because the app cannot start until migrations pass:
 

migrations/models/6_20260603000000_add_scheduled_messages.py

@@ -0,0 +1,38 @@
+"""Create scheduled_messages, which no committed migration ever created.
+
+Migrations 6–10 were generated outside the repository and never committed, and
+one of them created this table. Deployments that ran them already have it, so
+this is IF NOT EXISTS and a no-op there; fresh databases need it before
+migration 20 alters the table. Aerich cannot generate this file because its
+saved model state already includes the table: the DDL below is Tortoise's
+schema output for the model as of migration 19.
+"""
+
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        CREATE TABLE IF NOT EXISTS "scheduled_messages" (
+    "id" UUID NOT NULL PRIMARY KEY,
+    "recipient" VARCHAR(255) NOT NULL,
+    "channel" VARCHAR(20) NOT NULL,
+    "content" TEXT NOT NULL,
+    "subject" VARCHAR(255),
+    "scheduled_at" TIMESTAMPTZ NOT NULL,
+    "status" VARCHAR(20) NOT NULL DEFAULT 'pending',
+    "recurrence" VARCHAR(20) NOT NULL DEFAULT 'none',
+    "use_agent" BOOL NOT NULL DEFAULT False,
+    "error_message" TEXT,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "created_by_id" UUID NOT NULL REFERENCES "users" ("id") ON DELETE CASCADE
+);"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    # Existing deployments created this table in an uncommitted migration;
+    # never drop it from under them.
+    return ""

startup.sh

@@ -19,27 +19,36 @@ if [ "${OBSIDIAN_CONTINUOUS_SYNC:-false}" = "true" ]; then
     if [ -z "${OBSIDIAN_EMAIL:-}" ] || [ -z "${OBSIDIAN_PASSWORD:-}" ] || [ -z "${OBSIDIAN_VAULT_ID:-}" ]; then
         echo "WARNING: OBSIDIAN_EMAIL, OBSIDIAN_PASSWORD, or OBSIDIAN_VAULT_ID not set. Skipping sync."
     else
-        echo "Setting up Obsidian sync..."
-
         VAULT_PATH="${OBSIDIAN_VAULT_PATH:-/app/data/obsidian}"
+        SYNC_MODE="${OBSIDIAN_SYNC_MODE:-bidirectional}"
+        mkdir -p "${VAULT_PATH}"
 
-        # Login and setup sync (foreground, must complete before sync starts)
-        if ob login --email "${OBSIDIAN_EMAIL}" --password "${OBSIDIAN_PASSWORD}" && \
-           ob sync-setup \
-               --vault "${OBSIDIAN_VAULT_ID}" \
-               --path "${VAULT_PATH}" \
-               --password "${OBSIDIAN_E2E_PASSWORD:-}" \
-               --device-name "${OBSIDIAN_DEVICE_NAME:-simbarag}"; then
-            # Remove stale lock from previous container run
-            rm -rf "${VAULT_PATH}/.obsidian/.sync.lock"
-            # Set sync to pull-only (read-only) mode
-            ob sync-config --mode pull-only --path "${VAULT_PATH}"
-            # Start continuous sync in background
-            echo "Starting Obsidian continuous sync (pull-only)..."
-            ob sync --continuous --path "${VAULT_PATH}" &
-        else
-            echo "WARNING: Obsidian sync setup failed. Continuing without sync."
-        fi
+        # Blue and green share the vault volume. flock on a file in it works
+        # across containers, so this container only sets up and runs sync once
+        # the previous container's sync process has exited. Two clients writing
+        # one vault and its sync state would otherwise race. Setup runs in the
+        # background so the app starts while it waits.
+        (
+            flock 9
+            echo "Setting up Obsidian sync (${SYNC_MODE})..."
+            if ob login --email "${OBSIDIAN_EMAIL}" --password "${OBSIDIAN_PASSWORD}" && \
+               ob sync-setup \
+                   --vault "${OBSIDIAN_VAULT_ID}" \
+                   --path "${VAULT_PATH}" \
+                   --password "${OBSIDIAN_E2E_PASSWORD:-}" \
+                   --device-name "${OBSIDIAN_DEVICE_NAME:-simbarag}"; then
+                # Safe now: the lock means no other container's sync is running.
+                rm -rf "${VAULT_PATH}/.obsidian/.sync.lock"
+                if ob sync-config --mode "${SYNC_MODE}" --conflict-strategy merge \
+                    --path "${VAULT_PATH}"; then
+                    echo "Starting Obsidian continuous sync (${SYNC_MODE})..."
+                    exec ob sync --continuous --path "${VAULT_PATH}"
+                fi
+                echo "WARNING: Obsidian sync-config failed. Continuing without sync."
+            else
+                echo "WARNING: Obsidian sync setup failed. Continuing without sync."
+            fi
+        ) 9>"${VAULT_PATH}/.raggr-sync.lock" &
     fi
 fi
 

tests/unit/test_artifacts.py

@@ -85,6 +85,29 @@ class TestCreateArtifact:
         assert result is None
         assert await Artifact.all().count() == 0
 
+    async def test_drops_model_context_before_persisting(self, owner):
+        """model_context carries the full chunk text for the model's own view;
+        it must never reach the DB (or, from there, the UI)."""
+        user, conversation, message = owner
+        artifact = await create_artifact_from_output(
+            output=_output(
+                {
+                    "kind": "search.documents",
+                    "title": "Search: vet",
+                    "data": {"results": [{"snippet": "short"}]},
+                    "model_context": {"results": [{"content": "x" * 5000}]},
+                }
+            ),
+            tool="simba_search",
+            user=user,
+            conversation=conversation,
+            message=message,
+        )
+        assert artifact is not None
+        saved = await Artifact.get(id=artifact.id)
+        assert saved.data == {"results": [{"snippet": "short"}]}
+        assert "model_context" not in saved.data
+
     async def test_plain_string_output(self, owner):
         user, conversation, message = owner
         result = await create_artifact_from_output(

tests/unit/test_checkpoints.py

@@ -0,0 +1,226 @@
+"""LangGraph checkpoint pruning/deletion: bounds the on-disk checkpoint DB."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import aiosqlite
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langgraph.checkpoint.base import empty_checkpoint
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+from langgraph.graph import END, START, MessagesState, StateGraph
+from quart import Quart
+from quart_jwt_extended import JWTManager, create_access_token
+from tortoise import Tortoise
+
+from blueprints.conversation import conversation_blueprint
+from blueprints.conversation.checkpoints import (
+    DEFAULT_KEEP_PER_THREAD,
+    delete_thread_checkpoints,
+    keep_per_thread,
+    prune_thread_checkpoints,
+)
+from blueprints.conversation.models import Conversation
+from blueprints.users.models import User
+
+
+async def _checkpoint_ids(saver, thread_id):
+    async with saver.conn.execute(
+        "SELECT checkpoint_id FROM checkpoints WHERE thread_id = ? ORDER BY checkpoint_id",
+        (str(thread_id),),
+    ) as cur:
+        return [row[0] async for row in cur]
+
+
+async def _write_rows(saver, thread_id):
+    async with saver.conn.execute(
+        "SELECT checkpoint_id FROM writes WHERE thread_id = ?",
+        (str(thread_id),),
+    ) as cur:
+        return [row[0] async for row in cur]
+
+
+async def _add_checkpoint(saver, thread_id):
+    checkpoint = empty_checkpoint()
+    config = {"configurable": {"thread_id": str(thread_id), "checkpoint_ns": ""}}
+    await saver.aput(config, checkpoint, {}, {})
+    return checkpoint["id"]
+
+
+@pytest.fixture
+async def saver(tmp_path):
+    async with aiosqlite.connect(tmp_path / "checkpoints.db") as conn:
+        s = AsyncSqliteSaver(conn)
+        await s.setup()
+        yield s
+
+
+async def test_prune_keeps_newest_n_and_drops_orphaned_writes_only_for_pruned_thread(
+    saver,
+):
+    kept_ids = []
+    pruned_ids = []
+    for i in range(5):
+        cid = await _add_checkpoint(saver, "pruned-thread")
+        (pruned_ids if i < 3 else kept_ids).append(cid)
+        await saver.aput_writes(
+            {
+                "configurable": {
+                    "thread_id": "pruned-thread",
+                    "checkpoint_ns": "",
+                    "checkpoint_id": cid,
+                }
+            },
+            [("some_channel", "value")],
+            f"task-{i}",
+        )
+    other_id = await _add_checkpoint(saver, "other-thread")
+    await saver.aput_writes(
+        {
+            "configurable": {
+                "thread_id": "other-thread",
+                "checkpoint_ns": "",
+                "checkpoint_id": other_id,
+            }
+        },
+        [("some_channel", "value")],
+        "task-other",
+    )
+
+    await prune_thread_checkpoints(saver, "pruned-thread", keep=2)
+
+    remaining = await _checkpoint_ids(saver, "pruned-thread")
+    assert sorted(remaining) == sorted(kept_ids)
+    remaining_writes = await _write_rows(saver, "pruned-thread")
+    assert sorted(remaining_writes) == sorted(kept_ids)
+    for cid in pruned_ids:
+        assert cid not in remaining
+        assert cid not in remaining_writes
+
+    # Untouched: pruning one thread must not affect another.
+    assert await _checkpoint_ids(saver, "other-thread") == [other_id]
+    assert await _write_rows(saver, "other-thread") == [other_id]
+
+
+async def test_prune_preserves_full_latest_messages(saver):
+    def echo(state):
+        last = state["messages"][-1].content
+        return {"messages": [AIMessage(content=f"reply {last}")]}
+
+    graph = StateGraph(MessagesState)
+    graph.add_node("answer", echo)
+    graph.add_edge(START, "answer")
+    graph.add_edge("answer", END)
+    agent = graph.compile(checkpointer=saver)
+
+    thread_id = "many-turns"
+    config = {"configurable": {"thread_id": thread_id}}
+    turn_count = 25
+    for i in range(turn_count):
+        await agent.ainvoke(
+            {"messages": [HumanMessage(content=f"msg {i}")]}, config=config
+        )
+
+    assert len(await _checkpoint_ids(saver, thread_id)) > DEFAULT_KEEP_PER_THREAD
+
+    await prune_thread_checkpoints(saver, thread_id, keep=5)
+
+    assert len(await _checkpoint_ids(saver, thread_id)) == 5
+    state = await agent.aget_state(config)
+    # Every turn's human + AI message is still present: pruning dropped old
+    # checkpoint rows, but the newest row's channel_values already holds the
+    # full accumulated message list (not a delta).
+    assert len(state.values["messages"]) == turn_count * 2
+    assert state.values["messages"][0].content == "msg 0"
+    assert state.values["messages"][-1].content == f"reply msg {turn_count - 1}"
+
+
+@pytest.fixture
+async def conv_app(tmp_path):
+    await Tortoise.init(
+        db_url=f"sqlite://{tmp_path / 'conv.sqlite'}",
+        modules={
+            "models": ["blueprints.users.models", "blueprints.conversation.models"]
+        },
+    )
+    await Tortoise.generate_schemas()
+    app = Quart(__name__)
+    app.config.update(
+        TESTING=True,
+        JWT_SECRET_KEY="checkpoint-test-secret-at-least-thirty-two-characters",
+    )
+    JWTManager(app)
+    app.register_blueprint(conversation_blueprint)
+    user = await User.create(username="owner", email="owner@test")
+    async with app.app_context():
+        headers = {
+            "Authorization": f"Bearer {create_access_token(identity=str(user.id))}"
+        }
+    try:
+        yield app, headers, user
+    finally:
+        await Tortoise.close_connections()
+
+
+async def test_delete_conversation_via_api_removes_its_thread_and_leaves_others(
+    conv_app, tmp_path
+):
+    app, headers, user = conv_app
+    conv1 = await Conversation.create(user=user, name="one")
+    conv2 = await Conversation.create(user=user, name="two")
+
+    async with aiosqlite.connect(tmp_path / "checkpoints.db") as conn:
+        s = AsyncSqliteSaver(conn)
+        await s.setup()
+        await _add_checkpoint(s, str(conv1.id))
+        await _add_checkpoint(s, str(conv2.id))
+        app.main_agent = SimpleNamespace(checkpointer=s)
+
+        response = await app.test_client().delete(
+            f"/api/conversation/{conv1.id}", headers=headers
+        )
+        assert response.status_code == 200
+        assert (await response.get_json())["success"] is True
+        assert await Conversation.get_or_none(id=conv1.id) is None
+
+        assert await _checkpoint_ids(s, str(conv1.id)) == []
+        assert len(await _checkpoint_ids(s, str(conv2.id))) == 1
+
+
+async def test_delete_conversation_succeeds_even_when_checkpoint_delete_fails(
+    conv_app,
+):
+    app, headers, user = conv_app
+    conv = await Conversation.create(user=user, name="doomed")
+    app.main_agent = SimpleNamespace(
+        checkpointer=SimpleNamespace(
+            adelete_thread=AsyncMock(side_effect=RuntimeError("sqlite is gone"))
+        )
+    )
+
+    response = await app.test_client().delete(
+        f"/api/conversation/{conv.id}", headers=headers
+    )
+    assert response.status_code == 200
+    assert (await response.get_json())["success"] is True
+    assert await Conversation.get_or_none(id=conv.id) is None
+
+
+async def test_delete_thread_checkpoints_swallows_errors(caplog):
+    checkpointer = SimpleNamespace(
+        adelete_thread=AsyncMock(side_effect=RuntimeError("boom"))
+    )
+    await delete_thread_checkpoints(checkpointer, "t")
+    assert "Failed to delete checkpoint thread" in caplog.text
+
+
+@pytest.mark.parametrize("raw", ["0", "-5", "not-a-number"])
+def test_keep_per_thread_rejects_invalid_values(monkeypatch, caplog, raw):
+    monkeypatch.setenv("CHECKPOINT_KEEP_PER_THREAD", raw)
+    assert keep_per_thread() == DEFAULT_KEEP_PER_THREAD
+    assert "invalid" in caplog.text
+
+
+def test_keep_per_thread_accepts_valid_override(monkeypatch):
+    monkeypatch.setenv("CHECKPOINT_KEEP_PER_THREAD", "7")
+    assert keep_per_thread() == 7

tests/unit/test_configuration.py

@@ -63,6 +63,7 @@ def test_missing_required_settings(production_env, name):
         ("CHECKPOINT_DB_PATH", "/data/invalid\x00.db"),
         ("CONVERSATION_SUMMARY_DB_PATH", ""),
         ("OBSIDIAN_CONTINUOUS_SYNC", "yes"),
+        ("OBSIDIAN_SYNC_MODE", "push-only"),
     ],
 )
 def test_malformed_and_unsafe_settings(production_env, name, value):
@@ -202,6 +203,7 @@ def test_compose_forwards_runtime_settings():
         "CHECKPOINT_DB_PATH",
         "CONVERSATION_SUMMARY_DB_PATH",
         "OBSIDIAN_SYNC_INTERVAL",
+        "OBSIDIAN_SYNC_MODE",
         "OIDC_AUTHORIZATION_ENDPOINT",
         "OIDC_TOKEN_ENDPOINT",
         "OIDC_USERINFO_ENDPOINT",

tests/unit/test_context_trim.py

@@ -464,6 +464,64 @@ def test_focused_note_heading_index_uses_absolute_content_lines():
     assert msg.artifact["data"]["markdown"] == "full original"
 
 
+def _search_artifact(chunk_text: str) -> dict:
+    return {
+        "kind": "search.documents",
+        "title": "Search: vet",
+        "data": {
+            "query": "vet",
+            "source": "paperless",
+            "results": [
+                {
+                    "snippet": chunk_text[:50],
+                    "title": "vet_invoice",
+                    "source_id": "chunk-1",
+                    "url": "/sources/chunk-1",
+                }
+            ],
+        },
+        "model_context": {
+            "results": [
+                {
+                    "content": chunk_text,
+                    "title": "vet_invoice",
+                    "source_id": "chunk-1",
+                    "url": "/sources/chunk-1",
+                }
+            ]
+        },
+    }
+
+
+def test_search_result_compaction_keeps_full_chunk_text_under_a_generous_limit():
+    import json
+
+    chunk_text = "Rabies vaccine administered. " + "detail " * 200
+    msg = ToolMessage(
+        content="display", artifact=_search_artifact(chunk_text), tool_call_id="a"
+    )
+    projected = context.compact_tool_result(msg, 100_000)
+    parsed = json.loads(projected.content)
+    (result,) = parsed["data"]["results"]
+    assert result["content"] == chunk_text
+    assert "snippet" not in result
+
+
+def test_search_result_compaction_shortens_content_but_keeps_source_ids_under_pressure():
+    import json
+
+    chunk_text = "Rabies vaccine administered. " + "detail " * 200
+    msg = ToolMessage(
+        content="display", artifact=_search_artifact(chunk_text), tool_call_id="a"
+    )
+    projected = context.compact_tool_result(msg, 700)
+    parsed = json.loads(projected.content)
+    (result,) = parsed["data"]["results"]
+    assert len(result["content"]) < len(chunk_text)
+    assert result["source_id"] == "chunk-1"
+    assert result["url"] == "/sources/chunk-1"
+
+
 def _conversation(turns, size=1000):
     messages = []
     for i in range(turns):
@@ -542,3 +600,18 @@ def test_trim_drops_to_low_watermark_and_prefix_stays_stable():
     # ...so the retained prefix is unchanged for several turns in between.
     assert all(b - a >= 3 for a, b in zip(cuts, cuts[1:]))
     assert len(cuts) <= 5
+
+
+def test_search_result_compaction_keeps_source_ids_when_no_rows_fit():
+    import json
+
+    chunk_text = "Rabies vaccine administered. " + "detail " * 200
+    msg = ToolMessage(
+        content="display", artifact=_search_artifact(chunk_text), tool_call_id="a"
+    )
+    projected = json.loads(context.compact_tool_result(msg, 300).content)
+    data = projected["data"]
+    assert data["results"] == []
+    assert data["results_compaction"]["omitted_source_refs"] == [
+        {"source_id": "chunk-1", "url": "/sources/chunk-1"}
+    ]

tests/unit/test_file_upload.py

@@ -301,3 +301,45 @@ class TestStreamQueryFileKey:
                 f"conversations/{CONV_ID}/files/u.txt"
             )
         assert name == "u.txt"
+
+
+class TestImageOwnership:
+    async def test_presigns_only_keys_in_owned_conversations(self, app, headers, s3):
+        presign = AsyncMock(return_value="https://s3.example/signed")
+        with patch.object(conv, "s3_presigned_url", presign):
+            client = app.test_client()
+            own = await client.get(
+                f"/api/conversation/image/conversations/{CONV_ID}/a.jpg",
+                headers=headers,
+            )
+            assert own.status_code == 200
+            for key in (
+                f"conversations/{OTHER_ID}/a.jpg",
+                f"conversations/{CONV_ID}/../{OTHER_ID}/a.jpg",
+                "conversations/not-a-uuid/a.jpg",
+                "beli/profile.jpg",
+            ):
+                resp = await client.get(
+                    f"/api/conversation/image/{key}", headers=headers
+                )
+                assert resp.status_code == 404, key
+        presign.assert_awaited_once_with(f"conversations/{CONV_ID}/a.jpg")
+
+    async def test_image_upload_requires_owned_conversation(self, app, headers, s3):
+        with patch.object(conv.blueprints.users.models.User, "get", AsyncMock()):
+            resp = await self._upload_image(app, headers)
+        assert resp.status_code == 404
+        s3.upload_image.assert_not_awaited()
+
+    @staticmethod
+    async def _upload_image(app, headers):
+        return await app.test_client().post(
+            "/api/conversation/upload-image",
+            headers=headers,
+            form={"conversation_id": OTHER_ID},
+            files={
+                "file": FileStorage(
+                    io.BytesIO(_png_bytes()), filename="a.png", content_type="image/png"
+                )
+            },
+        )

tests/unit/test_obsidian_sync.py

@@ -0,0 +1,188 @@
+"""Tests for the incremental Obsidian sync: off-loop execution and delete SQL."""
+
+import json
+import os
+import threading
+from unittest.mock import patch
+
+import pytest
+
+from blueprints.rag import logic
+
+
+class FakeConnection:
+    def __init__(self, store):
+        self.store = store
+
+    def execute(self, stmt, params=None):
+        self.store.append((str(stmt), params or {}))
+
+    def commit(self):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        return False
+
+
+class FakeEngine:
+    def __init__(self):
+        self.executed: list[tuple[str, dict]] = []
+
+    def connect(self):
+        return FakeConnection(self.executed)
+
+
+class FakeVectorStore:
+    def __init__(self):
+        self.calls: list[list[str]] = []
+
+    async def aadd_documents(self, documents):
+        self.calls.append([d.page_content for d in documents])
+
+
+@pytest.fixture
+def vault_dir(tmp_path):
+    (tmp_path / "new.md").write_text("# New\nNew note body")
+    (tmp_path / "updated.md").write_text("# Updated\nUpdated note body")
+    with patch.dict(os.environ, {"OBSIDIAN_VAULT_PATH": str(tmp_path)}):
+        yield tmp_path
+
+
+@pytest.fixture
+async def synced(monkeypatch, vault_dir):
+    """Run a sync with a fake vault: one new file, one updated file, one
+    file that's been deleted off disk since the last sync."""
+    updated_path = str(vault_dir / "updated.md")
+    missing_path = str(vault_dir / "gone.md")
+
+    monkeypatch.setattr(
+        logic, "_obsidian_index_cache", {updated_path: 1.0, missing_path: 1.0}
+    )
+
+    engine = FakeEngine()
+    monkeypatch.setattr(logic, "_get_engine", lambda: engine)
+    monkeypatch.setattr(logic, "_get_collection_id", lambda: "collection-uuid")
+
+    def _unexpected_cache_load():
+        raise AssertionError("cache load should be skipped when already populated")
+
+    monkeypatch.setattr(logic, "_load_obsidian_index_cache", _unexpected_cache_load)
+
+    store = FakeVectorStore()
+    monkeypatch.setattr(logic, "_get_vector_store", lambda: store)
+
+    threads: dict[str, threading.Thread] = {}
+
+    real_walk = logic._walk_vault_mtimes
+
+    def recording_walk(service):
+        threads["walk"] = threading.current_thread()
+        return real_walk(service)
+
+    monkeypatch.setattr(logic, "_walk_vault_mtimes", recording_walk)
+
+    real_read = logic._read_obsidian_documents
+
+    def recording_read(service, filepaths, now):
+        threads["read"] = threading.current_thread()
+        return real_read(service, filepaths, now)
+
+    monkeypatch.setattr(logic, "_read_obsidian_documents", recording_read)
+
+    real_delete = logic.delete_documents_by_filepaths
+
+    def recording_delete(filepaths):
+        threads["delete"] = threading.current_thread()
+        return real_delete(filepaths)
+
+    monkeypatch.setattr(logic, "delete_documents_by_filepaths", recording_delete)
+
+    result = await logic.sync_obsidian_documents()
+    return result, threads, engine, store, updated_path, missing_path
+
+
+class TestSyncOffLoop:
+    async def test_counts_match_vault_changes(self, synced):
+        result, *_ = synced
+        assert result == {"added": 1, "updated": 1, "deleted": 1}
+
+    async def test_walk_runs_off_the_main_thread(self, synced):
+        _, threads, *_ = synced
+        assert threads["walk"] is not threading.main_thread()
+
+    async def test_reads_run_off_the_main_thread(self, synced):
+        _, threads, *_ = synced
+        assert threads["read"] is not threading.main_thread()
+
+    async def test_deletes_run_off_the_main_thread(self, synced):
+        _, threads, *_ = synced
+        assert threads["delete"] is not threading.main_thread()
+
+    async def test_embeds_new_and_updated_content(self, synced):
+        _, _, _, store, *_ = synced
+        embedded = {text for call in store.calls for text in call}
+        assert any("New note body" in t for t in embedded)
+        assert any("Updated note body" in t for t in embedded)
+
+    async def test_cache_reflects_add_update_delete(self, synced):
+        result, _, _, _, updated_path, missing_path = synced
+        assert missing_path not in logic._obsidian_index_cache
+        assert updated_path in logic._obsidian_index_cache
+
+    async def test_deletes_use_jsonb_containment_with_bound_params(self, synced):
+        _, _, engine, *_ = synced
+        deletes = [(sql, params) for sql, params in engine.executed if "DELETE" in sql]
+        assert deletes, "expected at least one delete statement"
+        for sql, params in deletes:
+            assert "cmetadata @> CAST(:filter AS jsonb)" in sql
+            assert "->>:" not in sql  # not the old, unindexed text-extraction filter
+            assert "filter" in params
+            parsed_filter = json.loads(params["filter"])
+            assert "filepath" in parsed_filter
+
+
+async def test_delete_failure_keeps_deleted_path_in_cache_and_propagates(
+    monkeypatch, vault_dir
+):
+    """If the batched delete raises, the sync must not have already forgotten
+    the deleted file: otherwise the next sync never retries the delete and
+    its chunks stay searchable forever with no record that they're stale."""
+    updated_path = str(vault_dir / "updated.md")
+    missing_path = str(vault_dir / "gone.md")
+
+    monkeypatch.setattr(
+        logic, "_obsidian_index_cache", {updated_path: 1.0, missing_path: 1.0}
+    )
+
+    def _raise_on_delete(filepaths):
+        raise RuntimeError("db unavailable")
+
+    monkeypatch.setattr(logic, "delete_documents_by_filepaths", _raise_on_delete)
+
+    store = FakeVectorStore()
+    monkeypatch.setattr(logic, "_get_vector_store", lambda: store)
+
+    with pytest.raises(RuntimeError, match="db unavailable"):
+        await logic.sync_obsidian_documents()
+
+    assert missing_path in logic._obsidian_index_cache
+    # The updated file's old chunks were never actually deleted, so its cache
+    # entry must not be bumped to look freshly indexed.
+    assert logic._obsidian_index_cache[updated_path] == 1.0
+    assert store.calls == []
+
+
+async def test_delete_documents_by_metadata_uses_containment(monkeypatch):
+    engine = FakeEngine()
+    monkeypatch.setattr(logic, "_get_engine", lambda: engine)
+    monkeypatch.setattr(logic, "_get_collection_id", lambda: "collection-uuid")
+
+    logic.delete_documents_by_metadata("source", "obsidian")
+
+    assert len(engine.executed) == 1
+    sql, params = engine.executed[0]
+    assert "cmetadata @> CAST(:filter AS jsonb)" in sql
+    assert json.loads(params["filter"]) == {"source": "obsidian"}

tests/unit/test_retrieval.py

@@ -94,6 +94,32 @@ def test_artifact_citations_identify_exact_retrieved_chunk(monkeypatch):
     assert result["headings"] == ["Budget", "Savings"]
 
 
+async def test_serialized_chunk_header_is_compact_not_raw_metadata(monkeypatch):
+    identifier = str(uuid4())
+    doc = Document(
+        id=identifier,
+        page_content="Rabies vaccine administered",
+        metadata={
+            "source": "paperless",
+            "filename": "vet_invoice.pdf",
+            "created_date": datetime(2026, 9, 1, tzinfo=timezone.utc).timestamp(),
+            "document_type": "Invoice",
+        },
+    )
+    store = SimpleNamespace(
+        amax_marginal_relevance_search=AsyncMock(return_value=[doc])
+    )
+    monkeypatch.setattr(logic, "_get_vector_store", lambda: store)
+    text, _ = await logic.query_vector_store("vet", source="paperless")
+    assert "{'source'" not in text
+    assert str(doc.metadata) not in text
+    assert "Title: vet_invoice" in text
+    assert "Date: 2026-09-01" in text
+    assert "Source: paperless" in text
+    assert f"Citation: /sources/{identifier}" in text
+    assert "Content: Rabies vaccine administered" in text
+
+
 def test_document_date_and_unknown_citations():
     assert (
         document_date(

tests/unit/test_tool_artifacts.py

@@ -521,6 +521,13 @@ async def test_simba_search(monkeypatch):
     assert result["snippet"].startswith("Rabies vaccine administered. ")
     assert len(result["snippet"]) <= SNIPPET_MAX_CHARS
 
+    (model_result,) = art["model_context"]["results"]
+    assert model_result["content"] == PAPERLESS_DOCS[0].page_content
+    assert len(model_result["content"]) > SNIPPET_MAX_CHARS
+    assert model_result["title"] == result["title"]
+    assert model_result["filename"] == result["filename"]
+    assert "snippet" not in model_result
+
 
 async def test_simba_search_empty(monkeypatch):
     monkeypatch.setattr(agents, "query_vector_store", AsyncMock(return_value=("", [])))
@@ -560,6 +567,14 @@ async def test_obsidian_search_notes(obsidian, monkeypatch):
             "date": "2026-01-02",
         }
     ]
+    assert art["model_context"]["results"] == [
+        {
+            "content": "Simba's vet is Dr Smith.",
+            "title": "simba",
+            "path": "pets/simba.md",
+            "date": "2026-01-02",
+        }
+    ]
 
 
 async def test_obsidian_search_notes_error(obsidian, monkeypatch):