What changed, and why

A standing command — "always categorize X as Y" — grants the strongest thing the agent can hold: standing autonomy to auto-apply a payee. handle_command granted it inline: it parsed the message and signalled the registry's bless straight away, minting a human_explicit auto-apply rule with no confirmation.

SPEC §0.6 and §5c are explicit that this is the wrong shape. High-impact verbs — standing rules, trust elevation — must be echoed back and confirmed with a one-word reply. And §5c notes the sharp edge: a command can arrive on a brand-new thread, where the allow-list is the only gate. A misread parse, or a mistaken send, must not silently grant auto-apply.

So the command now goes through a read-back: echo the interpretation, wait for a one-word yes, and only then bless.

The command opens a confirm, instead of blessing

handle_command no longer signals bless. It parses the command as before, then starts a one-shot confirm workflow keyed by (payee, category). The bless moves into that workflow, behind the owner's yes.

Parse → start CommandConfirmWorkflow. ALLOW_DUPLICATE + catching the already-started error makes a resend a no-op while one is pending.

src/ynab_agent/workflow/dispatch_activities.py · 239 lines
src/ynab_agent/workflow/dispatch_activities.py239 lines · Python
⋯ 180 lines hidden (lines 1–180)
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 
81 
82@activity.defn
83async def resolve_balance_thread(thread_id: str | None) -> str | None:
84 """Resolve an AgentMail thread to a *running* balance workflow id (§8).
85 
86 The balance workflow stamps the overspend thread into ``BalanceThreadId``,
87 so a reply there maps back to it the same way ``resolve_offer_thread`` maps
88 an offer. Filtered to ``Running`` so a reply after the offer has closed is
89 not routed (it falls through to the command path). ``None`` when no live
90 balance offer owns the thread.
91 """
92 if thread_id is None:
93 return None
94 from ynab_agent.workflow.balance_types import BALANCE_THREAD_ID
95 
96 temporal = await client()
97 safe = thread_id.replace('"', '\\"')
98 query = f'{BALANCE_THREAD_ID} = "{safe}" AND ExecutionStatus = "Running"'
99 async for execution in temporal.list_workflows(query=query):
100 return execution.id
101 return None
102 
103 
104@activity.defn
105async def signal_balance(balance_id: str, message: InboundMessage) -> None:
106 """Deliver a reply to its balance workflow (SPEC §8).
107 
108 A plain signal: the workflow must be live to receive it —
109 ``resolve_balance_thread`` already filtered to running executions. A missing
110 handle is a benign no-op.
111 """
112 from temporalio.service import RPCError
113 
114 temporal = await client()
115 handle = temporal.get_workflow_handle(balance_id)
116 with contextlib.suppress(RPCError):
117 await handle.signal("submit_response", message)
118 
119 
120@activity.defn
121async def classify_inbound(message: InboundMessage) -> InboundKind:
122 """Agentically classify a non-thread message: receipt, command, or noise.
123 
124 The model stack is imported lazily here so it never enters the workflow
125 sandbox. The model only labels the message; the deterministic dispatcher
126 routes it (SPEC §5).
127 """
128 from ynab_agent.agentic.classify import classify_inbound as run_classifier
129 from ynab_agent.agentic.classify import to_kind
130 
131 return to_kind(await run_classifier(message))
132 
133 
134@activity.defn
135async def signal_transaction(txn_id: str, message: InboundMessage) -> None:
136 """Deliver a reply to the transaction's W2 (SPEC §5a).
137 
138 Signal-with-start on ``submit_inbound``: a running W2 (the common case — it
139 asked the question) receives the reply and wakes; if the transaction's run
140 has closed, a fresh W2 starts with the reply buffered and re-triages. The
141 workflow id is the YNAB transaction id, so no thread↔txn table is needed.
142 """
143 from ynab_agent.domain.ids import YnabTransactionId
144 from ynab_agent.domain.signals import ReplySignal
145 from ynab_agent.workflow.types import TransactionParams
146 
147 if message.thread_id is None:
148 # RouteToTransaction only fires for a thread-matched message, so a
149 # missing thread id here is a routing bug, not an expected input.
150 msg = f"signal for {txn_id} has no thread id"
151 raise RuntimeError(msg)
152 reply = ReplySignal(
153 thread_id=message.thread_id,
154 message_id=message.message_id,
155 from_address=message.from_address,
156 text=message.body,
157 )
158 temporal = await client()
159 await temporal.start_workflow(
160 "TransactionWorkflow",
161 TransactionParams(ynab_id=YnabTransactionId(txn_id)),
162 id=txn_id,
163 task_queue=task_queue(),
164 start_signal="submit_inbound",
165 start_signal_args=[reply],
166 )
167 
168 
169@activity.defn
170async def route_receipt(_message: InboundMessage) -> None:
171 """Hand a forwarded receipt to the W4 join (SPEC §5b).
172 
173 No-op in v1: the receipt join (W4) is out of the core-triage scope, and the
174 agent does not advertise receipt forwarding, so receipts do not arrive yet.
175 Wired when W4 lands (the message arg is kept for that contract).
176 """
177 return None
178 
179 
180@activity.defn
181async def handle_command(message: InboundMessage) -> None:
182 """Parse a standing command and open a read-back to confirm it (SPEC §5c).
183 
184 The owner's direct opt-in: "always categorize X as Y" becomes an
185 ``ExplicitCommand``. Because a standing command grants autonomy — and can
186 arrive on a brand-new thread where the allow-list is the only gate — it is
187 *not* blessed inline (SPEC §0.6). Instead a one-shot
188 ``CommandConfirmWorkflow`` opens a read-back ("Auto-handle X as Y? reply
189 yes") and blesses only on a one-word confirm. The confirm is keyed by
190 (payee, category): a resend while one is pending is a no-op; a later command
191 after it closed re-opens a fresh one. Anything not a clear bless — a
192 question or comment — is a deliberate no-op (the parser declines it), so a
193 stray message never starts one.
194 """
195 import asyncio
196 
197 from temporalio.common import WorkflowIDReusePolicy
198 from temporalio.exceptions import WorkflowAlreadyStartedError
199 
200 from ynab_agent.agentic.command import (
201 CommandRequest,
202 parse_command,
203 to_explicit_command,
204 )
205 from ynab_agent.agentic.enrich import CandidateCategory
206 from ynab_agent.workflow.command_types import (
207 CommandConfirmParams,
208 command_confirm_id,
209 )
210 from ynab_agent.workflow.temporal_client import client, task_queue
211 from ynab_agent.ynab.client import YnabClient
212 
213 ynab = YnabClient.from_env()
214 spends = await asyncio.to_thread(ynab.category_spends)
215 candidates = tuple(
216 CandidateCategory(id=str(spend.category), name=spend.name)
217 for spend in spends
218 )
219 if not candidates:
220 return
221 reading = await parse_command(
222 CommandRequest(command_text=message.body, candidates=candidates)
223 )
224 command = to_explicit_command(reading)
225 if command is None:
226 return
227 temporal = await client()
228 try:
229 await temporal.start_workflow(
230 "CommandConfirmWorkflow",
231 CommandConfirmParams(command=command),
232 id=command_confirm_id(command),
233 task_queue=task_queue(),
234 id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
235 )
236 except WorkflowAlreadyStartedError:
237 # A confirm for this (payee, category) is already pending: a resend is a
238 # no-op (SPEC §5c, idempotent against resends).
⋯ 1 line hidden (lines 239–239)
239 return

