The shadow gate (Phase 3) — a code-level reading

What changed, and why

froot opens dependency- and security-patch PRs and a human approves every merge. It has been record-only: the dashboard reports the track record but draws no conclusion from it. Phase 3 makes that record legible as trust.

For each (repo, loop) class it now derives whether the class has earned a hypothetical auto-merge gate move, and for each open PR whether it would auto-merge under that grant. This is the shadow gate — a dry run a steward watches before any authority is granted. Nothing acts. Phase 4 will flip the same pure functions to action; this PR only computes and renders the verdict.

The change is grounded in the Many Hands Engineering (MHE) trust economy. Read it in four beats: the pure verdict (policy/autonomy.py), how the read-model derives each class's standing, what the page shows, and a separate run-ledger instrumentation beat that makes the scan stage legible.

The pure verdict — policy/autonomy.py

The whole decision lives in one pure, dependency-free module so Phase 4 can reuse it verbatim. class_earned answers "has this class earned its gate?" from a track record; pr_autonomy answers "would this PR ride the grant?".

The module docstring names MHE's five properties the grant carries — earned, narrow, conditional, revocable, expiring — and is blunt about the caveat that keeps this advisory: the approval rate "is the headline, not the whole truth".

The five properties, then class_earned: enough decided PRs at a high enough rate, measured over a recent window.

src/froot/policy/autonomy.py · 125 lines

src/froot/policy/autonomy.py125 lines · Python

⋯ 10 lines hidden (lines 1–10)1"""The earned-autonomy policy — read-only today, the gate's logic tomorrow.
2 
3froot is record-only: every PR is human-approved. This is the pure logic that
4says, per the Many Hands Engineering trust economy (§3.6-3.7), whether a *class*
5of work (a loop, on a repo) has earned its gate move, and whether a given PR
6*would* auto-merge under that grant. Today the dashboard renders the verdict
7and nothing acts on it — the "shadow gate", the dry run that lets a steward
8watch the decision for weeks before granting authority. Phase 4 flips it to
9action by reusing these same functions; nothing here mutates anything.
10 
11The grant has MHE's five properties. **Earned**: a class earns its gate by a
12track record — a high enough *approval rate* over enough *decided* PRs.
13**Narrow**: the class is one loop on one repo; dependency-patch and
14security-patch are separate trust classes (§3.9), never sharing a record.
15**Conditional**: a PR rides the grant only if its own changelog read clean and
16its CI went green. **Revocable**: the grant is gated behind an explicit
17per-repo allowlist a steward controls, and lapses on its own when the rate
18falls. **Expiring**: the rate is measured over a recent window, not all of
19history (§2.11) — a class that has not acted lately starts from a lower
20baseline.
⋯ 44 lines hidden (lines 21–64)21 
22A caveat MHE is blunt about: the approval rate is the headline, *not the whole
23truth* — "track record alone can lie" (§3.7). A real grant triangulates it with
24post-merge rollback rate and sampled review, and resets when the *environment*
25it was earned in changes (a judge-model swap, a refactor) — MHE's fuller sense
26of "conditional". Those are deliberately not here yet, which is exactly why
27this stays advisory.
28"""
29 
30from __future__ import annotations
31 
32from froot.domain.base import Frozen
33 
34 
35class AutonomyPolicy(Frozen):
36    """The thresholds a class must clear to earn its gate move (advisory).
37 
38    Attributes:
39        min_rate: The approval (merge) rate a class needs over the window.
40        min_decided: How many PRs must have been decided in the window before
41            the rate means anything — a 100% rate on one PR is not a record.
42        window_days: The look-back window; trust is recent, not lifetime.
43        allowlisted_repos: The repos a steward has opted into auto-merge for.
44            Empty by default — the revocable switch, off until trust is granted.
45    """
46 
47    min_rate: float = 0.95
48    min_decided: int = 5
49    window_days: int = 90
50    allowlisted_repos: frozenset[str] = frozenset()
51 
52 
53class AutonomyVerdict(Frozen):
54    """Whether a PR would auto-merge under the grant, and the reason either way.
55 
56    Attributes:
57        would_merge: True iff every condition is met (advisory — nothing acts).
58        reason: The deciding factor — the grant met, or the first blocker.
59    """
60 
61    would_merge: bool
62    reason: str
63 
64 
65def class_earned(
66    decided: int, merged: int, policy: AutonomyPolicy
67) -> tuple[bool, str | None]:
68    """Whether a class has earned its gate move, and why not if it hasn't.
69 
70    Args:
71        decided: PRs of this class decided (merged or closed) in the window.
72        merged: How many of those were merged.
73        policy: The thresholds.
74 
75    Returns:
76        ``(earned, blocker)`` — ``blocker`` is ``None`` when earned, else the
77        short reason the gate has not moved.
78    """
79    if decided < policy.min_decided or decided == 0:
80        return False, f"only {decided}/{policy.min_decided} decided recently"
81    rate = merged / decided
82    if rate < policy.min_rate:
83        return False, (f"approval rate {rate:.0%} < {policy.min_rate:.0%}")
84    return True, None
⋯ 41 lines hidden (lines 85–125)85 
86 
87def pr_autonomy(
88    *,
89    repo: str,
90    verdict: str | None,
91    ci: str | None,
92    earned: bool,
93    blocker: str | None,
94    policy: AutonomyPolicy,
95) -> AutonomyVerdict:
96    """Whether one open PR would auto-merge under its class's grant.
97 
98    Reports the *substantive* blockers first — the earned grant, then this PR's
99    own clean-and-green conditions — and the steward's own revocable switch
100    (the allowlist) *last*. That ordering is what makes the shadow gate
101    watchable in its default, allowlist-off state: a held PR shows the real
102    thing to fix (``CI pending`` / ``class not earned``), and the bare
103    ``auto-merge not enabled`` reason appears only once a PR is otherwise fully
104    ready — i.e. exactly when flipping the switch would change the outcome.
105    ``would_merge`` still requires *every* condition, the allowlist included.
106    """
107    if not earned:
108        return AutonomyVerdict(
109            would_merge=False, reason=f"class not earned ({blocker})"
110        )
111    if verdict != "clean":
112        return AutonomyVerdict(
113            would_merge=False, reason=f"verdict is {verdict or 'unknown'}"
114        )
115    if ci != "passed":
116        return AutonomyVerdict(
117            would_merge=False, reason=f"CI {ci or 'pending'}"
118        )
119    if repo not in policy.allowlisted_repos:
120        return AutonomyVerdict(
121            would_merge=False, reason="auto-merge not enabled for this repo"
122        )
123    return AutonomyVerdict(
124        would_merge=True, reason="clean + green on an earned class"
125    )

pr_autonomy: the substantive blockers first, the steward's own allowlist switch last.

src/froot/policy/autonomy.py · 125 lines

src/froot/policy/autonomy.py125 lines · Python

⋯ 86 lines hidden (lines 1–86)1"""The earned-autonomy policy — read-only today, the gate's logic tomorrow.
2 
3froot is record-only: every PR is human-approved. This is the pure logic that
4says, per the Many Hands Engineering trust economy (§3.6-3.7), whether a *class*
5of work (a loop, on a repo) has earned its gate move, and whether a given PR
6*would* auto-merge under that grant. Today the dashboard renders the verdict
7and nothing acts on it — the "shadow gate", the dry run that lets a steward
8watch the decision for weeks before granting authority. Phase 4 flips it to
9action by reusing these same functions; nothing here mutates anything.
10 
11The grant has MHE's five properties. **Earned**: a class earns its gate by a
12track record — a high enough *approval rate* over enough *decided* PRs.
13**Narrow**: the class is one loop on one repo; dependency-patch and
14security-patch are separate trust classes (§3.9), never sharing a record.
15**Conditional**: a PR rides the grant only if its own changelog read clean and
16its CI went green. **Revocable**: the grant is gated behind an explicit
17per-repo allowlist a steward controls, and lapses on its own when the rate
18falls. **Expiring**: the rate is measured over a recent window, not all of
19history (§2.11) — a class that has not acted lately starts from a lower
20baseline.
21 
22A caveat MHE is blunt about: the approval rate is the headline, *not the whole
23truth* — "track record alone can lie" (§3.7). A real grant triangulates it with
24post-merge rollback rate and sampled review, and resets when the *environment*
25it was earned in changes (a judge-model swap, a refactor) — MHE's fuller sense
26of "conditional". Those are deliberately not here yet, which is exactly why
27this stays advisory.
28"""
29 
30from __future__ import annotations
31 
32from froot.domain.base import Frozen
33 
34 
35class AutonomyPolicy(Frozen):
36    """The thresholds a class must clear to earn its gate move (advisory).
37 
38    Attributes:
39        min_rate: The approval (merge) rate a class needs over the window.
40        min_decided: How many PRs must have been decided in the window before
41            the rate means anything — a 100% rate on one PR is not a record.
42        window_days: The look-back window; trust is recent, not lifetime.
43        allowlisted_repos: The repos a steward has opted into auto-merge for.
44            Empty by default — the revocable switch, off until trust is granted.
45    """
46 
47    min_rate: float = 0.95
48    min_decided: int = 5
49    window_days: int = 90
50    allowlisted_repos: frozenset[str] = frozenset()
51 
52 
53class AutonomyVerdict(Frozen):
54    """Whether a PR would auto-merge under the grant, and the reason either way.
55 
56    Attributes:
57        would_merge: True iff every condition is met (advisory — nothing acts).
58        reason: The deciding factor — the grant met, or the first blocker.
59    """
60 
61    would_merge: bool
62    reason: str
63 
64 
65def class_earned(
66    decided: int, merged: int, policy: AutonomyPolicy
67) -> tuple[bool, str | None]:
68    """Whether a class has earned its gate move, and why not if it hasn't.
69 
70    Args:
71        decided: PRs of this class decided (merged or closed) in the window.
72        merged: How many of those were merged.
73        policy: The thresholds.
74 
75    Returns:
76        ``(earned, blocker)`` — ``blocker`` is ``None`` when earned, else the
77        short reason the gate has not moved.
78    """
79    if decided < policy.min_decided or decided == 0:
80        return False, f"only {decided}/{policy.min_decided} decided recently"
81    rate = merged / decided
82    if rate < policy.min_rate:
83        return False, (f"approval rate {rate:.0%} < {policy.min_rate:.0%}")
84    return True, None
85 
86 
87def pr_autonomy(
88    *,
89    repo: str,
90    verdict: str | None,
91    ci: str | None,
92    earned: bool,
93    blocker: str | None,
94    policy: AutonomyPolicy,
95) -> AutonomyVerdict:
96    """Whether one open PR would auto-merge under its class's grant.
97 
98    Reports the *substantive* blockers first — the earned grant, then this PR's
99    own clean-and-green conditions — and the steward's own revocable switch
100    (the allowlist) *last*. That ordering is what makes the shadow gate
101    watchable in its default, allowlist-off state: a held PR shows the real
102    thing to fix (``CI pending`` / ``class not earned``), and the bare
103    ``auto-merge not enabled`` reason appears only once a PR is otherwise fully
104    ready — i.e. exactly when flipping the switch would change the outcome.
105    ``would_merge`` still requires *every* condition, the allowlist included.
106    """
107    if not earned:
108        return AutonomyVerdict(
109            would_merge=False, reason=f"class not earned ({blocker})"
110        )
111    if verdict != "clean":
112        return AutonomyVerdict(
113            would_merge=False, reason=f"verdict is {verdict or 'unknown'}"
114        )
115    if ci != "passed":
116        return AutonomyVerdict(
117            would_merge=False, reason=f"CI {ci or 'pending'}"
118        )
119    if repo not in policy.allowlisted_repos:
120        return AutonomyVerdict(
121            would_merge=False, reason="auto-merge not enabled for this repo"
122        )
123    return AutonomyVerdict(
124        would_merge=True, reason="clean + green on an earned class"
125    )

Deriving each class's standing — read_model.py

The read-model joins GitHub outcomes to Temporal runs and now folds in the earned-autonomy figures. Two small helpers window the record to recent decisions (trust is recent, not lifetime): _decided_at dates a PR by its merge instant, falling back to its open time because a BumpRow carries no close timestamp; _within is the inclusive look-back test.

The windowing helpers — the closed-PR proxy is noted in the docstring, not hidden.

src/froot/dashboard/read_model.py · 587 lines

src/froot/dashboard/read_model.py587 lines · Python

⋯ 250 lines hidden (lines 1–250)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,
  ClassGate,
  DashboardModel,
  Failure,
  Judgment,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
30)
31from froot.domain.loop import Loop
32from froot.domain.repo import RepoRef, TargetRepo
33from froot.policy.autonomy import AutonomyPolicy, class_earned, pr_autonomy
34from froot.policy.naming import review_workflow_id, scan_workflow_id
35from froot.result import Ok
36 
37if TYPE_CHECKING:
  from datetime import datetime
