Create scheduled_messages for databases that never had it

No committed migration creates the table: migrations 6-10 were generated
outside the repository, so fresh databases fail once anything alters it.
The new migration is CREATE TABLE IF NOT EXISTS, a no-op where the table
already exists. Aerich cannot generate it because its saved state already
includes the table; the DDL is Tortoise's schema output for the model.

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

.env.example

@@ -32,6 +32,9 @@ TTFT_TIMEOUT_SECONDS=30
 CONTEXT_WINDOW_TOKENS=32768
 PROMPT_BUDGET_TOKENS=16000
 RESPONSE_RESERVE_TOKENS=4096
+# static: bind every role-appropriate tool on each call (stable, cacheable prompt
+# prefix). dynamic: keyword-route tool groups per call with a discovery tool.
+TOOL_ROUTING_MODE=static
 # Optional durable summary sidecar; defaults to CHECKPOINT_DB_PATH + .summaries.db
 # Store on the same persistent shared volume as the checkpoints.
 # CONVERSATION_SUMMARY_DB_PATH=/app/checkpoints/conversation-summaries.db
@@ -72,8 +75,9 @@ OIDC_USE_DISCOVERY=true
 YNAB_ACCESS_TOKEN=your-ynab-personal-access-token
 # Optional: Specify a budget ID, or leave empty to use the default/first budget
 YNAB_BUDGET_ID=
-# Seconds between background YNAB delta syncs (agent tools also sync on demand)
-YNAB_SYNC_INTERVAL=900
+# Seconds between background YNAB delta syncs. Keep under the agent tools' 120s
+# freshness window so tools rarely wait on an inline YNAB fetch.
+YNAB_SYNC_INTERVAL=90
 
 # Mealie Configuration
 # Base URL for your Mealie instance (e.g., http://192.168.1.5:9000 or https://mealie.example.com)

blueprints/artifacts/models.py

@@ -8,13 +8,17 @@ class Artifact(Model):
     id = fields.UUIDField(primary_key=True)
     user = fields.ForeignKeyField("models.User", related_name="artifacts")
     conversation = fields.ForeignKeyField(
-        "models.Conversation", related_name="artifacts", on_delete=fields.CASCADE
+        "models.Conversation",
+        related_name="artifacts",
+        on_delete=fields.CASCADE,
+        db_index=True,
     )
     message = fields.ForeignKeyField(
         "models.ConversationMessage",
         related_name="artifacts",
         null=True,
         on_delete=fields.SET_NULL,
+        db_index=True,
     )
     tool = fields.CharField(max_length=255)
     kind = fields.CharField(max_length=64)

blueprints/conversation/__init__.py

@@ -75,6 +75,7 @@ def _model_call_summary(call: dict) -> dict:
         "prefill_ms": round((first_chunk - start) * 1000) if first_chunk else None,
         "ms": round((end - start) * 1000) if end else None,
         "prompt_tokens": call.get("prompt_tokens"),
+        "cached_prompt_tokens": call.get("cached_prompt_tokens"),
         "output_tokens": call.get("output_tokens"),
         "model": call.get("model"),
         "provider": call.get("provider"),
@@ -257,7 +258,9 @@ async def upload_image():
     content_type = file.content_type or "image/jpeg"
 
     try:
-        processed_bytes, output_content_type = process_image(file_bytes, content_type)
+        processed_bytes, output_content_type = await asyncio.to_thread(
+            process_image, file_bytes, content_type
+        )
     except ImageValidationError as e:
         return jsonify({"error": str(e)}), 400
 
@@ -311,8 +314,8 @@ async def upload_attachment():
 
     if content_type in IMAGE_TYPES:
         try:
