Real copy and carried payloads for every lifecycle email

What changed, and why

The proposal email was good — and almost everything after it was broken. Six of ten message purposes rendered the placeholder "A quick note on this transaction." at exactly the moments the SPEC promises specific copy: the 7-day handoff, the auto-apply FYI, the revision summary, the couldn't-confirm warning, the diverged which-wins question, the archive notice. The payloads those emails needed were computed upstream — the model's clarify question, the diverged summaries, the decided category — then dropped before compose. W2 follow-ups omitted recipients, so when the agent spoke last, AgentMail addressed the handoff back to the agent's own inbox. The auto path never opened a thread, so the first blessed auto-apply crashed its workflow after the YNAB write. And a flagged AwaitingHuman answered every reply with the same canned question, forever.

One fix through one seam: the effect carries the payload, the state machine populates it, the workflow passes it through (and opens a thread when none exists), the activity resolves names and recipients, and compose renders real copy per purpose.

The payload rides the effect

SendThreadMessage gains two optional fields: detail — free text specific to this message (a clarify question, a which-wins comparison) — and decision — what was actually written, so confirms and FYIs can name it.

The effect: payloads travel with the purpose instead of being dropped.

src/ynab_agent/domain/effects.py · 149 lines

src/ynab_agent/domain/effects.py149 lines · Python

⋯ 58 lines hidden (lines 1–58)1"""Effects: data describing what the spine should do after a transition.
2 
3The state machine is pure and performs no I/O. It returns *effects* — commit a
4write, send a thread message, set or cancel a timer, feed rule learning. The
5Temporal spine interprets and executes them (commit→verify, idempotent sends,
6durable timers). Effects are values, so a transition is fully testable without
7touching YNAB or email.
8"""
9 
10from __future__ import annotations
11 
12import datetime
13from enum import StrEnum
14from typing import Annotated, Literal
15 
16from pydantic import Field
17 
18from ynab_agent.domain.base import Frozen
19from ynab_agent.domain.proposal import Decision
20from ynab_agent.domain.signals import InboundSignal
21 
22 
23class MessagePurpose(StrEnum):
24    """Why a thread message is being sent (shapes the prose; SPEC §3, §5)."""
25 
26    PROPOSAL = "proposal"
27    FYI = "fyi"
28    CONFIRM = "confirm"
29    CLARIFY = "clarify"
30    HANDOFF = "handoff"
31    REVISE_SUMMARY = "revise_summary"
32    POSSIBLY_INCONSISTENT = "possibly_inconsistent"
33    DIVERGED_READBACK = "diverged_readback"
34    ARCHIVE_NOTICE = "archive_notice"
35    OVERRIDE_NOTICE = "override_notice"
36 
37 
38class TimerKind(StrEnum):
39    """The lifecycle timers (SPEC §3)."""
40 
41    AMAZON_HOLD = "amazon_hold"
42    PATIENCE = "patience"
43    ARCHIVE = "archive"
44 
45 
46class RuleLearningKind(StrEnum):
47    """The human-decision events rule learning consumes (SPEC §9)."""
48 
49    CONFIRM = "confirm"
50    CORRECT = "correct"
51 
52 
53class OpenThread(Frozen):
54    """Create the AgentMail thread for this transaction (SPEC §5)."""
55 
56    kind: Literal["open_thread"] = "open_thread"
57 
58 
59class SendThreadMessage(Frozen):
60    """Send a message on the transaction's thread.
61 
62    ``detail`` is the message-specific content the template wraps (the model's
63    clarifying question, a diverged which-wins comparison); ``decision`` lets
64    the send name what was actually written (the confirm/FYI/revise-summary
65    category). Without these the spine once rendered every post-proposal email
66    as a contentless placeholder — the payloads were computed, then dropped.
67    """
68 
69    kind: Literal["send_message"] = "send_message"
70    purpose: MessagePurpose
71    detail: str | None = None
72    decision: Decision | None = None
73 
⋯ 76 lines hidden (lines 74–149)74 
75class CommitToYnab(Frozen):
76    """Commit a decision to YNAB (the spine does commit→verify; SPEC §0.5)."""
77 
78    kind: Literal["commit_to_ynab"] = "commit_to_ynab"
79    decision: Decision
80 
81 
82class SetTimer(Frozen):
83    """Arm a durable timer with an absolute deadline."""
84 
85    kind: Literal["set_timer"] = "set_timer"
86    timer: TimerKind
87    deadline: datetime.datetime
88 
89 
90class CancelTimer(Frozen):
91    """Cancel a previously armed timer."""
92 
93    kind: Literal["cancel_timer"] = "cancel_timer"
94    timer: TimerKind
95 
96 
97class FeedRuleLearning(Frozen):
98    """Feed a confirm/correct event to rule learning (W5; SPEC §9).
99 
100    ``payee`` is carried so W5 can key the rule's match on it without re-reading
101    the snapshot. For a correction, ``prior`` carries the decision being
102    overturned, so W5 can demote *the rule that produced the prior decision*
103    (§3 rule 6), not the new one.
104    """
105 
106    kind: Literal["feed_rule_learning"] = "feed_rule_learning"
107    event: RuleLearningKind
108    payee: str
109    decision: Decision | None = None
110    prior: Decision | None = None
111 
112 
113class RecordAutoAction(Frozen):
114    """Record a landed auto-action in the circuit-breaker ledger (SPEC §0.6).
115 
116    Emitted alongside the commit when a blessed rule auto-applies, so the hard
117    floor's per-run / per-day counts are real and the breaker can trip. Keyed by
118    ``ynab_id`` so a retry or re-enrichment counts the transaction once.
119    """
120 
121    kind: Literal["record_auto_action"] = "record_auto_action"
122    ynab_id: str
123 
124 
125class ReplayBuffered(Frozen):
126    """Re-deliver the signals buffered while in DISCOVERED (SPEC §3)."""
127 
128    kind: Literal["replay_buffered"] = "replay_buffered"
129    signals: tuple[InboundSignal, ...]
130 
131 
132class CloseThread(Frozen):
133    """Label and close the AgentMail thread on archive (SPEC §13)."""
134 
135    kind: Literal["close_thread"] = "close_thread"
136 
137 
138Effect = Annotated[
139    OpenThread
140    | SendThreadMessage
141    | CommitToYnab
142    | SetTimer
143    | CancelTimer
144    | FeedRuleLearning
145    | RecordAutoAction
146    | ReplayBuffered
147    | CloseThread,
148    Field(discriminator="kind"),
149]

The state machine populates them at every emit site. Two examples: the clarify branch carries the model's actual question, and the diverged branch builds the which-wins comparison from the summaries the converge activity already resolved — the signature safety moment finally says what it knows.

Clarify carries the question; diverged carries both sides; NeedsHuman asks a question carrying its reason instead of an empty proposal shell.

src/ynab_agent/domain/state_machine.py · 652 lines

src/ynab_agent/domain/state_machine.py652 lines · Python

⋯ 403 lines hidden (lines 1–403)1"""The transaction lifecycle state machine (SPEC §3).
2 
3``advance(txn, event, now, policy)`` is a pure function: given the current
4transaction, a decided event, the current time, and the timer policy, it returns
5a :class:`Transition` — the next transaction plus the effects the spine should
6execute. It performs no I/O and reads no clock; ``now`` is passed in, so replay
7is deterministic.
8 
9Dispatch is per-state (one handler each), and ``advance`` ends in
10``assert_never`` so mypy proves all ten states are handled. An event a state
11does not expect yields a ``REJECTED`` transition (no state change), never an
12exception, so the spine decides how to handle the surprise.
13 
14Interpretation notes (where the SPEC's diagram is collapsed for a pure model):
15 
16* The REVISING converge step already bundles commit *and* verify (SPEC §3 rules
3-4), so its ``Reapplied`` outcome lands directly in ``OPEN`` (re-approved and
verified), collapsing the degenerate ``APPLIED → OPEN`` hop the diagram draws.
19* ``AWAITING_HUMAN`` reached from a verify failure carries a ``flag`` and no
proposal; per the SPEC such flagged entries do *not* lapse into the generic
hand-off note, so ``PatienceExpired`` there is ignored (the §13 sweep tracks
it).
23"""
24 
25from __future__ import annotations
26 
27from enum import StrEnum
28from typing import TYPE_CHECKING, assert_never
29 
30from ynab_agent.domain.base import Frozen
31from ynab_agent.domain.effects import (
  CancelTimer,
  CloseThread,
  CommitToYnab,
  Effect,
  FeedRuleLearning,
  MessagePurpose,
  OpenThread,
  RecordAutoAction,
  ReplayBuffered,
  RuleLearningKind,
  SendThreadMessage,
  SetTimer,
  TimerKind,
45)
46from ynab_agent.domain.enums import AwaitingFlag, RevisingOrigin
47from ynab_agent.domain.events import (
  AnswerReceived,
  ArchiveWindowReached,
  AskHuman,
  AutoApply,
  ClarifyRequested,
  Converged,
  CouldNotConfirm,
  Diverged,
  Enriched,
  HoldDeadlineReached,
  HoldResolved,
  InboundReceived,
  LifecycleEvent,
  NeedsHuman,
  NoChange,
  OverrideDetected,
  PatienceExpired,
  Reapplied,
  SnapshotMaterialized,
  SnapshotUnavailable,
  VerifyOutcome,
  WriteVerified,
70)
71from ynab_agent.domain.signals import ReplySignal
72from ynab_agent.domain.transaction import (
  Applied,
  Archived,
  AutoApplied,
  AwaitingHuman,
  Discovered,
  Enriching,
  HoldAmazon,
  Lapsed,
  Open,
  Revising,
  Transaction,
  TxnCore,
85)
86 
87if TYPE_CHECKING:
  import datetime
89 
  from ynab_agent.domain.config import LifecyclePolicy
  from ynab_agent.domain.proposal import Decision
92 
93 
94class TransitionKind(StrEnum):
  """The disposition of an ``advance`` call."""
96 
  ADVANCED = "advanced"
  IGNORED = "ignored"
  REJECTED = "rejected"
100 
101 
102class Transition(Frozen):
  """The result of advancing a transaction.
104 
  Attributes:
      kind: ``ADVANCED`` (state and/or effects), ``IGNORED`` (legal no-op), or
          ``REJECTED`` (the event is not valid in this state).
      next: The resulting transaction (unchanged for IGNORED/REJECTED).
      effects: The effects the spine should execute, in order.
      reason: A short explanation for IGNORED/REJECTED.
  """
112 
  kind: TransitionKind
  next: Transaction
  effects: tuple[Effect, ...] = ()
  reason: str | None = None
117 
118 
119def _advanced(nxt: Transaction, *effects: Effect) -> Transition:
  return Transition(kind=TransitionKind.ADVANCED, next=nxt, effects=effects)
121 
122 
123def _ignored(txn: Transaction, reason: str) -> Transition:
  return Transition(kind=TransitionKind.IGNORED, next=txn, reason=reason)
125 
126 
127def _rejected(txn: Transaction, event: LifecycleEvent) -> Transition:
  reason = f"{type(event).__name__} is not valid in {txn.state}"
  return Transition(kind=TransitionKind.REJECTED, next=txn, reason=reason)
