An ops dashboard the worker serves itself — a code-level reading (PR #11)

What changed, and why

The agent has enough moving parts now — the W2 lifecycle, the rule registry and its autonomy ladder, proactive offers, YNAB writes, per-transaction email threads — that there was no single place to see the system as a whole. This adds one: a private, read-only dashboard the worker hosts in-process, reached over kubectl port-forward.

The shape (borrowed from froot's dashboard): a tiny dependency-free asyncio HTTP server reuses the worker's own Temporal client, fans five readers out concurrently on each GET, assembles a pure read-model, and renders one self-contained HTML page. It is derived live and stores nothing, and every reader degrades to an error string → a red dot, so a source being down yields a warning rather than a crash — and never touches the worker's real work. New code only; the worker's existing behaviour is untouched.

The server: derive on request, store nothing

build_html is the whole per-request job: asyncio.gather the five readers (each returns (data, error)), assemble, render. The HTTP layer is deliberately minimal — GET-only, Cache-Control: no-store, Connection: close, a 15s read timeout and a 64 KB header cap — because the only client is one human behind a port-forward.

One gather, one assemble, one render — nothing held between requests.

src/ynab_agent/dashboard/server.py · 170 lines

src/ynab_agent/dashboard/server.py170 lines · Python

⋯ 47 lines hidden (lines 1–47)1"""The dashboard HTTP server — a tiny dependency-free asyncio responder.
2 
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"""
11 
12from __future__ import annotations
13 
14import asyncio
15import contextlib
16import logging
17from datetime import UTC, datetime
18from typing import TYPE_CHECKING, Final
19 
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
30 
31if TYPE_CHECKING:
32    from temporalio.client import Client
33 
34_log = logging.getLogger("ynab_agent.dashboard")
35 
36_READ_TIMEOUT: Final = 15.0
37_MAX_HEAD_BYTES: Final = 64 * 1024
38 
39 
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"
46 
47 
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    model = read_model.assemble(
59        now=now,
60        repo=_repo(),
61        temporal=temporal,
62        clickhouse=clickhouse,
63        ynab=ynab,
64        agentmail=agentmail,
65        github=github,
66    )
67    return render.page(model)
68 
69 
70def _parse_path(request_line: bytes) -> tuple[str, str]:
⋯ 100 lines hidden (lines 71–170)71    """Return ``(method, path)`` from a raw HTTP request line."""
72    parts = request_line.decode("latin-1", "replace").split()
73    if len(parts) < 2:
74        return "", "/"
75    return parts[0].upper(), parts[1].split("?", 1)[0]
76 
77 
78async def _drain_headers(reader: asyncio.StreamReader) -> None:
79    """Read and discard the request headers, bounded in size and time."""
80    total = 0
81    while True:
82        line = await asyncio.wait_for(reader.readline(), _READ_TIMEOUT)
83        total += len(line)
84        if line in (b"\r\n", b"\n", b"") or total > _MAX_HEAD_BYTES:
85            return
86 
87 
88async def _respond(
89    writer: asyncio.StreamWriter,
90    status: str,
91    content_type: str,
92    body: bytes,
93) -> None:
94    """Write a complete HTTP/1.1 response and close the connection."""
95    head = (
96        f"HTTP/1.1 {status}\r\n"
97        f"Content-Type: {content_type}\r\n"
98        f"Content-Length: {len(body)}\r\n"
99        "Cache-Control: no-store\r\n"
100        "Connection: close\r\n\r\n"
101    ).encode("latin-1")
102    writer.write(head + body)
103    await writer.drain()
104 
105 
106async def _handle(
107    reader: asyncio.StreamReader,
108    writer: asyncio.StreamWriter,
109    client: Client,
110) -> None:
111    """Serve one request: route a GET to the dashboard, else 404/405/500."""
112    try:
113        request_line = await asyncio.wait_for(reader.readline(), _READ_TIMEOUT)
114        await _drain_headers(reader)
115        method, path = _parse_path(request_line)
116        if method != "GET":
117            await _respond(writer, "405 Method Not Allowed", "text/plain", b"")
118        elif path in ("/", "/index.html", "/dashboard"):
119            html = await build_html(client)
120            await _respond(
121                writer, "200 OK", "text/html; charset=utf-8", html.encode()
122            )
123        elif path == "/healthz":
124            await _respond(writer, "200 OK", "text/plain", b"ok")
125        else:
126            await _respond(writer, "404 Not Found", "text/plain", b"")
127    except (TimeoutError, ConnectionError):
128        pass
129    except Exception as exc:  # never let a request take the worker down
130        _log.exception("dashboard request failed")
131        with contextlib.suppress(Exception):
132            await _respond(
133                writer,
134                "500 Internal Server Error",
135                "text/html; charset=utf-8",
136                _error_html(exc).encode(),
137            )
138    finally:
139        with contextlib.suppress(Exception):
140            writer.close()
141            await writer.wait_closed()
142 
143 
144def _error_html(exc: Exception) -> str:
145    """A minimal error page (the detail is for a human at a port-forward)."""
146    from html import escape
147 
148    return (
149        "<!doctype html><meta charset=utf-8>"
150        "<title>ynab-agent &middot; error</title>"
151        '<body style="font:15px system-ui;max-width:640px;margin:40px auto">'
152        "<h1>dashboard error</h1><p>Could not build the read-model.</p>"
153        f"<pre>{escape(type(exc).__name__)}: {escape(str(exc))}</pre></body>"
154    )
155 
156 
157async def start(client: Client) -> asyncio.Server:
158    """Start the dashboard server on the configured host/port (returns it)."""
159    settings = DashboardSettings()
160 
161    async def on_connect(
162        reader: asyncio.StreamReader, writer: asyncio.StreamWriter
163    ) -> None:
164        await _handle(reader, writer, client)
165 
166    server = await asyncio.start_server(
167        on_connect, host=settings.host, port=settings.port
168    )
169    _log.info("dashboard listening on %s:%d", settings.host, settings.port)
170    return server

Reusing the worker's client (and not stopping it)

The server is started from run_worker after the Temporal client connects and before the worker runs, then closed in the same finally that flushes telemetry. It reuses that connected client (so the Temporal reader needs no second connection), and — the load-bearing detail — a start failure is logged, never raised: the dashboard must never be the reason the worker doesn't run.

Gated on YNAB_AGENT_DASHBOARD_ENABLED; try/except so a failure is non-fatal.

src/ynab_agent/worker.py · 137 lines

src/ynab_agent/worker.py137 lines · Python

⋯ 113 lines hidden (lines 1–113)1"""The Temporal worker entrypoint — the runnable assembly (SPEC §0.5).
2 
3Connects to Temporal with the Pydantic data converter and registers the whole
4runtime: every workflow and every activity port. The activities are
5progressively wired to the real clients (YNAB reads are live; the model and mail
6sends land as each is connected); registering them all here is the single place
7a deployment turns on.
8 
9Run it once a Temporal server is reachable::
10 
11    python -m ynab_agent.worker
12 
13The connection is env-configured so the same image runs anywhere:
14``TEMPORAL_HOST`` (default ``localhost:7233``), ``TEMPORAL_NAMESPACE`` (default
15``default``), and ``TEMPORAL_TASK_QUEUE`` (default ``ynab-agent``). The clients
16read their own keys (``YNAB_API_KEY``, ``AGENTMAIL_API_KEY``), the model
17endpoint (``YNAB_AGENT_OLLAMA_URL`` / ``YNAB_AGENT_MODEL``), and ``Settings``
18reads the recipient config — all from the environment, never the repo.
19"""
20 
21from __future__ import annotations
22 
23import asyncio
24import contextlib
25import logging
26import os
27import signal
28 
29from temporalio.client import Client
30from temporalio.worker import Worker
31 
32from ynab_agent.telemetry import (
33    metrics_runtime,
34    setup_tracing,
35    shutdown_tracing,
36    tracing_interceptors,
37)
38from ynab_agent.workflow.runtime import (
39    ALL_ACTIVITIES,
40    DATA_CONVERTER,
41    WORKFLOWS,
42)
43 
44DEFAULT_HOST = "localhost:7233"
45DEFAULT_NAMESPACE = "default"
46DEFAULT_TASK_QUEUE = "ynab-agent"
47# A small activity pool. Model calls effectively serialize at the single local
48# Gemma regardless (Ollama runs one generation at a time), but the fast I/O
49# activities (open_thread, the commit, the mail sends) have no reason to queue
50# behind a 20-min generation. At concurrency 1 a sub-second send could wait
51# minutes for the one slot — head-of-line blocking that, once activities had a
52# wall-clock budget, surfaced as a spurious SCHEDULE_TO_START page (#6 fix).
53# A few slots let the fast work start immediately while model work still piles
54# against the Ollama lock; keep it small so a burst can't fan many concurrent
55# generations at the model.
56_MAX_CONCURRENT_ACTIVITIES = 4
57 
58 
59async def run_worker(
60    *,
61    target_host: str | None = None,
62    namespace: str | None = None,
63    task_queue: str | None = None,
64) -> None:
65    """Connect to Temporal and run the worker until cancelled (SPEC §0.5).
66 
67    Each parameter falls back to its ``TEMPORAL_*`` environment variable, then
68    to the in-cluster default, so a deployment configures the worker purely
69    through the environment.
70    """
71    host = target_host or os.environ.get("TEMPORAL_HOST", DEFAULT_HOST)
72    ns = namespace or os.environ.get("TEMPORAL_NAMESPACE", DEFAULT_NAMESPACE)
73    queue = task_queue or os.environ.get(
74        "TEMPORAL_TASK_QUEUE", DEFAULT_TASK_QUEUE
75    )
76    setup_tracing("ynab-agent-worker")
77    client = await Client.connect(
78        host,
79        namespace=ns,
80        data_converter=DATA_CONVERTER,
81        interceptors=tracing_interceptors(),
82        runtime=metrics_runtime(),
83    )
84    worker = Worker(
85        client,
86        task_queue=queue,
87        workflows=WORKFLOWS,
88        activities=ALL_ACTIVITIES,
89        max_concurrent_activities=_MAX_CONCURRENT_ACTIVITIES,
90    )
91    # Run until SIGTERM/SIGINT, then shut down gracefully and flush telemetry.
92    # `uv run` forwards SIGTERM to us, and Python's atexit does NOT run on an
93    # unhandled signal — so without this the worker's last span batch is dropped
94    # on every (Recreate) rollout.
95    stop = asyncio.Event()
96    loop = asyncio.get_running_loop()
97    for sig in (signal.SIGTERM, signal.SIGINT):
98        with contextlib.suppress(NotImplementedError):
99            loop.add_signal_handler(sig, stop.set)
100    # The in-process ops dashboard (SPEC §15): reuses this client, reached over
101    # kubectl port-forward. A start failure must never stop the worker.
102    dashboard = await _start_dashboard(client)
103    try:
104        async with worker:
105            await stop.wait()
106    finally:
107        if dashboard is not None:
108            dashboard.close()
109            with contextlib.suppress(Exception):
110                await dashboard.wait_closed()
111        shutdown_tracing()
112 
113 
114async def _start_dashboard(client: Client) -> asyncio.Server | None:
115    """Start the dashboard if enabled; a failure is logged, never fatal."""
116    from ynab_agent.settings import DashboardSettings
117 
118    if not DashboardSettings().enabled:
119        return None
120    try:
121        from ynab_agent.dashboard.server import start as start_dashboard
122 
123        return await start_dashboard(client)
124    except Exception:
125        logging.getLogger("ynab_agent.worker").exception(
126            "dashboard failed to start; running worker without it"
127        )
128        return None
129 
130 
131def main() -> None:
132    """Console entrypoint: run the worker, configured from the environment."""
133    asyncio.run(run_worker())
134 
⋯ 3 lines hidden (lines 135–137)135 
136if __name__ == "__main__":
137    main()

The Temporal reader: the live run ledger

This is the load-bearing source — it reads the agent's durable state directly. fetch is best-effort end to end: it gathers the poll heartbeat, the in-flight W2 funnel, the autonomy ladder, offers, the dispatch tally, and failures, and on any error returns whatever it gathered plus an error string.

Each sub-read contributes what it can; the registry's absence is normal, not an error.

src/ynab_agent/dashboard/temporal_source.py · 338 lines

src/ynab_agent/dashboard/temporal_source.py338 lines · Python

⋯ 273 lines hidden (lines 1–273)1"""Temporal reader: the live run ledger (reuses the worker's client).
2 
3Reads the agent's durable state straight from Temporal: the poll heartbeat, the
4in-flight W2 transactions counted by lifecycle state (each running workflow's
5``state`` query), the rule registry's autonomy ladder (its ``view`` query), the
6live autonomy offers, the recent inbound-dispatch tally, and any
7terminated/failed workflows with their recovered reason. Everything is bounded
8and best-effort: a failure returns whatever was gathered plus an error string,
9never raising into the page.
10"""
11 
12from __future__ import annotations
13 
14import contextlib
15from datetime import datetime
16from typing import TYPE_CHECKING, Final
17 
18from temporalio.client import WorkflowExecutionStatus
19 
20from ynab_agent.dashboard.model import (
  DispatchTally,
  Failure,
  OfferRow,
  QueueItem,
  RuleRow,
  StateCount,
27)
28from ynab_agent.domain.base import Frozen
29 
30if TYPE_CHECKING:
  from collections.abc import AsyncIterator
32 
  from temporalio.client import Client, WorkflowExecution
34 
35_MAX_PER_TYPE: Final = 500
36_MAX_STATE_QUERIES: Final = 200
37_REGISTRY_ID: Final = "ynab-rule-registry"
38 
39 
40class TemporalReadout(Frozen):
  """The Temporal-derived pieces of the dashboard, ready to slot in."""
42 
  poll_status: str = "none"
  poll_live: bool = False
  poll_last_start: datetime | None = None
  lifecycle_states: tuple[StateCount, ...] = ()
  in_flight: int = 0
  archived: int = 0
  terminated: int = 0
  rules: tuple[RuleRow, ...] = ()
  observe: int = 0
  eligible: int = 0
  blessed: int = 0
  offers: tuple[OfferRow, ...] = ()
  awaiting: tuple[QueueItem, ...] = ()
  dispatch: DispatchTally = DispatchTally()
  failures: tuple[Failure, ...] = ()
58 
59 
60def _status(execution: WorkflowExecution) -> str:
  status = execution.status
  return status.name.lower() if status is not None else "unknown"
63 
64 
65async def _take(
  iterator: AsyncIterator[WorkflowExecution], cap: int = _MAX_PER_TYPE
67) -> AsyncIterator[WorkflowExecution]:
  """Yield at most ``cap`` executions (a runaway-visibility backstop)."""
  seen = 0
  async for execution in iterator:
      yield execution
      seen += 1
      if seen >= cap:
          return
75 
76 
77async def _poll(client: Client) -> tuple[str, bool, datetime | None]:
  """The most recent poll tick: (status, live, last_start)."""
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'PollWorkflow' ORDER BY StartTime DESC"
      ),
      cap=1,
  ):
      status = _status(execution)
      live = execution.status in (
          WorkflowExecutionStatus.RUNNING,
          WorkflowExecutionStatus.COMPLETED,
          WorkflowExecutionStatus.CONTINUED_AS_NEW,
      )
      return status, live, execution.start_time
  return "none", False, None
