CLAUDE.md

@@ -80,7 +80,7 @@ make redeploy    # git pull + scripts/deploy.sh (build new color, wait healthy,
   - `imessage/` - Twilio webhook receiver for iMessage (HMAC-validated, number allowlist)
   - `whatsapp/` - Twilio webhook receiver for WhatsApp (TWIML replies, SendBlue send support)
   - `scheduled_messages/` - Scheduled message delivery with background scheduler
-  - `ynab/` - Postgres mirror of the YNAB budget (`ynab_*` tables), delta-synced via `server_knowledge` (background loop + on-demand before each tool call, full resync daily); `ynab_changes` is the edit history YNAB itself doesn't keep. Agent tools in `tools.py`; admin-only read-only `ynab_sql` in `sql_tool.py`. Manual sync: `make ynab-sync` / `make ynab-resync`
+  - `ynab/` - Postgres mirror of the YNAB budget (`ynab_*` tables), delta-synced via `server_knowledge` (background loop + on-demand before each tool call, full resync daily that also marks rows missing from the download as deleted); `ynab_changes` is the edit history YNAB itself doesn't keep. Agent tools in `tools.py`; admin-only read-only `ynab_sql` in `sql_tool.py`. Manual sync: `make ynab-sync` / `make ynab-resync`
 - `config/` - Configuration modules
   - `db.py` - Database setup
   - `oidc_config.py` - OIDC authentication configuration
@@ -89,7 +89,7 @@ make redeploy    # git pull + scripts/deploy.sh (build new color, wait healthy,
   - `cleaner.py` - PDF cleaning and summarization
   - `image_process.py` - Image description with LLM
   - `request.py` - Paperless-NGX API client
-  - `obsidian_service.py`, `mealie_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets, S3, and SimbaKit
+  - `obsidian_service.py`, `mealie_service.py`, `ynab_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets (API client for the `blueprints/ynab/` mirror), S3, and SimbaKit
   - `homebox_mcp.py` - Homebox inventory tools via the `homebox-mcp` stdio MCP server; admin-only (compiled into `app.admin_agent`, picked per user by `get_agent_for`)
   - `bookstack_mcp.py` - BookStack wiki tools (read/create/update, no deletes) via the `bookstack-mcp` npm stdio server; admin-only like Homebox
   - `mcp_tools.py` - shared allowlist + admin-gate wrapping for stdio MCP integrations

blueprints/ynab/sql_tool.py

@@ -15,11 +15,15 @@ from sqlglot import exp
 from sqlglot.optimizer.scope import traverse_scope
 from tortoise.transactions import in_transaction
 
+from utils.mcp_tools import admin_only_message
+
 from . import models
 from .sync import ensure_fresh
+from .tools import NOT_SYNCED, ynab_tool_errors
 
