froot · dashboard determinism coverage

Overview — a second loop on the page

The read-model dashboard was single-loop: it showed the dependency-patch scan/bump loop and nothing else. This change adds the determinism reviewer (the transitive ring) to the same page, derived the same way — Temporal as the recent ledger, every figure computed in the read-model and unit-tested apart from any I/O.

The view model

Three new frozen types, each a twin of an existing one: ReviewLoop (liveness, like ScanLoop), ReviewRecord (the headline, like TrackRecord), and ReviewRow (one per-PR review, like BumpRow). They carry already-computed numbers, so the renderer stays a projection.

ReviewLoop · ReviewRow · ReviewRecord · 308 lines

ReviewLoop · ReviewRow · ReviewRecord308 lines · Python

⋯ 204 lines hidden (lines 1–204)1"""The dashboard's view model — pure, frozen, fully derived.
2 
3These types are the shape the renderer projects to HTML. They carry *already
4computed* numbers (the aggregates live in :mod:`~froot.dashboard.read_model`),
5so the renderer is a dumb projection and every figure on the page is
6unit-tested apart from any I/O. Nothing here is persisted: a
class:`DashboardModel` is built fresh per request and discarded after the
8response (derive, never store).
9"""
10 
11from __future__ import annotations
12 
13from datetime import datetime
14 
15from froot.domain.base import Frozen
16 
17 
18class SourceHealth(Frozen):
  """Whether one external truth answered this request.
20 
  Attributes:
      name: The source label (``github`` / ``temporal`` / ``clickhouse``).
      ok: True when the source returned data; False when it errored or is off.
      detail: A short human note — the row count, ``off``, or the error.
  """
26 
  name: str
  ok: bool
  detail: str
30 
31 
32class ScanLoop(Frozen):
  """The liveness of one repo's durable scan schedule (the signal stage).
34 
  Attributes:
      repo: The ``owner/name`` slug this loop watches.
      status: The current scan workflow status (``running`` /
          ``continued_as_new`` / ``terminated`` / ``none`` / ...), lowercased.
      live: True when the loop is actively self-scheduling (running / CAN).
      last_tick: When the current scan execution started (≈ the last tick),
          or ``None`` if no scan workflow exists for the repo.
  """
43 
  repo: str
  status: str
  live: bool
  last_tick: datetime | None
48 
49 
50class BumpRow(Frozen):
  """One proposed dependency bump, joined across GitHub and Temporal.
52 
  GitHub is authoritative for the outcome (state / timestamps / PR); Temporal
  enriches with the model verdict and the CI reading while it is still in the
  7-day window, with the PR body as the durable fallback for the verdict.
56 
  Attributes:
      repo: The ``owner/name`` slug.
      package: The bumped dependency.
      from_version: The version bumped from, if known.
      to_version: The version bumped to.
      state: The GitHub PR state — ``open`` / ``merged`` / ``closed``.
      verdict: The changelog verdict (``clean`` / ``risky`` / ``unknown``), or
          ``None`` when neither Temporal nor the PR body yields one.
      ci: The terminal CI reading (``passed`` / ``failed`` / ``absent`` /
          ``timed_out``), or ``None`` when only durable GitHub data remains.
      pr_number: The PR number, if a PR exists.
      pr_url: The PR URL, if a PR exists.
      opened_at: When the PR was opened.
      merged_at: When the PR was merged, if it was.
      ttm_minutes: Time-to-merge in minutes (merged - opened), if merged.
      age_hours: Age in hours for a still-open PR (now - opened), or ``None``.
  """
74 
  repo: str
  package: str
  from_version: str | None
  to_version: str
  state: str
  verdict: str | None
  ci: str | None
  pr_number: int | None
  pr_url: str | None
  opened_at: datetime | None
  merged_at: datetime | None
  ttm_minutes: float | None
  age_hours: float | None
88 
89 
90class Failure(Frozen):
  """A bump loop that did not close — the honest friction signal.
92 
  Attributes:
      workflow_id: The Temporal workflow id (encodes repo/package/target).
      kind: ``terminated`` or ``failed``.
      reason: The human-readable termination/failure reason, if recovered.
      when: When the workflow closed, if known.
  """
99 
  workflow_id: str
  kind: str
  reason: str | None
  when: datetime | None
104 
105 
106class ActivityStat(Frozen):
  """Latency of one activity stage, from traces (run-telemetry enrichment).
108 
  Attributes:
      name: The activity name (``scan_candidates``, ``open_pull_request``).
      count: How many executions in the window.
      avg_ms: Mean duration in milliseconds.
      max_ms: Max duration in milliseconds.
  """
115 
  name: str
  count: int
  avg_ms: float
  max_ms: float
120 
121 
122class RunTelemetry(Frozen):
  """Trace-derived run telemetry from ClickHouse, or an unavailable marker.
124 
  Attributes:
      available: True when ClickHouse answered with froot traces.
      total_spans: Total froot spans in the window.
      error_spans: Spans that ended in an error.
      last_activity: The most recent froot span timestamp.
      window_days: The look-back window the figures cover.
      activities: Per-stage latency rows.
  """
133 
  available: bool
  total_spans: int
  error_spans: int
  last_activity: datetime | None
  window_days: int
  activities: tuple[ActivityStat, ...]
140 
141 
142class TrackRecord(Frozen):
  """The reputation headline, derived from GitHub PR outcomes.
144 
  Attributes:
      opened: Total bumps froot has proposed.
      merged: How many a human merged.
      closed_unmerged: How many were closed without merging.
      open_now: How many are still awaiting a decision.
      merge_rate: ``merged / (merged + closed_unmerged)``, or ``None`` if none
          have been decided yet.
      median_ttm_minutes: Median time-to-merge across merged PRs, or ``None``.
  """
154 
  opened: int
  merged: int
  closed_unmerged: int
  open_now: int
  merge_rate: float | None
  median_ttm_minutes: float | None
161 
162 
163class Verification(Frozen):
  """The CI-oracle breakdown — kept honest about whether an oracle existed.
165 
  Attributes:
      passed: Bumps whose CI went green.
      failed: Bumps whose CI went red.
      absent: Bumps where no CI check existed (no oracle).
      timed_out: Bumps where froot stopped waiting on CI.
      unknown: Bumps with no recoverable CI reading (aged out of Temporal).
      oracle_existed: Bumps where a real oracle reported (passed + failed).
      with_reading: Bumps with any CI reading at all (excludes ``unknown``).
  """
175 
  passed: int
  failed: int
  absent: int
  timed_out: int
  unknown: int
  oracle_existed: int
  with_reading: int
183 
184 
185class Judgment(Frozen):
  """The model's changelog-verdict mix and its calibration against CI.
187 
  Attributes:
      clean: Verdicts of ``clean``.
      risky: Verdicts of ``risky``.
      unknown: Verdicts of ``unknown``.
      none: Bumps with no recoverable verdict.
      clean_but_failed: ``clean`` verdicts whose CI failed (mis-judged).
      flagged_but_passed: ``risky``/``unknown`` verdicts whose CI passed.
  """
196 
  clean: int
  risky: int
  unknown: int
  none: int
  clean_but_failed: int
  flagged_but_passed: int
203 
204 
205class ReviewLoop(Frozen):
  """Liveness of one repo's determinism-review loop (the transitive ring).
207 
  Attributes:
      repo: The ``owner/name`` slug this loop reviews.
      status: The review workflow status (``running`` /
          ``continued_as_new`` / ``terminated`` / ...), lowercased.
      live: True when the loop is actively self-scheduling.
      last_tick: When the current review execution started (≈ the last tick),
          or ``None`` if no review workflow exists for the repo.
  """
216 
  repo: str
  status: str
  live: bool
  last_tick: datetime | None
221 
222 
223class ReviewRow(Frozen):
  """One per-PR determinism review, from its ``PrReviewWorkflow`` result.
225 
  Attributes:
      repo: The ``owner/name`` slug the PR belongs to.
      pr_number: The reviewed PR number, if known.
      pr_url: The PR's web URL, if it can be formed.
      head_sha: The head commit the review ran against.
      findings: How many transitive hazards the review surfaced.
      rules: The distinct banned calls flagged (``datetime.datetime.now``…).
      comment_url: The advisory comment's URL, if one was posted.
      status: The review workflow status (``completed`` / ``running`` / ...).
      reviewed_at: When the review closed, or started if still running.
  """
237 
  repo: str
  pr_number: int | None
  pr_url: str | None
  head_sha: str | None
  findings: int
  rules: tuple[str, ...]
  comment_url: str | None
  status: str
  reviewed_at: datetime | None
247 
248 
249class ReviewRecord(Frozen):
  """The determinism reviewer's headline, derived from completed reviews.
251 
  The hazard-resolved rate (was a flagged hazard gone on a later commit?) is
  a later addition — it needs accumulated cross-commit history, so it is not
  here yet.
255 
  Attributes:
      reviewed: Completed per-PR reviews in the recent window.
      flagged: Reviews that surfaced at least one hazard.
      clean: Reviews that surfaced none.
      hazards: Total hazards surfaced across all reviews.
      repos_covered: Distinct repos with a live review loop.
  """
263 
  reviewed: int
  flagged: int
  clean: int
  hazards: int
  repos_covered: int
269 
⋯ 39 lines hidden (lines 270–308)270 
271class DashboardModel(Frozen):
  """The whole 10,000ft view, fully derived and ready to render.
273 
  Attributes:
      generated_at: When this view was assembled (UTC).
      repos_configured: The repos froot is pointed at (``FROOT_REPOS``).
      scan_interval_seconds: The configured gap between scan ticks.
      sources: Per-source health for this request.
      scan_loops: Liveness of each repo's scan schedule.
      track_record: The reputation headline.
      verification: The CI-oracle breakdown.
      judgment: The model-verdict mix and calibration.
      gate: Open PRs awaiting a human, the freshest last.
      bumps: Every proposed bump, newest first (the detail behind the stats).
      failures: Bump loops that did not close.
      review_interval_seconds: The configured gap between review poll ticks.
      review_loops: Liveness of each Temporal repo's determinism-review loop.
      review_record: The determinism reviewer's headline.
      reviews: Every per-PR determinism review, newest first.
      telemetry: Trace-derived run telemetry (best-effort).
  """
292 
  generated_at: datetime
  repos_configured: tuple[str, ...]
  scan_interval_seconds: int
  sources: tuple[SourceHealth, ...]
  scan_loops: tuple[ScanLoop, ...]
  track_record: TrackRecord
  verification: Verification
  judgment: Judgment
  gate: tuple[BumpRow, ...]
  bumps: tuple[BumpRow, ...]
  failures: tuple[Failure, ...]
  review_interval_seconds: int
  review_loops: tuple[ReviewLoop, ...]
  review_record: ReviewRecord
  reviews: tuple[ReviewRow, ...]
  telemetry: RunTelemetry

Reading the review out of Temporal

The Temporal reader already lists ScanWorkflow/BumpWorkflow; now it also lists ReviewWorkflow (loop liveness) and PrReviewWorkflow. For a completed per-PR review it decodes the PrReviewResult — the finding count, the flagged rules, and the advisory comment URL — defensively, so a bad decode is never fatal to the page.

PrReviewExecution · 335 lines

PrReviewExecution335 lines · Python

