Earned autonomy, end to end — a code-level reading (PR #10)

What changed, and why

This PR closes the loop on earned autonomy: the agent can now offer to take a payee over, guard every auto-apply, and revoke the grant when it gets something wrong — and a smaller fix stops it ever reading an autoresponder as a "yes".

Before, a learned rule could climb to eligible but nothing ever surfaced it (eligible_for_bless had zero callers); the only path to autonomy was the owner proactively emailing a standing command. Now the lifecycle is complete:

- Earn — a rule reaches eligibility only after K (raised 3 → 5) consistent same-way confirmations. - Offer (SPEC §14.7 3b) — the registry volunteers a one-time "want me to auto-handle Spotify?" email; a free-form "yes" blesses the rule. - Guard (§0.6 Layer 2) — before any auto-apply lands, an independent model review can veto it. - Revoke (§14.8) — a silent in-YNAB edit, or an explicit correction, demotes the rule back to Observe.

The design choice that keeps it simple: the offer is its own email thread, so a reply there is unambiguously a bless-acceptance, never a category reply — the pure per-transaction state machine is untouched. Read the pieces below; each is small and independently safe.

The fix: an autoresponder is not a confirmation

The deterministic classifier already refused to act on an autoresponder — but it keyed on an is_auto_reply flag the webhook never set, so the guard was dead and an owner's out-of-office reply on a transaction thread was read as a categorization. The fix derives the flag from the headers AgentMail already surfaces, in the one place that builds the domain message.

RFC 3834 Auto-Submitted, the bulk/auto Precedence values, and the vendor markers — normalised case-insensitively — then wired into to_inbound.

src/ynab_agent/webhook/app.py · 249 lines

src/ynab_agent/webhook/app.py249 lines · Python

⋯ 82 lines hidden (lines 1–82)1"""W3 · the inbound webhook receiver (SPEC §5).
2 
3A tiny HTTP front door, deployed alongside the worker. AgentMail POSTs a
4Svix-signed event here when an email arrives; this app verifies the signature
5(proving the request really came from AgentMail — the ``signature_verified``
6provenance check, SPEC §0.6), maps the message onto the domain
7``InboundMessage``, and starts a ``DispatchWorkflow`` on Temporal. The
8*dispatcher* then does the routing (reply / receipt / command / quarantine);
9this only authenticates the transport and hands off.
10 
11AgentMail already authenticates the *sender* (SPF/DKIM/DMARC) and surfaces it as
12the event type, so only ``message.received`` is acted on — ``*.unauthenticated``
13/ ``*.spam`` / ``*.blocked`` are ignored here. The Temporal client lives on
14``app.state`` so tests inject one without a real connection.
15"""
16 
17from __future__ import annotations
18 
19import os
20from contextlib import asynccontextmanager
21from typing import TYPE_CHECKING
22 
23from fastapi import FastAPI, HTTPException, Request
24from pydantic import BaseModel, ConfigDict, Field
25 
26from ynab_agent.dispatch.classify import InboundMessage
27from ynab_agent.domain.ids import MessageId, ThreadId
28from ynab_agent.workflow.dispatch_types import DispatchParams
29from ynab_agent.workflow.dispatch_workflow import DispatchWorkflow
30 
31if TYPE_CHECKING:
32    from collections.abc import AsyncIterator, Mapping
33 
34    from temporalio.client import Client
35 
36WEBHOOK_SECRET_ENV = "AGENTMAIL_WEBHOOK_SECRET"
37_RECEIVED = "message.received"
38 
39 
40class _WireMessage(BaseModel):
41    """The message fields we read from an AgentMail webhook (loose)."""
42 
43    model_config = ConfigDict(extra="ignore", populate_by_name=True)
44 
45    message_id: str
46    from_address: str = Field(alias="from")
47    subject: str = ""
48    text: str | None = None
49    # The new reply text with the quoted history stripped — what the human
50    # actually wrote. Preferred over ``text`` so the interpreter isn't fed the
51    # whole quoted proposal back.
52    extracted_text: str | None = None
53    thread_id: str | None = None
54    # The raw RFC 822 headers AgentMail surfaces (a flat name→value dict). Read
55    # only to detect machine-generated mail (see ``_is_auto_reply``); ``None``
56    # when AgentMail omits them.
57    headers: dict[str, str] | None = None
58 
59 
60class _WebhookPayload(BaseModel):
61    """An AgentMail webhook event — only the parts we use.
62 
63    The Svix envelope's top-level ``type`` is always the literal ``"event"``;
64    the actual event name (``message.received`` and its ``.spam`` / ``.blocked``
65    / ``.unauthenticated`` variants) is in ``event_type``. We act only on an
66    exact ``message.received``.
67    """
68 
69    model_config = ConfigDict(extra="ignore")
70 
71    event_type: str = ""
72    message: _WireMessage | None = None
73 
74 
75# RFC 3834 + de-facto vendor markers that a message was generated by a machine
76# (a vacation/auto-reply, a bounce, or a bulk mailer) rather than typed by a
77# human. Such a message must never be read as a confirmation — "silence is not
78# consent" (SPEC §0.6 inbound boundary, §5d).
79_AUTO_PRECEDENCE = frozenset({"bulk", "auto_reply", "junk"})
80_AUTO_MARKER_HEADERS = ("x-autoreply", "x-autorespond", "list-id")
81 
82 
83def _is_auto_reply(headers: Mapping[str, str] | None) -> bool:
84    """True if the inbound headers mark the message as machine-generated.
85 
86    Per RFC 3834, ``Auto-Submitted`` present and set to anything but ``no``
87    means the message was produced automatically (the canonical vacation /
88    auto-reply signal). We also honor the de-facto ``Precedence:
89    bulk/auto_reply/junk`` convention and the vendor ``X-Autoreply`` /
90    ``X-Autorespond`` markers, plus ``List-Id`` (bulk/list traffic). Header
91    names are case-insensitive, so the lookup is normalized to lower case.
92    """
93    if not headers:
94        return False
95    lowered = {
96        name.lower(): value.strip().lower() for name, value in headers.items()
97    }
98    auto_submitted = lowered.get("auto-submitted")
99    if auto_submitted is not None and auto_submitted != "no":
100        return True
101    if lowered.get("precedence") in _AUTO_PRECEDENCE:
102        return True
103    return any(header in lowered for header in _AUTO_MARKER_HEADERS)
104 
105 
106def to_inbound(message: _WireMessage, *, verified: bool) -> InboundMessage:
107    """Map a verified webhook message onto the domain InboundMessage."""
108    return InboundMessage(
109        message_id=MessageId(message.message_id),
110        from_address=message.from_address,
111        subject=message.subject,
112        body=message.extracted_text or message.text or "",
113        thread_id=ThreadId(message.thread_id) if message.thread_id else None,
114        signature_verified=verified,
115        is_auto_reply=_is_auto_reply(message.headers),
116    )
⋯ 133 lines hidden (lines 117–249)117 
118 
119def verify(body: bytes, headers: Mapping[str, str], secret: str) -> object:
120    """Verify an AgentMail (Svix) webhook and return its parsed payload.
121 
122    Raises ``svix.webhooks.WebhookVerificationError`` if the signature is bad.
123    """
124    from svix.webhooks import Webhook
125 
126    return Webhook(secret).verify(body, dict(headers))
127 
128 
129async def start_dispatch(
130    client: Client,
131    inbound: InboundMessage,
132    *,
133    allowlist: frozenset[str],
134    task_queue: str,
135) -> bool:
136    """Start a DispatchWorkflow for one message; idempotent (SPEC §5).
137 
138    The workflow id is derived from the message id, and the id-reuse policy
139    rejects duplicates (running *or* completed), so an AgentMail webhook retry
140    is a no-op rather than a re-dispatch. Returns ``True`` if newly started,
141    ``False`` if it had already been dispatched.
142    """
143    from temporalio.common import WorkflowIDReusePolicy
144    from temporalio.exceptions import WorkflowAlreadyStartedError
145 
146    try:
147        await client.start_workflow(
148            DispatchWorkflow.run,
149            DispatchParams(message=inbound, allowlist=allowlist),
150            id=f"dispatch-{inbound.message_id}",
151            task_queue=task_queue,
152            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
153        )
154    except WorkflowAlreadyStartedError:
155        return False
156    return True
157 
158 
159def _allowlist_from_env() -> frozenset[str]:
160    from ynab_agent.settings import Settings
161 
162    return frozenset(address.lower() for address in Settings().owners)
163 
164 
165async def _connect() -> Client:
166    from temporalio.client import Client
167    from temporalio.contrib.pydantic import pydantic_data_converter
168 
169    from ynab_agent.telemetry import metrics_runtime, tracing_interceptors
170 
171    return await Client.connect(
172        os.environ.get("TEMPORAL_HOST", "localhost:7233"),
173        namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
174        data_converter=pydantic_data_converter,
175        interceptors=tracing_interceptors(),
176        runtime=metrics_runtime(),
177    )
178 
179 
180def create_app(
181    *,
182    webhook_secret: str | None = None,
183    allowlist: frozenset[str] | None = None,
184    task_queue: str | None = None,
185) -> FastAPI:
186    """Build the webhook app.
187 
188    Production reads its config from the environment; tests pass it in and set
189    ``app.state.temporal`` before startup.
190    """
191    secret = (
192        webhook_secret
193        if webhook_secret is not None
194        else os.environ.get(WEBHOOK_SECRET_ENV)
195    )
196    queue = task_queue or os.environ.get("TEMPORAL_TASK_QUEUE", "ynab-agent")
197    allow = allowlist if allowlist is not None else _allowlist_from_env()
198 
199    @asynccontextmanager
200    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
201        if not hasattr(app.state, "temporal"):
202            app.state.temporal = await _connect()
203        yield
204        # uvicorn fires this on SIGTERM; flush buffered spans before exit.
205        from ynab_agent.telemetry import shutdown_tracing
206 
207        shutdown_tracing()
208 
209    app = FastAPI(lifespan=lifespan, title="ynab-agent webhook")
210 
211    from ynab_agent.telemetry import instrument_fastapi, setup_tracing
212 
213    setup_tracing("ynab-agent-webhook")
214    instrument_fastapi(app)
215 
216    @app.get("/healthz")
217    async def healthz() -> dict[str, str]:
218        return {"status": "ok"}
219 
220    @app.post("/webhooks/agentmail")
221    async def agentmail_webhook(request: Request) -> dict[str, str]:
222        if not secret:
223            raise HTTPException(
224                status_code=500, detail="webhook secret not configured"
225            )
226        body = await request.body()
227        from svix.webhooks import WebhookVerificationError
228 
229        try:
230            payload = verify(body, request.headers, secret)
231        except WebhookVerificationError as err:
232            raise HTTPException(
233                status_code=401, detail="invalid signature"
234            ) from err
235 
236        event = _WebhookPayload.model_validate(payload)
237        if event.event_type != _RECEIVED or event.message is None:
238            return {"status": "ignored"}
239 
240        inbound = to_inbound(event.message, verified=True)
241        started = await start_dispatch(
242            request.app.state.temporal,
243            inbound,
244            allowlist=allow,
245            task_queue=queue,
246        )
247        return {"status": "dispatched" if started else "duplicate"}
248 
249    return app

The pure core: a one-time marker and three folds

Everything stateful about the offer lives in the registry's pure folds, so it is trivially testable and carries across continue-as-new like the rest of the domain. A rule gains an offered_at marker; three new folds read and advance it.

pending_offers (eligible AND never offered), mark_offered (idempotent), bless_by_id (grant by id — never creates, so a stale offer can't resurrect).

src/ynab_agent/learn/registry.py · 221 lines

src/ynab_agent/learn/registry.py221 lines · Python

⋯ 137 lines hidden (lines 1–137)1"""The durable rule registry's pure state and folds (SPEC §14, W5).
2 
3The registry is the agent's *only* persistent learning memory: the rule table
4that earns autonomy. Per the derived-state rule (SPEC §0.5) it lives as Temporal
5workflow state — never an external store — and everything else (what a payee
6*usually* maps to, how consistent it has been) is derived from YNAB's canonical
7history on demand, not duplicated here.
8 
9This module is the pure core the durable :class:`RuleRegistryWorkflow` wraps
10(mirroring how ``state_machine`` sits under ``txn_workflow``): given the current
11:class:`RegistryState` and one learning input, return the next state. It folds
12in the already-pure transition logic (``plan_rule_update`` / ``apply_learning``)
13and keeps a bounded audit tail so "why did it learn that?" is answerable without
14an external log.
15 
16Autonomy is opt-in (SPEC §14): a *learned* rule that reaches ``trusted`` by
17consistency is only **eligible** for auto-apply — :func:`eligible_for_bless`
18surfaces it for the one-time owner prompt — and the gate auto-applies it only
19once the owner blesses it (``source=human_explicit``). Nothing here grants
20autonomy on silence.
21"""
22 
23from __future__ import annotations
24 
25import datetime
26from typing import TYPE_CHECKING
27 
28from pydantic import Field
29 
30from ynab_agent.domain.base import Frozen
31from ynab_agent.domain.enums import RuleSource, TrustState
32from ynab_agent.domain.rule import Rule
33from ynab_agent.learn.handler import plan_rule_update
34from ynab_agent.learn.transitions import (
35    RuleChange,
36    RuleChangeKind,
37    apply_learning,
38)
39 
40if TYPE_CHECKING:
41    from ynab_agent.domain.effects import FeedRuleLearning
42    from ynab_agent.domain.ids import RuleId
43    from ynab_agent.learn.events import ExplicitCommand
44 
45# How many recent rule-table changes the registry keeps for the in-band audit
46# (SPEC §9's interim log until the externalized audit subsystem lands). The tail
47# is bounded so continued learning never grows the carried state without limit.
48AUDIT_CAP = 200
49 
50 
51class RegistryAuditEntry(Frozen):
52    """One rule-table change, time-stamped and keyed to its payee (SPEC §9)."""
53 
54    at: datetime.datetime
55    payee: str
56    change: RuleChange
57 
58 
59class RegistryState(Frozen):
60    """The learned rule table plus a bounded change history.
61 
62    The whole of the agent's persistent learning memory. Held as Temporal
63    workflow state and carried across continue-as-new, so it is a frozen value
64    like the rest of the domain.
65    """
66 
67    rules: tuple[Rule, ...] = ()
68    audit: tuple[RegistryAuditEntry, ...] = Field(default=())
69 
70 
71def _appended_audit(
72    state: RegistryState, entry: RegistryAuditEntry
73) -> tuple[RegistryAuditEntry, ...]:
74    """The audit tail with ``entry`` appended, capped to :data:`AUDIT_CAP`."""
75    return (*state.audit, entry)[-AUDIT_CAP:]
76 
77 
78def record_learning(
79    state: RegistryState,
80    feed: FeedRuleLearning,
81    *,
82    now: datetime.datetime,
83    next_id: RuleId,
84) -> RegistryState:
85    """Fold a W2 confirm/correct effect into the registry (SPEC §9). Pure.
86 
87    Returns the state unchanged when there is nothing learnable (no decision,
88    or a split whose share template is the agent's call — ``plan_rule_update``).
89    """
90    outcome = plan_rule_update(state.rules, feed, now=now, next_id=next_id)
91    if outcome is None:
92        return state
93    entry = RegistryAuditEntry(at=now, payee=feed.payee, change=outcome.change)
94    return state.model_copy(
95        update={"rules": outcome.rules, "audit": _appended_audit(state, entry)}
96    )
97 
98 
99def bless_rule(
100    state: RegistryState,
101    command: ExplicitCommand,
102    *,
103    now: datetime.datetime,
104    next_id: RuleId,
105) -> RegistryState:
106    """Bless a ``(match, action)`` straight to ``trusted`` (SPEC §14). Pure.
107 
108    The owner's opt-in: an eligible learned rule becomes auto-applicable, or an
109    explicit standing command ("always categorize X as Y") seeds one already
110    trusted. Either way ``source`` becomes ``human_explicit`` — granted, not
111    earned — which is what the gate requires before it acts alone.
112    """
113    outcome = apply_learning(state.rules, command, now=now, next_id=next_id)
114    entry = RegistryAuditEntry(
115        at=now, payee=command.match.payee_pattern, change=outcome.change
116    )
117    return state.model_copy(
118        update={"rules": outcome.rules, "audit": _appended_audit(state, entry)}
119    )
120 
121 
122def eligible_for_bless(state: RegistryState) -> tuple[Rule, ...]:
123    """Learned rules that reached ``trusted`` by consistency but aren't blessed.
124 
125    These are the on-ramp surface (SPEC §14): a payee the owner has confirmed
126    enough times that the agent *could* take it over, awaiting the one-time
127    "want me to auto-handle this from now on?" opt-in. A rule already blessed
128    (``human_explicit``) is excluded — it is past eligibility, already trusted.
129    """
130    return tuple(
131        rule
132        for rule in state.rules
133        if rule.trust is TrustState.TRUSTED
134        and rule.source is RuleSource.LEARNED
135    )
136 
137 
138def pending_offers(state: RegistryState) -> tuple[Rule, ...]:
139    """Eligible rules whose one-time autonomy offer has not been sent yet (3b).
140 
141    The registry workflow drives off this: a rule that has just become eligible
142    and was never offered is volunteered to the owner exactly once. Once the
143    offer goes out, :func:`mark_offered` stamps ``offered_at`` and it drops off
144    this list, so a later confirmation of the same payee never re-asks.
145    """
146    return tuple(
147        rule for rule in eligible_for_bless(state) if rule.offered_at is None
148    )
149 
150 
151def mark_offered(
152    state: RegistryState, rule_id: RuleId, *, now: datetime.datetime
153) -> RegistryState:
154    """Record that the autonomy offer was sent for ``rule_id`` (SPEC §14.7 3b).
155 
156    Idempotent: stamping a rule already marked (or one no longer present) leaves
157    the table unchanged. This is the one-time guard — combined with the offer
158    workflow's id-reuse rejection — that the proactive prompt is sent only once.
159    """
160    updated = tuple(
161        rule.model_copy(update={"offered_at": now})
162        if rule.id == rule_id and rule.offered_at is None
163        else rule
164        for rule in state.rules
165    )
166    if updated == state.rules:
167        return state
168    return state.model_copy(update={"rules": updated})
169 
170 
171def bless_by_id(
172    state: RegistryState, rule_id: RuleId, *, now: datetime.datetime
173) -> RegistryState:
174    """Bless an existing eligible rule by id — owner accepted the offer (3b).
175 
176    Flips that specific learned rule to ``trusted``/``human_explicit`` (the
177    grant the gate requires). Unlike :func:`bless_rule`, it never creates a
178    rule: if the rule is gone or was corrected away from eligibility in the
179    meantime, this is a no-op, so accepting a stale offer can never resurrect a
180    superseded action or duplicate the payee.
181    """
182    target = next((r for r in state.rules if r.id == rule_id), None)
183    if (
184        target is None
185        or target.trust is not TrustState.TRUSTED
186        or target.source is not RuleSource.LEARNED
187    ):
188        return state
189    updated = target.model_copy(
190        update={
191            "source": RuleSource.HUMAN_EXPLICIT,
192            "last_confirmed_at": now,
193        }
194    )
195    new_rules = tuple(
196        updated if rule.id == rule_id else rule for rule in state.rules
197    )
198    entry = RegistryAuditEntry(
199        at=now,
200        payee=target.match.payee_pattern,
201        change=RuleChange(
202            kind=RuleChangeKind.BLESSED, rule_id=rule_id, trust=updated.trust
203        ),
204    )
205    return state.model_copy(
206        update={"rules": new_rules, "audit": _appended_audit(state, entry)}
207    )
⋯ 14 lines hidden (lines 208–221)208 
209 
210def rules_for_payee(state: RegistryState, payee: str) -> tuple[Rule, ...]:
211    """The rules whose payee pattern matches ``payee`` (case-insensitive).
212 
213    The gate's load path: it asks the registry for the candidate rules on a
214    transaction's payee, then decides auto-vs-ask over just those.
215    """
216    lowered = payee.lower()
217    return tuple(
218        rule
219        for rule in state.rules
220        if rule.match.payee_pattern.lower() in lowered
221    )

The registry becomes a thin effect-dispatcher

The registry was a park-on-signals singleton. It now mirrors the W2 driver: each wake, if a rule is newly eligible-and-unoffered it starts that rule's one-time offer and marks it offered; only once the offer queue is drained (and history has grown) does it continue-as-new. Crucially it must never die or busy-loop on a failed offer — the gate reads its rule table — so a terminal failure is alerted best-effort and the rule is marked offered anyway.

The run loop derives pending offers from state each wake (no separate queue → replay-safe); _offer is failure-isolated; bless_existing blesses by id.

src/ynab_agent/workflow/registry_workflow.py · 164 lines

src/ynab_agent/workflow/registry_workflow.py164 lines · Python

⋯ 59 lines hidden (lines 1–59)1"""W5 · the durable rule registry — the agent's persistent learning memory.
2 
3A singleton, long-lived workflow (id :data:`REGISTRY_WORKFLOW_ID`) holding the
4learned rule table as Temporal state (SPEC §14, §0.5 derived-state). It is born
5on the first learning signal — the ``feed_rule_learning`` activity does a
6signal-with-start — and lives forever, continuing-as-new to keep its history
7bounded while carrying the rule table forward.
8 
9The workflow is a thin durable shell over the pure
mod:`ynab_agent.learn.registry` folds (the same pattern ``txn_workflow`` has
11over ``state_machine``): signals fold confirm/correct/bless inputs into the
12state; queries read it back for the autonomy gate and the on-ramp prompt. All
13clock and id reads go through ``workflow.*`` so replay stays deterministic.
14"""
15 
16from __future__ import annotations
17 
18import contextlib
19 
20from temporalio import workflow
21from temporalio.exceptions import ActivityError
22 
23from ynab_agent.workflow.registry_types import RegistryParams, RegistryView
24 
25with workflow.unsafe.imports_passed_through():
  from ynab_agent.domain.effects import FeedRuleLearning
  from ynab_agent.domain.ids import RuleId
  from ynab_agent.domain.rule import Rule
  from ynab_agent.learn.events import ExplicitCommand
  from ynab_agent.learn.registry import (
      RegistryState,
      bless_by_id,
      bless_rule,
      eligible_for_bless,
      mark_offered,
      pending_offers,
      record_learning,
      rules_for_payee,
  )
  from ynab_agent.workflow import alert_activities, offer_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,
  )
49 
50 
51@workflow.defn
52class RuleRegistryWorkflow:
  """The household's one durable rule table (one fold per learning signal)."""
54 
  def __init__(self) -> None:
      """Start empty; the run method adopts any carried-forward state."""
      self._state = RegistryState()
58 
  @workflow.run
  async def run(self, params: RegistryParams) -> None:
      """Hold the rule table; volunteer offers; roll when history grows.
62 
      A thin effect-dispatcher (mirroring the W2 driver): each wake, if a rule
      has newly become eligible and was never offered, start its one-time
      autonomy offer (SPEC §14.7 3b) and mark it offered; otherwise, once the
      offer queue is drained and history has grown, continue-as-new carrying
      the table forward. Draining before the roll mirrors W2 draining its
      inbound before continue-as-new.
      """
      self._state = params.state
      while True:
          await workflow.wait_condition(
              lambda: (
                  bool(pending_offers(self._state))
                  or workflow.info().is_continue_as_new_suggested()
              )
          )
          pending = pending_offers(self._state)
          if pending:
              await self._offer(pending[0])
              continue
          workflow.continue_as_new(RegistryParams(state=self._state))
83 
  async def _offer(self, rule: Rule) -> None:
      """Start a rule's one-time offer, then mark it offered (SPEC §14.7 3b).
86 
      The singleton must never die or busy-loop on a failed offer (the gate
      reads its rule table), so a terminal failure is alerted best-effort and
      the rule is marked offered regardless — we move on rather than retry the
      same prompt forever. The id-reuse rejection in ``start_autonomy_offer``
      keeps it one-time even across replay.
      """
      try:
          await workflow.execute_activity(
              offer_activities.start_autonomy_offer,
              rule,
              start_to_close_timeout=ACTIVITY_TIMEOUT,
              retry_policy=ACTIVITY_RETRY,
          )
      except ActivityError as exc:
          with contextlib.suppress(ActivityError):
              await workflow.execute_activity(
                  alert_activities.alert_failure,
                  build_failure_alert(
                      key=f"offer-start-{rule.id}",
                      context=(
                          f"starting autonomy offer for "
                          f"{rule.match.payee_pattern}"
                      ),
                      exc=exc,
                  ),
                  start_to_close_timeout=ALERT_TIMEOUT,
                  schedule_to_close_timeout=ALERT_BUDGET,
                  retry_policy=ALERT_RETRY,
              )
      self._state = mark_offered(self._state, rule.id, now=workflow.now())
⋯ 19 lines hidden (lines 117–135)117 
  def _next_id(self) -> RuleId:
      """A fresh rule id, minted deterministically for replay safety."""
      return RuleId(str(workflow.uuid4()))
121 
  @workflow.signal
  def record(self, feed: FeedRuleLearning) -> None:
      """Fold a W2 confirm/correct effect into the rule table (SPEC §9)."""
      self._state = record_learning(
          self._state, feed, now=workflow.now(), next_id=self._next_id()
      )
128 
  @workflow.signal
  def bless(self, command: ExplicitCommand) -> None:
      """Grant a rule autonomy — the owner's opt-in or standing command."""
      self._state = bless_rule(
          self._state, command, now=workflow.now(), next_id=self._next_id()
      )
135 
  @workflow.signal
  def bless_existing(self, rule_id: str) -> None:
      """Bless an eligible rule by id — the owner accepted its offer (3b).
139 
      Distinct from :meth:`bless`: it grants autonomy to *exactly* the offered
      rule (no match/action reconstruction), and no-ops if that rule is gone
      or no longer eligible, so a stale acceptance can never resurrect it.
      """
      self._state = bless_by_id(
          self._state, RuleId(rule_id), now=workflow.now()
      )
⋯ 18 lines hidden (lines 147–164)147 
  @workflow.query
  def rules(self) -> tuple[Rule, ...]:
      """The whole rule table (for tests and operators)."""
      return self._state.rules
152 
  @workflow.query
  def view(self) -> RegistryView:
      """The rule table plus the rules eligible for an opt-in bless."""
      return RegistryView(
          rules=self._state.rules,
          eligible=eligible_for_bless(self._state),
      )
160 
  @workflow.query
  def payee_rules(self, payee: str) -> tuple[Rule, ...]:
      """The candidate rules for one payee — the gate's load path."""
      return rules_for_payee(self._state, payee)

The offer workflow: ask, then read the reply

One short-lived workflow per eligible rule. It opens its own offer thread, stamps an OfferThreadId search attribute (so a reply routes back to it), and waits for an answer or a 14-day timeout. The reply is read by a model into accept/decline/unclear; accept blesses, decline sends a brief note, unclear keeps waiting, and silence simply ends the offer (a later "yes" still works via the command path).

All clock reads via workflow.* and side effects behind activities, so the model/mail stacks never enter this sandbox; the verdict branch is a pure enum.

src/ynab_agent/workflow/offer_workflow.py · 125 lines

src/ynab_agent/workflow/offer_workflow.py125 lines · Python

⋯ 50 lines hidden (lines 1–50)1"""The autonomy-offer workflow — the proactive eligibility prompt (§14.7 3b).
2 
3One short-lived workflow per eligible rule (id ``autonomy-offer-{rule_id}``,
4started ``REJECT_DUPLICATE`` so the prompt is one-time). It opens its own email
5thread asking "want me to auto-handle *Payee*?", stamps an ``OfferThreadId``
6search attribute so W3 routes the reply back here (a bless-acceptance, never a
7category reply), and waits for an answer. A yes blesses the rule and confirms; a
8no sends a brief note; an unclear reply keeps waiting; silence past the patience
9window simply ends the offer (the owner can still bless later via a command).
10 
11The seam mirrors the W2 driver: ``workflow.*`` for all clock reads, side effects
12behind activities, pure :class:`~ynab_agent.domain.enums.OfferVerdict` for the
13branch — so the model/mail stacks never enter this sandbox.
14"""
15 
16from __future__ import annotations
17 
18from collections import deque
19from datetime import timedelta
20 
21from temporalio import workflow
22from temporalio.common import SearchAttributeKey
23from temporalio.exceptions import ActivityError
24 
25from ynab_agent.workflow.offer_types import OFFER_PATIENCE, OFFER_THREAD_ID
26 
27_OFFER_THREAD_ID = SearchAttributeKey.for_keyword(OFFER_THREAD_ID)
28 
29with workflow.unsafe.imports_passed_through():
30    from ynab_agent.dispatch.classify import InboundMessage
31    from ynab_agent.domain.enums import OfferVerdict
32    from ynab_agent.workflow import alert_activities, offer_activities
33    from ynab_agent.workflow.alerting import build_failure_alert
34    from ynab_agent.workflow.constants import (
35        ACTIVITY_RETRY,
36        ACTIVITY_TIMEOUT,
37        ALERT_BUDGET,
38        ALERT_RETRY,
39        ALERT_TIMEOUT,
40    )
41    from ynab_agent.workflow.offer_types import OfferParams
42 
43 
44@workflow.defn
45class AutonomyOfferWorkflow:
46    """One eligible rule's one-time "may I auto-handle this?" offer."""
47 
48    def __init__(self) -> None:
49        """Start with no reply buffered; ``run`` opens the thread and waits."""
50        self._responses: deque[InboundMessage] = deque()
51 
52    @workflow.signal
53    def submit_response(self, message: InboundMessage) -> None:
54        """The owner replied to the offer (delivered by W3)."""
55        self._responses.append(message)
56 
57    @workflow.run
58    async def run(self, params: OfferParams) -> None:
59        """Make the offer and act on the reply, paging on terminal failure."""
60        try:
61            await self._run(params)
62        except ActivityError as exc:
63            payee = params.rule.match.payee_pattern
64            await workflow.execute_activity(
65                alert_activities.alert_failure,
66                build_failure_alert(
67                    key=f"offer-{params.rule.id}",
68                    context=f"autonomy offer for {payee}",
69                    exc=exc,
70                ),
71                start_to_close_timeout=ALERT_TIMEOUT,
72                schedule_to_close_timeout=ALERT_BUDGET,
73                retry_policy=ALERT_RETRY,
74            )
75            raise
76 
77    async def _run(self, params: OfferParams) -> None:
78        rule = params.rule
79        thread_id = await workflow.execute_activity(
80            offer_activities.open_offer_thread,
81            rule,
82            start_to_close_timeout=ACTIVITY_TIMEOUT,
83            retry_policy=ACTIVITY_RETRY,
84        )
85        # Index this workflow by its thread so a reply routes here (§5a-style).
86        workflow.upsert_search_attributes(
87            [_OFFER_THREAD_ID.value_set(thread_id)]
88        )
89 
90        deadline = workflow.now() + OFFER_PATIENCE
91        while True:
92            timeout = deadline - workflow.now()
93            if timeout < timedelta(0):
94                timeout = timedelta(0)
95            try:
96                await workflow.wait_condition(
97                    lambda: len(self._responses) > 0, timeout=timeout
98                )
99            except TimeoutError:
100                # No clear answer in the window; leave the rule eligible, end.
101                return
102            message = self._responses.popleft()
103            verdict = await workflow.execute_activity(
104                offer_activities.interpret_offer_reply,
105                args=[message.body, rule.match.payee_pattern],
106                start_to_close_timeout=ACTIVITY_TIMEOUT,
107                retry_policy=ACTIVITY_RETRY,
108            )
109            if verdict is OfferVerdict.ACCEPT:
110                await workflow.execute_activity(
111                    offer_activities.accept_offer,
112                    args=[rule, thread_id],
113                    start_to_close_timeout=ACTIVITY_TIMEOUT,
114                    retry_policy=ACTIVITY_RETRY,
115                )
116                return
117            if verdict is OfferVerdict.DECLINE:
118                await workflow.execute_activity(
119                    offer_activities.decline_offer,
120                    args=[rule, thread_id],
121                    start_to_close_timeout=ACTIVITY_TIMEOUT,
122                    retry_policy=ACTIVITY_RETRY,
123                )
124                return
⋯ 1 line hidden (lines 125–125)125            # UNCLEAR: keep waiting for a clearer reply until the deadline.

Routing a reply back by thread identity

This is what lets a bare "yes" work without confusing the spine. When an inbound thread isn't a transaction's, W3 asks whether it's a live offer thread; if so, classify routes it to that offer — after the same inbound-boundary guards (unsigned / autoresponder / non-allow-listed all still win first).

RouteToOffer joins the discriminated union; the guard order is unchanged.

src/ynab_agent/dispatch/classify.py · 138 lines

src/ynab_agent/dispatch/classify.py138 lines · Python

⋯ 71 lines hidden (lines 1–71)1"""W3 inbound classification: the deterministic routing decision (SPEC §5).
2 
3Every inbound AgentMail message is classified and routed. The parts that are
4*deterministic* live here and run first — the inbound boundary (§0.6): trust the
5signed webhook for provenance, act only on allow-listed senders, and never treat
6an autoresponder or bounce as a confirmation. A verified, allow-listed message
7on a known transaction thread routes straight to that W2; anything else without
8a thread is handed to the agentic classifier (receipt vs. command).
9"""
10 
11from __future__ import annotations
12 
13from email.utils import parseaddr
14from enum import StrEnum
15from typing import Annotated, Literal
16 
17from pydantic import Field
18 
19from ynab_agent.domain.base import Frozen
20from ynab_agent.domain.ids import MessageId, ThreadId, YnabTransactionId
21 
22# Local-part prefixes that mark a bounce/system message — never a real reply.
23_SYSTEM_SENDERS = ("mailer-daemon@", "postmaster@")
24 
25 
26class InboundMessage(Frozen):
27    """A normalized inbound email from the AgentMail webhook.
28 
29    ``signature_verified`` is the Svix webhook provenance check, and
30    ``is_auto_reply`` is derived from the headers (auto-submitted / vacation);
31    both are computed by the webhook handler before classification.
32    """
33 
34    message_id: MessageId
35    from_address: str
36    subject: str
37    body: str
38    thread_id: ThreadId | None = None
39    signature_verified: bool = False
40    is_auto_reply: bool = False
41 
42 
43class InboundKind(StrEnum):
44    """The agentic classifier's verdict for a non-thread message (SPEC §5)."""
45 
46    RECEIPT = "receipt"
47    COMMAND = "command"
48    NOISE = "noise"
49 
50 
51class Quarantine(Frozen):
52    """The message failed the inbound boundary; hold it, do not act."""
53 
54    kind: Literal["quarantine"] = "quarantine"
55    reason: str
56 
57 
58class Ignore(Frozen):
59    """A non-actionable autoresponder/bounce; drop it silently."""
60 
61    kind: Literal["ignore"] = "ignore"
62    reason: str
63 
64 
65class RouteToTransaction(Frozen):
66    """A reply on a known transaction thread; signal that W2 (SPEC §5a)."""
67 
68    kind: Literal["transaction"] = "transaction"
69    txn_id: YnabTransactionId
70 
71 
72class RouteToOffer(Frozen):
73    """A reply on an autonomy-offer thread; signal that offer workflow (3b)."""
74 
75    kind: Literal["offer"] = "offer"
76    offer_id: str
⋯ 53 lines hidden (lines 77–129)77 
78 
79class RouteToInterpret(Frozen):
80    """No thread: hand to the agentic classifier (receipt vs. command)."""
81 
82    kind: Literal["interpret"] = "interpret"
83 
84 
85DispatchDecision = Annotated[
86    Quarantine | Ignore | RouteToTransaction | RouteToOffer | RouteToInterpret,
87    Field(discriminator="kind"),
88]
89 
90 
91def _sender_address(raw: str) -> str:
92    """The bare, lower-cased email from a possibly display-named ``From``.
93 
94    AgentMail surfaces the raw ``From`` header, which a mail client may send as
95    ``"Real Name <addr@host>"`` rather than a bare address. The owner allow-list
96    holds bare addresses, so the display-name form must be reduced to its
97    address before the membership (and system-sender) checks — otherwise a
98    legitimate owner reply is quarantined as "sender not allow-listed".
99    ``parseaddr`` returns the address unchanged when there is no display name;
100    on a parse miss we fall back to the raw value so the allow-list never
101    silently widens.
102    """
103    return (parseaddr(raw)[1] or raw).lower()
104 
105 
106def classify(
107    message: InboundMessage,
108    allowlist: frozenset[str],
109    *,
110    txn_id: YnabTransactionId | None,
111    offer_id: str | None = None,
112) -> DispatchDecision:
113    """Decide how to route an inbound message. Pure (SPEC §5, §0.6).
114 
115    Args:
116        message: The normalized inbound message.
117        allowlist: Lower-cased sender addresses permitted to act.
118        txn_id: The transaction this thread maps to, or ``None`` if the message
119            is not on a known transaction thread.
120        offer_id: The autonomy-offer workflow this thread maps to, or ``None``.
121            A transaction thread takes precedence (a message is on at most one).
122 
123    Returns:
124        The routing decision; ``RouteToInterpret`` defers receipt-vs-command to
125        the agentic classifier.
126    """
127    if not message.signature_verified:
128        return Quarantine(reason="unsigned webhook")
129    sender = _sender_address(message.from_address)
130    if message.is_auto_reply or sender.startswith(_SYSTEM_SENDERS):
131        return Ignore(reason="autoresponder or bounce")
132    if sender not in allowlist:
133        return Quarantine(reason="sender not allow-listed")
134    if txn_id is not None:
135        return RouteToTransaction(txn_id=txn_id)
136    if offer_id is not None:
137        return RouteToOffer(offer_id=offer_id)
138    return RouteToInterpret()

resolve_offer_thread filters to ExecutionStatus=Running (a closed offer is never resurrected); signal_offer is a plain best-effort signal.

src/ynab_agent/workflow/dispatch_activities.py · 192 lines

src/ynab_agent/workflow/dispatch_activities.py192 lines · Python

⋯ 43 lines hidden (lines 1–43)1"""The I/O ports of the W3 inbound dispatcher, as Temporal activities.
2 
3Kept separate from the other workflows' activity modules so each workflow's
4sandbox import graph stays minimal (see ``poll_activities``). Heavy clients
5(Temporal, the model stack) are imported lazily inside the bodies so they never
6enter the workflow sandbox.
7"""
8 
9from __future__ import annotations
10 
11import contextlib
12 
13from temporalio import activity
14 
15from ynab_agent.dispatch.classify import InboundKind, InboundMessage
16from ynab_agent.workflow.temporal_client import client, task_queue
17 
18 
19@activity.defn
20async def resolve_thread(thread_id: str | None) -> str | None:
21    """Resolve an AgentMail thread id to its txn id, or None (SPEC §5).
22 
23    The per-transaction workflow stamps its AgentMail thread id into the
24    ``TxnThreadId`` search attribute, so a reply's thread maps back to that
25    workflow through a Temporal visibility query. The workflow id *is* the YNAB
26    transaction id (started ``REJECT_DUPLICATE`` on it), so the matching
27    execution's id is the answer — there is no stored thread↔txn table (SPEC
28    §0.5, store-free). ``None`` when the thread belongs to no live transaction.
29    """
30    if thread_id is None:
31        return None
32    temporal = await client()
33    # The thread id is an AgentMail token, but quote-escape defensively so the
34    # visibility query stays well-formed.
35    safe = thread_id.replace('"', '\\"')
36    async for execution in temporal.list_workflows(
37        query=f'TxnThreadId = "{safe}"'
38    ):
39        return execution.id
40    return None
41 
42 
43@activity.defn
44async def resolve_offer_thread(thread_id: str | None) -> str | None:
45    """Resolve an AgentMail thread to a *running* offer workflow id (3b).
46 
47    The autonomy-offer workflow stamps its thread into the ``OfferThreadId``
48    search attribute, so a reply on that thread maps back to it the same way
49    ``resolve_thread`` maps a transaction. Filtered to ``Running`` executions so
50    a reply landing after the offer has closed is *not* routed (it falls through
51    to the command path, the documented late-accept fallback) rather than
52    resurrecting a finished offer. ``None`` when no live offer owns the thread.
53    """
54    if thread_id is None:
55        return None
56    from ynab_agent.workflow.offer_types import OFFER_THREAD_ID
57 
58    temporal = await client()
59    safe = thread_id.replace('"', '\\"')
60    query = f'{OFFER_THREAD_ID} = "{safe}" AND ExecutionStatus = "Running"'
61    async for execution in temporal.list_workflows(query=query):
62        return execution.id
63    return None
64 
65 
66@activity.defn
67async def signal_offer(offer_id: str, message: InboundMessage) -> None:
68    """Deliver a reply to its autonomy-offer workflow (SPEC §14.7 3b).
69 
70    A plain signal (not signal-with-start): the offer must be live to receive it
71    — ``resolve_offer_thread`` already filtered to running executions — so a
72    closed offer is never resurrected. A missing handle is a benign no-op.
73    """
74    from temporalio.service import RPCError
75 
76    temporal = await client()
77    handle = temporal.get_workflow_handle(offer_id)
78    with contextlib.suppress(RPCError):
79        await handle.signal("submit_response", message)
80 
⋯ 112 lines hidden (lines 81–192)81 
82@activity.defn
83async def classify_inbound(message: InboundMessage) -> InboundKind:
84    """Agentically classify a non-thread message: receipt, command, or noise.
85 
86    The model stack is imported lazily here so it never enters the workflow
87    sandbox. The model only labels the message; the deterministic dispatcher
88    routes it (SPEC §5).
89    """
90    from ynab_agent.agentic.classify import classify_inbound as run_classifier
91    from ynab_agent.agentic.classify import to_kind
92 
93    return to_kind(await run_classifier(message))
94 
95 
96@activity.defn
97async def signal_transaction(txn_id: str, message: InboundMessage) -> None:
98    """Deliver a reply to the transaction's W2 (SPEC §5a).
99 
100    Signal-with-start on ``submit_inbound``: a running W2 (the common case — it
101    asked the question) receives the reply and wakes; if the transaction's run
102    has closed, a fresh W2 starts with the reply buffered and re-triages. The
103    workflow id is the YNAB transaction id, so no thread↔txn table is needed.
104    """
105    from ynab_agent.domain.ids import YnabTransactionId
106    from ynab_agent.domain.signals import ReplySignal
107    from ynab_agent.workflow.types import TransactionParams
108 
109    if message.thread_id is None:
110        # RouteToTransaction only fires for a thread-matched message, so a
111        # missing thread id here is a routing bug, not an expected input.
112        msg = f"signal for {txn_id} has no thread id"
113        raise RuntimeError(msg)
114    reply = ReplySignal(
115        thread_id=message.thread_id,
116        message_id=message.message_id,
117        from_address=message.from_address,
118        text=message.body,
119    )
120    temporal = await client()
121    await temporal.start_workflow(
122        "TransactionWorkflow",
123        TransactionParams(ynab_id=YnabTransactionId(txn_id)),
124        id=txn_id,
125        task_queue=task_queue(),
126        start_signal="submit_inbound",
127        start_signal_args=[reply],
128    )
129 
130 
131@activity.defn
132async def route_receipt(_message: InboundMessage) -> None:
133    """Hand a forwarded receipt to the W4 join (SPEC §5b).
134 
135    No-op in v1: the receipt join (W4) is out of the core-triage scope, and the
136    agent does not advertise receipt forwarding, so receipts do not arrive yet.
137    Wired when W4 lands (the message arg is kept for that contract).
138    """
139    return None
140 
141 
142@activity.defn
143async def handle_command(message: InboundMessage) -> None:
144    """Parse a standing command and bless the rule it grants (SPEC §5c, §14).
145 
146    The owner's direct opt-in: "always categorize X as Y" becomes an
147    ``ExplicitCommand`` signalled to the durable registry's ``bless``, which
148    trusts the rule for auto-apply (``source=human_explicit``). Anything not a
149    clear bless — a question or comment — is a deliberate no-op (the parser
150    declines it), so a stray message never grants autonomy.
151    """
152    import asyncio
153 
154    from temporalio.common import WorkflowIDConflictPolicy
155 
156    from ynab_agent.agentic.command import (
157        CommandRequest,
158        parse_command,
159        to_explicit_command,
160    )
161    from ynab_agent.agentic.enrich import CandidateCategory
162    from ynab_agent.workflow.registry_types import (
163        REGISTRY_WORKFLOW_ID,
164        RegistryParams,
165    )
166    from ynab_agent.workflow.temporal_client import client, task_queue
167    from ynab_agent.ynab.client import YnabClient
168 
169    ynab = YnabClient.from_env()
170    spends = await asyncio.to_thread(ynab.category_spends)
171    candidates = tuple(
172        CandidateCategory(id=str(spend.category), name=spend.name)
173        for spend in spends
174    )
175    if not candidates:
176        return
177    reading = await parse_command(
178        CommandRequest(command_text=message.body, candidates=candidates)
179    )
180    command = to_explicit_command(reading)
181    if command is None:
182        return
183    temporal = await client()
184    await temporal.start_workflow(
185        "RuleRegistryWorkflow",
186        RegistryParams(),
187        id=REGISTRY_WORKFLOW_ID,
188        task_queue=task_queue(),
189        id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
190        start_signal="bless",
191        start_signal_args=[command],
192    )

Guard: an independent, clean-context safety review

Before an auto-apply lands, the model judges it — but blind to the rule's choice, so it isn't anchored. If the independent categorization doesn't even consider the blessed category plausible, the auto-apply is held back to ASK. This is the doubt-detection for "this transaction looks different from the usual."

review_auto_apply is the one-way ratchet (wiring the dormant ReviewVerdict enum); decide_enrichment runs it only on the AUTO branch.

src/ynab_agent/agentic/enrich.py · 277 lines

src/ynab_agent/agentic/enrich.py277 lines · Python

⋯ 160 lines hidden (lines 1–160)1"""The enrichment agent: propose a category for a transaction (SPEC §4.1).
2 
3The agentic half of the ``enrich`` step. Given a transaction's facts and the
4candidate budget categories, a Pydantic AI agent picks the best category, rates
5its confidence, and explains why. The deterministic gate (``policy.gate``) then
6decides auto-apply vs. ask — confidence here is *framing only*, never a gate
7(principle 6). :func:`to_proposal` maps the agent's structured output onto the
8domain :class:`~ynab_agent.domain.proposal.Proposal`.
9 
10The model is injected per run so tests drive a ``TestModel``/``FunctionModel``
11offline; production uses :func:`~ynab_agent.agentic.model.build_model` (Ollama).
12"""
13 
14from __future__ import annotations
15 
16from typing import TYPE_CHECKING
17 
18from pydantic import Field
19from pydantic_ai import Agent
20 
21from ynab_agent.agentic.model import run_structured
22from ynab_agent.domain.allocations import ProposedCategory
23from ynab_agent.domain.base import Frozen
24from ynab_agent.domain.enums import Confidence, ReviewVerdict, SourceKind
25from ynab_agent.domain.events import AskHuman, AutoApply
26from ynab_agent.domain.ids import CategoryId
27from ynab_agent.domain.proposal import Proposal, ProposalSource
28from ynab_agent.policy.gate import (
29    GateVerdict,
30    build_auto_decision,
31    evaluate_gate,
32)
33 
34if TYPE_CHECKING:
35    import datetime
36    from collections.abc import Iterable
37 
38    from pydantic_ai.models import Model
39 
40    from ynab_agent.domain.events import EnrichmentOutcome
41    from ynab_agent.domain.rule import Rule
42    from ynab_agent.domain.transaction import YnabSnapshot
43    from ynab_agent.policy.floor import AutoActionCounters
44 
45 
46class CandidateCategory(Frozen):
47    """One budget category the agent may choose from."""
48 
49    id: str
50    name: str
51 
52 
53class EnrichmentRequest(Frozen):
54    """The facts the agent reasons over to propose a category."""
55 
56    payee: str
57    amount_display: str
58    candidates: tuple[CandidateCategory, ...] = Field(min_length=1)
59    memo: str | None = None
60    rule_hint: str | None = None
61 
62 
63class EnrichmentSuggestion(Frozen):
64    """The agent's structured proposal (mapped to a domain Proposal)."""
65 
66    category_id: str
67    confidence: Confidence
68    rationale: str
69    alternatives: tuple[str, ...] = ()
70 
71 
72_SYSTEM_PROMPT = """\
73You categorize a single bank transaction for a personal budget. You are given
74the payee, amount, an optional memo, an optional hint from a learned rule, and a
75list of candidate budget categories (each with an id and a name).
76 
77Choose the SINGLE best category for the transaction. Your `category_id` MUST be
78one of the provided candidate ids — never invent one. Also list up to 2
79`alternatives`: the next-most-likely candidate ids (also from the list, never
80the chosen one) the owner might prefer — or leave empty if none fit. Rate your
81confidence (high / medium / low) and give a one-sentence rationale. Confidence
82is framing only; a human or a trusted rule decides whether to auto-apply."""
83 
84_AGENT: Agent[None, EnrichmentSuggestion] = Agent(
85    output_type=EnrichmentSuggestion,
86    system_prompt=_SYSTEM_PROMPT,
87)
88 
89 
90def _format_request(request: EnrichmentRequest) -> str:
91    """Render the request as the agent's user prompt."""
92    lines = [
93        f"Payee: {request.payee}",
94        f"Amount: {request.amount_display}",
95    ]
96    if request.memo:
97        lines.append(f"Memo: {request.memo}")
98    if request.rule_hint:
99        lines.append(f"Rule hint: {request.rule_hint}")
100    lines.append("Candidate categories:")
101    lines.extend(f"  - {c.name} (id: {c.id})" for c in request.candidates)
102    return "\n".join(lines)
103 
104 
105async def propose(
106    request: EnrichmentRequest, *, model: Model | None = None
107) -> EnrichmentSuggestion:
108    """Run the enrichment agent for one transaction (SPEC §4.1).
109 
110    Args:
111        request: The transaction facts and candidate categories.
112        model: A model to use; defaults to the configured Ollama/Gemma.
113 
114    Returns:
115        The agent's structured category suggestion.
116    """
117    return await run_structured(
118        _AGENT,
119        _format_request(request),
120        output_type=EnrichmentSuggestion,
121        model=model,
122    )
123 
124 
125def to_proposal(suggestion: EnrichmentSuggestion) -> Proposal:
126    """Map the agent's suggestion onto a domain Proposal (SPEC §4.1).
127 
128    The model's runner-up ids become the proposal's alternatives (the chosen
129    one filtered out defensively), surfaced in the email so the owner can pick
130    one at a glance.
131    """
132    alternatives = tuple(
133        CategoryId(alt)
134        for alt in suggestion.alternatives
135        if alt and alt != suggestion.category_id
136    )
137    return Proposal(
138        allocation=ProposedCategory(
139            category=CategoryId(suggestion.category_id)
140        ),
141        confidence=suggestion.confidence,
142        rationale=suggestion.rationale,
143        sources=(ProposalSource(kind=SourceKind.MODEL),),
144        alternatives=alternatives,
145    )
146 
147 
148def _blessed_category_id(rule: Rule) -> str | None:
149    """The single category a rule auto-applies, or ``None`` for a split.
150 
151    The safety review compares a single independent category against the rule's;
152    a split action has no single category to judge, so the review is skipped for
153    it (learned-eligible rules are always single-category, so this is rare).
154    """
155    allocation = rule.action.allocation
156    if isinstance(allocation, ProposedCategory):
157        return str(allocation.category)
158    return None
159 
160 
161def review_auto_apply(
162    blessed_category_id: str, suggestion: EnrichmentSuggestion
163) -> ReviewVerdict:
164    """The agent-powered safety review's one-way ratchet (SPEC §0.6 Layer 2).
165 
166    An independent, *clean-context* model categorization (run with no knowledge
167    of the rule's choice, to stay unbiased) judges the impending auto-apply: it
168    ``PROCEED``s only if it considers the blessed category plausible — the
169    category it chose or listed as an alternative — and otherwise
170    ``ESCALATE_TO_HUMAN``. It can only hold an auto-apply back, never grant one,
171    so the deterministic gate still authorizes (principle 6).
172    """
173    plausible = {suggestion.category_id, *suggestion.alternatives}
174    if blessed_category_id in plausible:
175        return ReviewVerdict.PROCEED
176    return ReviewVerdict.ESCALATE_TO_HUMAN
⋯ 65 lines hidden (lines 177–241)177 
178 
179def _escalation_proposal(
180    suggestion: EnrichmentSuggestion, blessed_category_id: str
181) -> Proposal:
182    """The proposal emailed when the review holds an auto-apply back.
183 
184    Leads with the model's independent pick and offers the usual auto category
185    as an alternative, so the owner sees both views and the disagreement that
186    triggered the question.
187    """
188    base = to_proposal(suggestion)
189    blessed = CategoryId(blessed_category_id)
190    # to_proposal always builds a single-category allocation; narrow for mypy.
191    allocation = base.allocation
192    top = (
193        allocation.category
194        if isinstance(allocation, ProposedCategory)
195        else None
196    )
197    alternatives = base.alternatives
198    if blessed != top and blessed not in alternatives:
199        alternatives = (blessed, *alternatives)
200    return base.model_copy(
201        update={
202            "alternatives": alternatives,
203            "rationale": (
204                "I'd usually auto-file this payee, but this charge looked "
205                "different from the usual — confirming before I apply it."
206            ),
207        }
208    )
209 
210 
211async def decide_enrichment(
212    snapshot: YnabSnapshot,
213    candidates: tuple[CandidateCategory, ...],
214    rules: Iterable[Rule],
215    counters: AutoActionCounters,
216    *,
217    now: datetime.datetime,
218    model: Model | None = None,
219) -> EnrichmentOutcome:
220    """Compose the enrich step: gate, then a safety review (SPEC §4.1, §0.6).
221 
222    The deterministic gate decides autonomy from the rules alone — a single
223    blessed rule may auto-apply *its* action (the model never authorizes a
224    write, principle 6). Before an auto-apply lands, an independent
225    *clean-context* model review judges it (:func:`review_auto_apply`); a
226    disagreement holds it back to ASK (a one-way ratchet — it can only veto). A
227    gated ASK runs the same model to produce the best-guess proposal the email
228    shows.
229 
230    Args:
231        snapshot: The transaction being enriched.
232        candidates: The budget categories the agent may choose from.
233        rules: The rules in scope for the gate.
234        counters: The auto-action counters the hard floor reads.
235        now: The decision timestamp (the activity supplies it).
236        model: A model override for tests; defaults to Ollama/Gemma.
237 
238    Returns:
239        ``AutoApply`` when a blessed rule gates it *and* the review proceeds,
240        else ``AskHuman``.
241    """
242    rules = tuple(rules)
243    gate = evaluate_gate(snapshot, rules, counters)
244    if gate.verdict is GateVerdict.AUTO and gate.rule_id is not None:
245        rule = next((r for r in rules if r.id == gate.rule_id), None)
246        if rule is not None:
247            decision = build_auto_decision(rule, snapshot, now)
248            blessed_category_id = _blessed_category_id(rule)
249            if blessed_category_id is None:
250                return AutoApply(decision=decision)
251            # Clean context: the independent judge never sees the rule's choice.
252            suggestion = await propose(
253                EnrichmentRequest(
254                    payee=snapshot.payee,
255                    amount_display=str(snapshot.amount),
256                    candidates=candidates,
257                ),
258                model=model,
259            )
260            if (
261                review_auto_apply(blessed_category_id, suggestion)
262                is ReviewVerdict.PROCEED
263            ):
264                return AutoApply(decision=decision)
265            return AskHuman(
266                proposal=_escalation_proposal(suggestion, blessed_category_id)
⋯ 11 lines hidden (lines 267–277)267            )
268 
269    suggestion = await propose(
270        EnrichmentRequest(
271            payee=snapshot.payee,
272            amount_display=str(snapshot.amount),
273            candidates=candidates,
274        ),
275        model=model,
276    )
277    return AskHuman(proposal=to_proposal(suggestion))

Revoke: catching a silent manual edit

The owner might just recategorize an auto-applied transaction in YNAB directly — a silent correction the reply path never sees. At the OPEN→archive boundary the driver re-reads YNAB and compares the live category to what the agent applied; a mismatch becomes an OverrideDetected carrying the human's choice.

A category-only comparison (a memo-only edit is not an override); the human Decision is built from the read-back.

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

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

⋯ 461 lines hidden (lines 1–461)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,
      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
104 
105# History-length ceiling before a resting workflow continues-as-new. High enough
106# that ordinary flows never trip it; long-lived (30-45 day) transactions do.
107_CONTINUE_AS_NEW_AFTER = 4_000
108 
109# Resting states the workflow may sit in for days while waiting on signals or
110# timers; continue-as-new fires only from one of these. DISCOVERED counts: a
111# transaction born from a signal can wait here for a slow YNAB import (SPEC §3).
112_RESTING = (Discovered, AwaitingHuman, Open, Lapsed, HoldAmazon)
113 
114 
115def _is_amazon(payee: str) -> bool:
  return "amazon" in payee.lower()
117 
118 
119def _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
122 
123 
124@workflow.defn
125class TransactionWorkflow:
  """The durable per-transaction lifecycle."""
127 
  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
139 
  # ── 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)
145 
  @workflow.signal
  def notify_snapshot(self, snapshot: YnabSnapshot) -> None:
      """W1 materialized (or backfilled the memo of) the YNAB snapshot."""
      self._snapshot_ready = snapshot
150 
  @workflow.query
  def state(self) -> str:
      """The current lifecycle state (for observability)."""
      return self._txn.state.value
155 
  # ── 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()
169 
      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
200 
  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})"
