Converge-to-target reads current state before writing

What changed, and why

REVISING is the riskiest path in W2: it mutates an already-approved money record. SPEC §3 rule 3 spells out how it must write — converge to target: read the current YNAB state, write only if it differs, then verify. Rule 4 adds the guard for the dangerous case: if YNAB shows a different non-empty state than you expect (a spouse edited it directly), don't clobber it — ask.

The shipped code did neither. The converge activity committed blind: it wrote the target, then read back to classify. Two real harms followed. A no-op revision still issued a PATCH — re-stamping the agent review flag, able to overwrite a richer human memo — when nothing needed to change. And a concurrent edit was overwritten: divergence could only be noticed after the clobbering write had already landed, which is exactly when it's too late.

The pure helpers rule 3 calls for — needs_write, content_hash — already existed and were unit-tested. The write path simply never called them. This change routes the converge activity through them, deciding what to do before it writes.

The decision, made pure

The whole pre-write decision is one pure function — no I/O, fully testable, sitting in the policy layer next to the helpers it composes. Given the freshly read current state, the target the instruction asks for, and prior (what the agent last applied), it returns one of four actions.

The subtle case is ALREADY_TARGET. If current already equals target we skip the write — but we must distinguish a genuine no-op (current also equals the prior: nothing changed) from a converge whose write did land and is being retried (current equals the target but differs from the prior). The first is NO_CHANGE; the second is ALREADY_TARGET, which the activity reports as re-applied. Collapsing both to NO_CHANGE would revert the workflow's decision back to the stale prior on a retry — a state/YNAB mismatch.

The enum and the decision. prior is the baseline that separates a no-op from a landed retry, and that anchors divergence.

src/ynab_agent/policy/converge.py · 156 lines

src/ynab_agent/policy/converge.py156 lines · Python

⋯ 95 lines hidden (lines 1–95)1"""Converge-to-target reconciliation for REVISING (SPEC §3 rules 2-4).
2 
3REVISING is the riskiest path: it mutates approved money records. The spine
4converges to a single target end-state — read current YNAB state, write only if
5it differs, then verify field-by-field. These pure helpers compute the three
6decisions that procedure needs: the dedup hash, whether a write is needed, and
7how a read-after-write compares to the target.
8"""
9 
10from __future__ import annotations
11 
12import hashlib
13from enum import StrEnum
14from typing import TYPE_CHECKING
15 
16from ynab_agent.domain.allocations import ResolvedAllocation
17from ynab_agent.domain.base import Frozen
18from ynab_agent.domain.events import VerifyOutcome
19 
20if TYPE_CHECKING:
21    from ynab_agent.domain.proposal import Decision
22    from ynab_agent.domain.transaction import YnabSnapshot
23 
24 
25class TargetState(Frozen):
26    """The normalized end-state a write converges to.
27 
28    Only the fields a write actually sets — the allocation, the memo, and the
29    approved flag — so two decisions that produce the same YNAB end-state hash
30    identically (and dedup), regardless of incidental metadata.
31    """
32 
33    allocation: ResolvedAllocation
34    memo: str | None = None
35    approved: bool = True
36 
37 
38def target_of(decision: Decision) -> TargetState:
39    """Project a decision onto its YNAB end-state."""
40    return TargetState(
41        allocation=decision.allocation,
42        memo=decision.memo,
43        approved=decision.approved,
44    )
45 
46 
47def content_hash(target: TargetState) -> str:
48    """A stable hash of the end-state, for ``(ynab_id, content_hash)`` dedup.
49 
50    Deterministic across processes (Pydantic JSON with fixed field order and
51    integer milliunits), so it is safe under Temporal replay.
52    """
53    return hashlib.sha256(target.model_dump_json().encode()).hexdigest()
54 
55 
56def needs_write(current: TargetState | None, target: TargetState) -> bool:
57    """Whether YNAB must be written (SPEC §3 rule 3).
58 
59    ``False`` only when the current state already equals the target — the
60    no-op exit that skips a needless intermediate write.
61    """
62    return current is None or content_hash(current) != content_hash(target)
63 
64 
65def classify_verify(
66    read_back: TargetState | None, target: TargetState
67) -> VerifyOutcome:
68    """Compare a read-after-write against the target (SPEC §3 rule 4).
69 
70    Args:
71        read_back: The post-write YNAB state, or ``None`` if it could not be
72            read after retries.
73        target: The intended end-state.
74 
75    Returns:
76        ``MATCH`` when they agree, ``COULD_NOT_CONFIRM`` when the read failed,
77        else ``DIVERGED`` (YNAB shows a different non-empty state).
78    """
79    if read_back is None:
80        return VerifyOutcome.COULD_NOT_CONFIRM
81    if content_hash(read_back) == content_hash(target):
82        return VerifyOutcome.MATCH
83    return VerifyOutcome.DIVERGED
84 
85 
86def reconciliation_blocks(snapshot: YnabSnapshot) -> bool:
87    """Whether the reconciliation guard forbids a silent edit (SPEC §3 rule 2).
88 
89    A transaction that is reconciled *or in a closed month* must not be silently
90    un-approved or edited; the spine should propose to the human (an additive
91    correction) instead.
92    """
93    return snapshot.reconciled or snapshot.month_closed
94 
95 
96class PrecommitAction(StrEnum):
97    """The pre-write decision for a converge run (SPEC §3 rules 3-4)."""
98 
99    WRITE = "write"
100    NO_CHANGE = "no_change"
101    ALREADY_TARGET = "already_target"
102    DIVERGED = "diverged"
103 
104 
105def precommit_action(
106    current: TargetState | None,
107    target: TargetState,
108    prior: TargetState | None,
109) -> PrecommitAction:
110    """Decide what to do *before* writing, from a fresh read of current state.
111 
112    This is SPEC §3 rule 3 (converge-to-target: read current YNAB state, write
113    only if it differs) plus the *pre-write* half of rule 4 (a spouse edited it
114    directly → don't clobber). Reading current state before the commit is what
115    lets the spine skip a needless write — and surface a divergence — instead of
116    overwriting it first and only noticing on the read-back.
117 
118    Args:
119        current: The freshly-read current end-state, or ``None`` when there
120            is no single category to read (a split, or uncategorized).
121        target: The end-state the instruction asks for.
122        prior: The end-state the agent last applied — the divergence baseline —
123            or ``None`` when the txn was never applied (a revision from LAPSED).
124 
125    Returns:
126        ``NO_CHANGE`` when YNAB already holds exactly the prior state and the
127        instruction asks for nothing new (no write, return to rest);
128        ``ALREADY_TARGET`` when the target is already in place but differs from
129        the prior — a retried converge whose write landed, or an edit that
130        happens to match — so adopt it as re-applied without rewriting;
131        ``DIVERGED`` when YNAB drifted to a different non-empty category
132        than the agent last applied (don't clobber, ask which wins);
133        otherwise ``WRITE``.
134    """
135    if not needs_write(current, target):
136        # YNAB already holds the target. A genuine no-op (unchanged from the
137        # prior) returns to rest; a target that differs from the prior is a
138        # write that already landed — adopt it rather than re-applying.
139        if (
140            current is not None
141            and prior is not None
142            and content_hash(current) == content_hash(prior)
143        ):
144            return PrecommitAction.NO_CHANGE
145        return PrecommitAction.ALREADY_TARGET
146    # A write is needed. Divergence is judged on the *allocation* (matching the
147    # silent-edit guard, §14.2): a spouse recategorising out of band leaves a
148    # different non-empty category than the agent last applied — surface it,
149    # don't overwrite. A memo-only drift is not a divergence.
150    if (
151        current is not None
152        and prior is not None
153        and current.allocation != prior.allocation
154    ):
155        return PrecommitAction.DIVERGED
156    return PrecommitAction.WRITE

Reading before writing