⋯ 70 lines hidden (lines 1–70)1"""Temporal reader: the live run ledger (scan loops + bump outcomes).
2 
3Reuses the worker's own connected client. Lists the durable scan loops (the
4signal stage's heartbeat) and the per-bump workflows; for a completed bump it
5reads the structured outcome from the workflow result (verdict + CI reading + PR
6number), and for a terminated/failed one it recovers the human reason from
7history. Everything is keyed off the deterministic workflow ids, so the
8read-model joins to GitHub by PR number and to repos by id prefix without
9parsing ambiguous slugs. Temporal keeps ~7 days, so this is the *recent* ledger;
10GitHub is the durable one.
11 
12Never raises: a failure returns an error string and whatever was gathered.
13"""
14 
15from __future__ import annotations
16 
17import re
18from datetime import datetime
19from typing import TYPE_CHECKING, Any, Final
20 
21from temporalio.client import WorkflowExecutionStatus
22 
23from froot.domain.base import Frozen
24 
25if TYPE_CHECKING:
  from collections.abc import AsyncIterator
27 
  from temporalio.client import Client, WorkflowExecution
29 
30# A backstop so a runaway visibility store can never spin this forever.
31_MAX_PER_TYPE: Final = 500
32# The owner/name slug out of a PR url — the repo-aware join key (see fetch).
33_PR_URL: Final = re.compile(r"github\.com/([^/]+/[^/]+)/pull/\d+")
34 
35 
36class ScanExecution(Frozen):
  """One scan-loop execution (there may be several per id across CAN)."""
38 
  workflow_id: str
  status: str
  start: datetime | None
42 
43 
44class BumpExecution(Frozen):
  """One bump workflow, with its outcome (if completed) or reason (if not).
46 
  ``repo`` is the ``owner/name`` the PR was opened against, parsed from the
  outcome's PR url; it makes the GitHub join repo-aware so two repos' PRs that
  share a number cannot cross-attribute.
  """
51 
  workflow_id: str
  status: str
  start: datetime | None
  close: datetime | None
  verdict: str | None
  ci: str | None
  pr_number: int | None
  repo: str | None
  reason: str | None
61 
62 
63class ReviewExecution(Frozen):
  """One determinism-review-loop execution (several per id across CAN)."""
65 
  workflow_id: str
  status: str
  start: datetime | None
69 
70 
71class PrReviewExecution(Frozen):
  """One per-PR determinism review, with its result (if completed).
73 
  The findings count + flagged rules + advisory comment come from the
  ``PrReviewWorkflow`` result; the repo is recovered by the read-model from
  the deterministic workflow id (it encodes the slug).
  """
78 
  workflow_id: str
  status: str
  start: datetime | None
  close: datetime | None
  pr_number: int | None
  head_sha: str | None
  findings: int
  rules: tuple[str, ...]
  comment_url: str | None
88 
89 
⋯ 246 lines hidden (lines 90–335)90def _status(execution: WorkflowExecution) -> str:
  """Lowercase the Temporal status enum (``continued_as_new`` etc.)."""
  status = execution.status
  return status.name.lower() if status is not None else "unknown"
94 
95 
96def _as_dict(result: object) -> dict[str, Any] | None:
  """Normalise a decoded workflow result to a dict (converter-agnostic)."""
  if isinstance(result, dict):
      return result
  dump = getattr(result, "model_dump", None)
  if callable(dump):
      dumped = dump(mode="json")
      return dumped if isinstance(dumped, dict) else None
  return None
105 
106 
107def _nested_kind(data: dict[str, Any], key: str) -> str | None:
  """Read ``data[key]["kind"]`` defensively (a discriminated-union tag)."""
  section = data.get(key)
  if isinstance(section, dict):
      kind = section.get("kind")
      if isinstance(kind, str):
          return kind
  return None
115 
116 
117def _pr_number(data: dict[str, Any]) -> int | None:
  """Read ``data["pr"]["number"]`` defensively."""
  pr = data.get("pr")
  if isinstance(pr, dict):
      number = pr.get("number")
      if isinstance(number, int):
          return number
  return None
125 
126 
127def _pr_repo(data: dict[str, Any]) -> str | None:
  """The ``owner/name`` slug from ``data["pr"]["url"]`` defensively."""
  pr = data.get("pr")
  url = pr.get("url") if isinstance(pr, dict) else None
  if isinstance(url, str):
      match = _PR_URL.search(url)
      if match is not None:
          return match.group(1)
  return None
136 
137 
138class _Outcome(Frozen):
  """The bits of a completed bump's outcome the read-model joins on."""
140 
  verdict: str | None
  ci: str | None
  pr_number: int | None
  repo: str | None
145 
146 
147async def _outcome(client: Client, execution: WorkflowExecution) -> _Outcome:
  """A completed bump's outcome (verdict/ci/pr/repo), or all ``None``."""
  empty = _Outcome(verdict=None, ci=None, pr_number=None, repo=None)
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      data = _as_dict(await handle.result())
  except Exception:  # best-effort enrichment — a bad decode is never fatal
      return empty
  if data is None:
      return empty
  return _Outcome(
      verdict=_nested_kind(data, "verdict"),
      ci=_nested_kind(data, "ci"),
      pr_number=_pr_number(data),
      repo=_pr_repo(data),
  )
165 
166 
167async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
  """Recover a terminated/failed bump'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 page for it
      return None
184 
185 
186class _ReviewOutcome(Frozen):
  """The bits of a completed review's result the read-model joins on."""
188 
  pr_number: int | None
  head_sha: str | None
  findings: int
  rules: tuple[str, ...]
  comment_url: str | None
194 
195 
196async def _review_outcome(
  client: Client, execution: WorkflowExecution
198) -> _ReviewOutcome:
  """A completed review's result (pr/head/findings/rules/comment)."""
  empty = _ReviewOutcome(
      pr_number=None, head_sha=None, findings=0, rules=(), comment_url=None
  )
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      data = _as_dict(await handle.result())
  except Exception:  # best-effort enrichment — a bad decode is never fatal
      return empty
  if data is None:
      return empty
  raw = data.get("findings")
  findings = raw if isinstance(raw, list) else []
  rules = tuple(
      str(f["rule"])
      for f in findings
      if isinstance(f, dict) and isinstance(f.get("rule"), str)
  )
  number = data.get("pr_number")
  head = data.get("head_sha")
  comment = data.get("comment_url")
  return _ReviewOutcome(
      pr_number=number if isinstance(number, int) else None,
      head_sha=head if isinstance(head, str) else None,
      findings=len(findings),
      rules=rules,
      comment_url=comment if isinstance(comment, str) else None,
  )
229 
230 
231type _Ledger = tuple[
  tuple[ScanExecution, ...],
  tuple[BumpExecution, ...],
  tuple[ReviewExecution, ...],
  tuple[PrReviewExecution, ...],
236]
237 
238 
239async def fetch(client: Client) -> tuple[_Ledger, str | None]:
  """Read froot's scan/bump + review executions (degrades to an error)."""
  scans: list[ScanExecution] = []
  bumps: list[BumpExecution] = []
  reviews: list[ReviewExecution] = []
  pr_reviews: list[PrReviewExecution] = []
245 
  def ledger() -> _Ledger:
      return (tuple(scans), tuple(bumps), tuple(reviews), tuple(pr_reviews))
248 
  try:
      async for execution in _take(
          client.list_workflows("WorkflowType = 'ScanWorkflow'")
      ):
          scans.append(
              ScanExecution(
                  workflow_id=execution.id,
                  status=_status(execution),
                  start=execution.start_time,
              )
          )
      async for execution in _take(
          client.list_workflows("WorkflowType = 'BumpWorkflow'")
      ):
          status = _status(execution)
          outcome = _Outcome(verdict=None, ci=None, pr_number=None, repo=None)
          reason: str | None = None
          if execution.status is WorkflowExecutionStatus.COMPLETED:
              outcome = await _outcome(client, execution)
          elif execution.status in (
              WorkflowExecutionStatus.TERMINATED,
              WorkflowExecutionStatus.FAILED,
          ):
              reason = await _reason(client, execution)
          bumps.append(
              BumpExecution(
                  workflow_id=execution.id,
                  status=status,
                  start=execution.start_time,
                  close=execution.close_time,
                  verdict=outcome.verdict,
                  ci=outcome.ci,
                  pr_number=outcome.pr_number,
                  repo=outcome.repo,
                  reason=reason,
              )
          )
      async for execution in _take(
          client.list_workflows("WorkflowType = 'ReviewWorkflow'")
      ):
          reviews.append(
              ReviewExecution(
                  workflow_id=execution.id,
                  status=_status(execution),
                  start=execution.start_time,
              )
          )
      async for execution in _take(
          client.list_workflows("WorkflowType = 'PrReviewWorkflow'")
      ):
          review = _ReviewOutcome(
              pr_number=None,
              head_sha=None,
              findings=0,
              rules=(),
              comment_url=None,
          )
          if execution.status is WorkflowExecutionStatus.COMPLETED:
              review = await _review_outcome(client, execution)
          pr_reviews.append(
              PrReviewExecution(
                  workflow_id=execution.id,
                  status=_status(execution),
                  start=execution.start_time,
                  close=execution.close_time,
                  pr_number=review.pr_number,
                  head_sha=review.head_sha,
                  findings=review.findings,
                  rules=review.rules,
                  comment_url=review.comment_url,
              )
          )
  except Exception as exc:  # degrade to an error string, never crash the page
      return ledger(), f"{type(exc).__name__}: {exc}"
  return ledger(), None
324 
325 
326async def _take(
  iterator: AsyncIterator[WorkflowExecution],
328) -> AsyncIterator[WorkflowExecution]:
  """Yield at most :data:`_MAX_PER_TYPE` executions (a runaway backstop)."""
  seen = 0
  async for execution in iterator:
      yield execution
      seen += 1
      if seen >= _MAX_PER_TYPE:
          return

_review_outcome — decode the PrReviewResult · 335 lines

_review_outcome — decode the PrReviewResult335 lines · Python

⋯ 195 lines hidden (lines 1–195)1"""Temporal reader: the live run ledger (scan loops + bump outcomes).
2 
3Reuses the worker's own connected client. Lists the durable scan loops (the
4signal stage's heartbeat) and the per-bump workflows; for a completed bump it
5reads the structured outcome from the workflow result (verdict + CI reading + PR
6number), and for a terminated/failed one it recovers the human reason from
7history. Everything is keyed off the deterministic workflow ids, so the
8read-model joins to GitHub by PR number and to repos by id prefix without
9parsing ambiguous slugs. Temporal keeps ~7 days, so this is the *recent* ledger;
10GitHub is the durable one.
11 
12Never raises: a failure returns an error string and whatever was gathered.
13"""
14 
15from __future__ import annotations
16 
17import re
18from datetime import datetime
19from typing import TYPE_CHECKING, Any, Final
20 
21from temporalio.client import WorkflowExecutionStatus
22 
23from froot.domain.base import Frozen
24 
25if TYPE_CHECKING:
  from collections.abc import AsyncIterator
27 
  from temporalio.client import Client, WorkflowExecution
29 
30# A backstop so a runaway visibility store can never spin this forever.
31_MAX_PER_TYPE: Final = 500
32# The owner/name slug out of a PR url — the repo-aware join key (see fetch).
33_PR_URL: Final = re.compile(r"github\.com/([^/]+/[^/]+)/pull/\d+")
34 
35 
36class ScanExecution(Frozen):
  """One scan-loop execution (there may be several per id across CAN)."""
38 
  workflow_id: str
  status: str
  start: datetime | None