93 
94 
95async def _lifecycle(
  client: Client,
97) -> tuple[tuple[StateCount, ...], int, tuple[QueueItem, ...]]:
  """Count running W2s by state; collect the awaiting-human queue."""
  counts: dict[str, int] = {}
  awaiting: list[QueueItem] = []
  seen = 0
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'TransactionWorkflow' "
          "AND ExecutionStatus = 'Running'"
      ),
      cap=_MAX_STATE_QUERIES,
  ):
      seen += 1
      try:
          handle = client.get_workflow_handle(
              execution.id, run_id=execution.run_id
          )
          state = await handle.query("state", result_type=str)
      except Exception:  # a busy/odd workflow shouldn't drop the whole panel
          state = "unknown"
      counts[state] = counts.get(state, 0) + 1
      if state == "awaiting_human":
          awaiting.append(
              QueueItem(
                  kind="proposal",
                  label=execution.id,
                  ident=execution.id,
                  since=execution.start_time,
              )
          )
  states = tuple(
      StateCount(state=name, count=counts[name]) for name in sorted(counts)
  )
  return states, seen, tuple(awaiting)
131 
132 
133async def _registry(
  client: Client,
135) -> tuple[tuple[RuleRow, ...], int, int, int]:
  """The rule table + (observe, eligible, blessed) counts (view query)."""
  from ynab_agent.domain.allocations import ProposedCategory
  from ynab_agent.domain.enums import RuleSource, TrustState
  from ynab_agent.workflow.registry_types import RegistryView
140 
  handle = client.get_workflow_handle(_REGISTRY_ID)
  view = await handle.query("view", result_type=RegistryView)
  rows: list[RuleRow] = []
  observe = eligible = blessed = 0
  for rule in view.rules:
      allocation = rule.action.allocation
      category = (
          str(allocation.category)
          if isinstance(allocation, ProposedCategory)
          else "split"
      )
      rows.append(
          RuleRow(
              payee=rule.match.payee_pattern,
              category=category,
              trust=rule.trust.value,
              source=rule.source.value,
              hits=rule.hits,
              offered=rule.offered_at is not None,
              last_confirmed_at=rule.last_confirmed_at,
          )
      )
      if rule.source is RuleSource.HUMAN_EXPLICIT:
          blessed += 1
      elif rule.trust is TrustState.TRUSTED:
          eligible += 1
      else:
          observe += 1
  return tuple(rows), observe, eligible, blessed
170 
171 
172async def _offers(client: Client) -> tuple[OfferRow, ...]:
  """The live autonomy-offer workflows (awaiting the owner's yes/no)."""
  offers: list[OfferRow] = []
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'AutonomyOfferWorkflow' "
          "AND ExecutionStatus = 'Running'"
      )
  ):
      offers.append(
          OfferRow(
              rule_id=execution.id.removeprefix("autonomy-offer-"),
              payee="",
              status=_status(execution),
              started_at=execution.start_time,
          )
      )
  return tuple(offers)
190 
191 
192async def _dispatch(client: Client) -> DispatchTally:
  """Tally recent inbound dispatch results by routing action."""
  counts: dict[str, int] = {}
  total = 0
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'DispatchWorkflow' "
          "AND ExecutionStatus = 'Completed'"
      )
  ):
      total += 1
      action = "?"
      try:
          handle = client.get_workflow_handle(
              execution.id, run_id=execution.run_id
          )
          result = await handle.result()
          got = getattr(result, "action", None)
          if isinstance(got, str):
              action = got
      except Exception:  # a bad decode shouldn't drop the tally
          action = "?"
      counts[action] = counts.get(action, 0) + 1
  return DispatchTally(
      transaction=counts.get("transaction", 0),
      offer=counts.get("offer", 0),
      receipt=counts.get("receipt", 0),
      command=counts.get("command", 0),
      quarantine=counts.get("quarantine", 0),
      ignore=counts.get("ignore", 0),
      total=total,
  )
224 
225 
226async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
  """Recover a terminated/failed workflow's human reason from history."""
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      found: str | None = None
      async for event in handle.fetch_history_events():
          terminated = event.workflow_execution_terminated_event_attributes
          if terminated.reason:
              found = terminated.reason
          failed = event.workflow_execution_failed_event_attributes
          if failed.failure.message:
              found = failed.failure.message
      return found
  except Exception:  # the reason is a nicety — never fail the panel for it
      return None
243 
244 
245async def _terminal(client: Client) -> tuple[int, int, tuple[Failure, ...]]:
  """Recent archived (completed W2) + terminated/failed counts + failures."""
  archived = terminated = 0
  failures: list[Failure] = []
  async for _completed in _take(
      client.list_workflows(
          "WorkflowType = 'TransactionWorkflow' "
          "AND ExecutionStatus = 'Completed'"
      )
  ):
      archived += 1
  async for execution in _take(
      client.list_workflows(
          "ExecutionStatus = 'Terminated' OR ExecutionStatus = 'Failed'"
      )
  ):
      terminated += 1
      if len(failures) < 25:
          failures.append(
              Failure(
                  workflow_id=execution.id,
                  kind=_status(execution),
                  reason=await _reason(client, execution),
                  when=execution.close_time,
              )
          )
  return archived, terminated, tuple(failures)
