Phase 4 — the acting gate on a triangulated record

Overview

Phase 4 is froot's thesis: stop proposing and start acting — but only where a class of work has earned it, on a record that triangulates so no single gamed metric can buy autonomy. The loop now auto-merges a clean (judge) + green (CI) bump when its (repo, loop) class has earned the grant on an allowlisted repo and an independent deep review approves.

The allowlist is empty by default, so every un-opted repo behaves exactly as before — the same verdict is the advisory shadow gate a steward watches. Authority rests on four bearings that fail differently and can't all lie at once (§3.7–3.8): the approval rate, the post-merge defect rate, an adversarial gate self-test, and a deep review at the gate. This guide walks the pure-core extension first, then each bearing, then the honesty sweep.

The pure core, extended

The acting gate is added the way Phase 1's close-on-red was: as data in the pure state machine, not I/O sprinkled through the spine. Three new states, three new effects, two new events. A bump that auto-merges flows AwaitingCi → GateReviewing → Merging → Recorded, mirroring the AwaitingCi → Closing → Recorded red path.

src/froot/domain/state.py · 105 lines

src/froot/domain/state.py105 lines · Python

⋯ 59 lines hidden (lines 1–59)1"""The bump lifecycle states — one frozen model per stage of a single bump.
2 
3A bump moves Discovered -> Judged -> AwaitingCi -> Recorded, and each state
4carries exactly the data valid at that stage and no more: ``Discovered`` has
5only a candidate; ``AwaitingCi`` necessarily has a verdict *and* an open PR.
6A state that holds a PR but no verdict, or an outcome before CI resolved, is
7unrepresentable. ``Recorded`` is terminal.
8"""
9 
10from __future__ import annotations
11 
12from typing import Annotated, Literal
13 
14from pydantic import Field
15 
16from froot.domain.base import Frozen
17from froot.domain.candidate import Candidate
18from froot.domain.changelog import ChangelogVerdict
19from froot.domain.outcome import LoopOutcome
20from froot.domain.pull_request import PullRequestRef
21 
22 
23class Discovered(Frozen):
24    """A fresh candidate, not yet judged. The loop's entry state."""
25 
26    kind: Literal["discovered"] = "discovered"
27    candidate: Candidate
28 
29 
30class Judged(Frozen):
31    """The changelog has been assessed; ready to open the PR."""
32 
33    kind: Literal["judged"] = "judged"
34    candidate: Candidate
35    verdict: ChangelogVerdict
36 
37 
38class AwaitingCi(Frozen):
39    """The PR is open; the loop is durably waiting on the repo's CI."""
40 
41    kind: Literal["awaiting_ci"] = "awaiting_ci"
42    candidate: Candidate
43    verdict: ChangelogVerdict
44    pr: PullRequestRef
45 
46 
47class Closing(Frozen):
48    """CI failed and close-on-red is on: the PR is being closed before record.
49 
50    A transient stage between a red CI reading and the terminal record: the
51    spine closes the PR (and deletes its branch), then the loop records the
52    same outcome it would have anyway. Carries the already-built outcome so the
53    record step needs nothing more.
54    """
55 
56    kind: Literal["closing"] = "closing"
57    outcome: LoopOutcome
58 
59 
60class GateReviewing(Frozen):
61    """A green, clean, earned bump is being deep-reviewed before it merges.
62 
63    The transient stage between the green CI reading and the merge, where the
64    spine runs the independent adversarial gate review (the fourth trust leg,
65    §3.7). On approval the loop merges; on a hold it records and leaves the PR
66    for the human. Carries the already-built outcome so the next step needs
67    nothing more.
68    """
69 
70    kind: Literal["gate_reviewing"] = "gate_reviewing"
71    outcome: LoopOutcome
72 
73 
74class Merging(Frozen):
75    """The PR is being auto-merged before record (green, clean, earned class).
76 
77    The mirror of :class:`Closing` on the success side — a transient stage
78    between a passed gate review and the terminal record, where the spine merges
79    the PR (§3.4 Stage 5) and then records the same outcome. Reached only when a
80    steward has allowlisted the repo; otherwise the loop records and leaves the
81    PR for the human.
82    """
83 
84    kind: Literal["merging"] = "merging"
85    outcome: LoopOutcome
86 
⋯ 19 lines hidden (lines 87–105)87 
88class Recorded(Frozen):
89    """Terminal: CI resolved and the outcome was recorded."""
90 
91    kind: Literal["recorded"] = "recorded"
92    outcome: LoopOutcome
93 
94 
95# The state of a single bump's loop.
96BumpState = Annotated[
97    Discovered
98    | Judged
99    | AwaitingCi
100    | Closing
101    | GateReviewing
102    | Merging
103    | Recorded,
104    Field(discriminator="kind"),
105]

src/froot/domain/effects.py · 109 lines

src/froot/domain/effects.py109 lines · Python

⋯ 59 lines hidden (lines 1–59)1"""Effects: data describing what the spine should do next.
2 
3The loop's state machine is pure and performs no I/O. On each transition it
4emits an *effect* — judge the changelog, open the PR, wait on CI, record the
5outcome. The Temporal spine interprets each effect into an activity (or, for
6:class:`AwaitCi`, a durable poll-and-sleep loop) and feeds the resulting event
7back in. Effects are values, so a transition is fully testable without touching
8npm, GitHub, or a model.
9"""
10 
11from __future__ import annotations
12 
13from typing import Annotated, Literal
14 
15from pydantic import Field
16 
17from froot.domain.base import Frozen
18from froot.domain.candidate import Candidate
19from froot.domain.changelog import ChangelogVerdict
20from froot.domain.outcome import LoopOutcome
21from froot.domain.pull_request import PullRequestRef
22 
23 
24class JudgeChangelog(Frozen):
25    """Fetch the candidate's changelog and get the model's typed verdict."""
26 
27    kind: Literal["judge_changelog"] = "judge_changelog"
28    candidate: Candidate
29 
30 
31class OpenPullRequest(Frozen):
32    """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
33 
34    kind: Literal["open_pull_request"] = "open_pull_request"
35    candidate: Candidate
36    verdict: ChangelogVerdict
37 
38 
39class AwaitCi(Frozen):
40    """Durably wait on the PR's CI until it resolves or the deadline passes."""
41 
42    kind: Literal["await_ci"] = "await_ci"
43    pr: PullRequestRef
44 
45 
46class ClosePullRequest(Frozen):
47    """Close a red bump's PR (comment why, close it, delete its branch).
48 
49    Emitted in place of going straight to the record when CI came back failed
50    and close-on-red is enabled: the loop leaves no rotting red PR behind. The
51    failing check names ride along so the spine can comment what failed; the
52    record still follows (this state's outcome), so the red outcome is logged.
53    """
54 
55    kind: Literal["close_pull_request"] = "close_pull_request"
56    pr: PullRequestRef
57    failing: tuple[str, ...] = ()
58 
59 
60class ReviewBump(Frozen):
61    """Independently deep-review a bump at the gate, before it auto-merges.
62 
63    The fourth trust leg (§3.7's *sampled deep review*, here run on every
64    auto-merge candidate). Emitted when a green, clean, earned bump is about to
65    merge: the spine asks a second, independent, adversarial model pass to read
66    the changelog with a "find the reason to hold" stance. Only its approval
67    lets the merge proceed; anything else holds the PR for the human. The
68    expensive leg, spent only at the high-stakes moment.
69    """
70 
71    kind: Literal["review_bump"] = "review_bump"
72    candidate: Candidate
73    pr: PullRequestRef
74 
75 
76class MergePullRequest(Frozen):
77    """Auto-merge a clean+green bump whose class has earned the grant.
78 
79    Emitted in place of going straight to the record when CI came back green,
80    the changelog read clean, the (repo, loop) class has earned auto-merge on an
81    allowlisted repo, *and* the independent gate review approved (the acting
82    gate — §3.4 Stage 5). The record still follows (this state's outcome), so
83    the merged outcome is logged either way. Nothing reaches here unless a
84    steward has opted the repo into the allowlist; the default is empty, so the
85    loop stays propose-only.
86    """
87 
88    kind: Literal["merge_pull_request"] = "merge_pull_request"
89    pr: PullRequestRef
90 
91 
92class RecordOutcome(Frozen):
⋯ 17 lines hidden (lines 93–109)93    """Record the closed-loop outcome (label the PR, emit run telemetry)."""
94 
95    kind: Literal["record_outcome"] = "record_outcome"
96    outcome: LoopOutcome
97 
98 
99# What the spine should do after a transition.
100Effect = Annotated[
101    JudgeChangelog
102    | OpenPullRequest
103    | AwaitCi
104    | ClosePullRequest
105    | ReviewBump
106    | MergePullRequest
107    | RecordOutcome,
108    Field(discriminator="kind"),
109]

ReviewBump carries the candidate (to re-read its changelog) and the PR; MergePullRequest carries only the PR. GateReviewed carries a ChangelogVerdict — clean approves, anything else holds.

src/froot/domain/events.py · 83 lines

src/froot/domain/events.py83 lines · Python

⋯ 48 lines hidden (lines 1–48)1"""Lifecycle events: the decided inputs that drive the state machine.
2 
3Each event carries an already-decided result — the model judged the changelog,
4the PR was opened, CI resolved — so by the time an event reaches
5:func:`froot.policy.state_machine.advance`, the machine only decides *where the
6loop goes next*, never *what the judgment was*. All interpretation (the model
7call, the CI poll) happens in the spine's activities and arrives here as data.
8"""
9 
10from __future__ import annotations
11 
12from typing import Annotated, Literal
13 
14from pydantic import Field
15 
16from froot.domain.base import Frozen
17from froot.domain.changelog import ChangelogVerdict
18from froot.domain.ci import CIStatus
19from froot.domain.pull_request import PullRequestRef
20 
21 
22class ChangelogJudged(Frozen):
23    """The model returned its verdict on the candidate's changelog."""
24 
25    kind: Literal["changelog_judged"] = "changelog_judged"
26    verdict: ChangelogVerdict
27 
28 
29class PullRequestReady(Frozen):
30    """The PR for this bump is open (newly created or already existing)."""
31 
32    kind: Literal["pull_request_ready"] = "pull_request_ready"
33    pr: PullRequestRef
34 
35 
36class CiResolved(Frozen):
37    """CI reached a terminal status (never ``CIPending``; the spine waits)."""
38 
39    kind: Literal["ci_resolved"] = "ci_resolved"
40    status: CIStatus
41 
42 
43class PullRequestClosed(Frozen):
44    """The red bump's PR was closed (and its branch deleted)."""
45 
46    kind: Literal["pull_request_closed"] = "pull_request_closed"
47 
48 
49class GateReviewed(Frozen):
50    """The independent gate review returned its verdict on the bump.
51 
52    Carries a :data:`ChangelogVerdict` like the first judge pass: ``clean``
53    approves the merge; anything else (``risky``/``unknown``) holds the PR for
54    the human. Fail-closed — a review that could not run arrives non-clean.
55    """
56 
57    kind: Literal["gate_reviewed"] = "gate_reviewed"
58    verdict: ChangelogVerdict
59 
60 
61class PullRequestMerged(Frozen):
62    """The earned bump's PR was auto-merged by the acting gate."""
63 
64    kind: Literal["pull_request_merged"] = "pull_request_merged"
65 
66 
67class OutcomeRecorded(Frozen):
68    """The outcome was recorded; the loop has nothing left to do."""
69 
70    kind: Literal["outcome_recorded"] = "outcome_recorded"
71 
⋯ 12 lines hidden (lines 72–83)72 
73# A decided input to the loop state machine.
74LoopEvent = Annotated[
75    ChangelogJudged
76    | PullRequestReady
77    | CiResolved
78    | PullRequestClosed
79    | GateReviewed
80    | PullRequestMerged
81    | OutcomeRecorded,
82    Field(discriminator="kind"),
83]

The acting flip — where the machine decides to merge

In AwaitingCi, a terminal green CI plus a clean changelog plus the class-level grant now routes to the deep review rather than straight to record. The per-PR conditions (green + clean) are checked here in the pure machine; the class-level grant (auto_merge_eligible) is decided outside it and passed in, exactly like close_on_red — so the machine stays pure and replay-safe.

src/froot/policy/state_machine.py · 285 lines

src/froot/policy/state_machine.py285 lines · Python

⋯ 178 lines hidden (lines 1–178)1"""The bump loop state machine: pure transitions, effects as data.
2 
3``start`` and ``advance`` are pure: given the current state and a decided event,
4they return a :class:`Transition` — the next state plus the effects the spine
5should run. No I/O, no clock, so a transition replays deterministically and is
6fully testable. Dispatch is per-state and ends in ``assert_never`` so the type
7checker proves every state is handled; an event a state does not expect yields a
8``REJECTED`` transition (no state change), never an exception.
9 
10The loop is linear: each advance emits exactly one effect, the spine runs it to
11obtain the next event, and the cycle repeats until ``Recorded`` (no effects). A
class:`~froot.domain.events.CiResolved` that is still pending is rejected — the
13spine must finish waiting before the machine will close the loop, so an outcome
14can never be recorded against an unresolved CI status.
15"""
16 
17from __future__ import annotations
18 
19from enum import StrEnum
20from typing import TYPE_CHECKING, assert_never
21 
22from froot.domain.base import Frozen
23from froot.domain.ci import CIFailed, CIPassed, is_terminal
24from froot.domain.effects import (
  AwaitCi,
  ClosePullRequest,
  Effect,
  JudgeChangelog,
  MergePullRequest,
  OpenPullRequest,
  RecordOutcome,
  ReviewBump,
33)
34from froot.domain.events import (
  ChangelogJudged,
  CiResolved,
  GateReviewed,
  LoopEvent,
  OutcomeRecorded,
  PullRequestClosed,
  PullRequestMerged,
  PullRequestReady,
43)
44from froot.domain.outcome import LoopOutcome
45from froot.domain.state import (
  AwaitingCi,
  BumpState,
  Closing,
  Discovered,
  GateReviewing,
  Judged,
  Merging,
  Recorded,
54)
55 
56if TYPE_CHECKING:
  from froot.domain.candidate import Candidate
58 
59 
60class TransitionKind(StrEnum):
  """The disposition of an :func:`advance` call."""
62 
  ADVANCED = "advanced"
  IGNORED = "ignored"
  REJECTED = "rejected"
66 
67 
68class Transition(Frozen):
  """The result of a transition.
70 
  Attributes:
      kind: ``ADVANCED`` (moved and/or emitted effects), ``IGNORED`` (a legal
          no-op, e.g. the terminal acknowledgement), or ``REJECTED`` (the
          event is not valid in this state).
      next: The resulting state (unchanged for ``IGNORED``/``REJECTED``).
      effects: The effects the spine should run, in order. Empty terminates
          the loop driver.
      reason: A short explanation for an ``IGNORED``/``REJECTED`` transition.
  """
80 
  kind: TransitionKind
  next: BumpState
  effects: tuple[Effect, ...] = ()
  reason: str | None = None
85 
86 
87def _advanced(nxt: BumpState, *effects: Effect) -> Transition:
  return Transition(kind=TransitionKind.ADVANCED, next=nxt, effects=effects)
89 
90 
91def _rejected(state: BumpState, reason: str) -> Transition:
  return Transition(kind=TransitionKind.REJECTED, next=state, reason=reason)
93 
94 
95def start(candidate: Candidate) -> Transition:
  """The opening transition: enter ``Discovered`` and judge the changelog."""
  return _advanced(
      Discovered(candidate=candidate),
      JudgeChangelog(candidate=candidate),
  )
101 
102 
103def advance(
  state: BumpState,
  event: LoopEvent,
  *,
  close_on_red: bool = True,
  auto_merge_eligible: bool = False,
109) -> Transition:
  """Advance the loop one step (pure).
111 
  Args:
      state: The current bump state.
      event: The decided event that just occurred.
      close_on_red: Whether a terminal red CI should close the PR before
          recording. Passed in rather than read from config so the machine
          stays pure and replay-safe.
      auto_merge_eligible: Whether this PR's (repo, loop) *class* has earned
          the auto-merge grant on an allowlisted repo — the class-level half
          of the gate, decided by the spine (it needs the class's history).
          The per-PR half (clean changelog + green CI) the machine checks
          itself. Defaults False, so the loop stays propose-only unless a
          steward has granted the class autonomy.
124 
  Returns:
      The :class:`Transition` to apply.
  """
  match state:
      case Discovered():
          return _from_discovered(state, event)
      case Judged():
          return _from_judged(state, event)
      case AwaitingCi():
          return _from_awaiting_ci(
              state,
              event,
              close_on_red=close_on_red,
              auto_merge_eligible=auto_merge_eligible,
          )
      case Closing():
          return _from_closing(state, event)
      case GateReviewing():
          return _from_gate_reviewing(state, event)
      case Merging():
          return _from_merging(state, event)
      case Recorded():
          return _from_recorded(state, event)
  assert_never(state)