39 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      BumpExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
47 
48# The conservative fallback policy: empty allowlist, so the shadow gate holds
49# everything until a caller passes a real one (built from FROOT_AUTOMERGE_*).
50_DEFAULT_POLICY = AutonomyPolicy()
51 
52_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
53# Every way a bump can end without closing cleanly — all belong in the honest
54# Failures panel, not silently dropped.
55_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
56 
57 
58def _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
67 
68 
69def _scan_id(repo: str, loop: Loop) -> str | None:
  """The deterministic scan-loop id for an ``owner/name`` slug + loop."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return scan_workflow_id(TargetRepo(repo=ref), loop)
      case _:
          return None
76 
77 
78def _scan_loops(
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  scans: tuple[ScanExecution, ...],
82) -> tuple[ScanLoop, ...]:
  """One liveness row per configured (repo, loop) (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
  rows: list[ScanLoop] = []
  for loop in loops:
      for repo in repos:
          scan_id = _scan_id(repo, loop)
          execution = by_id.get(scan_id) if scan_id is not None else None
          if execution is None:
              rows.append(
                  ScanLoop(
                      repo=repo,
                      loop=loop.value,
                      status="none",
                      live=False,
                      last_tick=None,
                  )
              )
          else:
              rows.append(
                  ScanLoop(
                      repo=repo,
                      loop=loop.value,
                      status=execution.status,
                      live=execution.status in _LIVE_STATUSES,
                      last_tick=execution.start,
                  )
              )
  return tuple(rows)
115 
116 
117def _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
124 
125 
126def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
130) -> 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,
              loop=pr.loop,
              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)
167 
168 
169def _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
172 
173 
174def _minutes_between(
  start: datetime | None, end: datetime | None
176) -> 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)
181 
182 
183def _hours_between(
  start: datetime | None, end: datetime | None
185) -> 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)
190 
191 
192def _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),
  )
207 
208 
209def _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,
  )
225 
226 
227def _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,
  )
249 
250 
251def _decided_at(row: BumpRow) -> datetime | None:
  """When a decided PR was decided — merge time, else open time as a proxy.
253 
  A merged PR has an exact merge instant; a closed-unmerged one does not (the
  issues list froot reads carries no close timestamp), so it is windowed by
  when it opened. The proxy only nudges a closed PR's window membership by its
  own lifetime — close enough for a recency window measured in months.
  """
  return row.merged_at or row.opened_at
260 
261 
262def _within(when: datetime | None, now: datetime, window_days: int) -> bool:
  """Whether ``when`` falls within ``window_days`` before ``now`` (pure)."""
  if when is None:
      return False
  return (now - when).total_seconds() <= window_days * 86400.0
267 
⋯ 320 lines hidden (lines 268–587)268 
269def _class_gates(
  now: datetime,
  rows: tuple[BumpRow, ...],
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  policy: AutonomyPolicy,
275) -> tuple[ClassGate, ...]:
  """The earned-autonomy standing of each (repo, loop) class (advisory).
277 
  Counts only PRs *decided within the window* — trust is recent, not lifetime
  (§2.11) — so a class that stops shipping decays back below its gate. The
  budget figures (approvals / reclaim per week) translate the record into the
  steward-time MHE actually meters (§3.6): what the class costs now, and what
  moving its gate would hand back.
  """
  weeks = max(policy.window_days / 7.0, 1.0)
  gates: list[ClassGate] = []
  for loop in loops:
      for repo in repos:
          decided_rows = [
              r
              for r in rows
              if r.repo == repo
              and r.loop == loop.value
              and r.state in ("merged", "closed")
              and _within(_decided_at(r), now, policy.window_days)
          ]
          merged_rows = [r for r in decided_rows if r.state == "merged"]
          decided = len(decided_rows)
          merged = len(merged_rows)
          earned, blocker = class_earned(decided, merged, policy)
          # Reclaim is the budget a gate *move* hands back — so it is zero
          # until the class has actually earned the move. Counting the
          # clean-and-green merges of an un-earned class would imply savings
          # the gate would refuse (every PR stays held at "class not earned").
          reclaimable = (
              sum(
                  1
                  for r in merged_rows
                  if r.verdict == "clean" and r.ci == "passed"
              )
              if earned
              else 0
          )
          gates.append(
              ClassGate(
                  repo=repo,
                  loop=loop.value,
                  decided=decided,
                  merged=merged,
                  merge_rate=(merged / decided) if decided else None,
                  earned=earned,
                  blocker=blocker,
                  approvals_per_week=round(merged / weeks, 2),
                  reclaim_per_week=round(reclaimable / weeks, 2),
                  window_days=policy.window_days,
              )
          )
  return tuple(gates)
328 
329 
330def _gate(
  rows: tuple[BumpRow, ...],
  gates: tuple[ClassGate, ...],
  policy: AutonomyPolicy,
334) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, most-aged first, each carrying a verdict.
336 
  The verdict is the shadow gate's: *would* this PR auto-merge under its
  class's grant (advisory; nothing acts). The reason is the grant met, or the
  first blocker to fix.
  """
  earned_by_class = {(g.repo, g.loop): (g.earned, g.blocker) for g in gates}
  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,
  )
  annotated: list[BumpRow] = []
  for row in open_rows:
      earned, blocker = earned_by_class.get(
          (row.repo, row.loop), (False, "no record for this class")
      )
      verdict = pr_autonomy(
          repo=row.repo,
          verdict=row.verdict,
          ci=row.ci,
          earned=earned,
          blocker=blocker,
          policy=policy,
      )
      annotated.append(
          row.model_copy(
              update={
                  "would_auto_merge": verdict.would_merge,
                  "held_reason": (
                      None if verdict.would_merge else verdict.reason
                  ),
              }
          )
      )
  return tuple(annotated)
371 
372 
373def _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)
390 
391 
392def _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
399 
400 
401def _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-") + "-"
408 
409 
410def _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
419 
420 
421def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
423) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
425 
  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)
449 
450 
451def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
453) -> 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)
481 
482 
483def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
485) -> 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),
  )
497 
498 
499def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
506) -> 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
      ),
  )
531 
532 
533def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
  policy: AutonomyPolicy = _DEFAULT_POLICY,
  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],
552) -> 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
557 
  rows = _bump_rows(now, prs, bumps)
  class_gates = _class_gates(now, rows, repos, loops, policy)
  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, loops, scans),
      track_record=_track_record(rows),
      class_gates=class_gates,
      verification=_verification(rows),
      judgment=_judgment(rows),
      gate=_gate(rows, class_gates, policy),
      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,
  )

_class_gates builds one ClassGate per configured (repo, loop): the windowed decided/merged counts, the merge rate, the earned verdict, and a steward-time budget — approvals/week the class costs now, and the clean-and-green slice a gate move would reclaim.

Note the spotlight: reclaim is zeroed unless the class is earned.

src/froot/dashboard/read_model.py · 587 lines

src/froot/dashboard/read_model.py587 lines · Python

⋯ 268 lines hidden (lines 1–268)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,
  ClassGate,
  DashboardModel,
  Failure,
  Judgment,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
30)
31from froot.domain.loop import Loop
32from froot.domain.repo import RepoRef, TargetRepo
33from froot.policy.autonomy import AutonomyPolicy, class_earned, pr_autonomy
34from froot.policy.naming import review_workflow_id, scan_workflow_id
35from froot.result import Ok
36 
37if TYPE_CHECKING:
  from datetime import datetime
39 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      BumpExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
47 
48# The conservative fallback policy: empty allowlist, so the shadow gate holds
49# everything until a caller passes a real one (built from FROOT_AUTOMERGE_*).
50_DEFAULT_POLICY = AutonomyPolicy()
51 
52_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
53# Every way a bump can end without closing cleanly — all belong in the honest
54# Failures panel, not silently dropped.
55_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
56 
57 
58def _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
67 
68 
69def _scan_id(repo: str, loop: Loop) -> str | None:
  """The deterministic scan-loop id for an ``owner/name`` slug + loop."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return scan_workflow_id(TargetRepo(repo=ref), loop)
      case _:
          return None
76 
77 
78def _scan_loops(
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  scans: tuple[ScanExecution, ...],
82) -> tuple[ScanLoop, ...]:
  """One liveness row per configured (repo, loop) (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
  rows: list[ScanLoop] = []
  for loop in loops:
      for repo in repos:
          scan_id = _scan_id(repo, loop)
          execution = by_id.get(scan_id) if scan_id is not None else None
          if execution is None:
              rows.append(
                  ScanLoop(
                      repo=repo,
                      loop=loop.value,
                      status="none",
                      live=False,
                      last_tick=None,
                  )
              )
          else:
              rows.append(
                  ScanLoop(
                      repo=repo,
                      loop=loop.value,
                      status=execution.status,
                      live=execution.status in _LIVE_STATUSES,
                      last_tick=execution.start,
                  )
              )
  return tuple(rows)
115 
116 
117def _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
124 
125 
126def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
130) -> 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,
              loop=pr.loop,
              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)
167 
168 
169def _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
172 
173 
174def _minutes_between(
  start: datetime | None, end: datetime | None
176) -> 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)
181 
182 
183def _hours_between(
  start: datetime | None, end: datetime | None
185) -> 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)
190 
191 
192def _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),
  )
207 
208 
209def _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,
  )
225 
226 
227def _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,
  )
249 
250 
251def _decided_at(row: BumpRow) -> datetime | None:
  """When a decided PR was decided — merge time, else open time as a proxy.
253 
  A merged PR has an exact merge instant; a closed-unmerged one does not (the
  issues list froot reads carries no close timestamp), so it is windowed by
  when it opened. The proxy only nudges a closed PR's window membership by its
  own lifetime — close enough for a recency window measured in months.
  """
  return row.merged_at or row.opened_at
260 
261 
262def _within(when: datetime | None, now: datetime, window_days: int) -> bool:
  """Whether ``when`` falls within ``window_days`` before ``now`` (pure)."""
  if when is None:
      return False
  return (now - when).total_seconds() <= window_days * 86400.0
267 
268 
269def _class_gates(
  now: datetime,
  rows: tuple[BumpRow, ...],
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  policy: AutonomyPolicy,
275) -> tuple[ClassGate, ...]:
  """The earned-autonomy standing of each (repo, loop) class (advisory).
277 
  Counts only PRs *decided within the window* — trust is recent, not lifetime
  (§2.11) — so a class that stops shipping decays back below its gate. The
  budget figures (approvals / reclaim per week) translate the record into the
  steward-time MHE actually meters (§3.6): what the class costs now, and what
  moving its gate would hand back.
  """
  weeks = max(policy.window_days / 7.0, 1.0)
  gates: list[ClassGate] = []
  for loop in loops:
      for repo in repos:
          decided_rows = [
              r
              for r in rows
              if r.repo == repo
              and r.loop == loop.value
              and r.state in ("merged", "closed")
              and _within(_decided_at(r), now, policy.window_days)
          ]
          merged_rows = [r for r in decided_rows if r.state == "merged"]
          decided = len(decided_rows)
          merged = len(merged_rows)
          earned, blocker = class_earned(decided, merged, policy)
          # Reclaim is the budget a gate *move* hands back — so it is zero
          # until the class has actually earned the move. Counting the
          # clean-and-green merges of an un-earned class would imply savings
          # the gate would refuse (every PR stays held at "class not earned").
          reclaimable = (
              sum(
                  1
                  for r in merged_rows
                  if r.verdict == "clean" and r.ci == "passed"
              )
              if earned
              else 0
          )
          gates.append(
              ClassGate(
                  repo=repo,
                  loop=loop.value,
                  decided=decided,
                  merged=merged,
                  merge_rate=(merged / decided) if decided else None,
                  earned=earned,
                  blocker=blocker,
                  approvals_per_week=round(merged / weeks, 2),
                  reclaim_per_week=round(reclaimable / weeks, 2),
                  window_days=policy.window_days,
              )
          )
  return tuple(gates)
⋯ 260 lines hidden (lines 328–587)328 
329 
330def _gate(
  rows: tuple[BumpRow, ...],
  gates: tuple[ClassGate, ...],
  policy: AutonomyPolicy,
334) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, most-aged first, each carrying a verdict.
336 
  The verdict is the shadow gate's: *would* this PR auto-merge under its
  class's grant (advisory; nothing acts). The reason is the grant met, or the
  first blocker to fix.
  """
  earned_by_class = {(g.repo, g.loop): (g.earned, g.blocker) for g in gates}
  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,
  )
  annotated: list[BumpRow] = []
  for row in open_rows:
      earned, blocker = earned_by_class.get(
          (row.repo, row.loop), (False, "no record for this class")
      )
      verdict = pr_autonomy(
          repo=row.repo,
          verdict=row.verdict,
          ci=row.ci,
          earned=earned,
          blocker=blocker,
          policy=policy,
      )
      annotated.append(
          row.model_copy(
              update={
                  "would_auto_merge": verdict.would_merge,
                  "held_reason": (
                      None if verdict.would_merge else verdict.reason
                  ),
              }
          )
      )
  return tuple(annotated)
371 
372 
373def _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)
390 
391 
392def _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
399 
400 
401def _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-") + "-"
408 
409 
410def _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
419 
420 
421def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
423) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
425 
  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)
449 
450 
451def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
453) -> 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)
481 
482 
483def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
485) -> 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),
  )
497 
498 
499def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
506) -> 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
      ),
  )
531 
532 
533def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
  policy: AutonomyPolicy = _DEFAULT_POLICY,
  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],
552) -> 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
557 
  rows = _bump_rows(now, prs, bumps)
  class_gates = _class_gates(now, rows, repos, loops, policy)
  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, loops, scans),
      track_record=_track_record(rows),
      class_gates=class_gates,
      verification=_verification(rows),
      judgment=_judgment(rows),
      gate=_gate(rows, class_gates, policy),
      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,
  )

_gate annotates each open PR with the shadow verdict via pr_autonomy + an immutable model_copy.

src/froot/dashboard/read_model.py · 587 lines

src/froot/dashboard/read_model.py587 lines · Python

⋯ 329 lines hidden (lines 1–329)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,
  ClassGate,
  DashboardModel,
  Failure,
  Judgment,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
