Trusted rules answer the ASK — a code-level reading

What changed, and why

W2's ASK path — propose a category, email the owner, wait — always ran the model, even for a payee whose category the owner had already confirmed enough times for its learned rule to reach trusted. The rule's knowledge was demoted to a one-line prompt hint, and the model re-derived the owner's own repeated answer on every new charge from that payee.

This PR lets a trusted matching rule be the proposal: same category, the same honest rationale the hint used, sources=rule, no model call. The human still decides — the deterministic gate alone grants autonomy, so nothing about write authority moves. As rules accumulate, this removes the proposal call for every repeat payee, the most common model call in the system.

The rule-answered proposal

_proposal_from_rule walks the matching rules and returns a proposal from the first one that qualifies. The bar is deliberately specific: trust must be TRUSTED (consistency-proven, the same threshold bless-eligibility uses), the action must be a single category, and that category must still be a live candidate. Anything weaker returns None and the model path runs unchanged.

Trusted + single-category + still-a-candidate, or None.

src/ynab_agent/agentic/enrich.py · 387 lines

src/ynab_agent/agentic/enrich.py387 lines · Python

⋯ 214 lines hidden (lines 1–214)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,
  TrustState,
29)
30from ynab_agent.domain.events import AskHuman, AutoApply
31from ynab_agent.domain.ids import CategoryId
32from ynab_agent.domain.proposal import Proposal, ProposalSource
33from ynab_agent.policy.gate import (
  GateVerdict,
  build_auto_decision,
  evaluate_gate,
  matching_rules,
38)
39 
40if TYPE_CHECKING:
  import datetime
  from collections.abc import Iterable
43 
  from pydantic_ai.models import Model
45 
  from ynab_agent.domain.events import EnrichmentOutcome
  from ynab_agent.domain.rule import Rule
  from ynab_agent.domain.transaction import YnabSnapshot
  from ynab_agent.policy.floor import AutoActionCounters
50 
51 
52class CandidateCategory(Frozen):
  """One budget category the agent may choose from."""
54 
  id: str
  name: str
57 
58 
59class EnrichmentRequest(Frozen):
  """The facts the agent reasons over to propose a category."""
61 
  payee: str
  amount_display: str
  candidates: tuple[CandidateCategory, ...] = Field(min_length=1)
  memo: str | None = None
  rule_hint: str | None = None
67 
68 
69class EnrichmentSuggestion(Frozen):
  """The agent's structured proposal (mapped to a domain Proposal)."""
71 
  category_id: str
  confidence: Confidence
  rationale: str
  alternatives: tuple[str, ...] = ()
76 
77 
78_SYSTEM_PROMPT = """\
79You categorize a single bank transaction for a personal budget. You are given
80the payee, amount, an optional memo, an optional hint from a learned rule, and a
81list of candidate budget categories (each with an id and a name).
82 
83Choose the SINGLE best category for the transaction. Your `category_id` MUST be
84one of the provided candidate ids — never invent one. Also list up to 2
85`alternatives`: the next-most-likely candidate ids (also from the list, never
86the chosen one) the owner might prefer — or leave empty if none fit. Rate your
87confidence (high / medium / low) and give a one-sentence rationale. Confidence
88is framing only; a human or a trusted rule decides whether to auto-apply."""
89 
90_AGENT: Agent[None, EnrichmentSuggestion] = Agent(
  output_type=EnrichmentSuggestion,
  system_prompt=_SYSTEM_PROMPT,
93)
94 
95 
96def _format_request(request: EnrichmentRequest) -> str:
  """Render the request as the agent's user prompt."""
  lines = [
      f"Payee: {request.payee}",
      f"Amount: {request.amount_display}",
  ]
  if request.memo:
      lines.append(f"Memo: {request.memo}")
  if request.rule_hint:
      lines.append(f"Rule hint: {request.rule_hint}")
  lines.append("Candidate categories:")
  lines.extend(f"  - {c.name} (id: {c.id})" for c in request.candidates)
  return "\n".join(lines)
109 
110 
111async def propose(
  request: EnrichmentRequest, *, model: Model | None = None
113) -> EnrichmentSuggestion:
  """Run the enrichment agent for one transaction (SPEC §4.1).
115 
  Args:
      request: The transaction facts and candidate categories.
      model: A model to use; defaults to the configured Ollama/Gemma.
119 
  Returns:
      The agent's structured category suggestion.
  """
  return await run_structured(
      _AGENT,
      _format_request(request),
      output_type=EnrichmentSuggestion,
      model=model,
  )
129 
130 
131def validate_suggestion(
  suggestion: EnrichmentSuggestion,
  candidates: tuple[CandidateCategory, ...],
134) -> EnrichmentSuggestion:
  """Reject a hallucinated category id; drop hallucinated alternatives.
136 
  The model's ``category_id`` MUST be a candidate id — an invented one would
  flow into the proposal and surface as a raw id in the email subject (and a
  write against it would 400). Raising lets the enclosing activity retry,
  which re-runs the model. Alternatives are framing only, so invalid ones are
  silently dropped rather than failing the run.
  """
  valid = {c.id for c in candidates}
  if suggestion.category_id not in valid:
      msg = (
          f"model proposed unknown category id "
          f"{suggestion.category_id!r} (not a candidate)"
      )
      raise ValueError(msg)
  kept = tuple(a for a in suggestion.alternatives if a in valid)
  if kept == suggestion.alternatives:
      return suggestion
  return suggestion.model_copy(update={"alternatives": kept})
154 
155 
156def to_proposal(suggestion: EnrichmentSuggestion) -> Proposal:
  """Map the agent's suggestion onto a domain Proposal (SPEC §4.1).
158 
  The model's runner-up ids become the proposal's alternatives (the chosen
  one filtered out defensively), surfaced in the email so the owner can pick
  one at a glance.
  """
  alternatives = tuple(
      CategoryId(alt)
      for alt in suggestion.alternatives
      if alt and alt != suggestion.category_id
  )
  return Proposal(
      allocation=ProposedCategory(
          category=CategoryId(suggestion.category_id)
      ),
      confidence=suggestion.confidence,
      rationale=suggestion.rationale,
      sources=(ProposalSource(kind=SourceKind.MODEL),),
      alternatives=alternatives,
  )
177 
178 
179def _blessed_category_id(rule: Rule) -> str | None:
  """The single category a rule auto-applies, or ``None`` for a split.
181 
  The safety review compares a single independent category against the rule's;
  a split action has no single category to judge, so the review is skipped for
  it (learned-eligible rules are always single-category, so this is rare).
  """
  allocation = rule.action.allocation
  if isinstance(allocation, ProposedCategory):
      return str(allocation.category)
  return None
