What changed, and why

A blank Amazon transaction parks in HOLD_AMAZON to wait for Amazon's daily item-detail sync (principle 5: "wait, don't race, on Amazon"). The hold is meant to exit early "if a receipt matches or the memo backfills." The receipt path works. The memo-backfill path was dead.

A held transaction stays unapproved, so W1 re-addresses it every tick. But address_transaction only starts a W2 — a run already in flight raises WorkflowAlreadyStartedError and the activity returns. Nothing ever delivered the W2's notify_snapshot signal. So a held Amazon txn waited out the full ~36h deadline, even though the memo lands hours earlier at the 02:00 sync.

The W2 side was already built — _on_hold consumes notify_snapshot to resolve the hold. The missing half was W1 actually calling it. This change adds that call, gated on the one condition that makes it meaningful.

The plan decision

W1's planner is pure and stateless per tick. It can't observe the empty→present transition of a memo; it only sees the current snapshot. So "is this a backfill to deliver?" reduces to: an Amazon payee that currently carries a memo. That's memo_backfilled, surfaced as notify_existing on the action.

notify_existing rides on the action; is_amazon/memo_backfilled are the pure predicates, and plan_ingest sets the flag.

src/ynab_agent/ingest/plan.py · 105 lines
src/ynab_agent/ingest/plan.py105 lines · Python
⋯ 33 lines hidden (lines 1–33)
1"""Ingestion planning: turn a YNAB delta into per-transaction W2 actions.
2 
3W1 polls the YNAB ``transactions`` delta and, for each new/unapproved in-scope
4transaction, addresses its W2 by ``ynab_id`` via signal-with-start (SPEC §2).
5The *decision* of which transactions to address — and how — is pure and lives
6here; the spine does the signal-with-start. Two SPEC rules shape it:
7 
8* **Cold-start cutover (§13).** On the first run the cursor is captured *without
9 acting*, so the agent never emails about the entire pre-existing backlog.
10* **Import lifecycle (§13).** A YNAB-matched/duplicate import is not
11 auto-approved — it is routed to a human.
12"""
13 
14from __future__ import annotations
15 
16from typing import TYPE_CHECKING
17 
18from ynab_agent.domain.base import Frozen
19from ynab_agent.domain.transaction import YnabSnapshot
20from ynab_agent.ingest.scope import IngestScope, in_scope
21 
22if TYPE_CHECKING:
23 from collections.abc import Iterable
24 
25 
26class AddressTxn(Frozen):
27 """A decision to address one transaction's W2 (start-or-signal).
28 
29 Attributes:
30 snapshot: The polled YNAB snapshot to hand to the W2.
31 route_to_human: ``True`` for a matched/duplicate import — the W2 must
32 not auto-approve it (SPEC §13).
33 notify_existing: ``True`` when an Amazon item-detail memo has backfilled
34 (Amazon payee + a memo now present). If the W2 is already in flight,
35 W1 signals it the fresh snapshot so a ``HOLD_AMAZON`` run resolves
36 early instead of waiting out the ~36h deadline (SPEC §2, §3).
37 """
38 
39 snapshot: YnabSnapshot
40 route_to_human: bool = False
41 notify_existing: bool = False
42 
43 
44def is_duplicate_import(snapshot: YnabSnapshot) -> bool:
45 """Whether YNAB matched this import to an existing txn (SPEC §13)."""
46 return snapshot.matched_transaction_id is not None
47 
48 
49def is_amazon(payee: str) -> bool:
50 """Whether a payee is Amazon-ish, so the §3 item-detail hold applies.
51 
52 A deliberately loose substring match (SPEC §11 leaves the exact payee
53 patterns open). The *same* predicate gates W1's backfill signal here and
54 W2's hold entry, so the two never disagree about what counts as Amazon.
55 """
56 return "amazon" in payee.lower()
57 
58 
59def memo_backfilled(snapshot: YnabSnapshot) -> bool:
60 """Whether an Amazon hold can now resolve: Amazon payee + memo present.
61 
62 The condition W1 turns into a ``notify_snapshot`` to an already-running W2
63 (SPEC §2, §3). W1 is stateless per tick, so it cannot see the empty→present
64 *transition*; an Amazon txn that currently carries a memo is exactly the
65 resolvable case, and signalling a W2 that is already past the hold is a
66 harmless no-op.
67 """
68 return is_amazon(snapshot.payee) and snapshot.has_memo
69 
70 
71def plan_ingest(
72 snapshots: Iterable[YnabSnapshot],
⋯ 23 lines hidden (lines 73–95)
73 scope: IngestScope,
74 *,
75 cold_start: bool,
76) -> tuple[AddressTxn, ...]:
77 """Plan the W2 actions for one YNAB delta page. Pure.
78 
79 Args:
80 snapshots: The transactions in this delta page (already normalized).
81 scope: The fail-closed ingestion scope.
82 cold_start: Whether this is the first poll (capture-cursor-only).
83 
84 Returns:
85 One :class:`AddressTxn` per in-scope, unapproved txn — empty on cold
86 start.
87 
88 Only *unapproved* transactions are addressed (SPEC §2/§13 "new/unapproved").
89 An approved transaction is the owner's settled state — the agent never
90 re-triages or emails about something already approved (whether the owner
91 approved it in YNAB or the agent's own triage did). This is what makes "act
92 on the current backlog" safe: only the outstanding, unapproved transactions
93 surface.
94 """
95 if cold_start:
96 return ()
97 return tuple(
98 AddressTxn(
99 snapshot=snap,
100 route_to_human=is_duplicate_import(snap),
101 notify_existing=memo_backfilled(snap),
102 )
103 for snap in snapshots
104 if in_scope(snap, scope) and not snap.approved
⋯ 1 line hidden (lines 105–105)
105 )

W1 delivers the signal

On the already-running path, when the action is flagged, W1 signals the existing W2's notify_snapshot with the fresh (memo-carrying) snapshot. A settled run can no longer be signalled, so the RPCError is suppressed and the next tick re-tries while the txn stays unapproved — the ~36h deadline remains the backstop.

The one exception to the start-is-a-no-op rule: signal the running W2 on an Amazon memo backfill.

src/ynab_agent/workflow/poll_activities.py · 80 lines
src/ynab_agent/workflow/poll_activities.py80 lines · Python
⋯ 54 lines hidden (lines 1–54)
1"""The I/O ports of the W1 ingestion poller, as Temporal activities.
2 
3Kept separate from the W2 :mod:`ynab_agent.workflow.activities` so the W2
4workflow's sandbox import graph stays minimal — pulling the ingest/poll types
5into the W2 activity module duplicates domain classes under the sandbox and
6breaks discriminated-union validation. Heavy clients (YNAB, Temporal) are
7imported lazily inside the bodies so they never enter the workflow sandbox.
8"""
9 
10from __future__ import annotations
11 
12from temporalio import activity
13 
14from ynab_agent.domain.transaction import YnabSnapshot
15from ynab_agent.ingest.plan import AddressTxn
16 
17 
18@activity.defn
19async def fetch_unapproved() -> tuple[YnabSnapshot, ...]:
20 """Read YNAB's currently unapproved transactions (SPEC §2 W1).
21 
22 The agent's outstanding-work set — YNAB's own ``type=unapproved`` view. It
23 excludes tentatively scheduled/auto-matched imports until they land. The
24 poll re-reads it each tick; the set is small, so there is no cursor to carry
25 (what's outstanding is derived from YNAB, not stored — SPEC §0.5).
26 """
27 import asyncio
28 
29 from ynab_agent.ynab.client import YnabClient
30 
31 client = YnabClient.from_env()
32 return await asyncio.to_thread(client.unapproved)
33 
34 
35@activity.defn
36async def address_transaction(action: AddressTxn) -> None:
37 """Start the transaction's W2 by ``ynab_id`` (SPEC §2, §3).
38 
39 Idempotent by construction: the W2's workflow id *is* the YNAB transaction
40 id, so "what's new" is derived from whether a workflow already exists — no
41 stored seen-set. The reuse policy is ``ALLOW_DUPLICATE_FAILED_ONLY``: a txn
42 already in flight, or one that *completed* (archived — even un-replied),
43 raises :class:`WorkflowAlreadyStartedError` and is a no-op, so the agent
44 never re-triages settled work; but a run that was *terminated* (an operator
45 reset) can be re-created. The W2 reads its own snapshot in ``DISCOVERED``.
46 ``route_to_human`` is not yet differentiated in v1 — the autonomy gate
47 already routes conservatively — so it is carried for observability only.
48 
49 One exception to the no-op: when ``notify_existing`` is set (an Amazon
50 item-detail memo has backfilled, SPEC §2/§3), W1 *signals* the
51 already-running W2 the fresh snapshot via ``notify_snapshot`` so a
52 ``HOLD_AMAZON`` run resolves early instead of waiting out the ~36h deadline.
53 Other states ignore the signal; a settled run can no longer be signalled
54 (suppressed), and the next tick re-tries while the txn stays unapproved.
55 """
56 import contextlib
57 
58 from temporalio.common import WorkflowIDReusePolicy
59 from temporalio.exceptions import WorkflowAlreadyStartedError
60 from temporalio.service import RPCError
61 
62 from ynab_agent.workflow.temporal_client import client, task_queue
63 from ynab_agent.workflow.types import TransactionParams
64 
65 ynab_id = action.snapshot.ynab_id
66 temporal = await client()
67 try:
68 await temporal.start_workflow(
69 "TransactionWorkflow",
70 TransactionParams(ynab_id=ynab_id),
71 id=str(ynab_id),
72 task_queue=task_queue(),
73 id_reuse_policy=(WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY),
74 )
75 except WorkflowAlreadyStartedError:
76 if not action.notify_existing:
77 return
78 handle = temporal.get_workflow_handle(str(ynab_id))
79 with contextlib.suppress(RPCError):
80 await handle.signal("notify_snapshot", action.snapshot)

