One thread per overspend — a code-level reading

What changed, and why

PR #17 put the W7 balancer's coverage offer on the W6 overspend-alert thread, so the alert and its offer were one conversation. One gap remained: a worsening mid-period re-alert still opened a brand-new thread. The balancer runs one workflow per category-period (REJECT_DUPLICATE) and stays indexed by the first alert's thread, so a reply on the new thread had no BalanceThreadId to match and was orphaned — misrouted to the command path. The conversation forked again.

This change makes one thread per overspend, full stop. A worsening re-alert now replies an update on the existing alert thread instead of opening a new one, so the balancer's BalanceThreadId always matches and a reply always routes back. The fix is a label split in the pure layer, one new mail method, and a one-line change at the monitor's send site. W7 itself is untouched.

Split the label: one stable thread, per-alert dedup

The old single label folded the verdict and projected month-end into the thread key, so each worsening got a different key and a different thread. The fix splits it. overspend_thread_label is now stable for a category-period, so every alert and re-alert resolves to the same thread. A new overspend_update_label keeps the verdict and projected, and becomes the per-alert send-dedup key — distinct namespaces (yaspend- vs yaspend-update-) so the two never collide on the shared thread.

The thread key no longer varies; the update key carries the dedup.

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

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

⋯ 58 lines hidden (lines 1–58)1"""The overspend alert's wording and dedup labels — 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 labels are unit-testable without Temporal or AgentMail.
7 
8Two labels, two jobs (:meth:`MailClient.alert_on_thread`). The *thread* label is
9stable for a category-period, so every alert and re-alert lands on one email
10thread (with the W7 offer on it): the overspend is one conversation. The
11*update* label folds in the verdict and projected month-end, so it is the
12per-alert send-dedup key. A *retry* of the same alert collapses (no duplicate);
13a *worsening* re-alert (the one ``should_alert`` lets through mid-period)
14carries a new update label and posts a fresh update on that thread.
15"""
16 
17from __future__ import annotations
18 
19from ynab_agent.budget.overspend import OverspendAssessment, OverspendVerdict
20 
21 
22def _status_phrase(verdict: OverspendVerdict) -> str:
23    """The human verb for the verdict (``OK`` never reaches an alert)."""
24    if verdict is OverspendVerdict.ALREADY_OVER:
25        return "already over budget"
26    return "trending over budget"
27 
28 
29def _days_phrase(days_left: int) -> str:
30    """``"6 days"`` / ``"1 day"`` — the time left in the month, not negative."""
31    days = max(days_left, 0)
32    unit = "day" if days == 1 else "days"
33    return f"{days} {unit}"
34 
35 
36def overspend_subject(assessment: OverspendAssessment) -> str:
37    """The alert thread's subject — category + how it is tracking."""
38    return f"{assessment.name}: {_status_phrase(assessment.verdict)}"
39 
40 
41def overspend_body(assessment: OverspendAssessment, days_left: int) -> str:
42    """The alert body: spend against budget, time left, month-end projection.
43 
44    e.g. ``Dining is trending over budget: $250.00 spent of $400.00, 6 days
45    left, projected ~$500.00 by month-end.`` Money renders via ``Money``'s
46    ``__str__``; ``spent``/``budgeted``/``projected`` are positive magnitudes.
47    """
48    trailer = (
49        "trending to"
50        if assessment.verdict is OverspendVerdict.ALREADY_OVER
51        else "projected"
52    )
53    return (
54        f"{assessment.name} is {_status_phrase(assessment.verdict)}: "
55        f"{assessment.spent} spent of {assessment.budgeted}, "
56        f"{_days_phrase(days_left)} left, "
57        f"{trailer} ~{assessment.projected} by month-end."
58    )
59 
60 
61def overspend_thread_label(assessment: OverspendAssessment, period: str) -> str:
62    """The category's alert-thread key for the period (SPEC §7).
63 
64    Stable across the period — independent of verdict and projected — so every
65    alert and re-alert for a category lands on the *same* email thread (and the
66    W7 balancer's offer on it): one conversation per overspend. The per-alert
67    send dedup lives in :func:`overspend_update_label`, not here.
68    """
69    return f"yaspend-{assessment.category}-{period}"
70 
71 
72def overspend_update_label(assessment: OverspendAssessment, period: str) -> str:
73    """The per-alert send-dedup label (SPEC §7).
74 
75    Keyed on the verdict and projected the alert fired at, so a retry of the
76    same alert collapses (no duplicate) while a materially-worse re-alert (the
77    only kind ``should_alert`` admits mid-period) carries a new label and so
78    posts a fresh update on the thread.
79    """
80    return (
81        f"yaspend-update-{assessment.category}-{period}"
82        f"-{assessment.verdict.value}-{assessment.projected.milliunits}"
83    )

