Making the auto-action circuit breaker load-bearing

What changed, and why

The hard floor (SPEC §0.6 Layer 1) is the deterministic cap that bounds a runaway poller or a mis-blessed rule regardless of the model — the dumb layer that catches disasters. One of its three rules is a per-run / per-day circuit breaker on the number of auto-actions. It was inert.

The gate reads the breaker through check_floor(amount, counters). But the counters it got were always AutoActionCounters() — literal zeros, hardcoded in _load_enrichment_inputs. So check_floor's TRIP_BREAKER branch could never fire. The per-transaction ceiling bounded each write's magnitude; nothing bounded the count. A single bad blessed rule could auto-apply across the whole backlog, each write under the ceiling, and the breaker would never trip.

This change gives the breaker real counts to read: a durable ledger of recent auto-actions, queried before the gate and recorded after each auto-apply. It mirrors the alert and overspend ledgers already in the codebase.

The ledger, as pure folds

The counting logic is pure: a frozen tail of (ynab_id, at) entries and two functions over it. record appends — dropping any existing entry for the same ynab_id first, so a retry or a re-enriched transaction counts once — and prunes to the day horizon so the carried state stays bounded. counters projects the tail into the AutoActionCounters the floor checks.

Two windows, two folds. record dedups by ynab_id and prunes; counters splits the tail into the run and day counts.

src/ynab_agent/policy/auto_action_ledger.py · 89 lines

src/ynab_agent/policy/auto_action_ledger.py89 lines · Python

⋯ 25 lines hidden (lines 1–25)1"""The auto-action ledger's pure state and folds — the circuit-breaker memory.
2 
3The hard floor's per-run / per-day circuit breaker (SPEC §0.6 Layer 1) caps how
4many auto-actions the agent may take, bounding a runaway poller or a mis-blessed
5rule regardless of the model. That cap can only bind if the counts are *real*:
6this module is the durable memory those counts derive from — an append-only tail
7of recent auto-actions, pruned to a day, that :func:`counters` reads into the
8:class:`~ynab_agent.policy.floor.AutoActionCounters` the floor checks. Without
9it the counters are always zero and the breaker can never trip.
10 
11A "run" has no shared identity across the per-transaction workflows, so it is
12approximated as a short rolling window — about one poll cycle's burst (the
13poller defaults to hourly). "Today" is the trailing 24 h. Both are rate limits
14the floor compares to its caps. Entries are keyed by ``ynab_id`` and
15deduplicated, so a retried record (or re-enriched transaction) counts once.
16"""
17 
18from __future__ import annotations
19 
20import datetime
21 
22from ynab_agent.domain.base import Frozen
23from ynab_agent.policy.floor import AutoActionCounters
24 
25# "This run" has no shared id across per-txn workflows, so approximate it as a
26# short rolling window — about one poll cycle (the poller defaults to hourly).
27RUN_WINDOW = datetime.timedelta(hours=1)
28# "Today" — the trailing day the per-day cap bounds, and the prune horizon, so
29# the carried tail never grows without limit.
30DAY_WINDOW = datetime.timedelta(hours=24)
31 
32 
33class AutoActionEntry(Frozen):
34    """One auto-action that landed: its transaction id and when it landed."""
35 
36    ynab_id: str
37    at: datetime.datetime
38 
39 
40class AutoActionLedgerState(Frozen):
41    """The pruned tail of recent auto-actions — the whole breaker memory.
42 
43    Held as Temporal workflow state and carried across continue-as-new, so it is
44    a frozen value like the rest of the domain.
45    """
46 
47    entries: tuple[AutoActionEntry, ...] = ()
48 
49 
50def record(
51    state: AutoActionLedgerState,
52    ynab_id: str,
53    now: datetime.datetime,
54    *,
55    retention: datetime.timedelta = DAY_WINDOW,
56) -> AutoActionLedgerState:
57    """Record an auto-action for ``ynab_id``; dedup it and prune. Pure.
58 
59    Any existing entry for the same ``ynab_id`` is dropped before appending,
60    so a retried record or a re-enriched transaction counts exactly once.
61    Entries older than ``retention`` are pruned, so the carried tail stays
62    bounded.
63    """
64    kept = tuple(
65        entry
66        for entry in state.entries
67        if entry.ynab_id != ynab_id and now - entry.at < retention
68    )
69    return AutoActionLedgerState(
70        entries=(*kept, AutoActionEntry(ynab_id=ynab_id, at=now))
71    )
72 
73 
74def counters(
75    state: AutoActionLedgerState,
76    now: datetime.datetime,
77    *,
78    run_window: datetime.timedelta = RUN_WINDOW,
79    day_window: datetime.timedelta = DAY_WINDOW,
80) -> AutoActionCounters:
81    """Project the tail into the floor's counters at ``now``. Pure.
82 
83    ``this_run`` is the count within ``run_window`` (the recent burst);
84    ``today`` is the count within ``day_window``. The floor (``check_floor``)
85    compares these to its per-run / per-day caps.
86    """
87    this_run = sum(1 for entry in state.entries if now - entry.at < run_window)
88    today = sum(1 for entry in state.entries if now - entry.at < day_window)
89    return AutoActionCounters(this_run=this_run, today=today)