42 
43 
44class BumpExecution(Frozen):
  """One bump workflow, with its outcome (if completed) or reason (if not).
46 
  ``repo`` is the ``owner/name`` the PR was opened against, parsed from the
  outcome's PR url; it makes the GitHub join repo-aware so two repos' PRs that
  share a number cannot cross-attribute.
  """
51 
  workflow_id: str
  status: str
  start: datetime | None
  close: datetime | None
  verdict: str | None
  ci: str | None
  pr_number: int | None
  repo: str | None
  reason: str | None
61 
62 
63class ReviewExecution(Frozen):
  """One determinism-review-loop execution (several per id across CAN)."""
65 
  workflow_id: str
  status: str
  start: datetime | None
69 
70 
71class PrReviewExecution(Frozen):
  """One per-PR determinism review, with its result (if completed).
73 
  The findings count + flagged rules + advisory comment come from the
  ``PrReviewWorkflow`` result; the repo is recovered by the read-model from
  the deterministic workflow id (it encodes the slug).
  """
78 
  workflow_id: str
  status: str
  start: datetime | None
  close: datetime | None
  pr_number: int | None
  head_sha: str | None
  findings: int
  rules: tuple[str, ...]
  comment_url: str | None
88 
89 
90def _status(execution: WorkflowExecution) -> str:
  """Lowercase the Temporal status enum (``continued_as_new`` etc.)."""
  status = execution.status
  return status.name.lower() if status is not None else "unknown"
94 
95 
96def _as_dict(result: object) -> dict[str, Any] | None:
  """Normalise a decoded workflow result to a dict (converter-agnostic)."""
  if isinstance(result, dict):
      return result
  dump = getattr(result, "model_dump", None)
  if callable(dump):
      dumped = dump(mode="json")
      return dumped if isinstance(dumped, dict) else None
  return None
105 
106 
107def _nested_kind(data: dict[str, Any], key: str) -> str | None:
  """Read ``data[key]["kind"]`` defensively (a discriminated-union tag)."""
  section = data.get(key)
  if isinstance(section, dict):
      kind = section.get("kind")
      if isinstance(kind, str):
          return kind
  return None
115 
116 
117def _pr_number(data: dict[str, Any]) -> int | None:
  """Read ``data["pr"]["number"]`` defensively."""
  pr = data.get("pr")
  if isinstance(pr, dict):
      number = pr.get("number")
      if isinstance(number, int):
          return number
  return None
125 
126 
127def _pr_repo(data: dict[str, Any]) -> str | None:
  """The ``owner/name`` slug from ``data["pr"]["url"]`` defensively."""
  pr = data.get("pr")
  url = pr.get("url") if isinstance(pr, dict) else None
  if isinstance(url, str):
      match = _PR_URL.search(url)
      if match is not None:
          return match.group(1)
  return None
136 
137 
138class _Outcome(Frozen):
  """The bits of a completed bump's outcome the read-model joins on."""
140 
  verdict: str | None
  ci: str | None
  pr_number: int | None
  repo: str | None
145 
146 
147async def _outcome(client: Client, execution: WorkflowExecution) -> _Outcome:
  """A completed bump's outcome (verdict/ci/pr/repo), or all ``None``."""
  empty = _Outcome(verdict=None, ci=None, pr_number=None, repo=None)
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      data = _as_dict(await handle.result())
  except Exception:  # best-effort enrichment — a bad decode is never fatal
      return empty
  if data is None:
      return empty
  return _Outcome(
      verdict=_nested_kind(data, "verdict"),
      ci=_nested_kind(data, "ci"),
      pr_number=_pr_number(data),
      repo=_pr_repo(data),
  )
165 
166 
167async def _reason(client: Client, execution: WorkflowExecution) -> str | None:
  """Recover a terminated/failed bump'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 page for it
      return None
184 
185 
186class _ReviewOutcome(Frozen):
  """The bits of a completed review's result the read-model joins on."""
188 
  pr_number: int | None
  head_sha: str | None
  findings: int
  rules: tuple[str, ...]
  comment_url: str | None
194 
195 
196async def _review_outcome(
  client: Client, execution: WorkflowExecution
198) -> _ReviewOutcome:
  """A completed review's result (pr/head/findings/rules/comment)."""
  empty = _ReviewOutcome(
      pr_number=None, head_sha=None, findings=0, rules=(), comment_url=None
  )
  try:
      handle = client.get_workflow_handle(
          execution.id, run_id=execution.run_id
      )
      data = _as_dict(await handle.result())
  except Exception:  # best-effort enrichment — a bad decode is never fatal
      return empty
  if data is None:
      return empty
  raw = data.get("findings")
  findings = raw if isinstance(raw, list) else []
  rules = tuple(
      str(f["rule"])
      for f in findings
      if isinstance(f, dict) and isinstance(f.get("rule"), str)
  )
  number = data.get("pr_number")
  head = data.get("head_sha")
  comment = data.get("comment_url")
  return _ReviewOutcome(
      pr_number=number if isinstance(number, int) else None,
      head_sha=head if isinstance(head, str) else None,
      findings=len(findings),
      rules=rules,
      comment_url=comment if isinstance(comment, str) else None,
  )
229 
⋯ 106 lines hidden (lines 230–335)230 
231type _Ledger = tuple[
  tuple[ScanExecution, ...],
  tuple[BumpExecution, ...],
  tuple[ReviewExecution, ...],
  tuple[PrReviewExecution, ...],
236]
237 
238 
239async def fetch(client: Client) -> tuple[_Ledger, str | None]:
  """Read froot's scan/bump + review executions (degrades to an error)."""
  scans: list[ScanExecution] = []
  bumps: list[BumpExecution] = []
  reviews: list[ReviewExecution] = []
  pr_reviews: list[PrReviewExecution] = []
245 
  def ledger() -> _Ledger:
      return (tuple(scans), tuple(bumps), tuple(reviews), tuple(pr_reviews))
248 
  try:
      async for execution in _take(
          client.list_workflows("WorkflowType = 'ScanWorkflow'")
      ):
          scans.append(
              ScanExecution(
                  workflow_id=execution.id,
                  status=_status(execution),
                  start=execution.start_time,
              )
          )
      async for execution in _take(
          client.list_workflows("WorkflowType = 'BumpWorkflow'")
      ):
          status = _status(execution)
          outcome = _Outcome(verdict=None, ci=None, pr_number=None, repo=None)
          reason: str | None = None
          if execution.status is WorkflowExecutionStatus.COMPLETED:
              outcome = await _outcome(client, execution)
          elif execution.status in (
              WorkflowExecutionStatus.TERMINATED,
              WorkflowExecutionStatus.FAILED,
          ):
              reason = await _reason(client, execution)
          bumps.append(
              BumpExecution(
                  workflow_id=execution.id,
                  status=status,
                  start=execution.start_time,
                  close=execution.close_time,
                  verdict=outcome.verdict,
                  ci=outcome.ci,
                  pr_number=outcome.pr_number,
                  repo=outcome.repo,
                  reason=reason,
              )
          )
      async for execution in _take(
          client.list_workflows("WorkflowType = 'ReviewWorkflow'")
      ):
          reviews.append(
              ReviewExecution(
                  workflow_id=execution.id,
                  status=_status(execution),
                  start=execution.start_time,
              )
          )
      async for execution in _take(
          client.list_workflows("WorkflowType = 'PrReviewWorkflow'")
      ):
          review = _ReviewOutcome(
              pr_number=None,
              head_sha=None,
              findings=0,
              rules=(),
              comment_url=None,
          )
          if execution.status is WorkflowExecutionStatus.COMPLETED:
              review = await _review_outcome(client, execution)
          pr_reviews.append(
              PrReviewExecution(
                  workflow_id=execution.id,
                  status=_status(execution),
                  start=execution.start_time,
                  close=execution.close_time,
                  pr_number=review.pr_number,
                  head_sha=review.head_sha,
                  findings=review.findings,
                  rules=review.rules,
                  comment_url=review.comment_url,
              )
          )
  except Exception as exc:  # degrade to an error string, never crash the page
      return ledger(), f"{type(exc).__name__}: {exc}"
  return ledger(), None
324 
325 
326async def _take(
  iterator: AsyncIterator[WorkflowExecution],
328) -> AsyncIterator[WorkflowExecution]:
  """Yield at most :data:`_MAX_PER_TYPE` executions (a runaway backstop)."""
  seen = 0
  async for execution in iterator:
      yield execution
      seen += 1
      if seen >= _MAX_PER_TYPE:
          return

Deriving, and attributing each review to a repo

PrReviewResult doesn't carry the repo slug, but the workflow id does (froot-pr-review-<owner>-<name>-<pr>-<sha>). The read-model matches each review id against the configured repos by the id prefix it would have — longest match wins — exactly how scan/bump join to repos by id. Review loops are listed only for repos that actually have one (reviews are scoped to the Temporal repos), not as dead rows.

_pr_review_prefix · _attribute_repo · _review_loops · 453 lines

_pr_review_prefix · _attribute_repo · _review_loops453 lines · Python

⋯ 270 lines hidden (lines 1–270)1"""Assemble the dashboard view — pure, from the three readers' output.
2 
3This is the reputation read-model proper: it joins GitHub (authoritative
4outcomes) to Temporal (recent verdict + CI reading) by PR number, derives the
5MHE-framed aggregates (track record, verification, judgment, the approval
6queue), and returns a fully-computed
class:`~froot.dashboard.model.DashboardModel` the renderer just projects. No
8I/O and no clock of its own — ``now`` is passed in — so every figure on the
9page is unit-tested apart from the network.
10"""
11 
12from __future__ import annotations
13 
14from typing import TYPE_CHECKING
15 
16from froot.dashboard.model import (
  BumpRow,
  DashboardModel,
  Failure,
  Judgment,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
29)
30from froot.domain.repo import RepoRef, TargetRepo
31from froot.policy.naming import review_workflow_id, scan_workflow_id
32from froot.result import Ok
33 
34if TYPE_CHECKING:
  from datetime import datetime
36 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      BumpExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
44 
45_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
46# Every way a bump can end without closing cleanly — all belong in the honest
47# Failures panel, not silently dropped.
48_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
49 
50 
51def _median(values: list[float]) -> float | None:
  """The median of ``values``, or ``None`` if empty (pure)."""
  if not values:
      return None
  ordered = sorted(values)
  mid = len(ordered) // 2
  if len(ordered) % 2 == 1:
      return ordered[mid]
  return (ordered[mid - 1] + ordered[mid]) / 2