alert_on_thread: open once, then reply updates

The new mail method carries the open-or-reply logic. First alert: no thread exists for the stable label, so it opens one — tagging the opening message with both the thread label and the first update label, so a retry of that first alert finds the update label already on record and re-sends nothing. A later re-alert: the thread exists, so it replies the update on it (addressed to the owners, because the thread's latest message is the agent's own), deduped on the update label. Either way it returns the one stable thread id.

find(thread) → open with both labels, or reply the update (dedup on update_label).

src/ynab_agent/mail/client.py · 320 lines

src/ynab_agent/mail/client.py320 lines · Python

⋯ 210 lines hidden (lines 1–210)1"""AgentMail: the email plumbing for the per-transaction threads (SPEC §5).
2 
3A thin, deterministic client — sending and replying are direct API calls, not
4agentic. One inbox is the agent's address; each transaction is one thread within
5it (started by the first send, continued by replies). The model composes the
6prose elsewhere; this module only puts it on the wire.
7 
8The backend is a small :class:`MailBackend` protocol over *our* two operations
9(send a new thread / reply on one), so tests inject a fake and strict mypy never
10has to reason about the AgentMail SDK's surface. :func:`MailClient.from_env`
11wires the real SDK behind that protocol, reading ``AGENTMAIL_API_KEY``.
12"""
13 
14from __future__ import annotations
15 
16import os
17from typing import TYPE_CHECKING, Protocol
18 
19from pydantic import model_validator
20 
21from ynab_agent.domain.base import Frozen
22 
23if TYPE_CHECKING:
  from agentmail import AgentMail
25 
26_API_KEY_ENV = "AGENTMAIL_API_KEY"
27# Every agent message carries this label, plus per-transaction / per-action
28# labels that serve as idempotency keys (no separate store — derived from the
29# AgentMail thread, SPEC §5).
30_AGENT_LABEL = "ynab-agent"
31 
32 
33class SentEmail(Frozen):
  """The identifiers AgentMail returns for a sent message."""
35 
  message_id: str
  thread_id: str
38 
39 
40class OutboundEmail(Frozen):
  """One message to send: a new thread, or a reply continuing one."""
42 
  inbox_id: str
  subject: str
  text: str
  to: tuple[str, ...] = ()
  reply_to_message_id: str | None = None
48 
  @model_validator(mode="after")
  def _check_addressing(self) -> OutboundEmail:
      if self.reply_to_message_id is None and not self.to:
          msg = "a new-thread email needs at least one recipient"
          raise ValueError(msg)
      return self
55 
56 
57class MailBackend(Protocol):
  """The mail operations the client needs (implemented over AgentMail)."""
59 
  def send_new(
      self,
      inbox_id: str,
      to: list[str],
      subject: str,
      text: str,
      labels: list[str] | None = None,
  ) -> SentEmail:
      """Start a new thread."""
      ...
70 
  def send_reply(
      self,
      inbox_id: str,
      message_id: str,
      text: str,
      labels: list[str] | None = None,
      to: list[str] | None = None,
  ) -> SentEmail:
      """Reply on an existing thread.
80 
      ``to`` overrides the recipients. AgentMail otherwise addresses a reply
      to the *sender* of ``message_id``; when that message is the agent's own
      (replying on a thread the agent last spoke on), the reply loops back to
      the agent and the owner never sees it. Pass the owners to deliver it.
      """
      ...