190 
191 
192def _rule_hint(
  snapshot: YnabSnapshot,
  rules: tuple[Rule, ...],
  candidates: tuple[CandidateCategory, ...],
196) -> str | None:
  """A one-line hint from the first matching rule, for the ASK proposal.
198 
  A matching-but-not-blessed rule (learned, or eligible-awaiting-blessing)
  still encodes how this payee was handled before — exactly the context the
  proposing model should weigh. Only the ASK-path prompt gets it; the §0.6
  clean-context review must stay blind to the rule's choice.
  """
  names = {c.id: c.name for c in candidates}
  for rule in matching_rules(rules, snapshot):
      category_id = _blessed_category_id(rule)
      if category_id is not None and category_id in names:
          return (
              f"past transactions from this payee were filed under "
              f"'{names[category_id]}'"
          )
  return None
213 
214 
215def _proposal_from_rule(
  snapshot: YnabSnapshot,
  rules: tuple[Rule, ...],
  candidates: tuple[CandidateCategory, ...],
219) -> Proposal | None:
  """A deterministic ASK proposal from a trusted matching rule, or None.
221 
  A rule that reached ``trusted`` by consistency already encodes the
  owner's repeated answer for this payee; re-deriving it with a model call
  adds latency, not information — and the owner still decides, since the
  deterministic gate alone grants autonomy (SPEC §14). Only a
  single-category rule whose category is still a live candidate
  qualifies; splits, lower trust, and stale category ids fall through to
  the model. Confidence is framing only, and a consistency-proven rule is
  exactly the high-confidence case.
  """
  names = {c.id: c.name for c in candidates}
  for rule in matching_rules(rules, snapshot):
      if rule.trust is not TrustState.TRUSTED:
          continue
      category_id = _blessed_category_id(rule)
      if category_id is None or category_id not in names:
          continue
      return Proposal(
          allocation=ProposedCategory(category=CategoryId(category_id)),
          confidence=Confidence.HIGH,
          rationale=(
              f"past transactions from this payee were filed under "
              f"'{names[category_id]}'"
          ),
          sources=(ProposalSource(kind=SourceKind.RULE),),
          alternatives=(),
      )
  return None
249 
250 
⋯ 137 lines hidden (lines 251–387)251def review_auto_apply(
  blessed_category_id: str, suggestion: EnrichmentSuggestion
253) -> ReviewVerdict:
  """The agent-powered safety review's one-way ratchet (SPEC §0.6 Layer 2).
255 
  An independent, *clean-context* model categorization (run with no knowledge
  of the rule's choice, to stay unbiased) judges the impending auto-apply: it
  ``PROCEED``s only if it considers the blessed category plausible — the
  category it chose or listed as an alternative — and otherwise
  ``ESCALATE_TO_HUMAN``. It can only hold an auto-apply back, never grant one,
  so the deterministic gate still authorizes (principle 6).
  """
  plausible = {suggestion.category_id, *suggestion.alternatives}
  if blessed_category_id in plausible:
      return ReviewVerdict.PROCEED
  return ReviewVerdict.ESCALATE_TO_HUMAN
267 
268 
269def _escalation_proposal(
  suggestion: EnrichmentSuggestion, blessed_category_id: str
271) -> Proposal:
  """The proposal emailed when the review holds an auto-apply back.
273 
  Leads with the model's independent pick and offers the usual auto category
  as an alternative, so the owner sees both views and the disagreement that
  triggered the question.
  """
  base = to_proposal(suggestion)
  blessed = CategoryId(blessed_category_id)
  # to_proposal always builds a single-category allocation; narrow for mypy.
  allocation = base.allocation
  top = (
      allocation.category
      if isinstance(allocation, ProposedCategory)
      else None
  )
  alternatives = base.alternatives
  if blessed != top and blessed not in alternatives:
      alternatives = (blessed, *alternatives)
  return base.model_copy(
      update={
          "alternatives": alternatives,
          "rationale": (
              "I'd usually auto-file this payee, but this charge looked "
              "different from the usual — confirming before I apply it."
          ),
      }
  )
299 
300 
301async def decide_enrichment(
  snapshot: YnabSnapshot,
  candidates: tuple[CandidateCategory, ...],
  rules: Iterable[Rule],
  counters: AutoActionCounters,
  *,
  now: datetime.datetime,
  model: Model | None = None,
309) -> EnrichmentOutcome:
  """Compose the enrich step: gate, then a safety review (SPEC §4.1, §0.6).
311 
  The deterministic gate decides autonomy from the rules alone — a single
  blessed rule may auto-apply *its* action (the model never authorizes a
  write, principle 6). Before an auto-apply lands, an independent
  *clean-context* model review judges it (:func:`review_auto_apply`); a
  disagreement holds it back to ASK (a one-way ratchet — it can only veto). A
  gated ASK runs the same model to produce the best-guess proposal the email
  shows.
319 
  Args:
      snapshot: The transaction being enriched.
      candidates: The budget categories the agent may choose from.
      rules: The rules in scope for the gate.
      counters: The auto-action counters the hard floor reads.
      now: The decision timestamp (the activity supplies it).
      model: A model override for tests; defaults to Ollama/Gemma.
327 
  Returns:
      ``AutoApply`` when a blessed rule gates it *and* the review proceeds,
      else ``AskHuman``.
  """
  rules = tuple(rules)
  gate = evaluate_gate(snapshot, rules, counters)
  if gate.verdict is GateVerdict.AUTO and gate.rule_id is not None:
      rule = next((r for r in rules if r.id == gate.rule_id), None)
      if rule is not None:
          decision = build_auto_decision(rule, snapshot, now)
          blessed_category_id = _blessed_category_id(rule)
          if blessed_category_id is None:
              return AutoApply(decision=decision)
          # Clean context: the independent judge never sees the rule's
          # choice — the memo is a transaction fact, so it rides along.
          suggestion = validate_suggestion(
              await propose(
                  EnrichmentRequest(
                      payee=snapshot.payee,
                      amount_display=str(snapshot.amount),
                      candidates=candidates,
                      memo=snapshot.memo,
                  ),
                  model=model,
              ),
              candidates,
          )
          if (
              review_auto_apply(blessed_category_id, suggestion)
              is ReviewVerdict.PROCEED
          ):
              return AutoApply(decision=decision)
          return AskHuman(
              proposal=_escalation_proposal(suggestion, blessed_category_id)
          )
363 
  # A consistency-proven rule answers the ASK without a model call: the
  # owner has repeatedly confirmed this payee's category, so the model
  # would only re-derive what the rule already knows (SPEC §14 keeps the
  # human deciding either way). Anything weaker falls through.
  rule_proposal = _proposal_from_rule(snapshot, rules, candidates)
  if rule_proposal is not None:
      return AskHuman(proposal=rule_proposal)
371 
  # The model ASK-path proposal gets every fact available: the memo and any
  # matching rule's history (the clean-context review above never does).
  suggestion = validate_suggestion(
      await propose(
          EnrichmentRequest(
              payee=snapshot.payee,
              amount_display=str(snapshot.amount),
              candidates=candidates,
              memo=snapshot.memo,
              rule_hint=_rule_hint(snapshot, rules, candidates),
          ),
          model=model,
      ),
      candidates,
  )
  return AskHuman(proposal=to_proposal(suggestion))

Wiring: one early return on the ASK path