The activity now reads the current end-state first, asks precommit_action, and only reaches client.commit on the WRITE branch. NO_CHANGE, ALREADY_TARGET, and DIVERGED all return without touching YNAB. The write branch is unchanged from before — commit, read back, classify — so the verify path keeps its existing behavior.

The signature gains prior; the body decides before it writes. end_state/prior_state are the TargetState projections the pure helper compares.

src/ynab_agent/workflow/activities.py · 560 lines

src/ynab_agent/workflow/activities.py560 lines · Python

⋯ 402 lines hidden (lines 1–402)1"""The I/O ports of the transaction lifecycle, as Temporal activities.
2 
3Every side effect the workflow performs — reading YNAB, committing a write,
4sending email, the agentic enrichment/interpretation/converge steps — is an
5activity, so neither the model's nor the spine's I/O runs in workflow code
6(SPEC §0.5). These are *stubs*: typed signatures with no body. The real
7implementations (YNAB/AgentMail MCP, Pydantic AI) are wired in a later step; the
8workflow tests register mock implementations.
9"""
10 
11from __future__ import annotations
12 
13from typing import TYPE_CHECKING
14 
15from temporalio import activity
16 
17from ynab_agent.domain.allocations import (
  ProposedCategory,
  ProposedSplit,
  ResolvedCategory,
21)
22from ynab_agent.domain.effects import FeedRuleLearning, MessagePurpose
23from ynab_agent.domain.events import ConvergeOutcome, EnrichmentOutcome
24from ynab_agent.domain.ids import CategoryId
25from ynab_agent.domain.proposal import Decision, Proposal
26from ynab_agent.domain.signals import InboundSignal
27from ynab_agent.domain.transaction import YnabSnapshot
28from ynab_agent.policy.converge import TargetState
29from ynab_agent.workflow.types import ReplyOutcome
30 
31if TYPE_CHECKING:
  import datetime
33 
  # Annotation-only: never imported at runtime, so the agentic/model stack
  # (pydantic-ai) never enters the workflow sandbox. The enrich body imports
  # it lazily, inside the activity, where it runs outside the sandbox.
  from ynab_agent.agentic.enrich import CandidateCategory
  from ynab_agent.budget.overspend import CategorySpend
  from ynab_agent.domain.rule import Rule
  from ynab_agent.policy.floor import AutoActionCounters
41 
42 
43@activity.defn
44async def fetch_snapshot(ynab_id: str) -> YnabSnapshot | None:
  """Read the current YNAB snapshot, or ``None`` if not yet imported.
46 
  The YNAB client is imported lazily (keeping httpx out of the sandbox) and
  its blocking call runs off the event loop.
  """
  import asyncio
51 
  from ynab_agent.ynab.client import YnabClient
53 
  client = YnabClient.from_env()
  return await asyncio.to_thread(client.snapshot, ynab_id)
56 
57 
58def _candidates_from_spends(
  spends: tuple[CategorySpend, ...],
60) -> tuple[CandidateCategory, ...]:
  """Map the budget's live categories to the agent's candidate choices."""
  from ynab_agent.agentic.enrich import CandidateCategory
63 
  return tuple(
      CandidateCategory(id=str(spend.category), name=spend.name)
      for spend in spends
  )
68 
69 
70async def _load_payee_rules(payee: str) -> list[Rule]:
  """Query the durable rule registry for this payee's rules (W5, SPEC §14).
72 
  Returns ``[]`` when the registry has not been started yet (no learning has
  happened) or is unreachable — the conservative fallback, which routes the
  transaction to ASK rather than risking an auto-apply on stale knowledge.
  """
  from temporalio.service import RPCError
78 
  from ynab_agent.domain.rule import Rule
  from ynab_agent.workflow.registry_types import REGISTRY_WORKFLOW_ID
  from ynab_agent.workflow.temporal_client import client
82 
  temporal = await client()
  handle = temporal.get_workflow_handle(REGISTRY_WORKFLOW_ID)
  try:
      # ``result_type`` is load-bearing: the pydantic data converter decodes a
      # query payload to plain ``dict``s without it, and the gate reads real
      # ``Rule`` objects (``rule.match``) — a dict there raises AttributeError
      # and the enrich activity retries forever (SPEC §14 gate-load path).
      rules = await handle.query(
          "payee_rules", payee, result_type=tuple[Rule, ...]
      )
  except RPCError:
      return []
  return list(rules)
96 
97 
98async def _load_enrichment_inputs(
  snapshot: YnabSnapshot,
100) -> tuple[tuple[CandidateCategory, ...], list[Rule], AutoActionCounters]:
  """Fetch the candidate categories, in-scope rules, and auto counters.
102 
  The candidates are the budget's live categories, read from YNAB (the source
  of truth); the rules come from the durable registry, keyed on the payee. The
  auto-action counters start at zero in v1 (the per-run circuit breaker is a
  later stage); the gate (SPEC §4.2, §14) decides auto-vs-ask over the loaded
  rules — and auto-applies only a blessed one.
  """
  import asyncio
110 
  from ynab_agent.policy.floor import AutoActionCounters
  from ynab_agent.ynab.client import YnabClient
113 
  client = YnabClient.from_env()
  spends = await asyncio.to_thread(client.category_spends)
  rules = await _load_payee_rules(snapshot.payee)
  return _candidates_from_spends(spends), rules, AutoActionCounters()
118 
119 
120@activity.defn
121async def enrich(snapshot: YnabSnapshot) -> EnrichmentOutcome:
  """Assemble the proposal and route via the gate (the agentic middle).
123 
  The model stack is imported lazily, here in the activity body, so it never
  enters the workflow sandbox. The gate decides autonomy from the rules; the
  agent runs only to produce the proposal a human is asked to confirm.
  """
  from datetime import UTC, datetime
129 
  from ynab_agent.agentic.enrich import decide_enrichment
131 
  candidates, rules, counters = await _load_enrichment_inputs(snapshot)
  return await decide_enrichment(
      snapshot, candidates, rules, counters, now=datetime.now(UTC)
  )
136 
137 
138@activity.defn
139async def commit_to_ynab(ynab_id: str, decision: Decision) -> None:
  """Commit a decision to YNAB (the deterministic write).
141 
  Lazy YNAB client (httpx stays out of the sandbox); the blocking write runs
  off the event loop.
  """
  import asyncio
146 
  from ynab_agent.ynab.client import YnabClient
148 
  client = YnabClient.from_env()
  await asyncio.to_thread(client.commit, ynab_id, decision)
151 
152 
153@activity.defn
154async def read_back(ynab_id: str) -> TargetState | None:
  """Read the post-write end-state for verification, or ``None`` if unread."""
  import asyncio
157 
  from ynab_agent.ynab.client import YnabClient
159 
  client = YnabClient.from_env()
  return await asyncio.to_thread(client.read_back, ynab_id)
162 
163 
164def _txn_label(ynab_id: str) -> str:
  """The per-transaction idempotency label (open-thread dedup; SPEC §5)."""
  return f"yatxn-{ynab_id}"
167 
168 
169def _seq_label(ynab_id: str, action_seq: int) -> str:
  """The per-action idempotency label (send dedup; SPEC §3)."""
  return f"yaseq-{ynab_id}-{action_seq}"
172 
173 
174def _subject(snapshot: YnabSnapshot, category: str | None) -> str:
  """The thread's subject: payee + amount, and the suggested category.
176 
  No ``[YNAB]`` prefix — the sender address is a known contact. Naming the
  proposed category lets the owner act from the subject line alone.
  """
  base = f"{snapshot.payee} — {snapshot.amount}"
  return f"{base} · {category}?" if category else base
182 
183 
184def _date_display(day: datetime.date) -> str:
  """A short, friendly transaction date, e.g. ``May 29`` (no leading zero)."""
  return f"{day.strftime('%b')} {day.day}"
