froot · Phase 1·

Closing the first loop

A code-level review of one PR · hardening the dependency-patch loop

froot closes its first loop

Make the loop actually close — the SPEC's own precondition for replicating it.

froot is a Temporal worker that proposes dependency patches as pull requests and lets the repo's own CI verify each one. Its charter says: close one loop end-to-end before adding the next. This PR closes the gaps that kept the first loop from fully closing — a red PR left rotting, a superseded PR never cleaned up, a fault that retried forever, a model outage that could stall the spine. Expand any block to read the surrounding code.

🆕 new this PR ♻️ the pure spine · shape unchanged 🐙 GitHub · external truth

workstreams

new loop state (Closing)

219

tests green

determinism gates

review findings folded in

What "hardening" actually meant

froot's loop is built on a pure, effect-driven state machine: each transition emits one effect as data, a durable activity runs it, and the resulting event feeds the next transition. The four gaps this PR closes map cleanly onto that shape rather than fighting it.

- Close-on-red. A failed CI now closes the PR (and deletes its branch) before recording — modeled as a new effect on the state machine, not buried in an activity. - Reconcile. A pure policy closes PRs a newer patch superseded or the base already satisfied, re-derived from the repo each scan tick. - Failure discipline. Every activity runs under a bounded retry so a persistent fault surfaces instead of looping forever; the model degrades to "unknown" so it can never stall the spine. - Proof. An end-to-end test drives the real activities through both terminal shapes, and a coverage floor plus a transitive-determinism gate lock the bar in.

The tour follows the change the way the loop runs it: the new effect as data, the state machine that emits it, the forge that performs it, the reconcile policy beside it, the activities that bind them to the world, the retry discipline around them, and the proof.

Close-on-red1🎬 Three small additions to the union types

The close, modeled as data

src/froot/domain/effects.pysrc/froot/domain/events.pysrc/froot/domain/state.py

froot's defining choice is that every externally-visible action is an effect — a frozen value the pure state machine emits and a durable activity interprets. Closing a PR (and deleting its branch) is exactly that kind of action, so it becomes a first-class effect rather than a side effect hidden inside the record step. That is one new member in each of three discriminated unions.

The new ClosePullRequest effect carries the PR and the failing check names, and joins the Effect union.

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

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

⋯ 45 lines hidden (lines 1–45)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 PatchCandidate
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: PatchCandidate
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: PatchCandidate
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 RecordOutcome(Frozen):
61    """Record the closed-loop outcome (label the PR, emit run telemetry)."""
62 
63    kind: Literal["record_outcome"] = "record_outcome"
64    outcome: LoopOutcome
65 
66 
67# What the spine should do after a transition.
68Effect = Annotated[
69    JudgeChangelog
70    | OpenPullRequest
71    | AwaitCi
72    | ClosePullRequest
73    | RecordOutcome,
74    Field(discriminator="kind"),
75]

The matching event is the spine's report that the close happened. It carries no data — the state it returns to already holds the outcome to record.

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

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

⋯ 42 lines hidden (lines 1–42)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 OutcomeRecorded(Frozen):
50    """The outcome was recorded; the loop has nothing left to do."""
51 
52    kind: Literal["outcome_recorded"] = "outcome_recorded"
53 
54 
55# A decided input to the loop state machine.
56LoopEvent = Annotated[
57    ChangelogJudged
58    | PullRequestReady
59    | CiResolved
60    | PullRequestClosed
61    | OutcomeRecorded,
62    Field(discriminator="kind"),
63]

And one new state sits between "CI came back red" and "outcome recorded": the PR is being closed. It carries the already-built LoopOutcome, so once the close lands the loop records exactly what it would have recorded anyway.

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

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

⋯ 46 lines hidden (lines 1–46)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 PatchCandidate
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: PatchCandidate
28 
29 
30class Judged(Frozen):
31    """The changelog has been assessed; ready to open the PR."""
32 
33    kind: Literal["judged"] = "judged"
34    candidate: PatchCandidate
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: PatchCandidate
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 Recorded(Frozen):
61    """Terminal: CI resolved and the outcome was recorded."""
62 
63    kind: Literal["recorded"] = "recorded"
64    outcome: LoopOutcome
65 
66 
67# The state of a single bump's loop.
68BumpState = Annotated[
69    Discovered | Judged | AwaitingCi | Closing | Recorded,
70    Field(discriminator="kind"),
71]

Close-on-red2🎬 The pure transition, read closely

One branch, the invariant intact

src/froot/policy/state_machine.py

The driver enforces a hard rule: every transition emits exactly one effect. Closing then recording is two actions, so it cannot be one transition — it is a sequence through the new Closing state. The branch is the only new logic.

Red CI with close-on-red on routes through Closing; every other terminal reading records straight away. Closing then records.

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

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

⋯ 153 lines hidden (lines 1–153)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
12: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, is_terminal
24from froot.domain.effects import (
25    AwaitCi,
26    ClosePullRequest,
27    Effect,
28    JudgeChangelog,
29    OpenPullRequest,
30    RecordOutcome,
31)
32from froot.domain.events import (
33    ChangelogJudged,
34    CiResolved,
35    LoopEvent,
36    OutcomeRecorded,
37    PullRequestClosed,
38    PullRequestReady,
39)
40from froot.domain.outcome import LoopOutcome
41from froot.domain.state import (
42    AwaitingCi,
43    BumpState,
44    Closing,
45    Discovered,
46    Judged,
47    Recorded,
48)
49 
50if TYPE_CHECKING:
51    from froot.domain.candidate import PatchCandidate
52 
53 
54class TransitionKind(StrEnum):
55    """The disposition of an :func:`advance` call."""
56 
57    ADVANCED = "advanced"
58    IGNORED = "ignored"
59    REJECTED = "rejected"
60 
61 
62class Transition(Frozen):
63    """The result of a transition.
64 
65    Attributes:
66        kind: ``ADVANCED`` (moved and/or emitted effects), ``IGNORED`` (a legal
67            no-op, e.g. the terminal acknowledgement), or ``REJECTED`` (the
68            event is not valid in this state).
69        next: The resulting state (unchanged for ``IGNORED``/``REJECTED``).
70        effects: The effects the spine should run, in order. Empty terminates
71            the loop driver.
72        reason: A short explanation for an ``IGNORED``/``REJECTED`` transition.
73    """
74 
75    kind: TransitionKind
76    next: BumpState
77    effects: tuple[Effect, ...] = ()
78    reason: str | None = None
79 
80 
81def _advanced(nxt: BumpState, *effects: Effect) -> Transition:
82    return Transition(kind=TransitionKind.ADVANCED, next=nxt, effects=effects)
83 
84 
85def _rejected(state: BumpState, reason: str) -> Transition:
86    return Transition(kind=TransitionKind.REJECTED, next=state, reason=reason)
87 
88 
89def start(candidate: PatchCandidate) -> Transition:
90    """The opening transition: enter ``Discovered`` and judge the changelog."""
91    return _advanced(
92        Discovered(candidate=candidate),
93        JudgeChangelog(candidate=candidate),
94    )
95 
96 
97def advance(
98    state: BumpState, event: LoopEvent, *, close_on_red: bool = True
99) -> Transition:
100    """Advance the loop one step (pure).
101 
102    Args:
103        state: The current bump state.
104        event: The decided event that just occurred.
105        close_on_red: Whether a terminal red CI should close the PR before
106            recording (the only transition this affects). Passed in rather than
107            read from config so the machine stays pure and replay-safe.
108 
109    Returns:
110        The :class:`Transition` to apply.
111    """
112    match state:
113        case Discovered():
114            return _from_discovered(state, event)
115        case Judged():
116            return _from_judged(state, event)
117        case AwaitingCi():
118            return _from_awaiting_ci(state, event, close_on_red=close_on_red)
119        case Closing():
120            return _from_closing(state, event)
121        case Recorded():
122            return _from_recorded(state, event)
123    assert_never(state)
124 
125 
126def _from_discovered(state: Discovered, event: LoopEvent) -> Transition:
127    match event:
128        case ChangelogJudged():
129            return _advanced(
130                Judged(candidate=state.candidate, verdict=event.verdict),
131                OpenPullRequest(
132                    candidate=state.candidate, verdict=event.verdict
133                ),
134            )
135        case _:
136            return _rejected(state, f"unexpected {event.kind} in discovered")
137 
138 
139def _from_judged(state: Judged, event: LoopEvent) -> Transition:
140    match event:
141        case PullRequestReady():
142            return _advanced(
143                AwaitingCi(
144                    candidate=state.candidate,
145                    verdict=state.verdict,
146                    pr=event.pr,
147                ),
148                AwaitCi(pr=event.pr),
149            )
150        case _:
151            return _rejected(state, f"unexpected {event.kind} in judged")
152 
153 
154def _from_awaiting_ci(
155    state: AwaitingCi, event: LoopEvent, *, close_on_red: bool
156) -> Transition:
157    match event:
158        case CiResolved():
159            if not is_terminal(event.status):
160                return _rejected(state, "ci still pending; the spine must wait")
161            outcome = LoopOutcome(
162                candidate=state.candidate,
163                verdict=state.verdict,
164                pr=state.pr,
165                ci=event.status,
166            )
167            # Red CI with close-on-red on: close the PR first (the loop leaves
168            # no rotting red proposal), then record the same outcome. Every
169            # other terminal reading (passed / absent / timed out), and red with
170            # close-on-red off, records straight away and leaves the PR for the
171            # human.
172            if isinstance(event.status, CIFailed) and close_on_red:
173                return _advanced(
174                    Closing(outcome=outcome),
175                    ClosePullRequest(pr=state.pr, failing=event.status.failing),
176                )
177            return _advanced(
178                Recorded(outcome=outcome), RecordOutcome(outcome=outcome)
179            )
180        case _:
181            return _rejected(state, f"unexpected {event.kind} in awaiting_ci")
182 
183 
184def _from_closing(state: Closing, event: LoopEvent) -> Transition:
185    match event:
186        case PullRequestClosed():
187            # The PR is closed; record the outcome it was carrying, exactly as
188            # the non-closing path would have.
189            return _advanced(
190                Recorded(outcome=state.outcome),
191                RecordOutcome(outcome=state.outcome),
192            )
193        case _:
194            return _rejected(state, f"unexpected {event.kind} in closing")
⋯ 12 lines hidden (lines 195–206)195 
196 
197def _from_recorded(state: Recorded, event: LoopEvent) -> Transition:
198    # Terminal: the only expected event is the record acknowledgement, a no-op
199    # that ends the driver loop (no effects).
200    match event:
201        case OutcomeRecorded():
202            return Transition(
203                kind=TransitionKind.IGNORED, next=state, reason="loop complete"
204            )
205        case _:
206            return _rejected(state, f"unexpected {event.kind} after recorded")

The toggle is configuration, and the state machine is pure — it must never read the environment. So close_on_red arrives as an explicit parameter to advance, defaulted to the SPEC's behavior, and only the one red branch reads it. The workflow passes it in from the bump's params (next section).

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

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

⋯ 96 lines hidden (lines 1–96)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
12: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, is_terminal
24from froot.domain.effects import (
25    AwaitCi,
26    ClosePullRequest,
27    Effect,
28    JudgeChangelog,
29    OpenPullRequest,
30    RecordOutcome,
31)
32from froot.domain.events import (
33    ChangelogJudged,
34    CiResolved,
35    LoopEvent,
36    OutcomeRecorded,
37    PullRequestClosed,
38    PullRequestReady,
39)
40from froot.domain.outcome import LoopOutcome
41from froot.domain.state import (
42    AwaitingCi,
43    BumpState,
44    Closing,
45    Discovered,
46    Judged,
47    Recorded,
48)
49 
50if TYPE_CHECKING:
51    from froot.domain.candidate import PatchCandidate
52 
53 
54class TransitionKind(StrEnum):
55    """The disposition of an :func:`advance` call."""
56 
57    ADVANCED = "advanced"
58    IGNORED = "ignored"
59    REJECTED = "rejected"
60 
61 
62class Transition(Frozen):
63    """The result of a transition.
64 
65    Attributes:
66        kind: ``ADVANCED`` (moved and/or emitted effects), ``IGNORED`` (a legal
67            no-op, e.g. the terminal acknowledgement), or ``REJECTED`` (the
68            event is not valid in this state).
69        next: The resulting state (unchanged for ``IGNORED``/``REJECTED``).
70        effects: The effects the spine should run, in order. Empty terminates
71            the loop driver.
72        reason: A short explanation for an ``IGNORED``/``REJECTED`` transition.
73    """
74 
75    kind: TransitionKind
76    next: BumpState
77    effects: tuple[Effect, ...] = ()
78    reason: str | None = None
79 
80 
81def _advanced(nxt: BumpState, *effects: Effect) -> Transition:
82    return Transition(kind=TransitionKind.ADVANCED, next=nxt, effects=effects)
83 
84 
85def _rejected(state: BumpState, reason: str) -> Transition:
86    return Transition(kind=TransitionKind.REJECTED, next=state, reason=reason)
87 
88 
89def start(candidate: PatchCandidate) -> Transition:
90    """The opening transition: enter ``Discovered`` and judge the changelog."""
91    return _advanced(
92        Discovered(candidate=candidate),
93        JudgeChangelog(candidate=candidate),
94    )
95 
96 
97def advance(
98    state: BumpState, event: LoopEvent, *, close_on_red: bool = True
99) -> Transition:
100    """Advance the loop one step (pure).
101 
102    Args:
103        state: The current bump state.
104        event: The decided event that just occurred.
105        close_on_red: Whether a terminal red CI should close the PR before
106            recording (the only transition this affects). Passed in rather than
107            read from config so the machine stays pure and replay-safe.
108 
109    Returns:
110        The :class:`Transition` to apply.
111    """
112    match state:
113        case Discovered():
114            return _from_discovered(state, event)
115        case Judged():
116            return _from_judged(state, event)
117        case AwaitingCi():
118            return _from_awaiting_ci(state, event, close_on_red=close_on_red)
119        case Closing():
120            return _from_closing(state, event)
121        case Recorded():
122            return _from_recorded(state, event)
123    assert_never(state)
⋯ 83 lines hidden (lines 124–206)124 
125 
126def _from_discovered(state: Discovered, event: LoopEvent) -> Transition:
127    match event:
128        case ChangelogJudged():
129            return _advanced(
130                Judged(candidate=state.candidate, verdict=event.verdict),
131                OpenPullRequest(
132                    candidate=state.candidate, verdict=event.verdict
133                ),
134            )
135        case _:
136            return _rejected(state, f"unexpected {event.kind} in discovered")
137 
138 
139def _from_judged(state: Judged, event: LoopEvent) -> Transition:
140    match event:
141        case PullRequestReady():
142            return _advanced(
143                AwaitingCi(
144                    candidate=state.candidate,
145                    verdict=state.verdict,
146                    pr=event.pr,
147                ),
148                AwaitCi(pr=event.pr),
149            )
150        case _:
151            return _rejected(state, f"unexpected {event.kind} in judged")
152 
153 
154def _from_awaiting_ci(
155    state: AwaitingCi, event: LoopEvent, *, close_on_red: bool
156) -> Transition:
157    match event:
158        case CiResolved():
159            if not is_terminal(event.status):
160                return _rejected(state, "ci still pending; the spine must wait")
161            outcome = LoopOutcome(
162                candidate=state.candidate,
163                verdict=state.verdict,
164                pr=state.pr,
165                ci=event.status,
166            )
167            # Red CI with close-on-red on: close the PR first (the loop leaves
168            # no rotting red proposal), then record the same outcome. Every
169            # other terminal reading (passed / absent / timed out), and red with
170            # close-on-red off, records straight away and leaves the PR for the
171            # human.
172            if isinstance(event.status, CIFailed) and close_on_red:
173                return _advanced(
174                    Closing(outcome=outcome),
175                    ClosePullRequest(pr=state.pr, failing=event.status.failing),
176                )
177            return _advanced(
178                Recorded(outcome=outcome), RecordOutcome(outcome=outcome)
179            )
180        case _:
181            return _rejected(state, f"unexpected {event.kind} in awaiting_ci")
182 
183 
184def _from_closing(state: Closing, event: LoopEvent) -> Transition:
185    match event:
186        case PullRequestClosed():
187            # The PR is closed; record the outcome it was carrying, exactly as
188            # the non-closing path would have.
189            return _advanced(
190                Recorded(outcome=state.outcome),
191                RecordOutcome(outcome=state.outcome),
192            )
193        case _:
194            return _rejected(state, f"unexpected {event.kind} in closing")
195 
196 
197def _from_recorded(state: Recorded, event: LoopEvent) -> Transition:
198    # Terminal: the only expected event is the record acknowledgement, a no-op
199    # that ends the driver loop (no effects).
200    match event:
201        case OutcomeRecorded():
202            return Transition(
203                kind=TransitionKind.IGNORED, next=state, reason="loop complete"
204            )
205        case _:
206            return _rejected(state, f"unexpected {event.kind} after recorded")