The confirm workflow — a mirror of the proactive offer

The confirm workflow is deliberately the twin of the proactive AutonomyOfferWorkflow: open a read-back thread, stamp the shared OfferThreadId so W3 routes the reply back, then wait — a yes blesses, a no declines, an unclear reply keeps waiting, and silence past the window blesses nothing. The one difference from the offer is what a yes blesses: an explicit (payee, category) command rather than an already-eligible rule.

Open → stamp the shared routing attribute → wait → interpret → accept/decline. The reply interpreter is reused verbatim from the offer.

src/ynab_agent/workflow/command_workflow.py · 135 lines
src/ynab_agent/workflow/command_workflow.py135 lines · Python
⋯ 85 lines hidden (lines 1–85)
1"""The command-confirm workflow — read-back before blessing (SPEC §5c, §0.6).
2 
3A standing command ("always categorize X as Y") grants standing autonomy, so it
4is not blessed inline. This short-lived workflow opens a read-back thread that
5echoes the interpretation, stamps the shared ``OfferThreadId`` search attribute
6so W3 routes the reply back here (no new dispatch path — a command-confirm reply
7is an autonomy decision, like an offer reply), and waits for a one-word confirm.
8A yes blesses the rule (the registry ``bless``) and confirms; a no sends a brief
9note; an unclear reply keeps waiting; silence past the patience window simply
10never blesses.
11 
12Mirrors :class:`~ynab_agent.workflow.offer_workflow.AutonomyOfferWorkflow` (the
13proactive prompt); the difference is only *what* a yes blesses — an explicit
14(payee, category) command rather than an already-eligible rule — so the reply
15interpreter (``offer_activities.interpret_offer_reply``) is shared.
16"""
17 
18from __future__ import annotations
19 
20from collections import deque
21from datetime import timedelta
22 
23from temporalio import workflow
24from temporalio.common import SearchAttributeKey
25from temporalio.exceptions import ActivityError
26 
27from ynab_agent.workflow.offer_types import OFFER_THREAD_ID
28 
29_OFFER_THREAD_ID = SearchAttributeKey.for_keyword(OFFER_THREAD_ID)
30 
31with workflow.unsafe.imports_passed_through():
32 from ynab_agent.dispatch.classify import InboundMessage
33 from ynab_agent.domain.enums import OfferVerdict
34 from ynab_agent.workflow import (
35 alert_activities,
36 command_activities,
37 offer_activities,
38 )
39 from ynab_agent.workflow.alerting import build_failure_alert
40 from ynab_agent.workflow.command_types import (
41 COMMAND_CONFIRM_PATIENCE,
42 CommandConfirmParams,
43 )
44 from ynab_agent.workflow.constants import (
45 ACTIVITY_RETRY,
46 ACTIVITY_TIMEOUT,
47 ALERT_BUDGET,
48 ALERT_RETRY,
49 ALERT_TIMEOUT,
50 )
51 
52 
53@workflow.defn
54class CommandConfirmWorkflow:
55 """One standing command's read-back + one-word confirm before blessing."""
56 
57 def __init__(self) -> None:
58 """Start with no reply buffered; ``run`` opens the thread and waits."""
59 self._responses: deque[InboundMessage] = deque()
60 
61 @workflow.signal
62 def submit_response(self, message: InboundMessage) -> None:
63 """The owner replied to the read-back (delivered by W3)."""
64 self._responses.append(message)
65 
66 @workflow.run
67 async def run(self, params: CommandConfirmParams) -> None:
68 """Read back the command and act on the reply, paging on failure."""
69 try:
70 await self._run(params)
71 except ActivityError as exc:
72 payee = params.command.match.payee_pattern
73 await workflow.execute_activity(
74 alert_activities.alert_failure,
75 build_failure_alert(
76 key=f"command-confirm-{payee}",
77 context=f"command confirm for {payee}",
78 exc=exc,
79 ),
80 start_to_close_timeout=ALERT_TIMEOUT,
81 schedule_to_close_timeout=ALERT_BUDGET,
82 retry_policy=ALERT_RETRY,
83 )
84 raise
85 
86 async def _run(self, params: CommandConfirmParams) -> None:
87 command = params.command
88 thread_id = await workflow.execute_activity(
89 command_activities.open_command_thread,
90 command,
91 start_to_close_timeout=ACTIVITY_TIMEOUT,
92 retry_policy=ACTIVITY_RETRY,
93 )
94 # Index this workflow by its thread so a reply routes here, reusing the
95 # offer's routing (a command-confirm reply is an autonomy decision).
96 workflow.upsert_search_attributes(
97 [_OFFER_THREAD_ID.value_set(thread_id)]
98 )
99 
100 deadline = workflow.now() + COMMAND_CONFIRM_PATIENCE
101 while True:
102 timeout = deadline - workflow.now()
103 if timeout < timedelta(0):
104 timeout = timedelta(0)
105 try:
106 await workflow.wait_condition(
107 lambda: len(self._responses) > 0, timeout=timeout
108 )
109 except TimeoutError:
110 # No clear confirm in the window; never bless, end.
111 return
112 message = self._responses.popleft()
113 verdict = await workflow.execute_activity(
114 offer_activities.interpret_offer_reply,
115 args=[message.body, command.match.payee_pattern],
116 start_to_close_timeout=ACTIVITY_TIMEOUT,
117 retry_policy=ACTIVITY_RETRY,
118 )
119 if verdict is OfferVerdict.ACCEPT:
120 await workflow.execute_activity(
121 command_activities.accept_command,
122 args=[command, thread_id],
123 start_to_close_timeout=ACTIVITY_TIMEOUT,
124 retry_policy=ACTIVITY_RETRY,
125 )
126 return
127 if verdict is OfferVerdict.DECLINE:
128 await workflow.execute_activity(
129 command_activities.decline_command,
130 args=[command, thread_id],
131 start_to_close_timeout=ACTIVITY_TIMEOUT,
132 retry_policy=ACTIVITY_RETRY,
133 )
134 return
135 # UNCLEAR: keep waiting for a clearer reply until the deadline.