The durable singleton that holds it

The tail lives as Temporal workflow state in a singleton ledger — born on the first auto-action via signal-with-start, living forever via continue-as-new. A thin shell over the pure folds, identical in shape to AlertLedgerWorkflow: a record signal folds in, a counters query reads out. The signal stamps workflow.now; the query takes now in its request (a query can't read the wall clock).

Signal folds, query reads, continue-as-new bounds history.

src/ynab_agent/workflow/auto_action_ledger_workflow.py · 60 lines

src/ynab_agent/workflow/auto_action_ledger_workflow.py60 lines · Python

⋯ 35 lines hidden (lines 1–35)1"""The durable auto-action circuit-breaker ledger (SPEC §0.6 Layer 1).
2 
3A singleton, long-lived workflow (id :data:`AUTO_ACTION_LEDGER_WORKFLOW_ID`)
4holding the recent-auto-action tail as Temporal state — the memory that makes
5the hard floor's per-run / per-day breaker actually *bind*. Without it the
6counters are always zero and the cap can never trip (SPEC §0.6). Born on the
7first auto-action — ``record_auto_action`` does a signal-with-start — and
8lives forever, continuing-as-new to keep its history bounded. A thin durable
9shell over the pure :mod:`ynab_agent.policy.auto_action_ledger` folds, exactly
10like ``alert_ledger_workflow`` over ``alert.ledger``:
11 
12* the ``counters`` query projects the tail into the floor's counters at a given
13  ``now`` without mutating anything;
14* the ``record`` signal folds a landed auto-action into the tail.
15 
16``now`` travels in with the query (a query cannot read the wall clock); the
17signal stamps ``workflow.now`` so replay stays deterministic.
18"""
19 
20from __future__ import annotations
21 
22from temporalio import workflow
23 
24from ynab_agent.workflow.auto_action_types import CountersRequest, LedgerParams
25 
26with workflow.unsafe.imports_passed_through():
27    from ynab_agent.policy.auto_action_ledger import (
28        AutoActionLedgerState,
29        counters,
30        record,
31    )
32    from ynab_agent.policy.floor import AutoActionCounters
33 
34 
35@workflow.defn
36class AutoActionLedgerWorkflow:
37    """The deployment's one durable auto-action tail (one fold per action)."""
38 
39    def __init__(self) -> None:
40        """Start empty; the run method adopts any carried-forward state."""
41        self._state = AutoActionLedgerState()
42 
43    @workflow.run
44    async def run(self, params: LedgerParams) -> None:
45        """Hold the tail, folding signals until history wants rolling."""
46        self._state = params.state
47        await workflow.wait_condition(
48            lambda: workflow.info().is_continue_as_new_suggested()
49        )
50        workflow.continue_as_new(LedgerParams(state=self._state))
51 
52    @workflow.signal
53    def record(self, ynab_id: str) -> None:
54        """Record that an auto-action for ``ynab_id`` just landed."""
55        self._state = record(self._state, ynab_id, now=workflow.now())
56 
57    @workflow.query
58    def counters(self, request: CountersRequest) -> AutoActionCounters:
59        """Project the tail into the floor's counters at ``request.now``."""
⋯ 1 line hidden (lines 60–60)60        return counters(self._state, request.now)

Reading real counts before the gate

The enrich activity now reads the live counts from the ledger and hands them to the gate, in place of the old zeros. The fallback matters: an unreachable or never-started ledger returns zeros — which is correct here, because a ledger that has never started genuinely means zero auto-actions, so 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.)

