What changed, and why

Each W6 monitor activity used to recompute the budget period (YYYY-MM) from its own wall-clock. So a single monitor pass that straddled UTC midnight could hand the alert thread, the dedupe ledger key, and the W7 offer id three different periods — re-forking the overspend conversation, the same orphaning #18 fixed, reached by another route. It is unreachable at the current 13:00 UTC schedule, but it is a latent footgun and cuts against SPEC §0.5 (decide clocks in the workflow, via workflow.now()).

The fix is small: read the clock once in the workflow and thread that one period into every activity, so they can never disagree.

The workflow decides the period, once

run reads workflow.now() a single time, derives the period from it (and the month-position clock too), then passes that period into each activity it calls. workflow.now() is the replay-safe clock, so the period is deterministic and identical for every step in the pass.

One now → one period (and the clock), decided in the workflow.

src/ynab_agent/workflow/monitor_workflow.py · 102 lines
src/ynab_agent/workflow/monitor_workflow.py102 lines · Python
⋯ 39 lines hidden (lines 1–39)
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
5: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():
20 from ynab_agent.budget.overspend import (
21 MonthClock,
22 OverspendVerdict,
23 PriorAlert,
24 assess,
25 should_alert,
26 )
27 from ynab_agent.workflow import balance_activities, monitor_activities
28 from ynab_agent.workflow.constants import ACTIVITY_RETRY, ACTIVITY_TIMEOUT
29 from ynab_agent.workflow.monitor_types import MonitorParams, MonitorResult
30 
31 
32@workflow.defn
33class OverspendMonitorWorkflow:
34 """One daily overspend pass across all categories."""
35 
36 @workflow.run
37 async def run(self, params: MonitorParams) -> MonitorResult:
38 """Assess every category and alert on what is off track (SPEC §7)."""
39 # Read the deterministic clock once and derive the period from it, then
40 # pass that period into every activity in the pass — so the thread, the
41 # dedupe key, and the W7 offer id can't drift across a month boundary
42 # (SPEC §0.5: clocks via ``workflow.now()``, decided in the workflow).
43 now = workflow.now()
44 period = now.strftime("%Y-%m")
45 clock = params.clock or MonthClock(
46 day_of_month=now.day,
47 days_in_month=calendar.monthrange(now.year, now.month)[1],
48 )
49 spends = await workflow.execute_activity(
50 monitor_activities.fetch_category_spends,
51 start_to_close_timeout=ACTIVITY_TIMEOUT,
52 retry_policy=ACTIVITY_RETRY,
53 )
⋯ 49 lines hidden (lines 54–102)
54 
55 alerted: list[str] = []
56 for spend in spends:
57 assessment = assess(spend, clock)
58 if assessment.verdict is OverspendVerdict.OK:
59 continue
60 prior = await workflow.execute_activity(
61 monitor_activities.load_prior_alert,
62 args=[str(spend.category), period],
63 start_to_close_timeout=ACTIVITY_TIMEOUT,
64 retry_policy=ACTIVITY_RETRY,
65 )
66 if not should_alert(assessment, prior):
67 continue
68 thread_id = await workflow.execute_activity(
69 monitor_activities.send_overspend_alert,
70 args=[assessment, period],
71 start_to_close_timeout=ACTIVITY_TIMEOUT,
72 retry_policy=ACTIVITY_RETRY,
73 )
74 await workflow.execute_activity(
75 monitor_activities.save_alert,
76 args=[
77 str(spend.category),
78 PriorAlert(
79 verdict=assessment.verdict,
80 projected=assessment.projected,
81 ),
82 period,
83 ],
84 start_to_close_timeout=ACTIVITY_TIMEOUT,
85 retry_policy=ACTIVITY_RETRY,
86 )
87 # Offer a balancing move on the same alert thread (W6→W7, §8). A
88 # fire-and-forget start: REJECT_DUPLICATE keeps it one offer per
89 # category per period, and the monitor never waits on coverage.
90 await workflow.execute_activity(
91 balance_activities.start_balance_offer,
92 args=[assessment, thread_id, period],
93 start_to_close_timeout=ACTIVITY_TIMEOUT,
94 retry_policy=ACTIVITY_RETRY,
95 )
96 alerted.append(spend.name)
97 
98 return MonitorResult(
99 categories=len(spends),
100 alerts=len(alerted),
101 alerted=tuple(alerted),
102 )