decide_enrichment is untouched until after the AUTO branch. The clean-context review that judges an auto-apply (SPEC §0.6 Layer 2) still runs the model on purpose — that call is a safety layer, not overhead. Only then, where the ASK proposal used to be built, the rule gets first claim; the model call below it is byte-for-byte the old behavior, hint and all.

Rule first; the model ASK (with the rule hint) is the unchanged fallback.

src/ynab_agent/agentic/enrich.py · 387 lines

src/ynab_agent/agentic/enrich.py387 lines · Python

⋯ 363 lines hidden (lines 1–363)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,
  TrustState,
29)
30from ynab_agent.domain.events import AskHuman, AutoApply
31from ynab_agent.domain.ids import CategoryId
32from ynab_agent.domain.proposal import Proposal, ProposalSource
33from ynab_agent.policy.gate import (
  GateVerdict,
  build_auto_decision,
  evaluate_gate,
  matching_rules,
38)
39 
40if TYPE_CHECKING:
  import datetime
  from collections.abc import Iterable
43 
  from pydantic_ai.models import Model
45 
  from ynab_agent.domain.events import EnrichmentOutcome
  from ynab_agent.domain.rule import Rule
  from ynab_agent.domain.transaction import YnabSnapshot
  from ynab_agent.policy.floor import AutoActionCounters
50 
51 
52class CandidateCategory(Frozen):
  """One budget category the agent may choose from."""
54 
  id: str
  name: str
57 
58 
59class EnrichmentRequest(Frozen):
  """The facts the agent reasons over to propose a category."""
61 
  payee: str
  amount_display: str
  candidates: tuple[CandidateCategory, ...] = Field(min_length=1)
  memo: str | None = None
  rule_hint: str | None = None
67 
68 
69class EnrichmentSuggestion(Frozen):
  """The agent's structured proposal (mapped to a domain Proposal)."""
71 
  category_id: str
  confidence: Confidence
  rationale: str
  alternatives: tuple[str, ...] = ()
76 
77 
78_SYSTEM_PROMPT = """\
79You categorize a single bank transaction for a personal budget. You are given
80the payee, amount, an optional memo, an optional hint from a learned rule, and a
81list of candidate budget categories (each with an id and a name).
82 
83Choose the SINGLE best category for the transaction. Your `category_id` MUST be
84one of the provided candidate ids — never invent one. Also list up to 2
85`alternatives`: the next-most-likely candidate ids (also from the list, never
86the chosen one) the owner might prefer — or leave empty if none fit. Rate your
87confidence (high / medium / low) and give a one-sentence rationale. Confidence
88is framing only; a human or a trusted rule decides whether to auto-apply."""
89 
90_AGENT: Agent[None, EnrichmentSuggestion] = Agent(
  output_type=EnrichmentSuggestion,
  system_prompt=_SYSTEM_PROMPT,
93)
94 
95 
96def _format_request(request: EnrichmentRequest) -> str:
  """Render the request as the agent's user prompt."""
  lines = [
      f"Payee: {request.payee}",
      f"Amount: {request.amount_display}",
  ]
  if request.memo:
      lines.append(f"Memo: {request.memo}")
  if request.rule_hint:
      lines.append(f"Rule hint: {request.rule_hint}")
  lines.append("Candidate categories:")
  lines.extend(f"  - {c.name} (id: {c.id})" for c in request.candidates)
  return "\n".join(lines)
109 
110 
111async def propose(
  request: EnrichmentRequest, *, model: Model | None = None
113) -> EnrichmentSuggestion:
  """Run the enrichment agent for one transaction (SPEC §4.1).
115 
  Args:
      request: The transaction facts and candidate categories.
      model: A model to use; defaults to the configured Ollama/Gemma.
119 
  Returns:
      The agent's structured category suggestion.
  """
  return await run_structured(
      _AGENT,
      _format_request(request),
      output_type=EnrichmentSuggestion,
      model=model,
  )
129 
130 
131def validate_suggestion(
  suggestion: EnrichmentSuggestion,
  candidates: tuple[CandidateCategory, ...],
134) -> EnrichmentSuggestion:
  """Reject a hallucinated category id; drop hallucinated alternatives.
136 
  The model's ``category_id`` MUST be a candidate id — an invented one would
  flow into the proposal and surface as a raw id in the email subject (and a
  write against it would 400). Raising lets the enclosing activity retry,
  which re-runs the model. Alternatives are framing only, so invalid ones are
  silently dropped rather than failing the run.
  """
  valid = {c.id for c in candidates}
  if suggestion.category_id not in valid:
      msg = (
          f"model proposed unknown category id "
          f"{suggestion.category_id!r} (not a candidate)"
      )
      raise ValueError(msg)
  kept = tuple(a for a in suggestion.alternatives if a in valid)
  if kept == suggestion.alternatives:
      return suggestion
  return suggestion.model_copy(update={"alternatives": kept})
154 
155 
156def to_proposal(suggestion: EnrichmentSuggestion) -> Proposal:
  """Map the agent's suggestion onto a domain Proposal (SPEC §4.1).
158 
  The model's runner-up ids become the proposal's alternatives (the chosen
  one filtered out defensively), surfaced in the email so the owner can pick
  one at a glance.
  """
  alternatives = tuple(
      CategoryId(alt)
      for alt in suggestion.alternatives
      if alt and alt != suggestion.category_id
  )
  return Proposal(
      allocation=ProposedCategory(
          category=CategoryId(suggestion.category_id)
      ),
      confidence=suggestion.confidence,
      rationale=suggestion.rationale,
      sources=(ProposalSource(kind=SourceKind.MODEL),),
      alternatives=alternatives,
  )
177 
178 
179def _blessed_category_id(rule: Rule) -> str | None:
  """The single category a rule auto-applies, or ``None`` for a split.
181 
  The safety review compares a single independent category against the rule's;
  a split action has no single category to judge, so the review is skipped for
  it (learned-eligible rules are always single-category, so this is rare).
  """
  allocation = rule.action.allocation
  if isinstance(allocation, ProposedCategory):
      return str(allocation.category)
  return None
190 
191 
192def _rule_hint(
  snapshot: YnabSnapshot,
  rules: tuple[Rule, ...],
  candidates: tuple[CandidateCategory, ...],
196) -> str | None:
  """A one-line hint from the first matching rule, for the ASK proposal.
198 
  A matching-but-not-blessed rule (learned, or eligible-awaiting-blessing)
  still encodes how this payee was handled before — exactly the context the
  proposing model should weigh. Only the ASK-path prompt gets it; the §0.6
  clean-context review must stay blind to the rule's choice.
  """
  names = {c.id: c.name for c in candidates}
  for rule in matching_rules(rules, snapshot):
      category_id = _blessed_category_id(rule)
      if category_id is not None and category_id in names:
          return (
              f"past transactions from this payee were filed under "
              f"'{names[category_id]}'"
          )
  return None
