An open loop registry the spine consumes

What changed, and why

froot runs maintenance loops over GitHub repos. Until now the spine learned each loop's specifics by match-ing on a Loop enum: the candidate scan, the judge framing, the reconcile skip, the PR-title verb, the dashboard icon were all per-loop arms scattered across the codebase. Adding a loop meant editing all of them.

This change replaces that with an open registry the spine reads. A LoopSpec carries only the genuinely per-loop seams; the durable chassis — the scan/dispatch machinery, the merge gate, earned autonomy, namespacing, the dashboard — stays loop-agnostic. The three acting loops self-register, and activities, model_judge, and the dashboard look the loop up instead of branching on it.

The shape follows three rules worth holding while you read:

- The gate stays a spine stage, unreachable from any model-filled slot — a LoopSpec exposes only observe, never a merge primitive. - Admission is by placement, not by an LLM call — dead-code is a first-class loop with CI as its oracle and no changelog judge. - **disposition is a field**, not a second loop species — the advisory family folds in behind it next.

Cut #1 is the acting family (dependency-patch, security-patch, dead-code) and is zero behavior change: the existing scan/judge/reconcile/dashboard tests now route through the registry and pass.

The registry: one spec per loop

LoopSpec is a frozen dataclass of the per-loop seams and nothing else. The one callable is observe (signal → candidates); the rest is inert data — disposition, the changelog-judge judge_context (None when a loop does no changelog judging), reconciles, the PR-title title_prefix, the dashboard_icon. The workflow-id segment is deliberately not here — it stays a pure derivation in naming (see the next sections).

The disposition enum, and the spec — six declarative fields, one callable.

src/froot/loops/registry.py · 125 lines

src/froot/loops/registry.py125 lines · Python

⋯ 44 lines hidden (lines 1–44)1"""The loop registry: froot's open loop catalog, keyed by :class:`Loop`.
2 
3The chassis schedules, dispatches, verifies, and gates; a :class:`LoopSpec`
4fills only the genuinely per-loop seams — the signal→candidate ``observe``
5function, the changelog-judge framing line (when the loop judges changelogs as
6an in-loop effect), and the workflow-id namespace segment. Everything else (the
7scan/dispatch chassis, the merge gate, earned autonomy, the dashboard) is
8loop-agnostic and stays in the spine.
9 
10A loop's *disposition* declares how its work item terminates — COMMIT_OR_REVERT
11(propose a PR; the spine gates the merge) or EMIT_SIGNAL (upsert a decaying
12advisory; no merge, no gate). The commit-vs-emit fork will read that field when
13advisory loops fold in; this cut registers the three acting (COMMIT_OR_REVERT)
14loops, so nothing branches on it yet.
15 
16Loops self-register at import. The registry is populated lazily on first read
17(see :func:`_ensure`) so importing a consumer never forces a particular import
18order, and the per-loop adapter imports stay inside each ``observe`` body.
19"""
20 
21from __future__ import annotations
22 
23from dataclasses import dataclass
24from enum import StrEnum
25from typing import TYPE_CHECKING
26 
27if TYPE_CHECKING:
28    from collections.abc import Awaitable, Callable
29    from pathlib import Path
30 
31    from froot.domain.loop import Loop
32    from froot.domain.repo import TargetRepo
33    from froot.domain.work import WorkItem
34    from froot.ports.protocols import PackageManager
35 
36    # Check out a repo's manifest and select this loop's work items, returning
37    # (considered, items) — ``considered`` is the upstream signal size, so the
38    # scan tick can show how much was seen versus kept.
39    ObserveFn = Callable[
40        [TargetRepo, PackageManager, Path],
41        Awaitable[tuple[int, tuple[WorkItem, ...]]],
42    ]
43 
44 
45class Disposition(StrEnum):
46    """How a loop's work item terminates (the commit-vs-emit fork)."""
47 
48    # Propose a PR; the spine gates the merge (the acting loops).
49    COMMIT_OR_REVERT = "commit-or-revert"
50    # Upsert a decaying advisory comment/label; no merge, no gate (advisory).
51    EMIT_SIGNAL = "emit-signal"
52 
⋯ 26 lines hidden (lines 53–78)53 
54@dataclass(frozen=True)
55class LoopSpec:
56    """One loop's declarative entry — the per-loop seams, nothing chassis-owned.
57 
58    Attributes:
59        loop: The loop this spec specialises (the registry key).
60        disposition: How its work item terminates (commit vs emit-signal). The
61            commit-vs-emit fork will read this when advisory loops fold in;
62            today every acting loop is ``COMMIT_OR_REVERT``.
63        observe: The signal→candidate seam — the one genuinely per-loop body.
64        title_prefix: The PR-title verb (``deps`` / ``security`` /
65            ``dead-code``) — a per-loop label, not derivable from the loop name.
66        judge_context: The framing line for the in-loop changelog judge, or
67            ``None`` when the loop does no changelog judging (e.g. dead-code,
68            whose judgment is a safe-to-remove veto *at the signal*, inside
69            ``observe``).
70        reconciles: Whether version-supersession reconcile applies — ``True``
71            for bump loops, ``False`` for a loop whose work item carries no
72            version to be overtaken (dead-code removals). The reconcile activity
73            keys on this instead of naming the loop.
74        dashboard_icon: The icon key for this loop's dashboard tab (one of the
75            renderer's ``_ICONS``). Carried here so a new loop's tab is fully
76            presented from its spec — no per-loop arm in the renderer.
77 
78    The workflow-id/branch namespace segment is NOT carried here: it is a pure
79    derivation in :mod:`froot.policy.naming` (empty for ``dependency-patch``,
80    the loop name otherwise) that is already zero-edit for a new loop.
81    """
82 
83    loop: Loop
84    disposition: Disposition
85    observe: ObserveFn
86    title_prefix: str
87    judge_context: str | None = None
88    reconciles: bool = True
89    dashboard_icon: str = "package"
90 
91 
92_LOOPS: dict[Loop, LoopSpec] = {}
93_REGISTERED = False
94 
95 
⋯ 30 lines hidden (lines 96–125)96def register(spec: LoopSpec) -> None:
97    """Register a loop spec (idempotent on re-import; last write wins)."""
98    _LOOPS[spec.loop] = spec
99 
100 
101def _ensure() -> None:
102    """Import the loop modules once so they self-register (lazy, idempotent)."""
103    global _REGISTERED
104    if _REGISTERED:
105        return
106    _REGISTERED = True
107    # Importing each module runs its module-level register(...) call. Kept here,
108    # not at package import, so the registry has no import-order constraints.
109    from froot.loops import (  # noqa: F401
110        dead_code,
111        dependency_patch,
112        security_patch,
113    )
114 
115 
116def get(loop: Loop) -> LoopSpec:
117    """The registered spec for a loop (raises ``KeyError`` if unregistered)."""
118    _ensure()
119    return _LOOPS[loop]
120 
121 
122def all_specs() -> tuple[LoopSpec, ...]:
123    """Every registered loop spec, in registration order."""
124    _ensure()
125    return tuple(_LOOPS.values())

The registry populates lazily: get() calls _ensure(), which imports the loop modules once so they run their register() calls. That keeps the registry free of import-order constraints and lets each loop keep its heavy adapter imports inside observe, so importing the registry pulls nothing.

src/froot/loops/registry.py · 125 lines

src/froot/loops/registry.py125 lines · Python

⋯ 95 lines hidden (lines 1–95)1"""The loop registry: froot's open loop catalog, keyed by :class:`Loop`.
2 
3The chassis schedules, dispatches, verifies, and gates; a :class:`LoopSpec`
4fills only the genuinely per-loop seams — the signal→candidate ``observe``
5function, the changelog-judge framing line (when the loop judges changelogs as
6an in-loop effect), and the workflow-id namespace segment. Everything else (the
7scan/dispatch chassis, the merge gate, earned autonomy, the dashboard) is
8loop-agnostic and stays in the spine.
9 
10A loop's *disposition* declares how its work item terminates — COMMIT_OR_REVERT
11(propose a PR; the spine gates the merge) or EMIT_SIGNAL (upsert a decaying
12advisory; no merge, no gate). The commit-vs-emit fork will read that field when
13advisory loops fold in; this cut registers the three acting (COMMIT_OR_REVERT)
14loops, so nothing branches on it yet.
15 
16Loops self-register at import. The registry is populated lazily on first read
17(see :func:`_ensure`) so importing a consumer never forces a particular import
18order, and the per-loop adapter imports stay inside each ``observe`` body.
19"""
20 
21from __future__ import annotations
22 
23from dataclasses import dataclass
24from enum import StrEnum
25from typing import TYPE_CHECKING
26 
27if TYPE_CHECKING:
28    from collections.abc import Awaitable, Callable
29    from pathlib import Path
30 
31    from froot.domain.loop import Loop
32    from froot.domain.repo import TargetRepo
33    from froot.domain.work import WorkItem
34    from froot.ports.protocols import PackageManager
35 
36    # Check out a repo's manifest and select this loop's work items, returning
37    # (considered, items) — ``considered`` is the upstream signal size, so the
38    # scan tick can show how much was seen versus kept.
39    ObserveFn = Callable[
40        [TargetRepo, PackageManager, Path],
41        Awaitable[tuple[int, tuple[WorkItem, ...]]],
42    ]
43 
44 
45class Disposition(StrEnum):
46    """How a loop's work item terminates (the commit-vs-emit fork)."""
47 
48    # Propose a PR; the spine gates the merge (the acting loops).
49    COMMIT_OR_REVERT = "commit-or-revert"
50    # Upsert a decaying advisory comment/label; no merge, no gate (advisory).
51    EMIT_SIGNAL = "emit-signal"
52 
53 
54@dataclass(frozen=True)
55class LoopSpec:
56    """One loop's declarative entry — the per-loop seams, nothing chassis-owned.
57 
58    Attributes:
59        loop: The loop this spec specialises (the registry key).
60        disposition: How its work item terminates (commit vs emit-signal). The
61            commit-vs-emit fork will read this when advisory loops fold in;
62            today every acting loop is ``COMMIT_OR_REVERT``.
63        observe: The signal→candidate seam — the one genuinely per-loop body.
64        title_prefix: The PR-title verb (``deps`` / ``security`` /
65            ``dead-code``) — a per-loop label, not derivable from the loop name.
66        judge_context: The framing line for the in-loop changelog judge, or
67            ``None`` when the loop does no changelog judging (e.g. dead-code,
68            whose judgment is a safe-to-remove veto *at the signal*, inside
69            ``observe``).
70        reconciles: Whether version-supersession reconcile applies — ``True``
71            for bump loops, ``False`` for a loop whose work item carries no
72            version to be overtaken (dead-code removals). The reconcile activity
73            keys on this instead of naming the loop.
74        dashboard_icon: The icon key for this loop's dashboard tab (one of the
75            renderer's ``_ICONS``). Carried here so a new loop's tab is fully
76            presented from its spec — no per-loop arm in the renderer.
77 
78    The workflow-id/branch namespace segment is NOT carried here: it is a pure
79    derivation in :mod:`froot.policy.naming` (empty for ``dependency-patch``,
80    the loop name otherwise) that is already zero-edit for a new loop.
81    """
82 
83    loop: Loop
84    disposition: Disposition
85    observe: ObserveFn
86    title_prefix: str
87    judge_context: str | None = None
88    reconciles: bool = True
89    dashboard_icon: str = "package"
90 
91 
92_LOOPS: dict[Loop, LoopSpec] = {}
93_REGISTERED = False
94 
95 
96def register(spec: LoopSpec) -> None:
97    """Register a loop spec (idempotent on re-import; last write wins)."""
98    _LOOPS[spec.loop] = spec
99 
100 
101def _ensure() -> None:
102    """Import the loop modules once so they self-register (lazy, idempotent)."""
103    global _REGISTERED
104    if _REGISTERED:
105        return
106    _REGISTERED = True
107    # Importing each module runs its module-level register(...) call. Kept here,
108    # not at package import, so the registry has no import-order constraints.
109    from froot.loops import (  # noqa: F401
110        dead_code,
111        dependency_patch,
112        security_patch,
113    )
114 
115 
116def get(loop: Loop) -> LoopSpec:
117    """The registered spec for a loop (raises ``KeyError`` if unregistered)."""
118    _ensure()
119    return _LOOPS[loop]
120 
121 
122def all_specs() -> tuple[LoopSpec, ...]:
123    """Every registered loop spec, in registration order."""
124    _ensure()
⋯ 1 line hidden (lines 125–125)125    return tuple(_LOOPS.values())

A loop is a small module

Each loop is one file that defines its observe and registers its spec. dependency-patch is the whole pattern in miniature: read the available upgrades, pick the highest patch. The adapter import stays inside observe (lazy), and the spec is the loop's entire identity.

observe + the registered spec — the title verb and a default icon.

src/froot/loops/dependency_patch.py · 49 lines

src/froot/loops/dependency_patch.py49 lines · Python

⋯ 28 lines hidden (lines 1–28)1"""The dependency-patch loop: keep dependencies patched (the first loop).
2 
3Signal: the available upgrades the package manager reports. Candidate: the
4highest patch-level target per package (pure selection). Disposition: commit —
5the spine opens a PR and gates the merge.
6"""
7 
8from __future__ import annotations
9 
10from typing import TYPE_CHECKING
11 
12from froot.domain.loop import Loop
13from froot.loops.registry import Disposition, LoopSpec, register
14 
15if TYPE_CHECKING:
16    from pathlib import Path
17 
18    from froot.domain.repo import TargetRepo
19    from froot.domain.work import WorkItem
20    from froot.ports.protocols import PackageManager
21 
22# The framing line for the in-loop changelog judge.
23_JUDGE_CONTEXT = (
24    "This is a patch-level upgrade; weigh whether the notes hide "
25    "any behavioral change behind a 'patch'."
26)
27 
28 
29async def observe(
30    target: TargetRepo,
31    package_manager: PackageManager,
32    manifest_dir: Path,
33) -> tuple[int, tuple[WorkItem, ...]]:
34    """Read the available upgrades and pick the highest patch per package."""
35    from froot.policy.candidates import select_patch_candidates
36 
37    upgrades = await package_manager.list_upgrades(target, manifest_dir)
38    return len(upgrades), select_patch_candidates(upgrades)
39 
40 
41register(
42    LoopSpec(
43        loop=Loop.DEPENDENCY_PATCH,
44        disposition=Disposition.COMMIT_OR_REVERT,
45        observe=observe,
46        title_prefix="deps",
47        judge_context=_JUDGE_CONTEXT,
48    )
49)

dead-code is the interesting one. Its judgment runs at the signal: the safe-to-remove judge vetoes each flagged removal inside observe, so a tool used without an import (pytest, eslint) is dropped before any workflow starts. A judge error drops that removal (fail-safe). Because it does no changelog judging, its spec carries judge_context=None and reconciles=False (a removal has no version to be overtaken).

The veto-at-signal, and a spec that declares it judges elsewhere and skips reconcile.

src/froot/loops/dead_code.py · 84 lines

src/froot/loops/dead_code.py84 lines · Python

⋯ 28 lines hidden (lines 1–28)1"""The dead-code loop: remove unused dependencies.
2 
3Signal: the unused direct dependencies a static analyzer flags. Judgment: the
4safe-to-remove judge vetoes each *at the signal* (a tool used without an import
5— pytest, eslint — is dropped before a workflow ever starts), so this loop's one
6thin model call lives inside ``observe``, not as an in-loop changelog judge
7(``judge_context`` is therefore ``None``). Candidate: the surviving removals.
8Disposition: commit — the spine opens a PR and gates the merge (CI is oracle).
9"""
10 
11from __future__ import annotations
12 
13from typing import TYPE_CHECKING
14 
15from temporalio import activity
16 
17from froot.domain.loop import Loop
18from froot.loops.registry import Disposition, LoopSpec, register
19 
20if TYPE_CHECKING:
21    from pathlib import Path
22 
23    from froot.domain.removal import Removal
24    from froot.domain.repo import TargetRepo
25    from froot.domain.work import WorkItem
26    from froot.ports.protocols import PackageManager
27 
28 
29async def observe(
30    target: TargetRepo,
31    package_manager: PackageManager,
32    manifest_dir: Path,
33) -> tuple[int, tuple[WorkItem, ...]]:
34    """Unused deps flagged, then vetoed safe-to-remove (the veto is the judge).
35 
36    The static analyzer flags every unused direct dependency; the safe-to-remove
37    judge then vetoes each — only a ``clean`` survives to become a PR. A judge
38    error drops that removal (fail-safe: never propose what was not vetted).
39    ``considered`` is the flagged count; the kept count is the survivors, so the
40    scan tick shows how much the veto filtered.
41    """
42    from froot.adapters.model_judge import PydanticAiJudge
43 
44    flagged = await package_manager.list_unused(target, manifest_dir)
45    if not flagged:
46        return 0, ()
47    judge = PydanticAiJudge()
48    kept: list[Removal] = []
49    for removal in flagged:
50        try:
51            verdict = await judge.judge_removal(removal)
52        except Exception as exc:
53            activity.logger.warning(
54                "safe-to-remove judge unavailable for %s; skipping: %r",
55                removal.package,
56                exc,
57            )
58            continue
59        if verdict.kind != "clean":
60            continue
61        # Carry the judge's reasoning into the work item so the PR body explains
62        # why the removal is safe, beside the detector note.
63        enriched = (
64            f"{removal.justification}; {verdict.rationale}"
65            if removal.justification
66            else verdict.rationale
67        )
68        kept.append(removal.model_copy(update={"justification": enriched}))
69    return len(flagged), tuple(kept)
⋯ 2 lines hidden (lines 70–71)70 
71 
72register(
73    LoopSpec(
74        loop=Loop.DEAD_CODE,
75        disposition=Disposition.COMMIT_OR_REVERT,
76        observe=observe,
77        title_prefix="dead-code",
78        # Dead-code judges at the signal (the veto above), not the changelog.
79        judge_context=None,
80        # A removal carries no version to be overtaken — nothing to reconcile.
81        reconciles=False,
82        dashboard_icon="scissors",
83    )
84)

The spine reads the registry

The per-loop match statements collapse to lookups. Candidate selection — once a three-arm match plus two helpers — is now one line. The changelog-judge framing reads spec.judge_context. The reconcile activity reads spec.reconciles instead of naming dead-code.

_select_candidates: the one genuinely per-loop body now lives in the spec.

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

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

⋯ 118 lines hidden (lines 1–118)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.a11y import A11yAnalysis, A11yVerdict
22from froot.domain.candidate import Candidate
23from froot.domain.changelog import (
  ChangelogVerdict,
  CleanVerdict,
  UnknownVerdict,
27)
28from froot.domain.ci import CIStatus
29from froot.domain.determinism import AnalysisResult, FrontierVerdict
30from froot.domain.pull_request import PullRequestDraft, PullRequestRef
31from froot.domain.removal import Removal
32from froot.domain.repo import TargetRepo
33from froot.domain.work import WorkItem
34from froot.policy.a11y_comment import (
  A11Y_MARKER,
  render_a11y_comment,
  should_post,
38)
39from froot.policy.a11y_scan import scan_sources
40from froot.policy.compose import (
  pr_labels,
  pull_request_draft,
  removal_pull_request_draft,
44)
45from froot.policy.determinism import analyze_workflow_surface
46from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_a11y_review_workflow_id,
  pr_review_workflow_id,
51)
52from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
53from froot.workflow.types import (
  AdjudicateA11yInput,
  AdjudicateInput,
  AutoMergeInput,
  CiCheckInput,
  CloseInput,
  DispatchA11yInput,
  DispatchInput,
  DispatchReviewInput,
  GateReviewInput,
  GateSelfTestInput,
  JudgeInput,
  MergeInput,
  OpenPrInput,
  PostA11yInput,
  PostReviewInput,
  PrA11yReviewParams,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
74)
75 
76if TYPE_CHECKING:
  from froot.domain.loop import Loop
  from froot.ports.protocols import PackageManager
79 
80_log = logging.getLogger("froot.outcome")
81_review_log = logging.getLogger("froot.review")
82_a11y_log = logging.getLogger("froot.a11y")
83_reconcile_log = logging.getLogger("froot.reconcile")
84_scan_log = logging.getLogger("froot.scan")
85_gate_log = logging.getLogger("froot.gate")
86 
87 
88def _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
91 
92 
93def _draft_for(params: OpenPrInput) -> PullRequestDraft:
  """Build the PR draft for the work item — bump vs removal compose.
95 
  The PR-title verb is the loop's, resolved from its registered spec here (the
  impure boundary) and passed into the pure draft builders.
  """
  from froot.loops import registry