60 
61 
62def _scan_id(repo: str) -> str | None:
  """The deterministic scan-loop id for an ``owner/name`` slug, if valid."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return scan_workflow_id(TargetRepo(repo=ref))
      case _:
          return None
69 
70 
71def _scan_loops(
  repos: tuple[str, ...], scans: tuple[ScanExecution, ...]
73) -> tuple[ScanLoop, ...]:
  """One liveness row per configured repo (latest execution wins)."""
  by_id: dict[str, ScanExecution] = {}
  for scan in scans:
      current = by_id.get(scan.workflow_id)
      if current is None or _newer(scan.start, current.start):
          by_id[scan.workflow_id] = scan
  loops: list[ScanLoop] = []
  for repo in repos:
      scan_id = _scan_id(repo)
      execution = by_id.get(scan_id) if scan_id is not None else None
      if execution is None:
          loops.append(
              ScanLoop(repo=repo, status="none", live=False, last_tick=None)
          )
      else:
          loops.append(
              ScanLoop(
                  repo=repo,
                  status=execution.status,
                  live=execution.status in _LIVE_STATUSES,
                  last_tick=execution.start,
              )
          )
  return tuple(loops)
98 
99 
100def _newer(a: datetime | None, b: datetime | None) -> bool:
  """True if ``a`` is a later instant than ``b`` (``None`` is oldest)."""
  if a is None:
      return False
  if b is None:
      return True
  return a > b
107 
108 
109def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
113) -> tuple[BumpRow, ...]:
  """Join GitHub PRs (authoritative) to Temporal outcomes by PR number."""
  # Keyed on (repo, PR number), not the bare number: Temporal lists every
  # repo's bumps in the namespace, so two repos that each have a PR #N would
  # otherwise cross-attribute one's verdict/CI onto the other's row.
  by_pr: dict[tuple[str, int], BumpExecution] = {
      (bump.repo, bump.pr_number): bump
      for bump in bumps
      if bump.repo is not None and bump.pr_number is not None
  }
  rows: list[BumpRow] = []
  for pr in prs:
      execution = by_pr.get((pr.repo, pr.number))
      verdict = (execution.verdict if execution else None) or pr.verdict
      ci = execution.ci if execution else None
      ttm = _minutes_between(pr.opened_at, pr.merged_at)
      age = _hours_between(pr.opened_at, now) if pr.state == "open" else None
      rows.append(
          BumpRow(
              repo=pr.repo,
              package=pr.package or "?",
              from_version=pr.from_version,
              to_version=pr.to_version or "?",
              state=pr.state,
              verdict=verdict,
              ci=ci,
              pr_number=pr.number,
              pr_url=pr.url,
              opened_at=pr.opened_at,
              merged_at=pr.merged_at,
              ttm_minutes=ttm,
              age_hours=age,
          )
      )
  rows.sort(key=_opened_sort_key, reverse=True)
  return tuple(rows)
149 
150 
151def _opened_sort_key(row: BumpRow) -> float:
  """Sort key putting the most recently opened PR first (unknown last)."""
  return row.opened_at.timestamp() if row.opened_at is not None else 0.0
154 
155 
156def _minutes_between(
  start: datetime | None, end: datetime | None
158) -> float | None:
  """Whole-ish minutes from ``start`` to ``end``, or ``None``."""
  if start is None or end is None:
      return None
  return round((end - start).total_seconds() / 60, 1)
163 
164 
165def _hours_between(
  start: datetime | None, end: datetime | None
167) -> float | None:
  """Hours from ``start`` to ``end``, or ``None``."""
  if start is None or end is None:
      return None
  return round((end - start).total_seconds() / 3600, 1)
172 
173 
174def _track_record(rows: tuple[BumpRow, ...]) -> TrackRecord:
  """Counts, merge rate, and median time-to-merge from the bump rows."""
  merged = [r for r in rows if r.state == "merged"]
  closed = sum(1 for r in rows if r.state == "closed")
  open_now = sum(1 for r in rows if r.state == "open")
  decided = len(merged) + closed
  ttms = [r.ttm_minutes for r in merged if r.ttm_minutes is not None]
  return TrackRecord(
      opened=len(rows),
      merged=len(merged),
      closed_unmerged=closed,
      open_now=open_now,
      merge_rate=(len(merged) / decided) if decided else None,
      median_ttm_minutes=_median(ttms),
  )
189 
190 
191def _verification(rows: tuple[BumpRow, ...]) -> Verification:
  """The CI-oracle breakdown, keeping ``absent`` distinct from a failure."""
  passed = sum(1 for r in rows if r.ci == "passed")
  failed = sum(1 for r in rows if r.ci == "failed")
  absent = sum(1 for r in rows if r.ci == "absent")
  timed_out = sum(1 for r in rows if r.ci == "timed_out")
  unknown = sum(1 for r in rows if r.ci is None)
  return Verification(
      passed=passed,
      failed=failed,
      absent=absent,
      timed_out=timed_out,
      unknown=unknown,
      oracle_existed=passed + failed,
      with_reading=passed + failed + absent + timed_out,
  )
207 
208 
209def _judgment(rows: tuple[BumpRow, ...]) -> Judgment:
  """The verdict mix plus the two calibration cells worth flagging."""
  clean = sum(1 for r in rows if r.verdict == "clean")
  risky = sum(1 for r in rows if r.verdict == "risky")
  unknown = sum(1 for r in rows if r.verdict == "unknown")
  none = sum(1 for r in rows if r.verdict is None)
  clean_but_failed = sum(
      1 for r in rows if r.verdict == "clean" and r.ci == "failed"
  )
  flagged_but_passed = sum(
      1
      for r in rows
      if r.verdict in ("risky", "unknown") and r.ci == "passed"
  )
  return Judgment(
      clean=clean,
      risky=risky,
      unknown=unknown,
      none=none,
      clean_but_failed=clean_but_failed,
      flagged_but_passed=flagged_but_passed,
  )
231 
232 
233def _gate(rows: tuple[BumpRow, ...]) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, the most-aged first (the freshest last)."""
  open_rows = [r for r in rows if r.state == "open"]
  open_rows.sort(
      key=lambda r: r.age_hours if r.age_hours is not None else 0.0,
      reverse=True,
  )
  return tuple(open_rows)
241 
242 
243def _failures(bumps: tuple[BumpExecution, ...]) -> tuple[Failure, ...]:
  """Bump loops that did not close, newest first."""
  failures = [
      Failure(
          workflow_id=bump.workflow_id,
          kind=bump.status,
          reason=bump.reason,
          when=bump.close,
      )
      for bump in bumps
      if bump.status in _FAILURE_STATUSES
  ]
  failures.sort(
      key=lambda f: f.when.timestamp() if f.when is not None else 0.0,
      reverse=True,
  )
  return tuple(failures)
260 
261 
262def _review_id(repo: str) -> str | None:
  """The deterministic review-loop id for an ``owner/name`` slug, if valid."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return review_workflow_id(TargetRepo(repo=ref))
      case _:
          return None
269 
270 
271def _pr_review_prefix(repo: str) -> str | None:
  """The id prefix every per-PR review of ``repo`` shares (the join key)."""
  review_id = _review_id(repo)
  if review_id is None:
      return None
  # froot-review-<slug> -> froot-pr-review-<slug>- ; the pr/sha tail follows.
  return "froot-pr-review-" + review_id.removeprefix("froot-review-") + "-"
278 
279 
280def _attribute_repo(workflow_id: str, repos: tuple[str, ...]) -> str | None:
  """The configured repo a per-PR-review id belongs to (longest prefix)."""
  best: str | None = None
  best_len = -1
  for repo in repos:
      prefix = _pr_review_prefix(repo)
      if prefix and workflow_id.startswith(prefix) and len(prefix) > best_len:
          best, best_len = repo, len(prefix)
  return best
289 
290 
291def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
293) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
295 
  Reviews are scoped to the Temporal repos, so a configured npm repo with no
  review loop is omitted rather than shown as a dead one.
  """
  by_id: dict[str, ReviewExecution] = {}
  for review in reviews:
      current = by_id.get(review.workflow_id)
      if current is None or _newer(review.start, current.start):
          by_id[review.workflow_id] = review
  loops: list[ReviewLoop] = []
  for repo in repos:
      review_id = _review_id(repo)
      execution = by_id.get(review_id) if review_id is not None else None
      if execution is None:
          continue
      loops.append(
          ReviewLoop(
              repo=repo,
              status=execution.status,
              live=execution.status in _LIVE_STATUSES,
              last_tick=execution.start,
          )
      )
  return tuple(loops)
⋯ 135 lines hidden (lines 319–453)319 
320 
321def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
323) -> tuple[ReviewRow, ...]:
  """Project each per-PR review into a row, newest review first."""
  rows: list[ReviewRow] = []
  for execution in pr_reviews:
      repo = _attribute_repo(execution.workflow_id, repos)
      pr_url = (
          f"https://github.com/{repo}/pull/{execution.pr_number}"
          if repo is not None and execution.pr_number is not None
          else None
      )
      rows.append(
          ReviewRow(
              repo=repo or "?",
              pr_number=execution.pr_number,
              pr_url=pr_url,
              head_sha=execution.head_sha,
              findings=execution.findings,
              rules=execution.rules,
              comment_url=execution.comment_url,
              status=execution.status,
              reviewed_at=execution.close or execution.start,
          )
      )
  rows.sort(
      key=lambda r: r.reviewed_at.timestamp() if r.reviewed_at else 0.0,
      reverse=True,
  )
  return tuple(rows)
351 
352 
353def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
355) -> ReviewRecord:
  """Counts over the completed reviews (resolved-rate is a later loop)."""
  completed = [r for r in rows if r.status == "completed"]
  flagged = sum(1 for r in completed if r.findings > 0)
  hazards = sum(r.findings for r in completed)
  return ReviewRecord(
      reviewed=len(completed),
      flagged=flagged,
      clean=len(completed) - flagged,
      hazards=hazards,
      repos_covered=len(loops),
  )
367 
368 
369def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
376) -> tuple[SourceHealth, ...]:
  """Per-source health for the header strip."""
  clickhouse_ok = clickhouse_error is None
  if clickhouse_error == "off":
      clickhouse_detail = "off"
  elif clickhouse_error is not None:
      clickhouse_detail = clickhouse_error
  else:
      spans = telemetry.total_spans
      clickhouse_detail = f"{spans} spans / {telemetry.window_days}d"
  return (
      SourceHealth(
          name="github",
          ok=github_error is None,
          detail=github_error or f"{github_count} PRs",
      ),
      SourceHealth(
          name="temporal",
          ok=temporal_error is None,
          detail=temporal_error or f"{temporal_count} workflows",
      ),
      SourceHealth(
          name="clickhouse", ok=clickhouse_ok, detail=clickhouse_detail
      ),
  )
401 
402 
403def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  scan_interval_seconds: int,
  review_interval_seconds: int,
  github: tuple[tuple[GithubPr, ...], str | None],
  temporal: tuple[
      tuple[
          tuple[ScanExecution, ...],
          tuple[BumpExecution, ...],
          tuple[ReviewExecution, ...],
          tuple[PrReviewExecution, ...],
      ],
      str | None,
  ],
  telemetry: tuple[RunTelemetry, str | None],
420) -> DashboardModel:
  """Build the whole view from the readers' ``(data, error)`` outputs."""
  prs, github_error = github
  (scans, bumps, reviews, pr_reviews), temporal_error = temporal
  run_telemetry, clickhouse_error = telemetry
425 
  rows = _bump_rows(now, prs, bumps)
  review_loops = _review_loops(repos, reviews)
  review_rows = _review_rows(pr_reviews, repos)
  return DashboardModel(
      generated_at=now,
      repos_configured=repos,
      scan_interval_seconds=scan_interval_seconds,
      sources=_sources(
          github_error,
          len(prs),
          temporal_error,
          len(scans) + len(bumps) + len(reviews) + len(pr_reviews),
          run_telemetry,
          clickhouse_error,
      ),
      scan_loops=_scan_loops(repos, scans),
      track_record=_track_record(rows),
      verification=_verification(rows),
      judgment=_judgment(rows),
      gate=_gate(rows),
      bumps=rows,
      failures=_failures(bumps),
      review_interval_seconds=review_interval_seconds,
      review_loops=review_loops,
      review_record=_review_record(review_loops, review_rows),
      reviews=review_rows,
      telemetry=run_telemetry,
  )

_review_record — reviewed / flagged / clean / hazards · 453 lines