The same period flows into load, send, save, and the W7 offer.

src/ynab_agent/workflow/monitor_workflow.py · 102 lines
src/ynab_agent/workflow/monitor_workflow.py102 lines · Python
⋯ 59 lines hidden (lines 1–59)
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
5: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():
20 from ynab_agent.budget.overspend import (
21 MonthClock,
22 OverspendVerdict,
23 PriorAlert,
24 assess,
25 should_alert,
26 )
27 from ynab_agent.workflow import balance_activities, monitor_activities
28 from ynab_agent.workflow.constants import ACTIVITY_RETRY, ACTIVITY_TIMEOUT
29 from ynab_agent.workflow.monitor_types import MonitorParams, MonitorResult
30 
31 
32@workflow.defn
33class OverspendMonitorWorkflow:
34 """One daily overspend pass across all categories."""
35 
36 @workflow.run
37 async def run(self, params: MonitorParams) -> MonitorResult:
38 """Assess every category and alert on what is off track (SPEC §7)."""
39 # Read the deterministic clock once and derive the period from it, then
40 # pass that period into every activity in the pass — so the thread, the
41 # dedupe key, and the W7 offer id can't drift across a month boundary
42 # (SPEC §0.5: clocks via ``workflow.now()``, decided in the workflow).
43 now = workflow.now()
44 period = now.strftime("%Y-%m")
45 clock = params.clock or MonthClock(
46 day_of_month=now.day,
47 days_in_month=calendar.monthrange(now.year, now.month)[1],
48 )
49 spends = await workflow.execute_activity(
50 monitor_activities.fetch_category_spends,
51 start_to_close_timeout=ACTIVITY_TIMEOUT,
52 retry_policy=ACTIVITY_RETRY,
53 )
54 
55 alerted: list[str] = []
56 for spend in spends:
57 assessment = assess(spend, clock)
58 if assessment.verdict is OverspendVerdict.OK:
59 continue
60 prior = await workflow.execute_activity(
61 monitor_activities.load_prior_alert,
62 args=[str(spend.category), period],
63 start_to_close_timeout=ACTIVITY_TIMEOUT,
64 retry_policy=ACTIVITY_RETRY,
65 )
66 if not should_alert(assessment, prior):
67 continue
68 thread_id = await workflow.execute_activity(
69 monitor_activities.send_overspend_alert,
70 args=[assessment, period],
71 start_to_close_timeout=ACTIVITY_TIMEOUT,
72 retry_policy=ACTIVITY_RETRY,
73 )
74 await workflow.execute_activity(
75 monitor_activities.save_alert,
76 args=[
77 str(spend.category),
78 PriorAlert(
79 verdict=assessment.verdict,
80 projected=assessment.projected,
81 ),
82 period,
83 ],
84 start_to_close_timeout=ACTIVITY_TIMEOUT,
85 retry_policy=ACTIVITY_RETRY,
86 )
87 # Offer a balancing move on the same alert thread (W6→W7, §8). A
88 # fire-and-forget start: REJECT_DUPLICATE keeps it one offer per
89 # category per period, and the monitor never waits on coverage.
90 await workflow.execute_activity(
91 balance_activities.start_balance_offer,
92 args=[assessment, thread_id, period],
93 start_to_close_timeout=ACTIVITY_TIMEOUT,
94 retry_policy=ACTIVITY_RETRY,
⋯ 8 lines hidden (lines 95–102)
95 )
96 alerted.append(spend.name)
97 
98 return MonitorResult(
99 categories=len(spends),
100 alerts=len(alerted),
101 alerted=tuple(alerted),
102 )

The activities take the period as input

Each activity now accepts period instead of reading the clock for it. The _period_of helper and the per-activity datetime.now() reads are gone; only the cosmetic "days left" still reads the wall-clock at send time. Shown here on load_prior_alert; send_overspend_alert, save_alert, and start_balance_offer change the same way.