207 
  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,
      )
219 
  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
239 
  # ── 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)
253 
  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, CloseThread):
          if self._thread_id is not None:
              await workflow.execute_activity(
                  activities.close_thread,
                  self._thread_id,
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=ACTIVITY_RETRY,
              )
      elif isinstance(effect, SetTimer):
          self._deadlines[effect.timer] = effect.deadline
      elif isinstance(effect, CancelTimer):
          self._deadlines.pop(effect.timer, None)
      elif isinstance(effect, ReplayBuffered):
          self._inbound.extendleft(reversed(effect.signals))
      return None
317 
  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})
324 
  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)
331 
  def _proposal(self) -> Proposal | None:
      """The current best-guess proposal, for states that carry one.
334 
      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
342 
  # ── 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())
          )
376 
  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))
385 
  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())
          )
401 
  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)
417 
  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)
          )
434 
  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())
          )
461 
  async def _archive_or_override(self, st: Open) -> LifecycleEvent:
      """At archive time, detect a manual YNAB edit (SPEC §14.2).
464 
      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()
⋯ 62 lines hidden (lines 487–548)487 
  async def _on_lapsed(self) -> None:
      await self._wait_then_revise_or_archive()
490 
  def _has_inbound(self) -> bool:
      return len(self._inbound) > 0
