What changed, and why

Ollama's llama.cpp server keeps a KV prefix cache: when a new request shares a byte-identical prefix with a cached one, that prefix skips prefill entirely. Our two highest-traffic prompts — enrich (propose a category) and interpret (read the owner's reply) — were built backwards for it: per-call facts first, then the candidate category list. The category list is the one block that never changes between calls (one budget, one list) and it is most of the prompt, yet the cache could never match past the first line.

This PR only reorders. Candidates now lead, extending the static system prompt into one long stable prefix; the per-call facts trail. Same words, same facts — nothing about what the model sees is added or removed.

Enrich: candidates first, facts last

The system prompt's description of the input order is updated to match (stale docs are bugs), and _format_request flips: the stable block opens the user message, payee/amount/memo/hint close it.

The reordered prompt; the docstring states the prefix-cache contract.

src/ynab_agent/agentic/enrich.py · 344 lines
src/ynab_agent/agentic/enrich.py344 lines · Python
⋯ 72 lines hidden (lines 1–72)
1"""The enrichment agent: propose a category for a transaction (SPEC §4.1).
2 
3The agentic half of the ``enrich`` step. Given a transaction's facts and the
4candidate budget categories, a Pydantic AI agent picks the best category, rates
5its confidence, and explains why. The deterministic gate (``policy.gate``) then
6decides auto-apply vs. ask — confidence here is *framing only*, never a gate
7(principle 6). :func:`to_proposal` maps the agent's structured output onto the
8domain :class:`~ynab_agent.domain.proposal.Proposal`.
9 
10The model is injected per run so tests drive a ``TestModel``/``FunctionModel``
11offline; production uses :func:`~ynab_agent.agentic.model.build_model` (Ollama).
12"""
13 
14from __future__ import annotations
15 
16from typing import TYPE_CHECKING
17 
18from pydantic import Field
19from pydantic_ai import Agent
20 
21from ynab_agent.agentic.model import run_structured
22from ynab_agent.domain.allocations import ProposedCategory
23from ynab_agent.domain.base import Frozen
24from ynab_agent.domain.enums import Confidence, ReviewVerdict, SourceKind
25from ynab_agent.domain.events import AskHuman, AutoApply
26from ynab_agent.domain.ids import CategoryId
27from ynab_agent.domain.proposal import Proposal, ProposalSource
28from ynab_agent.policy.gate import (
29 GateVerdict,
30 build_auto_decision,
31 evaluate_gate,
32 matching_rules,
34 
35if TYPE_CHECKING:
36 import datetime
37 from collections.abc import Iterable
38 
39 from pydantic_ai.models import Model
40 
41 from ynab_agent.domain.events import EnrichmentOutcome
42 from ynab_agent.domain.rule import Rule
43 from ynab_agent.domain.transaction import YnabSnapshot
44 from ynab_agent.policy.floor import AutoActionCounters
45 
46 
47class CandidateCategory(Frozen):
48 """One budget category the agent may choose from."""
49 
50 id: str
51 name: str
52 
53 
54class EnrichmentRequest(Frozen):
55 """The facts the agent reasons over to propose a category."""
56 
57 payee: str
58 amount_display: str
59 candidates: tuple[CandidateCategory, ...] = Field(min_length=1)
60 memo: str | None = None
61 rule_hint: str | None = None
62 
63 
64class EnrichmentSuggestion(Frozen):
65 """The agent's structured proposal (mapped to a domain Proposal)."""
66 
67 category_id: str
68 confidence: Confidence
69 rationale: str
70 alternatives: tuple[str, ...] = ()
71 
72 
73_SYSTEM_PROMPT = """\
74You categorize a single bank transaction for a personal budget. You are given
75a list of candidate budget categories (each with an id and a name), then the
76transaction's facts: payee, amount, an optional memo, and an optional hint
77from a learned rule.
78 
79Choose the SINGLE best category for the transaction. Your `category_id` MUST be
80one of the provided candidate ids — never invent one. Also list up to 2
81`alternatives`: the next-most-likely candidate ids (also from the list, never
82the chosen one) the owner might prefer — or leave empty if none fit. Rate your
83confidence (high / medium / low) and give a one-sentence rationale. Confidence
84is framing only; a human or a trusted rule decides whether to auto-apply."""
85 
86_AGENT: Agent[None, EnrichmentSuggestion] = Agent(
87 output_type=EnrichmentSuggestion,
88 system_prompt=_SYSTEM_PROMPT,
90 
91 
92def _format_request(request: EnrichmentRequest) -> str:
93 """Render the request as the agent's user prompt.
94 
95 Ordered for KV prefix-cache reuse: the candidate category list is
96 byte-stable across calls (one budget, one category list), so it leads
97 and extends the shared prefix the server can skip re-prefilling; the
98 per-call facts (payee, amount, memo, hint) trail. With dozens of
99 categories the stable block is most of the prompt.
100 """
101 lines = ["Candidate categories:"]
102 lines.extend(f" - {c.name} (id: {c.id})" for c in request.candidates)
103 lines.append(f"Payee: {request.payee}")
104 lines.append(f"Amount: {request.amount_display}")
105 if request.memo:
106 lines.append(f"Memo: {request.memo}")
107 if request.rule_hint:
108 lines.append(f"Rule hint: {request.rule_hint}")
109 return "\n".join(lines)
⋯ 235 lines hidden (lines 110–344)
110 
111 
112async def propose(
113 request: EnrichmentRequest, *, model: Model | None = None
114) -> EnrichmentSuggestion:
115 """Run the enrichment agent for one transaction (SPEC §4.1).
116 
117 Args:
118 request: The transaction facts and candidate categories.
119 model: A model to use; defaults to the configured Ollama/Gemma.
120 
121 Returns:
122 The agent's structured category suggestion.
123 """
124 return await run_structured(
125 _AGENT,
126 _format_request(request),
127 output_type=EnrichmentSuggestion,
128 model=model,
129 )
130 
131 
132def validate_suggestion(
133 suggestion: EnrichmentSuggestion,
134 candidates: tuple[CandidateCategory, ...],
135) -> EnrichmentSuggestion:
136 """Reject a hallucinated category id; drop hallucinated alternatives.
137 
138 The model's ``category_id`` MUST be a candidate id — an invented one would
139 flow into the proposal and surface as a raw id in the email subject (and a
140 write against it would 400). Raising lets the enclosing activity retry,
141 which re-runs the model. Alternatives are framing only, so invalid ones are
142 silently dropped rather than failing the run.
143 """
144 valid = {c.id for c in candidates}
145 if suggestion.category_id not in valid:
146 msg = (
147 f"model proposed unknown category id "
148 f"{suggestion.category_id!r} (not a candidate)"
149 )
150 raise ValueError(msg)
151 kept = tuple(a for a in suggestion.alternatives if a in valid)
152 if kept == suggestion.alternatives:
153 return suggestion
154 return suggestion.model_copy(update={"alternatives": kept})
155 
156 
157def to_proposal(suggestion: EnrichmentSuggestion) -> Proposal:
158 """Map the agent's suggestion onto a domain Proposal (SPEC §4.1).
159 
160 The model's runner-up ids become the proposal's alternatives (the chosen
161 one filtered out defensively), surfaced in the email so the owner can pick
162 one at a glance.
163 """
164 alternatives = tuple(
165 CategoryId(alt)
166 for alt in suggestion.alternatives
167 if alt and alt != suggestion.category_id
168 )
169 return Proposal(
170 allocation=ProposedCategory(
171 category=CategoryId(suggestion.category_id)
172 ),
173 confidence=suggestion.confidence,
174 rationale=suggestion.rationale,
175 sources=(ProposalSource(kind=SourceKind.MODEL),),
176 alternatives=alternatives,
177 )
178 
179 
180def _blessed_category_id(rule: Rule) -> str | None:
181 """The single category a rule auto-applies, or ``None`` for a split.
182 
183 The safety review compares a single independent category against the rule's;
184 a split action has no single category to judge, so the review is skipped for
185 it (learned-eligible rules are always single-category, so this is rare).
186 """
187 allocation = rule.action.allocation
188 if isinstance(allocation, ProposedCategory):
189 return str(allocation.category)
190 return None
191 
192 
193def _rule_hint(
194 snapshot: YnabSnapshot,
195 rules: tuple[Rule, ...],
196 candidates: tuple[CandidateCategory, ...],
197) -> str | None:
198 """A one-line hint from the first matching rule, for the ASK proposal.
199 
200 A matching-but-not-blessed rule (learned, or eligible-awaiting-blessing)
201 still encodes how this payee was handled before — exactly the context the
202 proposing model should weigh. Only the ASK-path prompt gets it; the §0.6
203 clean-context review must stay blind to the rule's choice.
204 """
205 names = {c.id: c.name for c in candidates}
206 for rule in matching_rules(rules, snapshot):
207 category_id = _blessed_category_id(rule)
208 if category_id is not None and category_id in names:
209 return (
210 f"past transactions from this payee were filed under "
211 f"'{names[category_id]}'"
212 )
213 return None
214 
215 
216def review_auto_apply(
217 blessed_category_id: str, suggestion: EnrichmentSuggestion
218) -> ReviewVerdict:
219 """The agent-powered safety review's one-way ratchet (SPEC §0.6 Layer 2).
220 
221 An independent, *clean-context* model categorization (run with no knowledge
222 of the rule's choice, to stay unbiased) judges the impending auto-apply: it
223 ``PROCEED``s only if it considers the blessed category plausible — the
224 category it chose or listed as an alternative — and otherwise
225 ``ESCALATE_TO_HUMAN``. It can only hold an auto-apply back, never grant one,
226 so the deterministic gate still authorizes (principle 6).
227 """
228 plausible = {suggestion.category_id, *suggestion.alternatives}
229 if blessed_category_id in plausible:
230 return ReviewVerdict.PROCEED
231 return ReviewVerdict.ESCALATE_TO_HUMAN
232 
233 
234def _escalation_proposal(
235 suggestion: EnrichmentSuggestion, blessed_category_id: str
236) -> Proposal:
237 """The proposal emailed when the review holds an auto-apply back.
238 
239 Leads with the model's independent pick and offers the usual auto category
240 as an alternative, so the owner sees both views and the disagreement that
241 triggered the question.
242 """
243 base = to_proposal(suggestion)
244 blessed = CategoryId(blessed_category_id)
245 # to_proposal always builds a single-category allocation; narrow for mypy.
246 allocation = base.allocation
247 top = (
248 allocation.category
249 if isinstance(allocation, ProposedCategory)
250 else None
251 )
252 alternatives = base.alternatives
253 if blessed != top and blessed not in alternatives:
254 alternatives = (blessed, *alternatives)
255 return base.model_copy(
256 update={
257 "alternatives": alternatives,
258 "rationale": (
259 "I'd usually auto-file this payee, but this charge looked "
260 "different from the usual — confirming before I apply it."
261 ),
262 }
263 )
264 
265 
266async def decide_enrichment(
267 snapshot: YnabSnapshot,
268 candidates: tuple[CandidateCategory, ...],
269 rules: Iterable[Rule],
270 counters: AutoActionCounters,
271 *,
272 now: datetime.datetime,
273 model: Model | None = None,
274) -> EnrichmentOutcome:
275 """Compose the enrich step: gate, then a safety review (SPEC §4.1, §0.6).
276 
277 The deterministic gate decides autonomy from the rules alone — a single
278 blessed rule may auto-apply *its* action (the model never authorizes a
279 write, principle 6). Before an auto-apply lands, an independent
280 *clean-context* model review judges it (:func:`review_auto_apply`); a
281 disagreement holds it back to ASK (a one-way ratchet — it can only veto). A
282 gated ASK runs the same model to produce the best-guess proposal the email
283 shows.
284 
285 Args:
286 snapshot: The transaction being enriched.
287 candidates: The budget categories the agent may choose from.
288 rules: The rules in scope for the gate.
289 counters: The auto-action counters the hard floor reads.
290 now: The decision timestamp (the activity supplies it).
291 model: A model override for tests; defaults to Ollama/Gemma.
292 
293 Returns:
294 ``AutoApply`` when a blessed rule gates it *and* the review proceeds,
295 else ``AskHuman``.
296 """
297 rules = tuple(rules)
298 gate = evaluate_gate(snapshot, rules, counters)
299 if gate.verdict is GateVerdict.AUTO and gate.rule_id is not None:
300 rule = next((r for r in rules if r.id == gate.rule_id), None)
301 if rule is not None:
302 decision = build_auto_decision(rule, snapshot, now)
303 blessed_category_id = _blessed_category_id(rule)
304 if blessed_category_id is None:
305 return AutoApply(decision=decision)
306 # Clean context: the independent judge never sees the rule's
307 # choice — the memo is a transaction fact, so it rides along.
308 suggestion = validate_suggestion(
309 await propose(
310 EnrichmentRequest(
311 payee=snapshot.payee,
312 amount_display=str(snapshot.amount),
313 candidates=candidates,
314 memo=snapshot.memo,
315 ),
316 model=model,
317 ),
318 candidates,
319 )
320 if (
321 review_auto_apply(blessed_category_id, suggestion)
322 is ReviewVerdict.PROCEED
323 ):
324 return AutoApply(decision=decision)
325 return AskHuman(
326 proposal=_escalation_proposal(suggestion, blessed_category_id)
327 )
328 
329 # The ASK-path proposal gets every fact available: the memo and any
330 # matching rule's history (the clean-context review above never does).
331 suggestion = validate_suggestion(
332 await propose(
333 EnrichmentRequest(
334 payee=snapshot.payee,
335 amount_display=str(snapshot.amount),
336 candidates=candidates,
337 memo=snapshot.memo,
338 rule_hint=_rule_hint(snapshot, rules, candidates),
339 ),
340 model=model,
341 ),
342 candidates,
343 )
344 return AskHuman(proposal=to_proposal(suggestion))

Interpret: the reply goes last

Same flip, plus one deliberate placement: the reply — the most variable line of all — moves to the very end. That maximizes the shared prefix and also puts the text being classified nearest the generation point.

Candidates lead; the human's reply is the final line.

src/ynab_agent/agentic/interpret.py · 236 lines
src/ynab_agent/agentic/interpret.py236 lines · Python
⋯ 102 lines hidden (lines 1–102)
1"""The reply-interpreting agent: what did the human mean? (SPEC §3, §5).
2 
3The agentic half of W2's ``interpret_inbound``. Given a human's free-text reply
4on a transaction's thread, the model reads the *intent*: approve the current
5proposal, switch to a different category, or — when unclear — ask one clarifying
6question. The model only classifies intent and (for a switch) picks a category;
7the deterministic spine assembles the ``Decision`` (stamping the human as
8decider and the time), so the model never fabricates an approval timestamp or
9an allocation the spine cannot resolve.
10 
11:func:`to_reply_outcome` maps the intent onto the domain ``ReplyOutcome``,
12defaulting to a clarifying question whenever a switch arrives without a category
13— the safe move is always to ask, never to guess a write.
14"""
15 
16from __future__ import annotations
17 
18import re
19from enum import StrEnum
20from typing import TYPE_CHECKING, assert_never
21 
22from pydantic import Field
23from pydantic_ai import Agent
24 
25from ynab_agent.agentic.enrich import CandidateCategory
26from ynab_agent.agentic.model import run_structured
27from ynab_agent.domain.allocations import ResolvedCategory
28from ynab_agent.domain.base import Frozen
29from ynab_agent.domain.enums import DecidedBy
30from ynab_agent.domain.ids import CategoryId
31from ynab_agent.domain.proposal import Decision
32from ynab_agent.workflow.types import (
33 AnswerOutcome,
34 ClarifyOutcome,
35 ReplyOutcome,
37 
38if TYPE_CHECKING:
39 import datetime
40 
41 from pydantic_ai.models import Model
42 
43 
44class ReplyIntent(StrEnum):
45 """What a human's reply asked for."""
46 
47 APPROVE = "approve"
48 RECATEGORIZE = "recategorize"
49 CLARIFY = "clarify"
50 
51 
52class InterpretRequest(Frozen):
53 """A human reply and the context needed to read its intent.
54 
55 ``proposed_category_name`` is ``None`` when there is no live proposal (a
56 flagged verify-failure entry, a NeedsHuman wait). The owner can still name
57 a category outright; only a bare "approve" has nothing to approve then.
58 """
59 
60 reply_text: str
61 payee: str
62 amount_display: str
63 proposed_category_name: str | None = None
64 candidates: tuple[CandidateCategory, ...] = Field(min_length=1)
65 
66 
67class Interpretation(Frozen):
68 """The agent's read of the reply (mapped to a domain ReplyOutcome)."""
69 
70 intent: ReplyIntent
71 category_id: str | None = None
72 question: str | None = None
73 memo: str | None = None
74 
75 
76_SYSTEM_PROMPT = """\
77You read one reply a human sent about a proposed transaction categorization. You
78are given the candidate categories (id + name), the payee and amount, the
79category currently proposed, and finally the reply text itself.
80 
81Classify the reply's intent: `approve` if they accept the current proposal (e.g.
82"ok", "yes", "sounds good"); `recategorize` if they name or describe a category
83— then set `category_id` to the matching candidate id; or `clarify` if the reply
84is ambiguous or asks a question — then set a short `question` to send back. When
85in doubt, prefer `clarify`: asking again is safe, a wrong write is not.
86 
87There may be NO current proposal (shown as "(none)"): the agent asked an open
88question rather than proposing. A bare yes/ok then has nothing to approve —
89classify it `clarify`; a reply that names a category is still `recategorize`.
90 
91Also set `memo`: if the reply gives any *context or reasoning* beyond the bare
92category — what it was for, who it was for, an occasion ("gift for mom", "kids'
93soccer", "their shopping") — distil it into a short, factual memo (≤ a sentence)
94that will be saved on the transaction. If the reply is only a bare confirmation
95or category with no added context, leave `memo` null. Never invent detail."""
96 
97_AGENT: Agent[None, Interpretation] = Agent(
98 output_type=Interpretation,
99 system_prompt=_SYSTEM_PROMPT,
101 
102 
103def _format_request(request: InterpretRequest) -> str:
104 """Render the request as the agent's user prompt.
105 
106 Ordered for KV prefix-cache reuse: the candidate category list is
107 byte-stable across calls, so it leads and extends the shared prefix the
108 server can skip re-prefilling; the per-call facts follow, with the
109 reply — the most variable line — last (which also puts it nearest the
110 generation point).
111 """
112 lines = ["Candidate categories:"]
113 lines.extend(f" - {c.name} (id: {c.id})" for c in request.candidates)
114 lines.append(f"Payee: {request.payee}")
115 lines.append(f"Amount: {request.amount_display}")
116 lines.append(
117 f"Currently proposed: {request.proposed_category_name or '(none)'}"
118 )
119 lines.append(f"Reply: {request.reply_text}")
120 return "\n".join(lines)
121 
⋯ 115 lines hidden (lines 122–236)
122 
123async def interpret(
124 request: InterpretRequest, *, model: Model | None = None
125) -> Interpretation:
126 """Run the reply-interpreting agent for one reply (SPEC §3).
127 
128 Args:
129 request: The reply text and its categorization context.
130 model: A model to use; defaults to the configured Ollama/Gemma.
131 
132 Returns:
133 The agent's structured reading of the reply.
134 """
135 return await run_structured(
136 _AGENT,
137 _format_request(request),
138 output_type=Interpretation,
139 model=model,
140 )
141 
142 
143def _supported_memo(memo: str | None, reply_text: str) -> str | None:
144 """Keep a memo only when the reply actually contains its substance.
145 
146 The memo is *defined* as context distilled from the reply ("never invent
147 detail"), but the model occasionally fabricates one anyway — e.g. echoing
148 a category name — and a fabricated memo overwrites real YNAB detail (an
149 Amazon item list, gone). Deterministic guard: at least one substantive
150 word of the memo must appear in the reply, else the memo is dropped (the
151 safe loss — a nice-to-have note — over the unsafe write).
152 """
153 if memo is None:
154 return None
155 words = re.findall(r"[A-Za-z0-9']{3,}", memo.lower())
156 if not words:
157 return None
158 reply = reply_text.lower()
159 return memo if any(word in reply for word in words) else None
160 
161 
162def _human_decision(
163 category: CategoryId,
164 decided_at: datetime.datetime,
165 *,
166 memo: str | None = None,
167) -> Decision:
168 return Decision(
169 allocation=ResolvedCategory(category=category),
170 memo=memo,
171 approved=True,
172 decided_by=DecidedBy.HUMAN,
173 decided_at=decided_at,
174 )
175 
176 
177def to_reply_outcome(
178 interpretation: Interpretation,
179 *,
180 proposed_category: CategoryId | None,
181 decided_at: datetime.datetime,
182 candidates: tuple[CandidateCategory, ...],
183 reply_text: str,
184) -> ReplyOutcome:
185 """Map the agent's reading onto a domain ReplyOutcome (SPEC §3, §14.4).
186 
187 ``approve`` commits the proposed category (or asks, when nothing was
188 proposed — there is nothing to approve); ``recategorize`` commits the named
189 one (or asks, if none was given *or* the model invented an id — a write
190 against a hallucinated category would land wrong or 400); ``clarify`` sends
191 the question back. Any rationale the reply carried rides along as the
192 decision's ``memo`` (the spine writes it to YNAB) — but only when the
193 reply actually contains it (:func:`_supported_memo`). The spine, not the
194 model, sets decider and time.
195 """
196 memo = _supported_memo(interpretation.memo, reply_text)
197 match interpretation.intent:
198 case ReplyIntent.APPROVE:
199 if proposed_category is None:
200 return ClarifyOutcome(
201 question=(
202 "There's no pending suggestion to approve here — "
203 "which category should this be?"
204 )
205 )
206 return AnswerOutcome(
207 decision=_human_decision(
208 proposed_category, decided_at, memo=memo
209 )
210 )
211 case ReplyIntent.RECATEGORIZE:
212 if interpretation.category_id and any(
213 c.id == interpretation.category_id for c in candidates
214 ):
215 return AnswerOutcome(
216 decision=_human_decision(
217 CategoryId(interpretation.category_id),
218 decided_at,
219 memo=memo,
220 )
221 )
222 if interpretation.category_id:
223 return ClarifyOutcome(
224 question=(
225 "I couldn't match that to one of the budget's "
226 "categories — could you give the category name as "
227 "it appears in YNAB?"
228 )
229 )
230 return ClarifyOutcome(question="Which category should this be?")
231 case ReplyIntent.CLARIFY:
232 return ClarifyOutcome(
233 question=interpretation.question
234 or "Could you clarify how you'd like this categorized?"
235 )
236 assert_never(interpretation.intent)

The ordering contract, pinned

Two small tests assert the property the whole PR exists for: the candidate block precedes the per-call facts, and interpret's reply line is last. They fail on exactly the reordering regression, nothing else.

Enrich: candidates before payee; the hint stays the final line.

tests/agentic/test_enrich.py · 398 lines
tests/agentic/test_enrich.py398 lines · Python
⋯ 378 lines hidden (lines 1–378)
1"""Tests for the enrichment agent (SPEC §4.1, §0.5).
2 
3Offline tests drive a ``TestModel`` (no network, deterministic) to exercise the
4agent wiring and the domain mapping. One opt-in live test runs the real Ollama
5Gemma model — SPEC §0.5 spike #2 — and is skipped unless
6``YNAB_AGENT_LIVE_OLLAMA`` is set, so the gate stays fast and offline.
7"""
8 
9from __future__ import annotations
10 
11import datetime
12import os
13 
14import pytest
15from pydantic_ai.models import Model
16from pydantic_ai.models.test import TestModel
17 
18from ynab_agent.agentic.enrich import (
19 CandidateCategory,
20 EnrichmentRequest,
21 EnrichmentSuggestion,
22 decide_enrichment,
23 propose,
24 review_auto_apply,
25 to_proposal,
26 validate_suggestion,
28from ynab_agent.agentic.model import build_model
29from ynab_agent.domain.allocations import ProposedCategory
30from ynab_agent.domain.enums import (
31 Confidence,
32 ReviewVerdict,
33 RuleSource,
34 SourceKind,
35 TrustState,
37from ynab_agent.domain.events import AskHuman, AutoApply
38from ynab_agent.domain.ids import (
39 AccountId,
40 CategoryId,
41 RuleId,
42 YnabTransactionId,
44from ynab_agent.domain.money import Money
45from ynab_agent.domain.rule import Rule, RuleAction, RuleMatch
46from ynab_agent.domain.transaction import YnabSnapshot
47from ynab_agent.policy.floor import AutoActionCounters
48 
49_REQUEST = EnrichmentRequest(
50 payee="Blue Bottle Coffee",
51 amount_display="-$4.50",
52 candidates=(
53 CandidateCategory(id="dining", name="Dining Out"),
54 CandidateCategory(id="groceries", name="Groceries"),
55 ),
57 
58 
59async def test_propose_returns_the_models_structured_suggestion() -> None:
60 model = TestModel(
61 custom_output_args={
62 "category_id": "dining",
63 "confidence": "high",
64 "rationale": "a coffee shop",
65 }
66 )
67 out = await propose(_REQUEST, model=model)
68 assert isinstance(out, EnrichmentSuggestion)
69 assert out.category_id == "dining"
70 assert out.confidence is Confidence.HIGH
71 assert out.rationale == "a coffee shop"
72 
73 
74async def test_propose_wiring_smoke_with_default_testmodel() -> None:
75 # No custom output → TestModel autofills a valid suggestion; this just
76 # proves the agent/output-schema wiring round-trips.
77 out = await propose(_REQUEST, model=TestModel())
78 assert isinstance(out, EnrichmentSuggestion)
79 
80 
81def test_to_proposal_maps_onto_the_domain_proposal() -> None:
82 suggestion = EnrichmentSuggestion(
83 category_id="dining",
84 confidence=Confidence.MEDIUM,
85 rationale="coffee",
86 )
87 proposal = to_proposal(suggestion)
88 assert isinstance(proposal.allocation, ProposedCategory)
89 assert proposal.allocation.category == "dining"
90 assert proposal.confidence is Confidence.MEDIUM
91 assert proposal.sources[0].kind is SourceKind.MODEL
92 
93 
94def test_to_proposal_carries_alternatives_filtering_the_chosen() -> None:
95 suggestion = EnrichmentSuggestion(
96 category_id="dining",
97 confidence=Confidence.LOW,
98 rationale="maybe",
99 # the chosen id and an empty entry are filtered out defensively
100 alternatives=("coffee", "dining", "", "groceries"),
101 )
102 proposal = to_proposal(suggestion)
103 assert [str(a) for a in proposal.alternatives] == ["coffee", "groceries"]
104 
105 
106def test_build_model_constructs_an_ollama_model() -> None:
107 assert isinstance(build_model(model_name="gemma4:e4b"), Model)
108 
109 
110def test_validate_suggestion_rejects_a_hallucinated_id() -> None:
111 # An invented id would surface as a raw id in the email subject and 400 on
112 # write — raising lets the activity retry with a fresh model run.
113 bogus = EnrichmentSuggestion(
114 category_id="10683d916894", confidence=Confidence.HIGH, rationale="x"
115 )
116 with pytest.raises(ValueError, match="unknown category id"):
117 validate_suggestion(bogus, _REQUEST.candidates)
118 
119 
120def test_validate_suggestion_drops_hallucinated_alternatives() -> None:
121 # Alternatives are framing only — invalid ones are dropped, not fatal.
122 mixed = EnrichmentSuggestion(
123 category_id="dining",
124 confidence=Confidence.HIGH,
125 rationale="x",
126 alternatives=("groceries", "10683d916894"),
127 )
128 assert validate_suggestion(mixed, _REQUEST.candidates).alternatives == (
129 "groceries",
130 )
131 
132 
133def test_review_auto_apply_proceeds_when_blessed_is_plausible() -> None:
134 top = EnrichmentSuggestion(
135 category_id="dining", confidence=Confidence.HIGH, rationale="x"
136 )
137 assert review_auto_apply("dining", top) is ReviewVerdict.PROCEED
138 alt = EnrichmentSuggestion(
139 category_id="groceries",
140 confidence=Confidence.HIGH,
141 rationale="x",
142 alternatives=("dining",),
143 )
144 assert review_auto_apply("dining", alt) is ReviewVerdict.PROCEED
145 
146 
147def test_review_auto_apply_escalates_when_blessed_is_implausible() -> None:
148 other = EnrichmentSuggestion(
149 category_id="groceries",
150 confidence=Confidence.HIGH,
151 rationale="x",
152 alternatives=("software",),
153 )
154 assert review_auto_apply("dining", other) is ReviewVerdict.ESCALATE_TO_HUMAN
155 
156 
157_NOW = datetime.datetime(2026, 5, 31, 12, 0, tzinfo=datetime.UTC)
158 
159 
160def _snapshot() -> YnabSnapshot:
161 return YnabSnapshot(
162 ynab_id=YnabTransactionId("t1"),
163 account=AccountId("a1"),
164 payee="Blue Bottle Coffee",
165 amount=Money.from_currency("-4.50"),
166 txn_date=datetime.date(2026, 5, 28),
167 )
168 
169 
170def _blessed_rule() -> Rule:
171 return Rule(
172 id=RuleId("r1"),
173 match=RuleMatch(payee_pattern="Blue Bottle"),
174 action=RuleAction(
175 allocation=ProposedCategory(category=CategoryId("dining"))
176 ),
177 trust=TrustState.TRUSTED,
178 source=RuleSource.HUMAN_EXPLICIT,
179 )
180 
181 
182def _judge(category_id: str, alternatives: tuple[str, ...] = ()) -> TestModel:
183 """A TestModel whose suggestion stands in for the review's judgment."""
184 return TestModel(
185 custom_output_args={
186 "category_id": category_id,
187 "confidence": "high",
188 "rationale": "independent read",
189 "alternatives": list(alternatives),
190 }
191 )
192 
193 
194async def test_decide_enrichment_auto_applies_when_the_review_agrees() -> None:
195 # The independent review picks the same category the blessed rule would →
196 # the auto-apply proceeds (SPEC §0.6 Layer 2).
197 outcome = await decide_enrichment(
198 _snapshot(),
199 _REQUEST.candidates,
200 [_blessed_rule()],
201 AutoActionCounters(),
202 now=_NOW,
203 model=_judge("dining"),
204 )
205 assert isinstance(outcome, AutoApply)
206 assert outcome.decision.rule_id == "r1"
207 
208 
209async def test_decide_enrichment_review_proceeds_if_blessed_is_an_alt() -> None:
210 # The review's top pick differs but it still considers the blessed category
211 # plausible (an alternative) → proceed.
212 outcome = await decide_enrichment(
213 _snapshot(),
214 _REQUEST.candidates,
215 [_blessed_rule()],
216 AutoActionCounters(),
217 now=_NOW,
218 model=_judge("groceries", alternatives=("dining",)),
219 )
220 assert isinstance(outcome, AutoApply)
221 
222 
223async def test_decide_enrichment_review_escalates_on_disagreement() -> None:
224 # The independent review does not find the blessed category plausible →
225 # the one-way ratchet holds the auto-apply back to ASK.
226 outcome = await decide_enrichment(
227 _snapshot(),
228 _REQUEST.candidates,
229 [_blessed_rule()],
230 AutoActionCounters(),
231 now=_NOW,
232 model=_judge("groceries"),
233 )
234 assert isinstance(outcome, AskHuman)
235 assert isinstance(outcome.proposal.allocation, ProposedCategory)
236 # Leads with the model's independent pick; offers the usual auto category.
237 assert outcome.proposal.allocation.category == "groceries"
238 assert CategoryId("dining") in outcome.proposal.alternatives
239 
240 
241def _learned_rule() -> Rule:
242 # Matches the snapshot's payee but is NOT blessed — it cannot gate an
243 # auto-apply, only inform the ASK proposal.
244 return Rule(
245 id=RuleId("r2"),
246 match=RuleMatch(payee_pattern="Blue Bottle"),
247 action=RuleAction(
248 allocation=ProposedCategory(category=CategoryId("dining"))
249 ),
250 trust=TrustState.CONFIRMED,
251 source=RuleSource.LEARNED,
252 )
253 
254 
255def _capturing_model(prompts: list[str]) -> Model:
256 """A FunctionModel that records the user prompt and answers 'dining'."""
257 from pydantic_ai.messages import (
258 ModelMessage,
259 ModelResponse,
260 ToolCallPart,
261 UserPromptPart,
262 )
263 from pydantic_ai.models.function import AgentInfo, FunctionModel
264 
265 def call(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
266 for message in messages:
267 for part in message.parts:
268 if isinstance(part, UserPromptPart):
269 prompts.append(str(part.content))
270 assert info.output_tools
271 return ModelResponse(
272 parts=[
273 ToolCallPart(
274 tool_name=info.output_tools[0].name,
275 args={
276 "category_id": "dining",
277 "confidence": "high",
278 "rationale": "x",
279 },
280 )
281 ]
282 )
283 
284 return FunctionModel(call)
285 
286 
287async def test_ask_prompt_carries_the_memo_and_the_rule_hint() -> None:
288 # The model finally *sees* the memo and the payee's prior handling — the
289 # context that turns "10683d916894?" guesses into informed proposals.
290 prompts: list[str] = []
291 snapshot = _snapshot().model_copy(update={"memo": "oat latte"})
292 outcome = await decide_enrichment(
293 snapshot,
294 _REQUEST.candidates,
295 [_learned_rule()],
296 AutoActionCounters(),
297 now=_NOW,
298 model=_capturing_model(prompts),
299 )
300 assert isinstance(outcome, AskHuman)
301 (prompt,) = prompts
302 assert "Memo: oat latte" in prompt
303 assert "Rule hint:" in prompt
304 assert "Dining Out" in prompt
305 
306 
307async def test_clean_context_review_never_sees_the_rule_hint() -> None:
308 # §0.6: the independent judge must stay blind to the rule's choice — the
309 # memo (a transaction fact) rides along, the hint never does.
310 prompts: list[str] = []
311 snapshot = _snapshot().model_copy(update={"memo": "oat latte"})
312 outcome = await decide_enrichment(
313 snapshot,
314 _REQUEST.candidates,
315 [_blessed_rule()],
316 AutoActionCounters(),
317 now=_NOW,
318 model=_capturing_model(prompts),
319 )
320 assert isinstance(outcome, AutoApply)
321 (prompt,) = prompts
322 assert "Memo: oat latte" in prompt
323 assert "Rule hint:" not in prompt
324 
325 
326async def test_decide_enrichment_raises_on_a_hallucinated_id() -> None:
327 # The raw-id-in-subject bug at its source: a hallucinated id never becomes
328 # a proposal — the activity fails and retries instead.
329 model = TestModel(
330 custom_output_args={
331 "category_id": "10683d916894",
332 "confidence": "high",
333 "rationale": "??",
334 }
335 )
336 with pytest.raises(ValueError, match="unknown category id"):
337 await decide_enrichment(
338 _snapshot(),
339 _REQUEST.candidates,
340 [],
341 AutoActionCounters(),
342 now=_NOW,
343 model=model,
344 )
345 
346 
347async def test_decide_enrichment_asks_when_no_trusted_rule() -> None:
348 model = TestModel(
349 custom_output_args={
350 "category_id": "dining",
351 "confidence": "medium",
352 "rationale": "a coffee shop",
353 }
354 )
355 outcome = await decide_enrichment(
356 _snapshot(),
357 _REQUEST.candidates,
358 [],
359 AutoActionCounters(),
360 now=_NOW,
361 model=model,
362 )
363 assert isinstance(outcome, AskHuman)
364 assert isinstance(outcome.proposal.allocation, ProposedCategory)
365 assert outcome.proposal.allocation.category == "dining"
366 
367 
368@pytest.mark.skipif(
369 not os.environ.get("YNAB_AGENT_LIVE_OLLAMA"),
370 reason="set YNAB_AGENT_LIVE_OLLAMA=1 to run the live Gemma smoke",
372async def test_live_gemma_categorizes_a_coffee_shop() -> None:
373 # SPEC §0.5 spike #2: real Gemma over Ollama returns usable content.
374 out = await propose(_REQUEST)
375 assert out.category_id in {"dining", "groceries"}
376 assert out.rationale
377 
378 
379def test_prompt_leads_with_the_stable_candidate_block() -> None:
380 # Prefix-cache contract: the candidate list is byte-stable across calls
381 # (one budget, one category list), so it must precede the per-call facts
382 # — together with the static system prompt it forms the shared prefix
383 # the model server can skip re-prefilling.
384 from ynab_agent.agentic.enrich import _format_request
385 
386 prompt = _format_request(
387 EnrichmentRequest(
388 payee="Hulu",
389 amount_display="-$13.07",
390 memo="monthly",
391 rule_hint="past transactions were filed under 'Streaming'",
392 candidates=(CandidateCategory(id="c1", name="Streaming"),),
393 )
394 )
395 assert prompt.index("Candidate categories:") < prompt.index("Payee:")
396 assert prompt.rstrip().endswith(
397 "Rule hint: past transactions were filed under 'Streaming'"
398 )

Interpret: candidates before payee; the reply ends the prompt.

tests/agentic/test_interpret.py · 192 lines
tests/agentic/test_interpret.py192 lines · Python
⋯ 184 lines hidden (lines 1–184)
1"""Tests for the reply-interpreting agent (SPEC §3, §0.5)."""
2 
3from __future__ import annotations
4 
5import datetime
6import os
7 
8import pytest
9from pydantic_ai.models.test import TestModel
10 
11from ynab_agent.agentic.enrich import CandidateCategory
12from ynab_agent.agentic.interpret import (
13 Interpretation,
14 InterpretRequest,
15 ReplyIntent,
16 interpret,
17 to_reply_outcome,
19from ynab_agent.domain.allocations import ResolvedCategory
20from ynab_agent.domain.enums import DecidedBy
21from ynab_agent.domain.ids import CategoryId
22from ynab_agent.workflow.types import (
23 AnswerOutcome,
24 ClarifyOutcome,
25 ReplyOutcome,
27 
28_NOW = datetime.datetime(2026, 5, 31, 12, 0, tzinfo=datetime.UTC)
29_PROPOSED = CategoryId("dining")
30_REQUEST = InterpretRequest(
31 reply_text="ok",
32 payee="Blue Bottle Coffee",
33 amount_display="-$4.50",
34 proposed_category_name="Dining Out",
35 candidates=(
36 CandidateCategory(id="dining", name="Dining Out"),
37 CandidateCategory(id="coffee", name="Coffee Shops"),
38 ),
40 
41 
42def _outcome(
43 interpretation: Interpretation, reply_text: str = "ok"
44) -> ReplyOutcome:
45 return to_reply_outcome(
46 interpretation,
47 proposed_category=_PROPOSED,
48 decided_at=_NOW,
49 candidates=_REQUEST.candidates,
50 reply_text=reply_text,
51 )
52 
53 
54async def test_approve_intent_round_trips_through_the_agent() -> None:
55 model = TestModel(custom_output_args={"intent": "approve"})
56 out = await interpret(_REQUEST, model=model)
57 assert out.intent is ReplyIntent.APPROVE
58 
59 
60def test_approve_commits_the_proposed_category() -> None:
61 outcome = _outcome(Interpretation(intent=ReplyIntent.APPROVE))
62 assert isinstance(outcome, AnswerOutcome)
63 assert isinstance(outcome.decision.allocation, ResolvedCategory)
64 assert outcome.decision.allocation.category == "dining"
65 assert outcome.decision.decided_by is DecidedBy.HUMAN
66 assert outcome.decision.decided_at == _NOW
67 
68 
69def test_approve_without_a_proposal_asks_instead() -> None:
70 # A flagged/NeedsHuman wait has no proposal: a bare "ok" has nothing to
71 # approve, so the safe move is to ask — never to guess a write.
72 outcome = to_reply_outcome(
73 Interpretation(intent=ReplyIntent.APPROVE),
74 proposed_category=None,
75 decided_at=_NOW,
76 candidates=_REQUEST.candidates,
77 reply_text="ok",
78 )
79 assert isinstance(outcome, ClarifyOutcome)
80 
81 
82def test_recategorize_without_a_proposal_still_commits() -> None:
83 # The owner can name a category outright even when nothing was proposed —
84 # this is what makes flagged entries answerable by email at all.
85 outcome = to_reply_outcome(
86 Interpretation(intent=ReplyIntent.RECATEGORIZE, category_id="coffee"),
87 proposed_category=None,
88 decided_at=_NOW,
89 candidates=_REQUEST.candidates,
90 reply_text="coffee shops please",
91 )
92 assert isinstance(outcome, AnswerOutcome)
93 assert isinstance(outcome.decision.allocation, ResolvedCategory)
94 assert outcome.decision.allocation.category == "coffee"
95 
96 
97def test_recategorize_with_a_hallucinated_id_asks_instead() -> None:
98 # The model mapped the reply onto an id that is not a real candidate — a
99 # write against it would land wrong or 400, so ask for the name instead.
100 outcome = _outcome(
101 Interpretation(
102 intent=ReplyIntent.RECATEGORIZE, category_id="10683d916894"
103 )
104 )
105 assert isinstance(outcome, ClarifyOutcome)
106 assert "couldn't match" in outcome.question
107 
108 
109def test_request_renders_a_missing_proposal_as_none() -> None:
110 request = _REQUEST.model_copy(update={"proposed_category_name": None})
111 from ynab_agent.agentic.interpret import _format_request
112 
113 assert "Currently proposed: (none)" in _format_request(request)
114 
115 
116def test_recategorize_commits_the_named_category() -> None:
117 outcome = _outcome(
118 Interpretation(intent=ReplyIntent.RECATEGORIZE, category_id="coffee")
119 )
120 assert isinstance(outcome, AnswerOutcome)
121 assert isinstance(outcome.decision.allocation, ResolvedCategory)
122 assert outcome.decision.allocation.category == "coffee"
123 
124 
125def test_recategorize_without_a_category_asks_instead() -> None:
126 outcome = _outcome(Interpretation(intent=ReplyIntent.RECATEGORIZE))
127 assert isinstance(outcome, ClarifyOutcome)
128 
129 
130def test_reply_rationale_is_carried_onto_the_decision_memo() -> None:
131 # Context the human gives beyond the category rides along as the memo, which
132 # the spine writes to YNAB (SPEC §14.4).
133 outcome = _outcome(
134 Interpretation(
135 intent=ReplyIntent.RECATEGORIZE,
136 category_id="coffee",
137 memo="Gift for mom's birthday",
138 ),
139 reply_text="coffee — it was a gift for mom's birthday",
140 )
141 assert isinstance(outcome, AnswerOutcome)
142 assert outcome.decision.memo == "Gift for mom's birthday"
143 
144 
145def test_fabricated_memo_is_dropped_not_written() -> None:
146 # The model echoed a category name into the memo field — nothing in the
147 # reply supports it. Writing it would overwrite real YNAB detail (the
148 # Amazon item list), so it is dropped.
149 outcome = _outcome(
150 Interpretation(
151 intent=ReplyIntent.RECATEGORIZE,
152 category_id="coffee",
153 memo="Amex Cash",
154 ),
155 reply_text="Groceries",
156 )
157 assert isinstance(outcome, AnswerOutcome)
158 assert outcome.decision.memo is None
159 
160 
161def test_bare_approval_leaves_the_memo_unset() -> None:
162 outcome = _outcome(Interpretation(intent=ReplyIntent.APPROVE))
163 assert isinstance(outcome, AnswerOutcome)
164 assert outcome.decision.memo is None
165 
166 
167def test_clarify_sends_the_question_back() -> None:
168 outcome = _outcome(
169 Interpretation(intent=ReplyIntent.CLARIFY, question="Split it how?")
170 )
171 assert isinstance(outcome, ClarifyOutcome)
172 assert outcome.question == "Split it how?"
173 
174 
175@pytest.mark.skipif(
176 not os.environ.get("YNAB_AGENT_LIVE_OLLAMA"),
177 reason="set YNAB_AGENT_LIVE_OLLAMA=1 to run the live Gemma smoke",
179async def test_live_gemma_reads_an_approval() -> None:
180 # SPEC §0.5 spike #2: real Gemma reads a plain "ok" as some valid intent.
181 out = await interpret(_REQUEST)
182 assert out.intent in set(ReplyIntent)
183 
184 
185def test_prompt_leads_with_the_stable_candidate_block() -> None:
186 # Prefix-cache contract: the byte-stable candidate list precedes the
187 # per-call facts, and the reply (the most variable line) comes last.
188 from ynab_agent.agentic.interpret import _format_request
189 
190 prompt = _format_request(_REQUEST)
191 assert prompt.index("Candidate categories:") < prompt.index("Payee:")
192 assert prompt.rstrip().split("\n")[-1].startswith("Reply:")

Gate: 744 passed, 7 skipped; ruff, mypy, and the workflow-determinism check clean.