Close-on-red3🎬 The adapter method, and why a retry is safe

The close itself: idempotent by construction

src/froot/ports/protocols.pysrc/froot/adapters/github.py

The Forge port gains one method. The signature states the contract: close the PR and, by default, delete its head branch.

src/froot/ports/protocols.py · 160 lines

src/froot/ports/protocols.py160 lines · Python

⋯ 126 lines hidden (lines 1–126)1"""Typed Protocols for the impure world the spine talks to.
2 
3Methods are ``async`` so an activity simply awaits a port; an adapter that wraps
4a blocking tool (``npm``, ``git``) runs it off the event loop internally, and
5one backed by an HTTP API uses an async client. The pure core and the spine
6depend on these abstractions; :mod:`froot.adapters` provides the concrete
7implementations and tests pass fakes.
8"""
9 
10from __future__ import annotations
11 
12from typing import TYPE_CHECKING, Protocol
13 
14if TYPE_CHECKING:
15    from pathlib import Path
16 
17    from froot.domain.candidate import AvailableUpgrade, PatchCandidate
18    from froot.domain.changelog import Changelog, ChangelogVerdict
19    from froot.domain.ci import CIStatus
20    from froot.domain.pull_request import (
21        BranchName,
22        PullRequestDraft,
23        PullRequestRef,
24    )
25    from froot.domain.repo import TargetRepo
26 
27 
28class PackageManager(Protocol):
29    """Reads available upgrades and regenerates the manifest + lockfile.
30 
31    The adapter carries the package manager (e.g. ``npm``) but never runs the
32    project's tests or install scripts — lockfile regeneration only, so the
33    worker stays light and no third-party dependency code executes in it.
34    """
35 
36    async def list_upgrades(
37        self, target: TargetRepo, workspace: Path
38    ) -> tuple[AvailableUpgrade, ...]:
39        """Report each outdated dependency and the versions available to it."""
40        ...
41 
42    async def apply_patch_bump(
43        self, candidate: PatchCandidate, workspace: Path
44    ) -> None:
45        """Rewrite the manifest + lockfile in ``workspace`` to the target.
46 
47        Lockfile-only and with install scripts disabled: it resolves and
48        writes the dependency tree but runs no project or dependency code.
49        """
50        ...
51 
52 
53class Forge(Protocol):
54    """Git + GitHub: checkout, branch/PR, CI status, labels.
55 
56    The verification oracle is the repo's own CI (:meth:`ci_status`); froot
57    never runs tests itself. PR creation is idempotent against the deterministic
58    branch — see :meth:`find_open_pull_request`.
59    """
60 
61    async def checkout(self, target: TargetRepo, workspace: Path) -> None:
62        """Materialize the repo's default branch into ``workspace``."""
63        ...
64 
65    async def checkout_pull_request(
66        self, target: TargetRepo, workspace: Path, number: int
67    ) -> None:
68        """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
69 
70        Works uniformly for same-repo and fork PRs — the base repo exposes the
71        head of every PR under ``refs/pull/<number>/head``, so no fork URL or
72        cross-repo auth is needed.
73        """
74        ...
75 
76    async def push_branch(
77        self, workspace: Path, branch: BranchName, commit_message: str
78    ) -> str:
79        """Commit the workspace changes onto ``branch`` and push it.
80 
81        The workspace's ``origin`` already authenticates against the repo (set
82        up by :meth:`checkout`), so no target is needed here.
83 
84        Returns:
85            The pushed head commit SHA.
86        """
87        ...
88 
89    async def find_open_pull_request(
90        self, target: TargetRepo, branch: BranchName
91    ) -> PullRequestRef | None:
92        """Return the open PR for ``branch`` if one already exists (dedup)."""
93        ...
94 
95    async def list_open_pull_requests(
96        self, target: TargetRepo
97    ) -> tuple[PullRequestRef, ...]:
98        """List the repo's open PRs (the determinism reviewer's work feed)."""
99        ...
100 
101    async def upsert_issue_comment(
102        self, target: TargetRepo, number: int, marker: str, body: str
103    ) -> str:
104        """Create or update the PR's ``marker``-tagged comment; return its URL.
105 
106        Finds the existing comment containing ``marker`` and edits it in place,
107        else posts a new one — so re-reviewing a PR never stacks comments.
108        """
109        ...
110 
111    async def open_pull_request(
112        self, target: TargetRepo, draft: PullRequestDraft
113    ) -> PullRequestRef:
114        """Open the PR for an already-pushed branch."""
115        ...
116 
117    async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
118        """Read the repo's combined CI status for a commit (the oracle)."""
119        ...
120 
121    async def add_labels(
122        self, target: TargetRepo, number: int, labels: tuple[str, ...]
123    ) -> None:
124        """Attach labels to a PR (the human-readable signal-update)."""
125        ...
126 
127    async def close_pull_request(
128        self,
129        target: TargetRepo,
130        number: int,
131        branch: BranchName,
132        *,
133        delete_branch: bool = True,
134    ) -> None:
135        """Close the PR and (by default) delete its head branch.
136 
137        Idempotent: closing an already-closed PR is a no-op, and a missing
138        branch is tolerated — so a retried close never fails on a half-done
139        prior attempt. Deleting the branch keeps a re-derived bump from later
140        colliding with a stale ref (a non-fast-forward push). Any human-facing
141        explanation is posted separately via :meth:`upsert_issue_comment`, so
142        this stays a pure lifecycle action.
143        """
144        ...
145 
146 
147class ChangelogSource(Protocol):
148    """Best-effort fetch of a target version's changelog / release notes."""
⋯ 12 lines hidden (lines 149–160)149 
150    async def fetch(self, candidate: PatchCandidate) -> Changelog | None:
151        """Return the changelog for the candidate's target, or ``None``."""
152        ...
153 
154 
155class ModelJudge(Protocol):
156    """The thin model judgment: is this changelog a clean patch?"""
157 
158    async def judge(self, changelog: Changelog) -> ChangelogVerdict:
159        """Assess a changelog and return a typed verdict."""
160        ...

The implementation is two GitHub calls, both chosen so a retried close never fails on a half-done prior attempt — which matters because every activity now runs under a bounded retry (§6).

src/froot/adapters/github.py · 382 lines

src/froot/adapters/github.py382 lines · Python

⋯ 315 lines hidden (lines 1–315)1"""The GitHub forge adapter: git (checkout/push) + the GitHub REST API.
2 
3Backs :class:`~froot.ports.protocols.Forge`. Checkout and branch push go
4through ``git``; pull requests, CI status, and labels go through the REST API
5(httpx). The verification oracle is the repo's own CI — :meth:`ci_status` reads
6it, froot never runs tests. PR creation is idempotent against the deterministic
7head branch (:meth:`find_open_pull_request`), so a re-run never double-opens.
8 
9The CI mapping (:func:`ci_status_from_checks`) is a pure function over typed
10check rows, unit-tested apart from the network; the GitHub JSON shapes are
11read at the boundary as untyped payloads and coerced into the domain.
12"""
13 
14from __future__ import annotations
15 
16from dataclasses import dataclass
17from typing import TYPE_CHECKING, Any, final
18 
19import httpx
20from temporalio.exceptions import ApplicationError
21 
22from froot.config.settings import GitHubSettings
23from froot.domain.ci import (
  CIAbsent,
  CIFailed,
  CIPassed,
  CIPending,
  CIStatus,
29)
30from froot.domain.pull_request import BranchName, PullRequestRef
31 
32if TYPE_CHECKING:
  from pathlib import Path
34 
  from froot.domain.pull_request import PullRequestDraft
  from froot.domain.repo import TargetRepo
37 
38from froot.adapters._proc import run_text
39 
40_API = "https://api.github.com"
41_API_VERSION = "2022-11-28"
42_TIMEOUT = 30.0
43_COMMITTER_NAME = "froot"
44_COMMITTER_EMAIL = "froot@users.noreply.github.com"
45 
46# Check-run conclusions that mean the change is not safe to merge.
47_BAD_CONCLUSIONS = frozenset(
  {
      "failure",
      "timed_out",
      "cancelled",
      "action_required",
      "startup_failure",
      "stale",
  }
56)
57 
58 
59@final
60@dataclass(frozen=True, slots=True)
61class CheckRow:
  """One GitHub check run, reduced to what the CI mapping needs."""
63 
  name: str
  status: str
  conclusion: str | None
67 
68 
69def ci_status_from_checks(
  checks: tuple[CheckRow, ...], combined_state: str | None
71) -> CIStatus:
  """Map GitHub check rows + the combined commit status to a CI status.
73 
  Args:
      checks: The commit's check runs (GitHub Checks API).
      combined_state: The legacy combined commit status (``success`` /
          ``failure`` / ``pending``), or ``None`` when no statuses exist.
78 
  Returns:
      ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
      unresolved, ``CIFailed`` (with the failing check names) on any bad
      conclusion or a failed combined status, else ``CIPassed``.
  """
  if not checks and combined_state is None:
      return CIAbsent()
  unresolved = combined_state == "pending" or any(
      row.status != "completed" for row in checks
  )
  if unresolved:
      return CIPending()
  failing = tuple(
      row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
  )
  if failing or combined_state == "failure":
      return CIFailed(failing=failing)
  return CIPassed()
97 
98 
99def _token() -> str:
  token = GitHubSettings().github_token
  if token is None:
      # A missing token is a permanent misconfiguration, not a transient
      # fault — fail the activity fast instead of retrying forever.
      raise ApplicationError(
          "FROOT_GITHUB_TOKEN is required", non_retryable=True
      )
  return token.get_secret_value()
108 
109 
110def _auth_remote(target: TargetRepo) -> str:
  return (
      f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
  )
114 
115 
116def _client() -> httpx.AsyncClient:
  return httpx.AsyncClient(
      base_url=_API,
      timeout=_TIMEOUT,
      headers={
          "Authorization": f"Bearer {_token()}",
          "Accept": "application/vnd.github+json",
          "X-GitHub-Api-Version": _API_VERSION,
      },
  )
126 
127 
128def _raise_for_status(response: httpx.Response) -> None:
  """Raise on error; 401/403 is a permanent (non-retryable) auth fault."""
  if response.status_code in (401, 403):
      raise ApplicationError(
          f"GitHub auth failed ({response.status_code})", non_retryable=True
      )
  response.raise_for_status()
135 
136 
137def _pull_request_ref(payload: Any) -> PullRequestRef:
  """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
  head = payload["head"]
  return PullRequestRef(
      number=int(payload["number"]),
      url=str(payload["html_url"]),
      branch=BranchName(value=str(head["ref"])),
      head_sha=str(head["sha"]),
  )
146 
147 
148@final
149class GitHubForge:
  """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
151 
  async def checkout(self, target: TargetRepo, workspace: Path) -> None:
      """Shallow-clone the repo's default branch into ``workspace``."""
      code, out, err = await run_text(
          "git",
          "clone",
          "--depth",
          "1",
          "--branch",
          target.default_branch,
          _auth_remote(target),
          ".",
          cwd=workspace,
      )
      if code != 0:
          raise RuntimeError(f"git clone failed ({code}): {err or out}")
167 
  async def checkout_pull_request(
      self, target: TargetRepo, workspace: Path, number: int
  ) -> None:
      """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
172 
      Fetches the PR head ref the base repo exposes for every PR (fork or
      not), so a community PR from a fork checks out the same way a same-repo
      one does — no fork URL, no cross-repo auth.
      """
      ref = f"pull/{number}/head"
      steps: tuple[tuple[str, ...], ...] = (
          ("git", "init", "-q"),
          ("git", "remote", "add", "origin", _auth_remote(target)),
          ("git", "fetch", "--depth", "1", "origin", ref),
          ("git", "checkout", "-q", "FETCH_HEAD"),
      )
      for step in steps:
          code, out, err = await run_text(*step, cwd=workspace)
          if code != 0:
              raise RuntimeError(
                  f"git {step[1]} ({ref}) failed ({code}): {err or out}"
              )
190 
  async def push_branch(
      self, workspace: Path, branch: BranchName, commit_message: str
  ) -> str:
      """Commit the workspace changes onto ``branch`` and push; return SHA."""
      steps: tuple[tuple[str, ...], ...] = (
          ("git", "checkout", "-b", branch.value),
          ("git", "add", "-A"),
          (
              "git",
              "-c",
              f"user.name={_COMMITTER_NAME}",
              "-c",
              f"user.email={_COMMITTER_EMAIL}",
              "commit",
              "-m",
              commit_message,
          ),
          ("git", "push", "-u", "origin", branch.value),
      )
      for step in steps:
          code, out, err = await run_text(*step, cwd=workspace)
          if code != 0:
              raise RuntimeError(f"{step[0:2]} failed ({code}): {err or out}")
      code, sha, _ = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
      if code != 0:
          raise RuntimeError(f"git rev-parse failed ({code})")
      return sha.strip()
218 
  async def find_open_pull_request(
      self, target: TargetRepo, branch: BranchName
  ) -> PullRequestRef | None:
      """Return the open PR for ``branch`` if one already exists (dedup)."""
      async with _client() as client:
          resp = await client.get(
              f"/repos/{target.repo.slug}/pulls",
              params={
                  "head": f"{target.repo.owner}:{branch.value}",
                  "state": "open",
              },
          )
      _raise_for_status(resp)
      payloads = resp.json()
      if isinstance(payloads, list) and payloads:
          return _pull_request_ref(payloads[0])
      return None
236 
  async def list_open_pull_requests(
      self, target: TargetRepo
  ) -> tuple[PullRequestRef, ...]:
      """List the repo's open PRs (the determinism reviewer's work feed)."""
      async with _client() as client:
          resp = await client.get(
              f"/repos/{target.repo.slug}/pulls",
              params={"state": "open", "per_page": 100},
          )
      _raise_for_status(resp)
      payloads = resp.json()
      if not isinstance(payloads, list):
          return ()
      return tuple(_pull_request_ref(payload) for payload in payloads)