_review_record — reviewed / flagged / clean / hazards453 lines · Python

⋯ 352 lines hidden (lines 1–352)1"""Assemble the dashboard view — pure, from the three readers' output.
2 
3This is the reputation read-model proper: it joins GitHub (authoritative
4outcomes) to Temporal (recent verdict + CI reading) by PR number, derives the
5MHE-framed aggregates (track record, verification, judgment, the approval
6queue), and returns a fully-computed
class:`~froot.dashboard.model.DashboardModel` the renderer just projects. No
8I/O and no clock of its own — ``now`` is passed in — so every figure on the
9page is unit-tested apart from the network.
10"""
11 
12from __future__ import annotations
13 
14from typing import TYPE_CHECKING
15 
16from froot.dashboard.model import (
  BumpRow,
  DashboardModel,
  Failure,
  Judgment,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
29)
30from froot.domain.repo import RepoRef, TargetRepo
31from froot.policy.naming import review_workflow_id, scan_workflow_id
32from froot.result import Ok
33 
34if TYPE_CHECKING:
  from datetime import datetime
36 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      BumpExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
44 
45_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
46# Every way a bump can end without closing cleanly — all belong in the honest
47# Failures panel, not silently dropped.
48_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
49 
50 
51def _median(values: list[float]) -> float | None:
  """The median of ``values``, or ``None`` if empty (pure)."""
  if not values:
      return None
  ordered = sorted(values)
  mid = len(ordered) // 2
  if len(ordered) % 2 == 1:
      return ordered[mid]
  return (ordered[mid - 1] + ordered[mid]) / 2
60 
61 
62def _scan_id(repo: str) -> str | None:
  """The deterministic scan-loop id for an ``owner/name`` slug, if valid."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return scan_workflow_id(TargetRepo(repo=ref))
      case _:
          return None
69 
70 
71def _scan_loops(
  repos: tuple[str, ...], scans: tuple[ScanExecution, ...]
73) -> tuple[ScanLoop, ...]:
  """One liveness row per configured repo (latest execution wins)."""
  by_id: dict[str, ScanExecution] = {}
  for scan in scans:
      current = by_id.get(scan.workflow_id)
      if current is None or _newer(scan.start, current.start):
          by_id[scan.workflow_id] = scan
  loops: list[ScanLoop] = []
  for repo in repos:
      scan_id = _scan_id(repo)
      execution = by_id.get(scan_id) if scan_id is not None else None
      if execution is None:
          loops.append(
              ScanLoop(repo=repo, status="none", live=False, last_tick=None)
          )
      else:
          loops.append(
              ScanLoop(
                  repo=repo,
                  status=execution.status,
                  live=execution.status in _LIVE_STATUSES,
                  last_tick=execution.start,
              )
          )
  return tuple(loops)
98 
99 
100def _newer(a: datetime | None, b: datetime | None) -> bool:
  """True if ``a`` is a later instant than ``b`` (``None`` is oldest)."""
  if a is None:
      return False
  if b is None:
      return True
  return a > b
107 
108 
109def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
113) -> tuple[BumpRow, ...]:
  """Join GitHub PRs (authoritative) to Temporal outcomes by PR number."""
  # Keyed on (repo, PR number), not the bare number: Temporal lists every
  # repo's bumps in the namespace, so two repos that each have a PR #N would
  # otherwise cross-attribute one's verdict/CI onto the other's row.
  by_pr: dict[tuple[str, int], BumpExecution] = {
      (bump.repo, bump.pr_number): bump
      for bump in bumps
      if bump.repo is not None and bump.pr_number is not None
  }
  rows: list[BumpRow] = []
  for pr in prs:
      execution = by_pr.get((pr.repo, pr.number))
      verdict = (execution.verdict if execution else None) or pr.verdict
      ci = execution.ci if execution else None
      ttm = _minutes_between(pr.opened_at, pr.merged_at)
      age = _hours_between(pr.opened_at, now) if pr.state == "open" else None
      rows.append(
          BumpRow(
              repo=pr.repo,
              package=pr.package or "?",
              from_version=pr.from_version,
              to_version=pr.to_version or "?",
              state=pr.state,
              verdict=verdict,
              ci=ci,
              pr_number=pr.number,
              pr_url=pr.url,
              opened_at=pr.opened_at,
              merged_at=pr.merged_at,
              ttm_minutes=ttm,
              age_hours=age,
          )
      )
  rows.sort(key=_opened_sort_key, reverse=True)
  return tuple(rows)
149 
150 
151def _opened_sort_key(row: BumpRow) -> float:
  """Sort key putting the most recently opened PR first (unknown last)."""
  return row.opened_at.timestamp() if row.opened_at is not None else 0.0
154 
155 
156def _minutes_between(
  start: datetime | None, end: datetime | None
158) -> float | None:
  """Whole-ish minutes from ``start`` to ``end``, or ``None``."""
  if start is None or end is None:
      return None
  return round((end - start).total_seconds() / 60, 1)
163 
164 
165def _hours_between(
  start: datetime | None, end: datetime | None
167) -> float | None:
  """Hours from ``start`` to ``end``, or ``None``."""
  if start is None or end is None:
      return None
  return round((end - start).total_seconds() / 3600, 1)
172 
173 
174def _track_record(rows: tuple[BumpRow, ...]) -> TrackRecord:
  """Counts, merge rate, and median time-to-merge from the bump rows."""
  merged = [r for r in rows if r.state == "merged"]
  closed = sum(1 for r in rows if r.state == "closed")
  open_now = sum(1 for r in rows if r.state == "open")
  decided = len(merged) + closed
  ttms = [r.ttm_minutes for r in merged if r.ttm_minutes is not None]
  return TrackRecord(
      opened=len(rows),
      merged=len(merged),
      closed_unmerged=closed,
      open_now=open_now,
      merge_rate=(len(merged) / decided) if decided else None,
      median_ttm_minutes=_median(ttms),
  )
189 
190 
191def _verification(rows: tuple[BumpRow, ...]) -> Verification:
  """The CI-oracle breakdown, keeping ``absent`` distinct from a failure."""
  passed = sum(1 for r in rows if r.ci == "passed")
  failed = sum(1 for r in rows if r.ci == "failed")
  absent = sum(1 for r in rows if r.ci == "absent")
  timed_out = sum(1 for r in rows if r.ci == "timed_out")
  unknown = sum(1 for r in rows if r.ci is None)
  return Verification(
      passed=passed,
      failed=failed,
      absent=absent,
      timed_out=timed_out,
      unknown=unknown,
      oracle_existed=passed + failed,
      with_reading=passed + failed + absent + timed_out,
  )
207 
208 
209def _judgment(rows: tuple[BumpRow, ...]) -> Judgment:
  """The verdict mix plus the two calibration cells worth flagging."""
  clean = sum(1 for r in rows if r.verdict == "clean")
  risky = sum(1 for r in rows if r.verdict == "risky")
  unknown = sum(1 for r in rows if r.verdict == "unknown")
  none = sum(1 for r in rows if r.verdict is None)
  clean_but_failed = sum(
      1 for r in rows if r.verdict == "clean" and r.ci == "failed"
  )
  flagged_but_passed = sum(
      1
      for r in rows
      if r.verdict in ("risky", "unknown") and r.ci == "passed"
  )
  return Judgment(
      clean=clean,
      risky=risky,
      unknown=unknown,
      none=none,
      clean_but_failed=clean_but_failed,
      flagged_but_passed=flagged_but_passed,
  )
231 
232 
233def _gate(rows: tuple[BumpRow, ...]) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, the most-aged first (the freshest last)."""
  open_rows = [r for r in rows if r.state == "open"]
  open_rows.sort(
      key=lambda r: r.age_hours if r.age_hours is not None else 0.0,
      reverse=True,
  )
  return tuple(open_rows)
241 
242 
243def _failures(bumps: tuple[BumpExecution, ...]) -> tuple[Failure, ...]:
  """Bump loops that did not close, newest first."""
  failures = [
      Failure(
          workflow_id=bump.workflow_id,
          kind=bump.status,
          reason=bump.reason,
          when=bump.close,
      )
      for bump in bumps
      if bump.status in _FAILURE_STATUSES
  ]
  failures.sort(
      key=lambda f: f.when.timestamp() if f.when is not None else 0.0,
      reverse=True,
  )
  return tuple(failures)
260 
261 
262def _review_id(repo: str) -> str | None:
  """The deterministic review-loop id for an ``owner/name`` slug, if valid."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return review_workflow_id(TargetRepo(repo=ref))
      case _:
          return None
269 
270 
271def _pr_review_prefix(repo: str) -> str | None:
  """The id prefix every per-PR review of ``repo`` shares (the join key)."""
  review_id = _review_id(repo)
  if review_id is None:
      return None
  # froot-review-<slug> -> froot-pr-review-<slug>- ; the pr/sha tail follows.
  return "froot-pr-review-" + review_id.removeprefix("froot-review-") + "-"
278 
279 
280def _attribute_repo(workflow_id: str, repos: tuple[str, ...]) -> str | None:
  """The configured repo a per-PR-review id belongs to (longest prefix)."""
  best: str | None = None
  best_len = -1
  for repo in repos:
      prefix = _pr_review_prefix(repo)
      if prefix and workflow_id.startswith(prefix) and len(prefix) > best_len:
          best, best_len = repo, len(prefix)
  return best
289 
290 
291def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
293) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
295 
  Reviews are scoped to the Temporal repos, so a configured npm repo with no
  review loop is omitted rather than shown as a dead one.
  """
  by_id: dict[str, ReviewExecution] = {}
  for review in reviews:
      current = by_id.get(review.workflow_id)
      if current is None or _newer(review.start, current.start):
          by_id[review.workflow_id] = review
  loops: list[ReviewLoop] = []
  for repo in repos:
      review_id = _review_id(repo)
      execution = by_id.get(review_id) if review_id is not None else None
      if execution is None:
          continue
      loops.append(
          ReviewLoop(
              repo=repo,
              status=execution.status,
              live=execution.status in _LIVE_STATUSES,
              last_tick=execution.start,
          )
      )
  return tuple(loops)
319 
320 
321def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
323) -> tuple[ReviewRow, ...]:
  """Project each per-PR review into a row, newest review first."""
  rows: list[ReviewRow] = []
  for execution in pr_reviews:
      repo = _attribute_repo(execution.workflow_id, repos)
      pr_url = (
          f"https://github.com/{repo}/pull/{execution.pr_number}"
          if repo is not None and execution.pr_number is not None
          else None
      )
      rows.append(
          ReviewRow(
              repo=repo or "?",
              pr_number=execution.pr_number,
              pr_url=pr_url,
              head_sha=execution.head_sha,
              findings=execution.findings,
              rules=execution.rules,
              comment_url=execution.comment_url,
              status=execution.status,
              reviewed_at=execution.close or execution.start,
          )
      )
  rows.sort(
      key=lambda r: r.reviewed_at.timestamp() if r.reviewed_at else 0.0,
      reverse=True,
  )
  return tuple(rows)
351 
352 
353def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
355) -> ReviewRecord:
  """Counts over the completed reviews (resolved-rate is a later loop)."""
  completed = [r for r in rows if r.status == "completed"]
  flagged = sum(1 for r in completed if r.findings > 0)
  hazards = sum(r.findings for r in completed)
  return ReviewRecord(
      reviewed=len(completed),
      flagged=flagged,
      clean=len(completed) - flagged,
      hazards=hazards,
      repos_covered=len(loops),
  )