272 
273 
274async def fetch(client: Client) -> tuple[TemporalReadout, str | None]:
  """Read the agent's Temporal state; degrades to an error string."""
  poll_status = "none"
  poll_live = False
  poll_last: datetime | None = None
  states: tuple[StateCount, ...] = ()
  in_flight = 0
  awaiting: tuple[QueueItem, ...] = ()
  rules: tuple[RuleRow, ...] = ()
  observe = eligible = blessed = 0
  offers: tuple[OfferRow, ...] = ()
  dispatch = DispatchTally()
  archived = terminated = 0
  failures: tuple[Failure, ...] = ()
288 
  try:
      poll_status, poll_live, poll_last = await _poll(client)
      states, in_flight, awaiting = await _lifecycle(client)
      # The registry may not exist until the first learning signal — a normal
      # state, not an error, so its absence is simply suppressed.
      with contextlib.suppress(Exception):
          rules, observe, eligible, blessed = await _registry(client)
      offers = await _offers(client)
      dispatch = await _dispatch(client)
      archived, terminated, failures = await _terminal(client)
  except Exception as exc:  # degrade to an error, never crash the page
      readout = TemporalReadout(
⋯ 38 lines hidden (lines 301–338)301            poll_status=poll_status,
          poll_live=poll_live,
          poll_last_start=poll_last,
          lifecycle_states=states,
          in_flight=in_flight,
          rules=rules,
          observe=observe,
          eligible=eligible,
          blessed=blessed,
          offers=offers,
          awaiting=awaiting,
          dispatch=dispatch,
          archived=archived,
          terminated=terminated,
          failures=failures,
      )
      return readout, f"{type(exc).__name__}: {exc}"
318 
  return (
      TemporalReadout(
          poll_status=poll_status,
          poll_live=poll_live,
          poll_last_start=poll_last,
          lifecycle_states=states,
          in_flight=in_flight,
          archived=archived,
          terminated=terminated,
          rules=rules,
          observe=observe,
          eligible=eligible,
          blessed=blessed,
          offers=offers,
          awaiting=awaiting,
          dispatch=dispatch,
          failures=failures,
      ),
      None,
  )

The funnel can't come from a search attribute (lifecycle state isn't indexed), so it queries each running TransactionWorkflow's state query and tallies — bounded by _MAX_STATE_QUERIES, and a stuck workflow degrades to unknown rather than dropping the panel.

Per-workflow state queries → the funnel + the awaiting-human queue.

src/ynab_agent/dashboard/temporal_source.py · 338 lines

src/ynab_agent/dashboard/temporal_source.py338 lines · Python

⋯ 94 lines hidden (lines 1–94)1"""Temporal reader: the live run ledger (reuses the worker's client).
2 
3Reads the agent's durable state straight from Temporal: the poll heartbeat, the
4in-flight W2 transactions counted by lifecycle state (each running workflow's
5``state`` query), the rule registry's autonomy ladder (its ``view`` query), the
6live autonomy offers, the recent inbound-dispatch tally, and any
7terminated/failed workflows with their recovered reason. Everything is bounded
8and best-effort: a failure returns whatever was gathered plus an error string,
9never raising into the page.
10"""
11 
12from __future__ import annotations
13 
14import contextlib
15from datetime import datetime
16from typing import TYPE_CHECKING, Final
17 
18from temporalio.client import WorkflowExecutionStatus
19 
20from ynab_agent.dashboard.model import (
  DispatchTally,
  Failure,
  OfferRow,
  QueueItem,
  RuleRow,
  StateCount,
27)
28from ynab_agent.domain.base import Frozen
29 
30if TYPE_CHECKING:
  from collections.abc import AsyncIterator
32 
  from temporalio.client import Client, WorkflowExecution
34 
35_MAX_PER_TYPE: Final = 500
36_MAX_STATE_QUERIES: Final = 200
37_REGISTRY_ID: Final = "ynab-rule-registry"
38 
39 
40class TemporalReadout(Frozen):
  """The Temporal-derived pieces of the dashboard, ready to slot in."""
42 
  poll_status: str = "none"
  poll_live: bool = False
  poll_last_start: datetime | None = None
  lifecycle_states: tuple[StateCount, ...] = ()
  in_flight: int = 0
  archived: int = 0
  terminated: int = 0
  rules: tuple[RuleRow, ...] = ()
  observe: int = 0
  eligible: int = 0
  blessed: int = 0
  offers: tuple[OfferRow, ...] = ()
  awaiting: tuple[QueueItem, ...] = ()
  dispatch: DispatchTally = DispatchTally()
  failures: tuple[Failure, ...] = ()
58 
59 
60def _status(execution: WorkflowExecution) -> str:
  status = execution.status
  return status.name.lower() if status is not None else "unknown"
63 
64 
65async def _take(
  iterator: AsyncIterator[WorkflowExecution], cap: int = _MAX_PER_TYPE
67) -> AsyncIterator[WorkflowExecution]:
  """Yield at most ``cap`` executions (a runaway-visibility backstop)."""
  seen = 0
  async for execution in iterator:
      yield execution
      seen += 1
      if seen >= cap:
          return
75 
76 
77async def _poll(client: Client) -> tuple[str, bool, datetime | None]:
  """The most recent poll tick: (status, live, last_start)."""
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'PollWorkflow' ORDER BY StartTime DESC"
      ),
      cap=1,
  ):
      status = _status(execution)
      live = execution.status in (
          WorkflowExecutionStatus.RUNNING,
          WorkflowExecutionStatus.COMPLETED,
          WorkflowExecutionStatus.CONTINUED_AS_NEW,
      )
      return status, live, execution.start_time
  return "none", False, None
93 
94 
95async def _lifecycle(
  client: Client,
97) -> tuple[tuple[StateCount, ...], int, tuple[QueueItem, ...]]:
  """Count running W2s by state; collect the awaiting-human queue."""
  counts: dict[str, int] = {}
  awaiting: list[QueueItem] = []
  seen = 0
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'TransactionWorkflow' "
          "AND ExecutionStatus = 'Running'"
      ),
      cap=_MAX_STATE_QUERIES,
  ):
      seen += 1
      try:
          handle = client.get_workflow_handle(
              execution.id, run_id=execution.run_id
          )
          state = await handle.query("state", result_type=str)
      except Exception:  # a busy/odd workflow shouldn't drop the whole panel
          state = "unknown"
      counts[state] = counts.get(state, 0) + 1
      if state == "awaiting_human":
          awaiting.append(
              QueueItem(
                  kind="proposal",
                  label=execution.id,
                  ident=execution.id,
                  since=execution.start_time,
              )
          )
  states = tuple(
      StateCount(state=name, count=counts[name]) for name in sorted(counts)
  )
⋯ 209 lines hidden (lines 130–338)130    return states, seen, tuple(awaiting)
131 
132 
133async def _registry(
  client: Client,
135) -> tuple[tuple[RuleRow, ...], int, int, int]:
  """The rule table + (observe, eligible, blessed) counts (view query)."""
  from ynab_agent.domain.allocations import ProposedCategory
  from ynab_agent.domain.enums import RuleSource, TrustState
  from ynab_agent.workflow.registry_types import RegistryView
140 
  handle = client.get_workflow_handle(_REGISTRY_ID)
  view = await handle.query("view", result_type=RegistryView)
  rows: list[RuleRow] = []
  observe = eligible = blessed = 0
  for rule in view.rules:
      allocation = rule.action.allocation
      category = (
          str(allocation.category)
          if isinstance(allocation, ProposedCategory)
          else "split"
      )
      rows.append(
          RuleRow(
              payee=rule.match.payee_pattern,
              category=category,
              trust=rule.trust.value,
              source=rule.source.value,
              hits=rule.hits,
              offered=rule.offered_at is not None,
              last_confirmed_at=rule.last_confirmed_at,
          )
      )
      if rule.source is RuleSource.HUMAN_EXPLICIT:
          blessed += 1
      elif rule.trust is TrustState.TRUSTED:
          eligible += 1
      else:
          observe += 1
  return tuple(rows), observe, eligible, blessed
170 
171 
172async def _offers(client: Client) -> tuple[OfferRow, ...]:
  """The live autonomy-offer workflows (awaiting the owner's yes/no)."""
  offers: list[OfferRow] = []
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'AutonomyOfferWorkflow' "
          "AND ExecutionStatus = 'Running'"
      )
  ):
      offers.append(
          OfferRow(
              rule_id=execution.id.removeprefix("autonomy-offer-"),
              payee="",
              status=_status(execution),
              started_at=execution.start_time,
          )
      )
  return tuple(offers)
190 
191 
192async def _dispatch(client: Client) -> DispatchTally:
  """Tally recent inbound dispatch results by routing action."""
  counts: dict[str, int] = {}
  total = 0
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'DispatchWorkflow' "
          "AND ExecutionStatus = 'Completed'"
      )
  ):
      total += 1
      action = "?"
      try:
          handle = client.get_workflow_handle(
              execution.id, run_id=execution.run_id
          )
          result = await handle.result()
          got = getattr(result, "action", None)
          if isinstance(got, str):
              action = got
      except Exception:  # a bad decode shouldn't drop the tally
          action = "?"
      counts[action] = counts.get(action, 0) + 1
  return DispatchTally(
      transaction=counts.get("transaction", 0),
      offer=counts.get("offer", 0),
      receipt=counts.get("receipt", 0),
      command=counts.get("command", 0),
      quarantine=counts.get("quarantine", 0),
      ignore=counts.get("ignore", 0),
      total=total,
  )
224 
225 
226async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
  """Recover a terminated/failed workflow's human reason from history."""
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      found: str | None = None
      async for event in handle.fetch_history_events():
          terminated = event.workflow_execution_terminated_event_attributes
          if terminated.reason:
              found = terminated.reason
          failed = event.workflow_execution_failed_event_attributes
          if failed.failure.message:
              found = failed.failure.message
      return found
  except Exception:  # the reason is a nicety — never fail the panel for it
      return None
243 
244 
245async def _terminal(client: Client) -> tuple[int, int, tuple[Failure, ...]]:
  """Recent archived (completed W2) + terminated/failed counts + failures."""
  archived = terminated = 0
  failures: list[Failure] = []
  async for _completed in _take(
      client.list_workflows(
          "WorkflowType = 'TransactionWorkflow' "
          "AND ExecutionStatus = 'Completed'"
      )
  ):
      archived += 1
  async for execution in _take(
      client.list_workflows(
          "ExecutionStatus = 'Terminated' OR ExecutionStatus = 'Failed'"
      )
  ):
      terminated += 1
      if len(failures) < 25:
          failures.append(
              Failure(
                  workflow_id=execution.id,
                  kind=_status(execution),
                  reason=await _reason(client, execution),
                  when=execution.close_time,
              )
          )
  return archived, terminated, tuple(failures)