_load_auto_action_counters queries the ledger (zeros on absence); _load_enrichment_inputs threads the counts into the gate inputs.

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

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

⋯ 97 lines hidden (lines 1–97)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 
⋯ 454 lines hidden (lines 154–607)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 
237async def _read_for_compose(
  ynab_id: str,
239) -> tuple[YnabSnapshot, dict[str, str]]:
  """Re-read the YNAB snapshot and a category-id→name map for composing.
241 
  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
246 
  from ynab_agent.ynab.client import YnabClient
248 
  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
257 
258 
259def _render_message(
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
  purpose: MessagePurpose,
  names: dict[str, str],
264) -> str:
  """Lay out one message body for the thread (deterministic; SPEC §5).
266 
  The proposal's category + alternatives (model-chosen upstream) are resolved
  to names here and handed to the template — no model call at send time.
  """
  from ynab_agent.agentic.compose import ComposeRequest, render_body
271 
  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 ()
  )
  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,
  )
  return render_body(request)
293 
294 
295@activity.defn
296async def open_thread(ynab_id: str, proposal: Proposal | None) -> str:
  """Open the AgentMail thread by sending the proposal; returns its id.
298 
  A thread starts on its first send (AgentMail has no empty-thread create), so
  this composes + sends the proposal as the opening email. ``proposal`` is the
  current best guess (carried from workflow state); the txn facts + category
  names are re-read from YNAB (the source of truth) at compose time. The open
  is idempotent on the per-transaction label, so a retry re-finds the thread
  rather than sending a duplicate.
  """
  import asyncio
307 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
310 
  settings = Settings()
  snapshot, names = await _read_for_compose(ynab_id)
  body = _render_message(snapshot, proposal, MessagePurpose.PROPOSAL, names)
  proposed = (
      _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, proposed),
      body=body,
      txn_label=_txn_label(ynab_id),
  )
328 
329 
330@activity.defn
331async def send_thread_message(
  ynab_id: str,
  thread_id: str | None,
  purpose: MessagePurpose,
  action_seq: int,
  proposal: Proposal | None,
337) -> None:
  """Send a follow-up message on the transaction's thread.
339 
  ``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); other purposes derive their
  content from a re-read of the YNAB snapshot.
  """
  import asyncio
346 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
349 
  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)
  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),
  )
364 
365 
366def _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
374 
375 
376@activity.defn
377async def interpret_inbound(
  signal: InboundSignal,
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
381) -> ReplyOutcome:
  """Interpret a human reply into an answer or a question (SPEC §3, §5).
383 
  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, or a
  missing/split proposal, can't be answered directly, so ask rather than guess
  a write. Names + candidates are re-read from YNAB at interpret time.
  """
  import asyncio
  from datetime import UTC, datetime
392 
  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
401 
  proposed_id = _proposed_category_id(proposal)
  if not isinstance(signal, ReplySignal) or proposed_id is None:
      return ClarifyOutcome(
          question="Could you say which category this should be?"
      )
407 
  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)),
      candidates=_candidates_from_spends(spends),
  )
  interpretation = await interpret(request)
  return to_reply_outcome(
      interpretation,
      proposed_category=proposed_id,
      decided_at=datetime.now(UTC),
  )
424 
425 
426def _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
436 
437 
438@activity.defn
439async def converge(
  snapshot: YnabSnapshot, instruction: InboundSignal
441) -> ConvergeOutcome:
  """Converge a REVISING run to its target and verify it (SPEC §3).
443 
  The agent reads the revision instruction into a target (retarget / memo /
  no-change); the spine commits, then reads back and classifies the result. 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
