What changed, and why

Splits were broken on the verify side. The write half worked: a split decision set category_id = null and a subtransaction per line. The verify half could not see it. to_target — which turns a read-back snapshot into the end-state to compare — returned None for any snapshot whose category_id was None, and a split parent's category is always null.

So classify_verify saw read = None, returned COULD_NOT_CONFIRM, and a perfectly-applied split escalated to a human every time: "I tried to change this and couldn't confirm it landed." A split could be written but never confirmed.

The fix reads the subtransactions back and reconstructs the split end-state, so a split write verifies field-by-field like any other.

Reading the subtransactions back

YNAB returns a split's subtransactions on the parent transaction; the wire model just wasn't reading them (it ignores unknown fields). Adding the field, the snapshot now carries the split lines, mapped from the wire — deleted and uncategorized lines dropped, since the agent only writes fully-categorized splits.

The wire gains WireSubtransaction and the parent's subtransactions.

src/ynab_agent/ynab/wire.py · 75 lines
src/ynab_agent/ynab/wire.py75 lines · Python
⋯ 19 lines hidden (lines 1–19)
1"""The slice of the YNAB JSON wire format this agent consumes (SPEC §1, §7).
2 
3Loose, ``extra='ignore'`` models — YNAB sends far more fields than we read, and
4a new field must never break a parse. These are the boundary types; the client
5maps them onto the strict domain (``YnabSnapshot`` etc.), so nothing downstream
6sees a wire shape.
7"""
8 
9from __future__ import annotations
10 
11from pydantic import BaseModel, ConfigDict
12 
13 
14class _Wire(BaseModel):
15 """Base for wire models: frozen, and tolerant of unknown fields."""
16 
17 model_config = ConfigDict(frozen=True, extra="ignore")
18 
19 
20class WireSubtransaction(_Wire):
21 """One subtransaction of a YNAB split (the fields we verify against, §3).
22 
23 A split parent carries ``category_id = null`` and one of these per line, so
24 the read-back reconstructs the split end-state from them — letting a split
25 write be confirmed field-by-field rather than always could-not-confirm
26 (SPEC §3 r4).
27 """
28 
29 amount: int
30 category_id: str | None = None
31 memo: str | None = None
32 deleted: bool = False
33 
34 
35class WireTransaction(_Wire):
36 """A YNAB transaction as the API returns it (the fields we use)."""
37 
38 id: str
39 account_id: str
40 date: str
41 amount: int
42 approved: bool
43 deleted: bool = False
44 memo: str | None = None
45 cleared: str = "uncleared"
46 flag_color: str | None = None
47 payee_id: str | None = None
48 payee_name: str | None = None
49 category_id: str | None = None
50 import_id: str | None = None
51 matched_transaction_id: str | None = None
52 subtransactions: tuple[WireSubtransaction, ...] = ()
53 
⋯ 22 lines hidden (lines 54–75)
54 
55class WireCategory(_Wire):
56 """A YNAB category as the API returns it (the budget figures we use)."""
57 
58 id: str
59 name: str
60 budgeted: int
61 activity: int
62 balance: int
63 deleted: bool = False
64 hidden: bool = False
65 
66 
67class WireMonth(_Wire):
68 """A YNAB budget month (the ``to_be_budgeted`` we read for W7, SPEC §8).
69 
70 ``to_be_budgeted`` is the month's Ready-to-Assign in milliunits — the first
71 source the balancer pulls from.
72 """
73 
74 month: str
75 to_be_budgeted: int

_to_split_lines maps the wire subtransactions onto domain split lines.