272 
273 
274async def fetch(client: Client) -> tuple[TemporalReadout, str | None]:
  """Read the agent's Temporal state; degrades to an error string."""
  poll_status = "none"
  poll_live = False
  poll_last: datetime | None = None
  states: tuple[StateCount, ...] = ()
  in_flight = 0
  awaiting: tuple[QueueItem, ...] = ()
  rules: tuple[RuleRow, ...] = ()
  observe = eligible = blessed = 0
  offers: tuple[OfferRow, ...] = ()
  dispatch = DispatchTally()
  archived = terminated = 0
  failures: tuple[Failure, ...] = ()
288 
  try:
      poll_status, poll_live, poll_last = await _poll(client)
      states, in_flight, awaiting = await _lifecycle(client)
      # The registry may not exist until the first learning signal — a normal
      # state, not an error, so its absence is simply suppressed.
      with contextlib.suppress(Exception):
          rules, observe, eligible, blessed = await _registry(client)
      offers = await _offers(client)
      dispatch = await _dispatch(client)
      archived, terminated, failures = await _terminal(client)
  except Exception as exc:  # degrade to an error, never crash the page
      readout = TemporalReadout(
          poll_status=poll_status,
          poll_live=poll_live,
          poll_last_start=poll_last,
          lifecycle_states=states,
          in_flight=in_flight,
          rules=rules,
          observe=observe,
          eligible=eligible,
          blessed=blessed,
          offers=offers,
          awaiting=awaiting,
          dispatch=dispatch,
          archived=archived,
          terminated=terminated,
          failures=failures,
      )
      return readout, f"{type(exc).__name__}: {exc}"
318 
  return (
      TemporalReadout(
          poll_status=poll_status,
          poll_live=poll_live,
          poll_last_start=poll_last,
          lifecycle_states=states,
          in_flight=in_flight,
          archived=archived,
          terminated=terminated,
          rules=rules,
          observe=observe,
          eligible=eligible,
          blessed=blessed,
          offers=offers,
          awaiting=awaiting,
          dispatch=dispatch,
          failures=failures,
      ),
      None,
  )

src/ynab_agent/dashboard/temporal_source.py · 338 lines

src/ynab_agent/dashboard/temporal_source.py338 lines · Python

⋯ 132 lines hidden (lines 1–132)1"""Temporal reader: the live run ledger (reuses the worker's client).
2 
3Reads the agent's durable state straight from Temporal: the poll heartbeat, the
4in-flight W2 transactions counted by lifecycle state (each running workflow's
5``state`` query), the rule registry's autonomy ladder (its ``view`` query), the
6live autonomy offers, the recent inbound-dispatch tally, and any
7terminated/failed workflows with their recovered reason. Everything is bounded
8and best-effort: a failure returns whatever was gathered plus an error string,
9never raising into the page.
10"""
11 
12from __future__ import annotations
13 
14import contextlib
15from datetime import datetime
16from typing import TYPE_CHECKING, Final
17 
18from temporalio.client import WorkflowExecutionStatus
19 
20from ynab_agent.dashboard.model import (
  DispatchTally,
  Failure,
  OfferRow,
  QueueItem,
  RuleRow,
  StateCount,
27)
28from ynab_agent.domain.base import Frozen
29 
30if TYPE_CHECKING:
  from collections.abc import AsyncIterator
32 
  from temporalio.client import Client, WorkflowExecution
34 
35_MAX_PER_TYPE: Final = 500
36_MAX_STATE_QUERIES: Final = 200
37_REGISTRY_ID: Final = "ynab-rule-registry"
38 
39 
40class TemporalReadout(Frozen):
  """The Temporal-derived pieces of the dashboard, ready to slot in."""
42 
  poll_status: str = "none"
  poll_live: bool = False
  poll_last_start: datetime | None = None
  lifecycle_states: tuple[StateCount, ...] = ()
  in_flight: int = 0
  archived: int = 0
  terminated: int = 0
  rules: tuple[RuleRow, ...] = ()
  observe: int = 0
  eligible: int = 0
  blessed: int = 0
  offers: tuple[OfferRow, ...] = ()
  awaiting: tuple[QueueItem, ...] = ()
  dispatch: DispatchTally = DispatchTally()
  failures: tuple[Failure, ...] = ()
58 
59 
60def _status(execution: WorkflowExecution) -> str:
  status = execution.status
  return status.name.lower() if status is not None else "unknown"
63 
64 
65async def _take(
  iterator: AsyncIterator[WorkflowExecution], cap: int = _MAX_PER_TYPE
67) -> AsyncIterator[WorkflowExecution]:
  """Yield at most ``cap`` executions (a runaway-visibility backstop)."""
  seen = 0
  async for execution in iterator:
      yield execution
      seen += 1
      if seen >= cap:
          return
75 
76 
77async def _poll(client: Client) -> tuple[str, bool, datetime | None]:
  """The most recent poll tick: (status, live, last_start)."""
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'PollWorkflow' ORDER BY StartTime DESC"
      ),
      cap=1,
  ):
      status = _status(execution)
      live = execution.status in (
          WorkflowExecutionStatus.RUNNING,
          WorkflowExecutionStatus.COMPLETED,
          WorkflowExecutionStatus.CONTINUED_AS_NEW,
      )
      return status, live, execution.start_time
  return "none", False, None
93 
94 
95async def _lifecycle(
  client: Client,
97) -> tuple[tuple[StateCount, ...], int, tuple[QueueItem, ...]]:
  """Count running W2s by state; collect the awaiting-human queue."""
  counts: dict[str, int] = {}
  awaiting: list[QueueItem] = []
  seen = 0
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'TransactionWorkflow' "
          "AND ExecutionStatus = 'Running'"
      ),
      cap=_MAX_STATE_QUERIES,
  ):
      seen += 1
      try:
          handle = client.get_workflow_handle(
              execution.id, run_id=execution.run_id
          )
          state = await handle.query("state", result_type=str)
      except Exception:  # a busy/odd workflow shouldn't drop the whole panel
          state = "unknown"
      counts[state] = counts.get(state, 0) + 1
      if state == "awaiting_human":
          awaiting.append(
              QueueItem(
                  kind="proposal",
                  label=execution.id,
                  ident=execution.id,
                  since=execution.start_time,
              )
          )
  states = tuple(
      StateCount(state=name, count=counts[name]) for name in sorted(counts)
  )
  return states, seen, tuple(awaiting)
131 
132 
133async def _registry(
  client: Client,
135) -> tuple[tuple[RuleRow, ...], int, int, int]:
  """The rule table + (observe, eligible, blessed) counts (view query)."""
  from ynab_agent.domain.allocations import ProposedCategory
  from ynab_agent.domain.enums import RuleSource, TrustState
  from ynab_agent.workflow.registry_types import RegistryView
140 
  handle = client.get_workflow_handle(_REGISTRY_ID)
  view = await handle.query("view", result_type=RegistryView)
  rows: list[RuleRow] = []
  observe = eligible = blessed = 0
  for rule in view.rules:
      allocation = rule.action.allocation
      category = (
          str(allocation.category)
          if isinstance(allocation, ProposedCategory)
          else "split"
      )
      rows.append(
          RuleRow(
              payee=rule.match.payee_pattern,
              category=category,
              trust=rule.trust.value,
              source=rule.source.value,
              hits=rule.hits,
              offered=rule.offered_at is not None,
              last_confirmed_at=rule.last_confirmed_at,
          )
      )
      if rule.source is RuleSource.HUMAN_EXPLICIT:
          blessed += 1
      elif rule.trust is TrustState.TRUSTED:
          eligible += 1
      else:
          observe += 1
⋯ 170 lines hidden (lines 169–338)169    return tuple(rows), observe, eligible, blessed
170 
171 
172async def _offers(client: Client) -> tuple[OfferRow, ...]:
  """The live autonomy-offer workflows (awaiting the owner's yes/no)."""
  offers: list[OfferRow] = []
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'AutonomyOfferWorkflow' "
          "AND ExecutionStatus = 'Running'"
      )
  ):
      offers.append(
          OfferRow(
              rule_id=execution.id.removeprefix("autonomy-offer-"),
              payee="",
              status=_status(execution),
              started_at=execution.start_time,
          )
      )
  return tuple(offers)
190 
191 
192async def _dispatch(client: Client) -> DispatchTally:
  """Tally recent inbound dispatch results by routing action."""
  counts: dict[str, int] = {}
  total = 0
  async for execution in _take(
      client.list_workflows(
          "WorkflowType = 'DispatchWorkflow' "
          "AND ExecutionStatus = 'Completed'"
      )
  ):
      total += 1
      action = "?"
      try:
          handle = client.get_workflow_handle(
              execution.id, run_id=execution.run_id
          )
          result = await handle.result()
          got = getattr(result, "action", None)
          if isinstance(got, str):
              action = got
      except Exception:  # a bad decode shouldn't drop the tally
          action = "?"
      counts[action] = counts.get(action, 0) + 1
  return DispatchTally(
      transaction=counts.get("transaction", 0),
      offer=counts.get("offer", 0),
      receipt=counts.get("receipt", 0),
      command=counts.get("command", 0),
      quarantine=counts.get("quarantine", 0),
      ignore=counts.get("ignore", 0),
      total=total,
  )
224 
225 
226async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
  """Recover a terminated/failed workflow's human reason from history."""
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      found: str | None = None
      async for event in handle.fetch_history_events():
          terminated = event.workflow_execution_terminated_event_attributes
          if terminated.reason:
              found = terminated.reason
          failed = event.workflow_execution_failed_event_attributes
          if failed.failure.message:
              found = failed.failure.message
      return found
  except Exception:  # the reason is a nicety — never fail the panel for it
      return None
243 
244 
245async def _terminal(client: Client) -> tuple[int, int, tuple[Failure, ...]]:
  """Recent archived (completed W2) + terminated/failed counts + failures."""
  archived = terminated = 0
  failures: list[Failure] = []
  async for _completed in _take(
      client.list_workflows(
          "WorkflowType = 'TransactionWorkflow' "
          "AND ExecutionStatus = 'Completed'"
      )
  ):
      archived += 1
  async for execution in _take(
      client.list_workflows(
          "ExecutionStatus = 'Terminated' OR ExecutionStatus = 'Failed'"
      )
  ):
      terminated += 1
      if len(failures) < 25:
          failures.append(
              Failure(
                  workflow_id=execution.id,
                  kind=_status(execution),
                  reason=await _reason(client, execution),
                  when=execution.close_time,
              )
          )
  return archived, terminated, tuple(failures)
