Overview
This PR adds the kernel of the Temporal determinism review: a stdlib-only AST check that fails when a known-nondeterministic call appears lexically inside an @workflow.defn class. It is two files — the checker (scripts/check_determinism.py) and a dedicated Determinism CI workflow that runs it on every push and PR.
The design choice that defines the kernel: it looks only at code written directly in a workflow class body, never chasing imports or helper calls. That trades recall for precision — it will miss a hazard hidden three modules down, but it will almost never flag something that is fine.
The gate wiring
A dedicated workflow, not a step folded into the existing CI job — so it surfaces as a distinct, independently-gateable check named Determinism. The script needs no install (stdlib ast only), so the job is just checkout → setup-python → run.
.github/workflows/determinism.yml · 28 lines
.github/workflows/determinism.yml
1name: Determinism
3on:
4 push:
5 branches: [main]
6 pull_request:
8# Cancel superseded runs on the same ref.
9concurrency:
10 group: determinism-${{ github.ref }}
11 cancel-in-progress: true
13permissions:
14 contents: read
16jobs:
17 determinism:
18 name: temporal determinism (workflow boundary)
19 runs-on: ubuntu-latest
20 steps:
21 - uses: actions/checkout@v4
22 - uses: actions/setup-python@v5
23 with:
24 python-version: "3.13"
25 # Stdlib-only AST check — no install needed. Flags known-nondeterministic
26 # calls written lexically inside an @workflow.defn class. High precision by
27 # design (lexical scope only); transitive analysis is a later advisory ring.
28 - run: python scripts/check_determinism.py src
The banned table
Four small tables drive everything. Each entry maps a hazard to its sanctioned replacement, so a failure tells the author exactly what to do.
The four banned tables · 242 lines
The four banned tables
⋯ 37 lines hidden (lines 1–37)
1#!/usr/bin/env python3
2"""Temporal determinism kernel — the lexical workflow-boundary check.
4Fails when a known-nondeterministic call appears *lexically inside* a
5``@workflow.defn`` class. This is the high-precision / low-recall seed of the
6determinism review: it inspects only code written directly in a workflow class
7body — not transitive helpers or imported functions (that is the "brain's" job,
8a later ring) — so a *blocking* CI gate built on it has near-zero false
9positives, which is the property a gate must have or humans disable it.
11Scope: every ``@workflow.defn`` class under the given paths. Activities
12(``@activity.defn``), client setup, type modules, and module-level helpers are
13out of scope by construction — nondeterminism is legal there.
15Usage:
16 python check_determinism.py [--json] PATH [PATH ...]
18Exit 0 if clean, 1 if any hazard is found. ``--json`` prints the findings as a
19JSON array (file, line, col, rule, hint, source) — the structured seam a later
20analysis ring consumes instead of re-parsing.
21"""
23from __future__ import annotations
25import argparse
26import ast
27import json
28import sys
29from dataclasses import asdict, dataclass
30from pathlib import Path
32# --- the banned table -------------------------------------------------------
33# Each entry maps a resolved callee / attribute to its sanctioned replacement.
34# Kept deliberately tight: every entry must be a near-certain hazard inside a
35# workflow body, so the gate stays precise enough to block on.
37# Exact-match callee paths (resolved through the file's imports).
38BANNED_CALLS: dict[str, str] = {
39 "datetime.datetime.now": "use workflow.now()",
40 "datetime.datetime.utcnow": "use workflow.now()",
41 "datetime.datetime.today": "use workflow.now()",
42 "datetime.date.today": "use workflow.now().date()",
43 "time.time": "use workflow.now()",
44 "time.time_ns": "use workflow.now()",
45 "time.monotonic": "use workflow.now()",
46 "time.monotonic_ns": "use workflow.now()",
47 "time.perf_counter": "use workflow.now()",
48 "time.perf_counter_ns": "use workflow.now()",
49 "time.process_time": "use workflow.now()",
50 "time.sleep": "use workflow.sleep()",
51 "uuid.uuid1": "use workflow.uuid4()",
52 "uuid.uuid3": "use workflow.uuid4()",
53 "uuid.uuid4": "use workflow.uuid4()",
54 "uuid.uuid5": "use workflow.uuid4()",
55 "os.getenv": "read env in an activity and pass it in",
56 "os.urandom": "use workflow.random()",
57 "asyncio.sleep": "use workflow.sleep()",
58 "socket.socket": "do network I/O in an activity",
59}
61# Any call whose resolved root module is one of these (module-wide impurity).
62# Only modules with *no* deterministic surface worth keeping belong here.
63BANNED_CALL_MODULES: dict[str, str] = {
64 "random": "use workflow.random()",
65 "threading": "workflows are single-threaded; use workflow APIs",
66 "requests": "do network I/O in an activity",
67 "httpx": "do network I/O in an activity",
68 "subprocess": "run external processes in an activity",
69}
71# Attribute *accesses* (need not be calls), e.g. os.environ["X"].
72BANNED_ATTRS: dict[str, str] = {
73 "os.environ": "read env in an activity and pass it in",
74}
76# Bare builtin names — flagged only when not shadowed by an import binding.
77BANNED_BUILTINS: dict[str, str] = {
78 "open": "do file I/O in an activity",
79 "input": "workflows cannot block on input",
80}
⋯ 162 lines hidden (lines 81–242)
83@dataclass(frozen=True)
84class Finding:
85 file: str
86 line: int
87 col: int
88 rule: str # the resolved hazard, e.g. "datetime.datetime.now"
89 hint: str
90 source: str
93def _import_map(tree: ast.Module) -> dict[str, str]:
94 """Map each bound name to the dotted module/symbol path it refers to."""
95 bindings: dict[str, str] = {}
96 for node in ast.walk(tree):
97 if isinstance(node, ast.Import):
98 for alias in node.names:
99 if alias.asname:
100 bindings[alias.asname] = alias.name
101 else:
102 root = alias.name.split(".")[0]
103 bindings[root] = root
104 elif isinstance(node, ast.ImportFrom):
105 if node.module is None: # relative import — out of scope
106 continue
107 for alias in node.names:
108 bound = alias.asname or alias.name
109 bindings[bound] = f"{node.module}.{alias.name}"
110 return bindings
113def _dotted(node: ast.AST) -> str | None:
114 """The literal dotted path of a Name/Attribute chain, or None."""
115 parts: list[str] = []
116 cur = node
117 while isinstance(cur, ast.Attribute):
118 parts.append(cur.attr)
119 cur = cur.value
120 if not isinstance(cur, ast.Name):
121 return None
122 parts.append(cur.id)
123 return ".".join(reversed(parts))
126def _resolve(node: ast.AST, imports: dict[str, str]) -> str | None:
127 """Dotted path with the root name substituted via the import map.
129 Returns None when the root is not an imported symbol — local variables and
130 attributes on instances (e.g. a stored Random) are out of scope, which is
131 what keeps the check precise.
132 """
133 dotted = _dotted(node)
134 if dotted is None:
135 return None
136 root, _, rest = dotted.partition(".")
137 if root not in imports:
138 return None
139 base = imports[root]
140 return f"{base}.{rest}" if rest else base
143def _is_workflow_class(node: ast.ClassDef, imports: dict[str, str]) -> bool:
144 """True if the class carries an ``@workflow.defn`` (or aliased) decorator."""
145 for dec in node.decorator_list:
146 target = dec.func if isinstance(dec, ast.Call) else dec
147 resolved = _resolve(target, imports) or _dotted(target)
148 if resolved and resolved.endswith("workflow.defn"):
149 return True
150 return False
153def scan_file(path: Path) -> list[Finding]:
154 text = path.read_text(encoding="utf-8")
155 try:
156 tree = ast.parse(text, filename=str(path))
157 except SyntaxError:
158 return []
159 imports = _import_map(tree)
160 lines = text.splitlines()
161 findings: list[Finding] = []
162 seen: set[tuple[int, int, str]] = set()
164 def record(node: ast.expr, rule: str, hint: str) -> None:
165 key = (node.lineno, node.col_offset, rule)
166 if key in seen:
167 return
168 seen.add(key)
169 src = lines[node.lineno - 1].strip() if 0 < node.lineno <= len(lines) else ""
170 findings.append(Finding(str(path), node.lineno, node.col_offset, rule, hint, src))
172 for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
173 if not _is_workflow_class(cls, imports):
174 continue
175 for node in ast.walk(cls):
176 if isinstance(node, ast.Call):
177 func = node.func
178 if (
179 isinstance(func, ast.Name)
180 and func.id in BANNED_BUILTINS
181 and func.id not in imports
182 ):
183 record(node, func.id, BANNED_BUILTINS[func.id])
184 continue
185 resolved = _resolve(func, imports)
186 if resolved is None:
187 continue
188 if resolved in BANNED_CALLS:
189 record(node, resolved, BANNED_CALLS[resolved])
190 elif resolved.split(".")[0] in BANNED_CALL_MODULES:
191 record(node, resolved, BANNED_CALL_MODULES[resolved.split(".")[0]])
192 elif isinstance(node, ast.Attribute):
193 resolved = _resolve(node, imports)
194 if resolved in BANNED_ATTRS:
195 record(node, resolved, BANNED_ATTRS[resolved])
197 return findings
200def _iter_py(roots: list[str]) -> list[Path]:
201 files: list[Path] = []
202 for root in roots:
203 p = Path(root)
204 if p.is_file() and p.suffix == ".py":
205 files.append(p)
206 elif p.is_dir():
207 files.extend(
208 f
209 for f in sorted(p.rglob("*.py"))
210 if "__pycache__" not in f.parts
211 )
212 return files
215def main(argv: list[str] | None = None) -> int:
216 parser = argparse.ArgumentParser(description=__doc__)
217 parser.add_argument("paths", nargs="+", help="files or directories to scan")
218 parser.add_argument("--json", action="store_true", help="emit findings as JSON")
219 args = parser.parse_args(argv)
221 findings: list[Finding] = []
222 for path in _iter_py(args.paths):
223 findings.extend(scan_file(path))
225 if args.json:
226 print(json.dumps([asdict(f) for f in findings], indent=2))
227 return 1 if findings else 0
229 if not findings:
230 print("No determinism hazards in workflow code. ✓")
231 return 0
233 for f in findings:
234 print(f"{f.file}:{f.line}:{f.col} {f.rule} -> {f.hint}")
235 print(f" {f.source}")
236 noun = "hazard" if len(findings) == 1 else "hazards"
237 print(f"\n{len(findings)} determinism {noun} in workflow code.", file=sys.stderr)
238 return 1
241if __name__ == "__main__":
242 raise SystemExit(main())
| Category | How it's matched | Replacement |
|---|---|---|
datetime.* / time.* wall clock | exact callee path | workflow.now() / workflow.sleep() |
uuid.uuid1/3/4/5, os.urandom | exact callee path | workflow.uuid4() / workflow.random() |
random / threading / requests / httpx / subprocess | any call on the module | workflow.random() / move to an activity |
os.environ access | attribute access (need not be a call) | read env in an activity, pass it in |
open, input | bare builtin, only if not shadowed | do I/O in an activity |
Scope is the precision lever
Three small functions do the precise work. _import_map records what each bound name refers to. _resolve rebuilds a call's dotted path through that map — and returns None when the root name was never imported. _is_workflow_class is the scope gate: only classes decorated with @workflow.defn (or an alias) are scanned.
Import map · resolution · scope · 242 lines
Import map · resolution · scope
⋯ 92 lines hidden (lines 1–92)
1#!/usr/bin/env python3
2"""Temporal determinism kernel — the lexical workflow-boundary check.
4Fails when a known-nondeterministic call appears *lexically inside* a
5``@workflow.defn`` class. This is the high-precision / low-recall seed of the
6determinism review: it inspects only code written directly in a workflow class
7body — not transitive helpers or imported functions (that is the "brain's" job,
8a later ring) — so a *blocking* CI gate built on it has near-zero false
9positives, which is the property a gate must have or humans disable it.
11Scope: every ``@workflow.defn`` class under the given paths. Activities
12(``@activity.defn``), client setup, type modules, and module-level helpers are
13out of scope by construction — nondeterminism is legal there.
15Usage:
16 python check_determinism.py [--json] PATH [PATH ...]
18Exit 0 if clean, 1 if any hazard is found. ``--json`` prints the findings as a
19JSON array (file, line, col, rule, hint, source) — the structured seam a later
20analysis ring consumes instead of re-parsing.
21"""
23from __future__ import annotations
25import argparse
26import ast
27import json
28import sys
29from dataclasses import asdict, dataclass
30from pathlib import Path
32# --- the banned table -------------------------------------------------------
33# Each entry maps a resolved callee / attribute to its sanctioned replacement.
34# Kept deliberately tight: every entry must be a near-certain hazard inside a
35# workflow body, so the gate stays precise enough to block on.
37# Exact-match callee paths (resolved through the file's imports).
38BANNED_CALLS: dict[str, str] = {
39 "datetime.datetime.now": "use workflow.now()",
40 "datetime.datetime.utcnow": "use workflow.now()",
41 "datetime.datetime.today": "use workflow.now()",
42 "datetime.date.today": "use workflow.now().date()",
43 "time.time": "use workflow.now()",
44 "time.time_ns": "use workflow.now()",
45 "time.monotonic": "use workflow.now()",
46 "time.monotonic_ns": "use workflow.now()",
47 "time.perf_counter": "use workflow.now()",
48 "time.perf_counter_ns": "use workflow.now()",
49 "time.process_time": "use workflow.now()",
50 "time.sleep": "use workflow.sleep()",
51 "uuid.uuid1": "use workflow.uuid4()",
52 "uuid.uuid3": "use workflow.uuid4()",
53 "uuid.uuid4": "use workflow.uuid4()",
54 "uuid.uuid5": "use workflow.uuid4()",
55 "os.getenv": "read env in an activity and pass it in",
56 "os.urandom": "use workflow.random()",
57 "asyncio.sleep": "use workflow.sleep()",
58 "socket.socket": "do network I/O in an activity",
59}
61# Any call whose resolved root module is one of these (module-wide impurity).
62# Only modules with *no* deterministic surface worth keeping belong here.
63BANNED_CALL_MODULES: dict[str, str] = {
64 "random": "use workflow.random()",
65 "threading": "workflows are single-threaded; use workflow APIs",
66 "requests": "do network I/O in an activity",
67 "httpx": "do network I/O in an activity",
68 "subprocess": "run external processes in an activity",
69}
71# Attribute *accesses* (need not be calls), e.g. os.environ["X"].
72BANNED_ATTRS: dict[str, str] = {
73 "os.environ": "read env in an activity and pass it in",
74}
76# Bare builtin names — flagged only when not shadowed by an import binding.
77BANNED_BUILTINS: dict[str, str] = {
78 "open": "do file I/O in an activity",
79 "input": "workflows cannot block on input",
80}
83@dataclass(frozen=True)
84class Finding:
85 file: str
86 line: int
87 col: int
88 rule: str # the resolved hazard, e.g. "datetime.datetime.now"
89 hint: str
90 source: str
93def _import_map(tree: ast.Module) -> dict[str, str]:
94 """Map each bound name to the dotted module/symbol path it refers to."""
95 bindings: dict[str, str] = {}
96 for node in ast.walk(tree):
97 if isinstance(node, ast.Import):
98 for alias in node.names:
99 if alias.asname:
100 bindings[alias.asname] = alias.name
101 else:
102 root = alias.name.split(".")[0]
103 bindings[root] = root
104 elif isinstance(node, ast.ImportFrom):
105 if node.module is None: # relative import — out of scope
106 continue
107 for alias in node.names:
108 bound = alias.asname or alias.name
109 bindings[bound] = f"{node.module}.{alias.name}"
110 return bindings
⋯ 15 lines hidden (lines 111–125)
113def _dotted(node: ast.AST) -> str | None:
114 """The literal dotted path of a Name/Attribute chain, or None."""
115 parts: list[str] = []
116 cur = node
117 while isinstance(cur, ast.Attribute):
118 parts.append(cur.attr)
119 cur = cur.value
120 if not isinstance(cur, ast.Name):
121 return None
122 parts.append(cur.id)
123 return ".".join(reversed(parts))
126def _resolve(node: ast.AST, imports: dict[str, str]) -> str | None:
127 """Dotted path with the root name substituted via the import map.
129 Returns None when the root is not an imported symbol — local variables and
130 attributes on instances (e.g. a stored Random) are out of scope, which is
131 what keeps the check precise.
132 """
133 dotted = _dotted(node)
134 if dotted is None:
135 return None
136 root, _, rest = dotted.partition(".")
137 if root not in imports:
138 return None
139 base = imports[root]
140 return f"{base}.{rest}" if rest else base
143def _is_workflow_class(node: ast.ClassDef, imports: dict[str, str]) -> bool:
144 """True if the class carries an ``@workflow.defn`` (or aliased) decorator."""
145 for dec in node.decorator_list:
146 target = dec.func if isinstance(dec, ast.Call) else dec
147 resolved = _resolve(target, imports) or _dotted(target)
148 if resolved and resolved.endswith("workflow.defn"):
149 return True
150 return False
⋯ 92 lines hidden (lines 151–242)
153def scan_file(path: Path) -> list[Finding]:
154 text = path.read_text(encoding="utf-8")
155 try:
156 tree = ast.parse(text, filename=str(path))
157 except SyntaxError:
158 return []
159 imports = _import_map(tree)
160 lines = text.splitlines()
161 findings: list[Finding] = []
162 seen: set[tuple[int, int, str]] = set()
164 def record(node: ast.expr, rule: str, hint: str) -> None:
165 key = (node.lineno, node.col_offset, rule)
166 if key in seen:
167 return
168 seen.add(key)
169 src = lines[node.lineno - 1].strip() if 0 < node.lineno <= len(lines) else ""
170 findings.append(Finding(str(path), node.lineno, node.col_offset, rule, hint, src))
172 for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
173 if not _is_workflow_class(cls, imports):
174 continue
175 for node in ast.walk(cls):
176 if isinstance(node, ast.Call):
177 func = node.func
178 if (
179 isinstance(func, ast.Name)
180 and func.id in BANNED_BUILTINS
181 and func.id not in imports
182 ):
183 record(node, func.id, BANNED_BUILTINS[func.id])
184 continue
185 resolved = _resolve(func, imports)
186 if resolved is None:
187 continue
188 if resolved in BANNED_CALLS:
189 record(node, resolved, BANNED_CALLS[resolved])
190 elif resolved.split(".")[0] in BANNED_CALL_MODULES:
191 record(node, resolved, BANNED_CALL_MODULES[resolved.split(".")[0]])
192 elif isinstance(node, ast.Attribute):
193 resolved = _resolve(node, imports)
194 if resolved in BANNED_ATTRS:
195 record(node, resolved, BANNED_ATTRS[resolved])
197 return findings
200def _iter_py(roots: list[str]) -> list[Path]:
201 files: list[Path] = []
202 for root in roots:
203 p = Path(root)
204 if p.is_file() and p.suffix == ".py":
205 files.append(p)
206 elif p.is_dir():
207 files.extend(
208 f
209 for f in sorted(p.rglob("*.py"))
210 if "__pycache__" not in f.parts
211 )
212 return files
215def main(argv: list[str] | None = None) -> int:
216 parser = argparse.ArgumentParser(description=__doc__)
217 parser.add_argument("paths", nargs="+", help="files or directories to scan")
218 parser.add_argument("--json", action="store_true", help="emit findings as JSON")
219 args = parser.parse_args(argv)
221 findings: list[Finding] = []
222 for path in _iter_py(args.paths):
223 findings.extend(scan_file(path))
225 if args.json:
226 print(json.dumps([asdict(f) for f in findings], indent=2))
227 return 1 if findings else 0
229 if not findings:
230 print("No determinism hazards in workflow code. ✓")
231 return 0
233 for f in findings:
234 print(f"{f.file}:{f.line}:{f.col} {f.rule} -> {f.hint}")
235 print(f" {f.source}")
236 noun = "hazard" if len(findings) == 1 else "hazards"
237 print(f"\n{len(findings)} determinism {noun} in workflow code.", file=sys.stderr)
238 return 1
241if __name__ == "__main__":
242 raise SystemExit(main())
The scan, and the gate
scan_file walks every @workflow.defn class body. Calls are checked against the exact table first, then the module list; bare builtins are checked by name (only when not shadowed); attribute accesses catch os.environ. A seen set dedupes overlapping nested nodes. The exit code is the gate: non-zero on any finding.
Visitor + exit code · 242 lines
Visitor + exit code
⋯ 171 lines hidden (lines 1–171)
1#!/usr/bin/env python3
2"""Temporal determinism kernel — the lexical workflow-boundary check.
4Fails when a known-nondeterministic call appears *lexically inside* a
5``@workflow.defn`` class. This is the high-precision / low-recall seed of the
6determinism review: it inspects only code written directly in a workflow class
7body — not transitive helpers or imported functions (that is the "brain's" job,
8a later ring) — so a *blocking* CI gate built on it has near-zero false
9positives, which is the property a gate must have or humans disable it.
11Scope: every ``@workflow.defn`` class under the given paths. Activities
12(``@activity.defn``), client setup, type modules, and module-level helpers are
13out of scope by construction — nondeterminism is legal there.
15Usage:
16 python check_determinism.py [--json] PATH [PATH ...]
18Exit 0 if clean, 1 if any hazard is found. ``--json`` prints the findings as a
19JSON array (file, line, col, rule, hint, source) — the structured seam a later
20analysis ring consumes instead of re-parsing.
21"""
23from __future__ import annotations
25import argparse
26import ast
27import json
28import sys
29from dataclasses import asdict, dataclass
30from pathlib import Path
32# --- the banned table -------------------------------------------------------
33# Each entry maps a resolved callee / attribute to its sanctioned replacement.
34# Kept deliberately tight: every entry must be a near-certain hazard inside a
35# workflow body, so the gate stays precise enough to block on.
37# Exact-match callee paths (resolved through the file's imports).
38BANNED_CALLS: dict[str, str] = {
39 "datetime.datetime.now": "use workflow.now()",
40 "datetime.datetime.utcnow": "use workflow.now()",
41 "datetime.datetime.today": "use workflow.now()",
42 "datetime.date.today": "use workflow.now().date()",
43 "time.time": "use workflow.now()",
44 "time.time_ns": "use workflow.now()",
45 "time.monotonic": "use workflow.now()",
46 "time.monotonic_ns": "use workflow.now()",
47 "time.perf_counter": "use workflow.now()",
48 "time.perf_counter_ns": "use workflow.now()",
49 "time.process_time": "use workflow.now()",
50 "time.sleep": "use workflow.sleep()",
51 "uuid.uuid1": "use workflow.uuid4()",
52 "uuid.uuid3": "use workflow.uuid4()",
53 "uuid.uuid4": "use workflow.uuid4()",
54 "uuid.uuid5": "use workflow.uuid4()",
55 "os.getenv": "read env in an activity and pass it in",
56 "os.urandom": "use workflow.random()",
57 "asyncio.sleep": "use workflow.sleep()",
58 "socket.socket": "do network I/O in an activity",
59}
61# Any call whose resolved root module is one of these (module-wide impurity).
62# Only modules with *no* deterministic surface worth keeping belong here.
63BANNED_CALL_MODULES: dict[str, str] = {
64 "random": "use workflow.random()",
65 "threading": "workflows are single-threaded; use workflow APIs",
66 "requests": "do network I/O in an activity",
67 "httpx": "do network I/O in an activity",
68 "subprocess": "run external processes in an activity",
69}
71# Attribute *accesses* (need not be calls), e.g. os.environ["X"].
72BANNED_ATTRS: dict[str, str] = {
73 "os.environ": "read env in an activity and pass it in",
74}
76# Bare builtin names — flagged only when not shadowed by an import binding.
77BANNED_BUILTINS: dict[str, str] = {
78 "open": "do file I/O in an activity",
79 "input": "workflows cannot block on input",
80}
83@dataclass(frozen=True)
84class Finding:
85 file: str
86 line: int
87 col: int
88 rule: str # the resolved hazard, e.g. "datetime.datetime.now"
89 hint: str
90 source: str
93def _import_map(tree: ast.Module) -> dict[str, str]:
94 """Map each bound name to the dotted module/symbol path it refers to."""
95 bindings: dict[str, str] = {}
96 for node in ast.walk(tree):
97 if isinstance(node, ast.Import):
98 for alias in node.names:
99 if alias.asname:
100 bindings[alias.asname] = alias.name
101 else:
102 root = alias.name.split(".")[0]
103 bindings[root] = root
104 elif isinstance(node, ast.ImportFrom):
105 if node.module is None: # relative import — out of scope
106 continue
107 for alias in node.names:
108 bound = alias.asname or alias.name
109 bindings[bound] = f"{node.module}.{alias.name}"
110 return bindings
113def _dotted(node: ast.AST) -> str | None:
114 """The literal dotted path of a Name/Attribute chain, or None."""
115 parts: list[str] = []
116 cur = node
117 while isinstance(cur, ast.Attribute):
118 parts.append(cur.attr)
119 cur = cur.value
120 if not isinstance(cur, ast.Name):
121 return None
122 parts.append(cur.id)
123 return ".".join(reversed(parts))
126def _resolve(node: ast.AST, imports: dict[str, str]) -> str | None:
127 """Dotted path with the root name substituted via the import map.
129 Returns None when the root is not an imported symbol — local variables and
130 attributes on instances (e.g. a stored Random) are out of scope, which is
131 what keeps the check precise.
132 """
133 dotted = _dotted(node)
134 if dotted is None:
135 return None
136 root, _, rest = dotted.partition(".")
137 if root not in imports:
138 return None
139 base = imports[root]
140 return f"{base}.{rest}" if rest else base
143def _is_workflow_class(node: ast.ClassDef, imports: dict[str, str]) -> bool:
144 """True if the class carries an ``@workflow.defn`` (or aliased) decorator."""
145 for dec in node.decorator_list:
146 target = dec.func if isinstance(dec, ast.Call) else dec
147 resolved = _resolve(target, imports) or _dotted(target)
148 if resolved and resolved.endswith("workflow.defn"):
149 return True
150 return False
153def scan_file(path: Path) -> list[Finding]:
154 text = path.read_text(encoding="utf-8")
155 try:
156 tree = ast.parse(text, filename=str(path))
157 except SyntaxError:
158 return []
159 imports = _import_map(tree)
160 lines = text.splitlines()
161 findings: list[Finding] = []
162 seen: set[tuple[int, int, str]] = set()
164 def record(node: ast.expr, rule: str, hint: str) -> None:
165 key = (node.lineno, node.col_offset, rule)
166 if key in seen:
167 return
168 seen.add(key)
169 src = lines[node.lineno - 1].strip() if 0 < node.lineno <= len(lines) else ""
170 findings.append(Finding(str(path), node.lineno, node.col_offset, rule, hint, src))
172 for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
173 if not _is_workflow_class(cls, imports):
174 continue
175 for node in ast.walk(cls):
176 if isinstance(node, ast.Call):
177 func = node.func
178 if (
179 isinstance(func, ast.Name)
180 and func.id in BANNED_BUILTINS
181 and func.id not in imports
182 ):
183 record(node, func.id, BANNED_BUILTINS[func.id])
184 continue
185 resolved = _resolve(func, imports)
186 if resolved is None:
187 continue
188 if resolved in BANNED_CALLS:
189 record(node, resolved, BANNED_CALLS[resolved])
190 elif resolved.split(".")[0] in BANNED_CALL_MODULES:
191 record(node, resolved, BANNED_CALL_MODULES[resolved.split(".")[0]])
192 elif isinstance(node, ast.Attribute):
193 resolved = _resolve(node, imports)
194 if resolved in BANNED_ATTRS:
195 record(node, resolved, BANNED_ATTRS[resolved])
197 return findings
⋯ 27 lines hidden (lines 198–224)
200def _iter_py(roots: list[str]) -> list[Path]:
201 files: list[Path] = []
202 for root in roots:
203 p = Path(root)
204 if p.is_file() and p.suffix == ".py":
205 files.append(p)
206 elif p.is_dir():
207 files.extend(
208 f
209 for f in sorted(p.rglob("*.py"))
210 if "__pycache__" not in f.parts
211 )
212 return files
215def main(argv: list[str] | None = None) -> int:
216 parser = argparse.ArgumentParser(description=__doc__)
217 parser.add_argument("paths", nargs="+", help="files or directories to scan")
218 parser.add_argument("--json", action="store_true", help="emit findings as JSON")
219 args = parser.parse_args(argv)
221 findings: list[Finding] = []
222 for path in _iter_py(args.paths):
223 findings.extend(scan_file(path))
225 if args.json:
226 print(json.dumps([asdict(f) for f in findings], indent=2))
227 return 1 if findings else 0
229 if not findings:
230 print("No determinism hazards in workflow code. ✓")
231 return 0
233 for f in findings:
234 print(f"{f.file}:{f.line}:{f.col} {f.rule} -> {f.hint}")
235 print(f" {f.source}")
236 noun = "hazard" if len(findings) == 1 else "hazards"
237 print(f"\n{len(findings)} determinism {noun} in workflow code.", file=sys.stderr)
238 return 1
Why ynab-agent's boundary is clean today
ynab-agent has six workflows (registry, receipt, monitor, dispatch, poll, txn). All read time via workflow.now() and ids via workflow.uuid4(), so the check passes green — and it does so without tripping on three patterns that look like hazards but are not:
Verified locally before opening this PR: 0 findings on src, and a planted-hazard fixture flags all six injected calls while ignoring the two sanctioned workflow.* calls and a datetime.datetime used only as a return annotation.