src/ynab_agent/ynab/client.py · 432 lines
src/ynab_agent/ynab/client.py432 lines · Python
⋯ 63 lines hidden (lines 1–63)
1"""The YNAB REST client: read snapshots, commit decisions, read budgets (§1).
2 
3A thin, deterministic boundary. The pure mappings (``to_snapshot``,
4``to_category_spend``, ``to_patch``) turn the wire format into the strict domain
5and a decision into a YNAB patch — they carry the sign/field conventions and are
6unit-tested against canned wire data. ``YnabClient`` wires them over a
7``YnabBackend`` protocol, so tests inject a fake and strict mypy never reasons
8about httpx. ``YnabClient.from_env`` builds the real httpx adapter, reading
9``YNAB_API_KEY`` (and optional ``YNAB_BUDGET_ID``); the token never lives here.
10"""
11 
12from __future__ import annotations
13 
14import datetime
15import os
16from typing import TYPE_CHECKING, Protocol
17 
18from ynab_agent.budget.overspend import CategorySpend
19from ynab_agent.domain.allocations import (
20 ResolvedCategory,
21 ResolvedSplit,
22 ResolvedSplitLine,
24from ynab_agent.domain.enums import ClearedState, DecidedBy, FlagColor
25from ynab_agent.domain.ids import (
26 AccountId,
27 CategoryId,
28 ImportId,
29 PayeeId,
30 YnabTransactionId,
32from ynab_agent.domain.money import Money
33from ynab_agent.domain.transaction import YnabSnapshot
34from ynab_agent.policy.converge import TargetState
35from ynab_agent.ynab.wire import (
36 WireCategory,
37 WireMonth,
38 WireSubtransaction,
39 WireTransaction,
41 
42if TYPE_CHECKING:
43 import httpx
44 
45 from ynab_agent.domain.proposal import Decision
46 
47_API_KEY_ENV = "YNAB_API_KEY"
48_BUDGET_ENV = "YNAB_BUDGET_ID"
49_DEFAULT_BUDGET = "last-used"
50_BASE_URL = "https://api.ynab.com/v1"
51# The month identifier the balancer operates on (SPEC §8): YNAB's "current"
52# shorthand for the live budget month, matching the W6 monitor's current-month
53# figures. ``/months/{month}/...`` also accepts ``YYYY-MM-01`` for later use.
54CURRENT_MONTH = "current"
55# How far back the snapshot fallback list scans when the single GET 404s (an
56# unapproved txn). The agent only reads recent transactions, so a year is ample.
57_FALLBACK_LOOKBACK_DAYS = 365
58_HTTP_NOT_FOUND = 404
59# The flag an agent-applied write carries so the owner sees it in the YNAB app
60# and can clear it as implicit review of an auto-action (SPEC §14.5).
61AGENT_REVIEW_FLAG = FlagColor.PURPLE
62 
63 
64def _to_split_lines(
65 subs: tuple[WireSubtransaction, ...],
66) -> tuple[ResolvedSplitLine, ...]:
67 """Map a split's wire subtransactions onto domain split lines (SPEC §3).
68 
69 Skips deleted lines and any line without a category (an uncategorized line
70 cannot be a :class:`ResolvedSplitLine`); the agent only writes fully
71 categorized splits, so a clean read-back maps every line.
72 """
73 return tuple(
74 ResolvedSplitLine(
75 category=CategoryId(sub.category_id),
76 amount=Money.from_milliunits(sub.amount),
77 memo=sub.memo,
78 )
79 for sub in subs
80 if not sub.deleted and sub.category_id
81 )
82 
83 
⋯ 349 lines hidden (lines 84–432)
84def to_snapshot(wire: WireTransaction) -> YnabSnapshot:
85 """Map a YNAB transaction onto the domain snapshot (SPEC §1)."""
86 return YnabSnapshot(
87 ynab_id=YnabTransactionId(wire.id),
88 account=AccountId(wire.account_id),
89 payee=wire.payee_name or "",
90 payee_id=PayeeId(wire.payee_id) if wire.payee_id else None,
91 amount=Money.from_milliunits(wire.amount),
92 txn_date=datetime.date.fromisoformat(wire.date),
93 memo=wire.memo,
94 flag=FlagColor(wire.flag_color) if wire.flag_color else None,
95 category_id=CategoryId(wire.category_id) if wire.category_id else None,
96 cleared=ClearedState(wire.cleared),
97 approved=wire.approved,
98 import_id=ImportId(wire.import_id) if wire.import_id else None,
99 matched_transaction_id=YnabTransactionId(wire.matched_transaction_id)
100 if wire.matched_transaction_id
101 else None,
102 subtransactions=_to_split_lines(wire.subtransactions),
103 )
104 
105 
106def to_category_spend(wire: WireCategory) -> CategorySpend:
107 """Map a YNAB category onto the domain spend figures (SPEC §7)."""
108 return CategorySpend(
109 category=CategoryId(wire.id),
110 name=wire.name,
111 budgeted=Money.from_milliunits(wire.budgeted),
112 activity=Money.from_milliunits(wire.activity),
113 balance=Money.from_milliunits(wire.balance),
114 )
115 
116 
117def to_target(snapshot: YnabSnapshot) -> TargetState | None:
118 """The read-back end-state of a transaction for verification (SPEC §3 r4).
119 
120 Reconstructs a split from its subtransactions so a split write verifies
121 field-by-field (a clean read-back hashes equal to the target); falls back to
122 a whole-category target; returns ``None`` only for an uncategorized txn (no
123 end-state to confirm), so the spine treats that as could-not-confirm rather
124 than a false divergence.
125 """
126 if len(snapshot.subtransactions) > 1: # a split has at least two lines
127 return TargetState(
128 allocation=ResolvedSplit(lines=snapshot.subtransactions),
129 memo=snapshot.memo,
130 approved=snapshot.approved,
131 )
132 if snapshot.category_id is None:
133 return None
134 return TargetState(
135 allocation=ResolvedCategory(category=snapshot.category_id),
136 memo=snapshot.memo,
137 approved=snapshot.approved,
138 )
139 
140 
141def to_patch(decision: Decision) -> dict[str, object]:
142 """Map a committed decision onto a YNAB transaction patch (SPEC §3).
143 
144 A whole-category decision sets ``category_id``; a split sets a null category
145 and the subtransactions (YNAB's split shape). ``approved`` and an optional
146 memo ride along.
147 """
148 allocation = decision.allocation
149 fields: dict[str, object] = {"approved": decision.approved}
150 if decision.memo is not None:
151 fields["memo"] = decision.memo
152 # Flag an agent-applied write so it surfaces in the YNAB app for review
153 # (SPEC §14.5); a human decision leaves the existing flag untouched.
154 if decision.decided_by is DecidedBy.AGENT:
155 fields["flag_color"] = AGENT_REVIEW_FLAG.value
156 if isinstance(allocation, ResolvedCategory):
157 fields["category_id"] = str(allocation.category)
158 else:
159 fields["category_id"] = None
160 fields["subtransactions"] = [
161 {
162 "amount": line.amount.milliunits,
163 "category_id": str(line.category),
164 "memo": line.memo,
165 }
166 for line in allocation.lines
167 ]
168 return fields
169 
170 
171class YnabBackend(Protocol):
172 """The YNAB operations the client needs (implemented over httpx)."""
173 
174 def get_transaction(self, txn_id: str) -> WireTransaction | None:
175 """Fetch one transaction, or ``None`` if the GET 404s (unapproved)."""
176 ...
177 
178 def patch_transaction(self, txn_id: str, fields: dict[str, object]) -> None:
179 """Patch one transaction with the given fields."""
180 ...
181 
182 def list_categories(self) -> tuple[WireCategory, ...]:
183 """List the budget's live categories."""
184 ...
185 
186 def list_transactions(
187 self, since_date: str, server_knowledge: int | None
188 ) -> tuple[tuple[WireTransaction, ...], int]:
189 """The transactions delta and the advanced ``server_knowledge``."""
190 ...
191 
192 def list_unapproved(self) -> tuple[WireTransaction, ...]:
193 """The budget's ``type=unapproved`` transactions."""
194 ...
195 
196 def get_month(self, month: str) -> WireMonth:
197 """The budget month, for its ``to_be_budgeted`` (Ready-to-Assign)."""
198 ...
199 
200 def get_month_category(
201 self, month: str, category_id: str
202 ) -> WireCategory | None:
203 """One category's figures for ``month``, or ``None`` if the GET 404s."""
204 ...
205 
206 def set_category_budgeted(
207 self, month: str, category_id: str, budgeted_milliunits: int
208 ) -> None:
209 """Set a month category's ``budgeted`` to a milliunit value."""
210 ...
211 
212 
213# The process-wide cached client built by ``from_env`` (see its docstring).
214# ``tests/conftest.py`` resets this between tests.
215_CACHED: YnabClient | None = None
216 
217 
218class YnabClient:
219 """Reads and writes YNAB through a backend, in domain terms (SPEC §1)."""
220 
221 def __init__(self, backend: YnabBackend) -> None:
222 """Wrap a backend (the real httpx adapter, or a test fake)."""
223 self._backend = backend
224 
225 @classmethod
226 def from_env(cls) -> YnabClient:
227 """Build (once) a client backed by the real YNAB REST API.
228 
229 The pooled httpx client and its OTel instrumentation are built on first
230 use and cached: every activity invocation calls this on the worker's hot
231 path, so a fresh-per-call client (never closed) would leak connections +
232 file descriptors. Tests reset the cache (see ``tests/conftest.py``).
233 
234 Raises:
235 RuntimeError: If ``YNAB_API_KEY`` is not set.
236 """
237 global _CACHED
238 if _CACHED is not None:
239 return _CACHED
240 token = os.environ.get(_API_KEY_ENV)
241 if not token:
242 msg = f"{_API_KEY_ENV} is not set"
243 raise RuntimeError(msg)
244 budget = os.environ.get(_BUDGET_ENV, _DEFAULT_BUDGET)
245 import httpx
246 
247 client = httpx.Client(
248 base_url=_BASE_URL,
249 headers={"Authorization": f"Bearer {token}"},
250 timeout=30.0,
251 )
252 from ynab_agent.telemetry import instrument_httpx
253 
254 instrument_httpx(client)
255 _CACHED = cls(_HttpxBackend(client, budget))
256 return _CACHED
257 
258 def snapshot(self, txn_id: str) -> YnabSnapshot | None:
259 """The current snapshot of a transaction, or ``None`` if not found.
260 
261 YNAB's single-transaction GET returns 404 for *unapproved* transactions
262 — including matched/scheduled imports — which are exactly the ones the
263 agent triages. Those do appear in the transactions list, so on a miss we
264 fall back to a bounded list scan (the agent only ever reads recent
265 transactions). An approved transaction uses the cheap single GET.
266 """
267 wire = self._backend.get_transaction(txn_id)
268 if wire is None:
269 since = (
270 datetime.datetime.now(datetime.UTC).date()
271 - datetime.timedelta(days=_FALLBACK_LOOKBACK_DAYS)
272 ).isoformat()
273 # ``last_knowledge_of_server=0`` is required: matched/scheduled
274 # unapproved transactions appear only in the delta-from-zero, not a
275 # plain since_date list (a YNAB quirk).
276 wires, _ = self._backend.list_transactions(since, 0)
277 wire = next((w for w in wires if w.id == txn_id), None)
278 if wire is None or wire.deleted:
279 return None
280 return to_snapshot(wire)
281 
282 def category_spends(self) -> tuple[CategorySpend, ...]:
283 """Every live category's month-to-date budget figures (SPEC §7)."""
284 return tuple(
285 to_category_spend(c) for c in self._backend.list_categories()
286 )
287 
288 def unapproved(self) -> tuple[YnabSnapshot, ...]:
289 """The budget's currently *unapproved* transactions (SPEC §2 W1).
290 
291 This is YNAB's own ``type=unapproved`` view — the transactions awaiting
292 the owner's review. It is the agent's outstanding-work set: a
293 tentatively scheduled/auto-matched import is *not* in it (YNAB excludes
294 it until it is combined), so the agent ignores those until they truly
295 land. Deleted rows drop out. The poll re-reads this each tick — the set
296 is small, so no cursor is needed (and none is stored, SPEC §0.5).
297 """
298 return tuple(
299 to_snapshot(wire)
300 for wire in self._backend.list_unapproved()
301 if not wire.deleted
302 )
303 
304 def commit(self, txn_id: str, decision: Decision) -> None:
305 """Commit a decision to a transaction (SPEC §3)."""
306 self._backend.patch_transaction(txn_id, to_patch(decision))
307 
308 def read_back(self, txn_id: str) -> TargetState | None:
309 """Re-read a transaction's end-state for verification (SPEC §3 r4)."""
310 snapshot = self.snapshot(txn_id)
311 return to_target(snapshot) if snapshot is not None else None
312 
313 def ready_to_assign(self, month: str = CURRENT_MONTH) -> Money:
314 """The month's Ready-to-Assign — the first balancer source (SPEC §8)."""
315 return Money.from_milliunits(
316 self._backend.get_month(month).to_be_budgeted
317 )
318 
319 def set_budgeted(
320 self, category_id: str, target: Money, month: str = CURRENT_MONTH
321 ) -> None:
322 """Set a category's month ``budgeted`` to ``target`` (SPEC §8).
323 
324 An *absolute* write: the workflow computes each category's new budgeted
325 from a snapshot and sets it directly, so an activity retry is idempotent
326 (re-setting the same value is a no-op) — unlike a relative add.
327 """
328 self._backend.set_category_budgeted(
329 month, category_id, target.milliunits
330 )
331 
332 def read_budgeted(
333 self, category_id: str, month: str = CURRENT_MONTH
334 ) -> Money | None:
335 """Re-read a category's month ``budgeted`` for verification (SPEC §8).
336 
337 ``None`` when the category can't be read, so the spine treats it as
338 could-not-confirm rather than false divergence (cf. :meth:`read_back`).
339 """
340 wire = self._backend.get_month_category(month, category_id)
341 return (
342 Money.from_milliunits(wire.budgeted) if wire is not None else None
343 )
344 
345 
346class _HttpxBackend:
347 """Adapts the YNAB REST API to the :class:`YnabBackend` protocol."""
348 
349 def __init__(self, client: httpx.Client, budget_id: str) -> None:
350 self._client = client
351 self._budget = budget_id
352 
353 def get_transaction(self, txn_id: str) -> WireTransaction | None:
354 path = f"/budgets/{self._budget}/transactions/{txn_id}"
355 response = self._client.get(path)
356 if response.status_code == _HTTP_NOT_FOUND:
357 return None
358 response.raise_for_status()
359 return WireTransaction.model_validate(
360 response.json()["data"]["transaction"]
361 )
362 
363 def patch_transaction(self, txn_id: str, fields: dict[str, object]) -> None:
364 path = f"/budgets/{self._budget}/transactions/{txn_id}"
365 response = self._client.patch(path, json={"transaction": fields})
366 response.raise_for_status()
367 
368 def list_transactions(
369 self, since_date: str, server_knowledge: int | None
370 ) -> tuple[tuple[WireTransaction, ...], int]:
371 path = f"/budgets/{self._budget}/transactions"
372 params: dict[str, str | int] = {"since_date": since_date}
373 if server_knowledge is not None:
374 params["last_knowledge_of_server"] = server_knowledge
375 response = self._client.get(path, params=params)
376 response.raise_for_status()
377 data = response.json()["data"]
378 transactions = tuple(
379 WireTransaction.model_validate(item)
380 for item in data["transactions"]
381 )
382 return transactions, int(data["server_knowledge"])
383 
384 def list_unapproved(self) -> tuple[WireTransaction, ...]:
385 path = f"/budgets/{self._budget}/transactions"
386 response = self._client.get(path, params={"type": "unapproved"})
387 response.raise_for_status()
388 return tuple(
389 WireTransaction.model_validate(item)
390 for item in response.json()["data"]["transactions"]
391 )
392 
393 def list_categories(self) -> tuple[WireCategory, ...]:
394 path = f"/budgets/{self._budget}/categories"
395 response = self._client.get(path)
396 response.raise_for_status()
397 groups = response.json()["data"]["category_groups"]
398 return tuple(
399 WireCategory.model_validate(category)
400 for group in groups
401 for category in group["categories"]
402 if not category.get("deleted") and not category.get("hidden")
403 )
404 
405 def get_month(self, month: str) -> WireMonth:
406 path = f"/budgets/{self._budget}/months/{month}"
407 response = self._client.get(path)
408 response.raise_for_status()
409 return WireMonth.model_validate(response.json()["data"]["month"])
410 
411 def get_month_category(
412 self, month: str, category_id: str
413 ) -> WireCategory | None:
414 path = (
415 f"/budgets/{self._budget}/months/{month}/categories/{category_id}"
416 )
417 response = self._client.get(path)
418 if response.status_code == _HTTP_NOT_FOUND:
419 return None
420 response.raise_for_status()
421 return WireCategory.model_validate(response.json()["data"]["category"])
422 
423 def set_category_budgeted(
424 self, month: str, category_id: str, budgeted_milliunits: int
425 ) -> None:
426 path = (
427 f"/budgets/{self._budget}/months/{month}/categories/{category_id}"
428 )
429 response = self._client.patch(
430 path, json={"category": {"budgeted": budgeted_milliunits}}
431 )
432 response.raise_for_status()

Reconstructing the split end-state

With the lines in hand, to_target rebuilds a ResolvedSplit for a split parent — so its end-state can be compared field-by-field — and falls back to the whole-category case. It returns None now only for a genuinely uncategorized txn (nothing to confirm).

A split parent reconstructs to a ResolvedSplit; the old code returned None here, forcing could-not-confirm.

src/ynab_agent/ynab/client.py · 432 lines
src/ynab_agent/ynab/client.py432 lines · Python
⋯ 116 lines hidden (lines 1–116)
1"""The YNAB REST client: read snapshots, commit decisions, read budgets (§1).
2 
3A thin, deterministic boundary. The pure mappings (``to_snapshot``,
4``to_category_spend``, ``to_patch``) turn the wire format into the strict domain
5and a decision into a YNAB patch — they carry the sign/field conventions and are
6unit-tested against canned wire data. ``YnabClient`` wires them over a
7``YnabBackend`` protocol, so tests inject a fake and strict mypy never reasons
8about httpx. ``YnabClient.from_env`` builds the real httpx adapter, reading
9``YNAB_API_KEY`` (and optional ``YNAB_BUDGET_ID``); the token never lives here.
10"""
11 
12from __future__ import annotations
13 
14import datetime
15import os
16from typing import TYPE_CHECKING, Protocol
17 
18from ynab_agent.budget.overspend import CategorySpend
19from ynab_agent.domain.allocations import (
20 ResolvedCategory,
21 ResolvedSplit,
22 ResolvedSplitLine,
24from ynab_agent.domain.enums import ClearedState, DecidedBy, FlagColor
25from ynab_agent.domain.ids import (
26 AccountId,
27 CategoryId,
28 ImportId,
29 PayeeId,
30 YnabTransactionId,
32from ynab_agent.domain.money import Money
33from ynab_agent.domain.transaction import YnabSnapshot
34from ynab_agent.policy.converge import TargetState
35from ynab_agent.ynab.wire import (
36 WireCategory,
37 WireMonth,
38 WireSubtransaction,
39 WireTransaction,
41 
42if TYPE_CHECKING:
43 import httpx
44 
45 from ynab_agent.domain.proposal import Decision
46 
47_API_KEY_ENV = "YNAB_API_KEY"
48_BUDGET_ENV = "YNAB_BUDGET_ID"
49_DEFAULT_BUDGET = "last-used"
50_BASE_URL = "https://api.ynab.com/v1"
51# The month identifier the balancer operates on (SPEC §8): YNAB's "current"
52# shorthand for the live budget month, matching the W6 monitor's current-month
53# figures. ``/months/{month}/...`` also accepts ``YYYY-MM-01`` for later use.
54CURRENT_MONTH = "current"
55# How far back the snapshot fallback list scans when the single GET 404s (an
56# unapproved txn). The agent only reads recent transactions, so a year is ample.
57_FALLBACK_LOOKBACK_DAYS = 365
58_HTTP_NOT_FOUND = 404
59# The flag an agent-applied write carries so the owner sees it in the YNAB app
60# and can clear it as implicit review of an auto-action (SPEC §14.5).
61AGENT_REVIEW_FLAG = FlagColor.PURPLE
62 
63 
64def _to_split_lines(
65 subs: tuple[WireSubtransaction, ...],
66) -> tuple[ResolvedSplitLine, ...]:
67 """Map a split's wire subtransactions onto domain split lines (SPEC §3).
68 
69 Skips deleted lines and any line without a category (an uncategorized line
70 cannot be a :class:`ResolvedSplitLine`); the agent only writes fully
71 categorized splits, so a clean read-back maps every line.
72 """
73 return tuple(
74 ResolvedSplitLine(
75 category=CategoryId(sub.category_id),
76 amount=Money.from_milliunits(sub.amount),
77 memo=sub.memo,
78 )
79 for sub in subs
80 if not sub.deleted and sub.category_id
81 )
82 
83 
84def to_snapshot(wire: WireTransaction) -> YnabSnapshot:
85 """Map a YNAB transaction onto the domain snapshot (SPEC §1)."""
86 return YnabSnapshot(
87 ynab_id=YnabTransactionId(wire.id),
88 account=AccountId(wire.account_id),
89 payee=wire.payee_name or "",
90 payee_id=PayeeId(wire.payee_id) if wire.payee_id else None,
91 amount=Money.from_milliunits(wire.amount),
92 txn_date=datetime.date.fromisoformat(wire.date),
93 memo=wire.memo,
94 flag=FlagColor(wire.flag_color) if wire.flag_color else None,
95 category_id=CategoryId(wire.category_id) if wire.category_id else None,
96 cleared=ClearedState(wire.cleared),
97 approved=wire.approved,
98 import_id=ImportId(wire.import_id) if wire.import_id else None,
99 matched_transaction_id=YnabTransactionId(wire.matched_transaction_id)
100 if wire.matched_transaction_id
101 else None,
102 subtransactions=_to_split_lines(wire.subtransactions),
103 )
104 
105 
106def to_category_spend(wire: WireCategory) -> CategorySpend:
107 """Map a YNAB category onto the domain spend figures (SPEC §7)."""
108 return CategorySpend(
109 category=CategoryId(wire.id),
110 name=wire.name,
111 budgeted=Money.from_milliunits(wire.budgeted),
112 activity=Money.from_milliunits(wire.activity),
113 balance=Money.from_milliunits(wire.balance),
114 )
115 
116 
117def to_target(snapshot: YnabSnapshot) -> TargetState | None:
118 """The read-back end-state of a transaction for verification (SPEC §3 r4).
119 
120 Reconstructs a split from its subtransactions so a split write verifies
121 field-by-field (a clean read-back hashes equal to the target); falls back to
122 a whole-category target; returns ``None`` only for an uncategorized txn (no
123 end-state to confirm), so the spine treats that as could-not-confirm rather
124 than a false divergence.
125 """
126 if len(snapshot.subtransactions) > 1: # a split has at least two lines
127 return TargetState(
128 allocation=ResolvedSplit(lines=snapshot.subtransactions),
129 memo=snapshot.memo,
130 approved=snapshot.approved,
131 )
132 if snapshot.category_id is None:
133 return None
134 return TargetState(
135 allocation=ResolvedCategory(category=snapshot.category_id),
136 memo=snapshot.memo,
137 approved=snapshot.approved,
138 )
⋯ 294 lines hidden (lines 139–432)
139 
140 
141def to_patch(decision: Decision) -> dict[str, object]:
142 """Map a committed decision onto a YNAB transaction patch (SPEC §3).
143 
144 A whole-category decision sets ``category_id``; a split sets a null category
145 and the subtransactions (YNAB's split shape). ``approved`` and an optional
146 memo ride along.
147 """
148 allocation = decision.allocation
149 fields: dict[str, object] = {"approved": decision.approved}
150 if decision.memo is not None:
151 fields["memo"] = decision.memo
152 # Flag an agent-applied write so it surfaces in the YNAB app for review
153 # (SPEC §14.5); a human decision leaves the existing flag untouched.
154 if decision.decided_by is DecidedBy.AGENT:
155 fields["flag_color"] = AGENT_REVIEW_FLAG.value
156 if isinstance(allocation, ResolvedCategory):
157 fields["category_id"] = str(allocation.category)
158 else:
159 fields["category_id"] = None
160 fields["subtransactions"] = [
161 {
162 "amount": line.amount.milliunits,
163 "category_id": str(line.category),
164 "memo": line.memo,
165 }
166 for line in allocation.lines
167 ]
168 return fields
169 
170 
171class YnabBackend(Protocol):
172 """The YNAB operations the client needs (implemented over httpx)."""
173 
174 def get_transaction(self, txn_id: str) -> WireTransaction | None:
175 """Fetch one transaction, or ``None`` if the GET 404s (unapproved)."""
176 ...
177 
178 def patch_transaction(self, txn_id: str, fields: dict[str, object]) -> None:
179 """Patch one transaction with the given fields."""
180 ...
181 
182 def list_categories(self) -> tuple[WireCategory, ...]:
183 """List the budget's live categories."""
184 ...
185 
186 def list_transactions(
187 self, since_date: str, server_knowledge: int | None
188 ) -> tuple[tuple[WireTransaction, ...], int]:
189 """The transactions delta and the advanced ``server_knowledge``."""
190 ...
191 
192 def list_unapproved(self) -> tuple[WireTransaction, ...]:
193 """The budget's ``type=unapproved`` transactions."""
194 ...
195 
196 def get_month(self, month: str) -> WireMonth:
197 """The budget month, for its ``to_be_budgeted`` (Ready-to-Assign)."""
198 ...
199 
200 def get_month_category(
201 self, month: str, category_id: str
202 ) -> WireCategory | None:
203 """One category's figures for ``month``, or ``None`` if the GET 404s."""
204 ...
205 
206 def set_category_budgeted(
207 self, month: str, category_id: str, budgeted_milliunits: int
208 ) -> None:
209 """Set a month category's ``budgeted`` to a milliunit value."""
210 ...
211 
212 
213# The process-wide cached client built by ``from_env`` (see its docstring).
214# ``tests/conftest.py`` resets this between tests.
215_CACHED: YnabClient | None = None
216 
217 
218class YnabClient:
219 """Reads and writes YNAB through a backend, in domain terms (SPEC §1)."""
220 
221 def __init__(self, backend: YnabBackend) -> None:
222 """Wrap a backend (the real httpx adapter, or a test fake)."""
223 self._backend = backend
224 
225 @classmethod
226 def from_env(cls) -> YnabClient:
227 """Build (once) a client backed by the real YNAB REST API.
228 
229 The pooled httpx client and its OTel instrumentation are built on first
230 use and cached: every activity invocation calls this on the worker's hot
231 path, so a fresh-per-call client (never closed) would leak connections +
232 file descriptors. Tests reset the cache (see ``tests/conftest.py``).
233 
234 Raises:
235 RuntimeError: If ``YNAB_API_KEY`` is not set.
236 """
237 global _CACHED
238 if _CACHED is not None:
239 return _CACHED
240 token = os.environ.get(_API_KEY_ENV)
241 if not token:
242 msg = f"{_API_KEY_ENV} is not set"
243 raise RuntimeError(msg)
244 budget = os.environ.get(_BUDGET_ENV, _DEFAULT_BUDGET)
245 import httpx
246 
247 client = httpx.Client(
248 base_url=_BASE_URL,
249 headers={"Authorization": f"Bearer {token}"},
250 timeout=30.0,
251 )
252 from ynab_agent.telemetry import instrument_httpx
253 
254 instrument_httpx(client)
255 _CACHED = cls(_HttpxBackend(client, budget))
256 return _CACHED
257 
258 def snapshot(self, txn_id: str) -> YnabSnapshot | None:
259 """The current snapshot of a transaction, or ``None`` if not found.
260 
261 YNAB's single-transaction GET returns 404 for *unapproved* transactions
262 — including matched/scheduled imports — which are exactly the ones the
263 agent triages. Those do appear in the transactions list, so on a miss we
264 fall back to a bounded list scan (the agent only ever reads recent
265 transactions). An approved transaction uses the cheap single GET.
266 """
267 wire = self._backend.get_transaction(txn_id)
268 if wire is None:
269 since = (
270 datetime.datetime.now(datetime.UTC).date()
271 - datetime.timedelta(days=_FALLBACK_LOOKBACK_DAYS)
272 ).isoformat()
273 # ``last_knowledge_of_server=0`` is required: matched/scheduled
274 # unapproved transactions appear only in the delta-from-zero, not a
275 # plain since_date list (a YNAB quirk).
276 wires, _ = self._backend.list_transactions(since, 0)
277 wire = next((w for w in wires if w.id == txn_id), None)
278 if wire is None or wire.deleted:
279 return None
280 return to_snapshot(wire)
281 
282 def category_spends(self) -> tuple[CategorySpend, ...]:
283 """Every live category's month-to-date budget figures (SPEC §7)."""
284 return tuple(
285 to_category_spend(c) for c in self._backend.list_categories()
286 )
287 
288 def unapproved(self) -> tuple[YnabSnapshot, ...]:
289 """The budget's currently *unapproved* transactions (SPEC §2 W1).
290 
291 This is YNAB's own ``type=unapproved`` view — the transactions awaiting
292 the owner's review. It is the agent's outstanding-work set: a
293 tentatively scheduled/auto-matched import is *not* in it (YNAB excludes
294 it until it is combined), so the agent ignores those until they truly
295 land. Deleted rows drop out. The poll re-reads this each tick — the set
296 is small, so no cursor is needed (and none is stored, SPEC §0.5).
297 """
298 return tuple(
299 to_snapshot(wire)
300 for wire in self._backend.list_unapproved()
301 if not wire.deleted
302 )
303 
304 def commit(self, txn_id: str, decision: Decision) -> None:
305 """Commit a decision to a transaction (SPEC §3)."""
306 self._backend.patch_transaction(txn_id, to_patch(decision))
307 
308 def read_back(self, txn_id: str) -> TargetState | None:
309 """Re-read a transaction's end-state for verification (SPEC §3 r4)."""
310 snapshot = self.snapshot(txn_id)
311 return to_target(snapshot) if snapshot is not None else None
312 
313 def ready_to_assign(self, month: str = CURRENT_MONTH) -> Money:
314 """The month's Ready-to-Assign — the first balancer source (SPEC §8)."""
315 return Money.from_milliunits(
316 self._backend.get_month(month).to_be_budgeted
317 )
318 
319 def set_budgeted(
320 self, category_id: str, target: Money, month: str = CURRENT_MONTH
321 ) -> None:
322 """Set a category's month ``budgeted`` to ``target`` (SPEC §8).
323 
324 An *absolute* write: the workflow computes each category's new budgeted
325 from a snapshot and sets it directly, so an activity retry is idempotent
326 (re-setting the same value is a no-op) — unlike a relative add.
327 """
328 self._backend.set_category_budgeted(
329 month, category_id, target.milliunits
330 )
331 
332 def read_budgeted(
333 self, category_id: str, month: str = CURRENT_MONTH
334 ) -> Money | None:
335 """Re-read a category's month ``budgeted`` for verification (SPEC §8).
336 
337 ``None`` when the category can't be read, so the spine treats it as
338 could-not-confirm rather than false divergence (cf. :meth:`read_back`).
339 """
340 wire = self._backend.get_month_category(month, category_id)
341 return (
342 Money.from_milliunits(wire.budgeted) if wire is not None else None
343 )
344 
345 
346class _HttpxBackend:
347 """Adapts the YNAB REST API to the :class:`YnabBackend` protocol."""
348 
349 def __init__(self, client: httpx.Client, budget_id: str) -> None:
350 self._client = client
351 self._budget = budget_id
352 
353 def get_transaction(self, txn_id: str) -> WireTransaction | None:
354 path = f"/budgets/{self._budget}/transactions/{txn_id}"
355 response = self._client.get(path)
356 if response.status_code == _HTTP_NOT_FOUND:
357 return None
358 response.raise_for_status()
359 return WireTransaction.model_validate(
360 response.json()["data"]["transaction"]
361 )
362 
363 def patch_transaction(self, txn_id: str, fields: dict[str, object]) -> None:
364 path = f"/budgets/{self._budget}/transactions/{txn_id}"
365 response = self._client.patch(path, json={"transaction": fields})
366 response.raise_for_status()
367 
368 def list_transactions(
369 self, since_date: str, server_knowledge: int | None
370 ) -> tuple[tuple[WireTransaction, ...], int]:
371 path = f"/budgets/{self._budget}/transactions"
372 params: dict[str, str | int] = {"since_date": since_date}
373 if server_knowledge is not None:
374 params["last_knowledge_of_server"] = server_knowledge
375 response = self._client.get(path, params=params)
376 response.raise_for_status()
377 data = response.json()["data"]
378 transactions = tuple(
379 WireTransaction.model_validate(item)
380 for item in data["transactions"]
381 )
382 return transactions, int(data["server_knowledge"])
383 
384 def list_unapproved(self) -> tuple[WireTransaction, ...]:
385 path = f"/budgets/{self._budget}/transactions"
386 response = self._client.get(path, params={"type": "unapproved"})
387 response.raise_for_status()
388 return tuple(
389 WireTransaction.model_validate(item)
390 for item in response.json()["data"]["transactions"]
391 )
392 
393 def list_categories(self) -> tuple[WireCategory, ...]:
394 path = f"/budgets/{self._budget}/categories"
395 response = self._client.get(path)
396 response.raise_for_status()
397 groups = response.json()["data"]["category_groups"]
398 return tuple(
399 WireCategory.model_validate(category)
400 for group in groups
401 for category in group["categories"]
402 if not category.get("deleted") and not category.get("hidden")
403 )
404 
405 def get_month(self, month: str) -> WireMonth:
406 path = f"/budgets/{self._budget}/months/{month}"
407 response = self._client.get(path)
408 response.raise_for_status()
409 return WireMonth.model_validate(response.json()["data"]["month"])
410 
411 def get_month_category(
412 self, month: str, category_id: str
413 ) -> WireCategory | None:
414 path = (
415 f"/budgets/{self._budget}/months/{month}/categories/{category_id}"
416 )
417 response = self._client.get(path)
418 if response.status_code == _HTTP_NOT_FOUND:
419 return None
420 response.raise_for_status()
421 return WireCategory.model_validate(response.json()["data"]["category"])
422 
423 def set_category_budgeted(
424 self, month: str, category_id: str, budgeted_milliunits: int
425 ) -> None:
426 path = (
427 f"/budgets/{self._budget}/months/{month}/categories/{category_id}"
428 )
429 response = self._client.patch(
430 path, json={"category": {"budgeted": budgeted_milliunits}}
431 )
432 response.raise_for_status()

Order-insensitive comparison

One subtlety: YNAB doesn't promise to return a split's subtransactions in the order they were written. A naive hash of the line tuple would then read a correct split as diverged. So content_hash canonically orders the lines before hashing — two equal splits hash the same regardless of order, while a genuinely different split still differs.

Canonicalize split-line order, then hash — the verify is order-insensitive.

src/ynab_agent/policy/converge.py · 123 lines
src/ynab_agent/policy/converge.py123 lines · Python
⋯ 49 lines hidden (lines 1–49)
1"""Converge-to-target reconciliation for REVISING (SPEC §3 rules 2-4).
2 
3REVISING is the riskiest path: it mutates approved money records. The spine
4converges to a single target end-state — read current YNAB state, write only if
5it differs, then verify field-by-field. These pure helpers compute the three
6decisions that procedure needs: the dedup hash, whether a write is needed, and
7how a read-after-write compares to the target.
8"""
9 
10from __future__ import annotations
11 
12import hashlib
13from typing import TYPE_CHECKING
14 
15from ynab_agent.domain.allocations import (
16 ResolvedAllocation,
17 ResolvedSplit,
18 ResolvedSplitLine,
20from ynab_agent.domain.base import Frozen
21from ynab_agent.domain.events import VerifyOutcome
22 
23if TYPE_CHECKING:
24 from ynab_agent.domain.proposal import Decision
25 from ynab_agent.domain.transaction import YnabSnapshot
26 
27 
28class TargetState(Frozen):
29 """The normalized end-state a write converges to.
30 
31 Only the fields a write actually sets — the allocation, the memo, and the
32 approved flag — so two decisions that produce the same YNAB end-state hash
33 identically (and dedup), regardless of incidental metadata.
34 """
35 
36 allocation: ResolvedAllocation
37 memo: str | None = None
38 approved: bool = True
39 
40 
41def target_of(decision: Decision) -> TargetState:
42 """Project a decision onto its YNAB end-state."""
43 return TargetState(
44 allocation=decision.allocation,
45 memo=decision.memo,
46 approved=decision.approved,
47 )
48 
49 
50def _split_line_key(line: ResolvedSplitLine) -> tuple[str, int, str]:
51 """A canonical sort key for a split line (category, amount, memo)."""
52 return (str(line.category), line.amount.milliunits, line.memo or "")
53 
54 
55def _canonical(target: TargetState) -> TargetState:
56 """Put a split's lines in a canonical order before hashing.
57 
58 YNAB does not promise to return a split's subtransactions in the order they
59 were written, so two equal splits must hash the same regardless of order. A
60 whole-category target is already canonical.
61 """
62 allocation = target.allocation
63 if isinstance(allocation, ResolvedSplit):
64 ordered = tuple(sorted(allocation.lines, key=_split_line_key))
65 return target.model_copy(
66 update={
67 "allocation": allocation.model_copy(update={"lines": ordered})
68 }
69 )
70 return target
71 
72 
73def content_hash(target: TargetState) -> str:
74 """A stable hash of the end-state, for ``(ynab_id, content_hash)`` dedup.
75 
76 Deterministic across processes (Pydantic JSON with fixed field order and
77 integer milliunits), so it is safe under Temporal replay. Split lines are
78 canonically ordered first, so a read-back that reorders them still hashes
79 equal to the target (SPEC §3 r4).
80 """
81 return hashlib.sha256(
82 _canonical(target).model_dump_json().encode()
83 ).hexdigest()
84 
⋯ 39 lines hidden (lines 85–123)
85 
86def needs_write(current: TargetState | None, target: TargetState) -> bool:
87 """Whether YNAB must be written (SPEC §3 rule 3).
88 
89 ``False`` only when the current state already equals the target — the
90 no-op exit that skips a needless intermediate write.
91 """
92 return current is None or content_hash(current) != content_hash(target)
93 
94 
95def classify_verify(
96 read_back: TargetState | None, target: TargetState
97) -> VerifyOutcome:
98 """Compare a read-after-write against the target (SPEC §3 rule 4).
99 
100 Args:
101 read_back: The post-write YNAB state, or ``None`` if it could not be
102 read after retries.
103 target: The intended end-state.
104 
105 Returns:
106 ``MATCH`` when they agree, ``COULD_NOT_CONFIRM`` when the read failed,
107 else ``DIVERGED`` (YNAB shows a different non-empty state).
108 """
109 if read_back is None:
110 return VerifyOutcome.COULD_NOT_CONFIRM
111 if content_hash(read_back) == content_hash(target):
112 return VerifyOutcome.MATCH
113 return VerifyOutcome.DIVERGED
114 
115 
116def reconciliation_blocks(snapshot: YnabSnapshot) -> bool:
117 """Whether the reconciliation guard forbids a silent edit (SPEC §3 rule 2).
118 
119 A transaction that is reconciled *or in a closed month* must not be silently
120 un-approved or edited; the spine should propose to the human (an additive
121 correction) instead.
122 """
123 return snapshot.reconciled or snapshot.month_closed

Tests

The discriminating test is end-to-end: a written split MATCHes on read-back even when YNAB returns the lines in the opposite order — the exact bug. The mapping and the order-insensitive hash are pinned directly too.

A split parent reconstructs to a ResolvedSplit.

tests/ynab/test_ynab_client.py · 400 lines
tests/ynab/test_ynab_client.py400 lines · Python
⋯ 262 lines hidden (lines 1–262)
1"""Tests for the YNAB client mappings and backend wiring (SPEC §1, §3, §7).
2 
3The pure mappings are tested against canned wire data; the client is tested over
4a fake backend. One opt-in live test (skipped unless ``YNAB_API_KEY`` is set)
5hits the real API.
6"""
7 
8from __future__ import annotations
9 
10import datetime
11import os
12 
13import pytest
14 
15from ynab_agent.domain.allocations import (
16 ResolvedCategory,
17 ResolvedSplit,
18 ResolvedSplitLine,
20from ynab_agent.domain.enums import ClearedState, DecidedBy
21from ynab_agent.domain.events import VerifyOutcome
22from ynab_agent.domain.ids import CategoryId, RuleId
23from ynab_agent.domain.money import Money
24from ynab_agent.domain.proposal import Decision
25from ynab_agent.policy.converge import classify_verify, target_of
26from ynab_agent.ynab.client import (
27 AGENT_REVIEW_FLAG,
28 YnabClient,
29 to_category_spend,
30 to_patch,
31 to_snapshot,
32 to_target,
34from ynab_agent.ynab.wire import WireCategory, WireMonth, WireTransaction
35 
36_NOW = datetime.datetime(2026, 5, 31, 12, 0, tzinfo=datetime.UTC)
37 
38 
39def _wire_txn(**kw: object) -> WireTransaction:
40 base: dict[str, object] = {
41 "id": "t1",
42 "account_id": "a1",
43 "date": "2026-05-28",
44 "amount": -4500,
45 "approved": False,
46 "payee_name": "Blue Bottle Coffee",
47 "cleared": "reconciled",
48 "category_id": "dining",
49 }
50 base.update(kw)
51 return WireTransaction.model_validate(base)
52 
53 
54def test_to_snapshot_maps_core_fields() -> None:
55 snap = to_snapshot(_wire_txn())
56 assert snap.ynab_id == "t1"
57 assert snap.payee == "Blue Bottle Coffee"
58 assert snap.amount == Money.from_currency("-4.50")
59 assert snap.txn_date == datetime.date(2026, 5, 28)
60 assert snap.cleared is ClearedState.RECONCILED
61 assert snap.reconciled is True
62 assert snap.category_id == "dining"
63 
64 
65def test_to_snapshot_handles_nulls() -> None:
66 snap = to_snapshot(
67 _wire_txn(payee_name=None, category_id=None, cleared="uncleared")
68 )
69 assert snap.payee == ""
70 assert snap.category_id is None
71 assert snap.cleared is ClearedState.UNCLEARED
72 
73 
74def test_to_category_spend_maps_figures() -> None:
75 spend = to_category_spend(
76 WireCategory(
77 id="dining",
78 name="Dining Out",
79 budgeted=400000,
80 activity=-210000,
81 balance=190000,
82 )
83 )
84 assert spend.category == "dining"
85 assert spend.budgeted == Money.from_currency("400")
86 assert spend.activity == Money.from_currency("-210")
87 
88 
89def test_to_patch_for_a_category_decision() -> None:
90 decision = Decision(
91 allocation=ResolvedCategory(category=CategoryId("dining")),
92 memo="coffee",
93 approved=True,
94 decided_by=DecidedBy.HUMAN,
95 decided_at=_NOW,
96 )
97 patch = to_patch(decision)
98 assert patch == {
99 "approved": True,
100 "memo": "coffee",
101 "category_id": "dining",
102 }
103 # A human decision leaves the flag untouched (no auto-action marker).
104 assert "flag_color" not in patch
105 
106 
107def test_to_patch_agent_decision_is_flagged_for_review() -> None:
108 # An agent-applied write carries the review flag so it surfaces in the YNAB
109 # app for the owner to clear as implicit review (SPEC §14.5).
110 decision = Decision(
111 allocation=ResolvedCategory(category=CategoryId("subscriptions")),
112 approved=True,
113 decided_by=DecidedBy.AGENT,
114 decided_at=_NOW,
115 rule_id=RuleId("r1"),
116 )
117 patch = to_patch(decision)
118 assert patch["flag_color"] == AGENT_REVIEW_FLAG.value
119 
120 
121def test_to_patch_for_a_split_decision() -> None:
122 decision = Decision(
123 allocation=ResolvedSplit(
124 lines=(
125 ResolvedSplitLine(
126 category=CategoryId("groceries"),
127 amount=Money.from_milliunits(-6000),
128 ),
129 ResolvedSplitLine(
130 category=CategoryId("household"),
131 amount=Money.from_milliunits(-4000),
132 ),
133 )
134 ),
135 approved=True,
136 decided_by=DecidedBy.HUMAN,
137 decided_at=_NOW,
138 )
139 patch = to_patch(decision)
140 assert patch["category_id"] is None
141 assert patch["subtransactions"] == [
142 {"amount": -6000, "category_id": "groceries", "memo": None},
143 {"amount": -4000, "category_id": "household", "memo": None},
144 ]
145 
146 
147class _FakeBackend:
148 def __init__(
149 self,
150 txn: WireTransaction,
151 *,
152 delta: tuple[tuple[WireTransaction, ...], int] = ((), 0),
153 unapproved: tuple[WireTransaction, ...] = (),
154 get_returns_none: bool = False,
155 to_be_budgeted: int = 0,
156 month_categories: dict[str, WireCategory] | None = None,
157 ) -> None:
158 self._txn = txn
159 self._delta = delta
160 self._unapproved = unapproved
161 self._get_returns_none = get_returns_none
162 self._to_be_budgeted = to_be_budgeted
163 self._month_categories = month_categories or {}
164 self.patched: list[tuple[str, dict[str, object]]] = []
165 self.delta_calls: list[tuple[str, int | None]] = []
166 self.budget_sets: list[tuple[str, str, int]] = []
167 
168 def get_transaction(self, txn_id: str) -> WireTransaction | None:
169 return None if self._get_returns_none else self._txn
170 
171 def patch_transaction(self, txn_id: str, fields: dict[str, object]) -> None:
172 self.patched.append((txn_id, fields))
173 
174 def list_categories(self) -> tuple[WireCategory, ...]:
175 return ()
176 
177 def list_transactions(
178 self, since_date: str, server_knowledge: int | None
179 ) -> tuple[tuple[WireTransaction, ...], int]:
180 self.delta_calls.append((since_date, server_knowledge))
181 return self._delta
182 
183 def list_unapproved(self) -> tuple[WireTransaction, ...]:
184 return self._unapproved
185 
186 def get_month(self, month: str) -> WireMonth:
187 return WireMonth(month=month, to_be_budgeted=self._to_be_budgeted)
188 
189 def get_month_category(
190 self, month: str, category_id: str
191 ) -> WireCategory | None:
192 return self._month_categories.get(category_id)
193 
194 def set_category_budgeted(
195 self, month: str, category_id: str, budgeted_milliunits: int
196 ) -> None:
197 self.budget_sets.append((month, category_id, budgeted_milliunits))
198 
199 
200def test_client_snapshot_maps_through_the_backend() -> None:
201 client = YnabClient(_FakeBackend(_wire_txn()))
202 snap = client.snapshot("t1")
203 assert snap is not None
204 assert snap.payee == "Blue Bottle Coffee"
205 
206 
207def test_client_snapshot_is_none_when_deleted() -> None:
208 client = YnabClient(_FakeBackend(_wire_txn(deleted=True)))
209 assert client.snapshot("t1") is None
210 
211 
212def test_client_snapshot_falls_back_to_list_for_unapproved() -> None:
213 # YNAB's single GET 404s (None) for unapproved txns; snapshot finds it in
214 # the transactions list instead.
215 wire = _wire_txn(approved=False)
216 backend = _FakeBackend(wire, get_returns_none=True, delta=((wire,), 5))
217 snap = YnabClient(backend).snapshot("t1")
218 assert snap is not None
219 assert snap.ynab_id == "t1"
220 # The fallback list is the delta-from-zero (cursor 0), which is the only
221 # form that surfaces matched/scheduled unapproved transactions.
222 assert backend.delta_calls and backend.delta_calls[0][1] == 0
223 
224 
225def test_client_snapshot_none_when_missing_everywhere() -> None:
226 # 404 on the single GET and absent from the list → genuinely gone.
227 backend = _FakeBackend(_wire_txn(), get_returns_none=True, delta=((), 0))
228 assert YnabClient(backend).snapshot("t1") is None
229 
230 
231def test_to_target_maps_a_categorized_snapshot() -> None:
232 target = to_target(to_snapshot(_wire_txn(memo="lunch")))
233 assert target is not None
234 assert isinstance(target.allocation, ResolvedCategory)
235 assert target.allocation.category == "dining"
236 assert target.memo == "lunch"
237 
238 
239def test_to_target_is_none_for_an_uncategorized_snapshot() -> None:
240 # No category and no subtransactions to verify → could-not-confirm.
241 assert to_target(to_snapshot(_wire_txn(category_id=None))) is None
242 
243 
244def test_to_snapshot_maps_split_subtransactions() -> None:
245 snap = to_snapshot(
246 _wire_txn(
247 category_id=None,
248 subtransactions=[
249 {"amount": -3000, "category_id": "groceries", "memo": "food"},
250 {"amount": -1500, "category_id": "gifts"},
251 # deleted and uncategorized lines are dropped.
252 {"amount": 0, "category_id": "x", "deleted": True},
253 {"amount": -500, "category_id": None},
254 ],
255 )
256 )
257 assert {str(line.category) for line in snap.subtransactions} == {
258 "groceries",
259 "gifts",
260 }
261 
262 
263def test_to_target_reconstructs_a_split() -> None:
264 # A split parent (null category) now verifies field-by-field from its
265 # subtransactions, rather than always reading as could-not-confirm.
266 target = to_target(
267 to_snapshot(
268 _wire_txn(
269 category_id=None,
270 subtransactions=[
271 {"amount": -3000, "category_id": "groceries"},
272 {"amount": -1500, "category_id": "gifts"},
273 ],
274 )
275 )
276 )
277 assert target is not None
278 assert isinstance(target.allocation, ResolvedSplit)
279 assert {str(line.category) for line in target.allocation.lines} == {
280 "groceries",
281 "gifts",
282 }
283 
284 
⋯ 116 lines hidden (lines 285–400)
285def test_split_write_verifies_match_even_if_lines_reorder() -> None:
286 # The end-to-end split bug: a written split must MATCH on read-back. YNAB
287 # may return the subtransactions in a different order, so the verify is
288 # order-insensitive (SPEC §3 r4).
289 decision = Decision(
290 allocation=ResolvedSplit(
291 lines=(
292 ResolvedSplitLine(
293 category=CategoryId("gifts"),
294 amount=Money.from_milliunits(-1500),
295 memo="present",
296 ),
297 ResolvedSplitLine(
298 category=CategoryId("groceries"),
299 amount=Money.from_milliunits(-3000),
300 memo="food",
301 ),
302 )
303 ),
304 approved=True,
305 decided_by=DecidedBy.HUMAN,
306 decided_at=_NOW,
307 )
308 # Read-back returns the lines in the opposite order, with the parent
309 # category nulled (YNAB's split shape).
310 read = to_target(
311 to_snapshot(
312 _wire_txn(
313 approved=True, # YNAB shows it approved after the write
314 category_id=None,
315 subtransactions=[
316 {
317 "amount": -3000,
318 "category_id": "groceries",
319 "memo": "food",
320 },
321 {
322 "amount": -1500,
323 "category_id": "gifts",
324 "memo": "present",
325 },
326 ],
327 )
328 )
329 )
330 assert classify_verify(read, target_of(decision)) is VerifyOutcome.MATCH
331 
332 
333def test_client_read_back_round_trips_to_a_target() -> None:
334 target = YnabClient(_FakeBackend(_wire_txn())).read_back("t1")
335 assert target is not None
336 assert isinstance(target.allocation, ResolvedCategory)
337 assert target.allocation.category == "dining"
338 
339 
340def test_client_commit_patches_the_transaction() -> None:
341 backend = _FakeBackend(_wire_txn())
342 decision = Decision(
343 allocation=ResolvedCategory(category=CategoryId("dining")),
344 approved=True,
345 decided_by=DecidedBy.AGENT,
346 decided_at=_NOW,
347 )
348 YnabClient(backend).commit("t1", decision)
349 assert backend.patched[0][0] == "t1"
350 assert backend.patched[0][1]["category_id"] == "dining"
351 
352 
353def test_client_unapproved_maps_snapshots_and_drops_deleted() -> None:
354 backend = _FakeBackend(
355 _wire_txn(),
356 unapproved=(_wire_txn(id="t1"), _wire_txn(id="t2", deleted=True)),
357 )
358 snapshots = YnabClient(backend).unapproved()
359 assert [s.ynab_id for s in snapshots] == ["t1"] # deleted t2 dropped
360 
361 
362def test_client_unapproved_is_empty_when_none() -> None:
363 backend = _FakeBackend(_wire_txn(), unapproved=())
364 assert YnabClient(backend).unapproved() == ()
365 
366 
367def test_client_ready_to_assign_reads_to_be_budgeted() -> None:
368 backend = _FakeBackend(_wire_txn(), to_be_budgeted=125000)
369 assert YnabClient(backend).ready_to_assign() == Money.from_currency("125")
370 
371 
372def test_client_set_budgeted_writes_an_absolute_value() -> None:
373 backend = _FakeBackend(_wire_txn())
374 YnabClient(backend).set_budgeted("dining", Money.from_currency("520"))
375 # An absolute milliunit write to the current month, idempotent on retry.
376 assert backend.budget_sets == [("current", "dining", 520000)]
377 
378 
379def test_client_read_budgeted_round_trips() -> None:
380 cat = WireCategory(
381 id="dining", name="Dining", budgeted=520000, activity=0, balance=520000
382 )
383 backend = _FakeBackend(_wire_txn(), month_categories={"dining": cat})
384 assert YnabClient(backend).read_budgeted("dining") == Money.from_currency(
385 "520"
386 )
387 
388 
389def test_client_read_budgeted_is_none_when_unread() -> None:
390 backend = _FakeBackend(_wire_txn(), month_categories={})
391 assert YnabClient(backend).read_budgeted("dining") is None
392 
393 
394@pytest.mark.skipif(
395 not os.environ.get("YNAB_API_KEY"),
396 reason="set YNAB_API_KEY to run the live YNAB smoke",
398def test_live_ynab_lists_categories() -> None:
399 spends = YnabClient.from_env().category_spends()
400 assert spends # a real budget has categories

content_hash is order-insensitive for splits.

tests/policy/test_converge.py · 110 lines
tests/policy/test_converge.py110 lines · Python
⋯ 54 lines hidden (lines 1–54)
1"""Tests for the converge-to-target reconciliation (SPEC §3 rules 2-4)."""
2 
3from __future__ import annotations
4 
5import datetime
6 
7from ynab_agent.domain.allocations import (
8 ResolvedCategory,
9 ResolvedSplit,
10 ResolvedSplitLine,
12from ynab_agent.domain.enums import ClearedState, DecidedBy
13from ynab_agent.domain.events import VerifyOutcome
14from ynab_agent.domain.ids import AccountId, CategoryId, YnabTransactionId
15from ynab_agent.domain.money import Money
16from ynab_agent.domain.proposal import Decision
17from ynab_agent.domain.transaction import YnabSnapshot
18from ynab_agent.policy.converge import (
19 TargetState,
20 classify_verify,
21 content_hash,
22 needs_write,
23 reconciliation_blocks,
24 target_of,
26 
27_EPOCH = datetime.datetime(2026, 5, 28, tzinfo=datetime.UTC)
28 
29 
30def _target(category: str = "dining", memo: str | None = None) -> TargetState:
31 return TargetState(
32 allocation=ResolvedCategory(category=CategoryId(category)), memo=memo
33 )
34 
35 
36def _split_target(*cats: str) -> TargetState:
37 return TargetState(
38 allocation=ResolvedSplit(
39 lines=tuple(
40 ResolvedSplitLine(
41 category=CategoryId(c), amount=Money.from_milliunits(-1000)
42 )
43 for c in cats
44 )
45 )
46 )
47 
48 
49def test_content_hash_is_stable_and_distinguishing() -> None:
50 assert content_hash(_target()) == content_hash(_target())
51 assert content_hash(_target("dining")) != content_hash(_target("gifts"))
52 assert content_hash(_target(memo="a")) != content_hash(_target(memo="b"))
53 
54 
55def test_content_hash_is_order_insensitive_for_splits() -> None:
56 # YNAB may return a split's lines in any order; two equal splits must hash
57 # the same so a read-back verifies as MATCH (SPEC §3 r4).
58 assert content_hash(_split_target("gifts", "groceries")) == content_hash(
59 _split_target("groceries", "gifts")
60 )
61 # A genuinely different split still hashes differently.
62 assert content_hash(_split_target("gifts", "groceries")) != content_hash(
63 _split_target("gifts", "dining")
64 )
65 
66 
⋯ 44 lines hidden (lines 67–110)
67def test_needs_write_skips_an_equal_state() -> None:
68 assert not needs_write(_target(), _target())
69 assert needs_write(_target("dining"), _target("gifts"))
70 assert needs_write(None, _target())
71 
72 
73def test_classify_verify_outcomes() -> None:
74 target = _target("dining")
75 assert classify_verify(target, target) is VerifyOutcome.MATCH
76 assert classify_verify(None, target) is VerifyOutcome.COULD_NOT_CONFIRM
77 assert classify_verify(_target("gifts"), target) is VerifyOutcome.DIVERGED
78 
79 
80def test_target_of_projects_a_decision() -> None:
81 decision = Decision(
82 allocation=ResolvedCategory(category=CategoryId("dining")),
83 memo="coffee",
84 approved=True,
85 decided_by=DecidedBy.AGENT,
86 decided_at=_EPOCH,
87 )
88 target = target_of(decision)
89 assert target.memo == "coffee"
90 assert content_hash(target) == content_hash(_target(memo="coffee"))
91 
92 
93def test_reconciliation_guard() -> None:
94 base: dict[str, object] = {
95 "ynab_id": YnabTransactionId("t1"),
96 "account": AccountId("a1"),
97 "payee": "Blue Bottle",
98 "amount": Money.from_currency("-4.50"),
99 "txn_date": datetime.date(2026, 5, 28),
100 }
101 reconciled = YnabSnapshot.model_validate(
102 {**base, "cleared": ClearedState.RECONCILED}
103 )
104 closed_month = YnabSnapshot.model_validate({**base, "month_closed": True})
105 cleared = YnabSnapshot.model_validate(
106 {**base, "cleared": ClearedState.CLEARED}
107 )
108 assert reconciliation_blocks(reconciled)
109 assert reconciliation_blocks(closed_month)
110 assert not reconciliation_blocks(cleared)