100 
  item = params.candidate
  title_prefix = registry.get(params.loop).title_prefix
  match item:
      case Candidate():
          return pull_request_draft(
              params.target,
              item,
              params.verdict,
              params.loop,
              title_prefix=title_prefix,
          )
      case Removal():
          return removal_pull_request_draft(
              params.target, item, params.loop, title_prefix=title_prefix
          )
  assert_never(item)
117 
118 
119async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
124) -> tuple[int, tuple[WorkItem, ...]]:
  """Gather this loop's signal from the checkout and select its work items.
126 
  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;
  dead-code reads the unused dependencies a static analyzer flags and vetoes
  each with the safe-to-remove judge. The impure sources are lazy-imported per
  arm so none drags another's stack into a sandbox. Each feeds a pure (or, for
  dead-code, model-vetoed) selection.
134 
  Returns ``(considered, items)`` — ``considered`` is the size of the upstream
  signal (available upgrades / advisories found / unused deps flagged) so the
  scan can make its selectivity legible (how much was seen versus kept).
138 
  The one genuinely per-loop body now lives in each loop's spec ``observe``
  (see :mod:`froot.loops`); the spine looks the loop up and runs it, so this
  selection seam needs no per-loop arm.
  """
  from froot.loops import registry
144 
  return await registry.get(loop).observe(
      target, package_manager, manifest_dir
  )
⋯ 678 lines hidden (lines 148–825)148 
149 
150@activity.defn
151async def scan_candidates(
  params: ScanCandidatesInput,
153) -> tuple[WorkItem, ...]:
  """Check out the repo and select this loop's candidates.
155 
  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
164 
  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
198 
199 
200@activity.defn
201async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
203 
  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
215 
  candidate = params.candidate
  if isinstance(candidate, Removal):
      # A removal already cleared the safe-to-remove veto at scan; there is no
      # changelog to read, so carry that conclusion straight into the PR
      # framing (the rationale the veto recorded on the work item).
      return CleanVerdict(
          rationale=candidate.justification or "unused; safe to remove"
      )
  changelog = await HttpChangelogSource().fetch(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__})."
      )
238 
239 
240@activity.defn
241async def gate_review(params: GateReviewInput) -> ChangelogVerdict:
  """Independently deep-review a bump at the gate; fail-closed to a hold.
243 
  The fourth trust leg (§3.7): a second, adversarial model pass over the
  changelog, run only when a bump is about to auto-merge. ``clean`` approves
  the merge; ``risky``/``unknown`` hold the PR for the human. Fail-CLOSED — a
  missing changelog or a model error returns a non-clean verdict, so an
  unreviewable bump never merges unattended. This is the opposite disposition
  from :func:`judge_changelog` (which degrades-to-proceed): here the safe
  direction is to hold, and a non-clean verdict already means hold.
  """
  from froot.adapters.changelog_http import HttpChangelogSource
  from froot.adapters.model_judge import PydanticAiJudge
254 
  candidate = params.candidate
  verdict: ChangelogVerdict
  if isinstance(candidate, Removal):
      # The fourth leg for a removal: independently re-judge that it is safe
      # to remove (the same check the scan veto ran). No changelog to read;
      # fail-CLOSED to a hold on a model error, like the bump path.
      try:
          verdict = await PydanticAiJudge().judge_removal(candidate)
      except Exception as exc:
          activity.logger.warning(
              "safe-to-remove gate review unavailable for %s; holding: %r",
              candidate.package,
              exc,
          )
          verdict = UnknownVerdict(
              rationale=(
                  f"Removal reviewer unavailable ({type(exc).__name__})."
              )
          )
  else:
      changelog = await HttpChangelogSource().fetch(candidate)
      if changelog is None:
          verdict = UnknownVerdict(
              rationale="No changelog to review; holding (fail-closed)."
          )
      else:
          try:
              verdict = await PydanticAiJudge().gate_review(
                  changelog, params.loop
              )
          except Exception as exc:
              activity.logger.warning(
                  "gate reviewer unavailable for %s; holding: %r",
                  candidate.package,
                  exc,
              )
              verdict = UnknownVerdict(
                  rationale=(
                      f"Gate reviewer unavailable ({type(exc).__name__})."
                  )
              )
  _gate_log.info(
      json.dumps(
          {
              "event": "gate_review",
              "loop": params.loop.value,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
              "package": params.candidate.package,
              "verdict": verdict.kind,
              "approved": verdict.kind == "clean",
          }
      )
  )
  return verdict
310 
311 
312@activity.defn
313async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Rewrite the manifest+lockfile and open (idempotently) the work item's PR.
315 
  The one place the action differs by work-item kind: a bump regenerates the
  lockfile at the target version, a removal deletes the dependency. Both are
  lockfile-only (no install, no scripts); everything around them — the dedup
  branch, the checkout, the push, the open — is the same chassis.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
323 
  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 = _draft_for(params)
  item = params.candidate
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      manifest_dir = _manifest_dir(params.target, workspace)
      match item:
          case Candidate():
              await package_manager.apply_patch_bump(item, manifest_dir)
          case Removal():
              await package_manager.remove_dependency(item, manifest_dir)
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
343 
344 
345@activity.defn
346async 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
349 
  return await GitHubForge().ci_status(params.target, params.head_sha)
351 
352 
353@activity.defn
354async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update.
356 
  The labels carry the loop *and* the judgment environment (the judge model)
  the PR was opened under, so the gate can count only the track record earned
  under the current environment and reset it when the model changes (§3.7).
  """
  from froot.adapters.github import GitHubForge
  from froot.config.settings import ModelSettings
  from froot.policy.environment import env_label
364 
  outcome = params.outcome
  item = outcome.candidate
  labels = (
      *pr_labels(params.loop),
      env_label(ModelSettings().ollama_model),
  )
  await GitHubForge().add_labels(params.target, outcome.pr.number, labels)
  record = {
      "event": "loop_outcome",
      "loop": params.loop.value,
      "repo": params.target.repo.slug,
      "package": item.package,
      "changelog": outcome.verdict.kind,
      "ci": outcome.ci.kind,
      "ci_passed": outcome.ci_passed,
      "pr": outcome.pr.number,
      "pr_url": outcome.pr.url,
  }
  match item:
      case Candidate():
          record |= {
              "action": "bump",
              "from": str(item.current),
              "to": str(item.target),
          }
      case Removal():
          record |= {"action": "remove", "dev": item.dev}
  _log.info(json.dumps(record))
393 
394 
395@activity.defn
396async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
398 
  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
405 
  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
410 
  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
429 
430 
431@activity.defn
432async def auto_merge_eligible(params: AutoMergeInput) -> bool:
  """Whether this (repo, loop) class has earned the auto-merge grant.
434 
  Short-circuits to ``False`` for any repo a steward has not allowlisted (the
  default), so the common case costs nothing. Otherwise it re-derives the
  class's standing from the live GitHub history — the same triangulated,
  windowed, environment-scoped computation the dashboard's shadow gate shows
  (``read_model.earned_now``) — so the acting gate and the advisory panel can
  never disagree. Best-effort: any read failure degrades to ``False`` (hold),
  never an auto-merge.
  """
  from datetime import UTC, datetime
444 
  from froot.config.settings import AutonomySettings, ModelSettings
  from froot.dashboard import github_source, read_model
  from froot.policy.environment import environment_slug
448 
  policy = AutonomySettings().policy()
  repo = params.target.repo.slug
  if repo not in policy.allowlisted_repos:
      return False
  now = datetime.now(UTC)
  prs, prs_error = await github_source.fetch((repo,))
  if prs_error is not None:
      return False  # can't confirm the record -> hold, never merge blind
  outcomes, _ = await github_source.fetch_outcomes(
      (repo,), prs, now=now, window_days=policy.window_days
  )
  return read_model.earned_now(
      now,
      prs,
      outcomes,
      repo,
      params.loop,
      policy,
      environment_slug(ModelSettings().ollama_model),
  )
469 
470 
471@activity.defn
472async def merge_pull_request(params: MergeInput) -> None:
  """Auto-merge an earned, clean+green bump's PR (the acting gate's write).
474 
  Reached only after the pure machine confirmed clean+green and the class
  earned the grant on an allowlisted repo. Passes the head SHA so GitHub
  refuses the merge if the head moved since the gate decided.
  """
  from froot.adapters.github import GitHubForge
480 
  await GitHubForge().merge_pull_request(
      params.target, params.pr.number, head_sha=params.pr.head_sha
  )
  _log.info(
      json.dumps(
          {
              "event": "pr_merged",
              "loop": params.loop.value,
              "reason": "auto_merge",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
          }
      )
  )
496 
497 
498@activity.defn
499async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
501 
  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
509 
  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),
          }
      )
  )
531 
532 
533@activity.defn
534async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
536 
  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
549 
  if not BehaviorSettings().reconcile:
      return 0
  # Reconcile is version-supersession cleanup, which only bump loops have: a
  # removal carries no version to be overtaken. A loop that does not reconcile
  # (dead-code) is skipped rather than re-running its signal (knip + the veto
  # judge) every tick only to close nothing. (A removal-specific reconcile —
  # close when no longer unused — is future work.) The trait is on the spec.
  from froot.loops import registry
558 
  if not registry.get(params.loop).reconciles:
      return 0
561 
  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)
593 
594 
595@activity.defn
596async def gate_selftest(params: GateSelfTestInput) -> tuple[str, ...]:
  """Run the adversarial gate probe against the live policy; alarm on escape.
598 
  The §2.11 deliberate disturbance for the acting gate: a battery of synthetic
  known-bad class histories a healthy gate must refuse, scored against the
  policy froot is *actually running* (config and all). Any escape — a bad
  class the live gate would grant — is logged at ERROR (the alarm) so it
  surfaces in telemetry the moment config drifts; a clean pass logs a
  heartbeat at INFO. Returns the escaped scenario names. Pure compute,
  repo-independent (the ``target``/``loop`` are only the log's context).
  """
  from froot.config.settings import AutonomySettings
  from froot.policy.gate_probe import gate_escapes
609 
  escaped = gate_escapes(AutonomySettings().policy())
  record = json.dumps(
      {
          "event": "gate_selftest",
          "loop": params.loop.value,
          "repo": params.target.repo.slug,
          "healthy": not escaped,
          "escaped": list(escaped),
      }
  )
  if escaped:
      _gate_log.error(record)  # an alarm: the gate would trust a bad class
  else:
      _gate_log.info(record)
  return escaped
625 
626 
627# ── The determinism reviewer (the transitive ring) ──────────────────────────
628@activity.defn
629async 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
632 
  return await GitHubForge().list_open_pull_requests(target)
634 
635 
636@activity.defn
637async 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
641 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
644 
  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
660 
661 
662@activity.defn
663async 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
668 
  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)
679 
680 
681@activity.defn
682async def adjudicate_frontier(
  params: AdjudicateInput,
684) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
687 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
693 
694 
695@activity.defn
696async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
699 
  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
721 
722 
723# ── The a11y reviewer (the source-level design-system ring) ──────────────────
724@activity.defn
725async def scan_pr_a11y(params: PrA11yReviewParams) -> A11yAnalysis:
  """Check out the PR head and scan its changed templates for a11y risks.
727 
  Scoped to the PR's changed Vue/JSX templates (shift-left: review what came
  in), so the advisory stays bounded and high-signal. The source lines are
  read into memory in the activity, so the pure scan is unaffected by the
  workspace being cleaned up.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.web_source import load_web_sources
735 
  forge = GitHubForge()
  changed = await forge.list_pull_request_files(
      params.target, params.pr.number
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout_pull_request(
          params.target, workspace, params.pr.number
      )
      sources = load_web_sources(workspace, changed)
      candidates = scan_sources(sources)
  return A11yAnalysis(candidates=candidates, scanned_files=len(sources))
748 
749 
750@activity.defn
751async def adjudicate_a11y(
  params: AdjudicateA11yInput,
753) -> tuple[A11yVerdict, ...]:
  """Run the model over each flagged candidate; return aligned verdicts."""
  from froot.adapters.a11y_judge import A11ySourceJudge
756 
  judge = A11ySourceJudge()
  verdicts: list[A11yVerdict] = []
  for candidate in params.candidates:
      verdicts.append(await judge.adjudicate(candidate))
  return tuple(verdicts)
762 
763 
764@activity.defn
765async def post_a11y_review(params: PostA11yInput) -> str | None:
  """Upsert (or clear) the advisory comment; log the ledger row.
767 
  Posts when there are findings, or when a prior comment must be cleared to
  "all clear" (true decay — a PR whose gaps were fixed never keeps a stale
  finding list, the bug the determinism reviewer leaves). A clean PR with
  no prior comment stays silent, so a clean PR is never spammed.
  """
  from froot.adapters.github import GitHubForge
774 
  forge = GitHubForge()
  exists = await forge.find_marked_comment(
      params.target, params.pr.number, A11Y_MARKER
  )
  url: str | None = None
  if should_post(has_findings=bool(params.findings), comment_exists=exists):
      body = render_a11y_comment(params.findings, params.pr.head_sha)
      url = await forge.upsert_issue_comment(
          params.target, params.pr.number, A11Y_MARKER, body
      )
  _a11y_log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": "a11y-review",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "head_sha": params.pr.head_sha,
              "findings": len(params.findings),
              "kinds": sorted({f.kind for f in params.findings}),
              "comment_url": url,
          }
      )
  )
  return url
800 
801 
802@activity.defn
803async def dispatch_pr_a11y_review(params: DispatchA11yInput) -> None:
  """Start a PR's a11y review (idempotent per PR + head SHA)."""
  from temporalio.common import WorkflowIDReusePolicy
  from temporalio.exceptions import WorkflowAlreadyStartedError
807 
  from froot.workflow.pr_a11y_review_workflow import PrA11yReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
810 
  temporal = await client()
  try:
      await temporal.start_workflow(
          PrA11yReviewWorkflow.run,
          PrA11yReviewParams(target=params.target, pr=params.pr),
          id=pr_a11y_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 an a11y review — a no-op, so
      # re-polling never double-reviews the same commit.
      return

The changelog framing line, sourced from the loop's spec.

src/froot/adapters/model_judge.py · 218 lines

src/froot/adapters/model_judge.py218 lines · Python

⋯ 93 lines hidden (lines 1–93)1"""The thin model judgment: how risky is this dependency bump's changelog?
2 
3froot's one model call. A Pydantic AI agent reads the changelog and returns a
4typed :class:`_Assessment`, which :func:`assessment_to_verdict` maps to the
5domain :data:`~froot.domain.changelog.ChangelogVerdict`. The verdict is framing,
6not a gate — the spine proposes the bump regardless — so even a "risky" reading
7just shapes the PR. The *loop* shapes what the model is asked: a patch bump asks
8"is this clean?", a security bump (often a minor/major) asks "what breaks?". The
9model is injected, so tests run it offline with a ``TestModel`` /
10``FunctionModel``; the mapping is a pure function, tested apart from any model.
11"""
12 
13from __future__ import annotations
14 
15from typing import TYPE_CHECKING, Literal, assert_never
16 
17from pydantic import BaseModel, Field
18from pydantic_ai import Agent
19 
20from froot.adapters.model import build_model
21from froot.domain.changelog import (
22    CleanVerdict,
23    RiskyVerdict,
24    UnknownVerdict,
25)
26from froot.domain.loop import Loop
27 
28if TYPE_CHECKING:
29    from pydantic_ai.models import Model
30 
31    from froot.domain.changelog import Changelog, ChangelogVerdict
32    from froot.domain.removal import Removal
33 
34_SYSTEM_PROMPT = (
35    "You assess the changelog of a dependency upgrade for a code-maintenance "
36    "bot. The bot proposes the bump either way; your job is only to frame the "
37    "risk for the human reviewer.\n"
38    "Return one verdict:\n"
39    "- clean: the notes describe only fixes / docs / internal changes with no "
40    "behavioral or API impact.\n"
41    "- risky: the notes hint at behavior change, deprecations, or anything a "
42    "reviewer should look at closely; list the concerns.\n"
43    "- unknown: the notes are empty or uninformative.\n"
44    "Quote-or-omit: base 'risky' concerns on what the text actually says; do "
45    "not speculate. Keep the rationale to one or two sentences."
46)
47 
48# The gate reviewer's stance is adversarial and asymmetric: this bump is about
49# to merge with NO human in the loop, so the burden is on the changelog to prove
50# it is safe, not on the reviewer to prove it is dangerous. The same _Assessment
51# shape, read as approve/hold (clean = approve).
52_GATE_SYSTEM_PROMPT = (
53    "You are the LAST line of defense before a dependency upgrade auto-merges "
54    "with no human review. Re-read the changelog adversarially and decide "
55    "whether it is safe to merge unattended.\n"
56    "Return one verdict:\n"
57    "- clean: ONLY if the notes clearly describe nothing beyond fixes / docs / "
58    "internal changes — no behavioral change, no API change, no deprecation, "
59    "no ambiguity. This approves the merge.\n"
60    "- risky: any hint of behavior change, removal, deprecation, or surface a "
61    "reviewer should see; list the concerns. This holds the PR.\n"
62    "- unknown: the notes are empty, uninformative, or you are unsure. This "
63    "holds the PR.\n"
64    "The burden is on the changelog to prove safety: when in doubt, do NOT "
65    "return clean. Base concerns on what the text says; keep it to one or two "
66    "sentences."
67)
68 
69 
70# The dead-code loop's thin judgment: a static analyzer flagged a dependency as
71# never imported, but "not imported" is not "unused" — CLI/test/build tools
72# (pytest, eslint, tsc), framework plugins, type-only/peer deps, and dynamically
73# loaded packages are all used without an import. This judge vetoes those so the
74# loop never opens a noisy "remove pytest" PR; CI is still the final oracle, but
75# a wrong removal wastes a human's time, so the burden is on "safe".
76_REMOVAL_SYSTEM_PROMPT = (
77    "You decide whether an UNUSED dependency is safe to remove from a project, "
78    "for a code-maintenance bot. A static analyzer flagged it as never "
79    "imported in the source. But 'not imported' does not always mean "
80    "'unused'.\n"
81    "Return one verdict:\n"
82    "- clean: a normal library that, if genuinely never imported, the build no "
83    "longer needs — safe to remove. This proposes the removal.\n"
84    "- risky: plausibly used WITHOUT an import — a CLI/test/build tool run as "
85    "a command (e.g. pytest, eslint, prettier, tsc, ruff), a framework or "
86    "bundler plugin, a type-only or peer dependency, or something loaded "
87    "dynamically or via config. Say which. This holds it back (no PR).\n"
88    "- unknown: you cannot tell. This holds it back.\n"
89    "CI is the final check, but a wrong removal wastes a human's time, so when "
90    "in doubt do NOT return clean. Keep the rationale to one or two sentences."
91)
92 
93 
94def _loop_context(loop: Loop) -> str:
95    """The framing line telling the model what kind of bump it is judging.
96 
97    Sourced from the loop's registered spec (see :mod:`froot.loops`). Only
98    changelog-judging loops reach this; a loop without a changelog judge
99    (dead-code, whose judgment is a signal-time safe-to-remove veto) has a
100    ``judge_context`` of ``None`` and never gets here — the empty fallback just
101    keeps the call total.
102    """
103    from froot.loops import registry
104 
105    return registry.get(loop).judge_context or ""
106 
⋯ 112 lines hidden (lines 107–218)107 
108class _Assessment(BaseModel):
109    """The model's structured output, mapped to a domain verdict."""
110 
111    verdict: Literal["clean", "risky", "unknown"]
112    rationale: str
113    concerns: list[str] = Field(default_factory=list)
114 
115 
116def assessment_to_verdict(assessment: _Assessment) -> ChangelogVerdict:
117    """Map the model's structured assessment to a domain verdict (pure)."""
118    match assessment.verdict:
119        case "clean":
120            return CleanVerdict(rationale=assessment.rationale)
121        case "risky":
122            return RiskyVerdict(
123                rationale=assessment.rationale,
124                concerns=tuple(assessment.concerns),
125            )
126        case "unknown":
127            return UnknownVerdict(rationale=assessment.rationale)
128    assert_never(assessment.verdict)
129 
130 
131def _gate_model() -> Model:
132    """Build the gate reviewer's model: the override, else the judge model."""
133    from froot.config.settings import ModelSettings
134 
135    return build_model(model_name=ModelSettings().gate_review_model or None)
136 
137 
138def _changelog_prompt(changelog: Changelog, loop: Loop) -> str:
139    """The shared user prompt: the bump's context and its changelog text."""
140    return (
141        f"{_loop_context(loop)}\n"
142        f"Package: {changelog.package}\n"
143        f"Target version: {changelog.version}\n"
144        f"Changelog:\n{changelog.text}"
145    )
146 
147 
148def _removal_prompt(removal: Removal) -> str:
149    """The user prompt for the safe-to-remove judge: the flagged dependency."""
150    declared = "a dev dependency" if removal.dev else "a runtime dependency"
151    return (
152        f"Package: {removal.package}\n"
153        f"Ecosystem: {removal.ecosystem.value}\n"
154        f"Declared as: {declared}\n"
155        f"Detector note: {removal.justification or 'flagged as unused'}\n"
156        "Is this safe to remove?"
157    )
158 
159 
160class PydanticAiJudge:
161    """A :class:`~froot.ports.protocols.ModelJudge` backed by Pydantic AI.
162 
163    Two agents share the structured ``_Assessment`` output but differ in stance:
164    the neutral framing judge (:meth:`judge`) and the adversarial gate reviewer
165    (:meth:`gate_review`), each its own model so a steward can make the deep
166    review independent in capability, not just prompt.
167    """
168 
169    def __init__(
170        self, model: Model | None = None, gate_model: Model | None = None
171    ) -> None:
172        """Build both agents.
173 
174        ``model`` defaults to the configured Ollama; ``gate_model`` to the
175        gate-review override, else ``model``, else the gate-review model.
176        """
177        self._agent: Agent[None, _Assessment] = Agent(
178            model or build_model(),
179            output_type=_Assessment,
180            system_prompt=_SYSTEM_PROMPT,
181        )
182        self._gate_agent: Agent[None, _Assessment] = Agent(
183            gate_model or model or _gate_model(),
184            output_type=_Assessment,
185            system_prompt=_GATE_SYSTEM_PROMPT,
186        )
187        # The dead-code loop's safe-to-remove judge; the neutral model, its own
188        # prompt. It is a veto (clean = propose), so it carries real weight —
189        # but CI remains the oracle.
190        self._removal_agent: Agent[None, _Assessment] = Agent(
191            model or build_model(),
192            output_type=_Assessment,
193            system_prompt=_REMOVAL_SYSTEM_PROMPT,
194        )
195 
196    async def judge(
197        self, changelog: Changelog, loop: Loop = Loop.DEPENDENCY_PATCH
198    ) -> ChangelogVerdict:
199        """Assess a changelog into a verdict, framed by the loop."""
200        result = await self._agent.run(_changelog_prompt(changelog, loop))
201        return assessment_to_verdict(result.output)
202 
203    async def gate_review(
204        self, changelog: Changelog, loop: Loop = Loop.DEPENDENCY_PATCH
205    ) -> ChangelogVerdict:
206        """Adversarially re-review a bump at the gate (clean = approve)."""
207        result = await self._gate_agent.run(_changelog_prompt(changelog, loop))
208        return assessment_to_verdict(result.output)
209 
210    async def judge_removal(self, removal: Removal) -> ChangelogVerdict:
211        """Assess whether an unused dependency is safe to remove (clean = yes).
212 
213        The dead-code loop's veto: ``clean`` lets the removal become a PR; any
214        other verdict holds it back. Same ``_Assessment`` shape as the changelog
215        judge, a different prompt.
216        """
217        result = await self._removal_agent.run(_removal_prompt(removal))
218        return assessment_to_verdict(result.output)

Reconcile keys on the trait, not the loop name.

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

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

⋯ 552 lines hidden (lines 1–552)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.a11y import A11yAnalysis, A11yVerdict
22from froot.domain.candidate import Candidate
23from froot.domain.changelog import (
  ChangelogVerdict,
  CleanVerdict,
  UnknownVerdict,
27)
28from froot.domain.ci import CIStatus
29from froot.domain.determinism import AnalysisResult, FrontierVerdict
30from froot.domain.pull_request import PullRequestDraft, PullRequestRef
31from froot.domain.removal import Removal
32from froot.domain.repo import TargetRepo
33from froot.domain.work import WorkItem
34from froot.policy.a11y_comment import (
  A11Y_MARKER,
  render_a11y_comment,
  should_post,
38)
39from froot.policy.a11y_scan import scan_sources
40from froot.policy.compose import (
  pr_labels,
  pull_request_draft,
  removal_pull_request_draft,
44)
45from froot.policy.determinism import analyze_workflow_surface
46from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_a11y_review_workflow_id,
  pr_review_workflow_id,
51)
52from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
53from froot.workflow.types import (
  AdjudicateA11yInput,
  AdjudicateInput,
  AutoMergeInput,
  CiCheckInput,
  CloseInput,
  DispatchA11yInput,
  DispatchInput,
  DispatchReviewInput,
  GateReviewInput,
  GateSelfTestInput,
  JudgeInput,
  MergeInput,
  OpenPrInput,
  PostA11yInput,
  PostReviewInput,
  PrA11yReviewParams,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
74)
75 
76if TYPE_CHECKING:
  from froot.domain.loop import Loop
  from froot.ports.protocols import PackageManager