451 
  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 (
      classify_verify,
      reconciliation_blocks,
      target_of,
  )
  from ynab_agent.ynab.client import YnabClient
473 
  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")
480 
  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()
500 
  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),
  )
  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, target_of(decision))
  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(target_of(decision), names),
  )
526 
527 
528@activity.defn
529async def feed_rule_learning(feed: FeedRuleLearning) -> None:
  """Persist a confirm/correct event into the durable rule registry (W5).
531 
  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
539 
  from ynab_agent.workflow.registry_types import (
      REGISTRY_WORKFLOW_ID,
      RegistryParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
545 
  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],
  )
556 
557 
558@activity.defn
559async def record_auto_action(ynab_id: str) -> None:
  """Record a landed auto-action in the durable breaker ledger (SPEC §0.6).
561 
  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
571 
  from ynab_agent.workflow.auto_action_types import (
      AUTO_ACTION_LEDGER_WORKFLOW_ID,
      LedgerParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
577 
  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
      )
593 
594 
595@activity.defn
596async def close_thread(thread_id: str) -> None:
  """Label and close the AgentMail thread on archive."""
  import asyncio
599 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
602 
  settings = Settings()
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.close, inbox_id=settings.inbox, thread_id=thread_id
  )

Recording an auto-action, through the layers

Recording follows the codebase's effect discipline rather than reaching out from the activity. The state machine emits a new RecordAutoAction effect when a blessed rule auto-applies (alongside the commit); the workflow dispatches it to a best-effort activity that signal-with-starts the ledger.

The new effect — a value the pure state machine can emit and a test can assert.

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

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

⋯ 103 lines hidden (lines 1–103)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    kind: Literal["send_message"] = "send_message"
63    purpose: MessagePurpose
64 
65 
66class CommitToYnab(Frozen):
67    """Commit a decision to YNAB (the spine does commit→verify; SPEC §0.5)."""
68 
69    kind: Literal["commit_to_ynab"] = "commit_to_ynab"
70    decision: Decision
71 
72 
73class SetTimer(Frozen):
74    """Arm a durable timer with an absolute deadline."""
75 
76    kind: Literal["set_timer"] = "set_timer"
77    timer: TimerKind
78    deadline: datetime.datetime
79 
80 
81class CancelTimer(Frozen):
82    """Cancel a previously armed timer."""
83 
84    kind: Literal["cancel_timer"] = "cancel_timer"
85    timer: TimerKind
86 
87 
88class FeedRuleLearning(Frozen):
89    """Feed a confirm/correct event to rule learning (W5; SPEC §9).
90 
91    ``payee`` is carried so W5 can key the rule's match on it without re-reading
92    the snapshot. For a correction, ``prior`` carries the decision being
93    overturned, so W5 can demote *the rule that produced the prior decision*
94    (§3 rule 6), not the new one.
95    """
96 
97    kind: Literal["feed_rule_learning"] = "feed_rule_learning"
98    event: RuleLearningKind
99    payee: str
100    decision: Decision | None = None
101    prior: Decision | None = None
102 
103 
104class RecordAutoAction(Frozen):
105    """Record a landed auto-action in the circuit-breaker ledger (SPEC §0.6).
106 
107    Emitted alongside the commit when a blessed rule auto-applies, so the hard
108    floor's per-run / per-day counts are real and the breaker can trip. Keyed by
109    ``ynab_id`` so a retry or re-enrichment counts the transaction once.
110    """
111 
112    kind: Literal["record_auto_action"] = "record_auto_action"
113    ynab_id: str
114 
⋯ 26 lines hidden (lines 115–140)115 
116class ReplayBuffered(Frozen):
117    """Re-deliver the signals buffered while in DISCOVERED (SPEC §3)."""
118 
119    kind: Literal["replay_buffered"] = "replay_buffered"
120    signals: tuple[InboundSignal, ...]
121 
122 
123class CloseThread(Frozen):
124    """Label and close the AgentMail thread on archive (SPEC §13)."""
125 
126    kind: Literal["close_thread"] = "close_thread"
127 
128 
129Effect = Annotated[
130    OpenThread
131    | SendThreadMessage
132    | CommitToYnab
133    | SetTimer
134    | CancelTimer
135    | FeedRuleLearning
136    | RecordAutoAction
137    | ReplayBuffered
138    | CloseThread,
139    Field(discriminator="kind"),
140]