149 
150 
151def _from_discovered(state: Discovered, event: LoopEvent) -> Transition:
  match event:
      case ChangelogJudged():
          return _advanced(
              Judged(candidate=state.candidate, verdict=event.verdict),
              OpenPullRequest(
                  candidate=state.candidate, verdict=event.verdict
              ),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in discovered")
162 
163 
164def _from_judged(state: Judged, event: LoopEvent) -> Transition:
  match event:
      case PullRequestReady():
          return _advanced(
              AwaitingCi(
                  candidate=state.candidate,
                  verdict=state.verdict,
                  pr=event.pr,
              ),
              AwaitCi(pr=event.pr),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in judged")
177 
178 
179def _from_awaiting_ci(
  state: AwaitingCi,
  event: LoopEvent,
  *,
  close_on_red: bool,
  auto_merge_eligible: bool,
185) -> Transition:
  match event:
      case CiResolved():
          if not is_terminal(event.status):
              return _rejected(state, "ci still pending; the spine must wait")
          outcome = LoopOutcome(
              candidate=state.candidate,
              verdict=state.verdict,
              pr=state.pr,
              ci=event.status,
          )
          # Green CI, a clean changelog, and the class has earned the grant:
          # deep-review the bump before merging (the fourth leg, §3.7), then
          # merge on approval (the acting gate, §3.4 Stage 5). The per-PR
          # conditions are checked here; the class-level grant arrives in
          # auto_merge_eligible.
          if (
              isinstance(event.status, CIPassed)
              and state.verdict.kind == "clean"
              and auto_merge_eligible
          ):
              return _advanced(
                  GateReviewing(outcome=outcome),
                  ReviewBump(candidate=state.candidate, pr=state.pr),
              )
          # Red CI with close-on-red on: close the PR first (the loop leaves
          # no rotting red proposal), then record the same outcome. Every
          # other terminal reading (passed / absent / timed out), and red with
          # close-on-red off, records straight away and leaves the PR for the
          # human.
          if isinstance(event.status, CIFailed) and close_on_red:
              return _advanced(
                  Closing(outcome=outcome),
                  ClosePullRequest(pr=state.pr, failing=event.status.failing),
              )
          return _advanced(
              Recorded(outcome=outcome), RecordOutcome(outcome=outcome)
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in awaiting_ci")
225 
⋯ 60 lines hidden (lines 226–285)226 
227def _from_gate_reviewing(state: GateReviewing, event: LoopEvent) -> Transition:
  match event:
      case GateReviewed():
          # The independent deep review approved (clean): merge. Any other
          # verdict holds — record the outcome and leave the PR for the human,
          # exactly as a non-earned bump would. Fail-closed: a review that
          # could not run arrives non-clean, so an unreviewable bump never
          # auto-merges.
          if event.verdict.kind == "clean":
              return _advanced(
                  Merging(outcome=state.outcome),
                  MergePullRequest(pr=state.outcome.pr),
              )
          return _advanced(
              Recorded(outcome=state.outcome),
              RecordOutcome(outcome=state.outcome),
          )
      case _:
          return _rejected(
              state, f"unexpected {event.kind} in gate_reviewing"
          )
248 
249 
250def _from_merging(state: Merging, event: LoopEvent) -> Transition:
  match event:
      case PullRequestMerged():
          # The PR is merged; record the outcome it was carrying, exactly as
          # the non-merging path would have.
          return _advanced(
              Recorded(outcome=state.outcome),
              RecordOutcome(outcome=state.outcome),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in merging")
261 
262 
263def _from_closing(state: Closing, event: LoopEvent) -> Transition:
  match event:
      case PullRequestClosed():
          # The PR is closed; record the outcome it was carrying, exactly as
          # the non-closing path would have.
          return _advanced(
              Recorded(outcome=state.outcome),
              RecordOutcome(outcome=state.outcome),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in closing")
274 
275 
276def _from_recorded(state: Recorded, event: LoopEvent) -> Transition:
  # Terminal: the only expected event is the record acknowledgement, a no-op
  # that ends the driver loop (no effects).
  match event:
      case OutcomeRecorded():
          return Transition(
              kind=TransitionKind.IGNORED, next=state, reason="loop complete"
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} after recorded")

GateReviewing then forks on the review's verdict: clean → Merging + MergePullRequest; anything else → Recorded, leaving the green PR open for the human. Fail-closed: an unreviewable bump arrives non-clean and never merges.

src/froot/policy/state_machine.py · 285 lines

src/froot/policy/state_machine.py285 lines · Python

⋯ 226 lines hidden (lines 1–226)1"""The bump loop state machine: pure transitions, effects as data.
2 
3``start`` and ``advance`` are pure: given the current state and a decided event,
4they return a :class:`Transition` — the next state plus the effects the spine
5should run. No I/O, no clock, so a transition replays deterministically and is
6fully testable. Dispatch is per-state and ends in ``assert_never`` so the type
7checker proves every state is handled; an event a state does not expect yields a
8``REJECTED`` transition (no state change), never an exception.
9 
10The loop is linear: each advance emits exactly one effect, the spine runs it to
11obtain the next event, and the cycle repeats until ``Recorded`` (no effects). A
class:`~froot.domain.events.CiResolved` that is still pending is rejected — the
13spine must finish waiting before the machine will close the loop, so an outcome
14can never be recorded against an unresolved CI status.
15"""
16 
17from __future__ import annotations
18 
19from enum import StrEnum
20from typing import TYPE_CHECKING, assert_never
21 
22from froot.domain.base import Frozen
23from froot.domain.ci import CIFailed, CIPassed, is_terminal
24from froot.domain.effects import (
  AwaitCi,
  ClosePullRequest,
  Effect,
  JudgeChangelog,
  MergePullRequest,
  OpenPullRequest,
  RecordOutcome,
  ReviewBump,
33)
34from froot.domain.events import (
  ChangelogJudged,
  CiResolved,
  GateReviewed,
  LoopEvent,
  OutcomeRecorded,
  PullRequestClosed,
  PullRequestMerged,
  PullRequestReady,
43)
44from froot.domain.outcome import LoopOutcome
45from froot.domain.state import (
  AwaitingCi,
  BumpState,
  Closing,
  Discovered,
  GateReviewing,
  Judged,
  Merging,
  Recorded,
54)
55 
56if TYPE_CHECKING:
  from froot.domain.candidate import Candidate
58 
59 
60class TransitionKind(StrEnum):
  """The disposition of an :func:`advance` call."""
62 
  ADVANCED = "advanced"
  IGNORED = "ignored"
  REJECTED = "rejected"
66 
67 
68class Transition(Frozen):
  """The result of a transition.
70 
  Attributes:
      kind: ``ADVANCED`` (moved and/or emitted effects), ``IGNORED`` (a legal
          no-op, e.g. the terminal acknowledgement), or ``REJECTED`` (the
          event is not valid in this state).
      next: The resulting state (unchanged for ``IGNORED``/``REJECTED``).
      effects: The effects the spine should run, in order. Empty terminates
          the loop driver.
      reason: A short explanation for an ``IGNORED``/``REJECTED`` transition.
  """
80 
  kind: TransitionKind
  next: BumpState
  effects: tuple[Effect, ...] = ()
  reason: str | None = None
85 
86 
87def _advanced(nxt: BumpState, *effects: Effect) -> Transition:
  return Transition(kind=TransitionKind.ADVANCED, next=nxt, effects=effects)
89 
90 
91def _rejected(state: BumpState, reason: str) -> Transition:
  return Transition(kind=TransitionKind.REJECTED, next=state, reason=reason)
93 
94 
95def start(candidate: Candidate) -> Transition:
  """The opening transition: enter ``Discovered`` and judge the changelog."""
  return _advanced(
      Discovered(candidate=candidate),
      JudgeChangelog(candidate=candidate),
  )
101 
102 
103def advance(
  state: BumpState,
  event: LoopEvent,
  *,
  close_on_red: bool = True,
  auto_merge_eligible: bool = False,
109) -> Transition:
  """Advance the loop one step (pure).
111 
  Args:
      state: The current bump state.
      event: The decided event that just occurred.
      close_on_red: Whether a terminal red CI should close the PR before
          recording. Passed in rather than read from config so the machine
          stays pure and replay-safe.
      auto_merge_eligible: Whether this PR's (repo, loop) *class* has earned
          the auto-merge grant on an allowlisted repo — the class-level half
          of the gate, decided by the spine (it needs the class's history).
          The per-PR half (clean changelog + green CI) the machine checks
          itself. Defaults False, so the loop stays propose-only unless a
          steward has granted the class autonomy.
124 
  Returns:
      The :class:`Transition` to apply.
  """
  match state:
      case Discovered():
          return _from_discovered(state, event)
      case Judged():
          return _from_judged(state, event)
      case AwaitingCi():
          return _from_awaiting_ci(
              state,
              event,
              close_on_red=close_on_red,
              auto_merge_eligible=auto_merge_eligible,
          )
      case Closing():
          return _from_closing(state, event)
      case GateReviewing():
          return _from_gate_reviewing(state, event)
      case Merging():
          return _from_merging(state, event)
      case Recorded():
          return _from_recorded(state, event)
  assert_never(state)
149 
150 
151def _from_discovered(state: Discovered, event: LoopEvent) -> Transition:
  match event:
      case ChangelogJudged():
          return _advanced(
              Judged(candidate=state.candidate, verdict=event.verdict),
              OpenPullRequest(
                  candidate=state.candidate, verdict=event.verdict
              ),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in discovered")
162 
163 
164def _from_judged(state: Judged, event: LoopEvent) -> Transition:
  match event:
      case PullRequestReady():
          return _advanced(
              AwaitingCi(
                  candidate=state.candidate,
                  verdict=state.verdict,
                  pr=event.pr,
              ),
              AwaitCi(pr=event.pr),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in judged")
177 
178 
179def _from_awaiting_ci(
  state: AwaitingCi,
  event: LoopEvent,
  *,
  close_on_red: bool,
  auto_merge_eligible: bool,
185) -> Transition:
  match event:
      case CiResolved():
          if not is_terminal(event.status):
              return _rejected(state, "ci still pending; the spine must wait")
          outcome = LoopOutcome(
              candidate=state.candidate,
              verdict=state.verdict,
              pr=state.pr,
              ci=event.status,
          )
          # Green CI, a clean changelog, and the class has earned the grant:
          # deep-review the bump before merging (the fourth leg, §3.7), then
          # merge on approval (the acting gate, §3.4 Stage 5). The per-PR
          # conditions are checked here; the class-level grant arrives in
          # auto_merge_eligible.
          if (
              isinstance(event.status, CIPassed)
              and state.verdict.kind == "clean"
              and auto_merge_eligible
          ):
              return _advanced(
                  GateReviewing(outcome=outcome),
                  ReviewBump(candidate=state.candidate, pr=state.pr),
              )
          # Red CI with close-on-red on: close the PR first (the loop leaves
          # no rotting red proposal), then record the same outcome. Every
          # other terminal reading (passed / absent / timed out), and red with
          # close-on-red off, records straight away and leaves the PR for the
          # human.
          if isinstance(event.status, CIFailed) and close_on_red:
              return _advanced(
                  Closing(outcome=outcome),
                  ClosePullRequest(pr=state.pr, failing=event.status.failing),
              )
          return _advanced(
              Recorded(outcome=outcome), RecordOutcome(outcome=outcome)
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in awaiting_ci")
225 
226 
227def _from_gate_reviewing(state: GateReviewing, event: LoopEvent) -> Transition:
  match event:
      case GateReviewed():
          # The independent deep review approved (clean): merge. Any other
          # verdict holds — record the outcome and leave the PR for the human,
          # exactly as a non-earned bump would. Fail-closed: a review that
          # could not run arrives non-clean, so an unreviewable bump never
          # auto-merges.
          if event.verdict.kind == "clean":
              return _advanced(
                  Merging(outcome=state.outcome),
                  MergePullRequest(pr=state.outcome.pr),
              )
          return _advanced(
              Recorded(outcome=state.outcome),
              RecordOutcome(outcome=state.outcome),
          )
      case _:
          return _rejected(
              state, f"unexpected {event.kind} in gate_reviewing"
          )
248 
⋯ 37 lines hidden (lines 249–285)249 
250def _from_merging(state: Merging, event: LoopEvent) -> Transition:
  match event:
      case PullRequestMerged():
          # The PR is merged; record the outcome it was carrying, exactly as
          # the non-merging path would have.
          return _advanced(
              Recorded(outcome=state.outcome),
              RecordOutcome(outcome=state.outcome),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in merging")
261 
262 
263def _from_closing(state: Closing, event: LoopEvent) -> Transition:
  match event:
      case PullRequestClosed():
          # The PR is closed; record the outcome it was carrying, exactly as
          # the non-closing path would have.
          return _advanced(
              Recorded(outcome=state.outcome),
              RecordOutcome(outcome=state.outcome),
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} in closing")
274 
275 
276def _from_recorded(state: Recorded, event: LoopEvent) -> Transition:
  # Terminal: the only expected event is the record acknowledgement, a no-op
  # that ends the driver loop (no effects).
  match event:
      case OutcomeRecorded():
          return Transition(
              kind=TransitionKind.IGNORED, next=state, reason="loop complete"
          )
      case _:
          return _rejected(state, f"unexpected {event.kind} after recorded")

The class-level grant — computed once, outside the machine

The grant is about the class's history, not this one PR, so the workflow decides it once up front and threads it into advance(). The activity short-circuits to False for any non-allowlisted repo — the common case costs nothing and touches no network — and otherwise re-derives standing from live GitHub via the dashboard's own logic.

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

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

⋯ 356 lines hidden (lines 1–356)1"""Activities: the effect interpreters that wrap the adapters.
2 
3Each activity is the impure boundary for one effect. Adapter imports are LAZY
4(inside the bodies) so the model and HTTP stacks never enter a workflow sandbox;
5the pure policy and domain imports stay at module level. Activities return
6domain values — the workflow wraps them into events and feeds the pure state
7machine. The activity signatures' types are evaluated at runtime by Temporal's
8data converter, so the domain imports here are deliberately not deferred.
9"""
10 
11from __future__ import annotations
12 
13import json
14import logging
15import tempfile
16from pathlib import Path
17from typing import TYPE_CHECKING, assert_never
18 
19from temporalio import activity
20 
21from froot.domain.candidate import Candidate
22from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
23from froot.domain.ci import CIStatus
24from froot.domain.determinism import AnalysisResult, FrontierVerdict
25from froot.domain.loop import Loop
26from froot.domain.pull_request import PullRequestRef
27from froot.domain.repo import TargetRepo
28from froot.policy.compose import pr_labels, pull_request_draft
29from froot.policy.determinism import analyze_workflow_surface
30from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_review_workflow_id,
34)
35from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
36from froot.workflow.types import (
  AdjudicateInput,
  AutoMergeInput,
  CiCheckInput,
  CloseInput,
  DispatchInput,
  DispatchReviewInput,
  GateReviewInput,
  GateSelfTestInput,
  JudgeInput,
  MergeInput,
  OpenPrInput,
  PostReviewInput,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
53)
54 
55if TYPE_CHECKING:
  from froot.ports.protocols import PackageManager
57 
58_log = logging.getLogger("froot.outcome")
59_review_log = logging.getLogger("froot.review")
60_reconcile_log = logging.getLogger("froot.reconcile")
61_scan_log = logging.getLogger("froot.scan")
62_gate_log = logging.getLogger("froot.gate")
63 
64 
65def _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
68 
69 
70async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
75) -> tuple[int, tuple[Candidate, ...]]:
  """Gather this loop's signal from the checkout and select its candidates.
77 
  The one genuinely per-loop seam: dependency-patch reads the available
  upgrades and picks the highest patch; security-patch reads the installed,
  asks OSV for advisories, and picks the lowest version clearing each. The
  impure sources are lazy-imported per arm so neither drags the other's stack
  into a sandbox. Both feed a pure selection policy.
83 
  Returns ``(considered, candidates)`` — ``considered`` is the size of the
  upstream signal (available upgrades / advisories found) so the scan can make
  its selectivity legible (how much was seen versus how much was kept).
  """
  match loop:
      case Loop.DEPENDENCY_PATCH:
          from froot.policy.candidates import select_patch_candidates
91 
          upgrades = await package_manager.list_upgrades(target, manifest_dir)
          return len(upgrades), select_patch_candidates(upgrades)
      case Loop.SECURITY_PATCH:
          return await _select_security_candidates(
              target, package_manager, manifest_dir
          )
  assert_never(loop)
99 
100 
101async def _select_security_candidates(
  target: TargetRepo, package_manager: PackageManager, manifest_dir: Path
103) -> tuple[int, tuple[Candidate, ...]]:
  """Security signal: installed set, OSV advisories, clearing targets.
105 
  ``considered`` is the count of advisories OSV returned for the installed
  set — the vulnerabilities in scope this tick, before selection narrows to
  the ones a forward-stable bump can actually clear.
  """
  from froot.adapters.osv import OsvAdvisorySource
  from froot.policy.candidates import select_security_candidates
112 
  installed = await package_manager.list_installed(target, manifest_dir)
  advisories = await OsvAdvisorySource().advisories(installed)
  return len(advisories), select_security_candidates(installed, advisories)
116 
117 
118@activity.defn
119async def scan_candidates(
  params: ScanCandidatesInput,
121) -> tuple[Candidate, ...]:
  """Check out the repo and select this loop's candidates.
123 
  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
132 
  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
166 
167 
168@activity.defn
169async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
171 
  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
183 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      return UnknownVerdict(rationale="No changelog could be fetched.")
  try:
      return await PydanticAiJudge().judge(changelog, params.loop)
  except Exception as exc:
      activity.logger.warning(
          "changelog judge unavailable for %s; degrading to unknown: %r",
          params.candidate.package,
          exc,
      )
      return UnknownVerdict(
          rationale=f"Changelog judge unavailable ({type(exc).__name__})."
      )
198 
199 
200@activity.defn
201async def gate_review(params: GateReviewInput) -> ChangelogVerdict:
  """Independently deep-review a bump at the gate; fail-closed to a hold.
203 
  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
214 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      verdict: ChangelogVerdict = 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",
              params.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
248 
249 
250@activity.defn
251async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
255 
  forge = GitHubForge()
  package_manager = package_manager_for(params.target.ecosystem)
  branch = branch_name(params.candidate, params.loop)
  existing = await forge.find_open_pull_request(params.target, branch)
  if existing is not None:
      return existing
  draft = pull_request_draft(
      params.target, params.candidate, params.verdict, params.loop
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      await package_manager.apply_patch_bump(
          params.candidate, _manifest_dir(params.target, workspace)
      )
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
273 
274 
275@activity.defn
276async 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
279 
  return await GitHubForge().ci_status(params.target, params.head_sha)
281 
282 
283@activity.defn
284async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update.
286 
  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
294 
  outcome = params.outcome
  labels = (
      *pr_labels(params.loop),
      env_label(ModelSettings().ollama_model),
  )
  await GitHubForge().add_labels(params.target, outcome.pr.number, labels)
  _log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": params.loop.value,
              "repo": params.target.repo.slug,
              "package": outcome.candidate.package,
              "from": str(outcome.candidate.current),
              "to": str(outcome.candidate.target),
              "changelog": outcome.verdict.kind,
              "ci": outcome.ci.kind,
              "ci_passed": outcome.ci_passed,
              "pr": outcome.pr.number,
              "pr_url": outcome.pr.url,
          }
      )
  )
318 
319 
320@activity.defn
321async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
323 
  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
330 
  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
335 
  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
354 
355 
356@activity.defn
357async def auto_merge_eligible(params: AutoMergeInput) -> bool:
  """Whether this (repo, loop) class has earned the auto-merge grant.
359 
  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
369 
  from froot.config.settings import AutonomySettings, ModelSettings
  from froot.dashboard import github_source, read_model
  from froot.policy.environment import environment_slug
373 
  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),
  )
394 
395 
⋯ 241 lines hidden (lines 396–636)396@activity.defn
397async def merge_pull_request(params: MergeInput) -> None:
  """Auto-merge an earned, clean+green bump's PR (the acting gate's write).
399 
  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
405 
  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,
          }
      )
  )
421 
422 
423@activity.defn
424async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
426 
  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
434 
  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),
          }
      )
  )
456 
457 
458@activity.defn
459async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
461 
  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
474 
  if not BehaviorSettings().reconcile:
      return 0
477 
  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)