79 
80_log = logging.getLogger("froot.outcome")
81_review_log = logging.getLogger("froot.review")
82_a11y_log = logging.getLogger("froot.a11y")
83_reconcile_log = logging.getLogger("froot.reconcile")
84_scan_log = logging.getLogger("froot.scan")
85_gate_log = logging.getLogger("froot.gate")
86 
87 
88def _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
91 
92 
93def _draft_for(params: OpenPrInput) -> PullRequestDraft:
  """Build the PR draft for the work item — bump vs removal compose.
95 
  The PR-title verb is the loop's, resolved from its registered spec here (the
  impure boundary) and passed into the pure draft builders.
  """
  from froot.loops import registry
100 
  item = params.candidate
  title_prefix = registry.get(params.loop).title_prefix
  match item:
      case Candidate():
          return pull_request_draft(
              params.target,
              item,
              params.verdict,
              params.loop,
              title_prefix=title_prefix,
          )
      case Removal():
          return removal_pull_request_draft(
              params.target, item, params.loop, title_prefix=title_prefix
          )
  assert_never(item)
117 
118 
119async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
124) -> tuple[int, tuple[WorkItem, ...]]:
  """Gather this loop's signal from the checkout and select its work items.
126 
  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;
  dead-code reads the unused dependencies a static analyzer flags and vetoes
  each with the safe-to-remove judge. The impure sources are lazy-imported per
  arm so none drags another's stack into a sandbox. Each feeds a pure (or, for
  dead-code, model-vetoed) selection.
134 
  Returns ``(considered, items)`` — ``considered`` is the size of the upstream
  signal (available upgrades / advisories found / unused deps flagged) so the
  scan can make its selectivity legible (how much was seen versus kept).
138 
  The one genuinely per-loop body now lives in each loop's spec ``observe``
  (see :mod:`froot.loops`); the spine looks the loop up and runs it, so this
  selection seam needs no per-loop arm.
  """
  from froot.loops import registry
144 
  return await registry.get(loop).observe(
      target, package_manager, manifest_dir
  )
148 
149 
150@activity.defn
151async def scan_candidates(
  params: ScanCandidatesInput,
153) -> tuple[WorkItem, ...]:
  """Check out the repo and select this loop's candidates.
155 
  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
164 
  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
198 
199 
200@activity.defn
201async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
203 
  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
215 
  candidate = params.candidate
  if isinstance(candidate, Removal):
      # A removal already cleared the safe-to-remove veto at scan; there is no
      # changelog to read, so carry that conclusion straight into the PR
      # framing (the rationale the veto recorded on the work item).
      return CleanVerdict(
          rationale=candidate.justification or "unused; safe to remove"
      )
  changelog = await HttpChangelogSource().fetch(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__})."
      )
238 
239 
240@activity.defn
241async def gate_review(params: GateReviewInput) -> ChangelogVerdict:
  """Independently deep-review a bump at the gate; fail-closed to a hold.
243 
  The fourth trust leg (§3.7): a second, adversarial model pass over the
  changelog, run only when a bump is about to auto-merge. ``clean`` approves
  the merge; ``risky``/``unknown`` hold the PR for the human. Fail-CLOSED — a
  missing changelog or a model error returns a non-clean verdict, so an
  unreviewable bump never merges unattended. This is the opposite disposition
  from :func:`judge_changelog` (which degrades-to-proceed): here the safe
  direction is to hold, and a non-clean verdict already means hold.
  """
  from froot.adapters.changelog_http import HttpChangelogSource
  from froot.adapters.model_judge import PydanticAiJudge
254 
  candidate = params.candidate
  verdict: ChangelogVerdict
  if isinstance(candidate, Removal):
      # The fourth leg for a removal: independently re-judge that it is safe
      # to remove (the same check the scan veto ran). No changelog to read;
      # fail-CLOSED to a hold on a model error, like the bump path.
      try:
          verdict = await PydanticAiJudge().judge_removal(candidate)
      except Exception as exc:
          activity.logger.warning(
              "safe-to-remove gate review unavailable for %s; holding: %r",
              candidate.package,
              exc,
          )
          verdict = UnknownVerdict(
              rationale=(
                  f"Removal reviewer unavailable ({type(exc).__name__})."
              )
          )
  else:
      changelog = await HttpChangelogSource().fetch(candidate)
      if changelog is None:
          verdict = UnknownVerdict(
              rationale="No changelog to review; holding (fail-closed)."
          )
      else:
          try:
              verdict = await PydanticAiJudge().gate_review(
                  changelog, params.loop
              )
          except Exception as exc:
              activity.logger.warning(
                  "gate reviewer unavailable for %s; holding: %r",
                  candidate.package,
                  exc,
              )
              verdict = UnknownVerdict(
                  rationale=(
                      f"Gate reviewer unavailable ({type(exc).__name__})."
                  )
              )
  _gate_log.info(
      json.dumps(
          {
              "event": "gate_review",
              "loop": params.loop.value,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
              "package": params.candidate.package,
              "verdict": verdict.kind,
              "approved": verdict.kind == "clean",
          }
      )
  )
  return verdict
310 
311 
312@activity.defn
313async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Rewrite the manifest+lockfile and open (idempotently) the work item's PR.
315 
  The one place the action differs by work-item kind: a bump regenerates the
  lockfile at the target version, a removal deletes the dependency. Both are
  lockfile-only (no install, no scripts); everything around them — the dedup
  branch, the checkout, the push, the open — is the same chassis.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
323 
  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 = _draft_for(params)
  item = params.candidate
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      manifest_dir = _manifest_dir(params.target, workspace)
      match item:
          case Candidate():
              await package_manager.apply_patch_bump(item, manifest_dir)
          case Removal():
              await package_manager.remove_dependency(item, manifest_dir)
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
343 
344 
345@activity.defn
346async 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
349 
  return await GitHubForge().ci_status(params.target, params.head_sha)
351 
352 
353@activity.defn
354async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update.
356 
  The labels carry the loop *and* the judgment environment (the judge model)
  the PR was opened under, so the gate can count only the track record earned
  under the current environment and reset it when the model changes (§3.7).
  """
  from froot.adapters.github import GitHubForge
  from froot.config.settings import ModelSettings
  from froot.policy.environment import env_label
364 
  outcome = params.outcome
  item = outcome.candidate
  labels = (
      *pr_labels(params.loop),
      env_label(ModelSettings().ollama_model),
  )
  await GitHubForge().add_labels(params.target, outcome.pr.number, labels)
  record = {
      "event": "loop_outcome",
      "loop": params.loop.value,
      "repo": params.target.repo.slug,
      "package": item.package,
      "changelog": outcome.verdict.kind,
      "ci": outcome.ci.kind,
      "ci_passed": outcome.ci_passed,
      "pr": outcome.pr.number,
      "pr_url": outcome.pr.url,
  }
  match item:
      case Candidate():
          record |= {
              "action": "bump",
              "from": str(item.current),
              "to": str(item.target),
          }
      case Removal():
          record |= {"action": "remove", "dev": item.dev}
  _log.info(json.dumps(record))
393 
394 
395@activity.defn
396async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
398 
  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
405 
  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
410 
  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
429 
430 
431@activity.defn
432async def auto_merge_eligible(params: AutoMergeInput) -> bool:
  """Whether this (repo, loop) class has earned the auto-merge grant.
434 
  Short-circuits to ``False`` for any repo a steward has not allowlisted (the
  default), so the common case costs nothing. Otherwise it re-derives the
  class's standing from the live GitHub history — the same triangulated,
  windowed, environment-scoped computation the dashboard's shadow gate shows
  (``read_model.earned_now``) — so the acting gate and the advisory panel can
  never disagree. Best-effort: any read failure degrades to ``False`` (hold),
  never an auto-merge.
  """
  from datetime import UTC, datetime
444 
  from froot.config.settings import AutonomySettings, ModelSettings
  from froot.dashboard import github_source, read_model
  from froot.policy.environment import environment_slug
448 
  policy = AutonomySettings().policy()
  repo = params.target.repo.slug
  if repo not in policy.allowlisted_repos:
      return False
  now = datetime.now(UTC)
  prs, prs_error = await github_source.fetch((repo,))
  if prs_error is not None:
      return False  # can't confirm the record -> hold, never merge blind
  outcomes, _ = await github_source.fetch_outcomes(
      (repo,), prs, now=now, window_days=policy.window_days
  )
  return read_model.earned_now(
      now,
      prs,
      outcomes,
      repo,
      params.loop,
      policy,
      environment_slug(ModelSettings().ollama_model),
  )
469 
470 
471@activity.defn
472async def merge_pull_request(params: MergeInput) -> None:
  """Auto-merge an earned, clean+green bump's PR (the acting gate's write).
474 
  Reached only after the pure machine confirmed clean+green and the class
  earned the grant on an allowlisted repo. Passes the head SHA so GitHub
  refuses the merge if the head moved since the gate decided.
  """
  from froot.adapters.github import GitHubForge
480 
  await GitHubForge().merge_pull_request(
      params.target, params.pr.number, head_sha=params.pr.head_sha
  )
  _log.info(
      json.dumps(
          {
              "event": "pr_merged",
              "loop": params.loop.value,
              "reason": "auto_merge",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
          }
      )
  )
496 
497 
498@activity.defn
499async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
501 
  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
509 
  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),
          }
      )
  )
531 
532 
533@activity.defn
534async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
536 
  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
549 
  if not BehaviorSettings().reconcile:
      return 0
  # Reconcile is version-supersession cleanup, which only bump loops have: a
  # removal carries no version to be overtaken. A loop that does not reconcile
  # (dead-code) is skipped rather than re-running its signal (knip + the veto
  # judge) every tick only to close nothing. (A removal-specific reconcile —
  # close when no longer unused — is future work.) The trait is on the spec.
  from froot.loops import registry
558 
  if not registry.get(params.loop).reconciles:
      return 0
⋯ 265 lines hidden (lines 561–825)561 
  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)
593 
594 
595@activity.defn
596async def gate_selftest(params: GateSelfTestInput) -> tuple[str, ...]:
  """Run the adversarial gate probe against the live policy; alarm on escape.
598 
  The §2.11 deliberate disturbance for the acting gate: a battery of synthetic
  known-bad class histories a healthy gate must refuse, scored against the
  policy froot is *actually running* (config and all). Any escape — a bad
  class the live gate would grant — is logged at ERROR (the alarm) so it
  surfaces in telemetry the moment config drifts; a clean pass logs a
  heartbeat at INFO. Returns the escaped scenario names. Pure compute,
  repo-independent (the ``target``/``loop`` are only the log's context).
  """
  from froot.config.settings import AutonomySettings
  from froot.policy.gate_probe import gate_escapes
609 
  escaped = gate_escapes(AutonomySettings().policy())
  record = json.dumps(
      {
          "event": "gate_selftest",
          "loop": params.loop.value,
          "repo": params.target.repo.slug,
          "healthy": not escaped,
          "escaped": list(escaped),
      }
  )
  if escaped:
      _gate_log.error(record)  # an alarm: the gate would trust a bad class
  else:
      _gate_log.info(record)
  return escaped
625 
626 
627# ── The determinism reviewer (the transitive ring) ──────────────────────────
628@activity.defn
629async 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
632 
  return await GitHubForge().list_open_pull_requests(target)
634 
635 
636@activity.defn
637async 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
641 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
644 
  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
660 
661 
662@activity.defn
663async 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
668 
  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)
679 
680 
681@activity.defn
682async def adjudicate_frontier(
  params: AdjudicateInput,
684) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
687 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
693 
694 
695@activity.defn
696async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
699 
  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
721 
722 
723# ── The a11y reviewer (the source-level design-system ring) ──────────────────
724@activity.defn
725async def scan_pr_a11y(params: PrA11yReviewParams) -> A11yAnalysis:
  """Check out the PR head and scan its changed templates for a11y risks.
727 
  Scoped to the PR's changed Vue/JSX templates (shift-left: review what came
  in), so the advisory stays bounded and high-signal. The source lines are
  read into memory in the activity, so the pure scan is unaffected by the
  workspace being cleaned up.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.web_source import load_web_sources
735 
  forge = GitHubForge()
  changed = await forge.list_pull_request_files(
      params.target, params.pr.number
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout_pull_request(
          params.target, workspace, params.pr.number
      )
      sources = load_web_sources(workspace, changed)
      candidates = scan_sources(sources)
  return A11yAnalysis(candidates=candidates, scanned_files=len(sources))
748 
749 
750@activity.defn
751async def adjudicate_a11y(
  params: AdjudicateA11yInput,
753) -> tuple[A11yVerdict, ...]:
  """Run the model over each flagged candidate; return aligned verdicts."""
  from froot.adapters.a11y_judge import A11ySourceJudge
756 
  judge = A11ySourceJudge()
  verdicts: list[A11yVerdict] = []
  for candidate in params.candidates:
      verdicts.append(await judge.adjudicate(candidate))
  return tuple(verdicts)
762 
763 
764@activity.defn
765async def post_a11y_review(params: PostA11yInput) -> str | None:
  """Upsert (or clear) the advisory comment; log the ledger row.
767 
  Posts when there are findings, or when a prior comment must be cleared to
  "all clear" (true decay — a PR whose gaps were fixed never keeps a stale
  finding list, the bug the determinism reviewer leaves). A clean PR with
  no prior comment stays silent, so a clean PR is never spammed.
  """
  from froot.adapters.github import GitHubForge
774 
  forge = GitHubForge()
  exists = await forge.find_marked_comment(
      params.target, params.pr.number, A11Y_MARKER
  )
  url: str | None = None
  if should_post(has_findings=bool(params.findings), comment_exists=exists):
      body = render_a11y_comment(params.findings, params.pr.head_sha)
      url = await forge.upsert_issue_comment(
          params.target, params.pr.number, A11Y_MARKER, body
      )
  _a11y_log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": "a11y-review",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "head_sha": params.pr.head_sha,
              "findings": len(params.findings),
              "kinds": sorted({f.kind for f in params.findings}),
              "comment_url": url,
          }
      )
  )
  return url
800 
801 
802@activity.defn
803async def dispatch_pr_a11y_review(params: DispatchA11yInput) -> None:
  """Start a PR's a11y review (idempotent per PR + head SHA)."""
  from temporalio.common import WorkflowIDReusePolicy
  from temporalio.exceptions import WorkflowAlreadyStartedError
807 
  from froot.workflow.pr_a11y_review_workflow import PrA11yReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
810 
  temporal = await client()
  try:
      await temporal.start_workflow(
          PrA11yReviewWorkflow.run,
          PrA11yReviewParams(target=params.target, pr=params.pr),
          id=pr_a11y_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 an a11y review — a no-op, so
      # re-polling never double-reviews the same commit.
      return

The dashboard derives from the registry

The acting-loop tabs already iterated the configured loops; the last hard-coded per-loop bit was the icon map keyed by loop name in the renderer. Now the spec carries dashboard_icon, read_model threads it onto each LoopView, and the renderer just uses view.icon.

The icon is resolved from the spec (read_model is the dashboard server context, not a workflow).

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

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

⋯ 559 lines hidden (lines 1–559)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 (
  A11yLoop,
  A11yRecord,
  A11yRow,
  BumpRow,
  ClassGate,
  DashboardModel,
  Failure,
  Judgment,
  LoopView,
  Probes,
  Reliability,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
36)
37from froot.domain.loop import Loop
38from froot.domain.repo import RepoRef, TargetRepo
39from froot.loops import registry
40from froot.policy.autonomy import AutonomyPolicy, class_earned, pr_autonomy
41from froot.policy.canary import is_canary, score_probe
42from froot.policy.naming import (
  a11y_review_workflow_id,
  review_workflow_id,
  scan_workflow_id,
46)
47from froot.result import Ok
48 
49if TYPE_CHECKING:
  from datetime import datetime
51 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      A11yExecution,
      BumpExecution,
      PrA11yExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
61 
62# The conservative fallback policy: empty allowlist, so the shadow gate holds
63# everything until a caller passes a real one (built from FROOT_AUTOMERGE_*).
64_DEFAULT_POLICY = AutonomyPolicy()
65 
66_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
67# Every way a bump can end without closing cleanly — all belong in the honest
68# Failures panel, not silently dropped.
69_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
70 
71 
72def _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
81 
82 
83def _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
90 
91 
92def _scan_loops(
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  scans: tuple[ScanExecution, ...],
96) -> 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)
129 
130 
131def _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
138 
139 
140def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
  outcomes: dict[tuple[str, int], str],