187 
188 
189def _allocation_display(
  allocation: ProposedCategory | ProposedSplit, names: dict[str, str]
191) -> str:
  """A human display of the proposed allocation, by category name."""
  if isinstance(allocation, ProposedCategory):
      return names.get(str(allocation.category), str(allocation.category))
  return " + ".join(
      names.get(str(line.category), str(line.category))
      for line in allocation.lines
  )
199 
200 
201async def _read_for_compose(
  ynab_id: str,
203) -> tuple[YnabSnapshot, dict[str, str]]:
  """Re-read the YNAB snapshot and a category-id→name map for composing.
205 
  Both come from YNAB (the source of truth) at send time — there is no stored
  copy of the facts or the category names (SPEC §0.5, store-free).
  """
  import asyncio
210 
  from ynab_agent.ynab.client import YnabClient
212 
  client = YnabClient.from_env()
  snapshot = await asyncio.to_thread(client.snapshot, ynab_id)
  if snapshot is None:
      msg = f"transaction {ynab_id} not found in YNAB at compose time"
      raise RuntimeError(msg)
  spends = await asyncio.to_thread(client.category_spends)
  names = {str(spend.category): spend.name for spend in spends}
  return snapshot, names
221 
222 
223def _render_message(
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
  purpose: MessagePurpose,
  names: dict[str, str],
228) -> str:
  """Lay out one message body for the thread (deterministic; SPEC §5).
230 
  The proposal's category + alternatives (model-chosen upstream) are resolved
  to names here and handed to the template — no model call at send time.
  """
  from ynab_agent.agentic.compose import ComposeRequest, render_body
235 
  proposed = (
      _allocation_display(proposal.allocation, names)
      if proposal is not None
      else None
  )
  alternatives = (
      tuple(names.get(str(alt), str(alt)) for alt in proposal.alternatives)
      if proposal is not None
      else ()
  )
  request = ComposeRequest(
      purpose=purpose.value,
      payee=snapshot.payee,
      amount_display=str(snapshot.amount),
      txn_date=_date_display(snapshot.txn_date),
      memo=snapshot.memo,
      proposed_category=proposed,
      alternatives=alternatives,
      rationale=proposal.rationale if proposal is not None else None,
  )
  return render_body(request)
257 
258 
259@activity.defn
260async def open_thread(ynab_id: str, proposal: Proposal | None) -> str:
  """Open the AgentMail thread by sending the proposal; returns its id.
262 
  A thread starts on its first send (AgentMail has no empty-thread create), so
  this composes + sends the proposal as the opening email. ``proposal`` is the
  current best guess (carried from workflow state); the txn facts + category
  names are re-read from YNAB (the source of truth) at compose time. The open
  is idempotent on the per-transaction label, so a retry re-finds the thread
  rather than sending a duplicate.
  """
  import asyncio
271 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
274 
  settings = Settings()
  snapshot, names = await _read_for_compose(ynab_id)
  body = _render_message(snapshot, proposal, MessagePurpose.PROPOSAL, names)
  proposed = (
      _allocation_display(proposal.allocation, names)
      if proposal is not None
      else None
  )
  mail = MailClient.from_env()
  return await asyncio.to_thread(
      mail.open_thread,
      inbox_id=settings.inbox,
      to=list(settings.owners),
      subject=_subject(snapshot, proposed),
      body=body,
      txn_label=_txn_label(ynab_id),
  )
292 
293 
294@activity.defn
295async def send_thread_message(
  ynab_id: str,
  thread_id: str | None,
  purpose: MessagePurpose,
  action_seq: int,
  proposal: Proposal | None,
301) -> None:
  """Send a follow-up message on the transaction's thread.
303 
  ``action_seq`` is the per-transaction idempotency key: the send dedups on it
  so a retry never double-sends (SPEC §3). ``proposal`` is the current best
  guess where the purpose needs it (re-proposal); other purposes derive their
  content from a re-read of the YNAB snapshot.
  """
  import asyncio
310 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
313 
  if thread_id is None:
      msg = f"cannot send {purpose.value} for {ynab_id}: no thread open yet"
      raise RuntimeError(msg)
  settings = Settings()
  snapshot, names = await _read_for_compose(ynab_id)
  body = _render_message(snapshot, proposal, purpose, names)
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.send_on_thread,
      inbox_id=settings.inbox,
      thread_id=thread_id,
      body=body,
      seq_label=_seq_label(ynab_id, action_seq),
  )
328 
329 
330def _proposed_category_id(proposal: Proposal | None) -> CategoryId | None:
  """The single proposed category id, or None (a split is not approvable)."""
  if proposal is None:
      return None
  allocation = proposal.allocation
  if isinstance(allocation, ProposedCategory):
      return allocation.category
  return None
338 
339 
340@activity.defn
341async def interpret_inbound(
  signal: InboundSignal,
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
345) -> ReplyOutcome:
  """Interpret a human reply into an answer or a question (SPEC §3, §5).
347 
  Free-form: the model reads whether the reply approves the proposal, names a
  different category, or asks a question. The spine — not the model — stamps
  the human + time into the resulting Decision. A non-reply inbound, or a
  missing/split proposal, can't be answered directly, so ask rather than guess
  a write. Names + candidates are re-read from YNAB at interpret time.
  """
  import asyncio
  from datetime import UTC, datetime
356 
  from ynab_agent.agentic.interpret import (
      InterpretRequest,
      interpret,
      to_reply_outcome,
  )
  from ynab_agent.domain.signals import ReplySignal
  from ynab_agent.workflow.types import ClarifyOutcome
  from ynab_agent.ynab.client import YnabClient
365 
  proposed_id = _proposed_category_id(proposal)
  if not isinstance(signal, ReplySignal) or proposed_id is None:
      return ClarifyOutcome(
          question="Could you say which category this should be?"
      )
371 
  client = YnabClient.from_env()
  spends = await asyncio.to_thread(client.category_spends)
  names = {str(spend.category): spend.name for spend in spends}
  request = InterpretRequest(
      reply_text=signal.text,
      payee=snapshot.payee,
      amount_display=str(snapshot.amount),
      proposed_category_name=names.get(str(proposed_id), str(proposed_id)),
      candidates=_candidates_from_spends(spends),
  )
  interpretation = await interpret(request)
  return to_reply_outcome(
      interpretation,
      proposed_category=proposed_id,
      decided_at=datetime.now(UTC),
  )
388 
389 
390def _target_summary(target: TargetState | None, names: dict[str, str]) -> str:
  """A short human description of an end-state, for a divergence note."""
  if target is None:
      return "(could not read)"
  allocation = target.allocation
  if isinstance(allocation, ResolvedCategory):
      label = names.get(str(allocation.category), str(allocation.category))
  else:
      label = "a split"
  return f"{label} — {target.memo}" if target.memo else label
400 
401 
402@activity.defn
403async def converge(
  snapshot: YnabSnapshot,
  instruction: InboundSignal,
  prior: Decision | None,
407) -> ConvergeOutcome:
⋯ 77 lines hidden (lines 408–484)408    """Converge a REVISING run to its target and verify it (SPEC §3 rules 3-4).
409 
  The agent reads the revision instruction into a target (retarget / memo /
  no-change). The spine then converges to it: it reads the *current* YNAB
  end-state first and, comparing against the agent's last-applied decision
  (``prior``), skips a needless write (the no-op exit), adopts a write that
  already landed, or surfaces a divergence (a spouse edited it directly) —
  *before* overwriting it — rather than committing blind and only noticing on
  the read-back. Only a genuine change writes, then verifies field-by-field. A
  reconciled or closed-month transaction, a non-reply instruction, or anything
  the model under-specifies routes to a human rather than a silent edit.
  """
  import asyncio
  from datetime import UTC, datetime