509 
510 
511@activity.defn
512async def gate_selftest(params: GateSelfTestInput) -> tuple[str, ...]:
  """Run the adversarial gate probe against the live policy; alarm on escape.
514 
  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
525 
  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
541 
542 
543# ── The determinism reviewer (the transitive ring) ──────────────────────────
544@activity.defn
545async 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
548 
  return await GitHubForge().list_open_pull_requests(target)
550 
551 
552@activity.defn
553async 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
557 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
560 
  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
576 
577 
578@activity.defn
579async 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
584 
  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)
595 
596 
597@activity.defn
598async def adjudicate_frontier(
  params: AdjudicateInput,
600) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
603 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
609 
610 
611@activity.defn
612async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
615 
  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

auto_merge_eligible reuses read_model.earned_now — the exact windowed, environment-scoped triangulation the dashboard panel shows — so the acting gate and the advisory panel can never disagree.

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

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

⋯ 410 lines hidden (lines 1–410)1"""Assemble the dashboard view — pure, from the three readers' output.
2 
3This is the reputation read-model proper: it joins GitHub (authoritative
4outcomes) to Temporal (recent verdict + CI reading) by PR number, derives the
5MHE-framed aggregates (track record, verification, judgment, the approval
6queue), and returns a fully-computed
class:`~froot.dashboard.model.DashboardModel` the renderer just projects. No
8I/O and no clock of its own — ``now`` is passed in — so every figure on the
9page is unit-tested apart from the network.
10"""
11 
12from __future__ import annotations
13 
14from typing import TYPE_CHECKING
15 
16from froot.dashboard.model import (
  BumpRow,
  ClassGate,
  DashboardModel,
  Failure,
  Judgment,
  Probes,
  Reliability,
  ReviewLoop,
  ReviewRecord,
  ReviewRow,
  RunTelemetry,
  ScanLoop,
  SourceHealth,
  TrackRecord,
  Verification,
32)
33from froot.domain.loop import Loop
34from froot.domain.repo import RepoRef, TargetRepo
35from froot.policy.autonomy import AutonomyPolicy, class_earned, pr_autonomy
36from froot.policy.canary import is_canary, score_probe
37from froot.policy.naming import review_workflow_id, scan_workflow_id
38from froot.result import Ok
39 
40if TYPE_CHECKING:
  from datetime import datetime
42 
  from froot.dashboard.github_source import GithubPr
  from froot.dashboard.temporal_source import (
      BumpExecution,
      PrReviewExecution,
      ReviewExecution,
      ScanExecution,
  )
50 
51# The conservative fallback policy: empty allowlist, so the shadow gate holds
52# everything until a caller passes a real one (built from FROOT_AUTOMERGE_*).
53_DEFAULT_POLICY = AutonomyPolicy()
54 
55_LIVE_STATUSES = frozenset({"running", "continued_as_new"})
56# Every way a bump can end without closing cleanly — all belong in the honest
57# Failures panel, not silently dropped.
58_FAILURE_STATUSES = frozenset({"terminated", "failed", "canceled", "timed_out"})
59 
60 
61def _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
70 
71 
72def _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
79 
80 
81def _scan_loops(
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  scans: tuple[ScanExecution, ...],
85) -> 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)
118 
119 
120def _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
127 
128 
129def _bump_rows(
  now: datetime,
  prs: tuple[GithubPr, ...],
  bumps: tuple[BumpExecution, ...],
  outcomes: dict[tuple[str, int], str],
134) -> tuple[BumpRow, ...]:
  """Join GitHub PRs (authoritative) to Temporal outcomes by PR number.
136 
  ``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
      rows.append(
          BumpRow(
              repo=pr.repo,
              loop=pr.loop,
              package=pr.package or "?",
              from_version=pr.from_version,
              to_version=pr.to_version or "?",
              state=pr.state,
              verdict=verdict,
              ci=ci,
              pr_number=pr.number,
              pr_url=pr.url,
              opened_at=pr.opened_at,
              merged_at=pr.merged_at,
              ttm_minutes=ttm,
              age_hours=age,
              post_merge=outcomes.get((pr.repo, pr.number)),
              env=pr.env,
          )
      )
  rows.sort(key=_opened_sort_key, reverse=True)
  return tuple(rows)
178 
179 
180def _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
183 
184 
185def _minutes_between(
  start: datetime | None, end: datetime | None
187) -> 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)
192 
193 
194def _hours_between(
  start: datetime | None, end: datetime | None
196) -> 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)
201 
202 
203def _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),
  )
218 
219 
220def _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,
  )
236 
237 
238def _reliability(rows: tuple[BumpRow, ...], window_days: int) -> Reliability:
  """The post-merge outcome breakdown — did the merges hold? (coarse).
240 
  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,
  )
260 
261 
262def _probes(canary_rows: tuple[BumpRow, ...]) -> Probes:
  """Tally the adversarial canary probes — caught / escaped / pending.
264 
  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),
  )
275 
276 
277def _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,
  )
299 
300 
301def _decided_at(row: BumpRow) -> datetime | None:
  """When a decided PR was decided — merge time, else open time as a proxy.
303 
  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
310 
311 
312def _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
317 
318 
319def _class_gates(
  now: datetime,
  rows: tuple[BumpRow, ...],
  repos: tuple[str, ...],
  loops: tuple[Loop, ...],
  policy: AutonomyPolicy,
  environment: str,
326) -> tuple[ClassGate, ...]:
  """The earned-autonomy standing of each (repo, loop) class.
328 
  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)
409 
410 
411def earned_now(
  now: datetime,
  prs: tuple[GithubPr, ...],
  outcomes: dict[tuple[str, int], str],
  repo: str,
  loop: Loop,
  policy: AutonomyPolicy,
  environment: str,
419) -> bool:
  """Whether one (repo, loop) class has earned its gate, from live data.
421 
  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)
⋯ 275 lines hidden (lines 429–703)429 
430 
431def _gate(
  rows: tuple[BumpRow, ...],
  gates: tuple[ClassGate, ...],
  policy: AutonomyPolicy,
435) -> tuple[BumpRow, ...]:
  """Open PRs awaiting a human, most-aged first, each carrying a verdict.
437 
  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)
473 
474 
475def _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)
492 
493 
494def _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
501 
502 
503def _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-") + "-"
510 
511 
512def _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
521 
522 
523def _review_loops(
  repos: tuple[str, ...], reviews: tuple[ReviewExecution, ...]
525) -> tuple[ReviewLoop, ...]:
  """One liveness row per repo that actually has a review loop.
527 
  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)
551 
552 
553def _review_rows(
  pr_reviews: tuple[PrReviewExecution, ...], repos: tuple[str, ...]
555) -> 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)
583 
584 
585def _review_record(
  loops: tuple[ReviewLoop, ...], rows: tuple[ReviewRow, ...]
587) -> 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),
  )
599 
600 
601def _sources(
  github_error: str | None,
  github_count: int,
  temporal_error: str | None,
  temporal_count: int,
  telemetry: RunTelemetry,
  clickhouse_error: str | None,
608) -> 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
      ),
  )
633 
634 
635def assemble(
  *,
  now: datetime,
  repos: tuple[str, ...],
  loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
  policy: AutonomyPolicy = _DEFAULT_POLICY,
  scan_interval_seconds: int,
  review_interval_seconds: int,
  github: tuple[tuple[GithubPr, ...], str | None],
  temporal: tuple[
      tuple[
          tuple[ScanExecution, ...],
          tuple[BumpExecution, ...],
          tuple[ReviewExecution, ...],
          tuple[PrReviewExecution, ...],
      ],
      str | None,
  ],
  telemetry: tuple[RunTelemetry, str | None],
  outcomes: dict[tuple[str, int], str] | None = None,
  reliability_window_days: int = 90,
  environment: str = "",
657) -> DashboardModel:
  """Build the whole view from the readers' ``(data, error)`` outputs.
659 
  ``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), temporal_error = temporal
  run_telemetry, clickhouse_error = telemetry
666 
  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)
  review_loops = _review_loops(repos, reviews)
  review_rows = _review_rows(pr_reviews, repos)
  return DashboardModel(
      generated_at=now,
      repos_configured=repos,
      scan_interval_seconds=scan_interval_seconds,
      sources=_sources(
          github_error,
          len(prs),
          temporal_error,
          len(scans) + len(bumps) + len(reviews) + len(pr_reviews),
          run_telemetry,
          clickhouse_error,
      ),
      scan_loops=_scan_loops(repos, loops, scans),
      track_record=_track_record(rows),
      class_gates=class_gates,
      verification=_verification(rows),
      reliability=_reliability(rows, reliability_window_days),
      probes=_probes(canary_rows),
      judgment=_judgment(rows),
      gate=_gate(rows, class_gates, policy),
      bumps=rows,
      failures=_failures(bumps),
      review_interval_seconds=review_interval_seconds,
      review_loops=review_loops,
      review_record=_review_record(review_loops, review_rows),
      reviews=review_rows,
      telemetry=run_telemetry,
  )

src/froot/workflow/bump_workflow.py · 213 lines

src/froot/workflow/bump_workflow.py213 lines · Python

⋯ 79 lines hidden (lines 1–79)1"""The per-bump lifecycle workflow — a thin driver over the pure core.
2 
3One durable workflow per (repo, package, target), keyed by the deterministic
func:`~froot.policy.naming.bump_workflow_id`, so a bump is proposed at most
5once. It loops: take the pure state machine's next effect, run it as an activity
6(or, for the CI wait, a durable poll-and-sleep), feed the resulting event back
7to :func:`~froot.policy.state_machine.advance`, and repeat until ``Recorded``.
8All nondeterminism is in the activities; the workflow uses only pure state and
9Temporal's own time APIs, so replay is deterministic.
10"""
11 
12from __future__ import annotations
13 
14from typing import TYPE_CHECKING, assert_never
15 
16from temporalio import workflow
17from temporalio.exceptions import ApplicationError
18 
19with workflow.unsafe.imports_passed_through():
  from froot.domain.ci import CIStatus, CITimedOut, is_terminal
  from froot.domain.effects import (
      AwaitCi,
      ClosePullRequest,
      Effect,
      JudgeChangelog,
      MergePullRequest,
      OpenPullRequest,
      RecordOutcome,
      ReviewBump,
  )
  from froot.domain.events import (
      ChangelogJudged,
      CiResolved,
      GateReviewed,
      LoopEvent,
      OutcomeRecorded,
      PullRequestClosed,
      PullRequestMerged,
      PullRequestReady,
  )
  from froot.domain.outcome import LoopOutcome
  from froot.domain.state import Recorded
  from froot.policy.state_machine import TransitionKind, advance, start
  from froot.workflow import activities
  from froot.workflow.constants import (
      ACTIVITY_TIMEOUT,
      CI_CHECK_TIMEOUT,
      CI_POLL_INTERVAL,
      CI_WAIT_DEADLINE,
      TOOL_RETRY,
  )
  from froot.workflow.types import (
      AutoMergeInput,
      BumpParams,
      CiCheckInput,
      CloseInput,
      GateReviewInput,
      JudgeInput,
      MergeInput,
      OpenPrInput,
      RecordInput,
  )
63 
64if TYPE_CHECKING:
  # Used only in the (non-workflow-decorated) helper signatures, so these are
  # type-only — unlike the run() signature, Temporal does not evaluate them.
  from froot.domain.loop import Loop
  from froot.domain.pull_request import PullRequestRef
  from froot.domain.repo import TargetRepo
70 
71 
72@workflow.defn
73class BumpWorkflow:
  """The durable loop for a single dependency patch bump."""
75 
  @workflow.run
  async def run(self, params: BumpParams) -> LoopOutcome:
      """Drive the pure state machine to a recorded outcome."""
      transition = start(params.candidate)
      # The class-level half of the gate: has this (repo, loop) earned the
      # auto-merge grant on an allowlisted repo? Decided once, up front (the
      # grant is about the class's history, not this PR), and cheap by default
      # — the activity short-circuits to False for any non-allowlisted repo.
      auto_merge_eligible = await workflow.execute_activity(
          activities.auto_merge_eligible,
          AutoMergeInput(target=params.target, loop=params.loop),
⋯ 80 lines hidden (lines 87–166)87            start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=TOOL_RETRY,
      )
      while transition.effects:
          state = transition.next
          if len(transition.effects) != 1:
              raise ApplicationError(
                  f"non-linear transition ({len(transition.effects)} "
                  "effects)",
                  non_retryable=True,
              )
          event = await self._execute(
              params.target, params.loop, transition.effects[0]
          )
          transition = advance(
              state,
              event,
              close_on_red=params.close_on_red,
              auto_merge_eligible=auto_merge_eligible,
          )
          if transition.kind is TransitionKind.REJECTED:
              raise ApplicationError(
                  f"rejected transition: {transition.reason}",
                  non_retryable=True,
              )
      final = transition.next
      if not isinstance(final, Recorded):
          raise ApplicationError(
              f"loop ended in non-terminal state: {final.kind}",
              non_retryable=True,
          )
      return final.outcome
