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 I/O write-primitive access lint.
6
7This check is CI mechanics, not a permissions model. The shared
8``solstone.think.journal_io`` write and lock primitives are mechanisms; the L2
9domain write-owner set is the policy. A non-owner module violates the rule only
10when it imports one of the gated journal_io primitives and calls through that
11import binding.
12
13Policy source: ``AGENTS.md`` §7, L2 — Domain write ownership. This script
14transcribes that owner set into path exclusions so the check can enforce the
15policy without restating it in prose.
16
17Design decisions:
18
19 D1 — Import-binding resolution. Bare-name matching is forbidden. For every
20 scanned module, the detector first records imports that bind names or module
21 aliases to ``solstone.think.journal_io`` or its write-primitive submodules,
22 then flags calls only when the call target resolves through one of those
23 bindings.
24
25 D2 — Allowed callers. The journal_io package itself and the L2 domain
26 write-owner files/directories are excluded during module discovery, so owner
27 modules are never scanned or counted as violations.
28
29 D3 — Violation kind and message. The violation ``kind`` is the primitive name
30 itself, and every failure line names both the file and primitive.
31
32 D4 — Script home and scope. This is a standalone sibling check for
33 journal_io access. It intentionally does not modify
34 ``scripts/check_layer_hygiene.py``.
35
36The check ships green via a committed ``ALLOWLIST`` keyed by ``(file, kind)``
37with an allowed **count**. A new violation that pushes any ``(file, kind)``
38count above its allowed number fails the check; fixing occurrences lets the
39allowed count be lowered, so the allowlist ratchets toward empty. It is never
40keyed by line number and never a blanket per-file disable.
41
42Exit codes:
43 0 — no un-allowlisted violations
44 1 — a (file, kind) count exceeds its allowlisted number
45"""
46
47from __future__ import annotations
48
49import argparse
50import ast
51import sys
52from pathlib import Path
53
54ROOT = Path(__file__).resolve().parent.parent
55
56GATED_PRIMITIVES: frozenset[str] = frozenset(
57 {
58 "append_jsonl",
59 "append_text",
60 "atomic_replace",
61 "hold_lock",
62 "install_file",
63 "save_npz",
64 "update_npz",
65 "write_json",
66 "write_jsonl",
67 "write_npz",
68 "write_text",
69 }
70)
71
72MODULE_PRIMITIVES: dict[str, frozenset[str]] = {
73 "solstone.think.journal_io": GATED_PRIMITIVES,
74 "solstone.think.journal_io.append": frozenset({"append_jsonl", "append_text"}),
75 "solstone.think.journal_io.atomic": frozenset(
76 {"atomic_replace", "install_file", "write_json", "write_jsonl", "write_text"}
77 ),
78 "solstone.think.journal_io.locking": frozenset({"hold_lock"}),
79 "solstone.think.journal_io.npz": frozenset({"save_npz", "update_npz", "write_npz"}),
80}
81
82OWNER_FILES: frozenset[str] = frozenset(
83 {
84 "solstone/apps/chat/config.py",
85 "solstone/convey/config.py",
86 "solstone/apps/speakers/attribution.py",
87 "solstone/apps/speakers/candidate_tracker.py",
88 "solstone/apps/speakers/discovery.py",
89 "solstone/apps/speakers/owner.py",
90 "solstone/apps/speakers/routes.py",
91 "solstone/apps/settings/vertex_credentials.py",
92 "solstone/apps/timeline/maintenance.py",
93 "solstone/apps/timeline/talent/segment_summary.py",
94 "solstone/apps/import/ingest.py",
95 "solstone/apps/import/facet_ingest.py",
96 "solstone/apps/import/journal_sources.py",
97 "solstone/apps/import/resolve.py",
98 "solstone/apps/observer/utils.py",
99 "solstone/think/activities.py",
100 "solstone/think/awareness.py",
101 "solstone/think/day_accumulator.py",
102 "solstone/think/entities/consolidation.py",
103 "solstone/think/entities/journal.py",
104 "solstone/think/entities/merge.py",
105 "solstone/think/entities/observations.py",
106 "solstone/think/entities/relationships.py",
107 "solstone/think/entities/review_candidates.py",
108 "solstone/think/entities/saving.py",
109 "solstone/think/entities/voiceprints.py",
110 "solstone/think/facet_review_candidates.py",
111 "solstone/think/speaker_review_candidates.py",
112 "solstone/think/facets.py",
113 "solstone/think/identity.py",
114 "solstone/think/journal_config.py",
115 "solstone/think/schedule_config.py",
116 "solstone/think/push/devices.py",
117 # Link domain — device-pairing service state.
118 "solstone/think/link/auth.py",
119 "solstone/think/link/ca.py",
120 "solstone/think/link/nonces.py",
121 "solstone/think/link/paths.py",
122 "solstone/think/sense_splitter.py",
123 "solstone/think/streams.py",
124 "solstone/think/thinking.py",
125 # Chronicle day content direct writers. Importer modules that merely
126 # route through importers/shared.py are not direct owners and are
127 # intentionally omitted.
128 "solstone/observe/depict.py",
129 "solstone/observe/describe.py",
130 "solstone/observe/extract_pdf.py",
131 "solstone/observe/transcribe/main.py",
132 "solstone/observe/transfer.py",
133 "solstone/think/importers/cli.py",
134 "solstone/think/importers/documents.py",
135 "solstone/think/importers/shared.py",
136 "solstone/convey/chat_stream.py",
137 # imports/** bundle + sync-cursor writers (local/CLI import flows).
138 "solstone/think/importers/plaud.py", # streamed imported-audio install.
139 "solstone/think/importers/sync.py",
140 "solstone/think/importers/utils.py",
141 }
142)
143
144OWNER_PREFIXES: tuple[str, ...] = (
145 "solstone/apps/facets/",
146 "solstone/think/indexer/",
147)
148
149# Committed allowlist of violations on the current tree, keyed by
150# (posix-relative-path, kind) -> allowed count. Ratchets toward empty: lower a
151# count as occurrences are fixed; never raise one to admit a new violation.
152# Never line-keyed; never a per-file blanket.
153ALLOWLIST: dict[tuple[str, str], int] = {}
154
155
156def _is_owner(rel: Path) -> bool:
157 rel_str = rel.as_posix()
158 return rel_str in OWNER_FILES or any(
159 rel_str.startswith(prefix) for prefix in OWNER_PREFIXES
160 )
161
162
163def _is_test_file(rel: Path) -> bool:
164 return (
165 "tests" in rel.parts
166 or rel.name == "conftest.py"
167 or (rel.name.startswith("test_") and rel.suffix == ".py")
168 )
169
170
171def discover_modules(root: Path) -> list[Path]:
172 """Return posix-relative non-owner, non-test modules under ``solstone/``."""
173 scope = root / "solstone"
174 if not scope.is_dir():
175 return []
176
177 found: list[Path] = []
178 for path in sorted(scope.rglob("*.py")):
179 rel = path.relative_to(root)
180 rel_str = rel.as_posix()
181 if "__pycache__" in rel.parts:
182 continue
183 if rel_str.startswith("solstone/think/journal_io/"):
184 continue
185 if _is_test_file(rel):
186 continue
187 if _is_owner(rel):
188 continue
189 found.append(rel)
190 return found
191
192
193def _dotted_name(node: ast.AST) -> str | None:
194 if isinstance(node, ast.Name):
195 return node.id
196 if isinstance(node, ast.Attribute):
197 base = _dotted_name(node.value)
198 if base:
199 return f"{base}.{node.attr}"
200 return None
201
202
203def _bind_from_import(
204 node: ast.ImportFrom,
205 direct_bindings: dict[str, str],
206 module_bindings: dict[str, frozenset[str]],
207) -> None:
208 module = node.module or ""
209 if module == "solstone.think":
210 for alias in node.names:
211 if alias.name == "journal_io":
212 module_bindings[alias.asname or alias.name] = GATED_PRIMITIVES
213 return
214
215 primitives = MODULE_PRIMITIVES.get(module)
216 if not primitives:
217 return
218
219 for alias in node.names:
220 if alias.name == "*":
221 for primitive in primitives:
222 direct_bindings[primitive] = primitive
223 elif alias.name in primitives:
224 direct_bindings[alias.asname or alias.name] = alias.name
225
226
227def _bind_import(
228 node: ast.Import,
229 module_bindings: dict[str, frozenset[str]],
230 dotted_bindings: dict[str, frozenset[str]],
231) -> None:
232 for alias in node.names:
233 primitives = MODULE_PRIMITIVES.get(alias.name)
234 if not primitives:
235 continue
236 if alias.asname:
237 module_bindings[alias.asname] = primitives
238 else:
239 dotted_bindings[alias.name] = primitives
240
241
242def _collect_bindings(
243 tree: ast.AST,
244) -> tuple[
245 dict[str, str],
246 dict[str, frozenset[str]],
247 dict[str, frozenset[str]],
248]:
249 direct_bindings: dict[str, str] = {}
250 module_bindings: dict[str, frozenset[str]] = {}
251 dotted_bindings: dict[str, frozenset[str]] = {}
252
253 for node in ast.walk(tree):
254 if isinstance(node, ast.ImportFrom):
255 _bind_from_import(node, direct_bindings, module_bindings)
256 elif isinstance(node, ast.Import):
257 _bind_import(node, module_bindings, dotted_bindings)
258
259 return direct_bindings, module_bindings, dotted_bindings
260
261
262def _called_primitive(
263 func: ast.expr,
264 direct_bindings: dict[str, str],
265 module_bindings: dict[str, frozenset[str]],
266 dotted_bindings: dict[str, frozenset[str]],
267) -> tuple[str, str] | None:
268 if isinstance(func, ast.Name):
269 primitive = direct_bindings.get(func.id)
270 if primitive:
271 return primitive, func.id
272 return None
273
274 if not isinstance(func, ast.Attribute):
275 return None
276
277 if (
278 isinstance(func.value, ast.Name)
279 and func.value.id in module_bindings
280 and func.attr in module_bindings[func.value.id]
281 ):
282 return func.attr, f"{func.value.id}.{func.attr}"
283
284 dotted = _dotted_name(func)
285 if not dotted:
286 return None
287 for prefix, primitives in dotted_bindings.items():
288 for primitive in primitives:
289 if dotted == f"{prefix}.{primitive}":
290 return primitive, dotted
291 return None
292
293
294def scan_source(source: str, filename: str = "<source>") -> list[tuple[int, str, str]]:
295 """Return ``(lineno, primitive, bound_name)`` violations for source."""
296 tree = ast.parse(source, filename=filename)
297 direct_bindings, module_bindings, dotted_bindings = _collect_bindings(tree)
298
299 findings: list[tuple[int, str, str]] = []
300 for node in ast.walk(tree):
301 if not isinstance(node, ast.Call):
302 continue
303 called = _called_primitive(
304 node.func,
305 direct_bindings,
306 module_bindings,
307 dotted_bindings,
308 )
309 if called:
310 primitive, bound_name = called
311 findings.append((node.lineno, primitive, bound_name))
312 findings.sort()
313 return findings
314
315
316def scan_file(path: Path) -> list[tuple[int, str, str]]:
317 return scan_source(path.read_text(encoding="utf-8"), filename=str(path))
318
319
320def count_violations(root: Path) -> dict[tuple[str, str], int]:
321 """Map ``(posix-relpath, primitive)`` -> occurrence count across the tree."""
322 counts: dict[tuple[str, str], int] = {}
323 for rel in discover_modules(root):
324 for _lineno, primitive, _bound_name in scan_file(root / rel):
325 key = (rel.as_posix(), primitive)
326 counts[key] = counts.get(key, 0) + 1
327 return counts
328
329
330def evaluate(
331 root: Path,
332 allowlist: dict[tuple[str, str], int],
333) -> tuple[list[str], list[str]]:
334 """Return ``(new_violations, tracked)`` human-readable lines."""
335 new: list[str] = []
336 tracked: list[str] = []
337 for rel in discover_modules(root):
338 rel_str = rel.as_posix()
339 findings = scan_file(root / rel)
340 by_primitive: dict[str, list[int]] = {}
341 for lineno, primitive, _bound_name in findings:
342 by_primitive.setdefault(primitive, []).append(lineno)
343 for primitive, linenos in sorted(by_primitive.items()):
344 count = len(linenos)
345 allowed = allowlist.get((rel_str, primitive), 0)
346 if count > allowed:
347 lines = ", ".join(str(n) for n in sorted(linenos))
348 new.append(
349 f"{rel_str}: {primitive} imported from journal_io and called "
350 f"by a non-owner ({count} occurrence(s), allowed {allowed}) "
351 f"at line(s) {lines}"
352 )
353 elif allowed:
354 tracked.append(
355 f"{rel_str}: {count}/{allowed} {primitive} (allowlisted)"
356 )
357 return new, tracked
358
359
360def main(argv: list[str] | None = None) -> int:
361 parser = argparse.ArgumentParser(description="Journal I/O access lint")
362 parser.add_argument(
363 "--root",
364 type=Path,
365 default=ROOT,
366 help="Repository root to scan (defaults to the checkout root).",
367 )
368 args = parser.parse_args(argv)
369
370 new, tracked = evaluate(args.root, ALLOWLIST)
371
372 if tracked:
373 print("journal-io-access: known violations (allowlisted, ratcheting down):")
374 for line in tracked:
375 print(f" {line}")
376 print()
377
378 if new:
379 print("journal-io-access: NEW violations:", file=sys.stderr)
380 for line in new:
381 print(f" {line}", file=sys.stderr)
382 print(file=sys.stderr)
383 print(
384 "Route writes through the L2 owner for that domain — see AGENTS.md §7.",
385 file=sys.stderr,
386 )
387 return 1
388
389 print("journal-io-access: pass")
390 return 0
391
392
393if __name__ == "__main__":
394 raise SystemExit(main())