Merge pull request 'Add Immich MCP technical design' (#90) from docs/immich-mcp-design into main

Design for an in-repo stdio MCP server that lets Simba (and Claude Code)
find existing albums, summarize trips by date clusters, and create or
fill albums behind a write flag and a confirmed-count check.

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/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
@@ -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 = []
@@ -507,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()
@@ -514,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:
 

docs/immich-mcp/design.md

@@ -0,0 +1,432 @@
+# Immich MCP: technical design
+
+Status: proposed · Owner: Ryan · Date: 2026-10-13
+
+## 1. Summary
+
+Give Simba (and Claude Code) a small set of tools for the household Immich
+library so a request like *"my Maine trip, here are the photos"* turns into
+either a link to the album that already exists or a confirmed, one-step
+album creation. The tools live in a stdio MCP server we write and keep in
+this repository (`immich_mcp/`). raggr loads it the same way it loads
+Homebox and BookStack; Claude Code can register the same server with
+`claude mcp add`.
+
+Work ships in two phases:
+
+- **Phase 1, read-only.** Find albums, search photos by date, place and person,
+  CLIP search, and spot photos missing from an existing album.
+- **Phase 2, writes.** Create an album and add photos to an album, behind a
+  server-side switch (`IMMICH_ENABLE_WRITES=true`) and a count check that
+  stops the model from writing something the user did not confirm.
+
+There are no delete, archive, trash, metadata-edit or sharing tools in either phase.
+
+## 2. Goals and non-goals
+
+### Goals
+
+1. **"Do I already have an album for X?"** returns the album name, photo count,
+   date span and a clickable link, in one tool call.
+2. **"Photos from my Maine trip"** with no album returns a summary the user can
+   check (how many, which days, which towns, a few sample filenames) without
+   moving hundreds of asset IDs through the model's context.
+3. **"Make an album for it"** creates the album with exactly the photos the user
+   saw in that summary, or does nothing and explains why.
+4. **"Is the Maine album complete?"** lists photos taken during the album's dates
+   that are not in it, such as camera imports or a partner's phone, and can add them after confirmation.
+5. The same server works from Claude Code on the laptop.
+6. Tool results stay small enough for the local Qwen model's 32k context: each
+   result is at most about 2,500 characters, and all six tool schemas together
+   cost at most about 1,500 tokens.
+
+### Non-goals
+
+- Deleting, trashing, archiving, hiding, or editing assets or albums. Removing photos from an album is out too.
+- Showing images inline in chat. The desktop client's `photo` artifact
+  (`docs/desktop-client.md`) can build on this later using the asset links
+  these tools return.
+- Shared links, inviting album users, face naming, tagging.
+- Replacing `scripts/index_immich.py`, the one-off script that describes Simba photos into the
+  vector store. It is left untouched; retiring it is an open question (§13).
+- Non-admin access. Photos are household-private, so like Homebox and BookStack
+  these tools exist only on the admin agent.
+
+## 3. Example conversations (acceptance scenarios)
+
+These double as the manual verification script for the final ticket.
+
+**A. The album already exists**
+
+> **User:** my maine trip, here are the photos
+> **Simba** → `immich_find_albums(query="maine")`
+> → `[{"name": "Maine 2026", "asset_count": 412, "start": "2026-08-02", "end": "2026-08-09", "url": "https://photos.example/albums/…"}]`
+> **Simba:** You already have **Maine 2026**: 412 photos, Aug 2–9. [link]
+
+**B. No album yet**
+
+> **User:** find my maine trip photos
+> `immich_find_albums("maine")` → `[]`
+> `immich_search_photos(state="Maine")` → total 398; clusters: Aug 2–9 2026 (391), Jul 14 2023 (7); top places Bar Harbor, Portland
+> **Simba:** I found 391 photos from Aug 2–9, 2026, mostly Bar Harbor and Portland, plus 7 from a day in July 2023. Want an album for the August trip?
+> **User:** yes, call it Maine 2026
+> `immich_search_photos(taken_after="2026-08-02", taken_before="2026-08-09")` → total 455 (this adds the drive up and photos with no GPS)
+> **Simba:** Counting everything from Aug 2–9, not only photos tagged in Maine, gives 455 (64 have no location or were taken on the way). Create **Maine 2026** with all 455?
+> **User:** yes
+> `immich_create_album(name="Maine 2026", criteria={taken_after, taken_before}, expected_count=455)` → created, 455 added, link
+
+**C. The album exists but is incomplete**
+
+> **User:** is the maine album missing anything?
+> `immich_find_albums("maine")` → Maine 2026, Aug 2–9
+> `immich_album_gaps(album_id=…)` → 37 photos taken Aug 1–10 are not in the album (mostly Aug 5, Canon EOS R6)
+> **Simba:** 37 photos from those days aren't in it, mostly camera shots from Aug 5. Add them?
+> `immich_add_to_album(album_id, criteria={taken_after: 2026-08-01, taken_before: 2026-08-10}, expected_new_count=37)`
+
+**D. Content search**
+
+> **User:** pictures of simba in the snow
+> `immich_smart_search(query="orange cat in snow", person="Simba")` → top 10, with dates and links
+
+## 4. Approach and alternatives
+
+| Option | Pros | Cons |
+|---|---|---|
+| **A. Our own stdio MCP server in this repo (chosen)** | One implementation serves raggr and Claude Code; we control tool names, descriptions and output size; write safety is enforced in code we own; no new runtime dependency (`mcp` 1.30 is already locked through `langchain-mcp-adapters`); fits the Homebox/BookStack loading pattern | We maintain it (≈500 lines plus tests) |
+| B. Third-party Immich MCP server (several exist: lidless-labs `immich-mcp`, `barryw/ImmichMCP`, `okopok/immich-mcp`, …), pinned and allowlisted like Homebox/BookStack | No code to write | Not evaluated in depth. Generic API-wrapping servers return per-asset JSON (a 400-photo trip won't fit in 32k) and add assets by ID list (the model would have to copy hundreds of UUIDs). Our key, which needs album write permissions, would go to a young third-party package. Tool descriptions we can't edit cost context. |
+| C. Plain LangChain tools in `utils/immich_service.py` (like Mealie/YNAB) | Simplest wiring | Simba only; Claude Code would need a second implementation |
+
+Option A keeps option C's control and adds reuse from Claude Code. Most of
+the work goes into compact summaries and write guards, and no off-the-shelf
+server gives us either.
+
+## 5. Architecture
+
+```
+            ┌───────────── raggr (admin agent) ─────────────┐
+            │ app.py lifespan                                │
+            │   load_immich_tools() ─┐                       │
+            │                        │ langchain-mcp-adapters│
+            │   wrap_admin_only()    │ (new stdio session    │
+            │   prefix immich_*      │  per tool call)       │
+            └────────────────────────┼───────────────────────┘
+                                     │ stdio
+ Claude Code ── claude mcp add ──────┤
+                                     ▼
+            ┌──────── immich_mcp (python -m immich_mcp) ─────┐
+            │ server.py   FastMCP tools (6)                  │
+            │ criteria.py PhotoCriteria → Immich search body │
+            │ summarize.py clusters / places / samples       │
+            │ client.py   ImmichClient (httpx, method+path   │
+            │             allowlist, no DELETE)              │
+            └────────────────────────┬───────────────────────┘
+                                     │ HTTPS, x-api-key
+                                     ▼
+                              Immich server (/api)
+```
+
+**Process lifetime.** `MultiServerMCPClient.get_tools()` without a session
+creates tools that open a **new stdio session for every call**
+(`langchain_mcp_adapters/tools.py`). The server therefore must be
+**stateless between calls**: no in-memory selections, caches or tokens. Write
+tools take the same *criteria* as the preview and resolve them again, and
+the `expected_count` check (§8) catches any drift. Starting a process costs
+roughly 0.5–1 s per call. That is acceptable for this use, and IMM-04 measures it.
+
+## 6. Package layout
+
+New top-level package (copied into the image by a new `COPY immich_mcp ./immich_mcp`
+line, because the Dockerfile copies directories explicitly):
+
+```
+immich_mcp/
+  __init__.py        # docstring only
+  __main__.py        # `python -m immich_mcp` → server.main()
+  config.py          # env parsing (§10), ImmichConfig dataclass
+  client.py          # ImmichClient: async httpx, allowlisted endpoints only
+  criteria.py        # PhotoCriteria (pydantic), date/time-zone handling, request bodies
+  summarize.py       # pure functions: clusters, places, samples, histogram
+  server.py          # FastMCP("immich"), tool definitions, JSON output shaping
+tests/unit/
+  test_immich_client.py
+  test_immich_criteria.py
+  test_immich_summarize.py
+  test_immich_server.py      # tools via an in-memory MCP session
+  test_immich_mcp.py         # raggr loader + admin gate (mirrors test_homebox_mcp.py)
+```
+
+raggr-side changes:
+
+- `utils/mcp_tools.py`: `load_stdio_tools(..., args: list[str] | None = None,
+  cwd: str | None = None)`. Today `args` is hard-coded to `[]` and no `cwd` is passed.
+  `StdioConnection` already supports both. Homebox and BookStack keep the defaults.
+- `utils/immich_mcp.py`: `IMMICH_TOOL_ALLOWLIST`, `select_immich_tools`,
+  `load_immich_tools()` (mirrors `utils/homebox_mcp.py`). The command defaults to
+  `sys.executable` (the app's venv interpreter) with `args=["-m", "immich_mcp"]`
+  and `cwd` set to the repo root (`Path(__file__).resolve().parents[1]`), so
+  `-m` finds the package whatever the parent's working directory is. The subprocess
+  env is only the `IMMICH_*` variables plus `PATH` (`load_stdio_tools` behaviour),
+  so no other secrets leak into it. `IMMICH_MCP_COMMAND` overrides the
+  command for debugging. Note the two names: `utils/immich_mcp.py` is raggr's
+  *loader*, and top-level `immich_mcp/` is the *server*. Neither imports the other.
+- `app.py`: add `await load_immich_tools()` to `admin_tools`.
+- `blueprints/conversation/tool_routing.py`: add a `"photos"` group with the terms
+  `immich photo photos picture pictures pic pics album albums trip trips vacation`,
+  map the prefix `("immich_", "photos")`, and add `"immich"` to the admin prefix
+  tuple in `ToolRouter.__init__`.
+- `blueprints/conversation/tool_labels.py`: `("immich_", "Looking through photos…")`.
+- `raggr-frontend/src/components/ActivityTimeline.tsx`: `["immich", Images]` (lucide).
+- `blueprints/conversation/prompts.py`: one sentence in INTEGRATIONS (§9).
+- `blueprints/integrations/__init__.py`: admin row `("Immich", ("IMMICH_URL", "IMMICH_API_KEY"))`.
+- `config/validation.py`: `IMMICH_URL` and `IMMICH_PUBLIC_URL` in the URL list;
+  `IMMICH_ENABLE_WRITES` in the true/false list.
+- `.env.example`, `docker-compose.yml`: env vars (§10). `IMMICH_URL` and
+  `IMMICH_API_KEY` already exist in `.env.example` for the indexer, so reuse them.
+
+## 7. Tool specifications
+
+Every tool returns **compact JSON** (`json.dumps(..., separators=(",", ":"),
+ensure_ascii=False)`), at most `MAX_RESULT_CHARS = 2500`. Dates in output are
+calendar dates (`YYYY-MM-DD`): a photo's date is its own local day (§7.4), and
+album `start`/`end` (UTC timestamps in Immich) are converted to
+`IMMICH_TIMEZONE`. If a result would go over the cap, lists are cut down (samples
+first, then places, then clusters) and `"trimmed": true` is set; the result is
+never cut mid-JSON. Links are built
+from `IMMICH_PUBLIC_URL`: albums at `/albums/{id}`, photos at `/photos/{id}`.
+Errors come back as `{"error": "<plain sentence>"}` and never raise, so the
+model can tell the user what went wrong.
+
+### 7.1 Shared input: `PhotoCriteria`
+
+| Field | Type | Meaning |
+|---|---|---|
+| `taken_after` | `YYYY-MM-DD` | inclusive, local day start |
+| `taken_before` | `YYYY-MM-DD` | **inclusive**, local day end (converted to the next day 00:00 local, exclusive) |
+| `state` | str | reverse-geocoded state/province, e.g. "Maine" |
+| `city` | str | e.g. "Bar Harbor" |
+| `country` | str | e.g. "United States of America" |
+| `person` | str | Immich person name, resolved via `GET /search/person?name=`; several matches → error that lists them |
+| `media` | `"photo" \| "video" \| "all"` | default `"all"` |
+| `favorites_only` | bool | default false |
+
+**The fields are flat tool parameters, never a nested object.** Every tool that
+takes criteria declares these eight parameters directly. A nested `criteria`
+object would produce `$defs`/`$ref` in the tool schema, which the local model's
+llama-server grammar handles poorly, and it costs more schema tokens. In the
+examples in §3, `criteria={…}` is shorthand for passing these parameters.
+
+For `immich_search_photos` and both write tools, at least one field must be
+set, so a request can never match the whole library. (`immich_smart_search` is
+already scoped by its query.)
+`taken_before < taken_after` is an error. Place names are matched
+case-insensitively: before searching, the server looks the value up in
+`GET /search/suggestions?type=state|city|country` and uses Immich's exact
+spelling. If nothing matches, it returns the three closest suggestions
+(`difflib.get_close_matches`) instead of searching.
+
+### 7.2 Phase 1: read-only
+
+**`immich_find_albums(query: str = "", limit: int = 10)`**
+Lists albums (`GET /albums`, both owned and shared-with-me) and ranks them by
+case-insensitive substring match, then `difflib` ratio ≥ 0.6. An empty query
+returns the most recently updated. Output per album: `id`, `name`,
+`asset_count`, `start`, `end`, `shared`, `owned` (false for albums shared with
+the key's user; adding to those may fail), `url`. Why match locally: Immich's `name` filter
+is an exact match, and "maine" should find "Maine 2026" and "Acadia / Maine".
+
+**`immich_search_photos(criteria…, samples: int = 5)`**
+1. `POST /search/statistics` gets the exact `total`.
+2. Pages `POST /search/metadata` (size 1000, `withExif: true`) up to
+   `MAX_SCAN = 5000` assets.
+3. Returns `total`, `scanned`, `truncated`, `first`, `last`,
+   `clusters` (§7.4, at most 8), `top_places` (city/state with counts, at most 5),
+   `no_location` (count with no city/state), `by_media` (photo/video counts),
+   `samples` (evenly spaced across the scan: date, city, filename, url), and
+   `criteria` (the normalized criteria, echoed so the model can pass it
+   straight to a write tool).
+
+**`immich_smart_search(query: str, criteria… (all optional), limit: int = 10)`**
+`POST /search/smart` (CLIP), with the same filters. Returns the top `limit` (at most 25)
+results in rank order: date, city, filename, url. There is no total, because
+smart search ranks results rather than filtering them. Use it to find content
+("in the snow", "sunset at the beach"), not to collect a trip.
+
+**`immich_album_gaps(album_id: str, pad_days: int = 1)`**
+Loads the album (`GET /albums/{id}` for `startDate`/`endDate`). It then collects
+the IDs already in the album (`POST /search/metadata` with `albumIds=[id]`,
+paged) and the photos in `[start − pad, end + pad]` (local dates), and returns the
+search summary from §7.2 computed over the photos **not** in the album. It also
+returns `criteria` for that window and `missing_count`, ready for
+`immich_add_to_album`. Returns an error if the album has no dates (it's empty).
+`missing_count` must be computed exactly as `immich_add_to_album` computes
+`expected_new_count`; both go through one shared helper,
+`resolve_missing(album_id, criteria)`. If the window has more than `MAX_SCAN`
+photos, `missing_count` is marked `truncated`, and a later add with that
+count fails the check. That's the safe direction.
+
+### 7.3 Phase 2: writes (only registered when `IMMICH_ENABLE_WRITES=true`)
+
+**`immich_create_album(name: str, expected_count: int, <criteria fields>, description: str = "")`**
+1. If any album already has this exact name (case-insensitive), return an error that names it
+   and suggests `immich_add_to_album`. Never create a duplicate.
+2. Resolve the criteria to asset IDs (full paging, no `MAX_SCAN` limit, but
+   refuse if `total > MAX_WRITE = 2000`).
+3. If `len(ids) != expected_count`, return `{"error": …, "actual_count": n}`
+   and write nothing.
+4. `POST /albums` with `albumName`, `description`, and `assetIds` for the first
+   500. Add the rest with `PUT /albums/{id}/assets` in chunks of 500.
+5. Return `id`, `name`, `added`, `url`.
+
+**`immich_add_to_album(album_id: str, expected_new_count: int, <criteria fields>)`**
+1. Resolve the criteria to IDs (same `MAX_WRITE` limit) and subtract the IDs already in the album.
+2. If the count of new IDs differs from `expected_new_count`, return an error and write nothing.
+3. `PUT /albums/{id}/assets` in chunks of 500. Count per-ID results from
+   `BulkIdResponseDto`: `success`, `duplicate`, other errors (with reasons).
+4. Return `added`, `already_present`, `failed` (at most 5 reasons), `url`.
+
+### 7.4 Trip clustering (`summarize.cluster_by_day`)
+
+Group scanned assets by the photographer's local day: the date part of
+`localDateTime`, taken **as-is**. Immich stores it as a "timezone-agnostic
+timestamp" of the wall-clock time, so converting it would shift days. If
+`localDateTime` is missing, fall back to `fileCreatedAt` converted to
+`IMMICH_TIMEZONE`. Sort the
+days and start a new cluster wherever the gap between consecutive days is more than
+`gap_days = 2` (one quiet day in the middle of a trip does not split it). Each
+cluster reports `start`, `end`, `count`, and its top city/state. Clusters are sorted
+by count descending; keep 8. This is how "my Maine trip" becomes
+"Aug 2–9, 2026" rather than "every photo ever taken in Maine". It's a pure function,
+tested with hand-built fixtures.
+
+## 8. Write safety
+
+These layers stack; any one of them blocks an unwanted write.
+
+1. **Not registered by default.** Write tools are defined only when
+   `IMMICH_ENABLE_WRITES=true`. Phase 1 can be deployed and used before
+   anyone decides to allow writes.
+2. **Admin-only.** The tools are compiled only into `app.admin_agent`, and each call
+   re-checks `is_admin` through `wrap_admin_only` (`utils/mcp_tools.py`).
+3. **Confirmed count.** `expected_count` / `expected_new_count` must equal the
+   number of photos the server resolves at write time. The model can only know the
+   number by having shown a search result, and the prompt (§9) tells it to get a yes
+   from the user first. If the library changed in between, the check fails.
+4. **Bounded size.** `MAX_WRITE = 2000` per call.
+5. **No destructive endpoints in the client.** `ImmichClient` exposes named
+   methods only, and its request helper accepts only the methods and path patterns
+   in `ALLOWED_ENDPOINTS`. A unit test asserts that no `DELETE`, `PATCH`, or
+   `/trash`, `/assets` (bulk) write is reachable.
+6. **A scoped API key.** Immich API keys carry permissions. The key needs only
+   `album.read`, `asset.read`, `asset.statistics`, `person.read`, and for Phase 2
+   `album.create` and `albumAsset.create`. Do not give it `all`.
+7. **No duplicate names.** See `immich_create_album` step 1.
+8. **Writes add and never remove.** At worst a write adds wrong photos to an
+   album. That is fixable in the Immich UI and loses no data.
+
+## 9. Prompt and routing
+
+Append to INTEGRATIONS in `blueprints/conversation/prompts.py`:
+
+> If immich_* tools are available, use them for photos and albums: call
+> immich_find_albums before searching or creating; for a trip, search by place,
+> read the date clusters, then search the whole date range so photos without
+> GPS are included; show the count and dates and get an explicit yes before
+> immich_create_album or immich_add_to_album, passing the confirmed count.
+
+Routing: the `"photos"` group (§6) makes these tools discoverable in
+`TOOL_ROUTING_MODE=dynamic`. In static mode they are always bound for admins, like
+Homebox.
+
+## 10. Configuration
+
+| Variable | Required | Default | Notes |
+|---|---|---|---|
+| `IMMICH_URL` | yes | – | Base URL the **server** uses (LAN/compose network). Already in `.env.example`. |
+| `IMMICH_API_KEY` | yes | – | Scoped key (§8.6). Already in `.env.example`. |
+| `IMMICH_PUBLIC_URL` | no | `IMMICH_URL` | Base for links shown to the user. |
+| `IMMICH_TIMEZONE` | no | `TZ`, then `America/New_York` | Local-day interpretation (§7.1, §7.4). |
+| `IMMICH_ENABLE_WRITES` | no | `false` | Registers the Phase 2 tools. |
+| `IMMICH_MCP_COMMAND` | no | `sys.executable -m immich_mcp` | Override for debugging. |
+
+All of them go in `.env`, `.env.example` and the `raggr` service environment in
+`docker-compose.yml` (project rule). If `IMMICH_URL` or `IMMICH_API_KEY` is unset,
+`load_immich_tools()` logs at info level and returns `[]`, like Homebox.
+
+Claude Code registration (documented in IMM-07):
+
+```bash
+claude mcp add immich \
+  -e IMMICH_URL=http://192.168.1.5:2283 -e IMMICH_API_KEY=… \
+  -e IMMICH_PUBLIC_URL=https://photos.… -e IMMICH_ENABLE_WRITES=true \
+  -- uv run --directory ~/Programs/raggr python -m immich_mcp
+```
+
+## 11. Immich API compatibility
+
+Verified against the upstream OpenAPI spec at `3.3.0-rc.0`:
+
+- Endpoints used: `GET /server/version`, `GET /albums`, `GET /albums/{id}`,
+  `POST /albums`, `PUT /albums/{id}/assets`, `POST /search/metadata`,
+  `POST /search/smart`, `POST /search/statistics`, `GET /search/suggestions`,
+  `GET /search/person`.
+- **Deprecation.** In v3.2.0 Immich deprecated the flat search fields
+  (`takenAfter`, `takenBefore`, `state`, `city`, `country`, `personIds`, `albumIds`,
+  `type`, `page`, …) in favour of a nested `filter` object (`SearchFilter`, with
+  `DateFilter {gte, lt}`, `StringFilterNullable {eq, in}`, `IdsFilter {any, all, none}`)
+  and `cursor` paging. The flat fields still work in 3.3, and they are the
+  only option before 3.2. Mapping: `takenAfter/takenBefore` → `filter.takenAt
+  {gte, lt}`, `state/city/country` → `{eq}`, `personIds` → `{all}`, `type` →
+  `{eq}`, `isFavorite` → `{eq: true}`, `albumIds` → `{any}`, and in `album_gaps`, `{none}`.
+- **Decision.** Build request bodies in a single function,
+  `criteria.to_search_body(criteria, version)`. Use the flat fields below 3.2 and
+  `filter`/`cursor` from 3.2 on, choosing by `GET /server/version` (read once per
+  process; a process lives for one call, so this is one extra request per call, or
+  `IMMICH_API_STYLE=legacy|filter` to skip it). Golden-body tests pin both shapes.
+  The `filter` style also gives `album_gaps` a server-side `albumIds: {none: [id]}`,
+  so it doesn't need to diff locally.
+- Production's Immich version is not yet known (it was unreachable from the design
+  machine). IMM-01 records it. If it is below 2.0, stop and revisit this design.
+
+## 12. Testing
+
+- **Pure units:** criteria validation, local-day to UTC conversion
+  (DST edges, inclusive `taken_before`), request-body golden tests for both API
+  styles, clustering fixtures (single trip, quiet day inside, two trips, empty),
+  sample spacing, output caps (a large fixture stays at most 2500 chars).
+- **Client:** `httpx.MockTransport`. Checks the endpoint allowlist (DELETE raises),
+  paging stops at `MAX_SCAN`, auth header, and that HTTP errors map to readable error strings.
+- **Tools:** call the FastMCP server in-process with
+  `mcp.shared.memory.create_connected_server_and_client_session` against the mock
+  transport. Covers scenarios A–D from §3, the `expected_count` mismatch (no write
+  happens), the duplicate-name refusal, write tools absent when the flag is off, and the
+  schema token budget (sum of `len(json.dumps(schema)) / 4` is at most 1500).
+- **raggr wiring:** mirror `tests/unit/test_homebox_mcp.py` (allowlist, admin
+  gate, missing env returns `[]`, prefix), plus `test_tool_routing.py` cases for the
+  `photos` group and admin hiding.
+- **Live (manual, IMM-08):** scenarios A–D against production with a scoped key.
+
+## 13. Open questions
+
+1. **Production Immich version** and whether `/search/statistics` and the
+   `filter` style are available (IMM-01).
+2. **Partner photos.** Immich search covers partners with "show in timeline"
+   enabled. Can the key's user add a partner's asset to their own album? If not,
+   `add` returns per-ID errors, which we report. Verify in IMM-01.
+3. **Time zone of typed dates.** Clustering uses each photo's own local day
+   (§7.4), but `taken_after`/`taken_before` filter on UTC `takenAt`, interpreted
+   in `IMMICH_TIMEZONE`. For a trip three hours west, a photo at 11 pm local
+   on the last day falls outside the window. `album_gaps` pads by a day, and
+   users can widen a range. A precise fix (filtering on `localDateTime`) is not
+   in the API. Accepted for now.
+4. **Retire `scripts/index_immich.py`?** Once smart search is live, the vector-store
+   copy of Simba photo descriptions may no longer be needed. Decide after Phase 1 has run for a while.
+5. **Sharing.** Should "send the Maine album to Mom" create a shared link? Not now;
+   it would need `sharedLink.create` and a separate design.
+
+## 14. Rollout
+
+1. Merge Phase 1 with `IMMICH_ENABLE_WRITES` unset. Deploy and run scenarios A, B (read
+   part), C (read part) and D.
+2. Create the Phase 2 key permissions, set `IMMICH_ENABLE_WRITES=true`, redeploy,
+   and run B and C end to end on a throwaway album name. Delete the test album by hand in Immich.
+3. Register the server in Claude Code.

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 ""

mkdocs.yml

@@ -25,3 +25,4 @@ nav:
       - Authentication: authentication.md
       - Desktop Client (proposal): desktop-client.md
       - Cowork UI plan: cowork-ui-plan.md
+      - Immich MCP (design): immich-mcp/design.md

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_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_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"}