Emitted on Enriching → AutoApplied, beside the commit. It records the attempt, keyed by ynab_id.

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

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

⋯ 230 lines hidden (lines 1–230)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
⋯ 385 lines hidden (lines 246–630)246                    # (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, ...] = (
              SendThreadMessage(purpose=applied_message),
              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)
364 
365 
366def _enter_flagged_awaiting(
  core: TxnCore,
  flag: AwaitingFlag,
  message: MessagePurpose,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
373) -> 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),
      SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
  )
381 
382 
383def _from_awaiting_human(
  txn: AwaitingHuman,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
389) -> Transition:
  match event:
      case AnswerReceived(decision=decision):
          return _advanced(
              Applied(core=txn.core, decision=decision),
              CommitToYnab(decision=decision),
              CancelTimer(timer=TimerKind.PATIENCE),
          )
      case ClarifyRequested():
          deadline = now + policy.patience_window
          return _advanced(
              AwaitingHuman(
                  core=txn.core,
                  proposal=txn.proposal,
                  patience_deadline=deadline,
                  flag=txn.flag,
              ),
              SendThreadMessage(purpose=MessagePurpose.CLARIFY),
              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
          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)
428 
429 
430def _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)
          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)
477 
478 
479def _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)
512 
513 
514def _from_revising(
  txn: Revising,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
520) -> 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():
                  return _enter_flagged_awaiting(
                      txn.core,
                      AwaitingFlag.DIVERGED,
                      MessagePurpose.DIVERGED_READBACK,
                      now=now,
                      policy=policy,
                  )
              case NeedsHuman():
                  deadline = now + policy.patience_window
                  return _advanced(
                      AwaitingHuman(
                          core=txn.core, patience_deadline=deadline
                      ),
                      SendThreadMessage(purpose=MessagePurpose.PROPOSAL),
                      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)
564 
565 
566def _reapply(
  txn: Revising,
  decision: Decision,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
572) -> Transition:
  """Re-applied revision: open, summarize, and feed learning (SPEC §3, §9).
574 
  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,
      SendThreadMessage(purpose=MessagePurpose.REVISE_SUMMARY),
      SetTimer(timer=TimerKind.ARCHIVE, deadline=archive),
  )
601 
602 
603def _resolve_no_change(
  txn: Revising,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
608) -> Transition:
  """No-change exit depends on history (SPEC §3 rule 5).
610 
  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),
  )
625 
626 
627def _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)

Signal-with-start on the singleton, mirroring feed_rule_learning. Swallows its own errors.

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

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

⋯ 558 lines hidden (lines 1–558)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 
237async def _read_for_compose(
  ynab_id: str,
239) -> tuple[YnabSnapshot, dict[str, str]]:
  """Re-read the YNAB snapshot and a category-id→name map for composing.
241 
  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
246 
  from ynab_agent.ynab.client import YnabClient
248 
  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
257 
258 
259def _render_message(
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
  purpose: MessagePurpose,
  names: dict[str, str],
264) -> str:
  """Lay out one message body for the thread (deterministic; SPEC §5).
266 
  The proposal's category + alternatives (model-chosen upstream) are resolved
  to names here and handed to the template — no model call at send time.
  """
  from ynab_agent.agentic.compose import ComposeRequest, render_body
271 
  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 ()
  )
  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,
  )
  return render_body(request)
293 
294 
295@activity.defn
296async def open_thread(ynab_id: str, proposal: Proposal | None) -> str:
  """Open the AgentMail thread by sending the proposal; returns its id.
