solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Static guard for direct durable writes to config/schedules.json.
5
6Known limitation: this detects direct durable-write sinks to a schedules path, not
7writes routed through a generic helper that receives the path as an opaque
8parameter. That indirect case is closed by the journal_io access gate and the
9raw-mechanic gate, which together require any new writer to surface.
10"""
11
12from __future__ import annotations
13
14import ast
15from pathlib import Path
16
17ROOT = Path(__file__).resolve().parents[1]
18WRITE_PRIMITIVES = {
19 "atomic_replace",
20 "install_file",
21 "write_json",
22 "write_jsonl",
23 "write_text",
24}
25WRITE_MODE_CHARS = frozenset({"w", "a", "x", "+"})
26
27
28def _production_modules() -> list[Path]:
29 return sorted(
30 path
31 for path in (ROOT / "solstone").rglob("*.py")
32 if path.is_file()
33 and "__pycache__" not in path.parts
34 and "tests" not in path.parts
35 and path.relative_to(ROOT).as_posix() != "solstone/think/schedule_config.py"
36 )
37
38
39def _call_name(node: ast.AST) -> str | None:
40 if isinstance(node, ast.Name):
41 return node.id
42 if isinstance(node, ast.Attribute):
43 return node.attr
44 return None
45
46
47def _is_schedules_path_expr(node: ast.AST, schedules_vars: set[str]) -> bool:
48 if isinstance(node, ast.Name):
49 return node.id in schedules_vars
50 if isinstance(node, ast.Call):
51 return _call_name(node.func) == "get_schedules_path"
52 if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div):
53 return (
54 isinstance(node.right, ast.Constant)
55 and node.right.value == "schedules.json"
56 )
57 return False
58
59
60def _is_write_mode(node: ast.AST | None) -> bool:
61 if node is None:
62 return False
63 if isinstance(node, ast.Constant) and isinstance(node.value, str):
64 return any(char in node.value for char in WRITE_MODE_CHARS)
65 return False
66
67
68def _open_mode(call: ast.Call) -> ast.AST | None:
69 if len(call.args) >= 2:
70 return call.args[1]
71 for keyword in call.keywords:
72 if keyword.arg == "mode":
73 return keyword.value
74 return None
75
76
77class SchedulesWriteVisitor(ast.NodeVisitor):
78 def __init__(self, path: Path) -> None:
79 self.path = path
80 self.schedules_vars: set[str] = set()
81 self.violations: list[str] = []
82
83 def visit_Assign(self, node: ast.Assign) -> None:
84 if _is_schedules_path_expr(node.value, self.schedules_vars):
85 for target in node.targets:
86 if isinstance(target, ast.Name):
87 self.schedules_vars.add(target.id)
88 self.generic_visit(node)
89
90 def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
91 if (
92 node.value is not None
93 and isinstance(node.target, ast.Name)
94 and _is_schedules_path_expr(node.value, self.schedules_vars)
95 ):
96 self.schedules_vars.add(node.target.id)
97 self.generic_visit(node)
98
99 def visit_Call(self, node: ast.Call) -> None:
100 self._check_open(node)
101 self._check_path_open(node)
102 self._check_write_primitive(node)
103 self._check_replace(node)
104 self.generic_visit(node)
105
106 def _record(self, node: ast.AST, kind: str) -> None:
107 rel = self.path.relative_to(ROOT)
108 self.violations.append(f"{rel}:{node.lineno}: {kind}")
109
110 def _check_open(self, node: ast.Call) -> None:
111 if _call_name(node.func) != "open" or not node.args:
112 return
113 if _is_schedules_path_expr(
114 node.args[0], self.schedules_vars
115 ) and _is_write_mode(_open_mode(node)):
116 self._record(node, "open-write")
117
118 def _check_path_open(self, node: ast.Call) -> None:
119 if not isinstance(node.func, ast.Attribute) or node.func.attr != "open":
120 return
121 if _is_schedules_path_expr(
122 node.func.value, self.schedules_vars
123 ) and _is_write_mode(_open_mode(node)):
124 self._record(node, "Path.open-write")
125
126 def _check_write_primitive(self, node: ast.Call) -> None:
127 if _call_name(node.func) not in WRITE_PRIMITIVES or not node.args:
128 return
129 if _is_schedules_path_expr(node.args[0], self.schedules_vars):
130 self._record(node, _call_name(node.func) or "write-primitive")
131
132 def _check_replace(self, node: ast.Call) -> None:
133 if isinstance(node.func, ast.Attribute) and node.func.attr == "replace":
134 if len(node.args) >= 2 and _is_schedules_path_expr(
135 node.args[1], self.schedules_vars
136 ):
137 self._record(node, "os.replace")
138 return
139 if node.args and _is_schedules_path_expr(node.args[0], self.schedules_vars):
140 self._record(node, "Path.replace")
141 return
142 if _call_name(node.func) == "replace" and len(node.args) >= 2:
143 if _is_schedules_path_expr(node.args[1], self.schedules_vars):
144 self._record(node, "os.replace")
145
146
147def test_only_schedule_config_directly_writes_schedules_json() -> None:
148 violations: list[str] = []
149 for path in _production_modules():
150 tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
151 visitor = SchedulesWriteVisitor(path)
152 visitor.visit(tree)
153 violations.extend(visitor.violations)
154
155 assert violations == []