Blessing only on a yes

The accept activity is where the bless finally happens — the same registry bless signal handle_command used to send inline, now reached only after the owner confirmed. The read-back and the confirmation prose round out the flow.

open_command_thread echoes the command; accept_command signals the registry bless and confirms — the inline bless, now gated.

src/ynab_agent/workflow/command_activities.py · 149 lines
src/ynab_agent/workflow/command_activities.py149 lines · Python
⋯ 49 lines hidden (lines 1–49)
1"""The I/O ports of the command-confirm subsystem, as Temporal activities.
2 
3A standing command ("always categorize X as Y") grants standing autonomy, so it
4is not blessed inline — it goes through a read-back + one-word confirm (SPEC
5§5c, §0.6). These wire that flow: the confirm workflow opens its read-back
6thread (:func:`open_command_thread`); the reply is read by the *shared* offer
7interpreter (``offer_activities.interpret_offer_reply``); a yes blesses the rule
8via the registry and confirms (:func:`accept_command`); a no sends a brief note
9(:func:`decline_command`). Heavy clients are imported lazily so they never enter
10a workflow sandbox.
11"""
12 
13from __future__ import annotations
14 
15from temporalio import activity
16 
17from ynab_agent.learn.events import ExplicitCommand
18 
19 
20def _category_id(command: ExplicitCommand) -> str | None:
21 """The single category a command names, or ``None`` for a split."""
22 from ynab_agent.domain.allocations import ProposedCategory
23 
24 allocation = command.action.allocation
25 if isinstance(allocation, ProposedCategory):
26 return str(allocation.category)
27 return None
28 
29 
30async def _category_names() -> dict[str, str]:
31 """A category-id→name map read from YNAB (the source of truth)."""
32 import asyncio
33 
34 from ynab_agent.ynab.client import YnabClient
35 
36 client = YnabClient.from_env()
37 spends = await asyncio.to_thread(client.category_spends)
38 return {str(spend.category): spend.name for spend in spends}
39 
40 
41def _category_display(command: ExplicitCommand, names: dict[str, str]) -> str:
42 """The display name of the category a command names (best effort)."""
43 cid = _category_id(command)
44 if cid is None:
45 return "a split"
46 return names.get(cid, cid)
47 
48 
49@activity.defn
50async def open_command_thread(command: ExplicitCommand) -> str:
51 """Open the read-back thread echoing the command and return its id (§5c).
52 
53 A new standalone thread (its own ``yacmd-`` idempotency label, so a retry
54 re-finds it rather than sending twice), addressed to the owners. The payee
55 comes from the command; the category name from a YNAB read at send time.
56 """
57 import asyncio
58 
59 from ynab_agent.agentic.compose import render_command_confirm
60 from ynab_agent.mail.client import MailClient
⋯ 20 lines hidden (lines 61–80)
61 from ynab_agent.settings import Settings
62 from ynab_agent.workflow.command_types import command_confirm_id
63 
64 settings = Settings()
65 names = await _category_names()
66 payee = command.match.payee_pattern
67 category = _category_display(command, names)
68 body = render_command_confirm(payee, category)
69 mail = MailClient.from_env()
70 return await asyncio.to_thread(
71 mail.open_thread,
72 inbox_id=settings.inbox,
73 to=list(settings.owners),
74 subject=f"Confirm: always {payee} as {category}?",
75 body=body,
76 txn_label=f"yacmd-{command_confirm_id(command)}",
77 )
78 
79 
80@activity.defn
81async def accept_command(command: ExplicitCommand, thread_id: str) -> None:
82 """Bless the command and confirm — the owner confirmed it (SPEC §5c, §14).
83 
84 Signals the durable registry's ``bless`` (signal-with-start, reusing the
85 running singleton) with the explicit command, which trusts the rule for
86 auto-apply (``source=human_explicit``), then sends a confirmation on the
87 read-back thread (idempotent on its accept label, so a retry never
88 double-sends).
89 """
90 import asyncio
91 
92 from temporalio.common import WorkflowIDConflictPolicy
93 
94 from ynab_agent.agentic.compose import render_offer_accepted
95 from ynab_agent.mail.client import MailClient
96 from ynab_agent.settings import Settings
97 from ynab_agent.workflow.command_types import command_confirm_id
98 from ynab_agent.workflow.registry_types import (
99 REGISTRY_WORKFLOW_ID,
100 RegistryParams,
101 )
102 from ynab_agent.workflow.temporal_client import client, task_queue
103 
104 temporal = await client()
105 await temporal.start_workflow(
106 "RuleRegistryWorkflow",
107 RegistryParams(),
108 id=REGISTRY_WORKFLOW_ID,
109 task_queue=task_queue(),
110 id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
111 start_signal="bless",
112 start_signal_args=[command],
⋯ 37 lines hidden (lines 113–149)
113 )
114 
115 settings = Settings()
116 names = await _category_names()
117 body = render_offer_accepted(
118 command.match.payee_pattern, _category_display(command, names)
119 )
120 mail = MailClient.from_env()
121 await asyncio.to_thread(
122 mail.send_on_thread,
123 inbox_id=settings.inbox,
124 thread_id=thread_id,
125 body=body,
126 seq_label=f"yacmd-accept-{command_confirm_id(command)}",
127 )
128 
129 
130@activity.defn
131async def decline_command(command: ExplicitCommand, thread_id: str) -> None:
132 """Send the brief "I'll keep proposing" note — owner declined (SPEC §5c)."""
133 import asyncio
134 
135 from ynab_agent.agentic.compose import render_offer_declined
136 from ynab_agent.mail.client import MailClient
137 from ynab_agent.settings import Settings
138 from ynab_agent.workflow.command_types import command_confirm_id
139 
140 settings = Settings()
141 body = render_offer_declined(command.match.payee_pattern)
142 mail = MailClient.from_env()
143 await asyncio.to_thread(
144 mail.send_on_thread,
145 inbox_id=settings.inbox,
146 thread_id=thread_id,
147 body=body,
148 seq_label=f"yacmd-decline-{command_confirm_id(command)}",
149 )

