Acknowledging a forwarded receipt instead of swallowing it

What changed, and why

The W3 dispatcher classifies a forwarded receipt and routes it to route_receipt — which was a silent no-op. So a forwarded receipt was accepted by the webhook and dropped on the floor, with the sender left assuming it had been handled.

The receipt⇄transaction join (W4) is a genuine deferred increment: the workflow and its pure planner exist, but the I/O activities are still stubs and there's no email→receipt parser yet. So the agent can't process the receipt — but it shouldn't vanish either. The fix is the honest middle ground: acknowledge it, and point the owner at the path that works today.

Why W4 isn't started: its activities still raise NotImplementedError — starting the workflow would crash and page.

src/ynab_agent/workflow/receipt_activities.py · 47 lines

src/ynab_agent/workflow/receipt_activities.py47 lines · Python

⋯ 19 lines hidden (lines 1–19)1"""The I/O ports of the W4 receipt join, as Temporal activities.
2 
3Kept in its own module so the join workflow's sandbox import graph stays minimal
4(see ``poll_activities`` / ``dispatch_activities``). The match itself is the
5agentic step (``match_receipt``); the rest are the spine's side effects. All
6stubbed; the real YNAB/AgentMail wiring lands later.
7"""
8 
9from __future__ import annotations
10 
11from temporalio import activity
12 
13from ynab_agent.domain.enums import ReceiptStatus
14from ynab_agent.domain.receipt import Receipt
15from ynab_agent.join.match import MatchOutcome
16 
17_STUB = "workflow activity stub — register a real or mock implementation"
18 
19 
20@activity.defn
21async def match_receipt(receipt: Receipt) -> MatchOutcome:
22    """Match a receipt against open transactions (the agentic step, SPEC §6)."""
23    raise NotImplementedError(_STUB)
24 
⋯ 23 lines hidden (lines 25–47)25 
26@activity.defn
27async def signal_match(txn_id: str, receipt_id: str) -> None:
28    """Signal-with-start the matched transaction's W2 with this receipt."""
29    raise NotImplementedError(_STUB)
30 
31 
32@activity.defn
33async def ask_disambiguation(receipt_id: str, candidates: list[str]) -> None:
34    """Ask the sender which candidate transaction the receipt belongs to."""
35    raise NotImplementedError(_STUB)
36 
37 
38@activity.defn
39async def ask_no_match(receipt_id: str) -> None:
40    """Tell the sender no matching transaction was found (TTL expiry)."""
41    raise NotImplementedError(_STUB)
42 
43 
44@activity.defn
45async def save_receipt_status(receipt_id: str, status: ReceiptStatus) -> None:
46    """Persist the receipt's new join status so re-checks dedup (SPEC §6)."""
47    raise NotImplementedError(_STUB)

Acknowledge, don't swallow

route_receipt now replies on the receipt's own thread with an honest note, addressed to the owners, idempotent on the message id. It deliberately does not start ReceiptJoinWorkflow (that would crash on the stubs); it just makes sure the forward isn't silently lost.

A reply instead of a no-op; no-op only when there's no thread to reply on.

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

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

⋯ 169 lines hidden (lines 1–169)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:
  """Resolve an AgentMail thread id to its txn id, or None (SPEC §5).
22 
  The per-transaction workflow stamps its AgentMail thread id into the
  ``TxnThreadId`` search attribute, so a reply's thread maps back to that
  workflow through a Temporal visibility query. The workflow id *is* the YNAB
  transaction id (started ``REJECT_DUPLICATE`` on it), so the matching
  execution's id is the answer — there is no stored thread↔txn table (SPEC
  §0.5, store-free). ``None`` when the thread belongs to no live transaction.
  """
  if thread_id is None:
      return None
  temporal = await client()
  # The thread id is an AgentMail token, but quote-escape defensively so the
  # visibility query stays well-formed.
  safe = thread_id.replace('"', '\\"')
  async for execution in temporal.list_workflows(
      query=f'TxnThreadId = "{safe}"'
  ):
      return execution.id
  return None