298 
  A thread starts on its first send (AgentMail has no empty-thread create), so
  this composes + sends the proposal as the opening email. ``proposal`` is the
  current best guess (carried from workflow state); the txn facts + category
  names are re-read from YNAB (the source of truth) at compose time. The open
  is idempotent on the per-transaction label, so a retry re-finds the thread
  rather than sending a duplicate.
  """
  import asyncio
307 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
310 
  settings = Settings()
  snapshot, names = await _read_for_compose(ynab_id)
  body = _render_message(snapshot, proposal, MessagePurpose.PROPOSAL, names)
  proposed = (
      _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, proposed),
      body=body,
      txn_label=_txn_label(ynab_id),
  )
328 
329 
330@activity.defn
331async def send_thread_message(
  ynab_id: str,
  thread_id: str | None,
  purpose: MessagePurpose,
  action_seq: int,
  proposal: Proposal | None,
337) -> None:
  """Send a follow-up message on the transaction's thread.
339 
  ``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); other purposes derive their
  content from a re-read of the YNAB snapshot.
  """
  import asyncio
346 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
349 
  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)
  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),
  )
364 
365 
366def _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
374 
375 
376@activity.defn
377async def interpret_inbound(
  signal: InboundSignal,
  snapshot: YnabSnapshot,
  proposal: Proposal | None,
381) -> ReplyOutcome:
  """Interpret a human reply into an answer or a question (SPEC §3, §5).
383 
  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, or a
  missing/split proposal, can't be answered directly, so ask rather than guess
  a write. Names + candidates are re-read from YNAB at interpret time.
  """
  import asyncio
  from datetime import UTC, datetime
392 
  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
401 
  proposed_id = _proposed_category_id(proposal)
  if not isinstance(signal, ReplySignal) or proposed_id is None:
      return ClarifyOutcome(
          question="Could you say which category this should be?"
      )
407 
  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)),
      candidates=_candidates_from_spends(spends),
  )
  interpretation = await interpret(request)
  return to_reply_outcome(
      interpretation,
      proposed_category=proposed_id,
      decided_at=datetime.now(UTC),
  )
424 
425 
426def _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
436 
437 
438@activity.defn
439async def converge(
  snapshot: YnabSnapshot, instruction: InboundSignal
441) -> ConvergeOutcome:
  """Converge a REVISING run to its target and verify it (SPEC §3).
443 
  The agent reads the revision instruction into a target (retarget / memo /
  no-change); the spine commits, then reads back and classifies the result. 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
451 
  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 (
      classify_verify,
      reconciliation_blocks,
      target_of,
  )
  from ynab_agent.ynab.client import YnabClient
473 
  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")
480 
  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()
500 
  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),
  )
  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, target_of(decision))
  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(target_of(decision), names),
  )
526 
527 
528@activity.defn
529async def feed_rule_learning(feed: FeedRuleLearning) -> None:
  """Persist a confirm/correct event into the durable rule registry (W5).
531 
  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
539 
  from ynab_agent.workflow.registry_types import (
      REGISTRY_WORKFLOW_ID,
      RegistryParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
545 
  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],
  )
556 
557 
558@activity.defn
559async def record_auto_action(ynab_id: str) -> None:
  """Record a landed auto-action in the durable breaker ledger (SPEC §0.6).
561 
  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
571 
  from ynab_agent.workflow.auto_action_types import (
      AUTO_ACTION_LEDGER_WORKFLOW_ID,
      LedgerParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
577 
  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
      )
⋯ 15 lines hidden (lines 593–607)593 
594 
595@activity.defn
596async def close_thread(thread_id: str) -> None:
  """Label and close the AgentMail thread on archive."""
  import asyncio
599 
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
602 
  settings = Settings()
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.close, inbox_id=settings.inbox, thread_id=thread_id
  )

The dispatch also swallows an ActivityError — a timeout the activity body can't catch.

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

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

⋯ 302 lines hidden (lines 1–302)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,
      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.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
105 
106# History-length ceiling before a resting workflow continues-as-new. High enough
107# that ordinary flows never trip it; long-lived (30-45 day) transactions do.
108_CONTINUE_AS_NEW_AFTER = 4_000
109 
110# Resting states the workflow may sit in for days while waiting on signals or
111# timers; continue-as-new fires only from one of these. DISCOVERED counts: a
112# transaction born from a signal can wait here for a slow YNAB import (SPEC §3).
113_RESTING = (Discovered, AwaitingHuman, Open, Lapsed, HoldAmazon)
114 
115 
116def _is_amazon(payee: str) -> bool:
  return "amazon" in payee.lower()