251 
  async def open_pull_request(
      self, target: TargetRepo, draft: PullRequestDraft
  ) -> PullRequestRef:
      """Open the PR for an already-pushed branch (idempotent on conflict)."""
      async with _client() as client:
          resp = await client.post(
              f"/repos/{target.repo.slug}/pulls",
              json={
                  "title": draft.title,
                  "head": draft.branch.value,
                  "base": draft.base,
                  "body": draft.body,
              },
          )
      if resp.status_code == 422:
          existing = await self.find_open_pull_request(target, draft.branch)
          if existing is not None:
              return existing
      _raise_for_status(resp)
      return _pull_request_ref(resp.json())
272 
  async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
      """Read the repo's combined CI status for a commit (the oracle)."""
      async with _client() as client:
          checks_resp = await client.get(
              f"/repos/{target.repo.slug}/commits/{head_sha}/check-runs",
              params={"per_page": 100},
          )
          status_resp = await client.get(
              f"/repos/{target.repo.slug}/commits/{head_sha}/status"
          )
      _raise_for_status(checks_resp)
      _raise_for_status(status_resp)
      rows = tuple(
          CheckRow(
              name=str(run["name"]),
              status=str(run["status"]),
              conclusion=(
                  str(run["conclusion"])
                  if run.get("conclusion") is not None
                  else None
              ),
          )
          for run in checks_resp.json().get("check_runs", [])
      )
      status_json = status_resp.json()
      combined = (
          str(status_json["state"])
          if int(status_json.get("total_count", 0)) > 0
          else None
      )
      return ci_status_from_checks(rows, combined)
304 
  async def add_labels(
      self, target: TargetRepo, number: int, labels: tuple[str, ...]
  ) -> None:
      """Attach labels to a PR (the human-readable signal-update)."""
      async with _client() as client:
          resp = await client.post(
              f"/repos/{target.repo.slug}/issues/{number}/labels",
              json={"labels": list(labels)},
          )
      _raise_for_status(resp)
315 
  async def close_pull_request(
      self,
      target: TargetRepo,
      number: int,
      branch: BranchName,
      *,
      delete_branch: bool = True,
  ) -> None:
      """Close the PR and (by default) delete its head branch.
325 
      Two idempotent GitHub calls: PATCH the PR to ``state=closed`` (a no-op
      if it is already closed), then DELETE the head ref. A 404/422 on the
      delete (branch already gone — e.g. the repo auto-deletes head branches
      on close) is tolerated, so a retried close never fails on a branch a
      prior attempt already removed.
      """
      slug = target.repo.slug
      async with _client() as client:
          close_resp = await client.patch(
              f"/repos/{slug}/pulls/{number}",
              json={"state": "closed"},
          )
          _raise_for_status(close_resp)
          if not delete_branch:
              return
          delete_resp = await client.delete(
              f"/repos/{slug}/git/refs/heads/{branch.value}"
          )
      if delete_resp.status_code not in (404, 422):
          _raise_for_status(delete_resp)
⋯ 37 lines hidden (lines 346–382)346 
  async def upsert_issue_comment(
      self, target: TargetRepo, number: int, marker: str, body: str
  ) -> str:
      """Create or update the PR's ``marker``-tagged comment; return its URL.
351 
      A PR conversation (issue) comment, not an inline review comment: the
      determinism findings are structural (a call path), so a single advisory
      summary fits better than line anchors. The marker lets a re-review edit
      its own prior comment in place rather than stack a new one.
      """
      slug = target.repo.slug
      async with _client() as client:
          listing = await client.get(
              f"/repos/{slug}/issues/{number}/comments",
              params={"per_page": 100},
          )
          _raise_for_status(listing)
          existing_id: int | None = None
          for comment in listing.json():
              if isinstance(comment, dict) and marker in str(
                  comment.get("body", "")
              ):
                  existing_id = int(comment["id"])
                  break
          if existing_id is not None:
              resp = await client.patch(
                  f"/repos/{slug}/issues/comments/{existing_id}",
                  json={"body": body},
              )
          else:
              resp = await client.post(
                  f"/repos/{slug}/issues/{number}/comments",
                  json={"body": body},
              )
      _raise_for_status(resp)
      return str(resp.json()["html_url"])

Reconcile4🎬 A pure policy with a careful matcher

Closing the stale PRs, fail-safe

src/froot/policy/reconcile.pysrc/froot/policy/naming.py

Branch names are per version, so a package that gets a newer patch (1.2.3 then 1.2.4) ends up with a second open PR the propose path never closes. The fix is a pure policy: given the repo's open PRs and the upgrade facts the scan already gathered, return the froot PRs to close — superseded by a newer target, or already satisfied by the base.

superseded fires when a newer candidate exists; satisfied when the base already sits at or past the PR's target.

src/froot/policy/reconcile.py · 137 lines

src/froot/policy/reconcile.py137 lines · Python

⋯ 51 lines hidden (lines 1–51)1"""Decide which of froot's own open bump PRs no longer deserve to stay open.
2 
3Branch names are per *version* (``froot/dependency-patch/<pkg>-<target>``), so a
4package that gets a newer patch (``1.2.3`` then ``1.2.4``) ends up with a
5*second* open PR — the stale one is never closed by the propose path. This pure
6policy is the cleanup: given the repo's open PRs and the same upgrade facts the
7scan tick gathered, it returns the froot PRs to close, each with its note.
8 
9Two reasons, both read off ground truth (the live ``AvailableUpgrade`` set), so
10the policy only closes on a *positive* match and otherwise leaves a PR alone:
11 
12* **superseded** — a newer patch target for that package is being proposed now,
13  so the older PR's version is below the current best.
14* **satisfied** — the base already caught up to (or past) the PR's target, so
15  merging it would be a no-op.
16 
17Like the rest of froot this stores nothing: the close set is re-derived from the
18repo each tick. A PR that can't be matched to a live upgrade (its package is no
19longer outdated at all, or its branch doesn't parse) is deliberately left open —
20reconcile fails safe, never guessing a close.
21"""
22 
23from __future__ import annotations
24 
25from typing import TYPE_CHECKING
26 
27from froot.domain.base import Frozen
28from froot.domain.pull_request import PullRequestRef
29from froot.domain.version import Version
30from froot.policy.candidates import select_patch_candidates
31from froot.policy.compose import CLOSE_MARKER
32from froot.policy.naming import branch_package_prefix
33from froot.result import Ok
34 
35if TYPE_CHECKING:
36    from froot.domain.candidate import AvailableUpgrade
37 
38 
39class ReconcileClosure(Frozen):
40    """A froot PR reconcile decided to close, plus the note to leave on it.
41 
42    Attributes:
43        pr: The open PR to close (its branch is deleted with it).
44        comment: The human-facing reason, carrying :data:`CLOSE_MARKER` so the
45            close posts through the idempotent comment path.
46    """
47 
48    pr: PullRequestRef
49    comment: str
50 
51 
52def reconciliations(
53    open_prs: tuple[PullRequestRef, ...],
54    upgrades: tuple[AvailableUpgrade, ...],
55) -> tuple[ReconcileClosure, ...]:
56    """The froot PRs to close this tick, derived from the live upgrade facts.
57 
58    Args:
59        open_prs: Every open PR on the repo (froot's and not — non-froot
60            branches simply never match a bump prefix).
61        upgrades: The outdated dependencies and their available versions, as
62            the scan gathered them. Both the current candidates and the
63            installed versions are derived from these.
64 
65    Returns:
66        One :class:`ReconcileClosure` per froot PR that is superseded by a newer
67        target or already satisfied by the base, in PR-number order.
68    """
69    candidates = {c.package: c for c in select_patch_candidates(upgrades)}
70    installed = {u.package: u.current for u in upgrades}
71    closures: list[ReconcileClosure] = []
72    for pr in sorted(open_prs, key=lambda p: p.number):
73        matched = _match_pr(pr, upgrades)
74        if matched is None:
75            continue
76        package, pr_target = matched
77        candidate = candidates.get(package)
78        if candidate is not None and pr_target < candidate.target:
79            closures.append(
80                ReconcileClosure(
81                    pr=pr, comment=_superseded_comment(candidate.target)
82                )
83            )
84        elif pr_target <= installed[package]:
85            closures.append(
86                ReconcileClosure(
87                    pr=pr, comment=_satisfied_comment(installed[package])
88                )
89            )
90    return tuple(closures)
91 
⋯ 46 lines hidden (lines 92–137)92 
93def _match_pr(
94    pr: PullRequestRef, upgrades: tuple[AvailableUpgrade, ...]
95) -> tuple[str, Version] | None:
96    """The ``(package, target)`` a froot bump PR is for, or ``None``.
97 
98    Matches the PR's branch against each upgrade's bump prefix and parses the
99    remainder as a version — so a branch is attributed to a package only when
100    the tail is a real version, which disambiguates packages whose slugs prefix
101    one another (``foo`` vs ``foo-bar``). The longest matching prefix wins, so a
102    pathological exact collision still resolves deterministically.
103    """
104    best: tuple[str, Version, int] | None = None
105    for upgrade in upgrades:
106        prefix = branch_package_prefix(upgrade.package)
107        if not pr.branch.value.startswith(prefix):
108            continue
109        match Version.parse(pr.branch.value[len(prefix) :]):
110            case Ok(version):
111                if best is None or len(prefix) > best[2]:
112                    best = (upgrade.package, version, len(prefix))
113            case _:
114                continue
115    return (best[0], best[1]) if best is not None else None
116 
117 
118def _superseded_comment(superseding: Version) -> str:
119    """The note for a PR a newer patch target has overtaken."""
120    return "\n".join(
121        (
122            CLOSE_MARKER,
123            f"froot closed this PR: a newer patch ({superseding}) supersedes "
124            "it. The newer bump is being proposed in its place.",
125        )
126    )
127 
128 
129def _satisfied_comment(installed: Version) -> str:
130    """The note for a PR the base has already caught up to."""
131    return "\n".join(
132        (
133            CLOSE_MARKER,
134            f"froot closed this PR: the base already has {installed}, so this "
135            "bump is moot.",
136        )
137    )

Everything hinges on matching an open PR back to its package and version. The branch is froot/dependency-patch/<slug>-<version>, and both the slug and the version contain hyphens — so a blind split is ambiguous. The matcher instead derives each package's prefix and parses the remainder as a version: a branch belongs to a package only when the tail is a real version.

src/froot/policy/reconcile.py · 137 lines

src/froot/policy/reconcile.py137 lines · Python

⋯ 92 lines hidden (lines 1–92)1"""Decide which of froot's own open bump PRs no longer deserve to stay open.
2 
3Branch names are per *version* (``froot/dependency-patch/<pkg>-<target>``), so a
4package that gets a newer patch (``1.2.3`` then ``1.2.4``) ends up with a
5*second* open PR — the stale one is never closed by the propose path. This pure
6policy is the cleanup: given the repo's open PRs and the same upgrade facts the
7scan tick gathered, it returns the froot PRs to close, each with its note.
8 
9Two reasons, both read off ground truth (the live ``AvailableUpgrade`` set), so
10the policy only closes on a *positive* match and otherwise leaves a PR alone:
11 
12* **superseded** — a newer patch target for that package is being proposed now,
13  so the older PR's version is below the current best.
14* **satisfied** — the base already caught up to (or past) the PR's target, so
15  merging it would be a no-op.
16 
17Like the rest of froot this stores nothing: the close set is re-derived from the
18repo each tick. A PR that can't be matched to a live upgrade (its package is no
19longer outdated at all, or its branch doesn't parse) is deliberately left open —
20reconcile fails safe, never guessing a close.
21"""
22 
23from __future__ import annotations
24 
25from typing import TYPE_CHECKING
26 
27from froot.domain.base import Frozen
28from froot.domain.pull_request import PullRequestRef
29from froot.domain.version import Version
30from froot.policy.candidates import select_patch_candidates
31from froot.policy.compose import CLOSE_MARKER
32from froot.policy.naming import branch_package_prefix
33from froot.result import Ok
34 
35if TYPE_CHECKING:
36    from froot.domain.candidate import AvailableUpgrade
37 
38 
39class ReconcileClosure(Frozen):
40    """A froot PR reconcile decided to close, plus the note to leave on it.
41 
42    Attributes:
43        pr: The open PR to close (its branch is deleted with it).
44        comment: The human-facing reason, carrying :data:`CLOSE_MARKER` so the
45            close posts through the idempotent comment path.
46    """
47 
48    pr: PullRequestRef
49    comment: str
50 
51 
52def reconciliations(
53    open_prs: tuple[PullRequestRef, ...],
54    upgrades: tuple[AvailableUpgrade, ...],
55) -> tuple[ReconcileClosure, ...]:
56    """The froot PRs to close this tick, derived from the live upgrade facts.
57 
58    Args:
59        open_prs: Every open PR on the repo (froot's and not — non-froot
60            branches simply never match a bump prefix).
61        upgrades: The outdated dependencies and their available versions, as
62            the scan gathered them. Both the current candidates and the
63            installed versions are derived from these.
64 
65    Returns:
66        One :class:`ReconcileClosure` per froot PR that is superseded by a newer
67        target or already satisfied by the base, in PR-number order.
68    """
69    candidates = {c.package: c for c in select_patch_candidates(upgrades)}
70    installed = {u.package: u.current for u in upgrades}
71    closures: list[ReconcileClosure] = []
72    for pr in sorted(open_prs, key=lambda p: p.number):
73        matched = _match_pr(pr, upgrades)
74        if matched is None:
75            continue
76        package, pr_target = matched
77        candidate = candidates.get(package)
78        if candidate is not None and pr_target < candidate.target:
79            closures.append(
80                ReconcileClosure(
81                    pr=pr, comment=_superseded_comment(candidate.target)
82                )
83            )
84        elif pr_target <= installed[package]:
85            closures.append(
86                ReconcileClosure(
87                    pr=pr, comment=_satisfied_comment(installed[package])
88                )
89            )
90    return tuple(closures)
91 
92 
93def _match_pr(
94    pr: PullRequestRef, upgrades: tuple[AvailableUpgrade, ...]
95) -> tuple[str, Version] | None:
96    """The ``(package, target)`` a froot bump PR is for, or ``None``.
97 
98    Matches the PR's branch against each upgrade's bump prefix and parses the
99    remainder as a version — so a branch is attributed to a package only when
100    the tail is a real version, which disambiguates packages whose slugs prefix
101    one another (``foo`` vs ``foo-bar``). The longest matching prefix wins, so a
102    pathological exact collision still resolves deterministically.
103    """
104    best: tuple[str, Version, int] | None = None
105    for upgrade in upgrades:
106        prefix = branch_package_prefix(upgrade.package)
107        if not pr.branch.value.startswith(prefix):
108            continue
109        match Version.parse(pr.branch.value[len(prefix) :]):
110            case Ok(version):
111                if best is None or len(prefix) > best[2]:
112                    best = (upgrade.package, version, len(prefix))
113            case _:
114                continue
115    return (best[0], best[1]) if best is not None else None
⋯ 22 lines hidden (lines 116–137)116 
117 
118def _superseded_comment(superseding: Version) -> str:
119    """The note for a PR a newer patch target has overtaken."""
120    return "\n".join(
121        (
122            CLOSE_MARKER,
123            f"froot closed this PR: a newer patch ({superseding}) supersedes "
124            "it. The newer bump is being proposed in its place.",
125        )
126    )
127 
128 
129def _satisfied_comment(installed: Version) -> str:
130    """The note for a PR the base has already caught up to."""
131    return "\n".join(
132        (
133            CLOSE_MARKER,
134            f"froot closed this PR: the base already has {installed}, so this "
135            "bump is moot.",
136        )
137    )