41 
42 
43@activity.defn
44async def resolve_offer_thread(thread_id: str | None) -> str | None:
  """Resolve an AgentMail thread to a *running* offer workflow id (3b).
46 
  The autonomy-offer workflow stamps its thread into the ``OfferThreadId``
  search attribute, so a reply on that thread maps back to it the same way
  ``resolve_thread`` maps a transaction. Filtered to ``Running`` executions so
  a reply landing after the offer has closed is *not* routed (it falls through
  to the command path, the documented late-accept fallback) rather than
  resurrecting a finished offer. ``None`` when no live offer owns the thread.
  """
  if thread_id is None:
      return None
  from ynab_agent.workflow.offer_types import OFFER_THREAD_ID
57 
  temporal = await client()
  safe = thread_id.replace('"', '\\"')
  query = f'{OFFER_THREAD_ID} = "{safe}" AND ExecutionStatus = "Running"'
  async for execution in temporal.list_workflows(query=query):
      return execution.id
  return None
64 
65 
66@activity.defn
67async def signal_offer(offer_id: str, message: InboundMessage) -> None:
  """Deliver a reply to its autonomy-offer workflow (SPEC §14.7 3b).
69 
  A plain signal (not signal-with-start): the offer must be live to receive it
  — ``resolve_offer_thread`` already filtered to running executions — so a
  closed offer is never resurrected. A missing handle is a benign no-op.
  """
  from temporalio.service import RPCError
75 
  temporal = await client()
  handle = temporal.get_workflow_handle(offer_id)
  with contextlib.suppress(RPCError):
      await handle.signal("submit_response", message)
80 
81 
82@activity.defn
83async def resolve_balance_thread(thread_id: str | None) -> str | None:
  """Resolve an AgentMail thread to a *running* balance workflow id (§8).
85 
  The balance workflow stamps the overspend thread into ``BalanceThreadId``,
  so a reply there maps back to it the same way ``resolve_offer_thread`` maps
  an offer. Filtered to ``Running`` so a reply after the offer has closed is
  not routed (it falls through to the command path). ``None`` when no live
  balance offer owns the thread.
  """
  if thread_id is None:
      return None
  from ynab_agent.workflow.balance_types import BALANCE_THREAD_ID
95 
  temporal = await client()
  safe = thread_id.replace('"', '\\"')
  query = f'{BALANCE_THREAD_ID} = "{safe}" AND ExecutionStatus = "Running"'
  async for execution in temporal.list_workflows(query=query):
      return execution.id
  return None
102 
103 
104@activity.defn
105async def signal_balance(balance_id: str, message: InboundMessage) -> None:
  """Deliver a reply to its balance workflow (SPEC §8).
107 
  A plain signal: the workflow must be live to receive it —
  ``resolve_balance_thread`` already filtered to running executions. A missing
  handle is a benign no-op.
  """
  from temporalio.service import RPCError
113 
  temporal = await client()
  handle = temporal.get_workflow_handle(balance_id)
  with contextlib.suppress(RPCError):
      await handle.signal("submit_response", message)
118 
119 
120@activity.defn
121async def classify_inbound(message: InboundMessage) -> InboundKind:
  """Agentically classify a non-thread message: receipt, command, or noise.
123 
  The model stack is imported lazily here so it never enters the workflow
  sandbox. The model only labels the message; the deterministic dispatcher
  routes it (SPEC §5).
  """
  from ynab_agent.agentic.classify import classify_inbound as run_classifier
  from ynab_agent.agentic.classify import to_kind
130 
  return to_kind(await run_classifier(message))
132 
133 
134@activity.defn
135async def signal_transaction(txn_id: str, message: InboundMessage) -> None:
  """Deliver a reply to the transaction's W2 (SPEC §5a).
137 
  Signal-with-start on ``submit_inbound``: a running W2 (the common case — it
  asked the question) receives the reply and wakes; if the transaction's run
  has closed, a fresh W2 starts with the reply buffered and re-triages. The
  workflow id is the YNAB transaction id, so no thread↔txn table is needed.
  """
  from ynab_agent.domain.ids import YnabTransactionId
  from ynab_agent.domain.signals import ReplySignal
  from ynab_agent.workflow.types import TransactionParams