422 
  from ynab_agent.agentic.converge import (
      RevisionRequest,
      interpret_revision,
      to_revision_plan,
  )
  from ynab_agent.domain.enums import DecidedBy
  from ynab_agent.domain.events import (
      CouldNotConfirm,
      Diverged,
      NeedsHuman,
      NoChange,
      Reapplied,
      VerifyOutcome,
  )
  from ynab_agent.domain.signals import ReplySignal
  from ynab_agent.policy.converge import (
      PrecommitAction,
      classify_verify,
      precommit_action,
      reconciliation_blocks,
      target_of,
  )
  from ynab_agent.ynab.client import YnabClient
446 
  if reconciliation_blocks(snapshot):
      return NeedsHuman(
          reason="reconciled or closed-month — propose, don't silently edit"
      )
  if not isinstance(instruction, ReplySignal):
      return NeedsHuman(reason="non-reply revision instruction unsupported")
453 
  client = YnabClient.from_env()
  spends = await asyncio.to_thread(client.category_spends)
  names = {str(spend.category): spend.name for spend in spends}
  current_name = (
      names.get(str(snapshot.category_id), str(snapshot.category_id))
      if snapshot.category_id is not None
      else "(uncategorized)"
  )
  target = await interpret_revision(
      RevisionRequest(
          instruction=instruction.text,
          current_category_name=current_name,
          candidates=_candidates_from_spends(spends),
          current_memo=snapshot.memo,
      )
  )
  plan = to_revision_plan(target)
  if not plan.changes:
      return NoChange()
473 
  category = (
      CategoryId(plan.category_id)
      if plan.category_id is not None
      else snapshot.category_id
  )
  if category is None:
      return NeedsHuman(reason="revision did not resolve a category")
  decision = Decision(
      allocation=ResolvedCategory(category=category),
      memo=plan.memo if plan.memo is not None else snapshot.memo,
      approved=True,
      decided_by=DecidedBy.HUMAN,
      decided_at=datetime.now(UTC),
  )
  end_state = target_of(decision)
  prior_state = target_of(prior) if prior is not None else None
490 
  # SPEC §3 r3-4: read the current end-state *before* writing, then decide.
  current = await asyncio.to_thread(client.read_back, snapshot.ynab_id)
  action = precommit_action(current, end_state, prior_state)
  if action is PrecommitAction.NO_CHANGE:
      return NoChange()
  if action is PrecommitAction.ALREADY_TARGET:
      return Reapplied(decision=decision)
  if action is PrecommitAction.DIVERGED:
      return Diverged(
          ynab_summary=_target_summary(current, names),
          requested_summary=_target_summary(end_state, names),
      )
503 
  # PrecommitAction.WRITE: converge to the target, then verify (SPEC §3 r3-4).
  await asyncio.to_thread(client.commit, snapshot.ynab_id, decision)
  read = await asyncio.to_thread(client.read_back, snapshot.ynab_id)
  verdict = classify_verify(read, end_state)
  if verdict is VerifyOutcome.MATCH:
      return Reapplied(decision=decision)
  if verdict is VerifyOutcome.COULD_NOT_CONFIRM:
      return CouldNotConfirm()
  return Diverged(
      ynab_summary=_target_summary(read, names),
      requested_summary=_target_summary(end_state, names),
⋯ 46 lines hidden (lines 515–560)515    )
516 
517 
518@activity.defn
519async def feed_rule_learning(feed: FeedRuleLearning) -> None:
  """Persist a confirm/correct event into the durable rule registry (W5).
521 
  Signal-with-start on the singleton :class:`RuleRegistryWorkflow`: the first
  learning event creates it, every later one just delivers the signal, and the
  workflow folds the event into the rule table the autonomy gate reads (SPEC
  §9, §14). The conflict policy reuses the running singleton rather than
  starting a second registry.
  """
  from temporalio.common import WorkflowIDConflictPolicy
529 
  from ynab_agent.workflow.registry_types import (
      REGISTRY_WORKFLOW_ID,
      RegistryParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
535 
  temporal = await client()
  await temporal.start_workflow(
      "RuleRegistryWorkflow",
      RegistryParams(),
      id=REGISTRY_WORKFLOW_ID,
      task_queue=task_queue(),
      id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
      start_signal="record",
      start_signal_args=[feed],
  )
546 
547 
548@activity.defn
549async def close_thread(thread_id: str) -> None:
  """Label and close the AgentMail thread on archive."""
  import asyncio
552 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
555 
  settings = Settings()
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.close, inbox_id=settings.inbox, thread_id=thread_id
  )

Threading the baseline