118 
119 
120def _hold_for_amazon(snapshot: YnabSnapshot) -> bool:
  """Whether to hold for Amazon item detail: Amazon-ish and no memo yet."""
  return _is_amazon(snapshot.payee) and not snapshot.has_memo
123 
124 
125@workflow.defn
126class TransactionWorkflow:
  """The durable per-transaction lifecycle."""
128 
  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
140 
  # ── 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)
146 
  @workflow.signal
  def notify_snapshot(self, snapshot: YnabSnapshot) -> None:
      """W1 materialized (or backfilled the memo of) the YNAB snapshot."""
      self._snapshot_ready = snapshot
151 
  @workflow.query
  def state(self) -> str:
      """The current lifecycle state (for observability)."""
      return self._txn.state.value
156 
  # ── 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()
170 
      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
201 
  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})"
208 
  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,
      )
220 
  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
240 
  # ── 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)
254 
  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):
          tid = await workflow.execute_activity(
              activities.open_thread,
              args=[self._ynab_id, self._proposal()],
              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)])
      elif isinstance(effect, SendThreadMessage):
          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(),
              ],
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
      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:
⋯ 245 lines hidden (lines 323–567)323                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
336 
  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})
343 
  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)
350 
  def _proposal(self) -> Proposal | None:
      """The current best-guess proposal, for states that carry one.
353 
      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
361 
  # ── 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())
          )
395 
  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))
404 
  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())
          )
420 
  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)
436 
  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)
          )
453 
  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())
          )
480 
  async def _archive_or_override(self, st: Open) -> LifecycleEvent:
      """At archive time, detect a manual YNAB edit (SPEC §14.2).
483 
      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()
506 
  async def _on_lapsed(self) -> None:
      await self._wait_then_revise_or_archive()
509 
  def _has_inbound(self) -> bool:
      return len(self._inbound) > 0
512 
  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())
          )
537 
  async def _wait_forever(self, predicate: Callable[[], bool]) -> bool:
      await workflow.wait_condition(predicate)
      return True
541 
  async def _on_revising(self, st: Revising) -> None:
      outcome = await workflow.execute_activity(
          activities.converge,
          args=[st.core.snapshot, st.instruction],
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=ACTIVITY_RETRY,
      )
      await self._dispatch(Converged(outcome=outcome))