src/froot/policy/naming.py · 92 lines

src/froot/policy/naming.py92 lines · Python

⋯ 37 lines hidden (lines 1–37)1"""Deterministic names — the loop's idempotency keys.
2 
3A bump's head branch and its per-bump Temporal workflow id are pure functions of
4the bump identity (repo + package + target version). Re-running the loop reuses
5the same names, so a duplicate PR or a duplicate in-flight workflow is
6impossible: the loop is idempotent by construction (SPEC: one PR per bump). The
7scan loop's id is a per-repo singleton for the same reason.
8"""
9 
10from __future__ import annotations
11 
12import re
13from typing import TYPE_CHECKING
14 
15from froot.domain.pull_request import BranchName
16 
17if TYPE_CHECKING:
18    from froot.domain.candidate import PatchCandidate
19    from froot.domain.repo import TargetRepo
20 
21_BRANCH_PREFIX = "froot/dependency-patch"
22# Anything outside the safe set collapses to a single hyphen.
23_UNSAFE = re.compile(r"[^a-z0-9._-]+")
24 
25 
26def _slug(text: str) -> str:
27    """Lowercase and reduce ref/id-unsafe runs to single hyphens."""
28    return _UNSAFE.sub("-", text.lower()).strip("-")
29 
30 
31def branch_name(candidate: PatchCandidate) -> BranchName:
32    """The deterministic head branch for a bump (also the PR dedup key)."""
33    return BranchName(
34        value=f"{_BRANCH_PREFIX}/{_slug(candidate.package)}-{candidate.target}"
35    )
36 
37 
38def branch_package_prefix(package: str) -> str:
39    """The branch-name prefix shared by every bump of ``package``.
40 
41    ``branch_name`` appends ``-<target>`` to this, so an open PR belongs to
42    ``package`` iff its branch starts with this prefix *and* the remainder
43    parses as a version (reconcile relies on that version-parse to tell apart
44    packages whose slugs prefix one another — ``foo`` vs ``foo-bar``).
45    """
46    return f"{_BRANCH_PREFIX}/{_slug(package)}-"
47 
48 
49def bump_workflow_id(repo: TargetRepo, candidate: PatchCandidate) -> str:
⋯ 43 lines hidden (lines 50–92)50    """The deterministic per-bump workflow id (the dispatch dedup key)."""
51    return "-".join(
52        (
53            "froot-bump",
54            _slug(repo.repo.owner),
55            _slug(repo.repo.name),
56            _slug(candidate.package),
57            _slug(str(candidate.target)),
58        )
59    )
60 
61 
62def scan_workflow_id(repo: TargetRepo) -> str:
63    """The deterministic per-repo scan-loop workflow id (a singleton)."""
64    return "-".join(
65        ("froot-scan", _slug(repo.repo.owner), _slug(repo.repo.name))
66    )
67 
68 
69def review_workflow_id(repo: TargetRepo) -> str:
70    """The deterministic per-repo determinism-review loop id (a singleton)."""
71    return "-".join(
72        ("froot-review", _slug(repo.repo.owner), _slug(repo.repo.name))
73    )
74 
75 
76def pr_review_workflow_id(
77    repo: TargetRepo, pr_number: int, head_sha: str
78) -> str:
79    """The deterministic per-(PR, head SHA) review id (the dispatch dedup key).
80 
81    Keyed on the head SHA so a new commit triggers a fresh review, and
82    re-dispatch of the same commit is a no-op (REJECT_DUPLICATE).
83    """
84    return "-".join(
85        (
86            "froot-pr-review",
87            _slug(repo.repo.owner),
88            _slug(repo.repo.name),
89            _slug(str(pr_number)),
90            _slug(head_sha[:12]),
91        )
92    )

Binding to the world5🎬 The impure boundary, lazy imports intact

Where the effects meet GitHub

src/froot/workflow/activities.pysrc/froot/workflow/scan_workflow.py

The activities are the only impure layer. Two are new. The close activity renders the "why", posts it through the idempotent comment path, then closes — mirroring the order the adapter is built for.

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

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