213 
214 
215def _proposal_from_rule(
  snapshot: YnabSnapshot,
  rules: tuple[Rule, ...],
  candidates: tuple[CandidateCategory, ...],
219) -> Proposal | None:
  """A deterministic ASK proposal from a trusted matching rule, or None.
221 
  A rule that reached ``trusted`` by consistency already encodes the
  owner's repeated answer for this payee; re-deriving it with a model call
  adds latency, not information — and the owner still decides, since the
  deterministic gate alone grants autonomy (SPEC §14). Only a
  single-category rule whose category is still a live candidate
  qualifies; splits, lower trust, and stale category ids fall through to
  the model. Confidence is framing only, and a consistency-proven rule is
  exactly the high-confidence case.
  """
  names = {c.id: c.name for c in candidates}
  for rule in matching_rules(rules, snapshot):
      if rule.trust is not TrustState.TRUSTED:
          continue
      category_id = _blessed_category_id(rule)
      if category_id is None or category_id not in names:
          continue
      return Proposal(
          allocation=ProposedCategory(category=CategoryId(category_id)),
          confidence=Confidence.HIGH,
          rationale=(
              f"past transactions from this payee were filed under "
              f"'{names[category_id]}'"
          ),
          sources=(ProposalSource(kind=SourceKind.RULE),),
          alternatives=(),
      )
  return None
249 
250 
251def review_auto_apply(
  blessed_category_id: str, suggestion: EnrichmentSuggestion
253) -> ReviewVerdict:
  """The agent-powered safety review's one-way ratchet (SPEC §0.6 Layer 2).
255 
  An independent, *clean-context* model categorization (run with no knowledge
  of the rule's choice, to stay unbiased) judges the impending auto-apply: it
  ``PROCEED``s only if it considers the blessed category plausible — the
  category it chose or listed as an alternative — and otherwise
  ``ESCALATE_TO_HUMAN``. It can only hold an auto-apply back, never grant one,
  so the deterministic gate still authorizes (principle 6).
  """
  plausible = {suggestion.category_id, *suggestion.alternatives}
  if blessed_category_id in plausible:
      return ReviewVerdict.PROCEED
  return ReviewVerdict.ESCALATE_TO_HUMAN
267 
268 
269def _escalation_proposal(
  suggestion: EnrichmentSuggestion, blessed_category_id: str
271) -> Proposal:
  """The proposal emailed when the review holds an auto-apply back.
273 
  Leads with the model's independent pick and offers the usual auto category
  as an alternative, so the owner sees both views and the disagreement that
  triggered the question.
  """
  base = to_proposal(suggestion)
  blessed = CategoryId(blessed_category_id)
  # to_proposal always builds a single-category allocation; narrow for mypy.
  allocation = base.allocation
  top = (
      allocation.category
      if isinstance(allocation, ProposedCategory)
      else None
  )
  alternatives = base.alternatives
  if blessed != top and blessed not in alternatives:
      alternatives = (blessed, *alternatives)
  return base.model_copy(
      update={
          "alternatives": alternatives,
          "rationale": (
              "I'd usually auto-file this payee, but this charge looked "
              "different from the usual — confirming before I apply it."
          ),
      }
  )
299 
300 
301async def decide_enrichment(
  snapshot: YnabSnapshot,
  candidates: tuple[CandidateCategory, ...],
  rules: Iterable[Rule],
  counters: AutoActionCounters,
  *,
  now: datetime.datetime,
  model: Model | None = None,
309) -> EnrichmentOutcome:
  """Compose the enrich step: gate, then a safety review (SPEC §4.1, §0.6).
311 
  The deterministic gate decides autonomy from the rules alone — a single
  blessed rule may auto-apply *its* action (the model never authorizes a
  write, principle 6). Before an auto-apply lands, an independent
  *clean-context* model review judges it (:func:`review_auto_apply`); a
  disagreement holds it back to ASK (a one-way ratchet — it can only veto). A
  gated ASK runs the same model to produce the best-guess proposal the email
  shows.
319 
  Args:
      snapshot: The transaction being enriched.
      candidates: The budget categories the agent may choose from.
      rules: The rules in scope for the gate.
      counters: The auto-action counters the hard floor reads.
      now: The decision timestamp (the activity supplies it).
      model: A model override for tests; defaults to Ollama/Gemma.
327 
  Returns:
      ``AutoApply`` when a blessed rule gates it *and* the review proceeds,
      else ``AskHuman``.
  """
  rules = tuple(rules)
  gate = evaluate_gate(snapshot, rules, counters)
  if gate.verdict is GateVerdict.AUTO and gate.rule_id is not None:
      rule = next((r for r in rules if r.id == gate.rule_id), None)
      if rule is not None:
          decision = build_auto_decision(rule, snapshot, now)
          blessed_category_id = _blessed_category_id(rule)
          if blessed_category_id is None:
              return AutoApply(decision=decision)
          # Clean context: the independent judge never sees the rule's
          # choice — the memo is a transaction fact, so it rides along.
          suggestion = validate_suggestion(
              await propose(
                  EnrichmentRequest(
                      payee=snapshot.payee,
                      amount_display=str(snapshot.amount),
                      candidates=candidates,
                      memo=snapshot.memo,
                  ),
                  model=model,
              ),
              candidates,
          )
          if (
              review_auto_apply(blessed_category_id, suggestion)
              is ReviewVerdict.PROCEED
          ):
              return AutoApply(decision=decision)
          return AskHuman(
              proposal=_escalation_proposal(suggestion, blessed_category_id)
          )
363 
  # A consistency-proven rule answers the ASK without a model call: the
  # owner has repeatedly confirmed this payee's category, so the model
  # would only re-derive what the rule already knows (SPEC §14 keeps the
  # human deciding either way). Anything weaker falls through.
  rule_proposal = _proposal_from_rule(snapshot, rules, candidates)
  if rule_proposal is not None:
      return AskHuman(proposal=rule_proposal)
371 
  # The model ASK-path proposal gets every fact available: the memo and any
  # matching rule's history (the clean-context review above never does).
  suggestion = validate_suggestion(
      await propose(
          EnrichmentRequest(
              payee=snapshot.payee,
              amount_display=str(snapshot.amount),
              candidates=candidates,
              memo=snapshot.memo,
              rule_hint=_rule_hint(snapshot, rules, candidates),
          ),
          model=model,
      ),
      candidates,
  )
⋯ 1 line hidden (lines 387–387)387    return AskHuman(proposal=to_proposal(suggestion))

The proof: an explosive model

The discriminating test replaces propose with a stub that raises if reached, so the no-model contract is enforced, not assumed. Two fallback tests pin the other side: a stale category id and a merely-confirmed rule both still consult the model.

A trusted-but-unblessed rule, and the stub that fails the test if the model is consulted.

tests/agentic/test_enrich.py · 469 lines

tests/agentic/test_enrich.py469 lines · Python