30)
31from froot.domain.loop import Loop
32from froot.domain.repo import RepoRef, TargetRepo
33from froot.policy.autonomy import AutonomyPolicy, class_earned, pr_autonomy
34from froot.policy.naming import review_workflow_id, scan_workflow_id
35from froot.result import Ok
36 
37if TYPE_CHECKING:
  from datetime import datetime
39 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      BumpExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
47 
48# The conservative fallback policy: empty allowlist, so the shadow gate holds
49# everything until a caller passes a real one (built from FROOT_AUTOMERGE_*).
50_DEFAULT_POLICY = AutonomyPolicy()
51 
52_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
53# Every way a bump can end without closing cleanly — all belong in the honest
54# Failures panel, not silently dropped.
55_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
56 
57 
58def _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
67 
68 
69def _scan_id(repo: str, loop: Loop) -> str | None:
  """The deterministic scan-loop id for an ``owner/name`` slug + loop."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return scan_workflow_id(TargetRepo(repo=ref), loop)
      case _:
          return None
76 
77 
78def _scan_loops(
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  scans: tuple[ScanExecution, ...],
82) -> tuple[ScanLoop, ...]:
  """One liveness row per configured (repo, loop) (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
  rows: list[ScanLoop] = []
  for loop in loops:
      for repo in repos:
          scan_id = _scan_id(repo, loop)
          execution = by_id.get(scan_id) if scan_id is not None else None
          if execution is None:
              rows.append(
                  ScanLoop(
                      repo=repo,
                      loop=loop.value,
                      status="none",
                      live=False,
                      last_tick=None,
                  )
              )
          else:
              rows.append(
                  ScanLoop(
                      repo=repo,
                      loop=loop.value,
                      status=execution.status,
                      live=execution.status in _LIVE_STATUSES,
                      last_tick=execution.start,
                  )
              )
  return tuple(rows)
115 
116 
117def _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
124 
125 
126def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
130) -> 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,
              loop=pr.loop,
              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)
167 
168 
169def _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
172 
173 
174def _minutes_between(
  start: datetime | None, end: datetime | None
176) -> 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)
181 
182 
183def _hours_between(
  start: datetime | None, end: datetime | None
185) -> 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)
190 
191 
192def _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),
  )
207 
208 
209def _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,
  )
225 
226 
227def _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,
  )
249 
250 
251def _decided_at(row: BumpRow) -> datetime | None:
  """When a decided PR was decided — merge time, else open time as a proxy.
253 
  A merged PR has an exact merge instant; a closed-unmerged one does not (the
  issues list froot reads carries no close timestamp), so it is windowed by
  when it opened. The proxy only nudges a closed PR's window membership by its
  own lifetime — close enough for a recency window measured in months.
  """
  return row.merged_at or row.opened_at
260 
261 
262def _within(when: datetime | None, now: datetime, window_days: int) -> bool:
  """Whether ``when`` falls within ``window_days`` before ``now`` (pure)."""
  if when is None:
      return False
  return (now - when).total_seconds() <= window_days * 86400.0
267 
268 
269def _class_gates(
  now: datetime,
  rows: tuple[BumpRow, ...],
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  policy: AutonomyPolicy,
275) -> tuple[ClassGate, ...]:
  """The earned-autonomy standing of each (repo, loop) class (advisory).
277 
  Counts only PRs *decided within the window* — trust is recent, not lifetime
  (§2.11) — so a class that stops shipping decays back below its gate. The
  budget figures (approvals / reclaim per week) translate the record into the
  steward-time MHE actually meters (§3.6): what the class costs now, and what
  moving its gate would hand back.
  """
  weeks = max(policy.window_days / 7.0, 1.0)
  gates: list[ClassGate] = []
  for loop in loops:
      for repo in repos:
          decided_rows = [
              r
              for r in rows
              if r.repo == repo
              and r.loop == loop.value
              and r.state in ("merged", "closed")
              and _within(_decided_at(r), now, policy.window_days)
          ]
          merged_rows = [r for r in decided_rows if r.state == "merged"]
          decided = len(decided_rows)
          merged = len(merged_rows)
          earned, blocker = class_earned(decided, merged, policy)
          # Reclaim is the budget a gate *move* hands back — so it is zero
          # until the class has actually earned the move. Counting the
          # clean-and-green merges of an un-earned class would imply savings
          # the gate would refuse (every PR stays held at "class not earned").
          reclaimable = (
              sum(
                  1
                  for r in merged_rows
                  if r.verdict == "clean" and r.ci == "passed"
              )
              if earned
              else 0
          )
          gates.append(
              ClassGate(
                  repo=repo,
                  loop=loop.value,
                  decided=decided,
                  merged=merged,
                  merge_rate=(merged / decided) if decided else None,
                  earned=earned,
                  blocker=blocker,
                  approvals_per_week=round(merged / weeks, 2),
                  reclaim_per_week=round(reclaimable / weeks, 2),
                  window_days=policy.window_days,
              )
          )
  return tuple(gates)
328 
329 
330def _gate(
  rows: tuple[BumpRow, ...],
  gates: tuple[ClassGate, ...],
  policy: AutonomyPolicy,
334) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, most-aged first, each carrying a verdict.
336 
  The verdict is the shadow gate's: *would* this PR auto-merge under its
  class's grant (advisory; nothing acts). The reason is the grant met, or the
  first blocker to fix.
  """
  earned_by_class = {(g.repo, g.loop): (g.earned, g.blocker) for g in gates}
  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,
  )
  annotated: list[BumpRow] = []
  for row in open_rows:
      earned, blocker = earned_by_class.get(
          (row.repo, row.loop), (False, "no record for this class")
      )
      verdict = pr_autonomy(
          repo=row.repo,
          verdict=row.verdict,
          ci=row.ci,
          earned=earned,
          blocker=blocker,
          policy=policy,
      )
      annotated.append(
          row.model_copy(
              update={
                  "would_auto_merge": verdict.would_merge,
                  "held_reason": (
                      None if verdict.would_merge else verdict.reason
                  ),
              }
          )
      )
  return tuple(annotated)
⋯ 217 lines hidden (lines 371–587)371 
372 
373def _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)
390 
391 
392def _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
399 
400 
401def _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-") + "-"
408 
409 
410def _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
419 
420 
421def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
423) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
425 
  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)
449 
450 
451def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
453) -> 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)
481 
482 
483def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
485) -> 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),
  )
497 
498 
499def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
506) -> 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
      ),
  )
531 
532 
533def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
  policy: AutonomyPolicy = _DEFAULT_POLICY,
  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],
552) -> 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
557 
  rows = _bump_rows(now, prs, bumps)
  class_gates = _class_gates(now, rows, repos, loops, policy)
  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, loops, scans),
      track_record=_track_record(rows),
      class_gates=class_gates,
      verification=_verification(rows),
      judgment=_judgment(rows),
      gate=_gate(rows, class_gates, policy),
      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,
  )

What the page shows — model.py + render.py

The view model gains a frozen ClassGate, and BumpRow gains would_auto_merge / held_reason (plus a loop label so the two loops' records never mix). Every figure is computed in the read-model, so the renderer stays a dumb projection.

ClassGate — the per-class standing, fully derived.

src/froot/dashboard/model.py · 357 lines

src/froot/dashboard/model.py357 lines · Python

⋯ 101 lines hidden (lines 1–101)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.
      loop: Which maintenance loop this row is for (``dependency-patch`` /
          ``security-patch``), so the two loops' scans show distinctly.
      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.
  """
45 
  repo: str
  loop: str = "dependency-patch"
  status: str
  live: bool
  last_tick: datetime | None
51 
52 
53class BumpRow(Frozen):
  """One proposed dependency bump, joined across GitHub and Temporal.
55 
  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.
59 
  Attributes:
      repo: The ``owner/name`` slug.
      loop: Which loop proposed it (``dependency-patch`` /
          ``security-patch``), from the PR's loop label — so the two loops'
          records never mix.
      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``.
      would_auto_merge: For an open PR, whether it would auto-merge under the
          advisory earned-autonomy grant (the shadow gate; nothing acts).
      held_reason: Why an open PR would *not* auto-merge, the first blocker.
  """
83 
  repo: str
  loop: str = "dependency-patch"
  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
  would_auto_merge: bool = False
  held_reason: str | None = None
100 
101 
102class ClassGate(Frozen):
  """The earned-autonomy standing of one (repo, loop) class — advisory only.
104 
  The MHE economics of approval (§3.6) made legible: a class earns its gate
  move with a high enough approval rate over enough recently-decided PRs,
  and the budget framing shows what moving the gate would reclaim.
108 
  Attributes:
      repo: The ``owner/name`` slug.
      loop: The loop this class is for.
      decided: PRs decided (merged or closed) in the recent window.
      merged: How many of those were merged.
      merge_rate: ``merged / decided``, or ``None`` if none decided.
      earned: Whether the class has earned its gate move under the policy.
      blocker: Why it has not, if it has not (else ``None``).
      approvals_per_week: The steward approvals this class costs now
          (merges per week over the window) — the current budget draw.
      reclaim_per_week: How many of those a gate move would auto-merge
          (the clean-and-green ones), i.e. the budget reclaimed — ``0`` until
          the class is ``earned``, since an un-earned class reclaims nothing.
      window_days: The look-back window the figures cover.
  """
124 
  repo: str
  loop: str
  decided: int
  merged: int
  merge_rate: float | None
  earned: bool
  blocker: str | None
  approvals_per_week: float
  reclaim_per_week: float
  window_days: int
⋯ 223 lines hidden (lines 135–357)135 
136 
137class Failure(Frozen):
  """A bump loop that did not close — the honest friction signal.
139 
  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.
  """
146 
  workflow_id: str
  kind: str
  reason: str | None
  when: datetime | None
151 
152 
153class ActivityStat(Frozen):
  """Latency of one activity stage, from traces (run-telemetry enrichment).
155 
  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.
  """
162 
  name: str
  count: int
  avg_ms: float
  max_ms: float
167 
168 
169class RunTelemetry(Frozen):
  """Trace-derived run telemetry from ClickHouse, or an unavailable marker.
171 
  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.
  """
180 
  available: bool
  total_spans: int
  error_spans: int
  last_activity: datetime | None
  window_days: int
  activities: tuple[ActivityStat, ...]
187 
188 
189class TrackRecord(Frozen):
  """The reputation headline, derived from GitHub PR outcomes.
191 
  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``.
  """
201 
  opened: int
  merged: int
  closed_unmerged: int
  open_now: int
  merge_rate: float | None
  median_ttm_minutes: float | None
208 
209 
210class Verification(Frozen):
  """The CI-oracle breakdown — kept honest about whether an oracle existed.
212 
  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``).
  """
222 
  passed: int
  failed: int
  absent: int
  timed_out: int
  unknown: int
  oracle_existed: int
  with_reading: int
230 
231 
232class Judgment(Frozen):
  """The model's changelog-verdict mix and its calibration against CI.
234 
  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.
  """
243 
  clean: int
  risky: int
  unknown: int
  none: int
  clean_but_failed: int
  flagged_but_passed: int
250 
251 
252class ReviewLoop(Frozen):
  """Liveness of one repo's determinism-review loop (the transitive ring).
254 
  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.
  """
263 
  repo: str
  status: str
  live: bool
  last_tick: datetime | None
268 
269 
270class ReviewRow(Frozen):
  """One per-PR determinism review, from its ``PrReviewWorkflow`` result.
272 
  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.
  """
284 
  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
294 
295 
296class ReviewRecord(Frozen):
  """The determinism reviewer's headline, derived from completed reviews.
298 
  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.
302 
  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.
  """
310 
  reviewed: int
  flagged: int
  clean: int
  hazards: int
  repos_covered: int
316 
317 
318class DashboardModel(Frozen):
  """The whole 10,000ft view, fully derived and ready to render.
320 
  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.
      class_gates: The earned-autonomy standing per (repo, loop) — advisory.
      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).
  """
340 
  generated_at: datetime
  repos_configured: tuple[str, ...]
  scan_interval_seconds: int
  sources: tuple[SourceHealth, ...]
  scan_loops: tuple[ScanLoop, ...]
  track_record: TrackRecord
  class_gates: tuple[ClassGate, ...]
  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

The renderer adds an "Earned autonomy · the shadow gate" panel and a per-PR badge in the approval queue. The badge reads would auto-merge or held · <reason>; both the panel note and the gate note are scrupulous that nothing acts and that a merge-rate record is not a confirmed-good (or security-reviewed) outcome.

The panel: a class row table plus a folded MHE framing note.

src/froot/dashboard/render.py · 763 lines

src/froot/dashboard/render.py763 lines · Python

⋯ 274 lines hidden (lines 1–274)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,
      ClassGate,
      DashboardModel,
      ReviewLoop,
      ReviewRow,
      RunTelemetry,
      ScanLoop,
  )