145) -> tuple[BumpRow, ...]:
  """Join GitHub PRs (authoritative) to Temporal outcomes by PR number.
147 
  ``outcomes`` carries the post-merge signal (``held`` / ``broke`` /
  ``reverted`` / ``unknown``) per recently-merged ``(repo, number)``; it is
  absent for older or non-merged PRs, which leaves ``post_merge`` ``None``.
  """
  # 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
      # A removal carries no version: show "unused" in the target column and
      # no from-version, so a dead-code row reads "left-pad -> unused" rather
      # than the bump-shaped "left-pad -> ?".
      is_removal = pr.loop == Loop.DEAD_CODE.value
      rows.append(
          BumpRow(
              repo=pr.repo,
              loop=pr.loop,
              package=pr.package or "?",
              from_version=None if is_removal else pr.from_version,
              to_version="unused" if is_removal else (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,
              post_merge=outcomes.get((pr.repo, pr.number)),
              env=pr.env,
          )
      )
  rows.sort(key=_opened_sort_key, reverse=True)
  return tuple(rows)
193 
194 
195def _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
198 
199 
200def _minutes_between(
  start: datetime | None, end: datetime | None
202) -> 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)
207 
208 
209def _hours_between(
  start: datetime | None, end: datetime | None
211) -> 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)
216 
217 
218def _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),
  )
233 
234 
235def _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,
  )
251 
252 
253def _reliability(rows: tuple[BumpRow, ...], window_days: int) -> Reliability:
  """The post-merge outcome breakdown — did the merges hold? (coarse).
255 
  Reads only the ``post_merge`` tag the outcome reader set: ``unknown`` means
  an in-window merge we could not classify (no branch oracle / aged out), kept
  distinct from a held one; ``None`` (older / non-merged) is not counted here.
  The defect rate is a floor — manual and bundled reverts are invisible.
  """
  held = sum(1 for r in rows if r.post_merge == "held")
  broke = sum(1 for r in rows if r.post_merge == "broke")
  reverted = sum(1 for r in rows if r.post_merge == "reverted")
  unverified = sum(1 for r in rows if r.post_merge == "unknown")
  determined = held + broke + reverted
  return Reliability(
      held=held,
      broke=broke,
      reverted=reverted,
      unverified=unverified,
      determined=determined,
      defect_rate=((broke + reverted) / determined) if determined else None,
      window_days=window_days,
  )
275 
276 
277def _probes(canary_rows: tuple[BumpRow, ...]) -> Probes:
  """Tally the adversarial canary probes — caught / escaped / pending.
279 
  These rows are the synthetic bad bumps; they are scored on the strict bar
  (a canary must never merge) and kept out of every genuine bearing.
  """
  scored = [score_probe(r.state) for r in canary_rows]
  return Probes(
      caught=sum(1 for s in scored if s == "caught"),
      escaped=sum(1 for s in scored if s == "escaped"),
      pending=sum(1 for s in scored if s == "pending"),
      total=len(scored),
  )
290 
291 
292def _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,
  )
314 
315 
316def _decided_at(row: BumpRow) -> datetime | None:
  """When a decided PR was decided — merge time, else open time as a proxy.
318 
  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
325 
326 
327def _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
332 
333 
334def _class_gates(
  now: datetime,
  rows: tuple[BumpRow, ...],
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  policy: AutonomyPolicy,
  environment: str,
341) -> tuple[ClassGate, ...]:
  """The earned-autonomy standing of each (repo, loop) class.
343 
  Counts only PRs *decided within the window* — trust is recent, not lifetime
  (§2.11) — *and* earned under the current ``environment`` (§3.7's conditional
  property): a PR opened under a different judge model no longer counts, so a
  model swap resets the class. ``prior_env_decided`` keeps that reset legible.
  The budget figures (approvals / reclaim per week) translate the record into
  the steward-time MHE 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:
          in_window = [
              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)
          ]
          # An empty ``environment`` means "don't filter" (none configured);
          # otherwise only PRs stamped with the current one count.
          decided_rows = [
              r for r in in_window if not environment or r.env == environment
          ]
          prior_env_decided = len(in_window) - len(decided_rows)
          merged_rows = [r for r in decided_rows if r.state == "merged"]
          decided = len(decided_rows)
          merged = len(merged_rows)
          # The post-merge defect bearing, per class (§3.8): confirmed-held
          # outcomes and how many of those went bad. Independent of the rate.
          determined = sum(
              1
              for r in merged_rows
              if r.post_merge in ("held", "broke", "reverted")
          )
          defects = sum(
              1 for r in merged_rows if r.post_merge in ("broke", "reverted")
          )
          earned, blocker = class_earned(
              decided=decided,
              merged=merged,
              determined=determined,
              defects=defects,
              policy=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,
                  determined=determined,
                  defects=defects,
                  defect_rate=(
                      (defects / determined) if determined else None
                  ),
                  prior_env_decided=prior_env_decided,
                  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)
424 
425 
426def earned_now(
  now: datetime,
  prs: tuple[GithubPr, ...],
  outcomes: dict[tuple[str, int], str],
  repo: str,
  loop: Loop,
  policy: AutonomyPolicy,
  environment: str,
434) -> bool:
  """Whether one (repo, loop) class has earned its gate, from live data.
436 
  The exact computation the dashboard's class-gate panel shows, exposed so the
  acting gate (the loop) reuses it rather than duplicating the windowing,
  environment filter, and triangulation. Pure over already-fetched data.
  """
  rows = _bump_rows(now, prs, (), outcomes)
  gates = _class_gates(now, rows, (repo,), (loop,), policy, environment)
  return any(g.earned for g in gates)
444 
445 
446def _gate(
  rows: tuple[BumpRow, ...],
  gates: tuple[ClassGate, ...],
  policy: AutonomyPolicy,
450) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, most-aged first, each carrying a verdict.
452 
  The verdict is the gate's: would this PR auto-merge under its class's grant
  — the loop's real decision where the repo is allowlisted, advisory (the
  shadow gate) where it is not. 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)
488 
489 
490def _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)
507 
508 
509def _loop_title(loop: Loop) -> str:
  """The human tab title for a loop key (``dependency-patch`` -> ...)."""
  return loop.value.capitalize()
512 
513 
514def _failure_loop(workflow_id: str, loops: tuple[Loop, ...]) -> Loop | None:
  """Which loop a failed bump belongs to, from its workflow id.
516 
  Non-default loops carry their name as an id segment
  (``froot-bump-security-patch-…``); dependency-patch is segmentless
  (``froot-bump-…``), so it is the fallback once the specific loops miss.
  """
  for loop in loops:
      if loop is not Loop.DEPENDENCY_PATCH and workflow_id.startswith(
          f"froot-bump-{loop.value}-"
      ):
          return loop
  if workflow_id.startswith("froot-bump-"):
      return next(
          (loop for loop in loops if loop is Loop.DEPENDENCY_PATCH), None
      )
  return None
531 
532 
533def _bump_loops(
  now: datetime,
  rows: tuple[BumpRow, ...],
  canary_rows: tuple[BumpRow, ...],
  failures: tuple[Failure, ...],
  scans: tuple[ScanExecution, ...],
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  policy: AutonomyPolicy,
  environment: str,
  scan_interval_seconds: int,
  reliability_window_days: int,
545) -> tuple[LoopView, ...]:
  """One self-contained :class:`LoopView` per loop — the per-loop tabs.
547 
  Partitions the already-computed bump rows by loop and runs the very same
  aggregates the combined view uses, so each loop's tab is the whole dashboard
  scoped to one trust class (§3.9). Failures are attributed by workflow id.
  """
  views: list[LoopView] = []
  for loop in loops:
      loop_rows = tuple(r for r in rows if r.loop == loop.value)
      loop_canary = tuple(r for r in canary_rows if r.loop == loop.value)
      loop_gates = _class_gates(
          now, loop_rows, repos, (loop,), policy, environment
      )
      loop_failures = tuple(
          f for f in failures if _failure_loop(f.workflow_id, loops) is loop
      )
      views.append(
          LoopView(
              loop=loop.value,
              title=_loop_title(loop),
              icon=registry.get(loop).dashboard_icon,
              scan_loops=_scan_loops(repos, (loop,), scans),
⋯ 373 lines hidden (lines 568–940)568                scan_interval_seconds=scan_interval_seconds,
              track_record=_track_record(loop_rows),
              class_gates=loop_gates,
              verification=_verification(loop_rows),
              reliability=_reliability(loop_rows, reliability_window_days),
              probes=_probes(loop_canary),
              judgment=_judgment(loop_rows),
              gate=_gate(loop_rows, loop_gates, policy),
              bumps=loop_rows,
              failures=loop_failures,
          )
      )
  return tuple(views)
581 
582 
583def _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
590 
591 
592def _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-") + "-"
599 
600 
601def _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
610 
611 
612def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
614) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
616 
  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)
640 
641 
642def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
644) -> 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)
672 
673 
674def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
676) -> 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),
  )
688 
689 
690def _a11y_id(repo: str) -> str | None:
  """The deterministic a11y-loop id for an ``owner/name`` slug, if valid."""
  match RepoRef.parse(repo):
      case Ok(ref):
          return a11y_review_workflow_id(TargetRepo(repo=ref))
      case _:
          return None
697 
698 
699def _pr_a11y_prefix(repo: str) -> str | None:
  """The id prefix every per-PR a11y review of ``repo`` shares (join key)."""
  a11y_id = _a11y_id(repo)
  if a11y_id is None:
      return None
  # froot-a11y-<slug> -> froot-pr-a11y-<slug>- ; the pr/sha tail follows.
  return "froot-pr-a11y-" + a11y_id.removeprefix("froot-a11y-") + "-"
706 
707 
708def _attribute_a11y_repo(
  workflow_id: str, repos: tuple[str, ...]
710) -> str | None:
  """The configured repo a per-PR-a11y id belongs to (longest prefix)."""
  best: str | None = None
  best_len = -1
  for repo in repos:
      prefix = _pr_a11y_prefix(repo)
      if prefix and workflow_id.startswith(prefix) and len(prefix) > best_len:
          best, best_len = repo, len(prefix)
  return best
719 
720 
721def _a11y_loops(
  repos: tuple[str, ...], a11y_reviews: tuple[A11yExecution, ...]
723) -> tuple[A11yLoop, ...]:
  """One liveness row per repo that actually has an a11y loop.
725 
  Scoped to the Temporal repos, so a configured repo with no a11y loop is
  omitted rather than shown as a dead one.
  """
  by_id: dict[str, A11yExecution] = {}
  for review in a11y_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[A11yLoop] = []
  for repo in repos:
      a11y_id = _a11y_id(repo)
      execution = by_id.get(a11y_id) if a11y_id is not None else None
      if execution is None:
          continue
      loops.append(
          A11yLoop(
              repo=repo,
              status=execution.status,
              live=execution.status in _LIVE_STATUSES,
              last_tick=execution.start,
          )
      )
  return tuple(loops)
749 
750 
751def _a11y_rows(
  pr_a11y_reviews: tuple[PrA11yExecution, ...], repos: tuple[str, ...]
753) -> tuple[A11yRow, ...]:
  """Project each per-PR a11y review into a row, newest review first."""
  rows: list[A11yRow] = []
  for execution in pr_a11y_reviews:
      repo = _attribute_a11y_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(
          A11yRow(
              repo=repo or "?",
              pr_number=execution.pr_number,
              pr_url=pr_url,
              head_sha=execution.head_sha,
              findings=execution.findings,
              kinds=execution.kinds,
              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)
781 
782 
783def _a11y_record(
  loops: tuple[A11yLoop, ...], rows: tuple[A11yRow, ...]
785) -> A11yRecord:
  """Counts over the completed a11y reviews."""
  completed = [r for r in rows if r.status == "completed"]
  flagged = sum(1 for r in completed if r.findings > 0)
  issues = sum(r.findings for r in completed)
  return A11yRecord(
      reviewed=len(completed),
      flagged=flagged,
      clean=len(completed) - flagged,
      issues=issues,
      repos_covered=len(loops),
  )
797 
798 
799def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
806) -> 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
      ),
  )
831 
832 
833def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
  policy: AutonomyPolicy = _DEFAULT_POLICY,
  scan_interval_seconds: int,
  review_interval_seconds: int,
  a11y_interval_seconds: int,
  github: tuple[tuple[GithubPr, ...], str | None],
  temporal: tuple[
      tuple[
          tuple[ScanExecution, ...],
          tuple[BumpExecution, ...],
          tuple[ReviewExecution, ...],
          tuple[PrReviewExecution, ...],
          tuple[A11yExecution, ...],
          tuple[PrA11yExecution, ...],
      ],
      str | None,
  ],
  telemetry: tuple[RunTelemetry, str | None],
  outcomes: dict[tuple[str, int], str] | None = None,
  reliability_window_days: int = 90,
  environment: str = "",
858) -> DashboardModel:
  """Build the whole view from the readers' ``(data, error)`` outputs.
860 
  ``outcomes`` is the post-merge signal per ``(repo, number)`` from the
  outcome reader (best-effort); absent for a merge older than the window.
  """
  prs, github_error = github
  (
      (
          scans,
          bumps,
          reviews,
          pr_reviews,
          a11y_reviews,
          pr_a11y_reviews,
      ),
      temporal_error,
  ) = temporal
  run_telemetry, clickhouse_error = telemetry
877 
  all_rows = _bump_rows(now, prs, bumps, outcomes or {})
  # Canary probes are synthetic bad bumps — keep them out of every genuine
  # bearing (track record, defect rate, gates) so a planted failure can never
  # pollute the real reputation; they get their own tally.
  canary_rows = tuple(r for r in all_rows if is_canary(r.to_version))
  rows = tuple(r for r in all_rows if not is_canary(r.to_version))
  class_gates = _class_gates(now, rows, repos, loops, policy, environment)
  all_failures = _failures(bumps)
  bump_loops = _bump_loops(
      now,
      rows,
      canary_rows,
      all_failures,
      scans,
      repos,
      loops,
      policy,
      environment,
      scan_interval_seconds,
      reliability_window_days,
  )
  review_loops = _review_loops(repos, reviews)
  review_rows = _review_rows(pr_reviews, repos)
  a11y_loops = _a11y_loops(repos, a11y_reviews)
  a11y_rows = _a11y_rows(pr_a11y_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)
          + len(a11y_reviews)
          + len(pr_a11y_reviews),
          run_telemetry,
          clickhouse_error,
      ),
      scan_loops=_scan_loops(repos, loops, scans),
      track_record=_track_record(rows),
      class_gates=class_gates,
      verification=_verification(rows),
      reliability=_reliability(rows, reliability_window_days),
      probes=_probes(canary_rows),
      judgment=_judgment(rows),
      gate=_gate(rows, class_gates, policy),
      bumps=rows,
      failures=all_failures,
      review_interval_seconds=review_interval_seconds,
      review_loops=review_loops,
      review_record=_review_record(review_loops, review_rows),
      reviews=review_rows,
      a11y_interval_seconds=a11y_interval_seconds,
      a11y_loops=a11y_loops,
      a11y_record=_a11y_record(a11y_loops, a11y_rows),
      a11y_reviews=a11y_rows,
      telemetry=run_telemetry,
      bump_loops=bump_loops,
  )

No more {...}.get(view.loop, ...) — the renderer is loop-blind.

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

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

⋯ 887 lines hidden (lines 1–887)1"""Render the view model to one self-contained HTML page (pure).
2 
3All CSS is inline, there is no JavaScript (tabs are CSS-only, via hidden radio
4inputs), and the page makes no network request of its own — it is a static,
5full-screen projection of an already-computed
class:`~froot.dashboard.model.DashboardModel`. Every dynamic value is
7HTML-escaped at the boundary.
8 
9The shape is gate-first: each loop is a tab whose hero is the *gate* — a small
10flow of the four trust bearings into the earned/hold decision — over a compact
11metric grid and foldable detail. Each loop is a distinct trust class (§3.9), so
12each tab is the whole dashboard scoped to one loop; determinism-review and the
13cross-cutting run-telemetry get their own tabs.
14"""
15 
16from __future__ import annotations
17 
18from datetime import UTC, datetime, timedelta
19from html import escape
20from typing import TYPE_CHECKING
21 
22if TYPE_CHECKING:
  from froot.dashboard.model import (
      A11yRow,
      BumpRow,
      ClassGate,
      DashboardModel,
      LoopView,
      ReviewRow,
      RunTelemetry,
  )