130 
131 
132def advance(
  txn: Transaction,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
138) -> Transition:
  """Advance a transaction by one event. Pure; see the module docstring."""
  match txn:
      case Discovered():
          return _from_discovered(txn, event, now=now, policy=policy)
      case HoldAmazon():
          return _from_hold_amazon(txn, event)
      case Enriching():
          return _from_enriching(txn, event, now=now, policy=policy)
      case AutoApplied():
          return _from_auto_applied(txn, event, now=now, policy=policy)
      case AwaitingHuman():
          return _from_awaiting_human(txn, event, now=now, policy=policy)
      case Applied():
          return _from_applied(txn, event, now=now, policy=policy)
      case Open():
          return _from_open(txn, event)
      case Lapsed():
          return _from_lapsed(txn, event)
      case Revising():
          return _from_revising(txn, event, now=now, policy=policy)
      case Archived():
          return _from_archived(txn, event)
  assert_never(txn)
162 
163 
164def _from_discovered(
  txn: Discovered,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
170) -> Transition:
  match event:
      case SnapshotMaterialized(snapshot=snap, hold_for_amazon=hold):
          core = TxnCore(snapshot=snap, thread_id=txn.thread_id)
          replay: tuple[Effect, ...] = (
              (ReplayBuffered(signals=txn.pending),) if txn.pending else ()
          )
          if hold:
              deadline = now + policy.amazon_hold
              return _advanced(
                  HoldAmazon(core=core, amazon_deadline=deadline),
                  SetTimer(timer=TimerKind.AMAZON_HOLD, deadline=deadline),
                  *replay,
              )
          return _advanced(Enriching(core=core), *replay)
      case SnapshotUnavailable():
          return _ignored(txn, "snapshot not yet in YNAB; staying DISCOVERED")
      case InboundReceived(signal=sig):
          thread_id = txn.thread_id
          if thread_id is None and isinstance(sig, ReplySignal):
              thread_id = sig.thread_id
          buffered = Discovered(
              ynab_id=txn.ynab_id,
              thread_id=thread_id,
              pending=(*txn.pending, sig),
          )
          return _advanced(buffered)
      case _:
          return _rejected(txn, event)
199 
200 
201def _from_hold_amazon(txn: HoldAmazon, event: LifecycleEvent) -> Transition:
  match event:
      case HoldResolved(snapshot=snap):
          core = txn.core.model_copy(update={"snapshot": snap})
          return _advanced(
              Enriching(core=core),
              CancelTimer(timer=TimerKind.AMAZON_HOLD),
          )
      case HoldDeadlineReached():
          # Deadline hit: enrich and fall back to asking (SPEC §3).
          return _advanced(Enriching(core=txn.core))
      case InboundReceived(signal=sig):
          # A matched receipt (or reply) short-circuits the hold (SPEC §6).
          return _advanced(
              Enriching(core=txn.core),
              CancelTimer(timer=TimerKind.AMAZON_HOLD),
              ReplayBuffered(signals=(sig,)),
          )
      case _:
          return _rejected(txn, event)
221 
222 
223def _from_enriching(
  txn: Enriching,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
229) -> Transition:
  match event:
      case Enriched(outcome=outcome):
          match outcome:
              case AutoApply(decision=decision):
                  return _advanced(
                      AutoApplied(core=txn.core, decision=decision),
                      CommitToYnab(decision=decision),
                      RecordAutoAction(
                          ynab_id=str(txn.core.snapshot.ynab_id)
                      ),
                  )
              case AskHuman(proposal=proposal):
                  deadline = now + policy.patience_window
                  # First contact: open_thread sends the proposal as the
                  # thread's opening email (AgentMail starts a thread on its
                  # first send), so no separate send is emitted. A re-proposal
                  # (the thread is already open) sends it as a reply.
                  send_effect: tuple[Effect, ...] = (
                      (OpenThread(),)
                      if txn.core.thread_id is None
                      else (
                          SendThreadMessage(purpose=MessagePurpose.PROPOSAL),
                      )
                  )
                  return _advanced(
                      AwaitingHuman(
                          core=txn.core,
                          proposal=proposal,
                          patience_deadline=deadline,
                      ),
                      *send_effect,
                      SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
                  )
          assert_never(outcome)
      case InboundReceived(signal=sig):
          # Re-feed so enrichment incorporates the new signal; stay ENRICHING.
          return Transition(
              kind=TransitionKind.ADVANCED,
              next=txn,
              effects=(ReplayBuffered(signals=(sig,)),),
          )
      case _:
          return _rejected(txn, event)
273 
274 
275def _from_auto_applied(
  txn: AutoApplied,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
281) -> Transition:
  match event:
      case WriteVerified(outcome=outcome):
          return _resolve_write_verify(
              core=txn.core,
              decision=txn.decision,
              outcome=outcome,
              applied_message=MessagePurpose.FYI,
              learning=None,
              now=now,
              policy=policy,
          )
      case _:
          return _rejected(txn, event)
295 
296 
297def _from_applied(
  txn: Applied,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
303) -> Transition:
  match event:
      case WriteVerified(outcome=outcome):
          return _resolve_write_verify(
              core=txn.core,
              decision=txn.decision,
              outcome=outcome,
              applied_message=MessagePurpose.CONFIRM,
              learning=RuleLearningKind.CONFIRM,
              now=now,
              policy=policy,
          )
      case _:
          return _rejected(txn, event)
317 
318 
319def _resolve_write_verify(
  *,
  core: TxnCore,
  decision: Decision,
  outcome: VerifyOutcome,
  applied_message: MessagePurpose,
  learning: RuleLearningKind | None,
  now: datetime.datetime,
  policy: LifecyclePolicy,
328) -> Transition:
  """Shared AUTO_APPLIED/APPLIED → OPEN-or-AWAITING handling (SPEC §3)."""
  match outcome:
      case VerifyOutcome.MATCH:
          archive = now + policy.archive_window
          effects: tuple[Effect, ...] = (
              # Carry the decision so the confirm/FYI names what was written.
              SendThreadMessage(purpose=applied_message, decision=decision),
              SetTimer(timer=TimerKind.ARCHIVE, deadline=archive),
          )
          if learning is not None:
              effects = (
                  FeedRuleLearning(
                      event=learning,
                      payee=core.snapshot.payee,
                      decision=decision,
                  ),
                  *effects,
              )
          return _advanced(Open(core=core, decision=decision), *effects)
      case VerifyOutcome.COULD_NOT_CONFIRM:
          return _enter_flagged_awaiting(
              core,
              AwaitingFlag.POSSIBLY_INCONSISTENT,
              MessagePurpose.POSSIBLY_INCONSISTENT,
              now=now,
              policy=policy,
          )
      case VerifyOutcome.DIVERGED:
          return _enter_flagged_awaiting(
              core,
              AwaitingFlag.DIVERGED,
              MessagePurpose.DIVERGED_READBACK,
              now=now,
              policy=policy,
          )
  assert_never(outcome)
365 
366 
367def _enter_flagged_awaiting(
  core: TxnCore,
  flag: AwaitingFlag,
  message: MessagePurpose,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
  detail: str | None = None,
375) -> Transition:
  """Route a verify failure to a flagged AWAITING_HUMAN with a read-back."""
  deadline = now + policy.patience_window
  return _advanced(
      AwaitingHuman(core=core, patience_deadline=deadline, flag=flag),
      SendThreadMessage(purpose=message, detail=detail),
      SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
  )
383 
384 
385def _from_awaiting_human(
  txn: AwaitingHuman,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
391) -> Transition:
  match event:
      case AnswerReceived(decision=decision):
          return _advanced(
              Applied(core=txn.core, decision=decision),
              CommitToYnab(decision=decision),
              CancelTimer(timer=TimerKind.PATIENCE),
          )
      case ClarifyRequested(question=question):
          deadline = now + policy.patience_window
          return _advanced(
              AwaitingHuman(
                  core=txn.core,
                  proposal=txn.proposal,
                  patience_deadline=deadline,
                  flag=txn.flag,
              ),
              # Carry the model's actual question — without it the owner got
              # the same canned line no matter what was being asked.
              SendThreadMessage(
                  purpose=MessagePurpose.CLARIFY, detail=question
              ),
              SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
          )
      case PatienceExpired():
          if txn.flag is not AwaitingFlag.NONE:
              # Flagged (inconsistent/diverged) entries do not generic-lapse.
              return _ignored(txn, "flagged entry does not lapse (SPEC §3)")
          archive = now + policy.archive_window
⋯ 132 lines hidden (lines 420–551)420            return _advanced(
              Lapsed(core=txn.core, proposal=txn.proposal),
              SendThreadMessage(purpose=MessagePurpose.HANDOFF),
              SetTimer(timer=TimerKind.ARCHIVE, deadline=archive),
          )
      case InboundReceived(signal=sig):
          # Re-feed for interpretation into an answer/clarify; keep waiting.
          return Transition(
              kind=TransitionKind.ADVANCED,
              next=txn,
              effects=(ReplayBuffered(signals=(sig,)),),
          )
      case _:
          return _rejected(txn, event)
434 
435 
436def _from_open(txn: Open, event: LifecycleEvent) -> Transition:
  match event:
      case InboundReceived(signal=sig):
          return _advanced(
              Revising(
                  core=txn.core,
                  instruction=sig,
                  origin=RevisingOrigin.APPLIED,
                  prior=txn.decision,
              ),
              CancelTimer(timer=TimerKind.ARCHIVE),
          )
      case ArchiveWindowReached():
          if not txn.core.snapshot.reconciled:
              return _ignored(txn, "not reconciled yet; staying OPEN")
          return _advanced(
              Archived(core=txn.core, final=txn.decision),
              CloseThread(),
          )
      case OverrideDetected(decision=human_decision):
          # The owner recategorized in YNAB directly: a silent correction.
          # Demote the driving rule back to Observe (CORRECT, carrying the
          # prior agent decision so the *right* rule is demoted, §14.2) and
          # tell them we noticed and backed off. Close the book if the txn is
          # reconciled; otherwise adopt their value and wait (ARCHIVED needs a
          # reconciled snapshot), so we never archive prematurely.
          demote = FeedRuleLearning(
              event=RuleLearningKind.CORRECT,
              payee=txn.core.snapshot.payee,
              decision=human_decision,
              prior=txn.decision,
          )
          notice = SendThreadMessage(
              purpose=MessagePurpose.OVERRIDE_NOTICE, decision=human_decision
          )
          if not txn.core.snapshot.reconciled:
              return _advanced(
                  Open(core=txn.core, decision=human_decision),
                  demote,
                  notice,
              )
          return _advanced(
              Archived(core=txn.core, final=human_decision),
              demote,
              notice,
              CloseThread(),
          )
      case _:
          return _rejected(txn, event)
485 
486 
487def _from_lapsed(txn: Lapsed, event: LifecycleEvent) -> Transition:
  match event:
      case InboundReceived(signal=sig):
          return _advanced(
              Revising(
                  core=txn.core,
                  instruction=sig,
                  origin=RevisingOrigin.LAPSED,
              ),
              CancelTimer(timer=TimerKind.ARCHIVE),
          )
      case ArchiveWindowReached():
          snap = txn.core.snapshot
          if not snap.reconciled:
              return _ignored(txn, "not reconciled yet; staying LAPSED")
          if not snap.categorized:
              # Don't go silent: warn, then let the §13 sweep keep tracking.
              return Transition(
                  kind=TransitionKind.ADVANCED,
                  next=txn,
                  effects=(
                      SendThreadMessage(
                          purpose=MessagePurpose.ARCHIVE_NOTICE
                      ),
                  ),
                  reason="uncategorized; warned before archiving (SPEC §3)",
              )
          return _advanced(
              Archived(core=txn.core),
              CloseThread(),
          )
      case _:
          return _rejected(txn, event)