493 
  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())
          )
518 
  async def _wait_forever(self, predicate: Callable[[], bool]) -> bool:
      await workflow.wait_condition(predicate)
      return True
522 
  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))
531 
  async def _wait_until(
      self,
      deadline: datetime.datetime,
      predicate: Callable[[], bool],
  ) -> bool:
      """Wait for ``predicate`` until an absolute deadline.
538 
      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

The pure transition: demote the driving rule (CORRECT, carrying the prior agent decision), notify, and close on the owner's value — archiving only if reconciled, else adopt-and-wait (ARCHIVED requires a reconciled snapshot).

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

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

⋯ 444 lines hidden (lines 1–444)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,
  ReplayBuffered,
  RuleLearningKind,
  SendThreadMessage,
  SetTimer,
  TimerKind,
44)
45from ynab_agent.domain.enums import AwaitingFlag, RevisingOrigin
46from 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,
69)
70from ynab_agent.domain.signals import ReplySignal
71from ynab_agent.domain.transaction import (
  Applied,
  Archived,
  AutoApplied,
  AwaitingHuman,
  Discovered,
  Enriching,
  HoldAmazon,
  Lapsed,
  Open,
  Revising,
  Transaction,
  TxnCore,
84)
85 
86if TYPE_CHECKING:
  import datetime
88 
  from ynab_agent.domain.config import LifecyclePolicy
  from ynab_agent.domain.proposal import Decision
91 
92 
93class TransitionKind(StrEnum):
  """The disposition of an ``advance`` call."""
95 
  ADVANCED = "advanced"
  IGNORED = "ignored"
  REJECTED = "rejected"
99 
100 
101class Transition(Frozen):
  """The result of advancing a transaction.