26 
27_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
29--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#fafafa}
30@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
31--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
32--accent:#58a6ff;--card:#141414}}
33*{box-sizing:border-box}
34body{margin:0;background:var(--bg);color:var(--fg);
35font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
36main{max-width:820px;margin:0 auto;padding:34px 20px 72px}
37h1{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}
41section{margin:30px 0 0}
42h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
43color:var(--mut);margin:0 0 12px;font-weight:600;
44border-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%;
49margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
52table{width:100%;border-collapse:collapse;font-size:13px}
53th,td{text-align:left;padding:6px 12px 6px 0;
54border-bottom:1px solid var(--line);vertical-align:top}
55th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
56letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
58a{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}
63footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
64color:var(--mut);font-size:12px;line-height:1.7}
65footer b{color:var(--fg);font-weight:600}
66/* loop groups — collapsible (collapsed by default), two-level hierarchy */
67details.loop{margin:24px 0 0}
68summary.loophead{display:flex;align-items:baseline;gap:6px 12px;flex-wrap:wrap;
69cursor:pointer;list-style:none;margin:0;padding:0 0 7px;
70border-bottom:2px solid var(--fg)}
71summary.loophead::-webkit-details-marker{display:none}
72summary.loophead::before{content:"\\25B8";color:var(--mut);margin-right:5px;
73font-size:11px;position:relative;top:-1px}
74details.loop[open]>summary.loophead::before{content:"\\25BE"}
75summary.loophead h2{font-size:16px;text-transform:none;letter-spacing:-.01em;
76color:var(--fg);border:0;margin:0;padding:0;font-weight:700}
77summary.loophead .sub{color:var(--mut);font-size:12px}
78summary.loophead .glance{margin-left:auto;display:flex;flex-wrap:wrap;
79align-items:baseline;gap:3px 14px;font-size:12px;color:var(--mut)}
80summary.loophead .glance b{color:var(--fg);font-weight:600}
81/* the shared-infra group reads as not-a-loop: lighter, dashed, muted */
82details.loop.shared>summary.loophead{border-bottom:1px dashed var(--line)}
83details.loop.shared>summary.loophead h2{font-size:13px;color:var(--mut);
84font-weight:600;text-transform:uppercase;letter-spacing:.06em}
85/* subsections sit tighter inside a loop; the loop rule carries the weight */
86details.loop section{margin:18px 0 0}
87details.loop section>h2{border-bottom:0;padding-bottom:0;margin-bottom:9px}
88/* foldable framing — the MHE notes, one click away, so the page scans clean */
89details.why{margin:9px 0 0}
90details.why>summary{color:var(--accent);font-size:12px;cursor:pointer;
91list-style:none}
92details.why>summary::-webkit-details-marker{display:none}
93details.why>summary::before{content:"\\002b  why";font-weight:600}
94details.why[open]>summary::before{content:"\\2212  why"}
95details.why .note{margin:6px 0 0}
96"""
97 
98_CI_CLASS = {
  "passed": "ok",
  "failed": "bad",
  "absent": "mut",
  "timed_out": "warn",
103}
104_VERDICT_CLASS = {"clean": "ok", "risky": "warn", "unknown": "mut"}
105 
106 
107def _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)
110 
111 
112def _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"
124 
125 
126def _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"
138 
139 
140def _dot(kind: str) -> str:
  """A status dot span (``ok`` / ``warn`` / ``bad`` / ``mute``)."""
  return f'<span class="dot {kind}"></span>'
143 
144 
145def _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>'
151 
152 
153def _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>'
  )
158 
159 
160def _why(note_html: str) -> str:
  """Fold a framing note behind a native ``<details>`` so the page scans."""
  return f'<details class="why"><summary></summary>{note_html}</details>'
163 
164 
165def _glance(n: object, label: str) -> str:
  """One at-a-glance item for a loop header (``<b>12</b> proposed``)."""
  return f"<span><b>{escape(str(n))}</b> {escape(label)}</span>"
168 
169 
170def _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>"
  )
190 
191 
192def _heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.scan_interval_seconds
195 
  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>'
          f' <span class="mut">{escape(loop.loop)}</span>{tail}</div>'
      )
214 
  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 it alive?</h2>{body}</section>"
220 
221 
222def _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>{_why(note)}</section>'
  )
250 
251 
252def _rate_cell(rate: float | None) -> str:
  """A class's approval rate as a percent, or an em-dash if none decided."""
  if rate is None:
      return '<span class="mut">—</span>'
  return f"{rate * 100:.0f}%"
257 
258 
259def _earned_cell(g: ClassGate) -> str:
  """A class's gate standing: a green 'earned', else the muted blocker."""
  if g.earned:
      return f'{_dot("ok")}<span class="ok">earned</span>'
  blocker = escape(g.blocker or "not earned")
  return f'{_dot("mut")}<span class="mut">{blocker}</span>'
265 
266 
267def _budget_cell(g: ClassGate) -> str:
  """The gate in steward-time: approvals/week now &rarr; reclaimable/week."""
  return (
      f'<span class="mono">{g.approvals_per_week:.1f}</span> appr'
      f' <span class="mut">&rarr; {g.reclaim_per_week:.1f} reclaimable</span>'
  )
273 
274 
275def _class_gates(model: DashboardModel) -> str:
  """The earned-autonomy panel — the shadow gate's per-class standing.
277 
  Read-only: it renders whether each (repo, loop) class *would* have earned a
  gate move, and the steward-budget that move would reclaim. Nothing acts.
  """
  gates = model.class_gates
  if not gates:
      body = (
          '<p class="note">No classes yet &mdash; a (repo, loop) earns a '
          "gate move from its own track record.</p>"
      )
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(g.repo)}</td>'
          f'<td class="mut">{escape(g.loop)}</td>'
          f"<td>{g.decided} in {g.window_days}d</td>"
          f"<td>{_rate_cell(g.merge_rate)}</td>"
          f"<td>{_earned_cell(g)}</td>"
          f"<td>{_budget_cell(g)}</td>"
          "</tr>"
          for g in gates
      )
      body = (
          "<table><thead><tr><th>repo</th><th>loop</th><th>decided</th>"
          "<th>rate</th><th>gate</th><th>budget / week</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  note = (
      '<p class="note">Each <b>class</b> is one loop on one repo &mdash; a '
      "distinct trust unit (a security patch is not a version bump, and they "
      "never share a record). A class earns its gate from a high enough "
      "<b>approval rate</b> over enough recently-<b>decided</b> PRs; the "
      "window keeps trust recent, not lifetime. <b>Budget</b> reads the "
      "gate in steward-time &mdash; approvals/week is what the class costs "
      "a human now, reclaimable is the clean-and-green slice a gate move "
      "would hand back. <b>Advisory only</b>: auto-merge is allowlist-gated "
      "and off, and the approval rate counts a human <i>merge</i>, not a "
      "confirmed-good outcome &mdash; with no revert tracking yet, a high "
      "rate can still mask rubber-stamping (&sect;3.7). Rollback tracking "
      "and sampled review come before anything acts.</p>"
  )
  return (
      "<section><h2>Earned autonomy &middot; the shadow gate</h2>"
⋯ 444 lines hidden (lines 320–763)320        f"{body}{_why(note)}</section>"
  )
322 
323 
324def _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 = _why(
          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>'
  )
349 
350 
351def _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>{_why(note)}</section>'
  )
376 
377 
378def _shadow_badge(row: BumpRow) -> str:
  """Whether this open PR *would* auto-merge under its class's grant.
380 
  Advisory: a green 'would auto-merge' means every condition is met and the
  human approval is the only thing still in the way; otherwise the first
  blocker, muted, so the held reason is the thing to fix.
  """
  if row.would_auto_merge:
      return '<span class="ok">would auto-merge</span>'
  reason = row.held_reason or "held"
  return f'<span class="mut">held &middot; {escape(reason)}</span>'
389 
390 
391def _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>{_shadow_badge(row)}</td>"
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.gate
      )
      note = _why(
          '<p class="note">The <b>shadow gate</b> column is advisory &mdash; '
          "froot acts on nothing. A green &lsquo;would auto-merge&rsquo; "
          "means this PR met its class&rsquo;s grant and the human approval "
          "is the only remaining step; it rests on the single CI oracle and "
          "a merge-rate record, which (esp. for a security patch) is not the "
          "same as a confirmed-good or security-reviewed outcome. You still "
          "own every merge.</p>"
      )
      body = (
          "<table><thead><tr><th>package</th><th>waiting</th>"
          "<th>shadow gate</th><th>pr</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>{note}"
      )
  return (
      "<section><h2>Approval gate &middot; what a human owns</h2>"
      f"{body}</section>"
  )
423 
424 
425def _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>"
449 
450 
451def _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>"
  )
469 
470 
471def _telem_glance(model: DashboardModel) -> str:
  """The telemetry group's at-a-glance (or an ``unavailable`` marker)."""
  t = model.telemetry
  if not t.available:
      return '<span class="mut">unavailable</span>'
  return _glance(t.total_spans, f"spans / {t.window_days}d")
477 
478 
479def _telemetry(model: DashboardModel) -> str:
  """The telemetry body — header lives in the shared loop group above it."""
  t: RunTelemetry = model.telemetry
  if not t.available:
      return (
          "<section>"
          '<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 f"<section>{summary}{table}</section>"
512 
513 
514def _review_heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.review_interval_seconds
517 
  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>'
      )
535 
  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 f"<section><h2>Is it alive?</h2>{body}</section>"
544 
545 
546def _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>Track record &middot; the reputation</h2>"
      f'<div class="stats">{stats}</div>{_why(note)}</section>'
  )
568 
569 
570def _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 f"<section><h2>Reviews &middot; the detail</h2>{body}</section>"
592 
593 
594def _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 a single loop on a single repo (a version bump is "
      "not a security patch; they never share a record), conditional on its "
      'environment (judge <span class="mono">gemma4:e4b</span>, '
      "lockfile-only regen), revocable, and time-expiring. Today it records "
      "the track "
      "record and renders a <b>shadow gate</b> &mdash; a dry run of the "
      "auto-merge decision that still acts on nothing.<br>"
      "Everything above is derived on this request from GitHub (outcomes) + "
      "Temporal (runs) + ClickHouse (telemetry). froot keeps no database; "
      "reload to recompute."
      "</footer>"
  )
612 
613 
614def _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>'
618 
619 
620def _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>'
624 
625 
626def _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}'
642 
643 
644def _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>'
655 
656 
657def _short_id(workflow_id: str) -> str:
  """Drop the ``froot-bump-`` prefix for a readable failures row."""
  return workflow_id.removeprefix("froot-bump-")
660 
661 
662def _dep_glance(model: DashboardModel) -> str:
  """The dependency-patch loop's at-a-glance, shown in its header."""
  t = model.track_record
  live = sum(1 for loop in model.scan_loops if loop.live)
  total = len(model.scan_loops)
  rate = "—" if t.merge_rate is None else f"{t.merge_rate * 100:.0f}%"
  items = [
      _glance(f"{live}/{total}", "live"),
      _glance(t.opened, "proposed"),
      _glance(rate, "merged"),
  ]
  if model.class_gates:
      earned = sum(1 for g in model.class_gates if g.earned)
      items.append(_glance(f"{earned}/{len(model.class_gates)}", "earned"))
  if t.open_now:
      items.append(_glance(t.open_now, "awaiting you"))
  return "".join(items)
679 
680 
681def _det_glance(model: DashboardModel) -> str:
  """The determinism-review loop's at-a-glance, shown in its header."""
  r = model.review_record
  return "".join(
      (
          _glance(r.repos_covered, "repos"),
          _glance(r.reviewed, "reviewed"),
          _glance(r.hazards, "hazards"),
      )
  )
691 
692 
693def _loop(
  title: str,
  sub: str,
  glance: str,
  sections: tuple[str, ...],
  *,
  shared: bool = False,
700) -> str:
  """A collapsible group: a prominent header + at-a-glance, collapsed.
702 
  The header (with its at-a-glance) is the ``<summary>``, so a collapsed group
  still shows the headline; expanding reveals the subsections. ``shared`` is a
  non-loop infra group (telemetry) with a lighter, distinct header.
  """
  cls = "loop shared" if shared else "loop"
  head = (
      '<summary class="loophead">'
      f"<h2>{escape(title)}</h2>"
      f'<span class="sub">{sub}</span>'
      f'<span class="glance">{glance}</span>'
      "</summary>"
  )
  return f'<details class="{cls}">{head}{"".join(sections)}</details>'
716 
717 
718def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _loop(
          "Dependency-patch",
          "npm + uv &middot; scan &rarr; bump &rarr; CI &rarr; merge",
          _dep_glance(model),
          (
              _heartbeat(model),
              _track_record(model),
              _class_gates(model),
              _verification(model),
              _judgment(model),
              _gate(model),
              _bumps(model),
              _failures(model),
          ),
      ),
      _loop(
          "Determinism review",
          "the transitive ring &middot; advisory",
          _det_glance(model),
          (
              _review_heartbeat(model),
              _review_record(model),
              _reviews(model),
          ),
      ),
      _loop(
          "Run telemetry",
          "ClickHouse &middot; trace-derived, best-effort",
          _telem_glance(model),
          (_telemetry(model),),
          shared=True,
      ),
      _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>"
  )

The shadow-gate badge — advisory; the green case means the human approval is the only remaining step.

src/froot/dashboard/render.py · 763 lines

src/froot/dashboard/render.py763 lines · Python

⋯ 377 lines hidden (lines 1–377)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,
      ClassGate,
      DashboardModel,
      ReviewLoop,
      ReviewRow,
      RunTelemetry,
      ScanLoop,
  )
