What changed, and why it's safe to merge
The in-worker operations dashboard was a wall of eleven panels organized by subsystem — Temporal, YNAB, ClickHouse, AgentMail, GitHub — so the operator had to assemble the two answers they actually want ("is it healthy?" and "do I need to do anything?") out of scattered tables. Several panels also misled: a benign backlog of resting workflows read as "40 in flight"; intentional go-live resets read as failures; the human queue listed bare UUIDs.
This PR keeps the same data flow — sources → read_model.assemble() (pure) → render.page() (pure) — but reshapes the view-model and the page around the operator's questions: a composite Health verdict, a deterministic plain-English Narrative, a humanized needs-you queue split by whether you already settled the transaction in YNAB, a health-and-flow zone, an activity feed, and the operator detail folded into progressive-disclosure rails.
Blast radius is small by construction. The dashboard runs in-process on the worker's event loop, gated and try/except'd, so a failure degrades the page, never the triage loop. Every new YNAB call is a read. The one new piece of I/O (resolve_queue) is bounded, runs off the event loop, has a wall-clock timeout, and falls back to bare ids on any error. make check is green (ruff, mypy --strict, 486 tests), and an adversarial multi-agent pre-ship review found zero ship-blockers.
The view-model: questions, not subsystems
The model now leads with the two headline projections. Narrative is the plain-English summary; Health is the single masthead verdict (one dot + word) plus the few figures shown beside it. Note the comment on Health.needs_you: an item waiting on the owner is normal operation, not a fault — so it never flips the verdict.
The two at-a-glance types that now lead the model.
src/ynab_agent/dashboard/model.py · 285 lines
src/ynab_agent/dashboard/model.py
⋯ 29 lines hidden (lines 1–29)
1"""The dashboard view-model — frozen values rendered to one HTML page.
3A pure projection assembled by :mod:`ynab_agent.dashboard.read_model` from the
4source readers and rendered by :mod:`ynab_agent.dashboard.render`. Nothing here
5performs I/O; every field is a plain value so the page is fully testable without
6a cluster.
8The shape is organized around the *questions an operator actually asks* — "is it
9healthy?", "do I need to do anything?", "what has it done?" — not around the
10subsystems the data came from. :class:`Narrative` and :class:`Health` are the
11headline projections that answer the first two at a glance.
12"""
14from __future__ import annotations
16from datetime import datetime
18from ynab_agent.domain.base import Frozen
21class SourceHealth(Frozen):
22 """One data source's liveness dot (green when ``ok``, else red)."""
24 name: str
25 ok: bool
26 detail: str
29# ── State of things (the narrative) ──────────────────────────────────────────
30class Narrative(Frozen):
31 """The plain-English 'state of things' — deterministic, render-ready.
33 A composed-from-the-numbers summary so a human comprehends the whole board
34 without decoding panels. ``tone`` drives its accent; ``paragraphs`` is one
35 or two friendly sentences. Built deterministically in
36 :func:`~ynab_agent.dashboard.read_model.narrate`; an optional LLM polish may
37 later rephrase the prose, always falling back to this text.
38 """
40 headline: str
41 paragraphs: tuple[str, ...] = ()
42 tone: str = "ok" # "ok" | "warn" | "bad"
45# ── Is it healthy? (the masthead rollup) ─────────────────────────────────────
46class Health(Frozen):
47 """The single composite health verdict shown in the masthead.
49 ``tone``/``label`` are the one status dot + word; the rest are the few
50 figures the masthead surfaces beside it. ``needs_you`` is *operation*, not
51 *fault* — a healthy agent can still have items waiting on the owner.
52 """
54 tone: str = "ok" # "ok" | "warn" | "bad"
55 label: str = "healthy"
56 poll_live: bool = False
57 poll_status: str = "none" # the W1 poll's latest execution status
58 poll_last_start: datetime | None = None
59 worker_last_span: datetime | None = None
60 span_error_rate: float | None = None
61 needs_you: int = 0
62 real_failures: int = 0
65# ── Transaction lifecycle ────────────────────────────────────────────────────
66class StateCount(Frozen):
⋯ 219 lines hidden (lines 67–285)
67 """A lifecycle state and how many in-flight transactions are in it."""
69 state: str
70 count: int
73class Lifecycle(Frozen):
74 """The in-flight W2 funnel plus recent terminal counts."""
76 states: tuple[StateCount, ...] = ()
77 in_flight: int = 0
78 archived: int = 0
79 terminated: int = 0
82# ── Autonomy ladder (the rule registry) ──────────────────────────────────────
83class RuleRow(Frozen):
84 """One learned/blessed rule, reduced to what the ladder shows."""
86 payee: str
87 category: str
88 trust: str
89 source: str
90 hits: int
91 offered: bool
92 last_confirmed_at: datetime | None = None
95class OfferRow(Frozen):
96 """One live autonomy-offer workflow (awaiting the owner's yes/no)."""
98 rule_id: str
99 payee: str
100 status: str
101 started_at: datetime | None = None
104class Autonomy(Frozen):
105 """The earned-autonomy picture: counts, the rule table, live offers."""
107 observe: int = 0
108 eligible: int = 0
109 blessed: int = 0
110 rules: tuple[RuleRow, ...] = ()
111 offers: tuple[OfferRow, ...] = ()
114# ── The human's queue ────────────────────────────────────────────────────────
115class TxnFacts(Frozen):
116 """YNAB facts resolved for a queued transaction (the humanizing join).
118 Keyed by ``ynab_id`` outside this type; produced by the YNAB source so the
119 read-model can purely turn a bare workflow id into a readable row.
120 ``approved`` is the pivotal signal: an awaiting proposal whose transaction
121 is already ``approved`` in YNAB is one the owner settled *in-app* — it will
122 lapse on its own and does not need them.
123 """
125 payee: str
126 amount: str
127 approved: bool
128 category: str | None = None
131class QueueItem(Frozen):
132 """Something waiting on the owner: an open proposal or a pending offer.
134 ``ident`` is the stable key (the YNAB txn id for a proposal, the rule id for
135 an offer); the optional humanizing fields are filled by the join when the
136 facts are reachable, and the renderer degrades to ``label`` (a short id)
137 when they are not.
138 """
140 kind: str # "proposal" | "offer"
141 label: str
142 ident: str
143 since: datetime | None = None
144 payee: str | None = None
145 amount: str | None = None
146 category: str | None = None
147 approved: bool | None = None # YNAB approved flag; None = unknown / offer
148 question: str | None = None # the proposal's email subject, when matched
151# ── Budget (YNAB) ────────────────────────────────────────────────────────────
152class CategoryRow(Frozen):
153 """A category's month-to-date balance (negative = overspent)."""
155 name: str
156 balance: str
157 overspent: bool
160class TxnRow(Frozen):
161 """A transaction reduced to a payee + amount display."""
163 payee: str
164 amount: str
167class Budget(Frozen):
168 """The budget surface: the unapproved backlog and overspent categories."""
170 available: bool = False
171 unapproved: int = 0
172 unapproved_sample: tuple[TxnRow, ...] = ()
173 overspent: tuple[CategoryRow, ...] = ()
176# ── Conversations (AgentMail) ────────────────────────────────────────────────
177class Conversation(Frozen):
178 """A recent AgentMail thread reduced to a one-line summary.
180 ``ref`` carries the transaction/rule id recovered from the agent's own
181 ``yatxn-``/``yaoffer-`` label, so the queue can borrow a thread's subject as
182 the most human description of the item it is waiting on.
183 """
185 subject: str
186 preview: str
187 kind: str # "proposal" | "offer" | "thread"
188 updated_at: datetime | None = None
189 ref: str | None = None
192# ── Inbound (W3 dispatch) ────────────────────────────────────────────────────
193class DispatchTally(Frozen):
194 """Recent inbound dispatch results, counted by routing action."""
196 transaction: int = 0
197 offer: int = 0
198 receipt: int = 0
199 command: int = 0
200 quarantine: int = 0
201 ignore: int = 0
202 total: int = 0
205# ── Run telemetry (ClickHouse) ───────────────────────────────────────────────
206class ActivityStat(Frozen):
207 """Per-activity run count and latency (from the trace spans)."""
209 name: str
210 count: int
211 avg_ms: float
212 max_ms: float
215class RunTelemetry(Frozen):
216 """Trace-derived health for the worker (best-effort enrichment)."""
218 available: bool = False
219 total_spans: int = 0
220 error_spans: int = 0
221 last_activity: datetime | None = None
222 window_days: int = 3
223 activities: tuple[ActivityStat, ...] = ()
224 recent_errors: tuple[str, ...] = ()
227# ── Failures ─────────────────────────────────────────────────────────────────
228class Failure(Frozen):
229 """A workflow that ended terminated/failed, with its recovered reason.
231 ``intentional`` marks the operator's own go-live / re-test / reset
232 terminations, so the page can keep them out of the fault headline (they are
233 housekeeping, not breakage) while still listing them under disclosure.
234 """
236 workflow_id: str
237 kind: str
238 reason: str | None = None
239 when: datetime | None = None
240 intentional: bool = False
243# ── Deploy (GitHub) ──────────────────────────────────────────────────────────
244class PrRow(Frozen):
245 """A recent pull request reduced to title + state + CI."""
247 number: int
248 title: str
249 state: str
250 ci: str | None
251 url: str
252 when: datetime | None = None
255class Deploy(Frozen):
256 """Recent repo activity: PRs and their CI."""
258 prs: tuple[PrRow, ...] = ()
261# ── The whole page ───────────────────────────────────────────────────────────
262class DashboardModel(Frozen):
263 """Everything one dashboard request renders.
265 ``narrative`` and ``health`` lead (the at-a-glance answer); ``needs_you`` is
266 the only acted-upon queue, ``handled`` the already-settled proposals still
267 winding down. The remaining fields back the supporting zones and the
268 progressive-disclosure rails.
269 """
271 generated_at: datetime
272 repo: str
273 narrative: Narrative
274 health: Health
275 sources: tuple[SourceHealth, ...]
276 needs_you: tuple[QueueItem, ...]
277 handled: tuple[QueueItem, ...]
278 lifecycle: Lifecycle
279 autonomy: Autonomy
280 budget: Budget
281 conversations: tuple[Conversation, ...]
282 dispatch: DispatchTally
283 telemetry: RunTelemetry
284 failures: tuple[Failure, ...]
285 deploy: Deploy
The queue gains the fields that make a row readable, and a dedicated facts type the YNAB source fills in. The pivotal field is TxnFacts.approved: a proposal whose transaction is already approved in YNAB is one you settled in-app, so it belongs in handled (it lapses on its own), not in your action list.
TxnFacts (the humanizing join's payload) and the enriched QueueItem.
src/ynab_agent/dashboard/model.py · 285 lines
src/ynab_agent/dashboard/model.py
⋯ 114 lines hidden (lines 1–114)
1"""The dashboard view-model — frozen values rendered to one HTML page.
3A pure projection assembled by :mod:`ynab_agent.dashboard.read_model` from the
4source readers and rendered by :mod:`ynab_agent.dashboard.render`. Nothing here
5performs I/O; every field is a plain value so the page is fully testable without
6a cluster.
8The shape is organized around the *questions an operator actually asks* — "is it
9healthy?", "do I need to do anything?", "what has it done?" — not around the
10subsystems the data came from. :class:`Narrative` and :class:`Health` are the
11headline projections that answer the first two at a glance.
12"""
14from __future__ import annotations
16from datetime import datetime
18from ynab_agent.domain.base import Frozen
21class SourceHealth(Frozen):
22 """One data source's liveness dot (green when ``ok``, else red)."""
24 name: str
25 ok: bool
26 detail: str
29# ── State of things (the narrative) ──────────────────────────────────────────
30class Narrative(Frozen):
31 """The plain-English 'state of things' — deterministic, render-ready.
33 A composed-from-the-numbers summary so a human comprehends the whole board
34 without decoding panels. ``tone`` drives its accent; ``paragraphs`` is one
35 or two friendly sentences. Built deterministically in
36 :func:`~ynab_agent.dashboard.read_model.narrate`; an optional LLM polish may
37 later rephrase the prose, always falling back to this text.
38 """
40 headline: str
41 paragraphs: tuple[str, ...] = ()
42 tone: str = "ok" # "ok" | "warn" | "bad"
45# ── Is it healthy? (the masthead rollup) ─────────────────────────────────────
46class Health(Frozen):
47 """The single composite health verdict shown in the masthead.
49 ``tone``/``label`` are the one status dot + word; the rest are the few
50 figures the masthead surfaces beside it. ``needs_you`` is *operation*, not
51 *fault* — a healthy agent can still have items waiting on the owner.
52 """
54 tone: str = "ok" # "ok" | "warn" | "bad"
55 label: str = "healthy"
56 poll_live: bool = False
57 poll_status: str = "none" # the W1 poll's latest execution status
58 poll_last_start: datetime | None = None
59 worker_last_span: datetime | None = None
60 span_error_rate: float | None = None
61 needs_you: int = 0
62 real_failures: int = 0
65# ── Transaction lifecycle ────────────────────────────────────────────────────
66class StateCount(Frozen):
67 """A lifecycle state and how many in-flight transactions are in it."""
69 state: str
70 count: int
73class Lifecycle(Frozen):
74 """The in-flight W2 funnel plus recent terminal counts."""
76 states: tuple[StateCount, ...] = ()
77 in_flight: int = 0
78 archived: int = 0
79 terminated: int = 0
82# ── Autonomy ladder (the rule registry) ──────────────────────────────────────
83class RuleRow(Frozen):
84 """One learned/blessed rule, reduced to what the ladder shows."""
86 payee: str
87 category: str
88 trust: str
89 source: str
90 hits: int
91 offered: bool
92 last_confirmed_at: datetime | None = None
95class OfferRow(Frozen):
96 """One live autonomy-offer workflow (awaiting the owner's yes/no)."""
98 rule_id: str
99 payee: str
100 status: str
101 started_at: datetime | None = None
104class Autonomy(Frozen):
105 """The earned-autonomy picture: counts, the rule table, live offers."""
107 observe: int = 0
108 eligible: int = 0
109 blessed: int = 0
110 rules: tuple[RuleRow, ...] = ()
111 offers: tuple[OfferRow, ...] = ()
114# ── The human's queue ────────────────────────────────────────────────────────
115class TxnFacts(Frozen):
116 """YNAB facts resolved for a queued transaction (the humanizing join).
118 Keyed by ``ynab_id`` outside this type; produced by the YNAB source so the
119 read-model can purely turn a bare workflow id into a readable row.
120 ``approved`` is the pivotal signal: an awaiting proposal whose transaction
121 is already ``approved`` in YNAB is one the owner settled *in-app* — it will
122 lapse on its own and does not need them.
123 """
125 payee: str
126 amount: str
127 approved: bool
128 category: str | None = None
131class QueueItem(Frozen):
132 """Something waiting on the owner: an open proposal or a pending offer.
134 ``ident`` is the stable key (the YNAB txn id for a proposal, the rule id for
135 an offer); the optional humanizing fields are filled by the join when the
136 facts are reachable, and the renderer degrades to ``label`` (a short id)
137 when they are not.
138 """
140 kind: str # "proposal" | "offer"
141 label: str
142 ident: str
143 since: datetime | None = None
144 payee: str | None = None
145 amount: str | None = None
146 category: str | None = None
147 approved: bool | None = None # YNAB approved flag; None = unknown / offer
148 question: str | None = None # the proposal's email subject, when matched
151# ── Budget (YNAB) ────────────────────────────────────────────────────────────
152class CategoryRow(Frozen):
⋯ 133 lines hidden (lines 153–285)
153 """A category's month-to-date balance (negative = overspent)."""
155 name: str
156 balance: str
157 overspent: bool
160class TxnRow(Frozen):
161 """A transaction reduced to a payee + amount display."""
163 payee: str
164 amount: str
167class Budget(Frozen):
168 """The budget surface: the unapproved backlog and overspent categories."""
170 available: bool = False
171 unapproved: int = 0
172 unapproved_sample: tuple[TxnRow, ...] = ()
173 overspent: tuple[CategoryRow, ...] = ()
176# ── Conversations (AgentMail) ────────────────────────────────────────────────
177class Conversation(Frozen):
178 """A recent AgentMail thread reduced to a one-line summary.
180 ``ref`` carries the transaction/rule id recovered from the agent's own
181 ``yatxn-``/``yaoffer-`` label, so the queue can borrow a thread's subject as
182 the most human description of the item it is waiting on.
183 """
185 subject: str
186 preview: str
187 kind: str # "proposal" | "offer" | "thread"
188 updated_at: datetime | None = None
189 ref: str | None = None
192# ── Inbound (W3 dispatch) ────────────────────────────────────────────────────
193class DispatchTally(Frozen):
194 """Recent inbound dispatch results, counted by routing action."""
196 transaction: int = 0
197 offer: int = 0
198 receipt: int = 0
199 command: int = 0
200 quarantine: int = 0
201 ignore: int = 0
202 total: int = 0
205# ── Run telemetry (ClickHouse) ───────────────────────────────────────────────
206class ActivityStat(Frozen):
207 """Per-activity run count and latency (from the trace spans)."""
209 name: str
210 count: int
211 avg_ms: float
212 max_ms: float
215class RunTelemetry(Frozen):
216 """Trace-derived health for the worker (best-effort enrichment)."""
218 available: bool = False
219 total_spans: int = 0
220 error_spans: int = 0
221 last_activity: datetime | None = None
222 window_days: int = 3
223 activities: tuple[ActivityStat, ...] = ()
224 recent_errors: tuple[str, ...] = ()
227# ── Failures ─────────────────────────────────────────────────────────────────
228class Failure(Frozen):
229 """A workflow that ended terminated/failed, with its recovered reason.
231 ``intentional`` marks the operator's own go-live / re-test / reset
232 terminations, so the page can keep them out of the fault headline (they are
233 housekeeping, not breakage) while still listing them under disclosure.
234 """
236 workflow_id: str
237 kind: str
238 reason: str | None = None
239 when: datetime | None = None
240 intentional: bool = False
243# ── Deploy (GitHub) ──────────────────────────────────────────────────────────
244class PrRow(Frozen):
245 """A recent pull request reduced to title + state + CI."""
247 number: int
248 title: str
249 state: str
250 ci: str | None
251 url: str
252 when: datetime | None = None
255class Deploy(Frozen):
256 """Recent repo activity: PRs and their CI."""
258 prs: tuple[PrRow, ...] = ()
261# ── The whole page ───────────────────────────────────────────────────────────
262class DashboardModel(Frozen):
263 """Everything one dashboard request renders.
265 ``narrative`` and ``health`` lead (the at-a-glance answer); ``needs_you`` is
266 the only acted-upon queue, ``handled`` the already-settled proposals still
267 winding down. The remaining fields back the supporting zones and the
268 progressive-disclosure rails.
269 """
271 generated_at: datetime
272 repo: str
273 narrative: Narrative
274 health: Health
275 sources: tuple[SourceHealth, ...]
276 needs_you: tuple[QueueItem, ...]
277 handled: tuple[QueueItem, ...]
278 lifecycle: Lifecycle
279 autonomy: Autonomy
280 budget: Budget
281 conversations: tuple[Conversation, ...]
282 dispatch: DispatchTally
283 telemetry: RunTelemetry
284 failures: tuple[Failure, ...]
285 deploy: Deploy
DashboardModel is reshaped to match: the old heartbeat/queue give way to narrative, health, and the two-way needs_you / handled split.
The whole-page model: narrative + health lead; the queue is split.
src/ynab_agent/dashboard/model.py · 285 lines
src/ynab_agent/dashboard/model.py
⋯ 261 lines hidden (lines 1–261)
1"""The dashboard view-model — frozen values rendered to one HTML page.
3A pure projection assembled by :mod:`ynab_agent.dashboard.read_model` from the
4source readers and rendered by :mod:`ynab_agent.dashboard.render`. Nothing here
5performs I/O; every field is a plain value so the page is fully testable without
6a cluster.
8The shape is organized around the *questions an operator actually asks* — "is it
9healthy?", "do I need to do anything?", "what has it done?" — not around the
10subsystems the data came from. :class:`Narrative` and :class:`Health` are the
11headline projections that answer the first two at a glance.
12"""
14from __future__ import annotations
16from datetime import datetime
18from ynab_agent.domain.base import Frozen
21class SourceHealth(Frozen):
22 """One data source's liveness dot (green when ``ok``, else red)."""
24 name: str
25 ok: bool
26 detail: str
29# ── State of things (the narrative) ──────────────────────────────────────────
30class Narrative(Frozen):
31 """The plain-English 'state of things' — deterministic, render-ready.
33 A composed-from-the-numbers summary so a human comprehends the whole board
34 without decoding panels. ``tone`` drives its accent; ``paragraphs`` is one
35 or two friendly sentences. Built deterministically in
36 :func:`~ynab_agent.dashboard.read_model.narrate`; an optional LLM polish may
37 later rephrase the prose, always falling back to this text.
38 """
40 headline: str
41 paragraphs: tuple[str, ...] = ()
42 tone: str = "ok" # "ok" | "warn" | "bad"
45# ── Is it healthy? (the masthead rollup) ─────────────────────────────────────
46class Health(Frozen):
47 """The single composite health verdict shown in the masthead.
49 ``tone``/``label`` are the one status dot + word; the rest are the few
50 figures the masthead surfaces beside it. ``needs_you`` is *operation*, not
51 *fault* — a healthy agent can still have items waiting on the owner.
52 """
54 tone: str = "ok" # "ok" | "warn" | "bad"
55 label: str = "healthy"
56 poll_live: bool = False
57 poll_status: str = "none" # the W1 poll's latest execution status
58 poll_last_start: datetime | None = None
59 worker_last_span: datetime | None = None
60 span_error_rate: float | None = None
61 needs_you: int = 0
62 real_failures: int = 0
65# ── Transaction lifecycle ────────────────────────────────────────────────────
66class StateCount(Frozen):
67 """A lifecycle state and how many in-flight transactions are in it."""
69 state: str
70 count: int
73class Lifecycle(Frozen):
74 """The in-flight W2 funnel plus recent terminal counts."""
76 states: tuple[StateCount, ...] = ()
77 in_flight: int = 0
78 archived: int = 0
79 terminated: int = 0
82# ── Autonomy ladder (the rule registry) ──────────────────────────────────────
83class RuleRow(Frozen):
84 """One learned/blessed rule, reduced to what the ladder shows."""
86 payee: str
87 category: str
88 trust: str
89 source: str
90 hits: int
91 offered: bool
92 last_confirmed_at: datetime | None = None
95class OfferRow(Frozen):
96 """One live autonomy-offer workflow (awaiting the owner's yes/no)."""
98 rule_id: str
99 payee: str
100 status: str
101 started_at: datetime | None = None
104class Autonomy(Frozen):
105 """The earned-autonomy picture: counts, the rule table, live offers."""
107 observe: int = 0
108 eligible: int = 0
109 blessed: int = 0
110 rules: tuple[RuleRow, ...] = ()
111 offers: tuple[OfferRow, ...] = ()
114# ── The human's queue ────────────────────────────────────────────────────────
115class TxnFacts(Frozen):
116 """YNAB facts resolved for a queued transaction (the humanizing join).
118 Keyed by ``ynab_id`` outside this type; produced by the YNAB source so the
119 read-model can purely turn a bare workflow id into a readable row.
120 ``approved`` is the pivotal signal: an awaiting proposal whose transaction
121 is already ``approved`` in YNAB is one the owner settled *in-app* — it will
122 lapse on its own and does not need them.
123 """
125 payee: str
126 amount: str
127 approved: bool
128 category: str | None = None
131class QueueItem(Frozen):
132 """Something waiting on the owner: an open proposal or a pending offer.
134 ``ident`` is the stable key (the YNAB txn id for a proposal, the rule id for
135 an offer); the optional humanizing fields are filled by the join when the
136 facts are reachable, and the renderer degrades to ``label`` (a short id)
137 when they are not.
138 """
140 kind: str # "proposal" | "offer"
141 label: str
142 ident: str
143 since: datetime | None = None
144 payee: str | None = None
145 amount: str | None = None
146 category: str | None = None
147 approved: bool | None = None # YNAB approved flag; None = unknown / offer
148 question: str | None = None # the proposal's email subject, when matched
151# ── Budget (YNAB) ────────────────────────────────────────────────────────────
152class CategoryRow(Frozen):
153 """A category's month-to-date balance (negative = overspent)."""
155 name: str
156 balance: str
157 overspent: bool
160class TxnRow(Frozen):
161 """A transaction reduced to a payee + amount display."""
163 payee: str
164 amount: str
167class Budget(Frozen):
168 """The budget surface: the unapproved backlog and overspent categories."""
170 available: bool = False
171 unapproved: int = 0
172 unapproved_sample: tuple[TxnRow, ...] = ()
173 overspent: tuple[CategoryRow, ...] = ()
176# ── Conversations (AgentMail) ────────────────────────────────────────────────
177class Conversation(Frozen):
178 """A recent AgentMail thread reduced to a one-line summary.
180 ``ref`` carries the transaction/rule id recovered from the agent's own
181 ``yatxn-``/``yaoffer-`` label, so the queue can borrow a thread's subject as
182 the most human description of the item it is waiting on.
183 """
185 subject: str
186 preview: str
187 kind: str # "proposal" | "offer" | "thread"
188 updated_at: datetime | None = None
189 ref: str | None = None
192# ── Inbound (W3 dispatch) ────────────────────────────────────────────────────
193class DispatchTally(Frozen):
194 """Recent inbound dispatch results, counted by routing action."""
196 transaction: int = 0
197 offer: int = 0
198 receipt: int = 0
199 command: int = 0
200 quarantine: int = 0
201 ignore: int = 0
202 total: int = 0
205# ── Run telemetry (ClickHouse) ───────────────────────────────────────────────
206class ActivityStat(Frozen):
207 """Per-activity run count and latency (from the trace spans)."""
209 name: str
210 count: int
211 avg_ms: float
212 max_ms: float
215class RunTelemetry(Frozen):
216 """Trace-derived health for the worker (best-effort enrichment)."""
218 available: bool = False
219 total_spans: int = 0
220 error_spans: int = 0
221 last_activity: datetime | None = None
222 window_days: int = 3
223 activities: tuple[ActivityStat, ...] = ()
224 recent_errors: tuple[str, ...] = ()
227# ── Failures ─────────────────────────────────────────────────────────────────
228class Failure(Frozen):
229 """A workflow that ended terminated/failed, with its recovered reason.
231 ``intentional`` marks the operator's own go-live / re-test / reset
232 terminations, so the page can keep them out of the fault headline (they are
233 housekeeping, not breakage) while still listing them under disclosure.
234 """
236 workflow_id: str
237 kind: str
238 reason: str | None = None
239 when: datetime | None = None
240 intentional: bool = False
243# ── Deploy (GitHub) ──────────────────────────────────────────────────────────
244class PrRow(Frozen):
245 """A recent pull request reduced to title + state + CI."""
247 number: int
248 title: str
249 state: str
250 ci: str | None
251 url: str
252 when: datetime | None = None
255class Deploy(Frozen):
256 """Recent repo activity: PRs and their CI."""
258 prs: tuple[PrRow, ...] = ()
261# ── The whole page ───────────────────────────────────────────────────────────
262class DashboardModel(Frozen):
263 """Everything one dashboard request renders.
265 ``narrative`` and ``health`` lead (the at-a-glance answer); ``needs_you`` is
266 the only acted-upon queue, ``handled`` the already-settled proposals still
267 winding down. The remaining fields back the supporting zones and the
268 progressive-disclosure rails.
269 """
271 generated_at: datetime
272 repo: str
273 narrative: Narrative
274 health: Health
275 sources: tuple[SourceHealth, ...]
276 needs_you: tuple[QueueItem, ...]
277 handled: tuple[QueueItem, ...]
278 lifecycle: Lifecycle
279 autonomy: Autonomy
280 budget: Budget
281 conversations: tuple[Conversation, ...]
282 dispatch: DispatchTally
The humanizing join — pure, and conservative
_humanize_queue turns bare awaiting ids into readable rows and performs the needs-you / handled split. It is a pure function: the I/O layer hands it a facts map and a conversations-by-ref map, and it joins by dict lookup. The label degrades in a clear order — the proposal's email subject, then payee · amount, then a short id — so a row is always readable even when YNAB or AgentMail is unreachable.
The pure join: humanize each awaiting id, then split on approved is True.
src/ynab_agent/dashboard/read_model.py · 397 lines
src/ynab_agent/dashboard/read_model.py
⋯ 86 lines hidden (lines 1–86)
1"""Assemble the source readouts into one view-model (pure).
3Each reader hands in ``(data, error)``; this composes the
4:class:`~ynab_agent.dashboard.model.DashboardModel`, turning each error into a
5source-health dot and each Temporal/YNAB/AgentMail piece into the operator's
6shape: a composite :class:`~ynab_agent.dashboard.model.Health` verdict, a
7humanized owner queue (split into *needs you* vs *already settled*), and a
8deterministic plain-English :class:`~ynab_agent.dashboard.model.Narrative`. No
9I/O — fully unit-testable from in-memory readouts.
10"""
12from __future__ import annotations
14from typing import TYPE_CHECKING
16from ynab_agent.dashboard.model import (
17 Autonomy,
18 DashboardModel,
19 Health,
20 Lifecycle,
21 Narrative,
22 QueueItem,
23 SourceHealth,
24)
26if TYPE_CHECKING:
27 from datetime import datetime
29 from ynab_agent.dashboard.model import (
30 Budget,
31 Conversation,
32 Deploy,
33 Failure,
34 RunTelemetry,
35 TxnFacts,
36 )
37 from ynab_agent.dashboard.temporal_source import TemporalReadout
39# A worker span older than this reads as a stale heartbeat (warn).
40_STALE_WORKER_SECS = 3600.0
41# A trace error rate above this contributes a warn to the health rollup.
42_ERROR_RATE_WARN = 0.05
43# The activities whose run-count means "a category was written to YNAB".
44_APPLIED_ACTIVITY = "commit_to_ynab"
47def _health(name: str, error: str | None, ok_detail: str) -> SourceHealth:
48 """A source dot: green with a summary, or red with the error/'off'."""
49 if error is None:
50 return SourceHealth(name=name, ok=True, detail=ok_detail)
51 return SourceHealth(name=name, ok=False, detail=error)
54def _s(n: int) -> str:
55 """The plural suffix for a count (1 → '', else 's')."""
56 return "" if n == 1 else "s"
59def _short(ident: str) -> str:
60 """A short, readable stand-in when an id can't be humanized."""
61 return ident[:8] if len(ident) > 8 else ident
64def _age_phrase(seconds: float) -> str:
65 """A coarse, friendly age ('4 days', '3 hours', 'moments')."""
66 if seconds < 90:
67 return "moments"
68 if seconds < 5400:
69 n = int(seconds // 60)
70 return f"{n} minute{_s(n)}"
71 if seconds < 129600:
72 n = int(seconds // 3600)
73 return f"{n} hour{_s(n)}"
74 n = int(seconds // 86400)
75 return f"{n} day{_s(n)}"
78def _oldest_age(items: tuple[QueueItem, ...], now: datetime) -> str | None:
79 """The age of the longest-waiting item, as a phrase (None if empty)."""
80 # Normalize each to aware BEFORE min() — comparing aware vs naive raises.
81 sinces = [_aware(q.since, now) for q in items if q.since is not None]
82 if not sinces:
83 return None
84 return _age_phrase((now - min(sinces)).total_seconds())
87def _humanize_queue(
88 readout: TemporalReadout,
89 queue_facts: dict[str, TxnFacts],
90 convo_by_ref: dict[str, Conversation],
91) -> tuple[tuple[QueueItem, ...], tuple[QueueItem, ...]]:
92 """Turn bare awaiting ids into readable rows; split needs-you vs settled.
94 A proposal whose transaction is already ``approved`` in YNAB is one the
95 owner settled in-app — it lands in *handled* (it lapses on its own); every
96 other proposal, and every live autonomy offer, lands in *needs you*. The
97 best label is the proposal's email subject, then ``payee · amount``, then a
98 short id.
99 """
100 needs_you: list[QueueItem] = []
101 handled: list[QueueItem] = []
103 for item in readout.awaiting:
104 facts = queue_facts.get(item.ident)
105 convo = convo_by_ref.get(item.ident)
106 question = convo.subject if convo is not None else None
107 payee = facts.payee if facts is not None else None
108 amount = facts.amount if facts is not None else None
109 approved = facts.approved if facts is not None else None
110 if question:
111 label = question
112 elif payee:
113 label = f"{payee} {amount}".strip()
114 else:
115 label = _short(item.ident)
116 row = QueueItem(
117 kind="proposal",
118 label=label,
119 ident=item.ident,
120 since=item.since,
121 payee=payee,
122 amount=amount,
123 category=facts.category if facts is not None else None,
124 approved=approved,
125 question=question,
126 )
127 (handled if approved is True else needs_you).append(row)
129 for offer in readout.offers:
130 convo = convo_by_ref.get(offer.rule_id)
131 needs_you.append(
132 QueueItem(
133 kind="offer",
134 label=(convo.subject if convo else None)
135 or offer.payee
136 or offer.rule_id,
137 ident=offer.rule_id,
138 since=offer.started_at,
139 payee=offer.payee or None,
140 question=convo.subject if convo else None,
141 )
142 )
144 return tuple(needs_you), tuple(handled)
⋯ 252 lines hidden (lines 146–397)
147def _rollup(
148 *,
149 now: datetime,
150 poll_live: bool,
151 poll_status: str,
152 poll_last_start: datetime | None,
153 temporal_error: str | None,
154 ynab_error: str | None,
155 telemetry: RunTelemetry,
156 failures: tuple[Failure, ...],
157 needs_you: int,
158) -> Health:
159 """Reduce the readouts to the single masthead health verdict."""
160 worker_last = telemetry.last_activity
161 rate = (
162 telemetry.error_spans / telemetry.total_spans
163 if telemetry.available and telemetry.total_spans > 0
164 else None
165 )
166 real_failures = sum(1 for f in failures if not f.intentional)
167 worker_stale = (
168 worker_last is not None
169 and (now - _aware(worker_last, now)).total_seconds()
170 > _STALE_WORKER_SECS
171 )
173 # Tone is *current* operational health — can it do its job right now. An
174 # isolated historical failure is surfaced (chip + narrative) but doesn't
175 # flip the verdict; a stuck worker, an unreachable money source, or a high
176 # live error rate does.
177 if not poll_live or temporal_error is not None:
178 tone, label = "bad", "down"
179 elif (
180 worker_stale
181 or ynab_error is not None
182 or (rate is not None and rate > _ERROR_RATE_WARN)
183 ):
184 tone, label = "warn", "degraded"
185 else:
186 tone, label = "ok", "healthy"
188 return Health(
189 tone=tone,
190 label=label,
191 poll_live=poll_live,
192 poll_status=poll_status,
193 poll_last_start=poll_last_start,
194 worker_last_span=worker_last,
195 span_error_rate=rate,
196 needs_you=needs_you,
197 real_failures=real_failures,
198 )
201def _aware(when: datetime, now: datetime) -> datetime:
202 return when if when.tzinfo is not None else when.replace(tzinfo=now.tzinfo)
205def _applied_count(telemetry: RunTelemetry) -> int:
206 """Recent writes to YNAB, from the commit activity's run count."""
207 for activity in telemetry.activities:
208 if activity.name == _APPLIED_ACTIVITY:
209 return activity.count
210 return 0
213def narrate(
214 *,
215 now: datetime,
216 health: Health,
217 lifecycle: Lifecycle,
218 needs_you: tuple[QueueItem, ...],
219 handled: tuple[QueueItem, ...],
220 budget: Budget,
221 autonomy: Autonomy,
222 telemetry: RunTelemetry,
223) -> Narrative:
224 """Compose the deterministic 'state of things' summary from the numbers.
226 Every clause is guarded, so the paragraph reads naturally whether or not a
227 given source is on. This is the trustworthy baseline; an optional LLM polish
228 may later rephrase it for warmth, always falling back to exactly this text.
229 """
230 states = {s.state: s.count for s in lifecycle.states}
231 proposals = tuple(q for q in needs_you if q.kind == "proposal")
232 offers = tuple(q for q in needs_you if q.kind == "offer")
233 p, h, offers_n = len(proposals), len(handled), len(offers)
234 real_failures = health.real_failures
236 # Headline — the single most important thing.
237 if health.tone == "bad":
238 headline = "Attention needed."
239 elif p or offers_n:
240 waiting = p + offers_n
241 headline = f"{waiting} thing{_s(waiting)} waiting on you."
242 else:
243 headline = "All caught up — nothing needs you."
245 sentences: list[str] = []
247 if budget.available:
248 u = budget.unapproved
249 if u == 0:
250 sentences.append("You're caught up in YNAB.")
251 elif u <= 5:
252 sentences.append(
253 f"You're nearly caught up in YNAB ({u} unapproved)."
254 )
255 else:
256 sentences.append(f"{u} transactions await approval in YNAB.")
258 in_flight = lifecycle.in_flight
259 if in_flight:
260 open_n = states.get("open", 0)
261 awaiting_total = p + h
262 other = max(0, in_flight - open_n - awaiting_total)
263 parts = [f"{open_n} categorized and resting"] if open_n else []
264 if h:
265 parts.append(f"{h} you already settled in YNAB")
266 if p:
267 parts.append(f"{p} waiting on your reply")
268 if other:
269 parts.append(f"{other} still being processed")
270 body = ", ".join(parts) if parts else "all winding down"
271 sentences.append(
272 f"The agent is tracking {in_flight} "
273 f"transaction{_s(in_flight)}: {body}."
274 )
275 else:
276 sentences.append("No transactions are in flight.")
278 applied = _applied_count(telemetry)
279 if telemetry.available and applied:
280 obs = autonomy.observe
281 tail = f" and is learning ({obs} rule{_s(obs)} observed)" if obs else ""
282 sentences.append(
283 f"It has applied {applied} categor{'y' if applied == 1 else 'ies'} "
284 f"in the last {telemetry.window_days} days{tail}."
285 )
287 closing: list[str] = []
288 if offers_n:
289 verb = "s" if offers_n == 1 else ""
290 closing.append(
291 f"{offers_n} autonomy offer{_s(offers_n)} await{verb} your yes/no"
292 )
293 oldest = _oldest_age(proposals, now)
294 if p and oldest:
295 closing.append(f"the oldest reply has been waiting {oldest}")
296 if h:
297 closing.append("the settled ones lapse on their own")
298 if real_failures:
299 closing.append(
300 f"{real_failures} workflow{_s(real_failures)} failed "
301 "recently — check Failures"
302 )
303 if not p and not offers_n and not closing:
304 closing.append("nothing needs you right now")
305 if closing:
306 joined = closing[0] + (
307 "" if len(closing) == 1 else "; " + "; ".join(closing[1:])
308 )
309 sentences.append(joined[0].upper() + joined[1:] + ".")
311 return Narrative(
312 headline=headline, paragraphs=(" ".join(sentences),), tone=health.tone
313 )
316def assemble(
317 *,
318 now: datetime,
319 repo: str,
320 temporal: tuple[TemporalReadout, str | None],
321 ynab: tuple[Budget, str | None],
322 clickhouse: tuple[RunTelemetry, str | None],
323 agentmail: tuple[tuple[Conversation, ...], str | None],
324 github: tuple[Deploy, str | None],
325 queue_facts: dict[str, TxnFacts] | None = None,
326) -> DashboardModel:
327 """Compose the whole dashboard view from the source readouts."""
328 t, t_err = temporal
329 budget, ynab_err = ynab
330 telemetry, ch_err = clickhouse
331 conversations, mail_err = agentmail
332 deploy, gh_err = github
333 facts = queue_facts or {}
335 sources = (
336 _health("temporal", t_err, f"{t.in_flight} in-flight"),
337 _health("ynab", ynab_err, f"{budget.unapproved} unapproved"),
338 _health("clickhouse", ch_err, f"{telemetry.total_spans} spans"),
339 _health("agentmail", mail_err, f"{len(conversations)} threads"),
340 _health("github", gh_err, f"{len(deploy.prs)} PRs"),
341 )
343 convo_by_ref = {c.ref: c for c in conversations if c.ref}
344 needs_you, handled = _humanize_queue(t, facts, convo_by_ref)
346 lifecycle = Lifecycle(
347 states=t.lifecycle_states,
348 in_flight=t.in_flight,
349 archived=t.archived,
350 terminated=t.terminated,
351 )
352 autonomy = Autonomy(
353 observe=t.observe,
354 eligible=t.eligible,
355 blessed=t.blessed,
356 rules=t.rules,
357 offers=t.offers,
358 )
359 health = _rollup(
360 now=now,
361 poll_live=t.poll_live,
362 poll_status=t.poll_status,
363 poll_last_start=t.poll_last_start,
364 temporal_error=t_err,
365 ynab_error=ynab_err,
366 telemetry=telemetry,
367 failures=t.failures,
368 needs_you=len(needs_you),
369 )
370 narrative = narrate(
371 now=now,
372 health=health,
373 lifecycle=lifecycle,
374 needs_you=needs_you,
375 handled=handled,
376 budget=budget,
377 autonomy=autonomy,
378 telemetry=telemetry,
379 )
381 return DashboardModel(
382 generated_at=now,
383 repo=repo,
384 narrative=narrative,
385 health=health,
386 sources=sources,
387 needs_you=needs_you,
388 handled=handled,
389 lifecycle=lifecycle,
390 autonomy=autonomy,
391 budget=budget,
392 conversations=conversations,
393 dispatch=t.dispatch,
394 telemetry=telemetry,
395 failures=t.failures,
396 deploy=deploy,
397 )
resolve_queue — bounded, read-only, timeout-guarded
This is the only new I/O, and the section a reviewer should scrutinize. It runs in the live worker, so it must be cheap, safe, and unable to hang or crash the page.
It resolves the still-unapproved ids (the genuine to-dos) from one unapproved index — no per-id list scan — and the rest, which are the proposals you already approved, with a cheap single GET each. Category names come from category_spends. Every call is a read; nothing here writes to YNAB.
The cap + timeout constants, the facts reducer, and the read-only resolve.
src/ynab_agent/dashboard/ynab_source.py · 130 lines
src/ynab_agent/dashboard/ynab_source.py
⋯ 23 lines hidden (lines 1–23)
1"""YNAB reader: the budget surface the agent acts on (best-effort).
3Reuses the production :class:`~ynab_agent.ynab.client.YnabClient` (its tested
4wire-parsing and budget-id resolution) off the event loop via
5``asyncio.to_thread`` — no parallel REST surface, and the client is never
6mutated. Surfaces the ``type=unapproved`` backlog (the W1 work queue) and the
7overspent categories. Degrades to "off" when ``YNAB_API_KEY`` is unset.
8"""
10from __future__ import annotations
12import asyncio
13import re
14from typing import TYPE_CHECKING
16from ynab_agent.dashboard.model import Budget, CategoryRow, TxnFacts, TxnRow
18if TYPE_CHECKING:
19 from ynab_agent.domain.transaction import YnabSnapshot
21_MAX_SAMPLE = 8
22_MAX_OVERSPENT = 8
24# Cap on how many awaiting ids one render will resolve against YNAB (the queue
25# is already bounded upstream; this is a runaway backstop).
26_MAX_RESOLVE = 40
28# Wall-clock bound on the resolve so a slow YNAB never hangs a page render (the
29# worker thread may finish in the background; the page proceeds with bare ids).
30_RESOLVE_TIMEOUT = 12.0
32# A W2 id is the bare YNAB txn id, but a few carry a ``_YYYY-MM-DD`` suffix
33# (scheduled/split origins); strip it for the YNAB lookup, keep the raw id key.
34_DATE_SUFFIX = re.compile(r"_\d{4}-\d{2}-\d{2}$")
⋯ 41 lines hidden (lines 35–75)
37def _read() -> Budget:
38 """Read the budget surface synchronously (run in a worker thread)."""
39 from ynab_agent.ynab.client import YnabClient
41 client = YnabClient.from_env()
42 unapproved = client.unapproved()
43 spends = client.category_spends()
45 sample = tuple(
46 TxnRow(payee=snap.payee, amount=str(snap.amount))
47 for snap in unapproved[:_MAX_SAMPLE]
48 )
49 overspent_spends = sorted(
50 (s for s in spends if s.balance.milliunits < 0),
51 key=lambda s: s.balance.milliunits,
52 )
53 overspent = tuple(
54 CategoryRow(name=s.name, balance=str(s.balance), overspent=True)
55 for s in overspent_spends[:_MAX_OVERSPENT]
56 )
57 return Budget(
58 available=True,
59 unapproved=len(unapproved),
60 unapproved_sample=sample,
61 overspent=overspent,
62 )
65async def fetch() -> tuple[Budget, str | None]:
66 """Read the budget surface; ``error`` is "off" when YNAB is unconfigured."""
67 try:
68 budget = await asyncio.to_thread(_read)
69 except RuntimeError:
70 return Budget(available=False), "off"
71 except Exception as exc: # any YNAB hiccup degrades to a red dot
72 return Budget(available=False), f"{type(exc).__name__}: {exc}"
73 return budget, None
76def _facts(snap: YnabSnapshot, categories: dict[str, str]) -> TxnFacts:
77 """Reduce a snapshot to the humanizing facts the queue shows."""
78 category = (
79 categories.get(str(snap.category_id))
80 if snap.category_id is not None
81 else None
82 )
83 return TxnFacts(
84 payee=snap.payee or "(no payee)",
85 amount=str(snap.amount),
86 approved=snap.approved,
87 category=category,
88 )
91def _resolve(ids: tuple[str, ...]) -> dict[str, TxnFacts]:
92 """Resolve queued workflow ids to YNAB facts (run in a worker thread).
94 The still-*unapproved* ids (the ones genuinely waiting on the owner) come
95 from one ``unapproved`` index — no per-id list scan. The rest are the
96 proposals the owner already approved in-app: a cheap single GET each
97 (``snapshot`` returns those without a scan). Keyed by the raw id so the
98 read-model can join on the workflow id verbatim.
99 """
100 from ynab_agent.ynab.client import YnabClient
102 client = YnabClient.from_env()
103 categories = {str(s.category): s.name for s in client.category_spends()}
104 unapproved = {str(s.ynab_id): s for s in client.unapproved()}
106 out: dict[str, TxnFacts] = {}
107 for raw in ids:
108 bare = _DATE_SUFFIX.sub("", raw)
109 snap = unapproved.get(bare) or client.snapshot(bare)
110 if snap is not None:
111 out[raw] = _facts(snap, categories)
112 return out
115async def resolve_queue(ids: tuple[str, ...]) -> dict[str, TxnFacts]:
116 """Humanize the awaiting-queue ids; ``{}`` whenever YNAB is unreachable.
118 Best-effort by design: any failure (or YNAB being off) yields an empty map
119 and the renderer degrades each row to its short id — the queue never fails
120 the page.
121 """
122 unique = tuple(dict.fromkeys(i for i in ids if i))[:_MAX_RESOLVE]
123 if not unique:
124 return {}
125 try:
126 return await asyncio.wait_for(
127 asyncio.to_thread(_resolve, unique), _RESOLVE_TIMEOUT
128 )
129 except Exception: # timeout / YNAB off / any hiccup → bare-id fallback
130 return {}
The most human label — the actual question the agent emailed — is recovered without any extra call. The agent stamps each thread with a yatxn-{id} / yaoffer-{id} label; _ref pulls that id back out so the read-model can borrow the thread's subject for the queue row.
The id the queue joins on rides in the agent's own idempotency label.
src/ynab_agent/dashboard/agentmail_source.py · 90 lines
src/ynab_agent/dashboard/agentmail_source.py
⋯ 29 lines hidden (lines 1–29)
1"""AgentMail reader: the recent conversation surface (best-effort).
3Lists the inbox's recent threads via the AgentMail SDK, run off the event loop
4in a worker thread, and reduces each to a one-line summary tagged by the agent's
5own labels (a ``yaoffer-`` thread is an autonomy offer, a ``yatxn-`` thread a
6transaction proposal, else a plain thread). Read-only; degrades to "off" when
7``AGENTMAIL_API_KEY`` is unset and to a red dot on any API hiccup.
8"""
10from __future__ import annotations
12import asyncio
13import os
14from typing import Any
16from ynab_agent.dashboard.model import Conversation
18_MAX_THREADS = 12
21def _kind(labels: tuple[str, ...]) -> str:
22 """Classify a thread from the agent's own idempotency labels."""
23 if any(label.startswith("yaoffer-") for label in labels):
24 return "offer"
25 if any(label.startswith("yatxn-") for label in labels):
26 return "proposal"
27 return "thread"
30def _ref(labels: tuple[str, ...]) -> str | None:
31 """Recover the txn/rule id from the agent's idempotency label.
33 The W2 stamps ``yatxn-{ynab_id}`` and an offer ``yaoffer-{rule_id}`` — the
34 same ids the queue is keyed by — so the queue can borrow this thread's
35 subject as the most human description of what it waits on.
36 """
37 for label in labels:
38 for prefix in ("yatxn-", "yaoffer-"):
39 if label.startswith(prefix):
40 return label[len(prefix) :]
41 return None
⋯ 49 lines hidden (lines 42–90)
44def _timestamp(item: Any) -> object:
45 """The most recent timestamp on a thread item (SDK-shape tolerant)."""
46 for attr in (
47 "updated_at",
48 "timestamp",
49 "sent_timestamp",
50 "received_timestamp",
51 ):
52 value = getattr(item, attr, None)
53 if value is not None:
54 return value
55 return None
58def _read(inbox: str, key: str) -> tuple[Conversation, ...]:
59 """List recent threads synchronously (run in a worker thread)."""
60 from agentmail import AgentMail
62 client = AgentMail(api_key=key)
63 result = client.inboxes.threads.list(inbox, limit=_MAX_THREADS)
64 threads = getattr(result, "threads", None) or []
65 conversations: list[Conversation] = []
66 for item in threads:
67 labels = tuple(str(x) for x in (getattr(item, "labels", None) or ()))
68 conversations.append(
69 Conversation(
70 subject=str(getattr(item, "subject", "") or "(no subject)"),
71 preview=str(getattr(item, "preview", "") or "")[:160],
72 kind=_kind(labels),
73 updated_at=_timestamp(item), # type: ignore[arg-type]
74 ref=_ref(labels),
75 )
76 )
77 return tuple(conversations)
80async def fetch() -> tuple[tuple[Conversation, ...], str | None]:
81 """List recent conversations; ``error`` is "off" when unconfigured."""
82 key = os.environ.get("AGENTMAIL_API_KEY")
83 inbox = os.environ.get("YNAB_AGENT_INBOX")
84 if not key or not inbox:
85 return (), "off"
86 try:
87 conversations = await asyncio.to_thread(_read, inbox, key)
88 except Exception as exc: # any AgentMail hiccup degrades to a red dot
89 return (), f"{type(exc).__name__}: {exc}"
90 return conversations, None
The narrator and the health verdict — deterministic
narrate composes the summary paragraph from the numbers. It is deterministic and fully guarded: every clause is conditional, so the prose reads naturally whether or not a given source is on. The headline keys off the health tone first, then the count of things waiting on you. (An optional LLM polish may later rephrase it, but always falls back to exactly this text.)
Guarded clause assembly: headline by tone, then the YNAB caught-up line.
src/ynab_agent/dashboard/read_model.py · 397 lines
src/ynab_agent/dashboard/read_model.py
⋯ 212 lines hidden (lines 1–212)
1"""Assemble the source readouts into one view-model (pure).
3Each reader hands in ``(data, error)``; this composes the
4:class:`~ynab_agent.dashboard.model.DashboardModel`, turning each error into a
5source-health dot and each Temporal/YNAB/AgentMail piece into the operator's
6shape: a composite :class:`~ynab_agent.dashboard.model.Health` verdict, a
7humanized owner queue (split into *needs you* vs *already settled*), and a
8deterministic plain-English :class:`~ynab_agent.dashboard.model.Narrative`. No
9I/O — fully unit-testable from in-memory readouts.
10"""
12from __future__ import annotations
14from typing import TYPE_CHECKING
16from ynab_agent.dashboard.model import (
17 Autonomy,
18 DashboardModel,
19 Health,
20 Lifecycle,
21 Narrative,
22 QueueItem,
23 SourceHealth,
24)
26if TYPE_CHECKING:
27 from datetime import datetime
29 from ynab_agent.dashboard.model import (
30 Budget,
31 Conversation,
32 Deploy,
33 Failure,
34 RunTelemetry,
35 TxnFacts,
36 )
37 from ynab_agent.dashboard.temporal_source import TemporalReadout
39# A worker span older than this reads as a stale heartbeat (warn).
40_STALE_WORKER_SECS = 3600.0
41# A trace error rate above this contributes a warn to the health rollup.
42_ERROR_RATE_WARN = 0.05
43# The activities whose run-count means "a category was written to YNAB".
44_APPLIED_ACTIVITY = "commit_to_ynab"
47def _health(name: str, error: str | None, ok_detail: str) -> SourceHealth:
48 """A source dot: green with a summary, or red with the error/'off'."""
49 if error is None:
50 return SourceHealth(name=name, ok=True, detail=ok_detail)
51 return SourceHealth(name=name, ok=False, detail=error)
54def _s(n: int) -> str:
55 """The plural suffix for a count (1 → '', else 's')."""
56 return "" if n == 1 else "s"
59def _short(ident: str) -> str:
60 """A short, readable stand-in when an id can't be humanized."""
61 return ident[:8] if len(ident) > 8 else ident
64def _age_phrase(seconds: float) -> str:
65 """A coarse, friendly age ('4 days', '3 hours', 'moments')."""
66 if seconds < 90:
67 return "moments"
68 if seconds < 5400:
69 n = int(seconds // 60)
70 return f"{n} minute{_s(n)}"
71 if seconds < 129600:
72 n = int(seconds // 3600)
73 return f"{n} hour{_s(n)}"
74 n = int(seconds // 86400)
75 return f"{n} day{_s(n)}"
78def _oldest_age(items: tuple[QueueItem, ...], now: datetime) -> str | None:
79 """The age of the longest-waiting item, as a phrase (None if empty)."""
80 # Normalize each to aware BEFORE min() — comparing aware vs naive raises.
81 sinces = [_aware(q.since, now) for q in items if q.since is not None]
82 if not sinces:
83 return None
84 return _age_phrase((now - min(sinces)).total_seconds())
87def _humanize_queue(
88 readout: TemporalReadout,
89 queue_facts: dict[str, TxnFacts],
90 convo_by_ref: dict[str, Conversation],
91) -> tuple[tuple[QueueItem, ...], tuple[QueueItem, ...]]:
92 """Turn bare awaiting ids into readable rows; split needs-you vs settled.
94 A proposal whose transaction is already ``approved`` in YNAB is one the
95 owner settled in-app — it lands in *handled* (it lapses on its own); every
96 other proposal, and every live autonomy offer, lands in *needs you*. The
97 best label is the proposal's email subject, then ``payee · amount``, then a
98 short id.
99 """
100 needs_you: list[QueueItem] = []
101 handled: list[QueueItem] = []
103 for item in readout.awaiting:
104 facts = queue_facts.get(item.ident)
105 convo = convo_by_ref.get(item.ident)
106 question = convo.subject if convo is not None else None
107 payee = facts.payee if facts is not None else None
108 amount = facts.amount if facts is not None else None
109 approved = facts.approved if facts is not None else None
110 if question:
111 label = question
112 elif payee:
113 label = f"{payee} {amount}".strip()
114 else:
115 label = _short(item.ident)
116 row = QueueItem(
117 kind="proposal",
118 label=label,
119 ident=item.ident,
120 since=item.since,
121 payee=payee,
122 amount=amount,
123 category=facts.category if facts is not None else None,
124 approved=approved,
125 question=question,
126 )
127 (handled if approved is True else needs_you).append(row)
129 for offer in readout.offers:
130 convo = convo_by_ref.get(offer.rule_id)
131 needs_you.append(
132 QueueItem(
133 kind="offer",
134 label=(convo.subject if convo else None)
135 or offer.payee
136 or offer.rule_id,
137 ident=offer.rule_id,
138 since=offer.started_at,
139 payee=offer.payee or None,
140 question=convo.subject if convo else None,
141 )
142 )
144 return tuple(needs_you), tuple(handled)
147def _rollup(
148 *,
149 now: datetime,
150 poll_live: bool,
151 poll_status: str,
152 poll_last_start: datetime | None,
153 temporal_error: str | None,
154 ynab_error: str | None,
155 telemetry: RunTelemetry,
156 failures: tuple[Failure, ...],
157 needs_you: int,
158) -> Health:
159 """Reduce the readouts to the single masthead health verdict."""
160 worker_last = telemetry.last_activity
161 rate = (
162 telemetry.error_spans / telemetry.total_spans
163 if telemetry.available and telemetry.total_spans > 0
164 else None
165 )
166 real_failures = sum(1 for f in failures if not f.intentional)
167 worker_stale = (
168 worker_last is not None
169 and (now - _aware(worker_last, now)).total_seconds()
170 > _STALE_WORKER_SECS
171 )
173 # Tone is *current* operational health — can it do its job right now. An
174 # isolated historical failure is surfaced (chip + narrative) but doesn't
175 # flip the verdict; a stuck worker, an unreachable money source, or a high
176 # live error rate does.
177 if not poll_live or temporal_error is not None:
178 tone, label = "bad", "down"
179 elif (
180 worker_stale
181 or ynab_error is not None
182 or (rate is not None and rate > _ERROR_RATE_WARN)
183 ):
184 tone, label = "warn", "degraded"
185 else:
186 tone, label = "ok", "healthy"
188 return Health(
189 tone=tone,
190 label=label,
191 poll_live=poll_live,
192 poll_status=poll_status,
193 poll_last_start=poll_last_start,
194 worker_last_span=worker_last,
195 span_error_rate=rate,
196 needs_you=needs_you,
197 real_failures=real_failures,
198 )
201def _aware(when: datetime, now: datetime) -> datetime:
202 return when if when.tzinfo is not None else when.replace(tzinfo=now.tzinfo)
205def _applied_count(telemetry: RunTelemetry) -> int:
206 """Recent writes to YNAB, from the commit activity's run count."""
207 for activity in telemetry.activities:
208 if activity.name == _APPLIED_ACTIVITY:
209 return activity.count
210 return 0
213def narrate(
214 *,
215 now: datetime,
216 health: Health,
217 lifecycle: Lifecycle,
218 needs_you: tuple[QueueItem, ...],
219 handled: tuple[QueueItem, ...],
220 budget: Budget,
221 autonomy: Autonomy,
222 telemetry: RunTelemetry,
223) -> Narrative:
224 """Compose the deterministic 'state of things' summary from the numbers.
226 Every clause is guarded, so the paragraph reads naturally whether or not a
227 given source is on. This is the trustworthy baseline; an optional LLM polish
228 may later rephrase it for warmth, always falling back to exactly this text.
229 """
230 states = {s.state: s.count for s in lifecycle.states}
231 proposals = tuple(q for q in needs_you if q.kind == "proposal")
232 offers = tuple(q for q in needs_you if q.kind == "offer")
233 p, h, offers_n = len(proposals), len(handled), len(offers)
234 real_failures = health.real_failures
236 # Headline — the single most important thing.
237 if health.tone == "bad":
238 headline = "Attention needed."
239 elif p or offers_n:
240 waiting = p + offers_n
241 headline = f"{waiting} thing{_s(waiting)} waiting on you."
242 else:
243 headline = "All caught up — nothing needs you."
245 sentences: list[str] = []
247 if budget.available:
248 u = budget.unapproved
249 if u == 0:
250 sentences.append("You're caught up in YNAB.")
251 elif u <= 5:
252 sentences.append(
253 f"You're nearly caught up in YNAB ({u} unapproved)."
254 )
255 else:
256 sentences.append(f"{u} transactions await approval in YNAB.")
⋯ 141 lines hidden (lines 257–397)
258 in_flight = lifecycle.in_flight
259 if in_flight:
260 open_n = states.get("open", 0)
261 awaiting_total = p + h
262 other = max(0, in_flight - open_n - awaiting_total)
263 parts = [f"{open_n} categorized and resting"] if open_n else []
264 if h:
265 parts.append(f"{h} you already settled in YNAB")
266 if p:
267 parts.append(f"{p} waiting on your reply")
268 if other:
269 parts.append(f"{other} still being processed")
270 body = ", ".join(parts) if parts else "all winding down"
271 sentences.append(
272 f"The agent is tracking {in_flight} "
273 f"transaction{_s(in_flight)}: {body}."
274 )
275 else:
276 sentences.append("No transactions are in flight.")
278 applied = _applied_count(telemetry)
279 if telemetry.available and applied:
280 obs = autonomy.observe
281 tail = f" and is learning ({obs} rule{_s(obs)} observed)" if obs else ""
282 sentences.append(
283 f"It has applied {applied} categor{'y' if applied == 1 else 'ies'} "
284 f"in the last {telemetry.window_days} days{tail}."
285 )
287 closing: list[str] = []
288 if offers_n:
289 verb = "s" if offers_n == 1 else ""
290 closing.append(
291 f"{offers_n} autonomy offer{_s(offers_n)} await{verb} your yes/no"
292 )
293 oldest = _oldest_age(proposals, now)
294 if p and oldest:
295 closing.append(f"the oldest reply has been waiting {oldest}")
296 if h:
297 closing.append("the settled ones lapse on their own")
298 if real_failures:
299 closing.append(
300 f"{real_failures} workflow{_s(real_failures)} failed "
301 "recently — check Failures"
302 )
303 if not p and not offers_n and not closing:
304 closing.append("nothing needs you right now")
305 if closing:
306 joined = closing[0] + (
307 "" if len(closing) == 1 else "; " + "; ".join(closing[1:])
308 )
309 sentences.append(joined[0].upper() + joined[1:] + ".")
311 return Narrative(
312 headline=headline, paragraphs=(" ".join(sentences),), tone=health.tone
313 )
316def assemble(
317 *,
318 now: datetime,
319 repo: str,
320 temporal: tuple[TemporalReadout, str | None],
321 ynab: tuple[Budget, str | None],
322 clickhouse: tuple[RunTelemetry, str | None],
323 agentmail: tuple[tuple[Conversation, ...], str | None],
324 github: tuple[Deploy, str | None],
325 queue_facts: dict[str, TxnFacts] | None = None,
326) -> DashboardModel:
327 """Compose the whole dashboard view from the source readouts."""
328 t, t_err = temporal
329 budget, ynab_err = ynab
330 telemetry, ch_err = clickhouse
331 conversations, mail_err = agentmail
332 deploy, gh_err = github
333 facts = queue_facts or {}
335 sources = (
336 _health("temporal", t_err, f"{t.in_flight} in-flight"),
337 _health("ynab", ynab_err, f"{budget.unapproved} unapproved"),
338 _health("clickhouse", ch_err, f"{telemetry.total_spans} spans"),
339 _health("agentmail", mail_err, f"{len(conversations)} threads"),
340 _health("github", gh_err, f"{len(deploy.prs)} PRs"),
341 )
343 convo_by_ref = {c.ref: c for c in conversations if c.ref}
344 needs_you, handled = _humanize_queue(t, facts, convo_by_ref)
346 lifecycle = Lifecycle(
347 states=t.lifecycle_states,
348 in_flight=t.in_flight,
349 archived=t.archived,
350 terminated=t.terminated,
351 )
352 autonomy = Autonomy(
353 observe=t.observe,
354 eligible=t.eligible,
355 blessed=t.blessed,
356 rules=t.rules,
357 offers=t.offers,
358 )
359 health = _rollup(
360 now=now,
361 poll_live=t.poll_live,
362 poll_status=t.poll_status,
363 poll_last_start=t.poll_last_start,
364 temporal_error=t_err,
365 ynab_error=ynab_err,
366 telemetry=telemetry,
367 failures=t.failures,
368 needs_you=len(needs_you),
369 )
370 narrative = narrate(
371 now=now,
372 health=health,
373 lifecycle=lifecycle,
374 needs_you=needs_you,
375 handled=handled,
376 budget=budget,
377 autonomy=autonomy,
378 telemetry=telemetry,
379 )
381 return DashboardModel(
382 generated_at=now,
383 repo=repo,
384 narrative=narrative,
385 health=health,
386 sources=sources,
387 needs_you=needs_you,
388 handled=handled,
389 lifecycle=lifecycle,
390 autonomy=autonomy,
391 budget=budget,
392 conversations=conversations,
393 dispatch=t.dispatch,
394 telemetry=telemetry,
395 failures=t.failures,
396 deploy=deploy,
397 )
_rollup reduces the readouts to the one verdict. The precedence is the important part: the tone reflects current operational health — poll dead or Temporal down is bad; a stuck worker, an unreachable money source, or a high live error rate is warn. An isolated historical failure is surfaced as a chip and in the narrative, but it does not flip the verdict.
The verdict: bad → warn → ok, with failures deliberately excluded from tone.
src/ynab_agent/dashboard/read_model.py · 397 lines
src/ynab_agent/dashboard/read_model.py
⋯ 146 lines hidden (lines 1–146)
1"""Assemble the source readouts into one view-model (pure).
3Each reader hands in ``(data, error)``; this composes the
4:class:`~ynab_agent.dashboard.model.DashboardModel`, turning each error into a
5source-health dot and each Temporal/YNAB/AgentMail piece into the operator's
6shape: a composite :class:`~ynab_agent.dashboard.model.Health` verdict, a
7humanized owner queue (split into *needs you* vs *already settled*), and a
8deterministic plain-English :class:`~ynab_agent.dashboard.model.Narrative`. No
9I/O — fully unit-testable from in-memory readouts.
10"""
12from __future__ import annotations
14from typing import TYPE_CHECKING
16from ynab_agent.dashboard.model import (
17 Autonomy,
18 DashboardModel,
19 Health,
20 Lifecycle,
21 Narrative,
22 QueueItem,
23 SourceHealth,
24)
26if TYPE_CHECKING:
27 from datetime import datetime
29 from ynab_agent.dashboard.model import (
30 Budget,
31 Conversation,
32 Deploy,
33 Failure,
34 RunTelemetry,
35 TxnFacts,
36 )
37 from ynab_agent.dashboard.temporal_source import TemporalReadout
39# A worker span older than this reads as a stale heartbeat (warn).
40_STALE_WORKER_SECS = 3600.0
41# A trace error rate above this contributes a warn to the health rollup.
42_ERROR_RATE_WARN = 0.05
43# The activities whose run-count means "a category was written to YNAB".
44_APPLIED_ACTIVITY = "commit_to_ynab"
47def _health(name: str, error: str | None, ok_detail: str) -> SourceHealth:
48 """A source dot: green with a summary, or red with the error/'off'."""
49 if error is None:
50 return SourceHealth(name=name, ok=True, detail=ok_detail)
51 return SourceHealth(name=name, ok=False, detail=error)
54def _s(n: int) -> str:
55 """The plural suffix for a count (1 → '', else 's')."""
56 return "" if n == 1 else "s"
59def _short(ident: str) -> str:
60 """A short, readable stand-in when an id can't be humanized."""
61 return ident[:8] if len(ident) > 8 else ident
64def _age_phrase(seconds: float) -> str:
65 """A coarse, friendly age ('4 days', '3 hours', 'moments')."""
66 if seconds < 90:
67 return "moments"
68 if seconds < 5400:
69 n = int(seconds // 60)
70 return f"{n} minute{_s(n)}"
71 if seconds < 129600:
72 n = int(seconds // 3600)
73 return f"{n} hour{_s(n)}"
74 n = int(seconds // 86400)
75 return f"{n} day{_s(n)}"
78def _oldest_age(items: tuple[QueueItem, ...], now: datetime) -> str | None:
79 """The age of the longest-waiting item, as a phrase (None if empty)."""
80 # Normalize each to aware BEFORE min() — comparing aware vs naive raises.
81 sinces = [_aware(q.since, now) for q in items if q.since is not None]
82 if not sinces:
83 return None
84 return _age_phrase((now - min(sinces)).total_seconds())
87def _humanize_queue(
88 readout: TemporalReadout,
89 queue_facts: dict[str, TxnFacts],
90 convo_by_ref: dict[str, Conversation],
91) -> tuple[tuple[QueueItem, ...], tuple[QueueItem, ...]]:
92 """Turn bare awaiting ids into readable rows; split needs-you vs settled.
94 A proposal whose transaction is already ``approved`` in YNAB is one the
95 owner settled in-app — it lands in *handled* (it lapses on its own); every
96 other proposal, and every live autonomy offer, lands in *needs you*. The
97 best label is the proposal's email subject, then ``payee · amount``, then a
98 short id.
99 """
100 needs_you: list[QueueItem] = []
101 handled: list[QueueItem] = []
103 for item in readout.awaiting:
104 facts = queue_facts.get(item.ident)
105 convo = convo_by_ref.get(item.ident)
106 question = convo.subject if convo is not None else None
107 payee = facts.payee if facts is not None else None
108 amount = facts.amount if facts is not None else None
109 approved = facts.approved if facts is not None else None
110 if question:
111 label = question
112 elif payee:
113 label = f"{payee} {amount}".strip()
114 else:
115 label = _short(item.ident)
116 row = QueueItem(
117 kind="proposal",
118 label=label,
119 ident=item.ident,
120 since=item.since,
121 payee=payee,
122 amount=amount,
123 category=facts.category if facts is not None else None,
124 approved=approved,
125 question=question,
126 )
127 (handled if approved is True else needs_you).append(row)
129 for offer in readout.offers:
130 convo = convo_by_ref.get(offer.rule_id)
131 needs_you.append(
132 QueueItem(
133 kind="offer",
134 label=(convo.subject if convo else None)
135 or offer.payee
136 or offer.rule_id,
137 ident=offer.rule_id,
138 since=offer.started_at,
139 payee=offer.payee or None,
140 question=convo.subject if convo else None,
141 )
142 )
144 return tuple(needs_you), tuple(handled)
147def _rollup(
148 *,
149 now: datetime,
150 poll_live: bool,
151 poll_status: str,
152 poll_last_start: datetime | None,
153 temporal_error: str | None,
154 ynab_error: str | None,
155 telemetry: RunTelemetry,
156 failures: tuple[Failure, ...],
157 needs_you: int,
158) -> Health:
159 """Reduce the readouts to the single masthead health verdict."""
160 worker_last = telemetry.last_activity
161 rate = (
162 telemetry.error_spans / telemetry.total_spans
163 if telemetry.available and telemetry.total_spans > 0
164 else None
165 )
166 real_failures = sum(1 for f in failures if not f.intentional)
167 worker_stale = (
168 worker_last is not None
169 and (now - _aware(worker_last, now)).total_seconds()
170 > _STALE_WORKER_SECS
171 )
173 # Tone is *current* operational health — can it do its job right now. An
174 # isolated historical failure is surfaced (chip + narrative) but doesn't
175 # flip the verdict; a stuck worker, an unreachable money source, or a high
176 # live error rate does.
177 if not poll_live or temporal_error is not None:
178 tone, label = "bad", "down"
179 elif (
180 worker_stale
181 or ynab_error is not None
182 or (rate is not None and rate > _ERROR_RATE_WARN)
183 ):
184 tone, label = "warn", "degraded"
185 else:
186 tone, label = "ok", "healthy"
188 return Health(
189 tone=tone,
190 label=label,
191 poll_live=poll_live,
192 poll_status=poll_status,
193 poll_last_start=poll_last_start,
194 worker_last_span=worker_last,
195 span_error_rate=rate,
196 needs_you=needs_you,
197 real_failures=real_failures,
198 )
201def _aware(when: datetime, now: datetime) -> datetime:
202 return when if when.tzinfo is not None else when.replace(tzinfo=now.tzinfo)
205def _applied_count(telemetry: RunTelemetry) -> int:
⋯ 192 lines hidden (lines 206–397)
206 """Recent writes to YNAB, from the commit activity's run count."""
207 for activity in telemetry.activities:
208 if activity.name == _APPLIED_ACTIVITY:
209 return activity.count
210 return 0
213def narrate(
214 *,
215 now: datetime,
216 health: Health,
217 lifecycle: Lifecycle,
218 needs_you: tuple[QueueItem, ...],
219 handled: tuple[QueueItem, ...],
220 budget: Budget,
221 autonomy: Autonomy,
222 telemetry: RunTelemetry,
223) -> Narrative:
224 """Compose the deterministic 'state of things' summary from the numbers.
226 Every clause is guarded, so the paragraph reads naturally whether or not a
227 given source is on. This is the trustworthy baseline; an optional LLM polish
228 may later rephrase it for warmth, always falling back to exactly this text.
229 """
230 states = {s.state: s.count for s in lifecycle.states}
231 proposals = tuple(q for q in needs_you if q.kind == "proposal")
232 offers = tuple(q for q in needs_you if q.kind == "offer")
233 p, h, offers_n = len(proposals), len(handled), len(offers)
234 real_failures = health.real_failures
236 # Headline — the single most important thing.
237 if health.tone == "bad":
238 headline = "Attention needed."
239 elif p or offers_n:
240 waiting = p + offers_n
241 headline = f"{waiting} thing{_s(waiting)} waiting on you."
242 else:
243 headline = "All caught up — nothing needs you."
245 sentences: list[str] = []
247 if budget.available:
248 u = budget.unapproved
249 if u == 0:
250 sentences.append("You're caught up in YNAB.")
251 elif u <= 5:
252 sentences.append(
253 f"You're nearly caught up in YNAB ({u} unapproved)."
254 )
255 else:
256 sentences.append(f"{u} transactions await approval in YNAB.")
258 in_flight = lifecycle.in_flight
259 if in_flight:
260 open_n = states.get("open", 0)
261 awaiting_total = p + h
262 other = max(0, in_flight - open_n - awaiting_total)
263 parts = [f"{open_n} categorized and resting"] if open_n else []
264 if h:
265 parts.append(f"{h} you already settled in YNAB")
266 if p:
267 parts.append(f"{p} waiting on your reply")
268 if other:
269 parts.append(f"{other} still being processed")
270 body = ", ".join(parts) if parts else "all winding down"
271 sentences.append(
272 f"The agent is tracking {in_flight} "
273 f"transaction{_s(in_flight)}: {body}."
274 )
275 else:
276 sentences.append("No transactions are in flight.")
278 applied = _applied_count(telemetry)
279 if telemetry.available and applied:
280 obs = autonomy.observe
281 tail = f" and is learning ({obs} rule{_s(obs)} observed)" if obs else ""
282 sentences.append(
283 f"It has applied {applied} categor{'y' if applied == 1 else 'ies'} "
284 f"in the last {telemetry.window_days} days{tail}."
285 )
287 closing: list[str] = []
288 if offers_n:
289 verb = "s" if offers_n == 1 else ""
290 closing.append(
291 f"{offers_n} autonomy offer{_s(offers_n)} await{verb} your yes/no"
292 )
293 oldest = _oldest_age(proposals, now)
294 if p and oldest:
295 closing.append(f"the oldest reply has been waiting {oldest}")
296 if h:
297 closing.append("the settled ones lapse on their own")
298 if real_failures:
299 closing.append(
300 f"{real_failures} workflow{_s(real_failures)} failed "
301 "recently — check Failures"
302 )
303 if not p and not offers_n and not closing:
304 closing.append("nothing needs you right now")
305 if closing:
306 joined = closing[0] + (
307 "" if len(closing) == 1 else "; " + "; ".join(closing[1:])
308 )
309 sentences.append(joined[0].upper() + joined[1:] + ".")
311 return Narrative(
312 headline=headline, paragraphs=(" ".join(sentences),), tone=health.tone
313 )
316def assemble(
317 *,
318 now: datetime,
319 repo: str,
320 temporal: tuple[TemporalReadout, str | None],
321 ynab: tuple[Budget, str | None],
322 clickhouse: tuple[RunTelemetry, str | None],
323 agentmail: tuple[tuple[Conversation, ...], str | None],
324 github: tuple[Deploy, str | None],
325 queue_facts: dict[str, TxnFacts] | None = None,
326) -> DashboardModel:
327 """Compose the whole dashboard view from the source readouts."""
328 t, t_err = temporal
329 budget, ynab_err = ynab
330 telemetry, ch_err = clickhouse
331 conversations, mail_err = agentmail
332 deploy, gh_err = github
333 facts = queue_facts or {}
335 sources = (
336 _health("temporal", t_err, f"{t.in_flight} in-flight"),
337 _health("ynab", ynab_err, f"{budget.unapproved} unapproved"),
338 _health("clickhouse", ch_err, f"{telemetry.total_spans} spans"),
339 _health("agentmail", mail_err, f"{len(conversations)} threads"),
340 _health("github", gh_err, f"{len(deploy.prs)} PRs"),
341 )
343 convo_by_ref = {c.ref: c for c in conversations if c.ref}
344 needs_you, handled = _humanize_queue(t, facts, convo_by_ref)
346 lifecycle = Lifecycle(
347 states=t.lifecycle_states,
348 in_flight=t.in_flight,
349 archived=t.archived,
350 terminated=t.terminated,
351 )
352 autonomy = Autonomy(
353 observe=t.observe,
354 eligible=t.eligible,
355 blessed=t.blessed,
356 rules=t.rules,
357 offers=t.offers,
358 )
359 health = _rollup(
360 now=now,
361 poll_live=t.poll_live,
362 poll_status=t.poll_status,
363 poll_last_start=t.poll_last_start,
364 temporal_error=t_err,
365 ynab_error=ynab_err,
366 telemetry=telemetry,
367 failures=t.failures,
368 needs_you=len(needs_you),
369 )
370 narrative = narrate(
371 now=now,
372 health=health,
373 lifecycle=lifecycle,
374 needs_you=needs_you,
375 handled=handled,
376 budget=budget,
377 autonomy=autonomy,
378 telemetry=telemetry,
379 )
381 return DashboardModel(
382 generated_at=now,
383 repo=repo,
384 narrative=narrative,
385 health=health,
386 sources=sources,
387 needs_you=needs_you,
388 handled=handled,
389 lifecycle=lifecycle,
390 autonomy=autonomy,
391 budget=budget,
392 conversations=conversations,
393 dispatch=t.dispatch,
394 telemetry=telemetry,
395 failures=t.failures,
396 deploy=deploy,
397 )
Operator resets, kept out of the fault headline
At go-live the operator terminated a batch of test/reset workflows. Those are housekeeping, not breakage, but they were swelling the failure list. _intentional classifies them out — and does so carefully: only a terminated run carrying a reset-vocabulary reason counts as housekeeping. A failed run is always real breakage, so a coincidental "connection reset" in an exception message is never miscounted.
The reset vocabulary, the scoped classifier, and where each Failure is tagged.
src/ynab_agent/dashboard/temporal_source.py · 379 lines
src/ynab_agent/dashboard/temporal_source.py
⋯ 41 lines hidden (lines 1–41)
1"""Temporal reader: the live run ledger (reuses the worker's client).
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"""
12from __future__ import annotations
14import contextlib
15from datetime import datetime
16from typing import TYPE_CHECKING, Final
18from temporalio.client import WorkflowExecutionStatus
20from ynab_agent.dashboard.model import (
21 DispatchTally,
22 Failure,
23 OfferRow,
24 QueueItem,
25 RuleRow,
26 StateCount,
27)
28from ynab_agent.domain.base import Frozen
30if TYPE_CHECKING:
31 from collections.abc import AsyncIterator
33 from temporalio.client import Client, WorkflowExecution
35_MAX_PER_TYPE: Final = 500
36_MAX_STATE_QUERIES: Final = 200
37_REGISTRY_ID: Final = "ynab-rule-registry"
39# Operator-termination vocabulary: the human reasons attached to a deliberate
40# go-live / re-test / reset wipe. Used to keep housekeeping out of the fault
41# headline (the page still lists them under disclosure).
42_RESET_MARKERS: Final = (
43 "go-live",
44 "go live",
45 "reset",
46 "re-test",
47 "retest",
48 "cleanup",
49 "clean up",
50 "scaffold",
51 "teardown",
52 "redeploy",
53 "wipe",
54)
57def _intentional(kind: str, reason: str | None) -> bool:
58 """Whether a terminal workflow was the operator's own housekeeping.
60 Scoped to *terminated* runs carrying a human reason — a ``failed`` run is
61 real breakage, never housekeeping — so a coincidental word in an exception
62 message (e.g. "connection reset") is never miscounted as intentional.
63 """
64 if kind != "terminated" or not reason:
65 return False
66 low = reason.lower()
67 return any(marker in low for marker in _RESET_MARKERS)
⋯ 234 lines hidden (lines 68–301)
70class TemporalReadout(Frozen):
71 """The Temporal-derived pieces of the dashboard, ready to slot in."""
73 poll_status: str = "none"
74 poll_live: bool = False
75 poll_last_start: datetime | None = None
76 lifecycle_states: tuple[StateCount, ...] = ()
77 in_flight: int = 0
78 archived: int = 0
79 terminated: int = 0
80 rules: tuple[RuleRow, ...] = ()
81 observe: int = 0
82 eligible: int = 0
83 blessed: int = 0
84 offers: tuple[OfferRow, ...] = ()
85 awaiting: tuple[QueueItem, ...] = ()
86 dispatch: DispatchTally = DispatchTally()
87 failures: tuple[Failure, ...] = ()
90def _status(execution: WorkflowExecution) -> str:
91 status = execution.status
92 return status.name.lower() if status is not None else "unknown"
95async def _take(
96 iterator: AsyncIterator[WorkflowExecution], cap: int = _MAX_PER_TYPE
97) -> AsyncIterator[WorkflowExecution]:
98 """Yield at most ``cap`` executions (a runaway-visibility backstop)."""
99 seen = 0
100 async for execution in iterator:
101 yield execution
102 seen += 1
103 if seen >= cap:
104 return
107async def _poll(client: Client) -> tuple[str, bool, datetime | None]:
108 """The most recent poll tick: (status, live, last_start).
110 Standard (SQL) Temporal visibility rejects an ``ORDER BY`` clause, so we
111 take a bounded page and pick the latest start client-side rather than
112 sorting in the query.
113 """
114 latest: WorkflowExecution | None = None
115 async for execution in _take(
116 client.list_workflows("WorkflowType = 'PollWorkflow'"), cap=100
117 ):
118 start = execution.start_time
119 if latest is None or (
120 start is not None
121 and (latest.start_time is None or start > latest.start_time)
122 ):
123 latest = execution
124 if latest is None:
125 return "none", False, None
126 live = latest.status in (
127 WorkflowExecutionStatus.RUNNING,
128 WorkflowExecutionStatus.COMPLETED,
129 WorkflowExecutionStatus.CONTINUED_AS_NEW,
130 )
131 return _status(latest), live, latest.start_time
134async def _lifecycle(
135 client: Client,
136) -> tuple[tuple[StateCount, ...], int, tuple[QueueItem, ...]]:
137 """Count running W2s by state; collect the awaiting-human queue."""
138 counts: dict[str, int] = {}
139 awaiting: list[QueueItem] = []
140 seen = 0
141 async for execution in _take(
142 client.list_workflows(
143 "WorkflowType = 'TransactionWorkflow' "
144 "AND ExecutionStatus = 'Running'"
145 ),
146 cap=_MAX_STATE_QUERIES,
147 ):
148 seen += 1
149 try:
150 handle = client.get_workflow_handle(
151 execution.id, run_id=execution.run_id
152 )
153 state = await handle.query("state", result_type=str)
154 except Exception: # a busy/odd workflow shouldn't drop the whole panel
155 state = "unknown"
156 counts[state] = counts.get(state, 0) + 1
157 if state == "awaiting_human":
158 awaiting.append(
159 QueueItem(
160 kind="proposal",
161 label=execution.id,
162 ident=execution.id,
163 since=execution.start_time,
164 )
165 )
166 states = tuple(
167 StateCount(state=name, count=counts[name]) for name in sorted(counts)
168 )
169 return states, seen, tuple(awaiting)
172async def _registry(
173 client: Client,
174) -> tuple[tuple[RuleRow, ...], int, int, int]:
175 """The rule table + (observe, eligible, blessed) counts (view query)."""
176 from ynab_agent.domain.allocations import ProposedCategory
177 from ynab_agent.domain.enums import RuleSource, TrustState
178 from ynab_agent.workflow.registry_types import RegistryView
180 handle = client.get_workflow_handle(_REGISTRY_ID)
181 view = await handle.query("view", result_type=RegistryView)
182 rows: list[RuleRow] = []
183 observe = eligible = blessed = 0
184 for rule in view.rules:
185 allocation = rule.action.allocation
186 category = (
187 str(allocation.category)
188 if isinstance(allocation, ProposedCategory)
189 else "split"
190 )
191 rows.append(
192 RuleRow(
193 payee=rule.match.payee_pattern,
194 category=category,
195 trust=rule.trust.value,
196 source=rule.source.value,
197 hits=rule.hits,
198 offered=rule.offered_at is not None,
199 last_confirmed_at=rule.last_confirmed_at,
200 )
201 )
202 if rule.source is RuleSource.HUMAN_EXPLICIT:
203 blessed += 1
204 elif rule.trust is TrustState.TRUSTED:
205 eligible += 1
206 else:
207 observe += 1
208 return tuple(rows), observe, eligible, blessed
211async def _offers(client: Client) -> tuple[OfferRow, ...]:
212 """The live autonomy-offer workflows (awaiting the owner's yes/no)."""
213 offers: list[OfferRow] = []
214 async for execution in _take(
215 client.list_workflows(
216 "WorkflowType = 'AutonomyOfferWorkflow' "
217 "AND ExecutionStatus = 'Running'"
218 )
219 ):
220 offers.append(
221 OfferRow(
222 rule_id=execution.id.removeprefix("autonomy-offer-"),
223 payee="",
224 status=_status(execution),
225 started_at=execution.start_time,
226 )
227 )
228 return tuple(offers)
231async def _dispatch(client: Client) -> DispatchTally:
232 """Tally recent inbound dispatch results by routing action."""
233 counts: dict[str, int] = {}
234 total = 0
235 async for execution in _take(
236 client.list_workflows(
237 "WorkflowType = 'DispatchWorkflow' "
238 "AND ExecutionStatus = 'Completed'"
239 )
240 ):
241 total += 1
242 action = "?"
243 try:
244 handle = client.get_workflow_handle(
245 execution.id, run_id=execution.run_id
246 )
247 result = await handle.result()
248 got = getattr(result, "action", None)
249 if isinstance(got, str):
250 action = got
251 except Exception: # a bad decode shouldn't drop the tally
252 action = "?"
253 counts[action] = counts.get(action, 0) + 1
254 return DispatchTally(
255 transaction=counts.get("transaction", 0),
256 offer=counts.get("offer", 0),
257 receipt=counts.get("receipt", 0),
258 command=counts.get("command", 0),
259 quarantine=counts.get("quarantine", 0),
260 ignore=counts.get("ignore", 0),
261 total=total,
262 )
265async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
266 """Recover a terminated/failed workflow's human reason from history."""
267 try:
268 handle = client.get_workflow_handle(
269 execution.id, run_id=execution.run_id
270 )
271 found: str | None = None
272 async for event in handle.fetch_history_events():
273 terminated = event.workflow_execution_terminated_event_attributes
274 if terminated.reason:
275 found = terminated.reason
276 failed = event.workflow_execution_failed_event_attributes
277 if failed.failure.message:
278 found = failed.failure.message
279 return found
280 except Exception: # the reason is a nicety — never fail the panel for it
281 return None
284async def _terminal(client: Client) -> tuple[int, int, tuple[Failure, ...]]:
285 """Recent archived (completed W2) + terminated/failed counts + failures."""
286 archived = terminated = 0
287 failures: list[Failure] = []
288 async for _completed in _take(
289 client.list_workflows(
290 "WorkflowType = 'TransactionWorkflow' "
291 "AND ExecutionStatus = 'Completed'"
292 )
293 ):
294 archived += 1
295 async for execution in _take(
296 client.list_workflows(
297 "ExecutionStatus = 'Terminated' OR ExecutionStatus = 'Failed'"
298 )
299 ):
300 terminated += 1
301 if len(failures) < 25:
302 kind = _status(execution)
303 reason = await _reason(client, execution)
304 failures.append(
305 Failure(
306 workflow_id=execution.id,
307 kind=kind,
308 reason=reason,
309 when=execution.close_time,
310 intentional=_intentional(kind, reason),
311 )
312 )
⋯ 67 lines hidden (lines 313–379)
313 return archived, terminated, tuple(failures)
316async def fetch(client: Client) -> tuple[TemporalReadout, str | None]:
317 """Read the agent's Temporal state; each panel degrades independently.
319 Every sub-read is guarded on its own, so one unsupported visibility query
320 (or a transient failure) reddens only its panel and names itself in the
321 error — the rest of the page still fills in.
322 """
323 poll_status = "none"
324 poll_live = False
325 poll_last: datetime | None = None
326 states: tuple[StateCount, ...] = ()
327 in_flight = 0
328 awaiting: tuple[QueueItem, ...] = ()
329 rules: tuple[RuleRow, ...] = ()
330 observe = eligible = blessed = 0
331 offers: tuple[OfferRow, ...] = ()
332 dispatch = DispatchTally()
333 archived = terminated = 0
334 failures: tuple[Failure, ...] = ()
335 errors: list[str] = []
337 try:
338 poll_status, poll_live, poll_last = await _poll(client)
339 except Exception as exc:
340 errors.append(f"poll: {type(exc).__name__}")
341 try:
342 states, in_flight, awaiting = await _lifecycle(client)
343 except Exception as exc:
344 errors.append(f"lifecycle: {type(exc).__name__}")
345 # The registry may not exist until the first learning signal — a normal
346 # state, not an error, so its absence is simply suppressed.
347 with contextlib.suppress(Exception):
348 rules, observe, eligible, blessed = await _registry(client)
349 try:
350 offers = await _offers(client)
351 except Exception as exc:
352 errors.append(f"offers: {type(exc).__name__}")
353 try:
354 dispatch = await _dispatch(client)
355 except Exception as exc:
356 errors.append(f"dispatch: {type(exc).__name__}")
357 try:
358 archived, terminated, failures = await _terminal(client)
359 except Exception as exc:
360 errors.append(f"terminal: {type(exc).__name__}")
362 readout = TemporalReadout(
363 poll_status=poll_status,
364 poll_live=poll_live,
365 poll_last_start=poll_last,
366 lifecycle_states=states,
367 in_flight=in_flight,
368 archived=archived,
369 terminated=terminated,
370 rules=rules,
371 observe=observe,
372 eligible=eligible,
373 blessed=blessed,
374 offers=offers,
375 awaiting=awaiting,
376 dispatch=dispatch,
377 failures=failures,
378 )
379 return readout, "; ".join(errors) if errors else None
The render — zones, escaping, designed empty states
page composes the zones in reading order: masthead, the narrative, needs you, "is it working", the activity feed, budget, then the operator detail as rails. It's pure string assembly with all CSS inline and no JavaScript — a static projection of an already-computed model.
Reading order: the operator's questions first, the detail in rails last.
src/ynab_agent/dashboard/render.py · 651 lines
src/ynab_agent/dashboard/render.py
⋯ 627 lines hidden (lines 1–627)
1"""Render the view-model to one self-contained HTML page (pure).
3All CSS is inline, there is no JavaScript, and the page makes no network request
4of its own — it is a static projection of an already-computed
5:class:`~ynab_agent.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary.
8The page is organized around the operator's questions, not the subsystems the
9data came from: a masthead verdict, a plain-English **state of things**, then
10**needs you** (the only acted-upon queue), **is it working** (health + flow),
11**what it's done** (the conversation feed), the **budget**, and finally the
12operator detail folded into progressive-disclosure rails. Structure comes from
13hairlines, whitespace, type, and a single accent — not boxes; semantic colour
14lives in dots and pills, not fills.
15"""
17from __future__ import annotations
19from datetime import UTC, datetime
20from html import escape
21from typing import TYPE_CHECKING
23if TYPE_CHECKING:
24 from ynab_agent.dashboard.model import (
25 DashboardModel,
26 Lifecycle,
27 QueueItem,
28 RunTelemetry,
29 )
31_CSS = """
32:root{--fg:#1a1a1a;--mut:#6b6b6b;--faint:#8a8a8a;--line:#e6e6e6;--bg:#fff;
33--ok:#1a7f37;--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#f6f8fa;
34--serif:"Iowan Old Style","Palatino Linotype",Palatino,Georgia,ui-serif,serif}
35@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
36--faint:#7a7a7a;--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;
37--bad:#f85149;--accent:#58a6ff;--card:#161b22}}
38*{box-sizing:border-box}
39body{margin:0;background:var(--bg);color:var(--fg);
40font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
41main{max-width:820px;margin:0 auto;padding:30px 20px 72px}
42a{color:var(--accent);text-decoration:none}a:hover{text-decoration:underline}
43.mut{color:var(--mut)}.faint{color:var(--faint)}
44.ok{color:var(--ok)}.warn{color:var(--warn)}.bad{color:var(--bad)}
45.accent{color:var(--accent)}
46.mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
47.dot{display:inline-block;width:9px;height:9px;border-radius:50%;flex:none;
48vertical-align:baseline}
49.dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
50.dot.bad{background:var(--bad)}.dot.mute{background:var(--faint)}
51.dot.accent{background:var(--accent)}
53/* Masthead — wordmark + the one composite verdict + the few facts. */
54.mast{display:flex;align-items:flex-start;gap:14px;flex-wrap:wrap}
55.mast h1{font-size:19px;margin:0;letter-spacing:-.01em;font-weight:650}
56.mast .tag{color:var(--mut);font-size:12px;margin:2px 0 0}
57.verdict{display:inline-flex;align-items:center;gap:7px;font-weight:650;
58font-size:15px}
59.verdict .dot{width:11px;height:11px}
60.verdict.ok{color:var(--ok)}.verdict.warn{color:var(--warn)}
61.verdict.bad{color:var(--bad)}
62.facts{margin-left:auto;text-align:right;color:var(--mut);font-size:12px;
63line-height:1.7}
64.chips{display:flex;flex-wrap:wrap;gap:7px;margin:12px 0 0}
65.chip{display:inline-flex;align-items:center;gap:6px;padding:3px 10px;
66border-radius:999px;border:1px solid var(--line);font-size:12px;font-weight:550}
67.chip.ok{border-color:transparent;color:var(--ok);
68background:color-mix(in srgb,var(--ok) 14%,transparent)}
69.chip.warn{border-color:transparent;color:var(--warn);
70background:color-mix(in srgb,var(--warn) 16%,transparent)}
71.chip.bad{border-color:transparent;color:var(--bad);
72background:color-mix(in srgb,var(--bad) 14%,transparent)}
73.srcs{display:flex;flex-wrap:wrap;gap:14px;margin:14px 0 0;font-size:11px;
74color:var(--mut)}
75.srcs .dot{margin-right:5px}
77/* Section = an 11px uppercase label with a hairline underline. */
78section{margin:30px 0 0}
79h2{font-size:11px;text-transform:uppercase;letter-spacing:.09em;
80color:var(--mut);margin:0 0 13px;font-weight:650;padding-bottom:6px;
81border-bottom:1px solid var(--line);display:flex;align-items:baseline;gap:8px}
82h2 .h2-aside{margin-left:auto;font-size:11px;font-weight:500;letter-spacing:0;
83text-transform:none;color:var(--faint)}
85/* State of things — the one serif voice, a lead with an accent rule. */
86.narr{border-left:3px solid var(--line);padding:2px 0 2px 15px;margin:26px 0 0}
87.narr.ok{border-color:var(--ok)}.narr.warn{border-color:var(--warn)}
88.narr.bad{border-color:var(--bad)}
89.narr .head{font-weight:650;font-size:15px;margin:0 0 4px}
90.narr .body{font-family:var(--serif);font-size:16.5px;line-height:1.5;
91margin:0;color:var(--fg);max-width:62ch}
93/* Needs you — the humanized queue. */
94.qrow{display:flex;align-items:baseline;gap:11px;padding:9px 0;
95border-bottom:1px solid var(--line)}
96.qrow:last-child{border-bottom:0}
97.qrow .dot{position:relative;top:1px}
98.q-main{flex:1;min-width:0}
99.q-title{display:block;font-weight:550;overflow-wrap:anywhere}
100.q-sub{display:block;color:var(--mut);font-size:12.5px;margin-top:1px}
101.q-age{color:var(--faint);font-size:12px;white-space:nowrap;
102font-variant-numeric:tabular-nums}
103.tag-pill{display:inline-block;padding:0 7px;border-radius:10px;
104font-size:10.5px;font-weight:650;text-transform:uppercase;letter-spacing:.04em;
105background:var(--card);color:var(--mut);vertical-align:1px}
106.tag-pill.offer{color:var(--accent);
107background:color-mix(in srgb,var(--accent) 14%,transparent)}
108.empty{display:flex;align-items:center;gap:10px;color:var(--mut);
109font-size:14px;padding:6px 0}
110.empty .big{font-size:20px;line-height:1}
111.note{color:var(--mut);font-size:12.5px;margin:11px 0 0}
113/* Is it working — health pills + a single flow bar. */
114.kpis{display:flex;flex-wrap:wrap;gap:18px;margin:0 0 16px}
115.kpi{display:inline-flex;align-items:baseline;gap:7px;font-size:13px}
116.kpi b{font-weight:600}
117.flow{display:flex;height:22px;border-radius:5px;overflow:hidden;
118margin:4px 0 10px;background:var(--card)}
119.flow>span{display:block;min-width:2px}
120.legend{display:flex;flex-wrap:wrap;gap:7px 16px;font-size:12px;
121color:var(--mut)}
122.legend .it{display:inline-flex;align-items:baseline;gap:6px}
123.legend b{color:var(--fg);font-weight:600;font-variant-numeric:tabular-nums}
125/* What it's done — the conversation feed. */
126.feed{display:flex;flex-direction:column}
127.frow{display:flex;align-items:baseline;gap:10px;padding:8px 0;
128border-bottom:1px solid var(--line)}
129.frow:last-child{border-bottom:0}
130.frow .g{width:1.2em;text-align:center;flex:none;color:var(--faint)}
131.f-main{flex:1;min-width:0}
132.f-title{display:block;overflow-wrap:anywhere}
133.f-prev{display:block;color:var(--faint);font-size:12px;margin-top:1px;
134white-space:nowrap;overflow:hidden;text-overflow:ellipsis}
135.f-age{color:var(--faint);font-size:12px;white-space:nowrap}
137/* Budget — one tight line + overspent inline. */
138.over{display:flex;flex-wrap:wrap;gap:6px 14px;margin:8px 0 0;font-size:13px}
139.over .it{display:inline-flex;gap:6px}
141/* Rails — progressive disclosure, hairline-topped. */
142details.rail{border-top:1px solid var(--line);padding:9px 0 2px}
143details.rail>summary{cursor:pointer;list-style:none;display:flex;
144align-items:center;gap:8px;font-size:12px;color:var(--mut)}
145details.rail>summary::-webkit-details-marker{display:none}
146details.rail>summary::before{content:"\\25B8";color:var(--faint);font-size:10px}
147details.rail[open]>summary::before{content:"\\25BE"}
148details.rail>summary .rail-n{margin-left:auto;color:var(--faint)}
149.rail-body{padding:12px 0 6px}
150table{width:100%;border-collapse:collapse;font-size:12.5px}
151th,td{text-align:left;padding:5px 12px 5px 0;vertical-align:top;
152border-bottom:1px solid var(--line)}
153th{color:var(--mut);font-weight:600;font-size:10.5px;text-transform:uppercase;
154letter-spacing:.04em}
155.num{text-align:right;font-variant-numeric:tabular-nums}
156footer{margin:40px 0 0;padding-top:15px;border-top:1px solid var(--line);
157color:var(--mut);font-size:12px;line-height:1.7}
158footer b{color:var(--fg);font-weight:600}
159"""
161_LIFECYCLE_ORDER = (
162 "discovered",
163 "enriching",
164 "hold_amazon",
165 "revising",
166 "awaiting_human",
167 "open",
168 "lapsed",
169)
170# Each in-flight state's human label + flow colour (a cool ramp: accent =
171# the agent is working, warn = it's on you, ok = settled-and-resting).
172_STATE_META = {
173 "discovered": ("discovered", "var(--faint)"),
174 "enriching": ("enriching", "var(--accent)"),
175 "hold_amazon": ("holding for Amazon detail", "var(--faint)"),
176 "revising": ("revising", "var(--accent)"),
177 "awaiting_human": ("awaiting you", "var(--warn)"),
178 "open": ("resting (done)", "var(--ok)"),
179 "lapsed": ("lapsed", "var(--faint)"),
180}
181_TRUST_CLASS = {"trusted": "ok", "confirmed": "warn", "suggested": "mute"}
182_CI_CLASS = {"passed": "ok", "failed": "bad", "running": "warn"}
183_STATE_CLASS = {"merged": "ok", "open": "warn", "closed": "mut"}
184_KIND_GLYPH = {"proposal": "✉", "offer": "◆", "thread": "·"}
187def _aware(when: datetime) -> datetime:
188 return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
191def _ago(when: datetime | None, now: datetime) -> str:
192 if when is None:
193 return "—"
194 secs = (now - _aware(when)).total_seconds()
195 if secs < 90:
196 return "just now"
197 if secs < 5400:
198 return f"{int(secs // 60)}m ago"
199 if secs < 129600:
200 return f"{int(secs // 3600)}h ago"
201 return f"{int(secs // 86400)}d ago"
204def _dot(kind: str) -> str:
205 return f'<span class="dot {kind}"></span>'
208def _meta(state: str) -> tuple[str, str]:
209 return _STATE_META.get(state, (state, "var(--faint)"))
212# ── Masthead ─────────────────────────────────────────────────────────────────
213def _masthead(model: DashboardModel) -> str:
214 h = model.health
215 now = model.generated_at
216 stamp = now.strftime("%Y-%m-%d %H:%M UTC")
218 chips: list[str] = []
219 if h.needs_you:
220 suffix = "s" if h.needs_you == 1 else ""
221 chips.append(
222 f'<span class="chip warn">{h.needs_you} need{suffix} you</span>'
223 )
224 else:
225 chips.append('<span class="chip ok">✓ all caught up</span>')
226 poll_cls = "ok" if h.poll_live else "bad"
227 poll_age = escape(_ago(h.poll_last_start, now))
228 chips.append(f'<span class="chip">{_dot(poll_cls)} poll {poll_age}</span>')
229 if model.budget.available:
230 chips.append(
231 f'<span class="chip">{model.budget.unapproved} unapproved</span>'
232 )
233 if h.real_failures:
234 suffix = "s" if h.real_failures != 1 else ""
235 chips.append(
236 f'<span class="chip bad">{h.real_failures} failure{suffix}</span>'
237 )
239 src = "".join(
240 f"<span>{_dot('ok' if s.ok else 'bad')}{escape(s.name)} "
241 f'<span class="faint">{escape(s.detail)}</span></span>'
242 for s in model.sources
243 )
244 return (
245 "<header>"
246 '<div class="mast">'
247 "<div><h1>ynab-agent · ops</h1>"
248 '<p class="tag">durable transaction triage · '
249 "earned-autonomy read-model</p></div>"
250 f'<div class="facts"><span class="verdict {escape(h.tone)}">'
251 f"{_dot(h.tone)}{escape(h.label)}</span><br>{escape(model.repo)}<br>"
252 f"{escape(stamp)}</div></div>"
253 f'<div class="chips">{"".join(chips)}</div>'
254 f'<div class="srcs">{src}</div>'
255 "</header>"
256 )
259# ── State of things ──────────────────────────────────────────────────────────
260def _narrative(model: DashboardModel) -> str:
261 n = model.narrative
262 body = " ".join(n.paragraphs)
263 return (
264 f'<section class="narr {escape(n.tone)}">'
265 f'<p class="head">{escape(n.headline)}</p>'
266 f'<p class="body">{escape(body)}</p></section>'
267 )
270# ── Needs you ────────────────────────────────────────────────────────────────
271def _needs_you(model: DashboardModel) -> str:
272 now = model.generated_at
273 queue = model.needs_you
274 if not queue:
275 body = (
276 '<div class="empty"><span class="big">✓</span>'
277 "<span>You're all caught up — nothing is waiting on you."
278 "</span></div>"
279 )
280 else:
281 body = "".join(_qrow(q, now) for q in queue[:30])
282 if model.handled:
283 n = len(model.handled)
284 plural = "s" if n != 1 else ""
285 verb = "are" if n != 1 else "is"
286 body += (
287 f'<p class="note">{n} more proposal{plural} you already settled '
288 f"in YNAB {verb} winding down — they lapse on their own, no reply "
289 "needed.</p>"
290 )
291 aside = (
292 f'<span class="h2-aside">oldest '
293 f"{escape(_ago(_oldest(queue), now))}</span>"
294 if queue
295 else ""
296 )
297 return f"<section><h2>Needs you{aside}</h2>{body}</section>"
300def _oldest(queue: tuple[QueueItem, ...]) -> datetime | None:
301 # Normalize to aware before min() — mixed aware/naive comparison raises.
302 sinces = [_aware(q.since) for q in queue if q.since is not None]
303 return min(sinces) if sinces else None
306def _qrow(q: QueueItem, now: datetime) -> str:
307 dot = "accent" if q.kind == "offer" else "warn"
308 title = escape(q.question or q.label)
309 pill = (
310 '<span class="tag-pill offer">offer</span> '
311 if q.kind == "offer"
312 else ""
313 )
314 sub_bits: list[str] = []
315 if q.amount:
316 sub_bits.append(escape(q.amount))
317 if q.category:
318 sub_bits.append(escape(q.category))
319 if not sub_bits and q.payee:
320 sub_bits.append(escape(q.payee))
321 sub = (
322 f'<span class="q-sub">{" · ".join(sub_bits)}</span>'
323 if sub_bits
324 else ""
325 )
326 age = escape(_ago(q.since, now))
327 return (
328 f'<div class="qrow">{_dot(dot)}'
329 f'<span class="q-main"><span class="q-title">{pill}{title}</span>'
330 f'{sub}</span><span class="q-age">{age}</span></div>'
331 )
334# ── Is it working ────────────────────────────────────────────────────────────
335def _working(model: DashboardModel) -> str:
336 h = model.health
337 now = model.generated_at
338 poll_word = escape(h.poll_status) if h.poll_live else "stopped"
339 poll_age = escape(_ago(h.poll_last_start, now))
340 kpis = [
341 f'<span class="kpi">{_dot("ok" if h.poll_live else "bad")}'
342 f'<b>poll</b> <span class="mut">{poll_word} '
343 f"· {poll_age}</span></span>"
344 ]
345 if h.worker_last_span is not None:
346 wago = (now - _aware(h.worker_last_span)).total_seconds()
347 wdot = "ok" if wago < 3600 else "warn"
348 span_age = escape(_ago(h.worker_last_span, now))
349 kpis.append(
350 f'<span class="kpi">{_dot(wdot)}<b>worker</b> '
351 f'<span class="mut">span {span_age}</span></span>'
352 )
353 else:
354 kpis.append(
355 f'<span class="kpi">{_dot("mute")}<b>worker</b> '
356 '<span class="mut">no telemetry</span></span>'
357 )
358 if h.span_error_rate is not None:
359 rate = h.span_error_rate
360 rdot = "ok" if rate <= 0.05 else "warn"
361 kpis.append(
362 f'<span class="kpi">{_dot(rdot)}<b>{rate * 100:.1f}%</b> '
363 '<span class="mut">span errors</span></span>'
364 )
365 return (
366 "<section><h2>Is it working?</h2>"
367 f'<div class="kpis">{"".join(kpis)}</div>'
368 f"{_flow(model.lifecycle)}</section>"
369 )
372def _flow(lc: Lifecycle) -> str:
373 counts = {s.state: s.count for s in lc.states}
374 ordered = [s for s in _LIFECYCLE_ORDER if counts.get(s)]
375 ordered += [
376 s for s in sorted(counts) if counts.get(s) and s not in _LIFECYCLE_ORDER
377 ]
378 if not ordered:
379 return '<p class="empty"><span>No transactions in flight.</span></p>'
380 segs = "".join(
381 f'<span style="flex:{counts[s]};background:{_meta(s)[1]}" '
382 f'title="{escape(_meta(s)[0])}: {counts[s]}"></span>'
383 for s in ordered
384 )
385 legend = "".join(
386 '<span class="it"><span class="dot" '
387 f'style="background:{_meta(s)[1]}"></span>'
388 f"<b>{counts[s]}</b> {escape(_meta(s)[0])}</span>"
389 for s in ordered
390 )
391 note = (
392 " — the post-write window runs for weeks, so resting transactions "
393 "haven't closed out yet"
394 if lc.archived == 0 and counts.get("open")
395 else ""
396 )
397 gloss = (
398 ' · <span class="faint">'
399 f"{lc.archived} archived{note} · "
400 f"{lc.terminated} terminated (recent)</span>"
401 )
402 return (
403 f'<div class="flow">{segs}</div>'
404 f'<div class="legend">{legend}{gloss}</div>'
405 )
408# ── What it's done ───────────────────────────────────────────────────────────
409def _activity(model: DashboardModel) -> str:
410 now = model.generated_at
411 if not model.conversations:
412 return ""
413 rows = ""
414 for c in model.conversations[:10]:
415 glyph = _KIND_GLYPH.get(c.kind, "·")
416 prev = (
417 f'<span class="f-prev">{escape(c.preview)}</span>'
418 if c.preview
419 else ""
420 )
421 age = escape(_ago(c.updated_at, now))
422 rows += (
423 f'<div class="frow"><span class="g">{glyph}</span>'
424 f'<span class="f-main"><span class="f-title">'
425 f"{escape(c.subject)}</span>{prev}</span>"
426 f'<span class="f-age">{age}</span></div>'
427 )
428 return (
429 "<section><h2>What it's done"
430 '<span class="h2-aside">recent threads</span></h2>'
431 f'<div class="feed">{rows}</div></section>'
432 )
435# ── Budget ───────────────────────────────────────────────────────────────────
436def _budget(model: DashboardModel) -> str:
437 b = model.budget
438 if not b.available:
439 return ""
440 if b.overspent:
441 items = "".join(
442 f'<span class="it"><span class="bad mono">{escape(c.balance)}'
443 f"</span> {escape(c.name)}</span>"
444 for c in b.overspent
445 )
446 over = f'<div class="over">{items}</div>'
447 else:
448 over = '<p class="note">No overspent categories.</p>'
449 noun = "y" if len(b.overspent) == 1 else "ies"
450 head = (
451 f"{b.unapproved} unapproved · "
452 f"{len(b.overspent)} overspent categor{noun}"
453 )
454 return (
455 "<section><h2>Budget · YNAB"
456 f'<span class="h2-aside">{escape(head)}</span></h2>{over}</section>'
457 )
460# ── Rails (progressive disclosure) ───────────────────────────────────────────
461def _rail(title: str, count: str, inner: str) -> str:
462 return (
463 f'<details class="rail"><summary>{escape(title)}'
464 f'<span class="rail-n">{escape(count)}</span></summary>'
465 f'<div class="rail-body">{inner}</div></details>'
466 )
469def _autonomy_rail(model: DashboardModel) -> str:
470 a = model.autonomy
471 note = (
472 '<p class="note">Earned then granted: a rule becomes <b>eligible</b> '
473 "after K consistent confirms, and <b>blessed</b> (auto-applies) only "
474 "once you accept the offer. A correction demotes it to observe.</p>"
475 )
476 if a.rules:
477 rows = ""
478 for r in a.rules[:30]:
479 trust_cls = _TRUST_CLASS.get(r.trust, "mut")
480 rows += (
481 f"<tr><td>{escape(r.payee)}</td>"
482 f'<td class="mut">{escape(r.category)}</td>'
483 f'<td class="{trust_cls}">{escape(r.trust)}</td>'
484 f'<td class="num">{r.hits}</td></tr>'
485 )
486 table = (
487 "<table><thead><tr><th>payee</th><th>category</th><th>trust</th>"
488 "<th class='num'>hits</th></tr></thead>"
489 f"<tbody>{rows}</tbody></table>"
490 )
491 else:
492 table = '<p class="note">No learned rules yet.</p>'
493 count = f"{a.observe} observe · {a.eligible} eligible · {a.blessed} blessed"
494 return _rail("Autonomy ladder", count, table + note)
497def _telemetry_rail(model: DashboardModel) -> str:
498 t: RunTelemetry = model.telemetry
499 now = model.generated_at
500 if not t.available:
501 return _rail(
502 "Run telemetry",
503 "off",
504 '<p class="note">ClickHouse not configured or unreachable; '
505 "Temporal carries the page regardless.</p>",
506 )
507 rows = "".join(
508 f'<tr><td class="mono">{escape(a.name)}</td>'
509 f'<td class="num">{a.count}</td>'
510 f'<td class="num mut">{a.avg_ms:.0f}</td>'
511 f'<td class="num mut">{a.max_ms:.0f}</td></tr>'
512 for a in t.activities
513 )
514 table = (
515 "<table><thead><tr><th>activity</th><th class='num'>runs</th>"
516 "<th class='num'>avg ms</th><th class='num'>max ms</th></tr></thead>"
517 f"<tbody>{rows}</tbody></table>"
518 if t.activities
519 else '<p class="note">No spans in the window.</p>'
520 )
521 last = escape(_ago(t.last_activity, now))
522 summary = (
523 f'<p class="note">{t.total_spans} spans · {t.error_spans} '
524 f"errored · last {last} · {t.window_days}-day window.</p>"
525 )
526 errs = "".join(
527 f'<p class="note bad mono">{escape(e)}</p>' for e in t.recent_errors
528 )
529 return _rail(
530 "Run telemetry", f"{t.total_spans} spans", summary + table + errs
531 )
534def _failures_rail(model: DashboardModel) -> str:
535 if not model.failures:
536 return ""
537 now = model.generated_at
538 real = [f for f in model.failures if not f.intentional]
539 intentional = [f for f in model.failures if f.intentional]
541 def _rows(items: list) -> str: # type: ignore[type-arg]
542 out = ""
543 for f in items:
544 cls = "bad" if not f.intentional else "mut"
545 out += (
546 f'<tr><td class="mono">{escape(f.workflow_id[:18])}</td>'
547 f'<td class="{cls}">{escape(f.kind)}</td>'
548 f'<td class="mut">{escape(f.reason or "—")}</td>'
549 f'<td class="mut">{escape(_ago(f.when, now))}</td></tr>'
550 )
551 return out
553 head = "<tr><th>workflow</th><th>kind</th><th>reason</th><th>when</th></tr>"
554 real_html = (
555 f"<table><thead>{head}</thead><tbody>{_rows(real)}</tbody></table>"
556 if real
557 else '<p class="note">No real failures — nothing broke.</p>'
558 )
559 intentional_html = ""
560 if intentional:
561 plural = "s" if len(intentional) != 1 else ""
562 intentional_html = (
563 f'<p class="note">{len(intentional)} intentional go-live / '
564 f"re-test reset{plural} (housekeeping):</p>"
565 f"<table><tbody>{_rows(intentional)}</tbody></table>"
566 )
567 extra = f" · {len(intentional)} resets" if intentional else ""
568 count = f"{len(real)} real{extra}"
569 return _rail("Failures & resets", count, real_html + intentional_html)
572def _dispatch_rail(model: DashboardModel) -> str:
573 d = model.dispatch
574 plural = "es" if d.total != 1 else ""
575 inner = (
576 f'<p class="note">{d.total} inbound dispatch{plural} currently '
577 "retained in Temporal "
578 f"(→ {d.transaction} transaction · {d.offer} offer · {d.receipt} "
579 f"receipt · {d.command} command · {d.quarantine} quarantined · "
580 f"{d.ignore} ignored). Replies are processed continuously; this panel "
581 "only counts what Temporal still holds, so it reads low after "
582 "retention ages dispatches out — the telemetry above is the fuller "
583 "record.</p>"
584 )
585 return _rail("Inbound · W3 dispatch", f"{d.total} retained", inner)
588def _deploy_rail(model: DashboardModel) -> str:
589 if not model.deploy.prs:
590 return ""
591 now = model.generated_at
592 rows = ""
593 for p in model.deploy.prs:
594 state_cls = _STATE_CLASS.get(p.state, "mut")
595 rows += (
596 f'<tr><td><a href="{escape(p.url, quote=True)}">#{p.number}</a>'
597 f"</td><td>{escape(p.title)}</td>"
598 f'<td class="{state_cls}">{escape(p.state)}</td>'
599 f"<td>{_ci_cell(p.ci)}</td>"
600 f'<td class="mut">{escape(_ago(p.when, now))}</td></tr>'
601 )
602 table = (
603 "<table><thead><tr><th>pr</th><th>title</th><th>state</th><th>ci</th>"
604 f"<th>opened</th></tr></thead><tbody>{rows}</tbody></table>"
605 )
606 return _rail("Deploy · GitHub", f"{len(model.deploy.prs)} PRs", table)
609def _ci_cell(ci: str | None) -> str:
610 if ci is None:
611 return '<span class="mut">—</span>'
612 return f'<span class="{_CI_CLASS.get(ci, "mut")}">{escape(ci)}</span>'
615def _footer() -> str:
616 return (
617 "<footer><b>Safety envelope.</b> Every auto-apply passes the hard "
618 "floor (amount ceiling, unreadable amount, per-run/day breaker) and "
619 "the earned-autonomy gate (exactly one blessed rule), then a "
620 "clean-context model <b>safety review</b> that can only hold it back. "
621 "Autonomy is revoked — the rule demoted — on an explicit correction or "
622 "a silent in-YNAB edit. Everything above is derived on this request "
623 "from Temporal, YNAB, ClickHouse, AgentMail, and GitHub; nothing is "
624 "stored. Reload to recompute.</footer>"
625 )
628def page(model: DashboardModel) -> str:
629 """Render the whole dashboard as one self-contained HTML document."""
630 parts = (
631 _masthead(model),
632 _narrative(model),
633 _needs_you(model),
634 _working(model),
635 _activity(model),
636 _budget(model),
637 _autonomy_rail(model),
638 _telemetry_rail(model),
639 _failures_rail(model),
640 _dispatch_rail(model),
641 _deploy_rail(model),
642 _footer(),
643 )
644 return (
645 '<!doctype html><html lang="en"><head><meta charset="utf-8">'
646 '<meta name="viewport" content="width=device-width,initial-scale=1">'
647 "<title>ynab-agent · ops</title>"
648 f"<style>{_CSS}</style></head><body><main>"
649 + "".join(parts)
650 + "</main></body></html>"
651 )
_qrow is representative of the render's two safety habits. Every dynamic value — the question, amount, category, payee — is escape()'d at the boundary, and the row reads even with no facts (it falls back through q.label, which itself fell back to a short id upstream).
Escape-at-the-boundary on every interpolation; graceful fallback.
src/ynab_agent/dashboard/render.py · 651 lines
src/ynab_agent/dashboard/render.py
⋯ 305 lines hidden (lines 1–305)
1"""Render the view-model to one self-contained HTML page (pure).
3All CSS is inline, there is no JavaScript, and the page makes no network request
4of its own — it is a static projection of an already-computed
5:class:`~ynab_agent.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary.
8The page is organized around the operator's questions, not the subsystems the
9data came from: a masthead verdict, a plain-English **state of things**, then
10**needs you** (the only acted-upon queue), **is it working** (health + flow),
11**what it's done** (the conversation feed), the **budget**, and finally the
12operator detail folded into progressive-disclosure rails. Structure comes from
13hairlines, whitespace, type, and a single accent — not boxes; semantic colour
14lives in dots and pills, not fills.
15"""
17from __future__ import annotations
19from datetime import UTC, datetime
20from html import escape
21from typing import TYPE_CHECKING
23if TYPE_CHECKING:
24 from ynab_agent.dashboard.model import (
25 DashboardModel,
26 Lifecycle,
27 QueueItem,
28 RunTelemetry,
29 )
31_CSS = """
32:root{--fg:#1a1a1a;--mut:#6b6b6b;--faint:#8a8a8a;--line:#e6e6e6;--bg:#fff;
33--ok:#1a7f37;--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#f6f8fa;
34--serif:"Iowan Old Style","Palatino Linotype",Palatino,Georgia,ui-serif,serif}
35@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
36--faint:#7a7a7a;--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;
37--bad:#f85149;--accent:#58a6ff;--card:#161b22}}
38*{box-sizing:border-box}
39body{margin:0;background:var(--bg);color:var(--fg);
40font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
41main{max-width:820px;margin:0 auto;padding:30px 20px 72px}
42a{color:var(--accent);text-decoration:none}a:hover{text-decoration:underline}
43.mut{color:var(--mut)}.faint{color:var(--faint)}
44.ok{color:var(--ok)}.warn{color:var(--warn)}.bad{color:var(--bad)}
45.accent{color:var(--accent)}
46.mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
47.dot{display:inline-block;width:9px;height:9px;border-radius:50%;flex:none;
48vertical-align:baseline}
49.dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
50.dot.bad{background:var(--bad)}.dot.mute{background:var(--faint)}
51.dot.accent{background:var(--accent)}
53/* Masthead — wordmark + the one composite verdict + the few facts. */
54.mast{display:flex;align-items:flex-start;gap:14px;flex-wrap:wrap}
55.mast h1{font-size:19px;margin:0;letter-spacing:-.01em;font-weight:650}
56.mast .tag{color:var(--mut);font-size:12px;margin:2px 0 0}
57.verdict{display:inline-flex;align-items:center;gap:7px;font-weight:650;
58font-size:15px}
59.verdict .dot{width:11px;height:11px}
60.verdict.ok{color:var(--ok)}.verdict.warn{color:var(--warn)}
61.verdict.bad{color:var(--bad)}
62.facts{margin-left:auto;text-align:right;color:var(--mut);font-size:12px;
63line-height:1.7}
64.chips{display:flex;flex-wrap:wrap;gap:7px;margin:12px 0 0}
65.chip{display:inline-flex;align-items:center;gap:6px;padding:3px 10px;
66border-radius:999px;border:1px solid var(--line);font-size:12px;font-weight:550}
67.chip.ok{border-color:transparent;color:var(--ok);
68background:color-mix(in srgb,var(--ok) 14%,transparent)}
69.chip.warn{border-color:transparent;color:var(--warn);
70background:color-mix(in srgb,var(--warn) 16%,transparent)}
71.chip.bad{border-color:transparent;color:var(--bad);
72background:color-mix(in srgb,var(--bad) 14%,transparent)}
73.srcs{display:flex;flex-wrap:wrap;gap:14px;margin:14px 0 0;font-size:11px;
74color:var(--mut)}
75.srcs .dot{margin-right:5px}
77/* Section = an 11px uppercase label with a hairline underline. */
78section{margin:30px 0 0}
79h2{font-size:11px;text-transform:uppercase;letter-spacing:.09em;
80color:var(--mut);margin:0 0 13px;font-weight:650;padding-bottom:6px;
81border-bottom:1px solid var(--line);display:flex;align-items:baseline;gap:8px}
82h2 .h2-aside{margin-left:auto;font-size:11px;font-weight:500;letter-spacing:0;
83text-transform:none;color:var(--faint)}
85/* State of things — the one serif voice, a lead with an accent rule. */
86.narr{border-left:3px solid var(--line);padding:2px 0 2px 15px;margin:26px 0 0}
87.narr.ok{border-color:var(--ok)}.narr.warn{border-color:var(--warn)}
88.narr.bad{border-color:var(--bad)}
89.narr .head{font-weight:650;font-size:15px;margin:0 0 4px}
90.narr .body{font-family:var(--serif);font-size:16.5px;line-height:1.5;
91margin:0;color:var(--fg);max-width:62ch}
93/* Needs you — the humanized queue. */
94.qrow{display:flex;align-items:baseline;gap:11px;padding:9px 0;
95border-bottom:1px solid var(--line)}
96.qrow:last-child{border-bottom:0}
97.qrow .dot{position:relative;top:1px}
98.q-main{flex:1;min-width:0}
99.q-title{display:block;font-weight:550;overflow-wrap:anywhere}
100.q-sub{display:block;color:var(--mut);font-size:12.5px;margin-top:1px}
101.q-age{color:var(--faint);font-size:12px;white-space:nowrap;
102font-variant-numeric:tabular-nums}
103.tag-pill{display:inline-block;padding:0 7px;border-radius:10px;
104font-size:10.5px;font-weight:650;text-transform:uppercase;letter-spacing:.04em;
105background:var(--card);color:var(--mut);vertical-align:1px}
106.tag-pill.offer{color:var(--accent);
107background:color-mix(in srgb,var(--accent) 14%,transparent)}
108.empty{display:flex;align-items:center;gap:10px;color:var(--mut);
109font-size:14px;padding:6px 0}
110.empty .big{font-size:20px;line-height:1}
111.note{color:var(--mut);font-size:12.5px;margin:11px 0 0}
113/* Is it working — health pills + a single flow bar. */
114.kpis{display:flex;flex-wrap:wrap;gap:18px;margin:0 0 16px}
115.kpi{display:inline-flex;align-items:baseline;gap:7px;font-size:13px}
116.kpi b{font-weight:600}
117.flow{display:flex;height:22px;border-radius:5px;overflow:hidden;
118margin:4px 0 10px;background:var(--card)}
119.flow>span{display:block;min-width:2px}
120.legend{display:flex;flex-wrap:wrap;gap:7px 16px;font-size:12px;
121color:var(--mut)}
122.legend .it{display:inline-flex;align-items:baseline;gap:6px}
123.legend b{color:var(--fg);font-weight:600;font-variant-numeric:tabular-nums}
125/* What it's done — the conversation feed. */
126.feed{display:flex;flex-direction:column}
127.frow{display:flex;align-items:baseline;gap:10px;padding:8px 0;
128border-bottom:1px solid var(--line)}
129.frow:last-child{border-bottom:0}
130.frow .g{width:1.2em;text-align:center;flex:none;color:var(--faint)}
131.f-main{flex:1;min-width:0}
132.f-title{display:block;overflow-wrap:anywhere}
133.f-prev{display:block;color:var(--faint);font-size:12px;margin-top:1px;
134white-space:nowrap;overflow:hidden;text-overflow:ellipsis}
135.f-age{color:var(--faint);font-size:12px;white-space:nowrap}
137/* Budget — one tight line + overspent inline. */
138.over{display:flex;flex-wrap:wrap;gap:6px 14px;margin:8px 0 0;font-size:13px}
139.over .it{display:inline-flex;gap:6px}
141/* Rails — progressive disclosure, hairline-topped. */
142details.rail{border-top:1px solid var(--line);padding:9px 0 2px}
143details.rail>summary{cursor:pointer;list-style:none;display:flex;
144align-items:center;gap:8px;font-size:12px;color:var(--mut)}
145details.rail>summary::-webkit-details-marker{display:none}
146details.rail>summary::before{content:"\\25B8";color:var(--faint);font-size:10px}
147details.rail[open]>summary::before{content:"\\25BE"}
148details.rail>summary .rail-n{margin-left:auto;color:var(--faint)}
149.rail-body{padding:12px 0 6px}
150table{width:100%;border-collapse:collapse;font-size:12.5px}
151th,td{text-align:left;padding:5px 12px 5px 0;vertical-align:top;
152border-bottom:1px solid var(--line)}
153th{color:var(--mut);font-weight:600;font-size:10.5px;text-transform:uppercase;
154letter-spacing:.04em}
155.num{text-align:right;font-variant-numeric:tabular-nums}
156footer{margin:40px 0 0;padding-top:15px;border-top:1px solid var(--line);
157color:var(--mut);font-size:12px;line-height:1.7}
158footer b{color:var(--fg);font-weight:600}
159"""
161_LIFECYCLE_ORDER = (
162 "discovered",
163 "enriching",
164 "hold_amazon",
165 "revising",
166 "awaiting_human",
167 "open",
168 "lapsed",
169)
170# Each in-flight state's human label + flow colour (a cool ramp: accent =
171# the agent is working, warn = it's on you, ok = settled-and-resting).
172_STATE_META = {
173 "discovered": ("discovered", "var(--faint)"),
174 "enriching": ("enriching", "var(--accent)"),
175 "hold_amazon": ("holding for Amazon detail", "var(--faint)"),
176 "revising": ("revising", "var(--accent)"),
177 "awaiting_human": ("awaiting you", "var(--warn)"),
178 "open": ("resting (done)", "var(--ok)"),
179 "lapsed": ("lapsed", "var(--faint)"),
180}
181_TRUST_CLASS = {"trusted": "ok", "confirmed": "warn", "suggested": "mute"}
182_CI_CLASS = {"passed": "ok", "failed": "bad", "running": "warn"}
183_STATE_CLASS = {"merged": "ok", "open": "warn", "closed": "mut"}
184_KIND_GLYPH = {"proposal": "✉", "offer": "◆", "thread": "·"}
187def _aware(when: datetime) -> datetime:
188 return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
191def _ago(when: datetime | None, now: datetime) -> str:
192 if when is None:
193 return "—"
194 secs = (now - _aware(when)).total_seconds()
195 if secs < 90:
196 return "just now"
197 if secs < 5400:
198 return f"{int(secs // 60)}m ago"
199 if secs < 129600:
200 return f"{int(secs // 3600)}h ago"
201 return f"{int(secs // 86400)}d ago"
204def _dot(kind: str) -> str:
205 return f'<span class="dot {kind}"></span>'
208def _meta(state: str) -> tuple[str, str]:
209 return _STATE_META.get(state, (state, "var(--faint)"))
212# ── Masthead ─────────────────────────────────────────────────────────────────
213def _masthead(model: DashboardModel) -> str:
214 h = model.health
215 now = model.generated_at
216 stamp = now.strftime("%Y-%m-%d %H:%M UTC")
218 chips: list[str] = []
219 if h.needs_you:
220 suffix = "s" if h.needs_you == 1 else ""
221 chips.append(
222 f'<span class="chip warn">{h.needs_you} need{suffix} you</span>'
223 )
224 else:
225 chips.append('<span class="chip ok">✓ all caught up</span>')
226 poll_cls = "ok" if h.poll_live else "bad"
227 poll_age = escape(_ago(h.poll_last_start, now))
228 chips.append(f'<span class="chip">{_dot(poll_cls)} poll {poll_age}</span>')
229 if model.budget.available:
230 chips.append(
231 f'<span class="chip">{model.budget.unapproved} unapproved</span>'
232 )
233 if h.real_failures:
234 suffix = "s" if h.real_failures != 1 else ""
235 chips.append(
236 f'<span class="chip bad">{h.real_failures} failure{suffix}</span>'
237 )
239 src = "".join(
240 f"<span>{_dot('ok' if s.ok else 'bad')}{escape(s.name)} "
241 f'<span class="faint">{escape(s.detail)}</span></span>'
242 for s in model.sources
243 )
244 return (
245 "<header>"
246 '<div class="mast">'
247 "<div><h1>ynab-agent · ops</h1>"
248 '<p class="tag">durable transaction triage · '
249 "earned-autonomy read-model</p></div>"
250 f'<div class="facts"><span class="verdict {escape(h.tone)}">'
251 f"{_dot(h.tone)}{escape(h.label)}</span><br>{escape(model.repo)}<br>"
252 f"{escape(stamp)}</div></div>"
253 f'<div class="chips">{"".join(chips)}</div>'
254 f'<div class="srcs">{src}</div>'
255 "</header>"
256 )
259# ── State of things ──────────────────────────────────────────────────────────
260def _narrative(model: DashboardModel) -> str:
261 n = model.narrative
262 body = " ".join(n.paragraphs)
263 return (
264 f'<section class="narr {escape(n.tone)}">'
265 f'<p class="head">{escape(n.headline)}</p>'
266 f'<p class="body">{escape(body)}</p></section>'
267 )
270# ── Needs you ────────────────────────────────────────────────────────────────
271def _needs_you(model: DashboardModel) -> str:
272 now = model.generated_at
273 queue = model.needs_you
274 if not queue:
275 body = (
276 '<div class="empty"><span class="big">✓</span>'
277 "<span>You're all caught up — nothing is waiting on you."
278 "</span></div>"
279 )
280 else:
281 body = "".join(_qrow(q, now) for q in queue[:30])
282 if model.handled:
283 n = len(model.handled)
284 plural = "s" if n != 1 else ""
285 verb = "are" if n != 1 else "is"
286 body += (
287 f'<p class="note">{n} more proposal{plural} you already settled '
288 f"in YNAB {verb} winding down — they lapse on their own, no reply "
289 "needed.</p>"
290 )
291 aside = (
292 f'<span class="h2-aside">oldest '
293 f"{escape(_ago(_oldest(queue), now))}</span>"
294 if queue
295 else ""
296 )
297 return f"<section><h2>Needs you{aside}</h2>{body}</section>"
300def _oldest(queue: tuple[QueueItem, ...]) -> datetime | None:
301 # Normalize to aware before min() — mixed aware/naive comparison raises.
302 sinces = [_aware(q.since) for q in queue if q.since is not None]
303 return min(sinces) if sinces else None
306def _qrow(q: QueueItem, now: datetime) -> str:
307 dot = "accent" if q.kind == "offer" else "warn"
308 title = escape(q.question or q.label)
309 pill = (
310 '<span class="tag-pill offer">offer</span> '
311 if q.kind == "offer"
312 else ""
313 )
314 sub_bits: list[str] = []
315 if q.amount:
316 sub_bits.append(escape(q.amount))
317 if q.category:
318 sub_bits.append(escape(q.category))
319 if not sub_bits and q.payee:
320 sub_bits.append(escape(q.payee))
321 sub = (
322 f'<span class="q-sub">{" · ".join(sub_bits)}</span>'
323 if sub_bits
324 else ""
325 )
326 age = escape(_ago(q.since, now))
327 return (
328 f'<div class="qrow">{_dot(dot)}'
329 f'<span class="q-main"><span class="q-title">{pill}{title}</span>'
330 f'{sub}</span><span class="q-age">{age}</span></div>'
⋯ 321 lines hidden (lines 331–651)
331 )
334# ── Is it working ────────────────────────────────────────────────────────────
335def _working(model: DashboardModel) -> str:
336 h = model.health
337 now = model.generated_at
338 poll_word = escape(h.poll_status) if h.poll_live else "stopped"
339 poll_age = escape(_ago(h.poll_last_start, now))
340 kpis = [
341 f'<span class="kpi">{_dot("ok" if h.poll_live else "bad")}'
342 f'<b>poll</b> <span class="mut">{poll_word} '
343 f"· {poll_age}</span></span>"
344 ]
345 if h.worker_last_span is not None:
346 wago = (now - _aware(h.worker_last_span)).total_seconds()
347 wdot = "ok" if wago < 3600 else "warn"
348 span_age = escape(_ago(h.worker_last_span, now))
349 kpis.append(
350 f'<span class="kpi">{_dot(wdot)}<b>worker</b> '
351 f'<span class="mut">span {span_age}</span></span>'
352 )
353 else:
354 kpis.append(
355 f'<span class="kpi">{_dot("mute")}<b>worker</b> '
356 '<span class="mut">no telemetry</span></span>'
357 )
358 if h.span_error_rate is not None:
359 rate = h.span_error_rate
360 rdot = "ok" if rate <= 0.05 else "warn"
361 kpis.append(
362 f'<span class="kpi">{_dot(rdot)}<b>{rate * 100:.1f}%</b> '
363 '<span class="mut">span errors</span></span>'
364 )
365 return (
366 "<section><h2>Is it working?</h2>"
367 f'<div class="kpis">{"".join(kpis)}</div>'
368 f"{_flow(model.lifecycle)}</section>"
369 )
372def _flow(lc: Lifecycle) -> str:
373 counts = {s.state: s.count for s in lc.states}
374 ordered = [s for s in _LIFECYCLE_ORDER if counts.get(s)]
375 ordered += [
376 s for s in sorted(counts) if counts.get(s) and s not in _LIFECYCLE_ORDER
377 ]
378 if not ordered:
379 return '<p class="empty"><span>No transactions in flight.</span></p>'
380 segs = "".join(
381 f'<span style="flex:{counts[s]};background:{_meta(s)[1]}" '
382 f'title="{escape(_meta(s)[0])}: {counts[s]}"></span>'
383 for s in ordered
384 )
385 legend = "".join(
386 '<span class="it"><span class="dot" '
387 f'style="background:{_meta(s)[1]}"></span>'
388 f"<b>{counts[s]}</b> {escape(_meta(s)[0])}</span>"
389 for s in ordered
390 )
391 note = (
392 " — the post-write window runs for weeks, so resting transactions "
393 "haven't closed out yet"
394 if lc.archived == 0 and counts.get("open")
395 else ""
396 )
397 gloss = (
398 ' · <span class="faint">'
399 f"{lc.archived} archived{note} · "
400 f"{lc.terminated} terminated (recent)</span>"
401 )
402 return (
403 f'<div class="flow">{segs}</div>'
404 f'<div class="legend">{legend}{gloss}</div>'
405 )
408# ── What it's done ───────────────────────────────────────────────────────────
409def _activity(model: DashboardModel) -> str:
410 now = model.generated_at
411 if not model.conversations:
412 return ""
413 rows = ""
414 for c in model.conversations[:10]:
415 glyph = _KIND_GLYPH.get(c.kind, "·")
416 prev = (
417 f'<span class="f-prev">{escape(c.preview)}</span>'
418 if c.preview
419 else ""
420 )
421 age = escape(_ago(c.updated_at, now))
422 rows += (
423 f'<div class="frow"><span class="g">{glyph}</span>'
424 f'<span class="f-main"><span class="f-title">'
425 f"{escape(c.subject)}</span>{prev}</span>"
426 f'<span class="f-age">{age}</span></div>'
427 )
428 return (
429 "<section><h2>What it's done"
430 '<span class="h2-aside">recent threads</span></h2>'
431 f'<div class="feed">{rows}</div></section>'
432 )
435# ── Budget ───────────────────────────────────────────────────────────────────
436def _budget(model: DashboardModel) -> str:
437 b = model.budget
438 if not b.available:
439 return ""
440 if b.overspent:
441 items = "".join(
442 f'<span class="it"><span class="bad mono">{escape(c.balance)}'
443 f"</span> {escape(c.name)}</span>"
444 for c in b.overspent
445 )
446 over = f'<div class="over">{items}</div>'
447 else:
448 over = '<p class="note">No overspent categories.</p>'
449 noun = "y" if len(b.overspent) == 1 else "ies"
450 head = (
451 f"{b.unapproved} unapproved · "
452 f"{len(b.overspent)} overspent categor{noun}"
453 )
454 return (
455 "<section><h2>Budget · YNAB"
456 f'<span class="h2-aside">{escape(head)}</span></h2>{over}</section>'
457 )
460# ── Rails (progressive disclosure) ───────────────────────────────────────────
461def _rail(title: str, count: str, inner: str) -> str:
462 return (
463 f'<details class="rail"><summary>{escape(title)}'
464 f'<span class="rail-n">{escape(count)}</span></summary>'
465 f'<div class="rail-body">{inner}</div></details>'
466 )
469def _autonomy_rail(model: DashboardModel) -> str:
470 a = model.autonomy
471 note = (
472 '<p class="note">Earned then granted: a rule becomes <b>eligible</b> '
473 "after K consistent confirms, and <b>blessed</b> (auto-applies) only "
474 "once you accept the offer. A correction demotes it to observe.</p>"
475 )
476 if a.rules:
477 rows = ""
478 for r in a.rules[:30]:
479 trust_cls = _TRUST_CLASS.get(r.trust, "mut")
480 rows += (
481 f"<tr><td>{escape(r.payee)}</td>"
482 f'<td class="mut">{escape(r.category)}</td>'
483 f'<td class="{trust_cls}">{escape(r.trust)}</td>'
484 f'<td class="num">{r.hits}</td></tr>'
485 )
486 table = (
487 "<table><thead><tr><th>payee</th><th>category</th><th>trust</th>"
488 "<th class='num'>hits</th></tr></thead>"
489 f"<tbody>{rows}</tbody></table>"
490 )
491 else:
492 table = '<p class="note">No learned rules yet.</p>'
493 count = f"{a.observe} observe · {a.eligible} eligible · {a.blessed} blessed"
494 return _rail("Autonomy ladder", count, table + note)
497def _telemetry_rail(model: DashboardModel) -> str:
498 t: RunTelemetry = model.telemetry
499 now = model.generated_at
500 if not t.available:
501 return _rail(
502 "Run telemetry",
503 "off",
504 '<p class="note">ClickHouse not configured or unreachable; '
505 "Temporal carries the page regardless.</p>",
506 )
507 rows = "".join(
508 f'<tr><td class="mono">{escape(a.name)}</td>'
509 f'<td class="num">{a.count}</td>'
510 f'<td class="num mut">{a.avg_ms:.0f}</td>'
511 f'<td class="num mut">{a.max_ms:.0f}</td></tr>'
512 for a in t.activities
513 )
514 table = (
515 "<table><thead><tr><th>activity</th><th class='num'>runs</th>"
516 "<th class='num'>avg ms</th><th class='num'>max ms</th></tr></thead>"
517 f"<tbody>{rows}</tbody></table>"
518 if t.activities
519 else '<p class="note">No spans in the window.</p>'
520 )
521 last = escape(_ago(t.last_activity, now))
522 summary = (
523 f'<p class="note">{t.total_spans} spans · {t.error_spans} '
524 f"errored · last {last} · {t.window_days}-day window.</p>"
525 )
526 errs = "".join(
527 f'<p class="note bad mono">{escape(e)}</p>' for e in t.recent_errors
528 )
529 return _rail(
530 "Run telemetry", f"{t.total_spans} spans", summary + table + errs
531 )
534def _failures_rail(model: DashboardModel) -> str:
535 if not model.failures:
536 return ""
537 now = model.generated_at
538 real = [f for f in model.failures if not f.intentional]
539 intentional = [f for f in model.failures if f.intentional]
541 def _rows(items: list) -> str: # type: ignore[type-arg]
542 out = ""
543 for f in items:
544 cls = "bad" if not f.intentional else "mut"
545 out += (
546 f'<tr><td class="mono">{escape(f.workflow_id[:18])}</td>'
547 f'<td class="{cls}">{escape(f.kind)}</td>'
548 f'<td class="mut">{escape(f.reason or "—")}</td>'
549 f'<td class="mut">{escape(_ago(f.when, now))}</td></tr>'
550 )
551 return out
553 head = "<tr><th>workflow</th><th>kind</th><th>reason</th><th>when</th></tr>"
554 real_html = (
555 f"<table><thead>{head}</thead><tbody>{_rows(real)}</tbody></table>"
556 if real
557 else '<p class="note">No real failures — nothing broke.</p>'
558 )
559 intentional_html = ""
560 if intentional:
561 plural = "s" if len(intentional) != 1 else ""
562 intentional_html = (
563 f'<p class="note">{len(intentional)} intentional go-live / '
564 f"re-test reset{plural} (housekeeping):</p>"
565 f"<table><tbody>{_rows(intentional)}</tbody></table>"
566 )
567 extra = f" · {len(intentional)} resets" if intentional else ""
568 count = f"{len(real)} real{extra}"
569 return _rail("Failures & resets", count, real_html + intentional_html)
572def _dispatch_rail(model: DashboardModel) -> str:
573 d = model.dispatch
574 plural = "es" if d.total != 1 else ""
575 inner = (
576 f'<p class="note">{d.total} inbound dispatch{plural} currently '
577 "retained in Temporal "
578 f"(→ {d.transaction} transaction · {d.offer} offer · {d.receipt} "
579 f"receipt · {d.command} command · {d.quarantine} quarantined · "
580 f"{d.ignore} ignored). Replies are processed continuously; this panel "
581 "only counts what Temporal still holds, so it reads low after "
582 "retention ages dispatches out — the telemetry above is the fuller "
583 "record.</p>"
584 )
585 return _rail("Inbound · W3 dispatch", f"{d.total} retained", inner)
588def _deploy_rail(model: DashboardModel) -> str:
589 if not model.deploy.prs:
590 return ""
591 now = model.generated_at
592 rows = ""
593 for p in model.deploy.prs:
594 state_cls = _STATE_CLASS.get(p.state, "mut")
595 rows += (
596 f'<tr><td><a href="{escape(p.url, quote=True)}">#{p.number}</a>'
597 f"</td><td>{escape(p.title)}</td>"
598 f'<td class="{state_cls}">{escape(p.state)}</td>'
599 f"<td>{_ci_cell(p.ci)}</td>"
600 f'<td class="mut">{escape(_ago(p.when, now))}</td></tr>'
601 )
602 table = (
603 "<table><thead><tr><th>pr</th><th>title</th><th>state</th><th>ci</th>"
604 f"<th>opened</th></tr></thead><tbody>{rows}</tbody></table>"
605 )
606 return _rail("Deploy · GitHub", f"{len(model.deploy.prs)} PRs", table)
609def _ci_cell(ci: str | None) -> str:
610 if ci is None:
611 return '<span class="mut">—</span>'
612 return f'<span class="{_CI_CLASS.get(ci, "mut")}">{escape(ci)}</span>'
615def _footer() -> str:
616 return (
617 "<footer><b>Safety envelope.</b> Every auto-apply passes the hard "
618 "floor (amount ceiling, unreadable amount, per-run/day breaker) and "
619 "the earned-autonomy gate (exactly one blessed rule), then a "
620 "clean-context model <b>safety review</b> that can only hold it back. "
621 "Autonomy is revoked — the rule demoted — on an explicit correction or "
622 "a silent in-YNAB edit. Everything above is derived on this request "
623 "from Temporal, YNAB, ClickHouse, AgentMail, and GitHub; nothing is "
624 "stored. Reload to recompute.</footer>"
625 )
628def page(model: DashboardModel) -> str:
629 """Render the whole dashboard as one self-contained HTML document."""
630 parts = (
631 _masthead(model),
632 _narrative(model),
633 _needs_you(model),
634 _working(model),
635 _activity(model),
636 _budget(model),
637 _autonomy_rail(model),
638 _telemetry_rail(model),
639 _failures_rail(model),
640 _dispatch_rail(model),
641 _deploy_rail(model),
642 _footer(),
643 )
644 return (
645 '<!doctype html><html lang="en"><head><meta charset="utf-8">'
646 '<meta name="viewport" content="width=device-width,initial-scale=1">'
647 "<title>ynab-agent · ops</title>"
648 f"<style>{_CSS}</style></head><body><main>"
649 + "".join(parts)
650 + "</main></body></html>"
651 )
Wiring and verification
build_html keeps the parallel fan-out of the five sources, then adds one dependent hop: the awaiting ids are known only after Temporal answers, so it resolves them to YNAB facts and hands the map to the pure assemble. If YNAB is off the map is empty and the queue still renders.
Parallel fan-out, then the one dependent resolve; failure stays isolated.
src/ynab_agent/dashboard/server.py · 178 lines
src/ynab_agent/dashboard/server.py
⋯ 47 lines hidden (lines 1–47)
1"""The dashboard HTTP server — a tiny dependency-free asyncio responder.
3Runs on the worker's own event loop (same process, same pod) and reuses the
4worker's connected Temporal client. On each GET it fans the source readers out
5concurrently, assembles the view, and renders one self-contained page; it stores
6nothing between requests. GET-only, ``Connection: close``, ``Cache-Control:
7no-store`` — built for a single viewer behind ``kubectl port-forward``, not the
8public internet. Every reader degrades to an error string, so a source being
9down yields a page with a red dot, never a crash.
10"""
12from __future__ import annotations
14import asyncio
15import contextlib
16import logging
17from datetime import UTC, datetime
18from typing import TYPE_CHECKING, Final
20from ynab_agent.dashboard import (
21 agentmail_source,
22 clickhouse_source,
23 github_source,
24 read_model,
25 render,
26 temporal_source,
27 ynab_source,
28)
29from ynab_agent.settings import DashboardSettings, GitHubSettings
31if TYPE_CHECKING:
32 from temporalio.client import Client
34_log = logging.getLogger("ynab_agent.dashboard")
36_READ_TIMEOUT: Final = 15.0
37_MAX_HEAD_BYTES: Final = 64 * 1024
40def _repo() -> str:
41 """The watched repo, degrading to a default if config is unreadable."""
42 try:
43 return GitHubSettings().repo
44 except Exception: # never let a config read fail the page
45 return "mseeks/ynab-agent"
48async def build_html(client: Client) -> str:
49 """Derive the whole view live and render it (the per-request work)."""
50 now = datetime.now(UTC)
51 temporal, clickhouse, ynab, agentmail, github = await asyncio.gather(
52 temporal_source.fetch(client),
53 clickhouse_source.fetch(),
54 ynab_source.fetch(),
55 agentmail_source.fetch(),
56 github_source.fetch(),
57 )
58 # A dependent second hop: the awaiting ids are known only after Temporal
59 # answers, so resolve them to YNAB facts now (best-effort; an empty map just
60 # leaves the queue showing short ids — see ynab_source.resolve_queue).
61 readout, _ = temporal
62 queue_facts = await ynab_source.resolve_queue(
63 tuple(item.ident for item in readout.awaiting)
64 )
65 model = read_model.assemble(
66 now=now,
67 repo=_repo(),
68 temporal=temporal,
69 clickhouse=clickhouse,
70 ynab=ynab,
71 agentmail=agentmail,
72 github=github,
⋯ 106 lines hidden (lines 73–178)
73 queue_facts=queue_facts,
74 )
75 return render.page(model)
78def _parse_path(request_line: bytes) -> tuple[str, str]:
79 """Return ``(method, path)`` from a raw HTTP request line."""
80 parts = request_line.decode("latin-1", "replace").split()
81 if len(parts) < 2:
82 return "", "/"
83 return parts[0].upper(), parts[1].split("?", 1)[0]
86async def _drain_headers(reader: asyncio.StreamReader) -> None:
87 """Read and discard the request headers, bounded in size and time."""
88 total = 0
89 while True:
90 line = await asyncio.wait_for(reader.readline(), _READ_TIMEOUT)
91 total += len(line)
92 if line in (b"\r\n", b"\n", b"") or total > _MAX_HEAD_BYTES:
93 return
96async def _respond(
97 writer: asyncio.StreamWriter,
98 status: str,
99 content_type: str,
100 body: bytes,
101) -> None:
102 """Write a complete HTTP/1.1 response and close the connection."""
103 head = (
104 f"HTTP/1.1 {status}\r\n"
105 f"Content-Type: {content_type}\r\n"
106 f"Content-Length: {len(body)}\r\n"
107 "Cache-Control: no-store\r\n"
108 "Connection: close\r\n\r\n"
109 ).encode("latin-1")
110 writer.write(head + body)
111 await writer.drain()
114async def _handle(
115 reader: asyncio.StreamReader,
116 writer: asyncio.StreamWriter,
117 client: Client,
118) -> None:
119 """Serve one request: route a GET to the dashboard, else 404/405/500."""
120 try:
121 request_line = await asyncio.wait_for(reader.readline(), _READ_TIMEOUT)
122 await _drain_headers(reader)
123 method, path = _parse_path(request_line)
124 if method != "GET":
125 await _respond(writer, "405 Method Not Allowed", "text/plain", b"")
126 elif path in ("/", "/index.html", "/dashboard"):
127 html = await build_html(client)
128 await _respond(
129 writer, "200 OK", "text/html; charset=utf-8", html.encode()
130 )
131 elif path == "/healthz":
132 await _respond(writer, "200 OK", "text/plain", b"ok")
133 else:
134 await _respond(writer, "404 Not Found", "text/plain", b"")
135 except (TimeoutError, ConnectionError):
136 pass
137 except Exception as exc: # never let a request take the worker down
138 _log.exception("dashboard request failed")
139 with contextlib.suppress(Exception):
140 await _respond(
141 writer,
142 "500 Internal Server Error",
143 "text/html; charset=utf-8",
144 _error_html(exc).encode(),
145 )
146 finally:
147 with contextlib.suppress(Exception):
148 writer.close()
149 await writer.wait_closed()
152def _error_html(exc: Exception) -> str:
153 """A minimal error page (the detail is for a human at a port-forward)."""
154 from html import escape
156 return (
157 "<!doctype html><meta charset=utf-8>"
158 "<title>ynab-agent · error</title>"
159 '<body style="font:15px system-ui;max-width:640px;margin:40px auto">'
160 "<h1>dashboard error</h1><p>Could not build the read-model.</p>"
161 f"<pre>{escape(type(exc).__name__)}: {escape(str(exc))}</pre></body>"
162 )
165async def start(client: Client) -> asyncio.Server:
166 """Start the dashboard server on the configured host/port (returns it)."""
167 settings = DashboardSettings()
169 async def on_connect(
170 reader: asyncio.StreamReader, writer: asyncio.StreamWriter
171 ) -> None:
172 await _handle(reader, writer, client)
174 server = await asyncio.start_server(
175 on_connect, host=settings.host, port=settings.port
176 )
177 _log.info("dashboard listening on %s:%d", settings.host, settings.port)
178 return server
The behavior that matters is tested directly: the needs-you / handled split keyed on approved, the health verdict across its branches, and the read-only resolve's flagging of an already-approved transaction.
The split (approved → handled; unknown → needs you) and the verdict branches.
tests/dashboard/test_read_model.py · 297 lines
tests/dashboard/test_read_model.py
⋯ 92 lines hidden (lines 1–92)
1"""``assemble`` composes the view and turns errors into source dots."""
3from __future__ import annotations
5from datetime import UTC, datetime
7from ynab_agent.dashboard import read_model
8from ynab_agent.dashboard.model import (
9 ActivityStat,
10 Budget,
11 Conversation,
12 DashboardModel,
13 Deploy,
14 Failure,
15 OfferRow,
16 QueueItem,
17 RunTelemetry,
18 StateCount,
19 TxnFacts,
20)
21from ynab_agent.dashboard.temporal_source import TemporalReadout
23_NOW = datetime(2026, 6, 5, 12, 0, tzinfo=UTC)
24_OFF_TEMPORAL: tuple[TemporalReadout, str | None] = (TemporalReadout(), None)
25_OFF_YNAB: tuple[Budget, str | None] = (Budget(available=False), "off")
26_OFF_CH: tuple[RunTelemetry, str | None] = (
27 RunTelemetry(available=False),
28 "off",
29)
30_OFF_MAIL: tuple[tuple[Conversation, ...], str | None] = ((), "off")
31_OFF_GH: tuple[Deploy, str | None] = (Deploy(), "off")
34def _assemble(
35 *,
36 temporal: tuple[TemporalReadout, str | None] = _OFF_TEMPORAL,
37 ynab: tuple[Budget, str | None] = _OFF_YNAB,
38 clickhouse: tuple[RunTelemetry, str | None] = _OFF_CH,
39 agentmail: tuple[tuple[Conversation, ...], str | None] = _OFF_MAIL,
40 github: tuple[Deploy, str | None] = _OFF_GH,
41 queue_facts: dict[str, TxnFacts] | None = None,
42) -> DashboardModel:
43 return read_model.assemble(
44 now=_NOW,
45 repo="mseeks/ynab-agent",
46 temporal=temporal,
47 ynab=ynab,
48 clickhouse=clickhouse,
49 agentmail=agentmail,
50 github=github,
51 queue_facts=queue_facts,
52 )
55def test_errors_become_red_source_dots() -> None:
56 model = _assemble(temporal=(TemporalReadout(), "RPCError: down"))
57 by_name = {s.name: s for s in model.sources}
58 assert by_name["temporal"].ok is False
59 assert by_name["temporal"].detail == "RPCError: down"
60 # "off" is also not-ok (a muted/absent source), never silently green.
61 assert by_name["ynab"].ok is False
62 assert by_name["clickhouse"].ok is False
65def test_healthy_sources_show_a_summary() -> None:
66 model = _assemble(
67 temporal=(TemporalReadout(in_flight=3), None),
68 ynab=(Budget(available=True, unapproved=5), None),
69 agentmail=(
70 (Conversation(subject="s", preview="", kind="thread"),),
71 None,
72 ),
73 )
74 by_name = {s.name: s for s in model.sources}
75 assert by_name["temporal"].ok is True
76 assert by_name["temporal"].detail == "3 in-flight"
77 assert by_name["ynab"].detail == "5 unapproved"
78 assert by_name["agentmail"].detail == "1 threads"
81def test_queue_merges_proposals_and_offers() -> None:
82 readout = TemporalReadout(
83 awaiting=(QueueItem(kind="proposal", label="t1", ident="t1"),),
84 offers=(OfferRow(rule_id="r1", payee="Spotify", status="running"),),
85 )
86 model = _assemble(temporal=(readout, None))
87 kinds = sorted(q.kind for q in model.needs_you)
88 assert kinds == ["offer", "proposal"]
89 offer = next(q for q in model.needs_you if q.kind == "offer")
90 assert offer.label == "Spotify"
93def test_queue_splits_needs_you_from_already_handled() -> None:
94 readout = TemporalReadout(
95 awaiting=(
96 QueueItem(kind="proposal", label="t1", ident="t1"),
97 QueueItem(kind="proposal", label="t2", ident="t2"),
98 QueueItem(kind="proposal", label="t3", ident="t3"),
99 ),
100 )
101 facts = {
102 "t1": TxnFacts(payee="Amazon", amount="$-5.00", approved=False),
103 "t2": TxnFacts(payee="CP Energy", amount="$-61.00", approved=True),
104 # t3 unresolved (YNAB couldn't find it) → defaults to needs-you.
105 }
106 model = _assemble(temporal=(readout, None), queue_facts=facts)
107 assert {q.ident for q in model.needs_you} == {"t1", "t3"}
108 assert {q.ident for q in model.handled} == {"t2"} # approved → winding down
109 row = next(q for q in model.needs_you if q.ident == "t1")
110 assert row.payee == "Amazon"
111 assert row.amount == "$-5.00"
⋯ 65 lines hidden (lines 112–176)
114def test_proposal_borrows_its_thread_subject_as_the_label() -> None:
115 readout = TemporalReadout(
116 awaiting=(QueueItem(kind="proposal", label="t1", ident="t1"),),
117 )
118 convo = Conversation(
119 subject="Amazon — $-5.00 · Shopping?",
120 preview="",
121 kind="proposal",
122 ref="t1",
123 )
124 model = _assemble(temporal=(readout, None), agentmail=((convo,), None))
125 assert model.needs_you[0].question == "Amazon — $-5.00 · Shopping?"
126 assert model.needs_you[0].label == "Amazon — $-5.00 · Shopping?"
129def test_temporal_pieces_slot_into_the_model() -> None:
130 readout = TemporalReadout(
131 poll_status="completed",
132 poll_live=True,
133 observe=4,
134 eligible=1,
135 blessed=2,
136 archived=9,
137 )
138 model = _assemble(temporal=(readout, None))
139 assert model.health.poll_live is True
140 assert model.autonomy.eligible == 1
141 assert model.autonomy.blessed == 2
142 assert model.lifecycle.archived == 9
145def test_narrative_reflects_the_numbers() -> None:
146 awaiting = tuple(
147 QueueItem(kind="proposal", label=f"t{i}", ident=f"t{i}")
148 for i in range(1, 4)
149 )
150 readout = TemporalReadout(
151 poll_live=True,
152 in_flight=10,
153 lifecycle_states=(
154 StateCount(state="open", count=7),
155 StateCount(state="awaiting_human", count=3),
156 ),
157 awaiting=awaiting,
158 )
159 # t1 unapproved (needs you), t2/t3 already settled in YNAB.
160 facts = {
161 "t1": TxnFacts(payee="p", amount="$-1.00", approved=False),
162 "t2": TxnFacts(payee="p", amount="$-1.00", approved=True),
163 "t3": TxnFacts(payee="p", amount="$-1.00", approved=True),
164 }
165 model = _assemble(
166 temporal=(readout, None),
167 ynab=(Budget(available=True, unapproved=0), None),
168 queue_facts=facts,
169 )
170 assert "waiting on you" in model.narrative.headline
171 body = model.narrative.paragraphs[0]
172 assert "caught up in YNAB" in body
173 assert "tracking 10 transactions" in body
174 assert "2 you already settled in YNAB" in body
177def test_health_verdict_branches() -> None:
178 # Poll dead → the worst verdict.
179 m = _assemble(temporal=(TemporalReadout(poll_live=False), None))
180 assert (m.health.tone, m.health.label) == ("bad", "down")
181 # Poll live but YNAB off (the default) → degraded.
182 m = _assemble(temporal=(TemporalReadout(poll_live=True), None))
183 assert (m.health.tone, m.health.label) == ("warn", "degraded")
184 # Poll live + money source reachable → healthy; poll_status threads through.
185 m = _assemble(
186 temporal=(TemporalReadout(poll_live=True, poll_status="running"), None),
187 ynab=(Budget(available=True, unapproved=0), None),
188 )
189 assert (m.health.tone, m.health.label) == ("ok", "healthy")
190 assert m.health.poll_status == "running"
⋯ 107 lines hidden (lines 191–297)
193def test_health_warns_on_high_span_error_rate() -> None:
194 tele = RunTelemetry(
195 available=True, total_spans=1000, error_spans=200, last_activity=_NOW
196 )
197 m = _assemble(
198 temporal=(TemporalReadout(poll_live=True), None),
199 ynab=(Budget(available=True), None),
200 clickhouse=(tele, None),
201 )
202 assert m.health.tone == "warn" # 20% errored is over the 5% floor
205def test_narrative_headline_branches() -> None:
206 # Nothing pending → the calm headline.
207 m = _assemble(
208 temporal=(TemporalReadout(poll_live=True), None),
209 ynab=(Budget(available=True, unapproved=0), None),
210 )
211 assert "caught up" in m.narrative.headline.lower()
212 assert m.narrative.tone == "ok"
213 # A bad verdict overrides everything else.
214 m = _assemble(temporal=(TemporalReadout(poll_live=False), None))
215 assert m.narrative.headline == "Attention needed."
216 assert m.narrative.tone == "bad"
219def test_narrative_includes_offers_and_failures() -> None:
220 readout = TemporalReadout(
221 poll_live=True,
222 offers=(OfferRow(rule_id="r1", payee="Spotify", status="running"),),
223 failures=(
224 Failure(
225 workflow_id="x", kind="failed", reason="boom", intentional=False
226 ),
227 ),
228 )
229 m = _assemble(
230 temporal=(readout, None),
231 ynab=(Budget(available=True, unapproved=0), None),
232 )
233 body = m.narrative.paragraphs[0]
234 assert "autonomy offer" in body
235 assert "failed recently" in body
236 assert any(q.kind == "offer" for q in m.needs_you)
239def test_intentional_resets_stay_out_of_the_real_failure_count() -> None:
240 readout = TemporalReadout(
241 poll_live=True,
242 failures=(
243 Failure(
244 workflow_id="a",
245 kind="failed",
246 reason="timed out",
247 intentional=False,
248 ),
249 Failure(
250 workflow_id="b",
251 kind="terminated",
252 reason="go-live reset",
253 intentional=True,
254 ),
255 ),
256 )
257 m = _assemble(
258 temporal=(readout, None),
259 ynab=(Budget(available=True, unapproved=0), None),
260 )
261 assert m.health.real_failures == 1 # the reset doesn't count
264def test_offer_borrows_thread_subject() -> None:
265 readout = TemporalReadout(
266 offers=(OfferRow(rule_id="r9", payee="Spotify", status="running"),),
267 )
268 convo = Conversation(
269 subject="Trust Spotify → Subscriptions?",
270 preview="",
271 kind="offer",
272 ref="r9",
273 )
274 m = _assemble(temporal=(readout, None), agentmail=((convo,), None))
275 offer = next(q for q in m.needs_you if q.kind == "offer")
276 assert offer.question == "Trust Spotify → Subscriptions?"
277 assert offer.label == "Trust Spotify → Subscriptions?"
280def test_applied_count_drives_the_learning_clause() -> None:
281 tele = RunTelemetry(
282 available=True,
283 total_spans=10,
284 last_activity=_NOW,
285 window_days=3,
286 activities=(
287 ActivityStat(name="commit_to_ynab", count=7, avg_ms=1, max_ms=2),
288 ),
289 )
290 m = _assemble(
291 temporal=(TemporalReadout(poll_live=True, observe=4), None),
292 ynab=(Budget(available=True, unapproved=0), None),
293 clickhouse=(tele, None),
294 )
295 body = m.narrative.paragraphs[0]
296 assert "applied 7 categories in the last 3 days" in body
297 assert "4 rules observed" in body
resolve_queue humanizes, flags approved, and omits the unresolvable.
tests/dashboard/test_simple_sources.py · 256 lines
tests/dashboard/test_simple_sources.py
⋯ 222 lines hidden (lines 1–222)
1"""Source readers: pure helpers + the graceful-degradation 'off' paths."""
3from __future__ import annotations
5import asyncio
6import datetime
7from typing import TYPE_CHECKING
9from ynab_agent.dashboard import (
10 agentmail_source,
11 clickhouse_source,
12 github_source,
13 ynab_source,
14)
16if TYPE_CHECKING:
17 import pytest
19_CH_ENV = ("YNAB_AGENT_CLICKHOUSE_URL",)
20_GH_ENV = ("YNAB_AGENT_GITHUB_TOKEN",)
21_MAIL_ENV = ("AGENTMAIL_API_KEY", "YNAB_AGENT_INBOX")
24# ── ClickHouse ───────────────────────────────────────────────────────────────
25def test_clickhouse_scalar_coercion() -> None:
26 assert clickhouse_source._int("42") == 42
27 assert (
28 clickhouse_source._int("99") == 99
29 ) # JSON quotes a UInt64 as a string
30 assert clickhouse_source._int("nope") == 0
31 assert clickhouse_source._float("1.5") == 1.5
32 assert clickhouse_source._float(None) == 0.0
35def test_clickhouse_datetime_parsing() -> None:
36 parsed = clickhouse_source._ch_datetime("2026-06-05 12:00:00")
37 assert parsed is not None
38 assert parsed.tzinfo is datetime.UTC
39 assert clickhouse_source._ch_datetime("1970-01-01 00:00:00") is None
40 assert clickhouse_source._ch_datetime("") is None
43def test_clickhouse_off_when_unconfigured(
44 monkeypatch: pytest.MonkeyPatch,
45) -> None:
46 for name in _CH_ENV:
47 monkeypatch.delenv(name, raising=False)
48 telemetry, error = asyncio.run(clickhouse_source.fetch())
49 assert error == "off"
50 assert telemetry.available is False
53# ── GitHub ───────────────────────────────────────────────────────────────────
54def test_github_ci_reduction() -> None:
55 assert github_source._ci_of(["success", "success"]) == "passed"
56 assert github_source._ci_of(["success", "failure"]) == "failed"
57 assert github_source._ci_of(["success", None]) == "running"
58 assert github_source._ci_of([]) is None
61def test_github_state() -> None:
62 assert github_source._state({"merged_at": "x"}) == "merged"
63 assert github_source._state({"state": "open"}) == "open"
64 assert github_source._state({"state": "closed"}) == "closed"
67def test_github_off_without_a_token(monkeypatch: pytest.MonkeyPatch) -> None:
68 for name in _GH_ENV:
69 monkeypatch.delenv(name, raising=False)
70 deploy, error = asyncio.run(github_source.fetch())
71 assert error == "off"
72 assert deploy.prs == ()
75# ── AgentMail ────────────────────────────────────────────────────────────────
76def test_agentmail_kind_from_labels() -> None:
77 assert agentmail_source._kind(("yaoffer-r1",)) == "offer"
78 assert agentmail_source._kind(("ynab-agent", "yatxn-t1")) == "proposal"
79 assert agentmail_source._kind(("ynab-agent",)) == "thread"
82def test_agentmail_ref_from_labels() -> None:
83 # The id the queue joins on rides in the agent's own idempotency label.
84 assert agentmail_source._ref(("yatxn-t1",)) == "t1"
85 assert agentmail_source._ref(("ynab-agent", "yaoffer-r9")) == "r9"
86 assert agentmail_source._ref(("ynab-agent",)) is None
89def test_agentmail_off_when_unconfigured(
90 monkeypatch: pytest.MonkeyPatch,
91) -> None:
92 for name in _MAIL_ENV:
93 monkeypatch.delenv(name, raising=False)
94 conversations, error = asyncio.run(agentmail_source.fetch())
95 assert error == "off"
96 assert conversations == ()
99# ── YNAB ─────────────────────────────────────────────────────────────────────
100class _FakeYnab:
101 def unapproved(self) -> tuple[object, ...]:
102 from ynab_agent.domain.ids import AccountId, YnabTransactionId
103 from ynab_agent.domain.money import Money
104 from ynab_agent.domain.transaction import YnabSnapshot
106 return (
107 YnabSnapshot(
108 ynab_id=YnabTransactionId("t1"),
109 account=AccountId("a1"),
110 payee="Blue Bottle",
111 amount=Money.from_currency("-4.50"),
112 txn_date=datetime.date(2026, 6, 1),
113 ),
114 )
116 def category_spends(self) -> tuple[object, ...]:
117 from ynab_agent.budget.overspend import CategorySpend
118 from ynab_agent.domain.ids import CategoryId
119 from ynab_agent.domain.money import Money
121 return (
122 CategorySpend(
123 category=CategoryId("dining"),
124 name="Dining",
125 budgeted=Money.from_currency("50"),
126 activity=Money.from_currency("-62"),
127 balance=Money.from_currency("-12"),
128 ),
129 CategorySpend(
130 category=CategoryId("gas"),
131 name="Gas",
132 budgeted=Money.from_currency("40"),
133 activity=Money.from_currency("-10"),
134 balance=Money.from_currency("30"),
135 ),
136 )
139def test_ynab_reads_backlog_and_overspent(
140 monkeypatch: pytest.MonkeyPatch,
141) -> None:
142 from ynab_agent.ynab.client import YnabClient
144 monkeypatch.setattr(
145 YnabClient, "from_env", classmethod(lambda cls: _FakeYnab())
146 )
147 budget, error = asyncio.run(ynab_source.fetch())
148 assert error is None
149 assert budget.available is True
150 assert budget.unapproved == 1
151 assert budget.unapproved_sample[0].payee == "Blue Bottle"
152 # Only the negative-balance category is overspent.
153 assert [c.name for c in budget.overspent] == ["Dining"]
156def test_ynab_off_when_key_missing(monkeypatch: pytest.MonkeyPatch) -> None:
157 from ynab_agent.ynab.client import YnabClient
159 def _raise(cls: type) -> _FakeYnab:
160 msg = "YNAB_API_KEY is not set"
161 raise RuntimeError(msg)
163 monkeypatch.setattr(YnabClient, "from_env", classmethod(_raise))
164 budget, error = asyncio.run(ynab_source.fetch())
165 assert error == "off"
166 assert budget.available is False
169def _snap(
170 tid: str,
171 payee: str,
172 amount: str,
173 *,
174 approved: bool,
175 category: str | None,
176) -> object:
177 from ynab_agent.domain.ids import AccountId, CategoryId, YnabTransactionId
178 from ynab_agent.domain.money import Money
179 from ynab_agent.domain.transaction import YnabSnapshot
181 return YnabSnapshot(
182 ynab_id=YnabTransactionId(tid),
183 account=AccountId("a1"),
184 payee=payee,
185 amount=Money.from_currency(amount),
186 txn_date=datetime.date(2026, 6, 1),
187 category_id=CategoryId(category) if category else None,
188 approved=approved,
189 )
192class _FakeResolveYnab:
193 """Unapproved ``t1`` (still waiting) + approved ``t2`` (settled in-app)."""
195 def category_spends(self) -> tuple[object, ...]:
196 from ynab_agent.budget.overspend import CategorySpend
197 from ynab_agent.domain.ids import CategoryId
198 from ynab_agent.domain.money import Money
200 return (
201 CategorySpend(
202 category=CategoryId("c-shop"),
203 name="Shopping",
204 budgeted=Money.from_currency("0"),
205 activity=Money.from_currency("0"),
206 balance=Money.from_currency("0"),
207 ),
208 )
210 def unapproved(self) -> tuple[object, ...]:
211 return (
212 _snap("t1", "Amazon", "-5.00", approved=False, category="c-shop"),
213 )
215 def snapshot(self, txn_id: str) -> object:
216 if txn_id == "t2":
217 return _snap(
218 "t2", "CP Energy", "-61.00", approved=True, category=None
219 )
220 return None
223def test_resolve_queue_humanizes_and_flags_approved(
224 monkeypatch: pytest.MonkeyPatch,
225) -> None:
226 from ynab_agent.ynab.client import YnabClient
228 monkeypatch.setattr(
229 YnabClient, "from_env", classmethod(lambda cls: _FakeResolveYnab())
230 )
231 facts = asyncio.run(ynab_source.resolve_queue(("t1", "t2", "t9")))
232 # t1 came from the unapproved index; its category id resolved to a name.
233 assert facts["t1"].payee == "Amazon"
234 assert facts["t1"].approved is False
235 assert facts["t1"].category == "Shopping"
236 # t2 a cheap single GET; flagged approved → the read-model rests it.
237 assert facts["t2"].approved is True
238 # t9 unresolved → omitted (the renderer falls back to a short id).
239 assert "t9" not in facts
242def test_resolve_queue_strips_a_date_suffix(
243 monkeypatch: pytest.MonkeyPatch,
244) -> None:
245 from ynab_agent.ynab.client import YnabClient
247 monkeypatch.setattr(
248 YnabClient, "from_env", classmethod(lambda cls: _FakeResolveYnab())
⋯ 8 lines hidden (lines 249–256)
249 )
250 facts = asyncio.run(ynab_source.resolve_queue(("t1_2026-06-02",)))
251 # Looked up by the bare id, keyed by the raw workflow id.
252 assert facts["t1_2026-06-02"].payee == "Amazon"
255def test_resolve_queue_empty_without_ids() -> None:
256 assert asyncio.run(ynab_source.resolve_queue(())) == {}