87 
  def find_thread(self, inbox_id: str, label: str) -> str | None:
      """The thread id of the first message carrying ``label``, or None.
90 
      Labels are the agent's idempotency keys: a per-transaction label lets
      ``open_thread`` find an already-opened thread instead of duplicating it.
      """
      ...
95 
  def latest_message_id(self, inbox_id: str, thread_id: str) -> str | None:
      """The newest message id on a thread (the target to reply to)."""
      ...
99 
  def archive(self, inbox_id: str, thread_id: str) -> None:
      """Mark a thread closed (an ``ynab-archived`` label)."""
      ...
103 
104 
105# The process-wide cached client built by ``from_env`` (see its docstring).
106# ``tests/conftest.py`` resets this between tests.
107_CACHED: MailClient | None = None
108 
109 
110class MailClient:
  """Sends a transaction's outbound mail through AgentMail (SPEC §5)."""
112 
  def __init__(self, backend: MailBackend) -> None:
      """Wrap a backend (the real AgentMail adapter, or a test fake)."""
      self._backend = backend
116 
  @classmethod
  def from_env(cls) -> MailClient:
      """Build (once) a client backed by the real AgentMail SDK.
120 
      Cached like the YNAB client: the instrumented httpx client is built once
      and reused, so a per-call client (never closed) can't leak sockets.
      Tests reset the cache (see ``tests/conftest.py``).
124 
      Raises:
          RuntimeError: If ``AGENTMAIL_API_KEY`` is not set.
      """
      global _CACHED
      if _CACHED is not None:
          return _CACHED
      key = os.environ.get(_API_KEY_ENV)
      if not key:
          msg = f"{_API_KEY_ENV} is not set"
          raise RuntimeError(msg)
      import httpx
      from agentmail import AgentMail
137 
      from ynab_agent.telemetry import instrument_httpx
139 
      # Inject our own httpx client so trace context propagates on mail sends
      # (the SDK accepts one); instrument it the same way the YNAB client is.
      http_client = httpx.Client(timeout=60.0)
      instrument_httpx(http_client)
      _CACHED = cls(
          _AgentMailBackend(AgentMail(api_key=key, httpx_client=http_client))
      )
      return _CACHED
148 
  def send(self, email: OutboundEmail) -> SentEmail:
      """Send a new thread, or a reply if a message to reply to is set."""
      if email.reply_to_message_id is not None:
          return self._backend.send_reply(
              email.inbox_id,
              email.reply_to_message_id,
              email.text,
              to=list(email.to) or None,
          )
      return self._backend.send_new(
          email.inbox_id, list(email.to), email.subject, email.text
      )
161 
  def open_thread(
      self,
      *,
      inbox_id: str,
      to: list[str],
      subject: str,
      body: str,
      txn_label: str,
  ) -> str:
      """Open the transaction's thread (idempotent); return its id.
172 
      If a thread already carries ``txn_label`` (a retry of this open), return
      it rather than starting a duplicate; otherwise send the opening message.
      """
      existing = self._backend.find_thread(inbox_id, txn_label)
      if existing is not None:
          return existing
      sent = self._backend.send_new(
          inbox_id, to, subject, body, labels=[_AGENT_LABEL, txn_label]
      )
      return sent.thread_id
183 
  def send_on_thread(
      self,
      *,
      inbox_id: str,
      thread_id: str,
      body: str,
      seq_label: str,
      to: list[str] | None = None,
  ) -> bool:
      """Reply on a thread (idempotent on ``seq_label``); True if sent.
194 
      Skips if a message with ``seq_label`` is already on record, so a retry
      never double-sends (SPEC §3 outbound dedup). ``to`` overrides the
      recipients: when the thread's latest message is the agent's own (e.g.
      the W6 alert the balancer replies on), AgentMail would address the reply
      back to the agent, so pass the owners to deliver it to them (SPEC §8).
      """
      if self._backend.find_thread(inbox_id, seq_label) is not None:
          return False
      target = self._backend.latest_message_id(inbox_id, thread_id)
      if target is None:
          return False
      self._backend.send_reply(
          inbox_id, target, body, labels=[_AGENT_LABEL, seq_label], to=to
      )
      return True