32 
33_LIGHT = (
  "--fg:#1b1f24;--mut:#586273;--faint:#8b94a3;--line:#e9ebef;--hair:#d6dae1;"
  "--bg:#fcfcfd;--tint:#f5f6f9;--accent:#7c3aed;--ok:#2f7d4f;--warn:#8a6011;"
  "--bad:#b23a2e"
37)
38_DARK = (
  "--fg:#e7e9ee;--mut:#9aa3b2;--faint:#6b7484;--line:#22262e;--hair:#333a45;"
  "--bg:#0b0d10;--tint:#14171c;--accent:#a78bfa;--ok:#4cae6e;--warn:#d6a23f;"
  "--bad:#e07a6e"
42)
43# Editorial-minimal "read-model" idiom: structure from hairlines + whitespace +
44# type, never from box fills; semantic color rides on dots and the value text,
45# not on filled chips. Light is the default; a toggle (data-theme) forces either
46# theme, and bare system preference is honored when nothing is forced.
47_CSS = f"""
root{{{_LIGHT};
49--serif:"Iowan Old Style",Palatino,Georgia,"Times New Roman",ui-serif,serif;
50--mono:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace}}
51@media(prefers-color-scheme:dark){{:root:not([data-theme]){{{_DARK}}}}}
root[data-theme=dark]{{{_DARK}}}
53*{{box-sizing:border-box}}
54body{{margin:0;background:var(--bg);color:var(--fg);
55font:14px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif;
56-webkit-font-smoothing:antialiased}}
57main{{padding:0 clamp(16px,3vw,46px) 72px}}
58a{{color:var(--accent);text-decoration:none}}a:hover{{text-decoration:underline}}
mono{{font-family:var(--mono);font-size:.92em}}
serif{{font-family:var(--serif)}}
mut{{color:var(--mut)}}.ok{{color:var(--ok)}}.warn{{color:var(--warn)}}
bad{{color:var(--bad)}}.acc{{color:var(--accent)}}
num{{font-variant-numeric:tabular-nums}}
64/* status dot — the one carrier of valence besides the value text itself */
dot{{display:inline-block;width:8px;height:8px;border-radius:50%;flex:none}}
dot.ok{{background:var(--ok)}}.dot.warn{{background:var(--warn)}}
dot.bad{{background:var(--bad)}}.dot.mute{{background:var(--faint)}}
dot.acc{{background:var(--accent)}}
69/* header */
70header{{display:flex;align-items:center;flex-wrap:wrap;gap:9px 22px;
71padding:26px 0 18px;border-bottom:1px solid var(--line)}}
72h1{{font-family:var(--serif);font-size:23px;margin:0;letter-spacing:-.01em;
73font-weight:600;display:inline-flex;align-items:center;gap:9px}}
ic{{width:15px;height:15px;flex:none}}
mark{{width:18px;height:18px;flex:none;color:var(--accent)}}
hstatus{{display:flex;align-items:center;gap:7px;font-size:13px}}
hmeta{{margin-left:auto;display:flex;flex-wrap:wrap;align-items:center;
78gap:7px 16px;color:var(--mut);font-size:12px}}
srcs{{display:inline-flex;flex-wrap:wrap;align-items:center;gap:5px 14px}}
src{{display:inline-flex;align-items:center;gap:6px}}
themetgl{{background:transparent;border:1px solid var(--line);border-radius:
82999px;width:30px;height:30px;cursor:pointer;color:var(--mut);font-size:14px;
83line-height:1;display:inline-flex;align-items:center;justify-content:center;
84transition:border-color .14s,color .14s}}
themetgl:hover{{border-color:var(--fg);color:var(--fg)}}
86/* tabs (CSS-only radios) */
tabin{{position:absolute;opacity:0;pointer-events:none}}
88nav.tabbar{{display:flex;flex-wrap:wrap;gap:2px 6px;margin:16px 0 0;
89border-bottom:1px solid var(--line)}}
90nav.tabbar label{{display:inline-flex;align-items:center;gap:8px;cursor:pointer;
91padding:10px 4px;margin:0 12px -1px 0;font-size:13.5px;font-weight:600;
92color:var(--mut);border-bottom:2px solid transparent;white-space:nowrap;
93transition:color .14s}}
94nav.tabbar label:hover{{color:var(--fg)}}
95nav.tabbar label.aside{{margin-left:auto}}
96nav.tabbar label .badge{{font-family:var(--mono);font-weight:500;font-size:11px;
97color:var(--faint)}}
panel{{display:none;padding:26px 0 0;animation:fade .16s ease}}
99@keyframes fade{{from{{opacity:.45}}to{{opacity:1}}}}
100/* gate hero */
hero{{display:grid;align-items:start;margin:0 0 4px;gap:22px 40px;
102grid-template-columns:minmax(330px,1.05fr) minmax(300px,1fr)}}
103@media(max-width:780px){{.hero{{grid-template-columns:1fr}}}}
heroh{{font-size:11px;text-transform:uppercase;letter-spacing:.1em;
105color:var(--mut);font-weight:600;margin:0 0 14px;padding-bottom:7px;
106border-bottom:1px solid var(--line);display:flex;align-items:center;gap:7px}}
gateflow{{display:flex;align-items:center;gap:14px;flex-wrap:wrap}}
bearings{{display:flex;flex-direction:column;min-width:172px;flex:1 1 172px}}
bearing{{display:flex;align-items:center;justify-content:space-between;gap:12px;
110padding:7px 0;border-bottom:1px solid var(--line);font-size:12.5px}}
bearing:last-child{{border-bottom:0}}
bearing .bl{{color:var(--mut)}}
bearing .bv{{display:inline-flex;align-items:center;gap:7px;font-weight:600;
114font-variant-numeric:tabular-nums}}
bearing.armed .bv{{color:var(--faint);font-weight:500}}
flowarrow{{color:var(--hair);font-size:17px;line-height:1}}
gatenode{{flex:0 0 auto;text-align:center;padding:2px 6px}}
gatenode .gl{{font-size:10px;text-transform:uppercase;letter-spacing:.1em;
119color:var(--mut);font-weight:600}}
gatenode .gv{{font-family:var(--serif);font-size:30px;font-weight:600;
121line-height:1.1;color:var(--fg);font-variant-numeric:tabular-nums}}
gatenode .gs{{font-size:11px;color:var(--faint)}}
outcome{{flex:0 0 auto;display:flex;flex-direction:column;gap:3px}}
outcome .ot{{display:flex;align-items:center;gap:7px;font-size:15px;
125font-weight:700;letter-spacing:.02em}}
outcome .os{{font-size:11px;color:var(--faint)}}
outcome.act .ot{{color:var(--ok)}}
outcome.hold .ot{{color:var(--mut)}}
caption{{color:var(--mut);font-size:12px;margin:14px 0 0;line-height:1.5;
130max-width:62ch}}
131/* per-class table inside the hero */
classes{{width:100%;border-collapse:collapse;font-size:12.5px}}
classes th,.classes td{{text-align:left;padding:7px 12px 7px 0;
134border-bottom:1px solid var(--line);vertical-align:baseline}}
classes th{{color:var(--faint);font-weight:600;font-size:10.5px;
136text-transform:uppercase;letter-spacing:.06em}}
classes td.r{{font-variant-numeric:tabular-nums}}
pill{{display:inline-flex;align-items:center;gap:6px;font-size:12px;
139font-weight:600}}
pill.ok{{color:var(--ok)}}.pill.hold{{color:var(--mut);font-weight:500}}
141/* metric strip — bare stats, no boxes; a hairline frames the row */
cards{{display:grid;gap:20px 30px;margin:26px 0 0;padding:18px 0 0;
143border-top:1px solid var(--line);
144grid-template-columns:repeat(auto-fit,minmax(140px,1fr))}}
card .n{{font-size:26px;font-weight:600;line-height:1.05;
146font-variant-numeric:tabular-nums;letter-spacing:-.01em}}
card .l{{color:var(--mut);font-size:11px;text-transform:uppercase;
148letter-spacing:.05em;font-weight:600;margin-top:5px}}
card .x{{color:var(--faint);font-size:11.5px;margin-top:3px}}
150/* section + detail */
sec{{margin:30px 0 0}}
sech{{font-size:11px;text-transform:uppercase;letter-spacing:.1em;
153color:var(--mut);font-weight:600;margin:0 0 12px;padding-bottom:7px;
154border-bottom:1px solid var(--line);display:flex;align-items:center;gap:8px}}
sech .n{{color:var(--faint);font-weight:500;letter-spacing:0;
156text-transform:none}}
157details.fold{{margin:14px 0 0;border-top:1px solid var(--line)}}
158details.fold>summary{{cursor:pointer;list-style:none;padding:11px 0;
159font-size:12px;font-weight:600;color:var(--mut);display:flex;
160align-items:center;gap:9px}}
161details.fold>summary:hover{{color:var(--fg)}}
162details.fold>summary::-webkit-details-marker{{display:none}}
163details.fold>summary::before{{content:"\\25B8";color:var(--hair);font-size:10px}}
164details.fold[open]>summary::before{{content:"\\25BE"}}
165details.fold>summary .c{{color:var(--faint);font-weight:500}}
166details.fold .body{{padding:2px 0 16px}}
167table.data{{width:100%;border-collapse:collapse;font-size:12.5px}}
168table.data th,table.data td{{text-align:left;padding:7px 14px 7px 0;
169border-bottom:1px solid var(--line);vertical-align:baseline}}
170table.data th{{color:var(--faint);font-weight:600;font-size:10.5px;
171text-transform:uppercase;letter-spacing:.05em}}
172table.data td.r{{font-variant-numeric:tabular-nums;white-space:nowrap}}
empty{{color:var(--faint);font-size:13px;padding:16px 0}}
t{{font-size:12px;color:var(--mut)}}
t.ok{{color:var(--ok)}}.t.warn{{color:var(--warn)}}.t.bad{{color:var(--bad)}}
cad{{color:var(--mut);font-size:12px;margin:18px 0 0}}
cad b{{color:var(--fg);font-weight:600}}
178footer{{margin:46px 0 0;padding-top:18px;border-top:1px solid var(--line);
179color:var(--mut);font-size:12px;line-height:1.7;max-width:92ch}}
180footer b{{color:var(--fg);font-weight:600}}
181"""
182 
183# The page's only script: set the saved theme before first paint (no flash) and
184# expose the toggle. Inline + tiny, so the page stays one self-contained file
185# with no external request; absent JS, the page simply follows the OS theme.
186_THEME_JS = (
  "(function(){var k='froot-theme',r=document.documentElement,"
  "s=localStorage.getItem(k);if(s)r.setAttribute('data-theme',s);"
  "window.__toggleTheme=function(){var c=r.getAttribute('data-theme')||"
  "(matchMedia('(prefers-color-scheme:dark)').matches?'dark':'light');"
  "var n=c==='dark'?'light':'dark';r.setAttribute('data-theme',n);"
  "localStorage.setItem(k,n);};})();"
193)
194 
195# A small set of hairline line-icons (inline SVG, stroke = currentColor, so they
196# inherit the muted ink and theme themselves). Anchors for the eye — tabs,
197# section headers, the wordmark — never on the number strip. Paths are 24x24.
198_ICONS = {
  "package": (
      '<path d="M21 8v8a2 2 0 0 1-1 1.73l-7 4a2 2 0 0 1-2 0l-7-4A2 2 0 0 1 3 '
      '16V8a2 2 0 0 1 1-1.73l7-4a2 2 0 0 1 2 0l7 4A2 2 0 0 1 21 8Z"/>'
      '<path d="m3.3 7 8.7 5 8.7-5"/><path d="M12 22V12"/>'
  ),
  "shield": (
      '<path d="M20 13c0 5-3.5 7.5-7.66 8.95a1 1 0 0 1-.67-.01C7.5 20.5 4 18 '
      "4 13V6a1 1 0 0 1 1-1c2 0 4.5-1.2 6.24-2.72a1.17 1.17 0 0 1 1.52 "
      '0C14.51 3.81 17 5 19 5a1 1 0 0 1 1 1z"/>'
  ),
  "shield-check": (
      '<path d="M20 13c0 5-3.5 7.5-7.66 8.95a1 1 0 0 1-.67-.01C7.5 20.5 4 18 '
      "4 13V6a1 1 0 0 1 1-1c2 0 4.5-1.2 6.24-2.72a1.17 1.17 0 0 1 1.52 "
      '0C14.51 3.81 17 5 19 5a1 1 0 0 1 1 1z"/><path d="m9 12 2 2 4-4"/>'
  ),
  "search": '<circle cx="11" cy="11" r="8"/><path d="m21 21-4.3-4.3"/>',
  "activity": '<path d="M22 12h-4l-3 9L9 3l-3 9H2"/>',
  "layers": (
      '<path d="M12.83 2.18a2 2 0 0 0-1.66 0L2.6 6.08a1 1 0 0 0 0 1.83l8.58 '
      '3.91a2 2 0 0 0 1.66 0l8.58-3.9a1 1 0 0 0 0-1.83Z"/>'
      '<path d="m22 17.65-9.17 4.16a2 2 0 0 1-1.66 0L2 17.65"/>'
      '<path d="m22 12.65-9.17 4.16a2 2 0 0 1-1.66 0L2 12.65"/>'
  ),
  "inbox": (
      '<path d="M22 12h-6l-2 3h-4l-2-3H2"/>'
      '<path d="M5.45 5.11 2 12v6a2 2 0 0 0 2 2h16a2 2 0 0 0 2-2v-6l-3.45'
      '-6.89A2 2 0 0 0 16.76 4H7.24a2 2 0 0 0-1.79 1.11z"/>'
  ),
  "lock": (
      '<rect width="18" height="11" x="3" y="11" rx="2"/>'
      '<path d="M7 11V7a5 5 0 0 1 10 0v4"/>'
  ),
  "lock-open": (
      '<rect width="18" height="11" x="3" y="11" rx="2"/>'
      '<path d="M7 11V7a5 5 0 0 1 9.9-1"/>'
  ),
  "merge": (
      '<circle cx="18" cy="18" r="3"/><circle cx="6" cy="6" r="3"/>'
      '<path d="M6 21V9a9 9 0 0 0 9 9"/>'
  ),
  "scissors": (
      '<circle cx="6" cy="6" r="3"/><path d="M8.12 8.12 12 12"/>'
      '<path d="M20 4 8.12 15.88"/><circle cx="6" cy="18" r="3"/>'
      '<path d="M14.8 14.8 20 20"/>'
  ),
  "accessibility": (
      '<circle cx="16" cy="4" r="1"/><path d="m18 19 1-7-6 1"/>'
      '<path d="m5 8 3-3 5.5 3-2.36 3.5"/>'
      '<path d="M4.24 14.5a5 5 0 0 0 6.88 6"/>'
      '<path d="M13.76 17.5a5 5 0 0 0-6.88-6"/>'
  ),
250}
251 
252 
253def _icon(name: str, cls: str = "ic") -> str:
  return (
      f'<svg class="{cls}" viewBox="0 0 24 24" fill="none" '
      'stroke="currentColor" stroke-width="1.9" stroke-linecap="round" '
      f'stroke-linejoin="round" aria-hidden="true">{_ICONS[name]}</svg>'
  )
259 
260 
261def _wordmark() -> str:
  """The froot mark: a fruit in the accent, with a single green leaf.
263 
  Two-tone on purpose — the one spot color is allowed to be richer. The body
  and stem ride the accent (currentColor); the leaf is the semantic green,
  so the mark reads as a piece of fruit, not a logo abstraction.
  """
  return (
      '<svg class="mark" viewBox="0 0 24 24" fill="none" '
      'stroke="currentColor" stroke-width="1.9" stroke-linecap="round" '
      'stroke-linejoin="round" aria-hidden="true">'
      '<circle cx="11.5" cy="14.5" r="6.2"/>'
      '<path d="M11.5 8.3V6.15" stroke="var(--ok)"/>'
      '<path d="M11.5 6.15c1.4-2 3.5-2.6 5.8-2.2-.4 2.5-2.5 3.7-5.8 2.2z" '
      'stroke="var(--ok)"/></svg>'
  )
277 
278 
279_CI_CLASS = {
  "passed": "ok",
  "failed": "bad",
  "absent": "mut",
  "timed_out": "warn",
284}
285_VERDICT_CLASS = {"clean": "ok", "risky": "warn", "unknown": "mut"}
286_POST_MERGE_CLASS = {
  "held": "ok",
  "broke": "bad",
  "reverted": "bad",
  "unknown": "mut",
291}
292_STATE_CLASS = {"merged": "ok", "closed": "mut", "open": "warn"}
293_FAIL_CLASS = {"failed": "bad", "terminated": "warn"}
294 
295 
296# ── small formatters ─────────────────────────────────────────────────────────
297def _aware(when: datetime) -> datetime:
  return when if when.tzinfo else when.replace(tzinfo=UTC)
299 
300 
301def _ago(when: datetime | None, now: datetime) -> str:
  if when is None:
      return "—"
  secs = (now - _aware(when)).total_seconds()
  if secs < 90:
      return "just now"
  mins = secs / 60
  if mins < 90:
      return f"{round(mins)}m ago"
  hours = mins / 60
  if hours < 36:
      return f"{round(hours)}h ago"
  return f"{round(hours / 24)}d ago"
314 
315 
316def _until(when: datetime | None, now: datetime) -> str:
  if when is None:
      return "—"
  secs = (_aware(when) - now).total_seconds()
  if secs <= 0:
      return "due"
  mins = secs / 60
  if mins < 90:
      return f"~{round(mins)}m"
  hours = mins / 60
  if hours < 36:
      return f"~{round(hours)}h"
  return f"~{round(hours / 24)}d"
329 
330 
331def _pct(rate: float | None) -> str:
  return "—" if rate is None else f"{rate * 100:.0f}%"
333 
334 
335def _dot(kind: str) -> str:
  return f'<span class="dot {kind}"></span>'
337 
338 
339def _tag(value: str | None, classes: dict[str, str]) -> str:
  if value is None:
      return '<span class="t">—</span>'
  cls = classes.get(value, "")
  return f'<span class="t {cls}">{escape(value)}</span>'
344 
345 
346def _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>'
350 
351 
352def _short_id(workflow_id: str) -> str:
  return escape(workflow_id.removeprefix("froot-bump-"))
354 
355 
356# ── header ───────────────────────────────────────────────────────────────────
357def _alive(model: DashboardModel) -> tuple[str, str]:
  """Global liveness dot + label across every loop."""
  live = sum(1 for x in model.scan_loops if x.live) + sum(
      1 for x in model.review_loops if x.live
  )
  total = len(model.scan_loops) + len(model.review_loops)
  if total == 0:
      return "mute", "no loops configured"
  kind = "ok" if live == total else ("warn" if live else "bad")
  return kind, f"{live}/{total} loops live"
367 
368 
369def _header(model: DashboardModel) -> str:
  kind, label = _alive(model)
  sources = "".join(
      f'<span class="src">{_dot("ok" if s.ok else "bad")}'
      f"{escape(s.name)}</span>"
      for s in model.sources
  )
  return (
      "<header>"
      f"<h1>{_wordmark()}froot</h1>"
      f'<span class="hstatus">{_dot(kind)}{escape(label)}</span>'
      f'<span class="hmeta"><span class="srcs">{sources}</span>'
      '<button class="themetgl" type="button" onclick="__toggleTheme()"'
      ' title="Toggle light / dark"'
      ' aria-label="Toggle light or dark theme">&#9680;</button>'
      "</span></header>"
  )
386 
387 
388# ── the gate hero (per loop) ─────────────────────────────────────────────────
389def _bearing(label: str, value: str, kind: str, *, armed: bool = False) -> str:
  cls = "bearing armed" if armed else "bearing"
  dot = _dot("mute" if armed else kind)
  return (
      f'<div class="{cls}"><span class="bl">{escape(label)}</span>'
      f'<span class="bv {kind}">{dot}{escape(value)}</span></div>'
  )
396 
397 
398def _gate_hero(view: LoopView) -> str:
  t, rel, pr = view.track_record, view.reliability, view.probes
  earned = sum(1 for g in view.class_gates if g.earned)
  total = len(view.class_gates)
  acting = any(r.would_auto_merge for r in view.gate)
  # The four bearings: rate + defect come from the record; the adversarial
  # probe (canary) is the loop's escaped-count; the deep review runs per-PR at
  # the merge, so it is shown armed (always-on), not a record figure.
  rate_ok = "ok" if (t.merge_rate or 0) >= 0.95 else "warn"
  defect_ok = "ok" if not rel.defect_rate else "bad"
  probe_ok = "ok" if pr.escaped == 0 else "bad"
  bearings = "".join(
      (
          _bearing(
              "approval rate",
              _pct(t.merge_rate),
              rate_ok if t.merge_rate is not None else "mut",
          ),
          _bearing(
              "defect rate",
              _pct(rel.defect_rate),
              defect_ok if rel.defect_rate is not None else "mut",
          ),
          _bearing(
              "probe",
              f"{pr.escaped} escaped" if pr.total else "none",
              probe_ok if pr.total else "mut",
          ),
          _bearing("deep review", "armed", "mut", armed=True),
      )
  )
  if total == 0:
      node = (
          '<div class="gatenode"><div class="gl">gate</div>'
          '<div class="gv">—</div><div class="gs">no class yet</div></div>'
      )
  else:
      node = (
          f'<div class="gatenode"><div class="gl">earned</div>'
          f'<div class="gv">{earned}/{total}</div>'
          '<div class="gs">classes</div></div>'
      )
  if acting:
      out = (
          f'<div class="outcome act"><div class="ot">{_icon("merge")}'
          "AUTO-MERGE</div>"
          '<div class="os">a class is acting now</div></div>'
      )
  elif earned:
      out = (
          f'<div class="outcome act"><div class="ot">{_icon("lock-open")}'
          "EARNED</div>"
          '<div class="os">acts where allowlisted</div></div>'
      )
  else:
      out = (
          f'<div class="outcome hold"><div class="ot">{_icon("lock")}HOLD'
          "</div><div class="
          '"os">building the record</div></div>'
      )
  flow = (
      '<div class="gateflow">'
      f'<div class="bearings">{bearings}</div>'
      '<div class="flowarrow">&rarr;</div>'
      f"{node}"
      '<div class="flowarrow">&rarr;</div>'
      f"{out}</div>"
  )
  caption = (
      '<p class="caption">A class earns the gate by triangulation: a high '
      "<b>approval rate</b> and a low <b>defect rate</b>, over enough "
      "evidence. Two further legs guard the live merge — an adversarial "
      "<b>probe</b> and an independent <b>deep review</b> at merge. "
      "Auto-merge is allowlist-gated (off by default).</p>"
  )
  return (
      f'<div><div class="heroh">{_icon("shield-check")}'
      "Earned autonomy &middot; the gate</div>"
      f"{flow}{caption}</div>"
  )
478 
479 
480def _class_table(view: LoopView) -> str:
  if not view.class_gates:
      return (
          f'<div><div class="heroh">{_icon("layers")}Classes</div>'
          '<div class="empty">No classes yet &mdash; a (repo, loop) '
          "earns the gate from its own track record.</div></div>"
      )
  rows = "".join(_class_row(g) for g in view.class_gates)
  return (
      f'<div><div class="heroh">{_icon("layers")}Per-class standing</div>'
      '<table class="classes"><thead><tr><th>repo</th><th>rate</th>'
      "<th>defect</th><th>gate</th><th>budget/wk</th></tr></thead>"
      f"<tbody>{rows}</tbody></table></div>"
  )
494 
495 
496def _class_row(g: ClassGate) -> str:
  if g.earned:
      gate = f'<span class="pill ok">{_dot("ok")}earned</span>'
  else:
      gate = (
          f'<span class="pill hold">{_dot("mute")}hold</span> '
          f'<span class="mut">{escape(g.blocker or "")}</span>'
      )
  defect = "—" if g.defect_rate is None else _pct(g.defect_rate)
  budget = (
      f"{g.reclaim_per_week:.1f}/{g.approvals_per_week:.1f}"
      if g.approvals_per_week
      else "—"
  )
  return (
      f'<tr><td class="mono">{escape(g.repo)}</td>'
      f'<td class="r">{_pct(g.merge_rate)}</td>'
      f'<td class="r">{escape(defect)}</td>'
      f"<td>{gate}</td>"
      f'<td class="r mut">{escape(budget)}</td></tr>'
  )
517 
518 
519# ── metric cards (per loop) ──────────────────────────────────────────────────
520def _card(n: object, label: str, extra: str = "", kind: str = "") -> str:
  x = f'<div class="x">{extra}</div>' if extra else ""
  return (
      f'<div class="card"><div class="n {kind}">{escape(str(n))}</div>'
      f'<div class="l">{escape(label)}</div>{x}</div>'
  )