⋯ 367 lines hidden (lines 1–367)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 (
  CandidateCategory,
  EnrichmentRequest,
  EnrichmentSuggestion,
  decide_enrichment,
  propose,
  review_auto_apply,
  to_proposal,
  validate_suggestion,
27)
28from ynab_agent.agentic.model import build_model
29from ynab_agent.domain.allocations import ProposedCategory
30from ynab_agent.domain.enums import (
  Confidence,
  ReviewVerdict,
  RuleSource,
  SourceKind,
  TrustState,
36)
37from ynab_agent.domain.events import AskHuman, AutoApply
38from ynab_agent.domain.ids import (
  AccountId,
  CategoryId,
  RuleId,
  YnabTransactionId,
43)
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(
  payee="Blue Bottle Coffee",
  amount_display="-$4.50",
  candidates=(
      CandidateCategory(id="dining", name="Dining Out"),
      CandidateCategory(id="groceries", name="Groceries"),
  ),
56)
57 
58 
59async def test_propose_returns_the_models_structured_suggestion() -> None:
  model = TestModel(
      custom_output_args={
          "category_id": "dining",
          "confidence": "high",
          "rationale": "a coffee shop",
      }
  )
  out = await propose(_REQUEST, model=model)
  assert isinstance(out, EnrichmentSuggestion)
  assert out.category_id == "dining"
  assert out.confidence is Confidence.HIGH
  assert out.rationale == "a coffee shop"
72 
73 
74async def test_propose_wiring_smoke_with_default_testmodel() -> None:
  # No custom output → TestModel autofills a valid suggestion; this just
  # proves the agent/output-schema wiring round-trips.
  out = await propose(_REQUEST, model=TestModel())
  assert isinstance(out, EnrichmentSuggestion)
79 
80 
81def test_to_proposal_maps_onto_the_domain_proposal() -> None:
  suggestion = EnrichmentSuggestion(
      category_id="dining",
      confidence=Confidence.MEDIUM,
      rationale="coffee",
  )
  proposal = to_proposal(suggestion)
  assert isinstance(proposal.allocation, ProposedCategory)
  assert proposal.allocation.category == "dining"
  assert proposal.confidence is Confidence.MEDIUM
  assert proposal.sources[0].kind is SourceKind.MODEL
92 
93 
94def test_to_proposal_carries_alternatives_filtering_the_chosen() -> None:
  suggestion = EnrichmentSuggestion(
      category_id="dining",
      confidence=Confidence.LOW,
      rationale="maybe",
      # the chosen id and an empty entry are filtered out defensively
      alternatives=("coffee", "dining", "", "groceries"),
  )
  proposal = to_proposal(suggestion)
  assert [str(a) for a in proposal.alternatives] == ["coffee", "groceries"]
104 
105 
106def test_build_model_constructs_an_ollama_model() -> None:
  assert isinstance(build_model(model_name="gemma4:e4b"), Model)
108 
109 
110def test_validate_suggestion_rejects_a_hallucinated_id() -> None:
  # An invented id would surface as a raw id in the email subject and 400 on
  # write — raising lets the activity retry with a fresh model run.
  bogus = EnrichmentSuggestion(
      category_id="10683d916894", confidence=Confidence.HIGH, rationale="x"
  )
  with pytest.raises(ValueError, match="unknown category id"):
      validate_suggestion(bogus, _REQUEST.candidates)
118 
119 
120def test_validate_suggestion_drops_hallucinated_alternatives() -> None:
  # Alternatives are framing only — invalid ones are dropped, not fatal.
  mixed = EnrichmentSuggestion(
      category_id="dining",
      confidence=Confidence.HIGH,
      rationale="x",
      alternatives=("groceries", "10683d916894"),
  )
  assert validate_suggestion(mixed, _REQUEST.candidates).alternatives == (
      "groceries",
  )
131 
132 
133def test_review_auto_apply_proceeds_when_blessed_is_plausible() -> None:
  top = EnrichmentSuggestion(
      category_id="dining", confidence=Confidence.HIGH, rationale="x"
  )
  assert review_auto_apply("dining", top) is ReviewVerdict.PROCEED
  alt = EnrichmentSuggestion(
      category_id="groceries",
      confidence=Confidence.HIGH,
      rationale="x",
      alternatives=("dining",),
  )
  assert review_auto_apply("dining", alt) is ReviewVerdict.PROCEED
145 
146 
147def test_review_auto_apply_escalates_when_blessed_is_implausible() -> None:
  other = EnrichmentSuggestion(
      category_id="groceries",
      confidence=Confidence.HIGH,
      rationale="x",
      alternatives=("software",),
  )
  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:
  return YnabSnapshot(
      ynab_id=YnabTransactionId("t1"),
      account=AccountId("a1"),
      payee="Blue Bottle Coffee",
      amount=Money.from_currency("-4.50"),
      txn_date=datetime.date(2026, 5, 28),
  )
168 
169 
170def _blessed_rule() -> Rule:
  return Rule(
      id=RuleId("r1"),
      match=RuleMatch(payee_pattern="Blue Bottle"),
      action=RuleAction(
          allocation=ProposedCategory(category=CategoryId("dining"))
      ),
      trust=TrustState.TRUSTED,
      source=RuleSource.HUMAN_EXPLICIT,
  )
180 
181 
182def _judge(category_id: str, alternatives: tuple[str, ...] = ()) -> TestModel:
  """A TestModel whose suggestion stands in for the review's judgment."""
  return TestModel(
      custom_output_args={
          "category_id": category_id,
          "confidence": "high",
          "rationale": "independent read",
          "alternatives": list(alternatives),
      }
  )
192 
193 
194async def test_decide_enrichment_auto_applies_when_the_review_agrees() -> None:
  # The independent review picks the same category the blessed rule would →
  # the auto-apply proceeds (SPEC §0.6 Layer 2).
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_judge("dining"),
  )
  assert isinstance(outcome, AutoApply)
  assert outcome.decision.rule_id == "r1"
207 
208 
209async def test_decide_enrichment_review_proceeds_if_blessed_is_an_alt() -> None:
  # The review's top pick differs but it still considers the blessed category
  # plausible (an alternative) → proceed.
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_judge("groceries", alternatives=("dining",)),
  )
  assert isinstance(outcome, AutoApply)
221 
222 
223async def test_decide_enrichment_review_escalates_on_disagreement() -> None:
  # The independent review does not find the blessed category plausible →
  # the one-way ratchet holds the auto-apply back to ASK.
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_judge("groceries"),
  )
  assert isinstance(outcome, AskHuman)
  assert isinstance(outcome.proposal.allocation, ProposedCategory)
  # Leads with the model's independent pick; offers the usual auto category.
  assert outcome.proposal.allocation.category == "groceries"
  assert CategoryId("dining") in outcome.proposal.alternatives
239 
240 
241def _learned_rule() -> Rule:
  # Matches the snapshot's payee but is NOT blessed — it cannot gate an
  # auto-apply, only inform the ASK proposal.
  return Rule(
      id=RuleId("r2"),
      match=RuleMatch(payee_pattern="Blue Bottle"),
      action=RuleAction(
          allocation=ProposedCategory(category=CategoryId("dining"))
      ),
      trust=TrustState.CONFIRMED,
      source=RuleSource.LEARNED,
  )
