What changed, and why
froot's scan loops tick on a daily timer. There was no way to force a scan in between — which is exactly what you want after a fresh deploy, or to chase a finding you just introduced. This adds python -m froot.trigger: a non-destructive "scan now".
The design choice is the interesting part. The obvious mechanism — terminate the running scan loop and let it restart — is destructive (and was blocked as interfering with cluster workloads). A signal that interrupts the loop's sleep would need a non-backward-compatible sleep → wait_condition change plus Temporal patching plus a one-time transition before it even works. Instead, the trigger starts a separate one-shot scan. ScanWorkflow already supports it (continuous=False does a single tick and returns), so this needs zero workflow-code change and works on the currently-deployed image immediately.
The module's own framing: a separate one-shot, distinct id, same idempotent dispatch.
src/froot/trigger.py · 156 lines
src/froot/trigger.py
1"""Trigger an immediate one-shot scan of every configured acting loop.
3Run as a one-shot (a k8s Job, or ``python -m froot.trigger``). For each acting
4(commit-or-revert) loop in ``FROOT_LOOPS`` and each repo in ``FROOT_REPOS`` it
5starts a *separate* one-shot :class:`~froot.workflow.scan_workflow.ScanWorkflow`
6(``continuous=False``) that scans once and returns. It never touches the
7long-lived scan loops: the one-shot carries a distinct ``-now`` workflow id, and
8the bumps it dispatches use the same deterministic ids, so it opens PRs only for
9work the scheduled loop hasn't already handled (no duplicates).
11This is the non-destructive "scan now" the steward reaches for to verify a fresh
12deploy or chase a just-introduced finding, instead of waiting out the daily
13interval — no terminating or interrupting the running loops.
14``FROOT_TRIGGER_LOOPS`` optionally narrows it to specific loops (default: all
15acting loops in ``FROOT_LOOPS``); advisory loops are skipped (no ScanWorkflow).
17Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
18repo list and the loop list), plus the optional ``FROOT_TRIGGER_LOOPS`` filter.
19Re-running is safe — a one-shot still in flight is left to finish.
20"""
⋯ 135 lines hidden (lines 22–156)
22from __future__ import annotations
24import asyncio
25import os
26from dataclasses import dataclass
27from typing import TYPE_CHECKING
29from froot.domain.loop import Loop
30from froot.loops import registry
31from froot.loops.registry import Disposition
32from froot.policy.naming import scan_workflow_id
33from froot.workflow.types import ScanParams
35if TYPE_CHECKING:
36 from froot.domain.repo import TargetRepo
39@dataclass(frozen=True)
40class _OneShot:
41 """One off-cycle scan to start — the pure unit of work.
43 Attributes:
44 params: The one-shot scan input (``continuous=False``).
45 workflow_id: The ``-now`` id, distinct from the long-lived loop's.
46 label: A short human tag for the start log line.
47 slug: The ``owner/name`` this scan is for.
48 """
50 params: ScanParams
51 workflow_id: str
52 label: str
53 slug: str
56def parse_trigger_loops(value: str | None) -> tuple[Loop, ...] | None:
57 """Parse ``FROOT_TRIGGER_LOOPS`` (comma-separated) into a filter.
59 Returns ``None`` (meaning "every configured acting loop") for an empty or
60 unset value; else the named loops. An unknown name raises ``ValueError``
61 (via :class:`Loop`), so a typo fails loudly rather than scanning nothing.
62 """
63 if value is None or not value.strip():
64 return None
65 loops = tuple(
66 Loop(entry.strip()) for entry in value.split(",") if entry.strip()
67 )
68 return loops or None
71def oneshot_plans(
72 *,
73 repos: tuple[TargetRepo, ...],
74 loops: tuple[Loop, ...],
75 only: tuple[Loop, ...] | None = None,
76) -> tuple[_OneShot, ...]:
77 """Every one-shot scan to start (pure), one per (acting loop, repo).
79 An acting loop is included iff it is in ``loops`` and, when ``only`` is
80 given, also in ``only``. Advisory loops are skipped — they have no
81 ScanWorkflow. The set of loops and their families come from the registry,
82 so a new acting loop is reachable here by registration alone.
83 """
84 wanted = set(loops) if only is None else (set(loops) & set(only))
85 out: list[_OneShot] = []
86 for spec in registry.all_specs():
87 loop = spec.loop
88 if (
89 spec.disposition is not Disposition.COMMIT_OR_REVERT
90 or loop not in wanted
91 ):
92 continue
93 for target in repos:
94 out.append(
95 _OneShot(
96 params=ScanParams(
97 target=target, continuous=False, loop=loop
98 ),
99 workflow_id=f"{scan_workflow_id(target, loop)}-now",
100 label=f"{loop.value} scan-now",
101 slug=target.repo.slug,
102 )
103 )
104 return tuple(out)
107async def _trigger() -> None:
108 from temporalio.client import Client
109 from temporalio.common import WorkflowIDReusePolicy
110 from temporalio.contrib.pydantic import pydantic_data_converter
111 from temporalio.exceptions import WorkflowAlreadyStartedError
113 from froot.config.settings import Settings, TemporalSettings
115 settings = Settings() # non-secret values come from the environment
116 plans = oneshot_plans(
117 repos=settings.repos,
118 loops=settings.loops,
119 only=parse_trigger_loops(os.environ.get("FROOT_TRIGGER_LOOPS")),
120 )
121 if not plans:
122 print(
123 "no acting loops to scan "
124 "(check FROOT_LOOPS and FROOT_TRIGGER_LOOPS)"
125 )
126 return
128 temporal = TemporalSettings()
129 client = await Client.connect(
130 temporal.host,
131 namespace=temporal.namespace,
132 data_converter=pydantic_data_converter,
133 )
134 queue = temporal.task_queue
135 for plan in plans:
136 try:
137 handle = await client.start_workflow(
138 "ScanWorkflow",
139 plan.params,
140 id=plan.workflow_id,
141 task_queue=queue,
142 # A finished one-shot can re-run; one still in flight is left.
143 id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
144 )
145 print(f"scan-now {plan.label} {handle.id!r} for {plan.slug}")
146 except WorkflowAlreadyStartedError:
147 print(f"scan-now {plan.workflow_id!r} already in flight — skipped")
150def main() -> None:
151 """Console entrypoint: start a one-shot scan for every configured loop."""
152 asyncio.run(_trigger())
155if __name__ == "__main__":
156 main()
Planning the one-shots
The plan is pure: one one-shot per acting loop × repo, filtered to FROOT_LOOPS (and an optional only). The acting/advisory split comes straight from the registry, so a new acting loop is reachable here by registration alone — advisory loops have no ScanWorkflow and are skipped. Each one-shot carries the loop's scan id with a -now suffix, distinct from the long-lived loop so the two never collide.
One plan per (acting loop, repo); the -now id keeps it apart from the scheduled loop.
src/froot/trigger.py · 156 lines
src/froot/trigger.py
⋯ 70 lines hidden (lines 1–70)
1"""Trigger an immediate one-shot scan of every configured acting loop.
3Run as a one-shot (a k8s Job, or ``python -m froot.trigger``). For each acting
4(commit-or-revert) loop in ``FROOT_LOOPS`` and each repo in ``FROOT_REPOS`` it
5starts a *separate* one-shot :class:`~froot.workflow.scan_workflow.ScanWorkflow`
6(``continuous=False``) that scans once and returns. It never touches the
7long-lived scan loops: the one-shot carries a distinct ``-now`` workflow id, and
8the bumps it dispatches use the same deterministic ids, so it opens PRs only for
9work the scheduled loop hasn't already handled (no duplicates).
11This is the non-destructive "scan now" the steward reaches for to verify a fresh
12deploy or chase a just-introduced finding, instead of waiting out the daily
13interval — no terminating or interrupting the running loops.
14``FROOT_TRIGGER_LOOPS`` optionally narrows it to specific loops (default: all
15acting loops in ``FROOT_LOOPS``); advisory loops are skipped (no ScanWorkflow).
17Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
18repo list and the loop list), plus the optional ``FROOT_TRIGGER_LOOPS`` filter.
19Re-running is safe — a one-shot still in flight is left to finish.
20"""
22from __future__ import annotations
24import asyncio
25import os
26from dataclasses import dataclass
27from typing import TYPE_CHECKING
29from froot.domain.loop import Loop
30from froot.loops import registry
31from froot.loops.registry import Disposition
32from froot.policy.naming import scan_workflow_id
33from froot.workflow.types import ScanParams
35if TYPE_CHECKING:
36 from froot.domain.repo import TargetRepo
39@dataclass(frozen=True)
40class _OneShot:
41 """One off-cycle scan to start — the pure unit of work.
43 Attributes:
44 params: The one-shot scan input (``continuous=False``).
45 workflow_id: The ``-now`` id, distinct from the long-lived loop's.
46 label: A short human tag for the start log line.
47 slug: The ``owner/name`` this scan is for.
48 """
50 params: ScanParams
51 workflow_id: str
52 label: str
53 slug: str
56def parse_trigger_loops(value: str | None) -> tuple[Loop, ...] | None:
57 """Parse ``FROOT_TRIGGER_LOOPS`` (comma-separated) into a filter.
59 Returns ``None`` (meaning "every configured acting loop") for an empty or
60 unset value; else the named loops. An unknown name raises ``ValueError``
61 (via :class:`Loop`), so a typo fails loudly rather than scanning nothing.
62 """
63 if value is None or not value.strip():
64 return None
65 loops = tuple(
66 Loop(entry.strip()) for entry in value.split(",") if entry.strip()
67 )
68 return loops or None
71def oneshot_plans(
72 *,
73 repos: tuple[TargetRepo, ...],
74 loops: tuple[Loop, ...],
75 only: tuple[Loop, ...] | None = None,
76) -> tuple[_OneShot, ...]:
77 """Every one-shot scan to start (pure), one per (acting loop, repo).
79 An acting loop is included iff it is in ``loops`` and, when ``only`` is
80 given, also in ``only``. Advisory loops are skipped — they have no
81 ScanWorkflow. The set of loops and their families come from the registry,
82 so a new acting loop is reachable here by registration alone.
83 """
84 wanted = set(loops) if only is None else (set(loops) & set(only))
85 out: list[_OneShot] = []
86 for spec in registry.all_specs():
87 loop = spec.loop
88 if (
89 spec.disposition is not Disposition.COMMIT_OR_REVERT
90 or loop not in wanted
91 ):
92 continue
93 for target in repos:
94 out.append(
95 _OneShot(
96 params=ScanParams(
97 target=target, continuous=False, loop=loop
98 ),
99 workflow_id=f"{scan_workflow_id(target, loop)}-now",
100 label=f"{loop.value} scan-now",
101 slug=target.repo.slug,
102 )
103 )
104 return tuple(out)
⋯ 51 lines hidden (lines 106–156)
107async def _trigger() -> None:
108 from temporalio.client import Client
109 from temporalio.common import WorkflowIDReusePolicy
110 from temporalio.contrib.pydantic import pydantic_data_converter
111 from temporalio.exceptions import WorkflowAlreadyStartedError
113 from froot.config.settings import Settings, TemporalSettings
115 settings = Settings() # non-secret values come from the environment
116 plans = oneshot_plans(
117 repos=settings.repos,
118 loops=settings.loops,
119 only=parse_trigger_loops(os.environ.get("FROOT_TRIGGER_LOOPS")),
120 )
121 if not plans:
122 print(
123 "no acting loops to scan "
124 "(check FROOT_LOOPS and FROOT_TRIGGER_LOOPS)"
125 )
126 return
128 temporal = TemporalSettings()
129 client = await Client.connect(
130 temporal.host,
131 namespace=temporal.namespace,
132 data_converter=pydantic_data_converter,
133 )
134 queue = temporal.task_queue
135 for plan in plans:
136 try:
137 handle = await client.start_workflow(
138 "ScanWorkflow",
139 plan.params,
140 id=plan.workflow_id,
141 task_queue=queue,
142 # A finished one-shot can re-run; one still in flight is left.
143 id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
144 )
145 print(f"scan-now {plan.label} {handle.id!r} for {plan.slug}")
146 except WorkflowAlreadyStartedError:
147 print(f"scan-now {plan.workflow_id!r} already in flight — skipped")
150def main() -> None:
151 """Console entrypoint: start a one-shot scan for every configured loop."""
152 asyncio.run(_trigger())
155if __name__ == "__main__":
156 main()
Starting them, and operating it
Each one-shot starts with ALLOW_DUPLICATE reuse: a finished one re-runs on the next trigger, and one still in flight is left to finish (caught and skipped). The connection mirrors the starter (the pydantic data converter, the same task queue).
Connect, start each one-shot, tolerate one already in flight.
src/froot/trigger.py · 156 lines
src/froot/trigger.py
⋯ 106 lines hidden (lines 1–106)
1"""Trigger an immediate one-shot scan of every configured acting loop.
3Run as a one-shot (a k8s Job, or ``python -m froot.trigger``). For each acting
4(commit-or-revert) loop in ``FROOT_LOOPS`` and each repo in ``FROOT_REPOS`` it
5starts a *separate* one-shot :class:`~froot.workflow.scan_workflow.ScanWorkflow`
6(``continuous=False``) that scans once and returns. It never touches the
7long-lived scan loops: the one-shot carries a distinct ``-now`` workflow id, and
8the bumps it dispatches use the same deterministic ids, so it opens PRs only for
9work the scheduled loop hasn't already handled (no duplicates).
11This is the non-destructive "scan now" the steward reaches for to verify a fresh
12deploy or chase a just-introduced finding, instead of waiting out the daily
13interval — no terminating or interrupting the running loops.
14``FROOT_TRIGGER_LOOPS`` optionally narrows it to specific loops (default: all
15acting loops in ``FROOT_LOOPS``); advisory loops are skipped (no ScanWorkflow).
17Config from the environment: ``TEMPORAL_*`` and the ``FROOT_*`` settings (the
18repo list and the loop list), plus the optional ``FROOT_TRIGGER_LOOPS`` filter.
19Re-running is safe — a one-shot still in flight is left to finish.
20"""
22from __future__ import annotations
24import asyncio
25import os
26from dataclasses import dataclass
27from typing import TYPE_CHECKING
29from froot.domain.loop import Loop
30from froot.loops import registry
31from froot.loops.registry import Disposition
32from froot.policy.naming import scan_workflow_id
33from froot.workflow.types import ScanParams
35if TYPE_CHECKING:
36 from froot.domain.repo import TargetRepo
39@dataclass(frozen=True)
40class _OneShot:
41 """One off-cycle scan to start — the pure unit of work.
43 Attributes:
44 params: The one-shot scan input (``continuous=False``).
45 workflow_id: The ``-now`` id, distinct from the long-lived loop's.
46 label: A short human tag for the start log line.
47 slug: The ``owner/name`` this scan is for.
48 """
50 params: ScanParams
51 workflow_id: str
52 label: str
53 slug: str
56def parse_trigger_loops(value: str | None) -> tuple[Loop, ...] | None:
57 """Parse ``FROOT_TRIGGER_LOOPS`` (comma-separated) into a filter.
59 Returns ``None`` (meaning "every configured acting loop") for an empty or
60 unset value; else the named loops. An unknown name raises ``ValueError``
61 (via :class:`Loop`), so a typo fails loudly rather than scanning nothing.
62 """
63 if value is None or not value.strip():
64 return None
65 loops = tuple(
66 Loop(entry.strip()) for entry in value.split(",") if entry.strip()
67 )
68 return loops or None
71def oneshot_plans(
72 *,
73 repos: tuple[TargetRepo, ...],
74 loops: tuple[Loop, ...],
75 only: tuple[Loop, ...] | None = None,
76) -> tuple[_OneShot, ...]:
77 """Every one-shot scan to start (pure), one per (acting loop, repo).
79 An acting loop is included iff it is in ``loops`` and, when ``only`` is
80 given, also in ``only``. Advisory loops are skipped — they have no
81 ScanWorkflow. The set of loops and their families come from the registry,
82 so a new acting loop is reachable here by registration alone.
83 """
84 wanted = set(loops) if only is None else (set(loops) & set(only))
85 out: list[_OneShot] = []
86 for spec in registry.all_specs():
87 loop = spec.loop
88 if (
89 spec.disposition is not Disposition.COMMIT_OR_REVERT
90 or loop not in wanted
91 ):
92 continue
93 for target in repos:
94 out.append(
95 _OneShot(
96 params=ScanParams(
97 target=target, continuous=False, loop=loop
98 ),
99 workflow_id=f"{scan_workflow_id(target, loop)}-now",
100 label=f"{loop.value} scan-now",
101 slug=target.repo.slug,
102 )
103 )
104 return tuple(out)
107async def _trigger() -> None:
108 from temporalio.client import Client
109 from temporalio.common import WorkflowIDReusePolicy
110 from temporalio.contrib.pydantic import pydantic_data_converter
111 from temporalio.exceptions import WorkflowAlreadyStartedError
113 from froot.config.settings import Settings, TemporalSettings
115 settings = Settings() # non-secret values come from the environment
116 plans = oneshot_plans(
117 repos=settings.repos,
118 loops=settings.loops,
119 only=parse_trigger_loops(os.environ.get("FROOT_TRIGGER_LOOPS")),
120 )
121 if not plans:
122 print(
123 "no acting loops to scan "
124 "(check FROOT_LOOPS and FROOT_TRIGGER_LOOPS)"
125 )
126 return
128 temporal = TemporalSettings()
129 client = await Client.connect(
130 temporal.host,
131 namespace=temporal.namespace,
132 data_converter=pydantic_data_converter,
133 )
134 queue = temporal.task_queue
135 for plan in plans:
136 try:
137 handle = await client.start_workflow(
138 "ScanWorkflow",
139 plan.params,
140 id=plan.workflow_id,
141 task_queue=queue,
142 # A finished one-shot can re-run; one still in flight is left.
143 id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
144 )
145 print(f"scan-now {plan.label} {handle.id!r} for {plan.slug}")
146 except WorkflowAlreadyStartedError:
147 print(f"scan-now {plan.workflow_id!r} already in flight — skipped")
Operationally it's a k8s manage Job (infra/k8s/froot/manage/scan-now.yaml, in the infra repo) that runs python -m froot.trigger, mirroring start-loops: kubectl apply -f manage/scan-now.yaml. FROOT_TRIGGER_LOOPS narrows it to specific loops (e.g. just dead-code); empty scans every acting loop. Pure tests cover the plan and filter logic; full suite green at 517 tests + both determinism gates.