119 
  async def _execute(
      self, target: TargetRepo, loop: Loop, effect: Effect
  ) -> LoopEvent:
      """Interpret one effect into an activity (or a durable CI wait).
124 
      ``loop`` rides in from the workflow's params and is handed to the impure
      activities (the branch namespace, the labels, the judge's prompt) — so
      the pure state machine never learns which loop it is running.
      """
      match effect:
          case JudgeChangelog():
              verdict = await workflow.execute_activity(
                  activities.judge_changelog,
                  JudgeInput(candidate=effect.candidate, loop=loop),
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return ChangelogJudged(verdict=verdict)
          case OpenPullRequest():
              pr = await workflow.execute_activity(
                  activities.open_pull_request,
                  OpenPrInput(
                      target=target,
                      candidate=effect.candidate,
                      verdict=effect.verdict,
                      loop=loop,
                  ),
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return PullRequestReady(pr=pr)
          case AwaitCi():
              status = await self._await_ci(target, effect.pr)
              return CiResolved(status=status)
          case ClosePullRequest():
              await workflow.execute_activity(
                  activities.close_pull_request,
                  CloseInput(
                      target=target,
                      pr=effect.pr,
                      failing=effect.failing,
                      loop=loop,
                  ),
                  start_to_close_timeout=CI_CHECK_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return PullRequestClosed()
          case ReviewBump():
              verdict = await workflow.execute_activity(
                  activities.gate_review,
                  GateReviewInput(
                      candidate=effect.candidate, pr=effect.pr, loop=loop
                  ),
                  start_to_close_timeout=ACTIVITY_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return GateReviewed(verdict=verdict)
          case MergePullRequest():
              await workflow.execute_activity(
                  activities.merge_pull_request,
                  MergeInput(target=target, pr=effect.pr, loop=loop),
                  start_to_close_timeout=CI_CHECK_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return PullRequestMerged()
          case RecordOutcome():
⋯ 28 lines hidden (lines 186–213)186                await workflow.execute_activity(
                  activities.record_outcome,
                  RecordInput(
                      target=target, outcome=effect.outcome, loop=loop
                  ),
                  start_to_close_timeout=CI_CHECK_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return OutcomeRecorded()
      assert_never(effect)
196 
  async def _await_ci(
      self, target: TargetRepo, pr: PullRequestRef
  ) -> CIStatus:
      """Durably poll the repo's CI until it resolves or the deadline."""
      deadline = workflow.now() + CI_WAIT_DEADLINE
      while True:
          status = await workflow.execute_activity(
              activities.check_ci,
              CiCheckInput(target=target, head_sha=pr.head_sha),
              start_to_close_timeout=CI_CHECK_TIMEOUT,
              retry_policy=TOOL_RETRY,
          )
          if is_terminal(status):
              return status
          if workflow.now() >= deadline:
              return CITimedOut()
          await workflow.sleep(CI_POLL_INTERVAL)

Bearings 1 & 2 — rate and post-merge defect rate

class_earned is the gate's arithmetic: it triangulates two independent bearings. The approval rate over enough decided PRs (track record), and the post-merge defect rate over enough confirmed-held merges (did reality punish the merge). They fail differently — rate to rubber-stamping, defect rate to a weak oracle — so a class earns only when both clear with evidence behind each. Checks run cheapest-first; the first to fail is the blocker to fix.

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

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

⋯ 83 lines hidden (lines 1–83)1"""The earned-autonomy policy — the gate's logic, now load-bearing.
2 
3The pure logic that says, per the Many Hands Engineering trust economy
4(§3.6-3.7), whether a *class* of work (a loop, on a repo) has earned its gate
5move, and whether a given PR auto-merges under that grant. The acting gate
6(Phase 4) reuses these same functions: on an **allowlisted** repo, a class that
7clears them has its clean+green bumps merged by the loop itself. Everywhere else
8— the default, the allowlist empty — the very same verdict stays advisory and
9the dashboard renders it as the *shadow gate* a steward watches before opting a
10repo in. Pure throughout; nothing here mutates anything (the merge is an effect
11the spine runs only once every condition, allowlist included, is met).
12 
13The grant has MHE's five properties. **Earned**: a class earns its gate by a
14track record — a high enough *approval rate* over enough *decided* PRs.
15**Narrow**: the class is one loop on one repo; dependency-patch and
16security-patch are separate trust classes (§3.9), never sharing a record.
17**Conditional**: a PR rides the grant only if its own changelog read clean and
18its CI went green. **Revocable**: the grant is gated behind an explicit
19per-repo allowlist a steward controls, and lapses on its own when the rate
20falls. **Expiring**: the rate is measured over a recent window, not all of
21history (§2.11) — a class that has not acted lately starts from a lower
22baseline.
23 
24MHE is blunt that the approval rate is the headline, *not the whole truth* —
25"track record alone can lie" (§3.7). So the gate triangulates it (§3.8): a class
26earns only when the rate **and** the post-merge **defect rate** both clear,
27each with enough evidence behind it — two bearings that fail independently and
28cannot both lie at once. Two further legs MHE names — *sampled deep review* and
29*adversarial probing* — strengthen the triangle and arrive in later slices; the
30*environment*-reset (a judge-model swap or refactor resets the record, §3.7's
31fuller "conditional") rides alongside them.
32"""
33 
34from __future__ import annotations
35 
36from froot.domain.base import Frozen
37 
38 
39class AutonomyPolicy(Frozen):
40    """The thresholds a class must clear to earn its gate move.
41 
42    Attributes:
43        min_rate: The approval (merge) rate a class needs over the window —
44            the first bearing (track record). High alone is a *smell*, not
45            proof (a 95% gate "is rubber-stamping", §2.10), so it never earns
46            the gate by itself.
47        min_decided: How many PRs must have been decided in the window before
48            the rate means anything — a 100% rate on one PR is not a record.
49        window_days: The look-back window; trust is recent, not lifetime.
50        min_determined: How many merges must have a *confirmed* post-merge
51            outcome (held / broke / reverted) before the defect bearing counts
52            — the rate can lie, so the outcome bearing needs evidence too
53            (§3.7). Until then the class is not earned.
54        max_defect_rate: The ceiling on the post-merge defect rate — the second,
55            independent bearing (§3.8: a target needs ≥2 metrics that gaming
56            would harm). Zero by default: one confirmed merge that broke or was
57            reverted blocks the gate until it ages out of the window.
58        allowlisted_repos: The repos a steward has opted into auto-merge for.
59            Empty by default — the revocable switch, off until trust is granted.
60    """
61 
62    min_rate: float = 0.95
63    min_decided: int = 5
64    window_days: int = 90
65    min_determined: int = 3
66    max_defect_rate: float = 0.0
67    allowlisted_repos: frozenset[str] = frozenset()
68 
69 
70class AutonomyVerdict(Frozen):
71    """Whether a PR would auto-merge under the grant, and the reason either way.
72 
73    Attributes:
74        would_merge: True iff every condition is met, the allowlist included —
75        so on an allowlisted repo this is the loop's actual merge decision,
76        and elsewhere (the default) the advisory shadow-gate verdict.
77        reason: The deciding factor — the grant met, or the first blocker.
78    """
79 
80    would_merge: bool
81    reason: str
82 
83 
84def class_earned(
85    *,
86    decided: int,
87    merged: int,
88    determined: int,
89    defects: int,
90    policy: AutonomyPolicy,
91) -> tuple[bool, str | None]:
92    """Whether a class has earned its gate move, and why not if it hasn't.
93 
94    Triangulates two *independent* bearings (§3.8): the approval **rate** (did
95    the steward say yes) and the post-merge **defect rate** (did reality punish
96    the merge). They fail differently — the rate to rubber-stamping, the defect
97    rate to a weak oracle — so a class earns only when both clear *and* there is
98    enough evidence behind each. Checks run cheapest-first, and the first to
99    fail is the blocker, so the reason is the thing to fix.
100 
101    Args:
102        decided: PRs of this class decided (merged or closed) in the window.
103        merged: How many of those were merged.
104        determined: Merges with a confirmed post-merge outcome in the window
105            (held / broke / reverted) — the defect bearing's evidence.
106        defects: Of those, how many broke the branch or were reverted.
107        policy: The thresholds.
108 
109    Returns:
110        ``(earned, blocker)`` — ``blocker`` is ``None`` when earned, else the
111        short reason the gate has not moved.
112    """
113    if decided < policy.min_decided or decided == 0:
114        return False, f"only {decided}/{policy.min_decided} decided recently"
115    rate = merged / decided
116    if rate < policy.min_rate:
117        return False, (f"approval rate {rate:.0%} < {policy.min_rate:.0%}")
118    if determined < policy.min_determined:
119        return False, (
120            f"only {determined}/{policy.min_determined} merges confirmed held"
121        )
122    defect_rate = defects / determined
123    if defect_rate > policy.max_defect_rate:
⋯ 45 lines hidden (lines 124–168)124        return False, (
125            f"defect rate {defect_rate:.0%} > {policy.max_defect_rate:.0%}"
126        )
127    return True, None
128 
129 
130def pr_autonomy(
131    *,
132    repo: str,
133    verdict: str | None,
134    ci: str | None,
135    earned: bool,
136    blocker: str | None,
137    policy: AutonomyPolicy,
138) -> AutonomyVerdict:
139    """Whether one open PR would auto-merge under its class's grant.
140 
141    Reports the *substantive* blockers first — the earned grant, then this PR's
142    own clean-and-green conditions — and the steward's own revocable switch
143    (the allowlist) *last*. That ordering is what makes the shadow gate
144    watchable in its default, allowlist-off state: a held PR shows the real
145    thing to fix (``CI pending`` / ``class not earned``), and the bare
146    ``auto-merge not enabled`` reason appears only once a PR is otherwise fully
147    ready — i.e. exactly when flipping the switch would change the outcome.
148    ``would_merge`` still requires *every* condition, the allowlist included.
149    """
150    if not earned:
151        return AutonomyVerdict(
152            would_merge=False, reason=f"class not earned ({blocker})"
153        )
154    if verdict != "clean":
155        return AutonomyVerdict(
156            would_merge=False, reason=f"verdict is {verdict or 'unknown'}"
157        )
158    if ci != "passed":
159        return AutonomyVerdict(
160            would_merge=False, reason=f"CI {ci or 'pending'}"
161        )
162    if repo not in policy.allowlisted_repos:
163        return AutonomyVerdict(
164            would_merge=False, reason="auto-merge not enabled for this repo"
165        )
166    return AutonomyVerdict(
167        would_merge=True, reason="clean + green on an earned class"
168    )

The outcome detector — did the merge hold? (P4.1)

The defect bearing needs an oracle for what happened after a merge. fetch_outcomes reads the default branch's recent commits to match froot's (#N) squash tail to its merge commit and spot Revert commits, then reads each merge commit's check-runs. Deliberately coarse and low-recall: a manual or bundled revert is invisible, so the defect rate is a floor, not the truth — the adversarial leg is what exercises the gap.

src/froot/dashboard/github_source.py · 355 lines

src/froot/dashboard/github_source.py355 lines · Python

⋯ 189 lines hidden (lines 1–189)1"""GitHub reader: froot's PRs as the durable outcome ledger.
2 
3One issues-list call per repo (PRs are issues) filtered to the ``froot`` label
4returns every proposed bump with its outcome — state, timestamps, and the
5deterministic title/body froot writes (``compose.py``). Package + target parse
6from the title (``deps: bump <pkg> to <ver>``); the from-version and the model's
7changelog verdict parse from the body, so the judgment survives even after the
8bump ages out of Temporal's window. Read-only: the dashboard needs no write
9scope (it reuses ``FROOT_GITHUB_TOKEN`` only because it is already in the pod).
10 
11The parsers are pure and unit-tested apart from the network; the HTTP shape is
12read at the boundary as untyped JSON and coerced here.
13"""
14 
15from __future__ import annotations
16 
17import re
18from datetime import UTC, datetime
19from typing import Any, Final
20 
21import httpx
22 
23from froot.config.settings import GitHubSettings
24from froot.domain.base import Frozen
25from froot.domain.loop import Loop
26from froot.policy.environment import env_from_labels
27 
28_API: Final = "https://api.github.com"
29_API_VERSION: Final = "2022-11-28"
30_TIMEOUT: Final = 15.0
31_LABEL: Final = "froot"
32_PER_PAGE: Final = 100
33 
34_TITLE_RE: Final = re.compile(r"^deps: bump (?P<pkg>.+) to (?P<ver>\S+)\s*$")
35_FROM_RE: Final = re.compile(r"from (?P<from>\S+) to (?P<to>\S+)")
36# froot's squash merges leave a ``(#N)`` tail on the default-branch commit, so a
37# merged PR can be matched to its merge commit without a per-PR API call.
38_PR_REF_RE: Final = re.compile(r"\(#(\d+)\)")
39 
40# Check-run conclusions, split into "the branch held" vs "the branch broke". A
41# commit with no check runs at all is neither — it is ``unknown`` (no oracle on
42# the branch), never silently counted as a pass.
43_OK_CONCLUSIONS: Final = frozenset({"success", "neutral", "skipped", "stale"})
44_BAD_CONCLUSIONS: Final = frozenset(
  {"failure", "cancelled", "timed_out", "action_required", "startup_failure"}
46)
47 
48 
49class GithubPr(Frozen):
  """A froot pull request, reduced to what the read-model needs."""
51 
  repo: str
  number: int
  url: str
  # The loop label (``dependency-patch`` | ``security-patch``); defaults to
  # dependency-patch so a PR with no loop label (or a hand-built test row)
  # attributes to the original loop rather than failing to construct.
  loop: str = "dependency-patch"
  # The judgment environment (judge-model slug) the PR was opened under, from
  # its ``froot-env:`` label; ``None`` if unstamped (a prior environment).
  env: str | None = None
  package: str | None
  from_version: str | None
  to_version: str | None
  verdict: str | None
  state: str  # open | merged | closed
  opened_at: datetime | None
  merged_at: datetime | None
69 
70 
71def _label_names(payload: Any) -> set[str]:
  """The set of label names on an issues-API row (empty if none/malformed)."""
  labels = payload.get("labels")
  if not isinstance(labels, list):
      return set()
  return {
      name
      for lbl in labels
      if isinstance(lbl, dict) and isinstance((name := lbl.get("name")), str)
  }
81 
82 
83def _loop_from_labels(names: set[str]) -> str:
  """The loop a PR belongs to, from its labels (defaults dependency-patch).
85 
  Every froot PR carries its loop's label alongside the ``froot`` label, so
  the loop is durable on the PR even after its Temporal run ages out.
  """
  for loop in Loop:
      if loop.value in names:
          return loop.value
  return Loop.DEPENDENCY_PATCH.value
93 
94 
95def parse_title(title: str) -> tuple[str, str] | None:
  """Parse ``deps: bump <pkg> to <ver>`` into ``(package, target)`` (pure)."""
  match = _TITLE_RE.match(title.strip())
  if match is None:
      return None
  return match.group("pkg"), match.group("ver")
101 
102 
103def parse_from_version(body: str | None) -> str | None:
  """Pull the from-version out of froot's PR body template (pure)."""
  if not body:
      return None
  match = _FROM_RE.search(body)
  return match.group("from") if match else None
109 
110 
111def parse_verdict(body: str | None) -> str | None:
  """Recover the changelog verdict from froot's body template (pure).
113 
  Mirrors ``policy/compose.py``'s ``_verdict_summary`` openers, so an old PR
  whose Temporal run has aged out still shows the judgment it carried.
  """
  if not body:
      return None
  if "Changelog reads clean" in body:
      return "clean"
  if "Review carefully" in body:
      return "risky"
  if "Changelog unavailable" in body:
      return "unknown"
  return None
126 
127 
128def _parse_dt(value: Any) -> datetime | None:
  """Coerce a GitHub ISO-8601 timestamp into an *aware* datetime (boundary).
130 
  GitHub stamps everything ``...Z`` (UTC), which ``fromisoformat`` parses as
  aware — but a missing/odd offset would yield a naive datetime, and the
  read-model subtracts these against an aware ``now`` (a naive operand raises
  ``TypeError`` and would blank the whole page). So any naive result is
  pinned to UTC here, at the one boundary, the way the other readers do.
  """
  if not isinstance(value, str) or not value:
      return None
  try:
      parsed = datetime.fromisoformat(value)
  except ValueError:
      return None
  return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
144 
145 
146def _to_pr(repo: str, payload: Any) -> GithubPr | None:
  """Coerce one issues-API row into a :class:`GithubPr`, or skip non-PRs."""
  if not isinstance(payload, dict):
      return None
  pull_request = payload.get("pull_request")
  if not isinstance(pull_request, dict):
      return None  # a plain issue, not a PR
  merged_at = _parse_dt(pull_request.get("merged_at"))
  if merged_at is not None:
      state = "merged"
  elif payload.get("state") == "open":
      state = "open"
  else:
      state = "closed"
  parsed = parse_title(str(payload.get("title", "")))
  package, to_version = parsed if parsed is not None else (None, None)
  body = payload.get("body")
  names = _label_names(payload)
  return GithubPr(
      repo=repo,
      number=int(payload["number"]),
      url=str(payload.get("html_url", "")),
      loop=_loop_from_labels(names),
      env=env_from_labels(names),
      package=package,
      from_version=parse_from_version(body),
      to_version=to_version,
      verdict=parse_verdict(body),
      state=state,
      opened_at=_parse_dt(payload.get("created_at")),
      merged_at=merged_at,
  )
178 
179 
180def _commit_pr_refs(message: str) -> list[int]:
  """PR numbers referenced as ``(#N)`` in a commit message (pure)."""
  return [int(ref) for ref in _PR_REF_RE.findall(message)]
183 
184 
185def _is_revert(message: str) -> bool:
  """True if a commit message reads as a git revert (``Revert "..."``)."""
  return message.lstrip().lower().startswith("revert")
188 
189 
190def classify_check_runs(payload: Any) -> str:
  """Reduce a commit's check-runs to held / broke / unknown (pure boundary).
192 
  ``broke`` if any run reported a failing conclusion; ``held`` if runs exist
  and none failed; ``unknown`` if no check ran on the commit — i.e. there is
  no oracle on the branch, which is never conflated with a pass (the same
  discipline the verification panel keeps for ``absent`` CI).
  """
  runs = payload.get("check_runs") if isinstance(payload, dict) else None
  if not isinstance(runs, list) or not runs:
      return "unknown"
  conclusions = {r.get("conclusion") for r in runs if isinstance(r, dict)}
  if conclusions & _BAD_CONCLUSIONS:
      return "broke"
  if conclusions & _OK_CONCLUSIONS:
      return "held"
  return "unknown"
207 
⋯ 32 lines hidden (lines 208–239)208 
209def _merge_index(
  commits: Any, merged_numbers: frozenset[int]
211) -> tuple[dict[int, str], set[int]]:
  """From a default-branch commit list, map PR# → merge SHA and the reverted.
213 
  Commits arrive newest-first, so the first commit referencing ``(#N)`` is the
  merge; a ``Revert`` commit's refs mark those PRs reverted instead. Only
  froot's own merged numbers are considered, so an unrelated ``(#N)`` can't
  cross-attribute. Pure over already-fetched JSON.
  """
  merge_sha: dict[int, str] = {}
  reverted: set[int] = set()
  if not isinstance(commits, list):
      return merge_sha, reverted
  for entry in commits:
      if not isinstance(entry, dict):
          continue
      sha = entry.get("sha")
      commit = entry.get("commit")
      message = commit.get("message", "") if isinstance(commit, dict) else ""
      refs = [n for n in _commit_pr_refs(message) if n in merged_numbers]
      if not refs:
          continue
      if _is_revert(message):
          reverted.update(refs)
      elif isinstance(sha, str):
          for number in refs:
              merge_sha.setdefault(number, sha)
  return merge_sha, reverted
238 
239 
240async def fetch_outcomes(
  repos: tuple[str, ...],
  prs: tuple[GithubPr, ...],
  *,
  now: datetime,
  window_days: int,
246) -> tuple[dict[tuple[str, int], str], str | None]:
  """Best-effort post-merge outcome per recently-merged PR; never raises.
248 
  Per repo: read the default branch's recent commits once (to match froot's
  ``(#N)`` squash tail to the merge commit and spot ``Revert`` commits), then
  read each merge commit's check-runs to see whether the branch's CI held.
  Returns ``{(repo, number): "held"|"broke"|"reverted"|"unknown"}``.
253 
  Deliberately coarse and low-recall: a *manual* or *bundled* revert is
  invisible here (most are — git-reverts are the minority), and a merge older
  than one commit page falls to ``unknown`` rather than a false ``held``. This
  is the natural-traffic floor; the adversarial canary leg is what exercises
  it. ``(prs, error)``-style: degrades to a reason, keeping partial results.
  """
  token = GitHubSettings().github_token
  if token is None:
      return {}, "FROOT_GITHUB_TOKEN unset"
  headers = {
      "Authorization": f"Bearer {token.get_secret_value()}",
      "Accept": "application/vnd.github+json",
      "X-GitHub-Api-Version": _API_VERSION,
  }
  horizon = window_days * 86400.0
  outcomes: dict[tuple[str, int], str] = {}
  try:
      async with httpx.AsyncClient(
          base_url=_API, timeout=_TIMEOUT, headers=headers
      ) as client:
          for repo in repos:
              merged = [
                  pr
                  for pr in prs
                  if pr.repo == repo
                  and pr.state == "merged"
                  and pr.merged_at is not None
                  and (now - pr.merged_at).total_seconds() <= horizon
              ]
              if not merged:
                  continue
              resp = await client.get(
                  f"/repos/{repo}/commits", params={"per_page": _PER_PAGE}
              )
              resp.raise_for_status()
              merge_sha, reverted = _merge_index(
                  resp.json(), frozenset(pr.number for pr in merged)
              )
              for pr in merged:
                  if pr.number in reverted:
                      outcomes[(repo, pr.number)] = "reverted"
                      continue
                  sha = merge_sha.get(pr.number)
                  if sha is None:
                      outcomes[(repo, pr.number)] = "unknown"
                      continue
                  checks = await client.get(
                      f"/repos/{repo}/commits/{sha}/check-runs"
                  )
                  checks.raise_for_status()
                  outcomes[(repo, pr.number)] = classify_check_runs(
                      checks.json()
                  )
  except Exception as exc:  # never raise into gather — degrade to an error
      return outcomes, f"{type(exc).__name__}: {exc}"
  return outcomes, None
310 
311 
⋯ 44 lines hidden (lines 312–355)312async def fetch(
  repos: tuple[str, ...],
314) -> tuple[tuple[GithubPr, ...], str | None]:
  """Read every froot PR across ``repos``; never raises.
316 
  Returns:
      ``(prs, error)`` — ``error`` is ``None`` on success, else a short
      reason and ``prs`` is whatever was gathered before the failure.
  """
  token = GitHubSettings().github_token
  if token is None:
      return (), "FROOT_GITHUB_TOKEN unset"
  headers = {
      "Authorization": f"Bearer {token.get_secret_value()}",
      "Accept": "application/vnd.github+json",
      "X-GitHub-Api-Version": _API_VERSION,
  }
  prs: list[GithubPr] = []
  try:
      async with httpx.AsyncClient(
          base_url=_API, timeout=_TIMEOUT, headers=headers
      ) as client:
          for repo in repos:
              resp = await client.get(
                  f"/repos/{repo}/issues",
                  params={
                      "labels": _LABEL,
                      "state": "all",
                      "per_page": _PER_PAGE,
                      "sort": "created",
                      "direction": "desc",
                  },
              )
              resp.raise_for_status()
              rows = resp.json()
              if isinstance(rows, list):
                  prs.extend(
                      pr
                      for row in rows
                      if (pr := _to_pr(repo, row)) is not None
                  )
  except Exception as exc:  # never raise into gather — degrade to an error
      return tuple(prs), f"{type(exc).__name__}: {exc}"
  return tuple(prs), None

Bearing 3 — the adversarial gate self-test (P4.2)

The rate and defect bearings need volume to mean anything; froot's per-class volume is tiny. A probe needs none — it is informative at N=1. gate_probe is the §2.11 deliberate disturbance aimed at the gate's new risk surface: a battery of synthetic class histories a healthy gate MUST refuse, each engineered to fail a different threshold, scored against the live policy. Any it would grant is an escape — the alarm. It catches a gate loosened by config drift (§3.7 silent drift) the moment it happens.

src/froot/policy/gate_probe.py · 111 lines

src/froot/policy/gate_probe.py111 lines · Python

1"""The adversarial gate self-test — would the live gate refuse a bad class?
2 
3The acting flip (Phase 4.3c) lets an earned class auto-merge its own clean+green
4bumps. Its risk surface is a gate that *grants when it must not* — through a
5code regression, or (the insidious one) a production **config** loosening a
6steward makes by hand that no CI test ever sees: drop the ``MIN_RATE`` floor or
7raise the ``MAX_DEFECT_RATE`` ceiling (the ``FROOT_AUTOMERGE_*`` env) and the
8gate quietly starts trusting classes it shouldn't. That is §3.7's "silent
9drift": trust comes apart without a single visible failure.
10 
11This is the §2.11 deliberate disturbance aimed straight at that surface — the
12third trust leg, adversarial probing. A battery of synthetic class histories
13that a *healthy* gate must refuse, each engineered to fail a different
14threshold, run against the **live** policy. Any scenario the gate would grant is
15an escape: the alarm. It needs no volume (informative at N=1, unlike the rate
16and defect bearings) and runs every tick, so a loosened gate is caught the
17moment config drifts — not after a bad bump has merged.
18 
19Pure: scenarios and a policy in, the names that escaped out. The schedule and
20the structured alarm are the impure wiring (the scan tick + the activity log).
21"""
22 
23from __future__ import annotations
24 
25from typing import TYPE_CHECKING, Final
26 
27from froot.domain.base import Frozen
28from froot.policy.autonomy import class_earned
29 
30if TYPE_CHECKING:
31    from froot.policy.autonomy import AutonomyPolicy
32 
33 
34class GateScenario(Frozen):
35    """One synthetic class history a healthy gate must never auto-merge-grant.
36 
37    The same shape :func:`~froot.policy.autonomy.class_earned` reads — a class's
38    windowed record — so the probe exercises the *real* gate, not a model of it.
39 
40    Attributes:
41        name: What makes this history untrustworthy (the threshold it probes).
42        decided: PRs decided in the window.
43        merged: How many were merged (the rate's numerator).
44        determined: Merges with a confirmed post-merge outcome.
45        defects: Of those, how many broke or were reverted.
46    """
47 
48    name: str
49    decided: int
50    merged: int
51    determined: int
52    defects: int
53 
54 
55# The battery. Each scenario clears every threshold *but one*, so each guards a
56# distinct knob against being loosened: evidence, rate, confirmation, defects.
57KNOWN_BAD: Final[tuple[GateScenario, ...]] = (
58    # No record at all — the cold-start floor (zero evidence).
59    GateScenario(
60        name="no record", decided=0, merged=0, determined=0, defects=0
61    ),
62    # A perfect but thin record — a 100% rate on too few PRs is not a record.
63    GateScenario(
64        name="thin record", decided=1, merged=1, determined=1, defects=0
65    ),
66    # Plenty of volume, but the steward rejects half — a poor track record.
67    GateScenario(
68        name="low approval rate",
69        decided=20,
70        merged=10,
71        determined=10,
72        defects=0,
73    ),
74    # A spotless rate, but almost no merge has a *confirmed* outcome yet — the
75    # defect bearing has no evidence behind it.
76    GateScenario(
77        name="unconfirmed merges",
78        decided=20,
79        merged=20,
80        determined=1,
81        defects=0,
82    ),
83    # Everything clears except reality: one confirmed merge broke or reverted.
84    GateScenario(
85        name="a defect on record",
86        decided=20,
87        merged=20,
88        determined=10,
89        defects=1,
90    ),
91)
92 
93 
94def gate_escapes(policy: AutonomyPolicy) -> tuple[str, ...]:
95    """Names of known-bad scenarios the live gate would *wrongly* grant.
96 
97    Empty is healthy: every deliberately-bad class is refused. Non-empty is the
98    alarm — the gate has loosened (in code or, more likely, in config) enough to
99    trust a class it must not. Pure over the supplied policy.
100    """
101    return tuple(
102        s.name
103        for s in KNOWN_BAD
104        if class_earned(
105            decided=s.decided,
106            merged=s.merged,
107            determined=s.determined,
108            defects=s.defects,
109            policy=policy,
110        )[0]
111    )

The gate_selftest activity runs the battery against the live AutonomySettings every scan tick, logging healthy at INFO (a heartbeat) and any escape at ERROR (surfaces in ClickStack). Repo-independent and cheap.

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

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

⋯ 511 lines hidden (lines 1–511)1"""Activities: the effect interpreters that wrap the adapters.
2 
3Each activity is the impure boundary for one effect. Adapter imports are LAZY
4(inside the bodies) so the model and HTTP stacks never enter a workflow sandbox;
5the pure policy and domain imports stay at module level. Activities return
6domain values — the workflow wraps them into events and feeds the pure state
7machine. The activity signatures' types are evaluated at runtime by Temporal's
8data converter, so the domain imports here are deliberately not deferred.
9"""
10 
11from __future__ import annotations
12 
13import json
14import logging
15import tempfile
16from pathlib import Path
17from typing import TYPE_CHECKING, assert_never
18 
19from temporalio import activity
20 
21from froot.domain.candidate import Candidate
22from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
23from froot.domain.ci import CIStatus
24from froot.domain.determinism import AnalysisResult, FrontierVerdict
25from froot.domain.loop import Loop
26from froot.domain.pull_request import PullRequestRef
27from froot.domain.repo import TargetRepo
28from froot.policy.compose import pr_labels, pull_request_draft
29from froot.policy.determinism import analyze_workflow_surface
30from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_review_workflow_id,
34)
35from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
36from froot.workflow.types import (
  AdjudicateInput,
  AutoMergeInput,
  CiCheckInput,
  CloseInput,
  DispatchInput,
  DispatchReviewInput,
  GateReviewInput,
  GateSelfTestInput,
  JudgeInput,
  MergeInput,
  OpenPrInput,
  PostReviewInput,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
53)
54 
55if TYPE_CHECKING:
  from froot.ports.protocols import PackageManager
57 
58_log = logging.getLogger("froot.outcome")
59_review_log = logging.getLogger("froot.review")
60_reconcile_log = logging.getLogger("froot.reconcile")
61_scan_log = logging.getLogger("froot.scan")
62_gate_log = logging.getLogger("froot.gate")
63 
64 
65def _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
68 
69 
70async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
75) -> tuple[int, tuple[Candidate, ...]]:
  """Gather this loop's signal from the checkout and select its candidates.
77 
  The one genuinely per-loop seam: dependency-patch reads the available
  upgrades and picks the highest patch; security-patch reads the installed,
  asks OSV for advisories, and picks the lowest version clearing each. The
  impure sources are lazy-imported per arm so neither drags the other's stack
  into a sandbox. Both feed a pure selection policy.
83 
  Returns ``(considered, candidates)`` — ``considered`` is the size of the
  upstream signal (available upgrades / advisories found) so the scan can make
  its selectivity legible (how much was seen versus how much was kept).
  """
  match loop:
      case Loop.DEPENDENCY_PATCH:
          from froot.policy.candidates import select_patch_candidates
91 
          upgrades = await package_manager.list_upgrades(target, manifest_dir)
          return len(upgrades), select_patch_candidates(upgrades)
      case Loop.SECURITY_PATCH:
          return await _select_security_candidates(
              target, package_manager, manifest_dir
          )
  assert_never(loop)
99 
100 
101async def _select_security_candidates(
  target: TargetRepo, package_manager: PackageManager, manifest_dir: Path
103) -> tuple[int, tuple[Candidate, ...]]:
  """Security signal: installed set, OSV advisories, clearing targets.
105 
  ``considered`` is the count of advisories OSV returned for the installed
  set — the vulnerabilities in scope this tick, before selection narrows to
  the ones a forward-stable bump can actually clear.
  """
  from froot.adapters.osv import OsvAdvisorySource
  from froot.policy.candidates import select_security_candidates
112 
  installed = await package_manager.list_installed(target, manifest_dir)
  advisories = await OsvAdvisorySource().advisories(installed)
  return len(advisories), select_security_candidates(installed, advisories)
116 
117 
118@activity.defn
119async def scan_candidates(
  params: ScanCandidatesInput,
121) -> tuple[Candidate, ...]:
  """Check out the repo and select this loop's candidates.
123 
  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
132 
  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
166 
167 
168@activity.defn
169async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
171 
  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
183 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      return UnknownVerdict(rationale="No changelog could be fetched.")
  try:
      return await PydanticAiJudge().judge(changelog, params.loop)
  except Exception as exc:
      activity.logger.warning(
          "changelog judge unavailable for %s; degrading to unknown: %r",
          params.candidate.package,
          exc,
      )
      return UnknownVerdict(
          rationale=f"Changelog judge unavailable ({type(exc).__name__})."
      )
198 
199 
200@activity.defn
201async def gate_review(params: GateReviewInput) -> ChangelogVerdict:
  """Independently deep-review a bump at the gate; fail-closed to a hold.
203 
  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
214 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      verdict: ChangelogVerdict = 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",
              params.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
248 
249 
250@activity.defn
251async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
255 
  forge = GitHubForge()
  package_manager = package_manager_for(params.target.ecosystem)
  branch = branch_name(params.candidate, params.loop)
  existing = await forge.find_open_pull_request(params.target, branch)
  if existing is not None:
      return existing
  draft = pull_request_draft(
      params.target, params.candidate, params.verdict, params.loop
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      await package_manager.apply_patch_bump(
          params.candidate, _manifest_dir(params.target, workspace)
      )
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
273 
274 
275@activity.defn
276async 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
279 
  return await GitHubForge().ci_status(params.target, params.head_sha)
281 
282 
283@activity.defn
284async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update.
286 
  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
294 
  outcome = params.outcome
  labels = (
      *pr_labels(params.loop),
      env_label(ModelSettings().ollama_model),
  )
  await GitHubForge().add_labels(params.target, outcome.pr.number, labels)
  _log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": params.loop.value,
              "repo": params.target.repo.slug,
              "package": outcome.candidate.package,
              "from": str(outcome.candidate.current),
              "to": str(outcome.candidate.target),
              "changelog": outcome.verdict.kind,
              "ci": outcome.ci.kind,
              "ci_passed": outcome.ci_passed,
              "pr": outcome.pr.number,
              "pr_url": outcome.pr.url,
          }
      )
  )
318 
319 
320@activity.defn
321async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
323 
  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
330 
  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
335 
  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
354 
355 
356@activity.defn
357async def auto_merge_eligible(params: AutoMergeInput) -> bool:
  """Whether this (repo, loop) class has earned the auto-merge grant.
359 
  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
369 
  from froot.config.settings import AutonomySettings, ModelSettings
  from froot.dashboard import github_source, read_model
  from froot.policy.environment import environment_slug
373 
  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),
  )
394 
395 
396@activity.defn
397async def merge_pull_request(params: MergeInput) -> None:
  """Auto-merge an earned, clean+green bump's PR (the acting gate's write).
399 
  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
405 
  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,
          }
      )
  )