26 
27_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
29--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#fafafa}
30@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
31--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
32--accent:#58a6ff;--card:#141414}}
33*{box-sizing:border-box}
34body{margin:0;background:var(--bg);color:var(--fg);
35font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
36main{max-width:820px;margin:0 auto;padding:34px 20px 72px}
37h1{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}
41section{margin:30px 0 0}
42h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
43color:var(--mut);margin:0 0 12px;font-weight:600;
44border-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%;
49margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
52table{width:100%;border-collapse:collapse;font-size:13px}
53th,td{text-align:left;padding:6px 12px 6px 0;
54border-bottom:1px solid var(--line);vertical-align:top}
55th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
56letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
58a{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}
63footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
64color:var(--mut);font-size:12px;line-height:1.7}
65footer b{color:var(--fg);font-weight:600}
66/* loop groups — collapsible (collapsed by default), two-level hierarchy */
67details.loop{margin:24px 0 0}
68summary.loophead{display:flex;align-items:baseline;gap:6px 12px;flex-wrap:wrap;
69cursor:pointer;list-style:none;margin:0;padding:0 0 7px;
70border-bottom:2px solid var(--fg)}
71summary.loophead::-webkit-details-marker{display:none}
72summary.loophead::before{content:"\\25B8";color:var(--mut);margin-right:5px;
73font-size:11px;position:relative;top:-1px}
74details.loop[open]>summary.loophead::before{content:"\\25BE"}
75summary.loophead h2{font-size:16px;text-transform:none;letter-spacing:-.01em;
76color:var(--fg);border:0;margin:0;padding:0;font-weight:700}
77summary.loophead .sub{color:var(--mut);font-size:12px}
78summary.loophead .glance{margin-left:auto;display:flex;flex-wrap:wrap;
79align-items:baseline;gap:3px 14px;font-size:12px;color:var(--mut)}
80summary.loophead .glance b{color:var(--fg);font-weight:600}
81/* the shared-infra group reads as not-a-loop: lighter, dashed, muted */
82details.loop.shared>summary.loophead{border-bottom:1px dashed var(--line)}
83details.loop.shared>summary.loophead h2{font-size:13px;color:var(--mut);
84font-weight:600;text-transform:uppercase;letter-spacing:.06em}
85/* subsections sit tighter inside a loop; the loop rule carries the weight */
86details.loop section{margin:18px 0 0}
87details.loop section>h2{border-bottom:0;padding-bottom:0;margin-bottom:9px}
88/* foldable framing — the MHE notes, one click away, so the page scans clean */
89details.why{margin:9px 0 0}
90details.why>summary{color:var(--accent);font-size:12px;cursor:pointer;
91list-style:none}
92details.why>summary::-webkit-details-marker{display:none}
93details.why>summary::before{content:"\\002b  why";font-weight:600}
94details.why[open]>summary::before{content:"\\2212  why"}
95details.why .note{margin:6px 0 0}
96"""
97 
98_CI_CLASS = {
  "passed": "ok",
  "failed": "bad",
  "absent": "mut",
  "timed_out": "warn",
103}
104_VERDICT_CLASS = {"clean": "ok", "risky": "warn", "unknown": "mut"}
105 
106 
107def _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)
110 
111 
112def _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"
124 
125 
126def _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"
138 
139 
140def _dot(kind: str) -> str:
  """A status dot span (``ok`` / ``warn`` / ``bad`` / ``mute``)."""
  return f'<span class="dot {kind}"></span>'
143 
144 
145def _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>'
151 
152 
153def _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>'
  )
158 
159 
160def _why(note_html: str) -> str:
  """Fold a framing note behind a native ``<details>`` so the page scans."""
  return f'<details class="why"><summary></summary>{note_html}</details>'
163 
164 
165def _glance(n: object, label: str) -> str:
  """One at-a-glance item for a loop header (``<b>12</b> proposed``)."""
  return f"<span><b>{escape(str(n))}</b> {escape(label)}</span>"
168 
169 
170def _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>"
  )
190 
191 
192def _heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.scan_interval_seconds
195 
  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>'
          f' <span class="mut">{escape(loop.loop)}</span>{tail}</div>'
      )
214 
  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 it alive?</h2>{body}</section>"
220 
221 
222def _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>{_why(note)}</section>'
  )
250 
251 
252def _rate_cell(rate: float | None) -> str:
  """A class's approval rate as a percent, or an em-dash if none decided."""
  if rate is None:
      return '<span class="mut">—</span>'
  return f"{rate * 100:.0f}%"
257 
258 
259def _earned_cell(g: ClassGate) -> str:
  """A class's gate standing: a green 'earned', else the muted blocker."""
  if g.earned:
      return f'{_dot("ok")}<span class="ok">earned</span>'
  blocker = escape(g.blocker or "not earned")
  return f'{_dot("mut")}<span class="mut">{blocker}</span>'
265 
266 
267def _budget_cell(g: ClassGate) -> str:
  """The gate in steward-time: approvals/week now &rarr; reclaimable/week."""
  return (
      f'<span class="mono">{g.approvals_per_week:.1f}</span> appr'
      f' <span class="mut">&rarr; {g.reclaim_per_week:.1f} reclaimable</span>'
  )
273 
274 
275def _class_gates(model: DashboardModel) -> str:
  """The earned-autonomy panel — the shadow gate's per-class standing.
277 
  Read-only: it renders whether each (repo, loop) class *would* have earned a
  gate move, and the steward-budget that move would reclaim. Nothing acts.
  """
  gates = model.class_gates
  if not gates:
      body = (
          '<p class="note">No classes yet &mdash; a (repo, loop) earns a '
          "gate move from its own track record.</p>"
      )
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(g.repo)}</td>'
          f'<td class="mut">{escape(g.loop)}</td>'
          f"<td>{g.decided} in {g.window_days}d</td>"
          f"<td>{_rate_cell(g.merge_rate)}</td>"
          f"<td>{_earned_cell(g)}</td>"
          f"<td>{_budget_cell(g)}</td>"
          "</tr>"
          for g in gates
      )
      body = (
          "<table><thead><tr><th>repo</th><th>loop</th><th>decided</th>"
          "<th>rate</th><th>gate</th><th>budget / week</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  note = (
      '<p class="note">Each <b>class</b> is one loop on one repo &mdash; a '
      "distinct trust unit (a security patch is not a version bump, and they "
      "never share a record). A class earns its gate from a high enough "
      "<b>approval rate</b> over enough recently-<b>decided</b> PRs; the "
      "window keeps trust recent, not lifetime. <b>Budget</b> reads the "
      "gate in steward-time &mdash; approvals/week is what the class costs "
      "a human now, reclaimable is the clean-and-green slice a gate move "
      "would hand back. <b>Advisory only</b>: auto-merge is allowlist-gated "
      "and off, and the approval rate counts a human <i>merge</i>, not a "
      "confirmed-good outcome &mdash; with no revert tracking yet, a high "
      "rate can still mask rubber-stamping (&sect;3.7). Rollback tracking "
      "and sampled review come before anything acts.</p>"
  )
  return (
      "<section><h2>Earned autonomy &middot; the shadow gate</h2>"
      f"{body}{_why(note)}</section>"
  )
322 
323 
324def _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 = _why(
          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>'
  )
349 
350 
351def _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>{_why(note)}</section>'
  )
376 
377 
378def _shadow_badge(row: BumpRow) -> str:
  """Whether this open PR *would* auto-merge under its class's grant.
380 
  Advisory: a green 'would auto-merge' means every condition is met and the
  human approval is the only thing still in the way; otherwise the first
  blocker, muted, so the held reason is the thing to fix.
  """
  if row.would_auto_merge:
      return '<span class="ok">would auto-merge</span>'
  reason = row.held_reason or "held"
  return f'<span class="mut">held &middot; {escape(reason)}</span>'
389 
⋯ 374 lines hidden (lines 390–763)390 
391def _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>{_shadow_badge(row)}</td>"
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.gate
      )
      note = _why(
          '<p class="note">The <b>shadow gate</b> column is advisory &mdash; '
          "froot acts on nothing. A green &lsquo;would auto-merge&rsquo; "
          "means this PR met its class&rsquo;s grant and the human approval "
          "is the only remaining step; it rests on the single CI oracle and "
          "a merge-rate record, which (esp. for a security patch) is not the "
          "same as a confirmed-good or security-reviewed outcome. You still "
          "own every merge.</p>"
      )
      body = (
          "<table><thead><tr><th>package</th><th>waiting</th>"
          "<th>shadow gate</th><th>pr</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>{note}"
      )
  return (
      "<section><h2>Approval gate &middot; what a human owns</h2>"
      f"{body}</section>"
  )
423 
424 
425def _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>"
449 
450 
451def _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>"
  )
469 
470 
471def _telem_glance(model: DashboardModel) -> str:
  """The telemetry group's at-a-glance (or an ``unavailable`` marker)."""
  t = model.telemetry
  if not t.available:
      return '<span class="mut">unavailable</span>'
  return _glance(t.total_spans, f"spans / {t.window_days}d")
477 
478 
479def _telemetry(model: DashboardModel) -> str:
  """The telemetry body — header lives in the shared loop group above it."""
  t: RunTelemetry = model.telemetry
  if not t.available:
      return (
          "<section>"
          '<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 f"<section>{summary}{table}</section>"
512 
513 
514def _review_heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.review_interval_seconds
517 
  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>'
      )
535 
  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 f"<section><h2>Is it alive?</h2>{body}</section>"
544 
545 
546def _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>Track record &middot; the reputation</h2>"
      f'<div class="stats">{stats}</div>{_why(note)}</section>'
  )
568 
569 
570def _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 f"<section><h2>Reviews &middot; the detail</h2>{body}</section>"
592 
593 
594def _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 a single loop on a single repo (a version bump is "
      "not a security patch; they never share a record), conditional on its "
      'environment (judge <span class="mono">gemma4:e4b</span>, '
      "lockfile-only regen), revocable, and time-expiring. Today it records "
      "the track "
      "record and renders a <b>shadow gate</b> &mdash; a dry run of the "
      "auto-merge decision that still acts on nothing.<br>"
      "Everything above is derived on this request from GitHub (outcomes) + "
      "Temporal (runs) + ClickHouse (telemetry). froot keeps no database; "
      "reload to recompute."
      "</footer>"
  )
612 
613 
614def _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>'
618 
619 
620def _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>'
624 
625 
626def _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}'
642 
643 
644def _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>'
655 
656 
657def _short_id(workflow_id: str) -> str:
  """Drop the ``froot-bump-`` prefix for a readable failures row."""
  return workflow_id.removeprefix("froot-bump-")
660 
661 
662def _dep_glance(model: DashboardModel) -> str:
  """The dependency-patch loop's at-a-glance, shown in its header."""
  t = model.track_record
  live = sum(1 for loop in model.scan_loops if loop.live)
  total = len(model.scan_loops)
  rate = "—" if t.merge_rate is None else f"{t.merge_rate * 100:.0f}%"
  items = [
      _glance(f"{live}/{total}", "live"),
      _glance(t.opened, "proposed"),
      _glance(rate, "merged"),
  ]
  if model.class_gates:
      earned = sum(1 for g in model.class_gates if g.earned)
      items.append(_glance(f"{earned}/{len(model.class_gates)}", "earned"))
  if t.open_now:
      items.append(_glance(t.open_now, "awaiting you"))
  return "".join(items)
679 
680 
681def _det_glance(model: DashboardModel) -> str:
  """The determinism-review loop's at-a-glance, shown in its header."""
  r = model.review_record
  return "".join(
      (
          _glance(r.repos_covered, "repos"),
          _glance(r.reviewed, "reviewed"),
          _glance(r.hazards, "hazards"),
      )
  )
691 
692 
693def _loop(
  title: str,
  sub: str,
  glance: str,
  sections: tuple[str, ...],
  *,
  shared: bool = False,
700) -> str:
  """A collapsible group: a prominent header + at-a-glance, collapsed.
702 
  The header (with its at-a-glance) is the ``<summary>``, so a collapsed group
  still shows the headline; expanding reveals the subsections. ``shared`` is a
  non-loop infra group (telemetry) with a lighter, distinct header.
  """
  cls = "loop shared" if shared else "loop"
  head = (
      '<summary class="loophead">'
      f"<h2>{escape(title)}</h2>"
      f'<span class="sub">{sub}</span>'
      f'<span class="glance">{glance}</span>'
      "</summary>"
  )
  return f'<details class="{cls}">{head}{"".join(sections)}</details>'
716 
717 
718def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _loop(
          "Dependency-patch",
          "npm + uv &middot; scan &rarr; bump &rarr; CI &rarr; merge",
          _dep_glance(model),
          (
              _heartbeat(model),
              _track_record(model),
              _class_gates(model),
              _verification(model),
              _judgment(model),
              _gate(model),
              _bumps(model),
              _failures(model),
          ),
      ),
      _loop(
          "Determinism review",
          "the transitive ring &middot; advisory",
          _det_glance(model),
          (
              _review_heartbeat(model),
              _review_record(model),
              _reviews(model),
          ),
      ),
      _loop(
          "Run telemetry",
          "ClickHouse &middot; trace-derived, best-effort",
          _telem_glance(model),
          (_telemetry(model),),
          shared=True,
      ),
      _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>"
  )

Config and wiring — settings.py + server.py

AutonomySettings (FROOT_AUTOMERGE_*) tunes the thresholds and the allowlist, and builds the pure AutonomyPolicy the read-model consumes. The allowlist is a comma-separated list, empty by default — the revocable switch, off until a steward flips it. server.py reads it per request and degrades to conservative defaults so a bad value can never blank the page.

Settings at the edge; the pure policy is built in .policy().

src/froot/config/settings.py · 332 lines

src/froot/config/settings.py332 lines · Python