One predicate, two callers — and the W2 side it wakes

W1's signal-gate and W2's hold-entry must agree on what counts as Amazon, or a txn could enter the hold (W2's view) that W1 never signals (or vice versa). So is_amazon is promoted to ingest.plan and shared: W2's _hold_for_amazon now calls the same predicate.

The hold-entry test, now sharing W1's is_amazon — a single source of truth (SPEC §11 leaves the patterns open).

src/ynab_agent/workflow/txn_workflow.py · 549 lines
src/ynab_agent/workflow/txn_workflow.py549 lines · Python
⋯ 115 lines hidden (lines 1–115)
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
4: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:
33 from collections.abc import Callable
34 
35 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():
44 from ynab_agent.domain.config import DEFAULT_POLICY
45 from ynab_agent.domain.effects import (
46 CancelTimer,
47 CloseThread,
48 CommitToYnab,
49 Effect,
50 FeedRuleLearning,
51 OpenThread,
52 ReplayBuffered,
53 SendThreadMessage,
54 SetTimer,
55 TimerKind,
56 )
57 from ynab_agent.domain.enums import DecidedBy
58 from ynab_agent.domain.events import (
59 AnswerReceived,
60 ArchiveWindowReached,
61 ClarifyRequested,
62 Converged,
63 Enriched,
64 HoldDeadlineReached,
65 HoldResolved,
66 InboundReceived,
67 LifecycleEvent,
68 OverrideDetected,
69 PatienceExpired,
70 SnapshotMaterialized,
71 SnapshotUnavailable,
72 WriteVerified,
73 )
74 from ynab_agent.domain.ids import ThreadId, YnabTransactionId
75 from ynab_agent.domain.proposal import Decision
76 from ynab_agent.domain.signals import InboundSignal
77 from ynab_agent.domain.state_machine import advance
78 from ynab_agent.domain.transaction import (
79 Applied,
80 Archived,
81 AutoApplied,
82 AwaitingHuman,
83 Discovered,
84 Enriching,
85 HoldAmazon,
86 Lapsed,
87 Open,
88 Revising,
89 Transaction,
90 YnabSnapshot,
91 born,
92 )
93 from ynab_agent.ingest.plan import is_amazon
94 from ynab_agent.policy.converge import classify_verify, target_of
95 from ynab_agent.workflow import activities, alert_activities
96 from ynab_agent.workflow.alerting import build_failure_alert
97 from ynab_agent.workflow.constants import (
98 ACTIVITY_RETRY,
99 ACTIVITY_TIMEOUT,
100 ALERT_BUDGET,
101 ALERT_RETRY,
102 ALERT_TIMEOUT,
103 )
104 from ynab_agent.workflow.types import AnswerOutcome, TransactionParams
105 
106# History-length ceiling before a resting workflow continues-as-new. High enough
107# that ordinary flows never trip it; long-lived (30-45 day) transactions do.
108_CONTINUE_AS_NEW_AFTER = 4_000
109 
110# Resting states the workflow may sit in for days while waiting on signals or
111# timers; continue-as-new fires only from one of these. DISCOVERED counts: a
112# transaction born from a signal can wait here for a slow YNAB import (SPEC §3).
113_RESTING = (Discovered, AwaitingHuman, Open, Lapsed, HoldAmazon)
114 
115 
116def _hold_for_amazon(snapshot: YnabSnapshot) -> bool:
117 """Whether to hold for Amazon item detail: Amazon-ish and no memo yet.
118 
119 Shares :func:`~ynab_agent.ingest.plan.is_amazon` with W1's backfill signal,
120 so "what counts as Amazon" has a single source of truth (SPEC §3, §11).
121 """
122 return is_amazon(snapshot.payee) and not snapshot.has_memo
123 
⋯ 426 lines hidden (lines 124–549)
124 
125@workflow.defn
126class TransactionWorkflow:
127 """The durable per-transaction lifecycle."""
128 
129 def __init__(self) -> None:
130 """Initialize empty state; ``run`` populates it from the params."""
131 self._txn: Transaction
132 self._ynab_id: str = ""
133 self._thread_id: str | None = None
134 self._deadlines: dict[TimerKind, datetime.datetime] = {}
135 self._inbound: deque[InboundSignal] = deque()
136 self._snapshot_ready: YnabSnapshot | None = None
137 # Monotonic per-transaction counter: the outbound-send idempotency key
138 # so a replay/retry never double-emails (SPEC §3 outbound dedup).
139 self._action_seq: int = 0
140 
141 # ── Signals (the external world pushes in) ──────────────────────────────
142 @workflow.signal
143 def submit_inbound(self, signal: InboundSignal) -> None:
144 """A reply or matched receipt arrived (W3/W4)."""
145 self._inbound.append(signal)
146 
147 @workflow.signal
148 def notify_snapshot(self, snapshot: YnabSnapshot) -> None:
149 """W1 materialized (or backfilled the memo of) the YNAB snapshot."""
150 self._snapshot_ready = snapshot
151 
152 @workflow.query
153 def state(self) -> str:
154 """The current lifecycle state (for observability)."""
155 return self._txn.state.value
156 
157 # ── The durable loop ────────────────────────────────────────────────────
158 @workflow.run
159 async def run(self, params: TransactionParams) -> None:
160 """Drive the lifecycle until the transaction is archived (SPEC §3)."""
161 self._ynab_id = params.ynab_id
162 self._thread_id = params.thread_id
163 self._deadlines = dict(params.resume_deadlines)
164 self._inbound.extend(params.resume_inbound)
165 self._action_seq = params.resume_action_seq
166 self._txn = params.resume_txn or born(
167 YnabTransactionId(params.ynab_id), params.thread_id
168 )
169 self._sync_thread_id()
170 
171 try:
172 while not isinstance(self._txn, Archived):
173 await self._step()
174 if (
175 isinstance(self._txn, _RESTING)
176 and not self._inbound # drain pending signals (SPEC §0.5)
177 and workflow.info().get_current_history_length()
178 > _CONTINUE_AS_NEW_AFTER
179 ):
180 # continue_as_new raises to restart; nothing runs after it.
181 # (ContinueAsNewError is not an ActivityError, so it escapes
182 # the failure hook below untouched.)
183 workflow.continue_as_new(self._resume_params())
184 except ActivityError as exc:
185 # A terminal activity failure: a non-retryable bug (the constants
186 # denylist) or the schedule_to_close budget elapsing. Page the owner
187 # once — deduped, best-effort — then re-raise so the transaction
188 # still fails and stays visible in Temporal (SPEC §13).
189 await workflow.execute_activity(
190 alert_activities.alert_failure,
191 build_failure_alert(
192 key=self._ynab_id,
193 context=self._alert_context(),
194 exc=exc,
195 ),
196 start_to_close_timeout=ALERT_TIMEOUT,
197 schedule_to_close_timeout=ALERT_BUDGET,
198 retry_policy=ALERT_RETRY,
199 )
200 raise
201 
202 def _alert_context(self) -> str:
203 """A human locator for a failure alert: payee + txn id when known."""
204 st = self._txn
205 if isinstance(st, Discovered):
206 return f"txn {self._ynab_id}"
207 return f"{st.core.snapshot.payee} (txn {self._ynab_id})"
208 
209 def _resume_params(self) -> TransactionParams:
210 return TransactionParams(
211 ynab_id=YnabTransactionId(self._ynab_id),
212 thread_id=ThreadId(self._thread_id)
213 if self._thread_id is not None
214 else None,
215 resume_txn=self._txn,
216 resume_deadlines=dict(self._deadlines),
217 resume_inbound=tuple(self._inbound),
218 resume_action_seq=self._action_seq,
219 )
220 
221 async def _step(self) -> None:
222 st = self._txn
223 if isinstance(st, Discovered):
224 await self._on_discovered()
225 elif isinstance(st, Enriching):
226 await self._on_enriching(st)
227 elif isinstance(st, HoldAmazon):
228 await self._on_hold(st)
229 elif isinstance(st, AwaitingHuman):
230 await self._on_awaiting(st)
231 elif isinstance(st, Open):
232 await self._on_open(st)
233 elif isinstance(st, Lapsed):
234 await self._on_lapsed()
235 elif isinstance(st, Revising):
236 await self._on_revising(st)
237 elif isinstance(st, (AutoApplied, Applied, Archived)):
238 # Transient (handled via the commit→verify follow-up) or terminal.
239 return
240 
241 # ── apply + effect dispatch ─────────────────────────────────────────────
242 async def _dispatch(self, event: LifecycleEvent) -> None:
243 transition = advance(
244 self._txn, event, now=workflow.now(), policy=DEFAULT_POLICY
245 )
246 self._txn = transition.next
247 # Keep the thread-id mirror in step with the state machine, which can
248 # adopt a reply's thread on its own (e.g. a reply in DISCOVERED).
249 self._sync_thread_id()
250 for effect in transition.effects:
251 followup = await self._execute(effect)
252 if followup is not None:
253 await self._dispatch(followup)
254 
255 async def _execute(self, effect: Effect) -> LifecycleEvent | None:
256 if isinstance(effect, CommitToYnab):
257 await workflow.execute_activity(
258 activities.commit_to_ynab,
259 args=[self._ynab_id, effect.decision],
260 start_to_close_timeout=ACTIVITY_TIMEOUT,
261 retry_policy=ACTIVITY_RETRY,
262 )
263 read = await workflow.execute_activity(
264 activities.read_back,
265 self._ynab_id,
266 start_to_close_timeout=ACTIVITY_TIMEOUT,
267 retry_policy=ACTIVITY_RETRY,
268 )
269 return WriteVerified(
270 outcome=classify_verify(read, target_of(effect.decision))
271 )
272 if isinstance(effect, OpenThread):
273 tid = await workflow.execute_activity(
274 activities.open_thread,
275 args=[self._ynab_id, self._proposal()],
276 start_to_close_timeout=ACTIVITY_TIMEOUT,
277 retry_policy=ACTIVITY_RETRY,
278 )
279 self._set_thread_id(tid)
280 # Index this workflow by its thread for reply routing (§5a).
281 workflow.upsert_search_attributes([_TXN_THREAD_ID.value_set(tid)])
282 elif isinstance(effect, SendThreadMessage):
283 self._action_seq += 1
284 await workflow.execute_activity(
285 activities.send_thread_message,
286 args=[
287 self._ynab_id,
288 self._thread_id,
289 effect.purpose,
290 self._action_seq,
291 self._proposal(),
292 ],
293 start_to_close_timeout=ACTIVITY_TIMEOUT,
294 retry_policy=ACTIVITY_RETRY,
295 )
296 elif isinstance(effect, FeedRuleLearning):
297 await workflow.execute_activity(
298 activities.feed_rule_learning,
299 effect,
300 start_to_close_timeout=ACTIVITY_TIMEOUT,
301 retry_policy=ACTIVITY_RETRY,
302 )
303 elif isinstance(effect, CloseThread):
304 if self._thread_id is not None:
305 await workflow.execute_activity(
306 activities.close_thread,
307 self._thread_id,
308 start_to_close_timeout=ACTIVITY_TIMEOUT,
309 retry_policy=ACTIVITY_RETRY,
310 )
311 elif isinstance(effect, SetTimer):
312 self._deadlines[effect.timer] = effect.deadline
313 elif isinstance(effect, CancelTimer):
314 self._deadlines.pop(effect.timer, None)
315 elif isinstance(effect, ReplayBuffered):
316 self._inbound.extendleft(reversed(effect.signals))
317 return None
318 
319 def _set_thread_id(self, tid: str) -> None:
320 self._thread_id = tid
321 st = self._txn
322 if not isinstance(st, Discovered):
323 new_core = st.core.model_copy(update={"thread_id": ThreadId(tid)})
324 self._txn = st.model_copy(update={"core": new_core})
325 
326 def _sync_thread_id(self) -> None:
327 """Mirror the current transaction's thread id (SM may adopt it)."""
328 st = self._txn
329 tid = st.thread_id if isinstance(st, Discovered) else st.core.thread_id
330 if tid is not None:
331 self._thread_id = str(tid)
332 
333 def _proposal(self) -> Proposal | None:
334 """The current best-guess proposal, for states that carry one.
335 
336 Passed to the mail activities so the proposal email can name the guess +
337 alternatives; ``None`` for purposes whose content derives from YNAB.
338 """
339 st = self._txn
340 if isinstance(st, (Enriching, AwaitingHuman, Lapsed)):
341 return st.proposal
342 return None
343 
344 # ── per-state steps ─────────────────────────────────────────────────────
345 async def _on_discovered(self) -> None:
346 snapshot = await workflow.execute_activity(
347 activities.fetch_snapshot,
348 self._ynab_id,
349 start_to_close_timeout=ACTIVITY_TIMEOUT,
350 retry_policy=ACTIVITY_RETRY,
351 )
352 if snapshot is not None:
353 await self._dispatch(
354 SnapshotMaterialized(
355 snapshot=snapshot,
356 hold_for_amazon=_hold_for_amazon(snapshot),
357 )
358 )
359 return
360 # Signal beat the poll: wait for materialization or buffer an inbound.
361 await self._dispatch(SnapshotUnavailable())
362 await workflow.wait_condition(
363 lambda: self._snapshot_ready is not None or len(self._inbound) > 0
364 )
365 if self._snapshot_ready is not None:
366 snap = self._snapshot_ready
367 self._snapshot_ready = None
368 await self._dispatch(
369 SnapshotMaterialized(
370 snapshot=snap, hold_for_amazon=_hold_for_amazon(snap)
371 )
372 )
373 else:
374 await self._dispatch(
375 InboundReceived(signal=self._inbound.popleft())
376 )
377 
378 async def _on_enriching(self, st: Enriching) -> None:
379 outcome = await workflow.execute_activity(
380 activities.enrich,
381 st.core.snapshot,
382 start_to_close_timeout=ACTIVITY_TIMEOUT,
383 retry_policy=ACTIVITY_RETRY,
384 )
385 await self._dispatch(Enriched(outcome=outcome))
386 
387 async def _on_hold(self, st: HoldAmazon) -> None:
388 ready = await self._wait_until(
389 st.amazon_deadline,
390 lambda: self._snapshot_ready is not None or len(self._inbound) > 0,
391 )
392 if not ready:
393 await self._dispatch(HoldDeadlineReached())
394 elif self._snapshot_ready is not None:
395 snap = self._snapshot_ready
396 self._snapshot_ready = None
397 await self._dispatch(HoldResolved(snapshot=snap))
398 else:
399 await self._dispatch(
400 InboundReceived(signal=self._inbound.popleft())
401 )
402 
403 async def _on_awaiting(self, st: AwaitingHuman) -> None:
404 got_inbound = await self._wait_until(
405 st.patience_deadline, self._has_inbound
406 )
407 if not got_inbound:
408 before = type(self._txn)
409 await self._dispatch(PatienceExpired())
410 if type(self._txn) is not before:
411 return # lapsed (or otherwise transitioned)
412 # Flagged (verify-failure) entry: PatienceExpired is ignored and
413 # does not lapse (SPEC §3). Drop the passed timer and wait for an
414 # inbound instead of re-spinning the expired deadline.
415 self._deadlines.pop(TimerKind.PATIENCE, None)
416 await workflow.wait_condition(self._has_inbound)
417 await self._interpret_inbound(st.core.snapshot)
418 
419 async def _interpret_inbound(self, snapshot: YnabSnapshot) -> None:
420 signal = self._inbound.popleft()
421 interpretation = await workflow.execute_activity(
422 activities.interpret_inbound,
423 args=[signal, snapshot, self._proposal()],
424 start_to_close_timeout=ACTIVITY_TIMEOUT,
425 retry_policy=ACTIVITY_RETRY,
426 )
427 if isinstance(interpretation, AnswerOutcome):
428 await self._dispatch(
429 AnswerReceived(decision=interpretation.decision)
430 )
431 else:
432 await self._dispatch(
433 ClarifyRequested(question=interpretation.question)
434 )
435 
436 async def _on_open(self, st: Open) -> None:
437 deadline = self._deadlines.get(TimerKind.ARCHIVE)
438 got_inbound = (
439 await self._wait_until(deadline, self._has_inbound)
440 if deadline is not None
441 else await self._wait_forever(self._has_inbound)
442 )
443 if got_inbound:
444 await self._dispatch(
445 InboundReceived(signal=self._inbound.popleft())
446 )
447 return
448 # The archive window elapsed. Before closing the book, re-read YNAB to
449 # catch a silent manual recategorization — an out-of-band correction
450 # that must demote the driving rule (SPEC §14.2).
451 event = await self._archive_or_override(st)
452 before = type(self._txn)
453 await self._dispatch(event)
454 if type(self._txn) is before:
455 # Archive blocked (not reconciled) and no override: drop the stale
456 # timer and wait for an inbound rather than busy-looping (SPEC §3).
457 self._deadlines.pop(TimerKind.ARCHIVE, None)
458 await workflow.wait_condition(self._has_inbound)
459 await self._dispatch(
460 InboundReceived(signal=self._inbound.popleft())
461 )
462 
463 async def _archive_or_override(self, st: Open) -> LifecycleEvent:
464 """At archive time, detect a manual YNAB edit (SPEC §14.2).
465 
466 Re-reads the current end-state and, if its category no longer matches
467 the agent's applied decision, returns an ``OverrideDetected`` carrying
468 the human's choice (the spine then demotes the rule); otherwise the
469 ordinary ``ArchiveWindowReached``. A memo-only change is not an override
470 — only the allocation is compared.
471 """
472 read = await workflow.execute_activity(
473 activities.read_back,
474 self._ynab_id,
475 start_to_close_timeout=ACTIVITY_TIMEOUT,
476 retry_policy=ACTIVITY_RETRY,
477 )
478 if read is not None and read.allocation != st.decision.allocation:
479 human = Decision(
480 allocation=read.allocation,
481 memo=read.memo,
482 approved=read.approved,
483 decided_by=DecidedBy.HUMAN,
484 decided_at=workflow.now(),
485 )
486 return OverrideDetected(decision=human)
487 return ArchiveWindowReached()
488 
489 async def _on_lapsed(self) -> None:
490 await self._wait_then_revise_or_archive()
491 
492 def _has_inbound(self) -> bool:
493 return len(self._inbound) > 0
494 
495 async def _wait_then_revise_or_archive(self) -> None:
496 deadline = self._deadlines.get(TimerKind.ARCHIVE)
497 got_inbound = (
498 await self._wait_until(deadline, self._has_inbound)
499 if deadline is not None
500 else await self._wait_forever(self._has_inbound)
501 )
502 if got_inbound:
503 await self._dispatch(
504 InboundReceived(signal=self._inbound.popleft())
505 )
506 return
507 # The archive window elapsed. Attempt to archive; if it is blocked (not
508 # reconciled / not categorized) the state is unchanged, so drop the
509 # now-stale timer and wait for an inbound rather than busy-looping on
510 # the already-passed deadline.
511 before = type(self._txn)
512 await self._dispatch(ArchiveWindowReached())
513 if type(self._txn) is before:
514 self._deadlines.pop(TimerKind.ARCHIVE, None)
515 await workflow.wait_condition(self._has_inbound)
516 await self._dispatch(
517 InboundReceived(signal=self._inbound.popleft())
518 )
519 
520 async def _wait_forever(self, predicate: Callable[[], bool]) -> bool:
521 await workflow.wait_condition(predicate)
522 return True
523 
524 async def _on_revising(self, st: Revising) -> None:
525 outcome = await workflow.execute_activity(
526 activities.converge,
527 args=[st.core.snapshot, st.instruction],
528 start_to_close_timeout=ACTIVITY_TIMEOUT,
529 retry_policy=ACTIVITY_RETRY,
530 )
531 await self._dispatch(Converged(outcome=outcome))
532 
533 async def _wait_until(
534 self,
535 deadline: datetime.datetime,
536 predicate: Callable[[], bool],
537 ) -> bool:
538 """Wait for ``predicate`` until an absolute deadline.
539 
540 Returns ``True`` if the predicate fired first, ``False`` on timeout.
541 """
542 timeout = deadline - workflow.now()
543 if timeout < timedelta(0):
544 timeout = timedelta(0)
545 try:
546 await workflow.wait_condition(predicate, timeout=timeout)
547 except TimeoutError:
548 return False
549 return True

The receiving side was already there. _on_hold waits on _snapshot_ready; the notify_snapshot signal sets it, and the hold resolves to ENRICHING carrying the backfilled snapshot — exactly the early exit the SPEC describes.

Pre-existing W2 mechanism: a delivered snapshot resolves the hold to its memo, short of the deadline.

src/ynab_agent/workflow/txn_workflow.py · 549 lines
src/ynab_agent/workflow/txn_workflow.py549 lines · Python
⋯ 386 lines hidden (lines 1–386)
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
4: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:
33 from collections.abc import Callable
34 
35 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():
44 from ynab_agent.domain.config import DEFAULT_POLICY
45 from ynab_agent.domain.effects import (
46 CancelTimer,
47 CloseThread,
48 CommitToYnab,
49 Effect,
50 FeedRuleLearning,
51 OpenThread,
52 ReplayBuffered,
53 SendThreadMessage,
54 SetTimer,
55 TimerKind,
56 )
57 from ynab_agent.domain.enums import DecidedBy
58 from ynab_agent.domain.events import (
59 AnswerReceived,
60 ArchiveWindowReached,
61 ClarifyRequested,
62 Converged,
63 Enriched,
64 HoldDeadlineReached,
65 HoldResolved,
66 InboundReceived,
67 LifecycleEvent,
68 OverrideDetected,
69 PatienceExpired,
70 SnapshotMaterialized,
71 SnapshotUnavailable,
72 WriteVerified,
73 )
74 from ynab_agent.domain.ids import ThreadId, YnabTransactionId
75 from ynab_agent.domain.proposal import Decision
76 from ynab_agent.domain.signals import InboundSignal
77 from ynab_agent.domain.state_machine import advance
78 from ynab_agent.domain.transaction import (
79 Applied,
80 Archived,
81 AutoApplied,
82 AwaitingHuman,
83 Discovered,
84 Enriching,
85 HoldAmazon,
86 Lapsed,
87 Open,
88 Revising,
89 Transaction,
90 YnabSnapshot,
91 born,
92 )
93 from ynab_agent.ingest.plan import is_amazon
94 from ynab_agent.policy.converge import classify_verify, target_of
95 from ynab_agent.workflow import activities, alert_activities
96 from ynab_agent.workflow.alerting import build_failure_alert
97 from ynab_agent.workflow.constants import (
98 ACTIVITY_RETRY,
99 ACTIVITY_TIMEOUT,
100 ALERT_BUDGET,
101 ALERT_RETRY,
102 ALERT_TIMEOUT,
103 )
104 from ynab_agent.workflow.types import AnswerOutcome, TransactionParams
105 
106# History-length ceiling before a resting workflow continues-as-new. High enough
107# that ordinary flows never trip it; long-lived (30-45 day) transactions do.
108_CONTINUE_AS_NEW_AFTER = 4_000
109 
110# Resting states the workflow may sit in for days while waiting on signals or
111# timers; continue-as-new fires only from one of these. DISCOVERED counts: a
112# transaction born from a signal can wait here for a slow YNAB import (SPEC §3).
113_RESTING = (Discovered, AwaitingHuman, Open, Lapsed, HoldAmazon)
114 
115 
116def _hold_for_amazon(snapshot: YnabSnapshot) -> bool:
117 """Whether to hold for Amazon item detail: Amazon-ish and no memo yet.
118 
119 Shares :func:`~ynab_agent.ingest.plan.is_amazon` with W1's backfill signal,
120 so "what counts as Amazon" has a single source of truth (SPEC §3, §11).
121 """
122 return is_amazon(snapshot.payee) and not snapshot.has_memo
123 
124 
125@workflow.defn
126class TransactionWorkflow:
127 """The durable per-transaction lifecycle."""
128 
129 def __init__(self) -> None:
130 """Initialize empty state; ``run`` populates it from the params."""
131 self._txn: Transaction
132 self._ynab_id: str = ""
133 self._thread_id: str | None = None
134 self._deadlines: dict[TimerKind, datetime.datetime] = {}
135 self._inbound: deque[InboundSignal] = deque()
136 self._snapshot_ready: YnabSnapshot | None = None
137 # Monotonic per-transaction counter: the outbound-send idempotency key
138 # so a replay/retry never double-emails (SPEC §3 outbound dedup).
139 self._action_seq: int = 0
140 
141 # ── Signals (the external world pushes in) ──────────────────────────────
142 @workflow.signal
143 def submit_inbound(self, signal: InboundSignal) -> None:
144 """A reply or matched receipt arrived (W3/W4)."""
145 self._inbound.append(signal)
146 
147 @workflow.signal
148 def notify_snapshot(self, snapshot: YnabSnapshot) -> None:
149 """W1 materialized (or backfilled the memo of) the YNAB snapshot."""
150 self._snapshot_ready = snapshot
151 
152 @workflow.query
153 def state(self) -> str:
154 """The current lifecycle state (for observability)."""
155 return self._txn.state.value
156 
157 # ── The durable loop ────────────────────────────────────────────────────
158 @workflow.run
159 async def run(self, params: TransactionParams) -> None:
160 """Drive the lifecycle until the transaction is archived (SPEC §3)."""
161 self._ynab_id = params.ynab_id
162 self._thread_id = params.thread_id
163 self._deadlines = dict(params.resume_deadlines)
164 self._inbound.extend(params.resume_inbound)
165 self._action_seq = params.resume_action_seq
166 self._txn = params.resume_txn or born(
167 YnabTransactionId(params.ynab_id), params.thread_id
168 )
169 self._sync_thread_id()
170 
171 try:
172 while not isinstance(self._txn, Archived):
173 await self._step()
174 if (
175 isinstance(self._txn, _RESTING)
176 and not self._inbound # drain pending signals (SPEC §0.5)
177 and workflow.info().get_current_history_length()
178 > _CONTINUE_AS_NEW_AFTER
179 ):
180 # continue_as_new raises to restart; nothing runs after it.
181 # (ContinueAsNewError is not an ActivityError, so it escapes
182 # the failure hook below untouched.)
183 workflow.continue_as_new(self._resume_params())
184 except ActivityError as exc:
185 # A terminal activity failure: a non-retryable bug (the constants
186 # denylist) or the schedule_to_close budget elapsing. Page the owner
187 # once — deduped, best-effort — then re-raise so the transaction
188 # still fails and stays visible in Temporal (SPEC §13).
189 await workflow.execute_activity(
190 alert_activities.alert_failure,
191 build_failure_alert(
192 key=self._ynab_id,
193 context=self._alert_context(),
194 exc=exc,
195 ),
196 start_to_close_timeout=ALERT_TIMEOUT,
197 schedule_to_close_timeout=ALERT_BUDGET,
198 retry_policy=ALERT_RETRY,
199 )
200 raise
201 
202 def _alert_context(self) -> str:
203 """A human locator for a failure alert: payee + txn id when known."""
204 st = self._txn
205 if isinstance(st, Discovered):
206 return f"txn {self._ynab_id}"
207 return f"{st.core.snapshot.payee} (txn {self._ynab_id})"
208 
209 def _resume_params(self) -> TransactionParams:
210 return TransactionParams(
211 ynab_id=YnabTransactionId(self._ynab_id),
212 thread_id=ThreadId(self._thread_id)
213 if self._thread_id is not None
214 else None,
215 resume_txn=self._txn,
216 resume_deadlines=dict(self._deadlines),
217 resume_inbound=tuple(self._inbound),
218 resume_action_seq=self._action_seq,
219 )
220 
221 async def _step(self) -> None:
222 st = self._txn
223 if isinstance(st, Discovered):
224 await self._on_discovered()
225 elif isinstance(st, Enriching):
226 await self._on_enriching(st)
227 elif isinstance(st, HoldAmazon):
228 await self._on_hold(st)
229 elif isinstance(st, AwaitingHuman):
230 await self._on_awaiting(st)
231 elif isinstance(st, Open):
232 await self._on_open(st)
233 elif isinstance(st, Lapsed):
234 await self._on_lapsed()
235 elif isinstance(st, Revising):
236 await self._on_revising(st)
237 elif isinstance(st, (AutoApplied, Applied, Archived)):
238 # Transient (handled via the commit→verify follow-up) or terminal.
239 return
240 
241 # ── apply + effect dispatch ─────────────────────────────────────────────
242 async def _dispatch(self, event: LifecycleEvent) -> None:
243 transition = advance(
244 self._txn, event, now=workflow.now(), policy=DEFAULT_POLICY
245 )
246 self._txn = transition.next
247 # Keep the thread-id mirror in step with the state machine, which can
248 # adopt a reply's thread on its own (e.g. a reply in DISCOVERED).
249 self._sync_thread_id()
250 for effect in transition.effects:
251 followup = await self._execute(effect)
252 if followup is not None:
253 await self._dispatch(followup)
254 
255 async def _execute(self, effect: Effect) -> LifecycleEvent | None:
256 if isinstance(effect, CommitToYnab):
257 await workflow.execute_activity(
258 activities.commit_to_ynab,
259 args=[self._ynab_id, effect.decision],
260 start_to_close_timeout=ACTIVITY_TIMEOUT,
261 retry_policy=ACTIVITY_RETRY,
262 )
263 read = await workflow.execute_activity(
264 activities.read_back,
265 self._ynab_id,
266 start_to_close_timeout=ACTIVITY_TIMEOUT,
267 retry_policy=ACTIVITY_RETRY,
268 )
269 return WriteVerified(
270 outcome=classify_verify(read, target_of(effect.decision))
271 )
272 if isinstance(effect, OpenThread):
273 tid = await workflow.execute_activity(
274 activities.open_thread,
275 args=[self._ynab_id, self._proposal()],
276 start_to_close_timeout=ACTIVITY_TIMEOUT,
277 retry_policy=ACTIVITY_RETRY,
278 )
279 self._set_thread_id(tid)
280 # Index this workflow by its thread for reply routing (§5a).
281 workflow.upsert_search_attributes([_TXN_THREAD_ID.value_set(tid)])
282 elif isinstance(effect, SendThreadMessage):
283 self._action_seq += 1
284 await workflow.execute_activity(
285 activities.send_thread_message,
286 args=[
287 self._ynab_id,
288 self._thread_id,
289 effect.purpose,
290 self._action_seq,
291 self._proposal(),
292 ],
293 start_to_close_timeout=ACTIVITY_TIMEOUT,
294 retry_policy=ACTIVITY_RETRY,
295 )
296 elif isinstance(effect, FeedRuleLearning):
297 await workflow.execute_activity(
298 activities.feed_rule_learning,
299 effect,
300 start_to_close_timeout=ACTIVITY_TIMEOUT,
301 retry_policy=ACTIVITY_RETRY,
302 )
303 elif isinstance(effect, CloseThread):
304 if self._thread_id is not None:
305 await workflow.execute_activity(
306 activities.close_thread,
307 self._thread_id,
308 start_to_close_timeout=ACTIVITY_TIMEOUT,
309 retry_policy=ACTIVITY_RETRY,
310 )
311 elif isinstance(effect, SetTimer):
312 self._deadlines[effect.timer] = effect.deadline
313 elif isinstance(effect, CancelTimer):
314 self._deadlines.pop(effect.timer, None)
315 elif isinstance(effect, ReplayBuffered):
316 self._inbound.extendleft(reversed(effect.signals))
317 return None
318 
319 def _set_thread_id(self, tid: str) -> None:
320 self._thread_id = tid
321 st = self._txn
322 if not isinstance(st, Discovered):
323 new_core = st.core.model_copy(update={"thread_id": ThreadId(tid)})
324 self._txn = st.model_copy(update={"core": new_core})
325 
326 def _sync_thread_id(self) -> None:
327 """Mirror the current transaction's thread id (SM may adopt it)."""
328 st = self._txn
329 tid = st.thread_id if isinstance(st, Discovered) else st.core.thread_id
330 if tid is not None:
331 self._thread_id = str(tid)
332 
333 def _proposal(self) -> Proposal | None:
334 """The current best-guess proposal, for states that carry one.
335 
336 Passed to the mail activities so the proposal email can name the guess +
337 alternatives; ``None`` for purposes whose content derives from YNAB.
338 """
339 st = self._txn
340 if isinstance(st, (Enriching, AwaitingHuman, Lapsed)):
341 return st.proposal
342 return None
343 
344 # ── per-state steps ─────────────────────────────────────────────────────
345 async def _on_discovered(self) -> None:
346 snapshot = await workflow.execute_activity(
347 activities.fetch_snapshot,
348 self._ynab_id,
349 start_to_close_timeout=ACTIVITY_TIMEOUT,
350 retry_policy=ACTIVITY_RETRY,
351 )
352 if snapshot is not None:
353 await self._dispatch(
354 SnapshotMaterialized(
355 snapshot=snapshot,
356 hold_for_amazon=_hold_for_amazon(snapshot),
357 )
358 )
359 return
360 # Signal beat the poll: wait for materialization or buffer an inbound.
361 await self._dispatch(SnapshotUnavailable())
362 await workflow.wait_condition(
363 lambda: self._snapshot_ready is not None or len(self._inbound) > 0
364 )
365 if self._snapshot_ready is not None:
366 snap = self._snapshot_ready
367 self._snapshot_ready = None
368 await self._dispatch(
369 SnapshotMaterialized(
370 snapshot=snap, hold_for_amazon=_hold_for_amazon(snap)
371 )
372 )
373 else:
374 await self._dispatch(
375 InboundReceived(signal=self._inbound.popleft())
376 )
377 
378 async def _on_enriching(self, st: Enriching) -> None:
379 outcome = await workflow.execute_activity(
380 activities.enrich,
381 st.core.snapshot,
382 start_to_close_timeout=ACTIVITY_TIMEOUT,
383 retry_policy=ACTIVITY_RETRY,
384 )
385 await self._dispatch(Enriched(outcome=outcome))
386 
387 async def _on_hold(self, st: HoldAmazon) -> None:
388 ready = await self._wait_until(
389 st.amazon_deadline,
390 lambda: self._snapshot_ready is not None or len(self._inbound) > 0,
391 )
392 if not ready:
393 await self._dispatch(HoldDeadlineReached())
394 elif self._snapshot_ready is not None:
395 snap = self._snapshot_ready
396 self._snapshot_ready = None
397 await self._dispatch(HoldResolved(snapshot=snap))
398 else:
399 await self._dispatch(
400 InboundReceived(signal=self._inbound.popleft())
⋯ 149 lines hidden (lines 401–549)
401 )
402 
403 async def _on_awaiting(self, st: AwaitingHuman) -> None:
404 got_inbound = await self._wait_until(
405 st.patience_deadline, self._has_inbound
406 )
407 if not got_inbound:
408 before = type(self._txn)
409 await self._dispatch(PatienceExpired())
410 if type(self._txn) is not before:
411 return # lapsed (or otherwise transitioned)
412 # Flagged (verify-failure) entry: PatienceExpired is ignored and
413 # does not lapse (SPEC §3). Drop the passed timer and wait for an
414 # inbound instead of re-spinning the expired deadline.
415 self._deadlines.pop(TimerKind.PATIENCE, None)
416 await workflow.wait_condition(self._has_inbound)
417 await self._interpret_inbound(st.core.snapshot)
418 
419 async def _interpret_inbound(self, snapshot: YnabSnapshot) -> None:
420 signal = self._inbound.popleft()
421 interpretation = await workflow.execute_activity(
422 activities.interpret_inbound,
423 args=[signal, snapshot, self._proposal()],
424 start_to_close_timeout=ACTIVITY_TIMEOUT,
425 retry_policy=ACTIVITY_RETRY,
426 )
427 if isinstance(interpretation, AnswerOutcome):
428 await self._dispatch(
429 AnswerReceived(decision=interpretation.decision)
430 )
431 else:
432 await self._dispatch(
433 ClarifyRequested(question=interpretation.question)
434 )
435 
436 async def _on_open(self, st: Open) -> None:
437 deadline = self._deadlines.get(TimerKind.ARCHIVE)
438 got_inbound = (
439 await self._wait_until(deadline, self._has_inbound)
440 if deadline is not None
441 else await self._wait_forever(self._has_inbound)
442 )
443 if got_inbound:
444 await self._dispatch(
445 InboundReceived(signal=self._inbound.popleft())
446 )
447 return
448 # The archive window elapsed. Before closing the book, re-read YNAB to
449 # catch a silent manual recategorization — an out-of-band correction
450 # that must demote the driving rule (SPEC §14.2).
451 event = await self._archive_or_override(st)
452 before = type(self._txn)
453 await self._dispatch(event)
454 if type(self._txn) is before:
455 # Archive blocked (not reconciled) and no override: drop the stale
456 # timer and wait for an inbound rather than busy-looping (SPEC §3).
457 self._deadlines.pop(TimerKind.ARCHIVE, None)
458 await workflow.wait_condition(self._has_inbound)
459 await self._dispatch(
460 InboundReceived(signal=self._inbound.popleft())
461 )
462 
463 async def _archive_or_override(self, st: Open) -> LifecycleEvent:
464 """At archive time, detect a manual YNAB edit (SPEC §14.2).
465 
466 Re-reads the current end-state and, if its category no longer matches
467 the agent's applied decision, returns an ``OverrideDetected`` carrying
468 the human's choice (the spine then demotes the rule); otherwise the
469 ordinary ``ArchiveWindowReached``. A memo-only change is not an override
470 — only the allocation is compared.
471 """
472 read = await workflow.execute_activity(
473 activities.read_back,
474 self._ynab_id,
475 start_to_close_timeout=ACTIVITY_TIMEOUT,
476 retry_policy=ACTIVITY_RETRY,
477 )
478 if read is not None and read.allocation != st.decision.allocation:
479 human = Decision(
480 allocation=read.allocation,
481 memo=read.memo,
482 approved=read.approved,
483 decided_by=DecidedBy.HUMAN,
484 decided_at=workflow.now(),
485 )
486 return OverrideDetected(decision=human)
487 return ArchiveWindowReached()
488 
489 async def _on_lapsed(self) -> None:
490 await self._wait_then_revise_or_archive()
491 
492 def _has_inbound(self) -> bool:
493 return len(self._inbound) > 0
494 
495 async def _wait_then_revise_or_archive(self) -> None:
496 deadline = self._deadlines.get(TimerKind.ARCHIVE)
497 got_inbound = (
498 await self._wait_until(deadline, self._has_inbound)
499 if deadline is not None
500 else await self._wait_forever(self._has_inbound)
501 )
502 if got_inbound:
503 await self._dispatch(
504 InboundReceived(signal=self._inbound.popleft())
505 )
506 return
507 # The archive window elapsed. Attempt to archive; if it is blocked (not
508 # reconciled / not categorized) the state is unchanged, so drop the
509 # now-stale timer and wait for an inbound rather than busy-looping on
510 # the already-passed deadline.
511 before = type(self._txn)
512 await self._dispatch(ArchiveWindowReached())
513 if type(self._txn) is before:
514 self._deadlines.pop(TimerKind.ARCHIVE, None)
515 await workflow.wait_condition(self._has_inbound)
516 await self._dispatch(
517 InboundReceived(signal=self._inbound.popleft())
518 )
519 
520 async def _wait_forever(self, predicate: Callable[[], bool]) -> bool:
521 await workflow.wait_condition(predicate)
522 return True
523 
524 async def _on_revising(self, st: Revising) -> None:
525 outcome = await workflow.execute_activity(
526 activities.converge,
527 args=[st.core.snapshot, st.instruction],
528 start_to_close_timeout=ACTIVITY_TIMEOUT,
529 retry_policy=ACTIVITY_RETRY,
530 )
531 await self._dispatch(Converged(outcome=outcome))
532 
533 async def _wait_until(
534 self,
535 deadline: datetime.datetime,
536 predicate: Callable[[], bool],
537 ) -> bool:
538 """Wait for ``predicate`` until an absolute deadline.
539 
540 Returns ``True`` if the predicate fired first, ``False`` on timeout.
541 """
542 timeout = deadline - workflow.now()
543 if timeout < timedelta(0):
544 timeout = timedelta(0)
545 try:
546 await workflow.wait_condition(predicate, timeout=timeout)
547 except TimeoutError:
548 return False
549 return True

Tests

Each layer is pinned. The pure planner sets notify_existing only for Amazon + memo. The W1 activity (with a fake client) signals notify_snapshot on the already-running path when flagged, and leaves a running W2 untouched otherwise.

memo_backfilled only for Amazon-with-memo; the plan flags it.

tests/ingest/test_plan.py · 135 lines
tests/ingest/test_plan.py135 lines · Python
⋯ 107 lines hidden (lines 1–107)
1"""Tests for the W1 ingestion planning core (SPEC §2, §13)."""
2 
3from __future__ import annotations
4 
5import datetime
6 
7from ynab_agent.domain.ids import AccountId, YnabTransactionId
8from ynab_agent.domain.money import Money
9from ynab_agent.domain.transaction import YnabSnapshot
10from ynab_agent.ingest.plan import (
11 is_amazon,
12 is_duplicate_import,
13 memo_backfilled,
14 plan_ingest,
16from ynab_agent.ingest.scope import IngestScope, in_scope
17 
18_INSTALL = datetime.date(2026, 5, 1)
19 
20 
21def _snapshot(
22 *,
23 ynab_id: str = "t1",
24 account: str = "a1",
25 day: int = 15,
26 matched: str | None = None,
27) -> YnabSnapshot:
28 return YnabSnapshot(
29 ynab_id=YnabTransactionId(ynab_id),
30 account=AccountId(account),
31 payee="Blue Bottle",
32 amount=Money.from_currency("-4.50"),
33 txn_date=datetime.date(2026, 5, day),
34 matched_transaction_id=(
35 YnabTransactionId(matched) if matched is not None else None
36 ),
37 )
38 
39 
40def _scope(*, accounts: set[str] | None = None) -> IngestScope:
41 return IngestScope(
42 budget_id="b1",
43 install_date=_INSTALL,
44 account_ids=(
45 frozenset(AccountId(a) for a in accounts)
46 if accounts is not None
47 else None
48 ),
49 )
50 
51 
52def test_in_scope_excludes_pre_install_dates() -> None:
53 pre = _snapshot(day=15).model_copy(
54 update={"txn_date": datetime.date(2026, 4, 30)}
55 )
56 assert not in_scope(pre, _scope())
57 assert in_scope(_snapshot(day=15), _scope())
58 
59 
60def test_in_scope_respects_account_subset() -> None:
61 scope = _scope(accounts={"a1"})
62 assert in_scope(_snapshot(account="a1"), scope)
63 assert not in_scope(_snapshot(account="a2"), scope)
64 
65 
66def test_in_scope_all_accounts_when_none() -> None:
67 assert in_scope(_snapshot(account="anything"), _scope())
68 
69 
70def test_is_duplicate_import() -> None:
71 assert is_duplicate_import(_snapshot(matched="t9"))
72 assert not is_duplicate_import(_snapshot())
73 
74 
75def test_cold_start_addresses_nothing() -> None:
76 plan = plan_ingest([_snapshot()], _scope(), cold_start=True)
77 assert plan == ()
78 
79 
80def test_plan_addresses_in_scope_only() -> None:
81 snaps = [
82 _snapshot(ynab_id="t1", account="a1"),
83 _snapshot(ynab_id="t2", account="a2"), # out of account scope
84 ]
85 plan = plan_ingest(snaps, _scope(accounts={"a1"}), cold_start=False)
86 assert [a.snapshot.ynab_id for a in plan] == ["t1"]
87 
88 
89def test_plan_routes_duplicate_imports_to_human() -> None:
90 plan = plan_ingest([_snapshot(matched="t9")], _scope(), cold_start=False)
91 assert len(plan) == 1
92 assert plan[0].route_to_human
93 
94 
95def test_plan_skips_already_approved_transactions() -> None:
96 # An approved transaction is settled — never re-triaged (SPEC §2/§13). Only
97 # the outstanding, unapproved one is addressed.
98 approved = _snapshot(ynab_id="t1").model_copy(update={"approved": True})
99 unapproved = _snapshot(ynab_id="t2")
100 plan = plan_ingest([approved, unapproved], _scope(), cold_start=False)
101 assert [a.snapshot.ynab_id for a in plan] == ["t2"]
102 
103 
104# ── Amazon memo backfill (SPEC §2, §3) ──────────────────────────────────────
105def _amazon(ynab_id: str = "t1", memo: str | None = None) -> YnabSnapshot:
106 return _snapshot(ynab_id=ynab_id).model_copy(
107 update={"payee": "Amazon", "memo": memo}
108 )
109 
110 
111def test_is_amazon_matches_loosely() -> None:
112 assert is_amazon("Amazon")
113 assert is_amazon("Amazon.com")
114 assert is_amazon("AMAZON MKTPLACE")
115 assert not is_amazon("Blue Bottle")
116 
117 
118def test_memo_backfilled_only_for_amazon_with_a_memo() -> None:
119 assert memo_backfilled(_amazon(memo="AmazonBasics HDMI cable"))
120 assert not memo_backfilled(_amazon(memo=None)) # held, no detail yet
121 assert not memo_backfilled(_snapshot()) # not Amazon
122 # A non-Amazon txn that happens to carry a memo is not a backfill signal.
123 assert not memo_backfilled(
124 _snapshot().model_copy(update={"memo": "team lunch"})
125 )
126 
127 
128def test_plan_flags_amazon_backfill_for_notify() -> None:
129 plan = plan_ingest(
130 [_amazon("t1", memo="HDMI cable"), _amazon("t2", memo=None)],
131 _scope(),
132 cold_start=False,
133 )
134 flags = {str(a.snapshot.ynab_id): a.notify_existing for a in plan}
135 assert flags == {"t1": True, "t2": False}

The end-to-end test proves the resolution path is distinguishable from the deadline fallback: it asserts enrichment ran on the snapshot with the memo. The deadline path would have enriched the blank held snapshot, so the memo assertion can only pass if the signal resolved the hold.

A HOLD_AMAZON run resolves on the backfill signal; enrichment sees the item detail.

tests/workflow/test_txn_workflow.py · 495 lines
tests/workflow/test_txn_workflow.py495 lines · Python
⋯ 242 lines hidden (lines 1–242)
1"""End-to-end tests for the W2 workflow on Temporal's time-skipping server.
2 
3These run the real durable workflow (signals, absolute-deadline timers, the
4commit→verify follow-up, continue-as-new guard) with mock activity
5implementations standing in for the stubbed I/O ports.
6"""
7 
8from __future__ import annotations
9 
10import asyncio
11import datetime
12from typing import TYPE_CHECKING
13 
14from temporalio import activity
15from temporalio.api.enums.v1 import IndexedValueType
16from temporalio.api.operatorservice.v1 import AddSearchAttributesRequest
17from temporalio.testing import WorkflowEnvironment
18from temporalio.worker import Worker
19 
20from ynab_agent.domain.allocations import ProposedCategory, ResolvedCategory
21from ynab_agent.domain.effects import FeedRuleLearning, RuleLearningKind
22from ynab_agent.domain.enums import ClearedState, DecidedBy, TrustState
23from ynab_agent.domain.events import (
24 AskHuman,
25 AutoApply,
26 ConvergeOutcome,
27 EnrichmentOutcome,
28 Reapplied,
30from ynab_agent.domain.ids import (
31 AccountId,
32 CategoryId,
33 MessageId,
34 RuleId,
35 ThreadId,
36 YnabTransactionId,
38from ynab_agent.domain.money import Money
39from ynab_agent.domain.proposal import Decision, Proposal
40from ynab_agent.domain.signals import ReplySignal
41from ynab_agent.domain.transaction import YnabSnapshot
42from ynab_agent.learn.handler import plan_rule_update
43from ynab_agent.policy.converge import TargetState, target_of
44from ynab_agent.workflow.runtime import DATA_CONVERTER
45from ynab_agent.workflow.txn_workflow import TransactionWorkflow
46from ynab_agent.workflow.types import (
47 AnswerOutcome,
48 ReplyOutcome,
49 TransactionParams,
51 
52if TYPE_CHECKING:
53 from collections.abc import Callable
54 
55 from temporalio.client import WorkflowHandle
56 
57TASK_QUEUE = "ynab-test"
58_EPOCH = datetime.datetime(2026, 5, 28, tzinfo=datetime.UTC)
59 
60 
61async def _start_env() -> WorkflowEnvironment:
62 """A time-skipping env with the TxnThreadId search attribute registered.
63 
64 The workflow upserts ``TxnThreadId`` on ``open_thread``; the time-skipping
65 test server *hangs* the workflow task on an unregistered search attribute,
66 register it before any run (the real cluster registers it via
67 ``manage/search-attributes.yaml``).
68 """
69 env = await WorkflowEnvironment.start_time_skipping(
70 data_converter=DATA_CONVERTER
71 )
72 await env.client.operator_service.add_search_attributes(
73 AddSearchAttributesRequest(
74 namespace="default",
75 search_attributes={
76 "TxnThreadId": IndexedValueType.INDEXED_VALUE_TYPE_KEYWORD
77 },
78 )
79 )
80 return env
81 
82 
83def _snapshot(*, reconciled: bool = True) -> YnabSnapshot:
84 return YnabSnapshot(
85 ynab_id=YnabTransactionId("t1"),
86 account=AccountId("a1"),
87 payee="Blue Bottle",
88 amount=Money.from_currency("-4.50"),
89 txn_date=datetime.date(2026, 5, 28),
90 category_id=CategoryId("dining"),
91 cleared=ClearedState.RECONCILED if reconciled else ClearedState.CLEARED,
92 )
93 
94 
95def _decision(by: DecidedBy) -> Decision:
96 return Decision(
97 allocation=ResolvedCategory(category=CategoryId("dining")),
98 approved=True,
99 decided_by=by,
100 decided_at=_EPOCH,
101 )
102 
103 
104def _proposal() -> Proposal:
105 from ynab_agent.domain.allocations import ProposedCategory
106 from ynab_agent.domain.enums import Confidence
107 
108 return Proposal(
109 allocation=ProposedCategory(category=CategoryId("dining")),
110 confidence=Confidence.HIGH,
111 rationale="known merchant",
112 )
113 
114 
115def _reply() -> ReplySignal:
116 return ReplySignal(
117 thread_id=ThreadId("thread-1"),
118 message_id=MessageId("m1"),
119 from_address="matthew@example.com",
120 text="ok",
121 )
122 
123 
124def _activities(
125 *,
126 snapshot: YnabSnapshot,
127 enrich_outcome: EnrichmentOutcome,
128 interpret: ReplyOutcome | None = None,
129 converge_outcome: ConvergeOutcome | None = None,
130 read_back_seq: list[object] | None = None,
131 learning_sink: list[FeedRuleLearning] | None = None,
132 enrich_snapshot_sink: list[YnabSnapshot] | None = None,
133) -> list[Callable[..., object]]:
134 """Build mock activity implementations for one scenario."""
135 committed: dict[str, object] = {}
136 read_seq = list(read_back_seq or [])
137 
138 @activity.defn(name="fetch_snapshot")
139 async def fetch_snapshot(ynab_id: str) -> YnabSnapshot | None:
140 return snapshot
141 
142 @activity.defn(name="enrich")
143 async def enrich(snap: YnabSnapshot) -> EnrichmentOutcome:
144 if enrich_snapshot_sink is not None:
145 enrich_snapshot_sink.append(snap)
146 return enrich_outcome
147 
148 @activity.defn(name="commit_to_ynab")
149 async def commit_to_ynab(ynab_id: str, decision: Decision) -> None:
150 committed["target"] = target_of(decision)
151 
152 @activity.defn(name="read_back")
153 async def read_back(ynab_id: str) -> object:
154 if read_seq:
155 return read_seq.pop(0)
156 return committed.get("target")
157 
158 @activity.defn(name="open_thread")
159 async def open_thread(ynab_id: str, proposal: object) -> str:
160 return "thread-1"
161 
162 @activity.defn(name="send_thread_message")
163 async def send_thread_message(
164 ynab_id: str,
165 thread_id: str | None,
166 purpose: object,
167 action_seq: int,
168 proposal: object,
169 ) -> None:
170 return None
171 
172 @activity.defn(name="interpret_inbound")
173 async def interpret_inbound(
174 signal: object, snap: object, proposal: object
175 ) -> ReplyOutcome:
176 assert interpret is not None
177 return interpret
178 
179 @activity.defn(name="converge")
180 async def converge(snap: object, instruction: object) -> ConvergeOutcome:
181 assert converge_outcome is not None
182 return converge_outcome
183 
184 @activity.defn(name="feed_rule_learning")
185 async def feed_rule_learning(feed: FeedRuleLearning) -> None:
186 if learning_sink is not None:
187 learning_sink.append(feed)
188 
189 @activity.defn(name="close_thread")
190 async def close_thread(thread_id: str) -> None:
191 return None
192 
193 return [
194 fetch_snapshot,
195 enrich,
196 commit_to_ynab,
197 read_back,
198 open_thread,
199 send_thread_message,
200 interpret_inbound,
201 converge,
202 feed_rule_learning,
203 close_thread,
204 ]
205 
206 
207async def _wait_for_state(
208 handle: WorkflowHandle[TransactionWorkflow, None],
209 target: str,
210 tries: int = 60,
211) -> None:
212 for _ in range(tries):
213 if await handle.query(TransactionWorkflow.state) == target:
214 return
215 await asyncio.sleep(0.05)
216 raise AssertionError(f"workflow never reached state {target!r}")
217 
218 
219async def test_auto_apply_flows_to_open_then_archives() -> None:
220 acts = _activities(
221 snapshot=_snapshot(reconciled=True),
222 enrich_outcome=AutoApply(decision=_decision(DecidedBy.AGENT)),
223 )
224 async with (
225 await _start_env() as env,
226 Worker(
227 env.client,
228 task_queue=TASK_QUEUE,
229 workflows=[TransactionWorkflow],
230 activities=acts,
231 ),
232 ):
233 handle = await env.client.start_workflow(
234 TransactionWorkflow.run,
235 TransactionParams(ynab_id=YnabTransactionId("t1")),
236 id="txn-auto",
237 task_queue=TASK_QUEUE,
238 )
239 # Reconciled → the archive timer (time-skipped) drives it to done.
240 await handle.result()
241 
242 
243async def test_hold_amazon_resolves_on_memo_backfill_signal() -> None:
244 # A blank Amazon txn holds for item detail; a notify_snapshot carrying the
245 # backfilled memo (what W1 delivers, SPEC §2/§3) resolves the hold, and
246 # enrichment runs on the snapshot *with* the memo — not the held blank one
247 # the ~36h deadline fallback would use.
248 held = YnabSnapshot(
249 ynab_id=YnabTransactionId("t1"),
250 account=AccountId("a1"),
251 payee="Amazon",
252 amount=Money.from_currency("-31.40"),
253 txn_date=datetime.date(2026, 5, 28),
254 cleared=ClearedState.RECONCILED,
255 )
256 backfilled = held.model_copy(update={"memo": "AmazonBasics HDMI cable"})
257 enriched: list[YnabSnapshot] = []
258 acts = _activities(
259 snapshot=held,
260 enrich_outcome=AskHuman(proposal=_proposal()),
261 enrich_snapshot_sink=enriched,
262 )
263 async with (
264 await _start_env() as env,
265 Worker(
266 env.client,
267 task_queue=TASK_QUEUE,
268 workflows=[TransactionWorkflow],
269 activities=acts,
270 ),
271 ):
272 # signal-with-start delivers the backfill snapshot as W1 would, so the
273 # hold resolves to its memo rather than waiting out the deadline.
274 handle = await env.client.start_workflow(
275 TransactionWorkflow.run,
276 TransactionParams(ynab_id=YnabTransactionId("t1")),
277 id="txn-hold",
278 task_queue=TASK_QUEUE,
279 start_signal="notify_snapshot",
280 start_signal_args=[backfilled],
281 )
282 await _wait_for_state(handle, "awaiting_human")
283 assert enriched
284 assert enriched[-1].memo == "AmazonBasics HDMI cable"
285 
286 
287async def test_ask_then_answer_reaches_open_and_archives() -> None:
⋯ 208 lines hidden (lines 288–495)
288 acts = _activities(
289 snapshot=_snapshot(reconciled=True),
290 enrich_outcome=AskHuman(proposal=_proposal()),
291 interpret=AnswerOutcome(decision=_decision(DecidedBy.HUMAN)),
292 )
293 async with (
294 await _start_env() as env,
295 Worker(
296 env.client,
297 task_queue=TASK_QUEUE,
298 workflows=[TransactionWorkflow],
299 activities=acts,
300 ),
301 ):
302 # signal-with-start: the reply is buffered before AWAITING_HUMAN.
303 handle = await env.client.start_workflow(
304 TransactionWorkflow.run,
305 TransactionParams(ynab_id=YnabTransactionId("t1")),
306 id="txn-answer",
307 task_queue=TASK_QUEUE,
308 start_signal="submit_inbound",
309 start_signal_args=[_reply()],
310 )
311 await handle.result()
312 
313 
314async def test_human_confirm_feeds_rule_learning() -> None:
315 # A human answer drives APPLIED → OPEN, which emits the W5 effect carrying
316 # the payee + the human decision; plan_rule_update then learns the rule.
317 sink: list[FeedRuleLearning] = []
318 acts = _activities(
319 snapshot=_snapshot(reconciled=True),
320 enrich_outcome=AskHuman(proposal=_proposal()),
321 interpret=AnswerOutcome(decision=_decision(DecidedBy.HUMAN)),
322 learning_sink=sink,
323 )
324 async with (
325 await _start_env() as env,
326 Worker(
327 env.client,
328 task_queue=TASK_QUEUE,
329 workflows=[TransactionWorkflow],
330 activities=acts,
331 ),
332 ):
333 handle = await env.client.start_workflow(
334 TransactionWorkflow.run,
335 TransactionParams(ynab_id=YnabTransactionId("t1")),
336 id="txn-learn",
337 task_queue=TASK_QUEUE,
338 start_signal="submit_inbound",
339 start_signal_args=[_reply()],
340 )
341 await handle.result()
342 
343 assert len(sink) == 1
344 feed = sink[0]
345 assert feed.event is RuleLearningKind.CONFIRM
346 assert feed.payee == "Blue Bottle"
347 
348 # The captured effect, fed to the handler, learns a confirmed dining rule.
349 outcome = plan_rule_update((), feed, now=_EPOCH, next_id=RuleId("r-new"))
350 assert outcome is not None
351 rule = outcome.rules[0]
352 assert rule.trust is TrustState.CONFIRMED
353 assert isinstance(rule.action.allocation, ProposedCategory)
354 assert rule.action.allocation.category == "dining"
355 
356 
357async def test_patience_timeout_lapses() -> None:
358 acts = _activities(
359 snapshot=_snapshot(reconciled=False),
360 enrich_outcome=AskHuman(proposal=_proposal()),
361 )
362 async with (
363 await _start_env() as env,
364 Worker(
365 env.client,
366 task_queue=TASK_QUEUE,
367 workflows=[TransactionWorkflow],
368 activities=acts,
369 ),
370 ):
371 handle = await env.client.start_workflow(
372 TransactionWorkflow.run,
373 TransactionParams(ynab_id=YnabTransactionId("t1")),
374 id="txn-lapse",
375 task_queue=TASK_QUEUE,
376 )
377 # Reach the resting ask, then skip past the ~7d patience window.
378 await _wait_for_state(handle, "awaiting_human")
379 await env.sleep(datetime.timedelta(days=8))
380 await _wait_for_state(handle, "lapsed")
381 
382 
383async def test_open_inbound_revises_then_archives() -> None:
384 acts = _activities(
385 snapshot=_snapshot(reconciled=True),
386 enrich_outcome=AutoApply(decision=_decision(DecidedBy.AGENT)),
387 converge_outcome=Reapplied(decision=_decision(DecidedBy.HUMAN)),
388 )
389 async with (
390 await _start_env() as env,
391 Worker(
392 env.client,
393 task_queue=TASK_QUEUE,
394 workflows=[TransactionWorkflow],
395 activities=acts,
396 ),
397 ):
398 # The buffered reply is consumed in OPEN → REVISING → converge.
399 handle = await env.client.start_workflow(
400 TransactionWorkflow.run,
401 TransactionParams(ynab_id=YnabTransactionId("t1")),
402 id="txn-revise",
403 task_queue=TASK_QUEUE,
404 start_signal="submit_inbound",
405 start_signal_args=[_reply()],
406 )
407 await handle.result()
408 
409 
410async def test_open_manual_edit_at_archive_demotes_and_archives() -> None:
411 # AUTO applies "dining"; the verify read matches (→ OPEN). At the archive
412 # boundary a re-read shows the owner recategorized to "gifts" in YNAB — a
413 # silent override that must feed a CORRECT demotion before closing (§14.2).
414 sink: list[FeedRuleLearning] = []
415 auto = _decision(DecidedBy.AGENT).model_copy(
416 update={"rule_id": RuleId("r1")}
417 )
418 matches = TargetState(
419 allocation=ResolvedCategory(category=CategoryId("dining")),
420 approved=True,
421 )
422 overridden = TargetState(
423 allocation=ResolvedCategory(category=CategoryId("gifts")),
424 approved=True,
425 )
426 acts = _activities(
427 snapshot=_snapshot(reconciled=True),
428 enrich_outcome=AutoApply(decision=auto),
429 read_back_seq=[matches, overridden],
430 learning_sink=sink,
431 )
432 async with (
433 await _start_env() as env,
434 Worker(
435 env.client,
436 task_queue=TASK_QUEUE,
437 workflows=[TransactionWorkflow],
438 activities=acts,
439 ),
440 ):
441 handle = await env.client.start_workflow(
442 TransactionWorkflow.run,
443 TransactionParams(ynab_id=YnabTransactionId("t1")),
444 id="txn-override",
445 task_queue=TASK_QUEUE,
446 )
447 await handle.result()
448 
449 assert len(sink) == 1
450 feed = sink[0]
451 assert feed.event is RuleLearningKind.CORRECT
452 # The prior (the agent's auto decision) names the rule to demote.
453 assert feed.prior is not None
454 assert feed.prior.rule_id == "r1"
455 assert feed.decision is not None
456 assert isinstance(feed.decision.allocation, ResolvedCategory)
457 assert feed.decision.allocation.category == "gifts"
458 
459 
460async def test_diverged_verify_flags_awaiting_and_does_not_livelock() -> None:
461 # First read-back diverges (forcing a flagged AWAITING_HUMAN); the second
462 # (after the human reply re-commits) matches.
463 divergent = TargetState(
464 allocation=ResolvedCategory(category=CategoryId("gifts")),
465 approved=True,
466 )
467 acts = _activities(
468 snapshot=_snapshot(reconciled=True),
469 enrich_outcome=AutoApply(decision=_decision(DecidedBy.AGENT)),
470 interpret=AnswerOutcome(decision=_decision(DecidedBy.HUMAN)),
471 read_back_seq=[divergent],
472 )
473 async with (
474 await _start_env() as env,
475 Worker(
476 env.client,
477 task_queue=TASK_QUEUE,
478 workflows=[TransactionWorkflow],
479 activities=acts,
480 ),
481 ):
482 handle = await env.client.start_workflow(
483 TransactionWorkflow.run,
484 TransactionParams(ynab_id=YnabTransactionId("t1")),
485 id="txn-diverge",
486 task_queue=TASK_QUEUE,
487 )
488 # The diverged verify parks it in a flagged AWAITING_HUMAN.
489 await _wait_for_state(handle, "awaiting_human")
490 # Past the patience window it must NOT lapse and must NOT livelock.
491 await env.sleep(datetime.timedelta(days=8))
492 assert await handle.query(TransactionWorkflow.state) == "awaiting_human"
493 # A reply still resolves it, all the way to archived.
494 await handle.signal(TransactionWorkflow.submit_inbound, _reply())
495 await handle.result()