Overview — the second ring
The determinism kernel (PR #2) is a lexical CI gate: it flags a nondeterministic call written directly inside an @workflow.defn class. This PR adds the transitive ring — what the kernel structurally can't see: a hazard reached through a first-party helper (up to two call levels out), or a risky third-party import at a workflow module's scope.
It's a durable, PR-reactive reviewer loop, built in froot's pure-spine style — a sibling of the scan/bump loops. Two new workflows, a pure call-graph analyzer, one model call, and an advisory PR comment.
The analyzer — matcher + bounded call-graph
policy/determinism.py is froot's own copy of the kernel matcher (check_node — the four banned tables + import resolution), wrapped in a depth-bounded walk. For each workflow class it scans the class body lexically (those hits are the kernel's, kept as lexical), then chases first-party free-function calls out of the class, running the same matcher on each reachable body. A hit there is a confirmed transitive HazardPath.
check_node — the shared banned-call matcher · 427 lines
check_node — the shared banned-call matcher
⋯ 156 lines hidden (lines 1–156)
1"""The transitive determinism analyzer — pure logic over parsed modules.
3This is froot's own embedded copy of the kernel matcher (the
4``scripts/check_determinism.py`` tables + resolution), wrapped in a bounded
5call-graph walk. The kernel finds banned calls *lexically* inside an
6``@workflow.defn`` class; here we additionally chase first-party free-function
7calls **out** of each workflow, up to ``max_depth`` levels, and run the same
8matcher on each reachable body — so an imported helper that hides
9``datetime.now()`` is caught even though it is invisible to the lexical kernel.
11The matcher is deliberately a copy, not an import: the kernel script is vendored
12per-repo (it must run standalone in any monitored repo's CI), so froot keeps its
13own authoritative tables here. The two are kept in sync by the same rule set.
15All functions are pure: the file/AST I/O is done by an adapter
16(:mod:`froot.adapters.source_tree`), which hands this module already-parsed
17:class:`LoadedModule` values. Self-method calls (``self.helper()``) are *not*
18chased — their bodies are lexically inside the workflow class, so the kernel
19already covers them; this loop only follows calls that leave the class.
20"""
22from __future__ import annotations
24import ast
25from collections import deque
26from dataclasses import dataclass
27from typing import TYPE_CHECKING
29from froot.domain.determinism import (
30 AnalysisResult,
31 FrontierItem,
32 HazardPath,
33 Impurity,
34)
36if TYPE_CHECKING:
37 from collections.abc import Iterable, Mapping
39# ── The banned tables (kept in sync with scripts/check_determinism.py) ───────
40BANNED_CALLS: dict[str, str] = {
41 "datetime.datetime.now": "use workflow.now()",
42 "datetime.datetime.utcnow": "use workflow.now()",
43 "datetime.datetime.today": "use workflow.now()",
44 "datetime.date.today": "use workflow.now().date()",
45 "time.time": "use workflow.now()",
46 "time.time_ns": "use workflow.now()",
47 "time.monotonic": "use workflow.now()",
48 "time.monotonic_ns": "use workflow.now()",
49 "time.perf_counter": "use workflow.now()",
50 "time.perf_counter_ns": "use workflow.now()",
51 "time.process_time": "use workflow.now()",
52 "time.sleep": "use workflow.sleep()",
53 "uuid.uuid1": "use workflow.uuid4()",
54 "uuid.uuid3": "use workflow.uuid4()",
55 "uuid.uuid4": "use workflow.uuid4()",
56 "uuid.uuid5": "use workflow.uuid4()",
57 "os.getenv": "read env in an activity and pass it in",
58 "os.urandom": "use workflow.random()",
59 "asyncio.sleep": "use workflow.sleep()",
60 "socket.socket": "do network I/O in an activity",
61}
62BANNED_CALL_MODULES: dict[str, str] = {
63 "random": "use workflow.random()",
64 "threading": "workflows are single-threaded; use workflow APIs",
65 "requests": "do network I/O in an activity",
66 "httpx": "do network I/O in an activity",
67 "subprocess": "run external processes in an activity",
68}
69BANNED_ATTRS: dict[str, str] = {
70 "os.environ": "read env in an activity and pass it in",
71}
72BANNED_BUILTINS: dict[str, str] = {
73 "open": "do file I/O in an activity",
74 "input": "workflows cannot block on input",
75}
77# Third-party roots with no deterministic surface worth importing into a
78# workflow module. Their presence at module scope is the model's frontier: is it
79# actually reached from the workflow, or dead weight?
80_RISKY_THIRD_PARTY: frozenset[str] = frozenset(
81 {
82 "httpx",
83 "requests",
84 "aiohttp",
85 "urllib",
86 "http",
87 "socket",
88 "subprocess",
89 "boto3",
90 "redis",
91 "psycopg",
92 "psycopg2",
93 "pymongo",
94 "sqlalchemy",
95 "smtplib",
96 "ftplib",
97 "paramiko",
98 }
99)
102@dataclass(frozen=True)
103class LoadedModule:
104 """A first-party module parsed by the source-tree adapter."""
106 qualname: str
107 tree: ast.Module
108 lines: tuple[str, ...]
111# ── The matcher (ported verbatim in spirit from the kernel script) ───────────
112def _import_map(tree: ast.Module) -> dict[str, str]:
113 """Map each bound name to the dotted module/symbol path it refers to."""
114 bindings: dict[str, str] = {}
115 for node in ast.walk(tree):
116 if isinstance(node, ast.Import):
117 for alias in node.names:
118 if alias.asname:
119 bindings[alias.asname] = alias.name
120 else:
121 root = alias.name.split(".")[0]
122 bindings[root] = root
123 elif isinstance(node, ast.ImportFrom):
124 # Relative imports (``from . import x`` / ``from ..pkg import x``)
125 # are out of scope: a bare/level path can't be resolved to a loaded
126 # module, and a bogus binding could collide with a real one.
127 if node.module is None or node.level > 0:
128 continue
129 for alias in node.names:
130 bound = alias.asname or alias.name
131 bindings[bound] = f"{node.module}.{alias.name}"
132 return bindings
135def _dotted(node: ast.AST) -> str | None:
136 """The literal dotted path of a Name/Attribute chain, or None."""
137 parts: list[str] = []
138 cur: ast.AST = node
139 while isinstance(cur, ast.Attribute):
140 parts.append(cur.attr)
141 cur = cur.value
142 if not isinstance(cur, ast.Name):
143 return None
144 parts.append(cur.id)
145 return ".".join(reversed(parts))
148def _resolve(node: ast.AST, imports: Mapping[str, str]) -> str | None:
149 """Dotted path with the root name substituted via the import map."""
150 dotted = _dotted(node)
151 if dotted is None:
152 return None
153 root, _, rest = dotted.partition(".")
154 if root not in imports:
155 return None
156 base = imports[root]
157 return f"{base}.{rest}" if rest else base
160def check_node(
161 node: ast.expr, imports: Mapping[str, str]
162) -> tuple[str, str] | None:
163 """Return ``(rule, hint)`` if ``node`` is a banned call/attr, else None."""
164 if isinstance(node, ast.Call):
165 func = node.func
166 if (
167 isinstance(func, ast.Name)
168 and func.id in BANNED_BUILTINS
169 and func.id not in imports
170 ):
171 return func.id, BANNED_BUILTINS[func.id]
172 resolved = _resolve(func, imports)
173 if resolved is None:
174 return None
175 if resolved in BANNED_CALLS:
176 return resolved, BANNED_CALLS[resolved]
177 if resolved.split(".")[0] in BANNED_CALL_MODULES:
178 return resolved, BANNED_CALL_MODULES[resolved.split(".")[0]]
179 return None
180 if isinstance(node, ast.Attribute):
181 resolved = _resolve(node, imports)
182 if resolved in BANNED_ATTRS:
183 return resolved, BANNED_ATTRS[resolved]
184 return None
⋯ 241 lines hidden (lines 187–427)
187def _is_workflow_class(node: ast.ClassDef, imports: Mapping[str, str]) -> bool:
188 """True if the class carries an ``@workflow.defn`` decorator."""
189 for dec in node.decorator_list:
190 target = dec.func if isinstance(dec, ast.Call) else dec
191 resolved = _resolve(target, imports) or _dotted(target)
192 if resolved and resolved.endswith("workflow.defn"):
193 return True
194 return False
197# ── The call graph ───────────────────────────────────────────────────────────
198def _module_functions(
199 tree: ast.Module,
200) -> dict[str, ast.FunctionDef | ast.AsyncFunctionDef]:
201 """Index the module's top-level free functions by name."""
202 out: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {}
203 for node in tree.body:
204 if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
205 out[node.name] = node
206 return out
209def _split_qual(qual: str) -> tuple[str, str] | None:
210 """Split ``a.b.c`` into module ``a.b`` and symbol ``c`` (None if no dot)."""
211 module, _, symbol = qual.rpartition(".")
212 if not module or not symbol:
213 return None
214 return module, symbol
217def _scan_body(
218 node: ast.AST, imports: Mapping[str, str], module: str
219) -> list[Impurity]:
220 """Find every banned call lexically inside ``node`` (a def or class)."""
221 found: list[Impurity] = []
222 seen: set[tuple[int, str]] = set()
223 for child in ast.walk(node):
224 if not isinstance(child, ast.Call | ast.Attribute):
225 continue
226 hit = check_node(child, imports)
227 if hit is None:
228 continue
229 rule, hint = hit
230 key = (child.lineno, rule)
231 if key in seen:
232 continue
233 seen.add(key)
234 found.append(
235 Impurity(rule=rule, hint=hint, module=module, line=child.lineno)
236 )
237 return found
240def _first_party_callees(
241 node: ast.AST,
242 imports: Mapping[str, str],
243 modules: Mapping[str, LoadedModule],
244 functions: Mapping[str, Mapping[str, object]],
245 current: str,
246) -> list[tuple[str, str, str]]:
247 """First-party functions ``node`` calls, as ``(module, symbol, label)``.
249 Only calls that *leave* the current scope to a known first-party free
250 function are returned; ``self.method()`` and stdlib/third-party calls are
251 not (the former is the kernel's lexical territory, the latter is not a
252 first-party edge to chase).
253 """
254 edges: list[tuple[str, str, str]] = []
255 seen: set[tuple[str, str]] = set()
256 for child in ast.walk(node):
257 if not isinstance(child, ast.Call):
258 continue
259 func = child.func
260 target: tuple[str, str] | None = None
261 if isinstance(func, ast.Name):
262 bound = imports.get(func.id)
263 if bound is not None:
264 target = _split_qual(bound)
265 elif func.id in functions.get(current, {}):
266 target = (current, func.id)
267 elif isinstance(func, ast.Attribute):
268 resolved = _resolve(func, imports)
269 if resolved is not None:
270 target = _split_qual(resolved)
271 if target is None:
272 continue
273 module, symbol = target
274 if module not in modules or symbol not in functions.get(module, {}):
275 continue
276 if (module, symbol) in seen:
277 continue
278 seen.add((module, symbol))
279 edges.append((module, symbol, symbol))
280 return edges
283def _risky_imports(
284 tree: ast.Module, first_party: frozenset[str]
285) -> list[tuple[int, str]]:
286 """Module-scope risky third-party imports as ``(line, dotted_root)``."""
287 out: list[tuple[int, str]] = []
288 seen: set[tuple[int, str]] = set()
289 for node in ast.walk(tree):
290 if isinstance(node, ast.Import):
291 names = [alias.name for alias in node.names]
292 elif (
293 isinstance(node, ast.ImportFrom)
294 and node.module is not None
295 and node.level == 0 # absolute only; a relative tail isn't a module
296 ):
297 names = [node.module]
298 else:
299 continue
300 for name in names:
301 root = name.split(".")[0]
302 if root in _RISKY_THIRD_PARTY and root not in first_party:
303 key = (node.lineno, name)
304 if key not in seen:
305 seen.add(key)
306 out.append((node.lineno, name))
307 return out
310def _line(lines: tuple[str, ...], lineno: int) -> str:
311 """The stripped source at ``lineno`` (1-based), or empty if out of range."""
312 return lines[lineno - 1].strip() if 0 < lineno <= len(lines) else ""
315def analyze_workflow_surface(
316 modules: Mapping[str, LoadedModule], *, max_depth: int = 2
317) -> AnalysisResult:
318 """Find transitive determinism hazards across a repo's workflow surface.
320 Args:
321 modules: Every first-party module, parsed, keyed by dotted qualname.
322 max_depth: How many first-party call levels to chase out of each
323 workflow method (2 = the helper a workflow calls, and the helper
324 *that* calls).
326 Returns:
327 The lexical hits (the kernel's set), the confirmed transitive hazards,
328 and the ambiguous frontier (risky third-party imports) for the model.
329 """
330 functions = {qn: _module_functions(lm.tree) for qn, lm in modules.items()}
331 first_party = frozenset(qn.split(".")[0] for qn in modules)
333 workflows: list[tuple[str, ast.ClassDef, dict[str, str]]] = []
334 for qn, lm in modules.items():
335 imports = _import_map(lm.tree)
336 for node in ast.walk(lm.tree):
337 if isinstance(node, ast.ClassDef) and _is_workflow_class(
338 node, imports
339 ):
340 workflows.append((qn, node, imports))
342 lexical: list[Impurity] = []
343 hazards: list[HazardPath] = []
344 for qn, cls, imports in workflows:
345 label = f"{qn}:{cls.name}"
346 lexical.extend(_scan_body(cls, imports, qn))
347 seen: set[tuple[str, str]] = set()
348 queue: deque[tuple[tuple[str, str], tuple[str, ...], int]] = deque(
349 ((m, s), (lbl,), 1)
350 for m, s, lbl in _first_party_callees(
351 cls, imports, modules, functions, qn
352 )
353 )
354 while queue:
355 (mod, sym), via, depth = queue.popleft()
356 if (mod, sym) in seen:
357 continue
358 seen.add((mod, sym))
359 fn = functions[mod].get(sym)
360 if fn is None:
361 continue
362 fn_imports = _import_map(modules[mod].tree)
363 for imp in _scan_body(fn, fn_imports, mod):
364 hazards.append(
365 HazardPath(workflow=label, via=via, impurity=imp)
366 )
367 if depth < max_depth:
368 for m2, s2, lbl2 in _first_party_callees(
369 fn, fn_imports, modules, functions, mod
370 ):
371 if (m2, s2) not in seen:
372 queue.append(((m2, s2), (*via, lbl2), depth + 1))
374 frontier: list[FrontierItem] = []
375 wf_class_of: dict[str, str] = {}
376 for qn, cls, _ in workflows:
377 wf_class_of.setdefault(qn, cls.name)
378 for qn in sorted(wf_class_of):
379 lm = modules[qn]
380 for lineno, root in _risky_imports(lm.tree, first_party):
381 frontier.append(
382 FrontierItem(
383 kind="third_party_import",
384 workflow=f"{qn}:{wf_class_of[qn]}",
385 module=qn,
386 line=lineno,
387 symbol=root,
388 snippet=_line(lm.lines, lineno),
389 )
390 )
392 return AnalysisResult(
393 lexical=tuple(_dedupe_impurities(lexical)),
394 hazards=tuple(_dedupe_hazards(hazards)),
395 frontier=tuple(frontier),
396 )
399def _dedupe_impurities(items: Iterable[Impurity]) -> list[Impurity]:
400 seen: set[tuple[str, int, str]] = set()
401 out: list[Impurity] = []
402 for i in sorted(items, key=lambda x: (x.module, x.line, x.rule)):
403 key = (i.module, i.line, i.rule)
404 if key not in seen:
405 seen.add(key)
406 out.append(i)
407 return out
410def _dedupe_hazards(items: Iterable[HazardPath]) -> list[HazardPath]:
411 seen: set[tuple[str, str, int, str]] = set()
412 out: list[HazardPath] = []
413 ordered = sorted(
414 items,
415 key=lambda h: (
416 h.workflow,
417 h.impurity.module,
418 h.impurity.line,
419 h.impurity.rule,
420 ),
421 )
422 for h in ordered:
423 key = (h.workflow, h.impurity.module, h.impurity.line, h.impurity.rule)
424 if key not in seen:
425 seen.add(key)
426 out.append(h)
427 return out
analyze_workflow_surface — the bounded BFS · 427 lines
analyze_workflow_surface — the bounded BFS
⋯ 314 lines hidden (lines 1–314)
1"""The transitive determinism analyzer — pure logic over parsed modules.
3This is froot's own embedded copy of the kernel matcher (the
4``scripts/check_determinism.py`` tables + resolution), wrapped in a bounded
5call-graph walk. The kernel finds banned calls *lexically* inside an
6``@workflow.defn`` class; here we additionally chase first-party free-function
7calls **out** of each workflow, up to ``max_depth`` levels, and run the same
8matcher on each reachable body — so an imported helper that hides
9``datetime.now()`` is caught even though it is invisible to the lexical kernel.
11The matcher is deliberately a copy, not an import: the kernel script is vendored
12per-repo (it must run standalone in any monitored repo's CI), so froot keeps its
13own authoritative tables here. The two are kept in sync by the same rule set.
15All functions are pure: the file/AST I/O is done by an adapter
16(:mod:`froot.adapters.source_tree`), which hands this module already-parsed
17:class:`LoadedModule` values. Self-method calls (``self.helper()``) are *not*
18chased — their bodies are lexically inside the workflow class, so the kernel
19already covers them; this loop only follows calls that leave the class.
20"""
22from __future__ import annotations
24import ast
25from collections import deque
26from dataclasses import dataclass
27from typing import TYPE_CHECKING
29from froot.domain.determinism import (
30 AnalysisResult,
31 FrontierItem,
32 HazardPath,
33 Impurity,
34)
36if TYPE_CHECKING:
37 from collections.abc import Iterable, Mapping
39# ── The banned tables (kept in sync with scripts/check_determinism.py) ───────
40BANNED_CALLS: dict[str, str] = {
41 "datetime.datetime.now": "use workflow.now()",
42 "datetime.datetime.utcnow": "use workflow.now()",
43 "datetime.datetime.today": "use workflow.now()",
44 "datetime.date.today": "use workflow.now().date()",
45 "time.time": "use workflow.now()",
46 "time.time_ns": "use workflow.now()",
47 "time.monotonic": "use workflow.now()",
48 "time.monotonic_ns": "use workflow.now()",
49 "time.perf_counter": "use workflow.now()",
50 "time.perf_counter_ns": "use workflow.now()",
51 "time.process_time": "use workflow.now()",
52 "time.sleep": "use workflow.sleep()",
53 "uuid.uuid1": "use workflow.uuid4()",
54 "uuid.uuid3": "use workflow.uuid4()",
55 "uuid.uuid4": "use workflow.uuid4()",
56 "uuid.uuid5": "use workflow.uuid4()",
57 "os.getenv": "read env in an activity and pass it in",
58 "os.urandom": "use workflow.random()",
59 "asyncio.sleep": "use workflow.sleep()",
60 "socket.socket": "do network I/O in an activity",
61}
62BANNED_CALL_MODULES: dict[str, str] = {
63 "random": "use workflow.random()",
64 "threading": "workflows are single-threaded; use workflow APIs",
65 "requests": "do network I/O in an activity",
66 "httpx": "do network I/O in an activity",
67 "subprocess": "run external processes in an activity",
68}
69BANNED_ATTRS: dict[str, str] = {
70 "os.environ": "read env in an activity and pass it in",
71}
72BANNED_BUILTINS: dict[str, str] = {
73 "open": "do file I/O in an activity",
74 "input": "workflows cannot block on input",
75}
77# Third-party roots with no deterministic surface worth importing into a
78# workflow module. Their presence at module scope is the model's frontier: is it
79# actually reached from the workflow, or dead weight?
80_RISKY_THIRD_PARTY: frozenset[str] = frozenset(
81 {
82 "httpx",
83 "requests",
84 "aiohttp",
85 "urllib",
86 "http",
87 "socket",
88 "subprocess",
89 "boto3",
90 "redis",
91 "psycopg",
92 "psycopg2",
93 "pymongo",
94 "sqlalchemy",
95 "smtplib",
96 "ftplib",
97 "paramiko",
98 }
99)
102@dataclass(frozen=True)
103class LoadedModule:
104 """A first-party module parsed by the source-tree adapter."""
106 qualname: str
107 tree: ast.Module
108 lines: tuple[str, ...]
111# ── The matcher (ported verbatim in spirit from the kernel script) ───────────
112def _import_map(tree: ast.Module) -> dict[str, str]:
113 """Map each bound name to the dotted module/symbol path it refers to."""
114 bindings: dict[str, str] = {}
115 for node in ast.walk(tree):
116 if isinstance(node, ast.Import):
117 for alias in node.names:
118 if alias.asname:
119 bindings[alias.asname] = alias.name
120 else:
121 root = alias.name.split(".")[0]
122 bindings[root] = root
123 elif isinstance(node, ast.ImportFrom):
124 # Relative imports (``from . import x`` / ``from ..pkg import x``)
125 # are out of scope: a bare/level path can't be resolved to a loaded
126 # module, and a bogus binding could collide with a real one.
127 if node.module is None or node.level > 0:
128 continue
129 for alias in node.names:
130 bound = alias.asname or alias.name
131 bindings[bound] = f"{node.module}.{alias.name}"
132 return bindings
135def _dotted(node: ast.AST) -> str | None:
136 """The literal dotted path of a Name/Attribute chain, or None."""
137 parts: list[str] = []
138 cur: ast.AST = node
139 while isinstance(cur, ast.Attribute):
140 parts.append(cur.attr)
141 cur = cur.value
142 if not isinstance(cur, ast.Name):
143 return None
144 parts.append(cur.id)
145 return ".".join(reversed(parts))
148def _resolve(node: ast.AST, imports: Mapping[str, str]) -> str | None:
149 """Dotted path with the root name substituted via the import map."""
150 dotted = _dotted(node)
151 if dotted is None:
152 return None
153 root, _, rest = dotted.partition(".")
154 if root not in imports:
155 return None
156 base = imports[root]
157 return f"{base}.{rest}" if rest else base
160def check_node(
161 node: ast.expr, imports: Mapping[str, str]
162) -> tuple[str, str] | None:
163 """Return ``(rule, hint)`` if ``node`` is a banned call/attr, else None."""
164 if isinstance(node, ast.Call):
165 func = node.func
166 if (
167 isinstance(func, ast.Name)
168 and func.id in BANNED_BUILTINS
169 and func.id not in imports
170 ):
171 return func.id, BANNED_BUILTINS[func.id]
172 resolved = _resolve(func, imports)
173 if resolved is None:
174 return None
175 if resolved in BANNED_CALLS:
176 return resolved, BANNED_CALLS[resolved]
177 if resolved.split(".")[0] in BANNED_CALL_MODULES:
178 return resolved, BANNED_CALL_MODULES[resolved.split(".")[0]]
179 return None
180 if isinstance(node, ast.Attribute):
181 resolved = _resolve(node, imports)
182 if resolved in BANNED_ATTRS:
183 return resolved, BANNED_ATTRS[resolved]
184 return None
187def _is_workflow_class(node: ast.ClassDef, imports: Mapping[str, str]) -> bool:
188 """True if the class carries an ``@workflow.defn`` decorator."""
189 for dec in node.decorator_list:
190 target = dec.func if isinstance(dec, ast.Call) else dec
191 resolved = _resolve(target, imports) or _dotted(target)
192 if resolved and resolved.endswith("workflow.defn"):
193 return True
194 return False
197# ── The call graph ───────────────────────────────────────────────────────────
198def _module_functions(
199 tree: ast.Module,
200) -> dict[str, ast.FunctionDef | ast.AsyncFunctionDef]:
201 """Index the module's top-level free functions by name."""
202 out: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {}
203 for node in tree.body:
204 if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
205 out[node.name] = node
206 return out
209def _split_qual(qual: str) -> tuple[str, str] | None:
210 """Split ``a.b.c`` into module ``a.b`` and symbol ``c`` (None if no dot)."""
211 module, _, symbol = qual.rpartition(".")
212 if not module or not symbol:
213 return None
214 return module, symbol
217def _scan_body(
218 node: ast.AST, imports: Mapping[str, str], module: str
219) -> list[Impurity]:
220 """Find every banned call lexically inside ``node`` (a def or class)."""
221 found: list[Impurity] = []
222 seen: set[tuple[int, str]] = set()
223 for child in ast.walk(node):
224 if not isinstance(child, ast.Call | ast.Attribute):
225 continue
226 hit = check_node(child, imports)
227 if hit is None:
228 continue
229 rule, hint = hit
230 key = (child.lineno, rule)
231 if key in seen:
232 continue
233 seen.add(key)
234 found.append(
235 Impurity(rule=rule, hint=hint, module=module, line=child.lineno)
236 )
237 return found
240def _first_party_callees(
241 node: ast.AST,
242 imports: Mapping[str, str],
243 modules: Mapping[str, LoadedModule],
244 functions: Mapping[str, Mapping[str, object]],
245 current: str,
246) -> list[tuple[str, str, str]]:
247 """First-party functions ``node`` calls, as ``(module, symbol, label)``.
249 Only calls that *leave* the current scope to a known first-party free
250 function are returned; ``self.method()`` and stdlib/third-party calls are
251 not (the former is the kernel's lexical territory, the latter is not a
252 first-party edge to chase).
253 """
254 edges: list[tuple[str, str, str]] = []
255 seen: set[tuple[str, str]] = set()
256 for child in ast.walk(node):
257 if not isinstance(child, ast.Call):
258 continue
259 func = child.func
260 target: tuple[str, str] | None = None
261 if isinstance(func, ast.Name):
262 bound = imports.get(func.id)
263 if bound is not None:
264 target = _split_qual(bound)
265 elif func.id in functions.get(current, {}):
266 target = (current, func.id)
267 elif isinstance(func, ast.Attribute):
268 resolved = _resolve(func, imports)
269 if resolved is not None:
270 target = _split_qual(resolved)
271 if target is None:
272 continue
273 module, symbol = target
274 if module not in modules or symbol not in functions.get(module, {}):
275 continue
276 if (module, symbol) in seen:
277 continue
278 seen.add((module, symbol))
279 edges.append((module, symbol, symbol))
280 return edges
283def _risky_imports(
284 tree: ast.Module, first_party: frozenset[str]
285) -> list[tuple[int, str]]:
286 """Module-scope risky third-party imports as ``(line, dotted_root)``."""
287 out: list[tuple[int, str]] = []
288 seen: set[tuple[int, str]] = set()
289 for node in ast.walk(tree):
290 if isinstance(node, ast.Import):
291 names = [alias.name for alias in node.names]
292 elif (
293 isinstance(node, ast.ImportFrom)
294 and node.module is not None
295 and node.level == 0 # absolute only; a relative tail isn't a module
296 ):
297 names = [node.module]
298 else:
299 continue
300 for name in names:
301 root = name.split(".")[0]
302 if root in _RISKY_THIRD_PARTY and root not in first_party:
303 key = (node.lineno, name)
304 if key not in seen:
305 seen.add(key)
306 out.append((node.lineno, name))
307 return out
310def _line(lines: tuple[str, ...], lineno: int) -> str:
311 """The stripped source at ``lineno`` (1-based), or empty if out of range."""
312 return lines[lineno - 1].strip() if 0 < lineno <= len(lines) else ""
315def analyze_workflow_surface(
316 modules: Mapping[str, LoadedModule], *, max_depth: int = 2
317) -> AnalysisResult:
318 """Find transitive determinism hazards across a repo's workflow surface.
320 Args:
321 modules: Every first-party module, parsed, keyed by dotted qualname.
322 max_depth: How many first-party call levels to chase out of each
323 workflow method (2 = the helper a workflow calls, and the helper
324 *that* calls).
326 Returns:
327 The lexical hits (the kernel's set), the confirmed transitive hazards,
328 and the ambiguous frontier (risky third-party imports) for the model.
329 """
330 functions = {qn: _module_functions(lm.tree) for qn, lm in modules.items()}
331 first_party = frozenset(qn.split(".")[0] for qn in modules)
333 workflows: list[tuple[str, ast.ClassDef, dict[str, str]]] = []
334 for qn, lm in modules.items():
335 imports = _import_map(lm.tree)
336 for node in ast.walk(lm.tree):
337 if isinstance(node, ast.ClassDef) and _is_workflow_class(
338 node, imports
339 ):
340 workflows.append((qn, node, imports))
342 lexical: list[Impurity] = []
343 hazards: list[HazardPath] = []
344 for qn, cls, imports in workflows:
345 label = f"{qn}:{cls.name}"
346 lexical.extend(_scan_body(cls, imports, qn))
347 seen: set[tuple[str, str]] = set()
348 queue: deque[tuple[tuple[str, str], tuple[str, ...], int]] = deque(
349 ((m, s), (lbl,), 1)
350 for m, s, lbl in _first_party_callees(
351 cls, imports, modules, functions, qn
352 )
353 )
354 while queue:
355 (mod, sym), via, depth = queue.popleft()
356 if (mod, sym) in seen:
357 continue
358 seen.add((mod, sym))
359 fn = functions[mod].get(sym)
360 if fn is None:
361 continue
362 fn_imports = _import_map(modules[mod].tree)
363 for imp in _scan_body(fn, fn_imports, mod):
364 hazards.append(
365 HazardPath(workflow=label, via=via, impurity=imp)
366 )
367 if depth < max_depth:
368 for m2, s2, lbl2 in _first_party_callees(
369 fn, fn_imports, modules, functions, mod
370 ):
371 if (m2, s2) not in seen:
372 queue.append(((m2, s2), (*via, lbl2), depth + 1))
374 frontier: list[FrontierItem] = []
375 wf_class_of: dict[str, str] = {}
376 for qn, cls, _ in workflows:
377 wf_class_of.setdefault(qn, cls.name)
378 for qn in sorted(wf_class_of):
379 lm = modules[qn]
380 for lineno, root in _risky_imports(lm.tree, first_party):
381 frontier.append(
382 FrontierItem(
383 kind="third_party_import",
384 workflow=f"{qn}:{wf_class_of[qn]}",
385 module=qn,
386 line=lineno,
387 symbol=root,
388 snippet=_line(lm.lines, lineno),
389 )
390 )
392 return AnalysisResult(
393 lexical=tuple(_dedupe_impurities(lexical)),
394 hazards=tuple(_dedupe_hazards(hazards)),
395 frontier=tuple(frontier),
396 )
⋯ 30 lines hidden (lines 398–427)
399def _dedupe_impurities(items: Iterable[Impurity]) -> list[Impurity]:
400 seen: set[tuple[str, int, str]] = set()
401 out: list[Impurity] = []
402 for i in sorted(items, key=lambda x: (x.module, x.line, x.rule)):
403 key = (i.module, i.line, i.rule)
404 if key not in seen:
405 seen.add(key)
406 out.append(i)
407 return out
410def _dedupe_hazards(items: Iterable[HazardPath]) -> list[HazardPath]:
411 seen: set[tuple[str, str, int, str]] = set()
412 out: list[HazardPath] = []
413 ordered = sorted(
414 items,
415 key=lambda h: (
416 h.workflow,
417 h.impurity.module,
418 h.impurity.line,
419 h.impurity.rule,
420 ),
421 )
422 for h in ordered:
423 key = (h.workflow, h.impurity.module, h.impurity.line, h.impurity.rule)
424 if key not in seen:
425 seen.add(key)
426 out.append(h)
427 return out
The edge resolver is the precision lever
_first_party_callees decides what counts as an edge to chase: a call whose callee resolves — through the module's import map — to a known first-party free function. A Name via a from x import f, a same-module function, or a mod.func attribute all resolve; everything else is dropped.
_first_party_callees — resolve, then keep only first-party defs · 427 lines
_first_party_callees — resolve, then keep only first-party defs
⋯ 239 lines hidden (lines 1–239)
1"""The transitive determinism analyzer — pure logic over parsed modules.
3This is froot's own embedded copy of the kernel matcher (the
4``scripts/check_determinism.py`` tables + resolution), wrapped in a bounded
5call-graph walk. The kernel finds banned calls *lexically* inside an
6``@workflow.defn`` class; here we additionally chase first-party free-function
7calls **out** of each workflow, up to ``max_depth`` levels, and run the same
8matcher on each reachable body — so an imported helper that hides
9``datetime.now()`` is caught even though it is invisible to the lexical kernel.
11The matcher is deliberately a copy, not an import: the kernel script is vendored
12per-repo (it must run standalone in any monitored repo's CI), so froot keeps its
13own authoritative tables here. The two are kept in sync by the same rule set.
15All functions are pure: the file/AST I/O is done by an adapter
16(:mod:`froot.adapters.source_tree`), which hands this module already-parsed
17:class:`LoadedModule` values. Self-method calls (``self.helper()``) are *not*
18chased — their bodies are lexically inside the workflow class, so the kernel
19already covers them; this loop only follows calls that leave the class.
20"""
22from __future__ import annotations
24import ast
25from collections import deque
26from dataclasses import dataclass
27from typing import TYPE_CHECKING
29from froot.domain.determinism import (
30 AnalysisResult,
31 FrontierItem,
32 HazardPath,
33 Impurity,
34)
36if TYPE_CHECKING:
37 from collections.abc import Iterable, Mapping
39# ── The banned tables (kept in sync with scripts/check_determinism.py) ───────
40BANNED_CALLS: dict[str, str] = {
41 "datetime.datetime.now": "use workflow.now()",
42 "datetime.datetime.utcnow": "use workflow.now()",
43 "datetime.datetime.today": "use workflow.now()",
44 "datetime.date.today": "use workflow.now().date()",
45 "time.time": "use workflow.now()",
46 "time.time_ns": "use workflow.now()",
47 "time.monotonic": "use workflow.now()",
48 "time.monotonic_ns": "use workflow.now()",
49 "time.perf_counter": "use workflow.now()",
50 "time.perf_counter_ns": "use workflow.now()",
51 "time.process_time": "use workflow.now()",
52 "time.sleep": "use workflow.sleep()",
53 "uuid.uuid1": "use workflow.uuid4()",
54 "uuid.uuid3": "use workflow.uuid4()",
55 "uuid.uuid4": "use workflow.uuid4()",
56 "uuid.uuid5": "use workflow.uuid4()",
57 "os.getenv": "read env in an activity and pass it in",
58 "os.urandom": "use workflow.random()",
59 "asyncio.sleep": "use workflow.sleep()",
60 "socket.socket": "do network I/O in an activity",
61}
62BANNED_CALL_MODULES: dict[str, str] = {
63 "random": "use workflow.random()",
64 "threading": "workflows are single-threaded; use workflow APIs",
65 "requests": "do network I/O in an activity",
66 "httpx": "do network I/O in an activity",
67 "subprocess": "run external processes in an activity",
68}
69BANNED_ATTRS: dict[str, str] = {
70 "os.environ": "read env in an activity and pass it in",
71}
72BANNED_BUILTINS: dict[str, str] = {
73 "open": "do file I/O in an activity",
74 "input": "workflows cannot block on input",
75}
77# Third-party roots with no deterministic surface worth importing into a
78# workflow module. Their presence at module scope is the model's frontier: is it
79# actually reached from the workflow, or dead weight?
80_RISKY_THIRD_PARTY: frozenset[str] = frozenset(
81 {
82 "httpx",
83 "requests",
84 "aiohttp",
85 "urllib",
86 "http",
87 "socket",
88 "subprocess",
89 "boto3",
90 "redis",
91 "psycopg",
92 "psycopg2",
93 "pymongo",
94 "sqlalchemy",
95 "smtplib",
96 "ftplib",
97 "paramiko",
98 }
99)
102@dataclass(frozen=True)
103class LoadedModule:
104 """A first-party module parsed by the source-tree adapter."""
106 qualname: str
107 tree: ast.Module
108 lines: tuple[str, ...]
111# ── The matcher (ported verbatim in spirit from the kernel script) ───────────
112def _import_map(tree: ast.Module) -> dict[str, str]:
113 """Map each bound name to the dotted module/symbol path it refers to."""
114 bindings: dict[str, str] = {}
115 for node in ast.walk(tree):
116 if isinstance(node, ast.Import):
117 for alias in node.names:
118 if alias.asname:
119 bindings[alias.asname] = alias.name
120 else:
121 root = alias.name.split(".")[0]
122 bindings[root] = root
123 elif isinstance(node, ast.ImportFrom):
124 # Relative imports (``from . import x`` / ``from ..pkg import x``)
125 # are out of scope: a bare/level path can't be resolved to a loaded
126 # module, and a bogus binding could collide with a real one.
127 if node.module is None or node.level > 0:
128 continue
129 for alias in node.names:
130 bound = alias.asname or alias.name
131 bindings[bound] = f"{node.module}.{alias.name}"
132 return bindings
135def _dotted(node: ast.AST) -> str | None:
136 """The literal dotted path of a Name/Attribute chain, or None."""
137 parts: list[str] = []
138 cur: ast.AST = node
139 while isinstance(cur, ast.Attribute):
140 parts.append(cur.attr)
141 cur = cur.value
142 if not isinstance(cur, ast.Name):
143 return None
144 parts.append(cur.id)
145 return ".".join(reversed(parts))
148def _resolve(node: ast.AST, imports: Mapping[str, str]) -> str | None:
149 """Dotted path with the root name substituted via the import map."""
150 dotted = _dotted(node)
151 if dotted is None:
152 return None
153 root, _, rest = dotted.partition(".")
154 if root not in imports:
155 return None
156 base = imports[root]
157 return f"{base}.{rest}" if rest else base
160def check_node(
161 node: ast.expr, imports: Mapping[str, str]
162) -> tuple[str, str] | None:
163 """Return ``(rule, hint)`` if ``node`` is a banned call/attr, else None."""
164 if isinstance(node, ast.Call):
165 func = node.func
166 if (
167 isinstance(func, ast.Name)
168 and func.id in BANNED_BUILTINS
169 and func.id not in imports
170 ):
171 return func.id, BANNED_BUILTINS[func.id]
172 resolved = _resolve(func, imports)
173 if resolved is None:
174 return None
175 if resolved in BANNED_CALLS:
176 return resolved, BANNED_CALLS[resolved]
177 if resolved.split(".")[0] in BANNED_CALL_MODULES:
178 return resolved, BANNED_CALL_MODULES[resolved.split(".")[0]]
179 return None
180 if isinstance(node, ast.Attribute):
181 resolved = _resolve(node, imports)
182 if resolved in BANNED_ATTRS:
183 return resolved, BANNED_ATTRS[resolved]
184 return None
187def _is_workflow_class(node: ast.ClassDef, imports: Mapping[str, str]) -> bool:
188 """True if the class carries an ``@workflow.defn`` decorator."""
189 for dec in node.decorator_list:
190 target = dec.func if isinstance(dec, ast.Call) else dec
191 resolved = _resolve(target, imports) or _dotted(target)
192 if resolved and resolved.endswith("workflow.defn"):
193 return True
194 return False
197# ── The call graph ───────────────────────────────────────────────────────────
198def _module_functions(
199 tree: ast.Module,
200) -> dict[str, ast.FunctionDef | ast.AsyncFunctionDef]:
201 """Index the module's top-level free functions by name."""
202 out: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {}
203 for node in tree.body:
204 if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
205 out[node.name] = node
206 return out
209def _split_qual(qual: str) -> tuple[str, str] | None:
210 """Split ``a.b.c`` into module ``a.b`` and symbol ``c`` (None if no dot)."""
211 module, _, symbol = qual.rpartition(".")
212 if not module or not symbol:
213 return None
214 return module, symbol
217def _scan_body(
218 node: ast.AST, imports: Mapping[str, str], module: str
219) -> list[Impurity]:
220 """Find every banned call lexically inside ``node`` (a def or class)."""
221 found: list[Impurity] = []
222 seen: set[tuple[int, str]] = set()
223 for child in ast.walk(node):
224 if not isinstance(child, ast.Call | ast.Attribute):
225 continue
226 hit = check_node(child, imports)
227 if hit is None:
228 continue
229 rule, hint = hit
230 key = (child.lineno, rule)
231 if key in seen:
232 continue
233 seen.add(key)
234 found.append(
235 Impurity(rule=rule, hint=hint, module=module, line=child.lineno)
236 )
237 return found
240def _first_party_callees(
241 node: ast.AST,
242 imports: Mapping[str, str],
243 modules: Mapping[str, LoadedModule],
244 functions: Mapping[str, Mapping[str, object]],
245 current: str,
246) -> list[tuple[str, str, str]]:
247 """First-party functions ``node`` calls, as ``(module, symbol, label)``.
249 Only calls that *leave* the current scope to a known first-party free
250 function are returned; ``self.method()`` and stdlib/third-party calls are
251 not (the former is the kernel's lexical territory, the latter is not a
252 first-party edge to chase).
253 """
254 edges: list[tuple[str, str, str]] = []
255 seen: set[tuple[str, str]] = set()
256 for child in ast.walk(node):
257 if not isinstance(child, ast.Call):
258 continue
259 func = child.func
260 target: tuple[str, str] | None = None
261 if isinstance(func, ast.Name):
262 bound = imports.get(func.id)
263 if bound is not None:
264 target = _split_qual(bound)
265 elif func.id in functions.get(current, {}):
266 target = (current, func.id)
267 elif isinstance(func, ast.Attribute):
268 resolved = _resolve(func, imports)
269 if resolved is not None:
270 target = _split_qual(resolved)
271 if target is None:
272 continue
273 module, symbol = target
274 if module not in modules or symbol not in functions.get(module, {}):
275 continue
276 if (module, symbol) in seen:
277 continue
278 seen.add((module, symbol))
279 edges.append((module, symbol, symbol))
280 return edges
⋯ 146 lines hidden (lines 282–427)
283def _risky_imports(
284 tree: ast.Module, first_party: frozenset[str]
285) -> list[tuple[int, str]]:
286 """Module-scope risky third-party imports as ``(line, dotted_root)``."""
287 out: list[tuple[int, str]] = []
288 seen: set[tuple[int, str]] = set()
289 for node in ast.walk(tree):
290 if isinstance(node, ast.Import):
291 names = [alias.name for alias in node.names]
292 elif (
293 isinstance(node, ast.ImportFrom)
294 and node.module is not None
295 and node.level == 0 # absolute only; a relative tail isn't a module
296 ):
297 names = [node.module]
298 else:
299 continue
300 for name in names:
301 root = name.split(".")[0]
302 if root in _RISKY_THIRD_PARTY and root not in first_party:
303 key = (node.lineno, name)
304 if key not in seen:
305 seen.add(key)
306 out.append((node.lineno, name))
307 return out
310def _line(lines: tuple[str, ...], lineno: int) -> str:
311 """The stripped source at ``lineno`` (1-based), or empty if out of range."""
312 return lines[lineno - 1].strip() if 0 < lineno <= len(lines) else ""
315def analyze_workflow_surface(
316 modules: Mapping[str, LoadedModule], *, max_depth: int = 2
317) -> AnalysisResult:
318 """Find transitive determinism hazards across a repo's workflow surface.
320 Args:
321 modules: Every first-party module, parsed, keyed by dotted qualname.
322 max_depth: How many first-party call levels to chase out of each
323 workflow method (2 = the helper a workflow calls, and the helper
324 *that* calls).
326 Returns:
327 The lexical hits (the kernel's set), the confirmed transitive hazards,
328 and the ambiguous frontier (risky third-party imports) for the model.
329 """
330 functions = {qn: _module_functions(lm.tree) for qn, lm in modules.items()}
331 first_party = frozenset(qn.split(".")[0] for qn in modules)
333 workflows: list[tuple[str, ast.ClassDef, dict[str, str]]] = []
334 for qn, lm in modules.items():
335 imports = _import_map(lm.tree)
336 for node in ast.walk(lm.tree):
337 if isinstance(node, ast.ClassDef) and _is_workflow_class(
338 node, imports
339 ):
340 workflows.append((qn, node, imports))
342 lexical: list[Impurity] = []
343 hazards: list[HazardPath] = []
344 for qn, cls, imports in workflows:
345 label = f"{qn}:{cls.name}"
346 lexical.extend(_scan_body(cls, imports, qn))
347 seen: set[tuple[str, str]] = set()
348 queue: deque[tuple[tuple[str, str], tuple[str, ...], int]] = deque(
349 ((m, s), (lbl,), 1)
350 for m, s, lbl in _first_party_callees(
351 cls, imports, modules, functions, qn
352 )
353 )
354 while queue:
355 (mod, sym), via, depth = queue.popleft()
356 if (mod, sym) in seen:
357 continue
358 seen.add((mod, sym))
359 fn = functions[mod].get(sym)
360 if fn is None:
361 continue
362 fn_imports = _import_map(modules[mod].tree)
363 for imp in _scan_body(fn, fn_imports, mod):
364 hazards.append(
365 HazardPath(workflow=label, via=via, impurity=imp)
366 )
367 if depth < max_depth:
368 for m2, s2, lbl2 in _first_party_callees(
369 fn, fn_imports, modules, functions, mod
370 ):
371 if (m2, s2) not in seen:
372 queue.append(((m2, s2), (*via, lbl2), depth + 1))
374 frontier: list[FrontierItem] = []
375 wf_class_of: dict[str, str] = {}
376 for qn, cls, _ in workflows:
377 wf_class_of.setdefault(qn, cls.name)
378 for qn in sorted(wf_class_of):
379 lm = modules[qn]
380 for lineno, root in _risky_imports(lm.tree, first_party):
381 frontier.append(
382 FrontierItem(
383 kind="third_party_import",
384 workflow=f"{qn}:{wf_class_of[qn]}",
385 module=qn,
386 line=lineno,
387 symbol=root,
388 snippet=_line(lm.lines, lineno),
389 )
390 )
392 return AnalysisResult(
393 lexical=tuple(_dedupe_impurities(lexical)),
394 hazards=tuple(_dedupe_hazards(hazards)),
395 frontier=tuple(frontier),
396 )
399def _dedupe_impurities(items: Iterable[Impurity]) -> list[Impurity]:
400 seen: set[tuple[str, int, str]] = set()
401 out: list[Impurity] = []
402 for i in sorted(items, key=lambda x: (x.module, x.line, x.rule)):
403 key = (i.module, i.line, i.rule)
404 if key not in seen:
405 seen.add(key)
406 out.append(i)
407 return out
410def _dedupe_hazards(items: Iterable[HazardPath]) -> list[HazardPath]:
411 seen: set[tuple[str, str, int, str]] = set()
412 out: list[HazardPath] = []
413 ordered = sorted(
414 items,
415 key=lambda h: (
416 h.workflow,
417 h.impurity.module,
418 h.impurity.line,
419 h.impurity.rule,
420 ),
421 )
422 for h in ordered:
423 key = (h.workflow, h.impurity.module, h.impurity.line, h.impurity.rule)
424 if key not in seen:
425 seen.add(key)
426 out.append(h)
427 return out
Two workflows — and why the brain passes its own gate
ReviewWorkflow is the durable per-repo loop: list open PRs, dispatch a review each (idempotent per PR + head SHA), sleep, continue-as-new — a near-exact mirror of ScanWorkflow. PrReviewWorkflow is the per-PR pipeline: analyze → adjudicate the frontier (only if there is one) → synthesize (pure) → post.
PrReviewWorkflow.run — the linear pipeline · 65 lines
PrReviewWorkflow.run — the linear pipeline
⋯ 33 lines hidden (lines 1–33)
1"""A single PR's determinism review — analyze, adjudicate, comment (advisory).
3A linear, durable pipeline (not a branching state machine — the flow is three
4steps): analyze the workflow surface at the PR head, adjudicate the ambiguous
5frontier with the model (only when there is one), synthesize the findings
6(pure), and upsert one advisory comment. Idempotent per (PR, head SHA) via its
7workflow id, so a new commit is a new review that edits the comment in place.
9All AST, model, and GitHub work happens in activities; the workflow body only
10orchestrates and runs one pure policy call (:func:`synthesize_findings`), so it
11stays deterministic and passes the determinism gate itself.
12"""
14from __future__ import annotations
16from temporalio import workflow
18with workflow.unsafe.imports_passed_through():
19 from froot.domain.determinism import FrontierVerdict, PrReviewResult
20 from froot.policy.review_comment import synthesize_findings
21 from froot.workflow import activities
22 from froot.workflow.constants import ACTIVITY_TIMEOUT, CI_CHECK_TIMEOUT
23 from froot.workflow.types import (
24 AdjudicateInput,
25 PostReviewInput,
26 PrReviewParams,
27 )
30@workflow.defn
31class PrReviewWorkflow:
32 """Review one PR for transitive determinism hazards (advisory)."""
34 @workflow.run
35 async def run(self, params: PrReviewParams) -> PrReviewResult:
36 """Analyze, adjudicate the frontier, then post the advisory comment."""
37 analysis = await workflow.execute_activity(
38 activities.analyze_pr,
39 params,
40 start_to_close_timeout=ACTIVITY_TIMEOUT,
41 )
42 verdicts: tuple[FrontierVerdict, ...] = ()
43 if analysis.frontier:
44 verdicts = await workflow.execute_activity(
45 activities.adjudicate_frontier,
46 AdjudicateInput(frontier=analysis.frontier),
47 start_to_close_timeout=ACTIVITY_TIMEOUT,
48 )
49 findings = synthesize_findings(
50 analysis.hazards, analysis.frontier, verdicts
51 )
52 comment_url = await workflow.execute_activity(
53 activities.post_review,
54 PostReviewInput(
55 target=params.target, pr=params.pr, findings=findings
56 ),
57 start_to_close_timeout=CI_CHECK_TIMEOUT,
58 )
59 return PrReviewResult(
60 pr_number=params.pr.number,
61 head_sha=params.pr.head_sha,
62 lexical_count=len(analysis.lexical),
63 findings=findings,
64 comment_url=comment_url,
65 )
The activities — the impure boundary
Five new activities, each a thin wrapper with lazy adapter imports (keeping HTTP/model stacks out of the sandbox), mirroring the bump loop's conventions. analyze_pr checks out the PR head, loads the first-party module ASTs, and runs the pure analyzer. dispatch_pr_review is idempotent per (PR, head SHA) via REJECT_DUPLICATE.
dispatch_pr_review + analyze_pr · 261 lines
dispatch_pr_review + analyze_pr
⋯ 177 lines hidden (lines 1–177)
1"""Activities: the effect interpreters that wrap the adapters.
3Each activity is the impure boundary for one effect. Adapter imports are LAZY
4(inside the bodies) so the model and HTTP stacks never enter a workflow sandbox;
5the pure policy and domain imports stay at module level. Activities return
6domain values — the workflow wraps them into events and feeds the pure state
7machine. The activity signatures' types are evaluated at runtime by Temporal's
8data converter, so the domain imports here are deliberately not deferred.
9"""
11from __future__ import annotations
13import json
14import logging
15import tempfile
16from pathlib import Path
18from temporalio import activity
20from froot.domain.candidate import PatchCandidate
21from froot.domain.changelog import ChangelogVerdict, UnknownVerdict
22from froot.domain.ci import CIStatus
23from froot.domain.determinism import AnalysisResult, FrontierVerdict
24from froot.domain.pull_request import PullRequestRef
25from froot.domain.repo import TargetRepo
26from froot.policy.candidates import select_patch_candidates
27from froot.policy.compose import PR_LABELS, pull_request_draft
28from froot.policy.determinism import analyze_workflow_surface
29from froot.policy.naming import (
30 branch_name,
31 bump_workflow_id,
32 pr_review_workflow_id,
33)
34from froot.policy.review_comment import REVIEW_MARKER, render_review_comment
35from froot.workflow.types import (
36 AdjudicateInput,
37 CiCheckInput,
38 DispatchInput,
39 DispatchReviewInput,
40 OpenPrInput,
41 PostReviewInput,
42 PrReviewParams,
43 RecordInput,
44)
46_log = logging.getLogger("froot.outcome")
47_review_log = logging.getLogger("froot.review")
50def _manifest_dir(target: TargetRepo, workspace: Path) -> Path:
51 """The directory the manifest lives in (a monorepo subdir, or the root)."""
52 return workspace / target.manifest_dir if target.manifest_dir else workspace
55@activity.defn
56async def scan_candidates(
57 target: TargetRepo,
58) -> tuple[PatchCandidate, ...]:
59 """Check out the repo, read available upgrades, select patch candidates."""
60 from froot.adapters.github import GitHubForge
61 from froot.adapters.registry import package_manager_for
63 forge = GitHubForge()
64 package_manager = package_manager_for(target.ecosystem)
65 with tempfile.TemporaryDirectory() as tmp:
66 workspace = Path(tmp)
67 await forge.checkout(target, workspace)
68 upgrades = await package_manager.list_upgrades(
69 target, _manifest_dir(target, workspace)
70 )
71 return select_patch_candidates(upgrades)
74@activity.defn
75async def judge_changelog(candidate: PatchCandidate) -> ChangelogVerdict:
76 """Fetch the candidate's changelog and get the model's typed verdict."""
77 from froot.adapters.changelog_http import HttpChangelogSource
78 from froot.adapters.model_judge import PydanticAiJudge
80 changelog = await HttpChangelogSource().fetch(candidate)
81 if changelog is None:
82 return UnknownVerdict(rationale="No changelog could be fetched.")
83 return await PydanticAiJudge().judge(changelog)
86@activity.defn
87async def open_pull_request(params: OpenPrInput) -> PullRequestRef:
88 """Regenerate manifest+lockfile and open (idempotently) the bump's PR."""
89 from froot.adapters.github import GitHubForge
90 from froot.adapters.registry import package_manager_for
92 forge = GitHubForge()
93 package_manager = package_manager_for(params.target.ecosystem)
94 branch = branch_name(params.candidate)
95 existing = await forge.find_open_pull_request(params.target, branch)
96 if existing is not None:
97 return existing
98 draft = pull_request_draft(params.target, params.candidate, params.verdict)
99 with tempfile.TemporaryDirectory() as tmp:
100 workspace = Path(tmp)
101 await forge.checkout(params.target, workspace)
102 await package_manager.apply_patch_bump(
103 params.candidate, _manifest_dir(params.target, workspace)
104 )
105 await forge.push_branch(workspace, branch, draft.title)
106 return await forge.open_pull_request(params.target, draft)
109@activity.defn
110async def check_ci(params: CiCheckInput) -> CIStatus:
111 """Read the repo's CI status for the PR's head commit (the oracle)."""
112 from froot.adapters.github import GitHubForge
114 return await GitHubForge().ci_status(params.target, params.head_sha)
117@activity.defn
118async def record_outcome(params: RecordInput) -> None:
119 """Label the PR and log the run telemetry — the signal-update."""
120 from froot.adapters.github import GitHubForge
122 outcome = params.outcome
123 await GitHubForge().add_labels(params.target, outcome.pr.number, PR_LABELS)
124 _log.info(
125 json.dumps(
126 {
127 "event": "loop_outcome",
128 "loop": "dependency-patch",
129 "repo": params.target.repo.slug,
130 "package": outcome.candidate.package,
131 "from": str(outcome.candidate.current),
132 "to": str(outcome.candidate.target),
133 "changelog": outcome.verdict.kind,
134 "ci": outcome.ci.kind,
135 "ci_passed": outcome.ci_passed,
136 "pr": outcome.pr.number,
137 "pr_url": outcome.pr.url,
138 }
139 )
140 )
143@activity.defn
144async def dispatch_bump(params: DispatchInput) -> None:
145 """Start the bump loop for a candidate (idempotent per bump identity)."""
146 from temporalio.common import WorkflowIDReusePolicy
147 from temporalio.exceptions import WorkflowAlreadyStartedError
149 from froot.workflow.bump_workflow import BumpWorkflow
150 from froot.workflow.temporal_client import client, task_queue
151 from froot.workflow.types import BumpParams
153 temporal = await client()
154 try:
155 await temporal.start_workflow(
156 BumpWorkflow.run,
157 BumpParams(target=params.target, candidate=params.candidate),
158 id=bump_workflow_id(params.target, params.candidate),
159 task_queue=task_queue(),
160 id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
161 )
162 except WorkflowAlreadyStartedError:
163 # This bump already has a loop (running or completed) — a no-op, so
164 # re-scanning never opens a second PR for the same bump.
165 return
168# ── The determinism reviewer (the transitive ring) ──────────────────────────
169@activity.defn
170async def list_review_prs(target: TargetRepo) -> tuple[PullRequestRef, ...]:
171 """List the repo's open PRs for the determinism reviewer to consider."""
172 from froot.adapters.github import GitHubForge
174 return await GitHubForge().list_open_pull_requests(target)
177@activity.defn
178async def dispatch_pr_review(params: DispatchReviewInput) -> None:
179 """Start a PR's determinism review (idempotent per PR + head SHA)."""
180 from temporalio.common import WorkflowIDReusePolicy
181 from temporalio.exceptions import WorkflowAlreadyStartedError
183 from froot.workflow.pr_review_workflow import PrReviewWorkflow
184 from froot.workflow.temporal_client import client, task_queue
186 temporal = await client()
187 try:
188 await temporal.start_workflow(
189 PrReviewWorkflow.run,
190 PrReviewParams(target=params.target, pr=params.pr),
191 id=pr_review_workflow_id(
192 params.target, params.pr.number, params.pr.head_sha
193 ),
194 task_queue=task_queue(),
195 id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
196 )
197 except WorkflowAlreadyStartedError:
198 # This (PR, head SHA) already has a review — a no-op, so re-polling
199 # never double-reviews the same commit.
200 return
203@activity.defn
204async def analyze_pr(params: PrReviewParams) -> AnalysisResult:
205 """Check out the PR head and analyze the workflow surface for hazards."""
206 from froot.adapters.github import GitHubForge
207 from froot.adapters.source_tree import load_modules
208 from froot.config.settings import ReviewSettings
210 forge = GitHubForge()
211 with tempfile.TemporaryDirectory() as tmp:
212 workspace = Path(tmp)
213 await forge.checkout_pull_request(
214 params.target, workspace, params.pr.number
215 )
216 # The ASTs and source lines are read into memory here, so the analysis
217 # below is unaffected by the workspace being cleaned up.
218 modules = load_modules(workspace)
219 return analyze_workflow_surface(modules, max_depth=ReviewSettings().depth)
⋯ 40 lines hidden (lines 222–261)
222@activity.defn
223async def adjudicate_frontier(
224 params: AdjudicateInput,
225) -> tuple[FrontierVerdict, ...]:
226 """Run the model over each frontier item; return aligned verdicts."""
227 from froot.adapters.determinism_judge import DeterminismFrontierJudge
229 judge = DeterminismFrontierJudge()
230 verdicts: list[FrontierVerdict] = []
231 for item in params.frontier:
232 verdicts.append(await judge.adjudicate(item))
233 return tuple(verdicts)
236@activity.defn
237async def post_review(params: PostReviewInput) -> str | None:
238 """Upsert the advisory comment (when there are findings); log the ledger."""
239 from froot.adapters.github import GitHubForge
241 body = render_review_comment(params.findings, params.pr.head_sha)
242 url: str | None = None
243 if body is not None:
244 url = await GitHubForge().upsert_issue_comment(
245 params.target, params.pr.number, REVIEW_MARKER, body
246 )
247 _review_log.info(
248 json.dumps(
249 {
250 "event": "loop_outcome",
251 "loop": "determinism-review",
252 "repo": params.target.repo.slug,
253 "pr": params.pr.number,
254 "head_sha": params.pr.head_sha,
255 "findings": len(params.findings),
256 "rules": sorted({f.rule for f in params.findings}),
257 "comment_url": url,
258 }
259 )
260 )
261 return url
The model frontier and the advisory comment
The static pass does the heavy lifting. The model adjudicates only the residual it can't resolve: a risky third-party import at a workflow module's scope (is it actually reached, or dead weight?). synthesize_findings then combines static hazards with the model-confirmed frontier, and render_review_comment emits one marker-tagged comment.
synthesize_findings — static ∪ (model == yes) · 103 lines
synthesize_findings — static ∪ (model == yes)
⋯ 27 lines hidden (lines 1–27)
1"""Synthesize and render the determinism reviewer's advisory PR comment (pure).
3One marker-tagged comment per PR, upserted in place on each new head SHA (the
4marker lets the forge find-and-update rather than stack comments — a reviewer
5that spams is the entropy it exists to prevent). Findings are synthesized from
6the static transitive hazards and the model-adjudicated frontier; the comment is
7advisory — the blocking gate is the kernel's ``Determinism`` CI check.
8"""
10from __future__ import annotations
12from typing import TYPE_CHECKING
14from froot.domain.determinism import ReviewFinding
16if TYPE_CHECKING:
17 from collections.abc import Sequence
19 from froot.domain.determinism import (
20 FrontierItem,
21 FrontierVerdict,
22 HazardPath,
23 )
25REVIEW_MARKER = "<!-- froot:determinism-review -->"
27_THIRD_PARTY_HINT = "move this dependency behind an activity"
30def synthesize_findings(
31 hazards: Sequence[HazardPath],
32 frontier: Sequence[FrontierItem],
33 verdicts: Sequence[FrontierVerdict],
34) -> tuple[ReviewFinding, ...]:
35 """Combine confirmed hazards and model-confirmed frontier into findings.
37 ``frontier`` and ``verdicts`` are index-aligned (one verdict per item). Only
38 items the model judged ``reaches == "yes"`` are surfaced; ``no`` and
39 ``uncertain`` are dropped to keep the advisory comment high-signal.
40 """
41 findings: list[ReviewFinding] = [
42 ReviewFinding(
43 origin="static",
44 workflow=h.workflow,
45 detail=" → ".join((*h.via, h.impurity.rule)),
46 rule=h.impurity.rule,
47 hint=h.impurity.hint,
48 module=h.impurity.module,
49 line=h.impurity.line,
50 )
51 for h in hazards
52 ]
53 for item, verdict in zip(frontier, verdicts, strict=True):
54 if verdict.reaches == "yes":
55 findings.append(
56 ReviewFinding(
57 origin="model",
58 workflow=item.workflow,
59 detail=verdict.rationale,
60 rule=item.symbol,
61 hint=_THIRD_PARTY_HINT,
62 module=item.module,
63 line=item.line,
64 )
65 )
66 return tuple(findings)
69def render_review_comment(
⋯ 34 lines hidden (lines 70–103)
70 findings: Sequence[ReviewFinding], head_sha: str
71) -> str | None:
72 """Render the marker-tagged comment body, or None when there's nothing."""
73 if not findings:
74 return None
75 out = [
76 REVIEW_MARKER,
77 "",
78 "### 🧭 froot determinism review",
79 "",
80 (
81 "Transitive determinism hazards reachable from this repo's "
82 f"Temporal workflows at `{head_sha[:7]}`. **Advisory** — the "
83 "blocking gate is the `Determinism` CI check; this loop catches "
84 "what that lexical check can't see across calls."
85 ),
86 "",
87 ]
88 for finding in findings:
89 origin = (
90 "static call-path"
91 if finding.origin == "static"
92 else "model-assessed"
93 )
94 out.append(f"- **`{finding.rule}`** — {finding.hint}")
95 out.append(
96 f" - reached from `{finding.workflow}`, "
97 f"at `{finding.module}:{finding.line}` ({origin})"
98 )
99 out.append(f" - path: `{finding.detail}`")
100 out.extend(
101 ["", "_Reviewed by froot · re-runs update this comment in place._"]
102 )
103 return "\n".join(out)
| Concern | How it's handled |
|---|---|
| Comment stacking | marker-tagged, upserted in place each new head SHA |
| Score / ledger | froot.review telemetry → derive hazard-resolved rate |
| Honesty | each finding tagged static call-path vs model-assessed; comment says advisory |
| Model cost | bounded — only the import frontier, not every reachable body |
Forge, settings, wiring
The Forge protocol gains three methods. checkout_pull_request uses refs/pull/N/head — the ref the base repo exposes for every PR — so a fork PR checks out exactly like a same-repo one. list_open_pull_requests feeds the loop; upsert_issue_comment finds-or-edits the marker comment.
checkout_pull_request — fork-safe via refs/pull/N/head · 351 lines
checkout_pull_request — fork-safe via refs/pull/N/head
⋯ 167 lines hidden (lines 1–167)
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
16from dataclasses import dataclass
17from typing import TYPE_CHECKING, Any, final
19import httpx
20from temporalio.exceptions import ApplicationError
22from froot.config.settings import GitHubSettings
23from froot.domain.ci import (
24 CIAbsent,
25 CIFailed,
26 CIPassed,
27 CIPending,
28 CIStatus,
29)
30from froot.domain.pull_request import BranchName, PullRequestRef
32if TYPE_CHECKING:
33 from pathlib import Path
35 from froot.domain.pull_request import PullRequestDraft
36 from froot.domain.repo import TargetRepo
38from froot.adapters._proc import run_text
40_API = "https://api.github.com"
41_API_VERSION = "2022-11-28"
42_TIMEOUT = 30.0
43_COMMITTER_NAME = "froot"
44_COMMITTER_EMAIL = "froot@users.noreply.github.com"
46# Check-run conclusions that mean the change is not safe to merge.
47_BAD_CONCLUSIONS = frozenset(
48 {
49 "failure",
50 "timed_out",
51 "cancelled",
52 "action_required",
53 "startup_failure",
54 "stale",
55 }
56)
59@final
60@dataclass(frozen=True, slots=True)
61class CheckRow:
62 """One GitHub check run, reduced to what the CI mapping needs."""
64 name: str
65 status: str
66 conclusion: str | None
69def ci_status_from_checks(
70 checks: tuple[CheckRow, ...], combined_state: str | None
71) -> CIStatus:
72 """Map GitHub check rows + the combined commit status to a CI status.
74 Args:
75 checks: The commit's check runs (GitHub Checks API).
76 combined_state: The legacy combined commit status (``success`` /
77 ``failure`` / ``pending``), or ``None`` when no statuses exist.
79 Returns:
80 ``CIAbsent`` when nothing reports, ``CIPending`` while anything is
81 unresolved, ``CIFailed`` (with the failing check names) on any bad
82 conclusion or a failed combined status, else ``CIPassed``.
83 """
84 if not checks and combined_state is None:
85 return CIAbsent()
86 unresolved = combined_state == "pending" or any(
87 row.status != "completed" for row in checks
88 )
89 if unresolved:
90 return CIPending()
91 failing = tuple(
92 row.name for row in checks if row.conclusion in _BAD_CONCLUSIONS
93 )
94 if failing or combined_state == "failure":
95 return CIFailed(failing=failing)
96 return CIPassed()
99def _token() -> str:
100 token = GitHubSettings().github_token
101 if token is None:
102 # A missing token is a permanent misconfiguration, not a transient
103 # fault — fail the activity fast instead of retrying forever.
104 raise ApplicationError(
105 "FROOT_GITHUB_TOKEN is required", non_retryable=True
106 )
107 return token.get_secret_value()
110def _auth_remote(target: TargetRepo) -> str:
111 return (
112 f"https://x-access-token:{_token()}@github.com/{target.repo.slug}.git"
113 )
116def _client() -> httpx.AsyncClient:
117 return httpx.AsyncClient(
118 base_url=_API,
119 timeout=_TIMEOUT,
120 headers={
121 "Authorization": f"Bearer {_token()}",
122 "Accept": "application/vnd.github+json",
123 "X-GitHub-Api-Version": _API_VERSION,
124 },
125 )
128def _raise_for_status(response: httpx.Response) -> None:
129 """Raise on error; 401/403 is a permanent (non-retryable) auth fault."""
130 if response.status_code in (401, 403):
131 raise ApplicationError(
132 f"GitHub auth failed ({response.status_code})", non_retryable=True
133 )
134 response.raise_for_status()
137def _pull_request_ref(payload: Any) -> PullRequestRef:
138 """Coerce a GitHub PR JSON payload into a domain ref (boundary)."""
139 head = payload["head"]
140 return PullRequestRef(
141 number=int(payload["number"]),
142 url=str(payload["html_url"]),
143 branch=BranchName(value=str(head["ref"])),
144 head_sha=str(head["sha"]),
145 )
148@final
149class GitHubForge:
150 """A :class:`~froot.ports.protocols.Forge` over git + the GitHub API."""
152 async def checkout(self, target: TargetRepo, workspace: Path) -> None:
153 """Shallow-clone the repo's default branch into ``workspace``."""
154 code, out = await run_text(
155 "git",
156 "clone",
157 "--depth",
158 "1",
159 "--branch",
160 target.default_branch,
161 _auth_remote(target),
162 ".",
163 cwd=workspace,
164 )
165 if code != 0:
166 raise RuntimeError(f"git clone failed ({code}): {out}")
168 async def checkout_pull_request(
169 self, target: TargetRepo, workspace: Path, number: int
170 ) -> None:
171 """Materialize a PR's head into ``workspace`` via ``refs/pull/N/head``.
173 Fetches the PR head ref the base repo exposes for every PR (fork or
174 not), so a community PR from a fork checks out the same way a same-repo
175 one does — no fork URL, no cross-repo auth.
176 """
177 ref = f"pull/{number}/head"
178 steps: tuple[tuple[str, ...], ...] = (
179 ("git", "init", "-q"),
180 ("git", "remote", "add", "origin", _auth_remote(target)),
181 ("git", "fetch", "--depth", "1", "origin", ref),
182 ("git", "checkout", "-q", "FETCH_HEAD"),
183 )
184 for step in steps:
185 code, out = await run_text(*step, cwd=workspace)
186 if code != 0:
187 raise RuntimeError(
188 f"git {step[1]} ({ref}) failed ({code}): {out}"
189 )
⋯ 162 lines hidden (lines 190–351)
191 async def push_branch(
192 self, workspace: Path, branch: BranchName, commit_message: str
193 ) -> str:
194 """Commit the workspace changes onto ``branch`` and push; return SHA."""
195 steps: tuple[tuple[str, ...], ...] = (
196 ("git", "checkout", "-b", branch.value),
197 ("git", "add", "-A"),
198 (
199 "git",
200 "-c",
201 f"user.name={_COMMITTER_NAME}",
202 "-c",
203 f"user.email={_COMMITTER_EMAIL}",
204 "commit",
205 "-m",
206 commit_message,
207 ),
208 ("git", "push", "-u", "origin", branch.value),
209 )
210 for step in steps:
211 code, out = await run_text(*step, cwd=workspace)
212 if code != 0:
213 raise RuntimeError(f"{step[0:2]} failed ({code}): {out}")
214 code, sha = await run_text("git", "rev-parse", "HEAD", cwd=workspace)
215 if code != 0:
216 raise RuntimeError(f"git rev-parse failed ({code})")
217 return sha.strip()
219 async def find_open_pull_request(
220 self, target: TargetRepo, branch: BranchName
221 ) -> PullRequestRef | None:
222 """Return the open PR for ``branch`` if one already exists (dedup)."""
223 async with _client() as client:
224 resp = await client.get(
225 f"/repos/{target.repo.slug}/pulls",
226 params={
227 "head": f"{target.repo.owner}:{branch.value}",
228 "state": "open",
229 },
230 )
231 _raise_for_status(resp)
232 payloads = resp.json()
233 if isinstance(payloads, list) and payloads:
234 return _pull_request_ref(payloads[0])
235 return None
237 async def list_open_pull_requests(
238 self, target: TargetRepo
239 ) -> tuple[PullRequestRef, ...]:
240 """List the repo's open PRs (the determinism reviewer's work feed)."""
241 async with _client() as client:
242 resp = await client.get(
243 f"/repos/{target.repo.slug}/pulls",
244 params={"state": "open", "per_page": 100},
245 )
246 _raise_for_status(resp)
247 payloads = resp.json()
248 if not isinstance(payloads, list):
249 return ()
250 return tuple(_pull_request_ref(payload) for payload in payloads)
252 async def open_pull_request(
253 self, target: TargetRepo, draft: PullRequestDraft
254 ) -> PullRequestRef:
255 """Open the PR for an already-pushed branch (idempotent on conflict)."""
256 async with _client() as client:
257 resp = await client.post(
258 f"/repos/{target.repo.slug}/pulls",
259 json={
260 "title": draft.title,
261 "head": draft.branch.value,
262 "base": draft.base,
263 "body": draft.body,
264 },
265 )
266 if resp.status_code == 422:
267 existing = await self.find_open_pull_request(target, draft.branch)
268 if existing is not None:
269 return existing
270 _raise_for_status(resp)
271 return _pull_request_ref(resp.json())
273 async def ci_status(self, target: TargetRepo, head_sha: str) -> CIStatus:
274 """Read the repo's combined CI status for a commit (the oracle)."""
275 async with _client() as client:
276 checks_resp = await client.get(
277 f"/repos/{target.repo.slug}/commits/{head_sha}/check-runs",
278 params={"per_page": 100},
279 )
280 status_resp = await client.get(
281 f"/repos/{target.repo.slug}/commits/{head_sha}/status"
282 )
283 _raise_for_status(checks_resp)
284 _raise_for_status(status_resp)
285 rows = tuple(
286 CheckRow(
287 name=str(run["name"]),
288 status=str(run["status"]),
289 conclusion=(
290 str(run["conclusion"])
291 if run.get("conclusion") is not None
292 else None
293 ),
294 )
295 for run in checks_resp.json().get("check_runs", [])
296 )
297 status_json = status_resp.json()
298 combined = (
299 str(status_json["state"])
300 if int(status_json.get("total_count", 0)) > 0
301 else None
302 )
303 return ci_status_from_checks(rows, combined)
305 async def add_labels(
306 self, target: TargetRepo, number: int, labels: tuple[str, ...]
307 ) -> None:
308 """Attach labels to a PR (the human-readable signal-update)."""
309 async with _client() as client:
310 resp = await client.post(
311 f"/repos/{target.repo.slug}/issues/{number}/labels",
312 json={"labels": list(labels)},
313 )
314 _raise_for_status(resp)
316 async def upsert_issue_comment(
317 self, target: TargetRepo, number: int, marker: str, body: str
318 ) -> str:
319 """Create or update the PR's ``marker``-tagged comment; return its URL.
321 A PR conversation (issue) comment, not an inline review comment: the
322 determinism findings are structural (a call path), so a single advisory
323 summary fits better than line anchors. The marker lets a re-review edit
324 its own prior comment in place rather than stack a new one.
325 """
326 slug = target.repo.slug
327 async with _client() as client:
328 listing = await client.get(
329 f"/repos/{slug}/issues/{number}/comments",
330 params={"per_page": 100},
331 )
332 _raise_for_status(listing)
333 existing_id: int | None = None
334 for comment in listing.json():
335 if isinstance(comment, dict) and marker in str(
336 comment.get("body", "")
337 ):
338 existing_id = int(comment["id"])
339 break
340 if existing_id is not None:
341 resp = await client.patch(
342 f"/repos/{slug}/issues/comments/{existing_id}",
343 json={"body": body},
344 )
345 else:
346 resp = await client.post(
347 f"/repos/{slug}/issues/{number}/comments",
348 json={"body": body},
349 )
350 _raise_for_status(resp)
351 return str(resp.json()["html_url"])
Rounding it out: ReviewSettings (FROOT_REVIEW_* — enabled / poll interval / depth), review_starter (boots one loop per repo, gated by enabled), the two workflows + five activities registered in runtime.py, and the review_workflow_id / pr_review_workflow_id naming keys.
What the adversarial review caught
Before this PR, four independent reviewers (analyzer correctness, Temporal replay-safety, GitHub I/O, spec alignment) swept the diff and each finding was verified. Four real defects were confirmed and fixed:
| Defect | Fix |
|---|---|
from ..random import x mis-binds (analyzer _import_map) | skip all relative imports (level > 0) |
| relative import falsely flagged as risky third-party | _risky_imports requires level == 0 |
| kernel matcher: same relative-import bug → a false positive in the blocking gate | same guard in scripts/check_determinism.py |
checkout_branch hard-fails on fork PRs | fetch refs/pull/N/head instead |
Verified end to end: ruff, mypy strict (92 files), 171 tests, the determinism gate on the brain's own workflows, and zero false positives running the analyzer across 143 real modules in froot + ynab-agent.