520 
521 
522def _from_revising(
  txn: Revising,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
528) -> Transition:
  match event:
      case Converged(outcome=outcome):
          match outcome:
              case Reapplied(decision=decision):
                  return _reapply(txn, decision, now=now, policy=policy)
              case NoChange():
                  return _resolve_no_change(txn, now=now, policy=policy)
              case CouldNotConfirm():
                  return _enter_flagged_awaiting(
                      txn.core,
                      AwaitingFlag.POSSIBLY_INCONSISTENT,
                      MessagePurpose.POSSIBLY_INCONSISTENT,
                      now=now,
                      policy=policy,
                  )
              case Diverged(ynab_summary=ynab, requested_summary=asked):
                  # The signature safety moment (SPEC §3 r4): the owner must
                  # see both sides to break the tie, not a contentless note.
                  return _enter_flagged_awaiting(
                      txn.core,
                      AwaitingFlag.DIVERGED,
                      MessagePurpose.DIVERGED_READBACK,
                      now=now,
                      policy=policy,
                      detail=(
                          f"YNAB now shows {ynab}, but your reply asked for "
                          f"{asked} — which should win? Reply with your "
                          "choice and I'll set it."
                      ),
                  )
              case NeedsHuman(reason=reason):
                  deadline = now + policy.patience_window
                  return _advanced(
                      AwaitingHuman(
                          core=txn.core, patience_deadline=deadline
                      ),
                      # A question, not a bare PROPOSAL shell — there is no
                      # proposal here, and the reason says what's needed.
                      SendThreadMessage(
                          purpose=MessagePurpose.CLARIFY, detail=reason
                      ),
                      SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
                  )
          assert_never(outcome)
      case InboundReceived(signal=sig):
          # Newest instruction wins (SPEC §3 rule 1): retain the correction
          # and re-target the in-flight converge, rather than dropping it.
          return Transition(
              kind=TransitionKind.ADVANCED,
              next=txn.model_copy(update={"instruction": sig}),
              effects=(ReplayBuffered(signals=(sig,)),),
          )
      case _:
          return _rejected(txn, event)
583 
584 
585def _reapply(
  txn: Revising,
  decision: Decision,
  *,
⋯ 64 lines hidden (lines 589–652)589    now: datetime.datetime,
  policy: LifecyclePolicy,
591) -> Transition:
  """Re-applied revision: open, summarize, and feed learning (SPEC §3, §9).
593 
  A real category change demotes the prior rule (CORRECT, carrying the prior
  decision so W5 demotes the *right* rule); a same-category revision — a
  receipt adding a memo, or a late first answer from LAPSED — only confirms.
  """
  archive = now + policy.archive_window
  is_correction = (
      txn.prior is not None and txn.prior.allocation != decision.allocation
  )
  payee = txn.core.snapshot.payee
  if is_correction:
      learning = FeedRuleLearning(
          event=RuleLearningKind.CORRECT,
          payee=payee,
          decision=decision,
          prior=txn.prior,
      )
  else:
      learning = FeedRuleLearning(
          event=RuleLearningKind.CONFIRM, payee=payee, decision=decision
      )
  return _advanced(
      Open(core=txn.core, decision=decision),
      learning,
      # Carry the re-decided outcome so the summary says what changed.
      SendThreadMessage(
          purpose=MessagePurpose.REVISE_SUMMARY, decision=decision
      ),
      SetTimer(timer=TimerKind.ARCHIVE, deadline=archive),
  )
623 
624 
625def _resolve_no_change(
  txn: Revising,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
630) -> Transition:
  """No-change exit depends on history (SPEC §3 rule 5).
632 
  Already applied → OPEN (resting). Entered from LAPSED (never applied) →
  AWAITING_HUMAN, re-arming patience, so an unhandled txn is not mislabeled.
  """
  if txn.origin is RevisingOrigin.APPLIED and txn.prior is not None:
      archive = now + policy.archive_window
      return _advanced(
          Open(core=txn.core, decision=txn.prior),
          SetTimer(timer=TimerKind.ARCHIVE, deadline=archive),
      )
  deadline = now + policy.patience_window
  return _advanced(
      AwaitingHuman(core=txn.core, patience_deadline=deadline),
      SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
  )
647 
648 
649def _from_archived(txn: Archived, event: LifecycleEvent) -> Transition:
  # Terminal. A late edit is handled by signal-with-start re-instantiating a
  # fresh REVISING run under the same logical id, not by a transition here.
  return _rejected(txn, event)

No thread? Open one — the auto-apply crash

A blessed auto-apply commits, verifies, then sends its FYI — but the auto path never proposed anything, so no thread exists, and the send raised ('no thread open yet'), retried ~15 minutes, paged the operator, and killed the workflow after the money write. The fix is one spine-level guard: when a message must be sent and no thread exists, open the thread with that message as its first email. The FYI becomes the thread — which is what makes the SPEC §14.5 'one-reply undo' possible at all.

The fallback covers every future no-thread send, not just the FYI.

src/ynab_agent/workflow/txn_workflow.py · 590 lines

src/ynab_agent/workflow/txn_workflow.py590 lines · Python

⋯ 273 lines hidden (lines 1–273)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
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:
  from collections.abc import Callable
34 
  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():
  from ynab_agent.domain.config import DEFAULT_POLICY
  from ynab_agent.domain.effects import (
      CancelTimer,
      CloseThread,
      CommitToYnab,
      Effect,
      FeedRuleLearning,
      MessagePurpose,
      OpenThread,
      RecordAutoAction,
      ReplayBuffered,
      SendThreadMessage,
      SetTimer,
      TimerKind,
  )
  from ynab_agent.domain.enums import DecidedBy
  from ynab_agent.domain.events import (
      AnswerReceived,
      ArchiveWindowReached,
      ClarifyRequested,
      Converged,
      Enriched,
      HoldDeadlineReached,
      HoldResolved,
      InboundReceived,
      LifecycleEvent,
      OverrideDetected,
      PatienceExpired,
      SnapshotMaterialized,
      SnapshotUnavailable,
      WriteVerified,
  )
  from ynab_agent.domain.ids import ThreadId, YnabTransactionId
  from ynab_agent.domain.proposal import Decision
  from ynab_agent.domain.signals import InboundSignal
  from ynab_agent.domain.state_machine import advance
  from ynab_agent.domain.transaction import (
      Applied,
      Archived,
      AutoApplied,
      AwaitingHuman,
      Discovered,
      Enriching,
      HoldAmazon,
      Lapsed,
      Open,
      Revising,
      Transaction,
      YnabSnapshot,
      born,
  )
  from ynab_agent.ingest.plan import is_amazon
  from ynab_agent.policy.converge import classify_verify, target_of
  from ynab_agent.workflow import activities, alert_activities
  from ynab_agent.workflow.alerting import build_failure_alert
  from ynab_agent.workflow.constants import (
      ACTIVITY_RETRY,
      ACTIVITY_TIMEOUT,
      ALERT_BUDGET,
      ALERT_RETRY,
      ALERT_TIMEOUT,
  )
  from ynab_agent.workflow.types import AnswerOutcome, TransactionParams
107 
108# History-length ceiling before a resting workflow continues-as-new. High enough
109# that ordinary flows never trip it; long-lived (30-45 day) transactions do.
110_CONTINUE_AS_NEW_AFTER = 4_000
111 
112# Resting states the workflow may sit in for days while waiting on signals or
113# timers; continue-as-new fires only from one of these. DISCOVERED counts: a
114# transaction born from a signal can wait here for a slow YNAB import (SPEC §3).
115_RESTING = (Discovered, AwaitingHuman, Open, Lapsed, HoldAmazon)
116 
117 
118def _hold_for_amazon(snapshot: YnabSnapshot) -> bool:
  """Whether to hold for Amazon item detail: Amazon-ish and no memo yet.
120 
  Shares :func:`~ynab_agent.ingest.plan.is_amazon` with W1's backfill signal,
  so "what counts as Amazon" has a single source of truth (SPEC §3, §11).
  """
  return is_amazon(snapshot.payee) and not snapshot.has_memo
125 
126 
127@workflow.defn
128class TransactionWorkflow:
  """The durable per-transaction lifecycle."""
130 
  def __init__(self) -> None:
      """Initialize empty state; ``run`` populates it from the params."""
      self._txn: Transaction
      self._ynab_id: str = ""
      self._thread_id: str | None = None
      self._deadlines: dict[TimerKind, datetime.datetime] = {}
      self._inbound: deque[InboundSignal] = deque()
      self._snapshot_ready: YnabSnapshot | None = None
      # Monotonic per-transaction counter: the outbound-send idempotency key
      # so a replay/retry never double-emails (SPEC §3 outbound dedup).
      self._action_seq: int = 0
142 
  # ── Signals (the external world pushes in) ──────────────────────────────
  @workflow.signal
  def submit_inbound(self, signal: InboundSignal) -> None:
      """A reply or matched receipt arrived (W3/W4)."""
      self._inbound.append(signal)
148 
  @workflow.signal
  def notify_snapshot(self, snapshot: YnabSnapshot) -> None:
      """W1 materialized (or backfilled the memo of) the YNAB snapshot."""
      self._snapshot_ready = snapshot
153 
  @workflow.query
  def state(self) -> str:
      """The current lifecycle state (for observability)."""
      return self._txn.state.value
158 
  # ── The durable loop ────────────────────────────────────────────────────
  @workflow.run
  async def run(self, params: TransactionParams) -> None:
      """Drive the lifecycle until the transaction is archived (SPEC §3)."""
      self._ynab_id = params.ynab_id
      self._thread_id = params.thread_id
      self._deadlines = dict(params.resume_deadlines)
      self._inbound.extend(params.resume_inbound)
      self._action_seq = params.resume_action_seq
      self._txn = params.resume_txn or born(
          YnabTransactionId(params.ynab_id), params.thread_id
      )
      self._sync_thread_id()
172 
      try:
          while not isinstance(self._txn, Archived):
              await self._step()
              if (
                  isinstance(self._txn, _RESTING)
                  and not self._inbound  # drain pending signals (SPEC §0.5)
                  and workflow.info().get_current_history_length()
                  > _CONTINUE_AS_NEW_AFTER
              ):
                  # continue_as_new raises to restart; nothing runs after it.
                  # (ContinueAsNewError is not an ActivityError, so it escapes
                  # the failure hook below untouched.)
                  workflow.continue_as_new(self._resume_params())
      except ActivityError as exc:
          # A terminal activity failure: a non-retryable bug (the constants
          # denylist) or the schedule_to_close budget elapsing. Page the owner
          # once — deduped, best-effort — then re-raise so the transaction
          # still fails and stays visible in Temporal (SPEC §13).
          await workflow.execute_activity(
              alert_activities.alert_failure,
              build_failure_alert(
                  key=self._ynab_id,
                  context=self._alert_context(),
                  exc=exc,
              ),
              start_to_close_timeout=ALERT_TIMEOUT,
              schedule_to_close_timeout=ALERT_BUDGET,
              retry_policy=ALERT_RETRY,
          )
          raise
