Fix second-round YNAB review findings

- Failure backoff is re-checked under the sync lock, so parallel tool
  calls share one failed YNAB request instead of retrying in turn.
- Budget id: YNAB_BUDGET_ID aliases ("last-used", "default") count as
  unset; state is always keyed by the real budget id, a mismatched
  response errors instead of silently skipping every sync, and more
  than one stored budget is an error rather than an arbitrary pick.
- Change log renames only id fields that resolve to names (account,
  payee, category, group); subtransaction transaction_id stays as is.
- ynab_recent_changes: correct singular for "categories", multi-word
  types, and the "month" type.
- The SQL wrap test now parses the wrapped query.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-12 18:03:48 -04:00
parent 9249f92db8
commit d1c3b7693e
3 changed files with 151 additions and 29 deletions
+51 -23
View File
@@ -53,8 +53,15 @@ def ynab_configured() -> bool:
return bool(os.getenv("YNAB_ACCESS_TOKEN"))
def _configured_budget_id() -> str:
return os.getenv("YNAB_BUDGET_ID") or "last-used"
# YNAB accepts these in place of a budget id; they can point at different
# budgets over time, so the mirror resolves them once and keeps the real id
BUDGET_ID_ALIASES = ("last-used", "default")
def _configured_budget_id() -> str | None:
"""YNAB_BUDGET_ID if it names a specific budget, else None."""
budget_id = os.getenv("YNAB_BUDGET_ID", "").strip()
return None if budget_id in ("", *BUDGET_ID_ALIASES) else budget_id
def _money(milliunits: int | None) -> Decimal | None:
@@ -266,6 +273,17 @@ def _budget_entities(budget: dict) -> dict[str, list[dict]]:
}
# Id fields that reference a named entity (account, payee, category, group);
# the change log records these by name
NAMED_ID_FIELDS = (
"account_id",
"payee_id",
"category_id",
"category_group_id",
"transfer_account_id",
)
class _Labeler:
"""Resolves ids to names for change log labels and diffs."""
@@ -289,9 +307,9 @@ class _Labeler:
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)]
"""Log e.g. ``category_id`` changes as ``category`` names, not ids."""
if field in NAMED_ID_FIELDS:
return field.removesuffix("_id"), [self.name(old), self.name(new)]
return field, [_jsonable(old), _jsonable(new)]
@@ -480,23 +498,26 @@ async def apply_budget(
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.
The mirror holds a single budget, keyed by its real YNAB id. With
``YNAB_BUDGET_ID`` unset or an alias, the first sync resolves "last-used"
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)
configured = _configured_budget_id()
states = await YnabSyncState.all().order_by("budget_id")
ids = [s.budget_id for s in states]
if len(ids) > 1 or (configured and ids and ids[0] != configured):
raise ValueError(
f"YNAB mirror holds budget(s) {', '.join(ids)} but YNAB_BUDGET_ID "
f"is {configured or 'unset'}; clear the ynab_* tables to switch budgets"
)
return states[0] if states else None
def _backing_off() -> bool:
return time.monotonic() - _last_failure < FAILURE_BACKOFF_SECONDS
def _is_fresh(state: YnabSyncState | None, max_age_seconds: int) -> bool:
return (
state is not None
@@ -509,8 +530,9 @@ def _is_fresh(state: YnabSyncState | None, max_age_seconds: int) -> bool:
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.
``max_age_seconds`` skips the sync if the mirror is already that fresh,
checked under the lock so parallel tool calls share one sync.
``max_age_seconds`` (on-demand syncs from tools) skips the sync if the
mirror is already that fresh or YNAB failed recently. Both are checked
under the lock, so parallel tool calls share one sync or one failure.
The YNAB request runs before the DB transaction. An asyncio lock
serializes syncs in this process; a Postgres advisory lock plus a
@@ -520,7 +542,9 @@ async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> d
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):
if max_age_seconds is not None and (
_is_fresh(state, max_age_seconds) or _backing_off()
):
return {"skipped": True}
now = datetime.now(timezone.utc)
@@ -533,6 +557,7 @@ async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> d
or now - _norm(state.last_full_sync_at) > FULL_RESYNC_INTERVAL
)
budget_id = state.budget_id if state else _configured_budget_id()
budget_id = budget_id or "last-used"
try:
data = await YNABService().fetch_budget(
@@ -540,6 +565,10 @@ async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> d
)
budget = data["budget"]
knowledge = data["server_knowledge"]
if state is not None and budget["id"] != state.budget_id:
raise ValueError(
f"YNAB returned budget {budget['id']} for {state.budget_id}"
)
async with in_transaction() as conn:
await conn.execute_query(
"SELECT pg_advisory_xact_lock(hashtext('ynab_sync'))"
@@ -588,8 +617,7 @@ async def ensure_fresh(max_age_seconds: int = 120) -> YnabSyncState | None:
been populated.
"""
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:
if not _is_fresh(state, max_age_seconds) and not _backing_off():
try:
await sync_ynab(max_age_seconds=max_age_seconds)
except Exception:
+10 -3
View File
@@ -33,6 +33,7 @@ ENTITY_TYPE_ALIASES = {
}
ENTITY_TYPES = (
"transaction",
"month",
"subtransaction",
"month_category",
"category",
@@ -42,6 +43,12 @@ ENTITY_TYPES = (
)
def _singular(word: str) -> str:
if word.endswith("ies"):
return word[:-3] + "y" # categories -> category
return word.removesuffix("s")
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."""
@@ -450,14 +457,14 @@ async def ynab_recent_changes(days: int = 7, entity_type: str = "") -> str:
Args:
days: Look back this many days (default 7)
entity_type: Optional filter: transaction, assignment, category, payee, account
entity_type: Optional filter: transaction, assignment, category, payee, account, month
"""
if await ensure_fresh() is None:
return NOT_SYNCED
entity_type = entity_type.strip().lower().removesuffix("s")
entity_type = _singular(entity_type.strip().lower().replace(" ", "_"))
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."
return f"Unknown entity_type '{entity_type}'. Use one of: transaction, assignment, category, category_group, payee, account, month."
rows = await _query(
"""SELECT synced_at, entity_type, action, label, changes FROM ynab_changes
WHERE synced_at >= now() - make_interval(days => $1)
+90 -3
View File
@@ -1,8 +1,10 @@
"""Tests for the YNAB Postgres mirror: delta application, change log, SQL guard."""
import asyncio
from decimal import Decimal
import pytest
import sqlglot
from tortoise import Tortoise
from tortoise.transactions import in_transaction
@@ -248,7 +250,12 @@ def test_executed_sql_drops_comments_and_semicolon():
)
assert error is None
assert "--" not in sql and not sql.endswith(";")
assert f"SELECT * FROM ({sql}) AS q" # wraps cleanly
# The LIMIT wrapper must still parse: a leftover line comment would
# swallow its closing parenthesis
wrapped = sqlglot.parse_one(
f"SELECT * FROM ({sql}) AS q LIMIT 100", read="postgres"
)
assert wrapped.args["limit"] is not None
async def test_full_sync_marks_missing_rows_deleted(db):
@@ -328,8 +335,14 @@ async def test_recent_changes_accepts_display_names(monkeypatch):
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"
for given, stored in [
("Assignments", "month_category"),
("categories", "category"),
("category groups", "category_group"),
("month", "month"),
]:
await tools.ynab_recent_changes.ainvoke({"entity_type": given})
assert seen[-1][1] == stored
result = await tools.ynab_recent_changes.ainvoke({"entity_type": "bogus"})
assert result.startswith("Unknown entity_type")
@@ -345,3 +358,77 @@ async def test_budget_summary_rejects_bad_month(monkeypatch):
assert await tools.ynab_budget_summary.ainvoke({"month": month}) == (
"Month must be YYYY-MM."
)
async def test_subtransaction_parent_id_is_not_renamed(db):
split = {
"id": "sub-1",
"transaction_id": "txn-1",
"amount": -10_000,
"memo": None,
"payee_id": None,
"category_id": "cat-1",
"transfer_account_id": None,
"deleted": False,
}
budget = _budget([_txn()])
budget["subtransactions"] = [split]
await _apply(budget, 1, log_changes=False)
budget["subtransactions"] = [{**split, "amount": -12_000}]
await _apply(budget, 2)
change = await YnabChange.get(entity_type="subtransaction")
assert change.changes == {"amount": [-10.0, -12.0]}
class TestBudgetPinning:
@pytest.fixture(autouse=True)
async def _db(self, db):
yield
async def _state(self, budget_id):
from blueprints.ynab.models import YnabSyncState
await YnabSyncState.create(budget_id=budget_id, server_knowledge=1)
@pytest.mark.parametrize("configured", ["", "last-used", "default"])
async def test_alias_or_unset_uses_stored_budget(self, monkeypatch, configured):
from blueprints.ynab.sync import _current_state
monkeypatch.setenv("YNAB_BUDGET_ID", configured)
await self._state("uuid-a")
assert (await _current_state()).budget_id == "uuid-a"
async def test_different_configured_budget_is_an_error(self, monkeypatch):
from blueprints.ynab.sync import _current_state
monkeypatch.setenv("YNAB_BUDGET_ID", "uuid-b")
await self._state("uuid-a")
with pytest.raises(ValueError, match="clear the ynab_"):
await _current_state()
async def test_multiple_stored_budgets_is_an_error(self, monkeypatch):
from blueprints.ynab.sync import _current_state
monkeypatch.delenv("YNAB_BUDGET_ID", raising=False)
await self._state("uuid-a")
await self._state("uuid-b")
with pytest.raises(ValueError):
await _current_state()
async def test_parallel_syncs_share_one_failure(db, monkeypatch):
from blueprints.ynab import sync
calls = []
class FailingYNAB:
async def fetch_budget(self, budget_id, since):
calls.append(budget_id)
raise TimeoutError("ynab down")
monkeypatch.setattr(sync, "YNABService", FailingYNAB)
monkeypatch.setattr(sync, "_last_failure", float("-inf"))
await asyncio.gather(
*(sync.ensure_fresh() for _ in range(3)), return_exceptions=True
)
assert len(calls) == 1