367 
368 
⋯ 85 lines hidden (lines 369–453)369def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
376) -> tuple[SourceHealth, ...]:
  """Per-source health for the header strip."""
  clickhouse_ok = clickhouse_error is None
  if clickhouse_error == "off":
      clickhouse_detail = "off"
  elif clickhouse_error is not None:
      clickhouse_detail = clickhouse_error
  else:
      spans = telemetry.total_spans
      clickhouse_detail = f"{spans} spans / {telemetry.window_days}d"
  return (
      SourceHealth(
          name="github",
          ok=github_error is None,
          detail=github_error or f"{github_count} PRs",
      ),
      SourceHealth(
          name="temporal",
          ok=temporal_error is None,
          detail=temporal_error or f"{temporal_count} workflows",
      ),
      SourceHealth(
          name="clickhouse", ok=clickhouse_ok, detail=clickhouse_detail
      ),
  )
401 
402 
403def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  scan_interval_seconds: int,
  review_interval_seconds: int,
  github: tuple[tuple[GithubPr, ...], str | None],
  temporal: tuple[
      tuple[
          tuple[ScanExecution, ...],
          tuple[BumpExecution, ...],
          tuple[ReviewExecution, ...],
          tuple[PrReviewExecution, ...],
      ],
      str | None,
  ],
  telemetry: tuple[RunTelemetry, str | None],
420) -> DashboardModel:
  """Build the whole view from the readers' ``(data, error)`` outputs."""
  prs, github_error = github
  (scans, bumps, reviews, pr_reviews), temporal_error = temporal
  run_telemetry, clickhouse_error = telemetry
425 
  rows = _bump_rows(now, prs, bumps)
  review_loops = _review_loops(repos, reviews)
  review_rows = _review_rows(pr_reviews, repos)
  return DashboardModel(
      generated_at=now,
      repos_configured=repos,
      scan_interval_seconds=scan_interval_seconds,
      sources=_sources(
          github_error,
          len(prs),
          temporal_error,
          len(scans) + len(bumps) + len(reviews) + len(pr_reviews),
          run_telemetry,
          clickhouse_error,
      ),
      scan_loops=_scan_loops(repos, scans),
      track_record=_track_record(rows),
      verification=_verification(rows),
      judgment=_judgment(rows),
      gate=_gate(rows),
      bumps=rows,
      failures=_failures(bumps),
      review_interval_seconds=review_interval_seconds,
      review_loops=review_loops,
      review_record=_review_record(review_loops, review_rows),
      reviews=review_rows,
      telemetry=run_telemetry,
  )

The three sections it renders

Three sections slot in after the bump detail, before telemetry: the review-loop heartbeat (is it alive?), the headline stats (the transitive ring), and the per-review table. A review's findings cell reads clean when none, else the hazard count, the flagged rules, and a one-click link to the advisory comment.

_review_record — the headline + advisory note · 545 lines

_review_record — the headline + advisory note545 lines · Python