526 
527 
528def _loop_cards(view: LoopView) -> str:
  t, v, rel, j, pr = (
      view.track_record,
      view.verification,
      view.reliability,
      view.judgment,
      view.probes,
  )
  defects = rel.broke + rel.reverted
  cards = [
      _card(
          t.opened,
          "proposed",
          f"{t.merged} merged · {t.closed_unmerged} closed",
      ),
      _card(_pct(t.merge_rate), "approval rate", "the first bearing"),
      _card(
          t.open_now,
          "awaiting you",
          "open, needs a human",
          "warn" if t.open_now else "",
      ),
      _card(
          _pct(rel.defect_rate) if rel.determined else "—",
          "defect rate",
          f"{rel.held} held · {defects} broke",
          "bad" if defects else "",
      ),
      _card(
          f"{v.passed}/{v.oracle_existed}" if v.oracle_existed else "—",
          "CI passed",
          "the oracle",
      ),
      _card(
          pr.escaped if pr.total else "0",
          "probes escaped",
          f"{pr.caught} caught" if pr.total else "no probes yet",
          "bad" if pr.escaped else "",
      ),
      _card(
          j.clean,
          "clean verdicts",
          f"{j.clean_but_failed} mis-judged"
          if j.clean_but_failed
          else "judge calibrated",
      ),
      _card(rel.held, "merges held", "post-merge, stayed green"),
  ]
  return f'<div class="cards">{"".join(cards)}</div>'
577 
578 
579# ── detail tables ────────────────────────────────────────────────────────────
580def _cadence(view: LoopView, now: datetime) -> str:
  live = sum(1 for s in view.scan_loops if s.live)
  total = len(view.scan_loops)
  nxt = ""
  last = max(
      (s.last_tick for s in view.scan_loops if s.last_tick is not None),
      default=None,
  )
  if last is not None:
      due = last + timedelta(seconds=view.scan_interval_seconds)
      nxt = f" &middot; next {escape(_until(due, now))}"
  every = round(view.scan_interval_seconds / 3600, 1)
  return (
      f'<p class="cad">Scan loop &middot; <b>{live}/{total}</b> live '
      f"&middot; every <b>{every}h</b>{nxt}</p>"
  )
596 
597 
598def _bumps_fold(view: LoopView) -> str:
  if not view.bumps:
      return ""
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(r.package)}</td>'
      f'<td class="mono mut">{escape(r.to_version)}</td>'
      f"<td>{_tag(r.state, _STATE_CLASS)}</td>"
      f"<td>{_tag(r.verdict, _VERDICT_CLASS)}</td>"
      f"<td>{_tag(r.ci, _CI_CLASS)}</td>"
      f"<td>{_tag(r.post_merge, _POST_MERGE_CLASS)}</td>"
      f"<td>{_pr_link(r)}</td>"
      "</tr>"
      for r in view.bumps
  )
  return (
      '<details class="fold"><summary>Bumps '
      f'<span class="c">{len(view.bumps)}</span></summary><div class="body">'
      '<table class="data"><thead><tr><th>package</th><th>&rarr;</th>'
      "<th>state</th><th>verdict</th><th>ci</th><th>post-merge</th>"
      "<th>pr</th></tr></thead>"
      f"<tbody>{rows}</tbody></table></div></details>"
  )
621 
622 
623def _failures_fold(view: LoopView) -> str:
  if not view.failures:
      return ""
  rows = "".join(
      "<tr>"
      f'<td class="mono">{_short_id(f.workflow_id)}</td>'
      f"<td>{_tag(f.kind, _FAIL_CLASS)}</td>"
      f'<td class="mut">{escape(f.reason or "—")}</td>'
      "</tr>"
      for f in view.failures
  )
  return (
      '<details class="fold"><summary class="bad">Failures '
      f'<span class="c">{len(view.failures)}</span></summary>'
      '<div class="body"><table class="data"><thead><tr><th>bump</th>'
      "<th>kind</th><th>reason</th></tr></thead>"
      f"<tbody>{rows}</tbody></table></div></details>"
  )
641 
642 
643# ── panels ───────────────────────────────────────────────────────────────────
644def _loop_panel(view: LoopView, pid: str, now: datetime) -> str:
  queue = _queue_sec(view)
  return (
      f'<section class="panel" id="{pid}">'
      f'<div class="hero">{_gate_hero(view)}{_class_table(view)}</div>'
      f"{_loop_cards(view)}"
      f"{queue}"
      f"{_cadence(view, now)}"
      f"{_bumps_fold(view)}{_failures_fold(view)}"
      "</section>"
  )
655 
656 
657def _queue_sec(view: LoopView) -> str:
  if not view.gate:
      return (
          f'<div class="sec"><div class="sech">{_icon("inbox")}'
          'Approval queue <span class="n">empty</span></div>'
          '<div class="empty">Nothing awaiting you.</div></div>'
      )
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(r.package)}</td>'
      f'<td class="mono mut">{escape(r.to_version)}</td>'
      f"<td>{_queue_badge(r)}</td>"
      f"<td>{_pr_link(r)}</td>"
      "</tr>"
      for r in view.gate
  )
  return (
      f'<div class="sec"><div class="sech">{_icon("inbox")}Approval queue '
      f'<span class="n">{len(view.gate)} yours</span></div>'
      '<table class="data"><thead><tr><th>package</th><th>&rarr;</th>'
      "<th>gate</th><th>pr</th></tr></thead>"
      f"<tbody>{rows}</tbody></table></div>"
  )
680 
681 
682def _queue_badge(row: BumpRow) -> str:
  if row.would_auto_merge:
      return f'<span class="pill ok">{_dot("ok")}would auto-merge</span>'
  reason = row.held_reason or "held"
  return f'<span class="mut">held &middot; {escape(reason)}</span>'
687 
688 
689def _review_panel(model: DashboardModel, pid: str) -> str:
  r = model.review_record
  live = sum(1 for x in model.review_loops if x.live)
  haz = "bad" if r.hazards else ""
  cards = (
      '<div class="cards">'
      f"{_card(r.repos_covered, 'repos covered')}"
      f"{_card(f'{live}/{len(model.review_loops)}', 'loops live')}"
      f"{_card(r.reviewed, 'reviewed', f'{r.flagged} flagged')}"
      f"{_card(r.hazards, 'hazards', 'transitive', haz)}"
      "</div>"
  )
  body = (
      '<div class="empty">No PRs reviewed yet.</div>'
      if not model.reviews
      else _reviews_table(model.reviews)
  )
  note = (
      '<p class="caption">The transitive ring — advisory. It '
      "re-derives each open PR's reachable determinism hazards and "
      "comments; it never blocks a merge.</p>"
  )
  every = round(model.review_interval_seconds / 60, 1)
  cad = (
      f'<p class="cad">Review loop &middot; <b>{live}</b> live &middot; '
      f"every <b>{every}m</b></p>"
  )
  return (
      f'<section class="panel" id="{pid}">'
      f'<div class="heroh">{_icon("search")}'
      "Determinism review &middot; the transitive ring"
      f"</div>{note}{cards}{cad}"
      f'<div class="sec"><div class="sech">Reviews '
      f'<span class="n">{len(model.reviews)}</span></div>{body}</div>'
      "</section>"
  )
725 
726 
727def _reviews_table(reviews: tuple[ReviewRow, ...]) -> str:
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(row.repo)}</td>'
      f"<td>{_review_pr_link(row)}</td>"
      f"<td>{_findings_cell(row)}</td>"
      f'<td class="mono mut">{escape(", ".join(row.rules) or "—")}</td>'
      "</tr>"
      for row in reviews
  )
  return (
      '<table class="data"><thead><tr><th>repo</th><th>pr</th>'
      "<th>findings</th><th>rules</th></tr></thead>"
      f"<tbody>{rows}</tbody></table>"
  )
742 
743 
744def _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>'
748 
749 
750def _findings_cell(row: ReviewRow) -> str:
  if row.findings == 0:
      return '<span class="ok">clean</span>'
  label = "hazard" if row.findings == 1 else "hazards"
  link = ""
  if row.comment_url:
      link = f' <a href="{escape(row.comment_url, quote=True)}">comment</a>'
  return f'<span class="bad">{row.findings} {label}</span>{link}'
758 
759 
760def _a11y_panel(model: DashboardModel, pid: str) -> str:
  r = model.a11y_record
  live = sum(1 for x in model.a11y_loops if x.live)
  iss = "bad" if r.issues else ""
  cards = (
      '<div class="cards">'
      f"{_card(r.repos_covered, 'repos covered')}"
      f"{_card(f'{live}/{len(model.a11y_loops)}', 'loops live')}"
      f"{_card(r.reviewed, 'reviewed', f'{r.flagged} flagged')}"
      f"{_card(r.issues, 'gaps', 'source-level', iss)}"
      "</div>"
  )
  body = (
      '<div class="empty">No PRs reviewed yet.</div>'
      if not model.a11y_reviews
      else _a11y_table(model.a11y_reviews)
  )
  note = (
      '<p class="caption">Source-level a11y — advisory. It re-derives each '
      "open PR's changed Vue/JSX templates for accessibility gaps and "
      "comments; it never blocks a merge.</p>"
  )
  every = round(model.a11y_interval_seconds / 60, 1)
  cad = (
      f'<p class="cad">A11y loop &middot; <b>{live}</b> live &middot; '
      f"every <b>{every}m</b></p>"
  )
  return (
      f'<section class="panel" id="{pid}">'
      f'<div class="heroh">{_icon("accessibility")}'
      "A11y review &middot; source-level gaps"
      f"</div>{note}{cards}{cad}"
      f'<div class="sec"><div class="sech">Reviews '
      f'<span class="n">{len(model.a11y_reviews)}</span></div>{body}</div>'
      "</section>"
  )
796 
797 
798def _a11y_table(reviews: tuple[A11yRow, ...]) -> str:
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(row.repo)}</td>'
      f"<td>{_a11y_pr_link(row)}</td>"
      f"<td>{_a11y_findings_cell(row)}</td>"
      f'<td class="mono mut">{escape(", ".join(row.kinds) or "—")}</td>'
      "</tr>"
      for row in reviews
  )
  return (
      '<table class="data"><thead><tr><th>repo</th><th>pr</th>'
      "<th>findings</th><th>kinds</th></tr></thead>"
      f"<tbody>{rows}</tbody></table>"
  )
813 
814 
815def _a11y_pr_link(row: A11yRow) -> 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>'
819 
820 
821def _a11y_findings_cell(row: A11yRow) -> str:
  if row.findings == 0:
      return '<span class="ok">clean</span>'
  label = "gap" if row.findings == 1 else "gaps"
  link = ""
  if row.comment_url:
      link = f' <a href="{escape(row.comment_url, quote=True)}">comment</a>'
  return f'<span class="bad">{row.findings} {label}</span>{link}'
829 
830 
831def _telemetry_panel(model: DashboardModel, pid: str, now: datetime) -> str:
  tel: RunTelemetry = model.telemetry
  if not tel.available:
      body = (
          '<div class="empty">Unavailable &mdash; ClickHouse off or no '
          "froot traces in the window.</div>"
      )
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(a.name)}</td>'
          f'<td class="r">{a.count}</td>'
          f'<td class="r">{a.avg_ms:.0f} ms</td>'
          f'<td class="r mut">{a.max_ms:.0f} ms</td>'
          "</tr>"
          for a in tel.activities
      )
      last = f"last {_ago(tel.last_activity, now)}"
      err = "bad" if tel.error_spans else ""
      head = (
          '<div class="cards">'
          f"{_card(tel.total_spans, 'spans', last)}"
          f"{_card(tel.error_spans, 'errors', 'in the window', err)}"
          f"{_card(f'{tel.window_days}d', 'window')}</div>"
      )
      table = (
          '<div class="sec"><div class="sech">Activity latency</div>'
          '<table class="data"><thead><tr><th>activity</th><th>runs</th>'
          "<th>avg</th><th>max</th></tr></thead>"
          f"<tbody>{rows}</tbody></table></div>"
      )
      body = head + table
  note = (
      '<p class="caption">Cross-cutting run telemetry from ClickHouse '
      "(trace-derived, best-effort) — latency per activity across "
      "every loop.</p>"
  )
  return (
      f'<section class="panel" id="{pid}">'
      f'<div class="heroh">{_icon("activity")}Run telemetry '
      "&middot; ClickHouse</div>"
      f"{note}{body}</section>"
  )
874 
875 
876# ── tabs + page ──────────────────────────────────────────────────────────────
877def _loop_badge(view: LoopView) -> str:
  earned = sum(1 for g in view.class_gates if g.earned)
  if view.class_gates and earned:
      return f"{earned}/{len(view.class_gates)} earned"
  if view.track_record.open_now:
      return f"{view.track_record.open_now} open"
  return str(view.track_record.opened)
884 
885 
886def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  now = model.generated_at
  # (tab-id, panel-id, icon, label, badge, panel-html)
  tabs: list[tuple[str, str, str, str, str, str]] = []
  for i, view in enumerate(model.bump_loops):
      pid, tid = f"panel-{i}", f"tab-{i}"
      icon = view.icon
      tabs.append(
          (
              tid,
⋯ 84 lines hidden (lines 897–980)897                pid,
              icon,
              view.title,
              _loop_badge(view),
              _loop_panel(view, pid, now),
          )
      )
  tabs.append(
      (
          "tab-det",
          "panel-det",
          "search",
          "Determinism review",
          str(model.review_record.reviewed),
          _review_panel(model, "panel-det"),
      )
  )
  tabs.append(
      (
          "tab-a11y",
          "panel-a11y",
          "accessibility",
          "A11y review",
          str(model.a11y_record.reviewed),
          _a11y_panel(model, "panel-a11y"),
      )
  )
  tabs.append(
      (
          "tab-tel",
          "panel-tel",
          "activity",
          "Telemetry",
          "live" if model.telemetry.available else "off",
          _telemetry_panel(model, "panel-tel", now),
      )
  )
934 
  inputs, labels, panels, rules = [], [], [], []
  for idx, (tid, pid, icon, label, badge, panel) in enumerate(tabs):
      checked = " checked" if idx == 0 else ""
      inputs.append(
          f'<input class="tabin" type="radio" name="tab" id="{tid}"{checked}>'
      )
      # Telemetry is cross-cutting, not a loop — float it to the far right.
      lcls = ' class="aside"' if tid == "tab-tel" else ""
      labels.append(
          f'<label{lcls} for="{tid}">{_icon(icon)}{escape(label)}'
          f'<span class="badge">{escape(badge)}</span></label>'
      )
      panels.append(panel)
      rules.append(f"#{tid}:checked~main #{pid}{{display:block}}")
      rules.append(
          f"#{tid}:checked~main nav.tabbar label[for={tid}]"
          "{color:var(--fg);border-bottom-color:var(--accent)}"
      )
  tabcss = "".join(rules)
  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}{tabcss}</style>"
      f"<script>{_THEME_JS}</script></head><body>"
      + "".join(inputs)
      + "<main>"
      + _header(model)
      + f'<nav class="tabbar">{"".join(labels)}</nav>'
      + "".join(panels)
      + _footer()
      + "</main></body></html>"
  )
968 
969 
970def _footer() -> str:
  return (
      "<footer><b>Authority envelope.</b> froot opens PRs everywhere and "
      "auto-merges only on an allowlisted repo where a class has earned its "
      "gate (the allowlist is empty by default, so commit authority is none "
      "until a steward opts in). Trust is earned, narrow to one loop on one "
      "repo, conditional on its environment "
      '(<span class="mono">gemma4:12b</span>), revocable, and time-expiring. '
      "Everything here is derived per request from GitHub + Temporal + "
      "ClickHouse; froot keeps no database.</footer>"
  )

The PR-title verb lives in the spec

The title verb (deps / security / dead-code) is a per-loop label like the icon, so it belongs in the spec too. But the draft builder is pure policy — it can't read the registry. So the impure open-PR activity resolves spec.title_prefix and passes it in; compose stays a pure function of its inputs.

A required keyword-only title_prefix; compose no longer knows about loops.

src/froot/policy/compose.py · 191 lines

src/froot/policy/compose.py191 lines · Python

⋯ 85 lines hidden (lines 1–85)1"""Compose the PR content and the PR labels — pure, no model call.
2 
3Spine-heavy: the PR title/body are deterministic templates over the candidate
4and the model's changelog verdict (the model already did its one job — the
5verdict — so rendering the text costs no model round-trip). froot tags every PR
6with one fixed pair of labels; the per-run changelog/CI signal lives in the
7durable workflow history (and the structured outcome log), not as accumulating
8labels on the PR.
9"""
10 
11from __future__ import annotations
12 
13from typing import TYPE_CHECKING, assert_never
14 
15from froot.domain.changelog import CleanVerdict, RiskyVerdict, UnknownVerdict
16from froot.domain.ecosystem import (
17    Ecosystem,
18    lockfile_filename,
19    manifest_filename,
20)
21from froot.domain.loop import Loop
22from froot.domain.pull_request import PullRequestDraft
23from froot.policy.naming import branch_name
24 
25if TYPE_CHECKING:
26    from froot.domain.candidate import Candidate
27    from froot.domain.changelog import ChangelogVerdict
28    from froot.domain.removal import Removal
29    from froot.domain.repo import TargetRepo
30 
31_LABEL_NAMESPACE = "froot"
32 
33 
34def pr_labels(loop: Loop = Loop.DEPENDENCY_PATCH) -> tuple[str, str]:
35    """The fixed labels for a loop's PRs: ``(froot, <loop>)``.
36 
37    Deliberately just these two — they mark the PR as froot's work for *this*
38    loop, and nothing more. How the proposal fared (the changelog verdict, the
39    CI result) is recorded durably in the workflow history, not layered onto the
40    PR as labels that pile up across re-runs. The loop label also keeps the two
41    loops' PRs distinguishable to a human filtering the repo.
42    """
43    return (_LABEL_NAMESPACE, loop.value)
44 
45 
46# Tags the comment froot leaves when it closes one of its own PRs (red CI, or a
47# reconcile sweep). The marker lets the close go through the idempotent
48# upsert_issue_comment path, so a retried close edits its note in place instead
49# of stacking a second one.
50CLOSE_MARKER = "<!-- froot:closed -->"
51 
52 
53def _changed_files(ecosystem: Ecosystem) -> str:
54    """Describe which files a bump rewrites, for the PR body.
55 
56    The phrase must match the diff the human approver actually sees. npm
57    rewrites both the manifest and the lockfile (``npm install
58    --package-lock-only`` updates the dependency spec too); uv rewrites only the
59    lockfile, because a patch stays within the existing ``pyproject.toml``
60    constraint, so the manifest is left untouched.
61    """
62    match ecosystem:
63        case Ecosystem.NPM:
64            return f"{manifest_filename(ecosystem)} + lockfile"
65        case Ecosystem.UV:
66            return (
67                f"{lockfile_filename(ecosystem)} only; "
68                f"{manifest_filename(ecosystem)} unchanged"
69            )
70    assert_never(ecosystem)
71 
72 
73def _verdict_summary(verdict: ChangelogVerdict) -> str:
74    """Render the model's changelog framing for the human reviewer."""
75    match verdict:
76        case CleanVerdict():
77            return f"Changelog reads clean. {verdict.rationale}"
78        case RiskyVerdict():
79            concerns = "".join(f"\n- {concern}" for concern in verdict.concerns)
80            return f"Review carefully. {verdict.rationale}{concerns}"
81        case UnknownVerdict():
82            return f"Changelog unavailable. {verdict.rationale}"
83    assert_never(verdict)
84 
85 
86def pull_request_draft(
87    target: TargetRepo,
88    candidate: Candidate,
89    verdict: ChangelogVerdict,
90    loop: Loop = Loop.DEPENDENCY_PATCH,
91    *,
92    title_prefix: str,
93) -> PullRequestDraft:
⋯ 26 lines hidden (lines 94–119)94    """Build the deterministic PR content for a bump (no model call).
95 
96    Args:
97        target: The repo the PR is opened against (gives the base branch).
98        candidate: The bump being proposed.
99        verdict: The model's changelog framing, surfaced for the reviewer.
100        loop: Which loop is proposing — sets the branch namespace and (via
101            ``candidate.justification``) the body's "why".
102        title_prefix: The PR-title verb (``deps`` / ``security`` …), from spec.
103 
104    Returns:
105        A :class:`PullRequestDraft` ready for the forge to open.
106    """
107    lines = [
108        f"Bumps `{candidate.package}` from {candidate.current} to "
109        f"{candidate.target} ({_changed_files(candidate.ecosystem)}).",
110    ]
111    if candidate.justification is not None:
112        lines += ["", candidate.justification]
113    lines += [
114        "",
115        _verdict_summary(verdict),
116        "",
117        "---",
118        "Opened by froot. froot does not merge; a human approves.",
119    ]
120    return PullRequestDraft(
121        branch=branch_name(candidate, loop),
122        base=target.default_branch,
123        title=(
124            f"{title_prefix}: bump {candidate.package} to {candidate.target}"
125        ),
126        body="\n".join(lines),
127    )
⋯ 64 lines hidden (lines 128–191)128 
129 
130def removal_pull_request_draft(
131    target: TargetRepo,
132    removal: Removal,
133    loop: Loop = Loop.DEPENDENCY_PATCH,
134    *,
135    title_prefix: str,
136) -> PullRequestDraft:
137    """Build the deterministic PR content for a removal (no model call).
138 
139    A removal carries no changelog verdict (there is no new version to read);
140    its "why" is the ``justification`` the safe-to-remove veto recorded, so that
141    is what the body surfaces. The action rewrites the manifest and lockfile
142    (``npm uninstall`` edits both), which the body states so the diff matches.
143 
144    Args:
145        target: The repo the PR is opened against (gives the base branch).
146        removal: The unused dependency being removed.
147        loop: Which loop is proposing — sets the branch namespace.
148        title_prefix: The PR-title verb (``dead-code`` …), from the loop's spec.
149 
150    Returns:
151        A :class:`PullRequestDraft` ready for the forge to open.
152    """
153    section = "dev dependency" if removal.dev else "dependency"
154    lines = [
155        f"Removes the unused {section} `{removal.package}` "
156        f"from the manifest and lockfile.",
157    ]
158    if removal.justification is not None:
159        lines += ["", removal.justification]
160    lines += [
161        "",
162        "---",
163        "Opened by froot. froot does not merge; a human approves.",
164    ]
165    return PullRequestDraft(
166        branch=branch_name(removal, loop),
167        base=target.default_branch,
168        title=f"{title_prefix}: remove {removal.package} (unused)",
169        body="\n".join(lines),
170    )
171 
172 
173def closed_on_red_comment(failing: tuple[str, ...]) -> str:
174    """The note froot leaves when it closes a PR for failing CI.
175 
176    Names the failing checks (when GitHub reported them) so the human sees why
177    without opening the checks tab, and states froot's contract: it will
178    re-propose the same bump if a newer patch is published. Carries
179    :data:`CLOSE_MARKER` so the close posts through the idempotent comment path.
180    """
181    checks = ", ".join(f"`{name}`" for name in failing)
182    reason = f"CI did not pass ({checks})." if checks else "CI did not pass."
183    return "\n".join(
184        (
185            CLOSE_MARKER,
186            f"froot closed this PR: {reason}",
187            "",
188            "The change wasn't safe to merge, so froot won't leave it open. "
189            "If a newer patch is published, froot will propose it fresh.",
190        )
191    )

