Why this PR exists

The determinism kernel's only CI is check_determinism.py src — run against clean code. That proves the gate passes on clean code; it never proves the gate fails on a hazard. For a blocking gate, the catch-property being pinned only by a since-deleted dev fixture is exactly what silently rots.

The kernel is a vendored, package-free script (it must run standalone in any repo's CI), so the test loads it by file path and calls into it directly. Registering it in sys.modules before exec_module is what lets the script's @dataclass resolve its own module.

loading the vendored script · 206 lines
loading the vendored script206 lines · Python
⋯ 29 lines hidden (lines 1–29)
1"""Catch-tests for the vendored kernel script, and the kernel/brain boundary.
2 
3``scripts/check_determinism.py`` runs standalone in CI (``check_determinism.py
4src``), which only proves it PASSES on clean code. These tests pin the other
5half — that it actually FAILS on hazards, across every banned category — and
6demonstrate the one thing the lexical kernel structurally cannot see: a
7transitive hazard, which only the brain's analyzer catches.
8 
9The kernel is a vendored, package-free script, so it is loaded by file path.
10"""
11 
12from __future__ import annotations
13 
14import ast
15import importlib.util
16import sys
17from pathlib import Path
18from typing import TYPE_CHECKING, Any
19 
20from froot.policy.determinism import LoadedModule, analyze_workflow_surface
21 
22if TYPE_CHECKING:
23 from collections.abc import Mapping
24 
25_SCRIPT = (
26 Path(__file__).resolve().parents[1] / "scripts" / "check_determinism.py"
28 
29 
30def _load_kernel() -> Any:
31 spec = importlib.util.spec_from_file_location("_froot_kernel", _SCRIPT)
32 assert spec is not None
33 assert spec.loader is not None
34 module = importlib.util.module_from_spec(spec)
35 # Register before exec so the script's @dataclass can resolve its module.
36 sys.modules[spec.name] = module
37 spec.loader.exec_module(module)
38 return module
39 
40 
41_KERNEL = _load_kernel()
⋯ 165 lines hidden (lines 42–206)
42 
43 
44# A workflow class touching every banned category, including the tricky
45# resolutions: an aliased import (dt.now) and a from-import (date.today).
46_HAZARDS = """
47import datetime
48import time
49import random
50import uuid
51import os
52import asyncio
53import threading
54import subprocess
55import socket
56import httpx
57from datetime import datetime as dt, date
58from temporalio import workflow
59 
60 
61@workflow.defn
62class Hazards:
63 @workflow.run
64 async def run(self) -> None:
65 datetime.datetime.now()
66 datetime.datetime.utcnow()
67 date.today()
68 dt.now()
69 time.time()
70 time.sleep(1)
71 random.randint(1, 6)
72 uuid.uuid4()
73 os.getenv("X")
74 os.environ["Y"]
75 await asyncio.sleep(1)
76 threading.Thread()
77 subprocess.run(["ls"])
78 socket.socket()
79 httpx.get("http://x")
80 open("/tmp/x")
81"""
82 
83# Things that LOOK like hazards but aren't: the sanctioned `workflow.*`
84# replacements, a type annotation, and wall-clock inside an @activity.defn.
85_NEGATIVES = """
86import datetime
87import os
88from temporalio import workflow, activity
89 
90 
91@workflow.defn
92class Clean:
93 @workflow.run
94 async def run(self) -> datetime.datetime: # annotation, not a call
95 workflow.now()
96 workflow.uuid4()
97 workflow.random()
98 return workflow.now()
99 
100 
101@activity.defn
102async def legit() -> None:
103 datetime.datetime.now() # wall-clock is legal in an activity
104 os.environ["Z"]
105"""
106 
107 
108def _write(tmp_path: Path, name: str, source: str) -> Path:
109 path = tmp_path / name
110 path.write_text(source)
111 return path
112 
113 
114def test_kernel_catches_every_banned_category(tmp_path: Path) -> None:
115 path = _write(tmp_path, "wf.py", _HAZARDS)
116 rules = {finding.rule for finding in _KERNEL.scan_file(path)}
117 assert rules == {
118 "datetime.datetime.now", # plain and via the `dt` alias
119 "datetime.datetime.utcnow",
120 "datetime.date.today", # via the `date` from-import
121 "time.time",
122 "time.sleep",
123 "random.randint",
124 "uuid.uuid4",
125 "os.getenv",
126 "os.environ", # attribute access, not a call
127 "asyncio.sleep",
128 "threading.Thread",
129 "subprocess.run",
130 "socket.socket",
131 "httpx.get",
132 "open",
133 }
134 
135 
136def test_kernel_exit_code_is_nonzero_on_hazards(tmp_path: Path) -> None:
137 path = _write(tmp_path, "wf.py", _HAZARDS)
138 assert _KERNEL.main([str(path)]) == 1
139 
140 
141def test_kernel_ignores_sanctioned_annotations_and_activities(
142 tmp_path: Path,
143) -> None:
144 path = _write(tmp_path, "wf.py", _NEGATIVES)
145 assert _KERNEL.scan_file(path) == []
146 assert _KERNEL.main([str(path)]) == 0
147 
148 
149# ── The kernel/brain boundary — only the brain catches this ──────────────────
150_TRANSITIVE_WORKFLOW = """
151from temporalio import workflow
152from app.util import stamp
153 
154 
155@workflow.defn
156class W:
157 @workflow.run
158 async def run(self):
159 return stamp()
160"""
161 
162_TRANSITIVE_HELPER = """
163import datetime
164 
165 
166def stamp():
167 return datetime.datetime.now()
168"""
169 
170 
171def _modules(sources: Mapping[str, str]) -> dict[str, LoadedModule]:
172 return {
173 qual: LoadedModule(
174 qualname=qual,
175 tree=ast.parse(src),
176 lines=tuple(src.splitlines()),
177 )
178 for qual, src in sources.items()
179 }
180 
181 
182def test_transitive_hazard_is_invisible_to_kernel_but_caught_by_brain(
183 tmp_path: Path,
184) -> None:
185 # Same code, two layers. The workflow file has NO banned call lexically —
186 # the hazard hides one call out, in a plain helper — and the helper file
187 # has no @workflow.defn, so the kernel ignores it there too. The lexical,
188 # workflow-scoped kernel is structurally blind to this.
189 wf_path = _write(tmp_path, "workflow.py", _TRANSITIVE_WORKFLOW)
190 helper_path = _write(tmp_path, "util.py", _TRANSITIVE_HELPER)
191 assert _KERNEL.scan_file(wf_path) == []
192 assert _KERNEL.scan_file(helper_path) == []
193 
194 # The brain's call-graph chases the import into the helper and finds it.
195 result = analyze_workflow_surface(
196 _modules(
197 {
198 "app.workflow": _TRANSITIVE_WORKFLOW,
199 "app.util": _TRANSITIVE_HELPER,
200 }
201 )
202 )
203 assert len(result.hazards) == 1
204 hazard = result.hazards[0]
205 assert hazard.impurity.rule == "datetime.datetime.now"
206 assert hazard.via == ("stamp",)

Pinning the catch-property

One workflow class touches all fifteen banned rules — including the resolutions a naive grep would miss: an aliased import (dt.now()) and a from-import (date.today()). The test asserts the exact rule set, so a regression that drops a category fails loudly.

every banned category, asserted exactly · 206 lines
every banned category, asserted exactly206 lines · Python
⋯ 113 lines hidden (lines 1–113)
1"""Catch-tests for the vendored kernel script, and the kernel/brain boundary.
2 
3``scripts/check_determinism.py`` runs standalone in CI (``check_determinism.py
4src``), which only proves it PASSES on clean code. These tests pin the other
5half — that it actually FAILS on hazards, across every banned category — and
6demonstrate the one thing the lexical kernel structurally cannot see: a
7transitive hazard, which only the brain's analyzer catches.
8 
9The kernel is a vendored, package-free script, so it is loaded by file path.
10"""
11 
12from __future__ import annotations
13 
14import ast
15import importlib.util
16import sys
17from pathlib import Path
18from typing import TYPE_CHECKING, Any
19 
20from froot.policy.determinism import LoadedModule, analyze_workflow_surface
21 
22if TYPE_CHECKING:
23 from collections.abc import Mapping
24 
25_SCRIPT = (
26 Path(__file__).resolve().parents[1] / "scripts" / "check_determinism.py"
28 
29 
30def _load_kernel() -> Any:
31 spec = importlib.util.spec_from_file_location("_froot_kernel", _SCRIPT)
32 assert spec is not None
33 assert spec.loader is not None
34 module = importlib.util.module_from_spec(spec)
35 # Register before exec so the script's @dataclass can resolve its module.
36 sys.modules[spec.name] = module
37 spec.loader.exec_module(module)
38 return module
39 
40 
41_KERNEL = _load_kernel()
42 
43 
44# A workflow class touching every banned category, including the tricky
45# resolutions: an aliased import (dt.now) and a from-import (date.today).
46_HAZARDS = """
47import datetime
48import time
49import random
50import uuid
51import os
52import asyncio
53import threading
54import subprocess
55import socket
56import httpx
57from datetime import datetime as dt, date
58from temporalio import workflow
59 
60 
61@workflow.defn
62class Hazards:
63 @workflow.run
64 async def run(self) -> None:
65 datetime.datetime.now()
66 datetime.datetime.utcnow()
67 date.today()
68 dt.now()
69 time.time()
70 time.sleep(1)
71 random.randint(1, 6)
72 uuid.uuid4()
73 os.getenv("X")
74 os.environ["Y"]
75 await asyncio.sleep(1)
76 threading.Thread()
77 subprocess.run(["ls"])
78 socket.socket()
79 httpx.get("http://x")
80 open("/tmp/x")
81"""
82 
83# Things that LOOK like hazards but aren't: the sanctioned `workflow.*`
84# replacements, a type annotation, and wall-clock inside an @activity.defn.
85_NEGATIVES = """
86import datetime
87import os
88from temporalio import workflow, activity
89 
90 
91@workflow.defn
92class Clean:
93 @workflow.run
94 async def run(self) -> datetime.datetime: # annotation, not a call
95 workflow.now()
96 workflow.uuid4()
97 workflow.random()
98 return workflow.now()
99 
100 
101@activity.defn
102async def legit() -> None:
103 datetime.datetime.now() # wall-clock is legal in an activity
104 os.environ["Z"]
105"""
106 
107 
108def _write(tmp_path: Path, name: str, source: str) -> Path:
109 path = tmp_path / name
110 path.write_text(source)
111 return path
112 
113 
114def test_kernel_catches_every_banned_category(tmp_path: Path) -> None:
115 path = _write(tmp_path, "wf.py", _HAZARDS)
116 rules = {finding.rule for finding in _KERNEL.scan_file(path)}
117 assert rules == {
118 "datetime.datetime.now", # plain and via the `dt` alias
119 "datetime.datetime.utcnow",
120 "datetime.date.today", # via the `date` from-import
121 "time.time",
122 "time.sleep",
123 "random.randint",
124 "uuid.uuid4",
125 "os.getenv",
126 "os.environ", # attribute access, not a call
127 "asyncio.sleep",
128 "threading.Thread",
129 "subprocess.run",
130 "socket.socket",
131 "httpx.get",
132 "open",
133 }
134 
⋯ 72 lines hidden (lines 135–206)
135 
136def test_kernel_exit_code_is_nonzero_on_hazards(tmp_path: Path) -> None:
137 path = _write(tmp_path, "wf.py", _HAZARDS)
138 assert _KERNEL.main([str(path)]) == 1
139 
140 
141def test_kernel_ignores_sanctioned_annotations_and_activities(
142 tmp_path: Path,
143) -> None:
144 path = _write(tmp_path, "wf.py", _NEGATIVES)
145 assert _KERNEL.scan_file(path) == []
146 assert _KERNEL.main([str(path)]) == 0
147 
148 
149# ── The kernel/brain boundary — only the brain catches this ──────────────────
150_TRANSITIVE_WORKFLOW = """
151from temporalio import workflow
152from app.util import stamp
153 
154 
155@workflow.defn
156class W:
157 @workflow.run
158 async def run(self):
159 return stamp()
160"""
161 
162_TRANSITIVE_HELPER = """
163import datetime
164 
165 
166def stamp():
167 return datetime.datetime.now()
168"""
169 
170 
171def _modules(sources: Mapping[str, str]) -> dict[str, LoadedModule]:
172 return {
173 qual: LoadedModule(
174 qualname=qual,
175 tree=ast.parse(src),
176 lines=tuple(src.splitlines()),
177 )
178 for qual, src in sources.items()
179 }
180 
181 
182def test_transitive_hazard_is_invisible_to_kernel_but_caught_by_brain(
183 tmp_path: Path,
184) -> None:
185 # Same code, two layers. The workflow file has NO banned call lexically —
186 # the hazard hides one call out, in a plain helper — and the helper file
187 # has no @workflow.defn, so the kernel ignores it there too. The lexical,
188 # workflow-scoped kernel is structurally blind to this.
189 wf_path = _write(tmp_path, "workflow.py", _TRANSITIVE_WORKFLOW)
190 helper_path = _write(tmp_path, "util.py", _TRANSITIVE_HELPER)
191 assert _KERNEL.scan_file(wf_path) == []
192 assert _KERNEL.scan_file(helper_path) == []
193 
194 # The brain's call-graph chases the import into the helper and finds it.
195 result = analyze_workflow_surface(
196 _modules(
197 {
198 "app.workflow": _TRANSITIVE_WORKFLOW,
199 "app.util": _TRANSITIVE_HELPER,
200 }
201 )
202 )
203 assert len(result.hazards) == 1
204 hazard = result.hazards[0]
205 assert hazard.impurity.rule == "datetime.datetime.now"
206 assert hazard.via == ("stamp",)

Two companions: main([...]) == 1 pins the gate's actual contract (a non-zero exit blocks the merge), and a negatives test pins zero false positives on the sanctioned workflow.* replacements, a -> datetime.datetime annotation, and wall-clock inside an @activity.defn (legal there).

The one only the brain can catch

The differential test runs the same code through both layers. A workflow calls a helper one call out; the helper hides datetime.datetime.now(). The workflow file has no banned call lexically, and the helper file has no @workflow.defn — so the lexical, workflow-scoped kernel reports both files clean. The brain's call-graph follows the import and catches it.

kernel blind, brain catches · 206 lines
kernel blind, brain catches206 lines · Python
⋯ 181 lines hidden (lines 1–181)
1"""Catch-tests for the vendored kernel script, and the kernel/brain boundary.
2 
3``scripts/check_determinism.py`` runs standalone in CI (``check_determinism.py
4src``), which only proves it PASSES on clean code. These tests pin the other
5half — that it actually FAILS on hazards, across every banned category — and
6demonstrate the one thing the lexical kernel structurally cannot see: a
7transitive hazard, which only the brain's analyzer catches.
8 
9The kernel is a vendored, package-free script, so it is loaded by file path.
10"""
11 
12from __future__ import annotations
13 
14import ast
15import importlib.util
16import sys
17from pathlib import Path
18from typing import TYPE_CHECKING, Any
19 
20from froot.policy.determinism import LoadedModule, analyze_workflow_surface
21 
22if TYPE_CHECKING:
23 from collections.abc import Mapping
24 
25_SCRIPT = (
26 Path(__file__).resolve().parents[1] / "scripts" / "check_determinism.py"
28 
29 
30def _load_kernel() -> Any:
31 spec = importlib.util.spec_from_file_location("_froot_kernel", _SCRIPT)
32 assert spec is not None
33 assert spec.loader is not None
34 module = importlib.util.module_from_spec(spec)
35 # Register before exec so the script's @dataclass can resolve its module.
36 sys.modules[spec.name] = module
37 spec.loader.exec_module(module)
38 return module
39 
40 
41_KERNEL = _load_kernel()
42 
43 
44# A workflow class touching every banned category, including the tricky
45# resolutions: an aliased import (dt.now) and a from-import (date.today).
46_HAZARDS = """
47import datetime
48import time
49import random
50import uuid
51import os
52import asyncio
53import threading
54import subprocess
55import socket
56import httpx
57from datetime import datetime as dt, date
58from temporalio import workflow
59 
60 
61@workflow.defn
62class Hazards:
63 @workflow.run
64 async def run(self) -> None:
65 datetime.datetime.now()
66 datetime.datetime.utcnow()
67 date.today()
68 dt.now()
69 time.time()
70 time.sleep(1)
71 random.randint(1, 6)
72 uuid.uuid4()
73 os.getenv("X")
74 os.environ["Y"]
75 await asyncio.sleep(1)
76 threading.Thread()
77 subprocess.run(["ls"])
78 socket.socket()
79 httpx.get("http://x")
80 open("/tmp/x")
81"""
82 
83# Things that LOOK like hazards but aren't: the sanctioned `workflow.*`
84# replacements, a type annotation, and wall-clock inside an @activity.defn.
85_NEGATIVES = """
86import datetime
87import os
88from temporalio import workflow, activity
89 
90 
91@workflow.defn
92class Clean:
93 @workflow.run
94 async def run(self) -> datetime.datetime: # annotation, not a call
95 workflow.now()
96 workflow.uuid4()
97 workflow.random()
98 return workflow.now()
99 
100 
101@activity.defn
102async def legit() -> None:
103 datetime.datetime.now() # wall-clock is legal in an activity
104 os.environ["Z"]
105"""
106 
107 
108def _write(tmp_path: Path, name: str, source: str) -> Path:
109 path = tmp_path / name
110 path.write_text(source)
111 return path
112 
113 
114def test_kernel_catches_every_banned_category(tmp_path: Path) -> None:
115 path = _write(tmp_path, "wf.py", _HAZARDS)
116 rules = {finding.rule for finding in _KERNEL.scan_file(path)}
117 assert rules == {
118 "datetime.datetime.now", # plain and via the `dt` alias
119 "datetime.datetime.utcnow",
120 "datetime.date.today", # via the `date` from-import
121 "time.time",
122 "time.sleep",
123 "random.randint",
124 "uuid.uuid4",
125 "os.getenv",
126 "os.environ", # attribute access, not a call
127 "asyncio.sleep",
128 "threading.Thread",
129 "subprocess.run",
130 "socket.socket",
131 "httpx.get",
132 "open",
133 }
134 
135 
136def test_kernel_exit_code_is_nonzero_on_hazards(tmp_path: Path) -> None:
137 path = _write(tmp_path, "wf.py", _HAZARDS)
138 assert _KERNEL.main([str(path)]) == 1
139 
140 
141def test_kernel_ignores_sanctioned_annotations_and_activities(
142 tmp_path: Path,
143) -> None:
144 path = _write(tmp_path, "wf.py", _NEGATIVES)
145 assert _KERNEL.scan_file(path) == []
146 assert _KERNEL.main([str(path)]) == 0
147 
148 
149# ── The kernel/brain boundary — only the brain catches this ──────────────────
150_TRANSITIVE_WORKFLOW = """
151from temporalio import workflow
152from app.util import stamp
153 
154 
155@workflow.defn
156class W:
157 @workflow.run
158 async def run(self):
159 return stamp()
160"""
161 
162_TRANSITIVE_HELPER = """
163import datetime
164 
165 
166def stamp():
167 return datetime.datetime.now()
168"""
169 
170 
171def _modules(sources: Mapping[str, str]) -> dict[str, LoadedModule]:
172 return {
173 qual: LoadedModule(
174 qualname=qual,
175 tree=ast.parse(src),
176 lines=tuple(src.splitlines()),
177 )
178 for qual, src in sources.items()
179 }
180 
181 
182def test_transitive_hazard_is_invisible_to_kernel_but_caught_by_brain(
183 tmp_path: Path,
184) -> None:
185 # Same code, two layers. The workflow file has NO banned call lexically —
186 # the hazard hides one call out, in a plain helper — and the helper file
187 # has no @workflow.defn, so the kernel ignores it there too. The lexical,
188 # workflow-scoped kernel is structurally blind to this.
189 wf_path = _write(tmp_path, "workflow.py", _TRANSITIVE_WORKFLOW)
190 helper_path = _write(tmp_path, "util.py", _TRANSITIVE_HELPER)
191 assert _KERNEL.scan_file(wf_path) == []
192 assert _KERNEL.scan_file(helper_path) == []
193 
194 # The brain's call-graph chases the import into the helper and finds it.
195 result = analyze_workflow_surface(
196 _modules(
197 {
198 "app.workflow": _TRANSITIVE_WORKFLOW,
199 "app.util": _TRANSITIVE_HELPER,
200 }
201 )
202 )
203 assert len(result.hazards) == 1
204 hazard = result.hazards[0]
205 assert hazard.impurity.rule == "datetime.datetime.now"
206 assert hazard.via == ("stamp",)