146 
  if message.thread_id is None:
      # RouteToTransaction only fires for a thread-matched message, so a
      # missing thread id here is a routing bug, not an expected input.
      msg = f"signal for {txn_id} has no thread id"
      raise RuntimeError(msg)
  reply = ReplySignal(
      thread_id=message.thread_id,
      message_id=message.message_id,
      from_address=message.from_address,
      text=message.body,
  )
  temporal = await client()
  await temporal.start_workflow(
      "TransactionWorkflow",
      TransactionParams(ynab_id=YnabTransactionId(txn_id)),
      id=txn_id,
      task_queue=task_queue(),
      start_signal="submit_inbound",
      start_signal_args=[reply],
  )
167 
168 
169@activity.defn
170async def route_receipt(message: InboundMessage) -> None:
  """Acknowledge a forwarded receipt — the W4 join is not built yet (§5b, §6).
172 
  The dispatcher classifies forwarded receipts, but the receipt⇄transaction
  join (W4) is a deferred increment (its match/park/ask activities are still
  stubs). Rather than swallow the forward silently, reply honestly and point
  the owner at the path that works — replying on the transaction's own thread.
  Idempotent on the message id (a retry never double-replies); a message with
  no thread to reply on is a no-op.
  """
  import asyncio
181 
  from ynab_agent.agentic.compose import render_receipt_unsupported
  from ynab_agent.mail.client import MailClient
  from ynab_agent.settings import Settings
185 
  if message.thread_id is None:
      return
  settings = Settings()
  mail = MailClient.from_env()
  await asyncio.to_thread(
      mail.send_on_thread,
      inbox_id=settings.inbox,
      thread_id=str(message.thread_id),
      body=render_receipt_unsupported(),
      seq_label=f"yarcpt-{message.message_id}",
      to=list(settings.owners),
  )
⋯ 53 lines hidden (lines 198–250)198 
199 
200@activity.defn
201async def handle_command(message: InboundMessage) -> None:
  """Parse a standing command and bless the rule it grants (SPEC §5c, §14).
203 
  The owner's direct opt-in: "always categorize X as Y" becomes an
  ``ExplicitCommand`` signalled to the durable registry's ``bless``, which
  trusts the rule for auto-apply (``source=human_explicit``). Anything not a
  clear bless — a question or comment — is a deliberate no-op (the parser
  declines it), so a stray message never grants autonomy.
  """
  import asyncio
211 
  from temporalio.common import WorkflowIDConflictPolicy
213 
  from ynab_agent.agentic.command import (
      CommandRequest,
      parse_command,
      to_explicit_command,
  )
  from ynab_agent.agentic.enrich import CandidateCategory
  from ynab_agent.workflow.registry_types import (
      REGISTRY_WORKFLOW_ID,
      RegistryParams,
  )
  from ynab_agent.workflow.temporal_client import client, task_queue
  from ynab_agent.ynab.client import YnabClient
226 
  ynab = YnabClient.from_env()
  spends = await asyncio.to_thread(ynab.category_spends)
  candidates = tuple(
      CandidateCategory(id=str(spend.category), name=spend.name)
      for spend in spends
  )
  if not candidates:
      return
  reading = await parse_command(
      CommandRequest(command_text=message.body, candidates=candidates)
  )
  command = to_explicit_command(reading)
  if command is None:
      return
  temporal = await client()
  await temporal.start_workflow(
      "RuleRegistryWorkflow",
      RegistryParams(),
      id=REGISTRY_WORKFLOW_ID,
      task_queue=task_queue(),
      id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
      start_signal="bless",
      start_signal_args=[command],
  )