272 
273 
274async def fetch(client: Client) -> tuple[TemporalReadout, str | None]:
  """Read the agent's Temporal state; degrades to an error string."""
  poll_status = "none"
  poll_live = False
  poll_last: datetime | None = None
  states: tuple[StateCount, ...] = ()
  in_flight = 0
  awaiting: tuple[QueueItem, ...] = ()
  rules: tuple[RuleRow, ...] = ()
  observe = eligible = blessed = 0
  offers: tuple[OfferRow, ...] = ()
  dispatch = DispatchTally()
  archived = terminated = 0
  failures: tuple[Failure, ...] = ()
288 
  try:
      poll_status, poll_live, poll_last = await _poll(client)
      states, in_flight, awaiting = await _lifecycle(client)
      # The registry may not exist until the first learning signal — a normal
      # state, not an error, so its absence is simply suppressed.
      with contextlib.suppress(Exception):
          rules, observe, eligible, blessed = await _registry(client)
      offers = await _offers(client)
      dispatch = await _dispatch(client)
      archived, terminated, failures = await _terminal(client)
  except Exception as exc:  # degrade to an error, never crash the page
      readout = TemporalReadout(
          poll_status=poll_status,
          poll_live=poll_live,
          poll_last_start=poll_last,
          lifecycle_states=states,
          in_flight=in_flight,
          rules=rules,
          observe=observe,
          eligible=eligible,
          blessed=blessed,
          offers=offers,
          awaiting=awaiting,
          dispatch=dispatch,
          archived=archived,
          terminated=terminated,
          failures=failures,
      )
      return readout, f"{type(exc).__name__}: {exc}"
318 
  return (
      TemporalReadout(
          poll_status=poll_status,
          poll_live=poll_live,
          poll_last_start=poll_last,
          lifecycle_states=states,
          in_flight=in_flight,
          archived=archived,
          terminated=terminated,
          rules=rules,
          observe=observe,
          eligible=eligible,
          blessed=blessed,
          offers=offers,
          awaiting=awaiting,
          dispatch=dispatch,
          failures=failures,
      ),
      None,
  )

Assemble: errors become dots, pieces slot in

assemble is pure: it turns each reader's error into a source-health dot (None → green with a summary, anything else — including "off" — → red) and slots the Temporal-derived pieces into the model. The human queue is the one join: awaiting-human transactions plus live offers.

No I/O — unit-tested straight from in-memory readouts.

src/ynab_agent/dashboard/read_model.py · 110 lines

src/ynab_agent/dashboard/read_model.py110 lines · Python

⋯ 40 lines hidden (lines 1–40)1"""Assemble the source readouts into one view-model (pure).
2 
3Each reader hands in ``(data, error)``; this composes the
class:`~ynab_agent.dashboard.model.DashboardModel`, turns each error into a
5source-health dot, and slots the Temporal-derived pieces into place. No I/O —
6fully unit-testable from in-memory readouts.
7"""
8 
9from __future__ import annotations
10 
11from typing import TYPE_CHECKING
12 
13from ynab_agent.dashboard.model import (
  Autonomy,
  DashboardModel,
  Heartbeat,
  Lifecycle,
  QueueItem,
  SourceHealth,
20)
21 
22if TYPE_CHECKING:
  from datetime import datetime
24 
  from ynab_agent.dashboard.model import (
      Budget,
      Conversation,
      Deploy,
      RunTelemetry,
  )
  from ynab_agent.dashboard.temporal_source import TemporalReadout
32 
33 
34def _health(name: str, error: str | None, ok_detail: str) -> SourceHealth:
  """A source dot: green with a summary, or red with the error/'off'."""
  if error is None:
      return SourceHealth(name=name, ok=True, detail=ok_detail)
  return SourceHealth(name=name, ok=False, detail=error)
39 
40 
41def assemble(
  *,
  now: datetime,
  repo: str,
  temporal: tuple[TemporalReadout, str | None],
  ynab: tuple[Budget, str | None],
  clickhouse: tuple[RunTelemetry, str | None],
  agentmail: tuple[tuple[Conversation, ...], str | None],
  github: tuple[Deploy, str | None],
50) -> DashboardModel:
  """Compose the whole dashboard view from the source readouts."""
  t, t_err = temporal
  budget, ynab_err = ynab
  telemetry, ch_err = clickhouse
  conversations, mail_err = agentmail
  deploy, gh_err = github
57 
  sources = (
      _health("temporal", t_err, f"{t.in_flight} in-flight"),
      _health("ynab", ynab_err, f"{budget.unapproved} unapproved"),
      _health("clickhouse", ch_err, f"{telemetry.total_spans} spans"),
      _health("agentmail", mail_err, f"{len(conversations)} threads"),
      _health("github", gh_err, f"{len(deploy.prs)} PRs"),
  )
65 
  heartbeat = Heartbeat(
      poll_status=t.poll_status,
      poll_live=t.poll_live,
      poll_last_start=t.poll_last_start,
      worker_last_span=telemetry.last_activity,
  )
  lifecycle = Lifecycle(
      states=t.lifecycle_states,
      in_flight=t.in_flight,
      archived=t.archived,
      terminated=t.terminated,
  )
  autonomy = Autonomy(
      observe=t.observe,
      eligible=t.eligible,
      blessed=t.blessed,
      rules=t.rules,
      offers=t.offers,
  )
  offer_queue = tuple(
      QueueItem(
          kind="offer",
          label=offer.payee or offer.rule_id,
          ident=offer.rule_id,
          since=offer.started_at,
      )
      for offer in t.offers
  )
  queue = (*t.awaiting, *offer_queue)
95 
⋯ 15 lines hidden (lines 96–110)96    return DashboardModel(
      generated_at=now,
      repo=repo,
      sources=sources,
      heartbeat=heartbeat,
      lifecycle=lifecycle,
      autonomy=autonomy,
      queue=queue,
      budget=budget,
      conversations=conversations,
      dispatch=t.dispatch,
      telemetry=telemetry,
      failures=t.failures,
      deploy=deploy,
  )

Render: the visuals

One self-contained page, inline CSS, no JS, every value html.escaped. Two panels carry the 'strong visuals' ask. The lifecycle funnel is proportional bars sized to each state's count; the autonomy ladder is a coloured observe→eligible→blessed strip above the rule table (with trust pills).

The lifecycle bar-chart: width ∝ count / peak.

src/ynab_agent/dashboard/render.py · 477 lines

src/ynab_agent/dashboard/render.py477 lines · Python

⋯ 171 lines hidden (lines 1–171)1"""Render the view-model to one self-contained HTML page (pure).
2 
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
class:`~ynab_agent.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary. Ordering is ops-first (is it alive → what's
7flowing → earned autonomy → the human's queue → money → conversations → inbound
8→ telemetry → failures → deploy), so it reads top to bottom.
9"""
10 
11from __future__ import annotations
12 
13from datetime import UTC, datetime
14from html import escape
15from typing import TYPE_CHECKING
16 
17if TYPE_CHECKING:
  from ynab_agent.dashboard.model import (
      DashboardModel,
      Lifecycle,
      RunTelemetry,
  )