period is a parameter; the ledger query uses it verbatim.

src/ynab_agent/workflow/monitor_activities.py · 158 lines
src/ynab_agent/workflow/monitor_activities.py158 lines · Python
⋯ 51 lines hidden (lines 1–51)
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`` (``"YYYY-MM"``) is computed once from the workflow's deterministic
13clock and passed *into* every activity in a pass, so they can never disagree on
14it across a month boundary (the thread, the dedupe key, and the W7 offer id all
15derive from it). "Days left" is cosmetic and still read at send time.
16"""
17 
18from __future__ import annotations
19 
20import calendar
21import datetime
22 
23from temporalio import activity
24 
25from ynab_agent.budget.overspend import (
26 CategorySpend,
27 OverspendAssessment,
28 PriorAlert,
30 
31 
32def _days_left_in_month(now: datetime.datetime) -> int:
33 """Calendar days remaining in ``now``'s month (0 on the last day)."""
34 days_in_month = calendar.monthrange(now.year, now.month)[1]
35 return days_in_month - now.day
36 
37 
38@activity.defn
39async def fetch_category_spends() -> list[CategorySpend]:
40 """Read each category's month-to-date budget figures from YNAB (§7).
41 
42 The YNAB client is imported lazily and its blocking call runs off the loop.
43 """
44 import asyncio
45 
46 from ynab_agent.ynab.client import YnabClient
47 
48 client = YnabClient.from_env()
49 return list(await asyncio.to_thread(client.category_spends))
50 
51 
52@activity.defn
53async def load_prior_alert(category_id: str, period: str) -> PriorAlert | None:
54 """Load the last alert raised for a category this period, for dedupe.
55 
56 Queries the durable ledger for ``period`` (supplied by the workflow from its
57 deterministic clock, so it matches the period every other activity in the
58 pass uses). Returns ``None`` when the ledger has not been started yet
59 (nothing has ever alerted) — so the first flag of a period always alerts
60 (SPEC §7).
61 
62 The client-side query decodes a pydantic model to a plain ``dict``, so the
63 result is hydrated back into a :class:`PriorAlert` here — ``should_alert``
64 reads attributes off it (the #4 dict-vs-object class of bug). We hydrate by
65 hand rather than via ``result_type`` because the result is optional and the
66 SDK types ``result_type`` as a plain ``type``, not a ``X | None`` union.
67 """
68 from temporalio.service import RPCError
69 
70 from ynab_agent.workflow.overspend_ledger_types import (
71 OVERSPEND_LEDGER_WORKFLOW_ID,
72 PriorRequest,
73 )
74 from ynab_agent.workflow.temporal_client import client
75 
76 temporal = await client()
77 handle = temporal.get_workflow_handle(OVERSPEND_LEDGER_WORKFLOW_ID)
78 try:
79 raw = await handle.query(
80 "prior", PriorRequest(category=category_id, period=period)
81 )
82 except RPCError:
83 return None
84 return None if raw is None else PriorAlert.model_validate(raw)
⋯ 74 lines hidden (lines 85–158)
85 
86 
87@activity.defn
88async def send_overspend_alert(
89 assessment: OverspendAssessment, period: str
90) -> str:
91 """Email the overspend alert and return its thread id (SPEC §7, §8).
92 
93 The body is deterministic (no model): the figures and verdict are already
94 decided. ``period`` is supplied by the workflow (from its deterministic
95 clock), so the thread and dedupe labels match the rest of the pass.
96 ``alert_on_thread`` keeps one thread per overspend: the first alert opens
97 it, a *worsening* re-alert (the one ``should_alert`` admits mid-period)
98 replies an update on that same thread, and a retry of either re-sends
99 nothing (idempotent on the update label). The id it returns is stable for
100 the period: what W7 replies on to offer a balancing move, and where a
101 worsening re-alert lands too, so the conversation never forks (the W6→W7
102 tie).
103 """
104 import asyncio
105 
106 from ynab_agent.budget.message import (
107 overspend_body,
108 overspend_subject,
109 overspend_thread_label,
110 overspend_update_label,
111 )
112 from ynab_agent.mail.client import MailClient
113 from ynab_agent.settings import Settings
114 
115 now = datetime.datetime.now(datetime.UTC)
116 settings = Settings()
117 mail = MailClient.from_env()
118 return await asyncio.to_thread(
119 mail.alert_on_thread,
120 inbox_id=settings.inbox,
121 to=list(settings.owners),
122 subject=overspend_subject(assessment),
123 body=overspend_body(assessment, _days_left_in_month(now)),
124 thread_label=overspend_thread_label(assessment, period),
125 update_label=overspend_update_label(assessment, period),
126 )
127 
128 
129@activity.defn
130async def save_alert(category_id: str, alert: PriorAlert, period: str) -> None:
131 """Record this period's alert so the next pass can dedupe against it.
132 
133 Signal-with-start on the singleton ledger: the first alert creates it, every
134 later one just delivers the signal (SPEC §7) — the same shape as
135 ``feed_rule_learning`` and the failure-alert ledger. ``period`` is supplied
136 by the workflow so the record matches what ``load_prior_alert`` queried.
137 """
138 from temporalio.common import WorkflowIDConflictPolicy
139 
140 from ynab_agent.workflow.overspend_ledger_types import (
141 OVERSPEND_LEDGER_WORKFLOW_ID,
142 LedgerParams,
143 RecordRequest,
144 )
145 from ynab_agent.workflow.temporal_client import client, task_queue
146 
147 temporal = await client()
148 await temporal.start_workflow(
149 "OverspendLedgerWorkflow",
150 LedgerParams(),
151 id=OVERSPEND_LEDGER_WORKFLOW_ID,
152 task_queue=task_queue(),
153 id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
154 start_signal="record",
155 start_signal_args=[
156 RecordRequest(category=category_id, period=period, alert=alert)
157 ],
158 )