The read-back: echoes the interpretation and asks for a one-word YES (SPEC §0.6).

src/ynab_agent/agentic/compose.py · 201 lines
src/ynab_agent/agentic/compose.py201 lines · Python
⋯ 89 lines hidden (lines 1–89)
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."
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_command_confirm(payee: str, category: str) -> str:
91 """The read-back for an explicit "always X as Y" command (SPEC §5c, §0.6).
92 
93 A standing command grants autonomy, so the agent echoes its interpretation
94 and waits for a one-word confirm before blessing — a command can arrive on a
95 brand-new thread where the allow-list is the only gate, so a misread or a
96 mistaken send must not silently grant auto-apply.
97 """
98 return (
99 f"You asked me to always categorize {payee} as {category}.\n\n"
100 f"Reply YES to confirm — I'll auto-handle {payee} as {category} from "
101 "now on, flag each one for you, and you can undo any with a one-word "
102 "reply. Reply NO to keep approving each one yourself."
103 )
104 
⋯ 97 lines hidden (lines 105–201)
105 
106def render_offer_accepted(payee: str, category: str) -> str:
107 """The confirmation sent when the owner accepts the offer (§14.7 3b)."""
108 return (
109 f"Great — I'll auto-handle {payee} as {category} from now on, and "
110 "flag each one so you can see (and undo) it. Reply any time to change "
111 "this."
112 )
113 
114 
115def render_offer_declined(payee: str) -> str:
116 """The brief note sent when the owner declines the offer (§14.7 3b)."""
117 return (
118 f"No problem — I'll keep proposing {payee} for you to approve, same "
119 "as before."
120 )
121 
122 
123def render_balance_options(
124 needy_name: str, options: tuple[BalanceOption, ...]
125) -> str:
126 """The balance offer: ways to cover an overspend, each explained (§8).
127 
128 Numbered so the owner can reply "option 2", but a free-text answer ("take it
129 from dining instead", "only $50") is read just as well. Each option leads
130 with the model's rationale, the plain-English description of the moves.
131 """
132 lines = [
133 f"{needy_name} is over budget. Here are some ways I can cover it by "
134 "moving money between categories (nothing leaves your accounts):",
135 "",
136 ]
137 for index, option in enumerate(options, start=1):
138 lines.append(f"{index}. {option.label} — {option.rationale}")
139 lines += [
140 "",
141 "Reply with the option you'd like (or your own tweak — e.g. "
142 '"option 2 but only $50", or "no thanks").',
143 ]
144 return "\n".join(lines)
145 
146 
147def render_balance_applied(needy_name: str, total: Money) -> str:
148 """The confirmation after a coverage plan is applied (SPEC §8)."""
149 return (
150 f"Done — moved {total} into {needy_name} to cover the overspend. "
151 "Reply any time to adjust it."
152 )
153 
154 
155def render_balance_declined(needy_name: str) -> str:
156 """The brief note when the owner declines to cover (SPEC §8)."""
157 return (
158 f"No problem — I'll leave {needy_name} as is. Reply any time if you "
159 "change your mind."
160 )
161 
162 
163def render_balance_could_not_cover(needy_name: str) -> str:
164 """The note when no safe coverage exists from current funds (SPEC §8)."""
165 return (
166 f"{needy_name} is over budget, but I couldn't find a safe way to cover "
167 "it from your current funds. You may want to move money in manually."
168 )
169 
170 
171def render_balance_failed(needy_name: str, reason: str) -> str:
172 """The note when an approved plan can't be applied (SPEC §8)."""
173 return (
174 f"I couldn't cover {needy_name}: {reason}. Nothing was changed — reply "
175 "and we can try another way."
176 )
177 
178 
179def render_body(request: ComposeRequest) -> str:
180 """Lay out the email body for one transaction message (SPEC §5).
181 
182 The subject is templated separately by the caller; this is the body.
183 """
184 facts = _facts(request)
185 if request.purpose == MessagePurpose.PROPOSAL.value:
186 return _proposal_body(request, facts)
187 if request.purpose == MessagePurpose.CONFIRM.value:
188 category = request.proposed_category or "the category you picked"
189 return f"{facts}\n\nDone — set to {category} and approved."
190 if request.purpose == MessagePurpose.CLARIFY.value:
191 question = request.question or "Which category should this be?"
192 return f"{facts}\n\n{question}"
193 if request.purpose == MessagePurpose.OVERRIDE_NOTICE.value:
194 return (
195 f"{facts}\n\nNoticed you recategorized this one yourself — I've "
196 "backed off auto-handling this payee and will go back to asking."
197 )
198 # fyi / archive_notice / revise_summary / handoff / possibly_inconsistent /
199 # diverged_readback — a brief, neutral note (question carries any detail).
200 note = request.question or "A quick note on this transaction."
201 return f"{facts}\n\n{note}"