253 
254 
255def _capturing_model(prompts: list[str]) -> Model:
  """A FunctionModel that records the user prompt and answers 'dining'."""
  from pydantic_ai.messages import (
      ModelMessage,
      ModelResponse,
      ToolCallPart,
      UserPromptPart,
  )
  from pydantic_ai.models.function import AgentInfo, FunctionModel
264 
  def call(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
      for message in messages:
          for part in message.parts:
              if isinstance(part, UserPromptPart):
                  prompts.append(str(part.content))
      assert info.output_tools
      return ModelResponse(
          parts=[
              ToolCallPart(
                  tool_name=info.output_tools[0].name,
                  args={
                      "category_id": "dining",
                      "confidence": "high",
                      "rationale": "x",
                  },
              )
          ]
      )
283 
  return FunctionModel(call)
285 
286 
287async def test_ask_prompt_carries_the_memo_and_the_rule_hint() -> None:
  # The model finally *sees* the memo and the payee's prior handling — the
  # context that turns "10683d916894?" guesses into informed proposals.
  prompts: list[str] = []
  snapshot = _snapshot().model_copy(update={"memo": "oat latte"})
  outcome = await decide_enrichment(
      snapshot,
      _REQUEST.candidates,
      [_learned_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_capturing_model(prompts),
  )
  assert isinstance(outcome, AskHuman)
  (prompt,) = prompts
  assert "Memo: oat latte" in prompt
  assert "Rule hint:" in prompt
  assert "Dining Out" in prompt
305 
306 
307async def test_clean_context_review_never_sees_the_rule_hint() -> None:
  # §0.6: the independent judge must stay blind to the rule's choice — the
  # memo (a transaction fact) rides along, the hint never does.
  prompts: list[str] = []
  snapshot = _snapshot().model_copy(update={"memo": "oat latte"})
  outcome = await decide_enrichment(
      snapshot,
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_capturing_model(prompts),
  )
  assert isinstance(outcome, AutoApply)
  (prompt,) = prompts
  assert "Memo: oat latte" in prompt
  assert "Rule hint:" not in prompt
324 
325 
326async def test_decide_enrichment_raises_on_a_hallucinated_id() -> None:
  # The raw-id-in-subject bug at its source: a hallucinated id never becomes
  # a proposal — the activity fails and retries instead.
  model = TestModel(
      custom_output_args={
          "category_id": "10683d916894",
          "confidence": "high",
          "rationale": "??",
      }
  )
  with pytest.raises(ValueError, match="unknown category id"):
      await decide_enrichment(
          _snapshot(),
          _REQUEST.candidates,
          [],
          AutoActionCounters(),
          now=_NOW,
          model=model,
      )
345 
346 
347async def test_decide_enrichment_asks_when_no_trusted_rule() -> None:
  model = TestModel(
      custom_output_args={
          "category_id": "dining",
          "confidence": "medium",
          "rationale": "a coffee shop",
      }
  )
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [],
      AutoActionCounters(),
      now=_NOW,
      model=model,
  )
  assert isinstance(outcome, AskHuman)
  assert isinstance(outcome.proposal.allocation, ProposedCategory)
  assert outcome.proposal.allocation.category == "dining"
366 
367 
368def _trusted_rule() -> Rule:
  # Consistency-proven but NOT blessed (source=learned): the gate still
  # routes it to ASK; the proposal itself may now come from the rule.
  return Rule(
      id=RuleId("r3"),
      match=RuleMatch(payee_pattern="Blue Bottle"),
      action=RuleAction(
          allocation=ProposedCategory(category=CategoryId("dining"))
      ),
      trust=TrustState.TRUSTED,
      source=RuleSource.LEARNED,
  )
380 
381 
382def _no_model(monkeypatch: pytest.MonkeyPatch) -> None:
  """A trusted-rule ASK must never reach the model — make it explosive."""
  from ynab_agent.agentic import enrich as enrich_mod
385 
  async def _boom(request: object, *, model: object = None) -> object:
      msg = "the model must not be called for a trusted-rule ASK"
      raise AssertionError(msg)
389 
  monkeypatch.setattr(enrich_mod, "propose", _boom)
391 
392 
393async def test_trusted_rule_answers_the_ask_without_a_model_call(
  monkeypatch: pytest.MonkeyPatch,
395) -> None:
  # The fewer-calls contract: a trusted matching rule already encodes the
  # owner's repeated answer, so the ASK proposal is built from it
  # deterministically — the model (explosive here) is never consulted.
  _no_model(monkeypatch)
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_trusted_rule()],
      AutoActionCounters(),
      now=_NOW,
  )
  assert isinstance(outcome, AskHuman)
  assert isinstance(outcome.proposal.allocation, ProposedCategory)
  assert outcome.proposal.allocation.category == "dining"
  assert outcome.proposal.sources[0].kind is SourceKind.RULE
  assert "Dining Out" in outcome.proposal.rationale
412 
⋯ 57 lines hidden (lines 413–469)413 
414async def test_trusted_rule_with_a_stale_category_falls_back() -> None:
  # The category was deleted from the budget since the rule learned it —
  # proposing it would render a raw id and 400 on write. Fall back to the
  # model instead of trusting the stale answer.
  stale = _trusted_rule().model_copy(
      update={
          "action": RuleAction(
              allocation=ProposedCategory(category=CategoryId("deleted"))
          )
      }
  )
  model = TestModel(
      custom_output_args={
          "category_id": "dining",
          "confidence": "medium",
          "rationale": "fallback",
      }
  )
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [stale],
      AutoActionCounters(),
      now=_NOW,
      model=model,
  )
  assert isinstance(outcome, AskHuman)
  assert outcome.proposal.sources[0].kind is SourceKind.MODEL
442 
443 
444async def test_confirmed_rule_still_consults_the_model() -> None:
  # Below trusted, the rule is a hypothesis, not an answer: it biases the
  # model as a hint (the existing behavior) rather than replacing it.
  prompts: list[str] = []
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_learned_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_capturing_model(prompts),
  )
  assert isinstance(outcome, AskHuman)
  assert len(prompts) == 1
  assert outcome.proposal.sources[0].kind is SourceKind.MODEL
459 
460 
461@pytest.mark.skipif(
  not os.environ.get("YNAB_AGENT_LIVE_OLLAMA"),
  reason="set YNAB_AGENT_LIVE_OLLAMA=1 to run the live Gemma smoke",
464)
465async def test_live_gemma_categorizes_a_coffee_shop() -> None:
  # SPEC §0.5 spike #2: real Gemma over Ollama returns usable content.
  out = await propose(_REQUEST)
  assert out.category_id in {"dining", "groceries"}
  assert out.rationale

Fallbacks: stale category id → model; confirmed-only rule → model with hint.

tests/agentic/test_enrich.py · 469 lines

tests/agentic/test_enrich.py469 lines · Python