Test: one period reaches every activity

The monitor test captures the period each of the four activities received and asserts a single, well-formed value reached all of them — the discriminating guard for "one period per pass." On the old code these were four independent wall-clock reads; the assertion pins them to one.

len == 4, one distinct value, matches YYYY-MM.

tests/workflow/test_monitor_workflow.py · 169 lines
tests/workflow/test_monitor_workflow.py169 lines · Python
⋯ 126 lines hidden (lines 1–126)
1"""End-to-end tests for the W6 overspend monitor (time-skipping server)."""
2 
3from __future__ import annotations
4 
5from typing import TYPE_CHECKING
6 
7from temporalio import activity
8from temporalio.testing import WorkflowEnvironment
9from temporalio.worker import Worker
10 
11from ynab_agent.budget.overspend import (
12 CategorySpend,
13 MonthClock,
14 OverspendAssessment,
15 OverspendVerdict,
16 PriorAlert,
18from ynab_agent.domain.ids import CategoryId
19from ynab_agent.domain.money import Money
20from ynab_agent.workflow.monitor_types import MonitorParams, MonitorResult
21from ynab_agent.workflow.monitor_workflow import OverspendMonitorWorkflow
22from ynab_agent.workflow.runtime import DATA_CONVERTER
23 
24if TYPE_CHECKING:
25 from collections.abc import Callable
26 
27TASK_QUEUE = "ynab-monitor-test"
28# Mid-month: a run-rate doubles the month-to-date spend.
29_CLOCK = MonthClock(day_of_month=15, days_in_month=30)
30 
31 
32def _spend(name: str, *, budgeted: str, activity: str) -> CategorySpend:
33 return CategorySpend(
34 category=CategoryId(name),
35 name=name,
36 budgeted=Money.from_currency(budgeted),
37 activity=Money.from_currency(activity),
38 balance=Money.from_currency(budgeted) + Money.from_currency(activity),
39 )
40 
41 
42def _activities(
43 *,
44 spends: list[CategorySpend],
45 prior: PriorAlert | None,
46 sent: list[str],
47 offered: list[str],
48 periods: list[str],
49) -> list[Callable[..., object]]:
50 @activity.defn(name="fetch_category_spends")
51 async def fetch_category_spends() -> list[CategorySpend]:
52 return spends
53 
54 @activity.defn(name="load_prior_alert")
55 async def load_prior_alert(
56 category_id: str, period: str
57 ) -> PriorAlert | None:
58 periods.append(period)
59 return prior
60 
61 @activity.defn(name="send_overspend_alert")
62 async def send_overspend_alert(
63 assessment: OverspendAssessment, period: str
64 ) -> str:
65 periods.append(period)
66 sent.append(assessment.name)
67 return f"thr-{assessment.category}"
68 
69 @activity.defn(name="save_alert")
70 async def save_alert(
71 category_id: str, alert: PriorAlert, period: str
72 ) -> None:
73 periods.append(period)
74 
75 @activity.defn(name="start_balance_offer")
76 async def start_balance_offer(
77 assessment: OverspendAssessment, thread_id: str, period: str
78 ) -> None:
79 periods.append(period)
80 offered.append(thread_id)
81 
82 return [
83 fetch_category_spends,
84 load_prior_alert,
85 send_overspend_alert,
86 save_alert,
87 start_balance_offer,
88 ]
89 
90 
91async def _run(
92 *,
93 wf_id: str,
94 spends: list[CategorySpend],
95 prior: PriorAlert | None = None,
96) -> tuple[MonitorResult, list[str], list[str], list[str]]:
97 sent: list[str] = []
98 offered: list[str] = []
99 periods: list[str] = []
100 acts = _activities(
101 spends=spends,
102 prior=prior,
103 sent=sent,
104 offered=offered,
105 periods=periods,
106 )
107 async with (
108 await WorkflowEnvironment.start_time_skipping(
109 data_converter=DATA_CONVERTER
110 ) as env,
111 Worker(
112 env.client,
113 task_queue=TASK_QUEUE,
114 workflows=[OverspendMonitorWorkflow],
115 activities=acts,
116 ),
117 ):
118 result = await env.client.execute_workflow(
119 OverspendMonitorWorkflow.run,
120 MonitorParams(clock=_CLOCK),
121 id=wf_id,
122 task_queue=TASK_QUEUE,
123 )
124 return result, sent, offered, periods
125 
126 
127async def test_overspending_category_is_alerted() -> None:
128 # $250 of $400 at mid-month → projects ~$500, trending over.
129 result, sent, offered, periods = await _run(
130 wf_id="mon-alert",
131 spends=[_spend("Dining", budgeted="400", activity="-250")],
132 )
133 assert result.alerts == 1
134 assert result.alerted == ("Dining",)
135 assert sent == ["Dining"]
136 # The alert hands its thread to a balancing offer (W6→W7, §8).
137 assert offered == ["thr-Dining"]
138 # One period, computed once in the workflow, reaches every activity in the
139 # pass (load + send + save + offer) — no per-activity wall-clock drift.
140 import re
141 
142 assert len(periods) == 4
143 assert len(set(periods)) == 1
144 assert re.fullmatch(r"\d{4}-\d{2}", periods[0])
145 
⋯ 24 lines hidden (lines 146–169)
146 
147async def test_on_track_category_is_silent() -> None:
148 result, sent, offered, _ = await _run(
149 wf_id="mon-ok",
150 spends=[_spend("Gas", budgeted="400", activity="-100")],
151 )
152 assert result.alerts == 0
153 assert sent == []
154 assert offered == []
155 
156 
157async def test_duplicate_alert_is_suppressed() -> None:
158 # An identical prior alert (same projection) → deduped, no send.
159 spends = [_spend("Dining", budgeted="400", activity="-250")]
160 prior = PriorAlert(
161 verdict=OverspendVerdict.TRENDING_OVER,
162 projected=Money.from_currency("500"),
163 )
164 result, sent, offered, _ = await _run(
165 wf_id="mon-dedupe", spends=spends, prior=prior
166 )
167 assert result.alerts == 0
168 assert sent == []
169 assert offered == []