What broke, and the fix

Verifying the live dashboard (PR #11) against the cluster's SQL (standard) Temporal visibility turned the Temporal panel red with operation is not supported: 'order by' clause. The poll-heartbeat query asked the visibility store to ORDER BY StartTime DESC, which standard visibility (unlike Elasticsearch) does not support. Worse, because fetch wrapped all the Temporal sub-reads in one try, that single bad query blanked the whole panel — heartbeat, lifecycle, autonomy ladder, everything.

No ORDER BY: take a bounded page, pick the latest tick client-side.

src/ynab_agent/dashboard/temporal_source.py · 346 lines
src/ynab_agent/dashboard/temporal_source.py346 lines · Python
⋯ 76 lines hidden (lines 1–76)
1"""Temporal reader: the live run ledger (reuses the worker's client).
2 
3Reads the agent's durable state straight from Temporal: the poll heartbeat, the
4in-flight W2 transactions counted by lifecycle state (each running workflow's
5``state`` query), the rule registry's autonomy ladder (its ``view`` query), the
6live autonomy offers, the recent inbound-dispatch tally, and any
7terminated/failed workflows with their recovered reason. Everything is bounded
8and best-effort: a failure returns whatever was gathered plus an error string,
9never raising into the page.
10"""
11 
12from __future__ import annotations
13 
14import contextlib
15from datetime import datetime
16from typing import TYPE_CHECKING, Final
17 
18from temporalio.client import WorkflowExecutionStatus
19 
20from ynab_agent.dashboard.model import (
21 DispatchTally,
22 Failure,
23 OfferRow,
24 QueueItem,
25 RuleRow,
26 StateCount,
28from ynab_agent.domain.base import Frozen
29 
30if TYPE_CHECKING:
31 from collections.abc import AsyncIterator
32 
33 from temporalio.client import Client, WorkflowExecution
34 
35_MAX_PER_TYPE: Final = 500
36_MAX_STATE_QUERIES: Final = 200
37_REGISTRY_ID: Final = "ynab-rule-registry"
38 
39 
40class TemporalReadout(Frozen):
41 """The Temporal-derived pieces of the dashboard, ready to slot in."""
42 
43 poll_status: str = "none"
44 poll_live: bool = False
45 poll_last_start: datetime | None = None
46 lifecycle_states: tuple[StateCount, ...] = ()
47 in_flight: int = 0
48 archived: int = 0
49 terminated: int = 0
50 rules: tuple[RuleRow, ...] = ()
51 observe: int = 0
52 eligible: int = 0
53 blessed: int = 0
54 offers: tuple[OfferRow, ...] = ()
55 awaiting: tuple[QueueItem, ...] = ()
56 dispatch: DispatchTally = DispatchTally()
57 failures: tuple[Failure, ...] = ()
58 
59 
60def _status(execution: WorkflowExecution) -> str:
61 status = execution.status
62 return status.name.lower() if status is not None else "unknown"
63 
64 
65async def _take(
66 iterator: AsyncIterator[WorkflowExecution], cap: int = _MAX_PER_TYPE
67) -> AsyncIterator[WorkflowExecution]:
68 """Yield at most ``cap`` executions (a runaway-visibility backstop)."""
69 seen = 0
70 async for execution in iterator:
71 yield execution
72 seen += 1
73 if seen >= cap:
74 return
75 
76 
77async def _poll(client: Client) -> tuple[str, bool, datetime | None]:
78 """The most recent poll tick: (status, live, last_start).
79 
80 Standard (SQL) Temporal visibility rejects an ``ORDER BY`` clause, so we
81 take a bounded page and pick the latest start client-side rather than
82 sorting in the query.
83 """
84 latest: WorkflowExecution | None = None
85 async for execution in _take(
86 client.list_workflows("WorkflowType = 'PollWorkflow'"), cap=100
87 ):
88 start = execution.start_time
89 if latest is None or (
90 start is not None
91 and (latest.start_time is None or start > latest.start_time)
92 ):
93 latest = execution
94 if latest is None:
95 return "none", False, None
96 live = latest.status in (
97 WorkflowExecutionStatus.RUNNING,
98 WorkflowExecutionStatus.COMPLETED,
99 WorkflowExecutionStatus.CONTINUED_AS_NEW,
100 )
101 return _status(latest), live, latest.start_time
⋯ 245 lines hidden (lines 102–346)
102 
103 
104async def _lifecycle(
105 client: Client,
106) -> tuple[tuple[StateCount, ...], int, tuple[QueueItem, ...]]:
107 """Count running W2s by state; collect the awaiting-human queue."""
108 counts: dict[str, int] = {}
109 awaiting: list[QueueItem] = []
110 seen = 0
111 async for execution in _take(
112 client.list_workflows(
113 "WorkflowType = 'TransactionWorkflow' "
114 "AND ExecutionStatus = 'Running'"
115 ),
116 cap=_MAX_STATE_QUERIES,
117 ):
118 seen += 1
119 try:
120 handle = client.get_workflow_handle(
121 execution.id, run_id=execution.run_id
122 )
123 state = await handle.query("state", result_type=str)
124 except Exception: # a busy/odd workflow shouldn't drop the whole panel
125 state = "unknown"
126 counts[state] = counts.get(state, 0) + 1
127 if state == "awaiting_human":
128 awaiting.append(
129 QueueItem(
130 kind="proposal",
131 label=execution.id,
132 ident=execution.id,
133 since=execution.start_time,
134 )
135 )
136 states = tuple(
137 StateCount(state=name, count=counts[name]) for name in sorted(counts)
138 )
139 return states, seen, tuple(awaiting)
140 
141 
142async def _registry(
143 client: Client,
144) -> tuple[tuple[RuleRow, ...], int, int, int]:
145 """The rule table + (observe, eligible, blessed) counts (view query)."""
146 from ynab_agent.domain.allocations import ProposedCategory
147 from ynab_agent.domain.enums import RuleSource, TrustState
148 from ynab_agent.workflow.registry_types import RegistryView
149 
150 handle = client.get_workflow_handle(_REGISTRY_ID)
151 view = await handle.query("view", result_type=RegistryView)
152 rows: list[RuleRow] = []
153 observe = eligible = blessed = 0
154 for rule in view.rules:
155 allocation = rule.action.allocation
156 category = (
157 str(allocation.category)
158 if isinstance(allocation, ProposedCategory)
159 else "split"
160 )
161 rows.append(
162 RuleRow(
163 payee=rule.match.payee_pattern,
164 category=category,
165 trust=rule.trust.value,
166 source=rule.source.value,
167 hits=rule.hits,
168 offered=rule.offered_at is not None,
169 last_confirmed_at=rule.last_confirmed_at,
170 )
171 )
172 if rule.source is RuleSource.HUMAN_EXPLICIT:
173 blessed += 1
174 elif rule.trust is TrustState.TRUSTED:
175 eligible += 1
176 else:
177 observe += 1
178 return tuple(rows), observe, eligible, blessed
179 
180 
181async def _offers(client: Client) -> tuple[OfferRow, ...]:
182 """The live autonomy-offer workflows (awaiting the owner's yes/no)."""
183 offers: list[OfferRow] = []
184 async for execution in _take(
185 client.list_workflows(
186 "WorkflowType = 'AutonomyOfferWorkflow' "
187 "AND ExecutionStatus = 'Running'"
188 )
189 ):
190 offers.append(
191 OfferRow(
192 rule_id=execution.id.removeprefix("autonomy-offer-"),
193 payee="",
194 status=_status(execution),
195 started_at=execution.start_time,
196 )
197 )
198 return tuple(offers)
199 
200 
201async def _dispatch(client: Client) -> DispatchTally:
202 """Tally recent inbound dispatch results by routing action."""
203 counts: dict[str, int] = {}
204 total = 0
205 async for execution in _take(
206 client.list_workflows(
207 "WorkflowType = 'DispatchWorkflow' "
208 "AND ExecutionStatus = 'Completed'"
209 )
210 ):
211 total += 1
212 action = "?"
213 try:
214 handle = client.get_workflow_handle(
215 execution.id, run_id=execution.run_id
216 )
217 result = await handle.result()
218 got = getattr(result, "action", None)
219 if isinstance(got, str):
220 action = got
221 except Exception: # a bad decode shouldn't drop the tally
222 action = "?"
223 counts[action] = counts.get(action, 0) + 1
224 return DispatchTally(
225 transaction=counts.get("transaction", 0),
226 offer=counts.get("offer", 0),
227 receipt=counts.get("receipt", 0),
228 command=counts.get("command", 0),
229 quarantine=counts.get("quarantine", 0),
230 ignore=counts.get("ignore", 0),
231 total=total,
232 )
233 
234 
235async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
236 """Recover a terminated/failed workflow's human reason from history."""
237 try:
238 handle = client.get_workflow_handle(
239 execution.id, run_id=execution.run_id
240 )
241 found: str | None = None
242 async for event in handle.fetch_history_events():
243 terminated = event.workflow_execution_terminated_event_attributes
244 if terminated.reason:
245 found = terminated.reason
246 failed = event.workflow_execution_failed_event_attributes
247 if failed.failure.message:
248 found = failed.failure.message
249 return found
250 except Exception: # the reason is a nicety — never fail the panel for it
251 return None
252 
253 
254async def _terminal(client: Client) -> tuple[int, int, tuple[Failure, ...]]:
255 """Recent archived (completed W2) + terminated/failed counts + failures."""
256 archived = terminated = 0
257 failures: list[Failure] = []
258 async for _completed in _take(
259 client.list_workflows(
260 "WorkflowType = 'TransactionWorkflow' "
261 "AND ExecutionStatus = 'Completed'"
262 )
263 ):
264 archived += 1
265 async for execution in _take(
266 client.list_workflows(
267 "ExecutionStatus = 'Terminated' OR ExecutionStatus = 'Failed'"
268 )
269 ):
270 terminated += 1
271 if len(failures) < 25:
272 failures.append(
273 Failure(
274 workflow_id=execution.id,
275 kind=_status(execution),
276 reason=await _reason(client, execution),
277 when=execution.close_time,
278 )
279 )
280 return archived, terminated, tuple(failures)
281 
282 
283async def fetch(client: Client) -> tuple[TemporalReadout, str | None]:
284 """Read the agent's Temporal state; each panel degrades independently.
285 
286 Every sub-read is guarded on its own, so one unsupported visibility query
287 (or a transient failure) reddens only its panel and names itself in the
288 error — the rest of the page still fills in.
289 """
290 poll_status = "none"
291 poll_live = False
292 poll_last: datetime | None = None
293 states: tuple[StateCount, ...] = ()
294 in_flight = 0
295 awaiting: tuple[QueueItem, ...] = ()
296 rules: tuple[RuleRow, ...] = ()
297 observe = eligible = blessed = 0
298 offers: tuple[OfferRow, ...] = ()
299 dispatch = DispatchTally()
300 archived = terminated = 0
301 failures: tuple[Failure, ...] = ()
302 errors: list[str] = []
303 
304 try:
305 poll_status, poll_live, poll_last = await _poll(client)
306 except Exception as exc:
307 errors.append(f"poll: {type(exc).__name__}")
308 try:
309 states, in_flight, awaiting = await _lifecycle(client)
310 except Exception as exc:
311 errors.append(f"lifecycle: {type(exc).__name__}")
312 # The registry may not exist until the first learning signal — a normal
313 # state, not an error, so its absence is simply suppressed.
314 with contextlib.suppress(Exception):
315 rules, observe, eligible, blessed = await _registry(client)
316 try:
317 offers = await _offers(client)
318 except Exception as exc:
319 errors.append(f"offers: {type(exc).__name__}")
320 try:
321 dispatch = await _dispatch(client)
322 except Exception as exc:
323 errors.append(f"dispatch: {type(exc).__name__}")
324 try:
325 archived, terminated, failures = await _terminal(client)
326 except Exception as exc:
327 errors.append(f"terminal: {type(exc).__name__}")
328 
329 readout = TemporalReadout(
330 poll_status=poll_status,
331 poll_live=poll_live,
332 poll_last_start=poll_last,
333 lifecycle_states=states,
334 in_flight=in_flight,
335 archived=archived,
336 terminated=terminated,
337 rules=rules,
338 observe=observe,
339 eligible=eligible,
340 blessed=blessed,
341 offers=offers,
342 awaiting=awaiting,
343 dispatch=dispatch,
344 failures=failures,
345 )
346 return readout, "; ".join(errors) if errors else None

Degrade each panel on its own

The deeper fix is resilience: every sub-read is now guarded independently, so an unsupported (or transient) query reddens only its own panel and names itself in the error string — the rest of the page still fills in. This is the same "every source degrades to a dot" contract the cross-source readers already had, now applied within the Temporal reader too.

src/ynab_agent/dashboard/temporal_source.py · 346 lines
src/ynab_agent/dashboard/temporal_source.py346 lines · Python
⋯ 282 lines hidden (lines 1–282)
1"""Temporal reader: the live run ledger (reuses the worker's client).
2 
3Reads the agent's durable state straight from Temporal: the poll heartbeat, the
4in-flight W2 transactions counted by lifecycle state (each running workflow's
5``state`` query), the rule registry's autonomy ladder (its ``view`` query), the
6live autonomy offers, the recent inbound-dispatch tally, and any
7terminated/failed workflows with their recovered reason. Everything is bounded
8and best-effort: a failure returns whatever was gathered plus an error string,
9never raising into the page.
10"""
11 
12from __future__ import annotations
13 
14import contextlib
15from datetime import datetime
16from typing import TYPE_CHECKING, Final
17 
18from temporalio.client import WorkflowExecutionStatus
19 
20from ynab_agent.dashboard.model import (
21 DispatchTally,
22 Failure,
23 OfferRow,
24 QueueItem,
25 RuleRow,
26 StateCount,
28from ynab_agent.domain.base import Frozen
29 
30if TYPE_CHECKING:
31 from collections.abc import AsyncIterator
32 
33 from temporalio.client import Client, WorkflowExecution
34 
35_MAX_PER_TYPE: Final = 500
36_MAX_STATE_QUERIES: Final = 200
37_REGISTRY_ID: Final = "ynab-rule-registry"
38 
39 
40class TemporalReadout(Frozen):
41 """The Temporal-derived pieces of the dashboard, ready to slot in."""
42 
43 poll_status: str = "none"
44 poll_live: bool = False
45 poll_last_start: datetime | None = None
46 lifecycle_states: tuple[StateCount, ...] = ()
47 in_flight: int = 0
48 archived: int = 0
49 terminated: int = 0
50 rules: tuple[RuleRow, ...] = ()
51 observe: int = 0
52 eligible: int = 0
53 blessed: int = 0
54 offers: tuple[OfferRow, ...] = ()
55 awaiting: tuple[QueueItem, ...] = ()
56 dispatch: DispatchTally = DispatchTally()
57 failures: tuple[Failure, ...] = ()
58 
59 
60def _status(execution: WorkflowExecution) -> str:
61 status = execution.status
62 return status.name.lower() if status is not None else "unknown"
63 
64 
65async def _take(
66 iterator: AsyncIterator[WorkflowExecution], cap: int = _MAX_PER_TYPE
67) -> AsyncIterator[WorkflowExecution]:
68 """Yield at most ``cap`` executions (a runaway-visibility backstop)."""
69 seen = 0
70 async for execution in iterator:
71 yield execution
72 seen += 1
73 if seen >= cap:
74 return
75 
76 
77async def _poll(client: Client) -> tuple[str, bool, datetime | None]:
78 """The most recent poll tick: (status, live, last_start).
79 
80 Standard (SQL) Temporal visibility rejects an ``ORDER BY`` clause, so we
81 take a bounded page and pick the latest start client-side rather than
82 sorting in the query.
83 """
84 latest: WorkflowExecution | None = None
85 async for execution in _take(
86 client.list_workflows("WorkflowType = 'PollWorkflow'"), cap=100
87 ):
88 start = execution.start_time
89 if latest is None or (
90 start is not None
91 and (latest.start_time is None or start > latest.start_time)
92 ):
93 latest = execution
94 if latest is None:
95 return "none", False, None
96 live = latest.status in (
97 WorkflowExecutionStatus.RUNNING,
98 WorkflowExecutionStatus.COMPLETED,
99 WorkflowExecutionStatus.CONTINUED_AS_NEW,
100 )
101 return _status(latest), live, latest.start_time
102 
103 
104async def _lifecycle(
105 client: Client,
106) -> tuple[tuple[StateCount, ...], int, tuple[QueueItem, ...]]:
107 """Count running W2s by state; collect the awaiting-human queue."""
108 counts: dict[str, int] = {}
109 awaiting: list[QueueItem] = []
110 seen = 0
111 async for execution in _take(
112 client.list_workflows(
113 "WorkflowType = 'TransactionWorkflow' "
114 "AND ExecutionStatus = 'Running'"
115 ),
116 cap=_MAX_STATE_QUERIES,
117 ):
118 seen += 1
119 try:
120 handle = client.get_workflow_handle(
121 execution.id, run_id=execution.run_id
122 )
123 state = await handle.query("state", result_type=str)
124 except Exception: # a busy/odd workflow shouldn't drop the whole panel
125 state = "unknown"
126 counts[state] = counts.get(state, 0) + 1
127 if state == "awaiting_human":
128 awaiting.append(
129 QueueItem(
130 kind="proposal",
131 label=execution.id,
132 ident=execution.id,
133 since=execution.start_time,
134 )
135 )
136 states = tuple(
137 StateCount(state=name, count=counts[name]) for name in sorted(counts)
138 )
139 return states, seen, tuple(awaiting)
140 
141 
142async def _registry(
143 client: Client,
144) -> tuple[tuple[RuleRow, ...], int, int, int]:
145 """The rule table + (observe, eligible, blessed) counts (view query)."""
146 from ynab_agent.domain.allocations import ProposedCategory
147 from ynab_agent.domain.enums import RuleSource, TrustState
148 from ynab_agent.workflow.registry_types import RegistryView
149 
150 handle = client.get_workflow_handle(_REGISTRY_ID)
151 view = await handle.query("view", result_type=RegistryView)
152 rows: list[RuleRow] = []
153 observe = eligible = blessed = 0
154 for rule in view.rules:
155 allocation = rule.action.allocation
156 category = (
157 str(allocation.category)
158 if isinstance(allocation, ProposedCategory)
159 else "split"
160 )
161 rows.append(
162 RuleRow(
163 payee=rule.match.payee_pattern,
164 category=category,
165 trust=rule.trust.value,
166 source=rule.source.value,
167 hits=rule.hits,
168 offered=rule.offered_at is not None,
169 last_confirmed_at=rule.last_confirmed_at,
170 )
171 )
172 if rule.source is RuleSource.HUMAN_EXPLICIT:
173 blessed += 1
174 elif rule.trust is TrustState.TRUSTED:
175 eligible += 1
176 else:
177 observe += 1
178 return tuple(rows), observe, eligible, blessed
179 
180 
181async def _offers(client: Client) -> tuple[OfferRow, ...]:
182 """The live autonomy-offer workflows (awaiting the owner's yes/no)."""
183 offers: list[OfferRow] = []
184 async for execution in _take(
185 client.list_workflows(
186 "WorkflowType = 'AutonomyOfferWorkflow' "
187 "AND ExecutionStatus = 'Running'"
188 )
189 ):
190 offers.append(
191 OfferRow(
192 rule_id=execution.id.removeprefix("autonomy-offer-"),
193 payee="",
194 status=_status(execution),
195 started_at=execution.start_time,
196 )
197 )
198 return tuple(offers)
199 
200 
201async def _dispatch(client: Client) -> DispatchTally:
202 """Tally recent inbound dispatch results by routing action."""
203 counts: dict[str, int] = {}
204 total = 0
205 async for execution in _take(
206 client.list_workflows(
207 "WorkflowType = 'DispatchWorkflow' "
208 "AND ExecutionStatus = 'Completed'"
209 )
210 ):
211 total += 1
212 action = "?"
213 try:
214 handle = client.get_workflow_handle(
215 execution.id, run_id=execution.run_id
216 )
217 result = await handle.result()
218 got = getattr(result, "action", None)
219 if isinstance(got, str):
220 action = got
221 except Exception: # a bad decode shouldn't drop the tally
222 action = "?"
223 counts[action] = counts.get(action, 0) + 1
224 return DispatchTally(
225 transaction=counts.get("transaction", 0),
226 offer=counts.get("offer", 0),
227 receipt=counts.get("receipt", 0),
228 command=counts.get("command", 0),
229 quarantine=counts.get("quarantine", 0),
230 ignore=counts.get("ignore", 0),
231 total=total,
232 )
233 
234 
235async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
236 """Recover a terminated/failed workflow's human reason from history."""
237 try:
238 handle = client.get_workflow_handle(
239 execution.id, run_id=execution.run_id
240 )
241 found: str | None = None
242 async for event in handle.fetch_history_events():
243 terminated = event.workflow_execution_terminated_event_attributes
244 if terminated.reason:
245 found = terminated.reason
246 failed = event.workflow_execution_failed_event_attributes
247 if failed.failure.message:
248 found = failed.failure.message
249 return found
250 except Exception: # the reason is a nicety — never fail the panel for it
251 return None
252 
253 
254async def _terminal(client: Client) -> tuple[int, int, tuple[Failure, ...]]:
255 """Recent archived (completed W2) + terminated/failed counts + failures."""
256 archived = terminated = 0
257 failures: list[Failure] = []
258 async for _completed in _take(
259 client.list_workflows(
260 "WorkflowType = 'TransactionWorkflow' "
261 "AND ExecutionStatus = 'Completed'"
262 )
263 ):
264 archived += 1
265 async for execution in _take(
266 client.list_workflows(
267 "ExecutionStatus = 'Terminated' OR ExecutionStatus = 'Failed'"
268 )
269 ):
270 terminated += 1
271 if len(failures) < 25:
272 failures.append(
273 Failure(
274 workflow_id=execution.id,
275 kind=_status(execution),
276 reason=await _reason(client, execution),
277 when=execution.close_time,
278 )
279 )
280 return archived, terminated, tuple(failures)
281 
282 
283async def fetch(client: Client) -> tuple[TemporalReadout, str | None]:
284 """Read the agent's Temporal state; each panel degrades independently.
285 
286 Every sub-read is guarded on its own, so one unsupported visibility query
287 (or a transient failure) reddens only its panel and names itself in the
288 error — the rest of the page still fills in.
289 """
290 poll_status = "none"
291 poll_live = False
292 poll_last: datetime | None = None
293 states: tuple[StateCount, ...] = ()
294 in_flight = 0
295 awaiting: tuple[QueueItem, ...] = ()
296 rules: tuple[RuleRow, ...] = ()
297 observe = eligible = blessed = 0
298 offers: tuple[OfferRow, ...] = ()
299 dispatch = DispatchTally()
300 archived = terminated = 0
301 failures: tuple[Failure, ...] = ()
302 errors: list[str] = []
303 
304 try:
305 poll_status, poll_live, poll_last = await _poll(client)
306 except Exception as exc:
307 errors.append(f"poll: {type(exc).__name__}")
308 try:
309 states, in_flight, awaiting = await _lifecycle(client)
310 except Exception as exc:
311 errors.append(f"lifecycle: {type(exc).__name__}")
312 # The registry may not exist until the first learning signal — a normal
313 # state, not an error, so its absence is simply suppressed.
314 with contextlib.suppress(Exception):
315 rules, observe, eligible, blessed = await _registry(client)
316 try:
317 offers = await _offers(client)
318 except Exception as exc:
319 errors.append(f"offers: {type(exc).__name__}")
320 try:
321 dispatch = await _dispatch(client)
322 except Exception as exc:
323 errors.append(f"dispatch: {type(exc).__name__}")
324 try:
325 archived, terminated, failures = await _terminal(client)
326 except Exception as exc:
327 errors.append(f"terminal: {type(exc).__name__}")
328 
329 readout = TemporalReadout(
330 poll_status=poll_status,
331 poll_live=poll_live,
332 poll_last_start=poll_last,
333 lifecycle_states=states,
⋯ 13 lines hidden (lines 334–346)
334 in_flight=in_flight,
335 archived=archived,
336 terminated=terminated,
337 rules=rules,
338 observe=observe,
339 eligible=eligible,
340 blessed=blessed,
341 offers=offers,
342 awaiting=awaiting,
343 dispatch=dispatch,
344 failures=failures,
345 )
346 return readout, "; ".join(errors) if errors else None