210 
  def alert_on_thread(
      self,
      *,
      inbox_id: str,
      to: list[str],
      subject: str,
      body: str,
      thread_label: str,
      update_label: str,
  ) -> str:
      """Open the alert thread, or reply a worsening update on it (SPEC §7).
222 
      One thread per overspend, keyed by the stable ``thread_label``. The
      first alert opens it (``body`` as the opening message), tagged with both
      the thread label and this ``update_label``. A later re-alert replies the
      update on that same thread (so the overspend stays one conversation and
      the W7 offer routes back unbroken), deduped on ``update_label`` so a
      retry never double-posts. It is addressed to ``to`` because the thread's
      latest message is the agent's own, so AgentMail needs the recipients
      spelled out (the same reason the balancer does; SPEC §8).
      """
      existing = self._backend.find_thread(inbox_id, thread_label)
      if existing is None:
          sent = self._backend.send_new(
              inbox_id,
              to,
              subject,
              body,
              labels=[_AGENT_LABEL, thread_label, update_label],
          )
          return sent.thread_id
      if self._backend.find_thread(inbox_id, update_label) is None:
          # ``existing`` came from a labelled message, so the thread is
          # non-empty and ``target`` is its latest message; the None guard
          # only satisfies the type (an empty thread cannot reach here).
          target = self._backend.latest_message_id(inbox_id, existing)
          if target is not None:
              self._backend.send_reply(
                  inbox_id,
                  target,
                  body,
                  labels=[_AGENT_LABEL, update_label],
                  to=to,
              )
      return existing
⋯ 65 lines hidden (lines 256–320)256 
  def close(self, *, inbox_id: str, thread_id: str) -> None:
      """Mark the transaction's thread closed."""
      self._backend.archive(inbox_id, thread_id)
260 
261 
262class _AgentMailBackend:
  """Adapts the AgentMail SDK to the :class:`MailBackend` protocol."""
264 
  def __init__(self, client: AgentMail) -> None:
      self._client = client
267 
  def send_new(
      self,
      inbox_id: str,
      to: list[str],
      subject: str,
      text: str,
      labels: list[str] | None = None,
  ) -> SentEmail:
      result = self._client.inboxes.messages.send(
          inbox_id, to=to, subject=subject, text=text, labels=labels
      )
      return SentEmail(
          message_id=result.message_id, thread_id=result.thread_id
      )
282 
  def send_reply(
      self,
      inbox_id: str,
      message_id: str,
      text: str,
      labels: list[str] | None = None,
      to: list[str] | None = None,
  ) -> SentEmail:
      # ``to`` overrides the recipients; omit it entirely (not None) when
      # unset so AgentMail keeps its own reply-addressing default.
      if to is not None:
          result = self._client.inboxes.messages.reply(
              inbox_id, message_id, text=text, labels=labels, to=to
          )
      else:
          result = self._client.inboxes.messages.reply(
              inbox_id, message_id, text=text, labels=labels
          )
      return SentEmail(
          message_id=result.message_id, thread_id=result.thread_id
      )
304 
  def find_thread(self, inbox_id: str, label: str) -> str | None:
      result = self._client.inboxes.messages.list(
          inbox_id, labels=[label], limit=1
      )
      messages = result.messages
      return messages[0].thread_id if messages else None
311 
  def latest_message_id(self, inbox_id: str, thread_id: str) -> str | None:
      thread = self._client.inboxes.threads.get(inbox_id, thread_id)
      message_id: str | None = thread.last_message_id
      return message_id
316 
  def archive(self, inbox_id: str, thread_id: str) -> None:
      self._client.inboxes.threads.update(
          inbox_id, thread_id, add_labels=["ynab-archived"]
      )

The monitor sends through it