-            processed_bytes, output_content_type = process_image(
-                file_bytes, content_type
+            processed_bytes, output_content_type = await asyncio.to_thread(
+                process_image, file_bytes, content_type
             )
         except Exception as e:
             return jsonify({"error": f"Could not process image: {e}"}), 400
@@ -402,7 +405,7 @@ async def stream_query():
     if image_key:
         try:
             image_bytes, _ = await s3_get_image(image_key)
-            image_block = image_content_block(image_bytes)
+            image_block = await asyncio.to_thread(image_content_block, image_bytes)
         except Exception as e:
             logging.error(f"Failed to load image {image_key}: {e}")
             query_text = f"{query_text or ''}\n\n[An image was sent but could not be loaded.]".strip()
@@ -600,6 +603,10 @@ async def stream_query():
                         if usage:
                             call["prompt_tokens"] = usage.get("input_tokens")
                             call["output_tokens"] = usage.get("output_tokens")
+                            # Prompt-cache hits, when the provider reports them.
+                            call["cached_prompt_tokens"] = (
+                                usage.get("input_token_details") or {}
+                            ).get("cache_read")
                             call["reasoning_tokens"] = (
                                 usage.get("output_token_details") or {}
                             ).get("reasoning")

blueprints/conversation/agents.py

@@ -29,7 +29,7 @@ from blueprints.conversation.artifacts_util import (
 from blueprints.conversation.context import RESPONSE_RESERVE_TOKENS, build_context
 from blueprints.conversation.memory import save_memory
 from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
-from blueprints.conversation.tool_routing import ToolRouter
+from blueprints.conversation.tool_routing import TOOL_ROUTING_MODES, ToolRouter
 from blueprints.rag.logic import query_vector_store
 from blueprints.ynab.sync import ynab_configured
 from blueprints.ynab.tools import YNAB_TOOLS
@@ -1160,16 +1160,24 @@ def build_agent(checkpointer=None, extra_tools=()):
     """
     extra_tools = list(extra_tools)
     # Keep the guarded Google tools executable for admins even when no MCP tools
-    # are configured and app.admin_agent aliases app.main_agent. The router never
-    # binds their schemas for a regular user. Extra tools exist only on the admin
-    # graph and are independently gated by the same invocation config.
+    # are configured and app.admin_agent aliases app.main_agent. Neither routing
+    # mode binds their schemas for a regular user. Extra tools exist only on the
+    # admin graph and are independently gated by the same invocation config.
     agent_tools = tools + extra_tools
     router = ToolRouter(agent_tools, admin_tools=extra_tools)
+    mode = os.getenv("TOOL_ROUTING_MODE", "static").strip().lower()
+    if mode not in TOOL_ROUTING_MODES:
+        raise ValueError(
+            f"TOOL_ROUTING_MODE must be one of {TOOL_ROUTING_MODES}, got {mode!r}"
+        )
+    dynamic = mode == "dynamic"
 
     def routed_model(messages, config: RunnableConfig):
-        selected = router.select(
-            messages,
-            is_admin=(config or {}).get("configurable", {}).get("is_admin", False),
+        is_admin = (config or {}).get("configurable", {}).get("is_admin", False)
+        selected = (
+            router.select(messages, is_admin=is_admin)
+            if dynamic
+            else router.select_static(is_admin=is_admin)
         )
         schemas = [convert_to_openai_tool(t) for t in selected]
         fallback = openai_fallback.bind_tools(selected)
@@ -1188,6 +1196,6 @@ def build_agent(checkpointer=None, extra_tools=()):
 
     return create_react_agent(
         model=select_model,
-        tools=agent_tools + [router.discovery],
+        tools=agent_tools + ([router.discovery] if dynamic else []),
         checkpointer=checkpointer,
     )

blueprints/conversation/context.py

@@ -29,6 +29,10 @@ EARLIER_MESSAGE_MAX_CHARS = 4000
 TOOL_RESULT_MAX_CHARS = 12000
 TRUNCATION_NOTE = "\n[truncated: earlier context]"
 IMAGE_PLACEHOLDER = "[image from an earlier message]"
+# On overflow, drop old turns down to about this share of the history budget.
+TRIM_LOW_WATERMARK = 0.65
+# Upper bound on the summary ledger message (summary.STATE_TOKEN_LIMIT + wrapper).
+SUMMARY_RESERVE_TOKENS = 2000
 
 
 def estimate_tokens(value) -> int:
@@ -340,46 +344,48 @@ def _summary(turns: list[list[BaseMessage]], limit: int = 2400) -> AIMessage | N
     )
 
 
-def trim_history(
-    messages: list[BaseMessage],
-    *,
-    token_budget: int = 12000,
-    summary_message: AIMessage | None = None,
-) -> list[BaseMessage]:
-    """Keep the current turn, then recent whole turns, within an estimated budget.
+def _stable_cut(costs: list[int], room: int, step: int) -> int:
+    """Return how many of the oldest prior turns to drop so the rest fit ``room``.
 
-    The current user message and tool call arguments are never truncated. Fail
-    explicitly if these mandatory inputs cannot fit, rather than lose the task.
+    Cut points sit where the cumulative cost from the oldest turn crosses a
+    multiple of ``step``. They depend only on turns already in history, so the
+    cut stays put as new turns arrive and only moves, by about ``step`` tokens,
+    once the retained turns overflow again. That keeps the prompt prefix
+    byte-identical across several turns for llama-server's prompt cache.
     """
-    if summary_message is not None:
-        if not isinstance(summary_message, AIMessage) or summary_message.tool_calls:
-            raise ValueError(
-                "Conversation summary must be assistant-role evidence without tool calls"
-            )
-        summary_cost = message_tokens(summary_message)
-        if summary_cost > token_budget:
-            logging.info("Omitting optional conversation summary: insufficient budget")
-            return trim_history(messages, token_budget=token_budget)
-        try:
-            trimmed = trim_history(messages, token_budget=token_budget - summary_cost)
-        except ValueError:
-            # Historical evidence must never evict a current request that fits
-            # without it. Required current-turn metadata retains priority.
-            logging.info(
-                "Omitting optional conversation summary to preserve current request"
-            )
-            return trim_history(messages, token_budget=token_budget)
-        if (
-            trimmed
-            and isinstance(trimmed[0], AIMessage)
-            and isinstance(trimmed[0].content, str)
-            and trimmed[0].content.startswith("Earlier conversation excerpts")
-        ):
-            trimmed = trimmed[1:]
-        return [summary_message, *trimmed]
+    remaining = sum(costs)
+    if remaining <= room:
+        return 0
+    total = 0
+    for i, turn_cost in enumerate(costs):
+        before = total
+        total += turn_cost
+        remaining -= turn_cost
+        if total // step > before // step and remaining <= room:
+            return i + 1
+    # No cut point leaves room for any prior turn (e.g. a heavy current turn).
+    # Fall back to keeping as many recent whole turns as fit.
+    remaining = sum(costs)
+    for i, turn_cost in enumerate(costs):
+        if remaining <= room:
+            return i
+        remaining -= turn_cost
+    return len(costs)
+
+
+def _trim(
+    messages: list[BaseMessage], token_budget: int, reserve: int = 0
+) -> tuple[list[BaseMessage], int]:
+    """Trim to ``token_budget - reserve``; return the messages and dropped turns.
+
+    ``reserve`` shrinks the space without moving the hysteresis cut points, so a
+    caller-supplied summary of varying size doesn't shift which turns are kept.
+    """
+    step = max(1, int(token_budget * (1 - TRIM_LOW_WATERMARK)))
+    token_budget -= reserve
     turns = [_complete_exchanges(turn) for turn in _turns(messages)]
     if not turns:
-        return []
+        return [], 0
     original_current = turns[-1]
     current = original_current
     current = [
@@ -421,19 +427,21 @@ def trim_history(
             else:
                 high = mid - 1
         current = compact(low)
-    selected = []
     used = cost(current)
-    prior = turns[:-1]
-    # Leave a small share for excerpts of older turns only when necessary.
-    summary_reserve = min(1000, max(0, token_budget - used))
-    for turn in reversed(prior):
-        shrunk = [_shrink(m, EARLIER_MESSAGE_MAX_CHARS, earlier=True) for m in turn]
-        turn_cost = cost(shrunk)
-        if used + turn_cost > token_budget - summary_reserve:
-            break
-        selected.insert(0, shrunk)
-        used += turn_cost
-    dropped = prior[: len(prior) - len(selected)]
+    prior = [
+        [_shrink(m, EARLIER_MESSAGE_MAX_CHARS, earlier=True) for m in turn]
+        for turn in turns[:-1]
+    ]
+    costs = [cost(turn) for turn in prior]
+    if used + sum(costs) <= token_budget:
+        cut = 0
+    else:
+        # Leave a small share for excerpts of older turns only when dropping.
+        summary_reserve = min(1000, max(0, token_budget - used))
+        cut = _stable_cut(costs, token_budget - summary_reserve - used, step)
+    selected = prior[cut:]
+    used += sum(costs[cut:])
+    dropped = turns[:cut]
     summary = _summary(dropped)
     result = [m for turn in selected for m in turn] + current
     if summary and used + message_tokens(summary) <= token_budget:
@@ -451,7 +459,68 @@ def trim_history(
             }
         )
     )
-    return result
+    return result, len(dropped)
+
+
+def trim_history(
+    messages: list[BaseMessage],
+    *,
+    token_budget: int = 12000,
+    summary_message: AIMessage | None = None,
+) -> list[BaseMessage]:
+    """Keep the current turn, then recent whole turns, within an estimated budget.
+
+    The current user message and tool call arguments are never truncated. Fail
+    explicitly if these mandatory inputs cannot fit, rather than lose the task.
+    When the budget overflows, old turns are dropped down to a lower watermark
+    so the retained prefix stays stable (and cacheable) for several turns.
+    """
+    if summary_message is not None:
+        if not isinstance(summary_message, AIMessage) or summary_message.tool_calls:
+            raise ValueError(
+                "Conversation summary must be assistant-role evidence without tool calls"
+            )
+    trimmed, dropped = _trim(messages, token_budget)
+    if summary_message is None or not dropped:
+        # The summary only duplicates retained turns when nothing was dropped,
+        # and it changes every turn, which would defeat prompt caching.
+        return trimmed
+    summary_cost = message_tokens(summary_message)
+    if summary_cost > token_budget:
+        logging.info("Omitting optional conversation summary: insufficient budget")
+        return trimmed
+    # Reserve a fixed ceiling, not the ledger's current size, so a shrinking
+    # ledger can't pull the cut back and change the retained prefix.
+    for reserve in dict.fromkeys(
+        (max(summary_cost, SUMMARY_RESERVE_TOKENS), summary_cost)
+    ):
+        try:
+            with_room, _ = _trim(messages, token_budget, reserve=reserve)
+            break
+        except ValueError:
+            continue
+    else:
+        # Historical evidence must never evict a current request that fits
+        # without it. Required current-turn metadata retains priority.
+        logging.info(
+            "Omitting optional conversation summary to preserve current request"
+        )
+        return trimmed
+    # The ledger supersedes excerpts of dropped turns; whenever it is included
+    # they are consistently absent, so this doesn't disturb the cached prefix.
+    if (
+        with_room
+        and isinstance(with_room[0], AIMessage)
+        and isinstance(with_room[0].content, str)
+        and with_room[0].content.startswith("Earlier conversation excerpts")
+    ):
+        with_room = with_room[1:]
+    # The ledger changes every turn, so place it just before the current turn:
+    # system, tools and retained history then stay byte-identical for the
+    # prompt cache. Turns start at a HumanMessage, so the last one opens the
+    # current turn and no tool exchange is split.
+    start = max(i for i, m in enumerate(with_room) if isinstance(m, HumanMessage))
+    return [*with_room[:start], summary_message, *with_room[start:]]
 
 
 def build_context(

blueprints/conversation/memory.py

@@ -1,3 +1,5 @@
+from collections import Counter
+
 from .models import UserMemory
 from .prompts import SIMBA_SYSTEM_PROMPT
 from .context import estimate_tokens
@@ -41,16 +43,23 @@ async def build_system_prompt_with_memories(user_id: str, query: str = "") -> st
     memories = await get_memories_for_user(user_id)
     if not memories:
         return SIMBA_SYSTEM_PROMPT
-    selected = []
     header = "\n\nUSER MEMORIES (facts the user has asked you to remember):\n"
     omission = "\n[Additional saved memories omitted from this prompt.]"
     # Reserve wrapper and omission costs so the entire injected memory block fits.
     used = estimate_tokens(header) + estimate_tokens(omission)
+    chosen = Counter()
     for memory in rank_memories(memories, query):
         cost = estimate_tokens(f"- {memory}\n")
         if used + cost <= MEMORY_BUDGET_TOKENS:
-            selected.append(memory)
+            chosen[memory] += 1
             used += cost
+    # Rank only chooses what fits; emit in stored order so the system prompt is
+    # byte-identical across queries and llama-server can reuse its KV cache.
+    selected = []
+    for memory in memories:
+        if chosen[memory]:
+            chosen[memory] -= 1
+            selected.append(memory)
     block = "\n".join(f"- {m}" for m in selected)
     if len(selected) < len(memories):
         block += omission

blueprints/conversation/models.py

@@ -26,6 +26,8 @@ class Conversation(Model):
 
     class Meta:
         table = "conversations"
+        # Conversation list: a user's conversations, most recent first.
+        indexes = (("user_id", "updated_at"),)
 
 
 class ConversationMessage(Model):
@@ -41,11 +43,13 @@ class ConversationMessage(Model):
 
     class Meta:
         table = "conversation_messages"
+        # Loading a conversation's history in order.
+        indexes = (("conversation_id", "created_at"),)
 
 
 class UserMemory(Model):
     id = fields.UUIDField(primary_key=True)
-    user = fields.ForeignKeyField("models.User", related_name="memories")
+    user = fields.ForeignKeyField("models.User", related_name="memories", db_index=True)
     content = fields.TextField()
     created_at = fields.DatetimeField(auto_now_add=True)
     updated_at = fields.DatetimeField(auto_now=True)

blueprints/conversation/tool_routing.py

@@ -1,4 +1,8 @@
-"""Deterministic per-call routing; discovery grants live in tool messages, not globals."""
+"""Tool selection per model call.
+
+Static mode (the default) binds the full role-appropriate set. Dynamic mode
+routes per call by keyword; discovery grants live in tool messages, not globals.
+"""
 
 import json
 import re
@@ -29,6 +33,7 @@ ADMIN_TOOL_NAMES = {
 }
 CORE_TOOL_NAMES = {"get_current_date", "save_user_memory"}
 DISCOVERY_NAME = "discover_tool_groups"
+TOOL_ROUTING_MODES = ("static", "dynamic")
 
 
 def words(text):
@@ -138,6 +143,16 @@ class ToolRouter:
         )
         return discover_tool_groups
 
+    def select_static(self, *, is_admin=False):
+        """Every role-appropriate tool in registration order, without discovery.
+
+        The bound set never varies by query or turn, so the rendered tool block
+        stays byte-identical and llama-server can reuse its cached prefix.
+        """
+        return [
+            t for t in self.tools if is_admin or tool_name(t) not in self.admin_names
+        ]
+
     def select(self, messages, *, is_admin=False):
         available = self.available_groups(is_admin)
         human_indices = [

blueprints/imessage/__init__.py

@@ -327,7 +327,7 @@ async def _process_and_reply(
             async with httpx.AsyncClient() as client:
                 resp = await client.get(media_url, timeout=30)
                 resp.raise_for_status()
-            image_block = image_content_block(resp.content)
+            image_block = await asyncio.to_thread(image_content_block, resp.content)
         except Exception as e:
             logger.error(f"Failed to load iMessage image: {e}")
             content = f"{content or ''}\n\n[An image was sent but could not be loaded.]".strip()

blueprints/rag/logic.py

@@ -67,6 +67,9 @@ def _get_engine():
     return _get_engine._engine
 
 
+# Chunks per embedding request when indexing Paperless documents.
+EMBED_BATCH_SIZE = 32
+
 text_splitter = RecursiveCharacterTextSplitter(
     chunk_size=1000,  # chunk size (characters)
     chunk_overlap=200,  # chunk overlap (characters)
@@ -200,15 +203,25 @@ async def index_documents():
     splits = _sanitize_documents(splits)
     logger.info(f"Indexing {len(splits)} chunks from {len(documents)} documents")
     vector_store = _get_vector_store()
-    for i, split in enumerate(splits):
+    for start in range(0, len(splits), EMBED_BATCH_SIZE):
+        batch = splits[start : start + EMBED_BATCH_SIZE]
         try:
-            await vector_store.aadd_documents(documents=[split])
-        except Exception as e:
-            logger.error(
-                f"Failed to embed chunk {i} from {split.metadata.get('filename', 'unknown')}: {e}"
+            await vector_store.aadd_documents(documents=batch)
+        except Exception as batch_error:
+            # Embedding fails before anything is inserted; retry the batch one
+            # chunk at a time to name the chunk the embedding server rejects.
+            logger.warning(
+                f"Batch of {len(batch)} chunks failed ({batch_error}); retrying one at a time"
             )
-            logger.debug(f"Chunk content preview: {split.page_content[:200]!r}")
-            raise
+            for i, split in enumerate(batch, start):
+                try:
+                    await vector_store.aadd_documents(documents=[split])
+                except Exception as e:
+                    logger.error(
+                        f"Failed to embed chunk {i} from {split.metadata.get('filename', 'unknown')}: {e}"
+                    )
+                    logger.debug(f"Chunk content preview: {split.page_content[:200]!r}")
+                    raise
 
 
 async def fetch_obsidian_documents() -> list[Document]:

blueprints/ynab/sync.py

@@ -630,7 +630,8 @@ async def ensure_fresh(max_age_seconds: int = 120) -> YnabSyncState | None:
 
 async def ynab_sync_loop() -> None:
     """Background task: delta-sync YNAB every ``YNAB_SYNC_INTERVAL`` seconds."""
-    interval = int(os.getenv("YNAB_SYNC_INTERVAL", "900"))
+    # Below ensure_fresh's 120s window, so tool calls rarely sync inline.
+    interval = int(os.getenv("YNAB_SYNC_INTERVAL", "90"))
     logger.info(f"YNAB sync loop started (interval={interval}s)")
     while True:
         try:

docker-compose.yml

@@ -15,6 +15,7 @@ x-raggr-common: &raggr-common
     - JWT_SECRET_KEY=${JWT_SECRET_KEY}
     - LLAMA_SERVER_URL=${LLAMA_SERVER_URL}
     - LLAMA_MODEL_NAME=${LLAMA_MODEL_NAME}
+    - TOOL_ROUTING_MODE=${TOOL_ROUTING_MODE:-static}
     - OIDC_ISSUER=${OIDC_ISSUER}
     - OIDC_CLIENT_ID=${OIDC_CLIENT_ID}
     - OIDC_CLIENT_SECRET=${OIDC_CLIENT_SECRET}
@@ -24,7 +25,7 @@ x-raggr-common: &raggr-common
     - TAVILY_API_KEY=${TAVILIY_API_KEY}
     - YNAB_ACCESS_TOKEN=${YNAB_ACCESS_TOKEN}
     - YNAB_BUDGET_ID=${YNAB_BUDGET_ID}
-    - YNAB_SYNC_INTERVAL=${YNAB_SYNC_INTERVAL:-900}
+    - YNAB_SYNC_INTERVAL=${YNAB_SYNC_INTERVAL:-90}
     - MEALIE_BASE_URL=${MEALIE_BASE_URL}
     - MEALIE_API_TOKEN=${MEALIE_API_TOKEN}
     - MEALIE_GROUP_SLUG=${MEALIE_GROUP_SLUG:-home}

migrations/models/19_20261012225640_add_fk_indexes.py

@@ -0,0 +1,142 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        ALTER TABLE "users" ALTER COLUMN "ldap_groups" TYPE JSONB USING "ldap_groups"::JSONB;
+        ALTER TABLE "beli_restaurants" ALTER COLUMN "cuisine" TYPE JSONB USING "cuisine"::JSONB;
+        ALTER TABLE "beli_restaurants" ALTER COLUMN "tags" TYPE JSONB USING "tags"::JSONB;
+        ALTER TABLE "artifacts" ALTER COLUMN "data" TYPE JSONB USING "data"::JSONB;
+        ALTER TABLE "artifacts" ALTER COLUMN "source_ref" TYPE JSONB USING "source_ref"::JSONB;
+        ALTER TABLE "ynab_changes" ALTER COLUMN "changes" TYPE JSONB USING "changes"::JSONB;
+        CREATE INDEX IF NOT EXISTS "idx_conversatio_user_id_1f0c4c" ON "conversations" ("user_id", "updated_at");
+        CREATE INDEX IF NOT EXISTS "idx_conversatio_convers_d1744c" ON "conversation_messages" ("conversation_id", "created_at");
+        CREATE INDEX IF NOT EXISTS "idx_user_memori_user_id_8278f0" ON "user_memories" ("user_id");
+        CREATE INDEX IF NOT EXISTS "idx_artifacts_message_d8728a" ON "artifacts" ("message_id");
+        CREATE INDEX IF NOT EXISTS "idx_artifacts_convers_188f03" ON "artifacts" ("conversation_id");"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        DROP INDEX IF EXISTS "idx_conversatio_convers_d1744c";
+        DROP INDEX IF EXISTS "idx_conversatio_user_id_1f0c4c";
+        DROP INDEX IF EXISTS "idx_user_memori_user_id_8278f0";
+        DROP INDEX IF EXISTS "idx_artifacts_convers_188f03";
+        DROP INDEX IF EXISTS "idx_artifacts_message_d8728a";
+        ALTER TABLE "users" ALTER COLUMN "ldap_groups" TYPE JSONB USING "ldap_groups"::JSONB;
+        ALTER TABLE "artifacts" ALTER COLUMN "data" TYPE JSONB USING "data"::JSONB;
+        ALTER TABLE "artifacts" ALTER COLUMN "source_ref" TYPE JSONB USING "source_ref"::JSONB;
+        ALTER TABLE "ynab_changes" ALTER COLUMN "changes" TYPE JSONB USING "changes"::JSONB;
+        ALTER TABLE "beli_restaurants" ALTER COLUMN "cuisine" TYPE JSONB USING "cuisine"::JSONB;
+        ALTER TABLE "beli_restaurants" ALTER COLUMN "tags" TYPE JSONB USING "tags"::JSONB;"""
+
+
+MODELS_STATE = (
+    "eJztXQtv4za2/iuEscBOgTQ3j0lmmntxASfjTtPmMZhkbne2KTS0RNvcyJRWlDI1uv3vl6"
+    "QkWw9KlmxJFj0EtttG4qHkjxR5znce/HMwdyxk08Mrh7wgj0IfO2RwAf4cEDhH7D+k9w/A"
+    "ALru6i6/4MOxLQTMREtxB46p70HTZzcn0KaIXbIQNT3sRg8jgW3zi47JGmIyXV0KCP53gA"
+    "zfmSJ/hjx247ff2WVMLPQHovzP3wYBRZ6BLf6kwLWgjywD+oPfeTv32ZhgZFupHxQ2FdcN"
+    "f+GKa58+Xb/7UbTk7zE2TMcO5mTV2l34M4csmwcBtg65DL83RQR5/LGJ38dfP8IjvhT+FH"
+    "bB9wK0/A3W6oKFJjCwOUqD/5kExOTgAPEk/n+v/3dQAzc2BhxzTHwO0p9/hb9q9ZvF1QF/"
+    "1NVPw4+vTs+/E7/Sof7UEzcFIoO/hCD0YSgqAF8BKf6dg/JqBj05lHH7DJjsRTeBMb6wwn"
+    "E1uWIgY4A2Q20wh38YNiJTf8b+PDk7K4Hx/4YfBZKslYDSYRM+/Bzuolsn4T0O6QpC00Px"
+    "hM0B+Y7d8fEcycFMS2YgtSLRw/g/egow+w3WPbEX0UdQgu/j9e3o4XF4+4H/kjml/7YFRM"
+    "PHEb9zIq4uMldfnWeGYtkJ+PX68SfA/wT/vL8bZef+st3jPwf8nWDgOwZxvhrQSnyv8dUY"
+    "mNTAJlaimgObltQDu9OBjV4+8cHOICHIrrPsJUQ2Wvmit6o1fIOvaLzFfpFZ+Y6qLHxHxe"
+    "veUXbZS2zZVffhhMgWm/EmWG73KWy+9XIFZvIs3Xk5GHnwfnQ8hKfkF7QQEF6z94DElG25"
+    "kU73Keqmr6Ctrq7ewoNflzpdclKwX8d+E/LDj3H4cDV8NxoIDMfQfP4KPcsoAHOOKIVTRP"
+    "OAXkaSP/7yEdlLzVeOZVI/vg177OfaXIRt6gOFno8nTGXfEpRh1I1iSPBZ45w4idmSmkf5"
+    "W/OTefYKJGwGWNGz+ZNK5skacysxnapZXUZySrdsfaWeG36KCdVUW2GtW2E++kOiYz6yq3"
+    "Io4/aqWGFluuPoH48ptTHWOV7dDv/xXUp1vLm/ex83T+goVzf3l9ok20fNPW+SURfBZ5ne"
+    "xFX3EQnmOa0pNboJ8R1/OoNPD6OPF4ArP0/k4fr2cngBKJ6P4RN5vL+/YWg4jjA36ir6x1"
+    "UU/eNiRf84p+jjOduGjGe0qGMwpYQ6M5ma5IrOjk8qIMlaFUIp7qWxjDZ1g5mTll2LeMtL"
+    "KolqOwxcXn+pqpRIRJvUUBS1Tc2Mo2BLGzXrd+gtimuNVcl02dxo7d5A6w8L0Kp9xjmRWz"
+    "R3vMVAYpYl7h6UWWOCmJjzhrgDK0wbWW0bWewhPiK17KyEiDa1tKklUW7219TS3q+9GNgc"
+    "Lbwzv02FUcxtLT3Ujbv12+wMtDb9Nm3qfvfX764efOhLGfnVzVLNz8GWaVDerpreN/iAiM"
+    "UwBbx/YDtTTC7AHLoUMGUPfBE9fQEu9OAc+A7APgUffrkaAWZKYPazvMMs0bN1h0/kiTz4"
+    "bE5aABPwIZrg4BVxfOB6jokoBUK1XXwHqANg+AzWO/QB691jHTJtziHoiYztAP0Xk0YEsB"
+    "XB8QAkFphggukMiVb8lRyu1TJJbNvgBdqYL+5U/KwtNF8aj1RVjmQp0Iyytl7zbZIZOT55"
+    "W4W5O3lbTN3xe+U6sPtsIiOeJXWgzQmqog+3BvIKVPSHi9nXtYGmlJbsRlPqbo/tk2IU41"
+    "Cq8mpbZi9U3nBgcyrcbvQRoecVsFDr+SfNO6nPO/FxrBtmnZRRc6dtxdHjQkq/Op5kZl5i"
+    "Ar1Fge6SkMqAOV5EOn5fiXMZeJechUsuvAyyy+u74cfPciIvbp+ct5efH0fDrB4zh7hWVO"
+    "xSQEl9+6iap7zMVZ7zlX9lBhRlS7pBgvm4noYtEW3Ew9stqqdVQD0txvRUEn4Qeb7rQyoR"
+    "1ZDG362BCP95srXUcWwESck3n5TNrqhR3Eob+1Nd7afGknp/f5NaUi+vs76PT7eXI7YeCJ"
+    "hZIxyyXdd3j1JwZ3NoMrXrGUn852sW1IysgjP2+LzKypq1BxIL63l2xoaMXDD+FzIltlkx"
+    "oFk5BcFsRZFi1tLMcD3nBVv1ltScYHfa6cB2TGhvod5n4ruqLKtnxcvqWW5ZtS3oGlPPCV"
+    "xJRMfPD/d3ckgzYlnyAJs++A+wMW0tAn/w2+/tGE38N5c7nLO+5YyWyjvQDuc9Jmm0w3nv"
+    "BnYZ3VWRfJNHHTaYydVn8748gysZfLY5Ful4t/7N/EpQhJoxNE0nINsGTI54X8OwK4Uhoe"
+    "YMWQGzwoxmsh8f4v7UT30cIxtzLXWC7W1RuWRdfQh7UhgQnQvaln9HLCYDiYNnFJOjxR4e"
+    "sahVDDER3YHoS2f/9iF3OIhgjChKNR9FUkFmGShCeTgJRRYQLwVEw6/Yn4HTo+8tuAAe4u"
+    "3ZyxyCG0yeKY87uZp5zhy+u3wiE8cDL8zGZv+iCHrmDJjQhWNsYx9vHwiiXVENuaJiPlQG"
+    "6PrUIymwSnD9rXAoG9BR2zJRO/fqnVVympyVOE3O8k6TCVtFuDnB5rFkfypGMyunJqStzE"
+    "1mjRUCWpJW7qgIZ9cZD5Y0KrGcoYhl1OQmFOEiKsV4jR1rYdStt5ASUiTvtuvPQmA08+eS"
+    "CII1wMZCGlh5hpXQstkT2avV1NwkooqArOs2ar55U0eCWvHYerMuHsmI762ZrpaW0uUcYk"
+    "DyGNbOWlOXOj/IpK+lJ0mfMthSGBcRi4kxWMMvppwmVXnGSIbThBM8DTzBCwPO8l3fDj/w"
+    "ywQJuosWsY41ekhwkEspD1mcbmTvGLKQiJjewmUTDcTRvAeJPgBFvs8eTw+eCGc34254Ql"
+    "hAD8FD4LoOT2qbsymGXRvFLShwkSeqMGmOsiccZTRp69NBOUHNVMaMBaauDRdG3SyErJy2"
+    "HRI10Vxj5tBa7G9KSBV2rSsw+fqcB/OaFBAHKZkMlrg93eSHH063gFKUif3+5Pj1m9dvT8"
+    "9fv2VNxKssr7wpATcfaSww2CS7KCeoJ2NmMhbmGY1iPaSY08r1oAq6XRNbmBnaTB95kczd"
+    "0hSElFyH6QfLjb/H2Qc2pL6BPM+RRHUXT9m0lCLbfOc8rOYL95Mv1IHHezGwSlW6UoI37F"
+    "2pq12RhY2cUbKK9tsydFktLNMxUwsSVtYKmgDigfX2sOxMIUhap48TyBQxyGnw1pHImYFb"
+    "zyM/stvPFHA5XnGMLzpURJ6KIl2CdkVJojjPI9fugfPItxxm9g8FXKcNhfney2Tm7kF4MS"
+    "qAhqxlaCxbCSPmeMI6DDwE+K975hXYfAfQkD0GmNl+aI5Y/3bYMRcQWjMQVdlZc80i94RF"
+    "FhaNmLSbhGvlpRvQ7fplIfVIlavkChZDEkciB7LvpJArlIl2RxkebaGaN0wYxhgUeIMLAc"
+    "zJfZPo8e6RGXDiyYi2CYkSUwhikfg3iaX4JCMUNl+hsx3oRXrHi7RmUvaTSdFxWHsWh9V2"
+    "UIP6UVi5bGiJHS3LmC42pOX52rq4p9J2podM7GLpsTLF/veUkCre4S4i8ssOZl9/uuO2h7"
+    "Q3B+qA77gPw/ejCxDXIHwio9vh9c0FWBbN3P0R7vpEpDb8xDvIC97LGLvVflnfnMjKqmlQ"
+    "KGJAVDINi5w+FQ/uXUp3WG8w0upCJ8nhh9Hdu+u793kNZBDduQBueGrLE3kY3T1eAMoWyi"
+    "fyI1v1R+8uhGcBWU/kanh3NboRl0z+W+2ojurutwOmlgSehyL8NxmmdA8dDtXH5YMP76IJ"
+    "nBkkfpkBK460eceG5PMFsNiILJ7Ir6PRL/zPrwg9879v7+8ef+IX5myvm9mLfgxOQJHBJq"
+    "Nsty4NlkvJ6Vq9mVq93HkXm2N11KCcoCLbto6Z04SgjpnTA1vI9Maf3XhR9yD1rKDmexOY"
+    "NED5qh9Fl5sifeJ7rzyH/OyMBxKaN751UMbumqyR8S9nrEld9UnduvlUOo0qn3KaeLMaen"
+    "VGTGvVUq3a9Zy5W4u2XUmoMke7xlQ7crQjR1lHzoZHOe3kECdlsii9gGxg3mZEdUhaH+KG"
+    "y5wPZRHg23kddu2Va+GEI51drJlSTahppvTbG9i6xxq1yVUlzySR8FWZI0uKOavcMSnN8l"
+    "a/LROIxYNCffl3zWZtsudswWYlwK+hAKWltG2bOVtomwJBuR5UQbdrnUgAVVjFYc3cLS7l"
+    "0FeVs6vJ66EJWz1mRccTV53BuW4UgbnrWRwyAiFYG+mB0g40sdAHYkHbwm18MZrObJbO1N"
+    "zCXpigmlvY04HVlct05bI+Vy7zePmkwINbn7zMmamPy87UQrbVol0ZYApIvTR0a3i9zKA1"
+    "Te25K4JRPG4cUEwQpdFkS3jcPIyIZWRtcvG3yUZ+yg8m16zgTljBzKDV4lcysqpwWB2QLA"
+    "Uzviq6BeIdZlNtMVm7g7dukGZGTBHTuytOcLkS110GkoIqTtHjKkESx8VBEse5IIlvK3q4"
+    "lROhTezXmolxe/1RLychM3RmY8ebOU6tzScrpxFdzskAc4UnD+bPD/d3BdNyJZJlWLDpg/"
+    "8AG9PWrKCEHjoOsO1jQg/581pSRTkK5URwlvPN6Ky8g3ygOzaRwWwYWXZu8STOiCk5h5vf"
+    "l9SrwCAPyb4cje4uwBgh8kR+Hd49Go/3xuNHnsDPbF1mpRq+15OcfTYDnw3xA/OoF9YJTQ"
+    "ttVB10B3O34fqg1HQ8WU1QZOI5tOXALWWya20odBgJ9xK/EnDeja6ub4c3r84OTjKOnnhO"
+    "vs6roI4vq01b7JZcCiiyVnbtkfThVIJn8c4ft9fb/ibb/gT55mahEmlJ7dbql1tL+6H3Yl"
+    "zzfujIGVDTZZmW0l7LpE9lS8dlJgS8t/it9V+m50ifCkYMPR9PQodazlm3vHdQ5qaDUauK"
+    "J+oMAWsRmH7gIQv4jmMDNrHYWIBXIg2ZSYwDa4r8A8C1uafg5Ojk/DvWho0hl8AE+DME4o"
+    "cCFxJk50/dae0purBFN+vLQYnTz48C4aoSKXF7NanpVmjAZ/b2dSCM26sJ4fnrCgievy4E"
+    "kN/KWFLYl+1wJXMwFlATwXaqq7APvI45Grfv3hxtTctqxfCkTuBx5hhN6qCbltoK414xKa"
+    "1A/II8Ki0KVEiIJiS6OyvpeIvvv+lzp7TVvJ9Ws47e3ouBzdNcDuFrloiFrVtFMy/aLi/S"
+    "zrk/G5kt+XMK62GXltoCtvWbcC9R00kDOmmgo6SBotWuAfSuMt31dYlbX+02v5SvB7OwFP"
+    "1WOCaOJ+vp0rcWzfTingLyYfQI7j7d3OyKCP5M4Dg+ik/CBSdvH5TRwQvW0IgOxOtZFeFi"
+    "dqbJ2PS2DyVMEzOn5xV4mdOsFrmiZfgtXUa4ZaJLYFADwri9mhA2X7CPYRC6a/IglubSp+"
+    "Q6zKZvTStsMpvedmjt4gQrIY1mLlQtj2V5pNrGX3iv6NV2SiZBG0oP8CoNnUxIbR082a+Z"
+    "GkdPHp8dnFYOn2T4Qo8fVLERmBJpDSp/9e1glcprYJflodhTTLzZaZnyHnSFqR1XmApN3L"
+    "qaRkJKH/cn4ft2RxJcxamyBSzB8v7BWpogyrrFfTtDXhMFmxAFcQ61MfWcwK1ZLkAq3Jb9"
+    "qwTQmoFpkIGZYcuSVQkt3YNWQtraTaHJnoY8AiVBmKV4JsX0pq4JhBYJhKkDbaMu7ZoSUg"
+    "TYtpN/Q0ygJyVfSy3djOR+prPWMnETiBjiVHa5hbsWzpVwmXGrHKTMWM0dv6hNx300Hd9z"
+    "HX+wxn4MGx1UNSIjy0FbksoYONrlrA2eNjcUbfD0GVK9t+/Z3j4ThaqKNvVZXMZq3W4uGl"
+    "bNJXZdRKzvHWIvgO1MgTMBX2fQBwiaM0AXxARhd9ZBWEUEjBfAwpMJGywAp2x8qC/yfOeY"
+    "H2UhySJutv8n8kQ+3w0v/07B8MM1mEEKiANmmPpMfTkA1GGNMQXsf1xIPJW7szyLPxhZ2K"
+    "fgFTqcHgLIOoKU4imZI+KDqcMfOPGcOfjb66Mj4Dvgb7wc4SG49gFlNgMThOGbTLDH3om/"
+    "ehvZzJd4Wpj/JNV3JKlPkeqyU3UnTH364eTk9PTNydHp+duz12/enL09WuZA5W+VJUNdXr"
+    "/nn2vqs44XxGIliA/SRn7QlGA3WTVND8C+JNXkvaEUeS/IM57ZfRtZshjpso9IJt1dNqE6"
+    "X1XyEB4f+4vaVGBGTE1HVBUy8LSYDDyVHNAuUKln6KaElASy+fIB0JTnmRSjuJJQ08htvh"
+    "Ao+2moVhmQpYCaCLZTonqlbqdhLCkHvBLRhRjyVQJ6ZJLdCjdBgUV2G/sQ1hhkwtegWdXW"
+    "N5mmFshig6K2y6kpN1P3+3c9N5P2vTfqe8fEdGT8fam3eCWkg6HZHiCylmTsaHkCREJMwx"
+    "gqzS/SQzhKYUyKaRh5/TxjjIwN52ReWEPKptgUGc6EK1dIMjsL+Z+s2DdaV1/7jgZ75TsS"
+    "tsi6vIJ0o4NqhkvdFIPBB+R9LyRBHFcC4FzUMjgEX77Ey9iXL9xLI3xB3KHD2to2BU+DoX"
+    "DKIOtpkHcmNdqzNqjWGlSla8pyZX5TaFDxW9qg2sigWoZkbZgDojBX3Hz2h7YFtC3QHxh1"
+    "Zn4DIGr9db/01w9wgVCR3hreXK+vurydJtqV2aF1+HKLfkn2NEInyIsrydXUJAvEFSHs29"
+    "Yo9fazX9vPQzAWEz6KESnYhzKt1m9INCVQkT+5wQRRHq8KAXUZYiDRxX+LEFQXeoj4f6cr"
+    "GgRzguOBt5bxJo30qPfFzqKcSs7tWQ3cJut5SlJJcqCFQLJ5XJG1jk27FNLGGK/JO3fqeP"
+    "/j9oooE117/4UZU/PzTsooAmvbOtoOKdTOq2e3jaU2JrQxoY2JqsbEgpgPPvQL+axVgwom"
+    "BGtr0GXjnfBaoSOk5jefEtLavEybjyCqS3ZlxJRcRVvJ2O9JZtjOooM6TgwT1WI3zrHMS+"
+    "s6szuuMyuGZMJ+ZrjrbDim2Q70sPZhWBHPn69jpKelFNlkujDVe6RpPq7nrB9rEda12Wrt"
+    "R+21hhlHw1UNr4vb73l0neZ8Nee7+41EctJILaJyJaJm2EPzpVbZ7uY5L7WptKSYLtuWQn"
+    "RiwylHSqY7Fk/NtJQin3zrk3Mj6rwBxnzXPtvmGfOeesgUQ1E7yLSDTAlw8dx1PD8M763N"
+    "lkuFlQS2lTDReMLVhTUrpySirXghNpumen4WlFeKd5u6gOYElcS0lRmqAw4G21tJPaGBh8"
+    "jDprREVXTnoIz0has2vWF696j+7ZblGop53Rfk0ZrlDxMiitJFbSyF/NOoo/WEzdUE8LjS"
+    "/nxcsj8fS/Znh/hIxqWXlD9ciejyhz0rf/jX/wO/ZAiL"
+)

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

tests/unit/test_context_summary.py

@@ -102,7 +102,8 @@ async def test_tools_and_incomplete_exchanges_do_not_become_summary_instructions
     assert "Now continue" not in message.content
     prompt = build_context(messages, "trusted system", [], summary_message=message)
     assert prompt[0].content == "trusted system"
-    assert prompt[1] == message
+    # Nothing was dropped, so the ledger would only duplicate retained turns.
+    assert message not in prompt
     assert prompt[-1] is messages[-1]
 
 
@@ -162,3 +163,17 @@ async def test_deletion_is_scoped_and_survives_reopen(tmp_path):
     await store.delete("owner", "c")
     assert (await SummaryStore(path).load("owner", "c")).cursor == 0
     assert (await SummaryStore(path).load("other", "c")).cursor == 4
+
+
+async def test_largest_ledger_fits_the_fixed_summary_reserve(tmp_path):
+    from blueprints.conversation.context import SUMMARY_RESERVE_TOKENS, message_tokens
+
+    messages = []
+    for i in range(100):
+        messages += [
+            HumanMessage(content=f"Must {i}: " + "猫😺" * 1000),
+            AIMessage(content=f"Decided {i}: " + "budget " * 1000),
+        ]
+    messages.append(HumanMessage(content="current"))
+    state = await SummaryStore(tmp_path / "summary.db").update("u", "c", messages)
+    assert message_tokens(state.as_message()) <= SUMMARY_RESERVE_TOKENS

tests/unit/test_context_trim.py

@@ -462,3 +462,83 @@ def test_focused_note_heading_index_uses_absolute_content_lines():
     metadata = json.loads(projected.content)["data"]["markdown_compaction"]
     assert metadata["headings"] == [{"line": 1001, "heading": "# Requested section"}]
     assert msg.artifact["data"]["markdown"] == "full original"
+
+
+def _conversation(turns, size=1000):
+    messages = []
+    for i in range(turns):
+        messages += [
+            HumanMessage(content=f"Question {i}"),
+            call(str(i)),
+            result(str(i), "data " * size),
+            AIMessage(content=f"Decision {i}"),
+        ]
+    return messages
+
+
+def test_summary_omitted_when_no_turns_dropped():
+    messages = _conversation(2, size=10) + [HumanMessage(content="next")]
+    summary = AIMessage(content="Earlier: keep dining low")
+    assert trim_history(messages, summary_message=summary) == trim_history(messages)
+    assert summary not in trim_history(messages, summary_message=summary)
+
+
+def assert_valid_sequence(out):
+    seen = set()
+    for m in out:
+        if isinstance(m, AIMessage):
+            seen.update(c["id"] for c in m.tool_calls)
+        if isinstance(m, ToolMessage):
+            assert m.tool_call_id in seen
+
+
+def test_summary_placed_before_current_turn_when_turns_dropped():
+    question = HumanMessage(content="next")
+    # The current turn has its own tool exchange the summary must not split.
+    messages = _conversation(12) + [question, call("now"), result("now")]
+    summary = AIMessage(content="Earlier: keep dining low")
+    out = trim_history(messages, token_budget=6000, summary_message=summary)
+    assert cost(out) <= 6000
+    assert out[-4:] == [summary, question, messages[-2], messages[-1]]
+    assert out.count(summary) == 1
+    # The ledger supersedes excerpts of dropped turns; don't send both.
+    assert not any(
+        isinstance(m.content, str)
+        and m.content.startswith("Earlier conversation excerpts")
+        for m in out
+    )
+    assert isinstance(out[0], HumanMessage)
+    # Without a ledger, the excerpts still cover the dropped turns.
+    plain = trim_history(messages, token_budget=6000)
+    assert plain[0].content.startswith("Earlier conversation excerpts")
+    assert_valid_sequence(out)
+
+
+def test_changing_summary_leaves_retained_prefix_identical():
+    messages = _conversation(12) + [HumanMessage(content="next")]
+    short = AIMessage(content="Earlier: keep dining low")
+    long = AIMessage(content="Earlier: " + "constraint " * 400)
+    a = trim_history(messages, token_budget=6000, summary_message=short)
+    b = trim_history(messages, token_budget=6000, summary_message=long)
+    assert a[: a.index(short)] == b[: b.index(long)]
+
+
+def test_trim_drops_to_low_watermark_and_prefix_stays_stable():
+    budget = 8000
+    messages, oldest, costs = [], [], []
+    for i in range(30):
+        messages.append(HumanMessage(content=f"Question {i}"))
+        out = trim_history(messages, token_budget=budget)
+        assert out[-1] is messages[-1]
+        assert cost(out) <= budget
+        oldest.append(next(m for m in out if isinstance(m, HumanMessage)).content)
+        costs.append(cost(out))
+        messages += [call(str(i)), result(str(i), "data " * 300), AIMessage("ok")]
+    cuts = [i for i in range(1, 30) if oldest[i] != oldest[i - 1]]
+    assert cuts, "history should overflow the budget"
+    # Each drop goes well below the budget instead of one turn under it...
+    for i in cuts:
+        assert costs[i] <= budget * (context.TRIM_LOW_WATERMARK + 0.1)
+    # ...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

tests/unit/test_file_upload.py

@@ -1,6 +1,7 @@
 """Tests for conversation file attachments: extraction, /upload and stream-query file_key."""
 
 import io
+import threading
 from datetime import timedelta
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
@@ -190,6 +191,21 @@ class TestUploadRoute:
         s3.upload_image.assert_awaited_once()
         s3.upload_file.assert_not_awaited()
 
+    async def test_image_processing_runs_off_the_event_loop(
+        self, app, headers, s3, monkeypatch
+    ):
+        threads = []
+        real = conv.process_image
+
+        def recording(*args):
+            threads.append(threading.current_thread())
+            return real(*args)
+
+        monkeypatch.setattr(conv, "process_image", recording)
+        resp = await _upload(app, headers, _png_bytes(), "cat.png", "image/png")
+        assert resp.status_code == 200
+        assert threads and threads[0] is not threading.main_thread()
+
     async def test_unsupported_type_is_415(self, app, headers, s3):
         resp = await _upload(app, headers, b"PK\x03\x04", "a.zip", "application/zip")
         assert resp.status_code == 415

tests/unit/test_memory_retrieval.py

@@ -46,3 +46,34 @@ async def test_backwards_compatible_no_query_and_empty_memories(monkeypatch):
     )
     fetch.return_value = ["Likes tea"]
     assert "Likes tea" in await memory.build_system_prompt_with_memories("user")
+
+
+@pytest.mark.asyncio
+async def test_memory_block_is_identical_across_queries_when_all_fit(monkeypatch):
+    memories = ["Favorite restaurant is Nori", "Cat Simba takes insulin", "Gym is Y"]
+    monkeypatch.setattr(
+        memory, "get_memories_for_user", AsyncMock(return_value=memories)
+    )
+    prompts = {
+        await memory.build_system_prompt_with_memories("user", query=q)
+        for q in ["", "Simba insulin", "gym", "Nori restaurant"]
+    }
+    assert len(prompts) == 1
+    prompt = prompts.pop()
+    assert prompt.index("Nori") < prompt.index("insulin") < prompt.index("Gym")
+
+
+@pytest.mark.asyncio
+async def test_rank_chooses_memories_but_stored_order_is_kept(monkeypatch):
+    filler = "Favorite restaurant is Nori. " * 20
+    memories = ["Likes tea", filler, "Simba insulin dose is two units."]
+    monkeypatch.setattr(
+        memory, "get_memories_for_user", AsyncMock(return_value=memories)
+    )
+    monkeypatch.setattr(memory, "MEMORY_BUDGET_TOKENS", 110)
+    prompt = await memory.build_system_prompt_with_memories(
+        "user", query="Simba insulin"
+    )
+    assert filler not in prompt
+    # Insulin ranks first but is emitted after the newer memory.
+    assert prompt.index("Likes tea") < prompt.index("insulin")

tests/unit/test_paperless_indexing.py

@@ -0,0 +1,56 @@
+from unittest.mock import AsyncMock
+
+import pytest
+from langchain_core.documents import Document
+
+from blueprints.rag import logic
+
+
+class FakeVectorStore:
+    def __init__(self, bad_text=None, fail_batches=False):
+        self.calls = []
+        self.bad_text = bad_text
+        self.fail_batches = fail_batches
+
+    async def aadd_documents(self, documents):
+        self.calls.append([d.page_content for d in documents])
+        if self.fail_batches and len(documents) > 1:
+            raise RuntimeError("batch too large")
+        if any(d.page_content == self.bad_text for d in documents):
+            raise RuntimeError("tokenizer error")
+
+
+def _setup(monkeypatch, store, count):
+    docs = [
+        Document(page_content=f"chunk {i}", metadata={"filename": f"doc{i}.pdf"})
+        for i in range(count)
+    ]
+    monkeypatch.setattr(
+        logic, "fetch_documents_from_paperless_ngx", AsyncMock(return_value=docs)
+    )
+    monkeypatch.setattr(logic.text_splitter, "split_documents", lambda d: d)
+    monkeypatch.setattr(logic, "_get_vector_store", lambda: store)
+    monkeypatch.setattr(logic, "EMBED_BATCH_SIZE", 4)
+
+
+async def test_chunks_are_embedded_in_batches(monkeypatch):
+    store = FakeVectorStore()
+    _setup(monkeypatch, store, 10)
+    await logic.index_documents()
+    assert [len(c) for c in store.calls] == [4, 4, 2]
+
+
+async def test_failed_batch_names_the_rejected_chunk(monkeypatch, caplog):
+    store = FakeVectorStore(bad_text="chunk 6")
+    _setup(monkeypatch, store, 10)
+    with pytest.raises(RuntimeError, match="tokenizer error"):
+        await logic.index_documents()
+    assert "Failed to embed chunk 6 from doc6.pdf" in caplog.text
+
+
+async def test_batch_failure_recovers_when_every_chunk_succeeds(monkeypatch):
+    store = FakeVectorStore(fail_batches=True)
+    _setup(monkeypatch, store, 5)
+    await logic.index_documents()
+    singles = [c[0] for c in store.calls if len(c) == 1]
+    assert singles == [f"chunk {i}" for i in range(5)]

tests/unit/test_stream_stall.py

@@ -61,7 +61,7 @@ def _chunk(run_id, text):
     }
 
 
-def _end(run_id, prompt_tokens=None):
+def _end(run_id, prompt_tokens=None, cached_tokens=None):
     usage = (
         {
             "input_tokens": prompt_tokens,
@@ -71,6 +71,8 @@ def _end(run_id, prompt_tokens=None):
         if prompt_tokens
         else None
     )
+    if usage and cached_tokens is not None:
+        usage["input_token_details"] = {"cache_read": cached_tokens}
     return {
         "event": "on_chat_model_end",
         "run_id": run_id,
@@ -207,7 +209,7 @@ async def test_logs_per_call_prompt_tokens(owner, monkeypatch, caplog):
     script = [
         (0, _start("m1")),
         (0, _chunk("m1", "Hi")),
-        (0, _end("m1", prompt_tokens=27262)),
+        (0, _end("m1", prompt_tokens=27262, cached_tokens=25000)),
         (0, _final("Hi")),
     ]
 
@@ -224,6 +226,7 @@ async def test_logs_per_call_prompt_tokens(owner, monkeypatch, caplog):
     assert summary["max_prompt_tokens"] == 27262
     (call,) = summary["model_calls"]
     assert call["prompt_tokens"] == 27262
+    assert call["cached_prompt_tokens"] == 25000
     assert call["prefill_ms"] is not None and call["ms"] is not None
 
 

tests/unit/test_tool_routing.py

@@ -136,11 +136,7 @@ def test_discovery_recovery_admin_gating_and_isolation(router):
     )
 
 
-@pytest.mark.asyncio
-@pytest.mark.parametrize("primary_fails", [False, True])
-async def test_real_graph_binds_discovered_schemas_and_accounts_same_subset(
-    monkeypatch, primary_fails
-):
+def import_agents():
     # Constructors are stubbed before agents import: no tokens, services or network.
     with (
         patch("langchain_openai.ChatOpenAI"),
@@ -154,7 +150,16 @@ async def test_real_graph_binds_discovered_schemas_and_accounts_same_subset(
             "utils.obsidian_service.ObsidianService", side_effect=ValueError("offline")
         ),
     ):
-        agents = importlib.import_module("blueprints.conversation.agents")
+        return importlib.import_module("blueprints.conversation.agents")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("primary_fails", [False, True])
+async def test_real_graph_binds_discovered_schemas_and_accounts_same_subset(
+    monkeypatch, primary_fails
+):
+    monkeypatch.setenv("TOOL_ROUTING_MODE", "dynamic")
+    agents = import_agents()
     bound_names, primary_names, accounted_names = [], [], []
 
     class FakeModel:
@@ -238,3 +243,90 @@ def test_schema_dict_tools_remain_routable_and_admin_gated():
     router = ToolRouter([extra], admin_tools=[extra])
     assert extra not in router.select([HumanMessage(content="extra")])
     assert extra in router.select([HumanMessage(content="extra")], is_admin=True)
+
+
+def test_static_selection_is_stable_role_gated_and_has_no_discovery(router):
+    queries = ["Review spending", "Find Simba's vet records", "Help", "calendar"]
+    user = [router.select_static() for _ in queries]
+    assert all(selected == user[0] for selected in user)
+    assert [t.name for t in user[0]] == [
+        "get_current_date",
+        "save_user_memory",
+        "ynab_budget_summary",
+        "mealie_get_recipe",
+        "obsidian_read_note",
+        "simba_search",
+        "web_search",
+    ]
+    admin = [t.name for t in router.select_static(is_admin=True)]
+    assert admin == [t.name for t in router.tools]
+    assert "discover_tool_groups" not in admin
+
+
+def test_static_selection_gates_schema_dict_admin_tools():
+    extra = {
+        "type": "function",
+        "function": {
+            "name": "extra",
+            "description": "docs",
+            "parameters": {"type": "object", "properties": {}},
+        },
+    }
+    router = ToolRouter([make_tool("web_search"), extra], admin_tools=[extra])
+    assert extra not in router.select_static()
+    assert extra in router.select_static(is_admin=True)
+
+
+@pytest.mark.asyncio
+async def test_static_graph_binds_full_role_set_every_call(monkeypatch):
+    monkeypatch.delenv("TOOL_ROUTING_MODE", raising=False)
+    agents = import_agents()
+    bound_names = []
+
+    class FakeModel:
+        def bind_tools(self, tools):
+            bound_names.append([t.name for t in tools])
+
+            def respond(messages):
+                if len(bound_names) == 1:
+                    return call("ynab_budget_summary")
+                return AIMessage(content="done")
+
+            return RunnableLambda(respond)
+
+    monkeypatch.setattr(
+        agents,
+        "tools",
+        [
+            make_tool("get_current_date"),
+            make_tool("ynab_budget_summary"),
+            make_tool("get_calendar_events"),
+        ],
+    )
+    monkeypatch.setattr(agents, "llama_chat", None)
+    monkeypatch.setattr(agents, "openai_fallback", FakeModel())
+    monkeypatch.setattr(agents, "build_context", lambda messages, *a, **k: messages)
+    graph = agents.build_agent()
+    # A query matching no keyword group still sees every tool, with no
+    # discovery round trip, and the bound set is identical on every call.
+    result = await graph.ainvoke({"messages": [HumanMessage(content="Help")]})
+    assert result["messages"][-1].content == "done"
+    assert bound_names == [["get_current_date", "ynab_budget_summary"]] * 2
+
+    bound_names.clear()
+    await graph.ainvoke(
+        {"messages": [HumanMessage(content="Help")]},
+        config={"configurable": {"is_admin": True}},
+    )
+    assert bound_names[0] == [
+        "get_current_date",
+        "ynab_budget_summary",
+        "get_calendar_events",
+    ]
+
+
+def test_unknown_routing_mode_fails_explicitly(monkeypatch):
+    agents = import_agents()
+    monkeypatch.setenv("TOOL_ROUTING_MODE", "sometimes")
+    with pytest.raises(ValueError, match="TOOL_ROUTING_MODE"):
+        agents.build_agent()

utils/google_service.py

@@ -25,8 +25,11 @@ class GoogleService:
         creds = Credentials.from_authorized_user_file(token_path, SCOPES)
         if creds.expired and creds.refresh_token:
             creds.refresh(Request())
-            with open(token_path, "w") as f:
-                f.write(creds.to_json())
+            try:
+                with open(token_path, "w") as f:
+                    f.write(creds.to_json())
+            except OSError:
+                pass  # read-only mount; refreshed creds live in memory
         self._creds = creds
 
         self._calendar = build("calendar", "v3", credentials=creds)