Overview

froot had three one-shot starter modules — scan_starter, review_starter, a11y_review_starter — each run after the worker boots to submit its loops' long-lived Temporal workflows. They were 98% identical: connect, loop over the repos, start a workflow with ALLOW_DUPLICATE_FAILED_ONLY, print. They differed only in the gate, the workflow, and the params.

This collapses them into one froot.starter that walks the loop registry once and branches on disposition: acting loops in FROOT_LOOPS get a ScanWorkflow; enabled advisory loops get their own review workflow. The decision is a pure, tested function; the Temporal I/O is thin glue. Net −226 lines, and adding a loop no longer means a new starter module.

The pure unit of work

A _Start is one workflow to submit: the registered workflow type name (started by name, so the starter imports no workflow classes), the params, the deterministic id, and a label. An _Advisory is the bespoke start-wiring for one advisory loop — its workflow type, id namer, params class, and its own enable flag and interval. The advisory roots are still their own (the per-PR engine is not unified here), so this is the one seam that stays per-loop.

src/froot/starter.py · 243 lines
src/froot/starter.py243 lines · Python
⋯ 46 lines hidden (lines 1–46)
1"""Start (once) every configured froot loop — acting and advisory, one pass.
2 
3Run as a one-shot after the worker is up (a k8s Job, or ``python -m
4froot.starter`` locally). It walks the loop registry: each acting
5(commit-or-revert) loop named in ``FROOT_LOOPS`` gets a long-lived
6``ScanWorkflow`` per repo; each advisory (emit-signal) loop that is enabled
7gets its own per-repo review workflow. Every start uses
8``ALLOW_DUPLICATE_FAILED_ONLY`` so a running loop is left untouched but a
9terminated one can restart, and each tick re-derives its work and
10continues-as-new (no cursor — derived, never stored).
11 
12This is the single entrypoint that replaced the per-family scan / review /
13a11y starters: adding a loop is a registry spec (plus, for an advisory loop,
14one row of start wiring), not a new starter module.
15 
16Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
17repo list, the per-loop intervals, and the advisory enable flags
18``FROOT_REVIEW_ENABLED`` / ``FROOT_A11Y_ENABLED``). Re-running is safe.
19"""
20 
21from __future__ import annotations
22 
23import asyncio
24from dataclasses import dataclass
25from typing import TYPE_CHECKING
26 
27from froot.domain.loop import Loop
28from froot.loops import registry
29from froot.loops.registry import Disposition
30from froot.policy.naming import (
31 a11y_review_workflow_id,
32 review_workflow_id,
33 scan_workflow_id,
35from froot.workflow.types import (
36 A11yReviewScanParams,
37 ReviewScanParams,
38 ScanParams,
40 
41if TYPE_CHECKING:
42 from collections.abc import Callable
43 
44 from froot.domain.repo import TargetRepo
45 
46# The input every loop's self-scheduling root workflow takes.
47_Params = ScanParams | ReviewScanParams | A11yReviewScanParams
48 
49 
50@dataclass(frozen=True)
51class _Start:
52 """One workflow to (idempotently) start — the pure unit of work.
53 
54 Attributes:
55 workflow_type: The registered workflow type to start by name
56 (``ScanWorkflow`` / ``ReviewWorkflow`` / ``A11yReviewWorkflow``).
57 params: The root workflow's input.
58 workflow_id: The deterministic id (the singleton / idempotency key).
59 label: A short human tag for the start log line.
60 slug: The ``owner/name`` this start is for.
61 """
62 
63 workflow_type: str
64 params: _Params
65 workflow_id: str
66 label: str
67 slug: str
68 
69 
70@dataclass(frozen=True)
71class _Advisory:
72 """The bespoke start wiring for one advisory (emit-signal) loop.
73 
74 The advisory loops' root workflows are still their own (the per-PR engine
75 is not unified), so each carries its workflow type, id namer, and params
76 class, plus its own enable flag and poll interval from settings.
77 
78 Attributes:
79 enabled: Whether this loop's enable flag is set.
80 interval_seconds: This loop's poll cadence.
81 workflow_type: The registered workflow type name.
82 namer: The deterministic per-repo id namer.
83 params: The params class this loop's workflow takes.
84 label: A short human tag for the start log line.
85 """
86 
87 enabled: bool
88 interval_seconds: int
89 workflow_type: str
90 namer: Callable[[TargetRepo], str]
91 params: type[ReviewScanParams] | type[A11yReviewScanParams]
92 label: str
93 
⋯ 150 lines hidden (lines 94–243)
94 
95def advisory_wiring(
96 *,
97 review_enabled: bool,
98 review_interval_seconds: int,
99 a11y_enabled: bool,
100 a11y_interval_seconds: int,
101) -> dict[Loop, _Advisory]:
102 """The advisory loops' bespoke start wiring, keyed by loop."""
103 return {
104 Loop.DETERMINISM_REVIEW: _Advisory(
105 enabled=review_enabled,
106 interval_seconds=review_interval_seconds,
107 workflow_type="ReviewWorkflow",
108 namer=review_workflow_id,
109 params=ReviewScanParams,
110 label="determinism",
111 ),
112 Loop.A11Y_REVIEW: _Advisory(
113 enabled=a11y_enabled,
114 interval_seconds=a11y_interval_seconds,
115 workflow_type="A11yReviewWorkflow",
116 namer=a11y_review_workflow_id,
117 params=A11yReviewScanParams,
118 label="a11y",
119 ),
120 }
121 
122 
123def plans(
124 *,
125 repos: tuple[TargetRepo, ...],
126 loops: tuple[Loop, ...],
127 scan_interval_seconds: int,
128 advisory: dict[Loop, _Advisory],
129) -> tuple[_Start, ...]:
130 """Every workflow to start, derived from the registry (pure).
131 
132 One pass over the registered loops, branched on disposition: an acting loop
133 is started iff it is in ``loops`` (``FROOT_LOOPS``); an advisory loop iff
134 its wiring is present and enabled. The set of loops, and which family each
135 is in, comes from the registry — so a new loop is a registration, not an
136 edit here.
137 """
138 out: list[_Start] = []
139 for spec in registry.all_specs():
140 loop = spec.loop
141 if spec.disposition is Disposition.COMMIT_OR_REVERT:
142 if loop not in loops:
143 continue
144 for target in repos:
145 out.append(
146 _Start(
147 workflow_type="ScanWorkflow",
148 params=ScanParams(
149 target=target,
150 interval_seconds=scan_interval_seconds,
151 continuous=True,
152 loop=loop,
153 ),
154 workflow_id=scan_workflow_id(target, loop),
155 label=f"{loop.value} scan",
156 slug=target.repo.slug,
157 )
158 )
159 continue
160 wiring = advisory.get(loop)
161 if wiring is None or not wiring.enabled:
162 continue
163 for target in repos:
164 out.append(
165 _Start(
166 workflow_type=wiring.workflow_type,
167 params=wiring.params(
168 target=target,
169 interval_seconds=wiring.interval_seconds,
170 continuous=True,
171 ),
172 workflow_id=wiring.namer(target),
173 label=f"{wiring.label} review",
174 slug=target.repo.slug,
175 )
176 )
177 return tuple(out)
178 
179 
180async def _start() -> None:
181 from temporalio.client import Client
182 from temporalio.common import WorkflowIDReusePolicy
183 from temporalio.contrib.pydantic import pydantic_data_converter
184 from temporalio.exceptions import WorkflowAlreadyStartedError
185 
186 from froot.config.settings import (
187 A11yReviewSettings,
188 ReviewSettings,
189 Settings,
190 TemporalSettings,
191 )
192 
193 settings = Settings() # non-secret values come from the environment
194 review = ReviewSettings()
195 a11y = A11yReviewSettings()
196 to_start = plans(
197 repos=settings.repos,
198 loops=settings.loops,
199 scan_interval_seconds=settings.scan_interval_seconds,
200 advisory=advisory_wiring(
201 review_enabled=review.enabled,
202 review_interval_seconds=review.poll_interval_seconds,
203 a11y_enabled=a11y.enabled,
204 a11y_interval_seconds=a11y.poll_interval_seconds,
205 ),
206 )
207 if not to_start:
208 print("no loops to start (check FROOT_LOOPS and the enable flags)")
209 return
210 
211 temporal = TemporalSettings()
212 client = await Client.connect(
213 temporal.host,
214 namespace=temporal.namespace,
215 data_converter=pydantic_data_converter,
216 )
217 queue = temporal.task_queue
218 for plan in to_start:
219 try:
220 handle = await client.start_workflow(
221 plan.workflow_type,
222 plan.params,
223 id=plan.workflow_id,
224 task_queue=queue,
225 id_reuse_policy=(
226 WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY
227 ),
228 )
229 print(f"started {plan.label} loop {handle.id!r} for {plan.slug}")
230 except WorkflowAlreadyStartedError:
231 print(
232 f"{plan.label} loop {plan.workflow_id!r} already running"
233 " — untouched"
234 )
235 
236 
237def main() -> None:
238 """Console entrypoint: start every configured loop, from the env."""
239 asyncio.run(_start())
240 
241 
242if __name__ == "__main__":
243 main()

The advisory wiring is one small table, keyed by loop — the two emit-signal loops and how each starts.

src/froot/starter.py · 243 lines
src/froot/starter.py243 lines · Python
⋯ 94 lines hidden (lines 1–94)
1"""Start (once) every configured froot loop — acting and advisory, one pass.
2 
3Run as a one-shot after the worker is up (a k8s Job, or ``python -m
4froot.starter`` locally). It walks the loop registry: each acting
5(commit-or-revert) loop named in ``FROOT_LOOPS`` gets a long-lived
6``ScanWorkflow`` per repo; each advisory (emit-signal) loop that is enabled
7gets its own per-repo review workflow. Every start uses
8``ALLOW_DUPLICATE_FAILED_ONLY`` so a running loop is left untouched but a
9terminated one can restart, and each tick re-derives its work and
10continues-as-new (no cursor — derived, never stored).
11 
12This is the single entrypoint that replaced the per-family scan / review /
13a11y starters: adding a loop is a registry spec (plus, for an advisory loop,
14one row of start wiring), not a new starter module.
15 
16Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
17repo list, the per-loop intervals, and the advisory enable flags
18``FROOT_REVIEW_ENABLED`` / ``FROOT_A11Y_ENABLED``). Re-running is safe.
19"""
20 
21from __future__ import annotations
22 
23import asyncio
24from dataclasses import dataclass
25from typing import TYPE_CHECKING
26 
27from froot.domain.loop import Loop
28from froot.loops import registry
29from froot.loops.registry import Disposition
30from froot.policy.naming import (
31 a11y_review_workflow_id,
32 review_workflow_id,
33 scan_workflow_id,
35from froot.workflow.types import (
36 A11yReviewScanParams,
37 ReviewScanParams,
38 ScanParams,
40 
41if TYPE_CHECKING:
42 from collections.abc import Callable
43 
44 from froot.domain.repo import TargetRepo
45 
46# The input every loop's self-scheduling root workflow takes.
47_Params = ScanParams | ReviewScanParams | A11yReviewScanParams
48 
49 
50@dataclass(frozen=True)
51class _Start:
52 """One workflow to (idempotently) start — the pure unit of work.
53 
54 Attributes:
55 workflow_type: The registered workflow type to start by name
56 (``ScanWorkflow`` / ``ReviewWorkflow`` / ``A11yReviewWorkflow``).
57 params: The root workflow's input.
58 workflow_id: The deterministic id (the singleton / idempotency key).
59 label: A short human tag for the start log line.
60 slug: The ``owner/name`` this start is for.
61 """
62 
63 workflow_type: str
64 params: _Params
65 workflow_id: str
66 label: str
67 slug: str
68 
69 
70@dataclass(frozen=True)
71class _Advisory:
72 """The bespoke start wiring for one advisory (emit-signal) loop.
73 
74 The advisory loops' root workflows are still their own (the per-PR engine
75 is not unified), so each carries its workflow type, id namer, and params
76 class, plus its own enable flag and poll interval from settings.
77 
78 Attributes:
79 enabled: Whether this loop's enable flag is set.
80 interval_seconds: This loop's poll cadence.
81 workflow_type: The registered workflow type name.
82 namer: The deterministic per-repo id namer.
83 params: The params class this loop's workflow takes.
84 label: A short human tag for the start log line.
85 """
86 
87 enabled: bool
88 interval_seconds: int
89 workflow_type: str
90 namer: Callable[[TargetRepo], str]
91 params: type[ReviewScanParams] | type[A11yReviewScanParams]
92 label: str
93 
94 
95def advisory_wiring(
96 *,
97 review_enabled: bool,
98 review_interval_seconds: int,
99 a11y_enabled: bool,
100 a11y_interval_seconds: int,
101) -> dict[Loop, _Advisory]:
102 """The advisory loops' bespoke start wiring, keyed by loop."""
103 return {
104 Loop.DETERMINISM_REVIEW: _Advisory(
105 enabled=review_enabled,
106 interval_seconds=review_interval_seconds,
107 workflow_type="ReviewWorkflow",
108 namer=review_workflow_id,
109 params=ReviewScanParams,
110 label="determinism",
111 ),
112 Loop.A11Y_REVIEW: _Advisory(
113 enabled=a11y_enabled,
114 interval_seconds=a11y_interval_seconds,
115 workflow_type="A11yReviewWorkflow",
116 namer=a11y_review_workflow_id,
117 params=A11yReviewScanParams,
118 label="a11y",
119 ),
120 }
121 
⋯ 122 lines hidden (lines 122–243)
122 
123def plans(
124 *,
125 repos: tuple[TargetRepo, ...],
126 loops: tuple[Loop, ...],
127 scan_interval_seconds: int,
128 advisory: dict[Loop, _Advisory],
129) -> tuple[_Start, ...]:
130 """Every workflow to start, derived from the registry (pure).
131 
132 One pass over the registered loops, branched on disposition: an acting loop
133 is started iff it is in ``loops`` (``FROOT_LOOPS``); an advisory loop iff
134 its wiring is present and enabled. The set of loops, and which family each
135 is in, comes from the registry — so a new loop is a registration, not an
136 edit here.
137 """
138 out: list[_Start] = []
139 for spec in registry.all_specs():
140 loop = spec.loop
141 if spec.disposition is Disposition.COMMIT_OR_REVERT:
142 if loop not in loops:
143 continue
144 for target in repos:
145 out.append(
146 _Start(
147 workflow_type="ScanWorkflow",
148 params=ScanParams(
149 target=target,
150 interval_seconds=scan_interval_seconds,
151 continuous=True,
152 loop=loop,
153 ),
154 workflow_id=scan_workflow_id(target, loop),
155 label=f"{loop.value} scan",
156 slug=target.repo.slug,
157 )
158 )
159 continue
160 wiring = advisory.get(loop)
161 if wiring is None or not wiring.enabled:
162 continue
163 for target in repos:
164 out.append(
165 _Start(
166 workflow_type=wiring.workflow_type,
167 params=wiring.params(
168 target=target,
169 interval_seconds=wiring.interval_seconds,
170 continuous=True,
171 ),
172 workflow_id=wiring.namer(target),
173 label=f"{wiring.label} review",
174 slug=target.repo.slug,
175 )
176 )
177 return tuple(out)
178 
179 
180async def _start() -> None:
181 from temporalio.client import Client
182 from temporalio.common import WorkflowIDReusePolicy
183 from temporalio.contrib.pydantic import pydantic_data_converter
184 from temporalio.exceptions import WorkflowAlreadyStartedError
185 
186 from froot.config.settings import (
187 A11yReviewSettings,
188 ReviewSettings,
189 Settings,
190 TemporalSettings,
191 )
192 
193 settings = Settings() # non-secret values come from the environment
194 review = ReviewSettings()
195 a11y = A11yReviewSettings()
196 to_start = plans(
197 repos=settings.repos,
198 loops=settings.loops,
199 scan_interval_seconds=settings.scan_interval_seconds,
200 advisory=advisory_wiring(
201 review_enabled=review.enabled,
202 review_interval_seconds=review.poll_interval_seconds,
203 a11y_enabled=a11y.enabled,
204 a11y_interval_seconds=a11y.poll_interval_seconds,
205 ),
206 )
207 if not to_start:
208 print("no loops to start (check FROOT_LOOPS and the enable flags)")
209 return
210 
211 temporal = TemporalSettings()
212 client = await Client.connect(
213 temporal.host,
214 namespace=temporal.namespace,
215 data_converter=pydantic_data_converter,
216 )
217 queue = temporal.task_queue
218 for plan in to_start:
219 try:
220 handle = await client.start_workflow(
221 plan.workflow_type,
222 plan.params,
223 id=plan.workflow_id,
224 task_queue=queue,
225 id_reuse_policy=(
226 WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY
227 ),
228 )
229 print(f"started {plan.label} loop {handle.id!r} for {plan.slug}")
230 except WorkflowAlreadyStartedError:
231 print(
232 f"{plan.label} loop {plan.workflow_id!r} already running"
233 " — untouched"
234 )
235 
236 
237def main() -> None:
238 """Console entrypoint: start every configured loop, from the env."""
239 asyncio.run(_start())
240 
241 
242if __name__ == "__main__":
243 main()

The decision: one pass over the registry

plans() is the whole decision, pure: given the repos, the configured FROOT_LOOPS, the scan interval, and the advisory wiring, it returns every workflow to start. One pass over the registered specs, branched on disposition — an acting loop is started iff it is in FROOT_LOOPS; an advisory loop iff its wiring is enabled.

src/froot/starter.py · 243 lines
src/froot/starter.py243 lines · Python
⋯ 122 lines hidden (lines 1–122)
1"""Start (once) every configured froot loop — acting and advisory, one pass.
2 
3Run as a one-shot after the worker is up (a k8s Job, or ``python -m
4froot.starter`` locally). It walks the loop registry: each acting
5(commit-or-revert) loop named in ``FROOT_LOOPS`` gets a long-lived
6``ScanWorkflow`` per repo; each advisory (emit-signal) loop that is enabled
7gets its own per-repo review workflow. Every start uses
8``ALLOW_DUPLICATE_FAILED_ONLY`` so a running loop is left untouched but a
9terminated one can restart, and each tick re-derives its work and
10continues-as-new (no cursor — derived, never stored).
11 
12This is the single entrypoint that replaced the per-family scan / review /
13a11y starters: adding a loop is a registry spec (plus, for an advisory loop,
14one row of start wiring), not a new starter module.
15 
16Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
17repo list, the per-loop intervals, and the advisory enable flags
18``FROOT_REVIEW_ENABLED`` / ``FROOT_A11Y_ENABLED``). Re-running is safe.
19"""
20 
21from __future__ import annotations
22 
23import asyncio
24from dataclasses import dataclass
25from typing import TYPE_CHECKING
26 
27from froot.domain.loop import Loop
28from froot.loops import registry
29from froot.loops.registry import Disposition
30from froot.policy.naming import (
31 a11y_review_workflow_id,
32 review_workflow_id,
33 scan_workflow_id,
35from froot.workflow.types import (
36 A11yReviewScanParams,
37 ReviewScanParams,
38 ScanParams,
40 
41if TYPE_CHECKING:
42 from collections.abc import Callable
43 
44 from froot.domain.repo import TargetRepo
45 
46# The input every loop's self-scheduling root workflow takes.
47_Params = ScanParams | ReviewScanParams | A11yReviewScanParams
48 
49 
50@dataclass(frozen=True)
51class _Start:
52 """One workflow to (idempotently) start — the pure unit of work.
53 
54 Attributes:
55 workflow_type: The registered workflow type to start by name
56 (``ScanWorkflow`` / ``ReviewWorkflow`` / ``A11yReviewWorkflow``).
57 params: The root workflow's input.
58 workflow_id: The deterministic id (the singleton / idempotency key).
59 label: A short human tag for the start log line.
60 slug: The ``owner/name`` this start is for.
61 """
62 
63 workflow_type: str
64 params: _Params
65 workflow_id: str
66 label: str
67 slug: str
68 
69 
70@dataclass(frozen=True)
71class _Advisory:
72 """The bespoke start wiring for one advisory (emit-signal) loop.
73 
74 The advisory loops' root workflows are still their own (the per-PR engine
75 is not unified), so each carries its workflow type, id namer, and params
76 class, plus its own enable flag and poll interval from settings.
77 
78 Attributes:
79 enabled: Whether this loop's enable flag is set.
80 interval_seconds: This loop's poll cadence.
81 workflow_type: The registered workflow type name.
82 namer: The deterministic per-repo id namer.
83 params: The params class this loop's workflow takes.
84 label: A short human tag for the start log line.
85 """
86 
87 enabled: bool
88 interval_seconds: int
89 workflow_type: str
90 namer: Callable[[TargetRepo], str]
91 params: type[ReviewScanParams] | type[A11yReviewScanParams]
92 label: str
93 
94 
95def advisory_wiring(
96 *,
97 review_enabled: bool,
98 review_interval_seconds: int,
99 a11y_enabled: bool,
100 a11y_interval_seconds: int,
101) -> dict[Loop, _Advisory]:
102 """The advisory loops' bespoke start wiring, keyed by loop."""
103 return {
104 Loop.DETERMINISM_REVIEW: _Advisory(
105 enabled=review_enabled,
106 interval_seconds=review_interval_seconds,
107 workflow_type="ReviewWorkflow",
108 namer=review_workflow_id,
109 params=ReviewScanParams,
110 label="determinism",
111 ),
112 Loop.A11Y_REVIEW: _Advisory(
113 enabled=a11y_enabled,
114 interval_seconds=a11y_interval_seconds,
115 workflow_type="A11yReviewWorkflow",
116 namer=a11y_review_workflow_id,
117 params=A11yReviewScanParams,
118 label="a11y",
119 ),
120 }
121 
122 
123def plans(
124 *,
125 repos: tuple[TargetRepo, ...],
126 loops: tuple[Loop, ...],
127 scan_interval_seconds: int,
128 advisory: dict[Loop, _Advisory],
129) -> tuple[_Start, ...]:
130 """Every workflow to start, derived from the registry (pure).
131 
132 One pass over the registered loops, branched on disposition: an acting loop
133 is started iff it is in ``loops`` (``FROOT_LOOPS``); an advisory loop iff
134 its wiring is present and enabled. The set of loops, and which family each
135 is in, comes from the registry — so a new loop is a registration, not an
136 edit here.
137 """
138 out: list[_Start] = []
139 for spec in registry.all_specs():
140 loop = spec.loop
141 if spec.disposition is Disposition.COMMIT_OR_REVERT:
142 if loop not in loops:
143 continue
144 for target in repos:
145 out.append(
146 _Start(
147 workflow_type="ScanWorkflow",
148 params=ScanParams(
149 target=target,
150 interval_seconds=scan_interval_seconds,
151 continuous=True,
152 loop=loop,
153 ),
154 workflow_id=scan_workflow_id(target, loop),
155 label=f"{loop.value} scan",
156 slug=target.repo.slug,
157 )
158 )
159 continue
160 wiring = advisory.get(loop)
161 if wiring is None or not wiring.enabled:
162 continue
163 for target in repos:
164 out.append(
165 _Start(
166 workflow_type=wiring.workflow_type,
167 params=wiring.params(
168 target=target,
169 interval_seconds=wiring.interval_seconds,
170 continuous=True,
171 ),
172 workflow_id=wiring.namer(target),
173 label=f"{wiring.label} review",
174 slug=target.repo.slug,
175 )
176 )
177 return tuple(out)
178 
⋯ 65 lines hidden (lines 179–243)
179 
180async def _start() -> None:
181 from temporalio.client import Client
182 from temporalio.common import WorkflowIDReusePolicy
183 from temporalio.contrib.pydantic import pydantic_data_converter
184 from temporalio.exceptions import WorkflowAlreadyStartedError
185 
186 from froot.config.settings import (
187 A11yReviewSettings,
188 ReviewSettings,
189 Settings,
190 TemporalSettings,
191 )
192 
193 settings = Settings() # non-secret values come from the environment
194 review = ReviewSettings()
195 a11y = A11yReviewSettings()
196 to_start = plans(
197 repos=settings.repos,
198 loops=settings.loops,
199 scan_interval_seconds=settings.scan_interval_seconds,
200 advisory=advisory_wiring(
201 review_enabled=review.enabled,
202 review_interval_seconds=review.poll_interval_seconds,
203 a11y_enabled=a11y.enabled,
204 a11y_interval_seconds=a11y.poll_interval_seconds,
205 ),
206 )
207 if not to_start:
208 print("no loops to start (check FROOT_LOOPS and the enable flags)")
209 return
210 
211 temporal = TemporalSettings()
212 client = await Client.connect(
213 temporal.host,
214 namespace=temporal.namespace,
215 data_converter=pydantic_data_converter,
216 )
217 queue = temporal.task_queue
218 for plan in to_start:
219 try:
220 handle = await client.start_workflow(
221 plan.workflow_type,
222 plan.params,
223 id=plan.workflow_id,
224 task_queue=queue,
225 id_reuse_policy=(
226 WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY
227 ),
228 )
229 print(f"started {plan.label} loop {handle.id!r} for {plan.slug}")
230 except WorkflowAlreadyStartedError:
231 print(
232 f"{plan.label} loop {plan.workflow_id!r} already running"
233 " — untouched"
234 )
235 
236 
237def main() -> None:
238 """Console entrypoint: start every configured loop, from the env."""
239 asyncio.run(_start())
240 
241 
242if __name__ == "__main__":
243 main()

The I/O: connect, then dispatch by name

_start() reads the settings, builds the plan, and — only if there is something to start — connects and submits each plan. It keeps every old starter's contract: ALLOW_DUPLICATE_FAILED_ONLY and a caught WorkflowAlreadyStartedError, so a running loop is left untouched and a terminated one can restart.

src/froot/starter.py · 243 lines
src/froot/starter.py243 lines · Python
⋯ 186 lines hidden (lines 1–186)
1"""Start (once) every configured froot loop — acting and advisory, one pass.
2 
3Run as a one-shot after the worker is up (a k8s Job, or ``python -m
4froot.starter`` locally). It walks the loop registry: each acting
5(commit-or-revert) loop named in ``FROOT_LOOPS`` gets a long-lived
6``ScanWorkflow`` per repo; each advisory (emit-signal) loop that is enabled
7gets its own per-repo review workflow. Every start uses
8``ALLOW_DUPLICATE_FAILED_ONLY`` so a running loop is left untouched but a
9terminated one can restart, and each tick re-derives its work and
10continues-as-new (no cursor — derived, never stored).
11 
12This is the single entrypoint that replaced the per-family scan / review /
13a11y starters: adding a loop is a registry spec (plus, for an advisory loop,
14one row of start wiring), not a new starter module.
15 
16Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
17repo list, the per-loop intervals, and the advisory enable flags
18``FROOT_REVIEW_ENABLED`` / ``FROOT_A11Y_ENABLED``). Re-running is safe.
19"""
20 
21from __future__ import annotations
22 
23import asyncio
24from dataclasses import dataclass
25from typing import TYPE_CHECKING
26 
27from froot.domain.loop import Loop
28from froot.loops import registry
29from froot.loops.registry import Disposition
30from froot.policy.naming import (
31 a11y_review_workflow_id,
32 review_workflow_id,
33 scan_workflow_id,
35from froot.workflow.types import (
36 A11yReviewScanParams,
37 ReviewScanParams,
38 ScanParams,
40 
41if TYPE_CHECKING:
42 from collections.abc import Callable
43 
44 from froot.domain.repo import TargetRepo
45 
46# The input every loop's self-scheduling root workflow takes.
47_Params = ScanParams | ReviewScanParams | A11yReviewScanParams
48 
49 
50@dataclass(frozen=True)
51class _Start:
52 """One workflow to (idempotently) start — the pure unit of work.
53 
54 Attributes:
55 workflow_type: The registered workflow type to start by name
56 (``ScanWorkflow`` / ``ReviewWorkflow`` / ``A11yReviewWorkflow``).
57 params: The root workflow's input.
58 workflow_id: The deterministic id (the singleton / idempotency key).
59 label: A short human tag for the start log line.
60 slug: The ``owner/name`` this start is for.
61 """
62 
63 workflow_type: str
64 params: _Params
65 workflow_id: str
66 label: str
67 slug: str
68 
69 
70@dataclass(frozen=True)
71class _Advisory:
72 """The bespoke start wiring for one advisory (emit-signal) loop.
73 
74 The advisory loops' root workflows are still their own (the per-PR engine
75 is not unified), so each carries its workflow type, id namer, and params
76 class, plus its own enable flag and poll interval from settings.
77 
78 Attributes:
79 enabled: Whether this loop's enable flag is set.
80 interval_seconds: This loop's poll cadence.
81 workflow_type: The registered workflow type name.
82 namer: The deterministic per-repo id namer.
83 params: The params class this loop's workflow takes.
84 label: A short human tag for the start log line.
85 """
86 
87 enabled: bool
88 interval_seconds: int
89 workflow_type: str
90 namer: Callable[[TargetRepo], str]
91 params: type[ReviewScanParams] | type[A11yReviewScanParams]
92 label: str
93 
94 
95def advisory_wiring(
96 *,
97 review_enabled: bool,
98 review_interval_seconds: int,
99 a11y_enabled: bool,
100 a11y_interval_seconds: int,
101) -> dict[Loop, _Advisory]:
102 """The advisory loops' bespoke start wiring, keyed by loop."""
103 return {
104 Loop.DETERMINISM_REVIEW: _Advisory(
105 enabled=review_enabled,
106 interval_seconds=review_interval_seconds,
107 workflow_type="ReviewWorkflow",
108 namer=review_workflow_id,
109 params=ReviewScanParams,
110 label="determinism",
111 ),
112 Loop.A11Y_REVIEW: _Advisory(
113 enabled=a11y_enabled,
114 interval_seconds=a11y_interval_seconds,
115 workflow_type="A11yReviewWorkflow",
116 namer=a11y_review_workflow_id,
117 params=A11yReviewScanParams,
118 label="a11y",
119 ),
120 }
121 
122 
123def plans(
124 *,
125 repos: tuple[TargetRepo, ...],
126 loops: tuple[Loop, ...],
127 scan_interval_seconds: int,
128 advisory: dict[Loop, _Advisory],
129) -> tuple[_Start, ...]:
130 """Every workflow to start, derived from the registry (pure).
131 
132 One pass over the registered loops, branched on disposition: an acting loop
133 is started iff it is in ``loops`` (``FROOT_LOOPS``); an advisory loop iff
134 its wiring is present and enabled. The set of loops, and which family each
135 is in, comes from the registry — so a new loop is a registration, not an
136 edit here.
137 """
138 out: list[_Start] = []
139 for spec in registry.all_specs():
140 loop = spec.loop
141 if spec.disposition is Disposition.COMMIT_OR_REVERT:
142 if loop not in loops:
143 continue
144 for target in repos:
145 out.append(
146 _Start(
147 workflow_type="ScanWorkflow",
148 params=ScanParams(
149 target=target,
150 interval_seconds=scan_interval_seconds,
151 continuous=True,
152 loop=loop,
153 ),
154 workflow_id=scan_workflow_id(target, loop),
155 label=f"{loop.value} scan",
156 slug=target.repo.slug,
157 )
158 )
159 continue
160 wiring = advisory.get(loop)
161 if wiring is None or not wiring.enabled:
162 continue
163 for target in repos:
164 out.append(
165 _Start(
166 workflow_type=wiring.workflow_type,
167 params=wiring.params(
168 target=target,
169 interval_seconds=wiring.interval_seconds,
170 continuous=True,
171 ),
172 workflow_id=wiring.namer(target),
173 label=f"{wiring.label} review",
174 slug=target.repo.slug,
175 )
176 )
177 return tuple(out)
178 
179 
180async def _start() -> None:
181 from temporalio.client import Client
182 from temporalio.common import WorkflowIDReusePolicy
183 from temporalio.contrib.pydantic import pydantic_data_converter
184 from temporalio.exceptions import WorkflowAlreadyStartedError
185 
186 from froot.config.settings import (
187 A11yReviewSettings,
188 ReviewSettings,
189 Settings,
190 TemporalSettings,
191 )
192 
193 settings = Settings() # non-secret values come from the environment
194 review = ReviewSettings()
195 a11y = A11yReviewSettings()
196 to_start = plans(
197 repos=settings.repos,
198 loops=settings.loops,
199 scan_interval_seconds=settings.scan_interval_seconds,
200 advisory=advisory_wiring(
201 review_enabled=review.enabled,
202 review_interval_seconds=review.poll_interval_seconds,
203 a11y_enabled=a11y.enabled,
204 a11y_interval_seconds=a11y.poll_interval_seconds,
205 ),
206 )
207 if not to_start:
208 print("no loops to start (check FROOT_LOOPS and the enable flags)")
209 return
210 
211 temporal = TemporalSettings()
212 client = await Client.connect(
213 temporal.host,
214 namespace=temporal.namespace,
215 data_converter=pydantic_data_converter,
216 )
217 queue = temporal.task_queue
218 for plan in to_start:
219 try:
220 handle = await client.start_workflow(
221 plan.workflow_type,
222 plan.params,
223 id=plan.workflow_id,
224 task_queue=queue,
225 id_reuse_policy=(
226 WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY
227 ),
228 )
229 print(f"started {plan.label} loop {handle.id!r} for {plan.slug}")
230 except WorkflowAlreadyStartedError:
231 print(
232 f"{plan.label} loop {plan.workflow_id!r} already running"
233 " — untouched"
234 )
⋯ 9 lines hidden (lines 235–243)
235 
236 
237def main() -> None:
238 """Console entrypoint: start every configured loop, from the env."""
239 asyncio.run(_start())
240 
241 
242if __name__ == "__main__":
243 main()

The tests the old starters never had

The starters were untested I/O glue. Factoring the decision into a pure plans() makes the gating and routing testable without Temporal — the discriminating behaviour. The defaults start exactly a dependency scan and a determinism review; an advisory loop in FROOT_LOOPS spawns no scan; each loop's interval is its own, not swapped.

tests/test_starter.py · 149 lines
tests/test_starter.py149 lines · Python
⋯ 58 lines hidden (lines 1–58)
1"""The unified starter's pure plan — which loops start, with what.
2 
3The starter's I/O (connect, ``start_workflow``) is thin glue; its *decision* is
4the pure :func:`~froot.starter.plans`, derived from the registry and the
5settings. These guard that decision: an acting loop starts iff it is in
6``FROOT_LOOPS``, an advisory loop iff its enable flag is on, and each lands on
7its own workflow type, id, and params — so the one entrypoint covers all five
8loops without a per-family starter.
9"""
10 
11from __future__ import annotations
12 
13from froot.domain.loop import Loop
14from froot.domain.repo import TargetRepo
15from froot.policy.naming import (
16 a11y_review_workflow_id,
17 review_workflow_id,
18 scan_workflow_id,
20from froot.starter import _Start, advisory_wiring, plans
21from froot.workflow.types import (
22 A11yReviewScanParams,
23 ReviewScanParams,
24 ScanParams,
26from tests.support import make_repo
27 
28REPO = make_repo("mseeks/revisionist")
29REPO2 = make_repo("mseeks/froot")
30 
31_ACTING = (Loop.DEPENDENCY_PATCH, Loop.SECURITY_PATCH, Loop.DEAD_CODE)
32 
33 
34def _plans(
35 *,
36 loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
37 repos: tuple[TargetRepo, ...] = (REPO,),
38 review: bool = True,
39 a11y: bool = False,
40 scan_interval: int = 86_400,
41):
42 return plans(
43 repos=repos,
44 loops=loops,
45 scan_interval_seconds=scan_interval,
46 advisory=advisory_wiring(
47 review_enabled=review,
48 review_interval_seconds=300,
49 a11y_enabled=a11y,
50 a11y_interval_seconds=600,
51 ),
52 )
53 
54 
55def _by_id(ps: tuple[_Start, ...]) -> dict[str, _Start]:
56 return {p.workflow_id: p for p in ps}
57 
58 
59def test_default_starts_dependency_scan_and_determinism_review():
60 # The defaults — FROOT_LOOPS=dependency-patch, review on, a11y off — start
61 # one acting scan and one advisory review per repo, and nothing else.
62 ps = _plans()
63 assert sorted(p.workflow_type for p in ps) == [
64 "ReviewWorkflow",
65 "ScanWorkflow",
66 ]
67 ids = _by_id(ps)
68 assert scan_workflow_id(REPO, Loop.DEPENDENCY_PATCH) in ids
69 assert review_workflow_id(REPO) in ids
70 
71 
72def test_acting_loop_not_in_froot_loops_is_skipped():
73 # security-patch and dead-code are registered but only start when listed.
74 ps = _plans(loops=(Loop.DEPENDENCY_PATCH,))
75 scan_loops = {p.params.loop for p in ps if isinstance(p.params, ScanParams)}
76 assert scan_loops == {Loop.DEPENDENCY_PATCH}
77 
78 
79def test_advisory_loop_in_froot_loops_does_not_spawn_a_scan():
80 # Routing is by disposition, not FROOT_LOOPS membership: an advisory loop
81 # mistakenly listed in FROOT_LOOPS must never get an acting ScanWorkflow
82 # (the old scan starter would have tried to start one).
83 ps = _plans(
84 loops=(Loop.DEPENDENCY_PATCH, Loop.DETERMINISM_REVIEW),
85 review=False,
86 )
87 assert [p.workflow_type for p in ps] == ["ScanWorkflow"]
88 scan_loops = {p.params.loop for p in ps if isinstance(p.params, ScanParams)}
89 assert scan_loops == {Loop.DEPENDENCY_PATCH}
⋯ 60 lines hidden (lines 90–149)
90 
91 
92def test_every_configured_acting_loop_gets_its_own_scan_namespace():
93 ps = _plans(loops=_ACTING)
94 scan_loops = {p.params.loop for p in ps if isinstance(p.params, ScanParams)}
95 assert scan_loops == set(_ACTING)
96 ids = {p.workflow_id for p in ps}
97 for loop in _ACTING:
98 assert scan_workflow_id(REPO, loop) in ids
99 
100 
101def test_a11y_starts_only_when_enabled():
102 off = {p.workflow_type for p in _plans(a11y=False)}
103 assert "A11yReviewWorkflow" not in off
104 on = _by_id(_plans(a11y=True))
105 assert a11y_review_workflow_id(REPO) in on
106 
107 
108def test_determinism_review_skipped_when_disabled():
109 ps = _plans(review=False)
110 assert all(p.workflow_type != "ReviewWorkflow" for p in ps)
111 
112 
113def test_no_acting_and_no_advisory_yields_nothing():
114 # The empty heartbeat: no configured acting loop, both advisory off.
115 assert _plans(loops=(), review=False, a11y=False) == ()
116 
117 
118def test_params_carry_each_loop_own_interval_and_continuous():
119 ps = _by_id(_plans(loops=(Loop.DEPENDENCY_PATCH,), a11y=True))
120 scan = ps[scan_workflow_id(REPO, Loop.DEPENDENCY_PATCH)]
121 assert isinstance(scan.params, ScanParams)
122 assert scan.params.interval_seconds == 86_400
123 assert scan.params.continuous is True
124 review = ps[review_workflow_id(REPO)]
125 assert isinstance(review.params, ReviewScanParams)
126 assert review.params.interval_seconds == 300
127 a11y = ps[a11y_review_workflow_id(REPO)]
128 assert isinstance(a11y.params, A11yReviewScanParams)
129 assert a11y.params.interval_seconds == 600
130 
131 
132def test_one_plan_per_repo():
133 ps = _plans(repos=(REPO, REPO2), loops=(Loop.DEPENDENCY_PATCH,), a11y=True)
134 # 2 repos x (dependency scan + determinism + a11y) = 6
135 assert len(ps) == 6
136 assert {p.slug for p in ps} == {"mseeks/revisionist", "mseeks/froot"}
137 
138 
139def test_advisory_wiring_routes_each_loop_to_its_own_workflow():
140 w = advisory_wiring(
141 review_enabled=True,
142 review_interval_seconds=1,
143 a11y_enabled=True,
144 a11y_interval_seconds=2,
145 )
146 assert w[Loop.DETERMINISM_REVIEW].workflow_type == "ReviewWorkflow"
147 assert w[Loop.DETERMINISM_REVIEW].params is ReviewScanParams
148 assert w[Loop.A11Y_REVIEW].workflow_type == "A11yReviewWorkflow"
149 assert w[Loop.A11Y_REVIEW].params is A11yReviewScanParams
tests/test_starter.py · 149 lines
tests/test_starter.py149 lines · Python
⋯ 117 lines hidden (lines 1–117)
1"""The unified starter's pure plan — which loops start, with what.
2 
3The starter's I/O (connect, ``start_workflow``) is thin glue; its *decision* is
4the pure :func:`~froot.starter.plans`, derived from the registry and the
5settings. These guard that decision: an acting loop starts iff it is in
6``FROOT_LOOPS``, an advisory loop iff its enable flag is on, and each lands on
7its own workflow type, id, and params — so the one entrypoint covers all five
8loops without a per-family starter.
9"""
10 
11from __future__ import annotations
12 
13from froot.domain.loop import Loop
14from froot.domain.repo import TargetRepo
15from froot.policy.naming import (
16 a11y_review_workflow_id,
17 review_workflow_id,
18 scan_workflow_id,
20from froot.starter import _Start, advisory_wiring, plans
21from froot.workflow.types import (
22 A11yReviewScanParams,
23 ReviewScanParams,
24 ScanParams,
26from tests.support import make_repo
27 
28REPO = make_repo("mseeks/revisionist")
29REPO2 = make_repo("mseeks/froot")
30 
31_ACTING = (Loop.DEPENDENCY_PATCH, Loop.SECURITY_PATCH, Loop.DEAD_CODE)
32 
33 
34def _plans(
35 *,
36 loops: tuple[Loop, ...] = (Loop.DEPENDENCY_PATCH,),
37 repos: tuple[TargetRepo, ...] = (REPO,),
38 review: bool = True,
39 a11y: bool = False,
40 scan_interval: int = 86_400,
41):
42 return plans(
43 repos=repos,
44 loops=loops,
45 scan_interval_seconds=scan_interval,
46 advisory=advisory_wiring(
47 review_enabled=review,
48 review_interval_seconds=300,
49 a11y_enabled=a11y,
50 a11y_interval_seconds=600,
51 ),
52 )
53 
54 
55def _by_id(ps: tuple[_Start, ...]) -> dict[str, _Start]:
56 return {p.workflow_id: p for p in ps}
57 
58 
59def test_default_starts_dependency_scan_and_determinism_review():
60 # The defaults — FROOT_LOOPS=dependency-patch, review on, a11y off — start
61 # one acting scan and one advisory review per repo, and nothing else.
62 ps = _plans()
63 assert sorted(p.workflow_type for p in ps) == [
64 "ReviewWorkflow",
65 "ScanWorkflow",
66 ]
67 ids = _by_id(ps)
68 assert scan_workflow_id(REPO, Loop.DEPENDENCY_PATCH) in ids
69 assert review_workflow_id(REPO) in ids
70 
71 
72def test_acting_loop_not_in_froot_loops_is_skipped():
73 # security-patch and dead-code are registered but only start when listed.
74 ps = _plans(loops=(Loop.DEPENDENCY_PATCH,))
75 scan_loops = {p.params.loop for p in ps if isinstance(p.params, ScanParams)}
76 assert scan_loops == {Loop.DEPENDENCY_PATCH}
77 
78 
79def test_advisory_loop_in_froot_loops_does_not_spawn_a_scan():
80 # Routing is by disposition, not FROOT_LOOPS membership: an advisory loop
81 # mistakenly listed in FROOT_LOOPS must never get an acting ScanWorkflow
82 # (the old scan starter would have tried to start one).
83 ps = _plans(
84 loops=(Loop.DEPENDENCY_PATCH, Loop.DETERMINISM_REVIEW),
85 review=False,
86 )
87 assert [p.workflow_type for p in ps] == ["ScanWorkflow"]
88 scan_loops = {p.params.loop for p in ps if isinstance(p.params, ScanParams)}
89 assert scan_loops == {Loop.DEPENDENCY_PATCH}
90 
91 
92def test_every_configured_acting_loop_gets_its_own_scan_namespace():
93 ps = _plans(loops=_ACTING)
94 scan_loops = {p.params.loop for p in ps if isinstance(p.params, ScanParams)}
95 assert scan_loops == set(_ACTING)
96 ids = {p.workflow_id for p in ps}
97 for loop in _ACTING:
98 assert scan_workflow_id(REPO, loop) in ids
99 
100 
101def test_a11y_starts_only_when_enabled():
102 off = {p.workflow_type for p in _plans(a11y=False)}
103 assert "A11yReviewWorkflow" not in off
104 on = _by_id(_plans(a11y=True))
105 assert a11y_review_workflow_id(REPO) in on
106 
107 
108def test_determinism_review_skipped_when_disabled():
109 ps = _plans(review=False)
110 assert all(p.workflow_type != "ReviewWorkflow" for p in ps)
111 
112 
113def test_no_acting_and_no_advisory_yields_nothing():
114 # The empty heartbeat: no configured acting loop, both advisory off.
115 assert _plans(loops=(), review=False, a11y=False) == ()
116 
117 
118def test_params_carry_each_loop_own_interval_and_continuous():
119 ps = _by_id(_plans(loops=(Loop.DEPENDENCY_PATCH,), a11y=True))
120 scan = ps[scan_workflow_id(REPO, Loop.DEPENDENCY_PATCH)]
121 assert isinstance(scan.params, ScanParams)
122 assert scan.params.interval_seconds == 86_400
123 assert scan.params.continuous is True
124 review = ps[review_workflow_id(REPO)]
125 assert isinstance(review.params, ReviewScanParams)
126 assert review.params.interval_seconds == 300
127 a11y = ps[a11y_review_workflow_id(REPO)]
128 assert isinstance(a11y.params, A11yReviewScanParams)
129 assert a11y.params.interval_seconds == 600
⋯ 20 lines hidden (lines 130–149)
130 
131 
132def test_one_plan_per_repo():
133 ps = _plans(repos=(REPO, REPO2), loops=(Loop.DEPENDENCY_PATCH,), a11y=True)
134 # 2 repos x (dependency scan + determinism + a11y) = 6
135 assert len(ps) == 6
136 assert {p.slug for p in ps} == {"mseeks/revisionist", "mseeks/froot"}
137 
138 
139def test_advisory_wiring_routes_each_loop_to_its_own_workflow():
140 w = advisory_wiring(
141 review_enabled=True,
142 review_interval_seconds=1,
143 a11y_enabled=True,
144 a11y_interval_seconds=2,
145 )
146 assert w[Loop.DETERMINISM_REVIEW].workflow_type == "ReviewWorkflow"
147 assert w[Loop.DETERMINISM_REVIEW].params is ReviewScanParams
148 assert w[Loop.A11Y_REVIEW].workflow_type == "A11yReviewWorkflow"
149 assert w[Loop.A11Y_REVIEW].params is A11yReviewScanParams