⋯ 404 lines hidden (lines 1–404)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
4request of its own — it is a static projection of an already-computed
class:`~froot.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary. The ordering is trust-first (is it alive → track
7record → oracle → judgment → the human's queue), so it reads top to bottom.
8"""
9 
10from __future__ import annotations
11 
12from datetime import UTC, datetime, timedelta
13from html import escape
14from typing import TYPE_CHECKING
15 
16if TYPE_CHECKING:
  from froot.dashboard.model import (
      BumpRow,
      DashboardModel,
      ReviewLoop,
      ReviewRow,
      RunTelemetry,
      ScanLoop,
  )
25 
26_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
28--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#fafafa}
29@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
30--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
31--accent:#58a6ff;--card:#141414}}
32*{box-sizing:border-box}
33body{margin:0;background:var(--bg);color:var(--fg);
34font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
35main{max-width:820px;margin:0 auto;padding:34px 20px 72px}
36h1{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}
40section{margin:30px 0 0}
41h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
42color:var(--mut);margin:0 0 12px;font-weight:600;
43border-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%;
48margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
51table{width:100%;border-collapse:collapse;font-size:13px}
52th,td{text-align:left;padding:6px 12px 6px 0;
53border-bottom:1px solid var(--line);vertical-align:top}
54th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
55letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
57a{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}
62footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
63color:var(--mut);font-size:12px;line-height:1.7}
64footer b{color:var(--fg);font-weight:600}
65"""
66 
67_CI_CLASS = {
  "passed": "ok",
  "failed": "bad",
  "absent": "mut",
  "timed_out": "warn",
72}
73_VERDICT_CLASS = {"clean": "ok", "risky": "warn", "unknown": "mut"}
74 
75 
76def _aware(when: datetime) -> datetime:
  """Treat a stray naive timestamp as UTC so arithmetic never raises."""
  return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
79 
80 
81def _ago(when: datetime | None, now: datetime) -> str:
  """A compact 'time since' label (``6h ago``)."""
  if when is None:
      return "—"
  secs = (now - _aware(when)).total_seconds()
  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"
93 
94 
95def _until(when: datetime | None, now: datetime) -> str:
  """A compact 'time until' label (``in 18h`` / ``due now``)."""
  if when is None:
      return "—"
  secs = (_aware(when) - now).total_seconds()
  if secs <= 0:
      return "due now"
  if secs < 5400:
      return f"in {int(secs // 60)}m"
  if secs < 129600:
      return f"in {int(secs // 3600)}h"
  return f"in {int(secs // 86400)}d"
107 
108 
109def _dot(kind: str) -> str:
  """A status dot span (``ok`` / ``warn`` / ``bad`` / ``mute``)."""
  return f'<span class="dot {kind}"></span>'
112 
113 
114def _tag(value: str | None, classes: dict[str, str]) -> str:
  """A small coloured label for a verdict/CI value, or an em-dash."""
  if value is None:
      return '<span class="mut">—</span>'
  cls = classes.get(value, "mut")
  return f'<span class="{cls}">{escape(value)}</span>'
120 
121 
122def _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>'
  )
127 
128 
129def _header(model: DashboardModel) -> str:
  now = model.generated_at
  repos = ", ".join(model.repos_configured) or "none configured"
  dots = "".join(
      f"<span>{_dot('ok' if s.ok else 'bad')}"
      f'{escape(s.name)} <span class="mut">{escape(s.detail)}</span></span>'
      for s in model.sources
  )
  stamp = now.strftime("%Y-%m-%d %H:%M UTC")
  return (
      "<header>"
      "<h1>froot</h1>"
      '<p class="tag">durable maintenance loops &middot; '
      "reputation read-model</p>"
      f'<p class="meta">watching <span class="mono">{escape(repos)}</span>'
      f" &middot; generated {escape(stamp)} &middot; "
      "derived live, stored nowhere</p>"
      f'<div class="sources">{dots}</div>'
      "</header>"
  )
149 
150 
151def _heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.scan_interval_seconds
154 
  def line(loop: ScanLoop) -> str:
      if loop.live:
          dot, tail = "ok", ""
          if loop.last_tick is not None:
              nxt = _aware(loop.last_tick) + timedelta(seconds=interval)
              last = _ago(loop.last_tick, now)
              tail = (
                  f' <span class="mut">&middot; last {last}'
                  f" &middot; next {_until(nxt, now)}</span>"
              )
      else:
          dot = "bad" if loop.status in ("terminated", "none") else "warn"
          tail = f' <span class="mut">&middot; {escape(loop.status)}</span>'
      return (
          f'<div class="row">{_dot(dot)}'
          f'<span class="mono">{escape(loop.repo)}</span>{tail}</div>'
      )
172 
  if not model.scan_loops:
      body = '<p class="note">No repos configured (FROOT_REPOS unset).</p>'
  else:
      body = "".join(line(loop) for loop in model.scan_loops)
  return f"<section><h2>Is the loop alive?</h2>{body}</section>"
178 
179 
180def _track_record(model: DashboardModel) -> str:
  t = model.track_record
  rate = "—" if t.merge_rate is None else f"{t.merge_rate * 100:.0f}%"
  ttm = (
      "—"
      if t.median_ttm_minutes is None
      else f"{t.median_ttm_minutes:.0f} min"
  )
  stats = "".join(
      (
          _stat(t.opened, "proposed"),
          _stat(t.merged, "merged"),
          _stat(t.open_now, "awaiting"),
          _stat(t.closed_unmerged, "closed"),
          _stat(rate, "merge rate"),
          _stat(ttm, "median time-to-merge"),
      )
  )
  note = (
      '<p class="note">Merge rate is the Stage-1 signal &mdash; narrow to '
      "npm patch bumps by construction. It counts a human merge, not a "
      "confirmed good outcome: revert tracking is a later loop, so merge is "
      "not yet proof of success.</p>"
  )
  return (
      "<section><h2>Track record &middot; the reputation</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
208 
209 
210def _verification(model: DashboardModel) -> str:
  v = model.verification
  stats = "".join(
      (
          _stat(v.passed, "CI passed"),
          _stat(v.failed, "CI failed"),
          _stat(v.absent, "no checks"),
          _stat(v.timed_out, "timed out"),
          _stat(v.unknown, "unknown"),
      )
  )
  if v.with_reading == 0:
      note = '<p class="note">No CI readings yet.</p>'
  else:
      note = (
          f'<p class="note">A real oracle reported on '
          f"<b>{v.oracle_existed}</b> of {v.with_reading} bumps with a "
          'reading. <span class="mut">&lsquo;no checks&rsquo; means CI was '
          "absent &mdash; not a pass; never conflated.</span>"
          "</p>"
      )
  return (
      "<section><h2>Verification &middot; CI is the oracle</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
235 
236 
237def _judgment(model: DashboardModel) -> str:
  j = model.judgment
  stats = "".join(
      (
          _stat(j.clean, "clean"),
          _stat(j.risky, "risky"),
          _stat(j.unknown, "unknown"),
          _stat(j.none, "no verdict"),
      )
  )
  if j.clean_but_failed or j.flagged_but_passed:
      note = (
          f'<p class="note">Calibration: <b>{j.clean_but_failed}</b> '
          "&lsquo;clean&rsquo; bumps whose CI failed, "
          f"<b>{j.flagged_but_passed}</b> flagged bumps whose CI passed.</p>"
      )
  else:
      note = (
          '<p class="note">The model&rsquo;s only job is the changelog '
          "verdict; the spine proposes the bump either way.</p>"
      )
  return (
      "<section><h2>Model judgment &middot; the one model call</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
262 
263 
264def _gate(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.gate:
      body = '<p class="note">Queue empty &mdash; nothing awaiting you.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.package)}</td>'
          f"<td>{escape(_ago(row.opened_at, now))}</td>"
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.gate
      )
      body = (
          "<table><thead><tr><th>package</th><th>waiting</th><th>pr</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>"
      )
  return (
      "<section><h2>Approval gate &middot; what a human owns</h2>"
      f"{body}</section>"
  )
285 
286 
287def _bumps(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.bumps:
      body = '<p class="note">No bumps proposed yet.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.package)}</td>'
          f'<td class="mono mut">{escape(row.from_version or "?")} &rarr; '
          f"{escape(row.to_version)}</td>"
          f"<td>{_tag(row.verdict, _VERDICT_CLASS)}</td>"
          f"<td>{_tag(row.ci, _CI_CLASS)}</td>"
          f"<td>{_state_tag(row.state)}</td>"
          f'<td class="mut">{escape(_ago(row.opened_at, now))}</td>'
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.bumps
      )
      body = (
          "<table><thead><tr><th>package</th><th>bump</th><th>verdict</th>"
          "<th>ci</th><th>state</th><th>opened</th><th>pr</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  return f"<section><h2>Bumps &middot; the detail</h2>{body}</section>"
311 
312 
313def _failures(model: DashboardModel) -> str:
  if not model.failures:
      return ""
  now = model.generated_at
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(_short_id(f.workflow_id))}</td>'
      f"<td>{_state_tag(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 the loop did not close</h2>"
      "<table><thead><tr><th>bump</th><th>kind</th><th>reason</th>"
      f"<th>when</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
331 
332 
333def _telemetry(model: DashboardModel) -> str:
  t: RunTelemetry = model.telemetry
  if not t.available:
      return (
          "<section><h2>Run telemetry &middot; ClickHouse</h2>"
          '<p class="note">Unavailable (not configured or unreachable). '
          "GitHub + Temporal carry the dashboard regardless.</p></section>"
      )
  now = model.generated_at
  if t.activities:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(a.name)}</td>'
          f"<td>{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 froot spans in the window.</p>'
  summary = (
      f'<p class="note">{t.total_spans} spans &middot; '
      f"{t.error_spans} errored &middot; last activity "
      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}</section>"
  )
368 
369 
370def _review_heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.review_interval_seconds
373 
  def line(loop: ReviewLoop) -> str:
      if loop.live:
          dot, tail = "ok", ""
          if loop.last_tick is not None:
              nxt = _aware(loop.last_tick) + timedelta(seconds=interval)
              last = _ago(loop.last_tick, now)
              tail = (
                  f' <span class="mut">&middot; last {last}'
                  f" &middot; next {_until(nxt, now)}</span>"
              )
      else:
          dot = "bad" if loop.status in ("terminated", "none") else "warn"
          tail = f' <span class="mut">&middot; {escape(loop.status)}</span>'
      return (
          f'<div class="row">{_dot(dot)}'
          f'<span class="mono">{escape(loop.repo)}</span>{tail}</div>'
      )
391 
  if not model.review_loops:
      body = (
          '<p class="note">No determinism-review loops running '
          "(the transitive ring watches the @workflow.defn repos).</p>"
      )
  else:
      body = "".join(line(loop) for loop in model.review_loops)
  return (
      "<section><h2>Determinism review &middot; is it alive?</h2>"
      f"{body}</section>"
  )
403 
404 
405def _review_record(model: DashboardModel) -> str:
  r = model.review_record
  stats = "".join(
      (
          _stat(r.reviewed, "reviewed"),
          _stat(r.flagged, "flagged"),
          _stat(r.clean, "clean"),
          _stat(r.hazards, "hazards"),
          _stat(r.repos_covered, "repos covered"),
      )
  )
  note = (
      '<p class="note">The transitive ring: it chases first-party helper '
      "calls out of each workflow to catch a hazard the lexical CI kernel "
      "can&rsquo;t see. <b>Advisory</b> &mdash; the blocking gate stays the "
      "kernel&rsquo;s CI check. The hazard-resolved rate (was a flag gone on "
      "a later commit?) is a later loop; it needs accumulated history.</p>"
  )
  return (
      "<section><h2>Determinism review &middot; the transitive ring</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
427 
428 
⋯ 117 lines hidden (lines 429–545)429def _reviews(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.reviews:
      body = '<p class="note">No PRs reviewed yet.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.repo)}</td>'
          f"<td>{_review_pr_link(row)}</td>"
          f'<td class="mono mut">{escape((row.head_sha or "")[:7]) or "—"}'
          "</td>"
          f"<td>{_findings_cell(row)}</td>"
          f'<td class="mut">{escape(_ago(row.reviewed_at, now))}</td>'
          "</tr>"
          for row in model.reviews
      )
      body = (
          "<table><thead><tr><th>repo</th><th>pr</th><th>head</th>"
          "<th>findings</th><th>reviewed</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  return (
      "<section><h2>Determinism reviews &middot; the detail</h2>"
      f"{body}</section>"
  )
454 
455 
456def _footer() -> str:
  return (
      "<footer>"
      "<b>Authority envelope.</b> Stage 1 &mdash; froot holds "
      "<b>write authority</b> only: it opens PRs, a human approves every "
      "merge (commit authority = none). Trust, when any is granted, is "
      "earned, narrow to npm patch bumps, conditional on its environment "
      '(judge <span class="mono">gemma4:e4b</span>, lockfile-only regen), '
      "revocable, and time-expiring. Today it records the track record; it "
      "does not yet act on it.<br>"
      "Everything above is derived on this request from GitHub (outcomes) + "
      "Temporal (runs) + ClickHouse (telemetry). froot keeps no database; "
      "reload to recompute."
      "</footer>"
  )
471 
472 
473def _pr_link(row: BumpRow) -> str:
  if row.pr_url is None or row.pr_number is None:
      return '<span class="mut">—</span>'
  return f'<a href="{escape(row.pr_url, quote=True)}">#{row.pr_number}</a>'
477 
478 
479def _review_pr_link(row: ReviewRow) -> str:
  if row.pr_url is None or row.pr_number is None:
      return '<span class="mut">—</span>'
  return f'<a href="{escape(row.pr_url, quote=True)}">#{row.pr_number}</a>'
483 
484 
485def _findings_cell(row: ReviewRow) -> str:
  """A review's findings: 'clean', or the hazard count + rules + comment."""
  if row.findings == 0:
      return '<span class="ok">clean</span>'
  rules = (
      f' <span class="mono mut">{escape(", ".join(row.rules))}</span>'
      if row.rules
      else ""
  )
  comment = (
      f' <a href="{escape(row.comment_url, quote=True)}">comment</a>'
      if row.comment_url
      else ""
  )
  noun = "hazard" if row.findings == 1 else "hazards"
  return f'<span class="bad">{row.findings} {noun}</span>{rules}{comment}'
501 
502 
503def _state_tag(state: str) -> str:
  cls = {
      "merged": "ok",
      "open": "warn",
      "closed": "mut",
      "terminated": "bad",
      "failed": "bad",
      "timed_out": "bad",
      "canceled": "warn",
  }.get(state, "mut")
  return f'<span class="{cls}">{escape(state)}</span>'
514 
515 
516def _short_id(workflow_id: str) -> str:
  """Drop the ``froot-bump-`` prefix for a readable failures row."""
  return workflow_id.removeprefix("froot-bump-")
519 
520 
521def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _heartbeat(model),
      _track_record(model),
      _verification(model),
      _judgment(model),
      _gate(model),
      _bumps(model),
      _failures(model),
      _review_heartbeat(model),
      _review_record(model),
      _reviews(model),
      _telemetry(model),
      _footer(),
  )
  return (
      '<!doctype html><html lang="en"><head><meta charset="utf-8">'
      '<meta name="viewport" content="width=device-width,initial-scale=1">'
      "<title>froot &middot; read-model</title>"
      f"<style>{_CSS}</style></head><body><main>"
      + "".join(parts)
      + "</main></body></html>"
  )

_findings_cell — clean, or count + rules + comment · 545 lines

_findings_cell — clean, or count + rules + comment545 lines · Python

⋯ 484 lines hidden (lines 1–484)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
4request of its own — it is a static projection of an already-computed
class:`~froot.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary. The ordering is trust-first (is it alive → track
7record → oracle → judgment → the human's queue), so it reads top to bottom.
8"""
9 
10from __future__ import annotations
11 
12from datetime import UTC, datetime, timedelta
13from html import escape
14from typing import TYPE_CHECKING
15 
16if TYPE_CHECKING:
  from froot.dashboard.model import (
      BumpRow,
      DashboardModel,
      ReviewLoop,
      ReviewRow,
      RunTelemetry,
      ScanLoop,
  )
25 
26_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
28--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#fafafa}
29@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
30--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
31--accent:#58a6ff;--card:#141414}}
32*{box-sizing:border-box}
33body{margin:0;background:var(--bg);color:var(--fg);
34font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
35main{max-width:820px;margin:0 auto;padding:34px 20px 72px}
36h1{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}
40section{margin:30px 0 0}
41h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
42color:var(--mut);margin:0 0 12px;font-weight:600;
43border-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%;
48margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
51table{width:100%;border-collapse:collapse;font-size:13px}
52th,td{text-align:left;padding:6px 12px 6px 0;
53border-bottom:1px solid var(--line);vertical-align:top}
54th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
55letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
57a{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}
62footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
63color:var(--mut);font-size:12px;line-height:1.7}
64footer b{color:var(--fg);font-weight:600}
65"""
66 
67_CI_CLASS = {
  "passed": "ok",
  "failed": "bad",
  "absent": "mut",
  "timed_out": "warn",
72}
73_VERDICT_CLASS = {"clean": "ok", "risky": "warn", "unknown": "mut"}
74 
75 
76def _aware(when: datetime) -> datetime:
  """Treat a stray naive timestamp as UTC so arithmetic never raises."""
  return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
79 
80 
81def _ago(when: datetime | None, now: datetime) -> str:
  """A compact 'time since' label (``6h ago``)."""
  if when is None:
      return "—"
  secs = (now - _aware(when)).total_seconds()
  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"
93 
94 
95def _until(when: datetime | None, now: datetime) -> str:
  """A compact 'time until' label (``in 18h`` / ``due now``)."""
  if when is None:
      return "—"
  secs = (_aware(when) - now).total_seconds()
  if secs <= 0:
      return "due now"
  if secs < 5400:
      return f"in {int(secs // 60)}m"
  if secs < 129600:
      return f"in {int(secs // 3600)}h"
  return f"in {int(secs // 86400)}d"
107 
108 
109def _dot(kind: str) -> str:
  """A status dot span (``ok`` / ``warn`` / ``bad`` / ``mute``)."""
  return f'<span class="dot {kind}"></span>'
112 
113 
114def _tag(value: str | None, classes: dict[str, str]) -> str:
  """A small coloured label for a verdict/CI value, or an em-dash."""
  if value is None:
      return '<span class="mut">—</span>'
  cls = classes.get(value, "mut")
  return f'<span class="{cls}">{escape(value)}</span>'
120 
121 
122def _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>'
  )
127 
128 
129def _header(model: DashboardModel) -> str:
  now = model.generated_at
  repos = ", ".join(model.repos_configured) or "none configured"
  dots = "".join(
      f"<span>{_dot('ok' if s.ok else 'bad')}"
      f'{escape(s.name)} <span class="mut">{escape(s.detail)}</span></span>'
      for s in model.sources
  )
  stamp = now.strftime("%Y-%m-%d %H:%M UTC")
  return (
      "<header>"
      "<h1>froot</h1>"
      '<p class="tag">durable maintenance loops &middot; '
      "reputation read-model</p>"
      f'<p class="meta">watching <span class="mono">{escape(repos)}</span>'
      f" &middot; generated {escape(stamp)} &middot; "
      "derived live, stored nowhere</p>"
      f'<div class="sources">{dots}</div>'
      "</header>"
  )
149 
150 
151def _heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.scan_interval_seconds
154 
  def line(loop: ScanLoop) -> str:
      if loop.live:
          dot, tail = "ok", ""
          if loop.last_tick is not None:
              nxt = _aware(loop.last_tick) + timedelta(seconds=interval)
              last = _ago(loop.last_tick, now)
              tail = (
                  f' <span class="mut">&middot; last {last}'
                  f" &middot; next {_until(nxt, now)}</span>"
              )
      else:
          dot = "bad" if loop.status in ("terminated", "none") else "warn"
          tail = f' <span class="mut">&middot; {escape(loop.status)}</span>'
      return (
          f'<div class="row">{_dot(dot)}'
          f'<span class="mono">{escape(loop.repo)}</span>{tail}</div>'
      )