send_overspend_alert now calls alert_on_thread with both labels and returns the stable thread id — what W7 replies on, and where every worsening re-alert lands too. The alert dedupe (whether to fire at all) is unchanged: it still lives in should_alert and the durable ledger, separate from the email labels. This change only moved where a re-alert lands, not when one fires.

One send site, both labels, stable id back.

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

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

⋯ 89 lines hidden (lines 1–89)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
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)
87 
88 
89@activity.defn
90async def send_overspend_alert(assessment: OverspendAssessment) -> 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. ``alert_on_thread`` keeps one thread per overspend: the first
95    alert opens it, a *worsening* re-alert (the one ``should_alert`` admits
96    mid-period) replies an update on that same thread, and a retry of either
97    re-sends nothing (idempotent on the update label). The id it returns is
98    stable for the period: what W7 replies on to offer a balancing move, and
99    where a worsening re-alert lands too, so the conversation never forks (the
100    W6→W7 tie).
101    """
102    import asyncio
103 
104    from ynab_agent.budget.message import (
105        overspend_body,
106        overspend_subject,
107        overspend_thread_label,
108        overspend_update_label,
109    )
110    from ynab_agent.mail.client import MailClient
111    from ynab_agent.settings import Settings
112 
113    now = datetime.datetime.now(datetime.UTC)
114    period = _period_of(now)
115    settings = Settings()
116    mail = MailClient.from_env()
117    return await asyncio.to_thread(
118        mail.alert_on_thread,
119        inbox_id=settings.inbox,
120        to=list(settings.owners),
121        subject=overspend_subject(assessment),
122        body=overspend_body(assessment, _days_left_in_month(now)),
123        thread_label=overspend_thread_label(assessment, period),
124        update_label=overspend_update_label(assessment, period),
125    )
⋯ 34 lines hidden (lines 126–159)126 
127 
128@activity.defn
129async def save_alert(category_id: str, alert: PriorAlert) -> None:
130    """Record this period's alert so the next pass can dedupe against it.
131 
132    Signal-with-start on the singleton ledger: the first alert creates it, every
133    later one just delivers the signal (SPEC §7) — the same shape as
134    ``feed_rule_learning`` and the failure-alert ledger.
135    """
136    from temporalio.common import WorkflowIDConflictPolicy
137 
138    from ynab_agent.workflow.overspend_ledger_types import (
139        OVERSPEND_LEDGER_WORKFLOW_ID,
140        LedgerParams,
141        RecordRequest,
142    )
143    from ynab_agent.workflow.temporal_client import client, task_queue
144 
145    now = datetime.datetime.now(datetime.UTC)
146    temporal = await client()
147    await temporal.start_workflow(
148        "OverspendLedgerWorkflow",
149        LedgerParams(),
150        id=OVERSPEND_LEDGER_WORKFLOW_ID,
151        task_queue=task_queue(),
152        id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
153        start_signal="record",
154        start_signal_args=[
155            RecordRequest(
156                category=category_id, period=_period_of(now), alert=alert
157            )
158        ],
159    )

Tests, and the edges left documented

The regression guard is at the mail layer: open the thread, then a worsening re-alert (new update label, same thread label), and assert both land on the same thread id and the reply is addressed to the owner. Companion tests pin the label semantics (thread stable, update varies, namespaces distinct) and the dedup on both the open and the reply.

first == second: one conversation across the re-alert.

tests/mail/test_mail_client.py · 382 lines

tests/mail/test_mail_client.py382 lines · Python

⋯ 223 lines hidden (lines 1–223)1"""Tests for the AgentMail client (SPEC §5).
2 
3Offline tests inject a fake backend (no network). One opt-in live test sends a
4real email through AgentMail and is skipped unless ``YNAB_AGENT_LIVE_EMAIL`` is
5set, so the gate stays offline.
6 
7The fake backend models the one property the client relies on for store-free
8idempotency: labels are sticky per thread, so ``find_thread(label)`` re-finds an
9already-opened thread (open dedup) or an already-sent action (send dedup).
10"""
11 
12from __future__ import annotations
13 
14import os
15 
16import pytest
17 
18from ynab_agent.mail.client import MailClient, OutboundEmail, SentEmail
19 
20 
21class _FakeBackend:
  """An in-memory AgentMail: threads, sticky labels, last-message tracking."""
23 
  def __init__(self) -> None:
      self.calls: list[tuple[str, ...]] = []
      self.reply_tos: list[list[str] | None] = []  # `to` per send_reply call
      self._labels: dict[str, str] = {}  # label -> thread_id (first to carry)
      self._last: dict[str, str] = {}  # thread_id -> last message_id
      self._msg_thread: dict[str, str] = {}  # message_id -> thread_id
      self._archived: set[str] = set()
      self._seq = 0
32 
  def _next(self, prefix: str) -> str:
      self._seq += 1
      return f"{prefix}{self._seq}"
36 
  def _register(self, thread_id: str, labels: list[str] | None) -> None:
      for label in labels or []:
          self._labels.setdefault(label, thread_id)
40 
  def send_new(
      self,
      inbox_id: str,
      to: list[str],
      subject: str,
      text: str,
      labels: list[str] | None = None,
  ) -> SentEmail:
      self.calls.append(("new", inbox_id, ",".join(to), subject, text))
      thread_id = self._next("t")
      message_id = self._next("m")
      self._last[thread_id] = message_id
      self._msg_thread[message_id] = thread_id
      self._register(thread_id, labels)
      return SentEmail(message_id=message_id, thread_id=thread_id)
56 
  def send_reply(
      self,
      inbox_id: str,
      message_id: str,
      text: str,
      labels: list[str] | None = None,
      to: list[str] | None = None,
  ) -> SentEmail:
      self.calls.append(("reply", inbox_id, message_id, text))
      self.reply_tos.append(to)
      thread_id = self._msg_thread.get(message_id, "t1")
      new_id = self._next("m")
      self._last[thread_id] = new_id
      self._msg_thread[new_id] = thread_id
      self._register(thread_id, labels)
      return SentEmail(message_id=new_id, thread_id=thread_id)
73 
  def find_thread(self, inbox_id: str, label: str) -> str | None:
      return self._labels.get(label)
76 
  def latest_message_id(self, inbox_id: str, thread_id: str) -> str | None:
      return self._last.get(thread_id)
79 
  def archive(self, inbox_id: str, thread_id: str) -> None:
      self.calls.append(("archive", inbox_id, thread_id))
      self._archived.add(thread_id)
83 
84 
85def test_new_thread_routes_to_send_new() -> None:
  backend = _FakeBackend()
  out = MailClient(backend).send(
      OutboundEmail(
          inbox_id="ib",
          to=("wife@example.com",),
          subject="[YNAB] $4.50",
          text="best guess: Dining",
      )
  )
  assert out.thread_id
  assert backend.calls[0][0] == "new"
  assert backend.calls[0][3] == "[YNAB] $4.50"
98 
99 
100def test_reply_routes_to_send_reply() -> None:
  backend = _FakeBackend()
  out = MailClient(backend).send(
      OutboundEmail(
          inbox_id="ib",
          subject="[YNAB] $4.50",
          text="applied, thanks",
          reply_to_message_id="m0",
      )
  )
  assert out.message_id
  assert backend.calls[0] == ("reply", "ib", "m0", "applied, thanks")
112 
113 
114def test_new_thread_without_recipients_is_rejected() -> None:
  with pytest.raises(ValueError, match="recipient"):
      OutboundEmail(inbox_id="ib", subject="x", text="y")
117 
118 
119def test_open_thread_sends_once_then_dedups() -> None:
  backend = _FakeBackend()
  client = MailClient(backend)
  first = client.open_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="[YNAB] $4.50",
      body="best guess: Dining",
      txn_label="yatxn-abc",
  )
  second = client.open_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="[YNAB] $4.50",
      body="best guess: Dining",
      txn_label="yatxn-abc",
  )
  assert first == second
  # Only the first call actually sent a new thread (idempotent open).
  assert [c[0] for c in backend.calls] == ["new"]
139 
140 
141def test_send_on_thread_dedups_on_seq_label() -> None:
  backend = _FakeBackend()
  client = MailClient(backend)
  thread_id = client.open_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="[YNAB] $4.50",
      body="proposal",
      txn_label="yatxn-abc",
  )
  sent_first = client.send_on_thread(
      inbox_id="ib",
      thread_id=thread_id,
      body="a reminder",
      seq_label="yaseq-abc-2",
  )
  sent_again = client.send_on_thread(
      inbox_id="ib",
      thread_id=thread_id,
      body="a reminder",
      seq_label="yaseq-abc-2",
  )
  assert sent_first is True
  assert sent_again is False
  assert [c[0] for c in backend.calls].count("reply") == 1
166 
167 
168def test_send_on_thread_false_when_thread_empty() -> None:
  backend = _FakeBackend()
  sent = MailClient(backend).send_on_thread(
      inbox_id="ib",
      thread_id="t-unknown",
      body="hi",
      seq_label="yaseq-x-1",
  )
  assert sent is False
  assert backend.calls == []
178 
179 
180def test_send_on_thread_forwards_recipients_to_the_reply() -> None:
  # The balancer replies on the W6 alert thread but must address the owners
  # explicitly, else AgentMail loops the reply back to the agent (SPEC §8).
  backend = _FakeBackend()
  client = MailClient(backend)
  thread_id = client.open_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: over budget",
      body="alert",
      txn_label="yaspend-dining",
  )
  sent = client.send_on_thread(
      inbox_id="ib",
      thread_id=thread_id,
      body="cover the overspend?",
      seq_label="yb-cover-dining",
      to=["wife@example.com"],
  )
  assert sent is True
  assert backend.reply_tos == [["wife@example.com"]]
201 
202 
203def test_send_on_thread_without_recipients_omits_the_override() -> None:
  # The other callers (W2, the autonomy offer) reply only after the owner has
  # spoken, so they leave AgentMail's default reply-addressing in place.
  backend = _FakeBackend()
  client = MailClient(backend)
  thread_id = client.open_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="[YNAB] $4.50",
      body="proposal",
      txn_label="yatxn-abc",
  )
  client.send_on_thread(
      inbox_id="ib",
      thread_id=thread_id,
      body="a reminder",
      seq_label="yaseq-abc-2",
  )
  assert backend.reply_tos == [None]
222 
223 
224def test_alert_on_thread_opens_then_replies_on_the_same_thread() -> None:
  # First alert opens the thread; a worsening re-alert (new update label, same
  # thread label) replies on that SAME thread, not a new one. The regression
  # guard for re-alert orphaning: one conversation, so a reply always routes
  # back to the W7 balancer indexed by that thread (SPEC §7).
  backend = _FakeBackend()
  client = MailClient(backend)
  first = client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: trending over budget",
      body="Dining is trending over budget...",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-trending_over-500000",
  )
  second = client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: already over budget",
      body="Dining is already over budget...",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-already_over-560000",
  )
  assert first == second  # one conversation across the re-alert
  assert [c[0] for c in backend.calls] == ["new", "reply"]
  # The re-alert reply is addressed to the owner — the thread's latest message
  # is the agent's own opening alert, so without this it would loop back.
⋯ 132 lines hidden (lines 251–382)251    assert backend.reply_tos == [["wife@example.com"]]
252 
253 
254def test_alert_on_thread_dedups_a_retried_alert() -> None:
  # A retry of the first alert (same thread + update label) re-sends nothing:
  # the update label rides on the opening message, so the retry's reply branch
  # finds it already on record and skips.
  backend = _FakeBackend()
  client = MailClient(backend)
  client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: trending over budget",
      body="alert",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-trending_over-500000",
  )
  client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: trending over budget",
      body="alert",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-trending_over-500000",
  )
  assert [c[0] for c in backend.calls] == ["new"]  # no duplicate send
277 
278 
279def test_alert_on_thread_dedups_a_retried_re_alert() -> None:
  # First alert opens; a worsening re-alert replies; a retry of THAT re-alert
  # (same update label) re-sends nothing. Proves the update label is recorded
  # on the reply message too, so the dedup holds for replies, not just opens.
  backend = _FakeBackend()
  client = MailClient(backend)
  client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: trending over budget",
      body="first alert",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-trending_over-500000",
  )
  client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: already over budget",
      body="worse now",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-already_over-560000",
  )
  client.alert_on_thread(
      inbox_id="ib",
      to=["wife@example.com"],
      subject="Dining: already over budget",
      body="worse now",
      thread_label="yaspend-dining-2026-06",
      update_label="yaspend-update-dining-2026-06-already_over-560000",
  )
  assert [c[0] for c in backend.calls] == ["new", "reply"]  # retry no-op
310 
311 
312def test_close_archives_the_thread() -> None:
  backend = _FakeBackend()
  MailClient(backend).close(inbox_id="ib", thread_id="t1")
  assert backend.calls == [("archive", "ib", "t1")]
316 
317 
318@pytest.mark.skipif(
  not os.environ.get("YNAB_AGENT_LIVE_EMAIL"),
  reason="set YNAB_AGENT_LIVE_EMAIL=1 to send a real AgentMail message",
321)
322def test_live_send_reaches_the_owner() -> None:
  inbox = os.environ.get("YNAB_AGENT_INBOX", "alivehoney93@agentmail.to")
  to = os.environ.get("YNAB_AGENT_OWNER", "matthew@mseeks.me")
  out = MailClient.from_env().send(
      OutboundEmail(
          inbox_id=inbox,
          to=(to,),
          subject="[YNAB agent] live mail smoke",
          text="This is the YNAB agent's AgentMail send path, working.",
      )
  )
  assert out.message_id
  assert out.thread_id
335 
336 
337@pytest.mark.skipif(
  not os.environ.get("YNAB_AGENT_LIVE_EMAIL"),
  reason="set YNAB_AGENT_LIVE_EMAIL=1 to send real AgentMail messages",
340)
341def test_live_reply_threads_and_addresses_the_owner() -> None:
  """The W7 fix's load-bearing claim, against the real server (SPEC §8).