203 
  def _alert_context(self) -> str:
      """A human locator for a failure alert: payee + txn id when known."""
      st = self._txn
      if isinstance(st, Discovered):
          return f"txn {self._ynab_id}"
      return f"{st.core.snapshot.payee} (txn {self._ynab_id})"
210 
  def _resume_params(self) -> TransactionParams:
      return TransactionParams(
          ynab_id=YnabTransactionId(self._ynab_id),
          thread_id=ThreadId(self._thread_id)
          if self._thread_id is not None
          else None,
          resume_txn=self._txn,
          resume_deadlines=dict(self._deadlines),
          resume_inbound=tuple(self._inbound),
          resume_action_seq=self._action_seq,
      )
222 
  async def _step(self) -> None:
      st = self._txn
      if isinstance(st, Discovered):
          await self._on_discovered()
      elif isinstance(st, Enriching):
          await self._on_enriching(st)
      elif isinstance(st, HoldAmazon):
          await self._on_hold(st)
      elif isinstance(st, AwaitingHuman):
          await self._on_awaiting(st)
      elif isinstance(st, Open):
          await self._on_open(st)
      elif isinstance(st, Lapsed):
          await self._on_lapsed()
      elif isinstance(st, Revising):
          await self._on_revising(st)
      elif isinstance(st, (AutoApplied, Applied, Archived)):
          # Transient (handled via the commit→verify follow-up) or terminal.
          return
242 
  # ── apply + effect dispatch ─────────────────────────────────────────────
  async def _dispatch(self, event: LifecycleEvent) -> None:
      transition = advance(
          self._txn, event, now=workflow.now(), policy=DEFAULT_POLICY
      )
      self._txn = transition.next
      # Keep the thread-id mirror in step with the state machine, which can
      # adopt a reply's thread on its own (e.g. a reply in DISCOVERED).
      self._sync_thread_id()
      for effect in transition.effects:
          followup = await self._execute(effect)
          if followup is not None:
              await self._dispatch(followup)