⋯ 413 lines hidden (lines 1–413)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 (
  CandidateCategory,
  EnrichmentRequest,
  EnrichmentSuggestion,
  decide_enrichment,
  propose,
  review_auto_apply,
  to_proposal,
  validate_suggestion,
27)
28from ynab_agent.agentic.model import build_model
29from ynab_agent.domain.allocations import ProposedCategory
30from ynab_agent.domain.enums import (
  Confidence,
  ReviewVerdict,
  RuleSource,
  SourceKind,
  TrustState,
36)
37from ynab_agent.domain.events import AskHuman, AutoApply
38from ynab_agent.domain.ids import (
  AccountId,
  CategoryId,
  RuleId,
  YnabTransactionId,
43)
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(
  payee="Blue Bottle Coffee",
  amount_display="-$4.50",
  candidates=(
      CandidateCategory(id="dining", name="Dining Out"),
      CandidateCategory(id="groceries", name="Groceries"),
  ),
56)
57 
58 
59async def test_propose_returns_the_models_structured_suggestion() -> None:
  model = TestModel(
      custom_output_args={
          "category_id": "dining",
          "confidence": "high",
          "rationale": "a coffee shop",
      }
  )
  out = await propose(_REQUEST, model=model)
  assert isinstance(out, EnrichmentSuggestion)
  assert out.category_id == "dining"
  assert out.confidence is Confidence.HIGH
  assert out.rationale == "a coffee shop"
72 
73 
74async def test_propose_wiring_smoke_with_default_testmodel() -> None:
  # No custom output → TestModel autofills a valid suggestion; this just
  # proves the agent/output-schema wiring round-trips.
  out = await propose(_REQUEST, model=TestModel())
  assert isinstance(out, EnrichmentSuggestion)
79 
80 
81def test_to_proposal_maps_onto_the_domain_proposal() -> None:
  suggestion = EnrichmentSuggestion(
      category_id="dining",
      confidence=Confidence.MEDIUM,
      rationale="coffee",
  )
  proposal = to_proposal(suggestion)
  assert isinstance(proposal.allocation, ProposedCategory)
  assert proposal.allocation.category == "dining"
  assert proposal.confidence is Confidence.MEDIUM
  assert proposal.sources[0].kind is SourceKind.MODEL
92 
93 
94def test_to_proposal_carries_alternatives_filtering_the_chosen() -> None:
  suggestion = EnrichmentSuggestion(
      category_id="dining",
      confidence=Confidence.LOW,
      rationale="maybe",
      # the chosen id and an empty entry are filtered out defensively
      alternatives=("coffee", "dining", "", "groceries"),
  )
  proposal = to_proposal(suggestion)
  assert [str(a) for a in proposal.alternatives] == ["coffee", "groceries"]
104 
105 
106def test_build_model_constructs_an_ollama_model() -> None:
  assert isinstance(build_model(model_name="gemma4:e4b"), Model)
108 
109 
110def test_validate_suggestion_rejects_a_hallucinated_id() -> None:
  # An invented id would surface as a raw id in the email subject and 400 on
  # write — raising lets the activity retry with a fresh model run.
  bogus = EnrichmentSuggestion(
      category_id="10683d916894", confidence=Confidence.HIGH, rationale="x"
  )
  with pytest.raises(ValueError, match="unknown category id"):
      validate_suggestion(bogus, _REQUEST.candidates)
118 
119 
120def test_validate_suggestion_drops_hallucinated_alternatives() -> None:
  # Alternatives are framing only — invalid ones are dropped, not fatal.
  mixed = EnrichmentSuggestion(
      category_id="dining",
      confidence=Confidence.HIGH,
      rationale="x",
      alternatives=("groceries", "10683d916894"),
  )
  assert validate_suggestion(mixed, _REQUEST.candidates).alternatives == (
      "groceries",
  )
131 
132 
133def test_review_auto_apply_proceeds_when_blessed_is_plausible() -> None:
  top = EnrichmentSuggestion(
      category_id="dining", confidence=Confidence.HIGH, rationale="x"
  )
  assert review_auto_apply("dining", top) is ReviewVerdict.PROCEED
  alt = EnrichmentSuggestion(
      category_id="groceries",
      confidence=Confidence.HIGH,
      rationale="x",
      alternatives=("dining",),
  )
  assert review_auto_apply("dining", alt) is ReviewVerdict.PROCEED
145 
146 
147def test_review_auto_apply_escalates_when_blessed_is_implausible() -> None:
  other = EnrichmentSuggestion(
      category_id="groceries",
      confidence=Confidence.HIGH,
      rationale="x",
      alternatives=("software",),
  )
  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:
  return YnabSnapshot(
      ynab_id=YnabTransactionId("t1"),
      account=AccountId("a1"),
      payee="Blue Bottle Coffee",
      amount=Money.from_currency("-4.50"),
      txn_date=datetime.date(2026, 5, 28),
  )
168 
169 
170def _blessed_rule() -> Rule:
  return Rule(
      id=RuleId("r1"),
      match=RuleMatch(payee_pattern="Blue Bottle"),
      action=RuleAction(
          allocation=ProposedCategory(category=CategoryId("dining"))
      ),
      trust=TrustState.TRUSTED,
      source=RuleSource.HUMAN_EXPLICIT,
  )
180 
181 
182def _judge(category_id: str, alternatives: tuple[str, ...] = ()) -> TestModel:
  """A TestModel whose suggestion stands in for the review's judgment."""
  return TestModel(
      custom_output_args={
          "category_id": category_id,
          "confidence": "high",
          "rationale": "independent read",
          "alternatives": list(alternatives),
      }
  )
192 
193 
194async def test_decide_enrichment_auto_applies_when_the_review_agrees() -> None:
  # The independent review picks the same category the blessed rule would →
  # the auto-apply proceeds (SPEC §0.6 Layer 2).
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_judge("dining"),
  )
  assert isinstance(outcome, AutoApply)
  assert outcome.decision.rule_id == "r1"
207 
208 
209async def test_decide_enrichment_review_proceeds_if_blessed_is_an_alt() -> None:
  # The review's top pick differs but it still considers the blessed category
  # plausible (an alternative) → proceed.
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_judge("groceries", alternatives=("dining",)),
  )
  assert isinstance(outcome, AutoApply)
221 
222 
223async def test_decide_enrichment_review_escalates_on_disagreement() -> None:
  # The independent review does not find the blessed category plausible →
  # the one-way ratchet holds the auto-apply back to ASK.
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_judge("groceries"),
  )
  assert isinstance(outcome, AskHuman)
  assert isinstance(outcome.proposal.allocation, ProposedCategory)
  # Leads with the model's independent pick; offers the usual auto category.
  assert outcome.proposal.allocation.category == "groceries"
  assert CategoryId("dining") in outcome.proposal.alternatives
239 
240 
241def _learned_rule() -> Rule:
  # Matches the snapshot's payee but is NOT blessed — it cannot gate an
  # auto-apply, only inform the ASK proposal.
  return Rule(
      id=RuleId("r2"),
      match=RuleMatch(payee_pattern="Blue Bottle"),
      action=RuleAction(
          allocation=ProposedCategory(category=CategoryId("dining"))
      ),
      trust=TrustState.CONFIRMED,
      source=RuleSource.LEARNED,
  )
