What changed, and why
froot is live now, so this batch is four correctness fixes — each one a place where the running loop was quietly doing the wrong thing — bundled because they share a theme and a discipline: every fix lands with a test that fails on the old code.
The four, each its own section below:
1. Logging that emitted nothing. The worker builds a structured loop_outcome JSON line per loop at INFO, but nothing configured logging — so the root logger's WARNING default dropped every line. kubectl logs and the ClickStack filelog were empty. 2. GitHub rate-limits killed the loop. A 403 was classified as a permanent auth fault and made non-retryable. But GitHub returns 403 for its secondary rate limit too, so a transient throttle permanently killed the call-heaviest path. 3. Pagination silently truncated. A single per_page=100 read of checks and comments stopped at item 100 — a failing 101st check read as green, and a marker past page 1 was missed, so froot posted a duplicate comment. 4. The two determinism tables could drift. froot's blocking CI gate and its advisory reviewer each hand-copy four BANNED_* tables. Nothing pinned them equal — froot could disagree with itself about what "deterministic" means.
A one-line Makefile target rounds it out. Read the four below; each is small, local, and guarded.
Fix 1 — the worker now actually emits its outcome lines
The activities emit one JSON loop_outcome line per closed loop at INFO on the froot.* loggers. That is the cheap, human-readable half of "derive, never store" — the stream the deploy points operators and the ClickStack filelog at. But a logger with no configured handler falls back to the root logger, which defaults to WARNING with no handler. So every INFO line was silently dropped.
configure_logging fixes it: set the root to INFO, attach one stdout handler that emits the record verbatim (the message is already JSON, so it stays machine-parseable), and pin the chatty temporalio SDK at WARNING so the outcome lines are not buried in poll/heartbeat noise.
The new entrypoint config: root to INFO, one verbatim stdout handler, SDK quieted.
src/froot/worker.py · 141 lines
src/froot/worker.py
⋯ 41 lines hidden (lines 1–41)
1"""The Temporal worker entrypoint — the runnable assembly.
3Connects to Temporal with the Pydantic data converter and registers the whole
4runtime: all four workflows and every activity. Run it once a Temporal server is
5reachable::
7 python -m froot.worker
9The connection is env-configured so the same image runs anywhere:
10``TEMPORAL_HOST`` (default ``localhost:7233``), ``TEMPORAL_NAMESPACE`` (default
11``default``), and ``TEMPORAL_TASK_QUEUE`` (default ``froot``). The adapters read
12their own keys (``FROOT_GITHUB_TOKEN``) and the model endpoint
13(``FROOT_OLLAMA_URL`` / ``FROOT_OLLAMA_MODEL``) from the environment.
14"""
16from __future__ import annotations
18import asyncio
19import contextlib
20import logging
21import signal
22import sys
24from temporalio.client import Client
25from temporalio.worker import Worker
27from froot.adapters.telemetry import (
28 metrics_runtime,
29 setup_tracing,
30 shutdown_tracing,
31 tracing_interceptors,
32)
33from froot.config.settings import DashboardSettings, TemporalSettings
34from froot.workflow.runtime import ALL_ACTIVITIES, DATA_CONVERTER, WORKFLOWS
36# Process one activity at a time: the model judge calls a single local Gemma
37# (which serializes anyway), and the household/hobby volume never needs more.
38# The durable CI wait sleeps between polls, so it does not hold this slot.
39_MAX_CONCURRENT_ACTIVITIES = 1
42def configure_logging() -> None:
43 """Send froot's structured outcome lines to stdout at INFO.
45 The activities emit one JSON ``loop_outcome`` line per closed loop on the
46 ``froot.*`` loggers at INFO — the cheap, human-readable half of "derive,
47 never store", and the stream the deploy points operators (and the ClickStack
48 filelog) at. A logger with no configured handler defaults to WARNING, so
49 without this those INFO lines are silently dropped. Attach a single stdout
50 handler at INFO that emits the record verbatim (the message is already JSON,
51 so it stays machine-parseable), and pin the chatty ``temporalio`` SDK at
52 WARNING so the outcome lines are not buried in poll/heartbeat noise.
54 Idempotent: a second call neither duplicates the handler nor lowers levels.
55 """
56 root = logging.getLogger()
57 root.setLevel(logging.INFO)
58 already = any(
59 isinstance(h, logging.StreamHandler) and h.stream is sys.stdout
60 for h in root.handlers
61 )
62 if not already:
63 handler = logging.StreamHandler(sys.stdout)
64 handler.setFormatter(logging.Formatter("%(message)s"))
65 root.addHandler(handler)
66 logging.getLogger("temporalio").setLevel(logging.WARNING)
⋯ 75 lines hidden (lines 67–141)
69async def run_worker(
70 *,
71 target_host: str | None = None,
72 namespace: str | None = None,
73 task_queue: str | None = None,
74) -> None:
75 """Connect to Temporal and run the worker until cancelled.
77 Each parameter falls back to its ``TEMPORAL_*`` environment variable
78 (via :class:`~froot.config.settings.TemporalSettings`), then to the default,
79 so a deployment configures the worker purely through env.
80 """
81 settings = TemporalSettings()
82 host = target_host or settings.host
83 ns = namespace or settings.namespace
84 queue = task_queue or settings.task_queue
85 setup_tracing("froot-worker")
86 client = await Client.connect(
87 host,
88 namespace=ns,
89 data_converter=DATA_CONVERTER,
90 interceptors=tracing_interceptors(),
91 runtime=metrics_runtime(),
92 )
93 worker = Worker(
94 client,
95 task_queue=queue,
96 workflows=WORKFLOWS,
97 activities=ALL_ACTIVITIES,
98 max_concurrent_activities=_MAX_CONCURRENT_ACTIVITIES,
99 )
100 # Run until SIGTERM/SIGINT, then shut down gracefully and flush telemetry
101 # (atexit does NOT run on an unhandled signal, so the last span batch would
102 # otherwise be dropped on every Recreate rollout).
103 stop = asyncio.Event()
104 loop = asyncio.get_running_loop()
105 for sig in (signal.SIGTERM, signal.SIGINT):
106 with contextlib.suppress(NotImplementedError):
107 loop.add_signal_handler(sig, stop.set)
108 # The read-model dashboard shares this process and this client (lazy import
109 # so httpx/the dashboard never load when it is switched off). It serves a
110 # derived 10,000ft view; reach it with `kubectl port-forward`. A failure to
111 # start it (e.g. the port is taken) must never take the worker down — it is
112 # a non-load-bearing debug surface, so degrade to running without it.
113 dashboard = None
114 if DashboardSettings().enabled:
115 from froot.dashboard.server import start as start_dashboard
117 try:
118 dashboard = await start_dashboard(client)
119 except Exception:
120 logging.getLogger("froot.worker").exception(
121 "dashboard failed to start; running worker without it"
122 )
123 try:
124 async with worker:
125 await stop.wait()
126 finally:
127 if dashboard is not None:
128 dashboard.close()
129 with contextlib.suppress(Exception):
130 await dashboard.wait_closed()
131 shutdown_tracing()
134def main() -> None:
135 """Console entrypoint: run the worker, configured from the environment."""
136 configure_logging()
137 asyncio.run(run_worker())
140if __name__ == "__main__":
141 main()
It is idempotent by construction. The handler is added only if no stdout StreamHandler is already attached, so a second call neither duplicates the handler nor re-lowers levels. And main() — the console entrypoint — now calls it before asyncio.run, so the configuration is in place before the first loop closes.
main() wires it in, before the event loop starts.
src/froot/worker.py · 141 lines
src/froot/worker.py
⋯ 133 lines hidden (lines 1–133)
1"""The Temporal worker entrypoint — the runnable assembly.
3Connects to Temporal with the Pydantic data converter and registers the whole
4runtime: all four workflows and every activity. Run it once a Temporal server is
5reachable::
7 python -m froot.worker
9The connection is env-configured so the same image runs anywhere:
10``TEMPORAL_HOST`` (default ``localhost:7233``), ``TEMPORAL_NAMESPACE`` (default
11``default``), and ``TEMPORAL_TASK_QUEUE`` (default ``froot``). The adapters read
12their own keys (``FROOT_GITHUB_TOKEN``) and the model endpoint
13(``FROOT_OLLAMA_URL`` / ``FROOT_OLLAMA_MODEL``) from the environment.
14"""
16from __future__ import annotations
18import asyncio
19import contextlib
20import logging
21import signal
22import sys
24from temporalio.client import Client
25from temporalio.worker import Worker
27from froot.adapters.telemetry import (
28 metrics_runtime,
29 setup_tracing,
30 shutdown_tracing,
31 tracing_interceptors,
32)
33from froot.config.settings import DashboardSettings, TemporalSettings
34from froot.workflow.runtime import ALL_ACTIVITIES, DATA_CONVERTER, WORKFLOWS
36# Process one activity at a time: the model judge calls a single local Gemma
37# (which serializes anyway), and the household/hobby volume never needs more.
38# The durable CI wait sleeps between polls, so it does not hold this slot.
39_MAX_CONCURRENT_ACTIVITIES = 1
42def configure_logging() -> None:
43 """Send froot's structured outcome lines to stdout at INFO.
45 The activities emit one JSON ``loop_outcome`` line per closed loop on the
46 ``froot.*`` loggers at INFO — the cheap, human-readable half of "derive,
47 never store", and the stream the deploy points operators (and the ClickStack
48 filelog) at. A logger with no configured handler defaults to WARNING, so
49 without this those INFO lines are silently dropped. Attach a single stdout
50 handler at INFO that emits the record verbatim (the message is already JSON,
51 so it stays machine-parseable), and pin the chatty ``temporalio`` SDK at
52 WARNING so the outcome lines are not buried in poll/heartbeat noise.
54 Idempotent: a second call neither duplicates the handler nor lowers levels.
55 """
56 root = logging.getLogger()
57 root.setLevel(logging.INFO)
58 already = any(
59 isinstance(h, logging.StreamHandler) and h.stream is sys.stdout
60 for h in root.handlers
61 )
62 if not already:
63 handler = logging.StreamHandler(sys.stdout)
64 handler.setFormatter(logging.Formatter("%(message)s"))
65 root.addHandler(handler)
66 logging.getLogger("temporalio").setLevel(logging.WARNING)
69async def run_worker(
70 *,
71 target_host: str | None = None,
72 namespace: str | None = None,
73 task_queue: str | None = None,
74) -> None:
75 """Connect to Temporal and run the worker until cancelled.
77 Each parameter falls back to its ``TEMPORAL_*`` environment variable
78 (via :class:`~froot.config.settings.TemporalSettings`), then to the default,
79 so a deployment configures the worker purely through env.
80 """
81 settings = TemporalSettings()
82 host = target_host or settings.host
83 ns = namespace or settings.namespace
84 queue = task_queue or settings.task_queue
85 setup_tracing("froot-worker")
86 client = await Client.connect(
87 host,
88 namespace=ns,
89 data_converter=DATA_CONVERTER,
90 interceptors=tracing_interceptors(),
91 runtime=metrics_runtime(),
92 )
93 worker = Worker(
94 client,
95 task_queue=queue,
96 workflows=WORKFLOWS,
97 activities=ALL_ACTIVITIES,
98 max_concurrent_activities=_MAX_CONCURRENT_ACTIVITIES,
99 )
100 # Run until SIGTERM/SIGINT, then shut down gracefully and flush telemetry
101 # (atexit does NOT run on an unhandled signal, so the last span batch would
102 # otherwise be dropped on every Recreate rollout).
103 stop = asyncio.Event()
104 loop = asyncio.get_running_loop()
105 for sig in (signal.SIGTERM, signal.SIGINT):
106 with contextlib.suppress(NotImplementedError):
107 loop.add_signal_handler(sig, stop.set)
108 # The read-model dashboard shares this process and this client (lazy import
109 # so httpx/the dashboard never load when it is switched off). It serves a
110 # derived 10,000ft view; reach it with `kubectl port-forward`. A failure to
111 # start it (e.g. the port is taken) must never take the worker down — it is
112 # a non-load-bearing debug surface, so degrade to running without it.
113 dashboard = None
114 if DashboardSettings().enabled:
115 from froot.dashboard.server import start as start_dashboard
117 try:
118 dashboard = await start_dashboard(client)
119 except Exception:
120 logging.getLogger("froot.worker").exception(
121 "dashboard failed to start; running worker without it"
122 )
123 try:
124 async with worker:
125 await stop.wait()
126 finally:
127 if dashboard is not None:
128 dashboard.close()
129 with contextlib.suppress(Exception):
130 await dashboard.wait_closed()
131 shutdown_tracing()
134def main() -> None:
135 """Console entrypoint: run the worker, configured from the environment."""
136 configure_logging()
137 asyncio.run(run_worker())
The guard: no config → INFO line dropped.
tests/test_worker.py · 85 lines
tests/test_worker.py
⋯ 65 lines hidden (lines 1–65)
1"""The worker entrypoint's process-wide logging configuration.
3These pin the one behaviour the deploy depends on: after ``configure_logging``
4the structured ``loop_outcome`` lines the activities emit at INFO actually reach
5stdout (so the ClickStack filelog and ``kubectl logs`` are not empty), while the
6chatty Temporal SDK is held at WARNING so they are not buried.
7"""
9from __future__ import annotations
11import logging
12import sys
13from typing import TYPE_CHECKING
15import pytest
17from froot.worker import configure_logging
19if TYPE_CHECKING:
20 from collections.abc import Iterator
23@pytest.fixture(autouse=True)
24def _restore_logging() -> Iterator[None]:
25 """Save and restore global logger state so these tests don't leak."""
26 root = logging.getLogger()
27 saved_handlers = root.handlers[:]
28 saved_root_level = root.level
29 saved_temporal_level = logging.getLogger("temporalio").level
30 # Start from a clean slate so a handler we add binds to the test's stdout.
31 root.handlers.clear()
32 try:
33 yield
34 finally:
35 root.handlers[:] = saved_handlers
36 root.setLevel(saved_root_level)
37 logging.getLogger("temporalio").setLevel(saved_temporal_level)
40def test_configure_logging_sets_info_stdout_handler_and_quiets_temporalio() -> (
41 None
42):
43 configure_logging()
45 root = logging.getLogger()
46 assert root.level == logging.INFO
47 assert any(
48 isinstance(h, logging.StreamHandler) and h.stream is sys.stdout
49 for h in root.handlers
50 )
51 assert logging.getLogger("temporalio").level == logging.WARNING
54def test_outcome_line_reaches_stdout_verbatim(
55 capsys: pytest.CaptureFixture[str],
56) -> None:
57 configure_logging()
59 # Mirrors what activities.record_outcome emits: a JSON line at INFO.
60 logging.getLogger("froot.outcome").info('{"event": "loop_outcome", "n": 1}')
62 out = capsys.readouterr().out
63 assert '{"event": "loop_outcome", "n": 1}' in out
66def test_info_is_dropped_without_configuration(
67 capsys: pytest.CaptureFixture[str],
68) -> None:
69 # The bug this guards against: an unconfigured logger defaults to WARNING,
70 # so the INFO outcome line is silently dropped. (No configure_logging call.)
71 logging.getLogger("froot.outcome").info('{"event": "loop_outcome"}')
73 assert "loop_outcome" not in capsys.readouterr().out
⋯ 12 lines hidden (lines 74–85)
Fix 2 — GitHub rate limits are retryable, not fatal
The old _raise_for_status had one rule for 401 and 403: permanent auth fault, non_retryable=True. That is right for 401 and for a real permission 403. It is wrong for the other thing GitHub returns 403 for — its secondary rate limit. Marking that non-retryable means a transient throttle permanently kills the activity, on the path that calls GitHub the most (the durable CI poll).
The fix splits the classification. 401 stays a hard fault. A 403 or 429 that carries a rate-limit marker becomes a retryable ApplicationError that honors the server's advised wait, so Temporal backs off instead of giving up. A 403 without those markers is still a real permission fault and stays non-retryable.
The new three-way split: hard 401, retryable rate-limit, hard 403.
src/froot/adapters/github.py · 461 lines
src/froot/adapters/github.py
⋯ 184 lines hidden (lines 1–184)
1"""The GitHub forge adapter: git (checkout/push) + the GitHub REST API.
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.
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"""
14from __future__ import annotations
16import time
17from dataclasses import dataclass
18from datetime import timedelta
19from typing import TYPE_CHECKING, Any, final
21import httpx
22from temporalio.exceptions import ApplicationError
24from froot.config.settings import GitHubSettings
25from froot.domain.ci import (
26 CIAbsent,
27 CIFailed,
28 CIPassed,
29 CIPending,
30 CIStatus,
31)
32from froot.domain.pull_request import BranchName, PullRequestRef
34if TYPE_CHECKING:
35 from collections.abc import AsyncIterator
36 from pathlib import Path
38 from froot.domain.pull_request import PullRequestDraft
39 from froot.domain.repo import TargetRepo
41from froot.adapters._proc import run_text
43_API = "https://api.github.com"
44_API_VERSION = "2022-11-28"
45_TIMEOUT = 30.0
46_COMMITTER_NAME = "froot"
47_COMMITTER_EMAIL = "froot@users.noreply.github.com"
49# Check-run conclusions that mean the change is not safe to merge.
50_BAD_CONCLUSIONS = frozenset(
51 {
52 "failure",
53 "timed_out",
54 "cancelled",
55 "action_required",
56 "startup_failure",
57 "stale",
58 }
59)
62@final
63@dataclass(frozen=True, slots=True)
64class CheckRow:
65 """One GitHub check run, reduced to what the CI mapping needs."""
67 name: str
68 status: str
69 conclusion: str | None
72def ci_status_from_checks(
73 checks: tuple[CheckRow, ...], combined_state: str | None
74) -> CIStatus:
75 """Map GitHub check rows + the combined commit status to a CI status.
77 Args:
78 checks: The commit's check runs (GitHub Checks API).
79 combined_state: The legacy combined commit status (``success`` /
80 ``failure`` / ``pending``), or ``None`` when no statuses exist.
82 Returns:
83 ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
84 unresolved, ``CIFailed`` (with the failing check names) on any bad
85 conclusion or a failed combined status, else ``CIPassed``.
86 """
87 if not checks and combined_state is None:
88 return CIAbsent()
89 unresolved = combined_state == "pending" or any(
90 row.status != "completed" for row in checks
91 )
92 if unresolved:
93 return CIPending()
94 failing = tuple(
95 row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
96 )
97 if failing or combined_state == "failure":
98 return CIFailed(failing=failing)
99 return CIPassed()
102def _token() -> str:
103 token = GitHubSettings().github_token
104 if token is None:
105 # A missing token is a permanent misconfiguration, not a transient
106 # fault — fail the activity fast instead of retrying forever.
107 raise ApplicationError(
108 "FROOT_GITHUB_TOKEN is required", non_retryable=True
109 )
110 return token.get_secret_value()
113def _auth_remote(target: TargetRepo) -> str:
114 return (
115 f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
116 )
119def _client() -> httpx.AsyncClient:
120 return httpx.AsyncClient(
121 base_url=_API,
122 timeout=_TIMEOUT,
123 headers={
124 "Authorization": f"Bearer {_token()}",
125 "Accept": "application/vnd.github+json",
126 "X-GitHub-Api-Version": _API_VERSION,
127 },
128 )
131def _retry_after_seconds(response: httpx.Response) -> float | None:
132 """The server-advised wait before retrying a rate-limited request, if any.
134 Honors the secondary-limit ``Retry-After`` header (delta-seconds form), then
135 the primary-limit ``X-RateLimit-Reset`` (epoch) when the remaining quota is
136 zero. Returns ``None`` when GitHub gives no usable hint (let Temporal's own
137 backoff decide).
138 """
139 retry_after = response.headers.get("retry-after")
140 if retry_after is not None:
141 try:
142 return max(0.0, float(retry_after))
143 except ValueError:
144 return None # HTTP-date form is rare from GitHub; fall through.
145 if response.headers.get("x-ratelimit-remaining") == "0":
146 reset = response.headers.get("x-ratelimit-reset")
147 if reset is not None:
148 try:
149 return max(0.0, float(reset) - time.time())
150 except ValueError:
151 return None
152 return None
155def _is_rate_limited(response: httpx.Response) -> bool:
156 """Whether a 403/429 is GitHub *rate limiting* (transient), not auth.
158 GitHub overloads 403 for both a permanent permission fault AND its secondary
159 rate limit, so the two are told apart by the rate-limit markers: an explicit
160 ``Retry-After``, an exhausted ``X-RateLimit-Remaining``, or a "rate limit"
161 note in the body (the secondary-limit form often only says so there).
162 """
163 if response.status_code == 429:
164 return True
165 if response.status_code != 403:
166 return False
167 if "retry-after" in response.headers:
168 return True
169 if response.headers.get("x-ratelimit-remaining") == "0":
170 return True
171 return "rate limit" in response.text.lower()
174def _raise_for_status(response: httpx.Response) -> None:
175 """Raise on error, classifying retryability the way the loop needs.
177 A 401 is always a permanent auth fault. A 403/429 doubles as GitHub's
178 rate-limit signal: when it carries a rate-limit marker the fault is
179 TRANSIENT, so raise a *retryable* error (honoring the server's advised wait)
180 and let Temporal back off rather than killing the loop — this is the
181 call-heaviest path (the durable CI poll), so a transient limit must not be
182 fatal. A 403 without those markers is a real permission fault and stays
183 non-retryable.
184 """
185 if response.status_code == 401:
186 raise ApplicationError("GitHub auth failed (401)", non_retryable=True)
187 if response.status_code in (403, 429) and _is_rate_limited(response):
188 delay = _retry_after_seconds(response)
189 raise ApplicationError(
190 f"GitHub rate-limited ({response.status_code}); backing off",
191 type="GitHubRateLimited",
192 next_retry_delay=(
193 timedelta(seconds=delay) if delay is not None else None
194 ),
195 )
196 if response.status_code == 403:
197 raise ApplicationError("GitHub auth failed (403)", non_retryable=True)
198 response.raise_for_status()
⋯ 263 lines hidden (lines 199–461)
201def _next_link(response: httpx.Response) -> str | None:
202 """The ``rel="next"`` page URL from the Link header, or None at the end."""
203 nxt = response.links.get("next")
204 return nxt.get("url") if nxt else None
207async def _iter_pages(
208 client: httpx.AsyncClient,
209 url: str,
210 params: dict[str, Any] | None = None,
211) -> AsyncIterator[httpx.Response]:
212 """Yield each page of a GitHub list endpoint, following Link ``rel=next``.
214 GitHub caps a page at 100 items; a single ``per_page=100`` read silently
215 truncates a busy repo — which here would *misread the CI oracle* (a failing
216 101st check goes unseen) or *break comment idempotency* (froot's marker on a
217 later page is missed, so it posts a duplicate). Following the Link header
218 reads the whole set; the data, not a fixed cap, bounds the loop.
219 """
220 next_url: str | None = url
221 # First page carries the caller's filters + per_page; later pages use the
222 # absolute Link URL verbatim (it already encodes them), so params drop off.
223 next_params: dict[str, Any] | None = {**(params or {}), "per_page": 100}
224 while next_url is not None:
225 resp = await client.get(next_url, params=next_params)
226 _raise_for_status(resp)
227 yield resp
228 next_url = _next_link(resp)
229 next_params = None
232async def _marked_comment_id(
233 client: httpx.AsyncClient, slug: str, number: int, marker: str
234) -> int | None:
235 """The id of froot's ``marker``-tagged comment on a PR, across all pages.
237 Searches every comment page (stopping at the first hit) so a chatty PR that
238 pushes the marker past the first 100 comments cannot fool the upsert into
239 posting a duplicate.
240 """
241 async for page in _iter_pages(
242 client, f"/repos/{slug}/issues/{number}/comments"
243 ):
244 for comment in page.json():
245 if isinstance(comment, dict) and marker in str(
246 comment.get("body", "")
247 ):
248 return int(comment["id"])
249 return None
252def _pull_request_ref(payload: Any) -> PullRequestRef:
253 """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
254 head = payload["head"]
255 return PullRequestRef(
256 number=int(payload["number"]),
257 url=str(payload["html_url"]),
258 branch=BranchName(value=str(head["ref"])),
259 head_sha=str(head["sha"]),
260 )
263@final
264class GitHubForge:
265 """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
267 async def checkout(self, target: TargetRepo, workspace: Path) -> None:
268 """Shallow-clone the repo's default branch into ``workspace``."""
269 code, out, err = await run_text(
270 "git",
271 "clone",
272 "--depth",
273 "1",
274 "--branch",
275 target.default_branch,
276 _auth_remote(target),
277 ".",
278 cwd=workspace,
279 )
280 if code != 0:
281 raise RuntimeError(f"git clone failed ({code}): {err or out}")
283 async def checkout_pull_request(
284 self, target: TargetRepo, workspace: Path, number: int
285 ) -> None:
286 """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
288 Fetches the PR head ref the base repo exposes for every PR (fork or
289 not), so a community PR from a fork checks out the same way a same-repo
290 one does — no fork URL, no cross-repo auth.
291 """
292 ref = f"pull/{number}/head"
293 steps: tuple[tuple[str, ...], ...] = (
294 ("git", "init", "-q"),
295 ("git", "remote", "add", "origin", _auth_remote(target)),
296 ("git", "fetch", "--depth", "1", "origin", ref),
297 ("git", "checkout", "-q", "FETCH_HEAD"),
298 )
299 for step in steps:
300 code, out, err = await run_text(*step, cwd=workspace)
301 if code != 0:
302 raise RuntimeError(
303 f"git {step[1]} ({ref}) failed ({code}): {err or out}"
304 )
306 async def push_branch(
307 self, workspace: Path, branch: BranchName, commit_message: str
308 ) -> str:
309 """Commit the workspace changes onto ``branch`` and push; return SHA."""
310 steps: tuple[tuple[str, ...], ...] = (
311 ("git", "checkout", "-b", branch.value),
312 ("git", "add", "-A"),
313 (
314 "git",
315 "-c",
316 f"user.name={_COMMITTER_NAME}",
317 "-c",
318 f"user.email={_COMMITTER_EMAIL}",
319 "commit",
320 "-m",
321 commit_message,
322 ),
323 ("git", "push", "-u", "origin", branch.value),
324 )
325 for step in steps:
326 code, out, err = await run_text(*step, cwd=workspace)
327 if code != 0:
328 raise RuntimeError(f"{step[0:2]} failed ({code}): {err or out}")
329 code, sha, _ = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
330 if code != 0:
331 raise RuntimeError(f"git rev-parse failed ({code})")
332 return sha.strip()
334 async def find_open_pull_request(
335 self, target: TargetRepo, branch: BranchName
336 ) -> PullRequestRef | None:
337 """Return the open PR for ``branch`` if one already exists (dedup)."""
338 async with _client() as client:
339 resp = await client.get(
340 f"/repos/{target.repo.slug}/pulls",
341 params={
342 "head": f"{target.repo.owner}:{branch.value}",
343 "state": "open",
344 },
345 )
346 _raise_for_status(resp)
347 payloads = resp.json()
348 if isinstance(payloads, list) and payloads:
349 return _pull_request_ref(payloads[0])
350 return None
352 async def list_open_pull_requests(
353 self, target: TargetRepo
354 ) -> tuple[PullRequestRef, ...]:
355 """List the repo's open PRs (the determinism reviewer's work feed)."""
356 refs: list[PullRequestRef] = []
357 async with _client() as client:
358 async for page in _iter_pages(
359 client,
360 f"/repos/{target.repo.slug}/pulls",
361 {"state": "open"},
362 ):
363 payload = page.json()
364 if isinstance(payload, list):
365 refs.extend(_pull_request_ref(p) for p in payload)
366 return tuple(refs)
368 async def open_pull_request(
369 self, target: TargetRepo, draft: PullRequestDraft
370 ) -> PullRequestRef:
371 """Open the PR for an already-pushed branch (idempotent on conflict)."""
372 async with _client() as client:
373 resp = await client.post(
374 f"/repos/{target.repo.slug}/pulls",
375 json={
376 "title": draft.title,
377 "head": draft.branch.value,
378 "base": draft.base,
379 "body": draft.body,
380 },
381 )
382 if resp.status_code == 422:
383 existing = await self.find_open_pull_request(target, draft.branch)
384 if existing is not None:
385 return existing
386 _raise_for_status(resp)
387 return _pull_request_ref(resp.json())
389 async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
390 """Read the repo's combined CI status for a commit (the oracle).
392 The check-runs are paginated (a commit can have >100 checks, and missing
393 a failing one would falsely read green); the legacy combined status is a
394 single rolled-up ``state`` over all statuses, so it needs no paging.
395 """
396 slug = target.repo.slug
397 rows: list[CheckRow] = []
398 async with _client() as client:
399 async for page in _iter_pages(
400 client, f"/repos/{slug}/commits/{head_sha}/check-runs"
401 ):
402 rows.extend(
403 CheckRow(
404 name=str(run["name"]),
405 status=str(run["status"]),
406 conclusion=(
407 str(run["conclusion"])
408 if run.get("conclusion") is not None
409 else None
410 ),
411 )
412 for run in page.json().get("check_runs", [])
413 )
414 status_resp = await client.get(
415 f"/repos/{slug}/commits/{head_sha}/status"
416 )
417 _raise_for_status(status_resp)
418 status_json = status_resp.json()
419 combined = (
420 str(status_json["state"])
421 if int(status_json.get("total_count", 0)) > 0
422 else None
423 )
424 return ci_status_from_checks(tuple(rows), combined)
426 async def add_labels(
427 self, target: TargetRepo, number: int, labels: tuple[str, ...]
428 ) -> None:
429 """Attach labels to a PR (the human-readable signal-update)."""
430 async with _client() as client:
431 resp = await client.post(
432 f"/repos/{target.repo.slug}/issues/{number}/labels",
433 json={"labels": list(labels)},
434 )
435 _raise_for_status(resp)
437 async def upsert_issue_comment(
438 self, target: TargetRepo, number: int, marker: str, body: str
439 ) -> str:
440 """Create or update the PR's ``marker``-tagged comment; return its URL.
442 A PR conversation (issue) comment, not an inline review comment: the
443 determinism findings are structural (a call path), so a single advisory
444 summary fits better than line anchors. The marker lets a re-review edit
445 its own prior comment in place rather than stack a new one.
446 """
447 slug = target.repo.slug
448 async with _client() as client:
449 existing_id = await _marked_comment_id(client, slug, number, marker)
450 if existing_id is not None:
451 resp = await client.patch(
452 f"/repos/{slug}/issues/comments/{existing_id}",
453 json={"body": body},
454 )
455 else:
456 resp = await client.post(
457 f"/repos/{slug}/issues/{number}/comments",
458 json={"body": body},
459 )
460 _raise_for_status(resp)
461 return str(resp.json()["html_url"])
_is_rate_limited is the discriminator: a 429 is always a limit; a 403 is one only if it carries a Retry-After header, an exhausted X-RateLimit-Remaining, or a "rate limit" note in the body (the secondary-limit form often only says so there). _retry_after_seconds then turns the server's hint into a concrete delay — the Retry-After delta-seconds, else the X-RateLimit-Reset epoch minus now — and returns None when there is no usable hint, deferring to Temporal's own backoff.
_is_rate_limited: the markers that tell a throttle from a permission fault.
src/froot/adapters/github.py · 461 lines
src/froot/adapters/github.py
⋯ 154 lines hidden (lines 1–154)
1"""The GitHub forge adapter: git (checkout/push) + the GitHub REST API.
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.
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"""
14from __future__ import annotations
16import time
17from dataclasses import dataclass
18from datetime import timedelta
19from typing import TYPE_CHECKING, Any, final
21import httpx
22from temporalio.exceptions import ApplicationError
24from froot.config.settings import GitHubSettings
25from froot.domain.ci import (
26 CIAbsent,
27 CIFailed,
28 CIPassed,
29 CIPending,
30 CIStatus,
31)
32from froot.domain.pull_request import BranchName, PullRequestRef
34if TYPE_CHECKING:
35 from collections.abc import AsyncIterator
36 from pathlib import Path
38 from froot.domain.pull_request import PullRequestDraft
39 from froot.domain.repo import TargetRepo
41from froot.adapters._proc import run_text
43_API = "https://api.github.com"
44_API_VERSION = "2022-11-28"
45_TIMEOUT = 30.0
46_COMMITTER_NAME = "froot"
47_COMMITTER_EMAIL = "froot@users.noreply.github.com"
49# Check-run conclusions that mean the change is not safe to merge.
50_BAD_CONCLUSIONS = frozenset(
51 {
52 "failure",
53 "timed_out",
54 "cancelled",
55 "action_required",
56 "startup_failure",
57 "stale",
58 }
59)
62@final
63@dataclass(frozen=True, slots=True)
64class CheckRow:
65 """One GitHub check run, reduced to what the CI mapping needs."""
67 name: str
68 status: str
69 conclusion: str | None
72def ci_status_from_checks(
73 checks: tuple[CheckRow, ...], combined_state: str | None
74) -> CIStatus:
75 """Map GitHub check rows + the combined commit status to a CI status.
77 Args:
78 checks: The commit's check runs (GitHub Checks API).
79 combined_state: The legacy combined commit status (``success`` /
80 ``failure`` / ``pending``), or ``None`` when no statuses exist.
82 Returns:
83 ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
84 unresolved, ``CIFailed`` (with the failing check names) on any bad
85 conclusion or a failed combined status, else ``CIPassed``.
86 """
87 if not checks and combined_state is None:
88 return CIAbsent()
89 unresolved = combined_state == "pending" or any(
90 row.status != "completed" for row in checks
91 )
92 if unresolved:
93 return CIPending()
94 failing = tuple(
95 row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
96 )
97 if failing or combined_state == "failure":
98 return CIFailed(failing=failing)
99 return CIPassed()
102def _token() -> str:
103 token = GitHubSettings().github_token
104 if token is None:
105 # A missing token is a permanent misconfiguration, not a transient
106 # fault — fail the activity fast instead of retrying forever.
107 raise ApplicationError(
108 "FROOT_GITHUB_TOKEN is required", non_retryable=True
109 )
110 return token.get_secret_value()
113def _auth_remote(target: TargetRepo) -> str:
114 return (
115 f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
116 )
119def _client() -> httpx.AsyncClient:
120 return httpx.AsyncClient(
121 base_url=_API,
122 timeout=_TIMEOUT,
123 headers={
124 "Authorization": f"Bearer {_token()}",
125 "Accept": "application/vnd.github+json",
126 "X-GitHub-Api-Version": _API_VERSION,
127 },
128 )
131def _retry_after_seconds(response: httpx.Response) -> float | None:
132 """The server-advised wait before retrying a rate-limited request, if any.
134 Honors the secondary-limit ``Retry-After`` header (delta-seconds form), then
135 the primary-limit ``X-RateLimit-Reset`` (epoch) when the remaining quota is
136 zero. Returns ``None`` when GitHub gives no usable hint (let Temporal's own
137 backoff decide).
138 """
139 retry_after = response.headers.get("retry-after")
140 if retry_after is not None:
141 try:
142 return max(0.0, float(retry_after))
143 except ValueError:
144 return None # HTTP-date form is rare from GitHub; fall through.
145 if response.headers.get("x-ratelimit-remaining") == "0":
146 reset = response.headers.get("x-ratelimit-reset")
147 if reset is not None:
148 try:
149 return max(0.0, float(reset) - time.time())
150 except ValueError:
151 return None
152 return None
155def _is_rate_limited(response: httpx.Response) -> bool:
156 """Whether a 403/429 is GitHub *rate limiting* (transient), not auth.
158 GitHub overloads 403 for both a permanent permission fault AND its secondary
159 rate limit, so the two are told apart by the rate-limit markers: an explicit
160 ``Retry-After``, an exhausted ``X-RateLimit-Remaining``, or a "rate limit"
161 note in the body (the secondary-limit form often only says so there).
162 """
163 if response.status_code == 429:
164 return True
165 if response.status_code != 403:
166 return False
167 if "retry-after" in response.headers:
168 return True
169 if response.headers.get("x-ratelimit-remaining") == "0":
170 return True
171 return "rate limit" in response.text.lower()
⋯ 290 lines hidden (lines 172–461)
174def _raise_for_status(response: httpx.Response) -> None:
175 """Raise on error, classifying retryability the way the loop needs.
177 A 401 is always a permanent auth fault. A 403/429 doubles as GitHub's
178 rate-limit signal: when it carries a rate-limit marker the fault is
179 TRANSIENT, so raise a *retryable* error (honoring the server's advised wait)
180 and let Temporal back off rather than killing the loop — this is the
181 call-heaviest path (the durable CI poll), so a transient limit must not be
182 fatal. A 403 without those markers is a real permission fault and stays
183 non-retryable.
184 """
185 if response.status_code == 401:
186 raise ApplicationError("GitHub auth failed (401)", non_retryable=True)
187 if response.status_code in (403, 429) and _is_rate_limited(response):
188 delay = _retry_after_seconds(response)
189 raise ApplicationError(
190 f"GitHub rate-limited ({response.status_code}); backing off",
191 type="GitHubRateLimited",
192 next_retry_delay=(
193 timedelta(seconds=delay) if delay is not None else None
194 ),
195 )
196 if response.status_code == 403:
197 raise ApplicationError("GitHub auth failed (403)", non_retryable=True)
198 response.raise_for_status()
201def _next_link(response: httpx.Response) -> str | None:
202 """The ``rel="next"`` page URL from the Link header, or None at the end."""
203 nxt = response.links.get("next")
204 return nxt.get("url") if nxt else None
207async def _iter_pages(
208 client: httpx.AsyncClient,
209 url: str,
210 params: dict[str, Any] | None = None,
211) -> AsyncIterator[httpx.Response]:
212 """Yield each page of a GitHub list endpoint, following Link ``rel=next``.
214 GitHub caps a page at 100 items; a single ``per_page=100`` read silently
215 truncates a busy repo — which here would *misread the CI oracle* (a failing
216 101st check goes unseen) or *break comment idempotency* (froot's marker on a
217 later page is missed, so it posts a duplicate). Following the Link header
218 reads the whole set; the data, not a fixed cap, bounds the loop.
219 """
220 next_url: str | None = url
221 # First page carries the caller's filters + per_page; later pages use the
222 # absolute Link URL verbatim (it already encodes them), so params drop off.
223 next_params: dict[str, Any] | None = {**(params or {}), "per_page": 100}
224 while next_url is not None:
225 resp = await client.get(next_url, params=next_params)
226 _raise_for_status(resp)
227 yield resp
228 next_url = _next_link(resp)
229 next_params = None
232async def _marked_comment_id(
233 client: httpx.AsyncClient, slug: str, number: int, marker: str
234) -> int | None:
235 """The id of froot's ``marker``-tagged comment on a PR, across all pages.
237 Searches every comment page (stopping at the first hit) so a chatty PR that
238 pushes the marker past the first 100 comments cannot fool the upsert into
239 posting a duplicate.
240 """
241 async for page in _iter_pages(
242 client, f"/repos/{slug}/issues/{number}/comments"
243 ):
244 for comment in page.json():
245 if isinstance(comment, dict) and marker in str(
246 comment.get("body", "")
247 ):
248 return int(comment["id"])
249 return None
252def _pull_request_ref(payload: Any) -> PullRequestRef:
253 """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
254 head = payload["head"]
255 return PullRequestRef(
256 number=int(payload["number"]),
257 url=str(payload["html_url"]),
258 branch=BranchName(value=str(head["ref"])),
259 head_sha=str(head["sha"]),
260 )
263@final
264class GitHubForge:
265 """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
267 async def checkout(self, target: TargetRepo, workspace: Path) -> None:
268 """Shallow-clone the repo's default branch into ``workspace``."""
269 code, out, err = await run_text(
270 "git",
271 "clone",
272 "--depth",
273 "1",
274 "--branch",
275 target.default_branch,
276 _auth_remote(target),
277 ".",
278 cwd=workspace,
279 )
280 if code != 0:
281 raise RuntimeError(f"git clone failed ({code}): {err or out}")
283 async def checkout_pull_request(
284 self, target: TargetRepo, workspace: Path, number: int
285 ) -> None:
286 """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
288 Fetches the PR head ref the base repo exposes for every PR (fork or
289 not), so a community PR from a fork checks out the same way a same-repo
290 one does — no fork URL, no cross-repo auth.
291 """
292 ref = f"pull/{number}/head"
293 steps: tuple[tuple[str, ...], ...] = (
294 ("git", "init", "-q"),
295 ("git", "remote", "add", "origin", _auth_remote(target)),
296 ("git", "fetch", "--depth", "1", "origin", ref),
297 ("git", "checkout", "-q", "FETCH_HEAD"),
298 )
299 for step in steps:
300 code, out, err = await run_text(*step, cwd=workspace)
301 if code != 0:
302 raise RuntimeError(
303 f"git {step[1]} ({ref}) failed ({code}): {err or out}"
304 )
306 async def push_branch(
307 self, workspace: Path, branch: BranchName, commit_message: str
308 ) -> str:
309 """Commit the workspace changes onto ``branch`` and push; return SHA."""
310 steps: tuple[tuple[str, ...], ...] = (
311 ("git", "checkout", "-b", branch.value),
312 ("git", "add", "-A"),
313 (
314 "git",
315 "-c",
316 f"user.name={_COMMITTER_NAME}",
317 "-c",
318 f"user.email={_COMMITTER_EMAIL}",
319 "commit",
320 "-m",
321 commit_message,
322 ),
323 ("git", "push", "-u", "origin", branch.value),
324 )
325 for step in steps:
326 code, out, err = await run_text(*step, cwd=workspace)
327 if code != 0:
328 raise RuntimeError(f"{step[0:2]} failed ({code}): {err or out}")
329 code, sha, _ = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
330 if code != 0:
331 raise RuntimeError(f"git rev-parse failed ({code})")
332 return sha.strip()
334 async def find_open_pull_request(
335 self, target: TargetRepo, branch: BranchName
336 ) -> PullRequestRef | None:
337 """Return the open PR for ``branch`` if one already exists (dedup)."""
338 async with _client() as client:
339 resp = await client.get(
340 f"/repos/{target.repo.slug}/pulls",
341 params={
342 "head": f"{target.repo.owner}:{branch.value}",
343 "state": "open",
344 },
345 )
346 _raise_for_status(resp)
347 payloads = resp.json()
348 if isinstance(payloads, list) and payloads:
349 return _pull_request_ref(payloads[0])
350 return None
352 async def list_open_pull_requests(
353 self, target: TargetRepo
354 ) -> tuple[PullRequestRef, ...]:
355 """List the repo's open PRs (the determinism reviewer's work feed)."""
356 refs: list[PullRequestRef] = []
357 async with _client() as client:
358 async for page in _iter_pages(
359 client,
360 f"/repos/{target.repo.slug}/pulls",
361 {"state": "open"},
362 ):
363 payload = page.json()
364 if isinstance(payload, list):
365 refs.extend(_pull_request_ref(p) for p in payload)
366 return tuple(refs)
368 async def open_pull_request(
369 self, target: TargetRepo, draft: PullRequestDraft
370 ) -> PullRequestRef:
371 """Open the PR for an already-pushed branch (idempotent on conflict)."""
372 async with _client() as client:
373 resp = await client.post(
374 f"/repos/{target.repo.slug}/pulls",
375 json={
376 "title": draft.title,
377 "head": draft.branch.value,
378 "base": draft.base,
379 "body": draft.body,
380 },
381 )
382 if resp.status_code == 422:
383 existing = await self.find_open_pull_request(target, draft.branch)
384 if existing is not None:
385 return existing
386 _raise_for_status(resp)
387 return _pull_request_ref(resp.json())
389 async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
390 """Read the repo's combined CI status for a commit (the oracle).
392 The check-runs are paginated (a commit can have >100 checks, and missing
393 a failing one would falsely read green); the legacy combined status is a
394 single rolled-up ``state`` over all statuses, so it needs no paging.
395 """
396 slug = target.repo.slug
397 rows: list[CheckRow] = []
398 async with _client() as client:
399 async for page in _iter_pages(
400 client, f"/repos/{slug}/commits/{head_sha}/check-runs"
401 ):
402 rows.extend(
403 CheckRow(
404 name=str(run["name"]),
405 status=str(run["status"]),
406 conclusion=(
407 str(run["conclusion"])
408 if run.get("conclusion") is not None
409 else None
410 ),
411 )
412 for run in page.json().get("check_runs", [])
413 )
414 status_resp = await client.get(
415 f"/repos/{slug}/commits/{head_sha}/status"
416 )
417 _raise_for_status(status_resp)
418 status_json = status_resp.json()
419 combined = (
420 str(status_json["state"])
421 if int(status_json.get("total_count", 0)) > 0
422 else None
423 )
424 return ci_status_from_checks(tuple(rows), combined)
426 async def add_labels(
427 self, target: TargetRepo, number: int, labels: tuple[str, ...]
428 ) -> None:
429 """Attach labels to a PR (the human-readable signal-update)."""
430 async with _client() as client:
431 resp = await client.post(
432 f"/repos/{target.repo.slug}/issues/{number}/labels",
433 json={"labels": list(labels)},
434 )
435 _raise_for_status(resp)
437 async def upsert_issue_comment(
438 self, target: TargetRepo, number: int, marker: str, body: str
439 ) -> str:
440 """Create or update the PR's ``marker``-tagged comment; return its URL.
442 A PR conversation (issue) comment, not an inline review comment: the
443 determinism findings are structural (a call path), so a single advisory
444 summary fits better than line anchors. The marker lets a re-review edit
445 its own prior comment in place rather than stack a new one.
446 """
447 slug = target.repo.slug
448 async with _client() as client:
449 existing_id = await _marked_comment_id(client, slug, number, marker)
450 if existing_id is not None:
451 resp = await client.patch(
452 f"/repos/{slug}/issues/comments/{existing_id}",
453 json={"body": body},
454 )
455 else:
456 resp = await client.post(
457 f"/repos/{slug}/issues/{number}/comments",
458 json={"body": body},
459 )
460 _raise_for_status(resp)
461 return str(resp.json()["html_url"])
Fix 3 — pagination, so the CI oracle and comment upsert see everything
GitHub caps a list page at 100 items. The old code read exactly one page (per_page=100) of check-runs, open PRs, and PR comments. On a busy repo that silently truncates, and here truncation is not cosmetic — it corrupts two decisions froot makes.
The new _iter_pages follows the Link: rel="next" header to read the whole set. The first request carries the caller's filters plus per_page=100; later requests use the absolute Link URL verbatim (it already encodes them), so the local params drop off. The data, not a fixed cap, bounds the loop.
_iter_pages: follow Link rel=next; first page filtered, rest verbatim.
src/froot/adapters/github.py · 461 lines
src/froot/adapters/github.py
⋯ 206 lines hidden (lines 1–206)
1"""The GitHub forge adapter: git (checkout/push) + the GitHub REST API.
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.
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"""
14from __future__ import annotations
16import time
17from dataclasses import dataclass
18from datetime import timedelta
19from typing import TYPE_CHECKING, Any, final
21import httpx
22from temporalio.exceptions import ApplicationError
24from froot.config.settings import GitHubSettings
25from froot.domain.ci import (
26 CIAbsent,
27 CIFailed,
28 CIPassed,
29 CIPending,
30 CIStatus,
31)
32from froot.domain.pull_request import BranchName, PullRequestRef
34if TYPE_CHECKING:
35 from collections.abc import AsyncIterator
36 from pathlib import Path
38 from froot.domain.pull_request import PullRequestDraft
39 from froot.domain.repo import TargetRepo
41from froot.adapters._proc import run_text
43_API = "https://api.github.com"
44_API_VERSION = "2022-11-28"
45_TIMEOUT = 30.0
46_COMMITTER_NAME = "froot"
47_COMMITTER_EMAIL = "froot@users.noreply.github.com"
49# Check-run conclusions that mean the change is not safe to merge.
50_BAD_CONCLUSIONS = frozenset(
51 {
52 "failure",
53 "timed_out",
54 "cancelled",
55 "action_required",
56 "startup_failure",
57 "stale",
58 }
59)
62@final
63@dataclass(frozen=True, slots=True)
64class CheckRow:
65 """One GitHub check run, reduced to what the CI mapping needs."""
67 name: str
68 status: str
69 conclusion: str | None
72def ci_status_from_checks(
73 checks: tuple[CheckRow, ...], combined_state: str | None
74) -> CIStatus:
75 """Map GitHub check rows + the combined commit status to a CI status.
77 Args:
78 checks: The commit's check runs (GitHub Checks API).
79 combined_state: The legacy combined commit status (``success`` /
80 ``failure`` / ``pending``), or ``None`` when no statuses exist.
82 Returns:
83 ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
84 unresolved, ``CIFailed`` (with the failing check names) on any bad
85 conclusion or a failed combined status, else ``CIPassed``.
86 """
87 if not checks and combined_state is None:
88 return CIAbsent()
89 unresolved = combined_state == "pending" or any(
90 row.status != "completed" for row in checks
91 )
92 if unresolved:
93 return CIPending()
94 failing = tuple(
95 row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
96 )
97 if failing or combined_state == "failure":
98 return CIFailed(failing=failing)
99 return CIPassed()
102def _token() -> str:
103 token = GitHubSettings().github_token
104 if token is None:
105 # A missing token is a permanent misconfiguration, not a transient
106 # fault — fail the activity fast instead of retrying forever.
107 raise ApplicationError(
108 "FROOT_GITHUB_TOKEN is required", non_retryable=True
109 )
110 return token.get_secret_value()
113def _auth_remote(target: TargetRepo) -> str:
114 return (
115 f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
116 )
119def _client() -> httpx.AsyncClient:
120 return httpx.AsyncClient(
121 base_url=_API,
122 timeout=_TIMEOUT,
123 headers={
124 "Authorization": f"Bearer {_token()}",
125 "Accept": "application/vnd.github+json",
126 "X-GitHub-Api-Version": _API_VERSION,
127 },
128 )
131def _retry_after_seconds(response: httpx.Response) -> float | None:
132 """The server-advised wait before retrying a rate-limited request, if any.
134 Honors the secondary-limit ``Retry-After`` header (delta-seconds form), then
135 the primary-limit ``X-RateLimit-Reset`` (epoch) when the remaining quota is
136 zero. Returns ``None`` when GitHub gives no usable hint (let Temporal's own
137 backoff decide).
138 """
139 retry_after = response.headers.get("retry-after")
140 if retry_after is not None:
141 try:
142 return max(0.0, float(retry_after))
143 except ValueError:
144 return None # HTTP-date form is rare from GitHub; fall through.
145 if response.headers.get("x-ratelimit-remaining") == "0":
146 reset = response.headers.get("x-ratelimit-reset")
147 if reset is not None:
148 try:
149 return max(0.0, float(reset) - time.time())
150 except ValueError:
151 return None
152 return None
155def _is_rate_limited(response: httpx.Response) -> bool:
156 """Whether a 403/429 is GitHub *rate limiting* (transient), not auth.
158 GitHub overloads 403 for both a permanent permission fault AND its secondary
159 rate limit, so the two are told apart by the rate-limit markers: an explicit
160 ``Retry-After``, an exhausted ``X-RateLimit-Remaining``, or a "rate limit"
161 note in the body (the secondary-limit form often only says so there).
162 """
163 if response.status_code == 429:
164 return True
165 if response.status_code != 403:
166 return False
167 if "retry-after" in response.headers:
168 return True
169 if response.headers.get("x-ratelimit-remaining") == "0":
170 return True
171 return "rate limit" in response.text.lower()
174def _raise_for_status(response: httpx.Response) -> None:
175 """Raise on error, classifying retryability the way the loop needs.
177 A 401 is always a permanent auth fault. A 403/429 doubles as GitHub's
178 rate-limit signal: when it carries a rate-limit marker the fault is
179 TRANSIENT, so raise a *retryable* error (honoring the server's advised wait)
180 and let Temporal back off rather than killing the loop — this is the
181 call-heaviest path (the durable CI poll), so a transient limit must not be
182 fatal. A 403 without those markers is a real permission fault and stays
183 non-retryable.
184 """
185 if response.status_code == 401:
186 raise ApplicationError("GitHub auth failed (401)", non_retryable=True)
187 if response.status_code in (403, 429) and _is_rate_limited(response):
188 delay = _retry_after_seconds(response)
189 raise ApplicationError(
190 f"GitHub rate-limited ({response.status_code}); backing off",
191 type="GitHubRateLimited",
192 next_retry_delay=(
193 timedelta(seconds=delay) if delay is not None else None
194 ),
195 )
196 if response.status_code == 403:
197 raise ApplicationError("GitHub auth failed (403)", non_retryable=True)
198 response.raise_for_status()
201def _next_link(response: httpx.Response) -> str | None:
202 """The ``rel="next"`` page URL from the Link header, or None at the end."""
203 nxt = response.links.get("next")
204 return nxt.get("url") if nxt else None
207async def _iter_pages(
208 client: httpx.AsyncClient,
209 url: str,
210 params: dict[str, Any] | None = None,
211) -> AsyncIterator[httpx.Response]:
212 """Yield each page of a GitHub list endpoint, following Link ``rel=next``.
214 GitHub caps a page at 100 items; a single ``per_page=100`` read silently
215 truncates a busy repo — which here would *misread the CI oracle* (a failing
216 101st check goes unseen) or *break comment idempotency* (froot's marker on a
217 later page is missed, so it posts a duplicate). Following the Link header
218 reads the whole set; the data, not a fixed cap, bounds the loop.
219 """
220 next_url: str | None = url
221 # First page carries the caller's filters + per_page; later pages use the
222 # absolute Link URL verbatim (it already encodes them), so params drop off.
223 next_params: dict[str, Any] | None = {**(params or {}), "per_page": 100}
224 while next_url is not None:
225 resp = await client.get(next_url, params=next_params)
226 _raise_for_status(resp)
227 yield resp
228 next_url = _next_link(resp)
229 next_params = None
⋯ 232 lines hidden (lines 230–461)
232async def _marked_comment_id(
233 client: httpx.AsyncClient, slug: str, number: int, marker: str
234) -> int | None:
235 """The id of froot's ``marker``-tagged comment on a PR, across all pages.
237 Searches every comment page (stopping at the first hit) so a chatty PR that
238 pushes the marker past the first 100 comments cannot fool the upsert into
239 posting a duplicate.
240 """
241 async for page in _iter_pages(
242 client, f"/repos/{slug}/issues/{number}/comments"
243 ):
244 for comment in page.json():
245 if isinstance(comment, dict) and marker in str(
246 comment.get("body", "")
247 ):
248 return int(comment["id"])
249 return None
252def _pull_request_ref(payload: Any) -> PullRequestRef:
253 """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
254 head = payload["head"]
255 return PullRequestRef(
256 number=int(payload["number"]),
257 url=str(payload["html_url"]),
258 branch=BranchName(value=str(head["ref"])),
259 head_sha=str(head["sha"]),
260 )
263@final
264class GitHubForge:
265 """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
267 async def checkout(self, target: TargetRepo, workspace: Path) -> None:
268 """Shallow-clone the repo's default branch into ``workspace``."""
269 code, out, err = await run_text(
270 "git",
271 "clone",
272 "--depth",
273 "1",
274 "--branch",
275 target.default_branch,
276 _auth_remote(target),
277 ".",
278 cwd=workspace,
279 )
280 if code != 0:
281 raise RuntimeError(f"git clone failed ({code}): {err or out}")
283 async def checkout_pull_request(
284 self, target: TargetRepo, workspace: Path, number: int
285 ) -> None:
286 """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
288 Fetches the PR head ref the base repo exposes for every PR (fork or
289 not), so a community PR from a fork checks out the same way a same-repo
290 one does — no fork URL, no cross-repo auth.
291 """
292 ref = f"pull/{number}/head"
293 steps: tuple[tuple[str, ...], ...] = (
294 ("git", "init", "-q"),
295 ("git", "remote", "add", "origin", _auth_remote(target)),
296 ("git", "fetch", "--depth", "1", "origin", ref),
297 ("git", "checkout", "-q", "FETCH_HEAD"),
298 )
299 for step in steps:
300 code, out, err = await run_text(*step, cwd=workspace)
301 if code != 0:
302 raise RuntimeError(
303 f"git {step[1]} ({ref}) failed ({code}): {err or out}"
304 )
306 async def push_branch(
307 self, workspace: Path, branch: BranchName, commit_message: str
308 ) -> str:
309 """Commit the workspace changes onto ``branch`` and push; return SHA."""
310 steps: tuple[tuple[str, ...], ...] = (
311 ("git", "checkout", "-b", branch.value),
312 ("git", "add", "-A"),
313 (
314 "git",
315 "-c",
316 f"user.name={_COMMITTER_NAME}",
317 "-c",
318 f"user.email={_COMMITTER_EMAIL}",
319 "commit",
320 "-m",
321 commit_message,
322 ),
323 ("git", "push", "-u", "origin", branch.value),
324 )
325 for step in steps:
326 code, out, err = await run_text(*step, cwd=workspace)
327 if code != 0:
328 raise RuntimeError(f"{step[0:2]} failed ({code}): {err or out}")
329 code, sha, _ = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
330 if code != 0:
331 raise RuntimeError(f"git rev-parse failed ({code})")
332 return sha.strip()
334 async def find_open_pull_request(
335 self, target: TargetRepo, branch: BranchName
336 ) -> PullRequestRef | None:
337 """Return the open PR for ``branch`` if one already exists (dedup)."""
338 async with _client() as client:
339 resp = await client.get(
340 f"/repos/{target.repo.slug}/pulls",
341 params={
342 "head": f"{target.repo.owner}:{branch.value}",
343 "state": "open",
344 },
345 )
346 _raise_for_status(resp)
347 payloads = resp.json()
348 if isinstance(payloads, list) and payloads:
349 return _pull_request_ref(payloads[0])
350 return None
352 async def list_open_pull_requests(
353 self, target: TargetRepo
354 ) -> tuple[PullRequestRef, ...]:
355 """List the repo's open PRs (the determinism reviewer's work feed)."""
356 refs: list[PullRequestRef] = []
357 async with _client() as client:
358 async for page in _iter_pages(
359 client,
360 f"/repos/{target.repo.slug}/pulls",
361 {"state": "open"},
362 ):
363 payload = page.json()
364 if isinstance(payload, list):
365 refs.extend(_pull_request_ref(p) for p in payload)
366 return tuple(refs)
368 async def open_pull_request(
369 self, target: TargetRepo, draft: PullRequestDraft
370 ) -> PullRequestRef:
371 """Open the PR for an already-pushed branch (idempotent on conflict)."""
372 async with _client() as client:
373 resp = await client.post(
374 f"/repos/{target.repo.slug}/pulls",
375 json={
376 "title": draft.title,
377 "head": draft.branch.value,
378 "base": draft.base,
379 "body": draft.body,
380 },
381 )
382 if resp.status_code == 422:
383 existing = await self.find_open_pull_request(target, draft.branch)
384 if existing is not None:
385 return existing
386 _raise_for_status(resp)
387 return _pull_request_ref(resp.json())
389 async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
390 """Read the repo's combined CI status for a commit (the oracle).
392 The check-runs are paginated (a commit can have >100 checks, and missing
393 a failing one would falsely read green); the legacy combined status is a
394 single rolled-up ``state`` over all statuses, so it needs no paging.
395 """
396 slug = target.repo.slug
397 rows: list[CheckRow] = []
398 async with _client() as client:
399 async for page in _iter_pages(
400 client, f"/repos/{slug}/commits/{head_sha}/check-runs"
401 ):
402 rows.extend(
403 CheckRow(
404 name=str(run["name"]),
405 status=str(run["status"]),
406 conclusion=(
407 str(run["conclusion"])
408 if run.get("conclusion") is not None
409 else None
410 ),
411 )
412 for run in page.json().get("check_runs", [])
413 )
414 status_resp = await client.get(
415 f"/repos/{slug}/commits/{head_sha}/status"
416 )
417 _raise_for_status(status_resp)
418 status_json = status_resp.json()
419 combined = (
420 str(status_json["state"])
421 if int(status_json.get("total_count", 0)) > 0
422 else None
423 )
424 return ci_status_from_checks(tuple(rows), combined)
426 async def add_labels(
427 self, target: TargetRepo, number: int, labels: tuple[str, ...]
428 ) -> None:
429 """Attach labels to a PR (the human-readable signal-update)."""
430 async with _client() as client:
431 resp = await client.post(
432 f"/repos/{target.repo.slug}/issues/{number}/labels",
433 json={"labels": list(labels)},
434 )
435 _raise_for_status(resp)
437 async def upsert_issue_comment(
438 self, target: TargetRepo, number: int, marker: str, body: str
439 ) -> str:
440 """Create or update the PR's ``marker``-tagged comment; return its URL.
442 A PR conversation (issue) comment, not an inline review comment: the
443 determinism findings are structural (a call path), so a single advisory
444 summary fits better than line anchors. The marker lets a re-review edit
445 its own prior comment in place rather than stack a new one.
446 """
447 slug = target.repo.slug
448 async with _client() as client:
449 existing_id = await _marked_comment_id(client, slug, number, marker)
450 if existing_id is not None:
451 resp = await client.patch(
452 f"/repos/{slug}/issues/comments/{existing_id}",
453 json={"body": body},
454 )
455 else:
456 resp = await client.post(
457 f"/repos/{slug}/issues/{number}/comments",
458 json={"body": body},
459 )
460 _raise_for_status(resp)
461 return str(resp.json()["html_url"])
ci_status now extends its CheckRow list across every check-runs page before mapping. The combined commit status stays a single GET — it is one rolled-up state over all statuses, not a paginated list, so it needs no paging.
ci_status: paginate check-runs; the combined status stays a single read.
src/froot/adapters/github.py · 461 lines
src/froot/adapters/github.py
⋯ 395 lines hidden (lines 1–395)
1"""The GitHub forge adapter: git (checkout/push) + the GitHub REST API.
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.
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"""
14from __future__ import annotations
16import time
17from dataclasses import dataclass
18from datetime import timedelta
19from typing import TYPE_CHECKING, Any, final
21import httpx
22from temporalio.exceptions import ApplicationError
24from froot.config.settings import GitHubSettings
25from froot.domain.ci import (
26 CIAbsent,
27 CIFailed,
28 CIPassed,
29 CIPending,
30 CIStatus,
31)
32from froot.domain.pull_request import BranchName, PullRequestRef
34if TYPE_CHECKING:
35 from collections.abc import AsyncIterator
36 from pathlib import Path
38 from froot.domain.pull_request import PullRequestDraft
39 from froot.domain.repo import TargetRepo
41from froot.adapters._proc import run_text
43_API = "https://api.github.com"
44_API_VERSION = "2022-11-28"
45_TIMEOUT = 30.0
46_COMMITTER_NAME = "froot"
47_COMMITTER_EMAIL = "froot@users.noreply.github.com"
49# Check-run conclusions that mean the change is not safe to merge.
50_BAD_CONCLUSIONS = frozenset(
51 {
52 "failure",
53 "timed_out",
54 "cancelled",
55 "action_required",
56 "startup_failure",
57 "stale",
58 }
59)
62@final
63@dataclass(frozen=True, slots=True)
64class CheckRow:
65 """One GitHub check run, reduced to what the CI mapping needs."""
67 name: str
68 status: str
69 conclusion: str | None
72def ci_status_from_checks(
73 checks: tuple[CheckRow, ...], combined_state: str | None
74) -> CIStatus:
75 """Map GitHub check rows + the combined commit status to a CI status.
77 Args:
78 checks: The commit's check runs (GitHub Checks API).
79 combined_state: The legacy combined commit status (``success`` /
80 ``failure`` / ``pending``), or ``None`` when no statuses exist.
82 Returns:
83 ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
84 unresolved, ``CIFailed`` (with the failing check names) on any bad
85 conclusion or a failed combined status, else ``CIPassed``.
86 """
87 if not checks and combined_state is None:
88 return CIAbsent()
89 unresolved = combined_state == "pending" or any(
90 row.status != "completed" for row in checks
91 )
92 if unresolved:
93 return CIPending()
94 failing = tuple(
95 row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
96 )
97 if failing or combined_state == "failure":
98 return CIFailed(failing=failing)
99 return CIPassed()
102def _token() -> str:
103 token = GitHubSettings().github_token
104 if token is None:
105 # A missing token is a permanent misconfiguration, not a transient
106 # fault — fail the activity fast instead of retrying forever.
107 raise ApplicationError(
108 "FROOT_GITHUB_TOKEN is required", non_retryable=True
109 )
110 return token.get_secret_value()
113def _auth_remote(target: TargetRepo) -> str:
114 return (
115 f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
116 )
119def _client() -> httpx.AsyncClient:
120 return httpx.AsyncClient(
121 base_url=_API,
122 timeout=_TIMEOUT,
123 headers={
124 "Authorization": f"Bearer {_token()}",
125 "Accept": "application/vnd.github+json",
126 "X-GitHub-Api-Version": _API_VERSION,
127 },
128 )
131def _retry_after_seconds(response: httpx.Response) -> float | None:
132 """The server-advised wait before retrying a rate-limited request, if any.
134 Honors the secondary-limit ``Retry-After`` header (delta-seconds form), then
135 the primary-limit ``X-RateLimit-Reset`` (epoch) when the remaining quota is
136 zero. Returns ``None`` when GitHub gives no usable hint (let Temporal's own
137 backoff decide).
138 """
139 retry_after = response.headers.get("retry-after")
140 if retry_after is not None:
141 try:
142 return max(0.0, float(retry_after))
143 except ValueError:
144 return None # HTTP-date form is rare from GitHub; fall through.
145 if response.headers.get("x-ratelimit-remaining") == "0":
146 reset = response.headers.get("x-ratelimit-reset")
147 if reset is not None:
148 try:
149 return max(0.0, float(reset) - time.time())
150 except ValueError:
151 return None
152 return None
155def _is_rate_limited(response: httpx.Response) -> bool:
156 """Whether a 403/429 is GitHub *rate limiting* (transient), not auth.
158 GitHub overloads 403 for both a permanent permission fault AND its secondary
159 rate limit, so the two are told apart by the rate-limit markers: an explicit
160 ``Retry-After``, an exhausted ``X-RateLimit-Remaining``, or a "rate limit"
161 note in the body (the secondary-limit form often only says so there).
162 """
163 if response.status_code == 429:
164 return True
165 if response.status_code != 403:
166 return False
167 if "retry-after" in response.headers:
168 return True
169 if response.headers.get("x-ratelimit-remaining") == "0":
170 return True
171 return "rate limit" in response.text.lower()
174def _raise_for_status(response: httpx.Response) -> None:
175 """Raise on error, classifying retryability the way the loop needs.
177 A 401 is always a permanent auth fault. A 403/429 doubles as GitHub's
178 rate-limit signal: when it carries a rate-limit marker the fault is
179 TRANSIENT, so raise a *retryable* error (honoring the server's advised wait)
180 and let Temporal back off rather than killing the loop — this is the
181 call-heaviest path (the durable CI poll), so a transient limit must not be
182 fatal. A 403 without those markers is a real permission fault and stays
183 non-retryable.
184 """
185 if response.status_code == 401:
186 raise ApplicationError("GitHub auth failed (401)", non_retryable=True)
187 if response.status_code in (403, 429) and _is_rate_limited(response):
188 delay = _retry_after_seconds(response)
189 raise ApplicationError(
190 f"GitHub rate-limited ({response.status_code}); backing off",
191 type="GitHubRateLimited",
192 next_retry_delay=(
193 timedelta(seconds=delay) if delay is not None else None
194 ),
195 )
196 if response.status_code == 403:
197 raise ApplicationError("GitHub auth failed (403)", non_retryable=True)
198 response.raise_for_status()
201def _next_link(response: httpx.Response) -> str | None:
202 """The ``rel="next"`` page URL from the Link header, or None at the end."""
203 nxt = response.links.get("next")
204 return nxt.get("url") if nxt else None
207async def _iter_pages(
208 client: httpx.AsyncClient,
209 url: str,
210 params: dict[str, Any] | None = None,
211) -> AsyncIterator[httpx.Response]:
212 """Yield each page of a GitHub list endpoint, following Link ``rel=next``.
214 GitHub caps a page at 100 items; a single ``per_page=100`` read silently
215 truncates a busy repo — which here would *misread the CI oracle* (a failing
216 101st check goes unseen) or *break comment idempotency* (froot's marker on a
217 later page is missed, so it posts a duplicate). Following the Link header
218 reads the whole set; the data, not a fixed cap, bounds the loop.
219 """
220 next_url: str | None = url
221 # First page carries the caller's filters + per_page; later pages use the
222 # absolute Link URL verbatim (it already encodes them), so params drop off.
223 next_params: dict[str, Any] | None = {**(params or {}), "per_page": 100}
224 while next_url is not None:
225 resp = await client.get(next_url, params=next_params)
226 _raise_for_status(resp)
227 yield resp
228 next_url = _next_link(resp)
229 next_params = None
232async def _marked_comment_id(
233 client: httpx.AsyncClient, slug: str, number: int, marker: str
234) -> int | None:
235 """The id of froot's ``marker``-tagged comment on a PR, across all pages.
237 Searches every comment page (stopping at the first hit) so a chatty PR that
238 pushes the marker past the first 100 comments cannot fool the upsert into
239 posting a duplicate.
240 """
241 async for page in _iter_pages(
242 client, f"/repos/{slug}/issues/{number}/comments"
243 ):
244 for comment in page.json():
245 if isinstance(comment, dict) and marker in str(
246 comment.get("body", "")
247 ):
248 return int(comment["id"])
249 return None
252def _pull_request_ref(payload: Any) -> PullRequestRef:
253 """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
254 head = payload["head"]
255 return PullRequestRef(
256 number=int(payload["number"]),
257 url=str(payload["html_url"]),
258 branch=BranchName(value=str(head["ref"])),
259 head_sha=str(head["sha"]),
260 )
263@final
264class GitHubForge:
265 """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
267 async def checkout(self, target: TargetRepo, workspace: Path) -> None:
268 """Shallow-clone the repo's default branch into ``workspace``."""
269 code, out, err = await run_text(
270 "git",
271 "clone",
272 "--depth",
273 "1",
274 "--branch",
275 target.default_branch,
276 _auth_remote(target),
277 ".",
278 cwd=workspace,
279 )
280 if code != 0:
281 raise RuntimeError(f"git clone failed ({code}): {err or out}")
283 async def checkout_pull_request(
284 self, target: TargetRepo, workspace: Path, number: int
285 ) -> None:
286 """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
288 Fetches the PR head ref the base repo exposes for every PR (fork or
289 not), so a community PR from a fork checks out the same way a same-repo
290 one does — no fork URL, no cross-repo auth.
291 """
292 ref = f"pull/{number}/head"
293 steps: tuple[tuple[str, ...], ...] = (
294 ("git", "init", "-q"),
295 ("git", "remote", "add", "origin", _auth_remote(target)),
296 ("git", "fetch", "--depth", "1", "origin", ref),
297 ("git", "checkout", "-q", "FETCH_HEAD"),
298 )
299 for step in steps:
300 code, out, err = await run_text(*step, cwd=workspace)
301 if code != 0:
302 raise RuntimeError(
303 f"git {step[1]} ({ref}) failed ({code}): {err or out}"
304 )
306 async def push_branch(
307 self, workspace: Path, branch: BranchName, commit_message: str
308 ) -> str:
309 """Commit the workspace changes onto ``branch`` and push; return SHA."""
310 steps: tuple[tuple[str, ...], ...] = (
311 ("git", "checkout", "-b", branch.value),
312 ("git", "add", "-A"),
313 (
314 "git",
315 "-c",
316 f"user.name={_COMMITTER_NAME}",
317 "-c",
318 f"user.email={_COMMITTER_EMAIL}",
319 "commit",
320 "-m",
321 commit_message,
322 ),
323 ("git", "push", "-u", "origin", branch.value),
324 )
325 for step in steps:
326 code, out, err = await run_text(*step, cwd=workspace)
327 if code != 0:
328 raise RuntimeError(f"{step[0:2]} failed ({code}): {err or out}")
329 code, sha, _ = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
330 if code != 0:
331 raise RuntimeError(f"git rev-parse failed ({code})")
332 return sha.strip()
334 async def find_open_pull_request(
335 self, target: TargetRepo, branch: BranchName
336 ) -> PullRequestRef | None:
337 """Return the open PR for ``branch`` if one already exists (dedup)."""
338 async with _client() as client:
339 resp = await client.get(
340 f"/repos/{target.repo.slug}/pulls",
341 params={
342 "head": f"{target.repo.owner}:{branch.value}",
343 "state": "open",
344 },
345 )
346 _raise_for_status(resp)
347 payloads = resp.json()
348 if isinstance(payloads, list) and payloads:
349 return _pull_request_ref(payloads[0])
350 return None
352 async def list_open_pull_requests(
353 self, target: TargetRepo
354 ) -> tuple[PullRequestRef, ...]:
355 """List the repo's open PRs (the determinism reviewer's work feed)."""
356 refs: list[PullRequestRef] = []
357 async with _client() as client:
358 async for page in _iter_pages(
359 client,
360 f"/repos/{target.repo.slug}/pulls",
361 {"state": "open"},
362 ):
363 payload = page.json()
364 if isinstance(payload, list):
365 refs.extend(_pull_request_ref(p) for p in payload)
366 return tuple(refs)
368 async def open_pull_request(
369 self, target: TargetRepo, draft: PullRequestDraft
370 ) -> PullRequestRef:
371 """Open the PR for an already-pushed branch (idempotent on conflict)."""
372 async with _client() as client:
373 resp = await client.post(
374 f"/repos/{target.repo.slug}/pulls",
375 json={
376 "title": draft.title,
377 "head": draft.branch.value,
378 "base": draft.base,
379 "body": draft.body,
380 },
381 )
382 if resp.status_code == 422:
383 existing = await self.find_open_pull_request(target, draft.branch)
384 if existing is not None:
385 return existing
386 _raise_for_status(resp)
387 return _pull_request_ref(resp.json())
389 async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
390 """Read the repo's combined CI status for a commit (the oracle).
392 The check-runs are paginated (a commit can have >100 checks, and missing
393 a failing one would falsely read green); the legacy combined status is a
394 single rolled-up ``state`` over all statuses, so it needs no paging.
395 """
396 slug = target.repo.slug
397 rows: list[CheckRow] = []
398 async with _client() as client:
399 async for page in _iter_pages(
400 client, f"/repos/{slug}/commits/{head_sha}/check-runs"
401 ):
402 rows.extend(
403 CheckRow(
404 name=str(run["name"]),
405 status=str(run["status"]),
406 conclusion=(
407 str(run["conclusion"])
408 if run.get("conclusion") is not None
409 else None
410 ),
411 )
412 for run in page.json().get("check_runs", [])
413 )
414 status_resp = await client.get(
415 f"/repos/{slug}/commits/{head_sha}/status"
416 )
417 _raise_for_status(status_resp)
418 status_json = status_resp.json()
419 combined = (
420 str(status_json["state"])
421 if int(status_json.get("total_count", 0)) > 0
422 else None
423 )
424 return ci_status_from_checks(tuple(rows), combined)
⋯ 37 lines hidden (lines 425–461)
426 async def add_labels(
427 self, target: TargetRepo, number: int, labels: tuple[str, ...]
428 ) -> None:
429 """Attach labels to a PR (the human-readable signal-update)."""
430 async with _client() as client:
431 resp = await client.post(
432 f"/repos/{target.repo.slug}/issues/{number}/labels",
433 json={"labels": list(labels)},
434 )
435 _raise_for_status(resp)
437 async def upsert_issue_comment(
438 self, target: TargetRepo, number: int, marker: str, body: str
439 ) -> str:
440 """Create or update the PR's ``marker``-tagged comment; return its URL.
442 A PR conversation (issue) comment, not an inline review comment: the
443 determinism findings are structural (a call path), so a single advisory
444 summary fits better than line anchors. The marker lets a re-review edit
445 its own prior comment in place rather than stack a new one.
446 """
447 slug = target.repo.slug
448 async with _client() as client:
449 existing_id = await _marked_comment_id(client, slug, number, marker)
450 if existing_id is not None:
451 resp = await client.patch(
452 f"/repos/{slug}/issues/comments/{existing_id}",
453 json={"body": body},
454 )
455 else:
456 resp = await client.post(
457 f"/repos/{slug}/issues/{number}/comments",
458 json={"body": body},
459 )
460 _raise_for_status(resp)
461 return str(resp.json()["html_url"])
The comment upsert delegates its lookup to _marked_comment_id, which iterates every comment page and stops at the first hit. Finding the marker on any page means the upsert PATCHes that comment; only a genuine absence falls through to a POST.
_marked_comment_id: search all pages, stop at the first hit.
src/froot/adapters/github.py · 461 lines
src/froot/adapters/github.py
⋯ 231 lines hidden (lines 1–231)
1"""The GitHub forge adapter: git (checkout/push) + the GitHub REST API.
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.
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"""
14from __future__ import annotations
16import time
17from dataclasses import dataclass
18from datetime import timedelta
19from typing import TYPE_CHECKING, Any, final
21import httpx
22from temporalio.exceptions import ApplicationError
24from froot.config.settings import GitHubSettings
25from froot.domain.ci import (
26 CIAbsent,
27 CIFailed,
28 CIPassed,
29 CIPending,
30 CIStatus,
31)
32from froot.domain.pull_request import BranchName, PullRequestRef
34if TYPE_CHECKING:
35 from collections.abc import AsyncIterator
36 from pathlib import Path
38 from froot.domain.pull_request import PullRequestDraft
39 from froot.domain.repo import TargetRepo
41from froot.adapters._proc import run_text
43_API = "https://api.github.com"
44_API_VERSION = "2022-11-28"
45_TIMEOUT = 30.0
46_COMMITTER_NAME = "froot"
47_COMMITTER_EMAIL = "froot@users.noreply.github.com"
49# Check-run conclusions that mean the change is not safe to merge.
50_BAD_CONCLUSIONS = frozenset(
51 {
52 "failure",
53 "timed_out",
54 "cancelled",
55 "action_required",
56 "startup_failure",
57 "stale",
58 }
59)
62@final
63@dataclass(frozen=True, slots=True)
64class CheckRow:
65 """One GitHub check run, reduced to what the CI mapping needs."""
67 name: str
68 status: str
69 conclusion: str | None
72def ci_status_from_checks(
73 checks: tuple[CheckRow, ...], combined_state: str | None
74) -> CIStatus:
75 """Map GitHub check rows + the combined commit status to a CI status.
77 Args:
78 checks: The commit's check runs (GitHub Checks API).
79 combined_state: The legacy combined commit status (``success`` /
80 ``failure`` / ``pending``), or ``None`` when no statuses exist.
82 Returns:
83 ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
84 unresolved, ``CIFailed`` (with the failing check names) on any bad
85 conclusion or a failed combined status, else ``CIPassed``.
86 """
87 if not checks and combined_state is None:
88 return CIAbsent()
89 unresolved = combined_state == "pending" or any(
90 row.status != "completed" for row in checks
91 )
92 if unresolved:
93 return CIPending()
94 failing = tuple(
95 row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
96 )
97 if failing or combined_state == "failure":
98 return CIFailed(failing=failing)
99 return CIPassed()
102def _token() -> str:
103 token = GitHubSettings().github_token
104 if token is None:
105 # A missing token is a permanent misconfiguration, not a transient
106 # fault — fail the activity fast instead of retrying forever.
107 raise ApplicationError(
108 "FROOT_GITHUB_TOKEN is required", non_retryable=True
109 )
110 return token.get_secret_value()
113def _auth_remote(target: TargetRepo) -> str:
114 return (
115 f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
116 )
119def _client() -> httpx.AsyncClient:
120 return httpx.AsyncClient(
121 base_url=_API,
122 timeout=_TIMEOUT,
123 headers={
124 "Authorization": f"Bearer {_token()}",
125 "Accept": "application/vnd.github+json",
126 "X-GitHub-Api-Version": _API_VERSION,
127 },
128 )
131def _retry_after_seconds(response: httpx.Response) -> float | None:
132 """The server-advised wait before retrying a rate-limited request, if any.
134 Honors the secondary-limit ``Retry-After`` header (delta-seconds form), then
135 the primary-limit ``X-RateLimit-Reset`` (epoch) when the remaining quota is
136 zero. Returns ``None`` when GitHub gives no usable hint (let Temporal's own
137 backoff decide).
138 """
139 retry_after = response.headers.get("retry-after")
140 if retry_after is not None:
141 try:
142 return max(0.0, float(retry_after))
143 except ValueError:
144 return None # HTTP-date form is rare from GitHub; fall through.
145 if response.headers.get("x-ratelimit-remaining") == "0":
146 reset = response.headers.get("x-ratelimit-reset")
147 if reset is not None:
148 try:
149 return max(0.0, float(reset) - time.time())
150 except ValueError:
151 return None
152 return None
155def _is_rate_limited(response: httpx.Response) -> bool:
156 """Whether a 403/429 is GitHub *rate limiting* (transient), not auth.
158 GitHub overloads 403 for both a permanent permission fault AND its secondary
159 rate limit, so the two are told apart by the rate-limit markers: an explicit
160 ``Retry-After``, an exhausted ``X-RateLimit-Remaining``, or a "rate limit"
161 note in the body (the secondary-limit form often only says so there).
162 """
163 if response.status_code == 429:
164 return True
165 if response.status_code != 403:
166 return False
167 if "retry-after" in response.headers:
168 return True
169 if response.headers.get("x-ratelimit-remaining") == "0":
170 return True
171 return "rate limit" in response.text.lower()
174def _raise_for_status(response: httpx.Response) -> None:
175 """Raise on error, classifying retryability the way the loop needs.
177 A 401 is always a permanent auth fault. A 403/429 doubles as GitHub's
178 rate-limit signal: when it carries a rate-limit marker the fault is
179 TRANSIENT, so raise a *retryable* error (honoring the server's advised wait)
180 and let Temporal back off rather than killing the loop — this is the
181 call-heaviest path (the durable CI poll), so a transient limit must not be
182 fatal. A 403 without those markers is a real permission fault and stays
183 non-retryable.
184 """
185 if response.status_code == 401:
186 raise ApplicationError("GitHub auth failed (401)", non_retryable=True)
187 if response.status_code in (403, 429) and _is_rate_limited(response):
188 delay = _retry_after_seconds(response)
189 raise ApplicationError(
190 f"GitHub rate-limited ({response.status_code}); backing off",
191 type="GitHubRateLimited",
192 next_retry_delay=(
193 timedelta(seconds=delay) if delay is not None else None
194 ),
195 )
196 if response.status_code == 403:
197 raise ApplicationError("GitHub auth failed (403)", non_retryable=True)
198 response.raise_for_status()
201def _next_link(response: httpx.Response) -> str | None:
202 """The ``rel="next"`` page URL from the Link header, or None at the end."""
203 nxt = response.links.get("next")
204 return nxt.get("url") if nxt else None
207async def _iter_pages(
208 client: httpx.AsyncClient,
209 url: str,
210 params: dict[str, Any] | None = None,
211) -> AsyncIterator[httpx.Response]:
212 """Yield each page of a GitHub list endpoint, following Link ``rel=next``.
214 GitHub caps a page at 100 items; a single ``per_page=100`` read silently
215 truncates a busy repo — which here would *misread the CI oracle* (a failing
216 101st check goes unseen) or *break comment idempotency* (froot's marker on a
217 later page is missed, so it posts a duplicate). Following the Link header
218 reads the whole set; the data, not a fixed cap, bounds the loop.
219 """
220 next_url: str | None = url
221 # First page carries the caller's filters + per_page; later pages use the
222 # absolute Link URL verbatim (it already encodes them), so params drop off.
223 next_params: dict[str, Any] | None = {**(params or {}), "per_page": 100}
224 while next_url is not None:
225 resp = await client.get(next_url, params=next_params)
226 _raise_for_status(resp)
227 yield resp
228 next_url = _next_link(resp)
229 next_params = None
232async def _marked_comment_id(
233 client: httpx.AsyncClient, slug: str, number: int, marker: str
234) -> int | None:
235 """The id of froot's ``marker``-tagged comment on a PR, across all pages.
237 Searches every comment page (stopping at the first hit) so a chatty PR that
238 pushes the marker past the first 100 comments cannot fool the upsert into
239 posting a duplicate.
240 """
241 async for page in _iter_pages(
242 client, f"/repos/{slug}/issues/{number}/comments"
243 ):
244 for comment in page.json():
245 if isinstance(comment, dict) and marker in str(
246 comment.get("body", "")
247 ):
248 return int(comment["id"])
249 return None
⋯ 212 lines hidden (lines 250–461)
252def _pull_request_ref(payload: Any) -> PullRequestRef:
253 """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
254 head = payload["head"]
255 return PullRequestRef(
256 number=int(payload["number"]),
257 url=str(payload["html_url"]),
258 branch=BranchName(value=str(head["ref"])),
259 head_sha=str(head["sha"]),
260 )
263@final
264class GitHubForge:
265 """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
267 async def checkout(self, target: TargetRepo, workspace: Path) -> None:
268 """Shallow-clone the repo's default branch into ``workspace``."""
269 code, out, err = await run_text(
270 "git",
271 "clone",
272 "--depth",
273 "1",
274 "--branch",
275 target.default_branch,
276 _auth_remote(target),
277 ".",
278 cwd=workspace,
279 )
280 if code != 0:
281 raise RuntimeError(f"git clone failed ({code}): {err or out}")
283 async def checkout_pull_request(
284 self, target: TargetRepo, workspace: Path, number: int
285 ) -> None:
286 """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
288 Fetches the PR head ref the base repo exposes for every PR (fork or
289 not), so a community PR from a fork checks out the same way a same-repo
290 one does — no fork URL, no cross-repo auth.
291 """
292 ref = f"pull/{number}/head"
293 steps: tuple[tuple[str, ...], ...] = (
294 ("git", "init", "-q"),
295 ("git", "remote", "add", "origin", _auth_remote(target)),
296 ("git", "fetch", "--depth", "1", "origin", ref),
297 ("git", "checkout", "-q", "FETCH_HEAD"),
298 )
299 for step in steps:
300 code, out, err = await run_text(*step, cwd=workspace)
301 if code != 0:
302 raise RuntimeError(
303 f"git {step[1]} ({ref}) failed ({code}): {err or out}"
304 )
306 async def push_branch(
307 self, workspace: Path, branch: BranchName, commit_message: str
308 ) -> str:
309 """Commit the workspace changes onto ``branch`` and push; return SHA."""
310 steps: tuple[tuple[str, ...], ...] = (
311 ("git", "checkout", "-b", branch.value),
312 ("git", "add", "-A"),
313 (
314 "git",
315 "-c",
316 f"user.name={_COMMITTER_NAME}",
317 "-c",
318 f"user.email={_COMMITTER_EMAIL}",
319 "commit",
320 "-m",
321 commit_message,
322 ),
323 ("git", "push", "-u", "origin", branch.value),
324 )
325 for step in steps:
326 code, out, err = await run_text(*step, cwd=workspace)
327 if code != 0:
328 raise RuntimeError(f"{step[0:2]} failed ({code}): {err or out}")
329 code, sha, _ = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
330 if code != 0:
331 raise RuntimeError(f"git rev-parse failed ({code})")
332 return sha.strip()
334 async def find_open_pull_request(
335 self, target: TargetRepo, branch: BranchName
336 ) -> PullRequestRef | None:
337 """Return the open PR for ``branch`` if one already exists (dedup)."""
338 async with _client() as client:
339 resp = await client.get(
340 f"/repos/{target.repo.slug}/pulls",
341 params={
342 "head": f"{target.repo.owner}:{branch.value}",
343 "state": "open",
344 },
345 )
346 _raise_for_status(resp)
347 payloads = resp.json()
348 if isinstance(payloads, list) and payloads:
349 return _pull_request_ref(payloads[0])
350 return None
352 async def list_open_pull_requests(
353 self, target: TargetRepo
354 ) -> tuple[PullRequestRef, ...]:
355 """List the repo's open PRs (the determinism reviewer's work feed)."""
356 refs: list[PullRequestRef] = []
357 async with _client() as client:
358 async for page in _iter_pages(
359 client,
360 f"/repos/{target.repo.slug}/pulls",
361 {"state": "open"},
362 ):
363 payload = page.json()
364 if isinstance(payload, list):
365 refs.extend(_pull_request_ref(p) for p in payload)
366 return tuple(refs)
368 async def open_pull_request(
369 self, target: TargetRepo, draft: PullRequestDraft
370 ) -> PullRequestRef:
371 """Open the PR for an already-pushed branch (idempotent on conflict)."""
372 async with _client() as client:
373 resp = await client.post(
374 f"/repos/{target.repo.slug}/pulls",
375 json={
376 "title": draft.title,
377 "head": draft.branch.value,
378 "base": draft.base,
379 "body": draft.body,
380 },
381 )
382 if resp.status_code == 422:
383 existing = await self.find_open_pull_request(target, draft.branch)
384 if existing is not None:
385 return existing
386 _raise_for_status(resp)
387 return _pull_request_ref(resp.json())
389 async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
390 """Read the repo's combined CI status for a commit (the oracle).
392 The check-runs are paginated (a commit can have >100 checks, and missing
393 a failing one would falsely read green); the legacy combined status is a
394 single rolled-up ``state`` over all statuses, so it needs no paging.
395 """
396 slug = target.repo.slug
397 rows: list[CheckRow] = []
398 async with _client() as client:
399 async for page in _iter_pages(
400 client, f"/repos/{slug}/commits/{head_sha}/check-runs"
401 ):
402 rows.extend(
403 CheckRow(
404 name=str(run["name"]),
405 status=str(run["status"]),
406 conclusion=(
407 str(run["conclusion"])
408 if run.get("conclusion") is not None
409 else None
410 ),
411 )
412 for run in page.json().get("check_runs", [])
413 )
414 status_resp = await client.get(
415 f"/repos/{slug}/commits/{head_sha}/status"
416 )
417 _raise_for_status(status_resp)
418 status_json = status_resp.json()
419 combined = (
420 str(status_json["state"])
421 if int(status_json.get("total_count", 0)) > 0
422 else None
423 )
424 return ci_status_from_checks(tuple(rows), combined)
426 async def add_labels(
427 self, target: TargetRepo, number: int, labels: tuple[str, ...]
428 ) -> None:
429 """Attach labels to a PR (the human-readable signal-update)."""
430 async with _client() as client:
431 resp = await client.post(
432 f"/repos/{target.repo.slug}/issues/{number}/labels",
433 json={"labels": list(labels)},
434 )
435 _raise_for_status(resp)
437 async def upsert_issue_comment(
438 self, target: TargetRepo, number: int, marker: str, body: str
439 ) -> str:
440 """Create or update the PR's ``marker``-tagged comment; return its URL.
442 A PR conversation (issue) comment, not an inline review comment: the
443 determinism findings are structural (a call path), so a single advisory
444 summary fits better than line anchors. The marker lets a re-review edit
445 its own prior comment in place rather than stack a new one.
446 """
447 slug = target.repo.slug
448 async with _client() as client:
449 existing_id = await _marked_comment_id(client, slug, number, marker)
450 if existing_id is not None:
451 resp = await client.patch(
452 f"/repos/{slug}/issues/comments/{existing_id}",
453 json={"body": body},
454 )
455 else:
456 resp = await client.post(
457 f"/repos/{slug}/issues/{number}/comments",
458 json={"body": body},
459 )
460 _raise_for_status(resp)
461 return str(resp.json()["html_url"])
Fix 4 — the determinism tables can no longer drift apart
froot enforces determinism two ways. scripts/check_determinism.py is a vendored, package-free script that runs as the blocking CI gate. froot.policy.determinism is the brain's advisory reviewer that posts findings on PRs. Each independently defines the same four BANNED_* tables — the symbols that make a workflow non-deterministic — by hand-copy.
If those tables drift, froot's gate and its reviewer disagree about what "deterministic" means. That is the exact class of slow decay froot exists to catch, uncaught in froot itself. The new test pins them equal.
The dogfood guard: all four BANNED_* tables, kernel == brain.
tests/test_kernel_script.py · 227 lines
tests/test_kernel_script.py
⋯ 208 lines hidden (lines 1–208)
1"""Catch-tests for the vendored kernel script, and the kernel/brain boundary.
3``scripts/check_determinism.py`` runs standalone in CI (``check_determinism.py
4src``), which only proves it PASSES on clean code. These tests pin the other
5half — that it actually FAILS on hazards, across every banned category — and
6demonstrate the one thing the lexical kernel structurally cannot see: a
7transitive hazard, which only the brain's analyzer catches.
9The kernel is a vendored, package-free script, so it is loaded by file path.
10"""
12from __future__ import annotations
14import ast
15import importlib.util
16import sys
17from pathlib import Path
18from typing import TYPE_CHECKING, Any
20from froot.policy.determinism import LoadedModule, analyze_workflow_surface
22if TYPE_CHECKING:
23 from collections.abc import Mapping
25_SCRIPT = (
26 Path(__file__).resolve().parents[1] / "scripts" / "check_determinism.py"
27)
30def _load_kernel() -> Any:
31 spec = importlib.util.spec_from_file_location("_froot_kernel", _SCRIPT)
32 assert spec is not None
33 assert spec.loader is not None
34 module = importlib.util.module_from_spec(spec)
35 # Register before exec so the script's @dataclass can resolve its module.
36 sys.modules[spec.name] = module
37 spec.loader.exec_module(module)
38 return module
41_KERNEL = _load_kernel()
44# A workflow class touching every banned category, including the tricky
45# resolutions: an aliased import (dt.now) and a from-import (date.today).
46_HAZARDS = """
47import datetime
48import time
49import random
50import uuid
51import os
52import asyncio
53import threading
54import subprocess
55import socket
56import httpx
57from datetime import datetime as dt, date
58from temporalio import workflow
61@workflow.defn
62class Hazards:
63 @workflow.run
64 async def run(self) -> None:
65 datetime.datetime.now()
66 datetime.datetime.utcnow()
67 date.today()
68 dt.now()
69 time.time()
70 time.sleep(1)
71 random.randint(1, 6)
72 uuid.uuid4()
73 os.getenv("X")
74 os.environ["Y"]
75 await asyncio.sleep(1)
76 threading.Thread()
77 subprocess.run(["ls"])
78 socket.socket()
79 httpx.get("http://x")
80 open("/tmp/x")
81"""
83# Things that LOOK like hazards but aren't: the sanctioned `workflow.*`
84# replacements, a type annotation, and wall-clock inside an @activity.defn.
85_NEGATIVES = """
86import datetime
87import os
88from temporalio import workflow, activity
91@workflow.defn
92class Clean:
93 @workflow.run
94 async def run(self) -> datetime.datetime: # annotation, not a call
95 workflow.now()
96 workflow.uuid4()
97 workflow.random()
98 return workflow.now()
101@activity.defn
102async def legit() -> None:
103 datetime.datetime.now() # wall-clock is legal in an activity
104 os.environ["Z"]
105"""
108def _write(tmp_path: Path, name: str, source: str) -> Path:
109 path = tmp_path / name
110 path.write_text(source)
111 return path
114def test_kernel_catches_every_banned_category(tmp_path: Path) -> None:
115 path = _write(tmp_path, "wf.py", _HAZARDS)
116 rules = {finding.rule for finding in _KERNEL.scan_file(path)}
117 assert rules == {
118 "datetime.datetime.now", # plain and via the `dt` alias
119 "datetime.datetime.utcnow",
120 "datetime.date.today", # via the `date` from-import
121 "time.time",
122 "time.sleep",
123 "random.randint",
124 "uuid.uuid4",
125 "os.getenv",
126 "os.environ", # attribute access, not a call
127 "asyncio.sleep",
128 "threading.Thread",
129 "subprocess.run",
130 "socket.socket",
131 "httpx.get",
132 "open",
133 }
136def test_kernel_exit_code_is_nonzero_on_hazards(tmp_path: Path) -> None:
137 path = _write(tmp_path, "wf.py", _HAZARDS)
138 assert _KERNEL.main([str(path)]) == 1
141def test_kernel_ignores_sanctioned_annotations_and_activities(
142 tmp_path: Path,
143) -> None:
144 path = _write(tmp_path, "wf.py", _NEGATIVES)
145 assert _KERNEL.scan_file(path) == []
146 assert _KERNEL.main([str(path)]) == 0
149# ── The kernel/brain boundary — only the brain catches this ──────────────────
150_TRANSITIVE_WORKFLOW = """
151from temporalio import workflow
152from app.util import stamp
155@workflow.defn
156class W:
157 @workflow.run
158 async def run(self):
159 return stamp()
160"""
162_TRANSITIVE_HELPER = """
163import datetime
166def stamp():
167 return datetime.datetime.now()
168"""
171def _modules(sources: Mapping[str, str]) -> dict[str, LoadedModule]:
172 return {
173 qual: LoadedModule(
174 qualname=qual,
175 tree=ast.parse(src),
176 lines=tuple(src.splitlines()),
177 )
178 for qual, src in sources.items()
179 }
182def test_transitive_hazard_is_invisible_to_kernel_but_caught_by_brain(
183 tmp_path: Path,
184) -> None:
185 # Same code, two layers. The workflow file has NO banned call lexically —
186 # the hazard hides one call out, in a plain helper — and the helper file
187 # has no @workflow.defn, so the kernel ignores it there too. The lexical,
188 # workflow-scoped kernel is structurally blind to this.
189 wf_path = _write(tmp_path, "workflow.py", _TRANSITIVE_WORKFLOW)
190 helper_path = _write(tmp_path, "util.py", _TRANSITIVE_HELPER)
191 assert _KERNEL.scan_file(wf_path) == []
192 assert _KERNEL.scan_file(helper_path) == []
194 # The brain's call-graph chases the import into the helper and finds it.
195 result = analyze_workflow_surface(
196 _modules(
197 {
198 "app.workflow": _TRANSITIVE_WORKFLOW,
199 "app.util": _TRANSITIVE_HELPER,
200 }
201 )
202 )
203 assert len(result.hazards) == 1
204 hazard = result.hazards[0]
205 assert hazard.impurity.rule == "datetime.datetime.now"
206 assert hazard.via == ("stamp",)
209def test_banned_tables_match_the_brain() -> None:
210 """The vendored kernel and the brain's analyzer must ban the SAME symbols.
212 ``scripts/check_determinism.py`` (the blocking CI gate) and
213 :mod:`froot.policy.determinism` (the advisory reviewer) each define four
214 hand-copied ``BANNED_*`` tables. If they drift, froot's own gate and its
215 reviewer disagree about what "deterministic" means — the exact class of
216 decay froot exists to catch, uncaught in froot itself. Pin them equal so the
217 duplication can never silently diverge.
218 """
219 from froot.policy import determinism as brain
221 for name in (
222 "BANNED_CALLS",
223 "BANNED_CALL_MODULES",
224 "BANNED_ATTRS",
225 "BANNED_BUILTINS",
226 ):
227 assert getattr(_KERNEL, name) == getattr(brain, name), name
And the Makefile
One additive target, mirroring the existing start-scan: start-review runs the durable determinism-review loop once per FROOT_REPOS repo. It is added to .PHONY and is a single uv run line — no behavior change to anything else.
start-review: the review loop's one-shot starter, parallel to start-scan.
Makefile · 44 lines
Makefile
⋯ 41 lines hidden (lines 1–41)
1# froot — developer terrain.
2# Short, single-purpose targets: each is one tool, easy to read and approve.
4.PHONY: sync fmt fmt-check lint type test check worker start-scan start-review
6# Install/refresh the dev env (dev tooling + ai + github + otel extras).
7sync:
8 uv sync --extra dev --extra ai --extra github --extra otel
10# Auto-format (ruff formatter) and fix lint where safe.
11fmt:
12 uv run ruff format src tests
13 uv run ruff check --fix src tests
15# Verify formatting without writing (CI-friendly).
16fmt-check:
17 uv run ruff format --check src tests
19# Lint only (no fixes).
20lint:
21 uv run ruff check src tests
23# Strict type check (the DDD safety net; covers src and tests).
24type:
25 uv run mypy
27# Run the test suite with coverage.
28test:
29 uv run pytest
31# The full gate: format check, lint, types, tests. Keep this green.
32check: fmt-check lint type test
34# Run the Temporal worker (needs a reachable Temporal server + env config).
35worker:
36 uv run python -m froot.worker
38# Start the durable scan loop for each FROOT_REPOS repo (one-shot).
39start-scan:
40 uv run python -m froot.scan_starter
42# Start the durable determinism-review loop for each FROOT_REPOS repo (one-shot).
43start-review:
44 uv run python -m froot.review_starter