The honest note: can't match receipts yet → reply on the transaction's thread.

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

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

⋯ 106 lines hidden (lines 1–106)1"""Render a transaction's email body (SPEC §5).
2 
3A deterministic template — *not* a model call. The model already did the
4thinking upstream (the proposal's category + one-line rationale + alternatives);
5this just lays it out cleanly so a glance is enough to act. Keeping it templated
6(rather than free-form prose) is a deliberate low-cognitive-load choice, and it
7drops a per-email model round-trip.
8 
9The layout, for a proposal:
10 
11    Hulu — $13.07 — May 29
12    <memo, when present — e.g. an Amazon item list>
13 
14    Suggested: Entertainment   (or: Streaming, Fun Money)
15    recurring streaming subscription
16 
17    Just reply in your own words — confirm it, suggest a different category, …
18"""
19 
20from __future__ import annotations
21 
22from typing import TYPE_CHECKING
23 
24from ynab_agent.domain.base import Frozen
25from ynab_agent.domain.effects import MessagePurpose
26 
27if TYPE_CHECKING:
28    from ynab_agent.budget.balance import BalanceOption
29    from ynab_agent.domain.money import Money
30 
31_REPLY_HINT = (
32    "Just reply in your own words — confirm it, suggest a different "
33    "category, or ask a question."
34)
35 
36 
37class ComposeRequest(Frozen):
38    """The facts the template lays out for one transaction email."""
39 
40    purpose: str  # the MessagePurpose value: proposal / confirm / clarify / ...
41    payee: str
42    amount_display: str
43    txn_date: str  # already display-formatted (e.g. "May 29")
44    memo: str | None = None
45    proposed_category: str | None = None  # the best-guess category NAME
46    alternatives: tuple[str, ...] = ()  # other category names to offer
47    rationale: str | None = None  # one-line reason for the best guess
48    question: str | None = (
49        None  # an explicit question, when the purpose has one
50    )
51 
52 
53def _facts(request: ComposeRequest) -> str:
54    """The transaction header line, plus the memo line when there is one."""
55    header = f"{request.payee} — {request.amount_display} — {request.txn_date}"
56    if request.memo and request.memo.strip():
57        return f"{header}\n{request.memo.strip()}"
58    return header
59 
60 
61def _proposal_body(request: ComposeRequest, facts: str) -> str:
62    suggested = (
63        f"Suggested: {request.proposed_category or '(needs a category)'}"
64    )
65    if request.alternatives:
66        suggested += f"   (or: {', '.join(request.alternatives)})"
67    lines = [facts, "", suggested]
68    if request.rationale:
69        lines.append(request.rationale)
70    lines += ["", _REPLY_HINT]
71    return "\n".join(lines)
72 
73 
74def render_autonomy_offer(payee: str, category: str) -> str:
75    """The one-time "want me to auto-handle this payee?" offer body (§14.7 3b).
76 
77    A standalone yes/no message (its own thread), so a plain "yes"/"no" reply is
78    unambiguous. The owner can also reply with the explicit standing command.
79    """
80    return (
81        f"You've consistently filed {payee} under {category}, and I haven't "
82        "had to correct it.\n\n"
83        f"Want me to start auto-handling {payee} as {category} from now on? "
84        "I'll apply it automatically, flag each one for you, and you can undo "
85        "any of them with a one-word reply.\n\n"
86        "Reply YES to let me, or NO to keep approving each one yourself."
87    )
88 
89 
90def render_offer_accepted(payee: str, category: str) -> str:
91    """The confirmation sent when the owner accepts the offer (§14.7 3b)."""
92    return (
93        f"Great — I'll auto-handle {payee} as {category} from now on, and "
94        "flag each one so you can see (and undo) it. Reply any time to change "
95        "this."
96    )
97 
98 
99def render_offer_declined(payee: str) -> str:
100    """The brief note sent when the owner declines the offer (§14.7 3b)."""
101    return (
102        f"No problem — I'll keep proposing {payee} for you to approve, same "
103        "as before."
104    )
105 
106 
107def render_receipt_unsupported() -> str:
108    """The honest note for a forwarded receipt the join can't process yet (§6).
109 
110    The receipt⇄transaction join (W4) is a deferred increment, so rather than
111    swallow a forwarded receipt silently, the agent acknowledges it and points
112    the owner at the path that does work: replying on the transaction's own
113    email thread.
114    """
115    return (
116        "Thanks for forwarding this. I can't match forwarded receipts to "
117        "transactions yet, so I haven't filed it.\n\n"
118        "To add detail to a specific charge — an item list, a split, or a note "
119        "— just reply on that transaction's own email thread and I'll fold it "
120        "in."
121    )
122 
123 
124def render_balance_options(
⋯ 78 lines hidden (lines 125–202)125    needy_name: str, options: tuple[BalanceOption, ...]
126) -> str:
127    """The balance offer: ways to cover an overspend, each explained (§8).
128 
129    Numbered so the owner can reply "option 2", but a free-text answer ("take it
130    from dining instead", "only $50") is read just as well. Each option leads
131    with the model's rationale, the plain-English description of the moves.
132    """
133    lines = [
134        f"{needy_name} is over budget. Here are some ways I can cover it by "
135        "moving money between categories (nothing leaves your accounts):",
136        "",
137    ]
138    for index, option in enumerate(options, start=1):
139        lines.append(f"{index}. {option.label} — {option.rationale}")
140    lines += [
141        "",
142        "Reply with the option you'd like (or your own tweak — e.g. "
143        '"option 2 but only $50", or "no thanks").',
144    ]
145    return "\n".join(lines)
146 
147 
148def render_balance_applied(needy_name: str, total: Money) -> str:
149    """The confirmation after a coverage plan is applied (SPEC §8)."""
150    return (
151        f"Done — moved {total} into {needy_name} to cover the overspend. "
152        "Reply any time to adjust it."
153    )
154 
155 
156def render_balance_declined(needy_name: str) -> str:
157    """The brief note when the owner declines to cover (SPEC §8)."""
158    return (
159        f"No problem — I'll leave {needy_name} as is. Reply any time if you "
160        "change your mind."
161    )
162 
163 
164def render_balance_could_not_cover(needy_name: str) -> str:
165    """The note when no safe coverage exists from current funds (SPEC §8)."""
166    return (
167        f"{needy_name} is over budget, but I couldn't find a safe way to cover "
168        "it from your current funds. You may want to move money in manually."
169    )
170 
171 
172def render_balance_failed(needy_name: str, reason: str) -> str:
173    """The note when an approved plan can't be applied (SPEC §8)."""
174    return (
175        f"I couldn't cover {needy_name}: {reason}. Nothing was changed — reply "
176        "and we can try another way."
177    )
178 
179 
180def render_body(request: ComposeRequest) -> str:
181    """Lay out the email body for one transaction message (SPEC §5).
182 
183    The subject is templated separately by the caller; this is the body.
184    """
185    facts = _facts(request)
186    if request.purpose == MessagePurpose.PROPOSAL.value:
187        return _proposal_body(request, facts)
188    if request.purpose == MessagePurpose.CONFIRM.value:
189        category = request.proposed_category or "the category you picked"
190        return f"{facts}\n\nDone — set to {category} and approved."
191    if request.purpose == MessagePurpose.CLARIFY.value:
192        question = request.question or "Which category should this be?"
193        return f"{facts}\n\n{question}"
194    if request.purpose == MessagePurpose.OVERRIDE_NOTICE.value:
195        return (
196            f"{facts}\n\nNoticed you recategorized this one yourself — I've "
197            "backed off auto-handling this payee and will go back to asking."
198        )
199    # fyi / archive_notice / revise_summary / handoff / possibly_inconsistent /
200    # diverged_readback — a brief, neutral note (question carries any detail).
201    note = request.question or "A quick note on this transaction."
202    return f"{facts}\n\n{note}"