What changed, and why

A deterministic bug — a dict where a Rule was expected — once made the enrich activity retry 500+ times with no signal that anything was wrong. This change closes both halves of that failure: the retrying that never gave up, and the silence while it happened.

Two coupled pieces. First, the retry policy now gives up on a wall-clock budget rather than an attempt count, and fails a wider set of bug-shaped exceptions immediately. Second, when an activity does fail for good, a W2 hook pushes one phone alert (via ntfy) naming the activity and payee, then lets the transaction fail visibly. The alert is immediate but deduped by a durable ledger, so it can't bombard — a single broken transaction pages once a day, a systemic break pages a few times then quiets.

Why a wall-clock budget, not maximum_attempts

This is the load-bearing decision, so it carries a long comment. The trap: per-attempt cost in this system spans ~1000x. A fast-fail (YNAB down, Ollama unreachable) fails in milliseconds; a hung model attempt runs to its 1200 s (20 min) request timeout before it fails. So one attempt count means radically different real-time windows — maximum_attempts=30 is ~25 min for the fast case but up to ~10 hours for a hung model, and that spread exists even within the single enrich activity.

schedule_to_close is the bound; 45 min is 'how long an outage lasts before you get paged'.

src/ynab_agent/workflow/constants.py · 119 lines
src/ynab_agent/workflow/constants.py119 lines · Python
⋯ 32 lines hidden (lines 1–32)
1"""Shared operational constants for the transaction workflows.
2 
3Kept import-light (stdlib + the Temporal SDK) so the workflow sandboxes can pass
4it through without pulling domain types. The poller (W1) sets its own longer
5window for delta fetches; these are the short request/response workflows'
6defaults.
7 
8These knobs encode the system's whole failure philosophy: retry a *transient*
9blip until it heals, fail a *deterministic* bug immediately, and — the part that
10earns this module its long comment — bound the retrying by the **wall clock**,
11not the attempt count, so a stuck activity always surfaces (and alerts) within a
12predictable window.
13"""
14 
15from __future__ import annotations
16 
17from datetime import timedelta
18 
19from temporalio.common import RetryPolicy
20 
21# Default per-attempt timeout for the short workflows (W2/W3/W4). One window
22# fits them all, sized very generously for the slowest: the agentic activities
23# call a local Gemma 4 31b over the tailnet *with reasoning on*, where a cold
24# 19 GB model load plus a long reasoned generation can run for several minutes.
25# We deliberately favour completion over speed (intelligence is the point), so
26# the ceiling sits well above the model's request timeout (see `agentic.model`,
27# 1200 s), which trips first and is retried. The fast I/O activities
28# (YNAB/AgentMail) finish in well under a second, so the high ceiling only
29# bounds a genuine hang. This is ``start_to_close`` — the budget below bounds
30# the whole retrying lifecycle.
31ACTIVITY_TIMEOUT = timedelta(seconds=1800)
32 
33# ── The retry budget: bound the wall clock, not the attempt count ─────────────
35# Why not ``maximum_attempts``? Because per-attempt cost in this system spans
36# ~1000x, so any single attempt count yields wildly different real-time windows:
38# * a fast-fail (YNAB down, Ollama *unreachable* -> connection refused) fails
39# in well under a second, so the backoff dominates;
40# * a *hung* model attempt runs to the 1200 s (20 min) request timeout in
41# `agentic.model` before it fails.
43# So ``maximum_attempts=30`` means ~25 min for the fast-fail case but up to ~10
44# *hours* for a hung model — and that 1000x spread exists even *within* the
45# single `enrich` activity ("Ollama asleep" fails in ms; "Ollama wedged" takes
46# 20 min). No attempt count can mean "give up after ~X minutes" across that.
48# ``schedule_to_close_timeout`` (applied at each ``execute_activity`` call, set
49# to this budget) bounds the activity's *entire* retrying life — all attempts
50# plus backoff — by elapsed time. It auto-adapts to each activity's per-attempt
51# cost: a deterministic bug fails at attempt 1 (the non-retryable list below)
52# and alerts in seconds regardless of the budget; anything transient retries
53# until the budget, then goes terminal and alerts. Same predictable window
54# whether the failure is fast-fail spam or a slow hang.
56# Why 45 minutes? It's "how long an outage lasts before you get paged." 45 min
57# rides out the common brief outages (a Mac Studio reboot, an Ollama restart, a
58# transient YNAB/AgentMail blip) without crying wolf, yet surfaces a *real*
59# outage well before the hourly poll re-fire. And it must exceed
60# ``ACTIVITY_TIMEOUT`` (one legitimate cold-load generation) so a slow attempt
61# is never guillotined as if it were a failure — 45 min > 30 min, with room for
62# a retry. The W1 poll loop re-addresses a failed W2 every tick (hourly, via
63# ``ALLOW_DUPLICATE_FAILED_ONLY``), so this budget need not survive a multi-hour
64# outage — the poll is the long-horizon retry; this is the short one. Net: one
65# alert ~45 min into a real outage, silence for blips shorter than that, instant
66# alert for actual bugs.
67ACTIVITY_BUDGET = timedelta(minutes=45)
68 
69# Cap the exponential backoff so fast-fail retries settle to a steady cadence
70# (~once every 2 min) instead of ballooning toward the budget on their own. With
71# this cap a fast-fail failure gets ~25 retries inside the 45 min budget —
72# plenty to ride out a blip — rather than a handful of ever-longer sleeps.
73_MAX_RETRY_INTERVAL = timedelta(seconds=120)
⋯ 46 lines hidden (lines 74–119)
74 
75# Deterministic failures a retry cannot fix: a malformed model output, a bad
76# payload, a programming error. Temporal records the raised exception's type
77# name, so we match by name — no need to import the activity-layer types into
78# the sandbox. This is a denylist (Temporal retries by default), so it must be
79# kept reasonably complete for the common bug classes — but ``ACTIVITY_BUDGET``
80# above is the real backstop for any type we forget: an unenumerated exception
81# still goes terminal when the budget elapses, instead of spinning forever.
82# (``AttributeError`` was the one missing here that let a registry-deserial-
83# ization bug retry 500+ times in production — #4.)
84_NON_RETRYABLE = (
85 "ValueError",
86 "TypeError",
87 "KeyError",
88 "AttributeError",
89 "IndexError",
90 "NameError",
91 "NotImplementedError", # the W4/W6 activity stubs raise this
92 "AssertionError",
93 "ValidationError", # pydantic
94 "UnexpectedModelBehavior", # pydantic-ai
96 
97# The retry policy every activity call uses (SPEC §0.5). Attempts stay unbounded
98# — ``ACTIVITY_BUDGET`` (the per-call ``schedule_to_close_timeout``) is the
99# bound, not a count — so the durable self-heal survives a transient YNAB /
100# AgentMail / model blip without dropping a transaction, while the non-retryable
101# list fails a deterministic bug fast and the budget stops everything else from
102# spinning past its window.
103ACTIVITY_RETRY = RetryPolicy(
104 non_retryable_error_types=list(_NON_RETRYABLE),
105 maximum_interval=_MAX_RETRY_INTERVAL,
107 
108# ── The alerting path: fast and best-effort ───────────────────────────────────
109# The failure-alert activities (the ntfy push and the dedup-ledger reads) run
110# *while a transaction is already failing*, so they must be quick and must never
111# become the thing that spins. Short ceiling, short total budget, few attempts —
112# a missed alert is acceptable (the send itself swallows its own errors); a
113# slow or looping alert path is not.
114ALERT_TIMEOUT = timedelta(seconds=30)
115ALERT_BUDGET = timedelta(seconds=90)
116ALERT_RETRY = RetryPolicy(
117 maximum_attempts=3,
118 maximum_interval=timedelta(seconds=10),

The denylist that fails a bug fast

Temporal retries by default, so the only lever is a denylist of exception type names that should not retry. The mechanism already existed; the gap was that AttributeError — exactly what the #4 bug raised — wasn't on it, so a deterministic crash was treated as a transient blip. The list now covers the common bug classes, and NotImplementedError is here too because the W4/W6 activity stubs raise it.

Denylist + the policy. Attempts stay unbounded — the budget is the bound.

src/ynab_agent/workflow/constants.py · 119 lines
src/ynab_agent/workflow/constants.py119 lines · Python
⋯ 74 lines hidden (lines 1–74)
1"""Shared operational constants for the transaction workflows.
2 
3Kept import-light (stdlib + the Temporal SDK) so the workflow sandboxes can pass
4it through without pulling domain types. The poller (W1) sets its own longer
5window for delta fetches; these are the short request/response workflows'
6defaults.
7 
8These knobs encode the system's whole failure philosophy: retry a *transient*
9blip until it heals, fail a *deterministic* bug immediately, and — the part that
10earns this module its long comment — bound the retrying by the **wall clock**,
11not the attempt count, so a stuck activity always surfaces (and alerts) within a
12predictable window.
13"""
14 
15from __future__ import annotations
16 
17from datetime import timedelta
18 
19from temporalio.common import RetryPolicy
20 
21# Default per-attempt timeout for the short workflows (W2/W3/W4). One window
22# fits them all, sized very generously for the slowest: the agentic activities
23# call a local Gemma 4 31b over the tailnet *with reasoning on*, where a cold
24# 19 GB model load plus a long reasoned generation can run for several minutes.
25# We deliberately favour completion over speed (intelligence is the point), so
26# the ceiling sits well above the model's request timeout (see `agentic.model`,
27# 1200 s), which trips first and is retried. The fast I/O activities
28# (YNAB/AgentMail) finish in well under a second, so the high ceiling only
29# bounds a genuine hang. This is ``start_to_close`` — the budget below bounds
30# the whole retrying lifecycle.
31ACTIVITY_TIMEOUT = timedelta(seconds=1800)
32 
33# ── The retry budget: bound the wall clock, not the attempt count ─────────────
35# Why not ``maximum_attempts``? Because per-attempt cost in this system spans
36# ~1000x, so any single attempt count yields wildly different real-time windows:
38# * a fast-fail (YNAB down, Ollama *unreachable* -> connection refused) fails
39# in well under a second, so the backoff dominates;
40# * a *hung* model attempt runs to the 1200 s (20 min) request timeout in
41# `agentic.model` before it fails.
43# So ``maximum_attempts=30`` means ~25 min for the fast-fail case but up to ~10
44# *hours* for a hung model — and that 1000x spread exists even *within* the
45# single `enrich` activity ("Ollama asleep" fails in ms; "Ollama wedged" takes
46# 20 min). No attempt count can mean "give up after ~X minutes" across that.
48# ``schedule_to_close_timeout`` (applied at each ``execute_activity`` call, set
49# to this budget) bounds the activity's *entire* retrying life — all attempts
50# plus backoff — by elapsed time. It auto-adapts to each activity's per-attempt
51# cost: a deterministic bug fails at attempt 1 (the non-retryable list below)
52# and alerts in seconds regardless of the budget; anything transient retries
53# until the budget, then goes terminal and alerts. Same predictable window
54# whether the failure is fast-fail spam or a slow hang.
56# Why 45 minutes? It's "how long an outage lasts before you get paged." 45 min
57# rides out the common brief outages (a Mac Studio reboot, an Ollama restart, a
58# transient YNAB/AgentMail blip) without crying wolf, yet surfaces a *real*
59# outage well before the hourly poll re-fire. And it must exceed
60# ``ACTIVITY_TIMEOUT`` (one legitimate cold-load generation) so a slow attempt
61# is never guillotined as if it were a failure — 45 min > 30 min, with room for
62# a retry. The W1 poll loop re-addresses a failed W2 every tick (hourly, via
63# ``ALLOW_DUPLICATE_FAILED_ONLY``), so this budget need not survive a multi-hour
64# outage — the poll is the long-horizon retry; this is the short one. Net: one
65# alert ~45 min into a real outage, silence for blips shorter than that, instant
66# alert for actual bugs.
67ACTIVITY_BUDGET = timedelta(minutes=45)
68 
69# Cap the exponential backoff so fast-fail retries settle to a steady cadence
70# (~once every 2 min) instead of ballooning toward the budget on their own. With
71# this cap a fast-fail failure gets ~25 retries inside the 45 min budget —
72# plenty to ride out a blip — rather than a handful of ever-longer sleeps.
73_MAX_RETRY_INTERVAL = timedelta(seconds=120)
74 
75# Deterministic failures a retry cannot fix: a malformed model output, a bad
76# payload, a programming error. Temporal records the raised exception's type
77# name, so we match by name — no need to import the activity-layer types into
78# the sandbox. This is a denylist (Temporal retries by default), so it must be
79# kept reasonably complete for the common bug classes — but ``ACTIVITY_BUDGET``
80# above is the real backstop for any type we forget: an unenumerated exception
81# still goes terminal when the budget elapses, instead of spinning forever.
82# (``AttributeError`` was the one missing here that let a registry-deserial-
83# ization bug retry 500+ times in production — #4.)
84_NON_RETRYABLE = (
85 "ValueError",
86 "TypeError",
87 "KeyError",
88 "AttributeError",
89 "IndexError",
90 "NameError",
91 "NotImplementedError", # the W4/W6 activity stubs raise this
92 "AssertionError",
93 "ValidationError", # pydantic
94 "UnexpectedModelBehavior", # pydantic-ai
96 
97# The retry policy every activity call uses (SPEC §0.5). Attempts stay unbounded
98# — ``ACTIVITY_BUDGET`` (the per-call ``schedule_to_close_timeout``) is the
99# bound, not a count — so the durable self-heal survives a transient YNAB /
100# AgentMail / model blip without dropping a transaction, while the non-retryable
101# list fails a deterministic bug fast and the budget stops everything else from
102# spinning past its window.
103ACTIVITY_RETRY = RetryPolicy(
104 non_retryable_error_types=list(_NON_RETRYABLE),
105 maximum_interval=_MAX_RETRY_INTERVAL,
⋯ 13 lines hidden (lines 107–119)
107 
108# ── The alerting path: fast and best-effort ───────────────────────────────────
109# The failure-alert activities (the ntfy push and the dedup-ledger reads) run
110# *while a transaction is already failing*, so they must be quick and must never
111# become the thing that spins. Short ceiling, short total budget, few attempts —
112# a missed alert is acceptable (the send itself swallows its own errors); a
113# slow or looping alert path is not.
114ALERT_TIMEOUT = timedelta(seconds=30)
115ALERT_BUDGET = timedelta(seconds=90)
116ALERT_RETRY = RetryPolicy(
117 maximum_attempts=3,
118 maximum_interval=timedelta(seconds=10),

The dedup ledger: immediate, but never a flood

"Alert on every failure" and "don't bombard me" meet in two pure rules. A per-key cooldown: the same transaction id, re-failing every hourly poll tick, alerts once per 24h, not hourly. A global rate cap: independent of any key, no more than ~5 alerts fire in a rolling hour, so a systemic break across many distinct transactions pings a handful of times then goes quiet.

should_notify: the cooldown check, then the rate-cap count. record prunes to keep the tail bounded.

src/ynab_agent/alert/ledger.py · 90 lines
src/ynab_agent/alert/ledger.py90 lines · Python
⋯ 54 lines hidden (lines 1–54)
1"""The alert ledger's pure state and folds — the anti-bombard policy (SPEC §13).
2 
3Two rules turn "alert immediately on every failure" into "alert promptly but
4never flood":
5 
6* **Per-key cooldown** (default 24 h): a given failure key — a transaction id —
7 alerts at most once per cooldown. The W1 poll re-fires a failed W2 every tick
8 (hourly), so without this a single deterministically-broken transaction would
9 ping every hour forever; with it, once a day.
10* **Global rate cap** (default 5 / hour): independent of the per-key rule, no
11 more than N alerts fire in any rolling hour. A systemic break (a bad deploy
12 failing *every* transaction — distinct keys, so the cooldown doesn't catch
13 them) pings a handful of times, then goes quiet. After a few "X failed"
14 pings it is obvious the whole thing is down; the (N+1)th adds no information.
15 
16The ledger is an append-only tail of ``(key, at)`` entries, pruned to the
17cooldown horizon so continued alerting never grows the carried state without
18bound (the same shape as the rule registry's audit tail).
19"""
20 
21from __future__ import annotations
22 
23import datetime
24 
25from ynab_agent.domain.base import Frozen
26 
27# How long the same failure key stays quiet after an alert. A re-fired failure
28# (the hourly poll re-addressing a still-broken transaction) is the same key, so
29# this is what collapses "every poll tick" down to "once a day".
30DEFAULT_COOLDOWN = datetime.timedelta(hours=24)
31 
32# The rolling window and ceiling for the global rate cap — the guard against a
33# systemic break (many *distinct* keys failing at once) turning into a flood.
34DEFAULT_RATE_WINDOW = datetime.timedelta(hours=1)
35DEFAULT_RATE_CAP = 5
36 
37 
38class AlertEntry(Frozen):
39 """One alert that fired: its dedup key and when it went out."""
40 
41 key: str
42 at: datetime.datetime
43 
44 
45class LedgerState(Frozen):
46 """The pruned tail of recent alerts — the whole dedup memory.
47 
48 Held as Temporal workflow state and carried across continue-as-new, so it is
49 a frozen value like the rest of the domain.
50 """
51 
52 entries: tuple[AlertEntry, ...] = ()
53 
54 
55def should_notify(
56 state: LedgerState,
57 key: str,
58 now: datetime.datetime,
59 *,
60 cooldown: datetime.timedelta = DEFAULT_COOLDOWN,
61 rate_window: datetime.timedelta = DEFAULT_RATE_WINDOW,
62 rate_cap: int = DEFAULT_RATE_CAP,
63) -> bool:
64 """Whether an alert for ``key`` should fire now (SPEC §13). Pure.
65 
66 ``False`` if this key alerted within ``cooldown`` (per-key dedup) or if the
67 rolling-``rate_window`` alert count has reached ``rate_cap`` (global flood
68 guard); ``True`` otherwise.
69 """
70 for entry in state.entries:
71 if entry.key == key and now - entry.at < cooldown:
72 return False
73 recent = sum(1 for entry in state.entries if now - entry.at < rate_window)
74 return recent < rate_cap
75 
76 
77def record(
78 state: LedgerState,
79 key: str,
80 now: datetime.datetime,
81 *,
82 retention: datetime.timedelta = DEFAULT_COOLDOWN,
83) -> LedgerState:
84 """Append ``key@now`` and prune entries older than ``retention``. Pure.
85 
86 The tail stays bounded — the cooldown is the longest horizon any rule looks
87 back — so continued alerting never grows the carried state without limit.
88 """
89 kept = tuple(entry for entry in state.entries if now - entry.at < retention)
90 return LedgerState(entries=(*kept, AlertEntry(key=key, at=now)))

This is pure and frozen like the rest of the domain; the durable AlertLedgerWorkflow wraps it as Temporal state, signalling record after a send and querying should_notify before one — the same shape the rule registry has over its folds.

The W2 hook: page once, then re-raise

The transaction loop is wrapped so a terminal activity failure is caught, alerted, and re-raised — the transaction still fails and stays visible in Temporal. The catch is narrow on purpose: only ActivityError (a non-retryable bug, or the budget elapsing). continue_as_new raises a different exception, so the normal restart path sails through untouched.

try around the loop; except ActivityError -> alert -> raise. _alert_context names the payee.

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

Pure: composes the push text from replay-safe fields off the error (the activity name and cause).

src/ynab_agent/workflow/alerting.py · 43 lines
src/ynab_agent/workflow/alerting.py43 lines · Python
⋯ 17 lines hidden (lines 1–17)
1"""Compose an operator alert from a terminal activity failure (SPEC §13).
2 
3A pure helper used by a workflow's failure hook: it reads only replay-safe
4fields off the ``ActivityError`` (the activity name and the underlying cause)
5and turns them into the :class:`FailureAlert` the ``alert_failure`` activity
6pushes. Kept free of Temporal commands so the hook stays deterministic; the SDK
7exception types it inspects are sandbox-safe.
8"""
9 
10from __future__ import annotations
11 
12from temporalio.exceptions import ActivityError, ApplicationError
13from temporalio.exceptions import TimeoutError as TemporalTimeoutError
14 
15from ynab_agent.workflow.alert_types import FailureAlert
16 
17 
18def build_failure_alert(
19 *, key: str, context: str, exc: ActivityError
20) -> FailureAlert:
21 """The operator alert for a terminal activity failure. Pure.
22 
23 ``key`` drives dedup (the transaction id); ``context`` is a human locator
24 (payee + txn id) for the push body.
25 """
26 activity_name = exc.activity_type or "an activity"
27 body = f"{context}\n{activity_name} failed after retries — {_explain(exc)}"
28 return FailureAlert(
29 key=key,
30 title=f"ynab-agent: {activity_name} failed",
31 body=body,
32 )
33 
34 
35def _explain(exc: ActivityError) -> str:
36 """A short, human reason from the error's cause (bug detail or timeout)."""
37 cause = exc.__cause__
38 if isinstance(cause, ApplicationError):
39 return f"{cause.type or 'error'}: {cause}"
40 if isinstance(cause, TemporalTimeoutError):
41 kind = cause.type.name if cause.type is not None else "timeout"
42 return f"timed out ({kind})"
43 return str(cause) if cause is not None else "unknown failure"

The alert activity: best-effort, never raises

This runs while a transaction is already failing, so its prime directive is to never become a new failure. The whole body is guarded: a dedup check, a send, and a record — and any error in them is logged and swallowed. Because it never raises, Temporal sees success and never retries it, so the push happens at most once. A missed page is acceptable; an alert path that raises (masking the real failure) or loops is not.

Suppressed? return. Unconfigured? log and return. Send off the event loop, then record — never raise.

src/ynab_agent/workflow/alert_activities.py · 123 lines
src/ynab_agent/workflow/alert_activities.py123 lines · Python
⋯ 30 lines hidden (lines 1–30)
1"""The failure-alert activity: a deduped ntfy push, best-effort (SPEC §13).
2 
3``alert_failure`` runs from the W2 terminal-failure hook *while a transaction is
4already failing*, so its prime directive is: **never become a new failure.**
5Every error here is logged and swallowed — a missed page is acceptable; an alert
6path that raises (masking the real failure) or loops is not. Because it never
7raises, Temporal sees success and never retries it, so the push happens at most
8once per hook invocation.
9 
10The dedup + rate cap live in the durable ``AlertLedgerWorkflow``: this activity
11asks it ``should_notify`` before sending and signals ``record`` after, exactly
12as ``feed_rule_learning`` talks to the rule registry. Heavy imports
13(``httpx`` via the notify client) stay lazy so this module — referenced from the
14workflow sandbox — never pulls them across the boundary.
15"""
16 
17from __future__ import annotations
18 
19import datetime
20from typing import TYPE_CHECKING
21 
22from temporalio import activity
23 
24from ynab_agent.workflow.alert_types import FailureAlert
25 
26if TYPE_CHECKING:
27 from ynab_agent.notify.client import Notification, NotifyClient
28 
29 
30@activity.defn
31async def alert_failure(alert: FailureAlert) -> None:
32 """Push one deduped operator alert. Best-effort: never raises."""
33 import asyncio
34 
35 try:
36 now = datetime.datetime.now(datetime.UTC)
37 if not await _should_notify(alert.key, now):
38 activity.logger.info(
39 "failure alert suppressed (dedup/rate cap): %s", alert.key
40 )
41 return
42 client = _client_or_none()
43 if client is None:
44 activity.logger.warning(
45 "NTFY_TOPIC unset; failure alert dropped: %s", alert.key
46 )
47 return
48 # The ntfy POST is blocking; keep it off the event loop. If it raises
49 # (ntfy down), the outer guard logs it and we do *not* record — so a
50 # later failure can try again rather than being deduped into silence.
51 await asyncio.to_thread(client.notify, _build_notification(alert))
52 await _record(alert.key)
53 except Exception:
54 # Best-effort: never let an alerting failure mask the real one.
55 activity.logger.warning(
56 "failure alert could not be delivered", exc_info=True
57 )
58 
⋯ 65 lines hidden (lines 59–123)
59 
60async def _should_notify(key: str, now: datetime.datetime) -> bool:
61 """Ask the durable ledger; default to *notify* if it doesn't exist yet."""
62 from temporalio.service import RPCError
63 
64 from ynab_agent.workflow.alert_types import (
65 ALERT_LEDGER_WORKFLOW_ID,
66 ShouldNotifyRequest,
67 )
68 from ynab_agent.workflow.temporal_client import client
69 
70 temporal = await client()
71 handle = temporal.get_workflow_handle(ALERT_LEDGER_WORKFLOW_ID)
72 try:
73 decision: bool = await handle.query(
74 "should_notify",
75 ShouldNotifyRequest(key=key, now=now),
76 result_type=bool,
77 )
78 except RPCError:
79 # No ledger started yet → nothing has ever alerted → notify.
80 return True
81 return decision
82 
83 
84def _build_notification(alert: FailureAlert) -> Notification:
85 from ynab_agent.notify.client import Notification
86 
87 return Notification(
88 title=alert.title,
89 body=alert.body,
90 priority="high",
91 tags=("rotating_light",),
92 )
93 
94 
95def _client_or_none() -> NotifyClient | None:
96 from ynab_agent.notify.client import NotifyClient
97 
98 try:
99 return NotifyClient.from_env()
100 except RuntimeError:
101 return None
102 
103 
104async def _record(key: str) -> None:
105 """Signal-with-start the ledger so this key's cooldown begins now."""
106 from temporalio.common import WorkflowIDConflictPolicy
107 
108 from ynab_agent.workflow.alert_types import (
109 ALERT_LEDGER_WORKFLOW_ID,
110 LedgerParams,
111 )
112 from ynab_agent.workflow.temporal_client import client, task_queue
113 
114 temporal = await client()
115 await temporal.start_workflow(
116 "AlertLedgerWorkflow",
117 LedgerParams(),
118 id=ALERT_LEDGER_WORKFLOW_ID,
119 task_queue=task_queue(),
120 id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
121 start_signal="record",
122 start_signal_args=[key],
123 )

Proof: one terminal failure, one push

The end-to-end test runs the real workflow on the time-skipping server with a mock enrich that raises a non-retryable error and the real alert_failure activity wired to a fake ntfy backend and a real ledger. It asserts the transaction still fails, and that exactly one push went out naming the activity, the payee, and the cause — the whole path, not a mock of it.

Real alert_failure + fake backend + real ledger; assert the txn fails AND one push carries enrich/Blue Bottle/ValueError.

tests/workflow/test_failure_alert.py · 115 lines
tests/workflow/test_failure_alert.py115 lines · Python
⋯ 78 lines hidden (lines 1–78)
1"""W2 pages the owner once when an activity fails terminally (SPEC §13).
2 
3The whole alert path, end to end on the time-skipping server: a non-retryable
4``enrich`` failure trips the workflow's terminal-failure hook, which runs the
5*real* ``alert_failure`` activity — it consults the durable ledger, pushes
6through a fake ntfy backend, and records the alert — then re-raises so the
7transaction still fails. Proves the hook fires, the push carries the failing
8activity + payee, and the workflow surfaces the failure rather than swallowing
9it.
10"""
11 
12from __future__ import annotations
13 
14import datetime
15from typing import TYPE_CHECKING
16 
17import pytest
18from temporalio import activity
19from temporalio.client import WorkflowFailureError
20from temporalio.testing import WorkflowEnvironment
21from temporalio.worker import Worker
22 
23import ynab_agent.notify.client as notify_client
24import ynab_agent.workflow.temporal_client as temporal_client
25from ynab_agent.domain.enums import ClearedState
26from ynab_agent.domain.ids import AccountId, CategoryId, YnabTransactionId
27from ynab_agent.domain.money import Money
28from ynab_agent.domain.transaction import YnabSnapshot
29from ynab_agent.notify.client import Notification, NotifyClient
30from ynab_agent.workflow import alert_activities
31from ynab_agent.workflow.alert_ledger_workflow import AlertLedgerWorkflow
32from ynab_agent.workflow.runtime import DATA_CONVERTER
33from ynab_agent.workflow.txn_workflow import TransactionWorkflow
34from ynab_agent.workflow.types import TransactionParams
35 
36if TYPE_CHECKING:
37 from collections.abc import Callable
38 
39_TASK_QUEUE = "ynab-failure-alert-test"
40 
41 
42class _RecordingBackend:
43 """A fake ntfy backend that captures every notification published."""
44 
45 def __init__(self) -> None:
46 self.sent: list[Notification] = []
47 
48 def publish(self, notification: Notification) -> None:
49 self.sent.append(notification)
50 
51 
52def _snapshot() -> YnabSnapshot:
53 return YnabSnapshot(
54 ynab_id=YnabTransactionId("t1"),
55 account=AccountId("a1"),
56 payee="Blue Bottle",
57 amount=Money.from_currency("-4.50"),
58 txn_date=datetime.date(2026, 5, 28),
59 category_id=CategoryId("dining"),
60 cleared=ClearedState.CLEARED,
61 )
62 
63 
64def _failing_enrich_activities() -> list[Callable[..., object]]:
65 @activity.defn(name="fetch_snapshot")
66 async def fetch_snapshot(ynab_id: str) -> YnabSnapshot:
67 return _snapshot()
68 
69 @activity.defn(name="enrich")
70 async def enrich(snapshot: YnabSnapshot) -> object:
71 # ValueError is on the non-retryable list → terminal on attempt 1.
72 msg = "boom in the model layer"
73 raise ValueError(msg)
74 
75 # The real alert_failure runs here (not a mock) — that is the point.
76 return [fetch_snapshot, enrich, alert_activities.alert_failure]
77 
78 
79async def test_terminal_enrich_failure_pages_once(
80 monkeypatch: pytest.MonkeyPatch,
81) -> None:
82 backend = _RecordingBackend()
83 monkeypatch.setattr(notify_client, "_CACHED", NotifyClient(backend))
84 # alert_failure reaches the ledger through these process globals.
85 monkeypatch.setenv("TEMPORAL_TASK_QUEUE", _TASK_QUEUE)
86 
87 async with (
88 await WorkflowEnvironment.start_time_skipping(
89 data_converter=DATA_CONVERTER
90 ) as env,
91 Worker(
92 env.client,
93 task_queue=_TASK_QUEUE,
94 workflows=[TransactionWorkflow, AlertLedgerWorkflow],
95 activities=_failing_enrich_activities(),
96 ),
97 ):
98 monkeypatch.setattr(temporal_client, "_CLIENT", env.client)
99 handle = await env.client.start_workflow(
100 TransactionWorkflow.run,
101 TransactionParams(ynab_id=YnabTransactionId("t1")),
102 id="txn-fail-1",
103 task_queue=_TASK_QUEUE,
104 )
105 
106 # The transaction still fails terminally (the alert never swallows it).
107 with pytest.raises(WorkflowFailureError):
108 await handle.result()
109 
110 # Exactly one push, naming the failing activity and the payee.
111 assert len(backend.sent) == 1
112 pushed = backend.sent[0]
113 assert "enrich" in pushed.title
114 assert "Blue Bottle" in pushed.body
115 assert "ValueError" in pushed.body