343 
  Reproduce the W6→W7 shape: open an alert thread (the agent's own outbound),
  then reply on it addressed to the owner — the exact case commit a60e676
  found broke (the reply's ``To`` looped back to the agent, so the owner never
  saw it). Assert the reply lands on the *same* thread (AgentMail threaded it)
  and its ``To`` is the owner (delivered, not looped back). These two server
  behaviors are what the offline fake cannot prove.
  """
  from agentmail import AgentMail
352 
  inbox = os.environ.get("YNAB_AGENT_INBOX", "alivehoney93@agentmail.to")
  owner = os.environ.get("YNAB_AGENT_OWNER", "matthew@mseeks.me")
  nonce = os.urandom(4).hex()
356 
  mail = MailClient.from_env()
  thread = mail.open_thread(
      inbox_id=inbox,
      to=[owner],
      subject="[YNAB agent] live reply-threading smoke",
      body="alert: Dining is over budget.",
      txn_label=f"live-alert-{nonce}",
  )
  sent = mail.send_on_thread(
      inbox_id=inbox,
      thread_id=thread,
      body="Cover the overspend? (live reply-threading test.)",
      seq_label=f"live-cover-{nonce}",
      to=[owner],
  )
  assert sent is True
373 
  raw = AgentMail(api_key=os.environ["AGENTMAIL_API_KEY"])
  last_id = raw.inboxes.threads.get(inbox, thread).last_message_id
  reply = raw.inboxes.messages.get(inbox, last_id)
  # Threaded on the server, and addressed to the owner (not looped back to the
  # agent — the a60e676 symptom). ``in_reply_to`` confirms the threading
  # headers the owner's mail client needs are set.
  assert reply.thread_id == thread
  assert owner in str(reply.to)
  assert reply.in_reply_to is not None