What changed, and why
In YNAB, approving an imported transaction accepts that import. So a YNAB-matched/duplicate import — one YNAB flagged as matching an existing txn — must never be auto-approved; it goes to a human (SPEC §13). The guard was unwired.
W1 did compute the condition: plan_ingest set route_to_human for a matched import. But address_transaction starts the W2 with the transaction id alone — the flag was dropped, "carried for observability only." And the autonomy gate never looked at the import state. So a matched import that happened to hit a blessed rule would be auto-applied, accepting the duplicate — the one thing §13 says not to do.
The guard, where autonomy is decided
The gate is the single path to auto-apply, so the guard belongs there. Right after the hard-floor check, a matched import forces ASK regardless of trust — before any blessed rule is even considered.
A second deterministic guard beside the floor: a matched import can never reach the blessed-rule AUTO path.
src/ynab_agent/policy/gate.py · 153 lines
src/ynab_agent/policy/gate.py
⋯ 99 lines hidden (lines 1–99)
1"""The autonomy gate: whether a proposal may auto-apply (SPEC §4.2, §1).
3Autonomy is authorized by *rules*, not raw model confidence (principle 6). The
4spine does not rank competing rules; it asks one question — does exactly one
5trusted rule clearly apply? If yes, that rule may gate auto-apply (still subject
6to the hard floor). If it is ambiguous (conflicting trusted rules, or none
7clearly applies), the transaction goes to ASK.
8"""
10from __future__ import annotations
12from enum import StrEnum
13from typing import TYPE_CHECKING
15from ynab_agent.domain.base import Frozen
16from ynab_agent.domain.enums import DecidedBy, RuleSource, TrustState
17from ynab_agent.domain.proposal import Decision
18from ynab_agent.policy.floor import (
19 CAUTIOUS_FLOOR,
20 AutoActionCounters,
21 FloorPolicy,
22 FloorVerdict,
23 check_floor,
24)
25from ynab_agent.policy.resolve import resolve_allocation
27if TYPE_CHECKING:
28 import datetime
29 from collections.abc import Iterable
31 from ynab_agent.domain.rule import Rule
32 from ynab_agent.domain.transaction import YnabSnapshot
35class GateVerdict(StrEnum):
36 """The autonomy gate's ruling."""
38 AUTO = "auto"
39 ASK = "ask"
42class GateOutcome(Frozen):
43 """The gate's decision and why.
45 Attributes:
46 verdict: ``AUTO`` (auto-apply eligible) or ``ASK`` (email a proposal).
47 rule_id: The single gating rule's id when ``AUTO``; ``None`` otherwise.
48 reason: A short, human-readable explanation (for the audit log).
49 """
51 verdict: GateVerdict
52 rule_id: str | None = None
53 reason: str = ""
56def rule_matches(rule: Rule, snapshot: YnabSnapshot) -> bool:
57 """Whether all of a rule's match conditions apply to a transaction (§1).
59 Payee matching is case-insensitive substring containment; an amount range is
60 compared in YNAB's signed convention; an item keyword is matched against the
61 memo (where item detail lands).
62 """
63 match = rule.match
64 if match.payee_pattern.lower() not in snapshot.payee.lower():
65 return False
66 if match.account is not None and match.account != snapshot.account:
67 return False
68 if match.amount_range is not None and not match.amount_range.contains(
69 snapshot.amount
70 ):
71 return False
72 if match.item_keyword is not None:
73 memo = snapshot.memo or ""
74 if match.item_keyword.lower() not in memo.lower():
75 return False
76 return True
79def matching_rules(rules: Iterable[Rule], snapshot: YnabSnapshot) -> list[Rule]:
80 """Return the rules whose conditions all apply to the transaction."""
81 return [rule for rule in rules if rule_matches(rule, snapshot)]
84def evaluate_gate(
85 snapshot: YnabSnapshot,
86 rules: Iterable[Rule],
87 counters: AutoActionCounters,
88 policy: FloorPolicy = CAUTIOUS_FLOOR,
89) -> GateOutcome:
90 """Decide whether the transaction may auto-apply (SPEC §4.2, §14). Pure.
92 The hard floor is consulted first; it can only force ASK, never grant AUTO.
93 A YNAB-matched/duplicate import is then routed to a human regardless of
94 trust (SPEC §13): auto-approving it would *accept the duplicate*. Only after
95 both deterministic guards is exactly one *blessed* matching rule required to
96 gate auto-apply. Per the §14 opt-in on-ramp, a learned rule that reaches
97 ``trusted`` by consistency is only *eligible* — it does not auto-apply until
98 the owner blesses it (``source=human_explicit``). So autonomy is always
99 granted, never taken: a trusted-but-unblessed rule still routes to ASK.
100 """
101 floor = check_floor(snapshot.amount, counters, policy)
102 if floor is not FloorVerdict.ALLOW:
103 return GateOutcome(verdict=GateVerdict.ASK, reason=f"floor: {floor}")
105 if snapshot.is_matched_import:
106 return GateOutcome(
107 verdict=GateVerdict.ASK,
108 reason="matched/duplicate import — accept by hand (SPEC §13)",
109 )
111 blessed = [
112 rule
113 for rule in matching_rules(rules, snapshot)
114 if rule.trust is TrustState.TRUSTED
115 and rule.source is RuleSource.HUMAN_EXPLICIT
116 ]
⋯ 37 lines hidden (lines 117–153)
117 if len(blessed) == 1:
118 return GateOutcome(
119 verdict=GateVerdict.AUTO,
120 rule_id=blessed[0].id,
121 reason="single blessed rule clearly applies",
122 )
123 if not blessed:
124 return GateOutcome(
125 verdict=GateVerdict.ASK, reason="no blessed rule applies"
126 )
127 return GateOutcome(
128 verdict=GateVerdict.ASK, reason="conflicting blessed rules"
129 )
132def build_auto_decision(
133 rule: Rule, snapshot: YnabSnapshot, decided_at: datetime.datetime
134) -> Decision:
135 """Resolve a gating rule's action into an approved agent decision.
137 Args:
138 rule: The single trusted rule the gate selected.
139 snapshot: The transaction to bind the rule's template against.
140 decided_at: The decision timestamp (passed in; the core reads no clock).
142 Returns:
143 An approved, agent-decided :class:`~.proposal.Decision`.
144 """
145 allocation = resolve_allocation(rule.action.allocation, snapshot.amount)
146 return Decision(
147 allocation=allocation,
148 memo=rule.action.memo_template,
149 approved=True,
150 decided_by=DecidedBy.AGENT,
151 decided_at=decided_at,
152 rule_id=rule.id,
153 )
One predicate, two readers
is_matched_import is a YnabSnapshot property, alongside reconciled / categorized / has_memo. It is the single source W1's observability flag and the gate's guard both read, so the two can't drift on what counts as a matched import.
The predicate, on the snapshot.
src/ynab_agent/domain/transaction.py · 265 lines
src/ynab_agent/domain/transaction.py
⋯ 84 lines hidden (lines 1–84)
1"""The transaction: a discriminated union over its lifecycle states (SPEC §3).
3Rather than one model with a ``state`` field and a pile of optional data (which
4admits illegal combinations like "applied, but no decision"), each state is its
5own frozen model carrying *exactly* the data valid in that state. mypy then
6proves, via ``assert_never`` in the state machine, that every state is handled.
8``Discovered`` is special: a transaction can be *born from a signal* before W1
9has polled its YNAB snapshot, so ``Discovered`` holds only the id (plus any
10buffered signals) and has no snapshot yet. Every other state embeds a
11:class:`TxnCore`, whose snapshot is always present.
12"""
14from __future__ import annotations
16import datetime
17from typing import Annotated, Literal
19from pydantic import Field, model_validator
21from ynab_agent.domain.allocations import ResolvedSplit
22from ynab_agent.domain.base import Frozen
23from ynab_agent.domain.enums import (
24 AwaitingFlag,
25 ClearedState,
26 DecidedBy,
27 FlagColor,
28 RevisingOrigin,
29 TxnState,
30)
31from ynab_agent.domain.ids import (
32 AccountId,
33 CategoryId,
34 ImportId,
35 PayeeId,
36 ReceiptId,
37 ThreadId,
38 YnabTransactionId,
39)
40from ynab_agent.domain.money import Money
41from ynab_agent.domain.proposal import Decision, Proposal
42from ynab_agent.domain.signals import InboundSignal
45class YnabSnapshot(Frozen):
46 """A point-in-time read of a YNAB transaction (re-read on each signal).
48 Reconciliation is derived, not stored alongside a contradictory flag: a
49 transaction is reconciled exactly when ``cleared is RECONCILED``, so the
50 illegal "reconciled but uncleared" combination cannot be represented.
51 """
53 ynab_id: YnabTransactionId
54 account: AccountId
55 payee: str
56 payee_id: PayeeId | None = None
57 amount: Money
58 txn_date: datetime.date
59 posted_at: datetime.datetime | None = None
60 memo: str | None = None
61 flag: FlagColor | None = None
62 category_id: CategoryId | None = None
63 cleared: ClearedState = ClearedState.UNCLEARED
64 approved: bool = False
65 month_closed: bool = False
66 import_id: ImportId | None = None
67 matched_transaction_id: YnabTransactionId | None = None
69 @property
70 def reconciled(self) -> bool:
71 """Whether YNAB considers this transaction reconciled."""
72 return self.cleared is ClearedState.RECONCILED
74 @property
75 def categorized(self) -> bool:
76 """Whether a YNAB category is assigned (LAPSED archive guard, §3)."""
77 return self.category_id is not None
79 @property
80 def has_memo(self) -> bool:
81 """Whether the memo is present and non-blank (Amazon-hold exit)."""
82 return bool(self.memo and self.memo.strip())
84 @property
85 def is_matched_import(self) -> bool:
86 """Whether YNAB matched this to an existing txn (SPEC §13).
88 Approving such an import *accepts the duplicate*, so it is never
89 auto-approved — the gate routes it to a human regardless of trust.
90 """
91 return self.matched_transaction_id is not None
⋯ 172 lines hidden (lines 94–265)
94class TxnCore(Frozen):
95 """The data common to every post-snapshot state."""
97 snapshot: YnabSnapshot
98 thread_id: ThreadId | None = None
99 receipt_ids: tuple[ReceiptId, ...] = ()
100 audit_log_ref: str | None = None
103def _ensure_committed(core: TxnCore, decision: Decision) -> None:
104 """Enforce the shared invariant of every post-write state (SPEC §3).
106 A written state must carry an approved decision, and a split's line amounts
107 must sum exactly to the transaction total.
109 Raises:
110 ValueError: If the decision is unapproved or a split is unbalanced.
111 """
112 if not decision.approved:
113 msg = "a post-write state requires an approved decision"
114 raise ValueError(msg)
115 allocation = decision.allocation
116 if (
117 isinstance(allocation, ResolvedSplit)
118 and allocation.total != core.snapshot.amount
119 ):
120 msg = "split lines must sum to the transaction amount"
121 raise ValueError(msg)
124# ── The lifecycle states ────────────────────────────────────────────────────
125class Discovered(Frozen):
126 """Born, snapshot possibly not yet polled. Buffers signals until it is."""
128 state: Literal[TxnState.DISCOVERED] = TxnState.DISCOVERED
129 ynab_id: YnabTransactionId
130 thread_id: ThreadId | None = None
131 pending: tuple[InboundSignal, ...] = ()
134class HoldAmazon(Frozen):
135 """An Amazon transaction with an empty memo, held for detail (SPEC §3)."""
137 state: Literal[TxnState.HOLD_AMAZON] = TxnState.HOLD_AMAZON
138 core: TxnCore
139 amazon_deadline: datetime.datetime
142class Enriching(Frozen):
143 """Assembling the proposal and routing via the autonomy gate."""
145 state: Literal[TxnState.ENRICHING] = TxnState.ENRICHING
146 core: TxnCore
147 proposal: Proposal | None = None
150class AutoApplied(Frozen):
151 """Written and approved by the agent under a blessed rule; pre-verify."""
153 state: Literal[TxnState.AUTO_APPLIED] = TxnState.AUTO_APPLIED
154 core: TxnCore
155 decision: Decision
157 @model_validator(mode="after")
158 def _check(self) -> AutoApplied:
159 if self.decision.decided_by is not DecidedBy.AGENT:
160 msg = "AUTO_APPLIED requires an agent decision"
161 raise ValueError(msg)
162 _ensure_committed(self.core, self.decision)
163 return self
166class AwaitingHuman(Frozen):
167 """A proposal or question is on the thread; a patience timer runs (§3)."""
169 state: Literal[TxnState.AWAITING_HUMAN] = TxnState.AWAITING_HUMAN
170 core: TxnCore
171 patience_deadline: datetime.datetime
172 proposal: Proposal | None = None
173 flag: AwaitingFlag = AwaitingFlag.NONE
176class Applied(Frozen):
177 """Written and approved via the thread (a reply, or a re-applied revision).
179 Unlike ``AUTO_APPLIED``, this is reached after thread interaction, so the
180 decision may be human- or agent-decided (e.g. a receipt-driven re-apply);
181 ``decision.decided_by`` records which.
182 """
184 state: Literal[TxnState.APPLIED] = TxnState.APPLIED
185 core: TxnCore
186 decision: Decision
188 @model_validator(mode="after")
189 def _check(self) -> Applied:
190 _ensure_committed(self.core, self.decision)
191 return self
194class Open(Frozen):
195 """The resting state after a verified write. A late inbound reopens it."""
197 state: Literal[TxnState.OPEN] = TxnState.OPEN
198 core: TxnCore
199 decision: Decision
201 @model_validator(mode="after")
202 def _check(self) -> Open:
203 _ensure_committed(self.core, self.decision)
204 return self
207class Lapsed(Frozen):
208 """Patience expired with no answer; handed back, never guessed (SPEC §3)."""
210 state: Literal[TxnState.LAPSED] = TxnState.LAPSED
211 core: TxnCore
212 proposal: Proposal | None = None
215class Revising(Frozen):
216 """One converge-to-target run; the newest instruction wins (SPEC §3)."""
218 state: Literal[TxnState.REVISING] = TxnState.REVISING
219 core: TxnCore
220 instruction: InboundSignal
221 origin: RevisingOrigin
222 prior: Decision | None = None
225class Archived(Frozen):
226 """Terminal: quiet past the window and reconciled (SPEC §3)."""
228 state: Literal[TxnState.ARCHIVED] = TxnState.ARCHIVED
229 core: TxnCore
230 final: Decision | None = None
232 @model_validator(mode="after")
233 def _check_archivable(self) -> Archived:
234 # Archiving is tied to reconciliation; a never-applied (LAPSED) archive
235 # additionally requires the txn be categorized, not merely reconciled
236 # (YNAB allows reconciling an uncategorized txn) (SPEC §3).
237 if not self.core.snapshot.reconciled:
238 msg = "ARCHIVED requires a reconciled snapshot"
239 raise ValueError(msg)
240 if self.final is None and not self.core.snapshot.categorized:
241 msg = "an unapplied ARCHIVED requires a categorized snapshot"
242 raise ValueError(msg)
243 return self
246Transaction = Annotated[
247 Discovered
248 | HoldAmazon
249 | Enriching
250 | AutoApplied
251 | AwaitingHuman
252 | Applied
253 | Open
254 | Lapsed
255 | Revising
256 | Archived,
257 Field(discriminator="state"),
258]
261def born(
262 ynab_id: YnabTransactionId, thread_id: ThreadId | None = None
263) -> Discovered:
264 """Create a freshly discovered transaction (snapshot not yet required)."""
265 return Discovered(ynab_id=ynab_id, thread_id=thread_id)
W1's is_duplicate_import now delegates to the property, rather than re-deriving matched_transaction_id is not None.
src/ynab_agent/ingest/plan.py · 78 lines
src/ynab_agent/ingest/plan.py
⋯ 38 lines hidden (lines 1–38)
1"""Ingestion planning: turn a YNAB delta into per-transaction W2 actions.
3W1 polls the YNAB ``transactions`` delta and, for each new/unapproved in-scope
4transaction, addresses its W2 by ``ynab_id`` via signal-with-start (SPEC §2).
5The *decision* of which transactions to address — and how — is pure and lives
6here; the spine does the signal-with-start. Two SPEC rules shape it:
8* **Cold-start cutover (§13).** On the first run the cursor is captured *without
9 acting*, so the agent never emails about the entire pre-existing backlog.
10* **Import lifecycle (§13).** A YNAB-matched/duplicate import is not
11 auto-approved — it is routed to a human.
12"""
14from __future__ import annotations
16from typing import TYPE_CHECKING
18from ynab_agent.domain.base import Frozen
19from ynab_agent.domain.transaction import YnabSnapshot
20from ynab_agent.ingest.scope import IngestScope, in_scope
22if TYPE_CHECKING:
23 from collections.abc import Iterable
26class AddressTxn(Frozen):
27 """A decision to address one transaction's W2 (start-or-signal).
29 Attributes:
30 snapshot: The polled YNAB snapshot to hand to the W2.
31 route_to_human: ``True`` for a matched/duplicate import — the W2 must
32 not auto-approve it (SPEC §13).
33 """
35 snapshot: YnabSnapshot
36 route_to_human: bool = False
39def is_duplicate_import(snapshot: YnabSnapshot) -> bool:
40 """Whether YNAB matched this import to an existing txn (SPEC §13).
42 Delegates to the snapshot's own predicate so W1's observability flag and the
43 gate's auto-apply guard read the *same* definition (no drift).
44 """
45 return snapshot.is_matched_import
⋯ 32 lines hidden (lines 47–78)
48def plan_ingest(
49 snapshots: Iterable[YnabSnapshot],
50 scope: IngestScope,
51 *,
52 cold_start: bool,
53) -> tuple[AddressTxn, ...]:
54 """Plan the W2 actions for one YNAB delta page. Pure.
56 Args:
57 snapshots: The transactions in this delta page (already normalized).
58 scope: The fail-closed ingestion scope.
59 cold_start: Whether this is the first poll (capture-cursor-only).
61 Returns:
62 One :class:`AddressTxn` per in-scope, unapproved txn — empty on cold
63 start.
65 Only *unapproved* transactions are addressed (SPEC §2/§13 "new/unapproved").
66 An approved transaction is the owner's settled state — the agent never
67 re-triages or emails about something already approved (whether the owner
68 approved it in YNAB or the agent's own triage did). This is what makes "act
69 on the current backlog" safe: only the outstanding, unapproved transactions
70 surface.
71 """
72 if cold_start:
73 return ()
74 return tuple(
75 AddressTxn(snapshot=snap, route_to_human=is_duplicate_import(snap))
76 for snap in snapshots
77 if in_scope(snap, scope) and not snap.approved
78 )
Tests
The discriminating test mirrors the floor-override test: a matched import ASKs even with a single blessed rule present — proving the guard sits ahead of the AUTO path. The property and the delegation are pinned too.
A blessed rule does not auto-apply a matched import.
tests/policy/test_gate.py · 177 lines
tests/policy/test_gate.py
⋯ 157 lines hidden (lines 1–157)
1"""Tests for the autonomy gate and rule matching (SPEC §4.2, §1)."""
3from __future__ import annotations
5import datetime
7from ynab_agent.domain.allocations import ProposedCategory, ResolvedCategory
8from ynab_agent.domain.enums import DecidedBy, RuleSource, TrustState
9from ynab_agent.domain.ids import (
10 AccountId,
11 CategoryId,
12 RuleId,
13 YnabTransactionId,
14)
15from ynab_agent.domain.money import Money
16from ynab_agent.domain.rule import AmountRange, Rule, RuleAction, RuleMatch
17from ynab_agent.domain.transaction import YnabSnapshot
18from ynab_agent.policy.floor import AutoActionCounters
19from ynab_agent.policy.gate import (
20 GateVerdict,
21 build_auto_decision,
22 evaluate_gate,
23 matching_rules,
24 rule_matches,
25)
27_EPOCH = datetime.datetime(2026, 5, 28, tzinfo=datetime.UTC)
30def _snapshot(**kw: object) -> YnabSnapshot:
31 base: dict[str, object] = {
32 "ynab_id": YnabTransactionId("t1"),
33 "account": AccountId("a1"),
34 "payee": "Blue Bottle Coffee",
35 "amount": Money.from_currency("-4.50"),
36 "txn_date": datetime.date(2026, 5, 28),
37 }
38 base.update(kw)
39 return YnabSnapshot.model_validate(base)
42def _rule(
43 trust: TrustState,
44 *,
45 payee: str = "Blue Bottle",
46 category: str = "dining",
47 rid: str = "r1",
48 match: RuleMatch | None = None,
49 source: RuleSource = RuleSource.LEARNED,
50) -> Rule:
51 return Rule(
52 id=RuleId(rid),
53 match=match or RuleMatch(payee_pattern=payee),
54 action=RuleAction(
55 allocation=ProposedCategory(category=CategoryId(category))
56 ),
57 trust=trust,
58 source=source,
59 )
62# A trusted, human-blessed rule is the only kind the gate auto-applies (§14).
63_BLESSED = RuleSource.HUMAN_EXPLICIT
66def test_rule_matches_payee_substring_case_insensitive() -> None:
67 assert rule_matches(
68 _rule(TrustState.TRUSTED, payee="blue bottle"), _snapshot()
69 )
72def test_rule_does_not_match_other_payee() -> None:
73 snap = _snapshot(payee="Costco")
74 assert not rule_matches(_rule(TrustState.TRUSTED, payee="Amazon"), snap)
77def test_rule_matches_amount_range() -> None:
78 band = RuleMatch(
79 payee_pattern="Blue Bottle",
80 amount_range=AmountRange(
81 low=Money.from_currency("-10"), high=Money.from_currency("0")
82 ),
83 )
84 assert rule_matches(_rule(TrustState.TRUSTED, match=band), _snapshot())
87# matching_rules — the filter that decides which rules reach the AUTO path.
88# A false match here auto-applies a wrong category with no human review, so the
89# AND logic and the None-memo guard are tested directly (test-backfill #1).
90def test_matching_rules_returns_matches() -> None:
91 rule = _rule(TrustState.SUGGESTED, payee="Blue Bottle")
92 assert matching_rules([rule], _snapshot()) == [rule]
95def test_matching_rules_excludes_payee_miss() -> None:
96 rule = _rule(TrustState.SUGGESTED, payee="Amazon")
97 assert matching_rules([rule], _snapshot(payee="Whole Foods")) == []
100def test_matching_rules_none_memo_with_item_keyword_is_excluded() -> None:
101 band = RuleMatch(payee_pattern="Blue Bottle", item_keyword="prime")
102 rule = _rule(TrustState.SUGGESTED, match=band)
103 # snapshot.memo is None → `memo or ""` guards against a crash and a match.
104 assert matching_rules([rule], _snapshot(memo=None)) == []
107def test_matching_rules_empty_input_returns_empty() -> None:
108 assert matching_rules([], _snapshot()) == []
111def test_single_blessed_rule_is_auto() -> None:
112 out = evaluate_gate(
113 _snapshot(),
114 [_rule(TrustState.TRUSTED, source=_BLESSED)],
115 AutoActionCounters(),
116 )
117 assert out.verdict is GateVerdict.AUTO
118 assert out.rule_id == "r1"
121def test_trusted_but_unblessed_rule_asks() -> None:
122 # Reaching `trusted` by consistency is eligibility, not autonomy: a learned
123 # rule still ASKs until the owner blesses it (§14 opt-in).
124 out = evaluate_gate(
125 _snapshot(), [_rule(TrustState.TRUSTED)], AutoActionCounters()
126 )
127 assert out.verdict is GateVerdict.ASK
128 assert "blessed" in out.reason
131def test_no_trusted_rule_asks() -> None:
132 out = evaluate_gate(
133 _snapshot(), [_rule(TrustState.CONFIRMED)], AutoActionCounters()
134 )
135 assert out.verdict is GateVerdict.ASK
138def test_conflicting_blessed_rules_ask() -> None:
139 rules = [
140 _rule(TrustState.TRUSTED, source=_BLESSED, rid="r1", category="dining"),
141 _rule(TrustState.TRUSTED, source=_BLESSED, rid="r2", category="coffee"),
142 ]
143 out = evaluate_gate(_snapshot(), rules, AutoActionCounters())
144 assert out.verdict is GateVerdict.ASK
147def test_floor_overrides_a_blessed_rule() -> None:
148 # Over the cautious ceiling, even a single blessed rule must ASK.
149 out = evaluate_gate(
150 _snapshot(amount=Money.from_currency("-200")),
151 [_rule(TrustState.TRUSTED, source=_BLESSED)],
152 AutoActionCounters(),
153 )
154 assert out.verdict is GateVerdict.ASK
155 assert "floor" in out.reason
158def test_matched_import_overrides_a_blessed_rule() -> None:
159 # A YNAB-matched/duplicate import is accepted by hand, never auto-approved —
160 # even a single blessed rule must ASK (SPEC §13 import lifecycle).
161 out = evaluate_gate(
162 _snapshot(matched_transaction_id=YnabTransactionId("t9")),
163 [_rule(TrustState.TRUSTED, source=_BLESSED)],
164 AutoActionCounters(),
165 )
166 assert out.verdict is GateVerdict.ASK
167 assert "import" in out.reason
⋯ 9 lines hidden (lines 169–177)
170def test_build_auto_decision() -> None:
171 rule = _rule(TrustState.TRUSTED)
172 decision = build_auto_decision(rule, _snapshot(), _EPOCH)
173 assert decision.approved
174 assert decision.decided_by is DecidedBy.AGENT
175 assert decision.rule_id == "r1"
176 assert isinstance(decision.allocation, ResolvedCategory)
177 assert decision.allocation.category == "dining"
The snapshot predicate flags a YNAB match.
tests/domain/test_entities.py · 189 lines
tests/domain/test_entities.py
⋯ 87 lines hidden (lines 1–87)
1"""Tests for domain entities and their construction invariants."""
3from __future__ import annotations
5import datetime
7import pytest
8from pydantic import ValidationError
10from ynab_agent.domain.allocations import (
11 ResolvedCategory,
12 ResolvedSplit,
13 ResolvedSplitLine,
14)
15from ynab_agent.domain.enums import ClearedState, DecidedBy
16from ynab_agent.domain.ids import (
17 AccountId,
18 CategoryId,
19 ReceiptId,
20 YnabTransactionId,
21)
22from ynab_agent.domain.money import Money
23from ynab_agent.domain.proposal import Decision
24from ynab_agent.domain.receipt import Receipt
25from ynab_agent.domain.rule import AmountRange, RuleMatch
26from ynab_agent.domain.transaction import (
27 Applied,
28 Archived,
29 AutoApplied,
30 Open,
31 TxnCore,
32 YnabSnapshot,
33)
35_EPOCH = datetime.datetime(2026, 5, 28, tzinfo=datetime.UTC)
38def _snapshot(**kw: object) -> YnabSnapshot:
39 base: dict[str, object] = {
40 "ynab_id": YnabTransactionId("t1"),
41 "account": AccountId("a1"),
42 "payee": "Blue Bottle",
43 "amount": Money.from_currency("-4.50"),
44 "txn_date": datetime.date(2026, 5, 28),
45 }
46 base.update(kw)
47 return YnabSnapshot.model_validate(base)
50def _decision(by: DecidedBy, *, approved: bool = True) -> Decision:
51 return Decision(
52 allocation=ResolvedCategory(category=CategoryId("dining")),
53 approved=approved,
54 decided_by=by,
55 decided_at=_EPOCH,
56 )
59def _split_decision(a: Money, b: Money) -> Decision:
60 return Decision(
61 allocation=ResolvedSplit(
62 lines=(
63 ResolvedSplitLine(category=CategoryId("x"), amount=a),
64 ResolvedSplitLine(category=CategoryId("y"), amount=b),
65 )
66 ),
67 approved=True,
68 decided_by=DecidedBy.HUMAN,
69 decided_at=_EPOCH,
70 )
73def test_reconciled_is_derived_from_cleared() -> None:
74 assert _snapshot(cleared=ClearedState.RECONCILED).reconciled
75 assert not _snapshot(cleared=ClearedState.CLEARED).reconciled
78def test_categorized_requires_a_category() -> None:
79 assert not _snapshot().categorized
80 assert _snapshot(category_id=CategoryId("c1")).categorized
83def test_has_memo_ignores_blank() -> None:
84 assert not _snapshot(memo=" ").has_memo
85 assert _snapshot(memo="AmazonBasics cable").has_memo
88def test_is_matched_import_flags_a_ynab_match() -> None:
89 assert not _snapshot().is_matched_import
90 assert _snapshot(
91 matched_transaction_id=YnabTransactionId("t9")
92 ).is_matched_import
⋯ 96 lines hidden (lines 94–189)
95def test_amount_range_rejects_inverted_bounds() -> None:
96 with pytest.raises(ValidationError):
97 AmountRange(low=Money.from_currency(100), high=Money.from_currency(10))
100def test_amount_range_contains() -> None:
101 band = AmountRange(
102 low=Money.from_currency(10), high=Money.from_currency(20)
103 )
104 assert band.contains(Money.from_currency(15))
105 assert not band.contains(Money.from_currency(5))
106 assert not band.contains(Money.from_currency(25))
109def test_open_band_contains() -> None:
110 band = AmountRange(low=Money.from_currency(10))
111 assert band.contains(Money.from_currency(1_000))
112 assert not band.contains(Money.from_currency(5))
115def test_rule_match_minimal() -> None:
116 match = RuleMatch(payee_pattern="Costco")
117 assert match.account is None
120def test_receipt_defaults_to_parked() -> None:
121 receipt = Receipt(id=ReceiptId("r1"), parked_at=_EPOCH)
122 assert receipt.status.value == "parked"
123 assert receipt.line_items == ()
126def test_auto_applied_requires_agent_decision() -> None:
127 core = TxnCore(snapshot=_snapshot())
128 with pytest.raises(ValidationError):
129 AutoApplied(core=core, decision=_decision(DecidedBy.HUMAN))
132def test_open_accepts_either_decider() -> None:
133 core = TxnCore(snapshot=_snapshot())
134 agent = Open(core=core, decision=_decision(DecidedBy.AGENT))
135 human = Open(core=core, decision=_decision(DecidedBy.HUMAN))
136 assert agent.decision.decided_by is DecidedBy.AGENT
137 assert human.decision.decided_by is DecidedBy.HUMAN
140def test_post_write_states_require_approved() -> None:
141 core = TxnCore(snapshot=_snapshot())
142 with pytest.raises(ValidationError):
143 AutoApplied(
144 core=core, decision=_decision(DecidedBy.AGENT, approved=False)
145 )
146 with pytest.raises(ValidationError):
147 Applied(core=core, decision=_decision(DecidedBy.HUMAN, approved=False))
148 with pytest.raises(ValidationError):
149 Open(core=core, decision=_decision(DecidedBy.HUMAN, approved=False))
152def test_applied_rejects_unbalanced_split() -> None:
153 # snapshot amount is -$4.50; this split sums to -$5.00.
154 core = TxnCore(snapshot=_snapshot())
155 bad = _split_decision(
156 Money.from_currency("-3.00"), Money.from_currency("-2.00")
157 )
158 with pytest.raises(ValidationError):
159 Applied(core=core, decision=bad)
162def test_applied_accepts_balanced_split() -> None:
163 core = TxnCore(snapshot=_snapshot()) # -$4.50
164 good = _split_decision(
165 Money.from_currency("-3.00"), Money.from_currency("-1.50")
166 )
167 assert Applied(core=core, decision=good).decision.approved
170def test_archived_requires_reconciled() -> None:
171 core = TxnCore(snapshot=_snapshot()) # uncleared
172 with pytest.raises(ValidationError):
173 Archived(core=core, final=_decision(DecidedBy.HUMAN))
176def test_archived_unapplied_requires_categorized() -> None:
177 # Reconciled but no category, and never applied (final=None).
178 core = TxnCore(snapshot=_snapshot(cleared=ClearedState.RECONCILED))
179 with pytest.raises(ValidationError):
180 Archived(core=core, final=None)
183def test_archived_valid_when_reconciled_and_categorized() -> None:
184 core = TxnCore(
185 snapshot=_snapshot(
186 cleared=ClearedState.RECONCILED, category_id=CategoryId("dining")
187 )
188 )
189 assert Archived(core=core, final=None).final is None