W6 overspend monitor — a code-level reading

What changed, and why

W6 is the daily budget guard: per category, project month-end spend by run-rate and email an alert for anything already over or trending over. The decision logic and the orchestration already existed and were tested — the pure projection (assess/should_alert in budget/overspend.py) and the OverspendMonitorWorkflow — and the daily Temporal Schedule is scaffolded in infra. What was missing was the I/O: three of the monitor's four activity ports were stubs that raised NotImplementedError.

This PR wires those three ports and adds the one piece they require: a durable place to remember what was last alerted, so the per-period dedupe survives between daily passes. It is v1 notify-only — the cover/balancing move is W7 and stays out of scope. Nothing about the workflow contract changes, so its existing test is untouched.

The orchestration that drives the new ports (unchanged here, shown for context): each pass fetches the category figures, runs the pure assess, and for anything off track that survives should_alert against the prior alert, sends and then records. The three activities this PR fills in are exactly the reads and writes in that loop.

The existing per-pass loop: assess → load prior → should_alert → send → save.

src/ynab_agent/workflow/monitor_workflow.py · 89 lines

src/ynab_agent/workflow/monitor_workflow.py89 lines · Python

⋯ 36 lines hidden (lines 1–36)1"""W6 · the overspend-monitor workflow (SPEC §7).
2 
3A short workflow per monitor pass — fired by a daily Temporal Schedule (the
4trigger is infrastructure). It reads each category's figures, runs the pure
func:`~ynab_agent.budget.overspend.assess` projection, and for anything off
6track that survives the dedupe (``should_alert`` against the last alert) emails
7an alert and records it. v1 is notify-only; no budget is moved (that is W7).
8 
9The month position comes from ``params.clock`` when set, else from
10``workflow.now()`` plus ``calendar.monthrange`` (both replay-deterministic).
11"""
12 
13from __future__ import annotations
14 
15import calendar
16 
17from temporalio import workflow
18 
19with workflow.unsafe.imports_passed_through():
  from ynab_agent.budget.overspend import (
      MonthClock,
      OverspendVerdict,
      PriorAlert,
      assess,
      should_alert,
  )
  from ynab_agent.workflow import monitor_activities
  from ynab_agent.workflow.constants import ACTIVITY_RETRY, ACTIVITY_TIMEOUT
  from ynab_agent.workflow.monitor_types import MonitorParams, MonitorResult
30 
31 
32@workflow.defn
33class OverspendMonitorWorkflow:
  """One daily overspend pass across all categories."""