256 
  async def _execute(self, effect: Effect) -> LifecycleEvent | None:
      if isinstance(effect, CommitToYnab):
          await workflow.execute_activity(
              activities.commit_to_ynab,
              args=[self._ynab_id, effect.decision],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          read = await workflow.execute_activity(
              activities.read_back,
              self._ynab_id,
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
          return WriteVerified(
              outcome=classify_verify(read, target_of(effect.decision))
          )
      if isinstance(effect, OpenThread):
          await self._open_thread(MessagePurpose.PROPOSAL, None, None)
      elif isinstance(effect, SendThreadMessage):
          if self._thread_id is None:
              # No thread yet — an auto-applied transaction reaches its FYI
              # (or a verify-failure note) without ever proposing. Open the
              # thread with THIS message as its first email instead of
              # crashing the send (SPEC §14.5: the FYI must exist for the
              # one-reply undo to exist).
              await self._open_thread(
                  effect.purpose, effect.detail, effect.decision
              )
              return None
          self._action_seq += 1
          await workflow.execute_activity(
              activities.send_thread_message,
              args=[
                  self._ynab_id,
                  self._thread_id,
                  effect.purpose,
                  self._action_seq,
                  self._proposal(),
                  effect.detail,
                  effect.decision,
              ],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
⋯ 42 lines hidden (lines 301–342)301            )
      elif isinstance(effect, FeedRuleLearning):
          await workflow.execute_activity(
              activities.feed_rule_learning,
              effect,
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
      elif isinstance(effect, RecordAutoAction):
          # Bound the auto-action in the durable breaker ledger (SPEC §0.6).
          # Best-effort: a ledger hiccup — even an activity timeout, which the
          # body's own try/except cannot catch — must never fail the
          # categorization it counts, so swallow it rather than trip the §13
          # failure hook. The per-txn ceiling still binds; the next tick
          # re-counts from the durable ledger.
          try:
              await workflow.execute_activity(
                  activities.record_auto_action,
                  effect.ynab_id,
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=ACTIVITY_RETRY,
              )
          except ActivityError:
              workflow.logger.warning(
                  "auto-action ledger record failed; continuing"
              )
      elif isinstance(effect, CloseThread):
          if self._thread_id is not None:
              await workflow.execute_activity(
                  activities.close_thread,
                  self._thread_id,
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=ACTIVITY_RETRY,
              )
      elif isinstance(effect, SetTimer):
          self._deadlines[effect.timer] = effect.deadline
      elif isinstance(effect, CancelTimer):
          self._deadlines.pop(effect.timer, None)
      elif isinstance(effect, ReplayBuffered):
          self._inbound.extendleft(reversed(effect.signals))
      return None
342 
  async def _open_thread(
      self,
      purpose: MessagePurpose,
      detail: str | None,
      decision: Decision | None,
  ) -> None:
      """Open the AgentMail thread with ``purpose`` as its first message."""
      tid = await workflow.execute_activity(
          activities.open_thread,
          args=[self._ynab_id, self._proposal(), purpose, detail, decision],
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      self._set_thread_id(tid)
      # Index this workflow by its thread for reply routing (§5a).
      workflow.upsert_search_attributes([_TXN_THREAD_ID.value_set(tid)])
359 
⋯ 231 lines hidden (lines 360–590)360    def _set_thread_id(self, tid: str) -> None:
      self._thread_id = tid
      st = self._txn
      if not isinstance(st, Discovered):
          new_core = st.core.model_copy(update={"thread_id": ThreadId(tid)})
          self._txn = st.model_copy(update={"core": new_core})
366 
  def _sync_thread_id(self) -> None:
      """Mirror the current transaction's thread id (SM may adopt it)."""
      st = self._txn
      tid = st.thread_id if isinstance(st, Discovered) else st.core.thread_id
      if tid is not None:
          self._thread_id = str(tid)
373 
  def _proposal(self) -> Proposal | None:
      """The current best-guess proposal, for states that carry one.
376 
      Passed to the mail activities so the proposal email can name the guess +
      alternatives; ``None`` for purposes whose content derives from YNAB.
      """
      st = self._txn
      if isinstance(st, (Enriching, AwaitingHuman, Lapsed)):
          return st.proposal
      return None
384 
  # ── per-state steps ─────────────────────────────────────────────────────
  async def _on_discovered(self) -> None:
      snapshot = await workflow.execute_activity(
          activities.fetch_snapshot,
          self._ynab_id,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      if snapshot is not None:
          await self._dispatch(
              SnapshotMaterialized(
                  snapshot=snapshot,
                  hold_for_amazon=_hold_for_amazon(snapshot),
              )
          )
          return
      # Signal beat the poll: wait for materialization or buffer an inbound.
      await self._dispatch(SnapshotUnavailable())
      await workflow.wait_condition(
          lambda: self._snapshot_ready is not None or len(self._inbound) > 0
      )
      if self._snapshot_ready is not None:
          snap = self._snapshot_ready
          self._snapshot_ready = None
          await self._dispatch(
              SnapshotMaterialized(
                  snapshot=snap, hold_for_amazon=_hold_for_amazon(snap)
              )
          )
      else:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
418 
  async def _on_enriching(self, st: Enriching) -> None:
      outcome = await workflow.execute_activity(
          activities.enrich,
          st.core.snapshot,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      await self._dispatch(Enriched(outcome=outcome))
427 
  async def _on_hold(self, st: HoldAmazon) -> None:
      ready = await self._wait_until(
          st.amazon_deadline,
          lambda: self._snapshot_ready is not None or len(self._inbound) > 0,
      )
      if not ready:
          await self._dispatch(HoldDeadlineReached())
      elif self._snapshot_ready is not None:
          snap = self._snapshot_ready
          self._snapshot_ready = None
          await self._dispatch(HoldResolved(snapshot=snap))
      else:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
443 
  async def _on_awaiting(self, st: AwaitingHuman) -> None:
      got_inbound = await self._wait_until(
          st.patience_deadline, self._has_inbound
      )
      if not got_inbound:
          before = type(self._txn)
          await self._dispatch(PatienceExpired())
          if type(self._txn) is not before:
              return  # lapsed (or otherwise transitioned)
          # Flagged (verify-failure) entry: PatienceExpired is ignored and
          # does not lapse (SPEC §3). Drop the passed timer and wait for an
          # inbound instead of re-spinning the expired deadline.
          self._deadlines.pop(TimerKind.PATIENCE, None)
          await workflow.wait_condition(self._has_inbound)
      await self._interpret_inbound(st.core.snapshot)
459 
  async def _interpret_inbound(self, snapshot: YnabSnapshot) -> None:
      signal = self._inbound.popleft()
      interpretation = await workflow.execute_activity(
          activities.interpret_inbound,
          args=[signal, snapshot, self._proposal()],
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      if isinstance(interpretation, AnswerOutcome):
          await self._dispatch(
              AnswerReceived(decision=interpretation.decision)
          )
      else:
          await self._dispatch(
              ClarifyRequested(question=interpretation.question)
          )
476 
  async def _on_open(self, st: Open) -> None:
      deadline = self._deadlines.get(TimerKind.ARCHIVE)
      got_inbound = (
          await self._wait_until(deadline, self._has_inbound)
          if deadline is not None
          else await self._wait_forever(self._has_inbound)
      )
      if got_inbound:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
          return
      # The archive window elapsed. Before closing the book, re-read YNAB to
      # catch a silent manual recategorization — an out-of-band correction
      # that must demote the driving rule (SPEC §14.2).
      event = await self._archive_or_override(st)
      before = type(self._txn)
      await self._dispatch(event)
      if type(self._txn) is before:
          # Archive blocked (not reconciled) and no override: drop the stale
          # timer and wait for an inbound rather than busy-looping (SPEC §3).
          self._deadlines.pop(TimerKind.ARCHIVE, None)
          await workflow.wait_condition(self._has_inbound)
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
503 
  async def _archive_or_override(self, st: Open) -> LifecycleEvent:
      """At archive time, detect a manual YNAB edit (SPEC §14.2).
506 
      Re-reads the current end-state and, if its category no longer matches
      the agent's applied decision, returns an ``OverrideDetected`` carrying
      the human's choice (the spine then demotes the rule); otherwise the
      ordinary ``ArchiveWindowReached``. A memo-only change is not an override
      — only the allocation is compared.
      """
      read = await workflow.execute_activity(
          activities.read_back,
          self._ynab_id,
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      if read is not None and read.allocation != st.decision.allocation:
          human = Decision(
              allocation=read.allocation,
              memo=read.memo,
              approved=read.approved,
              decided_by=DecidedBy.HUMAN,
              decided_at=workflow.now(),
          )
          return OverrideDetected(decision=human)
      return ArchiveWindowReached()
529 
  async def _on_lapsed(self) -> None:
      await self._wait_then_revise_or_archive()
532 
  def _has_inbound(self) -> bool:
      return len(self._inbound) > 0
535 
  async def _wait_then_revise_or_archive(self) -> None:
      deadline = self._deadlines.get(TimerKind.ARCHIVE)
      got_inbound = (
          await self._wait_until(deadline, self._has_inbound)
          if deadline is not None
          else await self._wait_forever(self._has_inbound)
      )
      if got_inbound:
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
          return
      # The archive window elapsed. Attempt to archive; if it is blocked (not
      # reconciled / not categorized) the state is unchanged, so drop the
      # now-stale timer and wait for an inbound rather than busy-looping on
      # the already-passed deadline.
      before = type(self._txn)
      await self._dispatch(ArchiveWindowReached())
      if type(self._txn) is before:
          self._deadlines.pop(TimerKind.ARCHIVE, None)
          await workflow.wait_condition(self._has_inbound)
          await self._dispatch(
              InboundReceived(signal=self._inbound.popleft())
          )
560 
  async def _wait_forever(self, predicate: Callable[[], bool]) -> bool:
      await workflow.wait_condition(predicate)
      return True
564 
  async def _on_revising(self, st: Revising) -> None:
      outcome = await workflow.execute_activity(
          activities.converge,
          args=[st.core.snapshot, st.instruction, st.prior],
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      await self._dispatch(Converged(outcome=outcome))
573 
  async def _wait_until(
      self,
      deadline: datetime.datetime,
      predicate: Callable[[], bool],
  ) -> bool:
      """Wait for ``predicate`` until an absolute deadline.
580 
      Returns ``True`` if the predicate fired first, ``False`` on timeout.
      """
      timeout = deadline - workflow.now()
      if timeout < timedelta(0):
          timeout = timedelta(0)
      try:
          await workflow.wait_condition(predicate, timeout=timeout)
      except TimeoutError:
          return False
      return True

Names, comparisons, and recipients at the send boundary

The send activity resolves the decided allocation to category names (Done — set to Groceries and approved, not 'the category you picked'), derives a which-wins comparison for the plain commit→verify path from a fresh snapshot read, and — the quiet killer — addresses every send to the owners explicitly. Without to=, AgentMail replies to the thread's last speaker; when that was the agent, the handoff/archive/override notices went to the agent's own inbox.

_decided_display names what was written; _diverged_detail builds the comparison for the non-converge path.

src/ynab_agent/workflow/activities.py · 706 lines

src/ynab_agent/workflow/activities.py706 lines · Python

⋯ 236 lines hidden (lines 1–236)1"""The I/O ports of the transaction lifecycle, as Temporal activities.
2 
3Every side effect the workflow performs — reading YNAB, committing a write,
4sending email, the agentic enrichment/interpretation/converge steps — is an
5activity, so neither the model's nor the spine's I/O runs in workflow code
6(SPEC §0.5). These are *stubs*: typed signatures with no body. The real
7implementations (YNAB/AgentMail MCP, Pydantic AI) are wired in a later step; the
8workflow tests register mock implementations.
9"""
10 
11from __future__ import annotations
12 
13from typing import TYPE_CHECKING
14 
15from temporalio import activity
16 
17from ynab_agent.domain.allocations import (
  ProposedCategory,
  ProposedSplit,
  ResolvedCategory,
21)
22from ynab_agent.domain.effects import FeedRuleLearning, MessagePurpose
23from ynab_agent.domain.events import ConvergeOutcome, EnrichmentOutcome
24from ynab_agent.domain.ids import CategoryId
25from ynab_agent.domain.proposal import Decision, Proposal
26from ynab_agent.domain.signals import InboundSignal
27from ynab_agent.domain.transaction import YnabSnapshot
28from ynab_agent.policy.converge import TargetState
29from ynab_agent.workflow.types import ReplyOutcome
30 
31if TYPE_CHECKING:
  import datetime
33 
  # Annotation-only: never imported at runtime, so the agentic/model stack
  # (pydantic-ai) never enters the workflow sandbox. The enrich body imports
  # it lazily, inside the activity, where it runs outside the sandbox.
  from ynab_agent.agentic.enrich import CandidateCategory
  from ynab_agent.budget.overspend import CategorySpend
  from ynab_agent.domain.rule import Rule
  from ynab_agent.policy.floor import AutoActionCounters
41 
42 
43@activity.defn
44async def fetch_snapshot(ynab_id: str) -> YnabSnapshot | None:
  """Read the current YNAB snapshot, or ``None`` if not yet imported.
46 
  The YNAB client is imported lazily (keeping httpx out of the sandbox) and
  its blocking call runs off the event loop.
  """
  import asyncio
51 
  from ynab_agent.ynab.client import YnabClient
53 
  client = YnabClient.from_env()
  return await asyncio.to_thread(client.snapshot, ynab_id)
56 
57 
58def _candidates_from_spends(
  spends: tuple[CategorySpend, ...],
60) -> tuple[CandidateCategory, ...]:
  """Map the budget's live categories to the agent's candidate choices."""
  from ynab_agent.agentic.enrich import CandidateCategory
63 
  return tuple(
      CandidateCategory(id=str(spend.category), name=spend.name)
      for spend in spends
  )
68 
69 
70async def _load_payee_rules(payee: str) -> list[Rule]:
  """Query the durable rule registry for this payee's rules (W5, SPEC §14).
72 
  Returns ``[]`` when the registry has not been started yet (no learning has
  happened) or is unreachable — the conservative fallback, which routes the
  transaction to ASK rather than risking an auto-apply on stale knowledge.
  """
  from temporalio.service import RPCError
78 
  from ynab_agent.domain.rule import Rule
  from ynab_agent.workflow.registry_types import REGISTRY_WORKFLOW_ID
  from ynab_agent.workflow.temporal_client import client
82 
  temporal = await client()
  handle = temporal.get_workflow_handle(REGISTRY_WORKFLOW_ID)
  try:
      # ``result_type`` is load-bearing: the pydantic data converter decodes a
      # query payload to plain ``dict``s without it, and the gate reads real
      # ``Rule`` objects (``rule.match``) — a dict there raises AttributeError
      # and the enrich activity retries forever (SPEC §14 gate-load path).
      rules = await handle.query(
          "payee_rules", payee, result_type=tuple[Rule, ...]
      )
  except RPCError:
      return []
  return list(rules)
96 
97 
98async def _load_auto_action_counters(
  now: datetime.datetime,
100) -> AutoActionCounters:
  """Read the live circuit-breaker counts from the durable ledger (SPEC §0.6).
102 
  Returns zeros when the ledger has not been started yet (no auto-action has
  ever happened) or is unreachable — the correct conservative default *here*,
  since a never-started ledger genuinely means zero auto-actions and the
  breaker must not trip on its own absence (contrast ``_load_payee_rules``,
  which fails *closed* to ASK because an unknown rule must never auto-apply).
  """
  from temporalio.service import RPCError
110 
  from ynab_agent.policy.floor import AutoActionCounters
  from ynab_agent.workflow.auto_action_types import (
      AUTO_ACTION_LEDGER_WORKFLOW_ID,
      CountersRequest,
  )
  from ynab_agent.workflow.temporal_client import client
117 
  temporal = await client()
  handle = temporal.get_workflow_handle(AUTO_ACTION_LEDGER_WORKFLOW_ID)
  try:
      result: AutoActionCounters = await handle.query(
          "counters",
          CountersRequest(now=now),
          result_type=AutoActionCounters,
      )
  except RPCError:
      return AutoActionCounters()
  return result
129 
130 
131async def _load_enrichment_inputs(
  snapshot: YnabSnapshot,
  now: datetime.datetime,
134) -> tuple[tuple[CandidateCategory, ...], list[Rule], AutoActionCounters]:
  """Fetch the candidate categories, in-scope rules, and live auto counters.
136 
  The candidates are the budget's live categories, read from YNAB (the source
  of truth); the rules come from the durable registry, keyed on the payee; the
  auto-action counters come from the durable circuit-breaker ledger (SPEC
  §0.6), so the per-run / per-day cap reads real counts and can trip. The gate
  (SPEC §4.2, §14) decides auto-vs-ask over the loaded rules — auto-applying
  only a blessed one, still bounded by the floor.
  """
  import asyncio
145 
  from ynab_agent.ynab.client import YnabClient
147 
  client = YnabClient.from_env()
  spends = await asyncio.to_thread(client.category_spends)
  rules = await _load_payee_rules(snapshot.payee)
  counters = await _load_auto_action_counters(now)
  return _candidates_from_spends(spends), rules, counters
153 
154 
155@activity.defn
156async def enrich(snapshot: YnabSnapshot) -> EnrichmentOutcome:
  """Assemble the proposal and route via the gate (the agentic middle).
158 
  The model stack is imported lazily, here in the activity body, so it never
  enters the workflow sandbox. The gate decides autonomy from the rules; the
  agent runs only to produce the proposal a human is asked to confirm.
  """
  from datetime import UTC, datetime
164 
  from ynab_agent.agentic.enrich import decide_enrichment
166 
  now = datetime.now(UTC)
  candidates, rules, counters = await _load_enrichment_inputs(snapshot, now)
  return await decide_enrichment(
      snapshot, candidates, rules, counters, now=now
  )
172 
173 
174@activity.defn
175async def commit_to_ynab(ynab_id: str, decision: Decision) -> None:
  """Commit a decision to YNAB (the deterministic write).
177 
  Lazy YNAB client (httpx stays out of the sandbox); the blocking write runs
  off the event loop.
  """
  import asyncio
182 
  from ynab_agent.ynab.client import YnabClient
184 
  client = YnabClient.from_env()
  await asyncio.to_thread(client.commit, ynab_id, decision)
187 
188 
189@activity.defn
190async def read_back(ynab_id: str) -> TargetState | None:
  """Read the post-write end-state for verification, or ``None`` if unread."""
  import asyncio
193 
  from ynab_agent.ynab.client import YnabClient
195 
  client = YnabClient.from_env()
  return await asyncio.to_thread(client.read_back, ynab_id)
198 
199 
200def _txn_label(ynab_id: str) -> str:
  """The per-transaction idempotency label (open-thread dedup; SPEC §5)."""
  return f"yatxn-{ynab_id}"
203 
204 
205def _seq_label(ynab_id: str, action_seq: int) -> str:
  """The per-action idempotency label (send dedup; SPEC §3)."""
  return f"yaseq-{ynab_id}-{action_seq}"
208 
209 
210def _subject(snapshot: YnabSnapshot, category: str | None) -> str:
  """The thread's subject: payee + amount, and the suggested category.
212 
  No ``[YNAB]`` prefix — the sender address is a known contact. Naming the
  proposed category lets the owner act from the subject line alone.
  """
  base = f"{snapshot.payee} — {snapshot.amount}"
  return f"{base} · {category}?" if category else base
218 
219 
220def _date_display(day: datetime.date) -> str:
  """A short, friendly transaction date, e.g. ``May 29`` (no leading zero)."""
  return f"{day.strftime('%b')} {day.day}"
223 
224 
225def _allocation_display(
  allocation: ProposedCategory | ProposedSplit, names: dict[str, str]
227) -> str:
  """A human display of the proposed allocation, by category name."""
  if isinstance(allocation, ProposedCategory):
      return names.get(str(allocation.category), str(allocation.category))
  return " + ".join(
      names.get(str(line.category), str(line.category))
      for line in allocation.lines
  )
235 
236 
237def _decided_display(
  decision: Decision | None, names: dict[str, str]
239) -> str | None:
  """The category name(s) a decision wrote, for naming it in an email."""
  if decision is None:
      return None
  allocation = decision.allocation
  if isinstance(allocation, ResolvedCategory):
      return names.get(str(allocation.category), str(allocation.category))
  return " + ".join(
      names.get(str(line.category), str(line.category))
      for line in allocation.lines
  )
250 
251 
252async def _read_for_compose(
  ynab_id: str,
254) -> tuple[YnabSnapshot, dict[str, str]]:
  """Re-read the YNAB snapshot and a category-id→name map for composing.
256 
  Both come from YNAB (the source of truth) at send time — there is no stored
  copy of the facts or the category names (SPEC §0.5, store-free).
  """
  import asyncio
261 
  from ynab_agent.ynab.client import YnabClient
263 
  client = YnabClient.from_env()
  snapshot = await asyncio.to_thread(client.snapshot, ynab_id)
  if snapshot is None:
      msg = f"transaction {ynab_id} not found in YNAB at compose time"
      raise RuntimeError(msg)
  spends = await asyncio.to_thread(client.category_spends)
  names = {str(spend.category): spend.name for spend in spends}
  return snapshot, names
272 
273 
274def _diverged_detail(
  snapshot: YnabSnapshot, decision: Decision, names: dict[str, str]
276) -> str:
  """A which-wins comparison from the live snapshot vs. the intended write.
278 
  The converge path carries its own comparison on the effect; this covers the
  plain commit→verify path, where only the intended decision is known — the
  current side comes from the fresh snapshot read at send time.
  """
  current = (
      names.get(str(snapshot.category_id), str(snapshot.category_id))
      if snapshot.category_id is not None
      else ("a split" if snapshot.subtransactions else "uncategorized")
  )
  intended = _decided_display(decision, names)
  return (
      f"YNAB now shows {current}, but I set {intended} — which should "
      "win? Reply with your choice and I'll sort it out."
  )
⋯ 414 lines hidden (lines 293–706)293 
294 
295def _render_message(
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
  purpose: MessagePurpose,
  names: dict[str, str],
  *,
  detail: str | None = None,
  decision: Decision | None = None,
303) -> str:
  """Lay out one message body for the thread (deterministic; SPEC §5).
305 
  The proposal's category + alternatives (model-chosen upstream) are resolved
  to names here and handed to the template — no model call at send time.
  ``detail`` and ``decision`` carry the message-specific payload from the
  state machine (the clarify question, the diverged comparison, what a
  confirm/FYI/revision actually wrote).
  """
  from ynab_agent.agentic.compose import ComposeRequest, render_body
313 
  proposed = (
      _allocation_display(proposal.allocation, names)
      if proposal is not None
      else None
  )
  alternatives = (
      tuple(names.get(str(alt), str(alt)) for alt in proposal.alternatives)
      if proposal is not None
      else ()
  )
  if (
      detail is None
      and decision is not None
      and purpose is MessagePurpose.DIVERGED_READBACK
  ):
      detail = _diverged_detail(snapshot, decision, names)
  request = ComposeRequest(
      purpose=purpose.value,
      payee=snapshot.payee,
      amount_display=str(snapshot.amount),
      txn_date=_date_display(snapshot.txn_date),
      memo=snapshot.memo,
      proposed_category=proposed,
      alternatives=alternatives,
      rationale=proposal.rationale if proposal is not None else None,
      detail=detail,
      decided_category=_decided_display(decision, names),
  )
  return render_body(request)
343 
344 
345@activity.defn
346async def open_thread(
  ynab_id: str,
  proposal: Proposal | None,
  purpose: MessagePurpose = MessagePurpose.PROPOSAL,
  detail: str | None = None,
  decision: Decision | None = None,
352) -> str:
  """Open the AgentMail thread with its first message; returns its id.
354 
  A thread starts on its first send (AgentMail has no empty-thread create).
  Usually that first message is the proposal, but an auto-applied transaction
  has no proposal email — its thread opens with the FYI (``purpose``/
  ``decision``), which is what makes the SPEC §14.5 per-action FYI + one-reply
  undo possible at all. The txn facts + category names are re-read from YNAB
  at compose time, and the open is idempotent on the per-transaction label, so
  a retry re-finds the thread rather than sending a duplicate.
  """
  import asyncio
364 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
367 
  settings = Settings()
  snapshot, names = await _read_for_compose(ynab_id)
  body = _render_message(
      snapshot, proposal, purpose, names, detail=detail, decision=decision
  )
  headline = _decided_display(decision, names) or (
      _allocation_display(proposal.allocation, names)
      if proposal is not None
      else None
  )
  mail = MailClient.from_env()
  return await asyncio.to_thread(
      mail.open_thread,
      inbox_id=settings.inbox,
      to=list(settings.owners),
      subject=_subject(snapshot, headline),
      body=body,
      txn_label=_txn_label(ynab_id),
  )
387 
388 
389@activity.defn
390async def send_thread_message(
  ynab_id: str,
  thread_id: str | None,
  purpose: MessagePurpose,
  action_seq: int,
  proposal: Proposal | None,
  detail: str | None = None,
  decision: Decision | None = None,
398) -> None:
  """Send a follow-up message on the transaction's thread.
400 
  ``action_seq`` is the per-transaction idempotency key: the send dedups on it
  so a retry never double-sends (SPEC §3). ``proposal`` is the current best
  guess where the purpose needs it (re-proposal); ``detail``/``decision``
  carry the message's specific payload (clarify question, diverged
  comparison, what was written). Recipients are set to the owners explicitly:
  when the agent was the last speaker on the thread, AgentMail would
  otherwise address the reply back to the agent's own inbox and the owners
  would never see it (the same fix the W7 balancer needed, #17).
  """
  import asyncio
411 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
414 
  if thread_id is None:
      msg = f"cannot send {purpose.value} for {ynab_id}: no thread open yet"
      raise RuntimeError(msg)
  settings = Settings()
  snapshot, names = await _read_for_compose(ynab_id)
  body = _render_message(
      snapshot, proposal, purpose, names, detail=detail, decision=decision
  )
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.send_on_thread,
      inbox_id=settings.inbox,
      thread_id=thread_id,
      body=body,
      seq_label=_seq_label(ynab_id, action_seq),
      to=list(settings.owners),
  )
432 
433 
434def _proposed_category_id(proposal: Proposal | None) -> CategoryId | None:
  """The single proposed category id, or None (a split is not approvable)."""
  if proposal is None:
      return None
  allocation = proposal.allocation
  if isinstance(allocation, ProposedCategory):
      return allocation.category
  return None
442 
443 
444@activity.defn
445async def interpret_inbound(
  signal: InboundSignal,
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
449) -> ReplyOutcome:
  """Interpret a human reply into an answer or a question (SPEC §3, §5).
451 
  Free-form: the model reads whether the reply approves the proposal, names a
  different category, or asks a question. The spine — not the model — stamps
  the human + time into the resulting Decision. A non-reply inbound can't be
  answered directly, so ask rather than guess a write. A *missing* proposal
  (a flagged verify-failure entry, a NeedsHuman wait) is fine: the owner can
  still name a category outright, and looping the same canned question at
  them made those states unanswerable by email. Names + candidates are
  re-read from YNAB at interpret time.
  """
  import asyncio
  from datetime import UTC, datetime
463 
  from ynab_agent.agentic.interpret import (
      InterpretRequest,
      interpret,
      to_reply_outcome,
  )
  from ynab_agent.domain.signals import ReplySignal
  from ynab_agent.workflow.types import ClarifyOutcome
  from ynab_agent.ynab.client import YnabClient
472 
  proposed_id = _proposed_category_id(proposal)
  if not isinstance(signal, ReplySignal):
      return ClarifyOutcome(
          question="Could you say which category this should be?"
      )
478 
  client = YnabClient.from_env()
  spends = await asyncio.to_thread(client.category_spends)
  names = {str(spend.category): spend.name for spend in spends}
  request = InterpretRequest(
      reply_text=signal.text,
      payee=snapshot.payee,
      amount_display=str(snapshot.amount),
      proposed_category_name=names.get(str(proposed_id), str(proposed_id))
      if proposed_id is not None
      else None,
      candidates=_candidates_from_spends(spends),
  )
  interpretation = await interpret(request)
  return to_reply_outcome(
      interpretation,
      proposed_category=proposed_id,
      decided_at=datetime.now(UTC),
  )
497 
498 
499def _target_summary(target: TargetState | None, names: dict[str, str]) -> str:
  """A short human description of an end-state, for a divergence note."""
  if target is None:
      return "(could not read)"
  allocation = target.allocation
  if isinstance(allocation, ResolvedCategory):
      label = names.get(str(allocation.category), str(allocation.category))
  else:
      label = "a split"
  return f"{label} — {target.memo}" if target.memo else label
509 
510 
511@activity.defn
512async def converge(
  snapshot: YnabSnapshot,
  instruction: InboundSignal,
  prior: Decision | None,
516) -> ConvergeOutcome:
  """Converge a REVISING run to its target and verify it (SPEC §3 rules 3-4).
518 
  The agent reads the revision instruction into a target (retarget / memo /
  no-change). The spine then converges to it: it reads the *current* YNAB
  end-state first and, comparing against the agent's last-applied decision
  (``prior``), skips a needless write (the no-op exit), adopts a write that
  already landed, or surfaces a divergence (a spouse edited it directly) —
  *before* overwriting it — rather than committing blind and only noticing on
  the read-back. Only a genuine change writes, then verifies field-by-field. A
  reconciled or closed-month transaction, a non-reply instruction, or anything
  the model under-specifies routes to a human rather than a silent edit.
  """
  import asyncio
  from datetime import UTC, datetime
531 
  from ynab_agent.agentic.converge import (
      RevisionRequest,
      interpret_revision,
      to_revision_plan,
  )
  from ynab_agent.domain.enums import DecidedBy
  from ynab_agent.domain.events import (
      CouldNotConfirm,
      Diverged,
      NeedsHuman,
      NoChange,
      Reapplied,
      VerifyOutcome,
  )
  from ynab_agent.domain.signals import ReplySignal
  from ynab_agent.policy.converge import (
      PrecommitAction,
      classify_verify,
      precommit_action,
      reconciliation_blocks,
      target_of,
  )
  from ynab_agent.ynab.client import YnabClient
555 
  if reconciliation_blocks(snapshot):
      return NeedsHuman(
          reason="reconciled or closed-month — propose, don't silently edit"
      )
  if not isinstance(instruction, ReplySignal):
      return NeedsHuman(reason="non-reply revision instruction unsupported")
562 
  client = YnabClient.from_env()
  spends = await asyncio.to_thread(client.category_spends)
  names = {str(spend.category): spend.name for spend in spends}
  current_name = (
      names.get(str(snapshot.category_id), str(snapshot.category_id))
      if snapshot.category_id is not None
      else "(uncategorized)"
  )
  target = await interpret_revision(
      RevisionRequest(
          instruction=instruction.text,
          current_category_name=current_name,
          candidates=_candidates_from_spends(spends),
          current_memo=snapshot.memo,
      )
  )
  plan = to_revision_plan(target)
  if not plan.changes:
      return NoChange()
582 
  category = (
      CategoryId(plan.category_id)
      if plan.category_id is not None
      else snapshot.category_id
  )
  if category is None:
      return NeedsHuman(reason="revision did not resolve a category")
  decision = Decision(
      allocation=ResolvedCategory(category=category),
      memo=plan.memo if plan.memo is not None else snapshot.memo,
      approved=True,
      decided_by=DecidedBy.HUMAN,
      decided_at=datetime.now(UTC),
  )
  end_state = target_of(decision)
  prior_state = target_of(prior) if prior is not None else None
599 
  # SPEC §3 r3-4: read the current end-state *before* writing, then decide.
  current = await asyncio.to_thread(client.read_back, snapshot.ynab_id)
  action = precommit_action(current, end_state, prior_state)
  if action is PrecommitAction.NO_CHANGE:
      return NoChange()
  if action is PrecommitAction.ALREADY_TARGET:
      return Reapplied(decision=decision)
  if action is PrecommitAction.DIVERGED:
      return Diverged(
          ynab_summary=_target_summary(current, names),
          requested_summary=_target_summary(end_state, names),
      )
612 
  # PrecommitAction.WRITE: converge to the target, then verify (SPEC §3 r3-4).
  await asyncio.to_thread(client.commit, snapshot.ynab_id, decision)
  read = await asyncio.to_thread(client.read_back, snapshot.ynab_id)
  verdict = classify_verify(read, end_state)
  if verdict is VerifyOutcome.MATCH:
      return Reapplied(decision=decision)
  if verdict is VerifyOutcome.COULD_NOT_CONFIRM:
      return CouldNotConfirm()
  return Diverged(
      ynab_summary=_target_summary(read, names),
      requested_summary=_target_summary(end_state, names),
  )
625 
626 
627@activity.defn
628async def feed_rule_learning(feed: FeedRuleLearning) -> None:
  """Persist a confirm/correct event into the durable rule registry (W5).
630 
  Signal-with-start on the singleton :class:`RuleRegistryWorkflow`: the first
  learning event creates it, every later one just delivers the signal, and the
  workflow folds the event into the rule table the autonomy gate reads (SPEC
  §9, §14). The conflict policy reuses the running singleton rather than
  starting a second registry.
  """
  from temporalio.common import WorkflowIDConflictPolicy
638 
  from ynab_agent.workflow.registry_types import (
      REGISTRY_WORKFLOW_ID,
      RegistryParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
644 
  temporal = await client()
  await temporal.start_workflow(
      "RuleRegistryWorkflow",
      RegistryParams(),
      id=REGISTRY_WORKFLOW_ID,
      task_queue=task_queue(),
      id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
      start_signal="record",
      start_signal_args=[feed],
  )
655 
656 
657@activity.defn
658async def record_auto_action(ynab_id: str) -> None:
  """Record a landed auto-action in the durable breaker ledger (SPEC §0.6).
660 
  Signal-with-start on the singleton ledger: the first auto-action creates it,
  every later one just delivers the signal, and the ledger folds it into the
  counts the hard floor reads (mirrors ``feed_rule_learning`` talking to the
  registry). Best-effort — a ledger hiccup must never block or fail the
  categorization it bounds, so any error is logged and swallowed. The breaker
  tolerates an occasional missed count; the per-txn ceiling still binds, and
  the ``ynab_id`` key dedups a retry.
  """
  from temporalio.common import WorkflowIDConflictPolicy
670 
  from ynab_agent.workflow.auto_action_types import (
      AUTO_ACTION_LEDGER_WORKFLOW_ID,
      LedgerParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
676 
  try:
      temporal = await client()
      await temporal.start_workflow(
          "AutoActionLedgerWorkflow",
          LedgerParams(),
          id=AUTO_ACTION_LEDGER_WORKFLOW_ID,
          task_queue=task_queue(),
          id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
          start_signal="record",
          start_signal_args=[ynab_id],
      )
  except Exception:
      activity.logger.warning(
          "auto-action ledger record failed (best-effort): %s", ynab_id
      )
692 
693 
694@activity.defn
695async def close_thread(thread_id: str) -> None:
  """Label and close the AgentMail thread on archive."""
  import asyncio
698 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
701 
  settings = Settings()
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.close, inbox_id=settings.inbox, thread_id=thread_id
  )

Real copy per purpose, and answerable dead-ends

render_body now has a branch per lifecycle moment, with the SPEC §3 copy: the handoff says it's yours now (and that a reply still works); the couldn't-confirm note says to check YNAB; the archive notice asks for a category or 'handled'; the FYI names the category, the standing rule, and the undo. A sweep test renders every purpose and asserts the placeholder never appears.

Every purpose has copy; the placeholder survives only as the fallback for an unmapped future purpose.

src/ynab_agent/agentic/compose.py · 286 lines

src/ynab_agent/agentic/compose.py286 lines · Python

⋯ 202 lines hidden (lines 1–202)1"""Render a transaction's email body (SPEC §5).
2 
3A deterministic template — *not* a model call. The model already did the
4thinking upstream (the proposal's category + one-line rationale + alternatives);
5this just lays it out cleanly so a glance is enough to act. Keeping it templated
6(rather than free-form prose) is a deliberate low-cognitive-load choice, and it
7drops a per-email model round-trip.
8 
9The layout, for a proposal:
10 
  Hulu — $13.07 — May 29
  <memo, when present — e.g. an Amazon item list>
13 
  Suggested: Entertainment   (or: Streaming, Fun Money)
  recurring streaming subscription
16 
  Just reply in your own words — confirm it, suggest a different category, …
18"""
19 
20from __future__ import annotations
21 
22from typing import TYPE_CHECKING
23 
24from ynab_agent.domain.base import Frozen
25from ynab_agent.domain.effects import MessagePurpose
26 
27if TYPE_CHECKING:
  from ynab_agent.budget.balance import BalanceOption
  from ynab_agent.domain.money import Money
30 
31_REPLY_HINT = (
  "Just reply in your own words — confirm it, suggest a different "
  "category, or ask a question."
34)
35 
36 
37class ComposeRequest(Frozen):
  """The facts the template lays out for one transaction email."""
39 
  purpose: str  # the MessagePurpose value: proposal / confirm / clarify / ...
  payee: str
  amount_display: str
  txn_date: str  # already display-formatted (e.g. "May 29")
  memo: str | None = None
  proposed_category: str | None = None  # the best-guess category NAME
  alternatives: tuple[str, ...] = ()  # other category names to offer
  rationale: str | None = None  # one-line reason for the best guess
  question: str | None = (
      None  # an explicit question, when the purpose has one
  )
  # Message-specific content carried from the state machine (the model's
  # clarify question, a diverged which-wins comparison) — see the
  # SendThreadMessage effect.
  detail: str | None = None
  # The category NAME a decision actually wrote (confirm / FYI / revision
  # summary / override notice), so those emails name what happened.
  decided_category: str | None = None
58 
59 
60def _facts(request: ComposeRequest) -> str:
  """The transaction header line, plus the memo line when there is one."""
  header = f"{request.payee} — {request.amount_display} — {request.txn_date}"
  if request.memo and request.memo.strip():
      return f"{header}\n{request.memo.strip()}"
  return header
66 
67 
68def _proposal_body(request: ComposeRequest, facts: str) -> str:
  suggested = (
      f"Suggested: {request.proposed_category or '(needs a category)'}"
  )
  if request.alternatives:
      suggested += f"   (or: {', '.join(request.alternatives)})"
  lines = [facts, "", suggested]
  if request.rationale:
      lines.append(request.rationale)
  lines += ["", _REPLY_HINT]
  return "\n".join(lines)
79 
80 
81def render_autonomy_offer(payee: str, category: str) -> str:
  """The one-time "want me to auto-handle this payee?" offer body (§14.7 3b).
83 
  A standalone yes/no message (its own thread), so a plain "yes"/"no" reply is
  unambiguous. The owner can also reply with the explicit standing command.
  """
  return (
      f"You've consistently filed {payee} under {category}, and I haven't "
      "had to correct it.\n\n"
      f"Want me to start auto-handling {payee} as {category} from now on? "
      "I'll apply it automatically, flag each one for you, and you can undo "
      "any of them with a one-word reply.\n\n"
      "Reply YES to let me, or NO to keep approving each one yourself."
  )
95 
96 
97def render_command_confirm(payee: str, category: str) -> str:
  """The read-back for an explicit "always X as Y" command (SPEC §5c, §0.6).
99 
  A standing command grants autonomy, so the agent echoes its interpretation
  and waits for a one-word confirm before blessing — a command can arrive on a
  brand-new thread where the allow-list is the only gate, so a misread or a
  mistaken send must not silently grant auto-apply.
  """
  return (
      f"You asked me to always categorize {payee} as {category}.\n\n"
      f"Reply YES to confirm — I'll auto-handle {payee} as {category} from "
      "now on, flag each one for you, and you can undo any with a one-word "
      "reply. Reply NO to keep approving each one yourself."
  )
111 
112 
113def render_offer_accepted(payee: str, category: str) -> str:
  """The confirmation sent when the owner accepts the offer (§14.7 3b)."""
  return (
      f"Great — I'll auto-handle {payee} as {category} from now on, and "
      "flag each one so you can see (and undo) it. Reply any time to change "
      "this."
  )
120 
121 
122def render_offer_declined(payee: str) -> str:
  """The brief note sent when the owner declines the offer (§14.7 3b)."""
  return (
      f"No problem — I'll keep proposing {payee} for you to approve, same "
      "as before."
  )
128 
129 
130def render_receipt_unsupported() -> str:
  """The honest note for a forwarded receipt the join can't process yet (§6).
132 
  The receipt⇄transaction join (W4) is a deferred increment, so rather than
  swallow a forwarded receipt silently, the agent acknowledges it and points
  the owner at the path that does work: replying on the transaction's own
  email thread.
  """
  return (
      "Thanks for forwarding this. I can't match forwarded receipts to "
      "transactions yet, so I haven't filed it.\n\n"
      "To add detail to a specific charge — an item list, a split, or a note "
      "— just reply on that transaction's own email thread and I'll fold it "
      "in."
  )
145 
146 
147def render_balance_options(
  needy_name: str, options: tuple[BalanceOption, ...]
149) -> str:
  """The balance offer: ways to cover an overspend, each explained (§8).
151 
  Numbered so the owner can reply "option 2", but a free-text answer ("take it
  from dining instead", "only $50") is read just as well. Each option leads
  with the model's rationale, the plain-English description of the moves.
  """
  lines = [
      f"{needy_name} is over budget. Here are some ways I can cover it by "
      "moving money between categories (nothing leaves your accounts):",
      "",
  ]
  for index, option in enumerate(options, start=1):
      lines.append(f"{index}. {option.label} — {option.rationale}")
  lines += [
      "",
      "Reply with the option you'd like (or your own tweak — e.g. "
      '"option 2 but only $50", or "no thanks").',
  ]
  return "\n".join(lines)
169 
170 
171def render_balance_applied(needy_name: str, total: Money) -> str:
  """The confirmation after a coverage plan is applied (SPEC §8)."""
  return (
      f"Done — moved {total} into {needy_name} to cover the overspend. "
      "Reply any time to adjust it."
  )
177 
178 
179def render_balance_declined(needy_name: str) -> str:
  """The brief note when the owner declines to cover (SPEC §8)."""
  return (
      f"No problem — I'll leave {needy_name} as is. Reply any time if you "
      "change your mind."
  )
185 
186 
187def render_balance_could_not_cover(needy_name: str) -> str:
  """The note when no safe coverage exists from current funds (SPEC §8)."""
  return (
      f"{needy_name} is over budget, but I couldn't find a safe way to cover "
      "it from your current funds. You may want to move money in manually."
  )
193 
194 
195def render_balance_failed(needy_name: str, reason: str) -> str:
  """The note when an approved plan can't be applied (SPEC §8)."""
  return (
      f"I couldn't cover {needy_name}: {reason}. Nothing was changed — reply "
      "and we can try another way."
  )
201 
202 
203def render_body(request: ComposeRequest) -> str:
  """Lay out the email body for one transaction message (SPEC §5).
205 
  The subject is templated separately by the caller; this is the body. Every
  purpose carries real copy for its lifecycle moment (SPEC §3 state notes) —
  none of these may fall back to a contentless placeholder.
  """
  facts = _facts(request)
  purpose = request.purpose
  if purpose == MessagePurpose.PROPOSAL.value:
      return _proposal_body(request, facts)
  if purpose == MessagePurpose.CONFIRM.value:
      category = (
          request.decided_category
          or request.proposed_category
          or "the category you picked"
      )
      return f"{facts}\n\nDone — set to {category} and approved."
  if purpose == MessagePurpose.FYI.value:
      filed = (
          f"Filed automatically as {request.decided_category}"
          if request.decided_category
          else "Filed automatically"
      )
      return (
          f"{facts}\n\n{filed} under your standing rule and flagged in "
          "YNAB so you can spot it. Reply here any time to change it."
      )
  if purpose == MessagePurpose.CLARIFY.value:
      question = (
          request.detail
          or request.question
          or "Which category should this be?"
      )
      return f"{facts}\n\n{question}"
  if purpose == MessagePurpose.REVISE_SUMMARY.value:
      updated = (
          f"Updated — now {request.decided_category}, re-approved."
          if request.decided_category
          else "Updated and re-approved."
      )
      return f"{facts}\n\n{updated} Reply if that's not right."
  if purpose == MessagePurpose.HANDOFF.value:
      return (
          f"{facts}\n\nI haven't heard back, so I'm leaving this one for "
          "you to handle in YNAB — no more nudges from me. A reply here "
          "still works any time."
      )
  if purpose == MessagePurpose.POSSIBLY_INCONSISTENT.value:
      return (
          f"{facts}\n\nI tried to update this in YNAB and couldn't confirm "
          "the change landed — please check it there, and reply with what "
          "you see so we end up in the right place."
      )
  if purpose == MessagePurpose.DIVERGED_READBACK.value:
      comparison = request.detail or (
          "YNAB now shows something different from what was asked for — "
          "which should win? Reply with your choice and I'll set it."
      )
      return f"{facts}\n\n{comparison}"
  if purpose == MessagePurpose.ARCHIVE_NOTICE.value:
      return (
          f"{facts}\n\nThis one is still uncategorized and its window is "
          "closing. Reply with a category and I'll file it — or reply "
          "'handled' if you've taken care of it."
      )
  if purpose == MessagePurpose.OVERRIDE_NOTICE.value:
      what = (
          f"you set this to {request.decided_category} yourself"
          if request.decided_category
          else "you recategorized this one yourself"
      )
      return (
          f"{facts}\n\nNoticed {what} — I've backed off auto-handling this "
          "payee and will go back to asking."
      )
  # An unmapped purpose would mean a new MessagePurpose without copy; say
  # something honest rather than nothing.
  note = (
      request.detail
      or request.question
      or ("A quick note on this transaction.")
  )
  return f"{facts}\n\n{note}"

Finally, interpret accepts a missing proposal. A flagged verify-failure entry or a NeedsHuman wait has no proposal; the old code answered every reply with the same canned question — an infinite loop. Now a reply that names a category commits it, and only a bare "ok" (nothing to approve) asks.

Approve-with-nothing-to-approve asks; a named category commits — flagged entries are answerable by email.

src/ynab_agent/agentic/interpret.py · 193 lines

src/ynab_agent/agentic/interpret.py193 lines · Python

⋯ 148 lines hidden (lines 1–148)1"""The reply-interpreting agent: what did the human mean? (SPEC §3, §5).
2 
3The agentic half of W2's ``interpret_inbound``. Given a human's free-text reply
4on a transaction's thread, the model reads the *intent*: approve the current
5proposal, switch to a different category, or — when unclear — ask one clarifying
6question. The model only classifies intent and (for a switch) picks a category;
7the deterministic spine assembles the ``Decision`` (stamping the human as
8decider and the time), so the model never fabricates an approval timestamp or
9an allocation the spine cannot resolve.
10 
11:func:`to_reply_outcome` maps the intent onto the domain ``ReplyOutcome``,
12defaulting to a clarifying question whenever a switch arrives without a category
13— the safe move is always to ask, never to guess a write.
14"""
15 
16from __future__ import annotations
17 
18from enum import StrEnum
19from typing import TYPE_CHECKING, assert_never
20 
21from pydantic import Field
22from pydantic_ai import Agent
23 
24from ynab_agent.agentic.enrich import CandidateCategory
25from ynab_agent.agentic.model import run_structured
26from ynab_agent.domain.allocations import ResolvedCategory
27from ynab_agent.domain.base import Frozen
28from ynab_agent.domain.enums import DecidedBy
29from ynab_agent.domain.ids import CategoryId
30from ynab_agent.domain.proposal import Decision
31from ynab_agent.workflow.types import (
32    AnswerOutcome,
33    ClarifyOutcome,
34    ReplyOutcome,
35)
36 
37if TYPE_CHECKING:
38    import datetime
39 
40    from pydantic_ai.models import Model
41 
42 
43class ReplyIntent(StrEnum):
44    """What a human's reply asked for."""
45 
46    APPROVE = "approve"
47    RECATEGORIZE = "recategorize"
48    CLARIFY = "clarify"
49 
50 
51class InterpretRequest(Frozen):
52    """A human reply and the context needed to read its intent.
53 
54    ``proposed_category_name`` is ``None`` when there is no live proposal (a
55    flagged verify-failure entry, a NeedsHuman wait). The owner can still name
56    a category outright; only a bare "approve" has nothing to approve then.
57    """
58 
59    reply_text: str
60    payee: str
61    amount_display: str
62    proposed_category_name: str | None = None
63    candidates: tuple[CandidateCategory, ...] = Field(min_length=1)
64 
65 
66class Interpretation(Frozen):
67    """The agent's read of the reply (mapped to a domain ReplyOutcome)."""
68 
69    intent: ReplyIntent
70    category_id: str | None = None
71    question: str | None = None
72    memo: str | None = None
73 
74 
75_SYSTEM_PROMPT = """\
76You read one reply a human sent about a proposed transaction categorization. You
77are given the reply text, the payee and amount, the category currently proposed,
78and the candidate categories (id + name).
79 
80Classify the reply's intent: `approve` if they accept the current proposal (e.g.
81"ok", "yes", "sounds good"); `recategorize` if they name or describe a category
82— then set `category_id` to the matching candidate id; or `clarify` if the reply
83is ambiguous or asks a question — then set a short `question` to send back. When
84in doubt, prefer `clarify`: asking again is safe, a wrong write is not.
85 
86There may be NO current proposal (shown as "(none)"): the agent asked an open
87question rather than proposing. A bare yes/ok then has nothing to approve —
88classify it `clarify`; a reply that names a category is still `recategorize`.
89 
90Also set `memo`: if the reply gives any *context or reasoning* beyond the bare
91category — what it was for, who it was for, an occasion ("gift for mom", "kids'
92soccer", "their shopping") — distil it into a short, factual memo (≤ a sentence)
93that will be saved on the transaction. If the reply is only a bare confirmation
94or category with no added context, leave `memo` null. Never invent detail."""
95 
96_AGENT: Agent[None, Interpretation] = Agent(
97    output_type=Interpretation,
98    system_prompt=_SYSTEM_PROMPT,
99)
100 
101 
102def _format_request(request: InterpretRequest) -> str:
103    """Render the request as the agent's user prompt."""
104    lines = [
105        f"Reply: {request.reply_text}",
106        f"Payee: {request.payee}",
107        f"Amount: {request.amount_display}",
108        f"Currently proposed: {request.proposed_category_name or '(none)'}",
109        "Candidate categories:",
110    ]
111    lines.extend(f"  - {c.name} (id: {c.id})" for c in request.candidates)
112    return "\n".join(lines)
113 
114 
115async def interpret(
116    request: InterpretRequest, *, model: Model | None = None
117) -> Interpretation:
118    """Run the reply-interpreting agent for one reply (SPEC §3).
119 
120    Args:
121        request: The reply text and its categorization context.
122        model: A model to use; defaults to the configured Ollama/Gemma.
123 
124    Returns:
125        The agent's structured reading of the reply.
126    """
127    return await run_structured(
128        _AGENT,
129        _format_request(request),
130        output_type=Interpretation,
131        model=model,
132    )
133 
134 
135def _human_decision(
136    category: CategoryId,
137    decided_at: datetime.datetime,
138    *,
139    memo: str | None = None,
140) -> Decision:
141    return Decision(
142        allocation=ResolvedCategory(category=category),
143        memo=memo,
144        approved=True,
145        decided_by=DecidedBy.HUMAN,
146        decided_at=decided_at,
147    )
148 
149 
150def to_reply_outcome(
151    interpretation: Interpretation,
152    *,
153    proposed_category: CategoryId | None,
154    decided_at: datetime.datetime,
155) -> ReplyOutcome:
156    """Map the agent's reading onto a domain ReplyOutcome (SPEC §3, §14.4).
157 
158    ``approve`` commits the proposed category (or asks, when nothing was
159    proposed — there is nothing to approve); ``recategorize`` commits the named
160    one (or asks, if none was given); ``clarify`` sends the question back. Any
161    rationale the reply carried rides along as the decision's ``memo`` (the
162    spine writes it to YNAB). The spine, not the model, sets decider and time.
163    """
164    match interpretation.intent:
165        case ReplyIntent.APPROVE:
166            if proposed_category is None:
167                return ClarifyOutcome(
168                    question=(
169                        "There's no pending suggestion to approve here — "
170                        "which category should this be?"
171                    )
172                )
173            return AnswerOutcome(
174                decision=_human_decision(
175                    proposed_category, decided_at, memo=interpretation.memo
176                )
177            )
178        case ReplyIntent.RECATEGORIZE:
179            if interpretation.category_id:
180                return AnswerOutcome(
181                    decision=_human_decision(
182                        CategoryId(interpretation.category_id),
183                        decided_at,
⋯ 10 lines hidden (lines 184–193)184                        memo=interpretation.memo,
185                    )
186                )
187            return ClarifyOutcome(question="Which category should this be?")
188        case ReplyIntent.CLARIFY:
189            return ClarifyOutcome(
190                question=interpretation.question
191                or "Could you clarify how you'd like this categorized?"
192            )
193    assert_never(interpretation.intent)