The activity resolves the verb from the spec and hands it to the pure builder.

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

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

⋯ 92 lines hidden (lines 1–92)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.a11y import A11yAnalysis, A11yVerdict
22from froot.domain.candidate import Candidate
23from froot.domain.changelog import (
  ChangelogVerdict,
  CleanVerdict,
  UnknownVerdict,
27)
28from froot.domain.ci import CIStatus
29from froot.domain.determinism import AnalysisResult, FrontierVerdict
30from froot.domain.pull_request import PullRequestDraft, PullRequestRef
31from froot.domain.removal import Removal
32from froot.domain.repo import TargetRepo
33from froot.domain.work import WorkItem
34from froot.policy.a11y_comment import (
  A11Y_MARKER,
  render_a11y_comment,
  should_post,
38)
39from froot.policy.a11y_scan import scan_sources
40from froot.policy.compose import (
  pr_labels,
  pull_request_draft,
  removal_pull_request_draft,
44)
45from froot.policy.determinism import analyze_workflow_surface
46from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_a11y_review_workflow_id,
  pr_review_workflow_id,
51)
52from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
53from froot.workflow.types import (
  AdjudicateA11yInput,
  AdjudicateInput,
  AutoMergeInput,
  CiCheckInput,
  CloseInput,
  DispatchA11yInput,
  DispatchInput,
  DispatchReviewInput,
  GateReviewInput,
  GateSelfTestInput,
  JudgeInput,
  MergeInput,
  OpenPrInput,
  PostA11yInput,
  PostReviewInput,
  PrA11yReviewParams,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
74)
75 
76if TYPE_CHECKING:
  from froot.domain.loop import Loop
  from froot.ports.protocols import PackageManager
79 
80_log = logging.getLogger("froot.outcome")
81_review_log = logging.getLogger("froot.review")
82_a11y_log = logging.getLogger("froot.a11y")
83_reconcile_log = logging.getLogger("froot.reconcile")
84_scan_log = logging.getLogger("froot.scan")
85_gate_log = logging.getLogger("froot.gate")
86 
87 
88def _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
91 
92 
93def _draft_for(params: OpenPrInput) -> PullRequestDraft:
  """Build the PR draft for the work item — bump vs removal compose.
95 
  The PR-title verb is the loop's, resolved from its registered spec here (the
  impure boundary) and passed into the pure draft builders.
  """
  from froot.loops import registry
100 
  item = params.candidate
  title_prefix = registry.get(params.loop).title_prefix
  match item:
      case Candidate():
          return pull_request_draft(
              params.target,
              item,
              params.verdict,
              params.loop,
              title_prefix=title_prefix,
          )
      case Removal():
          return removal_pull_request_draft(
              params.target, item, params.loop, title_prefix=title_prefix
          )
⋯ 710 lines hidden (lines 116–825)116    assert_never(item)
117 
118 
119async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
124) -> tuple[int, tuple[WorkItem, ...]]:
  """Gather this loop's signal from the checkout and select its work items.
126 
  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;
  dead-code reads the unused dependencies a static analyzer flags and vetoes
  each with the safe-to-remove judge. The impure sources are lazy-imported per
  arm so none drags another's stack into a sandbox. Each feeds a pure (or, for
  dead-code, model-vetoed) selection.
134 
  Returns ``(considered, items)`` — ``considered`` is the size of the upstream
  signal (available upgrades / advisories found / unused deps flagged) so the
  scan can make its selectivity legible (how much was seen versus kept).
138 
  The one genuinely per-loop body now lives in each loop's spec ``observe``
  (see :mod:`froot.loops`); the spine looks the loop up and runs it, so this
  selection seam needs no per-loop arm.
  """
  from froot.loops import registry
144 
  return await registry.get(loop).observe(
      target, package_manager, manifest_dir
  )
148 
149 
150@activity.defn
151async def scan_candidates(
  params: ScanCandidatesInput,
153) -> tuple[WorkItem, ...]:
  """Check out the repo and select this loop's candidates.
155 
  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
164 
  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
198 
199 
200@activity.defn
201async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
203 
  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
215 
  candidate = params.candidate
  if isinstance(candidate, Removal):
      # A removal already cleared the safe-to-remove veto at scan; there is no
      # changelog to read, so carry that conclusion straight into the PR
      # framing (the rationale the veto recorded on the work item).
      return CleanVerdict(
          rationale=candidate.justification or "unused; safe to remove"
      )
  changelog = await HttpChangelogSource().fetch(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__})."
      )
238 
239 
240@activity.defn
241async def gate_review(params: GateReviewInput) -> ChangelogVerdict:
  """Independently deep-review a bump at the gate; fail-closed to a hold.
243 
  The fourth trust leg (§3.7): a second, adversarial model pass over the
  changelog, run only when a bump is about to auto-merge. ``clean`` approves
  the merge; ``risky``/``unknown`` hold the PR for the human. Fail-CLOSED — a
  missing changelog or a model error returns a non-clean verdict, so an
  unreviewable bump never merges unattended. This is the opposite disposition
  from :func:`judge_changelog` (which degrades-to-proceed): here the safe
  direction is to hold, and a non-clean verdict already means hold.
  """
  from froot.adapters.changelog_http import HttpChangelogSource
  from froot.adapters.model_judge import PydanticAiJudge
254 
  candidate = params.candidate
  verdict: ChangelogVerdict
  if isinstance(candidate, Removal):
      # The fourth leg for a removal: independently re-judge that it is safe
      # to remove (the same check the scan veto ran). No changelog to read;
      # fail-CLOSED to a hold on a model error, like the bump path.
      try:
          verdict = await PydanticAiJudge().judge_removal(candidate)
      except Exception as exc:
          activity.logger.warning(
              "safe-to-remove gate review unavailable for %s; holding: %r",
              candidate.package,
              exc,
          )
          verdict = UnknownVerdict(
              rationale=(
                  f"Removal reviewer unavailable ({type(exc).__name__})."
              )
          )
  else:
      changelog = await HttpChangelogSource().fetch(candidate)
      if changelog is None:
          verdict = UnknownVerdict(
              rationale="No changelog to review; holding (fail-closed)."
          )
      else:
          try:
              verdict = await PydanticAiJudge().gate_review(
                  changelog, params.loop
              )
          except Exception as exc:
              activity.logger.warning(
                  "gate reviewer unavailable for %s; holding: %r",
                  candidate.package,
                  exc,
              )
              verdict = UnknownVerdict(
                  rationale=(
                      f"Gate reviewer unavailable ({type(exc).__name__})."
                  )
              )
  _gate_log.info(
      json.dumps(
          {
              "event": "gate_review",
              "loop": params.loop.value,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
              "package": params.candidate.package,
              "verdict": verdict.kind,
              "approved": verdict.kind == "clean",
          }
      )
  )
  return verdict
310 
311 
312@activity.defn
313async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Rewrite the manifest+lockfile and open (idempotently) the work item's PR.
315 
  The one place the action differs by work-item kind: a bump regenerates the
  lockfile at the target version, a removal deletes the dependency. Both are
  lockfile-only (no install, no scripts); everything around them — the dedup
  branch, the checkout, the push, the open — is the same chassis.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
323 
  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 = _draft_for(params)
  item = params.candidate
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      manifest_dir = _manifest_dir(params.target, workspace)
      match item:
          case Candidate():
              await package_manager.apply_patch_bump(item, manifest_dir)
          case Removal():
              await package_manager.remove_dependency(item, manifest_dir)
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
343 
344 
345@activity.defn
346async 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
349 
  return await GitHubForge().ci_status(params.target, params.head_sha)
351 
352 
353@activity.defn
354async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update.
356 
  The labels carry the loop *and* the judgment environment (the judge model)
  the PR was opened under, so the gate can count only the track record earned
  under the current environment and reset it when the model changes (§3.7).
  """
  from froot.adapters.github import GitHubForge
  from froot.config.settings import ModelSettings
  from froot.policy.environment import env_label
364 
  outcome = params.outcome
  item = outcome.candidate
  labels = (
      *pr_labels(params.loop),
      env_label(ModelSettings().ollama_model),
  )
  await GitHubForge().add_labels(params.target, outcome.pr.number, labels)
  record = {
      "event": "loop_outcome",
      "loop": params.loop.value,
      "repo": params.target.repo.slug,
      "package": item.package,
      "changelog": outcome.verdict.kind,
      "ci": outcome.ci.kind,
      "ci_passed": outcome.ci_passed,
      "pr": outcome.pr.number,
      "pr_url": outcome.pr.url,
  }
  match item:
      case Candidate():
          record |= {
              "action": "bump",
              "from": str(item.current),
              "to": str(item.target),
          }
      case Removal():
          record |= {"action": "remove", "dev": item.dev}
  _log.info(json.dumps(record))
393 
394 
395@activity.defn
396async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
398 
  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
405 
  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
410 
  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
429 
430 
431@activity.defn
432async def auto_merge_eligible(params: AutoMergeInput) -> bool:
  """Whether this (repo, loop) class has earned the auto-merge grant.
434 
  Short-circuits to ``False`` for any repo a steward has not allowlisted (the
  default), so the common case costs nothing. Otherwise it re-derives the
  class's standing from the live GitHub history — the same triangulated,
  windowed, environment-scoped computation the dashboard's shadow gate shows
  (``read_model.earned_now``) — so the acting gate and the advisory panel can
  never disagree. Best-effort: any read failure degrades to ``False`` (hold),
  never an auto-merge.
  """
  from datetime import UTC, datetime
444 
  from froot.config.settings import AutonomySettings, ModelSettings
  from froot.dashboard import github_source, read_model
  from froot.policy.environment import environment_slug
448 
  policy = AutonomySettings().policy()
  repo = params.target.repo.slug
  if repo not in policy.allowlisted_repos:
      return False
  now = datetime.now(UTC)
  prs, prs_error = await github_source.fetch((repo,))
  if prs_error is not None:
      return False  # can't confirm the record -> hold, never merge blind
  outcomes, _ = await github_source.fetch_outcomes(
      (repo,), prs, now=now, window_days=policy.window_days
  )
  return read_model.earned_now(
      now,
      prs,
      outcomes,
      repo,
      params.loop,
      policy,
      environment_slug(ModelSettings().ollama_model),
  )
469 
470 
471@activity.defn
472async def merge_pull_request(params: MergeInput) -> None:
  """Auto-merge an earned, clean+green bump's PR (the acting gate's write).
474 
  Reached only after the pure machine confirmed clean+green and the class
  earned the grant on an allowlisted repo. Passes the head SHA so GitHub
  refuses the merge if the head moved since the gate decided.
  """
  from froot.adapters.github import GitHubForge
480 
  await GitHubForge().merge_pull_request(
      params.target, params.pr.number, head_sha=params.pr.head_sha
  )
  _log.info(
      json.dumps(
          {
              "event": "pr_merged",
              "loop": params.loop.value,
              "reason": "auto_merge",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "pr_url": params.pr.url,
          }
      )
  )
496 
497 
498@activity.defn
499async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
501 
  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
509 
  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),
          }
      )
  )
531 
532 
533@activity.defn
534async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
536 
  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
549 
  if not BehaviorSettings().reconcile:
      return 0
  # Reconcile is version-supersession cleanup, which only bump loops have: a
  # removal carries no version to be overtaken. A loop that does not reconcile
  # (dead-code) is skipped rather than re-running its signal (knip + the veto
  # judge) every tick only to close nothing. (A removal-specific reconcile —
  # close when no longer unused — is future work.) The trait is on the spec.
  from froot.loops import registry
558 
  if not registry.get(params.loop).reconciles:
      return 0
561 
  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)
593 
594 
595@activity.defn
596async def gate_selftest(params: GateSelfTestInput) -> tuple[str, ...]:
  """Run the adversarial gate probe against the live policy; alarm on escape.
598 
  The §2.11 deliberate disturbance for the acting gate: a battery of synthetic
  known-bad class histories a healthy gate must refuse, scored against the
  policy froot is *actually running* (config and all). Any escape — a bad
  class the live gate would grant — is logged at ERROR (the alarm) so it
  surfaces in telemetry the moment config drifts; a clean pass logs a
  heartbeat at INFO. Returns the escaped scenario names. Pure compute,
  repo-independent (the ``target``/``loop`` are only the log's context).
  """
  from froot.config.settings import AutonomySettings
  from froot.policy.gate_probe import gate_escapes
609 
  escaped = gate_escapes(AutonomySettings().policy())
  record = json.dumps(
      {
          "event": "gate_selftest",
          "loop": params.loop.value,
          "repo": params.target.repo.slug,
          "healthy": not escaped,
          "escaped": list(escaped),
      }
  )
  if escaped:
      _gate_log.error(record)  # an alarm: the gate would trust a bad class
  else:
      _gate_log.info(record)
  return escaped
625 
626 
627# ── The determinism reviewer (the transitive ring) ──────────────────────────
628@activity.defn
629async 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
632 
  return await GitHubForge().list_open_pull_requests(target)
634 
635 
636@activity.defn
637async 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
641 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
644 
  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
660 
661 
662@activity.defn
663async 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
668 
  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)
679 
680 
681@activity.defn
682async def adjudicate_frontier(
  params: AdjudicateInput,
684) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
687 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
693 
694 
695@activity.defn
696async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
699 
  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
721 
722 
723# ── The a11y reviewer (the source-level design-system ring) ──────────────────
724@activity.defn
725async def scan_pr_a11y(params: PrA11yReviewParams) -> A11yAnalysis:
  """Check out the PR head and scan its changed templates for a11y risks.
727 
  Scoped to the PR's changed Vue/JSX templates (shift-left: review what came
  in), so the advisory stays bounded and high-signal. The source lines are
  read into memory in the activity, so the pure scan is unaffected by the
  workspace being cleaned up.
  """
  from froot.adapters.github import GitHubForge
  from froot.adapters.web_source import load_web_sources
735 
  forge = GitHubForge()
  changed = await forge.list_pull_request_files(
      params.target, params.pr.number
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout_pull_request(
          params.target, workspace, params.pr.number
      )
      sources = load_web_sources(workspace, changed)
      candidates = scan_sources(sources)
  return A11yAnalysis(candidates=candidates, scanned_files=len(sources))
748 
749 
750@activity.defn
751async def adjudicate_a11y(
  params: AdjudicateA11yInput,
753) -> tuple[A11yVerdict, ...]:
  """Run the model over each flagged candidate; return aligned verdicts."""
  from froot.adapters.a11y_judge import A11ySourceJudge
756 
  judge = A11ySourceJudge()
  verdicts: list[A11yVerdict] = []
  for candidate in params.candidates:
      verdicts.append(await judge.adjudicate(candidate))
  return tuple(verdicts)
762 
763 
764@activity.defn
765async def post_a11y_review(params: PostA11yInput) -> str | None:
  """Upsert (or clear) the advisory comment; log the ledger row.
767 
  Posts when there are findings, or when a prior comment must be cleared to
  "all clear" (true decay — a PR whose gaps were fixed never keeps a stale
  finding list, the bug the determinism reviewer leaves). A clean PR with
  no prior comment stays silent, so a clean PR is never spammed.
  """
  from froot.adapters.github import GitHubForge
774 
  forge = GitHubForge()
  exists = await forge.find_marked_comment(
      params.target, params.pr.number, A11Y_MARKER
  )
  url: str | None = None
  if should_post(has_findings=bool(params.findings), comment_exists=exists):
      body = render_a11y_comment(params.findings, params.pr.head_sha)
      url = await forge.upsert_issue_comment(
          params.target, params.pr.number, A11Y_MARKER, body
      )
  _a11y_log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": "a11y-review",
              "repo": params.target.repo.slug,
              "pr": params.pr.number,
              "head_sha": params.pr.head_sha,
              "findings": len(params.findings),
              "kinds": sorted({f.kind for f in params.findings}),
              "comment_url": url,
          }
      )
  )
  return url
800 
801 
802@activity.defn
803async def dispatch_pr_a11y_review(params: DispatchA11yInput) -> None:
  """Start a PR's a11y review (idempotent per PR + head SHA)."""
  from temporalio.common import WorkflowIDReusePolicy
  from temporalio.exceptions import WorkflowAlreadyStartedError
807 
  from froot.workflow.pr_a11y_review_workflow import PrA11yReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