23 
24_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
26--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#f2f2f2}
27@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
28--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
29--accent:#58a6ff;--card:#1b1b1b}}
30*{box-sizing:border-box}
31body{margin:0;background:var(--bg);color:var(--fg);
32font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
33main{max-width:860px;margin:0 auto;padding:34px 20px 72px}
34h1{font-size:22px;margin:0;letter-spacing:-.01em}
tag{color:var(--mut);margin:3px 0 0;font-size:13px}
meta{color:var(--mut);font-size:12px;margin:12px 0 0}
sources{display:flex;flex-wrap:wrap;gap:16px;margin:10px 0 0;font-size:12px}
38section{margin:30px 0 0}
39h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
40color:var(--mut);margin:0 0 12px;font-weight:600;
41border-bottom:1px solid var(--line);padding-bottom:6px}
stats{display:flex;flex-wrap:wrap;gap:26px}
stat .n{font-size:26px;font-weight:600;line-height:1.1}
stat .l{color:var(--mut);font-size:12px;margin-top:2px}
dot{display:inline-block;width:9px;height:9px;border-radius:50%;margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
48table{width:100%;border-collapse:collapse;font-size:13px}
49th,td{text-align:left;padding:6px 12px 6px 0;vertical-align:top;
50border-bottom:1px solid var(--line)}
51th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
52letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
54a{color:var(--accent);text-decoration:none}a:hover{text-decoration:underline}
row{display:flex;align-items:baseline;gap:8px;padding:4px 0}
mut{color:var(--mut)}.ok{color:var(--ok)}.warn{color:var(--warn)}.bad{color:var(--bad)}
note{color:var(--mut);font-size:12px;margin:10px 0 0}
barrow{display:grid;grid-template-columns:140px 1fr 34px;gap:10px;
59align-items:center;margin:5px 0}
barrow .lab{font-size:12px;color:var(--mut)}
barrow .num{font-size:12px;text-align:right;font-variant-numeric:tabular-nums}
bar{background:var(--card);border-radius:4px;height:16px;overflow:hidden}
bar>span{display:block;height:100%;background:var(--accent);min-width:2px}
ladder{display:flex;height:24px;border-radius:5px;overflow:hidden;
65margin:2px 0 10px;font-size:11px;font-weight:600}
ladder>span{display:flex;align-items:center;justify-content:center;color:#fff;
67white-space:nowrap;min-width:0}
ladder .obs{background:var(--mut)}.ladder .elig{background:var(--warn)}
ladder .bless{background:var(--ok)}
pill{display:inline-block;padding:1px 7px;border-radius:10px;font-size:11px;
71font-weight:600}
pill.ok{background:rgba(63,185,80,.16);color:var(--ok)}
pill.warn{background:rgba(210,153,34,.16);color:var(--warn)}
pill.mute{background:var(--card);color:var(--mut)}
75footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
76color:var(--mut);font-size:12px;line-height:1.7}
77footer b{color:var(--fg);font-weight:600}
78"""
79 
80# The canonical W2 lifecycle order for the funnel (any other state is appended).
81_LIFECYCLE_ORDER = (
  "discovered",
  "enriching",
  "hold_amazon",
  "awaiting_human",
  "open",
  "lapsed",
  "revising",
89)
90_TRUST_CLASS = {"trusted": "ok", "confirmed": "warn", "suggested": "mute"}
91_CI_CLASS = {"passed": "ok", "failed": "bad", "running": "warn"}
92_STATE_CLASS = {"merged": "ok", "open": "warn", "closed": "mut"}
93 
94 
95def _aware(when: datetime) -> datetime:
  return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
97 
98 
99def _ago(when: datetime | None, now: datetime) -> str:
  if when is None:
      return "—"
  secs = (now - _aware(when)).total_seconds()
  if secs < 0:
      return "just now"
  if secs < 90:
      return "just now"
  if secs < 5400:
      return f"{int(secs // 60)}m ago"
  if secs < 129600:
      return f"{int(secs // 3600)}h ago"
  return f"{int(secs // 86400)}d ago"
112 
113 
114def _dot(kind: str) -> str:
  return f'<span class="dot {kind}"></span>'
116 
117 
118def _stat(n: object, label: str) -> str:
  return (
      f'<div class="stat"><div class="n">{escape(str(n))}</div>'
      f'<div class="l">{escape(label)}</div></div>'
  )
123 
124 
125def _pill(text: str, cls: str) -> str:
  return f'<span class="pill {cls}">{escape(text)}</span>'
127 
128 
129def _header(model: DashboardModel) -> str:
  stamp = model.generated_at.strftime("%Y-%m-%d %H:%M UTC")
  dots = "".join(
      f"<span>{_dot('ok' if s.ok else 'bad')}{escape(s.name)} "
      f'<span class="mut">{escape(s.detail)}</span></span>'
      for s in model.sources
  )
  return (
      "<header><h1>ynab-agent</h1>"
      '<p class="tag">durable transaction triage &middot; '
      "earned-autonomy read-model</p>"
      f'<p class="meta">{escape(model.repo)} &middot; generated '
      f"{escape(stamp)} &middot; derived live, stored nowhere</p>"
      f'<div class="sources">{dots}</div></header>'
  )
144 
145 
146def _heartbeat(model: DashboardModel) -> str:
  h = model.heartbeat
  now = model.generated_at
  poll_dot = "ok" if h.poll_live else "bad"
  poll = (
      f'<div class="row">{_dot(poll_dot)}<span class="mono">W1 poll</span> '
      f'<span class="mut">&middot; {escape(h.poll_status)} &middot; last '
      f"{escape(_ago(h.poll_last_start, now))}</span></div>"
  )
  if h.worker_last_span is not None:
      span_ago = (now - _aware(h.worker_last_span)).total_seconds()
      wdot = "ok" if span_ago < 3600 else "warn"
      worker = (
          f'<div class="row">{_dot(wdot)}<span class="mono">worker</span> '
          f'<span class="mut">&middot; last span '
          f"{escape(_ago(h.worker_last_span, now))}</span></div>"
      )
  else:
      worker = (
          f'<div class="row">{_dot("mute")}'
          '<span class="mono">worker</span> '
          '<span class="mut">&middot; no telemetry</span></div>'
      )
  return f"<section><h2>Is it alive?</h2>{poll}{worker}</section>"
170 
171 
172def _lifecycle(model: DashboardModel) -> str:
  lc: Lifecycle = model.lifecycle
  counts = {s.state: s.count for s in lc.states}
  ordered = [s for s in _LIFECYCLE_ORDER if s in counts]
  ordered += [s for s in sorted(counts) if s not in _LIFECYCLE_ORDER]
  peak = max((counts[s] for s in ordered), default=0) or 1
  rows = "".join(
      f'<div class="barrow"><div class="lab">{escape(s)}</div>'
      f'<div class="bar"><span style="width:{counts[s] * 100 // peak}%">'
      f'</span></div><div class="num">{counts[s]}</div></div>'
      for s in ordered
  )
  if not ordered:
      rows = '<p class="note">No transactions in flight.</p>'
  stats = "".join(
      (
          _stat(lc.in_flight, "in flight"),
          _stat(lc.archived, "archived (recent)"),
          _stat(lc.terminated, "terminated (recent)"),
      )
  )
  return (
      "<section><h2>Transaction lifecycle</h2>"
      f'<div class="stats">{stats}</div>'
      f'<div style="margin-top:14px">{rows}</div></section>'
  )
198 
⋯ 279 lines hidden (lines 199–477)199 
200def _ladder(model: DashboardModel) -> str:
  a = model.autonomy
  total = a.observe + a.eligible + a.blessed
  if total:
      segs = "".join(
          f'<span class="{cls}" style="flex:{n}" title="{label}: {n}">'
          f"{n if n * 100 // total >= 8 else ''}</span>"
          for n, cls, label in (
              (a.observe, "obs", "observe"),
              (a.eligible, "elig", "eligible"),
              (a.blessed, "bless", "blessed"),
          )
          if n
      )
      bar = f'<div class="ladder">{segs}</div>'
  else:
      bar = ""
  stats = "".join(
      (
          _stat(a.observe, "observe"),
          _stat(a.eligible, "eligible"),
          _stat(a.blessed, "blessed (auto)"),
          _stat(len(a.offers), "live offers"),
      )
  )
  if a.rules:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(r.payee)}</td>'
          f'<td class="mono mut">{escape(r.category)}</td>'
          f"<td>{_pill(r.trust, _TRUST_CLASS.get(r.trust, 'mute'))}</td>"
          f"<td>{escape(r.source)}</td>"
          f'<td class="num">{r.hits}</td>'
          f"<td>{'yes' if r.offered else '—'}</td>"
          "</tr>"
          for r in a.rules[:30]
      )
      table = (
          "<table><thead><tr><th>payee</th><th>category</th><th>trust</th>"
          "<th>source</th><th>hits</th><th>offered</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  else:
      table = '<p class="note">No learned rules yet.</p>'
  note = (
      '<p class="note">Autonomy is earned then granted: a rule becomes '
      "<b>eligible</b> after K consistent confirms, and <b>blessed</b> "
      "(auto-applies) only once the owner accepts the offer. A correction "
      "demotes it back to observe.</p>"
  )
  return (
      "<section><h2>Autonomy ladder</h2>"
      f'<div class="stats">{stats}</div>{bar}{table}{note}</section>'
  )
254 
255 
256def _queue(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.queue:
      body = '<p class="note">Nothing awaiting the owner.</p>'
  else:
      rows = "".join(
          "<tr>"
          f"<td>{_pill(q.kind, 'warn')}</td>"
          f'<td class="mono">{escape(q.label)}</td>'
          f'<td class="mut">{escape(_ago(q.since, now))}</td>'
          "</tr>"
          for q in model.queue[:30]
      )
      body = (
          "<table><thead><tr><th>kind</th><th>what</th><th>waiting</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>"
      )
  return f"<section><h2>Awaiting a human</h2>{body}</section>"
274 
275 
276def _budget(model: DashboardModel) -> str:
  b = model.budget
  if not b.available:
      return (
          "<section><h2>Budget &middot; YNAB</h2>"
          '<p class="note">Unavailable (YNAB not configured or '
          "unreachable).</p></section>"
      )
  sample = ", ".join(
      f"{escape(t.payee)} {escape(t.amount)}" for t in b.unapproved_sample
  )
  sample_note = f'<p class="note">{escape(sample)}</p>' if sample else ""
  if b.overspent:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(c.name)}</td>'
          f'<td class="bad mono">{escape(c.balance)}</td>'
          "</tr>"
          for c in b.overspent
      )
      over = (
          "<table><thead><tr><th>overspent category</th><th>balance</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>"
      )
  else:
      over = '<p class="note">No overspent categories.</p>'
  stats = "".join((_stat(b.unapproved, "unapproved"),))
  return (
      "<section><h2>Budget &middot; YNAB</h2>"
      f'<div class="stats">{stats}</div>{sample_note}'
      f'<div style="margin-top:12px">{over}</div></section>'
  )
308 
309 
310def _conversations(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.conversations:
      return ""
  rows = "".join(
      "<tr>"
      f"<td>{_pill(c.kind, 'mute')}</td>"
      f"<td>{escape(c.subject)}</td>"
      f'<td class="mut">{escape(c.preview)}</td>'
      f'<td class="mut">{escape(_ago(c.updated_at, now))}</td>'
      "</tr>"
      for c in model.conversations[:12]
  )
  return (
      "<section><h2>Conversations &middot; AgentMail</h2>"
      "<table><thead><tr><th>kind</th><th>subject</th><th>preview</th>"
      f"<th>updated</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
328 
329 
330def _dispatch(model: DashboardModel) -> str:
  d = model.dispatch
  stats = "".join(
      (
          _stat(d.transaction, "→ transaction"),
          _stat(d.offer, "→ offer"),
          _stat(d.receipt, "→ receipt"),
          _stat(d.command, "→ command"),
          _stat(d.quarantine, "quarantined"),
          _stat(d.ignore, "ignored"),
      )
  )
  return (
      "<section><h2>Inbound &middot; W3 dispatch "
      f'({d.total} recent)</h2><div class="stats">{stats}</div></section>'
  )
346 
347 
348def _telemetry(model: DashboardModel) -> str:
  t: RunTelemetry = model.telemetry
  now = model.generated_at
  if not t.available:
      return (
          "<section><h2>Run telemetry &middot; ClickHouse</h2>"
          '<p class="note">Unavailable (not configured or unreachable); '
          "Temporal carries the dashboard regardless.</p></section>"
      )
  if t.activities:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(a.name)}</td>'
          f'<td class="num">{a.count}</td>'
          f'<td class="mut">{a.avg_ms:.0f} ms</td>'
          f'<td class="mut">{a.max_ms:.0f} ms</td>'
          "</tr>"
          for a in t.activities
      )
      table = (
          "<table><thead><tr><th>activity</th><th>runs</th><th>avg</th>"
          f"<th>max</th></tr></thead><tbody>{rows}</tbody></table>"
      )
  else:
      table = '<p class="note">No spans in the window.</p>'
  errors = (
      "".join(
          f'<div class="row mono bad">{escape(e)}</div>'
          for e in t.recent_errors
      )
      if t.recent_errors
      else ""
  )
  summary = (
      f'<p class="note">{t.total_spans} spans &middot; '
      f"{t.error_spans} errored &middot; last "
      f"{escape(_ago(t.last_activity, now))} &middot; "
      f"{t.window_days}-day window.</p>"
  )
  return (
      "<section><h2>Run telemetry &middot; ClickHouse</h2>"
      f"{summary}{table}{errors}</section>"
  )
391 
392 
393def _failures(model: DashboardModel) -> str:
  if not model.failures:
      return ""
  now = model.generated_at
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(f.workflow_id)}</td>'
      f'<td class="bad">{escape(f.kind)}</td>'
      f'<td class="mut">{escape(f.reason or "—")}</td>'
      f'<td class="mut">{escape(_ago(f.when, now))}</td>'
      "</tr>"
      for f in model.failures
  )
  return (
      "<section><h2>Failures &middot; where a workflow did not close</h2>"
      "<table><thead><tr><th>workflow</th><th>kind</th><th>reason</th>"
      f"<th>when</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
411 
412 
413def _deploy(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.deploy.prs:
      return ""
  rows = "".join(
      "<tr>"
      f'<td><a href="{escape(p.url, quote=True)}">#{p.number}</a></td>'
      f"<td>{escape(p.title)}</td>"
      f'<td class="{_STATE_CLASS.get(p.state, "mut")}">{escape(p.state)}</td>'
      f"<td>{_ci_cell(p.ci)}</td>"
      f'<td class="mut">{escape(_ago(p.when, now))}</td>'
      "</tr>"
      for p in model.deploy.prs
  )
  return (
      "<section><h2>Deploy &middot; GitHub</h2>"
      "<table><thead><tr><th>pr</th><th>title</th><th>state</th><th>ci</th>"
      f"<th>opened</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
432 
433 
434def _ci_cell(ci: str | None) -> str:
  if ci is None:
      return '<span class="mut">—</span>'
  return f'<span class="{_CI_CLASS.get(ci, "mut")}">{escape(ci)}</span>'
438 
439 
440def _footer() -> str:
  return (
      "<footer><b>Safety envelope.</b> Every auto-apply passes the hard "
      "floor (amount ceiling, unreadable amount, per-run/day breaker) and "
      "the earned-autonomy gate (exactly one blessed rule), then a "
      "clean-context "
      "model <b>safety review</b> that can only hold it back. Autonomy is "
      "revoked — the rule demoted — on an explicit correction or a silent "
      "in-YNAB edit. Everything above is derived on this request from "
      "Temporal, YNAB, ClickHouse, AgentMail, and GitHub; nothing is stored. "
      "Reload to recompute.</footer>"
  )
452 
453 
454def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _heartbeat(model),
      _lifecycle(model),
      _ladder(model),
      _queue(model),
      _budget(model),
      _conversations(model),
      _dispatch(model),
      _telemetry(model),
      _failures(model),
      _deploy(model),
      _footer(),
  )
  return (
      '<!doctype html><html lang="en"><head><meta charset="utf-8">'
      '<meta name="viewport" content="width=device-width,initial-scale=1">'
      "<title>ynab-agent &middot; ops</title>"
      f"<style>{_CSS}</style></head><body><main>"
      + "".join(parts)
      + "</main></body></html>"
  )

The autonomy-ladder strip: flex segments sized to the three counts.

src/ynab_agent/dashboard/render.py · 477 lines

src/ynab_agent/dashboard/render.py477 lines · Python

⋯ 199 lines hidden (lines 1–199)1"""Render the view-model to one self-contained HTML page (pure).
2 
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
class:`~ynab_agent.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary. Ordering is ops-first (is it alive → what's
7flowing → earned autonomy → the human's queue → money → conversations → inbound
8→ telemetry → failures → deploy), so it reads top to bottom.
9"""
10 
11from __future__ import annotations
12 
13from datetime import UTC, datetime
14from html import escape
15from typing import TYPE_CHECKING
16 
17if TYPE_CHECKING:
  from ynab_agent.dashboard.model import (
      DashboardModel,
      Lifecycle,
      RunTelemetry,
  )
23 
24_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
26--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#f2f2f2}
27@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
28--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
29--accent:#58a6ff;--card:#1b1b1b}}
30*{box-sizing:border-box}
31body{margin:0;background:var(--bg);color:var(--fg);
32font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
33main{max-width:860px;margin:0 auto;padding:34px 20px 72px}
34h1{font-size:22px;margin:0;letter-spacing:-.01em}
tag{color:var(--mut);margin:3px 0 0;font-size:13px}
meta{color:var(--mut);font-size:12px;margin:12px 0 0}
sources{display:flex;flex-wrap:wrap;gap:16px;margin:10px 0 0;font-size:12px}
38section{margin:30px 0 0}
39h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
40color:var(--mut);margin:0 0 12px;font-weight:600;
41border-bottom:1px solid var(--line);padding-bottom:6px}
stats{display:flex;flex-wrap:wrap;gap:26px}
stat .n{font-size:26px;font-weight:600;line-height:1.1}
stat .l{color:var(--mut);font-size:12px;margin-top:2px}
dot{display:inline-block;width:9px;height:9px;border-radius:50%;margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
48table{width:100%;border-collapse:collapse;font-size:13px}
49th,td{text-align:left;padding:6px 12px 6px 0;vertical-align:top;
50border-bottom:1px solid var(--line)}
51th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
52letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
54a{color:var(--accent);text-decoration:none}a:hover{text-decoration:underline}
row{display:flex;align-items:baseline;gap:8px;padding:4px 0}
mut{color:var(--mut)}.ok{color:var(--ok)}.warn{color:var(--warn)}.bad{color:var(--bad)}
note{color:var(--mut);font-size:12px;margin:10px 0 0}
barrow{display:grid;grid-template-columns:140px 1fr 34px;gap:10px;
59align-items:center;margin:5px 0}
barrow .lab{font-size:12px;color:var(--mut)}
barrow .num{font-size:12px;text-align:right;font-variant-numeric:tabular-nums}
bar{background:var(--card);border-radius:4px;height:16px;overflow:hidden}
bar>span{display:block;height:100%;background:var(--accent);min-width:2px}
ladder{display:flex;height:24px;border-radius:5px;overflow:hidden;
65margin:2px 0 10px;font-size:11px;font-weight:600}
ladder>span{display:flex;align-items:center;justify-content:center;color:#fff;
67white-space:nowrap;min-width:0}
ladder .obs{background:var(--mut)}.ladder .elig{background:var(--warn)}
ladder .bless{background:var(--ok)}
pill{display:inline-block;padding:1px 7px;border-radius:10px;font-size:11px;
71font-weight:600}
pill.ok{background:rgba(63,185,80,.16);color:var(--ok)}
pill.warn{background:rgba(210,153,34,.16);color:var(--warn)}
pill.mute{background:var(--card);color:var(--mut)}
75footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
76color:var(--mut);font-size:12px;line-height:1.7}
77footer b{color:var(--fg);font-weight:600}
78"""
79 
80# The canonical W2 lifecycle order for the funnel (any other state is appended).
81_LIFECYCLE_ORDER = (
  "discovered",
  "enriching",
  "hold_amazon",
  "awaiting_human",
  "open",
  "lapsed",
  "revising",
89)
90_TRUST_CLASS = {"trusted": "ok", "confirmed": "warn", "suggested": "mute"}
91_CI_CLASS = {"passed": "ok", "failed": "bad", "running": "warn"}
92_STATE_CLASS = {"merged": "ok", "open": "warn", "closed": "mut"}
93 
94 
95def _aware(when: datetime) -> datetime:
  return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
97 
98 
99def _ago(when: datetime | None, now: datetime) -> str:
  if when is None:
      return "—"
  secs = (now - _aware(when)).total_seconds()
  if secs < 0:
      return "just now"
  if secs < 90:
      return "just now"
  if secs < 5400:
      return f"{int(secs // 60)}m ago"
  if secs < 129600:
      return f"{int(secs // 3600)}h ago"
  return f"{int(secs // 86400)}d ago"
112 
113 
114def _dot(kind: str) -> str:
  return f'<span class="dot {kind}"></span>'
116 
117 
118def _stat(n: object, label: str) -> str:
  return (
      f'<div class="stat"><div class="n">{escape(str(n))}</div>'
      f'<div class="l">{escape(label)}</div></div>'
  )
123 
124 
125def _pill(text: str, cls: str) -> str:
  return f'<span class="pill {cls}">{escape(text)}</span>'
127 
128 
129def _header(model: DashboardModel) -> str:
  stamp = model.generated_at.strftime("%Y-%m-%d %H:%M UTC")
  dots = "".join(
      f"<span>{_dot('ok' if s.ok else 'bad')}{escape(s.name)} "
      f'<span class="mut">{escape(s.detail)}</span></span>'
      for s in model.sources
  )
  return (
      "<header><h1>ynab-agent</h1>"
      '<p class="tag">durable transaction triage &middot; '
      "earned-autonomy read-model</p>"
      f'<p class="meta">{escape(model.repo)} &middot; generated '
      f"{escape(stamp)} &middot; derived live, stored nowhere</p>"
      f'<div class="sources">{dots}</div></header>'
  )
144 
145 
146def _heartbeat(model: DashboardModel) -> str:
  h = model.heartbeat
  now = model.generated_at
  poll_dot = "ok" if h.poll_live else "bad"
  poll = (
      f'<div class="row">{_dot(poll_dot)}<span class="mono">W1 poll</span> '
      f'<span class="mut">&middot; {escape(h.poll_status)} &middot; last '
      f"{escape(_ago(h.poll_last_start, now))}</span></div>"
  )
  if h.worker_last_span is not None:
      span_ago = (now - _aware(h.worker_last_span)).total_seconds()
      wdot = "ok" if span_ago < 3600 else "warn"
      worker = (
          f'<div class="row">{_dot(wdot)}<span class="mono">worker</span> '
          f'<span class="mut">&middot; last span '
          f"{escape(_ago(h.worker_last_span, now))}</span></div>"
      )
  else:
      worker = (
          f'<div class="row">{_dot("mute")}'
          '<span class="mono">worker</span> '
          '<span class="mut">&middot; no telemetry</span></div>'
      )
  return f"<section><h2>Is it alive?</h2>{poll}{worker}</section>"
170 
171 
172def _lifecycle(model: DashboardModel) -> str:
  lc: Lifecycle = model.lifecycle
  counts = {s.state: s.count for s in lc.states}
  ordered = [s for s in _LIFECYCLE_ORDER if s in counts]
  ordered += [s for s in sorted(counts) if s not in _LIFECYCLE_ORDER]
  peak = max((counts[s] for s in ordered), default=0) or 1
  rows = "".join(
      f'<div class="barrow"><div class="lab">{escape(s)}</div>'
      f'<div class="bar"><span style="width:{counts[s] * 100 // peak}%">'
      f'</span></div><div class="num">{counts[s]}</div></div>'
      for s in ordered
  )
  if not ordered:
      rows = '<p class="note">No transactions in flight.</p>'
  stats = "".join(
      (
          _stat(lc.in_flight, "in flight"),
          _stat(lc.archived, "archived (recent)"),
          _stat(lc.terminated, "terminated (recent)"),
      )
  )
  return (
      "<section><h2>Transaction lifecycle</h2>"
      f'<div class="stats">{stats}</div>'
      f'<div style="margin-top:14px">{rows}</div></section>'
  )
198 
199 
200def _ladder(model: DashboardModel) -> str:
  a = model.autonomy
  total = a.observe + a.eligible + a.blessed
  if total:
      segs = "".join(
          f'<span class="{cls}" style="flex:{n}" title="{label}: {n}">'
          f"{n if n * 100 // total >= 8 else ''}</span>"
          for n, cls, label in (
              (a.observe, "obs", "observe"),
              (a.eligible, "elig", "eligible"),
              (a.blessed, "bless", "blessed"),
          )
          if n
      )
      bar = f'<div class="ladder">{segs}</div>'
  else:
      bar = ""
  stats = "".join(
      (
          _stat(a.observe, "observe"),
          _stat(a.eligible, "eligible"),
          _stat(a.blessed, "blessed (auto)"),
          _stat(len(a.offers), "live offers"),
      )
  )
⋯ 253 lines hidden (lines 225–477)225    if a.rules:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(r.payee)}</td>'
          f'<td class="mono mut">{escape(r.category)}</td>'
          f"<td>{_pill(r.trust, _TRUST_CLASS.get(r.trust, 'mute'))}</td>"
          f"<td>{escape(r.source)}</td>"
          f'<td class="num">{r.hits}</td>'
          f"<td>{'yes' if r.offered else '—'}</td>"
          "</tr>"
          for r in a.rules[:30]
      )
      table = (
          "<table><thead><tr><th>payee</th><th>category</th><th>trust</th>"
          "<th>source</th><th>hits</th><th>offered</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  else:
      table = '<p class="note">No learned rules yet.</p>'
  note = (
      '<p class="note">Autonomy is earned then granted: a rule becomes '
      "<b>eligible</b> after K consistent confirms, and <b>blessed</b> "
      "(auto-applies) only once the owner accepts the offer. A correction "
      "demotes it back to observe.</p>"
  )
  return (
      "<section><h2>Autonomy ladder</h2>"
      f'<div class="stats">{stats}</div>{bar}{table}{note}</section>'
  )
254 
255 
256def _queue(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.queue:
      body = '<p class="note">Nothing awaiting the owner.</p>'
  else:
      rows = "".join(
          "<tr>"
          f"<td>{_pill(q.kind, 'warn')}</td>"
          f'<td class="mono">{escape(q.label)}</td>'
          f'<td class="mut">{escape(_ago(q.since, now))}</td>'
          "</tr>"
          for q in model.queue[:30]
      )
      body = (
          "<table><thead><tr><th>kind</th><th>what</th><th>waiting</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>"
      )
  return f"<section><h2>Awaiting a human</h2>{body}</section>"
274 
275 
276def _budget(model: DashboardModel) -> str:
  b = model.budget
  if not b.available:
      return (
          "<section><h2>Budget &middot; YNAB</h2>"
          '<p class="note">Unavailable (YNAB not configured or '
          "unreachable).</p></section>"
      )
  sample = ", ".join(
      f"{escape(t.payee)} {escape(t.amount)}" for t in b.unapproved_sample
  )
  sample_note = f'<p class="note">{escape(sample)}</p>' if sample else ""
  if b.overspent:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(c.name)}</td>'
          f'<td class="bad mono">{escape(c.balance)}</td>'
          "</tr>"
          for c in b.overspent
      )
      over = (
          "<table><thead><tr><th>overspent category</th><th>balance</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>"
      )
  else:
      over = '<p class="note">No overspent categories.</p>'
  stats = "".join((_stat(b.unapproved, "unapproved"),))
  return (
      "<section><h2>Budget &middot; YNAB</h2>"
      f'<div class="stats">{stats}</div>{sample_note}'
      f'<div style="margin-top:12px">{over}</div></section>'
  )
308 
309 
310def _conversations(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.conversations:
      return ""
  rows = "".join(
      "<tr>"
      f"<td>{_pill(c.kind, 'mute')}</td>"
      f"<td>{escape(c.subject)}</td>"
      f'<td class="mut">{escape(c.preview)}</td>'
      f'<td class="mut">{escape(_ago(c.updated_at, now))}</td>'
      "</tr>"
      for c in model.conversations[:12]
  )
  return (
      "<section><h2>Conversations &middot; AgentMail</h2>"
      "<table><thead><tr><th>kind</th><th>subject</th><th>preview</th>"
      f"<th>updated</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
328 
329 
330def _dispatch(model: DashboardModel) -> str:
  d = model.dispatch
  stats = "".join(
      (
          _stat(d.transaction, "→ transaction"),
          _stat(d.offer, "→ offer"),
          _stat(d.receipt, "→ receipt"),
          _stat(d.command, "→ command"),
          _stat(d.quarantine, "quarantined"),
          _stat(d.ignore, "ignored"),
      )
  )
  return (
      "<section><h2>Inbound &middot; W3 dispatch "
      f'({d.total} recent)</h2><div class="stats">{stats}</div></section>'
  )
346 
347 
348def _telemetry(model: DashboardModel) -> str:
  t: RunTelemetry = model.telemetry
  now = model.generated_at
  if not t.available:
      return (
          "<section><h2>Run telemetry &middot; ClickHouse</h2>"
          '<p class="note">Unavailable (not configured or unreachable); '
          "Temporal carries the dashboard regardless.</p></section>"
      )
  if t.activities:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(a.name)}</td>'
          f'<td class="num">{a.count}</td>'
          f'<td class="mut">{a.avg_ms:.0f} ms</td>'
          f'<td class="mut">{a.max_ms:.0f} ms</td>'
          "</tr>"
          for a in t.activities
      )
      table = (
          "<table><thead><tr><th>activity</th><th>runs</th><th>avg</th>"
          f"<th>max</th></tr></thead><tbody>{rows}</tbody></table>"
      )
  else:
      table = '<p class="note">No spans in the window.</p>'
  errors = (
      "".join(
          f'<div class="row mono bad">{escape(e)}</div>'
          for e in t.recent_errors
      )
      if t.recent_errors
      else ""
  )
  summary = (
      f'<p class="note">{t.total_spans} spans &middot; '
      f"{t.error_spans} errored &middot; last "
      f"{escape(_ago(t.last_activity, now))} &middot; "
      f"{t.window_days}-day window.</p>"
  )
  return (
      "<section><h2>Run telemetry &middot; ClickHouse</h2>"
      f"{summary}{table}{errors}</section>"
  )
391 
392 
393def _failures(model: DashboardModel) -> str:
  if not model.failures:
      return ""
  now = model.generated_at
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(f.workflow_id)}</td>'
      f'<td class="bad">{escape(f.kind)}</td>'
      f'<td class="mut">{escape(f.reason or "—")}</td>'
      f'<td class="mut">{escape(_ago(f.when, now))}</td>'
      "</tr>"
      for f in model.failures
  )
  return (
      "<section><h2>Failures &middot; where a workflow did not close</h2>"
      "<table><thead><tr><th>workflow</th><th>kind</th><th>reason</th>"
      f"<th>when</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
411 
412 
413def _deploy(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.deploy.prs:
      return ""
  rows = "".join(
      "<tr>"
      f'<td><a href="{escape(p.url, quote=True)}">#{p.number}</a></td>'
      f"<td>{escape(p.title)}</td>"
      f'<td class="{_STATE_CLASS.get(p.state, "mut")}">{escape(p.state)}</td>'
      f"<td>{_ci_cell(p.ci)}</td>"
      f'<td class="mut">{escape(_ago(p.when, now))}</td>'
      "</tr>"
      for p in model.deploy.prs
  )
  return (
      "<section><h2>Deploy &middot; GitHub</h2>"
      "<table><thead><tr><th>pr</th><th>title</th><th>state</th><th>ci</th>"
      f"<th>opened</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
432 
433 
434def _ci_cell(ci: str | None) -> str:
  if ci is None:
      return '<span class="mut">—</span>'
  return f'<span class="{_CI_CLASS.get(ci, "mut")}">{escape(ci)}</span>'
438 
439 
440def _footer() -> str:
  return (
      "<footer><b>Safety envelope.</b> Every auto-apply passes the hard "
      "floor (amount ceiling, unreadable amount, per-run/day breaker) and "
      "the earned-autonomy gate (exactly one blessed rule), then a "
      "clean-context "
      "model <b>safety review</b> that can only hold it back. Autonomy is "
      "revoked — the rule demoted — on an explicit correction or a silent "
      "in-YNAB edit. Everything above is derived on this request from "
      "Temporal, YNAB, ClickHouse, AgentMail, and GitHub; nothing is stored. "
      "Reload to recompute.</footer>"
  )
452 
453 
454def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _heartbeat(model),
      _lifecycle(model),
      _ladder(model),
      _queue(model),
      _budget(model),
      _conversations(model),
      _dispatch(model),
      _telemetry(model),
      _failures(model),
      _deploy(model),
      _footer(),
  )
  return (
      '<!doctype html><html lang="en"><head><meta charset="utf-8">'
      '<meta name="viewport" content="width=device-width,initial-scale=1">'
      "<title>ynab-agent &middot; ops</title>"
      f"<style>{_CSS}</style></head><body><main>"
      + "".join(parts)
      + "</main></body></html>"
  )

Verification

make check is green — ruff, mypy, 468 tests (19 new). The pure core (render, read_model) renders a full page from any model and degrades cleanly; each source's helpers and its "off" path are unit-tested; and a fake-Temporal-client test drives fetch end to end (funnel counts, the ladder fold, the dispatch tally, failures) plus its degrade-to-error path. Live smoke: kubectl port-forward deploy/ynab-agent-worker 8080:8080 → open http://localhost:8080/.

The ClickHouse and GitHub panels are best-effort: they show "off" until YNAB_AGENT_CLICKHOUSE_PASSWORD / YNAB_AGENT_GITHUB_TOKEN are set (the infra change in the zo repo adds the container port + those optional secrets).