The activity needs the last-applied decision as its divergence baseline. The W2 state machine already carries it: a Revising state holds prior (the decision it's revising, or None when entered from LAPSED). The workflow just passes it through — a one-line change, no new state.

st.prior is already on the Revising state; this hands it to the activity.

src/ynab_agent/workflow/txn_workflow.py · 548 lines

src/ynab_agent/workflow/txn_workflow.py548 lines · Python

⋯ 522 lines hidden (lines 1–522)1"""W2 · the Transaction Lifecycle workflow (SPEC §3, §0.5).
2 
3One durable workflow per ``ynab_id``. It is a thin *driver* around the pure
func:`~ynab_agent.domain.state_machine.advance` core: each step produces one
5event — from an activity (fetch / enrich / commit+verify / converge) or from a
6signal or an absolute-deadline timer — feeds it to ``advance``, then dispatches
7the emitted effects back out through activities. All nondeterminism lives in
8activities; the workflow uses only pure state and Temporal APIs
9(``workflow.now`` for time), so it replays deterministically. Long-lived
10transactions survive via ``continue-as-new`` from a resting state, carrying
11their state forward.
12 
13Deferred (a cohesive subsystem for its own step, SPEC §3, §9): the externalized,
14append-only **audit log**. ``TxnCore.audit_log_ref`` and the
15``_action_seq`` outbound-dedup key are plumbed here, but no audit entries are
16written yet — Temporal's own event history provides replay-safety in the
17meantime, and ``_action_seq`` is the idempotency key the real send activity will
18use so a retry never double-emails.
19"""
20 
21from __future__ import annotations
22 
23import datetime
24from collections import deque
25from datetime import timedelta
26from typing import TYPE_CHECKING
27 
28from temporalio import workflow
29from temporalio.common import SearchAttributeKey
30from temporalio.exceptions import ActivityError
31 
32if TYPE_CHECKING:
  from collections.abc import Callable
34 
  from ynab_agent.domain.proposal import Proposal
36 
37# The reply-routing reverse index: each workflow stamps its AgentMail thread_id
38# here on open_thread, so the dispatcher resolves an inbound reply's thread back
39# to this workflow with a Temporal visibility query (no separate store, §5a).
40# Registered on the namespace by manage/search-attributes.yaml.
41_TXN_THREAD_ID = SearchAttributeKey.for_keyword("TxnThreadId")
42 
43with workflow.unsafe.imports_passed_through():
  from ynab_agent.domain.config import DEFAULT_POLICY
  from ynab_agent.domain.effects import (
      CancelTimer,
      CloseThread,
      CommitToYnab,
      Effect,
      FeedRuleLearning,
      OpenThread,
      ReplayBuffered,
      SendThreadMessage,
      SetTimer,
      TimerKind,
  )
  from ynab_agent.domain.enums import DecidedBy
  from ynab_agent.domain.events import (
      AnswerReceived,
      ArchiveWindowReached,
      ClarifyRequested,
      Converged,
      Enriched,
      HoldDeadlineReached,
      HoldResolved,
      InboundReceived,
      LifecycleEvent,
      OverrideDetected,
      PatienceExpired,
      SnapshotMaterialized,
      SnapshotUnavailable,
      WriteVerified,
  )
  from ynab_agent.domain.ids import ThreadId, YnabTransactionId
  from ynab_agent.domain.proposal import Decision
  from ynab_agent.domain.signals import InboundSignal
  from ynab_agent.domain.state_machine import advance
  from ynab_agent.domain.transaction import (
      Applied,
      Archived,
      AutoApplied,
      AwaitingHuman,
      Discovered,
      Enriching,
      HoldAmazon,
      Lapsed,
      Open,
      Revising,
      Transaction,
      YnabSnapshot,
      born,
  )
  from ynab_agent.policy.converge import classify_verify, target_of
  from ynab_agent.workflow import activities, alert_activities
  from ynab_agent.workflow.alerting import build_failure_alert
  from ynab_agent.workflow.constants import (
      ACTIVITY_RETRY,
      ACTIVITY_TIMEOUT,
      ALERT_BUDGET,
      ALERT_RETRY,
      ALERT_TIMEOUT,
  )
  from ynab_agent.workflow.types import AnswerOutcome, TransactionParams
104 
105# History-length ceiling before a resting workflow continues-as-new. High enough
106# that ordinary flows never trip it; long-lived (30-45 day) transactions do.
107_CONTINUE_AS_NEW_AFTER = 4_000
108 
109# Resting states the workflow may sit in for days while waiting on signals or
110# timers; continue-as-new fires only from one of these. DISCOVERED counts: a
111# transaction born from a signal can wait here for a slow YNAB import (SPEC §3).
112_RESTING = (Discovered, AwaitingHuman, Open, Lapsed, HoldAmazon)
113 
114 
115def _is_amazon(payee: str) -> bool:
  return "amazon" in payee.lower()
117 
118 
119def _hold_for_amazon(snapshot: YnabSnapshot) -> bool:
  """Whether to hold for Amazon item detail: Amazon-ish and no memo yet."""
  return _is_amazon(snapshot.payee) and not snapshot.has_memo
122 
123 
124@workflow.defn
125class TransactionWorkflow:
  """The durable per-transaction lifecycle."""
127 
  def __init__(self) -> None:
      """Initialize empty state; ``run`` populates it from the params."""
      self._txn: Transaction
      self._ynab_id: str = ""
      self._thread_id: str | None = None
      self._deadlines: dict[TimerKind, datetime.datetime] = {}
      self._inbound: deque[InboundSignal] = deque()
      self._snapshot_ready: YnabSnapshot | None = None
      # Monotonic per-transaction counter: the outbound-send idempotency key
      # so a replay/retry never double-emails (SPEC §3 outbound dedup).
      self._action_seq: int = 0
139 
  # ── Signals (the external world pushes in) ──────────────────────────────
  @workflow.signal
  def submit_inbound(self, signal: InboundSignal) -> None:
      """A reply or matched receipt arrived (W3/W4)."""
      self._inbound.append(signal)
145 
  @workflow.signal
  def notify_snapshot(self, snapshot: YnabSnapshot) -> None:
      """W1 materialized (or backfilled the memo of) the YNAB snapshot."""
      self._snapshot_ready = snapshot
150 
  @workflow.query
  def state(self) -> str:
      """The current lifecycle state (for observability)."""
      return self._txn.state.value
155 
  # ── The durable loop ────────────────────────────────────────────────────
  @workflow.run
  async def run(self, params: TransactionParams) -> None:
      """Drive the lifecycle until the transaction is archived (SPEC §3)."""
      self._ynab_id = params.ynab_id
      self._thread_id = params.thread_id
      self._deadlines = dict(params.resume_deadlines)
      self._inbound.extend(params.resume_inbound)
      self._action_seq = params.resume_action_seq
      self._txn = params.resume_txn or born(
          YnabTransactionId(params.ynab_id), params.thread_id
      )
      self._sync_thread_id()
169 
      try:
          while not isinstance(self._txn, Archived):
              await self._step()
              if (
                  isinstance(self._txn, _RESTING)
                  and not self._inbound  # drain pending signals (SPEC §0.5)
                  and workflow.info().get_current_history_length()
                  > _CONTINUE_AS_NEW_AFTER
              ):
                  # continue_as_new raises to restart; nothing runs after it.
                  # (ContinueAsNewError is not an ActivityError, so it escapes
                  # the failure hook below untouched.)
                  workflow.continue_as_new(self._resume_params())
      except ActivityError as exc:
          # A terminal activity failure: a non-retryable bug (the constants
          # denylist) or the schedule_to_close budget elapsing. Page the owner
          # once — deduped, best-effort — then re-raise so the transaction
          # still fails and stays visible in Temporal (SPEC §13).
          await workflow.execute_activity(
              alert_activities.alert_failure,
              build_failure_alert(
                  key=self._ynab_id,
                  context=self._alert_context(),
                  exc=exc,
              ),
              start_to_close_timeout=ALERT_TIMEOUT,
              schedule_to_close_timeout=ALERT_BUDGET,
              retry_policy=ALERT_RETRY,
          )
          raise
200 
  def _alert_context(self) -> str:
      """A human locator for a failure alert: payee + txn id when known."""
      st = self._txn
      if isinstance(st, Discovered):
          return f"txn {self._ynab_id}"
      return f"{st.core.snapshot.payee} (txn {self._ynab_id})"
207 
  def _resume_params(self) -> TransactionParams:
      return TransactionParams(
          ynab_id=YnabTransactionId(self._ynab_id),
          thread_id=ThreadId(self._thread_id)
          if self._thread_id is not None
          else None,
          resume_txn=self._txn,
          resume_deadlines=dict(self._deadlines),
          resume_inbound=tuple(self._inbound),
          resume_action_seq=self._action_seq,
      )
219 
  async def _step(self) -> None:
      st = self._txn
      if isinstance(st, Discovered):
          await self._on_discovered()
      elif isinstance(st, Enriching):
          await self._on_enriching(st)
      elif isinstance(st, HoldAmazon):
          await self._on_hold(st)
      elif isinstance(st, AwaitingHuman):
          await self._on_awaiting(st)
      elif isinstance(st, Open):
          await self._on_open(st)
      elif isinstance(st, Lapsed):
          await self._on_lapsed()
      elif isinstance(st, Revising):
          await self._on_revising(st)
      elif isinstance(st, (AutoApplied, Applied, Archived)):
          # Transient (handled via the commit→verify follow-up) or terminal.
          return
239 
  # ── apply + effect dispatch ─────────────────────────────────────────────
  async def _dispatch(self, event: LifecycleEvent) -> None:
      transition = advance(
          self._txn, event, now=workflow.now(), policy=DEFAULT_POLICY
      )
      self._txn = transition.next
      # Keep the thread-id mirror in step with the state machine, which can
      # adopt a reply's thread on its own (e.g. a reply in DISCOVERED).
      self._sync_thread_id()
      for effect in transition.effects:
          followup = await self._execute(effect)
          if followup is not None:
              await self._dispatch(followup)
253 
  async def _execute(self, effect: Effect) -> LifecycleEvent | None:
      if isinstance(effect, CommitToYnab):
          await workflow.execute_activity(
              activities.commit_to_ynab,
              args=[self._ynab_id, effect.decision],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          read = await workflow.execute_activity(
              activities.read_back,
              self._ynab_id,
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          return WriteVerified(
              outcome=classify_verify(read, target_of(effect.decision))
          )
      if isinstance(effect, OpenThread):
          tid = await workflow.execute_activity(
              activities.open_thread,
              args=[self._ynab_id, self._proposal()],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          self._set_thread_id(tid)
          # Index this workflow by its thread for reply routing (§5a).
          workflow.upsert_search_attributes([_TXN_THREAD_ID.value_set(tid)])
      elif isinstance(effect, SendThreadMessage):
          self._action_seq += 1
          await workflow.execute_activity(
              activities.send_thread_message,
              args=[
                  self._ynab_id,
                  self._thread_id,
                  effect.purpose,
                  self._action_seq,
                  self._proposal(),
              ],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
      elif isinstance(effect, FeedRuleLearning):
          await workflow.execute_activity(
              activities.feed_rule_learning,
              effect,
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
      elif isinstance(effect, CloseThread):
          if self._thread_id is not None:
              await workflow.execute_activity(
                  activities.close_thread,
                  self._thread_id,
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=ACTIVITY_RETRY,
              )
      elif isinstance(effect, SetTimer):
          self._deadlines[effect.timer] = effect.deadline
      elif isinstance(effect, CancelTimer):
          self._deadlines.pop(effect.timer, None)
      elif isinstance(effect, ReplayBuffered):
          self._inbound.extendleft(reversed(effect.signals))
      return None
317 
  def _set_thread_id(self, tid: str) -> None:
      self._thread_id = tid
      st = self._txn
      if not isinstance(st, Discovered):
          new_core = st.core.model_copy(update={"thread_id": ThreadId(tid)})
          self._txn = st.model_copy(update={"core": new_core})
324 
  def _sync_thread_id(self) -> None:
      """Mirror the current transaction's thread id (SM may adopt it)."""
      st = self._txn
      tid = st.thread_id if isinstance(st, Discovered) else st.core.thread_id
      if tid is not None:
          self._thread_id = str(tid)
331 
  def _proposal(self) -> Proposal | None:
      """The current best-guess proposal, for states that carry one.
334 
      Passed to the mail activities so the proposal email can name the guess +
      alternatives; ``None`` for purposes whose content derives from YNAB.
      """
      st = self._txn
      if isinstance(st, (Enriching, AwaitingHuman, Lapsed)):
          return st.proposal
      return None
342 
  # ── per-state steps ─────────────────────────────────────────────────────
  async def _on_discovered(self) -> None:
      snapshot = await workflow.execute_activity(
          activities.fetch_snapshot,
          self._ynab_id,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      if snapshot is not None:
          await self._dispatch(
              SnapshotMaterialized(
                  snapshot=snapshot,
                  hold_for_amazon=_hold_for_amazon(snapshot),
              )
          )
          return
      # Signal beat the poll: wait for materialization or buffer an inbound.
      await self._dispatch(SnapshotUnavailable())
      await workflow.wait_condition(
          lambda: self._snapshot_ready is not None or len(self._inbound) > 0
      )
      if self._snapshot_ready is not None:
          snap = self._snapshot_ready
          self._snapshot_ready = None
          await self._dispatch(
              SnapshotMaterialized(
                  snapshot=snap, hold_for_amazon=_hold_for_amazon(snap)
              )
          )
      else:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
376 
  async def _on_enriching(self, st: Enriching) -> None:
      outcome = await workflow.execute_activity(
          activities.enrich,
          st.core.snapshot,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      await self._dispatch(Enriched(outcome=outcome))
385 
  async def _on_hold(self, st: HoldAmazon) -> None:
      ready = await self._wait_until(
          st.amazon_deadline,
          lambda: self._snapshot_ready is not None or len(self._inbound) > 0,
      )
      if not ready:
          await self._dispatch(HoldDeadlineReached())
      elif self._snapshot_ready is not None:
          snap = self._snapshot_ready
          self._snapshot_ready = None
          await self._dispatch(HoldResolved(snapshot=snap))
      else:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
401 
  async def _on_awaiting(self, st: AwaitingHuman) -> None:
      got_inbound = await self._wait_until(
          st.patience_deadline, self._has_inbound
      )
      if not got_inbound:
          before = type(self._txn)
          await self._dispatch(PatienceExpired())
          if type(self._txn) is not before:
              return  # lapsed (or otherwise transitioned)
          # Flagged (verify-failure) entry: PatienceExpired is ignored and
          # does not lapse (SPEC §3). Drop the passed timer and wait for an
          # inbound instead of re-spinning the expired deadline.
          self._deadlines.pop(TimerKind.PATIENCE, None)
          await workflow.wait_condition(self._has_inbound)
      await self._interpret_inbound(st.core.snapshot)
417 
  async def _interpret_inbound(self, snapshot: YnabSnapshot) -> None:
      signal = self._inbound.popleft()
      interpretation = await workflow.execute_activity(
          activities.interpret_inbound,
          args=[signal, snapshot, self._proposal()],
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      if isinstance(interpretation, AnswerOutcome):
          await self._dispatch(
              AnswerReceived(decision=interpretation.decision)
          )
      else:
          await self._dispatch(
              ClarifyRequested(question=interpretation.question)
          )
434 
  async def _on_open(self, st: Open) -> None:
      deadline = self._deadlines.get(TimerKind.ARCHIVE)
      got_inbound = (
          await self._wait_until(deadline, self._has_inbound)
          if deadline is not None
          else await self._wait_forever(self._has_inbound)
      )
      if got_inbound:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
          return
      # The archive window elapsed. Before closing the book, re-read YNAB to
      # catch a silent manual recategorization — an out-of-band correction
      # that must demote the driving rule (SPEC §14.2).
      event = await self._archive_or_override(st)
      before = type(self._txn)
      await self._dispatch(event)
      if type(self._txn) is before:
          # Archive blocked (not reconciled) and no override: drop the stale
          # timer and wait for an inbound rather than busy-looping (SPEC §3).
          self._deadlines.pop(TimerKind.ARCHIVE, None)
          await workflow.wait_condition(self._has_inbound)
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
461 
  async def _archive_or_override(self, st: Open) -> LifecycleEvent:
      """At archive time, detect a manual YNAB edit (SPEC §14.2).
464 
      Re-reads the current end-state and, if its category no longer matches
      the agent's applied decision, returns an ``OverrideDetected`` carrying
      the human's choice (the spine then demotes the rule); otherwise the
      ordinary ``ArchiveWindowReached``. A memo-only change is not an override
      — only the allocation is compared.
      """
      read = await workflow.execute_activity(
          activities.read_back,
          self._ynab_id,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      if read is not None and read.allocation != st.decision.allocation:
          human = Decision(
              allocation=read.allocation,
              memo=read.memo,
              approved=read.approved,
              decided_by=DecidedBy.HUMAN,
              decided_at=workflow.now(),
          )
          return OverrideDetected(decision=human)
      return ArchiveWindowReached()
487 
  async def _on_lapsed(self) -> None:
      await self._wait_then_revise_or_archive()
490 
  def _has_inbound(self) -> bool:
      return len(self._inbound) > 0
493 
  async def _wait_then_revise_or_archive(self) -> None:
      deadline = self._deadlines.get(TimerKind.ARCHIVE)
      got_inbound = (
          await self._wait_until(deadline, self._has_inbound)
          if deadline is not None
          else await self._wait_forever(self._has_inbound)
      )
      if got_inbound:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
          return
      # The archive window elapsed. Attempt to archive; if it is blocked (not
      # reconciled / not categorized) the state is unchanged, so drop the
      # now-stale timer and wait for an inbound rather than busy-looping on
      # the already-passed deadline.
      before = type(self._txn)
      await self._dispatch(ArchiveWindowReached())
      if type(self._txn) is before:
          self._deadlines.pop(TimerKind.ARCHIVE, None)
          await workflow.wait_condition(self._has_inbound)
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
518 
  async def _wait_forever(self, predicate: Callable[[], bool]) -> bool:
      await workflow.wait_condition(predicate)
      return True
522 
  async def _on_revising(self, st: Revising) -> None:
      outcome = await workflow.execute_activity(
          activities.converge,
          args=[st.core.snapshot, st.instruction, st.prior],
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      await self._dispatch(Converged(outcome=outcome))
⋯ 18 lines hidden (lines 531–548)531 
  async def _wait_until(
      self,
      deadline: datetime.datetime,
      predicate: Callable[[], bool],
  ) -> bool:
      """Wait for ``predicate`` until an absolute deadline.
538 
      Returns ``True`` if the predicate fired first, ``False`` on timeout.
      """
      timeout = deadline - workflow.now()
      if timeout < timedelta(0):
          timeout = timedelta(0)
      try:
          await workflow.wait_condition(predicate, timeout=timeout)
      except TimeoutError:
          return False
      return True

The discriminating tests

Two layers of tests. The pure precommit_action cases pin every branch, including the two that are easy to get wrong: a landed retry must be ALREADY_TARGET (not NO_CHANGE), and a memo-only drift must not read as divergence.

The no-op / already-target / diverged trio; the write and edge cases follow (folded).

tests/policy/test_converge.py · 152 lines

tests/policy/test_converge.py152 lines · Python

⋯ 85 lines hidden (lines 1–85)1"""Tests for the converge-to-target reconciliation (SPEC §3 rules 2-4)."""
2 
3from __future__ import annotations
4 
5import datetime
6 
7from ynab_agent.domain.allocations import ResolvedCategory
8from ynab_agent.domain.enums import ClearedState, DecidedBy
9from ynab_agent.domain.events import VerifyOutcome
10from ynab_agent.domain.ids import AccountId, CategoryId, YnabTransactionId
11from ynab_agent.domain.money import Money
12from ynab_agent.domain.proposal import Decision
13from ynab_agent.domain.transaction import YnabSnapshot
14from ynab_agent.policy.converge import (
15    PrecommitAction,
16    TargetState,
17    classify_verify,
18    content_hash,
19    needs_write,
20    precommit_action,
21    reconciliation_blocks,
22    target_of,
23)
24 
25_EPOCH = datetime.datetime(2026, 5, 28, tzinfo=datetime.UTC)
26 
27 
28def _target(category: str = "dining", memo: str | None = None) -> TargetState:
29    return TargetState(
30        allocation=ResolvedCategory(category=CategoryId(category)), memo=memo
31    )
32 
33 
34def test_content_hash_is_stable_and_distinguishing() -> None:
35    assert content_hash(_target()) == content_hash(_target())
36    assert content_hash(_target("dining")) != content_hash(_target("gifts"))
37    assert content_hash(_target(memo="a")) != content_hash(_target(memo="b"))
38 
39 
40def test_needs_write_skips_an_equal_state() -> None:
41    assert not needs_write(_target(), _target())
42    assert needs_write(_target("dining"), _target("gifts"))
43    assert needs_write(None, _target())
44 
45 
46def test_classify_verify_outcomes() -> None:
47    target = _target("dining")
48    assert classify_verify(target, target) is VerifyOutcome.MATCH
49    assert classify_verify(None, target) is VerifyOutcome.COULD_NOT_CONFIRM
50    assert classify_verify(_target("gifts"), target) is VerifyOutcome.DIVERGED
51 
52 
53def test_target_of_projects_a_decision() -> None:
54    decision = Decision(
55        allocation=ResolvedCategory(category=CategoryId("dining")),
56        memo="coffee",
57        approved=True,
58        decided_by=DecidedBy.AGENT,
59        decided_at=_EPOCH,
60    )
61    target = target_of(decision)
62    assert target.memo == "coffee"
63    assert content_hash(target) == content_hash(_target(memo="coffee"))
64 
65 
66def test_reconciliation_guard() -> None:
67    base: dict[str, object] = {
68        "ynab_id": YnabTransactionId("t1"),
69        "account": AccountId("a1"),
70        "payee": "Blue Bottle",
71        "amount": Money.from_currency("-4.50"),
72        "txn_date": datetime.date(2026, 5, 28),
73    }
74    reconciled = YnabSnapshot.model_validate(
75        {**base, "cleared": ClearedState.RECONCILED}
76    )
77    closed_month = YnabSnapshot.model_validate({**base, "month_closed": True})
78    cleared = YnabSnapshot.model_validate(
79        {**base, "cleared": ClearedState.CLEARED}
80    )
81    assert reconciliation_blocks(reconciled)
82    assert reconciliation_blocks(closed_month)
83    assert not reconciliation_blocks(cleared)
84 
85 
86# ── precommit_action: the pre-write converge decision (SPEC §3 r3-4) ─────────
87def test_precommit_no_change_when_target_equals_prior() -> None:
88    # YNAB already holds the prior state and the target asks nothing new.
89    action = precommit_action(
90        _target("dining"), _target("dining"), _target("dining")
91    )
92    assert action is PrecommitAction.NO_CHANGE
93 
94 
95def test_precommit_already_target_when_write_already_landed() -> None:
96    # current == target but differs from the prior: a retried converge whose
97    # write landed (or an out-of-band edit that happens to match the target).
98    # Adopt it as re-applied rather than rewriting — and never as NO_CHANGE,
99    # which would revert the workflow's decision to the stale prior.
100    assert (
101        precommit_action(_target("gifts"), _target("gifts"), _target("dining"))
102        is PrecommitAction.ALREADY_TARGET
103    )
104    # No prior (a revision entered from LAPSED) with the target already present.
105    assert (
106        precommit_action(_target("gifts"), _target("gifts"), None)
107        is PrecommitAction.ALREADY_TARGET
108    )
109 
110 
111def test_precommit_diverged_on_out_of_band_recategorisation() -> None:
112    # A write is needed, but YNAB drifted to a different non-empty category than
113    # the agent last applied (a spouse edited it directly) — surface it BEFORE
114    # writing, never clobber.
115    action = precommit_action(
116        _target("groceries"), _target("gifts"), _target("dining")
117    )
118    assert action is PrecommitAction.DIVERGED
⋯ 34 lines hidden (lines 119–152)119 
120 
121def test_precommit_writes_when_current_matches_prior() -> None:
122    # The normal revision: YNAB still shows what the agent last applied, and the
123    # target differs — converge.
124    action = precommit_action(
125        _target("dining"), _target("gifts"), _target("dining")
126    )
127    assert action is PrecommitAction.WRITE
128 
129 
130def test_precommit_writes_from_lapsed_without_a_prior() -> None:
131    # No prior baseline (entered from LAPSED): there is nothing of ours to
132    # clobber, so a differing target simply writes.
133    action = precommit_action(_target("dining"), _target("gifts"), None)
134    assert action is PrecommitAction.WRITE
135 
136 
137def test_precommit_writes_when_current_is_unreadable() -> None:
138    # current None (a split or uncategorized) is not a divergence — there is no
139    # non-empty state to protect — so it writes (and the read-back classifies).
140    action = precommit_action(None, _target("gifts"), _target("dining"))
141    assert action is PrecommitAction.WRITE
142 
143 
144def test_precommit_memo_only_drift_is_not_divergence() -> None:
145    # A spouse changing only the memo (same category) is not a divergence; the
146    # allocation still matches the prior, so the converge proceeds.
147    action = precommit_action(
148        _target("dining", memo="their note"),
149        _target("gifts"),
150        _target("dining", memo="orig"),
151    )
152    assert action is PrecommitAction.WRITE

Then an integration test drives the real converge activity with a stub client and a stubbed agent, asserting the write is actually skipped. This is the test the old code would fail: it committed in every case, so fake.commits would never be empty for a no-op or a divergence.

The stub records commits; the no-op and divergence cases assert it stays empty.

tests/workflow/test_inbound_activities.py · 254 lines

tests/workflow/test_inbound_activities.py254 lines · Python

⋯ 107 lines hidden (lines 1–107)1"""Tests for the W2 inbound activities (SPEC §3, §5).
2 
3``interpret_inbound`` and ``converge`` are thin glue over the interpret/converge
4agents (``tests/agentic``) and the verify policy (``tests/policy``). Their pure
5helpers (the single proposed category, an end-state summary) are covered here.
6The ``converge`` *orchestration* is also covered directly: it reads the current
7YNAB state and decides before writing (SPEC §3 r3-4), so a no-op and a
8divergence never issue a clobbering write.
9"""
10 
11from __future__ import annotations
12 
13import datetime
14from typing import TYPE_CHECKING
15 
16from ynab_agent.budget.overspend import CategorySpend
17from ynab_agent.domain.allocations import (
18    PercentShare,
19    ProposedCategory,
20    ProposedSplit,
21    ResolvedCategory,
22    SplitLine,
23)
24from ynab_agent.domain.enums import Confidence, DecidedBy
25from ynab_agent.domain.events import Diverged, NoChange, Reapplied
26from ynab_agent.domain.ids import (
27    AccountId,
28    CategoryId,
29    MessageId,
30    ThreadId,
31    YnabTransactionId,
32)
33from ynab_agent.domain.money import Money
34from ynab_agent.domain.proposal import Decision, Proposal
35from ynab_agent.domain.signals import ReplySignal
36from ynab_agent.domain.transaction import YnabSnapshot
37from ynab_agent.policy.converge import TargetState, target_of
38from ynab_agent.workflow.activities import (
39    _proposed_category_id,
40    _target_summary,
41    converge,
42)
43 
44if TYPE_CHECKING:
45    import pytest
46 
47_NAMES = {"dining": "Dining Out", "coffee": "Coffee"}
48 
49 
50def _proposal(allocation: object) -> Proposal:
51    return Proposal(
52        allocation=allocation,  # type: ignore[arg-type]
53        confidence=Confidence.MEDIUM,
54        rationale="because",
55    )
56 
57 
58def test_proposed_category_id_for_single_category() -> None:
59    proposal = _proposal(ProposedCategory(category=CategoryId("dining")))
60    assert _proposed_category_id(proposal) == "dining"
61 
62 
63def test_proposed_category_id_none_for_split() -> None:
64    split = ProposedSplit(
65        lines=(
66            SplitLine(
67                share=PercentShare(percent=50), category=CategoryId("dining")
68            ),
69            SplitLine(
70                share=PercentShare(percent=50), category=CategoryId("coffee")
71            ),
72        )
73    )
74    assert _proposed_category_id(_proposal(split)) is None
75 
76 
77def test_proposed_category_id_none_for_no_proposal() -> None:
78    assert _proposed_category_id(None) is None
79 
80 
81def test_target_summary_names_category_and_memo() -> None:
82    target = TargetState(
83        allocation=ResolvedCategory(category=CategoryId("dining")),
84        memo="team lunch",
85    )
86    assert _target_summary(target, _NAMES) == "Dining Out — team lunch"
87 
88 
89def test_target_summary_without_memo() -> None:
90    target = TargetState(
91        allocation=ResolvedCategory(category=CategoryId("coffee"))
92    )
93    assert _target_summary(target, _NAMES) == "Coffee"
94 
95 
96def test_target_summary_handles_unreadable_state() -> None:
97    assert _target_summary(None, _NAMES) == "(could not read)"
98 
99 
100def test_target_summary_falls_back_to_id() -> None:
101    target = TargetState(
102        allocation=ResolvedCategory(category=CategoryId("mystery"))
103    )
104    assert _target_summary(target, _NAMES) == "mystery"
105 
106 
107# ── converge orchestration: read current state before writing (SPEC §3 r3-4) ──
108class _FakeYnab:
109    """A YNAB client stub that records commits and serves a current state.
110 
111    ``read_back`` returns the current end-state; a ``commit`` lands the write so
112    a subsequent ``read_back`` (the post-write verify) returns the target.
113    """
114 
115    def __init__(self, current: TargetState | None) -> None:
116        self._state = current
117        self.commits: list[Decision] = []
118 
119    def category_spends(self) -> tuple[CategorySpend, ...]:
120        zero = Money.from_milliunits(0)
121        return tuple(
122            CategorySpend(
123                category=CategoryId(name),
124                name=name.title(),
125                budgeted=zero,
126                activity=zero,
127                balance=zero,
128            )
129            for name in ("dining", "gifts", "groceries")
130        )
131 
132    def read_back(self, ynab_id: str) -> TargetState | None:
133        return self._state
134 
135    def commit(self, ynab_id: str, decision: Decision) -> None:
⋯ 60 lines hidden (lines 136–195)136        self.commits.append(decision)
137        self._state = target_of(decision)
138 
139 
140def _converge_snapshot() -> YnabSnapshot:
141    return YnabSnapshot(
142        ynab_id=YnabTransactionId("t-1"),
143        account=AccountId("a1"),
144        payee="Blue Bottle",
145        amount=Money.from_currency("-4.50"),
146        txn_date=datetime.date(2026, 5, 30),
147        category_id=CategoryId("dining"),
148    )
149 
150 
151def _reply(text: str) -> ReplySignal:
152    return ReplySignal(
153        thread_id=ThreadId("thread-1"),
154        message_id=MessageId("m-1"),
155        from_address="matthew@example.com",
156        text=text,
157    )
158 
159 
160def _decision(category: str) -> Decision:
161    return Decision(
162        allocation=ResolvedCategory(category=CategoryId(category)),
163        approved=True,
164        decided_by=DecidedBy.HUMAN,
165        decided_at=datetime.datetime(2026, 5, 28, tzinfo=datetime.UTC),
166    )
167 
168 
169async def _run_converge(
170    monkeypatch: pytest.MonkeyPatch,
171    *,
172    current: TargetState | None,
173    prior: Decision | None,
174    retarget_to: str = "gifts",
175) -> tuple[object, _FakeYnab]:
176    """Drive the converge activity with a stub client and a stubbed agent."""
177    from ynab_agent.agentic import converge as converge_mod
178    from ynab_agent.ynab.client import YnabClient
179 
180    fake = _FakeYnab(current)
181 
182    async def _fake_interpret(
183        request: object, *, model: object = None
184    ) -> converge_mod.RevisionTarget:
185        return converge_mod.RevisionTarget(
186            decision=converge_mod.RevisionDecision.RETARGET,
187            category_id=retarget_to,
188        )
189 
190    monkeypatch.setattr(converge_mod, "interpret_revision", _fake_interpret)
191    monkeypatch.setattr(YnabClient, "from_env", classmethod(lambda cls: fake))
192    outcome = await converge(
193        _converge_snapshot(), _reply("make it gifts"), prior
194    )
195    return outcome, fake
196 
197 
198async def test_converge_no_op_skips_the_write(
199    monkeypatch: pytest.MonkeyPatch,
200) -> None:
201    # The target already equals what the agent last applied: no write at all.
202    outcome, fake = await _run_converge(
203        monkeypatch,
204        current=target_of(_decision("gifts")),
205        prior=_decision("gifts"),
206    )
207    assert isinstance(outcome, NoChange)
208    assert fake.commits == []
209 
210 
211async def test_converge_adopts_a_landed_write_without_rewriting(
212    monkeypatch: pytest.MonkeyPatch,
213) -> None:
214    # The target is already in YNAB but differs from the prior (a retried
215    # converge whose write landed): re-applied, not NoChange, and not rewritten.
216    outcome, fake = await _run_converge(
217        monkeypatch,
218        current=target_of(_decision("gifts")),
219        prior=_decision("dining"),
220    )
221    assert isinstance(outcome, Reapplied)
222    assert fake.commits == []
223 
224 
225async def test_converge_surfaces_divergence_without_clobbering(
226    monkeypatch: pytest.MonkeyPatch,
227) -> None:
228    # A spouse recategorised the txn out of band: surface it, never overwrite.
229    outcome, fake = await _run_converge(
230        monkeypatch,
231        current=target_of(_decision("groceries")),
232        prior=_decision("dining"),
233    )
234    assert isinstance(outcome, Diverged)
235    assert outcome.ynab_summary == "Groceries"
236    assert outcome.requested_summary == "Gifts"
237    assert fake.commits == []
⋯ 17 lines hidden (lines 238–254)238 
239 
240async def test_converge_writes_then_verifies_a_real_change(
241    monkeypatch: pytest.MonkeyPatch,
242) -> None:
243    # YNAB still shows what the agent last applied and the target differs:
244    # converge (one write), then verify the read-back.
245    outcome, fake = await _run_converge(
246        monkeypatch,
247        current=target_of(_decision("dining")),
248        prior=_decision("dining"),
249    )
250    assert isinstance(outcome, Reapplied)
251    assert outcome.decision.allocation == ResolvedCategory(
252        category=CategoryId("gifts")
253    )
254    assert len(fake.commits) == 1