diff --git a/CLAUDE.md b/CLAUDE.md index 6aceebb..e42c9a0 100644 --- a/CLAUDE.md +++ b/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 diff --git a/blueprints/ynab/sql_tool.py b/blueprints/ynab/sql_tool.py index 58f49cc..de39457 100644 --- a/blueprints/ynab/sql_tool.py +++ b/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]) diff --git a/blueprints/ynab/sync.py b/blueprints/ynab/sync.py index 88bf15f..a551e0c 100644 --- a/blueprints/ynab/sync.py +++ b/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 diff --git a/blueprints/ynab/tools.py b/blueprints/ynab/tools.py index 5499aed..7fa5ee3 100644 --- a/blueprints/ynab/tools.py +++ b/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) diff --git a/raggr-frontend/src/hooks/useChat.ts b/raggr-frontend/src/hooks/useChat.ts index 21074c7..1d61a3d 100644 --- a/raggr-frontend/src/hooks/useChat.ts +++ b/raggr-frontend/src/hooks/useChat.ts @@ -14,9 +14,12 @@ const TOOL_MESSAGES: Record = { 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...", diff --git a/tests/unit/test_ynab_mirror.py b/tests/unit/test_ynab_mirror.py index 8d9bc4f..cd19517 100644 --- a/tests/unit/test_ynab_mirror.py +++ b/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): diff --git a/utils/ynab_service.py b/utils/ynab_service.py new file mode 100644 index 0000000..0ec4cfd --- /dev/null +++ b/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"]