810 
  temporal = await client()
  try:
      await temporal.start_workflow(
          PrA11yReviewWorkflow.run,
          PrA11yReviewParams(target=params.target, pr=params.pr),
          id=pr_a11y_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 an a11y review — a no-op, so
      # re-polling never double-reviews the same commit.
      return

Tests: zero behavior change, single-sourced seams

Behavior equivalence is carried by the existing suite: the scan, judge, reconcile, and dashboard tests now run through the registry and still pass — including test_reconcile_skips_dead_code (the skip happens without re-scanning) and test_scan_candidates_dead_code_vetoes_unsafe_removals (the veto-at-signal). The new registry tests guard the seams the spec now owns.

Disposition, the title verb, judge-context placement, the reconcile trait.

tests/test_loop_registry.py · 54 lines

tests/test_loop_registry.py54 lines · Python

⋯ 19 lines hidden (lines 1–19)1"""The loop registry — the spine reads it instead of branching on the enum.
2 
3These guard the per-loop seams the registry owns: the disposition, the
4changelog-judge framing, the reconcile trait, the PR-title verb, and the
5dashboard icon — so a loop's identity stays single-sourced as loops multiply.
6Behavior equivalence of each loop's ``observe`` is covered by the existing scan
7tests, which now route through the registry.
8"""
9 
10from __future__ import annotations
11 
12from froot.adapters.model_judge import _loop_context
13from froot.domain.loop import Loop
14from froot.loops import registry
15from froot.loops.registry import Disposition
16 
17_ACTING = (Loop.DEPENDENCY_PATCH, Loop.SECURITY_PATCH, Loop.DEAD_CODE)
18 
19 
20def test_acting_loops_registered_as_commit_or_revert() -> None:
21    specs = {spec.loop: spec for spec in registry.all_specs()}
22    for loop in _ACTING:
23        assert specs[loop].disposition is Disposition.COMMIT_OR_REVERT
24 
25 
26def test_title_prefix_is_the_per_loop_pr_verb() -> None:
27    # The PR-title verb is a per-loop label single-sourced in the spec (not
28    # derivable from the loop name), so a new loop carries its own.
29    assert registry.get(Loop.DEPENDENCY_PATCH).title_prefix == "deps"
30    assert registry.get(Loop.SECURITY_PATCH).title_prefix == "security"
31    assert registry.get(Loop.DEAD_CODE).title_prefix == "dead-code"
32 
33 
34def test_judge_context_present_for_changelog_loops_absent_for_dead_code() -> (
35    None
36):
37    # Dependency- and security-patch judge a changelog in-loop (a framing line);
38    # dead-code judges at the signal (a removal veto), so it carries no context.
39    assert registry.get(Loop.DEPENDENCY_PATCH).judge_context is not None
40    assert registry.get(Loop.SECURITY_PATCH).judge_context is not None
41    assert registry.get(Loop.DEAD_CODE).judge_context is None
42 
43 
44def test_loop_context_reads_the_registry() -> None:
45    for loop in (Loop.DEPENDENCY_PATCH, Loop.SECURITY_PATCH):
46        assert _loop_context(loop) == registry.get(loop).judge_context
47 
48 
49def test_reconciles_true_for_bump_loops_false_for_dead_code() -> None:
50    # Version-supersession reconcile only applies to version-bearing bumps; a
51    # removal has no version to be overtaken, so dead-code declares it skips.
52    assert registry.get(Loop.DEPENDENCY_PATCH).reconciles is True
53    assert registry.get(Loop.SECURITY_PATCH).reconciles is True
54    assert registry.get(Loop.DEAD_CODE).reconciles is False

The discriminator: a spec icon change flows to the rendered tab, no renderer edit.

tests/test_dashboard_render.py · 544 lines

tests/test_dashboard_render.py544 lines · Python

⋯ 150 lines hidden (lines 1–150)1from __future__ import annotations
2 
3from collections.abc import Sequence
4from datetime import UTC, datetime
5 
6from froot.dashboard import read_model, render
7from froot.dashboard.github_source import GithubPr
8from froot.dashboard.model import ActivityStat, DashboardModel, RunTelemetry
9from froot.dashboard.temporal_source import (
  A11yExecution,
  BumpExecution,
  PrA11yExecution,
  PrReviewExecution,
  ReviewExecution,
  ScanExecution,
16)
17from froot.domain.loop import Loop
18from froot.policy.autonomy import AutonomyPolicy
19 
20NOW = datetime(2026, 6, 3, 12, 0, tzinfo=UTC)
21REPO = "mseeks/revisionist"
22 
23 
24def _model(
  prs: Sequence[GithubPr] = (),
  scans: Sequence[ScanExecution] = (),
  telemetry: tuple[RunTelemetry, str | None] | None = None,
  reviews: Sequence[ReviewExecution] = (),
  pr_reviews: Sequence[PrReviewExecution] = (),
  a11y_reviews: Sequence[A11yExecution] = (),
  pr_a11y_reviews: Sequence[PrA11yExecution] = (),
32) -> DashboardModel:
  if telemetry is None:
      telemetry = (
          RunTelemetry(
              available=False,
              total_spans=0,
              error_spans=0,
              last_activity=None,
              window_days=3,
              activities=(),
          ),
          "off",
      )
  return read_model.assemble(
      now=NOW,
      repos=(REPO,),
      scan_interval_seconds=86_400,
      review_interval_seconds=300,
      a11y_interval_seconds=300,
      github=(tuple(prs), None),
      temporal=(
          (
              tuple(scans),
              (),
              tuple(reviews),
              tuple(pr_reviews),
              tuple(a11y_reviews),
              tuple(pr_a11y_reviews),
          ),
          None,
      ),
      telemetry=telemetry,
  )
65 
66 
67def _pr(number: int, package: str, state: str, **kw) -> GithubPr:
  return GithubPr(
      repo=REPO,
      number=number,
      url=f"https://github.com/{REPO}/pull/{number}",
      package=package,
      from_version=kw.get("from_version"),
      to_version=kw.get("to_version", "1.0.0"),
      verdict=kw.get("verdict"),
      state=state,
      opened_at=kw.get("opened", NOW),
      merged_at=kw.get("merged"),
  )
80 
81 
82# ── the page shell: self-contained, no JS, tabbed ────────────────────────────
83def test_page_is_a_self_contained_html_document():
  html = render.page(_model())
  assert html.startswith("<!doctype html>")
  assert html.rstrip().endswith("</html>")
  assert "http://" not in html and "https://" not in html  # no links
  assert "<script src" not in html.lower()  # no external JS; theme is inline
89 
90 
91def test_page_is_tabbed_one_per_loop_plus_determinism_and_telemetry():
  html = render.page(_model())
  assert '<nav class="tabbar">' in html
  # CSS-only tabs: hidden radio inputs drive the panels (no external JS).
  assert 'type="radio"' in html and "<script src" not in html.lower()
  for label in (
      "Dependency-patch",
      "Determinism review",
      "A11y review",
      "Telemetry",
  ):
      assert label in html
  # the footer's authority envelope, trimmed of the word-bomb
  assert "Authority envelope" in html
  assert "froot" in html
106 
107 
108def test_dead_code_loop_renders_with_scissors_and_unused():
  # The dead-code tab gets the scissors icon and its removal rows read
  # "<pkg> -> unused", not the bump-shaped "<pkg> -> <version>".
  rm = GithubPr(
      repo=REPO,
      number=5,
      url=f"https://github.com/{REPO}/pull/5",
      package="left-pad",
      from_version=None,
      to_version=None,
      verdict=None,
      state="open",
      opened_at=NOW,
      merged_at=None,
      loop="dead-code",
  )
  model = read_model.assemble(
      now=NOW,
      repos=(REPO,),
      loops=(Loop.DEAD_CODE,),
      scan_interval_seconds=86_400,
      review_interval_seconds=300,
      a11y_interval_seconds=300,
      github=((rm,), None),
      temporal=(((), (), (), (), (), ()), None),
      telemetry=(
          RunTelemetry(
              available=False,
              total_spans=0,
              error_spans=0,
              last_activity=None,
              window_days=3,
              activities=(),
          ),
          "off",
      ),
  )
  html = render.page(model)
  assert "Dead-code" in html  # the tab label
  assert "M20 4 8.12 15.88" in html  # the scissors icon path
  assert "unused" in html  # the removal's target column, not a version
149 
150 
151def test_loop_tab_icon_is_spec_driven_not_a_hardcoded_map() -> None:
  # The dashboard-from-registry milestone: a loop's tab presentation flows
  # from its registered spec, so changing the spec's icon changes the rendered
  # tab with NO renderer edit (the renderer no longer keys icons by loop).
  from dataclasses import replace
156 
  from froot.dashboard.render import _ICONS
  from froot.loops import registry
159 
  original = registry.get(Loop.DEPENDENCY_PATCH)
  assert original.dashboard_icon == "package"
  try:
      registry.register(replace(original, dashboard_icon="layers"))
      html = render.page(_model())
      assert _ICONS["layers"] in html  # the spec's new icon is rendered
      assert _ICONS["package"] not in html  # the old one is gone
  finally:
      registry.register(original)  # restore for the other tests
⋯ 376 lines hidden (lines 169–544)169 
170 
171def test_theme_toggle_is_present_and_light_is_the_default():
  html = render.page(_model())
  # Light is the :root default; dark applies via system pref or the toggle.
  assert "--fg:#1b1f24" in html  # the light palette is the base
  assert "[data-theme=dark]" in html  # the forced-dark override exists
  assert "__toggleTheme" in html and 'class="themetgl"' in html
177 
178 
179def test_gate_hero_shows_the_four_bearings_and_a_decision():
  html = render.page(_model())
  assert "Earned autonomy" in html and "the gate" in html
  for bearing in ("approval rate", "defect rate", "probe", "deep review"):
      assert bearing in html
  assert "HOLD" in html  # no record yet -> the gate holds
185 
186 
187def test_page_renders_track_record_numbers():
  prs = [
      _pr(1, "a", "merged", opened=NOW, merged=NOW),
      _pr(2, "b", "merged", opened=NOW, merged=NOW),
  ]
  html = render.page(_model(prs=prs))
  assert "100%" in html  # 2/2 approval rate
  assert ">2<" in html  # the proposed count as a card stat
  assert "2 merged" in html
196 
197 
198def test_page_links_open_prs_and_lists_them_in_the_queue():
  prs = [_pr(23, "vitest", "open", to_version="3.2.6")]
  html = render.page(_model(prs=prs))
  assert f"https://github.com/{REPO}/pull/23" in html
  assert "vitest" in html
  assert "#23" in html
204 
205 
206def test_page_escapes_dynamic_content():
  evil = "<script>alert(1)</script>"
  prs = [_pr(1, evil, "merged", opened=NOW, merged=NOW)]
  html = render.page(_model(prs=prs))
  assert "<script>alert(1)</script>" not in html
  assert "&lt;script&gt;" in html
212 
213 
214def test_empty_queue_is_explicit_not_blank():
  html = render.page(_model())
  assert "Nothing awaiting you" in html
217 
218 
219# ── telemetry tab ────────────────────────────────────────────────────────────
220def test_telemetry_panel_reports_unavailable_when_off():
  html = render.page(_model())
  assert "Unavailable" in html
223 
224 
225def test_telemetry_panel_renders_activity_rows_when_available():
  telemetry = (
      RunTelemetry(
          available=True,
          total_spans=75,
          error_spans=10,
          last_activity=datetime(2026, 6, 3, 6, 0, tzinfo=UTC),
          window_days=3,
          activities=(
              ActivityStat(
                  name="open_pull_request",
                  count=14,
                  avg_ms=14162.0,
                  max_ms=55828.0,
              ),
          ),
      ),
      None,
  )
  html = render.page(_model(telemetry=telemetry))
  assert "open_pull_request" in html
  assert ">75<" in html and "spans" in html
247 
248 
249def test_scan_cadence_shows_liveness_and_next_due():
  scans = [
      ScanExecution(
          workflow_id="froot-scan-mseeks-revisionist",
          status="running",
          start=datetime(2026, 6, 3, 6, 0, tzinfo=UTC),
      )
  ]
  html = render.page(_model(scans=scans))
  assert REPO in html  # the class row carries the repo
  assert "next" in html  # the next-due hint for the live loop
260 
261 
262# ── determinism review tab ───────────────────────────────────────────────────
263def _review(status: str = "running") -> ReviewExecution:
  return ReviewExecution(
      workflow_id="froot-review-mseeks-revisionist",
      status=status,
      start=datetime(2026, 6, 3, 6, 0, tzinfo=UTC),
  )
269 
270 
271def _pr_review(
  pr: int,
  findings: int,
  rules: tuple[str, ...],
  comment: str | None = None,
276) -> PrReviewExecution:
  return PrReviewExecution(
      workflow_id=f"froot-pr-review-mseeks-revisionist-{pr}-abc1234def56",
      status="completed",
      start=datetime(2026, 6, 3, 6, 0, tzinfo=UTC),
      close=datetime(2026, 6, 3, 6, 1, tzinfo=UTC),
      pr_number=pr,
      head_sha="abc1234def56",
      findings=findings,
      rules=rules,
      comment_url=comment,
  )
288 
289 
290def test_determinism_tab_has_its_own_treatment_and_empty_state():
  html = render.page(_model())
  assert "Determinism review" in html
  assert "transitive ring" in html
  assert "No PRs reviewed yet" in html
295 
296 
297def test_determinism_tab_counts_a_live_review_loop():
  html = render.page(_model(reviews=[_review("running")]))
  assert "loops live" in html  # the liveness card
  assert ">1<" in html  # one repo covered / one loop live
301 
302 
303def test_flagged_review_renders_rule_count_and_comment_link():
  comment = f"https://github.com/{REPO}/pull/7#issuecomment-1"
  pr_reviews = [_pr_review(7, 1, ("datetime.datetime.now",), comment=comment)]
  html = render.page(_model(pr_reviews=pr_reviews))
  assert "datetime.datetime.now" in html
  assert "1 hazard" in html
  assert "#7" in html
  assert comment in html  # the one-click comment link
311 
312 
313def test_clean_review_renders_clean_not_a_hazard():
  html = render.page(_model(pr_reviews=[_pr_review(8, 0, ())]))
  assert ">clean<" in html
316 
317 
318# ── source-level a11y review tab ─────────────────────────────────────────────
319def _a11y(status: str = "running") -> A11yExecution:
  return A11yExecution(
      workflow_id="froot-a11y-mseeks-revisionist",
      status=status,
      start=datetime(2026, 6, 3, 6, 0, tzinfo=UTC),
  )
325 
326 
327def _pr_a11y(
  pr: int,
  findings: int,
  kinds: tuple[str, ...],
  comment: str | None = None,
332) -> PrA11yExecution:
  return PrA11yExecution(
      workflow_id=f"froot-pr-a11y-mseeks-revisionist-{pr}-abc1234def56",
      status="completed",
      start=datetime(2026, 6, 3, 6, 0, tzinfo=UTC),
      close=datetime(2026, 6, 3, 6, 1, tzinfo=UTC),
      pr_number=pr,
      head_sha="abc1234def56",
      findings=findings,
      kinds=kinds,
      comment_url=comment,
  )
344 
345 
346def test_a11y_tab_has_its_own_treatment_and_empty_state():
  html = render.page(_model())
  assert "A11y review" in html
  assert "source-level gaps" in html
  assert "No PRs reviewed yet" in html
351 
352 
353def test_a11y_tab_counts_a_live_loop():
  html = render.page(_model(a11y_reviews=[_a11y("running")]))
  assert "A11y loop" in html  # the per-tab cadence line
  assert "loops live" in html  # the liveness card
357 
358 
359def test_flagged_a11y_renders_gap_count_and_comment_link():
  comment = f"https://github.com/{REPO}/pull/7#issuecomment-9"
  pr_a11y = [_pr_a11y(7, 1, ("missing-alt",), comment=comment)]
  html = render.page(_model(pr_a11y_reviews=pr_a11y))
  assert "missing-alt" in html
  assert "1 gap" in html
  assert "#7" in html
  assert comment in html  # the one-click comment link
367 
368 
369def test_clean_a11y_renders_clean_not_a_gap():
  html = render.page(_model(pr_a11y_reviews=[_pr_a11y(8, 0, ())]))
  assert ">clean<" in html
372 
373 
374# ── the gate: per-class standing + queue badge ───────────────────────────────
375def _model_p(
  prs: Sequence[GithubPr],
  bumps: Sequence[BumpExecution],
  policy: AutonomyPolicy,
  outcomes: dict[tuple[str, int], str] | None = None,
380) -> DashboardModel:
  telemetry = (
      RunTelemetry(
          available=False,
          total_spans=0,
          error_spans=0,
          last_activity=None,
          window_days=3,
          activities=(),
      ),
      "off",
  )
  return read_model.assemble(
      now=NOW,
      repos=(REPO,),
      policy=policy,
      scan_interval_seconds=86_400,
      review_interval_seconds=300,
      a11y_interval_seconds=300,
      github=(tuple(prs), None),
      temporal=(((), tuple(bumps), (), (), (), ()), None),
      telemetry=telemetry,
      outcomes=outcomes,
  )
404 
405 
406def _clean_green(number: int, package: str) -> tuple[GithubPr, BumpExecution]:
  opened = datetime(2026, 5, 20, 12, 0, tzinfo=UTC)
  merged = datetime(2026, 5, 20, 12, 15, tzinfo=UTC)
  pr = _pr(
      number, package, "merged", verdict="clean", opened=opened, merged=merged
  )
  bump = BumpExecution(
      workflow_id=f"froot-bump-mseeks-revisionist-{package}",
      status="completed",
      start=opened,
      close=merged,
      verdict="clean",
      ci="passed",
      pr_number=number,
      repo=REPO,
      reason=None,
  )
  return pr, bump
424 
425 
426def test_class_table_shows_the_unearned_blocker_with_no_history():
  html = render.page(_model())
  assert "Per-class standing" in html
  assert "only 0/5 decided recently" in html  # the honest blocker
430 
431 
432def test_hero_says_no_classes_when_no_repos_configured():
  model = read_model.assemble(
      now=NOW,
      repos=(),
      scan_interval_seconds=86_400,
      review_interval_seconds=300,
      a11y_interval_seconds=300,
      github=((), None),
      temporal=(((), (), (), (), (), ()), None),
      telemetry=(
          RunTelemetry(
              available=False,
              total_spans=0,
              error_spans=0,
              last_activity=None,
              window_days=3,
              activities=(),
          ),
          "off",
      ),
  )
  assert "No classes yet" in render.page(model)
454 
455 
456def test_queue_badge_holds_open_pr_with_substantive_reason():
  prs = [_pr(23, "vitest", "open", verdict="clean", opened=NOW)]
  html = render.page(_model_p(prs, [], AutonomyPolicy()))
  assert "held" in html
  assert "class not earned" in html
461 
462 
463def test_class_earned_pill_and_budget():
  pairs = [_clean_green(n, f"pkg{n}") for n in (1, 2, 3, 4, 5)]
  prs = [p for p, _ in pairs]
  bumps = [b for _, b in pairs]
  policy = AutonomyPolicy(min_decided=3, allowlisted_repos=frozenset({REPO}))
  held = {(REPO, n): "held" for n in (1, 2, 3, 4, 5)}  # defect bearing clean
  html = render.page(_model_p(prs, bumps, policy, outcomes=held))
  assert ">earned<" in html  # the class cleared its gate (the pill)
  assert "budget/wk" in html  # the budget column
472 
473 
474def test_queue_badge_would_auto_merge_on_earned_allowlisted_class():
  pairs = [_clean_green(n, f"pkg{n}") for n in (1, 2, 3)]
  prs = [p for p, _ in pairs]
  bumps = [b for _, b in pairs]
  _, open_bump = _clean_green(9, "axios")
  prs.append(_pr(9, "axios", "open", verdict="clean", opened=NOW))
  bumps.append(open_bump)
  policy = AutonomyPolicy(min_decided=3, allowlisted_repos=frozenset({REPO}))
  held = {(REPO, n): "held" for n in (1, 2, 3)}  # clear the defect bearing
  html = render.page(_model_p(prs, bumps, policy, outcomes=held))
  assert "would auto-merge" in html
485 
486 
487# ── reliability + probes surface in the loop's cards / detail ────────────────
488def _model_outcomes(
  prs: Sequence[GithubPr], outcomes: dict[tuple[str, int], str]
490) -> DashboardModel:
  telemetry = (
      RunTelemetry(
          available=False,
          total_spans=0,
          error_spans=0,
          last_activity=None,
          window_days=3,
          activities=(),
      ),
      "off",
  )
  return read_model.assemble(
      now=NOW,
      repos=(REPO,),
      scan_interval_seconds=86_400,
      review_interval_seconds=300,
      a11y_interval_seconds=300,
      github=(tuple(prs), None),
      temporal=(((), (), (), (), (), ()), None),
      telemetry=telemetry,
      outcomes=outcomes,
      reliability_window_days=90,
  )
514 
515 
516def test_defect_rate_card_and_post_merge_tags():
  prs = [
      _pr(1, "a", "merged", opened=NOW, merged=NOW),
      _pr(2, "b", "merged", opened=NOW, merged=NOW),
  ]
  outcomes = {(REPO, 1): "held", (REPO, 2): "broke"}
  html = render.page(_model_outcomes(prs, outcomes))
  assert "defect rate" in html
  assert "50%" in html  # 1 of 2 determined was a defect
  assert ">held<" in html and ">broke<" in html  # the post-merge tags
526 
527 
528def test_canary_escape_shows_in_probes_card_and_segregated():
  # A merged canary (to 99.99.99) is a guardrail hole: the probes card counts
  # the escape, and the canary stays out of the genuine bumps.
  prs = [
      _pr(7, "evil", "merged", to_version="99.99.99", opened=NOW, merged=NOW)
  ]
  html = render.page(_model_outcomes(prs, {}))
  assert "probes escaped" in html
  # the canary is not a real bump -> the track record stays at zero proposed
  assert "evil" not in html
538 
539 
540def test_bumps_fold_lists_bumps_with_post_merge_column():
  prs = [_pr(1, "a", "merged", opened=NOW, merged=NOW)]
  html = render.page(_model_outcomes(prs, {(REPO, 1): "reverted"}))
  assert "post-merge" in html  # the bumps-table column
  assert "reverted" in html  # the per-row post-merge tag

The whole cut is green at every commit: ruff, mypy, the full test suite, and both the lexical and transitive determinism gates. Adding an acting loop now needs a Loop enum member and a loops/<name>.py spec — nothing else.