Tests

The workflow is pinned end to end on the time-skipping server: a yes blesses, a no declines, an unclear-then-yes blesses, and silence blesses nothing. And the activity-level test proves the behavioural change directly — a bless command now starts the confirm workflow, never a direct registry bless.

Yes blesses; silence never does.

tests/workflow/test_command_workflow.py · 175 lines
tests/workflow/test_command_workflow.py175 lines · Python
⋯ 132 lines hidden (lines 1–132)
1"""End-to-end tests for the command-confirm workflow (SPEC §5c, §0.6).
2 
3A standing command is read back and blessed only on a one-word confirm. Run on
4the time-skipping server with mock activities (mirroring the offer workflow): a
5yes blesses, a no declines, an unclear reply keeps waiting, and silence ends
6without blessing. The workflow stamps the shared ``OfferThreadId`` on open, so
7the env registers it (as the cluster does via manage/search-attributes.yaml).
8"""
9 
10from __future__ import annotations
11 
12from typing import TYPE_CHECKING
13 
14from temporalio import activity
15from temporalio.api.enums.v1 import IndexedValueType
16from temporalio.api.operatorservice.v1 import AddSearchAttributesRequest
17from temporalio.testing import WorkflowEnvironment
18from temporalio.worker import Worker
19 
20from ynab_agent.dispatch.classify import InboundMessage
21from ynab_agent.domain.allocations import ProposedCategory
22from ynab_agent.domain.enums import OfferVerdict
23from ynab_agent.domain.ids import CategoryId, MessageId, ThreadId
24from ynab_agent.domain.rule import RuleAction, RuleMatch
25from ynab_agent.learn.events import ExplicitCommand
26from ynab_agent.workflow.command_types import CommandConfirmParams
27from ynab_agent.workflow.command_workflow import CommandConfirmWorkflow
28from ynab_agent.workflow.runtime import DATA_CONVERTER
29 
30if TYPE_CHECKING:
31 from collections.abc import Callable
32 
33_TASK_QUEUE = "command-wf-test"
34 
35 
36def _command() -> ExplicitCommand:
37 return ExplicitCommand(
38 match=RuleMatch(payee_pattern="Costco"),
39 action=RuleAction(
40 allocation=ProposedCategory(category=CategoryId("groceries"))
41 ),
42 )
43 
44 
45def _reply(text: str = "yes do it") -> InboundMessage:
46 return InboundMessage(
47 message_id=MessageId("m1"),
48 from_address="matthew@example.com",
49 subject="re: Confirm: always Costco as Groceries?",
50 body=text,
51 thread_id=ThreadId("thr-cmd"),
52 signature_verified=True,
53 )
54 
55 
56def _activities(
57 *, verdicts: list[OfferVerdict], calls: list[str]
58) -> list[Callable[..., object]]:
59 seq = list(verdicts)
60 
61 @activity.defn(name="open_command_thread")
62 async def open_command_thread(command: ExplicitCommand) -> str:
63 calls.append("open")
64 return "thr-cmd"
65 
66 @activity.defn(name="interpret_offer_reply")
67 async def interpret_offer_reply(
68 reply_text: str, payee: str
69 ) -> OfferVerdict:
70 return seq.pop(0)
71 
72 @activity.defn(name="accept_command")
73 async def accept_command(command: ExplicitCommand, thread_id: str) -> None:
74 calls.append(f"accept:{thread_id}")
75 
76 @activity.defn(name="decline_command")
77 async def decline_command(command: ExplicitCommand, thread_id: str) -> None:
78 calls.append(f"decline:{thread_id}")
79 
80 @activity.defn(name="alert_failure")
81 async def alert_failure(alert: object) -> None:
82 return None
83 
84 return [
85 open_command_thread,
86 interpret_offer_reply,
87 accept_command,
88 decline_command,
89 alert_failure,
90 ]
91 
92 
93async def _start_env() -> WorkflowEnvironment:
94 env = await WorkflowEnvironment.start_time_skipping(
95 data_converter=DATA_CONVERTER
96 )
97 await env.client.operator_service.add_search_attributes(
98 AddSearchAttributesRequest(
99 namespace="default",
100 search_attributes={
101 "OfferThreadId": IndexedValueType.INDEXED_VALUE_TYPE_KEYWORD
102 },
103 )
104 )
105 return env
106 
107 
108async def _run(*, verdicts: list[OfferVerdict], replies: int) -> list[str]:
109 calls: list[str] = []
110 async with (
111 await _start_env() as env,
112 Worker(
113 env.client,
114 task_queue=_TASK_QUEUE,
115 workflows=[CommandConfirmWorkflow],
116 activities=_activities(verdicts=verdicts, calls=calls),
117 ),
118 ):
119 handle = await env.client.start_workflow(
120 CommandConfirmWorkflow.run,
121 CommandConfirmParams(command=_command()),
122 id="command-confirm-test",
123 task_queue=_TASK_QUEUE,
124 )
125 for _ in range(replies):
126 await handle.signal(
127 CommandConfirmWorkflow.submit_response, _reply()
128 )
129 await handle.result()
130 return calls
131 
132 
133async def test_confirm_blesses_only_after_a_yes() -> None:
134 calls = await _run(verdicts=[OfferVerdict.ACCEPT], replies=1)
135 assert "open" in calls
136 assert "accept:thr-cmd" in calls
137 assert not any(c.startswith("decline") for c in calls)
138 
139 
140async def test_decline_does_not_bless() -> None:
141 calls = await _run(verdicts=[OfferVerdict.DECLINE], replies=1)
142 assert "decline:thr-cmd" in calls
143 assert not any(c.startswith("accept") for c in calls)
144 
145 
146async def test_unclear_keeps_waiting_then_a_clear_yes_blesses() -> None:
147 calls = await _run(
148 verdicts=[OfferVerdict.UNCLEAR, OfferVerdict.ACCEPT], replies=2
149 )
150 assert "accept:thr-cmd" in calls
151 assert not any(c.startswith("decline") for c in calls)
152 
153 
154async def test_silence_never_blesses() -> None:
155 # No confirm at all: the patience window elapses (time-skipped) and the
156 # command is never blessed — the read-back alone grants nothing.
157 calls = await _run(verdicts=[], replies=0)
158 assert calls == ["open"]
⋯ 17 lines hidden (lines 159–175)
159 
160 
161def test_command_confirm_id_is_stable_per_payee_category() -> None:
162 from ynab_agent.workflow.command_types import command_confirm_id
163 
164 same = command_confirm_id(_command())
165 assert same == command_confirm_id(
166 _command()
167 ) # deterministic: dedups resends
168 assert same.startswith("command-confirm-")
169 other = ExplicitCommand(
170 match=RuleMatch(payee_pattern="Costco"),
171 action=RuleAction(
172 allocation=ProposedCategory(category=CategoryId("dining"))
173 ),
174 )
175 assert command_confirm_id(other) != same # a different target → fresh id