253 
254 
255def _capturing_model(prompts: list[str]) -> Model:
  """A FunctionModel that records the user prompt and answers 'dining'."""
  from pydantic_ai.messages import (
      ModelMessage,
      ModelResponse,
      ToolCallPart,
      UserPromptPart,
  )
  from pydantic_ai.models.function import AgentInfo, FunctionModel
264 
  def call(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
      for message in messages:
          for part in message.parts:
              if isinstance(part, UserPromptPart):
                  prompts.append(str(part.content))
      assert info.output_tools
      return ModelResponse(
          parts=[
              ToolCallPart(
                  tool_name=info.output_tools[0].name,
                  args={
                      "category_id": "dining",
                      "confidence": "high",
                      "rationale": "x",
                  },
              )
          ]
      )
283 
  return FunctionModel(call)
285 
286 
287async def test_ask_prompt_carries_the_memo_and_the_rule_hint() -> None:
  # The model finally *sees* the memo and the payee's prior handling — the
  # context that turns "10683d916894?" guesses into informed proposals.
  prompts: list[str] = []
  snapshot = _snapshot().model_copy(update={"memo": "oat latte"})
  outcome = await decide_enrichment(
      snapshot,
      _REQUEST.candidates,
      [_learned_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_capturing_model(prompts),
  )
  assert isinstance(outcome, AskHuman)
  (prompt,) = prompts
  assert "Memo: oat latte" in prompt
  assert "Rule hint:" in prompt
  assert "Dining Out" in prompt
305 
306 
307async def test_clean_context_review_never_sees_the_rule_hint() -> None:
  # §0.6: the independent judge must stay blind to the rule's choice — the
  # memo (a transaction fact) rides along, the hint never does.
  prompts: list[str] = []
  snapshot = _snapshot().model_copy(update={"memo": "oat latte"})
  outcome = await decide_enrichment(
      snapshot,
      _REQUEST.candidates,
      [_blessed_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_capturing_model(prompts),
  )
  assert isinstance(outcome, AutoApply)
  (prompt,) = prompts
  assert "Memo: oat latte" in prompt
  assert "Rule hint:" not in prompt
324 
325 
326async def test_decide_enrichment_raises_on_a_hallucinated_id() -> None:
  # The raw-id-in-subject bug at its source: a hallucinated id never becomes
  # a proposal — the activity fails and retries instead.
  model = TestModel(
      custom_output_args={
          "category_id": "10683d916894",
          "confidence": "high",
          "rationale": "??",
      }
  )
  with pytest.raises(ValueError, match="unknown category id"):
      await decide_enrichment(
          _snapshot(),
          _REQUEST.candidates,
          [],
          AutoActionCounters(),
          now=_NOW,
          model=model,
      )
345 
346 
347async def test_decide_enrichment_asks_when_no_trusted_rule() -> None:
  model = TestModel(
      custom_output_args={
          "category_id": "dining",
          "confidence": "medium",
          "rationale": "a coffee shop",
      }
  )
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [],
      AutoActionCounters(),
      now=_NOW,
      model=model,
  )
  assert isinstance(outcome, AskHuman)
  assert isinstance(outcome.proposal.allocation, ProposedCategory)
  assert outcome.proposal.allocation.category == "dining"
366 
367 
368def _trusted_rule() -> Rule:
  # Consistency-proven but NOT blessed (source=learned): the gate still
  # routes it to ASK; the proposal itself may now come from the rule.
  return Rule(
      id=RuleId("r3"),
      match=RuleMatch(payee_pattern="Blue Bottle"),
      action=RuleAction(
          allocation=ProposedCategory(category=CategoryId("dining"))
      ),
      trust=TrustState.TRUSTED,
      source=RuleSource.LEARNED,
  )
380 
381 
382def _no_model(monkeypatch: pytest.MonkeyPatch) -> None:
  """A trusted-rule ASK must never reach the model — make it explosive."""
  from ynab_agent.agentic import enrich as enrich_mod
385 
  async def _boom(request: object, *, model: object = None) -> object:
      msg = "the model must not be called for a trusted-rule ASK"
      raise AssertionError(msg)
389 
  monkeypatch.setattr(enrich_mod, "propose", _boom)
391 
392 
393async def test_trusted_rule_answers_the_ask_without_a_model_call(
  monkeypatch: pytest.MonkeyPatch,
395) -> None:
  # The fewer-calls contract: a trusted matching rule already encodes the
  # owner's repeated answer, so the ASK proposal is built from it
  # deterministically — the model (explosive here) is never consulted.
  _no_model(monkeypatch)
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_trusted_rule()],
      AutoActionCounters(),
      now=_NOW,
  )
  assert isinstance(outcome, AskHuman)
  assert isinstance(outcome.proposal.allocation, ProposedCategory)
  assert outcome.proposal.allocation.category == "dining"
  assert outcome.proposal.sources[0].kind is SourceKind.RULE
  assert "Dining Out" in outcome.proposal.rationale
412 
413 
414async def test_trusted_rule_with_a_stale_category_falls_back() -> None:
  # The category was deleted from the budget since the rule learned it —
  # proposing it would render a raw id and 400 on write. Fall back to the
  # model instead of trusting the stale answer.
  stale = _trusted_rule().model_copy(
      update={
          "action": RuleAction(
              allocation=ProposedCategory(category=CategoryId("deleted"))
          )
      }
  )
  model = TestModel(
      custom_output_args={
          "category_id": "dining",
          "confidence": "medium",
          "rationale": "fallback",
      }
  )
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [stale],
      AutoActionCounters(),
      now=_NOW,
      model=model,
  )
  assert isinstance(outcome, AskHuman)
  assert outcome.proposal.sources[0].kind is SourceKind.MODEL
442 
443 
444async def test_confirmed_rule_still_consults_the_model() -> None:
  # Below trusted, the rule is a hypothesis, not an answer: it biases the
  # model as a hint (the existing behavior) rather than replacing it.
  prompts: list[str] = []
  outcome = await decide_enrichment(
      _snapshot(),
      _REQUEST.candidates,
      [_learned_rule()],
      AutoActionCounters(),
      now=_NOW,
      model=_capturing_model(prompts),
  )
  assert isinstance(outcome, AskHuman)
⋯ 13 lines hidden (lines 457–469)457    assert len(prompts) == 1
  assert outcome.proposal.sources[0].kind is SourceKind.MODEL
459 
460 
461@pytest.mark.skipif(
  not os.environ.get("YNAB_AGENT_LIVE_OLLAMA"),
  reason="set YNAB_AGENT_LIVE_OLLAMA=1 to run the live Gemma smoke",
464)
465async def test_live_gemma_categorizes_a_coffee_shop() -> None:
  # SPEC §0.5 spike #2: real Gemma over Ollama returns usable content.
  out = await propose(_REQUEST)
  assert out.category_id in {"dining", "groceries"}
  assert out.rationale

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