⋯ 199 lines hidden (lines 1–199)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
17 
18from temporalio import activity
19 
20from froot.domain.candidate import PatchCandidate
21from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
22from froot.domain.ci import CIStatus
23from froot.domain.determinism import AnalysisResult, FrontierVerdict
24from froot.domain.pull_request import PullRequestRef
25from froot.domain.repo import TargetRepo
26from froot.policy.candidates import select_patch_candidates
27from froot.policy.compose import PR_LABELS, pull_request_draft
28from froot.policy.determinism import analyze_workflow_surface
29from froot.policy.naming import (
30    branch_name,
31    bump_workflow_id,
32    pr_review_workflow_id,
33)
34from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
35from froot.workflow.types import (
36    AdjudicateInput,
37    CiCheckInput,
38    CloseInput,
39    DispatchInput,
40    DispatchReviewInput,
41    OpenPrInput,
42    PostReviewInput,
43    PrReviewParams,
44    RecordInput,
45)
46 
47_log = logging.getLogger("froot.outcome")
48_review_log = logging.getLogger("froot.review")
49_reconcile_log = logging.getLogger("froot.reconcile")
50 
51 
52def _manifest_dir(target: TargetRepo, workspace: Path) -> Path:
53    """The directory the manifest lives in (a monorepo subdir, or the root)."""
54    return workspace / target.manifest_dir if target.manifest_dir else workspace
55 
56 
57@activity.defn
58async def scan_candidates(
59    target: TargetRepo,
60) -> tuple[PatchCandidate, ...]:
61    """Check out the repo, read available upgrades, select patch candidates."""
62    from froot.adapters.github import GitHubForge
63    from froot.adapters.registry import package_manager_for
64 
65    forge = GitHubForge()
66    package_manager = package_manager_for(target.ecosystem)
67    with tempfile.TemporaryDirectory() as tmp:
68        workspace = Path(tmp)
69        await forge.checkout(target, workspace)
70        upgrades = await package_manager.list_upgrades(
71            target, _manifest_dir(target, workspace)
72        )
73    return select_patch_candidates(upgrades)
74 
75 
76@activity.defn
77async def judge_changelog(candidate: PatchCandidate) -> ChangelogVerdict:
78    """Fetch the candidate's changelog and get the model's typed verdict.
79 
80    The model is froot's one thin, non-load-bearing judgment: a clean verdict
81    never *gates* a PR (CI is the oracle), so a model that is down, slow, or
82    erroring must not stall the spine. A judge failure degrades to
83    ``UnknownVerdict`` — the bump proceeds, the human still gets the PR, and the
84    dashboard records the verdict as unknown — rather than failing (and then
85    retrying) the activity. Only the model call is guarded; the fetch is already
86    best-effort (returns ``None``, not an exception).
87    """
88    from froot.adapters.changelog_http import HttpChangelogSource
89    from froot.adapters.model_judge import PydanticAiJudge
90 
91    changelog = await HttpChangelogSource().fetch(candidate)
92    if changelog is None:
93        return UnknownVerdict(rationale="No changelog could be fetched.")
94    try:
95        return await PydanticAiJudge().judge(changelog)
96    except Exception as exc:
97        activity.logger.warning(
98            "changelog judge unavailable for %s; degrading to unknown: %r",
99            candidate.package,
100            exc,
101        )
102        return UnknownVerdict(
103            rationale=f"Changelog judge unavailable ({type(exc).__name__})."
104        )
105 
106 
107@activity.defn
108async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
109    """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
110    from froot.adapters.github import GitHubForge
111    from froot.adapters.registry import package_manager_for
112 
113    forge = GitHubForge()
114    package_manager = package_manager_for(params.target.ecosystem)
115    branch = branch_name(params.candidate)
116    existing = await forge.find_open_pull_request(params.target, branch)
117    if existing is not None:
118        return existing
119    draft = pull_request_draft(params.target, params.candidate, params.verdict)
120    with tempfile.TemporaryDirectory() as tmp:
121        workspace = Path(tmp)
122        await forge.checkout(params.target, workspace)
123        await package_manager.apply_patch_bump(
124            params.candidate, _manifest_dir(params.target, workspace)
125        )
126        await forge.push_branch(workspace, branch, draft.title)
127    return await forge.open_pull_request(params.target, draft)
128 
129 
130@activity.defn
131async def check_ci(params: CiCheckInput) -> CIStatus:
132    """Read the repo's CI status for the PR's head commit (the oracle)."""
133    from froot.adapters.github import GitHubForge
134 
135    return await GitHubForge().ci_status(params.target, params.head_sha)
136 
137 
138@activity.defn
139async def record_outcome(params: RecordInput) -> None:
140    """Label the PR and log the run telemetry — the signal-update."""
141    from froot.adapters.github import GitHubForge
142 
143    outcome = params.outcome
144    await GitHubForge().add_labels(params.target, outcome.pr.number, PR_LABELS)
145    _log.info(
146        json.dumps(
147            {
148                "event": "loop_outcome",
149                "loop": "dependency-patch",
150                "repo": params.target.repo.slug,
151                "package": outcome.candidate.package,
152                "from": str(outcome.candidate.current),
153                "to": str(outcome.candidate.target),
154                "changelog": outcome.verdict.kind,
155                "ci": outcome.ci.kind,
156                "ci_passed": outcome.ci_passed,
157                "pr": outcome.pr.number,
158                "pr_url": outcome.pr.url,
159            }
160        )
161    )
162 
163 
164@activity.defn
165async def dispatch_bump(params: DispatchInput) -> None:
166    """Start the bump loop for a candidate (idempotent per bump identity).
167 
168    Reads the close-on-red toggle here, at the impure boundary, and pins it onto
169    the bump's params — so the running workflow never reads config itself and an
170    in-flight bump keeps the value it was dispatched with.
171    """
172    from temporalio.common import WorkflowIDReusePolicy
173    from temporalio.exceptions import WorkflowAlreadyStartedError
174 
175    from froot.config.settings import BehaviorSettings
176    from froot.workflow.bump_workflow import BumpWorkflow
177    from froot.workflow.temporal_client import client, task_queue
178    from froot.workflow.types import BumpParams
179 
180    temporal = await client()
181    try:
182        await temporal.start_workflow(
183            BumpWorkflow.run,
184            BumpParams(
185                target=params.target,
186                candidate=params.candidate,
187                close_on_red=BehaviorSettings().close_on_red,
188            ),
189            id=bump_workflow_id(params.target, params.candidate),
190            task_queue=task_queue(),
191            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
192        )
193    except WorkflowAlreadyStartedError:
194        # This bump already has a loop (running or completed) — a no-op, so
195        # re-scanning never opens a second PR for the same bump.
196        return
197 
198 
199@activity.defn
200async def close_pull_request(params: CloseInput) -> None:
201    """Comment why, then close a red bump's PR and delete its branch.
202 
203    The note goes through the idempotent ``upsert_issue_comment`` and the close
204    itself is idempotent, so a retried close edits its comment in place and
205    never double-posts. The bump's record step still runs after this, so the red
206    outcome is logged either way.
207    """
208    from froot.adapters.github import GitHubForge
209    from froot.policy.compose import CLOSE_MARKER, closed_on_red_comment
210 
211    forge = GitHubForge()
212    body = closed_on_red_comment(params.failing)
213    await forge.upsert_issue_comment(
214        params.target, params.pr.number, CLOSE_MARKER, body
215    )
216    await forge.close_pull_request(
217        params.target, params.pr.number, params.pr.branch
218    )
219    _log.info(
220        json.dumps(
221            {
222                "event": "pr_closed",
223                "loop": "dependency-patch",
224                "reason": "ci_red",
225                "repo": params.target.repo.slug,
226                "pr": params.pr.number,
227                "pr_url": params.pr.url,
228                "failing": list(params.failing),
229            }
230        )
231    )
232 
233 
⋯ 145 lines hidden (lines 234–378)234@activity.defn
235async def reconcile_open_prs(target: TargetRepo) -> int:
236    """Close froot PRs a newer patch superseded or the base already satisfied.
237 
238    Self-contained: lists the repo's open PRs, re-derives the live upgrade facts
239    (a fresh checkout + the package manager's read — derive, never store), asks
240    the pure :func:`~froot.policy.reconcile.reconciliations` policy which to
241    close, and closes each (deleting its branch). A no-op when reconcile is off
242    (``FROOT_RECONCILE``). Returns the number closed.
243    """
244    from froot.adapters.github import GitHubForge
245    from froot.adapters.registry import package_manager_for
246    from froot.config.settings import BehaviorSettings
247    from froot.policy.compose import CLOSE_MARKER
248    from froot.policy.reconcile import reconciliations
249 
250    if not BehaviorSettings().reconcile:
251        return 0
252 
253    forge = GitHubForge()
254    package_manager = package_manager_for(target.ecosystem)
255    open_prs = await forge.list_open_pull_requests(target)
256    with tempfile.TemporaryDirectory() as tmp:
257        workspace = Path(tmp)
258        await forge.checkout(target, workspace)
259        upgrades = await package_manager.list_upgrades(
260            target, _manifest_dir(target, workspace)
261        )
262    closures = reconciliations(open_prs, upgrades)
263    for closure in closures:
264        await forge.upsert_issue_comment(
265            target, closure.pr.number, CLOSE_MARKER, closure.comment
266        )
267        await forge.close_pull_request(
268            target, closure.pr.number, closure.pr.branch
269        )
270    if closures:
271        _reconcile_log.info(
272            json.dumps(
273                {
274                    "event": "reconcile",
275                    "loop": "dependency-patch",
276                    "repo": target.repo.slug,
277                    "closed": len(closures),
278                    "prs": [closure.pr.number for closure in closures],
279                }
280            )
281        )
282    return len(closures)
283 
284 
285# ── The determinism reviewer (the transitive ring) ──────────────────────────
286@activity.defn
287async def list_review_prs(target: TargetRepo) -> tuple[PullRequestRef, ...]:
288    """List the repo's open PRs for the determinism reviewer to consider."""
289    from froot.adapters.github import GitHubForge
290 
291    return await GitHubForge().list_open_pull_requests(target)
292 
293 
294@activity.defn
295async def dispatch_pr_review(params: DispatchReviewInput) -> None:
296    """Start a PR's determinism review (idempotent per PR + head SHA)."""
297    from temporalio.common import WorkflowIDReusePolicy
298    from temporalio.exceptions import WorkflowAlreadyStartedError
299 
300    from froot.workflow.pr_review_workflow import PrReviewWorkflow
301    from froot.workflow.temporal_client import client, task_queue
302 
303    temporal = await client()
304    try:
305        await temporal.start_workflow(
306            PrReviewWorkflow.run,
307            PrReviewParams(target=params.target, pr=params.pr),
308            id=pr_review_workflow_id(
309                params.target, params.pr.number, params.pr.head_sha
310            ),
311            task_queue=task_queue(),
312            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
313        )
314    except WorkflowAlreadyStartedError:
315        # This (PR, head SHA) already has a review — a no-op, so re-polling
316        # never double-reviews the same commit.
317        return
318 
319 
320@activity.defn
321async def analyze_pr(params: PrReviewParams) -> AnalysisResult:
322    """Check out the PR head and analyze the workflow surface for hazards."""
323    from froot.adapters.github import GitHubForge
324    from froot.adapters.source_tree import load_modules
325    from froot.config.settings import ReviewSettings
326 
327    forge = GitHubForge()
328    with tempfile.TemporaryDirectory() as tmp:
329        workspace = Path(tmp)
330        await forge.checkout_pull_request(
331            params.target, workspace, params.pr.number
332        )
333        # The ASTs and source lines are read into memory here, so the analysis
334        # below is unaffected by the workspace being cleaned up.
335        modules = load_modules(workspace)
336    return analyze_workflow_surface(modules, max_depth=ReviewSettings().depth)
337 
338 
339@activity.defn
340async def adjudicate_frontier(
341    params: AdjudicateInput,
342) -> tuple[FrontierVerdict, ...]:
343    """Run the model over each frontier item; return aligned verdicts."""
344    from froot.adapters.determinism_judge import DeterminismFrontierJudge
345 
346    judge = DeterminismFrontierJudge()
347    verdicts: list[FrontierVerdict] = []
348    for item in params.frontier:
349        verdicts.append(await judge.adjudicate(item))
350    return tuple(verdicts)
351 
352 
353@activity.defn
354async def post_review(params: PostReviewInput) -> str | None:
355    """Upsert the advisory comment (when there are findings); log the ledger."""
356    from froot.adapters.github import GitHubForge
357 
358    body = render_review_comment(params.findings, params.pr.head_sha)
359    url: str | None = None
360    if body is not None:
361        url = await GitHubForge().upsert_issue_comment(
362            params.target, params.pr.number, REVIEW_MARKER, body
363        )
364    _review_log.info(
365        json.dumps(
366            {
367                "event": "loop_outcome",
368                "loop": "determinism-review",
369                "repo": params.target.repo.slug,
370                "pr": params.pr.number,
371                "head_sha": params.pr.head_sha,
372                "findings": len(params.findings),
373                "rules": sorted({f.rule for f in params.findings}),
374                "comment_url": url,
375            }
376        )
377    )
378    return url

The reconcile activity is self-contained: list the open PRs, re-derive the live upgrade facts (a fresh checkout — derive, never store), ask the pure policy which to close, and close each. It no-ops when reconcile is switched off.

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

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

⋯ 234 lines hidden (lines 1–234)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
17 
18from temporalio import activity
19 
20from froot.domain.candidate import PatchCandidate
21from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
22from froot.domain.ci import CIStatus
23from froot.domain.determinism import AnalysisResult, FrontierVerdict
24from froot.domain.pull_request import PullRequestRef
25from froot.domain.repo import TargetRepo
26from froot.policy.candidates import select_patch_candidates
27from froot.policy.compose import PR_LABELS, pull_request_draft
28from froot.policy.determinism import analyze_workflow_surface
29from froot.policy.naming import (
30    branch_name,
31    bump_workflow_id,
32    pr_review_workflow_id,
33)
34from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
35from froot.workflow.types import (
36    AdjudicateInput,
37    CiCheckInput,
38    CloseInput,
39    DispatchInput,
40    DispatchReviewInput,
41    OpenPrInput,
42    PostReviewInput,
43    PrReviewParams,
44    RecordInput,
45)
46 
47_log = logging.getLogger("froot.outcome")
48_review_log = logging.getLogger("froot.review")
49_reconcile_log = logging.getLogger("froot.reconcile")
50 
51 
52def _manifest_dir(target: TargetRepo, workspace: Path) -> Path:
53    """The directory the manifest lives in (a monorepo subdir, or the root)."""
54    return workspace / target.manifest_dir if target.manifest_dir else workspace
55 
56 
57@activity.defn
58async def scan_candidates(
59    target: TargetRepo,
60) -> tuple[PatchCandidate, ...]:
61    """Check out the repo, read available upgrades, select patch candidates."""
62    from froot.adapters.github import GitHubForge
63    from froot.adapters.registry import package_manager_for
64 
65    forge = GitHubForge()
66    package_manager = package_manager_for(target.ecosystem)
67    with tempfile.TemporaryDirectory() as tmp:
68        workspace = Path(tmp)
69        await forge.checkout(target, workspace)
70        upgrades = await package_manager.list_upgrades(
71            target, _manifest_dir(target, workspace)
72        )
73    return select_patch_candidates(upgrades)
74 
75 
76@activity.defn
77async def judge_changelog(candidate: PatchCandidate) -> ChangelogVerdict:
78    """Fetch the candidate's changelog and get the model's typed verdict.
79 
80    The model is froot's one thin, non-load-bearing judgment: a clean verdict
81    never *gates* a PR (CI is the oracle), so a model that is down, slow, or
82    erroring must not stall the spine. A judge failure degrades to
83    ``UnknownVerdict`` — the bump proceeds, the human still gets the PR, and the
84    dashboard records the verdict as unknown — rather than failing (and then
85    retrying) the activity. Only the model call is guarded; the fetch is already
86    best-effort (returns ``None``, not an exception).
87    """
88    from froot.adapters.changelog_http import HttpChangelogSource
89    from froot.adapters.model_judge import PydanticAiJudge
90 
91    changelog = await HttpChangelogSource().fetch(candidate)
92    if changelog is None:
93        return UnknownVerdict(rationale="No changelog could be fetched.")
94    try:
95        return await PydanticAiJudge().judge(changelog)
96    except Exception as exc:
97        activity.logger.warning(
98            "changelog judge unavailable for %s; degrading to unknown: %r",
99            candidate.package,
100            exc,
101        )
102        return UnknownVerdict(
103            rationale=f"Changelog judge unavailable ({type(exc).__name__})."
104        )
105 
106 
107@activity.defn
108async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
109    """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
110    from froot.adapters.github import GitHubForge
111    from froot.adapters.registry import package_manager_for
112 
113    forge = GitHubForge()
114    package_manager = package_manager_for(params.target.ecosystem)
115    branch = branch_name(params.candidate)
116    existing = await forge.find_open_pull_request(params.target, branch)
117    if existing is not None:
118        return existing
119    draft = pull_request_draft(params.target, params.candidate, params.verdict)
120    with tempfile.TemporaryDirectory() as tmp:
121        workspace = Path(tmp)
122        await forge.checkout(params.target, workspace)
123        await package_manager.apply_patch_bump(
124            params.candidate, _manifest_dir(params.target, workspace)
125        )
126        await forge.push_branch(workspace, branch, draft.title)
127    return await forge.open_pull_request(params.target, draft)
128 
129 
130@activity.defn
131async def check_ci(params: CiCheckInput) -> CIStatus:
132    """Read the repo's CI status for the PR's head commit (the oracle)."""
133    from froot.adapters.github import GitHubForge
134 
135    return await GitHubForge().ci_status(params.target, params.head_sha)
136 
137 
138@activity.defn
139async def record_outcome(params: RecordInput) -> None:
140    """Label the PR and log the run telemetry — the signal-update."""
141    from froot.adapters.github import GitHubForge
142 
143    outcome = params.outcome
144    await GitHubForge().add_labels(params.target, outcome.pr.number, PR_LABELS)
145    _log.info(
146        json.dumps(
147            {
148                "event": "loop_outcome",
149                "loop": "dependency-patch",
150                "repo": params.target.repo.slug,
151                "package": outcome.candidate.package,
152                "from": str(outcome.candidate.current),
153                "to": str(outcome.candidate.target),
154                "changelog": outcome.verdict.kind,
155                "ci": outcome.ci.kind,
156                "ci_passed": outcome.ci_passed,
157                "pr": outcome.pr.number,
158                "pr_url": outcome.pr.url,
159            }
160        )
161    )
162 
163 
164@activity.defn
165async def dispatch_bump(params: DispatchInput) -> None:
166    """Start the bump loop for a candidate (idempotent per bump identity).
167 
168    Reads the close-on-red toggle here, at the impure boundary, and pins it onto
169    the bump's params — so the running workflow never reads config itself and an
170    in-flight bump keeps the value it was dispatched with.
171    """
172    from temporalio.common import WorkflowIDReusePolicy
173    from temporalio.exceptions import WorkflowAlreadyStartedError
174 
175    from froot.config.settings import BehaviorSettings
176    from froot.workflow.bump_workflow import BumpWorkflow
177    from froot.workflow.temporal_client import client, task_queue
178    from froot.workflow.types import BumpParams
179 
180    temporal = await client()
181    try:
182        await temporal.start_workflow(
183            BumpWorkflow.run,
184            BumpParams(
185                target=params.target,
186                candidate=params.candidate,
187                close_on_red=BehaviorSettings().close_on_red,
188            ),
189            id=bump_workflow_id(params.target, params.candidate),
190            task_queue=task_queue(),
191            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
192        )
193    except WorkflowAlreadyStartedError:
194        # This bump already has a loop (running or completed) — a no-op, so
195        # re-scanning never opens a second PR for the same bump.
196        return
197 
198 
199@activity.defn
200async def close_pull_request(params: CloseInput) -> None:
201    """Comment why, then close a red bump's PR and delete its branch.
202 
203    The note goes through the idempotent ``upsert_issue_comment`` and the close
204    itself is idempotent, so a retried close edits its comment in place and
205    never double-posts. The bump's record step still runs after this, so the red
206    outcome is logged either way.
207    """
208    from froot.adapters.github import GitHubForge
209    from froot.policy.compose import CLOSE_MARKER, closed_on_red_comment
210 
211    forge = GitHubForge()
212    body = closed_on_red_comment(params.failing)
213    await forge.upsert_issue_comment(
214        params.target, params.pr.number, CLOSE_MARKER, body
215    )
216    await forge.close_pull_request(
217        params.target, params.pr.number, params.pr.branch
218    )
219    _log.info(
220        json.dumps(
221            {
222                "event": "pr_closed",
223                "loop": "dependency-patch",
224                "reason": "ci_red",
225                "repo": params.target.repo.slug,
226                "pr": params.pr.number,
227                "pr_url": params.pr.url,
228                "failing": list(params.failing),
229            }
230        )
231    )
232 
233 
234@activity.defn
235async def reconcile_open_prs(target: TargetRepo) -> int:
236    """Close froot PRs a newer patch superseded or the base already satisfied.
237 
238    Self-contained: lists the repo's open PRs, re-derives the live upgrade facts
239    (a fresh checkout + the package manager's read — derive, never store), asks
240    the pure :func:`~froot.policy.reconcile.reconciliations` policy which to
241    close, and closes each (deleting its branch). A no-op when reconcile is off
242    (``FROOT_RECONCILE``). Returns the number closed.
243    """
244    from froot.adapters.github import GitHubForge
245    from froot.adapters.registry import package_manager_for
246    from froot.config.settings import BehaviorSettings
247    from froot.policy.compose import CLOSE_MARKER
248    from froot.policy.reconcile import reconciliations
249 
250    if not BehaviorSettings().reconcile:
251        return 0
252 
253    forge = GitHubForge()
254    package_manager = package_manager_for(target.ecosystem)
255    open_prs = await forge.list_open_pull_requests(target)
256    with tempfile.TemporaryDirectory() as tmp:
257        workspace = Path(tmp)
258        await forge.checkout(target, workspace)
259        upgrades = await package_manager.list_upgrades(
260            target, _manifest_dir(target, workspace)
261        )
262    closures = reconciliations(open_prs, upgrades)
263    for closure in closures:
264        await forge.upsert_issue_comment(
265            target, closure.pr.number, CLOSE_MARKER, closure.comment
266        )
267        await forge.close_pull_request(
268            target, closure.pr.number, closure.pr.branch
269        )
270    if closures:
271        _reconcile_log.info(
272            json.dumps(
273                {
274                    "event": "reconcile",
275                    "loop": "dependency-patch",
276                    "repo": target.repo.slug,
277                    "closed": len(closures),
278                    "prs": [closure.pr.number for closure in closures],
279                }
280            )
281        )
282    return len(closures)
⋯ 96 lines hidden (lines 283–378)283 
284 
285# ── The determinism reviewer (the transitive ring) ──────────────────────────
286@activity.defn
287async def list_review_prs(target: TargetRepo) -> tuple[PullRequestRef, ...]:
288    """List the repo's open PRs for the determinism reviewer to consider."""
289    from froot.adapters.github import GitHubForge
290 
291    return await GitHubForge().list_open_pull_requests(target)
292 
293 
294@activity.defn
295async def dispatch_pr_review(params: DispatchReviewInput) -> None:
296    """Start a PR's determinism review (idempotent per PR + head SHA)."""
297    from temporalio.common import WorkflowIDReusePolicy
298    from temporalio.exceptions import WorkflowAlreadyStartedError
299 
300    from froot.workflow.pr_review_workflow import PrReviewWorkflow
301    from froot.workflow.temporal_client import client, task_queue
302 
303    temporal = await client()
304    try:
305        await temporal.start_workflow(
306            PrReviewWorkflow.run,
307            PrReviewParams(target=params.target, pr=params.pr),
308            id=pr_review_workflow_id(
309                params.target, params.pr.number, params.pr.head_sha
310            ),
311            task_queue=task_queue(),
312            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
313        )
314    except WorkflowAlreadyStartedError:
315        # This (PR, head SHA) already has a review — a no-op, so re-polling
316        # never double-reviews the same commit.
317        return
318 
319 
320@activity.defn
321async def analyze_pr(params: PrReviewParams) -> AnalysisResult:
322    """Check out the PR head and analyze the workflow surface for hazards."""
323    from froot.adapters.github import GitHubForge
324    from froot.adapters.source_tree import load_modules
325    from froot.config.settings import ReviewSettings
326 
327    forge = GitHubForge()
328    with tempfile.TemporaryDirectory() as tmp:
329        workspace = Path(tmp)
330        await forge.checkout_pull_request(
331            params.target, workspace, params.pr.number
332        )
333        # The ASTs and source lines are read into memory here, so the analysis
334        # below is unaffected by the workspace being cleaned up.
335        modules = load_modules(workspace)
336    return analyze_workflow_surface(modules, max_depth=ReviewSettings().depth)
337 
338 
339@activity.defn
340async def adjudicate_frontier(
341    params: AdjudicateInput,
342) -> tuple[FrontierVerdict, ...]:
343    """Run the model over each frontier item; return aligned verdicts."""
344    from froot.adapters.determinism_judge import DeterminismFrontierJudge
345 
346    judge = DeterminismFrontierJudge()
347    verdicts: list[FrontierVerdict] = []
348    for item in params.frontier:
349        verdicts.append(await judge.adjudicate(item))
350    return tuple(verdicts)
351 
352 
353@activity.defn
354async def post_review(params: PostReviewInput) -> str | None:
355    """Upsert the advisory comment (when there are findings); log the ledger."""
356    from froot.adapters.github import GitHubForge
357 
358    body = render_review_comment(params.findings, params.pr.head_sha)
359    url: str | None = None
360    if body is not None:
361        url = await GitHubForge().upsert_issue_comment(
362            params.target, params.pr.number, REVIEW_MARKER, body
363        )
364    _review_log.info(
365        json.dumps(
366            {
367                "event": "loop_outcome",
368                "loop": "determinism-review",
369                "repo": params.target.repo.slug,
370                "pr": params.pr.number,
371                "head_sha": params.pr.head_sha,
372                "findings": len(params.findings),
373                "rules": sorted({f.rule for f in params.findings}),
374                "comment_url": url,
375            }
376        )
377    )
378    return url

The scan loop calls it once per tick, after dispatching the bumps — so the stale-PR cleanup is re-derived from the repo on the same cadence as the scan.

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

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

⋯ 45 lines hidden (lines 1–45)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():
21    from froot.workflow import activities
22    from froot.workflow.constants import (
23        ACTIVITY_TIMEOUT,
24        DISPATCH_TIMEOUT,
25        TOOL_RETRY,
26    )
27    from froot.workflow.types import DispatchInput, ScanParams, ScanResult
28 
29 
30@workflow.defn
31class ScanWorkflow:
32    """The durable dependency-scan loop (one tick per continue-as-new)."""
33 
34    @workflow.run
35    async def run(self, params: ScanParams) -> ScanResult:
36        """Scan, dispatch each, reconcile stale PRs, then loop or return."""
37        candidates = await workflow.execute_activity(
38            activities.scan_candidates,
39            params.target,
40            start_to_close_timeout=ACTIVITY_TIMEOUT,
41            retry_policy=TOOL_RETRY,
42        )
43        for candidate in candidates:
44            await workflow.execute_activity(
45                activities.dispatch_bump,
46                DispatchInput(target=params.target, candidate=candidate),
47                start_to_close_timeout=DISPATCH_TIMEOUT,
48                retry_policy=TOOL_RETRY,
49            )
50        # Close froot PRs a newer patch superseded or the base already
51        # satisfied — re-derived from the repo each tick, same as the scan.
52        reconciled = await workflow.execute_activity(
53            activities.reconcile_open_prs,
54            params.target,
55            start_to_close_timeout=ACTIVITY_TIMEOUT,
56            retry_policy=TOOL_RETRY,
57        )
58        result = ScanResult(
59            found=len(candidates),
60            dispatched=len(candidates),
61            reconciled=reconciled,
62        )
⋯ 12 lines hidden (lines 63–74)63        if not params.continuous:
64            return result
65        # Durable loop: sleep, then restart fresh. continue_as_new raises, so
66        # nothing runs after it and history stays bounded to one tick.
67        await workflow.sleep(timedelta(seconds=params.interval_seconds))
68        workflow.continue_as_new(
69            ScanParams(
70                target=params.target,
71                interval_seconds=params.interval_seconds,
72                continuous=True,
73            )
74        )

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

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

⋯ 179 lines hidden (lines 1–179)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
17 
18from temporalio import activity
19 
20from froot.domain.candidate import PatchCandidate
21from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
22from froot.domain.ci import CIStatus
23from froot.domain.determinism import AnalysisResult, FrontierVerdict
24from froot.domain.pull_request import PullRequestRef
25from froot.domain.repo import TargetRepo
26from froot.policy.candidates import select_patch_candidates
27from froot.policy.compose import PR_LABELS, pull_request_draft
28from froot.policy.determinism import analyze_workflow_surface
29from froot.policy.naming import (
30    branch_name,
31    bump_workflow_id,
32    pr_review_workflow_id,
33)
34from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
35from froot.workflow.types import (
36    AdjudicateInput,
37    CiCheckInput,
38    CloseInput,
39    DispatchInput,
40    DispatchReviewInput,
41    OpenPrInput,
42    PostReviewInput,
43    PrReviewParams,
44    RecordInput,
45)
46 
47_log = logging.getLogger("froot.outcome")
48_review_log = logging.getLogger("froot.review")
49_reconcile_log = logging.getLogger("froot.reconcile")
50 
51 
52def _manifest_dir(target: TargetRepo, workspace: Path) -> Path:
53    """The directory the manifest lives in (a monorepo subdir, or the root)."""
54    return workspace / target.manifest_dir if target.manifest_dir else workspace
55 
56 
57@activity.defn
58async def scan_candidates(
59    target: TargetRepo,
60) -> tuple[PatchCandidate, ...]:
61    """Check out the repo, read available upgrades, select patch candidates."""
62    from froot.adapters.github import GitHubForge
63    from froot.adapters.registry import package_manager_for
64 
65    forge = GitHubForge()
66    package_manager = package_manager_for(target.ecosystem)
67    with tempfile.TemporaryDirectory() as tmp:
68        workspace = Path(tmp)
69        await forge.checkout(target, workspace)
70        upgrades = await package_manager.list_upgrades(
71            target, _manifest_dir(target, workspace)
72        )
73    return select_patch_candidates(upgrades)
74 
75 
76@activity.defn
77async def judge_changelog(candidate: PatchCandidate) -> ChangelogVerdict:
78    """Fetch the candidate's changelog and get the model's typed verdict.
79 
80    The model is froot's one thin, non-load-bearing judgment: a clean verdict
81    never *gates* a PR (CI is the oracle), so a model that is down, slow, or
82    erroring must not stall the spine. A judge failure degrades to
83    ``UnknownVerdict`` — the bump proceeds, the human still gets the PR, and the
84    dashboard records the verdict as unknown — rather than failing (and then
85    retrying) the activity. Only the model call is guarded; the fetch is already
86    best-effort (returns ``None``, not an exception).
87    """
88    from froot.adapters.changelog_http import HttpChangelogSource
89    from froot.adapters.model_judge import PydanticAiJudge
90 
91    changelog = await HttpChangelogSource().fetch(candidate)
92    if changelog is None:
93        return UnknownVerdict(rationale="No changelog could be fetched.")
94    try:
95        return await PydanticAiJudge().judge(changelog)
96    except Exception as exc:
97        activity.logger.warning(
98            "changelog judge unavailable for %s; degrading to unknown: %r",
99            candidate.package,
100            exc,
101        )
102        return UnknownVerdict(
103            rationale=f"Changelog judge unavailable ({type(exc).__name__})."
104        )
105 
106 
107@activity.defn
108async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
109    """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
110    from froot.adapters.github import GitHubForge
111    from froot.adapters.registry import package_manager_for
112 
113    forge = GitHubForge()
114    package_manager = package_manager_for(params.target.ecosystem)
115    branch = branch_name(params.candidate)
116    existing = await forge.find_open_pull_request(params.target, branch)
117    if existing is not None:
118        return existing
119    draft = pull_request_draft(params.target, params.candidate, params.verdict)
120    with tempfile.TemporaryDirectory() as tmp:
121        workspace = Path(tmp)
122        await forge.checkout(params.target, workspace)
123        await package_manager.apply_patch_bump(
124            params.candidate, _manifest_dir(params.target, workspace)
125        )
126        await forge.push_branch(workspace, branch, draft.title)
127    return await forge.open_pull_request(params.target, draft)
128 
129 
130@activity.defn
131async def check_ci(params: CiCheckInput) -> CIStatus:
132    """Read the repo's CI status for the PR's head commit (the oracle)."""
133    from froot.adapters.github import GitHubForge
134 
135    return await GitHubForge().ci_status(params.target, params.head_sha)
136 
137 
138@activity.defn
139async def record_outcome(params: RecordInput) -> None:
140    """Label the PR and log the run telemetry — the signal-update."""
141    from froot.adapters.github import GitHubForge
142 
143    outcome = params.outcome
144    await GitHubForge().add_labels(params.target, outcome.pr.number, PR_LABELS)
145    _log.info(
146        json.dumps(
147            {
148                "event": "loop_outcome",
149                "loop": "dependency-patch",
150                "repo": params.target.repo.slug,
151                "package": outcome.candidate.package,
152                "from": str(outcome.candidate.current),
153                "to": str(outcome.candidate.target),
154                "changelog": outcome.verdict.kind,
155                "ci": outcome.ci.kind,
156                "ci_passed": outcome.ci_passed,
157                "pr": outcome.pr.number,
158                "pr_url": outcome.pr.url,
159            }
160        )
161    )
162 
163 
164@activity.defn
165async def dispatch_bump(params: DispatchInput) -> None:
166    """Start the bump loop for a candidate (idempotent per bump identity).
167 
168    Reads the close-on-red toggle here, at the impure boundary, and pins it onto
169    the bump's params — so the running workflow never reads config itself and an
170    in-flight bump keeps the value it was dispatched with.
171    """
172    from temporalio.common import WorkflowIDReusePolicy
173    from temporalio.exceptions import WorkflowAlreadyStartedError
174 
175    from froot.config.settings import BehaviorSettings
176    from froot.workflow.bump_workflow import BumpWorkflow
177    from froot.workflow.temporal_client import client, task_queue
178    from froot.workflow.types import BumpParams
179 
180    temporal = await client()
181    try:
182        await temporal.start_workflow(
183            BumpWorkflow.run,
184            BumpParams(
185                target=params.target,
186                candidate=params.candidate,
187                close_on_red=BehaviorSettings().close_on_red,
188            ),
189            id=bump_workflow_id(params.target, params.candidate),
190            task_queue=task_queue(),
191            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
192        )
193    except WorkflowAlreadyStartedError:
194        # This bump already has a loop (running or completed) — a no-op, so
195        # re-scanning never opens a second PR for the same bump.
196        return
⋯ 182 lines hidden (lines 197–378)197 
198 
199@activity.defn
200async def close_pull_request(params: CloseInput) -> None:
201    """Comment why, then close a red bump's PR and delete its branch.
202 
203    The note goes through the idempotent ``upsert_issue_comment`` and the close
204    itself is idempotent, so a retried close edits its comment in place and
205    never double-posts. The bump's record step still runs after this, so the red
206    outcome is logged either way.
207    """
208    from froot.adapters.github import GitHubForge
209    from froot.policy.compose import CLOSE_MARKER, closed_on_red_comment
210 
211    forge = GitHubForge()
212    body = closed_on_red_comment(params.failing)
213    await forge.upsert_issue_comment(
214        params.target, params.pr.number, CLOSE_MARKER, body
215    )
216    await forge.close_pull_request(
217        params.target, params.pr.number, params.pr.branch
218    )
219    _log.info(
220        json.dumps(
221            {
222                "event": "pr_closed",
223                "loop": "dependency-patch",
224                "reason": "ci_red",
225                "repo": params.target.repo.slug,
226                "pr": params.pr.number,
227                "pr_url": params.pr.url,
228                "failing": list(params.failing),
229            }
230        )
231    )
232 
233 
234@activity.defn
235async def reconcile_open_prs(target: TargetRepo) -> int:
236    """Close froot PRs a newer patch superseded or the base already satisfied.
237 
238    Self-contained: lists the repo's open PRs, re-derives the live upgrade facts
239    (a fresh checkout + the package manager's read — derive, never store), asks
240    the pure :func:`~froot.policy.reconcile.reconciliations` policy which to
241    close, and closes each (deleting its branch). A no-op when reconcile is off
242    (``FROOT_RECONCILE``). Returns the number closed.
243    """
244    from froot.adapters.github import GitHubForge
245    from froot.adapters.registry import package_manager_for
246    from froot.config.settings import BehaviorSettings
247    from froot.policy.compose import CLOSE_MARKER
248    from froot.policy.reconcile import reconciliations
249 
250    if not BehaviorSettings().reconcile:
251        return 0
252 
253    forge = GitHubForge()
254    package_manager = package_manager_for(target.ecosystem)
255    open_prs = await forge.list_open_pull_requests(target)
256    with tempfile.TemporaryDirectory() as tmp:
257        workspace = Path(tmp)
258        await forge.checkout(target, workspace)
259        upgrades = await package_manager.list_upgrades(
260            target, _manifest_dir(target, workspace)
261        )
262    closures = reconciliations(open_prs, upgrades)
263    for closure in closures:
264        await forge.upsert_issue_comment(
265            target, closure.pr.number, CLOSE_MARKER, closure.comment
266        )
267        await forge.close_pull_request(
268            target, closure.pr.number, closure.pr.branch
269        )
270    if closures:
271        _reconcile_log.info(
272            json.dumps(
273                {
274                    "event": "reconcile",
275                    "loop": "dependency-patch",
276                    "repo": target.repo.slug,
277                    "closed": len(closures),
278                    "prs": [closure.pr.number for closure in closures],
279                }
280            )
281        )
282    return len(closures)
283 
284 
285# ── The determinism reviewer (the transitive ring) ──────────────────────────
286@activity.defn
287async def list_review_prs(target: TargetRepo) -> tuple[PullRequestRef, ...]:
288    """List the repo's open PRs for the determinism reviewer to consider."""
289    from froot.adapters.github import GitHubForge
290 
291    return await GitHubForge().list_open_pull_requests(target)
292 
293 
294@activity.defn
295async def dispatch_pr_review(params: DispatchReviewInput) -> None:
296    """Start a PR's determinism review (idempotent per PR + head SHA)."""
297    from temporalio.common import WorkflowIDReusePolicy
298    from temporalio.exceptions import WorkflowAlreadyStartedError
299 
300    from froot.workflow.pr_review_workflow import PrReviewWorkflow
301    from froot.workflow.temporal_client import client, task_queue
302 
303    temporal = await client()
304    try:
305        await temporal.start_workflow(
306            PrReviewWorkflow.run,
307            PrReviewParams(target=params.target, pr=params.pr),
308            id=pr_review_workflow_id(
309                params.target, params.pr.number, params.pr.head_sha
310            ),
311            task_queue=task_queue(),
312            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
313        )
314    except WorkflowAlreadyStartedError:
315        # This (PR, head SHA) already has a review — a no-op, so re-polling
316        # never double-reviews the same commit.
317        return
318 
319 
320@activity.defn
321async def analyze_pr(params: PrReviewParams) -> AnalysisResult:
322    """Check out the PR head and analyze the workflow surface for hazards."""
323    from froot.adapters.github import GitHubForge
324    from froot.adapters.source_tree import load_modules
325    from froot.config.settings import ReviewSettings
326 
327    forge = GitHubForge()
328    with tempfile.TemporaryDirectory() as tmp:
329        workspace = Path(tmp)
330        await forge.checkout_pull_request(
331            params.target, workspace, params.pr.number
332        )
333        # The ASTs and source lines are read into memory here, so the analysis
334        # below is unaffected by the workspace being cleaned up.
335        modules = load_modules(workspace)
336    return analyze_workflow_surface(modules, max_depth=ReviewSettings().depth)
337 
338 
339@activity.defn
340async def adjudicate_frontier(
341    params: AdjudicateInput,
342) -> tuple[FrontierVerdict, ...]:
343    """Run the model over each frontier item; return aligned verdicts."""
344    from froot.adapters.determinism_judge import DeterminismFrontierJudge
345 
346    judge = DeterminismFrontierJudge()
347    verdicts: list[FrontierVerdict] = []
348    for item in params.frontier:
349        verdicts.append(await judge.adjudicate(item))
350    return tuple(verdicts)
351 
352 
353@activity.defn
354async def post_review(params: PostReviewInput) -> str | None:
355    """Upsert the advisory comment (when there are findings); log the ledger."""
356    from froot.adapters.github import GitHubForge
357 
358    body = render_review_comment(params.findings, params.pr.head_sha)
359    url: str | None = None
360    if body is not None:
361        url = await GitHubForge().upsert_issue_comment(
362            params.target, params.pr.number, REVIEW_MARKER, body
363        )
364    _review_log.info(
365        json.dumps(
366            {
367                "event": "loop_outcome",
368                "loop": "determinism-review",
369                "repo": params.target.repo.slug,
370                "pr": params.pr.number,
371                "head_sha": params.pr.head_sha,
372                "findings": len(params.findings),
373                "rules": sorted({f.rule for f in params.findings}),
374                "comment_url": url,
375            }
376        )
377    )
378    return url

Failure discipline6🎬 Two resilience choices, opposite directions

Surface persistent faults; never stall on the model

src/froot/workflow/constants.pysrc/froot/workflow/activities.py

Before this PR, activities ran under Temporal's default retry — unbounded. A genuinely stuck fault (a deleted commit, a sustained outage) would loop forever, invisible. A bounded policy makes it eventually fail, where the dashboard's Failures panel can show it.

src/froot/workflow/constants.py · 35 lines

src/froot/workflow/constants.py35 lines · Python

⋯ 25 lines hidden (lines 1–25)1"""Operational constants for the froot workflows (sandbox-safe).
2 
3stdlib plus Temporal's own :class:`~temporalio.common.RetryPolicy` (a plain,
4deterministic value object) — nothing here touches I/O or the clock, so the
5module is safe to import inside a workflow.
6"""
7 
8from __future__ import annotations
9 
10from datetime import timedelta
11 
12from temporalio.common import RetryPolicy
13 
14# Generous per-activity ceiling for the tool-backed steps: a checkout + npm +
15# git + a (possibly cold) local model call all fit; the bound only trips a hang.
16ACTIVITY_TIMEOUT = timedelta(minutes=10)
17# Reading CI status or setting labels is a quick GitHub API call.
18CI_CHECK_TIMEOUT = timedelta(seconds=60)
19# Dispatching a bump loop is a fast Temporal client call.
20DISPATCH_TIMEOUT = timedelta(seconds=30)
21# The durable CI wait: how often to poll, and how long before giving up. The
22# wait is a workflow timer (durable, free while idle) — this is Temporal's
23# sweet spot, so froot can sit on a slow CI run for an hour without holding any
24# process open.
25CI_POLL_INTERVAL = timedelta(minutes=1)
26CI_WAIT_DEADLINE = timedelta(hours=1)
27 
28# Bound every activity's retries. The default policy retries forever, so a
29# persistent fault (a deleted commit, a sustained GitHub outage, a genuinely
30# broken tool) would loop invisibly; capping attempts makes it surface as a
31# failed workflow in the dashboard's Failures panel instead. Permanent faults
32# (a missing or invalid token) are already raised non-retryable by the adapters,
33# so they stop on the first attempt regardless of this bound. Six attempts with
34# the default exponential backoff (1s, 2s, 4s, …) ride out a transient blip.
35TOOL_RETRY = RetryPolicy(maximum_attempts=6)

TOOL_RETRY is attached to every execute_activity call across the four workflows. Permanent faults are unaffected — the adapter already raises a missing or invalid token as non-retryable, so it still stops on the first attempt.

The new ClosePullRequest arm, under the same bounded retry as every other activity call.

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

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

⋯ 119 lines hidden (lines 1–119)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,
      OpenPullRequest,
      RecordOutcome,
  )
  from froot.domain.events import (
      ChangelogJudged,
      CiResolved,
      LoopEvent,
      OutcomeRecorded,
      PullRequestClosed,
      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 (
      BumpParams,
      CiCheckInput,
      CloseInput,
      OpenPrInput,
      RecordInput,
  )
55 
56if 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.pull_request import PullRequestRef
  from froot.domain.repo import TargetRepo
61 
62 
63@workflow.defn
64class BumpWorkflow:
  """The durable loop for a single dependency patch bump."""
66 
  @workflow.run
  async def run(self, params: BumpParams) -> LoopOutcome:
      """Drive the pure state machine to a recorded outcome."""
      transition = start(params.candidate)
      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, transition.effects[0])
          transition = advance(state, event, close_on_red=params.close_on_red)
          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
93 
  async def _execute(self, target: TargetRepo, effect: Effect) -> LoopEvent:
      """Interpret one effect into an activity (or a durable CI wait)."""
      match effect:
          case JudgeChangelog():
              verdict = await workflow.execute_activity(
                  activities.judge_changelog,
                  effect.candidate,
                  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,
                  ),
                  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,
                  ),
                  start_to_close_timeout=CI_CHECK_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
              return PullRequestClosed()
          case RecordOutcome():
              await workflow.execute_activity(
                  activities.record_outcome,
                  RecordInput(target=target, outcome=effect.outcome),
                  start_to_close_timeout=CI_CHECK_TIMEOUT,
                  retry_policy=TOOL_RETRY,
              )
⋯ 20 lines hidden (lines 139–158)139                return OutcomeRecorded()
      assert_never(effect)
141 
  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)