⋯ 286 lines hidden (lines 1–286)1"""All deployment config, as pydantic-settings models (env or ``.env``), frozen.
2 
3* :class:`Settings` (``FROOT_*``) — loop config: repositories, scan interval,
and which maintenance loops (``FROOT_LOOPS``) to run on them.
5* :class:`TemporalSettings` (``TEMPORAL_*``) — connection: host / namespace /
task queue, shared by the worker, the scan starter, and the activity client.
7* :class:`GitHubSettings` — the API token (``FROOT_GITHUB_TOKEN``) as a
:class:`~pydantic.SecretStr`, so it is masked in ``repr``, logs, and
tracebacks and cannot leak accidentally.
10* :class:`ModelSettings` (``FROOT_OLLAMA_MODEL`` / ``FROOT_OLLAMA_URL``) — the
changelog-judge model endpoint.
12* :class:`TelemetrySettings` (``FROOT_OTEL``) — observability toggle.
13 
14Each consumer builds the small model it needs at its point of use; nothing
15secret lives in the repo. ``repos`` is ``NoDecode`` so ``FROOT_REPOS`` is a
16comma-separated list of ``owner/name`` slugs rather than JSON; a slug may carry
17an optional ``@<ecosystem>`` suffix (e.g. ``acme/pylib@uv``), defaulting to npm.
18"""
19 
20from __future__ import annotations
21 
22from typing import Annotated
23 
24from pydantic import Field, SecretStr, field_validator
25from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
26 
27from froot.domain.ecosystem import Ecosystem
28from froot.domain.loop import Loop
29from froot.domain.repo import RepoRef, TargetRepo
30from froot.policy.autonomy import AutonomyPolicy
31from froot.result import Ok
32 
33_DEFAULT_SCAN_INTERVAL_SECONDS = 86_400
34 
35 
36class Settings(BaseSettings):
  """Non-secret worker config from ``FROOT_*`` (env or ``.env``)."""
38 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_",
      env_file=".env",
      extra="ignore",
      frozen=True,
  )
45 
  repos: Annotated[tuple[TargetRepo, ...], NoDecode] = Field(min_length=1)
  scan_interval_seconds: int = Field(
      default=_DEFAULT_SCAN_INTERVAL_SECONDS, gt=0
  )
  # Which maintenance loops to run on each repo. ``FROOT_LOOPS`` is a
  # comma-separated list of loop names; it defaults to dependency-patch alone,
  # so an existing deployment keeps running exactly the loop it did before.
  loops: Annotated[tuple[Loop, ...], NoDecode] = Field(
      default=(Loop.DEPENDENCY_PATCH,), min_length=1
  )
56 
  @field_validator("loops", mode="before")
  @classmethod
  def _parse_loops(cls, value: object) -> object:
      """Parse ``FROOT_LOOPS`` as a comma-separated loop-name list."""
      if not isinstance(value, str):
          return value
      loops: list[Loop] = []
      for raw in value.split(","):
          entry = raw.strip()
          if not entry:
              continue
          try:
              loops.append(Loop(entry))
          except ValueError:
              raise ValueError(f"unknown loop: {entry!r}") from None
      return tuple(loops)
73 
  @field_validator("repos", mode="before")
  @classmethod
  def _parse_repos(cls, value: object) -> object:
      """Parse ``FROOT_REPOS`` as a comma-separated target list.
78 
      Each entry is an ``owner/name`` slug, optionally suffixed with
      ``@<ecosystem>`` (e.g. ``acme/pylib@uv``); the suffix is omitted for the
      default ``npm``.
      """
      if not isinstance(value, str):
          return value
      targets: list[TargetRepo] = []
      for raw in value.split(","):
          entry = raw.strip()
          if not entry:
              continue
          slug, _, eco = entry.partition("@")
          match RepoRef.parse(slug):
              case Ok(ref):
                  pass
              case _:
                  raise ValueError(f"invalid repo slug: {slug!r}")
          try:
              ecosystem = Ecosystem(eco) if eco else Ecosystem.NPM
          except ValueError:
              raise ValueError(f"unknown ecosystem: {eco!r}") from None
          targets.append(TargetRepo(repo=ref, ecosystem=ecosystem))
      return tuple(targets)
102 
103 
104class TemporalSettings(BaseSettings):
  """Temporal connection config from ``TEMPORAL_*`` (env or ``.env``).
106 
  The same image runs anywhere by configuring these; the in-cluster
  deployment sets them to the cluster's frontend, namespace, and queue.
  """
110 
  model_config = SettingsConfigDict(
      env_prefix="TEMPORAL_",
      env_file=".env",
      extra="ignore",
      frozen=True,
  )
117 
  host: str = Field(default="localhost:7233", min_length=1)
  namespace: str = Field(default="default", min_length=1)
  task_queue: str = Field(default="froot", min_length=1)
121 
122 
123class GitHubSettings(BaseSettings):
  """GitHub credentials, from ``FROOT_GITHUB_TOKEN``.
125 
  The token is a :class:`~pydantic.SecretStr`, so it is masked in ``repr``,
  logs, and tracebacks and cannot leak accidentally; call
  ``github_token.get_secret_value()`` only where the real value is sent.
  """
130 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_", env_file=".env", extra="ignore", frozen=True
  )
134 
  github_token: SecretStr | None = None
136 
137 
138class ModelSettings(BaseSettings):
  """The changelog-judge model endpoint (a local Ollama by default)."""
140 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_", env_file=".env", extra="ignore", frozen=True
  )
144 
  ollama_model: str = Field(default="gemma4:e4b", min_length=1)
  ollama_url: str = Field(default="http://localhost:11434/v1", min_length=1)
147 
148 
149class TelemetrySettings(BaseSettings):
  """OpenTelemetry toggle — off unless ``FROOT_OTEL`` is truthy."""
151 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_", env_file=".env", extra="ignore", frozen=True
  )
155 
  otel: bool = False
157 
  @field_validator("otel", mode="before")
  @classmethod
  def _blank_is_off(cls, value: object) -> object:
      """Treat an empty/whitespace ``FROOT_OTEL`` as off, not an error."""
      if isinstance(value, str) and not value.strip():
          return False
      return value
165 
166 
167class ClickHouseSettings(BaseSettings):
  """ClickHouse (the run ledger) connection for the dashboard read-model.
169 
  Every field is optional: when ``FROOT_CLICKHOUSE_URL`` is unset the
  dashboard renders the run-telemetry panel as *unavailable* rather than
  failing. That panel is best-effort enrichment — GitHub (outcomes) and
  Temporal (live runs) are the dependable sources; ClickHouse only adds
  trace-derived run telemetry, on a 3-day TTL at that. The password is a
  :class:`~pydantic.SecretStr`.
  """
177 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_CLICKHOUSE_",
      env_file=".env",
      extra="ignore",
      frozen=True,
  )
184 
  # Host only, e.g. ``http://clickhouse:8123`` — NEVER embed credentials as
  # userinfo (``http://user:pw@host``): they belong in ``user``/``password``
  # so they stay out of error strings the dashboard may surface.
  url: str | None = None
  user: str = Field(default="default", min_length=1)
  password: SecretStr | None = None
  database: str = Field(default="default", min_length=1)
192 
193 
194class DashboardSettings(BaseSettings):
  """The read-model dashboard's HTTP surface, served by the worker.
196 
  A read-only page the worker serves on ``FROOT_DASHBOARD_PORT``; reach it
  with ``kubectl port-forward``. It derives everything on request and stores
  nothing (froot's own derived-state invariant), so it is safe to leave on.
  """
201 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_DASHBOARD_",
      env_file=".env",
      extra="ignore",
      frozen=True,
  )
208 
  enabled: bool = True
  # Bind all interfaces so an in-cluster ``kubectl port-forward`` reaches it.
  host: str = Field(default="0.0.0.0", min_length=1)
  port: int = Field(default=8080, gt=0, le=65535)
213 
  @field_validator("enabled", mode="before")
  @classmethod
  def _blank_is_on(cls, value: object) -> object:
      """Treat an empty/whitespace value as the default (on), not an error."""
      if isinstance(value, str) and not value.strip():
          return True
      return value
221 
222 
223class ReviewSettings(BaseSettings):
  """The determinism-reviewer loop (``FROOT_REVIEW_*``).
225 
  The transitive ring: a per-repo loop that polls open PRs and leaves an
  advisory comment when a workflow reaches a determinism hazard through a
  first-party helper (``depth`` call levels) or a risky third-party import.
  Advisory only — the blocking gate is the kernel's ``Determinism`` CI check.
  """
231 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_REVIEW_",
      env_file=".env",
      extra="ignore",
      frozen=True,
  )
238 
  enabled: bool = True
  # PRs merge fast, but this loop is advisory (not a gate), so a little
  # latency is fine — it never needs to win the merge race.
  poll_interval_seconds: int = Field(default=300, gt=0)
  # How many first-party call levels to chase out of each workflow method.
  depth: int = Field(default=2, ge=1, le=4)
245 
  @field_validator("enabled", mode="before")
  @classmethod
  def _blank_is_on(cls, value: object) -> object:
      """Treat an empty/whitespace value as the default (on), not an error."""
      if isinstance(value, str) and not value.strip():
          return True
      return value
253 
254 
255class BehaviorSettings(BaseSettings):
  """Loop-hygiene toggles (``FROOT_*``), both defaulting on.
257 
  * ``close_on_red`` — close a bump's PR (and delete its branch) when its CI
    comes back red, so no rotting red proposal is left for the human. The
    outcome is still recorded either way. Read at dispatch and pinned onto the
    bump's params, so an in-flight bump keeps the value it started with.
  * ``reconcile`` — each scan tick, close froot PRs a newer patch has
    superseded or the base has already satisfied. Read by the reconcile
    activity, which no-ops when it is off.
265 
  Both close-then-delete a branch, so they are the destructive knobs; default
  on (the SPEC's behavior), but here to disable for cautious adoption on a
  flaky-CI repo.
  """
270 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_", env_file=".env", extra="ignore", frozen=True
  )
274 
  close_on_red: bool = True
  reconcile: bool = True
277 
  @field_validator("close_on_red", "reconcile", mode="before")
  @classmethod
  def _blank_is_on(cls, value: object) -> object:
      """Treat an empty/whitespace value as the default (on), not an error."""
      if isinstance(value, str) and not value.strip():
          return True
      return value
285 
286 
287class AutonomySettings(BaseSettings):
  """The earned-autonomy thresholds (``FROOT_AUTOMERGE_*``), advisory today.
289 
  These tune the *shadow gate*: the dashboard reads them to decide whether a
  (repo, loop) class has earned its gate move and whether each open PR would
  auto-merge under that grant. Nothing acts on the verdict yet — froot stays
  record-only — so the defaults are deliberately conservative and the
  allowlist is empty, the revocable switch left off until a steward flips it.
295 
  * ``min_rate`` / ``min_decided`` / ``window_days`` — the track-record bar a
    class must clear, measured over a recent window (trust is recent, §2.11).
  * ``allowlist`` (``FROOT_AUTOMERGE_ALLOWLIST``) — a comma-separated list of
    ``owner/name`` slugs a steward has opted into; ``NoDecode`` so it is a
    plain list, not JSON. Empty by default: no class can ride the grant.
  """
302 
  model_config = SettingsConfigDict(
      env_prefix="FROOT_AUTOMERGE_",
      env_file=".env",
      extra="ignore",
      frozen=True,
  )
309 
  min_rate: float = Field(default=0.95, ge=0.0, le=1.0)
  min_decided: int = Field(default=5, ge=1)
  window_days: int = Field(default=90, gt=0)
  allowlist: Annotated[tuple[str, ...], NoDecode] = ()
314 
  @field_validator("allowlist", mode="before")
  @classmethod
  def _parse_allowlist(cls, value: object) -> object:
      """Parse the allowlist as a comma-separated ``owner/name`` list."""
      if not isinstance(value, str):
          return value
      return tuple(
          entry.strip() for entry in value.split(",") if entry.strip()
      )
324 
  def policy(self) -> AutonomyPolicy:
      """Build the pure :class:`AutonomyPolicy` the read-model consumes."""
      return AutonomyPolicy(
          min_rate=self.min_rate,
          min_decided=self.min_decided,
          window_days=self.window_days,
          allowlisted_repos=frozenset(self.allowlist),
      )

The per-request policy, with a safe fallback.

src/froot/dashboard/server.py · 201 lines

src/froot/dashboard/server.py201 lines · Python