550 
  async def _wait_until(
      self,
      deadline: datetime.datetime,
      predicate: Callable[[], bool],
  ) -> bool:
      """Wait for ``predicate`` until an absolute deadline.
557 
      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

Tests

The pure folds pin the behaviour that's easy to get wrong: a re-recorded ynab_id counts once, old entries prune, and the run/day windows split a tail correctly.

Dedup, prune, and the split windows.

tests/policy/test_auto_action_ledger.py · 52 lines

tests/policy/test_auto_action_ledger.py52 lines · Python

⋯ 23 lines hidden (lines 1–23)1"""Tests for the auto-action circuit-breaker ledger folds (SPEC §0.6)."""
2 
3from __future__ import annotations
4 
5import datetime
6 
7from ynab_agent.policy.auto_action_ledger import (
8    AutoActionLedgerState,
9    counters,
10    record,
11)
12 
13_NOW = datetime.datetime(2026, 6, 8, 12, 0, tzinfo=datetime.UTC)
14 
15 
16def _ago(**kw: float) -> datetime.datetime:
17    return _NOW - datetime.timedelta(**kw)
18 
19 
20def test_record_appends_an_entry() -> None:
21    state = record(AutoActionLedgerState(), "t1", _NOW)
22    assert [entry.ynab_id for entry in state.entries] == ["t1"]
23 
24 
25def test_record_dedups_the_same_transaction() -> None:
26    # A retried record or a re-enriched txn must count once, not twice.
27    state = record(AutoActionLedgerState(), "t1", _ago(minutes=30))
28    state = record(state, "t1", _NOW)
29    assert len(state.entries) == 1
30    assert state.entries[0].at == _NOW  # the latest timestamp wins
31 
32 
33def test_record_prunes_entries_older_than_retention() -> None:
34    state = record(AutoActionLedgerState(), "old", _ago(hours=25))
35    state = record(state, "t2", _NOW)
36    assert [entry.ynab_id for entry in state.entries] == ["t2"]
37 
38 
39def test_counters_split_run_and_day_windows() -> None:
40    state = AutoActionLedgerState()
41    state = record(state, "t1", _ago(minutes=10))  # within run + day
42    state = record(state, "t2", _ago(minutes=50))  # within run + day
43    state = record(state, "t3", _ago(hours=5))  # within day only (run = 1 h)
44    result = counters(state, _NOW)
45    assert result.this_run == 2
46    assert result.today == 3
⋯ 6 lines hidden (lines 47–52)47 
48 
49def test_counters_empty_ledger_is_zero() -> None:
50    result = counters(AutoActionLedgerState(), _NOW)
51    assert result.this_run == 0
52    assert result.today == 0

The workflow test drives the singleton on the time-skipping server — record, count, then a query two hours on showing the run window rolled off while the day window holds. And the W2 workflow test asserts the auto-apply's record reaches the activity end-to-end.

Record → counts → window roll-off, on the durable workflow.

tests/workflow/test_auto_action_ledger_workflow.py · 68 lines

tests/workflow/test_auto_action_ledger_workflow.py68 lines · Python

⋯ 45 lines hidden (lines 1–45)1"""The durable auto-action circuit-breaker ledger on the time-skipping server.
2 
3Verifies the wiring around the pure folds (tested in ``tests/policy``): the
4``record`` signal folds an auto-action in, and the ``counters`` query reads the
5per-run / per-day counts back out. The query carries ``now`` so the window maths
6line up with the ``workflow.now()`` the signal records at.
7"""
8 
9from __future__ import annotations
10 
11import datetime
12 
13from temporalio.testing import WorkflowEnvironment
14from temporalio.worker import Worker
15 
16from ynab_agent.workflow.auto_action_ledger_workflow import (
17    AutoActionLedgerWorkflow,
18)
19from ynab_agent.workflow.auto_action_types import (
20    AUTO_ACTION_LEDGER_WORKFLOW_ID,
21    CountersRequest,
22    LedgerParams,
23)
24from ynab_agent.workflow.runtime import DATA_CONVERTER
25 
26_TASK_QUEUE = "auto-action-ledger-test"
27 
28 
29async def test_record_then_count_and_window() -> None:
30    async with (
31        await WorkflowEnvironment.start_time_skipping(
32            data_converter=DATA_CONVERTER
33        ) as env,
34        Worker(
35            env.client,
36            task_queue=_TASK_QUEUE,
37            workflows=[AutoActionLedgerWorkflow],
38        ),
39    ):
40        handle = await env.client.start_workflow(
41            AutoActionLedgerWorkflow.run,
42            LedgerParams(),
43            id=AUTO_ACTION_LEDGER_WORKFLOW_ID,
44            task_queue=_TASK_QUEUE,
45        )
46        now = await env.get_current_time()
47 
48        # Fresh ledger → zero counts.
49        empty = await handle.query(
50            AutoActionLedgerWorkflow.counters, CountersRequest(now=now)
51        )
52        assert (empty.this_run, empty.today) == (0, 0)
53 
54        # Two auto-actions land; the same txn signalled twice counts once.
55        await handle.signal(AutoActionLedgerWorkflow.record, "txn-1")
56        await handle.signal(AutoActionLedgerWorkflow.record, "txn-2")
57        await handle.signal(AutoActionLedgerWorkflow.record, "txn-1")
58        counts = await handle.query(
59            AutoActionLedgerWorkflow.counters, CountersRequest(now=now)
60        )
61        assert (counts.this_run, counts.today) == (2, 2)
62 
63        # Two hours on, the run window has rolled off but the day window holds.
64        later = now + datetime.timedelta(hours=2)
65        rolled = await handle.query(
66            AutoActionLedgerWorkflow.counters, CountersRequest(now=later)
67        )
68        assert (rolled.this_run, rolled.today) == (0, 2)