421 
422 
423@activity.defn
424async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
426 
  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
434 
  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),
          }
      )
  )
456 
457 
458@activity.defn
459async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
461 
  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
474 
  if not BehaviorSettings().reconcile:
      return 0
477 
  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)
509 
510 
511@activity.defn
512async def gate_selftest(params: GateSelfTestInput) -> tuple[str, ...]:
  """Run the adversarial gate probe against the live policy; alarm on escape.
514 
  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
525 
  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
541 
⋯ 95 lines hidden (lines 542–636)542 
543# ── The determinism reviewer (the transitive ring) ──────────────────────────
544@activity.defn
545async 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
548 
  return await GitHubForge().list_open_pull_requests(target)
550 
551 
552@activity.defn
553async 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
557 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
560 
  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
576 
577 
578@activity.defn
579async 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
584 
  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)
595 
596 
597@activity.defn
598async def adjudicate_frontier(
  params: AdjudicateInput,
600) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
603 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
609 
610 
611@activity.defn
612async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
615 
  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

src/froot/workflow/scan_workflow.py · 95 lines

src/froot/workflow/scan_workflow.py95 lines · Python

⋯ 59 lines hidden (lines 1–59)1"""The self-scheduling scan loop — froot's durable trigger (per repo).
2 
3One long-lived workflow per target repo. Each tick checks out the repo, selects
4the patch candidates, and dispatches a bump loop per candidate (idempotent, so
5re-dispatch is a no-op); then it sleeps and continues-as-new, keeping history
6bounded to one tick. There is no stored seen-set — the outstanding work is
7re-derived from the repo each tick (derive, never store), and the per-bump
8workflow id makes re-proposing an already-handled bump a no-op.
9 
10A one-shot run (``continuous=False``, the default) performs a single tick and
11returns; production starts it once with ``continuous=True`` and it runs forever.
12"""
13 
14from __future__ import annotations
15 
16from datetime import timedelta
17 
18from temporalio import workflow
19 
20with workflow.unsafe.imports_passed_through():
  from froot.workflow import activities
  from froot.workflow.constants import (
      ACTIVITY_TIMEOUT,
      DISPATCH_TIMEOUT,
      TOOL_RETRY,
  )
  from froot.workflow.types import (
      DispatchInput,
      GateSelfTestInput,
      ReconcileInput,
      ScanCandidatesInput,
      ScanParams,
      ScanResult,
  )
35 
36 
37@workflow.defn
38class ScanWorkflow:
  """The durable dependency-scan loop (one tick per continue-as-new)."""
40 
  @workflow.run
  async def run(self, params: ScanParams) -> ScanResult:
      """Scan, dispatch each, reconcile stale PRs, then loop or return."""
      candidates = await workflow.execute_activity(
          activities.scan_candidates,
          ScanCandidatesInput(target=params.target, loop=params.loop),
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=TOOL_RETRY,
      )
      for candidate in candidates:
          await workflow.execute_activity(
              activities.dispatch_bump,
              DispatchInput(
                  target=params.target,
                  candidate=candidate,
                  loop=params.loop,
              ),
              start_to_close_timeout=DISPATCH_TIMEOUT,
              retry_policy=TOOL_RETRY,
          )
      # Close this loop's PRs a newer target superseded — re-derived from the
      # repo each tick, same as the scan, and scoped to the loop's namespace.
      reconciled = await workflow.execute_activity(
          activities.reconcile_open_prs,
          ReconcileInput(target=params.target, loop=params.loop),
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=TOOL_RETRY,
      )
      # The adversarial probe (§2.11): every tick, confirm the *live* gate
      # still refuses a battery of known-bad classes. Cheap, repo-independent,
      # and alarms in telemetry the moment a config drift loosens the gate.
      await workflow.execute_activity(
          activities.gate_selftest,
          GateSelfTestInput(target=params.target, loop=params.loop),
          start_to_close_timeout=ACTIVITY_TIMEOUT,
          retry_policy=TOOL_RETRY,
      )
      result = ScanResult(
          found=len(candidates),
          dispatched=len(candidates),
          reconciled=reconciled,
      )
      if not params.continuous:
          return result
      # Durable loop: sleep, then restart fresh. continue_as_new raises, so
      # nothing runs after it and history stays bounded to one tick.
⋯ 9 lines hidden (lines 87–95)87        await workflow.sleep(timedelta(seconds=params.interval_seconds))
      workflow.continue_as_new(
          ScanParams(
              target=params.target,
              interval_seconds=params.interval_seconds,
              continuous=True,
              loop=params.loop,
          )
      )

Bearing 4 — the independent deep review at the gate (P4.4)

The fourth leg (§3.7's sampled deep review, here run on every auto-merge candidate). The first changelog judge frames risk neutrally and never gates; this one is the opposite — an independent, adversarial re-read whose approval is required to merge. Its system prompt makes the stance asymmetric: the burden is on the changelog to prove safety.

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

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

⋯ 50 lines hidden (lines 1–50)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 
33_SYSTEM_PROMPT = (
34    "You assess the changelog of a dependency upgrade for a code-maintenance "
35    "bot. The bot proposes the bump either way; your job is only to frame the "
36    "risk for the human reviewer.\n"
37    "Return one verdict:\n"
38    "- clean: the notes describe only fixes / docs / internal changes with no "
39    "behavioral or API impact.\n"
40    "- risky: the notes hint at behavior change, deprecations, or anything a "
41    "reviewer should look at closely; list the concerns.\n"
42    "- unknown: the notes are empty or uninformative.\n"
43    "Quote-or-omit: base 'risky' concerns on what the text actually says; do "
44    "not speculate. Keep the rationale to one or two sentences."
45)
46 
47# The gate reviewer's stance is adversarial and asymmetric: this bump is about
48# to merge with NO human in the loop, so the burden is on the changelog to prove
49# it is safe, not on the reviewer to prove it is dangerous. The same _Assessment
50# shape, read as approve/hold (clean = approve).
51_GATE_SYSTEM_PROMPT = (
52    "You are the LAST line of defense before a dependency upgrade auto-merges "
53    "with no human review. Re-read the changelog adversarially and decide "
54    "whether it is safe to merge unattended.\n"
55    "Return one verdict:\n"
56    "- clean: ONLY if the notes clearly describe nothing beyond fixes / docs / "
57    "internal changes — no behavioral change, no API change, no deprecation, "
58    "no ambiguity. This approves the merge.\n"
59    "- risky: any hint of behavior change, removal, deprecation, or surface a "
60    "reviewer should see; list the concerns. This holds the PR.\n"
61    "- unknown: the notes are empty, uninformative, or you are unsure. This "
62    "holds the PR.\n"
63    "The burden is on the changelog to prove safety: when in doubt, do NOT "
64    "return clean. Base concerns on what the text says; keep it to one or two "
65    "sentences."
66)
67 
68 
69def _loop_context(loop: Loop) -> str:
70    """The one line that tells the model what kind of bump it is judging."""
71    match loop:
72        case Loop.DEPENDENCY_PATCH:
73            return (
74                "This is a patch-level upgrade; weigh whether the notes hide "
⋯ 60 lines hidden (lines 75–134)75                "any behavioral change behind a 'patch'."
76            )
77        case Loop.SECURITY_PATCH:
78            return (
79                "This is a SECURITY upgrade that may cross a minor or major "
80                "line to clear a vulnerability; weigh breaking changes the "
81                "human should know before merging — the fix is still worth it."
82            )
83    assert_never(loop)
84 
85 
86class _Assessment(BaseModel):
87    """The model's structured output, mapped to a domain verdict."""
88 
89    verdict: Literal["clean", "risky", "unknown"]
90    rationale: str
91    concerns: list[str] = Field(default_factory=list)
92 
93 
94def assessment_to_verdict(assessment: _Assessment) -> ChangelogVerdict:
95    """Map the model's structured assessment to a domain verdict (pure)."""
96    match assessment.verdict:
97        case "clean":
98            return CleanVerdict(rationale=assessment.rationale)
99        case "risky":
100            return RiskyVerdict(
101                rationale=assessment.rationale,
102                concerns=tuple(assessment.concerns),
103            )
104        case "unknown":
105            return UnknownVerdict(rationale=assessment.rationale)
106    assert_never(assessment.verdict)
107 
108 
109def _gate_model() -> Model:
110    """Build the gate reviewer's model: the override, else the judge model."""
111    from froot.config.settings import ModelSettings
112 
113    return build_model(model_name=ModelSettings().gate_review_model or None)
114 
115 
116def _changelog_prompt(changelog: Changelog, loop: Loop) -> str:
117    """The shared user prompt: the bump's context and its changelog text."""
118    return (
119        f"{_loop_context(loop)}\n"
120        f"Package: {changelog.package}\n"
121        f"Target version: {changelog.version}\n"
122        f"Changelog:\n{changelog.text}"
123    )
124 
125 
126class PydanticAiJudge:
127    """A :class:`~froot.ports.protocols.ModelJudge` backed by Pydantic AI.
128 
129    Two agents share the structured ``_Assessment`` output but differ in stance:
130    the neutral framing judge (:meth:`judge`) and the adversarial gate reviewer
131    (:meth:`gate_review`), each its own model so a steward can make the deep
132    review independent in capability, not just prompt.
133    """
134 
135    def __init__(
136        self, model: Model | None = None, gate_model: Model | None = None
137    ) -> None:
138        """Build both agents.
139 
140        ``model`` defaults to the configured Ollama; ``gate_model`` to the
141        gate-review override, else ``model``, else the gate-review model.
142        """
143        self._agent: Agent[None, _Assessment] = Agent(
144            model or build_model(),
145            output_type=_Assessment,
146            system_prompt=_SYSTEM_PROMPT,
147        )
148        self._gate_agent: Agent[None, _Assessment] = Agent(
149            gate_model or model or _gate_model(),
150            output_type=_Assessment,
151            system_prompt=_GATE_SYSTEM_PROMPT,
152        )
153 
154    async def judge(
155        self, changelog: Changelog, loop: Loop = Loop.DEPENDENCY_PATCH
156    ) -> ChangelogVerdict:
157        """Assess a changelog into a verdict, framed by the loop."""
158        result = await self._agent.run(_changelog_prompt(changelog, loop))
159        return assessment_to_verdict(result.output)
160 
161    async def gate_review(
162        self, changelog: Changelog, loop: Loop = Loop.DEPENDENCY_PATCH
163    ) -> ChangelogVerdict:
164        """Adversarially re-review a bump at the gate (clean = approve)."""
165        result = await self._gate_agent.run(_changelog_prompt(changelog, loop))
166        return assessment_to_verdict(result.output)