-MAX_ROWS = 200
-MAX_OUTPUT_CHARS = 12000
+MAX_ROWS = 100
+# Tool results are cut at 4000 chars in blueprints/conversation/logic.py
+MAX_OUTPUT_CHARS = 3800
 STATEMENT_TIMEOUT = "10s"
 
 ALLOWED_TABLES = frozenset(
@@ -49,19 +53,33 @@ FORBIDDEN_FUNCTIONS = re.compile(
 
 def validate_sql(query: str) -> str | None:
     """Return an error message if ``query`` isn't allowed, else None."""
+    return _check(query)[0]
+
+
+def _check(query: str) -> tuple[str | None, str | None]:
+    """Validate ``query``; return ``(error, sql_to_run)``.
+
+    The SQL that runs is regenerated from the validated parse tree (minus
+    comments), so Postgres executes exactly what was checked.
+    """
     try:
-        statements = [s for s in sqlglot.parse(query, read="postgres") if s]
-    except sqlglot.errors.ParseError as e:
-        return f"Could not parse SQL: {e}"
+        statements = [
+            s
+            for s in sqlglot.parse(query, read="postgres")
+            # a trailing "; -- comment" parses as a bare Semicolon
+            if s and not isinstance(s, exp.Semicolon)
+        ]
+    except sqlglot.errors.SqlglotError as e:
+        return f"Could not parse SQL: {e}", None
     if len(statements) != 1:
-        return "Exactly one SQL statement is allowed."
+        return "Exactly one SQL statement is allowed.", None
     root = statements[0]
     if not isinstance(root, exp.Query):
-        return "Only SELECT queries are allowed."
+        return "Only SELECT queries are allowed.", None
     if root.find(*FORBIDDEN_NODES) or any(
         s.args.get("locks") for s in root.find_all(exp.Select)
     ):
-        return "Only read-only SELECT queries are allowed."
+        return "Only read-only SELECT queries are allowed.", None
 
     # A name matching a CTE isn't necessarily a CTE reference (e.g. inside the
     # CTE's own body, or outside the scope that defines it), so resolve each
@@ -70,7 +88,7 @@ def validate_sql(query: str) -> str | None:
     try:
         scopes = traverse_scope(root)
     except Exception as e:
-        return f"Could not analyze SQL: {e}"
+        return f"Could not analyze SQL: {e}", None
     cte_names = {cte.alias_or_name for cte in root.find_all(exp.CTE)}
     tables = [
         source
@@ -83,18 +101,19 @@ def validate_sql(query: str) -> str | None:
         if isinstance(table.this, exp.Func):  # e.g. generate_series(...)
             continue
         if table.catalog or table.db not in ("", "public"):
-            return f"Schema-qualified table '{table.sql()}' is not allowed."
+            return f"Schema-qualified table '{table.sql()}' is not allowed.", None
         if table.name not in ALLOWED_TABLES:
             return (
                 f"Table '{table.name}' is not allowed. "
-                f"Available: {', '.join(sorted(ALLOWED_TABLES))}"
+                f"Available: {', '.join(sorted(ALLOWED_TABLES))}",
+                None,
             )
 
     for func in root.find_all(exp.Func):
         name = func.name if isinstance(func, exp.Anonymous) else func.sql_name()
         if FORBIDDEN_FUNCTIONS.match(name or ""):
-            return f"Function '{name}' is not allowed."
-    return None
+            return f"Function '{name}' is not allowed.", None
+    return None, root.sql(dialect="postgres", comments=False).rstrip("; ")
 
 
 def _format_value(value) -> str:
@@ -107,21 +126,21 @@ def _format_value(value) -> str:
     return str(value).replace("\t", " ").replace("\n", " ")
 
 
-def format_rows(rows: list[dict]) -> str:
+def format_rows(columns: list[str], rows: list[tuple]) -> str:
     if not rows:
         return "Query returned no rows."
-    header = list(rows[0])
-    lines = ["\t".join(header)]
-    lines += ["\t".join(_format_value(r[c]) for c in header) for r in rows]
+    lines = ["\t".join(columns)]
+    lines += ["\t".join(_format_value(v) for v in row) for row in rows]
     text = "\n".join(lines)
     if len(text) > MAX_OUTPUT_CHARS:
-        text = text[:MAX_OUTPUT_CHARS] + "\n...(output truncated)"
-    if len(rows) == MAX_ROWS:
+        text = text[:MAX_OUTPUT_CHARS] + "\n...(output truncated; aggregate more)"
+    elif len(rows) == MAX_ROWS:
         text += f"\n(capped at {MAX_ROWS} rows; aggregate or add LIMIT)"
     return text
 
 
 @tool
+@ynab_tool_errors
 async def ynab_sql(query: str, config: RunnableConfig) -> str:
     """Run a read-only Postgres SELECT over the YNAB mirror for questions the other YNAB tools can't answer. Tables: ynab_transactions (date, amount, payee_name, category_name, account_name, memo, transfer_account_id, deleted), ynab_subtransactions (split lines), ynab_accounts, ynab_categories, ynab_category_groups, ynab_payees, ynab_months, ynab_month_categories (month, category_id, budgeted=assigned, activity, balance=available), ynab_changes (edit history: synced_at, entity_type, action, label, changes JSON). Amounts are dollars; outflows negative. Filter NOT deleted.
 
@@ -129,23 +148,25 @@ async def ynab_sql(query: str, config: RunnableConfig) -> str:
         query: A single PostgreSQL SELECT statement
     """
     if not (config or {}).get("configurable", {}).get("is_admin"):
-        return "YNAB SQL access is restricted to admin users."
-    error = validate_sql(query)
+        return admin_only_message("YNAB SQL")
+    error, sql = _check(query)
     if error:
         return error
     if await ensure_fresh() is None:
-        return "YNAB data is unavailable: the budget has not been synced yet."
+        return NOT_SYNCED
 
-    query = query.strip().rstrip(";")
     try:
         async with in_transaction() as conn:
             await conn.execute_query("SET TRANSACTION READ ONLY")
             await conn.execute_query(
                 f"SET LOCAL statement_timeout = '{STATEMENT_TIMEOUT}'"
             )
-            rows = await conn.execute_query_dict(
-                f"SELECT * FROM ({query}) AS q LIMIT {MAX_ROWS}"
+            # Raw records rather than dicts, which would merge same-named
+            # columns (e.g. c.name and g.name)
+            _, records = await conn.execute_query(
+                f"SELECT * FROM ({sql}) AS q LIMIT {MAX_ROWS}"
             )
     except Exception as e:
         return f"SQL error: {e}"
-    return format_rows(rows)
+    columns = list(records[0].keys()) if records else []
+    return format_rows(columns, [tuple(r.values()) for r in records])

blueprints/ynab/sync.py

@@ -13,15 +13,17 @@ to ``ynab_changes``, which is the only history of edits we get.
 import asyncio
 import logging
 import os
+import time
 from dataclasses import dataclass
 from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from typing import Any, Callable
 
-import httpx
 from tortoise.models import Model
 from tortoise.transactions import in_transaction
 
+from utils.ynab_service import YNABService
+
 from .models import (
     YnabAccount,
     YnabCategory,
@@ -37,19 +39,21 @@ from .models import (
 
 logger = logging.getLogger(__name__)
 
-YNAB_API_URL = "https://api.ynab.com/v1"
 FULL_RESYNC_INTERVAL = timedelta(hours=24)
+# After a failed sync, tools answer from the mirror without retrying YNAB
+FAILURE_BACKOFF_SECONDS = 60
 # Cap on rows passed to a single ``id IN (...)`` lookup
 LOOKUP_CHUNK = 2000
 
 _sync_lock = asyncio.Lock()
+_last_failure = float("-inf")
 
 
 def ynab_configured() -> bool:
     return bool(os.getenv("YNAB_ACCESS_TOKEN"))
 
 
-def _budget_id() -> str:
+def _configured_budget_id() -> str:
     return os.getenv("YNAB_BUDGET_ID") or "last-used"
 
 
@@ -189,6 +193,10 @@ def _map_subtransaction(s: dict) -> dict:
     }
 
 
+# Filled in from payees/categories/accounts after each sync, not from YNAB
+DENORMALIZED_FIELDS = ("account_name", "payee_name", "category_name")
+
+
 @dataclass
 class EntitySpec:
     entity_type: str
@@ -228,9 +236,7 @@ SPECS = {
         "transaction",
         YnabTransaction,
         _map_transaction,
-        _fields(
-            YnabTransaction, exclude=("account_name", "payee_name", "category_name")
-        ),
+        _fields(YnabTransaction, exclude=DENORMALIZED_FIELDS),
     ),
     "subtransactions": EntitySpec(
         "subtransaction",
@@ -261,25 +267,33 @@ def _budget_entities(budget: dict) -> dict[str, list[dict]]:
 
 
 class _Labeler:
-    """Builds human-readable change log labels from ids."""
+    """Resolves ids to names for change log labels and diffs."""
 
     def __init__(self, names: dict[str, str]):
         self.names = names
 
+    def name(self, entity_id: str | None) -> str | None:
+        return None if entity_id is None else self.names.get(entity_id, entity_id)
+
     def __call__(self, entity_type: str, row: dict) -> str:
-        n = self.names.get
         if entity_type == "transaction":
-            payee = n(row.get("payee_id") or "") or row.get("import_payee_name") or ""
-            return f"{row['date']} {payee} {row['amount']:+.2f}".strip()
+            payee = self.name(row.get("payee_id")) or row.get("import_payee_name") or ""
+            return f"{row['date']} {payee} {row['amount']:+.2f}"
         if entity_type == "subtransaction":
-            category = n(row.get("category_id") or "", "")
-            return f"split line {category} {row['amount']:+.2f}".strip()
+            category = self.name(row.get("category_id")) or ""
+            return f"split line {category} {row['amount']:+.2f}"
         if entity_type == "month_category":
-            return f"{n(row['category_id'], row['category_id'])} {row['month']:%Y-%m}"
+            return f"{self.name(row['category_id'])} {row['month']:%Y-%m}"
         if entity_type == "month":
             return row["id"][:7]
         return row.get("name") or row["id"]
 
+    def diff_entry(self, field: str, old: Any, new: Any) -> tuple[str, list]:
+        """Log ``category_id`` changes as ``category`` names, not opaque ids."""
+        if field.endswith("_id"):
+            return field[:-3], [self.name(old), self.name(new)]
+        return field, [_jsonable(old), _jsonable(new)]
+
 
 async def _load_names(conn) -> dict[str, str]:
     names: dict[str, str] = {}
@@ -289,6 +303,18 @@ async def _load_names(conn) -> dict[str, str]:
     return names
 
 
+async def _fetch_existing(spec: EntitySpec, ids: list[str], conn, fields) -> dict:
+    existing: dict[str, dict] = {}
+    for i in range(0, len(ids), LOOKUP_CHUNK):
+        rows = (
+            await spec.model.filter(id__in=ids[i : i + LOOKUP_CHUNK])
+            .using_db(conn)
+            .values("id", *fields)
+        )
+        existing.update({row["id"]: row for row in rows})
+    return existing
+
+
 async def _apply(
     spec: EntitySpec,
     raw_rows: list[dict],
@@ -296,28 +322,35 @@ async def _apply(
     server_knowledge: int,
     log_changes: bool,
     labeler: _Labeler,
+    complete: bool = False,
 ) -> int:
-    """Upsert rows that differ from the mirror; return how many changed."""
-    if not raw_rows:
-        return 0
+    """Upsert rows that differ from the mirror; return how many changed.
 
+    ``complete`` means ``raw_rows`` is every live entity of this type (a full
+    download). Full downloads omit deleted entities, so mirror rows missing
+    from it are marked deleted here; otherwise deletions that happened right
+    before a full sync would never reach the mirror.
+    """
     incoming = {row["id"]: row for row in map(spec.mapper, raw_rows)}
     # Only YNAB-sourced columns; denormalized names are filled in afterwards
-    sample = next(iter(incoming.values()))
-    field_names = tuple(f for f in _fields(spec.model) if f in sample)
-
-    ids = list(incoming)
-    existing: dict[str, dict] = {}
-    for i in range(0, len(ids), LOOKUP_CHUNK):
-        rows = (
-            await spec.model.filter(id__in=ids[i : i + LOOKUP_CHUNK])
-            .using_db(conn)
-            .values("id", *field_names)
-        )
-        existing.update({row["id"]: row for row in rows})
+    field_names = tuple(f for f in _fields(spec.model) if f not in DENORMALIZED_FIELDS)
+    existing = await _fetch_existing(spec, list(incoming), conn, field_names)
 
     to_write: list[Model] = []
     changes: list[YnabChange] = []
+
+    def log(entity_id: str, action: str, diff: dict, row: dict) -> None:
+        changes.append(
+            YnabChange(
+                server_knowledge=server_knowledge,
+                entity_type=spec.entity_type,
+                entity_id=entity_id,
+                action=action,
+                label=labeler(spec.entity_type, row)[:500],
+                changes=diff,
+            )
+        )
+
     for entity_id, new in incoming.items():
         old = existing.get(entity_id)
         if old is not None and all(_norm(old[f]) == _norm(new[f]) for f in field_names):
@@ -330,31 +363,45 @@ async def _apply(
             # Created and deleted between two syncs: nothing worth logging
             if new.get("deleted"):
                 continue
-            action = "created"
-            diff = {
-                f: [None, _jsonable(new[f])]
+            diff = dict(
+                labeler.diff_entry(f, None, new[f])
                 for f in spec.logged_fields
                 if new[f] is not None
-            }
+            )
+            # e.g. a month rollover creates every category's month row with
+            # nothing assigned; that's not an edit anyone made
+            if not any(values[1] for values in diff.values()):
+                continue
+            log(entity_id, "created", diff, new)
         else:
-            diff = {
-                f: [_jsonable(old[f]), _jsonable(new[f])]
+            diff = dict(
+                labeler.diff_entry(f, old[f], new[f])
                 for f in spec.logged_fields
                 if _norm(old[f]) != _norm(new[f])
-            }
-            if not diff:
-                continue
-            action = "deleted" if diff.get("deleted") == [False, True] else "updated"
-        changes.append(
-            YnabChange(
-                server_knowledge=server_knowledge,
-                entity_type=spec.entity_type,
-                entity_id=entity_id,
-                action=action,
-                label=labeler(spec.entity_type, new)[:500],
-                changes=diff,
             )
+            if diff:
+                deleted = diff.get("deleted") == [False, True]
+                log(entity_id, "deleted" if deleted else "updated", diff, new)
+
+    removed = 0
+    if complete:
+        live_ids = (
+            await spec.model.filter(deleted=False)
+            .using_db(conn)
+            .values_list("id", flat=True)
         )
+        missing = [i for i in live_ids if i not in incoming]
+        if missing:
+            if log_changes and "deleted" in spec.logged_fields:
+                gone = await _fetch_existing(spec, missing, conn, field_names)
+                for entity_id, row in gone.items():
+                    log(entity_id, "deleted", {"deleted": [False, True]}, row)
+            for i in range(0, len(missing), LOOKUP_CHUNK):
+                removed += (
+                    await spec.model.filter(id__in=missing[i : i + LOOKUP_CHUNK])
+                    .using_db(conn)
+                    .update(deleted=True)
+                )
 
     if to_write:
         await spec.model.bulk_create(
@@ -366,7 +413,7 @@ async def _apply(
         )
     if changes:
         await YnabChange.bulk_create(changes, batch_size=500, using_db=conn)
-    return len(to_write)
+    return len(to_write) + removed
 
 
 DENORMALIZE_SQL = (
@@ -381,92 +428,149 @@ DENORMALIZE_SQL = (
     """UPDATE ynab_transactions SET category_name = NULL
        WHERE category_id IS NULL AND category_name IS NOT NULL""",
 )
+# Entity types whose changes can affect the denormalized name columns
+NAME_SOURCES = ("accounts", "payees", "categories", "transactions")
+
+REFERENCE_KEYS = ("accounts", "payees", "category_groups", "categories")
+DATA_KEYS = ("months", "month_categories", "transactions", "subtransactions")
 
 
 async def apply_budget(
-    budget: dict, server_knowledge: int, conn, log_changes: bool
+    budget: dict,
+    server_knowledge: int,
+    conn,
+    log_changes: bool,
+    full: bool = False,
 ) -> dict[str, int]:
-    """Write a (full or delta) budget payload into the mirror."""
+    """Write a full or delta budget payload into the mirror.
+
+    With ``full``, entity types present in the payload are treated as
+    complete and anything missing from them is marked deleted.
+    """
     entities = _budget_entities(budget)
-    counts: dict[str, int] = {}
+    present = {key for key in SPECS if budget.get(key) is not None}
+    if budget.get("months") is not None:
+        present.add("month_categories")
 
-    # Reference data first so transaction labels can resolve names
     labeler = _Labeler({})
-    for key in ("accounts", "payees", "category_groups", "categories"):
-        counts[key] = await _apply(
-            SPECS[key], entities[key], conn, server_knowledge, log_changes, labeler
-        )
-    labeler.names = await _load_names(conn)
-    for key in ("months", "month_categories", "transactions", "subtransactions"):
-        counts[key] = await _apply(
-            SPECS[key], entities[key], conn, server_knowledge, log_changes, labeler
-        )
+    if log_changes and any(entities.values()):
+        labeler.names = await _load_names(conn)
 
-    if any(counts.values()):
+    counts: dict[str, int] = {}
+    for key in REFERENCE_KEYS + DATA_KEYS:
+        counts[key] = await _apply(
+            SPECS[key],
+            entities[key],
+            conn,
+            server_knowledge,
+            log_changes,
+            labeler,
+            complete=full and key in present,
+        )
+        if key in REFERENCE_KEYS:
+            # Later labels (e.g. a new payee on a new transaction) need these
+            labeler.names.update({row["id"]: row["name"] for row in entities[key]})
+
+    if any(counts[key] for key in NAME_SOURCES):
         for sql in DENORMALIZE_SQL:
             await conn.execute_query(sql)
     return counts
 
 
-async def fetch_budget(last_knowledge: int | None) -> dict:
-    """GET the budget from YNAB, as a delta when ``last_knowledge`` is set."""
-    params = {}
-    if last_knowledge is not None:
-        params["last_knowledge_of_server"] = last_knowledge
-    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0)) as client:
-        response = await client.get(
-            f"{YNAB_API_URL}/budgets/{_budget_id()}",
-            params=params,
-            headers={"Authorization": f"Bearer {os.environ['YNAB_ACCESS_TOKEN']}"},
-        )
-    response.raise_for_status()
-    return response.json()["data"]
+async def _current_state() -> YnabSyncState | None:
+    """The sync state for the mirrored budget.
+
+    The mirror holds a single budget. With ``YNAB_BUDGET_ID`` unset, the first
+    sync resolves YNAB's "last-used" alias and the mirror stays pinned to that
+    budget, so switching budgets in YNAB can't mix two budgets' rows.
+    """
+    configured = os.getenv("YNAB_BUDGET_ID")
+    states = await YnabSyncState.all()
+    if configured:
+        others = [s.budget_id for s in states if s.budget_id != configured]
+        if others:
+            raise ValueError(
+                f"YNAB mirror holds budget {others[0]} but YNAB_BUDGET_ID is "
+                f"{configured}; clear the ynab_* tables to switch budgets"
+            )
+        return next((s for s in states if s.budget_id == configured), None)
+    return states[0] if states else None
 
 
-async def sync_ynab(full: bool = False) -> dict:
+def _is_fresh(state: YnabSyncState | None, max_age_seconds: int) -> bool:
+    return (
+        state is not None
+        and state.last_synced_at is not None
+        and datetime.now(timezone.utc) - _norm(state.last_synced_at)
+        <= timedelta(seconds=max_age_seconds)
+    )
+
+
+async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> dict:
     """Sync YNAB into Postgres. Returns per-entity counts of changed rows.
 
-    Safe to call concurrently: an in-process lock plus a Postgres advisory
-    lock serialize syncs across both blue/green containers.
+    ``max_age_seconds`` skips the sync if the mirror is already that fresh,
+    checked under the lock so parallel tool calls share one sync.
+
+    The YNAB request runs before the DB transaction. An asyncio lock
+    serializes syncs in this process; a Postgres advisory lock plus a
+    server_knowledge check discards a payload if the other blue/green
+    container synced in the meantime.
     """
-    budget_id = _budget_id()
+    global _last_failure
     async with _sync_lock:
+        state = await _current_state()
+        if max_age_seconds is not None and _is_fresh(state, max_age_seconds):
+            return {"skipped": True}
+
+        now = datetime.now(timezone.utc)
+        base_knowledge = state.server_knowledge if state else None
+        first_sync = base_knowledge is None
+        full = (
+            full
+            or first_sync
+            or state.last_full_sync_at is None
+            or now - _norm(state.last_full_sync_at) > FULL_RESYNC_INTERVAL
+        )
+        budget_id = state.budget_id if state else _configured_budget_id()
+
         try:
+            data = await YNABService().fetch_budget(
+                budget_id, None if full else base_knowledge
+            )
+            budget = data["budget"]
+            knowledge = data["server_knowledge"]
             async with in_transaction() as conn:
                 await conn.execute_query(
                     "SELECT pg_advisory_xact_lock(hashtext('ynab_sync'))"
                 )
-                state, _ = await YnabSyncState.get_or_create(
-                    budget_id=budget_id, using_db=conn
-                )
-                first_sync = state.server_knowledge is None
-                now = datetime.now(timezone.utc)
-                full = (
-                    full
-                    or first_sync
-                    or (
-                        state.last_full_sync_at is None
-                        or now - _norm(state.last_full_sync_at) > FULL_RESYNC_INTERVAL
-                    )
-                )
+                latest = await YnabSyncState.get_or_none(
+                    budget_id=budget["id"]
+                ).using_db(conn)
+                if (latest.server_knowledge if latest else None) != base_knowledge:
+                    logger.info("YNAB sync skipped: mirror changed during fetch")
+                    return {"skipped": True}
 
-                data = await fetch_budget(None if full else state.server_knowledge)
-                knowledge = data["server_knowledge"]
                 counts = await apply_budget(
-                    data["budget"], knowledge, conn, log_changes=not first_sync
+                    budget, knowledge, conn, log_changes=not first_sync, full=full
+                )
+                await YnabSyncState.update_or_create(
+                    budget_id=budget["id"],
+                    defaults={
+                        "budget_name": budget.get("name"),
+                        "server_knowledge": knowledge,
+                        "last_synced_at": now,
+                        "last_error": None,
+                        **({"last_full_sync_at": now} if full else {}),
+                    },
+                    using_db=conn,
                 )
-
-                state.budget_name = data["budget"].get("name") or state.budget_name
-                state.server_knowledge = knowledge
-                state.last_synced_at = now
-                if full:
-                    state.last_full_sync_at = now
-                state.last_error = None
-                await state.save(using_db=conn)
         except Exception as e:
-            await YnabSyncState.filter(budget_id=budget_id).update(
-                last_error=f"{type(e).__name__}: {e}"[:2000]
-            )
+            _last_failure = time.monotonic()
+            if state is not None:
+                await YnabSyncState.filter(budget_id=state.budget_id).update(
+                    last_error=f"{type(e).__name__}: {e}"[:2000]
+                )
             raise
 
     changed = {k: v for k, v in counts.items() if v}
@@ -479,22 +583,18 @@ async def ensure_fresh(max_age_seconds: int = 120) -> YnabSyncState | None:
     """Sync if the mirror is older than ``max_age_seconds``.
 
     Used by agent tools before answering. A failed sync is logged and the
-    (possibly stale) mirror is used anyway; returns None only if the mirror
-    has never been populated.
+    (possibly stale) mirror is used anyway, and YNAB isn't retried for
+    ``FAILURE_BACKOFF_SECONDS``. Returns None only if the mirror has never
+    been populated.
     """
-    state = await YnabSyncState.get_or_none(budget_id=_budget_id())
-    stale = (
-        state is None
-        or state.last_synced_at is None
-        or datetime.now(timezone.utc) - _norm(state.last_synced_at)
-        > timedelta(seconds=max_age_seconds)
-    )
-    if stale:
+    state = await _current_state()
+    backing_off = time.monotonic() - _last_failure < FAILURE_BACKOFF_SECONDS
+    if not _is_fresh(state, max_age_seconds) and not backing_off:
         try:
-            await sync_ynab()
+            await sync_ynab(max_age_seconds=max_age_seconds)
         except Exception:
             logger.exception("YNAB sync failed; answering from existing mirror")
-        state = await YnabSyncState.get_or_none(budget_id=_budget_id())
+        state = await _current_state()
     if state is None or state.server_knowledge is None:
         return None
     return state

blueprints/ynab/tools.py

@@ -4,7 +4,9 @@ Every tool calls ``ensure_fresh`` first, which runs a cheap delta sync when
 the mirror is more than a couple of minutes old.
 """
 
+import functools
 import json
+import logging
 from datetime import date, datetime, timedelta
 from decimal import Decimal
 from zoneinfo import ZoneInfo
@@ -14,8 +16,47 @@ from tortoise import connections
 
 from .sync import ensure_fresh
 
+logger = logging.getLogger(__name__)
+
 NOT_SYNCED = "YNAB data is unavailable: the budget has not been synced yet."
 
+# Keep in step with TOOL_RESULT_MAX_CHARS in blueprints/conversation/logic.py;
+# longer results get truncated there anyway and crowd the 32k context.
+MAX_TRANSACTIONS = 40
+
+# ynab_recent_changes accepts the names it prints as well as stored types
+ENTITY_TYPE_ALIASES = {
+    "assignment": "month_category",
+    "transfer": "transaction",
+    "split": "subtransaction",
+    "group": "category_group",
+}
+ENTITY_TYPES = (
+    "transaction",
+    "subtransaction",
+    "month_category",
+    "category",
+    "category_group",
+    "payee",
+    "account",
+)
+
+
+def ynab_tool_errors(fn):
+    """Return errors to the model as text: LangGraph's ToolNode re-raises
+    anything else, which would fail the whole chat turn."""
+
+    @functools.wraps(fn)
+    async def wrapper(*args, **kwargs):
+        try:
+            return await fn(*args, **kwargs)
+        except Exception as e:
+            logger.exception(f"{fn.__name__} failed")
+            return f"YNAB error in {fn.__name__}: {type(e).__name__}: {e}"
+
+    return wrapper
+
+
 # Every transaction line, with split transactions expanded into their
 # subtransactions so category/payee filters see the real allocation.
 LINES_CTE = """
@@ -81,6 +122,7 @@ async def _query(sql: str, *params) -> list[dict]:
 
 
 @tool
+@ynab_tool_errors
 async def ynab_budget_summary(month: str = "") -> str:
     """Get YNAB budget overview for a month: Ready to Assign, income, assigned, spent, age of money, overspent categories, and net worth.
 
@@ -151,6 +193,7 @@ async def ynab_budget_summary(month: str = "") -> str:
 
 
 @tool
+@ynab_tool_errors
 async def ynab_account_balances(include_closed: bool = False) -> str:
     """List YNAB accounts (checking, savings, credit cards, investments) with current balances.
 
@@ -181,6 +224,7 @@ async def ynab_account_balances(include_closed: bool = False) -> str:
 
 
 @tool
+@ynab_tool_errors
 async def ynab_search_transactions(
     start_date: str = "",
     end_date: str = "",
@@ -190,7 +234,7 @@ async def ynab_search_transactions(
     memo: str = "",
     min_amount: float = 0,
     max_amount: float = 0,
-    limit: int = 25,
+    limit: int = 20,
 ) -> str:
     """Search YNAB transactions. Split transactions are expanded into their category lines.
 
@@ -203,7 +247,7 @@ async def ynab_search_transactions(
         memo: Partial match on memo
         min_amount: Minimum absolute dollar amount
         max_amount: Maximum absolute dollar amount
-        limit: Max transactions to list (default 25, max 100)
+        limit: Max transactions to list (default 20, max 40)
     """
     if await ensure_fresh() is None:
         return NOT_SYNCED
@@ -253,7 +297,7 @@ async def ynab_search_transactions(
     if not count:
         return f"No transactions found from {range_text}."
 
-    limit = min(max(1, limit), 100)
+    limit = min(max(1, limit), MAX_TRANSACTIONS)
     rows = await _query(
         f"""{LINES_CTE}
             SELECT * FROM lines l WHERE {where}
@@ -280,6 +324,7 @@ async def ynab_search_transactions(
 
 
 @tool
+@ynab_tool_errors
 async def ynab_spending_report(
     group_by: str = "category",
     start_date: str = "",
@@ -346,6 +391,7 @@ async def ynab_spending_report(
 
 
 @tool
+@ynab_tool_errors
 async def ynab_category_budget(month: str = "", category_group: str = "") -> str:
     """Get YNAB categories for a month grouped by category group: assigned, activity (negative = spent), available, and goal target. Flags overspending.
 
@@ -398,20 +444,25 @@ async def ynab_category_budget(month: str = "", category_group: str = "") -> str
 
 
 @tool
+@ynab_tool_errors
 async def ynab_recent_changes(days: int = 7, entity_type: str = "") -> str:
     """Show what changed in YNAB recently: new/edited/deleted transactions, assignment changes, renamed categories or payees.
 
     Args:
         days: Look back this many days (default 7)
-        entity_type: Optional filter: transaction, month_category (assignments), category, payee, account
+        entity_type: Optional filter: transaction, assignment, category, payee, account
     """
     if await ensure_fresh() is None:
         return NOT_SYNCED
+    entity_type = entity_type.strip().lower().removesuffix("s")
+    entity_type = ENTITY_TYPE_ALIASES.get(entity_type, entity_type)
+    if entity_type and entity_type not in ENTITY_TYPES:
+        return f"Unknown entity_type '{entity_type}'. Use one of: transaction, assignment, category, category_group, payee, account."
     rows = await _query(
         """SELECT synced_at, entity_type, action, label, changes FROM ynab_changes
            WHERE synced_at >= now() - make_interval(days => $1)
              AND ($2 = '' OR entity_type = $2)
-           ORDER BY id DESC LIMIT 50""",
+           ORDER BY id DESC LIMIT 40""",
         max(1, days),
         entity_type,
     )
@@ -433,7 +484,7 @@ async def ynab_recent_changes(days: int = 7, entity_type: str = "") -> str:
         entity = r["entity_type"].replace("month_category", "assignment")
         line = f"- {when:%Y-%m-%d %H:%M} {r['action']} {entity} '{r['label']}'"
         lines.append(line + (f": {detail}" if detail else ""))
-    note = " (newest 50)" if len(rows) == 50 else ""
+    note = " (newest 40)" if len(rows) == 40 else ""
     return f"YNAB changes in the last {days} days{note}:\n" + "\n".join(lines)
 
 

raggr-frontend/src/hooks/useChat.ts

@@ -14,9 +14,12 @@ const TOOL_MESSAGES: Record<string, string> = {
   web_search: "Searching the web...",
   get_current_date: "Checking today's date...",
   ynab_budget_summary: "Checking budget summary...",
+  ynab_account_balances: "Checking account balances...",
   ynab_search_transactions: "Looking up transactions...",
-  ynab_category_spending: "Analyzing category spending...",
-  ynab_insights: "Generating budget insights...",
+  ynab_spending_report: "Adding up spending...",
+  ynab_category_budget: "Checking category budgets...",
+  ynab_recent_changes: "Checking recent budget changes...",
+  ynab_sql: "Querying budget data...",
   obsidian_search_notes: "Searching notes...",
   obsidian_read_note: "Reading note...",
   obsidian_create_note: "Saving note...",

tests/unit/test_ynab_mirror.py

@@ -7,7 +7,7 @@ from tortoise import Tortoise
 from tortoise.transactions import in_transaction
 
 from blueprints.ynab.models import YnabChange, YnabMonthCategory, YnabTransaction
-from blueprints.ynab.sql_tool import format_rows, validate_sql
+from blueprints.ynab.sql_tool import _check, format_rows, validate_sql
 from blueprints.ynab.sync import _money, apply_budget
 
 ACCOUNT = {
@@ -105,9 +105,11 @@ async def db():
     await Tortoise.close_connections()
 
 
-async def _apply(budget, knowledge, log_changes=True):
+async def _apply(budget, knowledge, log_changes=True, full=False):
     async with in_transaction() as conn:
-        return await apply_budget(budget, knowledge, conn, log_changes=log_changes)
+        return await apply_budget(
+            budget, knowledge, conn, log_changes=log_changes, full=full
+        )
 
 
 def test_money_is_exact():
@@ -227,13 +229,109 @@ class TestValidateSql:
         assert result is not None and error in result
 
 
-def test_format_rows():
-    rows = [
-        {"payee": "Costco", "total": Decimal("-54.150")},
-        {"payee": None, "total": Decimal("10.000")},
-    ]
-    assert format_rows(rows) == "payee\ttotal\nCostco\t-54.150\n\t10"
-    assert format_rows([]) == "Query returned no rows."
+def test_format_rows_keeps_duplicate_column_names():
+    rows = [("Groceries", "Bills", Decimal("-54.150")), ("Rent", None, Decimal("10"))]
+    assert format_rows(["name", "name", "total"], rows) == (
+        "name\tname\ttotal\nGroceries\tBills\t-54.150\nRent\t\t10"
+    )
+    assert format_rows([], []) == "Query returned no rows."
+
+
+def test_unbalanced_quote_is_reported_not_raised():
+    error = validate_sql("SELECT * FROM ynab_payees WHERE name = 'Trader Joe's'")
+    assert error and error.startswith("Could not parse SQL")
+
+
+def test_executed_sql_drops_comments_and_semicolon():
+    error, sql = _check(
+        "SELECT payee_name, sum(amount) FROM ynab_transactions GROUP BY 1; -- top"
+    )
+    assert error is None
+    assert "--" not in sql and not sql.endswith(";")
+    assert f"SELECT * FROM ({sql}) AS q"  # wraps cleanly
+
+
+async def test_full_sync_marks_missing_rows_deleted(db):
+    await _apply(_budget([_txn(), _txn(id="txn-2")]), 1, log_changes=False, full=True)
+    # A full download omits deleted entities entirely
+    counts = await _apply(_budget([_txn(id="txn-2")]), 2, full=True)
+
+    assert counts["transactions"] == 1
+    assert (await YnabTransaction.get(id="txn-1")).deleted is True
+    assert (await YnabTransaction.get(id="txn-2")).deleted is False
+    change = await YnabChange.get(entity_id="txn-1")
+    assert change.action == "deleted" and "Costco" in change.label
+
+
+async def test_delta_sync_does_not_treat_missing_rows_as_deleted(db):
+    await _apply(_budget([_txn(), _txn(id="txn-2")]), 1, log_changes=False)
+    await _apply(_budget([_txn(id="txn-2")]), 2)
+    assert (await YnabTransaction.get(id="txn-1")).deleted is False
+
+
+async def test_full_sync_skips_entity_types_absent_from_payload(db):
+    await _apply(_budget([_txn()]), 1, log_changes=False, full=True)
+    budget = _budget([])
+    del budget["transactions"]
+    await _apply(budget, 2, full=True)
+    assert (await YnabTransaction.get(id="txn-1")).deleted is False
+
+
+async def test_new_month_with_nothing_assigned_is_not_logged(db):
+    await _apply(_budget([]), 1, log_changes=False)
+    next_month = {**_month(budgeted=0, activity=0), "month": "2026-11-01"}
+    await _apply(_budget([], next_month), 2)
+    assert await YnabChange.filter(entity_type="month_category").count() == 0
+
+
+async def test_recategorization_logs_names_not_ids(db):
+    household = {**CATEGORY, "id": "cat-2", "name": "Household"}
+    budget = _budget([_txn()])
+    budget["categories"].append(household)
+    await _apply(budget, 1, log_changes=False)
+
+    budget = _budget([_txn(category_id="cat-2")])
+    budget["categories"].append(household)
+    await _apply(budget, 2)
+
+    change = await YnabChange.get(entity_type="transaction")
+    assert change.changes == {"category": ["Groceries", "Household"]}
+    assert (await YnabTransaction.get(id="txn-1")).category_name == "Household"
+
+
+async def test_tool_errors_are_returned_as_text(monkeypatch):
+    from blueprints.ynab import tools
+
+    async def fresh():
+        return object()
+
+    async def broken_query(*args):
+        raise ConnectionError("db went away")
+
+    monkeypatch.setattr(tools, "ensure_fresh", fresh)
+    monkeypatch.setattr(tools, "_query", broken_query)
+    result = await tools.ynab_account_balances.ainvoke({})
+    assert "db went away" in result and result.startswith("YNAB error")
+
+
+async def test_recent_changes_accepts_display_names(monkeypatch):
+    from blueprints.ynab import tools
+
+    seen = []
+
+    async def fresh():
+        return object()
+
+    async def fake_query(sql, *params):
+        seen.append(params)
+        return []
+
+    monkeypatch.setattr(tools, "ensure_fresh", fresh)
+    monkeypatch.setattr(tools, "_query", fake_query)
+    await tools.ynab_recent_changes.ainvoke({"entity_type": "Assignments"})
+    assert seen[-1][1] == "month_category"
+    result = await tools.ynab_recent_changes.ainvoke({"entity_type": "bogus"})
+    assert result.startswith("Unknown entity_type")
 
 
 async def test_budget_summary_rejects_bad_month(monkeypatch):

utils/ynab_service.py

@@ -0,0 +1,37 @@
+"""YNAB API client used by the budget mirror (``blueprints.ynab.sync``)."""
+
+import os
+
+import httpx
+
+YNAB_API_URL = "https://api.ynab.com/v1"
+
+
+class YNABService:
+    """Thin async client for the YNAB endpoints the mirror needs."""
+
+    def __init__(self):
+        self.access_token = os.getenv("YNAB_ACCESS_TOKEN", "")
+        if not self.access_token:
+            raise ValueError("YNAB_ACCESS_TOKEN environment variable is required")
+
+    async def fetch_budget(self, budget_id: str, last_knowledge: int | None) -> dict:
+        """GET a whole budget, or only what changed since ``last_knowledge``.
+
+        Returns ``{"budget": {...}, "server_knowledge": int}``. Full responses
+        omit deleted entities; delta responses include them with
+        ``deleted: true``.
+        """
+        params = {}
+        if last_knowledge is not None:
+            params["last_knowledge_of_server"] = last_knowledge
+        async with httpx.AsyncClient(
+            timeout=httpx.Timeout(60.0, connect=10.0)
+        ) as client:
+            response = await client.get(
+                f"{YNAB_API_URL}/budgets/{budget_id}",
+                params=params,
+                headers={"Authorization": f"Bearer {self.access_token}"},
+            )
+        response.raise_for_status()
+        return response.json()["data"]