103 
  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.
  """
111 
  kind: TransitionKind
  next: Transaction
  effects: tuple[Effect, ...] = ()
  reason: str | None = None
116 
117 
118def _advanced(nxt: Transaction, *effects: Effect) -> Transition:
  return Transition(kind=TransitionKind.ADVANCED, next=nxt, effects=effects)
120 
121 
122def _ignored(txn: Transaction, reason: str) -> Transition:
  return Transition(kind=TransitionKind.IGNORED, next=txn, reason=reason)
124 
125 
126def _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)
129 
130 
131def advance(
  txn: Transaction,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
137) -> 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)
161 
162 
163def _from_discovered(
  txn: Discovered,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
169) -> 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)
198 
199 
200def _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)
220 
221 
222def _from_enriching(
  txn: Enriching,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
228) -> Transition:
  match event:
      case Enriched(outcome=outcome):
          match outcome:
              case AutoApply(decision=decision):
                  return _advanced(
                      AutoApplied(core=txn.core, decision=decision),
                      CommitToYnab(decision=decision),
                  )
              case AskHuman(proposal=proposal):
                  deadline = now + policy.patience_window
                  # First contact: open_thread sends the proposal as the
                  # thread's opening email (AgentMail starts a thread on its
                  # first send), so no separate send is emitted. A re-proposal
                  # (the thread is already open) sends it as a reply.
                  send_effect: tuple[Effect, ...] = (
                      (OpenThread(),)
                      if txn.core.thread_id is None
                      else (
                          SendThreadMessage(purpose=MessagePurpose.PROPOSAL),
                      )
                  )
                  return _advanced(
                      AwaitingHuman(
                          core=txn.core,
                          proposal=proposal,
                          patience_deadline=deadline,
                      ),
                      *send_effect,
                      SetTimer(timer=TimerKind.PATIENCE, deadline=deadline),
                  )
          assert_never(outcome)
      case InboundReceived(signal=sig):
          # Re-feed so enrichment incorporates the new signal; stay ENRICHING.
          return Transition(
              kind=TransitionKind.ADVANCED,
              next=txn,
              effects=(ReplayBuffered(signals=(sig,)),),
          )
      case _:
          return _rejected(txn, event)
269 
270 
271def _from_auto_applied(
  txn: AutoApplied,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
277) -> 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)
291 
292 
293def _from_applied(
  txn: Applied,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
299) -> 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)
313 
314 
315def _resolve_write_verify(
  *,
  core: TxnCore,
  decision: Decision,
  outcome: VerifyOutcome,
  applied_message: MessagePurpose,
  learning: RuleLearningKind | None,
  now: datetime.datetime,
  policy: LifecyclePolicy,
324) -> 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)
360 
361 
362def _enter_flagged_awaiting(
  core: TxnCore,
  flag: AwaitingFlag,
  message: MessagePurpose,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
369) -> 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),
  )
377 
378 
379def _from_awaiting_human(
  txn: AwaitingHuman,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
385) -> 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)
424 
425 
426def _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(),
          )
⋯ 156 lines hidden (lines 471–626)471        case _:
          return _rejected(txn, event)
473 
474 
475def _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)
508 
509 
510def _from_revising(
  txn: Revising,
  event: LifecycleEvent,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
516) -> 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)
560 
561 
562def _reapply(
  txn: Revising,
  decision: Decision,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
568) -> Transition:
  """Re-applied revision: open, summarize, and feed learning (SPEC §3, §9).
570 
  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),
  )
597 
598 
599def _resolve_no_change(
  txn: Revising,
  *,
  now: datetime.datetime,
  policy: LifecyclePolicy,
604) -> Transition:
  """No-change exit depends on history (SPEC §3 rule 5).
606 
  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),
  )
621 
622 
623def _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)

Verification

The full gate is green — ruff, mypy over 182 files, and 449 tests — including the "Many Hands" determinism, sandbox-imports, and variant-exhaustiveness sweeps, which scan the new workflow files. New coverage spans every layer: the pure folds and the K=5 ladder; the registry workflow (eligibility → exactly one offer → no re-offer; bless_existing); the offer workflow (accept / decline / unclear / timeout); classify's RouteToOffer and guard precedence; resolve_offer_thread's Running-only query; the safety review's proceed/escalate; and the manual-edit demotion end to end.

One deploy step rides alongside the code: the new OfferThreadId search attribute must be registered on the Temporal namespace (the same idempotent job that registers TxnThreadId).