The gate_review activity fetches the changelog and runs the second pass, fail-CLOSED throughout: a missing changelog or a model error returns a non-clean verdict, so an unreviewable bump never merges unattended (the opposite disposition from judge_changelog, which degrades-to-proceed).

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

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

⋯ 200 lines hidden (lines 1–200)1"""Activities: the effect interpreters that wrap the adapters.
2 
3Each activity is the impure boundary for one effect. Adapter imports are LAZY
4(inside the bodies) so the model and HTTP stacks never enter a workflow sandbox;
5the pure policy and domain imports stay at module level. Activities return
6domain values — the workflow wraps them into events and feeds the pure state
7machine. The activity signatures' types are evaluated at runtime by Temporal's
8data converter, so the domain imports here are deliberately not deferred.
9"""
10 
11from __future__ import annotations
12 
13import json
14import logging
15import tempfile
16from pathlib import Path
17from typing import TYPE_CHECKING, assert_never
18 
19from temporalio import activity
20 
21from froot.domain.candidate import Candidate
22from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
23from froot.domain.ci import CIStatus
24from froot.domain.determinism import AnalysisResult, FrontierVerdict
25from froot.domain.loop import Loop
26from froot.domain.pull_request import PullRequestRef
27from froot.domain.repo import TargetRepo
28from froot.policy.compose import pr_labels, pull_request_draft
29from froot.policy.determinism import analyze_workflow_surface
30from froot.policy.naming import (
  branch_name,
  bump_workflow_id,
  pr_review_workflow_id,
34)
35from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
36from froot.workflow.types import (
  AdjudicateInput,
  AutoMergeInput,
  CiCheckInput,
  CloseInput,
  DispatchInput,
  DispatchReviewInput,
  GateReviewInput,
  GateSelfTestInput,
  JudgeInput,
  MergeInput,
  OpenPrInput,
  PostReviewInput,
  PrReviewParams,
  ReconcileInput,
  RecordInput,
  ScanCandidatesInput,
53)
54 
55if TYPE_CHECKING:
  from froot.ports.protocols import PackageManager
57 
58_log = logging.getLogger("froot.outcome")
59_review_log = logging.getLogger("froot.review")
60_reconcile_log = logging.getLogger("froot.reconcile")
61_scan_log = logging.getLogger("froot.scan")
62_gate_log = logging.getLogger("froot.gate")
63 
64 
65def _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
68 
69 
70async def _select_candidates(
  loop: Loop,
  target: TargetRepo,
  package_manager: PackageManager,
  manifest_dir: Path,
75) -> tuple[int, tuple[Candidate, ...]]:
  """Gather this loop's signal from the checkout and select its candidates.
77 
  The one genuinely per-loop seam: dependency-patch reads the available
  upgrades and picks the highest patch; security-patch reads the installed,
  asks OSV for advisories, and picks the lowest version clearing each. The
  impure sources are lazy-imported per arm so neither drags the other's stack
  into a sandbox. Both feed a pure selection policy.
83 
  Returns ``(considered, candidates)`` — ``considered`` is the size of the
  upstream signal (available upgrades / advisories found) so the scan can make
  its selectivity legible (how much was seen versus how much was kept).
  """
  match loop:
      case Loop.DEPENDENCY_PATCH:
          from froot.policy.candidates import select_patch_candidates
91 
          upgrades = await package_manager.list_upgrades(target, manifest_dir)
          return len(upgrades), select_patch_candidates(upgrades)
      case Loop.SECURITY_PATCH:
          return await _select_security_candidates(
              target, package_manager, manifest_dir
          )
  assert_never(loop)
99 
100 
101async def _select_security_candidates(
  target: TargetRepo, package_manager: PackageManager, manifest_dir: Path
103) -> tuple[int, tuple[Candidate, ...]]:
  """Security signal: installed set, OSV advisories, clearing targets.
105 
  ``considered`` is the count of advisories OSV returned for the installed
  set — the vulnerabilities in scope this tick, before selection narrows to
  the ones a forward-stable bump can actually clear.
  """
  from froot.adapters.osv import OsvAdvisorySource
  from froot.policy.candidates import select_security_candidates
112 
  installed = await package_manager.list_installed(target, manifest_dir)
  advisories = await OsvAdvisorySource().advisories(installed)
  return len(advisories), select_security_candidates(installed, advisories)
116 
117 
118@activity.defn
119async def scan_candidates(
  params: ScanCandidatesInput,
121) -> tuple[Candidate, ...]:
  """Check out the repo and select this loop's candidates.
123 
  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
132 
  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
166 
167 
168@activity.defn
169async def judge_changelog(params: JudgeInput) -> ChangelogVerdict:
  """Fetch the candidate's changelog and get the model's typed verdict.
171 
  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
183 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      return UnknownVerdict(rationale="No changelog could be fetched.")
  try:
      return await PydanticAiJudge().judge(changelog, params.loop)
  except Exception as exc:
      activity.logger.warning(
          "changelog judge unavailable for %s; degrading to unknown: %r",
          params.candidate.package,
          exc,
      )
      return UnknownVerdict(
          rationale=f"Changelog judge unavailable ({type(exc).__name__})."
      )
198 
199 
200@activity.defn
201async def gate_review(params: GateReviewInput) -> ChangelogVerdict:
  """Independently deep-review a bump at the gate; fail-closed to a hold.
203 
  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
214 
  changelog = await HttpChangelogSource().fetch(params.candidate)
  if changelog is None:
      verdict: ChangelogVerdict = 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",
              params.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
⋯ 389 lines hidden (lines 248–636)248 
249 
250@activity.defn
251async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
  """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
  from froot.adapters.github import GitHubForge
  from froot.adapters.registry import package_manager_for
255 
  forge = GitHubForge()
  package_manager = package_manager_for(params.target.ecosystem)
  branch = branch_name(params.candidate, params.loop)
  existing = await forge.find_open_pull_request(params.target, branch)
  if existing is not None:
      return existing
  draft = pull_request_draft(
      params.target, params.candidate, params.verdict, params.loop
  )
  with tempfile.TemporaryDirectory() as tmp:
      workspace = Path(tmp)
      await forge.checkout(params.target, workspace)
      await package_manager.apply_patch_bump(
          params.candidate, _manifest_dir(params.target, workspace)
      )
      await forge.push_branch(workspace, branch, draft.title)
  return await forge.open_pull_request(params.target, draft)
273 
274 
275@activity.defn
276async 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
279 
  return await GitHubForge().ci_status(params.target, params.head_sha)
281 
282 
283@activity.defn
284async def record_outcome(params: RecordInput) -> None:
  """Label the PR and log the run telemetry — the signal-update.
286 
  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
294 
  outcome = params.outcome
  labels = (
      *pr_labels(params.loop),
      env_label(ModelSettings().ollama_model),
  )
  await GitHubForge().add_labels(params.target, outcome.pr.number, labels)
  _log.info(
      json.dumps(
          {
              "event": "loop_outcome",
              "loop": params.loop.value,
              "repo": params.target.repo.slug,
              "package": outcome.candidate.package,
              "from": str(outcome.candidate.current),
              "to": str(outcome.candidate.target),
              "changelog": outcome.verdict.kind,
              "ci": outcome.ci.kind,
              "ci_passed": outcome.ci_passed,
              "pr": outcome.pr.number,
              "pr_url": outcome.pr.url,
          }
      )
  )
318 
319 
320@activity.defn
321async def dispatch_bump(params: DispatchInput) -> None:
  """Start the bump loop for a candidate (idempotent per bump identity).
323 
  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
330 
  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
335 
  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
354 
355 
356@activity.defn
357async def auto_merge_eligible(params: AutoMergeInput) -> bool:
  """Whether this (repo, loop) class has earned the auto-merge grant.
359 
  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
369 
  from froot.config.settings import AutonomySettings, ModelSettings
  from froot.dashboard import github_source, read_model
  from froot.policy.environment import environment_slug
373 
  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),
  )
394 
395 
396@activity.defn
397async def merge_pull_request(params: MergeInput) -> None:
  """Auto-merge an earned, clean+green bump's PR (the acting gate's write).
399 
  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
405 
  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,
          }
      )
  )
421 
422 
423@activity.defn
424async def close_pull_request(params: CloseInput) -> None:
  """Comment why, then close a red bump's PR and delete its branch.
426 
  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
434 
  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),
          }
      )
  )
456 
457 
458@activity.defn
459async def reconcile_open_prs(params: ReconcileInput) -> int:
  """Close this loop's PRs that a newer candidate or the base has overtaken.
461 
  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
474 
  if not BehaviorSettings().reconcile:
      return 0
477 
  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)