The model goes the other way. It is froot's one thin judgment, and it never gates a PR — CI is the oracle. So a model that is down must not fail (and then retry) the activity and stall the bump. It degrades to an "unknown" verdict instead, logged for visibility.

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

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

⋯ 76 lines hidden (lines 1–76)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
17 
18from temporalio import activity
19 
20from froot.domain.candidate import PatchCandidate
21from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
22from froot.domain.ci import CIStatus
23from froot.domain.determinism import AnalysisResult, FrontierVerdict
24from froot.domain.pull_request import PullRequestRef
25from froot.domain.repo import TargetRepo
26from froot.policy.candidates import select_patch_candidates
27from froot.policy.compose import PR_LABELS, pull_request_draft
28from froot.policy.determinism import analyze_workflow_surface
29from froot.policy.naming import (
30    branch_name,
31    bump_workflow_id,
32    pr_review_workflow_id,
33)
34from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
35from froot.workflow.types import (
36    AdjudicateInput,
37    CiCheckInput,
38    CloseInput,
39    DispatchInput,
40    DispatchReviewInput,
41    OpenPrInput,
42    PostReviewInput,
43    PrReviewParams,
44    RecordInput,
45)
46 
47_log = logging.getLogger("froot.outcome")
48_review_log = logging.getLogger("froot.review")
49_reconcile_log = logging.getLogger("froot.reconcile")
50 
51 
52def _manifest_dir(target: TargetRepo, workspace: Path) -> Path:
53    """The directory the manifest lives in (a monorepo subdir, or the root)."""
54    return workspace / target.manifest_dir if target.manifest_dir else workspace
55 
56 
57@activity.defn
58async def scan_candidates(
59    target: TargetRepo,
60) -> tuple[PatchCandidate, ...]:
61    """Check out the repo, read available upgrades, select patch candidates."""
62    from froot.adapters.github import GitHubForge
63    from froot.adapters.registry import package_manager_for
64 
65    forge = GitHubForge()
66    package_manager = package_manager_for(target.ecosystem)
67    with tempfile.TemporaryDirectory() as tmp:
68        workspace = Path(tmp)
69        await forge.checkout(target, workspace)
70        upgrades = await package_manager.list_upgrades(
71            target, _manifest_dir(target, workspace)
72        )
73    return select_patch_candidates(upgrades)
74 
75 
76@activity.defn
77async def judge_changelog(candidate: PatchCandidate) -> ChangelogVerdict:
78    """Fetch the candidate's changelog and get the model's typed verdict.
79 
80    The model is froot's one thin, non-load-bearing judgment: a clean verdict
81    never *gates* a PR (CI is the oracle), so a model that is down, slow, or
82    erroring must not stall the spine. A judge failure degrades to
83    ``UnknownVerdict`` — the bump proceeds, the human still gets the PR, and the
84    dashboard records the verdict as unknown — rather than failing (and then
85    retrying) the activity. Only the model call is guarded; the fetch is already
86    best-effort (returns ``None``, not an exception).
87    """
88    from froot.adapters.changelog_http import HttpChangelogSource
89    from froot.adapters.model_judge import PydanticAiJudge
90 
91    changelog = await HttpChangelogSource().fetch(candidate)
92    if changelog is None:
93        return UnknownVerdict(rationale="No changelog could be fetched.")
94    try:
95        return await PydanticAiJudge().judge(changelog)
96    except Exception as exc:
97        activity.logger.warning(
98            "changelog judge unavailable for %s; degrading to unknown: %r",
99            candidate.package,
100            exc,
101        )
102        return UnknownVerdict(
103            rationale=f"Changelog judge unavailable ({type(exc).__name__})."
104        )
⋯ 274 lines hidden (lines 105–378)105 
106 
107@activity.defn
108async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
109    """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
110    from froot.adapters.github import GitHubForge
111    from froot.adapters.registry import package_manager_for
112 
113    forge = GitHubForge()
114    package_manager = package_manager_for(params.target.ecosystem)
115    branch = branch_name(params.candidate)
116    existing = await forge.find_open_pull_request(params.target, branch)
117    if existing is not None:
118        return existing
119    draft = pull_request_draft(params.target, params.candidate, params.verdict)
120    with tempfile.TemporaryDirectory() as tmp:
121        workspace = Path(tmp)
122        await forge.checkout(params.target, workspace)
123        await package_manager.apply_patch_bump(
124            params.candidate, _manifest_dir(params.target, workspace)
125        )
126        await forge.push_branch(workspace, branch, draft.title)
127    return await forge.open_pull_request(params.target, draft)
128 
129 
130@activity.defn
131async def check_ci(params: CiCheckInput) -> CIStatus:
132    """Read the repo's CI status for the PR's head commit (the oracle)."""
133    from froot.adapters.github import GitHubForge
134 
135    return await GitHubForge().ci_status(params.target, params.head_sha)
136 
137 
138@activity.defn
139async def record_outcome(params: RecordInput) -> None:
140    """Label the PR and log the run telemetry — the signal-update."""
141    from froot.adapters.github import GitHubForge
142 
143    outcome = params.outcome
144    await GitHubForge().add_labels(params.target, outcome.pr.number, PR_LABELS)
145    _log.info(
146        json.dumps(
147            {
148                "event": "loop_outcome",
149                "loop": "dependency-patch",
150                "repo": params.target.repo.slug,
151                "package": outcome.candidate.package,
152                "from": str(outcome.candidate.current),
153                "to": str(outcome.candidate.target),
154                "changelog": outcome.verdict.kind,
155                "ci": outcome.ci.kind,
156                "ci_passed": outcome.ci_passed,
157                "pr": outcome.pr.number,
158                "pr_url": outcome.pr.url,
159            }
160        )
161    )
162 
163 
164@activity.defn
165async def dispatch_bump(params: DispatchInput) -> None:
166    """Start the bump loop for a candidate (idempotent per bump identity).
167 
168    Reads the close-on-red toggle here, at the impure boundary, and pins it onto
169    the bump's params — so the running workflow never reads config itself and an
170    in-flight bump keeps the value it was dispatched with.
171    """
172    from temporalio.common import WorkflowIDReusePolicy
173    from temporalio.exceptions import WorkflowAlreadyStartedError
174 
175    from froot.config.settings import BehaviorSettings
176    from froot.workflow.bump_workflow import BumpWorkflow
177    from froot.workflow.temporal_client import client, task_queue
178    from froot.workflow.types import BumpParams
179 
180    temporal = await client()
181    try:
182        await temporal.start_workflow(
183            BumpWorkflow.run,
184            BumpParams(
185                target=params.target,
186                candidate=params.candidate,
187                close_on_red=BehaviorSettings().close_on_red,
188            ),
189            id=bump_workflow_id(params.target, params.candidate),
190            task_queue=task_queue(),
191            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
192        )
193    except WorkflowAlreadyStartedError:
194        # This bump already has a loop (running or completed) — a no-op, so
195        # re-scanning never opens a second PR for the same bump.
196        return
197 
198 
199@activity.defn
200async def close_pull_request(params: CloseInput) -> None:
201    """Comment why, then close a red bump's PR and delete its branch.
202 
203    The note goes through the idempotent ``upsert_issue_comment`` and the close
204    itself is idempotent, so a retried close edits its comment in place and
205    never double-posts. The bump's record step still runs after this, so the red
206    outcome is logged either way.
207    """
208    from froot.adapters.github import GitHubForge
209    from froot.policy.compose import CLOSE_MARKER, closed_on_red_comment
210 
211    forge = GitHubForge()
212    body = closed_on_red_comment(params.failing)
213    await forge.upsert_issue_comment(
214        params.target, params.pr.number, CLOSE_MARKER, body
215    )
216    await forge.close_pull_request(
217        params.target, params.pr.number, params.pr.branch
218    )
219    _log.info(
220        json.dumps(
221            {
222                "event": "pr_closed",
223                "loop": "dependency-patch",
224                "reason": "ci_red",
225                "repo": params.target.repo.slug,
226                "pr": params.pr.number,
227                "pr_url": params.pr.url,
228                "failing": list(params.failing),
229            }
230        )
231    )
232 
233 
234@activity.defn
235async def reconcile_open_prs(target: TargetRepo) -> int:
236    """Close froot PRs a newer patch superseded or the base already satisfied.
237 
238    Self-contained: lists the repo's open PRs, re-derives the live upgrade facts
239    (a fresh checkout + the package manager's read — derive, never store), asks
240    the pure :func:`~froot.policy.reconcile.reconciliations` policy which to
241    close, and closes each (deleting its branch). A no-op when reconcile is off
242    (``FROOT_RECONCILE``). Returns the number closed.
243    """
244    from froot.adapters.github import GitHubForge
245    from froot.adapters.registry import package_manager_for
246    from froot.config.settings import BehaviorSettings
247    from froot.policy.compose import CLOSE_MARKER
248    from froot.policy.reconcile import reconciliations
249 
250    if not BehaviorSettings().reconcile:
251        return 0
252 
253    forge = GitHubForge()
254    package_manager = package_manager_for(target.ecosystem)
255    open_prs = await forge.list_open_pull_requests(target)
256    with tempfile.TemporaryDirectory() as tmp:
257        workspace = Path(tmp)
258        await forge.checkout(target, workspace)
259        upgrades = await package_manager.list_upgrades(
260            target, _manifest_dir(target, workspace)
261        )
262    closures = reconciliations(open_prs, upgrades)
263    for closure in closures:
264        await forge.upsert_issue_comment(
265            target, closure.pr.number, CLOSE_MARKER, closure.comment
266        )
267        await forge.close_pull_request(
268            target, closure.pr.number, closure.pr.branch
269        )
270    if closures:
271        _reconcile_log.info(
272            json.dumps(
273                {
274                    "event": "reconcile",
275                    "loop": "dependency-patch",
276                    "repo": target.repo.slug,
277                    "closed": len(closures),
278                    "prs": [closure.pr.number for closure in closures],
279                }
280            )
281        )
282    return len(closures)
283 
284 
285# ── The determinism reviewer (the transitive ring) ──────────────────────────
286@activity.defn
287async def list_review_prs(target: TargetRepo) -> tuple[PullRequestRef, ...]:
288    """List the repo's open PRs for the determinism reviewer to consider."""
289    from froot.adapters.github import GitHubForge
290 
291    return await GitHubForge().list_open_pull_requests(target)
292 
293 
294@activity.defn
295async def dispatch_pr_review(params: DispatchReviewInput) -> None:
296    """Start a PR's determinism review (idempotent per PR + head SHA)."""
297    from temporalio.common import WorkflowIDReusePolicy
298    from temporalio.exceptions import WorkflowAlreadyStartedError
299 
300    from froot.workflow.pr_review_workflow import PrReviewWorkflow
301    from froot.workflow.temporal_client import client, task_queue
302 
303    temporal = await client()
304    try:
305        await temporal.start_workflow(
306            PrReviewWorkflow.run,
307            PrReviewParams(target=params.target, pr=params.pr),
308            id=pr_review_workflow_id(
309                params.target, params.pr.number, params.pr.head_sha
310            ),
311            task_queue=task_queue(),
312            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
313        )
314    except WorkflowAlreadyStartedError:
315        # This (PR, head SHA) already has a review — a no-op, so re-polling
316        # never double-reviews the same commit.
317        return
318 
319 
320@activity.defn
321async def analyze_pr(params: PrReviewParams) -> AnalysisResult:
322    """Check out the PR head and analyze the workflow surface for hazards."""
323    from froot.adapters.github import GitHubForge
324    from froot.adapters.source_tree import load_modules
325    from froot.config.settings import ReviewSettings
326 
327    forge = GitHubForge()
328    with tempfile.TemporaryDirectory() as tmp:
329        workspace = Path(tmp)
330        await forge.checkout_pull_request(
331            params.target, workspace, params.pr.number
332        )
333        # The ASTs and source lines are read into memory here, so the analysis
334        # below is unaffected by the workspace being cleaned up.
335        modules = load_modules(workspace)
336    return analyze_workflow_surface(modules, max_depth=ReviewSettings().depth)
337 
338 
339@activity.defn
340async def adjudicate_frontier(
341    params: AdjudicateInput,
342) -> tuple[FrontierVerdict, ...]:
343    """Run the model over each frontier item; return aligned verdicts."""
344    from froot.adapters.determinism_judge import DeterminismFrontierJudge
345 
346    judge = DeterminismFrontierJudge()
347    verdicts: list[FrontierVerdict] = []
348    for item in params.frontier:
349        verdicts.append(await judge.adjudicate(item))
350    return tuple(verdicts)
351 
352 
353@activity.defn
354async def post_review(params: PostReviewInput) -> str | None:
355    """Upsert the advisory comment (when there are findings); log the ledger."""
356    from froot.adapters.github import GitHubForge
357 
358    body = render_review_comment(params.findings, params.pr.head_sha)
359    url: str | None = None
360    if body is not None:
361        url = await GitHubForge().upsert_issue_comment(
362            params.target, params.pr.number, REVIEW_MARKER, body
363        )
364    _review_log.info(
365        json.dumps(
366            {
367                "event": "loop_outcome",
368                "loop": "determinism-review",
369                "repo": params.target.repo.slug,
370                "pr": params.pr.number,
371                "head_sha": params.pr.head_sha,
372                "findings": len(params.findings),
373                "rules": sorted({f.rule for f in params.findings}),
374                "comment_url": url,
375            }
376        )
377    )
378    return url

How it's proven7🎬 Evidence

The real loop, and a gate that holds the line

tests/test_e2e_workflow.pyscripts/check_determinism_transitive.py.github/workflows/determinism.yml

The existing workflow tests mock the activities to exercise the spine. This PR adds an end-to-end test that does the opposite: it runs the real activities and only fakes the adapters underneath them, so the actual effect interpreters drive the loop. Both terminal shapes are covered, plus the durable CI wait.

One shared fake forge wired under the real activities; the red case asserts the PR is closed, its branch deleted, and recorded.

tests/test_e2e_workflow.py · 154 lines

tests/test_e2e_workflow.py154 lines · Python

⋯ 54 lines hidden (lines 1–54)1"""End-to-end: the real bump loop closing through the real activities.
2 
3Unlike ``test_bump_workflow`` (which mocks the activities to exercise just the
4spine), this wires the *real* activities — judge, open, check-ci, record, close
5— and only swaps the adapters underneath them for in-memory fakes. So it proves
6the whole chain closes: the durable workflow drives the pure state machine,
7which drives the real effect interpreters, which drive the ports. Both terminal
8shapes are covered — a green bump that records and stays open for the human, and
9a red bump that the loop closes (and whose branch it deletes) before recording.
10"""
11 
12from __future__ import annotations
13 
14from collections.abc import Callable
15from typing import Any
16 
17import pytest
18from temporalio.client import Client
19from temporalio.testing import WorkflowEnvironment
20from temporalio.worker import Worker
21 
22import froot.adapters.changelog_http as changelog_mod
23import froot.adapters.github as github_mod
24import froot.adapters.model_judge as model_mod
25import froot.adapters.registry as registry_mod
26from froot.domain.changelog import Changelog, CleanVerdict
27from froot.domain.ci import CIFailed, CIPassed, CIPending
28from froot.domain.outcome import LoopOutcome
29from froot.workflow import activities
30from froot.workflow.bump_workflow import BumpWorkflow
31from froot.workflow.runtime import DATA_CONVERTER
32from froot.workflow.types import BumpParams
33from tests.support import (
34    FakeChangelogSource,
35    FakeForge,
36    FakeJudge,
37    FakePackageManager,
38    make_candidate,
39    make_pr,
40    make_repo,
41    ver,
42)
43 
44_TASK_QUEUE = "froot-e2e"
45# The real bump activities — only the adapters beneath them are faked.
46_REAL_ACTIVITIES: list[Callable[..., Any]] = [
47    activities.judge_changelog,
48    activities.open_pull_request,
49    activities.check_ci,
50    activities.record_outcome,
51    activities.close_pull_request,
52]
53 
54 
55def _wire_fakes(monkeypatch: pytest.MonkeyPatch, fake: FakeForge) -> None:
56    """Point every adapter the bump activities reach at the shared fake."""
57    monkeypatch.setattr(github_mod, "GitHubForge", lambda: fake)
58    monkeypatch.setattr(
59        registry_mod,
60        "package_manager_for",
61        lambda ecosystem: FakePackageManager(),
62    )
63    monkeypatch.setattr(
64        changelog_mod,
65        "HttpChangelogSource",
66        lambda: FakeChangelogSource(
67            Changelog(package="left-pad", version=ver("1.4.3"), text="fix")
68        ),
69    )
70    monkeypatch.setattr(
71        model_mod,
72        "PydanticAiJudge",
73        lambda: FakeJudge(CleanVerdict(rationale="clean")),
74    )
75 
76 
77async def _pydantic_client(env: WorkflowEnvironment) -> Client:
⋯ 54 lines hidden (lines 78–131)78    config = env.client.config()
79    config["data_converter"] = DATA_CONVERTER
80    return Client(**config)
81 
82 
83async def _run(*, close_on_red: bool = True) -> LoopOutcome:
84    async with await WorkflowEnvironment.start_time_skipping() as env:
85        client = await _pydantic_client(env)
86        async with Worker(
87            client,
88            task_queue=_TASK_QUEUE,
89            workflows=[BumpWorkflow],
90            activities=_REAL_ACTIVITIES,
91        ):
92            return await client.execute_workflow(
93                BumpWorkflow.run,
94                BumpParams(
95                    target=make_repo(),
96                    candidate=make_candidate(),
97                    close_on_red=close_on_red,
98                ),
99                id="bump-e2e",
100                task_queue=_TASK_QUEUE,
101            )
102 
103 
104async def test_green_bump_records_and_stays_open(
105    monkeypatch: pytest.MonkeyPatch,
106):
107    fake = FakeForge(opened_pr=make_pr(number=7), ci=CIPassed())
108    _wire_fakes(monkeypatch, fake)
109    outcome = await _run()
110    assert outcome.pr.number == 7
111    assert outcome.ci_passed
112    # The PR is recorded (labeled) and left open for the human — never closed.
113    assert set(fake.labeled or ()) == {"froot", "dependency-patch"}
114    assert fake.closed == []
115 
116 
117async def test_green_bump_waits_through_pending_then_records(
118    monkeypatch: pytest.MonkeyPatch,
119):
120    # The real check_ci activity polls the fake through pending -> pending ->
121    # passed; the workflow's durable wait (time-skipped here) carries it.
122    fake = FakeForge(
123        opened_pr=make_pr(number=7),
124        ci_sequence=(CIPending(), CIPending(), CIPassed()),
125    )
126    _wire_fakes(monkeypatch, fake)
127    outcome = await _run()
128    assert outcome.ci_passed
129    assert fake.closed == []
130 
131 
132async def test_red_bump_closes_pr_and_records(
133    monkeypatch: pytest.MonkeyPatch,
134):
135    fake = FakeForge(opened_pr=make_pr(number=7), ci=CIFailed(failing=("ci",)))
136    _wire_fakes(monkeypatch, fake)
137    outcome = await _run()
138    assert isinstance(outcome.ci, CIFailed)
139    # Closed (with its branch deleted) AND recorded: the loop closed cleanly.
140    assert fake.closed == [7]
141    assert fake.deleted_branches == [make_pr(number=7).branch]
142    assert set(fake.labeled or ()) == {"froot", "dependency-patch"}
143 
144 
145async def test_red_bump_stays_open_when_close_on_red_off(
146    monkeypatch: pytest.MonkeyPatch,
147):
148    fake = FakeForge(opened_pr=make_pr(number=7), ci=CIFailed(failing=("ci",)))
⋯ 6 lines hidden (lines 149–154)149    _wire_fakes(monkeypatch, fake)
150    outcome = await _run(close_on_red=False)
151    assert isinstance(outcome.ci, CIFailed)
152    # Toggle off: recorded but left open, nothing closed.
153    assert fake.closed == []
154    assert set(fake.labeled or ()) == {"froot", "dependency-patch"}

The second proof is a gate. froot already ships a transitive determinism analyzer — the one its reviewer loop points at other repos. This PR points it at froot's own source, so a hazard hidden behind a first-party helper (which the lexical kernel cannot see) fails froot's own CI too.

scripts/check_determinism_transitive.py · 71 lines

scripts/check_determinism_transitive.py71 lines · Python

⋯ 26 lines hidden (lines 1–26)1"""Transitive determinism gate — froot held to the bar it sets for others.
2 
3The kernel (``check_determinism.py``) is lexical: it flags banned calls written
4directly inside an ``@workflow.defn`` class, high-precision and install-free.
5This wrapper runs froot's *own* transitive analyzer — the same
6``analyze_workflow_surface`` the determinism-reviewer loop points at other repos
7— over froot's source, chasing first-party helpers OUT of each workflow up to a
8bounded depth. A confirmed hazard (a banned call reachable from a workflow
9through a first-party function) fails the gate.
10 
11The risky-third-party-import *frontier* (which froot adjudicates with a model for
12other repos) is printed as advisory only and never fails the build: a static gate
13must not cry wolf. Unlike the kernel this imports froot, so CI runs it in the
14synced environment (``uv run``), not the install-free lexical job.
15"""
16 
17from __future__ import annotations
18 
19import argparse
20import sys
21from pathlib import Path
22 
23from froot.adapters.source_tree import load_modules
24from froot.policy.determinism import analyze_workflow_surface
25 
26 
27def main(argv: list[str] | None = None) -> int:
28    parser = argparse.ArgumentParser(description=__doc__)
29    parser.add_argument(
30        "repo",
31        nargs="?",
32        default=".",
33        help="repo root containing src/ (default: .)",
34    )
35    parser.add_argument(
36        "--depth",
37        type=int,
38        default=2,
39        help="first-party call levels to chase out of each workflow",
40    )
41    args = parser.parse_args(argv)
42 
43    modules = load_modules(Path(args.repo))
44    result = analyze_workflow_surface(modules, max_depth=args.depth)
45 
46    for item in result.frontier:
47        print(
48            f"advisory (frontier): {item.module}:{item.line} imports "
49            f"{item.symbol} — reachability from {item.workflow} not confirmed"
50        )
51 
52    if not result.hazards:
53        print("No transitive determinism hazards reach a workflow. ✓")
54        return 0
55 
56    for hazard in result.hazards:
57        via = " -> ".join(hazard.via)
58        imp = hazard.impurity
59        print(f"{imp.module}:{imp.line}  {imp.rule}  ->  {imp.hint}")
60        print(f"    reached from {hazard.workflow} via {via}")
61    noun = "hazard" if len(result.hazards) == 1 else "hazards"
62    print(
63        f"\n{len(result.hazards)} transitive determinism {noun} "
64        "reachable from a workflow.",
65        file=sys.stderr,
66    )
67    return 1
68 
⋯ 3 lines hidden (lines 69–71)69 
70if __name__ == "__main__":
71    raise SystemExit(main())

.github/workflows/determinism.yml · 45 lines

.github/workflows/determinism.yml45 lines · YAML

⋯ 30 lines hidden (lines 1–30)1name: Determinism
2 
3on:
push:
  branches: [main]
pull_request:
7 
8# Cancel superseded runs on the same ref.
9concurrency:
group: determinism-${{ github.ref }}
cancel-in-progress: true
12 
13permissions:
contents: read
15 
16jobs:
lexical:
  name: temporal determinism (lexical kernel)
  runs-on: ubuntu-latest
  steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v5
      with:
        python-version: "3.13"
    # Stdlib-only AST check — no install needed (runs even if deps are broken).
    # Flags known-nondeterministic calls written lexically inside an
    # @workflow.defn class. High precision by design (lexical scope only); the
    # transitive job below is the second ring.
    - run: python scripts/check_determinism.py src
30 
transitive:
  name: temporal determinism (transitive)
  runs-on: ubuntu-latest
  steps:
    - uses: actions/checkout@v4
    - uses: astral-sh/setup-uv@v5
      with:
        python-version: "3.13"
        enable-cache: true
    # froot dogfooding its own transitive analyzer over its own workflows —
    # chasing first-party helpers out of each @workflow.defn, so a hazard
    # hidden behind an imported helper fails the gate too. Imports froot, so
    # it needs the synced env (unlike the install-free lexical kernel).
    - run: uv sync --extra dev --extra ai --extra github --extra otel
    - run: uv run python scripts/check_determinism_transitive.py

The verdict8🎬 Claims scored against the code

Does the loop close now?

The claim this PR makes is that the first loop now closes end-to-end: no rotting red PR, no stale superseded PR, no invisible infinite retry, no model outage that stalls the spine — and all of it expressed on the existing pure spine rather than bolted onto its side. Below, each claim is scored against the change.

Claim	Upheld by	Where
A red PR is closed, not left to rot	terminal `CIFailed` routes through `Closing → ClosePullRequest`, then records	`domain/{effects,events,state}.py`, `policy/state_machine.py`
The close is a visible effect, not a hidden mutation	modeled as data on the pure machine; `assert_never` proves exhaustiveness	`policy/state_machine.py`
Closing never drops the outcome	both paths build and record the same `LoopOutcome`	`policy/state_machine.py` `_from_closing`
Stale superseded/satisfied PRs are cleaned up	pure `reconciliations`, re-derived each scan tick	`policy/reconcile.py`, `workflow/scan_workflow.py`
Reconcile never closes a live PR by mistake	version-parse matcher; closes only on a positive match	`policy/reconcile.py`, `policy/naming.py`
The close survives a retry	idempotent PATCH + 404/422-tolerant branch delete + marker comment	`adapters/github.py`
Persistent faults surface; the model can't stall the spine	`TOOL_RETRY` on every activity; judge degrades to `UnknownVerdict`	`workflow/constants.py`, `workflow/activities.py`
The whole loop is proven to close	real-activity e2e (green / red-close / pending→pass) + transitive determinism gate	`tests/test_e2e_workflow.py`, `scripts/check_determinism_transitive.py`

Eight claims, eight pieces of code-level evidence. make check is green: ruff, strict mypy, 219 tests, coverage ≥ 72%, both determinism gates.

Three findings from the adversarial review are folded in: the UnknownVerdict docstring now names the model-failure path, and the end-to-end suite gained a pending→pass case so the durable CI wait is exercised with real activities. A third — a test that would specifically pin the retry bound — was considered and declined: Temporal's own default also retries, so such a test could not distinguish this change from the framework's behavior. From the code: the loop now closes on every terminal path, the cleanup is fail-safe, the failure modes are deliberate, and the durable spine is exactly the shape it was.