172 
  if not model.scan_loops:
      body = '<p class="note">No repos configured (FROOT_REPOS unset).</p>'
  else:
      body = "".join(line(loop) for loop in model.scan_loops)
  return f"<section><h2>Is the loop alive?</h2>{body}</section>"
178 
179 
180def _track_record(model: DashboardModel) -> str:
  t = model.track_record
  rate = "—" if t.merge_rate is None else f"{t.merge_rate * 100:.0f}%"
  ttm = (
      "—"
      if t.median_ttm_minutes is None
      else f"{t.median_ttm_minutes:.0f} min"
  )
  stats = "".join(
      (
          _stat(t.opened, "proposed"),
          _stat(t.merged, "merged"),
          _stat(t.open_now, "awaiting"),
          _stat(t.closed_unmerged, "closed"),
          _stat(rate, "merge rate"),
          _stat(ttm, "median time-to-merge"),
      )
  )
  note = (
      '<p class="note">Merge rate is the Stage-1 signal &mdash; narrow to '
      "npm patch bumps by construction. It counts a human merge, not a "
      "confirmed good outcome: revert tracking is a later loop, so merge is "
      "not yet proof of success.</p>"
  )
  return (
      "<section><h2>Track record &middot; the reputation</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
208 
209 
210def _verification(model: DashboardModel) -> str:
  v = model.verification
  stats = "".join(
      (
          _stat(v.passed, "CI passed"),
          _stat(v.failed, "CI failed"),
          _stat(v.absent, "no checks"),
          _stat(v.timed_out, "timed out"),
          _stat(v.unknown, "unknown"),
      )
  )
  if v.with_reading == 0:
      note = '<p class="note">No CI readings yet.</p>'
  else:
      note = (
          f'<p class="note">A real oracle reported on '
          f"<b>{v.oracle_existed}</b> of {v.with_reading} bumps with a "
          'reading. <span class="mut">&lsquo;no checks&rsquo; means CI was '
          "absent &mdash; not a pass; never conflated.</span>"
          "</p>"
      )
  return (
      "<section><h2>Verification &middot; CI is the oracle</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
235 
236 
237def _judgment(model: DashboardModel) -> str:
  j = model.judgment
  stats = "".join(
      (
          _stat(j.clean, "clean"),
          _stat(j.risky, "risky"),
          _stat(j.unknown, "unknown"),
          _stat(j.none, "no verdict"),
      )
  )
  if j.clean_but_failed or j.flagged_but_passed:
      note = (
          f'<p class="note">Calibration: <b>{j.clean_but_failed}</b> '
          "&lsquo;clean&rsquo; bumps whose CI failed, "
          f"<b>{j.flagged_but_passed}</b> flagged bumps whose CI passed.</p>"
      )
  else:
      note = (
          '<p class="note">The model&rsquo;s only job is the changelog '
          "verdict; the spine proposes the bump either way.</p>"
      )
  return (
      "<section><h2>Model judgment &middot; the one model call</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
262 
263 
264def _gate(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.gate:
      body = '<p class="note">Queue empty &mdash; nothing awaiting you.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.package)}</td>'
          f"<td>{escape(_ago(row.opened_at, now))}</td>"
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.gate
      )
      body = (
          "<table><thead><tr><th>package</th><th>waiting</th><th>pr</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>"
      )
  return (
      "<section><h2>Approval gate &middot; what a human owns</h2>"
      f"{body}</section>"
  )
285 
286 
287def _bumps(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.bumps:
      body = '<p class="note">No bumps proposed yet.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.package)}</td>'
          f'<td class="mono mut">{escape(row.from_version or "?")} &rarr; '
          f"{escape(row.to_version)}</td>"
          f"<td>{_tag(row.verdict, _VERDICT_CLASS)}</td>"
          f"<td>{_tag(row.ci, _CI_CLASS)}</td>"
          f"<td>{_state_tag(row.state)}</td>"
          f'<td class="mut">{escape(_ago(row.opened_at, now))}</td>'
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.bumps
      )
      body = (
          "<table><thead><tr><th>package</th><th>bump</th><th>verdict</th>"
          "<th>ci</th><th>state</th><th>opened</th><th>pr</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  return f"<section><h2>Bumps &middot; the detail</h2>{body}</section>"
311 
312 
313def _failures(model: DashboardModel) -> str:
  if not model.failures:
      return ""
  now = model.generated_at
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(_short_id(f.workflow_id))}</td>'
      f"<td>{_state_tag(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 the loop did not close</h2>"
      "<table><thead><tr><th>bump</th><th>kind</th><th>reason</th>"
      f"<th>when</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
331 
332 
333def _telemetry(model: DashboardModel) -> str:
  t: RunTelemetry = model.telemetry
  if not t.available:
      return (
          "<section><h2>Run telemetry &middot; ClickHouse</h2>"
          '<p class="note">Unavailable (not configured or unreachable). '
          "GitHub + Temporal carry the dashboard regardless.</p></section>"
      )
  now = model.generated_at
  if t.activities:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(a.name)}</td>'
          f"<td>{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 froot spans in the window.</p>'
  summary = (
      f'<p class="note">{t.total_spans} spans &middot; '
      f"{t.error_spans} errored &middot; last activity "
      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}</section>"
  )
368 
369 
370def _review_heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.review_interval_seconds
373 
  def line(loop: ReviewLoop) -> str:
      if loop.live:
          dot, tail = "ok", ""
          if loop.last_tick is not None:
              nxt = _aware(loop.last_tick) + timedelta(seconds=interval)
              last = _ago(loop.last_tick, now)
              tail = (
                  f' <span class="mut">&middot; last {last}'
                  f" &middot; next {_until(nxt, now)}</span>"
              )
      else:
          dot = "bad" if loop.status in ("terminated", "none") else "warn"
          tail = f' <span class="mut">&middot; {escape(loop.status)}</span>'
      return (
          f'<div class="row">{_dot(dot)}'
          f'<span class="mono">{escape(loop.repo)}</span>{tail}</div>'
      )
391 
  if not model.review_loops:
      body = (
          '<p class="note">No determinism-review loops running '
          "(the transitive ring watches the @workflow.defn repos).</p>"
      )
  else:
      body = "".join(line(loop) for loop in model.review_loops)
  return (
      "<section><h2>Determinism review &middot; is it alive?</h2>"
      f"{body}</section>"
  )
403 
404 
405def _review_record(model: DashboardModel) -> str:
  r = model.review_record
  stats = "".join(
      (
          _stat(r.reviewed, "reviewed"),
          _stat(r.flagged, "flagged"),
          _stat(r.clean, "clean"),
          _stat(r.hazards, "hazards"),
          _stat(r.repos_covered, "repos covered"),
      )
  )
  note = (
      '<p class="note">The transitive ring: it chases first-party helper '
      "calls out of each workflow to catch a hazard the lexical CI kernel "
      "can&rsquo;t see. <b>Advisory</b> &mdash; the blocking gate stays the "
      "kernel&rsquo;s CI check. The hazard-resolved rate (was a flag gone on "
      "a later commit?) is a later loop; it needs accumulated history.</p>"
  )
  return (
      "<section><h2>Determinism review &middot; the transitive ring</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
427 
428 
429def _reviews(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.reviews:
      body = '<p class="note">No PRs reviewed yet.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.repo)}</td>'
          f"<td>{_review_pr_link(row)}</td>"
          f'<td class="mono mut">{escape((row.head_sha or "")[:7]) or "—"}'
          "</td>"
          f"<td>{_findings_cell(row)}</td>"
          f'<td class="mut">{escape(_ago(row.reviewed_at, now))}</td>'
          "</tr>"
          for row in model.reviews
      )
      body = (
          "<table><thead><tr><th>repo</th><th>pr</th><th>head</th>"
          "<th>findings</th><th>reviewed</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  return (
      "<section><h2>Determinism reviews &middot; the detail</h2>"
      f"{body}</section>"
  )
454 
455 
456def _footer() -> str:
  return (
      "<footer>"
      "<b>Authority envelope.</b> Stage 1 &mdash; froot holds "
      "<b>write authority</b> only: it opens PRs, a human approves every "
      "merge (commit authority = none). Trust, when any is granted, is "
      "earned, narrow to npm patch bumps, conditional on its environment "
      '(judge <span class="mono">gemma4:e4b</span>, lockfile-only regen), '
      "revocable, and time-expiring. Today it records the track record; it "
      "does not yet act on it.<br>"
      "Everything above is derived on this request from GitHub (outcomes) + "
      "Temporal (runs) + ClickHouse (telemetry). froot keeps no database; "
      "reload to recompute."
      "</footer>"
  )
471 
472 
473def _pr_link(row: BumpRow) -> str:
  if row.pr_url is None or row.pr_number is None:
      return '<span class="mut">—</span>'
  return f'<a href="{escape(row.pr_url, quote=True)}">#{row.pr_number}</a>'
477 
478 
479def _review_pr_link(row: ReviewRow) -> str:
  if row.pr_url is None or row.pr_number is None:
      return '<span class="mut">—</span>'
  return f'<a href="{escape(row.pr_url, quote=True)}">#{row.pr_number}</a>'
483 
484 
485def _findings_cell(row: ReviewRow) -> str:
  """A review's findings: 'clean', or the hazard count + rules + comment."""
  if row.findings == 0:
      return '<span class="ok">clean</span>'
  rules = (
      f' <span class="mono mut">{escape(", ".join(row.rules))}</span>'
      if row.rules
      else ""
  )
  comment = (
      f' <a href="{escape(row.comment_url, quote=True)}">comment</a>'
      if row.comment_url
      else ""
  )
  noun = "hazard" if row.findings == 1 else "hazards"
  return f'<span class="bad">{row.findings} {noun}</span>{rules}{comment}'
501 
502 
503def _state_tag(state: str) -> str:
  cls = {
      "merged": "ok",
      "open": "warn",
⋯ 39 lines hidden (lines 507–545)507        "closed": "mut",
      "terminated": "bad",
      "failed": "bad",
      "timed_out": "bad",
      "canceled": "warn",
  }.get(state, "mut")
  return f'<span class="{cls}">{escape(state)}</span>'
514 
515 
516def _short_id(workflow_id: str) -> str:
  """Drop the ``froot-bump-`` prefix for a readable failures row."""
  return workflow_id.removeprefix("froot-bump-")
519 
520 
521def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _heartbeat(model),
      _track_record(model),
      _verification(model),
      _judgment(model),
      _gate(model),
      _bumps(model),
      _failures(model),
      _review_heartbeat(model),
      _review_record(model),
      _reviews(model),
      _telemetry(model),
      _footer(),
  )
  return (
      '<!doctype html><html lang="en"><head><meta charset="utf-8">'
      '<meta name="viewport" content="width=device-width,initial-scale=1">'
      "<title>froot &middot; read-model</title>"
      f"<style>{_CSS}</style></head><body><main>"
      + "".join(parts)
      + "</main></body></html>"
  )

Verified: ruff (format + lint), mypy strict (93 files), and 186 tests — 11 new, covering loop liveness, the record counts, the repo-prefix attribution, and the rendered sections (flagged vs. clean).