⋯ 65 lines hidden (lines 1–65)1"""The dashboard HTTP server — a tiny dependency-free asyncio responder.
2 
3Runs on the worker's own event loop (same process, same pod) and reuses the
4worker's connected Temporal client. On each GET it fans the three readers out
5concurrently, assembles the view, and renders one self-contained page; it stores
6nothing between requests. GET-only, ``Connection: close``, ``Cache-Control:
7no-store`` — built for a single viewer behind ``kubectl port-forward``, not the
8public internet. Every reader degrades to an error string, so a source being
9down yields a page with a red dot, never a crash.
10"""
11 
12from __future__ import annotations
13 
14import asyncio
15import contextlib
16import logging
17from datetime import UTC, datetime
18from typing import TYPE_CHECKING, Final
19 
20from froot.config.settings import (
21    AutonomySettings,
22    DashboardSettings,
23    ReviewSettings,
24    Settings,
25)
26from froot.dashboard import (
27    clickhouse_source,
28    github_source,
29    read_model,
30    render,
31    temporal_source,
32)
33from froot.domain.loop import Loop
34from froot.policy.autonomy import AutonomyPolicy
35 
36if TYPE_CHECKING:
37    from temporalio.client import Client
38 
39_log = logging.getLogger("froot.dashboard")
40 
41_READ_TIMEOUT: Final = 15.0
42_MAX_HEAD_BYTES: Final = 64 * 1024
43# Settings' own default, repeated here so a missing FROOT_REPOS still renders.
44_DEFAULT_INTERVAL: Final = 86_400
45_DEFAULT_REVIEW_INTERVAL: Final = 300
46 
47 
48def _config() -> tuple[tuple[str, ...], tuple[Loop, ...], int]:
49    """The watched repos, active loops, and scan interval (empty if unset)."""
50    try:
51        settings = Settings()
52    except Exception:  # FROOT_REPOS unset/invalid — show an empty heartbeat
53        return (), (Loop.DEPENDENCY_PATCH,), _DEFAULT_INTERVAL
54    repos = tuple(target.repo.slug for target in settings.repos)
55    return repos, settings.loops, settings.scan_interval_seconds
56 
57 
58def _review_interval() -> int:
59    """The determinism-review poll cadence, degrading to the default."""
60    try:
61        return ReviewSettings().poll_interval_seconds
62    except Exception:  # never let a config read fail the page
63        return _DEFAULT_REVIEW_INTERVAL
64 
65 
66def _autonomy_policy() -> AutonomyPolicy:
67    """The earned-autonomy thresholds, degrading to safe defaults.
68 
69    A bad ``FROOT_AUTOMERGE_*`` value must never blank the page: the fallback
70    is the conservative default with an empty allowlist, so the shadow gate
71    simply holds everything rather than erroring.
72    """
73    try:
74        return AutonomySettings().policy()
75    except Exception:  # never let a config read fail the page
76        return AutonomyPolicy()
77 
78 
79async def build_html(client: Client) -> str:
80    """Derive the whole view live and render it (the per-request work)."""
81    now = datetime.now(UTC)
82    repos, loops, interval = _config()
⋯ 119 lines hidden (lines 83–201)83    github_result, temporal_result, telemetry_result = await asyncio.gather(
84        github_source.fetch(repos),
85        temporal_source.fetch(client),
86        clickhouse_source.fetch(),
87    )
88    model = read_model.assemble(
89        now=now,
90        repos=repos,
91        loops=loops,
92        policy=_autonomy_policy(),
93        scan_interval_seconds=interval,
94        review_interval_seconds=_review_interval(),
95        github=github_result,
96        temporal=temporal_result,
97        telemetry=telemetry_result,
98    )
99    return render.page(model)
100 
101 
102def _parse_path(request_line: bytes) -> tuple[str, str]:
103    """Return ``(method, path)`` from a raw HTTP request line."""
104    parts = request_line.decode("latin-1", "replace").split()
105    if len(parts) < 2:
106        return "", "/"
107    return parts[0].upper(), parts[1].split("?", 1)[0]
108 
109 
110async def _drain_headers(reader: asyncio.StreamReader) -> None:
111    """Read and discard the request headers, bounded in size and time."""
112    total = 0
113    while True:
114        line = await asyncio.wait_for(reader.readline(), _READ_TIMEOUT)
115        total += len(line)
116        if line in (b"\r\n", b"\n", b"") or total > _MAX_HEAD_BYTES:
117            return
118 
119 
120async def _respond(
121    writer: asyncio.StreamWriter,
122    status: str,
123    content_type: str,
124    body: bytes,
125) -> None:
126    """Write a complete HTTP/1.1 response and close the connection."""
127    head = (
128        f"HTTP/1.1 {status}\r\n"
129        f"Content-Type: {content_type}\r\n"
130        f"Content-Length: {len(body)}\r\n"
131        "Cache-Control: no-store\r\n"
132        "Connection: close\r\n\r\n"
133    ).encode("latin-1")
134    writer.write(head + body)
135    await writer.drain()
136 
137 
138async def _handle(
139    reader: asyncio.StreamReader,
140    writer: asyncio.StreamWriter,
141    client: Client,
142) -> None:
143    """Serve one request: route a GET to the dashboard, else 404/405/500."""
144    try:
145        request_line = await asyncio.wait_for(reader.readline(), _READ_TIMEOUT)
146        await _drain_headers(reader)
147        method, path = _parse_path(request_line)
148        if method != "GET":
149            await _respond(writer, "405 Method Not Allowed", "text/plain", b"")
150        elif path in ("/", "/index.html", "/dashboard"):
151            html = await build_html(client)
152            await _respond(
153                writer, "200 OK", "text/html; charset=utf-8", html.encode()
154            )
155        elif path == "/healthz":
156            await _respond(writer, "200 OK", "text/plain", b"ok")
157        else:
158            await _respond(writer, "404 Not Found", "text/plain", b"")
159    except (TimeoutError, ConnectionError):
160        pass
161    except Exception as exc:  # never let a request take the worker down
162        _log.exception("dashboard request failed")
163        with contextlib.suppress(Exception):
164            await _respond(
165                writer,
166                "500 Internal Server Error",
167                "text/html; charset=utf-8",
168                _error_html(exc).encode(),
169            )
170    finally:
171        with contextlib.suppress(Exception):
172            writer.close()
173            await writer.wait_closed()
174 
175 
176def _error_html(exc: Exception) -> str:
177    """A minimal error page (the detail is for a human at a port-forward)."""
178    from html import escape
179 
180    return (
181        "<!doctype html><meta charset=utf-8><title>froot &middot; error</title>"
182        '<body style="font:15px system-ui;max-width:640px;margin:40px auto">'
183        "<h1>froot dashboard error</h1><p>Could not build the read-model.</p>"
184        f"<pre>{escape(type(exc).__name__)}: {escape(str(exc))}</pre></body>"
185    )
186 
187 
188async def start(client: Client) -> asyncio.Server:
189    """Start the dashboard server on the configured host/port (returns it)."""
190    settings = DashboardSettings()
191 
192    async def on_connect(
193        reader: asyncio.StreamReader, writer: asyncio.StreamWriter
194    ) -> None:
195        await _handle(reader, writer, client)
196 
197    server = await asyncio.start_server(
198        on_connect, host=settings.host, port=settings.port
199    )
200    _log.info("dashboard listening on %s:%d", settings.host, settings.port)
201    return server

Run-ledger instrumentation — telemetry.py + activities.py

A second, independent beat: make the signal stage legible. scan_candidates now reports its selectivity — how much upstream signal it considered versus how much it kept — so a tick that proposes nothing is still visible in the ledger.

set_span_attributes is a thin helper that writes froot-namespaced attributes onto the activity's span. It lazy-imports OpenTelemetry and no-ops when FROOT_OTEL is off, so nothing enters the Temporal workflow sandbox and local runs pay nothing.

No-op early-return, then a lazy otel import.

src/froot/adapters/telemetry.py · 156 lines

src/froot/adapters/telemetry.py156 lines · Python

⋯ 42 lines hidden (lines 1–42)1"""OpenTelemetry wiring: a tracer provider, SDK metrics, and instrumentation.
2 
3The run-telemetry half of "derive, never store": traces and the Temporal SDK's
4runtime metrics export OTLP/HTTP to the in-cluster collector, which adds the
5ClickStack token on forward. Metrics are
6CUMULATIVE to match what is already in ClickStack.
7 
8All of it is gated on :class:`~froot.config.settings.TelemetrySettings`
9(``FROOT_OTEL``) and is a no-op when off, so tests and local runs stay
10telemetry-free (no exporter threads). It is never imported by a workflow- or
11activity-decorated module, so no OpenTelemetry import enters the Temporal
12workflow sandbox graph.
13"""
14 
15from __future__ import annotations
16 
17from datetime import timedelta
18from typing import TYPE_CHECKING
19 
20from froot.config.settings import TelemetrySettings
21 
22if TYPE_CHECKING:
23    from collections.abc import Sequence
24 
25    import httpx
26    from temporalio.client import Interceptor as ClientInterceptor
27    from temporalio.runtime import Runtime
28 
29# In-cluster collector. Both signals export
30# OTLP/HTTP :4318; the collector adds the ClickStack token on forward.
31_COLLECTOR = "temporal-otel-collector.temporal.svc.cluster.local"
32_TRACES_ENDPOINT = f"http://{_COLLECTOR}:4318/v1/traces"
33_METRICS_ENDPOINT = f"http://{_COLLECTOR}:4318/v1/metrics"
34 
35_tracing_configured = False
36 
37 
38def otel_enabled() -> bool:
39    """True when telemetry is on (the Deployments set ``FROOT_OTEL``)."""
40    return TelemetrySettings().otel
41 
42 
43def set_span_attributes(**attributes: int | float | str | bool) -> None:
44    """Annotate the activity's current span with froot-namespaced attributes.
45 
46    A thin, dependency-light way for an activity to record *what it decided* —
47    e.g. how many upgrades it saw versus how many candidates it kept — onto the
48    span the Temporal tracing interceptor already opened. Keys are prefixed
49    ``froot.`` so they never collide with the interceptor's own attributes.
50 
51    Lazy-imports OpenTelemetry inside the body, so no otel import enters any
52    module's top-level graph (the Temporal workflow sandbox stays clean), and is
53    a no-op when ``FROOT_OTEL`` is off — so tests and local runs add no spans
54    and pay nothing. Call it from an activity body, never from a workflow.
55    """
56    if not otel_enabled():
57        return
58    from opentelemetry import trace
59 
60    span = trace.get_current_span()
61    for key, value in attributes.items():
62        span.set_attribute(f"froot.{key}", value)
⋯ 94 lines hidden (lines 63–156)63 
64 
65def setup_tracing(service_name: str) -> None:
66    """Install a global TracerProvider exporting OTLP/HTTP to the collector.
67 
68    Gated on ``FROOT_OTEL`` and idempotent: the provider (and its one background
69    export thread) is built at most once per process.
70 
71    Args:
72        service_name: The ``service.name`` resource attribute for this process.
73    """
74    global _tracing_configured
75    if _tracing_configured or not otel_enabled():
76        return
77    from opentelemetry import trace
78    from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
79        OTLPSpanExporter,
80    )
81    from opentelemetry.sdk.resources import SERVICE_NAME, Resource
82    from opentelemetry.sdk.trace import TracerProvider
83    from opentelemetry.sdk.trace.export import BatchSpanProcessor
84 
85    provider = TracerProvider(
86        resource=Resource.create({SERVICE_NAME: service_name})
87    )
88    provider.add_span_processor(
89        BatchSpanProcessor(OTLPSpanExporter(endpoint=_TRACES_ENDPOINT))
90    )
91    trace.set_tracer_provider(provider)
92    _tracing_configured = True
93 
94 
95def tracing_interceptors() -> Sequence[ClientInterceptor]:
96    """Temporal interceptors that carry trace context across the boundary.
97 
98    ``always_create_workflow_spans=True`` so the schedule-/loop-started
99    workflows (the scan loop) are traced too. Empty when telemetry is off.
100    """
101    if not otel_enabled():
102        return []
103    from temporalio.contrib.opentelemetry import TracingInterceptor
104 
105    return [TracingInterceptor(always_create_workflow_spans=True)]
106 
107 
108def metrics_runtime() -> Runtime | None:
109    """A Temporal Runtime that pushes SDK metrics OTLP, or ``None`` (off).
110 
111    ``None`` is accepted by ``Client.connect(runtime=...)`` (default runtime).
112    """
113    if not otel_enabled():
114        return None
115    from temporalio.runtime import (
116        OpenTelemetryConfig,
117        OpenTelemetryMetricTemporality,
118        Runtime,
119        TelemetryConfig,
120    )
121 
122    return Runtime(
123        telemetry=TelemetryConfig(
124            metrics=OpenTelemetryConfig(
125                url=_METRICS_ENDPOINT,
126                metric_temporality=OpenTelemetryMetricTemporality.CUMULATIVE,
127                metric_periodicity=timedelta(seconds=30),
128                http=True,
129            )
130        )
131    )
132 
133 
134def instrument_httpx(client: httpx.Client) -> None:
135    """Instrument one httpx client so W3C traceparent rides outbound calls."""
136    if not otel_enabled():
137        return
138    from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
139 
140    HTTPXClientInstrumentor.instrument_client(client)
141 
142 
143def shutdown_tracing() -> None:
144    """Flush + shut down the tracer provider on graceful termination.
145 
146    The BatchSpanProcessor buffers spans (~5s) and Python's ``atexit`` does NOT
147    run on an unhandled SIGTERM, so without this the worker's last span batch is
148    dropped on every (Recreate) rollout.
149    """
150    if not otel_enabled():
151        return
152    from opentelemetry import trace
153 
154    shutdown = getattr(trace.get_tracer_provider(), "shutdown", None)
155    if shutdown is not None:
156        shutdown()

scan_candidates: emit the considered/selected/dropped counts as span attributes and a structured scan_tick log.

src/froot/workflow/activities.py · 473 lines

src/froot/workflow/activities.py473 lines · Python

⋯ 113 lines hidden (lines 1–113)1"""Activities: the effect interpreters that wrap the adapters.
2 
3Each activity is the impure boundary for one effect. Adapter imports are LAZY
4(inside the bodies) so the model and HTTP stacks never enter a workflow sandbox;
5the pure policy and domain imports stay at module level. Activities return
6domain values — the workflow wraps them into events and feeds the pure state
7machine. The activity signatures' types are evaluated at runtime by Temporal's
8data converter, so the domain imports here are deliberately not deferred.
9"""
10 
11from __future__ import annotations
12 
13import json
14import logging
15import tempfile
16from pathlib import Path
17from typing import TYPE_CHECKING, assert_never
18 
19from temporalio import activity
20 
21from froot.domain.candidate import Candidate
22from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
23from froot.domain.ci import CIStatus
24from froot.domain.determinism import AnalysisResult, FrontierVerdict
25from froot.domain.loop import Loop
26from froot.domain.pull_request import PullRequestRef
27from froot.domain.repo import TargetRepo
28from froot.policy.compose import pr_labels, pull_request_draft
29from froot.policy.determinism import analyze_workflow_surface
30from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_review_workflow_id,
34)
35from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
36from froot.workflow.types import (
  AdjudicateInput,
  CiCheckInput,
  CloseInput,
  DispatchInput,
  DispatchReviewInput,
  JudgeInput,
  OpenPrInput,
  PostReviewInput,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
49)
50 
51if TYPE_CHECKING:
  from froot.ports.protocols import PackageManager