handle_command opens a confirm, not an inline bless.

tests/workflow/test_dispatch_activities.py · 255 lines
tests/workflow/test_dispatch_activities.py255 lines · Python
⋯ 226 lines hidden (lines 1–226)
1"""Tests for the W3 dispatch activities (SPEC §5).
2 
3``resolve_thread`` maps an AgentMail thread id back to its transaction via a
4Temporal visibility query on the ``TxnThreadId`` search attribute (store-free,
5SPEC §0.5); ``signal_transaction`` turns a verified reply into a
6``submit_inbound`` signal-with-start on that W2. Both are exercised against a
7fake client injected as the cached connection; the live calls are covered by
8the worker against a real Temporal server.
9"""
10 
11from __future__ import annotations
12 
13import asyncio
14from typing import TYPE_CHECKING
15 
16import pytest
17 
18from ynab_agent.dispatch.classify import InboundMessage
19from ynab_agent.domain.ids import MessageId, ThreadId
20from ynab_agent.workflow import dispatch_activities, temporal_client
21 
22if TYPE_CHECKING:
23 from collections.abc import AsyncIterator
24 
25 
26class _FakeExecution:
27 def __init__(self, workflow_id: str) -> None:
28 self.id = workflow_id
29 
30 
31class _FakeClient:
32 """Stands in for the Temporal client's ``list_workflows`` visibility API."""
33 
34 def __init__(self, executions: list[_FakeExecution]) -> None:
35 self._executions = executions
36 self.queries: list[str] = []
37 
38 def list_workflows(self, query: str) -> AsyncIterator[_FakeExecution]:
39 self.queries.append(query)
40 
41 async def _gen() -> AsyncIterator[_FakeExecution]:
42 for execution in self._executions:
43 yield execution
44 
45 return _gen()
46 
47 
48def test_resolve_thread_returns_matching_workflow_id(
49 monkeypatch: pytest.MonkeyPatch,
50) -> None:
51 fake = _FakeClient([_FakeExecution("txn-123")])
52 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
53 result = asyncio.run(dispatch_activities.resolve_thread("thread-abc"))
54 assert result == "txn-123"
55 assert fake.queries == ['TxnThreadId = "thread-abc"']
56 
57 
58def test_resolve_thread_none_when_no_workflow_matches(
59 monkeypatch: pytest.MonkeyPatch,
60) -> None:
61 fake = _FakeClient([])
62 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
63 assert asyncio.run(dispatch_activities.resolve_thread("orphan")) is None
64 
65 
66def test_resolve_thread_none_for_none_input() -> None:
67 # No client is touched when there is no thread id to resolve.
68 assert asyncio.run(dispatch_activities.resolve_thread(None)) is None
69 
70 
71def test_resolve_thread_escapes_quotes_in_query(
72 monkeypatch: pytest.MonkeyPatch,
73) -> None:
74 fake = _FakeClient([_FakeExecution("txn-9")])
75 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
76 asyncio.run(dispatch_activities.resolve_thread('th"read'))
77 assert fake.queries == ['TxnThreadId = "th\\"read"']
78 
79 
80def test_resolve_offer_thread_query_filters_to_running(
81 monkeypatch: pytest.MonkeyPatch,
82) -> None:
83 fake = _FakeClient([_FakeExecution("autonomy-offer-r1")])
84 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
85 result = asyncio.run(dispatch_activities.resolve_offer_thread("thread-abc"))
86 assert result == "autonomy-offer-r1"
87 # A closed offer must not be resurrected, so the query is Running-only.
88 assert fake.queries == [
89 'OfferThreadId = "thread-abc" AND ExecutionStatus = "Running"'
90 ]
91 
92 
93def test_resolve_offer_thread_none_for_none_input() -> None:
94 assert asyncio.run(dispatch_activities.resolve_offer_thread(None)) is None
95 
96 
97class _FakeHandle:
98 def __init__(self) -> None:
99 self.signals: list[tuple[str, object]] = []
100 
101 async def signal(self, name: str, arg: object) -> None:
102 self.signals.append((name, arg))
103 
104 
105class _FakeHandleClient:
106 """Stands in for ``get_workflow_handle`` + ``signal``."""
107 
108 def __init__(self) -> None:
109 self.handle = _FakeHandle()
110 self.requested: list[str] = []
111 
112 def get_workflow_handle(self, workflow_id: str) -> _FakeHandle:
113 self.requested.append(workflow_id)
114 return self.handle
115 
116 
117def test_signal_offer_signals_the_offer_workflow(
118 monkeypatch: pytest.MonkeyPatch,
119) -> None:
120 fake = _FakeHandleClient()
121 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
122 asyncio.run(
123 dispatch_activities.signal_offer(
124 "autonomy-offer-r1", _message(thread_id="t")
125 )
126 )
127 assert fake.requested == ["autonomy-offer-r1"]
128 assert len(fake.handle.signals) == 1
129 name, arg = fake.handle.signals[0]
130 assert name == "submit_response"
131 assert arg.body == "actually make it Dining" # type: ignore[attr-defined]
132 
133 
134class _FakeStartClient:
135 """Captures ``start_workflow`` (signal-with-start) calls."""
136 
137 def __init__(self) -> None:
138 self.started: list[tuple[object, object, dict[str, object]]] = []
139 
140 async def start_workflow(
141 self, workflow: object, arg: object, **kwargs: object
142 ) -> None:
143 self.started.append((workflow, arg, kwargs))
144 
145 
146def _message(*, thread_id: str | None) -> InboundMessage:
147 return InboundMessage(
148 message_id=MessageId("m1"),
149 from_address="owner@example.com",
150 subject="re: coffee",
151 body="actually make it Dining",
152 thread_id=ThreadId(thread_id) if thread_id is not None else None,
153 signature_verified=True,
154 )
155 
156 
157def test_signal_transaction_signals_with_the_reply(
158 monkeypatch: pytest.MonkeyPatch,
159) -> None:
160 fake = _FakeStartClient()
161 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
162 asyncio.run(
163 dispatch_activities.signal_transaction("txn-1", _message(thread_id="t"))
164 )
165 assert len(fake.started) == 1
166 workflow, _arg, kwargs = fake.started[0]
167 assert workflow == "TransactionWorkflow"
168 assert kwargs["id"] == "txn-1"
169 assert kwargs["start_signal"] == "submit_inbound"
170 reply = kwargs["start_signal_args"][0] # type: ignore[index]
171 assert reply.text == "actually make it Dining"
172 assert reply.thread_id == "t"
173 
174 
175def test_signal_transaction_requires_a_thread_id() -> None:
176 with pytest.raises(RuntimeError, match="no thread id"):
177 asyncio.run(
178 dispatch_activities.signal_transaction(
179 "txn-1", _message(thread_id=None)
180 )
181 )
182 
183 
184# ── handle_command: read-back + confirm before blessing (SPEC §5c, §0.6) ─────
185class _FakeYnab:
186 def category_spends(self) -> tuple[object, ...]:
187 from ynab_agent.budget.overspend import CategorySpend
188 from ynab_agent.domain.ids import CategoryId
189 from ynab_agent.domain.money import Money
190 
191 zero = Money.from_milliunits(0)
192 return (
193 CategorySpend(
194 category=CategoryId("groceries"),
195 name="Groceries",
196 budgeted=zero,
197 activity=zero,
198 balance=zero,
199 ),
200 )
201 
202 
203def _stub_command(monkeypatch: pytest.MonkeyPatch, *, bless: bool) -> None:
204 """Stub the YNAB read and the model parse so handle_command is offline."""
205 from ynab_agent.agentic import command as command_mod
206 from ynab_agent.ynab.client import YnabClient
207 
208 monkeypatch.setattr(
209 YnabClient, "from_env", classmethod(lambda cls: _FakeYnab())
210 )
211 
212 async def _parse(request: object) -> command_mod.CommandReading:
213 kind = (
214 command_mod.CommandKind.BLESS
215 if bless
216 else command_mod.CommandKind.OTHER
217 )
218 return command_mod.CommandReading(
219 kind=kind,
220 payee_pattern="Costco" if bless else None,
221 category_id="groceries" if bless else None,
222 )
223 
224 monkeypatch.setattr(command_mod, "parse_command", _parse)
225 
226 
227def test_handle_command_opens_a_confirm_not_an_inline_bless(
228 monkeypatch: pytest.MonkeyPatch,
229) -> None:
230 # A standing command no longer blesses inline (SPEC §0.6): it starts a
231 # CommandConfirmWorkflow read-back, never signals the registry directly.
232 _stub_command(monkeypatch, bless=True)
233 fake = _FakeStartClient()
234 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
235 
236 asyncio.run(dispatch_activities.handle_command(_message(thread_id="t")))
237 
238 assert len(fake.started) == 1
239 workflow, arg, kwargs = fake.started[0]
240 # Crucially the confirm workflow starts — not a direct registry bless.
241 assert workflow == "CommandConfirmWorkflow"
242 assert str(kwargs["id"]).startswith("command-confirm-")
243 assert arg.command.match.payee_pattern == "Costco" # type: ignore[attr-defined]
244 
245 
246def test_handle_command_ignores_a_non_bless(
247 monkeypatch: pytest.MonkeyPatch,
248) -> None:
⋯ 7 lines hidden (lines 249–255)
249 _stub_command(monkeypatch, bless=False)
250 fake = _FakeStartClient()
251 monkeypatch.setattr(temporal_client, "_CLIENT", fake)
252 
253 asyncio.run(dispatch_activities.handle_command(_message(thread_id="t")))
254 
255 assert fake.started == []