35 
  @workflow.run
  async def run(self, params: MonitorParams) -> MonitorResult:
      """Assess every category and alert on what is off track (SPEC §7)."""
      clock = params.clock or self._current_clock()
      spends = await workflow.execute_activity(
          monitor_activities.fetch_category_spends,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
45 
      alerted: list[str] = []
      for spend in spends:
          assessment = assess(spend, clock)
          if assessment.verdict is OverspendVerdict.OK:
              continue
          prior = await workflow.execute_activity(
              monitor_activities.load_prior_alert,
              str(spend.category),
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          if not should_alert(assessment, prior):
              continue
          await workflow.execute_activity(
              monitor_activities.send_overspend_alert,
              assessment,
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          await workflow.execute_activity(
              monitor_activities.save_alert,
              args=[
                  str(spend.category),
                  PriorAlert(
                      verdict=assessment.verdict,
                      projected=assessment.projected,
                  ),
              ],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          alerted.append(spend.name)
78 
      return MonitorResult(
          categories=len(spends),
          alerts=len(alerted),
          alerted=tuple(alerted),
      )
⋯ 6 lines hidden (lines 84–89)84 
  def _current_clock(self) -> MonthClock:
      """Derive the month position from the replay-safe workflow clock."""
      now = workflow.now()
      days_in_month = calendar.monthrange(now.year, now.month)[1]
      return MonthClock(day_of_month=now.day, days_in_month=days_in_month)

The dedup memory — budget/ledger.py

should_alert compares the current assessment against the last alert raised this period, so that prior alert has to persist across daily runs. This new pure module is that memory, kept to one entry per category: the verdict and projected month-end it last alerted at, stamped with the budget period as "YYYY-MM".

One frozen entry per category; the whole table is a tuple carried as Temporal state.

src/ynab_agent/budget/ledger.py · 82 lines

src/ynab_agent/budget/ledger.py82 lines · Python

⋯ 27 lines hidden (lines 1–27)1"""The overspend-alert dedup ledger — pure state and folds (SPEC §7).
2 
3W6 alerts a category *at most once per budget period* unless it materially
4worsens (the ``should_alert`` rule in :mod:`ynab_agent.budget.overspend`). That
5rule compares the current assessment against the *last alert raised this
6period*, so the prior alert must survive between daily passes. This module is
7that memory, kept tiny: one entry per category — the verdict and projected
8month-end it last alerted at, stamped with the period it belongs to.
9 
10* ``prior`` answers "what did we last alert this category, this period?" — and
11  returns ``None`` across a period boundary, so the first alert of a new month
12  always fires.
13* ``record`` folds a freshly-sent alert in, replacing the category's old entry
14  so the tail stays one-per-category (bounded without pruning).
15 
16Pure and frozen like the rest of the domain; the durable
17:class:`~ynab_agent.workflow.overspend_ledger_workflow.OverspendLedgerWorkflow`
18wraps it as Temporal state, mirroring how ``alert.ledger`` sits under the
19failure-alert ledger and ``learn.registry`` under the rule registry.
20"""
21 
22from __future__ import annotations
23 
24from ynab_agent.budget.overspend import PriorAlert
25from ynab_agent.domain.base import Frozen
26 
27 
28class LedgerEntry(Frozen):
29    """The last alert raised for one category, and the period it belongs to.
30 
31    ``period`` is the budget month as ``"YYYY-MM"`` (the caller derives it from
32    the household clock), so a last-month entry reads as absent this month.
33    """
34 
35    category: str
36    period: str
37    alert: PriorAlert
38 
39 
40class OverspendLedgerState(Frozen):
41    """The whole dedup memory — one entry per category, carried as state.
42 
43    Held as Temporal workflow state and carried across continue-as-new, so it is
44    a frozen value like the rest of the domain.
45    """
46 
47    entries: tuple[LedgerEntry, ...] = ()
⋯ 35 lines hidden (lines 48–82)48 
49 
50def prior(
51    state: OverspendLedgerState, category: str, period: str
52) -> PriorAlert | None:
53    """The category's last alert *this period*, or ``None``. Pure.
54 
55    A stored entry from a different period reads as absent, so a new month
56    resets the dedupe and the first flag of the period always alerts (SPEC §7).
57    """
58    for entry in state.entries:
59        if entry.category == category and entry.period == period:
60            return entry.alert
61    return None
62 
63 
64def record(
65    state: OverspendLedgerState,
66    category: str,
67    period: str,
68    alert: PriorAlert,
69) -> OverspendLedgerState:
70    """Record this period's alert for ``category``, replacing any prior. Pure.
71 
72    Dropping the category's previous entry (regardless of period) keeps the tail
73    at one entry per category, so the carried state never grows with continued
74    alerting.
75    """
76    kept = tuple(entry for entry in state.entries if entry.category != category)
77    return OverspendLedgerState(
78        entries=(
79            *kept,
80            LedgerEntry(category=category, period=period, alert=alert),
81        )
82    )

prior returns an entry only when its stored period matches the one asked for, and record drops any existing entry for the category before appending the new one. Those two choices do the real work: the period match makes a new month read as empty (so the first flag of a period always fires), and the drop-then-append keeps the table at one row per category — bounded without any pruning logic.

prior() scopes by period; record() upserts one-per-category.

src/ynab_agent/budget/ledger.py · 82 lines

src/ynab_agent/budget/ledger.py82 lines · Python

⋯ 49 lines hidden (lines 1–49)1"""The overspend-alert dedup ledger — pure state and folds (SPEC §7).
2 
3W6 alerts a category *at most once per budget period* unless it materially
4worsens (the ``should_alert`` rule in :mod:`ynab_agent.budget.overspend`). That
5rule compares the current assessment against the *last alert raised this
6period*, so the prior alert must survive between daily passes. This module is
7that memory, kept tiny: one entry per category — the verdict and projected
8month-end it last alerted at, stamped with the period it belongs to.
9 
10* ``prior`` answers "what did we last alert this category, this period?" — and
11  returns ``None`` across a period boundary, so the first alert of a new month
12  always fires.
13* ``record`` folds a freshly-sent alert in, replacing the category's old entry
14  so the tail stays one-per-category (bounded without pruning).
15 
16Pure and frozen like the rest of the domain; the durable
17:class:`~ynab_agent.workflow.overspend_ledger_workflow.OverspendLedgerWorkflow`
18wraps it as Temporal state, mirroring how ``alert.ledger`` sits under the
19failure-alert ledger and ``learn.registry`` under the rule registry.
20"""
21 
22from __future__ import annotations
23 
24from ynab_agent.budget.overspend import PriorAlert
25from ynab_agent.domain.base import Frozen
26 
27 
28class LedgerEntry(Frozen):
29    """The last alert raised for one category, and the period it belongs to.
30 
31    ``period`` is the budget month as ``"YYYY-MM"`` (the caller derives it from
32    the household clock), so a last-month entry reads as absent this month.
33    """
34 
35    category: str
36    period: str
37    alert: PriorAlert
38 
39 
40class OverspendLedgerState(Frozen):
41    """The whole dedup memory — one entry per category, carried as state.
42 
43    Held as Temporal workflow state and carried across continue-as-new, so it is
44    a frozen value like the rest of the domain.
45    """
46 
47    entries: tuple[LedgerEntry, ...] = ()
48 
49 
50def prior(
51    state: OverspendLedgerState, category: str, period: str
52) -> PriorAlert | None:
53    """The category's last alert *this period*, or ``None``. Pure.
54 
55    A stored entry from a different period reads as absent, so a new month
56    resets the dedupe and the first flag of the period always alerts (SPEC §7).
57    """
58    for entry in state.entries:
59        if entry.category == category and entry.period == period:
60            return entry.alert
61    return None
62 
63 
64def record(
65    state: OverspendLedgerState,
66    category: str,
67    period: str,
68    alert: PriorAlert,
69) -> OverspendLedgerState:
70    """Record this period's alert for ``category``, replacing any prior. Pure.
71 
72    Dropping the category's previous entry (regardless of period) keeps the tail
73    at one entry per category, so the carried state never grows with continued
74    alerting.
75    """
76    kept = tuple(entry for entry in state.entries if entry.category != category)
77    return OverspendLedgerState(
78        entries=(
79            *kept,
80            LedgerEntry(category=category, period=period, alert=alert),
81        )
82    )

The alert wording and the idempotency label — budget/message.py

The alert email is deterministic: every figure and the verdict are already decided upstream, so this is plain templating, not a model call. Keeping it in its own pure module makes the subject, body, and label unit-testable without Temporal or AgentMail. The body reads like Dining is trending over budget: $250.00 spent of $400.00, 6 days left, projected ~$500.00 by month-end.

The body, and the thread label that keys send-idempotency.

src/ynab_agent/budget/message.py · 69 lines

src/ynab_agent/budget/message.py69 lines · Python

⋯ 38 lines hidden (lines 1–38)1"""The overspend alert's wording and dedup label — pure, no model (SPEC §7).
2 
3The W6 alert is deterministic: the figures and the verdict are already decided
4by the pure projection (:mod:`ynab_agent.budget.overspend`), so the email is
5plain templating, not a model call. Kept apart from the activity glue so the
6subject, body, and label are unit-testable without Temporal or AgentMail.
7 
8The thread label is the send-idempotency key (:meth:`MailClient.open_thread`):
9it folds in the verdict and projected month-end so a *retry* of the same alert
10reuses the thread (no duplicate), while a *worsening* re-alert — the one case
11``should_alert`` lets through within a period — carries a different label and
12so opens a fresh alert thread.
13"""
14 
15from __future__ import annotations
16 
17from ynab_agent.budget.overspend import OverspendAssessment, OverspendVerdict
18 
19 
20def _status_phrase(verdict: OverspendVerdict) -> str:
21    """The human verb for the verdict (``OK`` never reaches an alert)."""
22    if verdict is OverspendVerdict.ALREADY_OVER:
23        return "already over budget"
24    return "trending over budget"
25 
26 
27def _days_phrase(days_left: int) -> str:
28    """``"6 days"`` / ``"1 day"`` — the time left in the month, not negative."""
29    days = max(days_left, 0)
30    unit = "day" if days == 1 else "days"
31    return f"{days} {unit}"
32 
33 
34def overspend_subject(assessment: OverspendAssessment) -> str:
35    """The alert thread's subject — category + how it is tracking."""
36    return f"{assessment.name}: {_status_phrase(assessment.verdict)}"
37 
38 
39def overspend_body(assessment: OverspendAssessment, days_left: int) -> str:
40    """The alert body: spend against budget, time left, month-end projection.
41 
42    e.g. ``Dining is trending over budget: $250.00 spent of $400.00, 6 days
43    left, projected ~$500.00 by month-end.`` Money renders via ``Money``'s
44    ``__str__``; ``spent``/``budgeted``/``projected`` are positive magnitudes.
45    """
46    trailer = (
47        "trending to"
48        if assessment.verdict is OverspendVerdict.ALREADY_OVER
49        else "projected"
50    )
51    return (
52        f"{assessment.name} is {_status_phrase(assessment.verdict)}: "
53        f"{assessment.spent} spent of {assessment.budgeted}, "
54        f"{_days_phrase(days_left)} left, "
55        f"{trailer} ~{assessment.projected} by month-end."
56    )
57 
58 
59def overspend_thread_label(assessment: OverspendAssessment, period: str) -> str:
60    """The per-alert idempotency label (send dedup; SPEC §7).
61 
62    Keyed on category + period + the verdict and projected it alerted at, so a
63    retry collapses onto the same thread while a materially-worse re-alert (the
64    only kind ``should_alert`` admits mid-period) gets a new one.
65    """
66    return (
67        f"yaspend-{assessment.category}-{period}"
68        f"-{assessment.verdict.value}-{assessment.projected.milliunits}"
69    )

The durable singleton — OverspendLedgerWorkflow

The ledger is held by a long-lived singleton workflow, born on the first alert via signal-with-start and living forever through continue-as-new. It is a thin shell over the pure folds: a record signal folds an alert in, a prior query reads one back. This is the third instance of a pattern already in the codebase (RuleRegistryWorkflow over learn.registry, AlertLedgerWorkflow over alert.ledger).

Pure folds passed through the sandbox; signal folds, query reads.

src/ynab_agent/workflow/overspend_ledger_workflow.py · 61 lines

src/ynab_agent/workflow/overspend_ledger_workflow.py61 lines · Python

⋯ 28 lines hidden (lines 1–28)1"""W6 · the durable overspend-alert dedup ledger (SPEC §7).
2 
3A singleton, long-lived workflow (id :data:`OVERSPEND_LEDGER_WORKFLOW_ID`)
4holding the per-category last-alert table as Temporal state (SPEC §0.5
5derived-state). It is born on the first alert — the ``save_alert`` activity does
6a signal-with-start — and lives forever, continuing-as-new to keep its history
7bounded while carrying the table forward. A thin durable shell over the pure
8:mod:`ynab_agent.budget.ledger` folds, exactly like ``alert_ledger_workflow``
9over ``alert.ledger``:
10 
11* the ``prior`` query answers "what did we last alert this category, this
12  period?" without mutating anything;
13* the ``record`` signal folds a fired alert into the table.
14 
15The period travels in with each request, so the workflow needs no clock and
16replay is trivially deterministic.
17"""
18 
19from __future__ import annotations
20 
21from temporalio import workflow
22 
23from ynab_agent.workflow.overspend_ledger_types import (
24    LedgerParams,
25    PriorRequest,
26    RecordRequest,
27)
28 
29with workflow.unsafe.imports_passed_through():
30    from ynab_agent.budget.ledger import OverspendLedgerState, prior, record
31    from ynab_agent.budget.overspend import PriorAlert
32 
33 
34@workflow.defn
35class OverspendLedgerWorkflow:
36    """The household's one durable overspend-alert dedup table."""
37 
38    def __init__(self) -> None:
39        """Start empty; the run method adopts any carried-forward state."""
40        self._state = OverspendLedgerState()
41 
42    @workflow.run
43    async def run(self, params: LedgerParams) -> None:
44        """Hold the table, folding signals until history wants rolling."""
45        self._state = params.state
46        await workflow.wait_condition(
47            lambda: workflow.info().is_continue_as_new_suggested()
48        )
49        workflow.continue_as_new(LedgerParams(state=self._state))
50 
51    @workflow.signal
52    def record(self, request: RecordRequest) -> None:
53        """Fold a freshly-sent alert into the table (SPEC §7)."""
54        self._state = record(
55            self._state, request.category, request.period, request.alert
56        )
57 
58    @workflow.query
59    def prior(self, request: PriorRequest) -> PriorAlert | None:
60        """The category's last alert this period, for the dedupe."""
61        return prior(self._state, request.category, request.period)

Wiring the ports — monitor_activities

With the store in place, the three stubs become thin glue. load_prior_alert queries the ledger; save_alert signal-with-starts it; send_overspend_alert opens the alert's own email thread. The period and "days left" are read from real time here, in the activity, and passed into the pure query/fold — which is why the workflow stays clock-free. Heavy clients (Temporal, mail) are imported lazily inside the bodies so they never cross into the workflow sandbox.

The real-clock helpers, the load (query + hydrate), and the send + save writes.

src/ynab_agent/workflow/monitor_activities.py · 153 lines

src/ynab_agent/workflow/monitor_activities.py153 lines · Python

⋯ 29 lines hidden (lines 1–29)1"""The I/O ports of the W6 overspend monitor, as Temporal activities.
2 
3Its own module so the monitor workflow's sandbox import graph stays minimal
4(see ``poll_activities`` / ``dispatch_activities``). Heavy clients (YNAB,
5AgentMail, the Temporal client) are imported lazily inside the bodies so they
6never enter the workflow sandbox.
7 
8The per-period dedupe store is the durable
9:class:`~ynab_agent.workflow.overspend_ledger_workflow.OverspendLedgerWorkflow`:
10``load_prior_alert`` queries it and ``save_alert`` signals it, exactly as the
11W2 activities talk to the rule registry and the failure-alert ledger. The
12``period`` and "days left" are read from real time here in the activity and
13passed *into* the pure query/fold, so the workflow stays clock-free.
14"""
15 
16from __future__ import annotations
17 
18import calendar
19import datetime
20 
21from temporalio import activity
22 
23from ynab_agent.budget.overspend import (
24    CategorySpend,
25    OverspendAssessment,
26    PriorAlert,
27)
28 
29 
30def _period_of(now: datetime.datetime) -> str:
31    """The budget period as ``"YYYY-MM"`` — the dedupe's reset boundary."""
32    return now.strftime("%Y-%m")
33 
34 
35def _days_left_in_month(now: datetime.datetime) -> int:
36    """Calendar days remaining in ``now``'s month (0 on the last day)."""
37    days_in_month = calendar.monthrange(now.year, now.month)[1]
38    return days_in_month - now.day
⋯ 16 lines hidden (lines 39–54)39 
40 
41@activity.defn
42async def fetch_category_spends() -> list[CategorySpend]:
43    """Read each category's month-to-date budget figures from YNAB (§7).
44 
45    The YNAB client is imported lazily and its blocking call runs off the loop.
46    """
47    import asyncio
48 
49    from ynab_agent.ynab.client import YnabClient
50 
51    client = YnabClient.from_env()
52    return list(await asyncio.to_thread(client.category_spends))
53 
54 
55@activity.defn
56async def load_prior_alert(category_id: str) -> PriorAlert | None:
57    """Load the last alert raised for a category this period, for dedupe.
58 
59    Queries the durable ledger (the period derived from real time). Returns
60    ``None`` when the ledger has not been started yet (nothing has ever
61    alerted) — so the first flag of a period always alerts (SPEC §7).
62 
63    The client-side query decodes a pydantic model to a plain ``dict``, so the
64    result is hydrated back into a :class:`PriorAlert` here — ``should_alert``
65    reads attributes off it (the #4 dict-vs-object class of bug). We hydrate by
66    hand rather than via ``result_type`` because the result is optional and the
67    SDK types ``result_type`` as a plain ``type``, not a ``X | None`` union.
68    """
69    from temporalio.service import RPCError
70 
71    from ynab_agent.workflow.overspend_ledger_types import (
72        OVERSPEND_LEDGER_WORKFLOW_ID,
73        PriorRequest,
74    )
75    from ynab_agent.workflow.temporal_client import client
76 
77    now = datetime.datetime.now(datetime.UTC)
78    temporal = await client()
79    handle = temporal.get_workflow_handle(OVERSPEND_LEDGER_WORKFLOW_ID)
80    try:
81        raw = await handle.query(
82            "prior", PriorRequest(category=category_id, period=_period_of(now))
83        )
84    except RPCError:
85        return None
86    return None if raw is None else PriorAlert.model_validate(raw)
⋯ 2 lines hidden (lines 87–88)87 
88 
89@activity.defn
90async def send_overspend_alert(assessment: OverspendAssessment) -> None:
91    """Email the overspend alert on its own thread (SPEC §7, notify-only).
92 
93    The body is deterministic (no model): the figures and verdict are already
94    decided. ``open_thread`` is idempotent on the thread label, so a retry
95    re-finds the thread rather than re-sending; the label folds in the verdict
96    and projected so a *worsening* re-alert (the one ``should_alert`` admits
97    mid-period) opens a fresh thread.
98    """
99    import asyncio
100 
101    from ynab_agent.budget.message import (
102        overspend_body,
103        overspend_subject,
104        overspend_thread_label,
105    )
106    from ynab_agent.mail.client import MailClient
107    from ynab_agent.settings import Settings
108 
109    now = datetime.datetime.now(datetime.UTC)
110    settings = Settings()
111    mail = MailClient.from_env()
112    await asyncio.to_thread(
113        mail.open_thread,
114        inbox_id=settings.inbox,
115        to=list(settings.owners),
116        subject=overspend_subject(assessment),
117        body=overspend_body(assessment, _days_left_in_month(now)),
118        txn_label=overspend_thread_label(assessment, _period_of(now)),
119    )
120 
121 
122@activity.defn
123async def save_alert(category_id: str, alert: PriorAlert) -> None:
124    """Record this period's alert so the next pass can dedupe against it.
125 
126    Signal-with-start on the singleton ledger: the first alert creates it, every
127    later one just delivers the signal (SPEC §7) — the same shape as
128    ``feed_rule_learning`` and the failure-alert ledger.
129    """
130    from temporalio.common import WorkflowIDConflictPolicy
131 
132    from ynab_agent.workflow.overspend_ledger_types import (
133        OVERSPEND_LEDGER_WORKFLOW_ID,
134        LedgerParams,
135        RecordRequest,
136    )
137    from ynab_agent.workflow.temporal_client import client, task_queue
138 
139    now = datetime.datetime.now(datetime.UTC)
140    temporal = await client()
141    await temporal.start_workflow(
142        "OverspendLedgerWorkflow",
143        LedgerParams(),
144        id=OVERSPEND_LEDGER_WORKFLOW_ID,
145        task_queue=task_queue(),
146        id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
147        start_signal="record",
148        start_signal_args=[
149            RecordRequest(
150                category=category_id, period=_period_of(now), alert=alert
151            )
152        ],
153    )

Registration and how it's verified

The new singleton is registered in the worker (the four monitor activities were already listed; only the workflow class is added). Everything else is tests.

OverspendLedgerWorkflow joins WORKFLOWS; the count test moves 8 → 9.

src/ynab_agent/workflow/runtime.py · 88 lines

src/ynab_agent/workflow/runtime.py88 lines · Python

⋯ 40 lines hidden (lines 1–40)1"""Worker wiring: the Pydantic data converter and activity/workflow registry.
2 
3A real worker (and the workflow tests) constructs its client with
4:data:`DATA_CONVERTER` so domain models serialize through Temporal, and
5registers :data:`WORKFLOWS` and :data:`ALL_ACTIVITIES`. Tests substitute mock
6activity implementations for the stubs.
7"""
8 
9from __future__ import annotations
10 
11from typing import TYPE_CHECKING
12 
13from temporalio.contrib.pydantic import pydantic_data_converter
14 
15from ynab_agent.workflow import (
16    activities,
17    alert_activities,
18    dispatch_activities,
19    monitor_activities,
20    offer_activities,
21    poll_activities,
22    receipt_activities,
23)
24 
25if TYPE_CHECKING:
26    from collections.abc import Callable
27from ynab_agent.workflow.alert_ledger_workflow import AlertLedgerWorkflow
28from ynab_agent.workflow.dispatch_workflow import DispatchWorkflow
29from ynab_agent.workflow.monitor_workflow import OverspendMonitorWorkflow
30from ynab_agent.workflow.offer_workflow import AutonomyOfferWorkflow
31from ynab_agent.workflow.overspend_ledger_workflow import (
32    OverspendLedgerWorkflow,
33)
34from ynab_agent.workflow.poll_workflow import PollWorkflow
35from ynab_agent.workflow.receipt_workflow import ReceiptJoinWorkflow
36from ynab_agent.workflow.registry_workflow import RuleRegistryWorkflow
37from ynab_agent.workflow.txn_workflow import TransactionWorkflow
38 
39DATA_CONVERTER = pydantic_data_converter
40 
41WORKFLOWS = [
42    TransactionWorkflow,
43    PollWorkflow,
44    DispatchWorkflow,
45    ReceiptJoinWorkflow,
46    OverspendMonitorWorkflow,
47    RuleRegistryWorkflow,
48    AlertLedgerWorkflow,
49    AutonomyOfferWorkflow,
50    OverspendLedgerWorkflow,
51]
⋯ 37 lines hidden (lines 52–88)52 
53ALL_ACTIVITIES: list[Callable[..., object]] = [
54    activities.fetch_snapshot,
55    activities.enrich,
56    activities.commit_to_ynab,
57    activities.read_back,
58    activities.open_thread,
59    activities.send_thread_message,
60    activities.interpret_inbound,
61    activities.converge,
62    activities.feed_rule_learning,
63    activities.close_thread,
64    poll_activities.fetch_unapproved,
65    poll_activities.address_transaction,
66    dispatch_activities.resolve_thread,
67    dispatch_activities.resolve_offer_thread,
68    dispatch_activities.classify_inbound,
69    dispatch_activities.signal_transaction,
70    dispatch_activities.signal_offer,
71    dispatch_activities.route_receipt,
72    dispatch_activities.handle_command,
73    offer_activities.start_autonomy_offer,
74    offer_activities.open_offer_thread,
75    offer_activities.interpret_offer_reply,
76    offer_activities.accept_offer,
77    offer_activities.decline_offer,
78    receipt_activities.match_receipt,
79    receipt_activities.signal_match,
80    receipt_activities.ask_disambiguation,
81    receipt_activities.ask_no_match,
82    receipt_activities.save_receipt_status,
83    monitor_activities.fetch_category_spends,
84    monitor_activities.load_prior_alert,
85    monitor_activities.send_overspend_alert,
86    monitor_activities.save_alert,
87    alert_activities.alert_failure,
88]

Coverage tracks the change at every layer: the pure folds and the formatter/label are unit-tested; the ledger workflow has a round-trip test on the time-skipping server; and test_monitor_activities drives save_alert → load_prior_alert through a real ledger over the pydantic converter, asserting the result is a real PriorAlert (the hydration guard above). That last test pre-starts the singleton's run, because in production the ledger is always already running when a later daily pass loads from it.