53 
54_log = logging.getLogger("froot.outcome")
55_review_log = logging.getLogger("froot.review")
56_reconcile_log = logging.getLogger("froot.reconcile")
57_scan_log = logging.getLogger("froot.scan")
58 
59 
60def _manifest_dir(target: TargetRepo, workspace: Path) -> Path:
  """The directory the manifest lives in (a monorepo subdir, or the root)."""
  return workspace / target.manifest_dir if target.manifest_dir else workspace
63 
64 
65async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
70) -> tuple[int, tuple[Candidate, ...]]:
  """Gather this loop's signal from the checkout and select its candidates.
72 
  The one genuinely per-loop seam: dependency-patch reads the available
  upgrades and picks the highest patch; security-patch reads the installed,
  asks OSV for advisories, and picks the lowest version clearing each. The
  impure sources are lazy-imported per arm so neither drags the other's stack
  into a sandbox. Both feed a pure selection policy.
78 
  Returns ``(considered, candidates)`` — ``considered`` is the size of the
  upstream signal (available upgrades / advisories found) so the scan can make
  its selectivity legible (how much was seen versus how much was kept).
  """
  match loop:
      case Loop.DEPENDENCY_PATCH:
          from froot.policy.candidates import select_patch_candidates
86 
          upgrades = await package_manager.list_upgrades(target, manifest_dir)
          return len(upgrades), select_patch_candidates(upgrades)
      case Loop.SECURITY_PATCH:
          return await _select_security_candidates(
              target, package_manager, manifest_dir
          )
  assert_never(loop)
94 
95 
96async def _select_security_candidates(
  target: TargetRepo, package_manager: PackageManager, manifest_dir: Path
98) -> tuple[int, tuple[Candidate, ...]]:
  """Security signal: installed set, OSV advisories, clearing targets.
100 
  ``considered`` is the count of advisories OSV returned for the installed
  set — the vulnerabilities in scope this tick, before selection narrows to
  the ones a forward-stable bump can actually clear.
  """
  from froot.adapters.osv import OsvAdvisorySource
  from froot.policy.candidates import select_security_candidates
107 
  installed = await package_manager.list_installed(target, manifest_dir)
  advisories = await OsvAdvisorySource().advisories(installed)
  return len(advisories), select_security_candidates(installed, advisories)
111 
112 
113@activity.defn
114async def scan_candidates(
  params: ScanCandidatesInput,
116) -> tuple[Candidate, ...]:
  """Check out the repo and select this loop's candidates.
118 
  Emits the tick's selectivity — how much upstream signal was considered
  versus how much was kept — as span attributes (when ``FROOT_OTEL`` is on)
  and as a structured ``scan_tick`` log, so the signal stage is legible in the
  run ledger even on a tick that proposes nothing.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
  from froot.adapters.telemetry import set_span_attributes
127 
  forge = GitHubForge()
  package_manager = package_manager_for(params.target.ecosystem)
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      considered, candidates = await _select_candidates(
          params.loop,
          params.target,
          package_manager,
          _manifest_dir(params.target, workspace),
      )
  selected = len(candidates)
  dropped = max(considered - selected, 0)
  set_span_attributes(
      scan_loop=params.loop.value,
      scan_repo=params.target.repo.slug,
      scan_considered=considered,
      scan_selected=selected,
      scan_dropped=dropped,
  )
  _scan_log.info(
      json.dumps(
          {
              "event": "scan_tick",
              "loop": params.loop.value,
              "repo": params.target.repo.slug,
              "considered": considered,
              "selected": selected,
              "dropped": dropped,
          }
      )
  )
  return candidates
⋯ 313 lines hidden (lines 161–473)161 
162 
163@activity.defn
164async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
166 
  The model is froot's one thin, non-load-bearing judgment: a clean verdict
  never *gates* a PR (CI is the oracle), so a model that is down, slow, or
  erroring must not stall the spine. A judge failure degrades to
  ``UnknownVerdict`` — the bump proceeds, the human still gets the PR, and the
  dashboard records the verdict as unknown — rather than failing (and then
  retrying) the activity. Only the model call is guarded; the fetch is already
  best-effort (returns ``None``, not an exception). The loop selects what the
  model is asked (clean-patch vs breaking-change-on-a-security-bump).
  """
  from froot.adapters.changelog_http import HttpChangelogSource
  from froot.adapters.model_judge import PydanticAiJudge
178 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      return UnknownVerdict(rationale="No changelog could be fetched.")
  try:
      return await PydanticAiJudge().judge(changelog, params.loop)
  except Exception as exc:
      activity.logger.warning(
          "changelog judge unavailable for %s; degrading to unknown: %r",
          params.candidate.package,
          exc,
      )
      return UnknownVerdict(
          rationale=f"Changelog judge unavailable ({type(exc).__name__})."
      )
193 
194 
195@activity.defn
196async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
200 
  forge = GitHubForge()
  package_manager = package_manager_for(params.target.ecosystem)
  branch = branch_name(params.candidate, params.loop)
  existing = await forge.find_open_pull_request(params.target, branch)
  if existing is not None:
      return existing
  draft = pull_request_draft(
      params.target, params.candidate, params.verdict, params.loop
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      await package_manager.apply_patch_bump(
          params.candidate, _manifest_dir(params.target, workspace)
      )
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
218 
219 
220@activity.defn
221async def check_ci(params: CiCheckInput) -> CIStatus:
  """Read the repo's CI status for the PR's head commit (the oracle)."""
  from froot.adapters.github import GitHubForge
224 
  return await GitHubForge().ci_status(params.target, params.head_sha)
226 
227 
228@activity.defn
229async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update."""
  from froot.adapters.github import GitHubForge
232 
  outcome = params.outcome
  await GitHubForge().add_labels(
      params.target, outcome.pr.number, pr_labels(params.loop)
  )
  _log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": params.loop.value,
              "repo": params.target.repo.slug,
              "package": outcome.candidate.package,
              "from": str(outcome.candidate.current),
              "to": str(outcome.candidate.target),
              "changelog": outcome.verdict.kind,
              "ci": outcome.ci.kind,
              "ci_passed": outcome.ci_passed,
              "pr": outcome.pr.number,
              "pr_url": outcome.pr.url,
          }
      )
  )
254 
255 
256@activity.defn
257async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
259 
  Reads the close-on-red toggle here, at the impure boundary, and pins it onto
  the bump's params — so the running workflow never reads config itself and an
  in-flight bump keeps the value it was dispatched with.
  """
  from temporalio.common import WorkflowIDReusePolicy
  from temporalio.exceptions import WorkflowAlreadyStartedError
266 
  from froot.config.settings import BehaviorSettings
  from froot.workflow.bump_workflow import BumpWorkflow
  from froot.workflow.temporal_client import client, task_queue
  from froot.workflow.types import BumpParams
271 
  temporal = await client()
  try:
      await temporal.start_workflow(
          BumpWorkflow.run,
          BumpParams(
              target=params.target,
              candidate=params.candidate,
              close_on_red=BehaviorSettings().close_on_red,
              loop=params.loop,
          ),
          id=bump_workflow_id(params.target, params.candidate, params.loop),
          task_queue=task_queue(),
          id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
      )
  except WorkflowAlreadyStartedError:
      # This bump already has a loop (running or completed) — a no-op, so
      # re-scanning never opens a second PR for the same bump.
      return
290 
291 
292@activity.defn
293async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
295 
  The note goes through the idempotent ``upsert_issue_comment`` and the close
  itself is idempotent, so a retried close edits its comment in place and
  never double-posts. The bump's record step still runs after this, so the red
  outcome is logged either way.
  """
  from froot.adapters.github import GitHubForge
  from froot.policy.compose import CLOSE_MARKER, closed_on_red_comment
303 
  forge = GitHubForge()
  body = closed_on_red_comment(params.failing)
  await forge.upsert_issue_comment(
      params.target, params.pr.number, CLOSE_MARKER, body
  )
  await forge.close_pull_request(
      params.target, params.pr.number, params.pr.branch
  )
  _log.info(
      json.dumps(
          {
              "event": "pr_closed",
              "loop": params.loop.value,
              "reason": "ci_red",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
              "failing": list(params.failing),
          }
      )
  )
325 
326 
327@activity.defn
328async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
330 
  Self-contained: lists the repo's open PRs, re-derives this loop's live
  candidates (a fresh checkout + the loop's signal, derive-never-store), asks
  the pure :func:`~froot.policy.reconcile.reconciliations` policy: which of
  loop's PRs to close, and closes each (deleting its branch). Scoped to the
  loop's own branch namespace, so the two loops never reconcile each other's
  PRs. A no-op when reconcile is off (``FROOT_RECONCILE``). Returns the count.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
  from froot.config.settings import BehaviorSettings
  from froot.policy.compose import CLOSE_MARKER
  from froot.policy.reconcile import reconciliations
343 
  if not BehaviorSettings().reconcile:
      return 0
346 
  target, loop = params.target, params.loop
  forge = GitHubForge()
  package_manager = package_manager_for(target.ecosystem)
  open_prs = await forge.list_open_pull_requests(target)
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(target, workspace)
      _considered, candidates = await _select_candidates(
          loop, target, package_manager, _manifest_dir(target, workspace)
      )
  closures = reconciliations(open_prs, candidates, loop)
  for closure in closures:
      await forge.upsert_issue_comment(
          target, closure.pr.number, CLOSE_MARKER, closure.comment
      )
      await forge.close_pull_request(
          target, closure.pr.number, closure.pr.branch
      )
  if closures:
      _reconcile_log.info(
          json.dumps(
              {
                  "event": "reconcile",
                  "loop": loop.value,
                  "repo": target.repo.slug,
                  "closed": len(closures),
                  "prs": [closure.pr.number for closure in closures],
              }
          )
      )
  return len(closures)
378 
379 
380# ── The determinism reviewer (the transitive ring) ──────────────────────────
381@activity.defn
382async def list_review_prs(target: TargetRepo) -> tuple[PullRequestRef, ...]:
  """List the repo's open PRs for the determinism reviewer to consider."""
  from froot.adapters.github import GitHubForge
385 
  return await GitHubForge().list_open_pull_requests(target)
387 
388 
389@activity.defn
390async def dispatch_pr_review(params: DispatchReviewInput) -> None:
  """Start a PR's determinism review (idempotent per PR + head SHA)."""
  from temporalio.common import WorkflowIDReusePolicy
  from temporalio.exceptions import WorkflowAlreadyStartedError
394 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
397 
  temporal = await client()
  try:
      await temporal.start_workflow(
          PrReviewWorkflow.run,
          PrReviewParams(target=params.target, pr=params.pr),
          id=pr_review_workflow_id(
              params.target, params.pr.number, params.pr.head_sha
          ),
          task_queue=task_queue(),
          id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
      )
  except WorkflowAlreadyStartedError:
      # This (PR, head SHA) already has a review — a no-op, so re-polling
      # never double-reviews the same commit.
      return
413 
414 
415@activity.defn
416async def analyze_pr(params: PrReviewParams) -> AnalysisResult:
  """Check out the PR head and analyze the workflow surface for hazards."""
  from froot.adapters.github import GitHubForge
  from froot.adapters.source_tree import load_modules
  from froot.config.settings import ReviewSettings
421 
  forge = GitHubForge()
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout_pull_request(
          params.target, workspace, params.pr.number
      )
      # The ASTs and source lines are read into memory here, so the analysis
      # below is unaffected by the workspace being cleaned up.
      modules = load_modules(workspace)
  return analyze_workflow_surface(modules, max_depth=ReviewSettings().depth)
432 
433 
434@activity.defn
435async def adjudicate_frontier(
  params: AdjudicateInput,
437) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
440 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
446 
447 
448@activity.defn
449async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
452 
  body = render_review_comment(params.findings, params.pr.head_sha)
  url: str | None = None
  if body is not None:
      url = await GitHubForge().upsert_issue_comment(
          params.target, params.pr.number, REVIEW_MARKER, body
      )
  _review_log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": "determinism-review",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "head_sha": params.pr.head_sha,
              "findings": len(params.findings),
              "rules": sorted({f.rule for f in params.findings}),
              "comment_url": url,
          }
      )
  )
  return url

How it was verified

make check is green — ruff format + lint, mypy strict, 290 tests at 77% coverage (floor 72%) — and both determinism gates pass. The feature was put through a three-dimension adversarial review (correctness · sandbox/determinism · MHE-honesty) with findings independently verified and folded in:

- a div-by-zero guard in class_earned for a degenerate min_decided; - an aware-datetime coercion at the GitHub boundary (a naive timestamp would have crashed the read-model's now - when subtraction); - the reclaim-only-when-earned honesty fix (above); - a class-accurate footer that names the shadow gate; - an advisory caveat on the gate badge; - the substantive-blocker-first reason ordering (above).

New tests cover the pure policy, the windowing and reclaim logic, the loop-label parse, the settings, the no-op telemetry helper, and the rendered panel + badge.