509 
510 
511@activity.defn
512async def gate_selftest(params: GateSelfTestInput) -> tuple[str, ...]:
  """Run the adversarial gate probe against the live policy; alarm on escape.
514 
  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
525 
  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
541 
542 
543# ── The determinism reviewer (the transitive ring) ──────────────────────────
544@activity.defn
545async 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
548 
  return await GitHubForge().list_open_pull_requests(target)
550 
551 
552@activity.defn
553async 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
557 
  from froot.workflow.pr_review_workflow import PrReviewWorkflow
  from froot.workflow.temporal_client import client, task_queue
560 
  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
576 
577 
578@activity.defn
579async 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
584 
  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)
595 
596 
597@activity.defn
598async def adjudicate_frontier(
  params: AdjudicateInput,
600) -> tuple[FrontierVerdict, ...]:
  """Run the model over each frontier item; return aligned verdicts."""
  from froot.adapters.determinism_judge import DeterminismFrontierJudge
603 
  judge = DeterminismFrontierJudge()
  verdicts: list[FrontierVerdict] = []
  for item in params.frontier:
      verdicts.append(await judge.adjudicate(item))
  return tuple(verdicts)
609 
610 
611@activity.defn
612async def post_review(params: PostReviewInput) -> str | None:
  """Upsert the advisory comment (when there are findings); log the ledger."""
  from froot.adapters.github import GitHubForge
615 
  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

The environment fingerprint — trust resets on a model swap (P4.3b)

Trust is conditional on the environment that earned it (§3.7). Every recorded PR is stamped with a froot-env:<model> label; the dashboard and the gate count only PRs from the current environment, so a judge-model swap resets each class's record rather than letting trust silently ride across a capability change.

src/froot/policy/environment.py · 56 lines

src/froot/policy/environment.py56 lines · Python

1"""The environment fingerprint — what trust was earned *under* (§3.7).
2 
3MHE's "conditional" property: trust belongs to a task class *in a specific
4environment*, and when the environment changes some trust resets. The dominant
5piece of froot's judgment environment is the judge model — a changelog verdict
6from ``gemma4:26b`` is not the same evidence as one from ``gemma4:e4b`` — so a
7track record earned under one model must not silently transfer to another
8("The model underneath changes... Trust drops from auto-merge back to
9propose-plus-review until the new model re-earns it", §3.7).
10 
11froot stamps each PR with the environment it was opened under (a ``froot-env:``
12label), and the gate counts only the record earned under the *current*
13environment. A model swap therefore resets every class to unearned until fresh
14PRs accrue — exactly the calibration window §3.7 prescribes, and with no failure
15required, just a changed anchor. The slug is kept human-readable rather than an
16opaque hash so a steward can see at a glance which environment earned the trust.
17 
18Today the fingerprint captures the model dimension; tools, permissions, and
19codebase shape are the other dimensions §3.7 names, and fold in here when froot
20learns to track them.
21"""
22 
23from __future__ import annotations
24 
25import re
26from typing import Final
27 
28_LABEL_PREFIX: Final = "froot-env:"
29_SLUG_RE: Final = re.compile(r"[^a-z0-9]+")
30 
31 
32def environment_slug(model: str) -> str:
33    """A short, legible slug of the judgment environment (currently the model).
34 
35    ``gemma4:26b`` -> ``gemma4-26b``. Readable on purpose: the steward should be
36    able to read which environment a class's trust was earned under.
37    """
38    slug = _SLUG_RE.sub("-", model.strip().lower()).strip("-")
39    return slug or "unknown"
40 
41 
42def env_label(model: str) -> str:
43    """The label that stamps a PR with the environment it was opened under."""
44    return _LABEL_PREFIX + environment_slug(model)
45 
46 
47def env_from_labels(names: set[str]) -> str | None:
48    """The environment slug recovered from a PR's ``froot-env:`` label.
49 
50    ``None`` if the PR carries no such stamp (opened before stamping existed — a
51    prior environment by definition, so it does not count for the current one).
52    """
53    for name in names:
54        if name.startswith(_LABEL_PREFIX):
55            return name[len(_LABEL_PREFIX) :]
56    return None

Honest framing & verification

The acting flip made the gate load-bearing, so the prose that called it a dry run that 'acts on nothing' became the most dangerous kind of stale doc — it understates the authority a steward is granting. The dashboard footer now states the conditional merge authority plainly and the class-gate note names all four legs.

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

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

⋯ 708 lines hidden (lines 1–708)1"""Render the view model to one self-contained HTML page (pure).
2 
3All CSS is inline, there is no JavaScript, and the page makes no network
4request of its own — it is a static projection of an already-computed
class:`~froot.dashboard.model.DashboardModel`. Every dynamic value is
6HTML-escaped at the boundary. The ordering is trust-first (is it alive → track
7record → oracle → judgment → the human's queue), so it reads top to bottom.
8"""
9 
10from __future__ import annotations
11 
12from datetime import UTC, datetime, timedelta
13from html import escape
14from typing import TYPE_CHECKING
15 
16if TYPE_CHECKING:
  from froot.dashboard.model import (
      BumpRow,
      ClassGate,
      DashboardModel,
      ReviewLoop,
      ReviewRow,
      RunTelemetry,
      ScanLoop,
  )
26 
27_CSS = """
root{--fg:#1a1a1a;--mut:#6b6b6b;--line:#e4e4e4;--bg:#fff;--ok:#1a7f37;
29--warn:#9a6700;--bad:#cf222e;--accent:#0969da;--card:#fafafa}
30@media(prefers-color-scheme:dark){:root{--fg:#e6e6e6;--mut:#9a9a9a;
31--line:#262626;--bg:#0d0d0d;--ok:#3fb950;--warn:#d29922;--bad:#f85149;
32--accent:#58a6ff;--card:#141414}}
33*{box-sizing:border-box}
34body{margin:0;background:var(--bg);color:var(--fg);
35font:15px/1.55 system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}
36main{max-width:820px;margin:0 auto;padding:34px 20px 72px}
37h1{font-size:22px;margin:0;letter-spacing:-.01em}
tag{color:var(--mut);margin:3px 0 0;font-size:13px}
meta{color:var(--mut);font-size:12px;margin:12px 0 0}
sources{display:flex;flex-wrap:wrap;gap:16px;margin:10px 0 0;font-size:12px}
41section{margin:30px 0 0}
42h2{font-size:12px;text-transform:uppercase;letter-spacing:.08em;
43color:var(--mut);margin:0 0 12px;font-weight:600;
44border-bottom:1px solid var(--line);padding-bottom:6px}
stats{display:flex;flex-wrap:wrap;gap:26px}
stat .n{font-size:26px;font-weight:600;line-height:1.1}
stat .l{color:var(--mut);font-size:12px;margin-top:2px}
dot{display:inline-block;width:9px;height:9px;border-radius:50%;
49margin-right:7px}
dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}
dot.bad{background:var(--bad)}.dot.mute{background:var(--mut)}
52table{width:100%;border-collapse:collapse;font-size:13px}
53th,td{text-align:left;padding:6px 12px 6px 0;
54border-bottom:1px solid var(--line);vertical-align:top}
55th{color:var(--mut);font-weight:600;font-size:11px;text-transform:uppercase;
56letter-spacing:.04em}
mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}
58a{color:var(--accent);text-decoration:none}a:hover{text-decoration:underline}
row{display:flex;align-items:baseline;gap:8px;padding:4px 0}
mut{color:var(--mut)}.ok{color:var(--ok)}.warn{color:var(--warn)}
bad{color:var(--bad)}
note{color:var(--mut);font-size:12px;margin:10px 0 0}
63footer{margin:44px 0 0;padding-top:16px;border-top:1px solid var(--line);
64color:var(--mut);font-size:12px;line-height:1.7}
65footer b{color:var(--fg);font-weight:600}
66/* loop groups — collapsible (collapsed by default), two-level hierarchy */
67details.loop{margin:24px 0 0}
68summary.loophead{display:flex;align-items:baseline;gap:6px 12px;flex-wrap:wrap;
69cursor:pointer;list-style:none;margin:0;padding:0 0 7px;
70border-bottom:2px solid var(--fg)}
71summary.loophead::-webkit-details-marker{display:none}
72summary.loophead::before{content:"\\25B8";color:var(--mut);margin-right:5px;
73font-size:11px;position:relative;top:-1px}
74details.loop[open]>summary.loophead::before{content:"\\25BE"}
75summary.loophead h2{font-size:16px;text-transform:none;letter-spacing:-.01em;
76color:var(--fg);border:0;margin:0;padding:0;font-weight:700}
77summary.loophead .sub{color:var(--mut);font-size:12px}
78summary.loophead .glance{margin-left:auto;display:flex;flex-wrap:wrap;
79align-items:baseline;gap:3px 14px;font-size:12px;color:var(--mut)}
80summary.loophead .glance b{color:var(--fg);font-weight:600}
81/* the shared-infra group reads as not-a-loop: lighter, dashed, muted */
82details.loop.shared>summary.loophead{border-bottom:1px dashed var(--line)}
83details.loop.shared>summary.loophead h2{font-size:13px;color:var(--mut);
84font-weight:600;text-transform:uppercase;letter-spacing:.06em}
85/* subsections sit tighter inside a loop; the loop rule carries the weight */
86details.loop section{margin:18px 0 0}
87details.loop section>h2{border-bottom:0;padding-bottom:0;margin-bottom:9px}
88/* foldable framing — the MHE notes, one click away, so the page scans clean */
89details.why{margin:9px 0 0}
90details.why>summary{color:var(--accent);font-size:12px;cursor:pointer;
91list-style:none}
92details.why>summary::-webkit-details-marker{display:none}
93details.why>summary::before{content:"\\002b  why";font-weight:600}
94details.why[open]>summary::before{content:"\\2212  why"}
95details.why .note{margin:6px 0 0}
96"""
97 
98_CI_CLASS = {
  "passed": "ok",
  "failed": "bad",
  "absent": "mut",
  "timed_out": "warn",
103}
104_VERDICT_CLASS = {"clean": "ok", "risky": "warn", "unknown": "mut"}
105_POST_MERGE_CLASS = {
  "held": "ok",
  "broke": "bad",
  "reverted": "bad",
  "unknown": "mut",
110}
111 
112 
113def _aware(when: datetime) -> datetime:
  """Treat a stray naive timestamp as UTC so arithmetic never raises."""
  return when if when.tzinfo is not None else when.replace(tzinfo=UTC)
116 
117 
118def _ago(when: datetime | None, now: datetime) -> str:
  """A compact 'time since' label (``6h ago``)."""
  if when is None:
      return "—"
  secs = (now - _aware(when)).total_seconds()
  if secs < 90:
      return "just now"
  if secs < 5400:
      return f"{int(secs // 60)}m ago"
  if secs < 129600:
      return f"{int(secs // 3600)}h ago"
  return f"{int(secs // 86400)}d ago"
130 
131 
132def _until(when: datetime | None, now: datetime) -> str:
  """A compact 'time until' label (``in 18h`` / ``due now``)."""
  if when is None:
      return "—"
  secs = (_aware(when) - now).total_seconds()
  if secs <= 0:
      return "due now"
  if secs < 5400:
      return f"in {int(secs // 60)}m"
  if secs < 129600:
      return f"in {int(secs // 3600)}h"
  return f"in {int(secs // 86400)}d"
144 
145 
146def _dot(kind: str) -> str:
  """A status dot span (``ok`` / ``warn`` / ``bad`` / ``mute``)."""
  return f'<span class="dot {kind}"></span>'
149 
150 
151def _tag(value: str | None, classes: dict[str, str]) -> str:
  """A small coloured label for a verdict/CI value, or an em-dash."""
  if value is None:
      return '<span class="mut">—</span>'
  cls = classes.get(value, "mut")
  return f'<span class="{cls}">{escape(value)}</span>'
157 
158 
159def _stat(n: object, label: str) -> str:
  return (
      f'<div class="stat"><div class="n">{escape(str(n))}</div>'
      f'<div class="l">{escape(label)}</div></div>'
  )
164 
165 
166def _why(note_html: str) -> str:
  """Fold a framing note behind a native ``<details>`` so the page scans."""
  return f'<details class="why"><summary></summary>{note_html}</details>'
169 
170 
171def _glance(n: object, label: str) -> str:
  """One at-a-glance item for a loop header (``<b>12</b> proposed``)."""
  return f"<span><b>{escape(str(n))}</b> {escape(label)}</span>"
174 
175 
176def _header(model: DashboardModel) -> str:
  now = model.generated_at
  repos = ", ".join(model.repos_configured) or "none configured"
  dots = "".join(
      f"<span>{_dot('ok' if s.ok else 'bad')}"
      f'{escape(s.name)} <span class="mut">{escape(s.detail)}</span></span>'
      for s in model.sources
  )
  stamp = now.strftime("%Y-%m-%d %H:%M UTC")
  return (
      "<header>"
      "<h1>froot</h1>"
      '<p class="tag">durable maintenance loops &middot; '
      "reputation read-model</p>"
      f'<p class="meta">watching <span class="mono">{escape(repos)}</span>'
      f" &middot; generated {escape(stamp)} &middot; "
      "derived live, stored nowhere</p>"
      f'<div class="sources">{dots}</div>'
      "</header>"
  )
196 
197 
198def _heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.scan_interval_seconds
201 
  def line(loop: ScanLoop) -> str:
      if loop.live:
          dot, tail = "ok", ""
          if loop.last_tick is not None:
              nxt = _aware(loop.last_tick) + timedelta(seconds=interval)
              last = _ago(loop.last_tick, now)
              tail = (
                  f' <span class="mut">&middot; last {last}'
                  f" &middot; next {_until(nxt, now)}</span>"
              )
      else:
          dot = "bad" if loop.status in ("terminated", "none") else "warn"
          tail = f' <span class="mut">&middot; {escape(loop.status)}</span>'
      return (
          f'<div class="row">{_dot(dot)}'
          f'<span class="mono">{escape(loop.repo)}</span>'
          f' <span class="mut">{escape(loop.loop)}</span>{tail}</div>'
      )
220 
  if not model.scan_loops:
      body = '<p class="note">No repos configured (FROOT_REPOS unset).</p>'
  else:
      body = "".join(line(loop) for loop in model.scan_loops)
  return f"<section><h2>Is it alive?</h2>{body}</section>"
226 
227 
228def _track_record(model: DashboardModel) -> str:
  t = model.track_record
  rate = "—" if t.merge_rate is None else f"{t.merge_rate * 100:.0f}%"
  ttm = (
      "—"
      if t.median_ttm_minutes is None
      else f"{t.median_ttm_minutes:.0f} min"
  )
  stats = "".join(
      (
          _stat(t.opened, "proposed"),
          _stat(t.merged, "merged"),
          _stat(t.open_now, "awaiting"),
          _stat(t.closed_unmerged, "closed"),
          _stat(rate, "merge rate"),
          _stat(ttm, "median time-to-merge"),
      )
  )
  note = (
      '<p class="note">Merge rate is the Stage-1 signal &mdash; narrow to '
      "npm patch bumps by construction. It counts a human merge, not a "
      "confirmed good outcome: revert tracking is a later loop, so merge is "
      "not yet proof of success.</p>"
  )
  return (
      "<section><h2>Track record &middot; the reputation</h2>"
      f'<div class="stats">{stats}</div>{_why(note)}</section>'
  )
256 
257 
258def _rate_cell(rate: float | None) -> str:
  """A class's approval rate as a percent, or an em-dash if none decided."""
  if rate is None:
      return '<span class="mut">—</span>'
  return f"{rate * 100:.0f}%"
263 
264 
265def _earned_cell(g: ClassGate) -> str:
  """A class's gate standing: a green 'earned', else the muted blocker.
267 
  A model change resets the record (§3.7), which can read as a mysterious drop
  to zero; when prior-environment history exists, it is named so the reset is
  legible rather than alarming.
  """
  reset = (
      f' <span class="mut">&middot; reset: {g.prior_env_decided} under a '
      "prior env</span>"
      if g.prior_env_decided
      else ""
  )
  if g.earned:
      return f'{_dot("ok")}<span class="ok">earned</span>{reset}'
  blocker = escape(g.blocker or "not earned")
  return f'{_dot("mut")}<span class="mut">{blocker}</span>{reset}'
282 
283 
284def _budget_cell(g: ClassGate) -> str:
  """The gate in steward-time: approvals/week now &rarr; reclaimable/week."""
  return (
      f'<span class="mono">{g.approvals_per_week:.1f}</span> appr'
      f' <span class="mut">&rarr; {g.reclaim_per_week:.1f} reclaimable</span>'
  )
290 
291 
292def _defect_cell(g: ClassGate) -> str:
  """The post-merge defect bearing: rate over confirmed-held outcomes."""
  if g.determined == 0:
      return '<span class="mut">— (0 confirmed)</span>'
  rate = g.defect_rate if g.defect_rate is not None else 0.0
  cls = "ok" if g.defects == 0 else "bad"
  return (
      f'<span class="{cls}">{rate * 100:.0f}%</span>'
      f' <span class="mut">({g.defects}/{g.determined})</span>'
  )
302 
303 
304def _class_gates(model: DashboardModel) -> str:
  """The earned-autonomy panel — the gate's per-class standing.
306 
  Renders whether each (repo, loop) class has earned its gate move (the two
  record bearings: rate + defect rate) and the steward-budget that move
  reclaims. On an allowlisted repo a met gate auto-merges; elsewhere — the
  default — it is the advisory shadow gate a steward watches.
  """
  gates = model.class_gates
  if not gates:
      body = (
          '<p class="note">No classes yet &mdash; a (repo, loop) earns a '
          "gate move from its own track record.</p>"
      )
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(g.repo)}</td>'
          f'<td class="mut">{escape(g.loop)}</td>'
          f"<td>{g.decided} in {g.window_days}d</td>"
          f"<td>{_rate_cell(g.merge_rate)}</td>"
          f"<td>{_defect_cell(g)}</td>"
          f"<td>{_earned_cell(g)}</td>"
          f"<td>{_budget_cell(g)}</td>"
          "</tr>"
          for g in gates
      )
      body = (
          "<table><thead><tr><th>repo</th><th>loop</th><th>decided</th>"
          "<th>rate</th><th>defect</th><th>gate</th><th>budget / week</th>"
          "</tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  note = (
      '<p class="note">Each <b>class</b> is one loop on one repo &mdash; a '
      "distinct trust unit (a security patch is not a version bump, and they "
      "never share a record). A class earns its gate by <b>triangulation</b> "
      "(&sect;3.8): a high enough <b>approval rate</b> over enough "
      "recently-<b>decided</b> PRs <i>and</i> a low <b>defect rate</b> over "
      "enough confirmed-held merges &mdash; two bearings that fail "
      "differently (the rate to rubber-stamping, the defect rate to a weak "
      "oracle), so they can&rsquo;t both lie at once. The window keeps trust "
      "recent. <b>Budget</b> reads the gate in steward-time: approvals/week "
      "the class costs now &rarr; the reclaimable slice a move hands back. "
      "Two further legs guard the live merge beyond this panel: an "
      "<b>adversarial gate self-test</b> (a known-bad class must never be "
      "granted) and an <b>independent deep review</b> of each bump at merge. "
      "Where a repo is <b>allowlisted</b> a met gate auto-merges; the "
      "allowlist is empty by default &mdash; the revocable switch &mdash; so "
      "elsewhere this stays advisory.</p>"
  )
  return (
      "<section><h2>Earned autonomy &middot; the gate</h2>"
      f"{body}{_why(note)}</section>"
  )
359 
360 
361def _verification(model: DashboardModel) -> str:
  v = model.verification
  stats = "".join(
      (
          _stat(v.passed, "CI passed"),
          _stat(v.failed, "CI failed"),
          _stat(v.absent, "no checks"),
          _stat(v.timed_out, "timed out"),
          _stat(v.unknown, "unknown"),
      )
  )
  if v.with_reading == 0:
      note = '<p class="note">No CI readings yet.</p>'
  else:
      note = _why(
          f'<p class="note">A real oracle reported on '
          f"<b>{v.oracle_existed}</b> of {v.with_reading} bumps with a "
          'reading. <span class="mut">&lsquo;no checks&rsquo; means CI was '
          "absent &mdash; not a pass; never conflated.</span>"
          "</p>"
      )
  return (
      "<section><h2>Verification &middot; CI is the oracle</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
386 
387 
388def _reliability(model: DashboardModel) -> str:
  r = model.reliability
  rate = "—" if r.defect_rate is None else f"{r.defect_rate * 100:.0f}%"
  stats = "".join(
      (
          _stat(r.held, "held"),
          _stat(r.broke, "broke"),
          _stat(r.reverted, "reverted"),
          _stat(r.unverified, "unverified"),
          _stat(rate, "defect rate"),
      )
  )
  if r.determined == 0:
      note = (
          '<p class="note">No post-merge outcomes determined yet '
          f"(last {r.window_days}d). A merge is only a *success* once it "
          "holds; until then it&rsquo;s just merged.</p>"
      )
  else:
      note = _why(
          '<p class="note">Did the merge <b>hold</b>? &mdash; the outcome '
          "leg, the only bearing that watches what actually shipped. "
          f"<b>{r.determined}</b> of the recent merges were classifiable; "
          "the defect rate is a <b>floor</b>, not the truth: it sees the "
          "branch&rsquo;s CI and git-reverts, but a <i>manual</i> or "
          "<i>bundled</i> revert (most of them) is invisible. "
          '<span class="mut">&lsquo;unverified&rsquo; means no branch '
          "oracle &mdash; never counted as held.</span></p>"
      )
  return (
      "<section><h2>Reliability &middot; did the merge hold?</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
421 
422 
423def _probes(model: DashboardModel) -> str:
  p = model.probes
  stats = "".join(
      (
          _stat(p.caught, "caught"),
          _stat(p.escaped, "escaped"),
          _stat(p.pending, "pending"),
      )
  )
  if p.total == 0:
      note = (
          '<p class="note">No adversarial probes yet. A <b>canary</b> '
          "&mdash; a deliberately-bad bump &mdash; tests whether the "
          "guardrail still bites; it needs no volume to be informative, so "
          "it carries weight while the rate-based bearings are thin "
          "(&sect;2.11).</p>"
      )
  elif p.escaped:
      note = (
          f'<p class="note"><span class="bad">&#9888; <b>{p.escaped}</b> '
          "canary(s) <b>escaped</b> &mdash; a known-bad bump landed, so the "
          "guardrail has a hole.</span> This is the alarm the probe exists "
          "to raise.</p>"
      )
  else:
      note = _why(
          '<p class="note">Every canary was <b>caught</b> &mdash; the '
          "guardrail refused the planted bad bumps. Probes are scored on a "
          "strict bar (a canary must never merge) and kept <b>out</b> of the "
          "track record and defect rate, so a synthetic failure never "
          "pollutes the real bearings.</p>"
      )
  return (
      "<section><h2>Adversarial probes &middot; does the guardrail bite?</h2>"
      f'<div class="stats">{stats}</div>{note}</section>'
  )
459 
460 
461def _judgment(model: DashboardModel) -> str:
  j = model.judgment
  stats = "".join(
      (
          _stat(j.clean, "clean"),
          _stat(j.risky, "risky"),
          _stat(j.unknown, "unknown"),
          _stat(j.none, "no verdict"),
      )
  )
  if j.clean_but_failed or j.flagged_but_passed:
      note = (
          f'<p class="note">Calibration: <b>{j.clean_but_failed}</b> '
          "&lsquo;clean&rsquo; bumps whose CI failed, "
          f"<b>{j.flagged_but_passed}</b> flagged bumps whose CI passed.</p>"
      )
  else:
      note = (
          '<p class="note">The model&rsquo;s only job is the changelog '
          "verdict; the spine proposes the bump either way.</p>"
      )
  return (
      "<section><h2>Model judgment &middot; the one model call</h2>"
      f'<div class="stats">{stats}</div>{_why(note)}</section>'
  )
486 
487 
488def _shadow_badge(row: BumpRow) -> str:
  """Whether this open PR meets its class's auto-merge grant.
490 
  These PRs are still open, so on an allowlisted repo a met grant would
  already have merged: a green 'would auto-merge' here means the grant is met
  but the repo is not allowlisted (or a deeper check held it), so it is still
  yours. Otherwise the first blocker, muted — the held reason to fix.
  """
  if row.would_auto_merge:
      return '<span class="ok">would auto-merge</span>'
  reason = row.held_reason or "held"
  return f'<span class="mut">held &middot; {escape(reason)}</span>'
500 
501 
502def _gate(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.gate:
      body = '<p class="note">Queue empty &mdash; nothing awaiting you.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.package)}</td>'
          f"<td>{escape(_ago(row.opened_at, now))}</td>"
          f"<td>{_shadow_badge(row)}</td>"
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.gate
      )
      note = _why(
          '<p class="note">These PRs are open and <b>yours</b>. A green '
          "&lsquo;would auto-merge&rsquo; means this PR met its "
          "class&rsquo;s grant &mdash; on an allowlisted repo the loop would "
          "have merged it already, so its presence here means the repo is "
          "not allowlisted (or a deeper check held it). The verdict rests on "
          "the single CI "
          "oracle and a merge-rate record, which (esp. for a security patch) "
          "is not the same as a confirmed-good or security-reviewed outcome. "
          "You own every open merge.</p>"
      )
      body = (
          "<table><thead><tr><th>package</th><th>waiting</th>"
          "<th>shadow gate</th><th>pr</th>"
          f"</tr></thead><tbody>{rows}</tbody></table>{note}"
      )
  return (
      "<section><h2>Approval gate &middot; what a human owns</h2>"
      f"{body}</section>"
  )
536 
537 
538def _bumps(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.bumps:
      body = '<p class="note">No bumps proposed yet.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.package)}</td>'
          f'<td class="mono mut">{escape(row.from_version or "?")} &rarr; '
          f"{escape(row.to_version)}</td>"
          f"<td>{_tag(row.verdict, _VERDICT_CLASS)}</td>"
          f"<td>{_tag(row.ci, _CI_CLASS)}</td>"
          f"<td>{_state_tag(row.state)}</td>"
          f"<td>{_tag(row.post_merge, _POST_MERGE_CLASS)}</td>"
          f'<td class="mut">{escape(_ago(row.opened_at, now))}</td>'
          f"<td>{_pr_link(row)}</td>"
          "</tr>"
          for row in model.bumps
      )
      body = (
          "<table><thead><tr><th>package</th><th>bump</th><th>verdict</th>"
          "<th>ci</th><th>state</th><th>held?</th><th>opened</th><th>pr</th>"
          "</tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  return f"<section><h2>Bumps &middot; the detail</h2>{body}</section>"
564 
565 
566def _failures(model: DashboardModel) -> str:
  if not model.failures:
      return ""
  now = model.generated_at
  rows = "".join(
      "<tr>"
      f'<td class="mono">{escape(_short_id(f.workflow_id))}</td>'
      f"<td>{_state_tag(f.kind)}</td>"
      f'<td class="mut">{escape(f.reason or "—")}</td>'
      f'<td class="mut">{escape(_ago(f.when, now))}</td>'
      "</tr>"
      for f in model.failures
  )
  return (
      "<section><h2>Failures &middot; where the loop did not close</h2>"
      "<table><thead><tr><th>bump</th><th>kind</th><th>reason</th>"
      f"<th>when</th></tr></thead><tbody>{rows}</tbody></table></section>"
  )
584 
585 
586def _telem_glance(model: DashboardModel) -> str:
  """The telemetry group's at-a-glance (or an ``unavailable`` marker)."""
  t = model.telemetry
  if not t.available:
      return '<span class="mut">unavailable</span>'
  return _glance(t.total_spans, f"spans / {t.window_days}d")
592 
593 
594def _telemetry(model: DashboardModel) -> str:
  """The telemetry body — header lives in the shared loop group above it."""
  t: RunTelemetry = model.telemetry
  if not t.available:
      return (
          "<section>"
          '<p class="note">Unavailable (not configured or unreachable). '
          "GitHub + Temporal carry the dashboard regardless.</p></section>"
      )
  now = model.generated_at
  if t.activities:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(a.name)}</td>'
          f"<td>{a.count}</td>"
          f'<td class="mut">{a.avg_ms:.0f} ms</td>'
          f'<td class="mut">{a.max_ms:.0f} ms</td>'
          "</tr>"
          for a in t.activities
      )
      table = (
          "<table><thead><tr><th>activity</th><th>runs</th><th>avg</th>"
          f"<th>max</th></tr></thead><tbody>{rows}</tbody></table>"
      )
  else:
      table = '<p class="note">No froot spans in the window.</p>'
  summary = (
      f'<p class="note">{t.total_spans} spans &middot; '
      f"{t.error_spans} errored &middot; last activity "
      f"{escape(_ago(t.last_activity, now))} &middot; "
      f"{t.window_days}-day window.</p>"
  )
  return f"<section>{summary}{table}</section>"
627 
628 
629def _review_heartbeat(model: DashboardModel) -> str:
  now = model.generated_at
  interval = model.review_interval_seconds
632 
  def line(loop: ReviewLoop) -> str:
      if loop.live:
          dot, tail = "ok", ""
          if loop.last_tick is not None:
              nxt = _aware(loop.last_tick) + timedelta(seconds=interval)
              last = _ago(loop.last_tick, now)
              tail = (
                  f' <span class="mut">&middot; last {last}'
                  f" &middot; next {_until(nxt, now)}</span>"
              )
      else:
          dot = "bad" if loop.status in ("terminated", "none") else "warn"
          tail = f' <span class="mut">&middot; {escape(loop.status)}</span>'
      return (
          f'<div class="row">{_dot(dot)}'
          f'<span class="mono">{escape(loop.repo)}</span>{tail}</div>'
      )
650 
  if not model.review_loops:
      body = (
          '<p class="note">No determinism-review loops running '
          "(the transitive ring watches the @workflow.defn repos).</p>"
      )
  else:
      body = "".join(line(loop) for loop in model.review_loops)
  return f"<section><h2>Is it alive?</h2>{body}</section>"
659 
660 
661def _review_record(model: DashboardModel) -> str:
  r = model.review_record
  stats = "".join(
      (
          _stat(r.reviewed, "reviewed"),
          _stat(r.flagged, "flagged"),
          _stat(r.clean, "clean"),
          _stat(r.hazards, "hazards"),
          _stat(r.repos_covered, "repos covered"),
      )
  )
  note = (
      '<p class="note">The transitive ring: it chases first-party helper '
      "calls out of each workflow to catch a hazard the lexical CI kernel "
      "can&rsquo;t see. <b>Advisory</b> &mdash; the blocking gate stays the "
      "kernel&rsquo;s CI check. The hazard-resolved rate (was a flag gone on "
      "a later commit?) is a later loop; it needs accumulated history.</p>"
  )
  return (
      "<section><h2>Track record &middot; the reputation</h2>"
      f'<div class="stats">{stats}</div>{_why(note)}</section>'
  )
683 
684 
685def _reviews(model: DashboardModel) -> str:
  now = model.generated_at
  if not model.reviews:
      body = '<p class="note">No PRs reviewed yet.</p>'
  else:
      rows = "".join(
          "<tr>"
          f'<td class="mono">{escape(row.repo)}</td>'
          f"<td>{_review_pr_link(row)}</td>"
          f'<td class="mono mut">{escape((row.head_sha or "")[:7]) or "—"}'
          "</td>"
          f"<td>{_findings_cell(row)}</td>"
          f'<td class="mut">{escape(_ago(row.reviewed_at, now))}</td>'
          "</tr>"
          for row in model.reviews
      )
      body = (
          "<table><thead><tr><th>repo</th><th>pr</th><th>head</th>"
          "<th>findings</th><th>reviewed</th></tr></thead>"
          f"<tbody>{rows}</tbody></table>"
      )
  return f"<section><h2>Reviews &middot; the detail</h2>{body}</section>"
707 
708 
709def _footer() -> str:
  return (
      "<footer>"
      "<b>Authority envelope.</b> froot holds <b>write authority</b> "
      "everywhere (it opens PRs) and conditional <b>merge authority</b> "
      "only on an allowlisted repo where the class has earned its gate "
      "&mdash; the allowlist is empty by default, so commit authority is "
      "none until a steward opts a repo in. Trust, when granted, is earned, "
      "narrow to a single loop on a single repo (a version bump is not a "
      "security patch; they never share a record), conditional on its "
      "environment (judge "
      '<span class="mono">gemma4:26b</span>, lockfile-only regen), revocable '
      "(the allowlist), and time-expiring. A met gate auto-merges only "
      "after passing an independent deep review; everywhere not allowlisted "
      "the same verdict is the advisory <b>shadow gate</b> a steward "
      "watches.<br>"
      "Everything above is derived on this request from GitHub (outcomes) + "
      "Temporal (runs) + ClickHouse (telemetry). froot keeps no database; "
⋯ 157 lines hidden (lines 727–883)727        "reload to recompute."
      "</footer>"
  )
730 
731 
732def _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>'
736 
737 
738def _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>'
742 
743 
744def _findings_cell(row: ReviewRow) -> str:
  """A review's findings: 'clean', or the hazard count + rules + comment."""
  if row.findings == 0:
      return '<span class="ok">clean</span>'
  rules = (
      f' <span class="mono mut">{escape(", ".join(row.rules))}</span>'
      if row.rules
      else ""
  )
  comment = (
      f' <a href="{escape(row.comment_url, quote=True)}">comment</a>'
      if row.comment_url
      else ""
  )
  noun = "hazard" if row.findings == 1 else "hazards"
  return f'<span class="bad">{row.findings} {noun}</span>{rules}{comment}'
760 
761 
762def _state_tag(state: str) -> str:
  cls = {
      "merged": "ok",
      "open": "warn",
      "closed": "mut",
      "terminated": "bad",
      "failed": "bad",
      "timed_out": "bad",
      "canceled": "warn",
  }.get(state, "mut")
  return f'<span class="{cls}">{escape(state)}</span>'
773 
774 
775def _short_id(workflow_id: str) -> str:
  """Drop the ``froot-bump-`` prefix for a readable failures row."""
  return workflow_id.removeprefix("froot-bump-")
778 
779 
780def _dep_glance(model: DashboardModel) -> str:
  """The dependency-patch loop's at-a-glance, shown in its header."""
  t = model.track_record
  live = sum(1 for loop in model.scan_loops if loop.live)
  total = len(model.scan_loops)
  rate = "—" if t.merge_rate is None else f"{t.merge_rate * 100:.0f}%"
  items = [
      _glance(f"{live}/{total}", "live"),
      _glance(t.opened, "proposed"),
      _glance(rate, "merged"),
  ]
  if model.class_gates:
      earned = sum(1 for g in model.class_gates if g.earned)
      items.append(_glance(f"{earned}/{len(model.class_gates)}", "earned"))
  if t.open_now:
      items.append(_glance(t.open_now, "awaiting you"))
  return "".join(items)
797 
798 
799def _det_glance(model: DashboardModel) -> str:
  """The determinism-review loop's at-a-glance, shown in its header."""
  r = model.review_record
  return "".join(
      (
          _glance(r.repos_covered, "repos"),
          _glance(r.reviewed, "reviewed"),
          _glance(r.hazards, "hazards"),
      )
  )
809 
810 
811def _loop(
  title: str,
  sub: str,
  glance: str,
  sections: tuple[str, ...],
  *,
  shared: bool = False,
818) -> str:
  """A collapsible group: a prominent header + at-a-glance, collapsed.
820 
  The header (with its at-a-glance) is the ``<summary>``, so a collapsed group
  still shows the headline; expanding reveals the subsections. ``shared`` is a
  non-loop infra group (telemetry) with a lighter, distinct header.
  """
  cls = "loop shared" if shared else "loop"
  head = (
      '<summary class="loophead">'
      f"<h2>{escape(title)}</h2>"
      f'<span class="sub">{sub}</span>'
      f'<span class="glance">{glance}</span>'
      "</summary>"
  )
  return f'<details class="{cls}">{head}{"".join(sections)}</details>'
834 
835 
836def page(model: DashboardModel) -> str:
  """Render the whole dashboard as one self-contained HTML document."""
  parts = (
      _header(model),
      _loop(
          "Dependency-patch",
          "npm + uv &middot; scan &rarr; bump &rarr; CI &rarr; merge",
          _dep_glance(model),
          (
              _heartbeat(model),
              _track_record(model),
              _class_gates(model),
              _verification(model),
              _reliability(model),
              _probes(model),
              _judgment(model),
              _gate(model),
              _bumps(model),
              _failures(model),
          ),
      ),
      _loop(
          "Determinism review",
          "the transitive ring &middot; advisory",
          _det_glance(model),
          (
              _review_heartbeat(model),
              _review_record(model),
              _reviews(model),
          ),
      ),
      _loop(
          "Run telemetry",
          "ClickHouse &middot; trace-derived, best-effort",
          _telem_glance(model),
          (_telemetry(model),),
          shared=True,
      ),
      _footer(),
  )
  return (
      '<!doctype html><html lang="en"><head><meta charset="utf-8">'
      '<meta name="viewport" content="width=device-width,initial-scale=1">'
      "<title>froot &middot; read-model</title>"
      f"<style>{_CSS}</style></head><body><main>"
      + "".join(parts)
      + "</main></body></html>"
  )

Full suite green: 351 tests, mypy clean, ruff format + lint clean, both determinism gates green (no hazard reaches a workflow). New discriminating tests cover the acting flip (state machine + workflow + e2e), the gate self-test (escapes under loosened config), the deep-review veto (eligible but held → not merged), and fail-closed degradation.

tests/test_bump_workflow.py · 225 lines

tests/test_bump_workflow.py225 lines · Python

⋯ 171 lines hidden (lines 1–171)1"""Integration tests for the bump workflow on a time-skipping test server.
2 
3The activities are mocked (by matching activity name), so this exercises the
4real spine: the driver loop, the effect interpretation, and the durable CI wait
5(whose ``workflow.sleep`` the time-skipping server fast-forwards).
6"""
7 
8from __future__ import annotations
9 
10from collections.abc import Callable
11from typing import Any
12 
13from temporalio import activity
14from temporalio.client import Client
15from temporalio.testing import WorkflowEnvironment
16from temporalio.worker import Worker
17 
18from froot.domain.changelog import ChangelogVerdict, CleanVerdict, RiskyVerdict
19from froot.domain.ci import (
20    CIFailed,
21    CIPassed,
22    CIPending,
23    CIStatus,
24    CITimedOut,
25)
26from froot.domain.outcome import LoopOutcome
27from froot.domain.pull_request import PullRequestRef
28from froot.workflow.bump_workflow import BumpWorkflow
29from froot.workflow.runtime import DATA_CONVERTER
30from froot.workflow.types import (
31    AutoMergeInput,
32    BumpParams,
33    CiCheckInput,
34    CloseInput,
35    GateReviewInput,
36    JudgeInput,
37    MergeInput,
38    OpenPrInput,
39    RecordInput,
40)
41from tests.support import make_candidate, make_pr, make_repo
42 
43_TASK_QUEUE = "froot-test"
44# A scripted CI reply sequence the mock pops through (then falls back to green).
45_ci_replies: list[CIStatus] = []
46# PR numbers the close-on-red path closed, in order.
47_closed: list[int] = []
48# PR numbers the acting gate auto-merged, in order, and the eligibility the
49# mock gate reports (a test flips it to exercise the auto-merge path).
50_merged: list[int] = []
51_eligible: bool = False
52# The verdict the mock gate reviewer returns (clean = approve the merge; a test
53# flips it to a hold to exercise the deep-review veto).
54_gate_verdict: ChangelogVerdict = CleanVerdict(rationale="re-read clean")
55 
56 
57@activity.defn(name="judge_changelog")
58async def _mock_judge(params: JudgeInput) -> ChangelogVerdict:
59    return CleanVerdict(rationale="patch only")
60 
61 
62@activity.defn(name="open_pull_request")
63async def _mock_open_pr(params: OpenPrInput) -> PullRequestRef:
64    return make_pr(number=7)
65 
66 
67@activity.defn(name="check_ci")
68async def _mock_check_ci(params: CiCheckInput) -> CIStatus:
69    return _ci_replies.pop(0) if _ci_replies else CIPassed()
70 
71 
72@activity.defn(name="record_outcome")
73async def _mock_record(params: RecordInput) -> None:
74    return None
75 
76 
77@activity.defn(name="close_pull_request")
78async def _mock_close(params: CloseInput) -> None:
79    _closed.append(params.pr.number)
80 
81 
82@activity.defn(name="auto_merge_eligible")
83async def _mock_eligible(params: AutoMergeInput) -> bool:
84    return _eligible
85 
86 
87@activity.defn(name="gate_review")
88async def _mock_gate_review(params: GateReviewInput) -> ChangelogVerdict:
89    return _gate_verdict
90 
91 
92@activity.defn(name="merge_pull_request")
93async def _mock_merge(params: MergeInput) -> None:
94    _merged.append(params.pr.number)
95 
96 
97_MOCKS: list[Callable[..., Any]] = [
98    _mock_judge,
99    _mock_open_pr,
100    _mock_check_ci,
101    _mock_record,
102    _mock_close,
103    _mock_eligible,
104    _mock_gate_review,
105    _mock_merge,
106]
107 
108 
109async def _pydantic_client(env: WorkflowEnvironment) -> Client:
110    config = env.client.config()
111    config["data_converter"] = DATA_CONVERTER
112    return Client(**config)
113 
114 
115async def _run_bump(*, close_on_red: bool = True) -> LoopOutcome:
116    async with await WorkflowEnvironment.start_time_skipping() as env:
117        client = await _pydantic_client(env)
118        async with Worker(
119            client,
120            task_queue=_TASK_QUEUE,
121            workflows=[BumpWorkflow],
122            activities=_MOCKS,
123        ):
124            return await client.execute_workflow(
125                BumpWorkflow.run,
126                BumpParams(
127                    target=make_repo(),
128                    candidate=make_candidate(),
129                    close_on_red=close_on_red,
130                ),
131                id="bump-test",
132                task_queue=_TASK_QUEUE,
133            )
134 
135 
136async def test_happy_path_green():
137    _ci_replies.clear()
138    _closed.clear()
139    outcome = await _run_bump()
140    assert outcome.pr.number == 7
141    assert outcome.ci_passed
142    assert isinstance(outcome.ci, CIPassed)
143    assert _closed == []  # green PRs are left for the human, never closed
144 
145 
146async def test_ci_failed_closes_pr_and_records_failure():
147    _ci_replies.clear()
148    _closed.clear()
149    _ci_replies.append(CIFailed(failing=("build",)))
150    outcome = await _run_bump()
151    assert not outcome.ci_passed
152    assert isinstance(outcome.ci, CIFailed)
153    # Close-on-red (the default): the red PR is closed, and the outcome is
154    # still recorded (the loop reached its terminal state).
155    assert _closed == [7]
156 
157 
158async def test_ci_failed_left_open_when_close_on_red_off():
159    _ci_replies.clear()
160    _closed.clear()
161    _ci_replies.append(CIFailed(failing=("build",)))
162    outcome = await _run_bump(close_on_red=False)
163    assert isinstance(outcome.ci, CIFailed)
164    # close-on-red off: the red PR is recorded but left open for the human.
165    assert _closed == []
166 
167 
168async def test_ci_pending_then_pass_waits_durably():
169    _ci_replies.clear()
170    _ci_replies.extend([CIPending(), CIPending(), CIPassed()])
171    outcome = await _run_bump()
172    assert outcome.ci_passed
173 
174 
175async def test_ci_timeout_when_never_resolves():
176    _ci_replies.clear()
177    _ci_replies.extend([CIPending()] * 100)
178    outcome = await _run_bump()
179    assert isinstance(outcome.ci, CITimedOut)
180 
181 
182async def test_green_clean_auto_merges_when_eligible_and_review_approves():
183    # The acting gate, full path: a clean+green bump on an earned class is
184    # deep-reviewed and, on approval, auto-merged by the loop.
185    global _eligible
186    _ci_replies.clear()
187    _closed.clear()
188    _merged.clear()
189    _eligible = True
190    try:
191        outcome = await _run_bump()
192    finally:
193        _eligible = False
194    assert outcome.ci_passed
195    assert _merged == [7]  # the loop merged it
196    assert _closed == []
197 
198 
199async def test_eligible_but_gate_review_holds_does_not_merge():
200    # The deep-review veto: the class is earned and CI is green, but the
201    # independent gate reviewer holds -> the loop records and leaves it open.
202    global _eligible, _gate_verdict
203    _ci_replies.clear()
204    _merged.clear()
205    _closed.clear()
206    _eligible = True
207    _gate_verdict = RiskyVerdict(rationale="a hidden behavior change")
208    try:
209        outcome = await _run_bump()
210    finally:
211        _eligible = False
⋯ 14 lines hidden (lines 212–225)212        _gate_verdict = CleanVerdict(rationale="re-read clean")
213    assert outcome.ci_passed
214    assert _merged == []  # the review vetoed the merge
215    assert _closed == []  # green: left open for the human, not closed
216 
217 
218async def test_green_clean_not_merged_when_class_not_eligible():
219    # The default: no grant -> propose-only, the loop never merges (and never
220    # even reaches the deep review).
221    _ci_replies.clear()
222    _merged.clear()
223    outcome = await _run_bump()
224    assert outcome.ci_passed
225    assert _merged == []