solpbc.org
personal memory agent
1#!/usr/bin/env python3
2# SPDX-License-Identifier: AGPL-3.0-only
3# Copyright (c) 2026 sol pbc
4
5"""Journal config single-owner transaction lint.
6
7Invariant: ``config/journal.json`` is a single-owner transactional domain.
8Production code outside ``solstone.think.journal_config`` must not import or
9call the private raw serializer, compose its own read/lock/write cycle, replace
10the config file directly, re-export or wrap the private serializer, or create a
11second lock for the same file.
12
13Design decisions:
14
15 D1 - Import-binding resolution. Bare-name matching is forbidden. The scanner
16 records direct imports, module aliases, and dotted imports before deciding
17 whether a call or assignment targets ``_write_journal_config``,
18 ``get_journal_config_path``, ``atomic_replace``, ``hold_lock``, or
19 ``fcntl.flock``.
20
21 D2 - Config-path-specific replacement. ``atomic_replace`` is legitimate for
22 many other domains. This gate flags replacement only when the replacement
23 target resolves to ``get_journal_config_path(...)`` or a path expression
24 ending in ``config/journal.json``.
25
26 D3 - Owner boundary. The only production owner file is
27 ``solstone/think/journal_config.py``. Tests are excluded via the same
28 ``_is_test_file`` convention as sibling gates.
29
30 D4 - Empty allowlist. The migrated tree must be green without production
31 exceptions. A new violation fails immediately.
32
33Exit codes:
34 0 - no production violations
35 1 - any production violation is found
36"""
37
38from __future__ import annotations
39
40import argparse
41import ast
42import sys
43from pathlib import Path
44
45ROOT = Path(__file__).resolve().parent.parent
46
47OWNER_FILES: frozenset[str] = frozenset({"solstone/think/journal_config.py"})
48ALLOWLIST: dict[tuple[str, str], int] = {}
49CONFIG_FILENAME = "journal.json"
50SIDECAR_LOCK_FILENAME = "." + CONFIG_FILENAME + ".lock"
51
52
53def _is_owner(rel: Path) -> bool:
54 return rel.as_posix() in OWNER_FILES
55
56
57def _is_test_file(rel: Path) -> bool:
58 return (
59 "tests" in rel.parts
60 or rel.name == "conftest.py"
61 or (rel.name.startswith("test_") and rel.suffix == ".py")
62 )
63
64
65def discover_modules(root: Path) -> list[Path]:
66 """Return posix-relative non-owner, non-test modules under ``solstone/``."""
67 scope = root / "solstone"
68 if not scope.is_dir():
69 return []
70
71 found: list[Path] = []
72 for path in sorted(scope.rglob("*.py")):
73 rel = path.relative_to(root)
74 if "__pycache__" in rel.parts:
75 continue
76 if _is_test_file(rel):
77 continue
78 if _is_owner(rel):
79 continue
80 found.append(rel)
81 return found
82
83
84def _dotted_name(node: ast.AST) -> str | None:
85 if isinstance(node, ast.Name):
86 return node.id
87 if isinstance(node, ast.Attribute):
88 base = _dotted_name(node.value)
89 if base:
90 return f"{base}.{node.attr}"
91 return None
92
93
94class _Bindings:
95 def __init__(self) -> None:
96 self.private_serializer_names: set[str] = set()
97 self.journal_config_modules: set[str] = set()
98 self.get_path_names: set[str] = set()
99 self.atomic_replace_names: set[str] = set()
100 self.journal_io_modules: set[str] = set()
101 self.hold_lock_names: set[str] = set()
102 self.fcntl_modules: set[str] = set()
103 self.flock_names: set[str] = set()
104 self.os_modules: set[str] = set()
105 self.os_replace_names: set[str] = set()
106
107
108def _collect_bindings(tree: ast.AST) -> _Bindings:
109 bindings = _Bindings()
110 for node in ast.walk(tree):
111 if isinstance(node, ast.Import):
112 for alias in node.names:
113 bound = alias.asname or alias.name
114 if alias.name == "solstone.think.journal_config":
115 bindings.journal_config_modules.add(bound)
116 elif alias.name == "solstone.think.journal_io":
117 bindings.journal_io_modules.add(bound)
118 elif alias.name == "solstone.think.journal_io.atomic":
119 bindings.journal_io_modules.add(bound)
120 elif alias.name == "solstone.think.journal_io.locking":
121 bindings.journal_io_modules.add(bound)
122 elif alias.name == "fcntl":
123 bindings.fcntl_modules.add(bound)
124 elif alias.name == "os":
125 bindings.os_modules.add(bound)
126 elif isinstance(node, ast.ImportFrom):
127 module = node.module or ""
128 for alias in node.names:
129 bound = alias.asname or alias.name
130 if module == "solstone.think" and alias.name == "journal_config":
131 bindings.journal_config_modules.add(bound)
132 elif module == "solstone.think.journal_config":
133 if alias.name == "_write_journal_config":
134 bindings.private_serializer_names.add(bound)
135 elif alias.name == "get_journal_config_path":
136 bindings.get_path_names.add(bound)
137 elif module in {
138 "solstone.think.journal_io",
139 "solstone.think.journal_io.atomic",
140 }:
141 if alias.name == "atomic_replace":
142 bindings.atomic_replace_names.add(bound)
143 elif alias.name == "hold_lock":
144 bindings.hold_lock_names.add(bound)
145 elif module == "solstone.think.journal_io.locking":
146 if alias.name == "hold_lock":
147 bindings.hold_lock_names.add(bound)
148 elif module == "fcntl" and alias.name == "flock":
149 bindings.flock_names.add(bound)
150 elif module == "os" and alias.name == "replace":
151 bindings.os_replace_names.add(bound)
152 return bindings
153
154
155def _called_attr(func: ast.expr, modules: set[str], attr: str, full_name: str) -> bool:
156 if not isinstance(func, ast.Attribute) or func.attr != attr:
157 return False
158 if isinstance(func.value, ast.Name) and func.value.id in modules:
159 return True
160 return _dotted_name(func) == full_name
161
162
163def _is_private_serializer_expr(node: ast.AST, bindings: _Bindings) -> bool:
164 if isinstance(node, ast.Name):
165 return node.id in bindings.private_serializer_names
166 if isinstance(node, ast.Attribute) and node.attr == "_write_journal_config":
167 if isinstance(node.value, ast.Name) and node.value.id in (
168 bindings.journal_config_modules
169 ):
170 return True
171 return (
172 _dotted_name(node) == "solstone.think.journal_config._write_journal_config"
173 )
174 return False
175
176
177def _is_private_serializer_call(node: ast.Call, bindings: _Bindings) -> bool:
178 return _is_private_serializer_expr(node.func, bindings)
179
180
181def _is_get_config_path_call(node: ast.AST, bindings: _Bindings) -> bool:
182 if not isinstance(node, ast.Call):
183 return False
184 func = node.func
185 if isinstance(func, ast.Name) and func.id in bindings.get_path_names:
186 return True
187 return _called_attr(
188 func,
189 bindings.journal_config_modules,
190 "get_journal_config_path",
191 "solstone.think.journal_config.get_journal_config_path",
192 )
193
194
195def _constant_path_parts(node: ast.AST) -> list[str]:
196 if isinstance(node, ast.Constant) and isinstance(node.value, str):
197 return [node.value]
198 if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div):
199 return _constant_path_parts(node.left) + _constant_path_parts(node.right)
200 if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
201 left = "".join(_constant_path_parts(node.left))
202 right = "".join(_constant_path_parts(node.right))
203 if left or right:
204 return [left + right]
205 if isinstance(node, ast.Call):
206 parts: list[str] = []
207 for arg in node.args:
208 parts.extend(_constant_path_parts(arg))
209 return parts
210 return []
211
212
213def _parts_end_with_config_path(parts: list[str]) -> bool:
214 normalized: list[str] = []
215 for part in parts:
216 normalized.extend(
217 piece for piece in part.replace("\\", "/").split("/") if piece
218 )
219 return len(normalized) >= 2 and normalized[-2:] == ["config", CONFIG_FILENAME]
220
221
222def _parts_contain_sidecar_lock(parts: list[str]) -> bool:
223 for part in parts:
224 if SIDECAR_LOCK_FILENAME in part.replace("\\", "/").split("/"):
225 return True
226 return False
227
228
229def _assigned_path_names(
230 tree: ast.AST, bindings: _Bindings
231) -> tuple[set[str], set[str]]:
232 config_path_names: set[str] = set()
233 sidecar_lock_names: set[str] = set()
234 changed = True
235 while changed:
236 changed = False
237 for node in ast.walk(tree):
238 value: ast.AST | None = None
239 targets: list[ast.expr] = []
240 if isinstance(node, ast.Assign):
241 value = node.value
242 targets = list(node.targets)
243 elif isinstance(node, ast.AnnAssign):
244 value = node.value
245 targets = [node.target]
246 if value is None:
247 continue
248 is_config = _is_config_path_expr(value, bindings, config_path_names)
249 is_sidecar = _is_sidecar_lock_expr(value, sidecar_lock_names)
250 for target in targets:
251 if not isinstance(target, ast.Name):
252 continue
253 if is_config and target.id not in config_path_names:
254 config_path_names.add(target.id)
255 changed = True
256 if is_sidecar and target.id not in sidecar_lock_names:
257 sidecar_lock_names.add(target.id)
258 changed = True
259 return config_path_names, sidecar_lock_names
260
261
262def _is_config_path_expr(
263 node: ast.AST,
264 bindings: _Bindings,
265 config_path_names: set[str],
266) -> bool:
267 if isinstance(node, ast.Name):
268 return node.id in config_path_names
269 if _is_get_config_path_call(node, bindings):
270 return True
271 return _parts_end_with_config_path(_constant_path_parts(node))
272
273
274def _is_sidecar_lock_expr(node: ast.AST, sidecar_lock_names: set[str]) -> bool:
275 if isinstance(node, ast.Name):
276 return node.id in sidecar_lock_names
277 return _parts_contain_sidecar_lock(_constant_path_parts(node))
278
279
280def _called_atomic_replace(func: ast.expr, bindings: _Bindings) -> bool:
281 if isinstance(func, ast.Name):
282 return func.id in bindings.atomic_replace_names
283 return _called_attr(
284 func,
285 bindings.journal_io_modules,
286 "atomic_replace",
287 "solstone.think.journal_io.atomic_replace",
288 )
289
290
291def _called_hold_lock(func: ast.expr, bindings: _Bindings) -> bool:
292 if isinstance(func, ast.Name):
293 return func.id in bindings.hold_lock_names
294 return _called_attr(
295 func,
296 bindings.journal_io_modules,
297 "hold_lock",
298 "solstone.think.journal_io.hold_lock",
299 )
300
301
302def _called_flock(func: ast.expr, bindings: _Bindings) -> bool:
303 if isinstance(func, ast.Name):
304 return func.id in bindings.flock_names
305 return _called_attr(func, bindings.fcntl_modules, "flock", "fcntl.flock")
306
307
308def _called_os_replace(func: ast.expr, bindings: _Bindings) -> bool:
309 if isinstance(func, ast.Name):
310 return func.id in bindings.os_replace_names
311 return _called_attr(func, bindings.os_modules, "replace", "os.replace")
312
313
314def _is_path_replace_call(node: ast.Call) -> bool:
315 return isinstance(node.func, ast.Attribute) and node.func.attr == "replace"
316
317
318def scan_source(source: str, filename: str = "<source>") -> list[tuple[int, str, str]]:
319 """Return sorted ``(lineno, kind, detail)`` production violations."""
320 tree = ast.parse(source, filename=filename)
321 bindings = _collect_bindings(tree)
322 config_path_names, sidecar_lock_names = _assigned_path_names(tree, bindings)
323 findings: list[tuple[int, str, str]] = []
324
325 for node in ast.walk(tree):
326 if isinstance(node, ast.ImportFrom):
327 if node.module == "solstone.think.journal_config":
328 for alias in node.names:
329 if alias.name == "_write_journal_config":
330 findings.append(
331 (
332 node.lineno,
333 "private_serializer",
334 "_write_journal_config import",
335 )
336 )
337 elif isinstance(node, (ast.Assign, ast.AnnAssign)):
338 value = node.value
339 if value is not None and _is_private_serializer_expr(value, bindings):
340 findings.append(
341 (
342 node.lineno,
343 "private_serializer_wrapper",
344 "_write_journal_config alias or re-export",
345 )
346 )
347 elif isinstance(node, ast.FunctionDef):
348 for child in ast.walk(node):
349 if isinstance(child, ast.Call) and _is_private_serializer_call(
350 child, bindings
351 ):
352 findings.append(
353 (
354 child.lineno,
355 "private_serializer_wrapper",
356 f"{node.name} wraps _write_journal_config",
357 )
358 )
359 elif isinstance(node, ast.Call):
360 if _is_private_serializer_call(node, bindings):
361 findings.append(
362 (
363 node.lineno,
364 "private_serializer",
365 "_write_journal_config call",
366 )
367 )
368 elif _called_atomic_replace(node.func, bindings):
369 if node.args and _is_config_path_expr(
370 node.args[0], bindings, config_path_names
371 ):
372 findings.append(
373 (
374 node.lineno,
375 "journal_config_replace",
376 "atomic_replace targets config/journal.json",
377 )
378 )
379 elif _called_os_replace(node.func, bindings):
380 if len(node.args) >= 2 and _is_config_path_expr(
381 node.args[1], bindings, config_path_names
382 ):
383 findings.append(
384 (
385 node.lineno,
386 "journal_config_replace",
387 "os.replace targets config/journal.json",
388 )
389 )
390 elif _is_path_replace_call(node):
391 if node.args and _is_config_path_expr(
392 node.args[0], bindings, config_path_names
393 ):
394 findings.append(
395 (
396 node.lineno,
397 "journal_config_replace",
398 "Path.replace targets config/journal.json",
399 )
400 )
401 elif _called_hold_lock(node.func, bindings):
402 if node.args and _is_config_path_expr(
403 node.args[0], bindings, config_path_names
404 ):
405 findings.append(
406 (
407 node.lineno,
408 "second_config_lock",
409 "hold_lock targets config/journal.json",
410 )
411 )
412 elif _called_flock(node.func, bindings):
413 if (
414 sidecar_lock_names
415 or SIDECAR_LOCK_FILENAME in source
416 or any(
417 _is_sidecar_lock_expr(arg, sidecar_lock_names)
418 for arg in node.args
419 )
420 ):
421 findings.append(
422 (
423 node.lineno,
424 "second_config_lock",
425 "fcntl.flock targets journal config sidecar lock",
426 )
427 )
428
429 findings.sort()
430 return findings
431
432
433def scan_file(path: Path) -> list[tuple[int, str, str]]:
434 return scan_source(path.read_text(encoding="utf-8"), filename=str(path))
435
436
437def count_violations(root: Path) -> dict[tuple[str, str], int]:
438 """Map ``(posix-relpath, kind)`` -> occurrence count across production."""
439 counts: dict[tuple[str, str], int] = {}
440 for rel in discover_modules(root):
441 for _lineno, kind, _detail in scan_file(root / rel):
442 key = (rel.as_posix(), kind)
443 counts[key] = counts.get(key, 0) + 1
444 return counts
445
446
447def evaluate(
448 root: Path,
449 allowlist: dict[tuple[str, str], int],
450) -> tuple[list[str], list[str]]:
451 """Return ``(over, tracked)`` human-readable lines."""
452 over: list[str] = []
453 tracked: list[str] = []
454
455 for rel in discover_modules(root):
456 rel_str = rel.as_posix()
457 by_kind: dict[str, list[tuple[int, str]]] = {}
458 for lineno, kind, detail in scan_file(root / rel):
459 by_kind.setdefault(kind, []).append((lineno, detail))
460 for kind, entries in sorted(by_kind.items()):
461 count = len(entries)
462 allowed = allowlist.get((rel_str, kind), 0)
463 if count > allowed:
464 details = ", ".join(
465 f"line {lineno} ({detail})" for lineno, detail in entries
466 )
467 over.append(
468 f"{rel_str}: {kind} count {count} exceeds allowed {allowed}: "
469 f"{details}"
470 )
471 elif allowed:
472 tracked.append(f"{rel_str}: {count}/{allowed} {kind} (allowlisted)")
473 return over, tracked
474
475
476def main(argv: list[str] | None = None) -> int:
477 parser = argparse.ArgumentParser(description="Journal config owner lint")
478 parser.add_argument(
479 "--root",
480 type=Path,
481 default=ROOT,
482 help="Repository root to scan (defaults to the checkout root).",
483 )
484 args = parser.parse_args(argv)
485
486 over, tracked = evaluate(args.root, ALLOWLIST)
487
488 if tracked:
489 print("journal-config-owner: known violations (allowlisted):")
490 for line in tracked:
491 print(f" {line}")
492 print()
493
494 if over:
495 print("journal-config-owner: NEW violations:", file=sys.stderr)
496 for line in over:
497 print(f" {line}", file=sys.stderr)
498 print(file=sys.stderr)
499 print(
500 "Route config/journal.json writes through "
501 "solstone.think.journal_config.mutate_journal_config.",
502 file=sys.stderr,
503 )
504 return 1
505
506 print("journal-config-owner: pass")
507 return 0
508
509
510if __name__ == "__main__":
511 raise SystemExit(main())