solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

test(config): gate journal config owner boundary

Add an AST-based check that blocks production imports or calls of the private raw serializer, config-specific direct replacement, wrappers or re-exports around the private writer, and second config locks. Wire it into install-checks and cover scanner plus end-to-end cases, including legitimate atomic_replace on non-config domains.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

author
Jer Miller co-author
Claude Opus 4.8 (1M context) date (Jul 18, 2026, 10:16 PM -0600) commit fd15b391 parent 11205961
+708 -1
+8 -1
Makefile
··· 14 14 PYTEST_BASETEMP_INIT := BASETEMP=$$(mktemp -d /var/tmp/solstone-pytest-XXXXXX); trap 'rm -rf "$$BASETEMP"' EXIT INT TERM; 15 15 PYTEST_BASETEMP_FLAG := --basetemp "$$BASETEMP" 16 16 17 - .PHONY: install hopper-install uninstall test test-cov test-integration test-performance test-app test-only format format-check install-checks ci clean clean-install coverage watch versions update update-prices preflight pre-commit skills render-packaging check-rust-fmt check-rust-clippy check-rust-test check-rust-ios check-rust-deny audit openapi check-openapi contract check-contract journal-resolution-vectors check-journal-resolution-vectors dev all sandbox sandbox-stop install-models parakeet-helper parakeet-helper-clean wheel-macos wheel-macos-clean verify verify-api verify-schemathesis update-api-baselines eval-schemas service-logs check-layer-hygiene check-api-conventions check-journal-io-access check-journal-io-mechanic check-call-http-only check-tools-http-only check-access-imports-clean check-convey-bind-imports-clean check-schema-bounds check-thin-base-install check-cogitate-prompts smoke-cogitate release release-test FORCE 17 + .PHONY: install hopper-install uninstall test test-cov test-integration test-performance test-app test-only format format-check install-checks ci clean clean-install coverage watch versions update update-prices preflight pre-commit skills render-packaging check-rust-fmt check-rust-clippy check-rust-test check-rust-ios check-rust-deny audit openapi check-openapi contract check-contract journal-resolution-vectors check-journal-resolution-vectors dev all sandbox sandbox-stop install-models parakeet-helper parakeet-helper-clean wheel-macos wheel-macos-clean verify verify-api verify-schemathesis update-api-baselines eval-schemas service-logs check-layer-hygiene check-api-conventions check-journal-io-access check-journal-io-mechanic check-journal-config-owner check-call-http-only check-tools-http-only check-access-imports-clean check-convey-bind-imports-clean check-schema-bounds check-thin-base-install check-cogitate-prompts smoke-cogitate release release-test FORCE 18 18 19 19 # Default target - install package in editable mode 20 20 all: install ··· 480 480 @echo "=== Running journal-io mechanic check ===" 481 481 @$(MAKE) check-journal-io-mechanic 482 482 @echo "" 483 + @echo "=== Running journal-config owner check ===" 484 + @$(MAKE) check-journal-config-owner 485 + @echo "" 483 486 @echo "=== Running call-http-only check ===" 484 487 @$(MAKE) check-call-http-only 485 488 @echo "" ··· 604 607 # Journal raw-mechanic check (see AGENTS.md §7 L2) 605 608 check-journal-io-mechanic: .installed 606 609 $(VENV_BIN)/python scripts/check_journal_io_mechanic.py 610 + 611 + # Journal config owner transaction boundary gate 612 + check-journal-config-owner: .installed 613 + $(VENV_BIN)/python scripts/check_journal_config_owner.py 607 614 608 615 # sol call HTTP-only gate (call.py reaches the journal only over HTTP) 609 616 check-call-http-only: .installed
+511
scripts/check_journal_config_owner.py
··· 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 + 7 + Invariant: ``config/journal.json`` is a single-owner transactional domain. 8 + Production code outside ``solstone.think.journal_config`` must not import or 9 + call the private raw serializer, compose its own read/lock/write cycle, replace 10 + the config file directly, re-export or wrap the private serializer, or create a 11 + second lock for the same file. 12 + 13 + Design 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 + 33 + Exit codes: 34 + 0 - no production violations 35 + 1 - any production violation is found 36 + """ 37 + 38 + from __future__ import annotations 39 + 40 + import argparse 41 + import ast 42 + import sys 43 + from pathlib import Path 44 + 45 + ROOT = Path(__file__).resolve().parent.parent 46 + 47 + OWNER_FILES: frozenset[str] = frozenset({"solstone/think/journal_config.py"}) 48 + ALLOWLIST: dict[tuple[str, str], int] = {} 49 + CONFIG_FILENAME = "journal.json" 50 + SIDECAR_LOCK_FILENAME = "." + CONFIG_FILENAME + ".lock" 51 + 52 + 53 + def _is_owner(rel: Path) -> bool: 54 + return rel.as_posix() in OWNER_FILES 55 + 56 + 57 + def _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 + 65 + def 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 + 84 + def _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 + 94 + class _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 + 108 + def _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 + 155 + def _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 + 163 + def _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 + 177 + def _is_private_serializer_call(node: ast.Call, bindings: _Bindings) -> bool: 178 + return _is_private_serializer_expr(node.func, bindings) 179 + 180 + 181 + def _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 + 195 + def _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 + 213 + def _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 + 222 + def _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 + 229 + def _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 + 262 + def _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 + 274 + def _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 + 280 + def _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 + 291 + def _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 + 302 + def _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 + 308 + def _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 + 314 + def _is_path_replace_call(node: ast.Call) -> bool: 315 + return isinstance(node.func, ast.Attribute) and node.func.attr == "replace" 316 + 317 + 318 + def 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 + 433 + def scan_file(path: Path) -> list[tuple[int, str, str]]: 434 + return scan_source(path.read_text(encoding="utf-8"), filename=str(path)) 435 + 436 + 437 + def 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 + 447 + def 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 + 476 + def 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 + 510 + if __name__ == "__main__": 511 + raise SystemExit(main())
+189
tests/test_check_journal_config_owner.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + import importlib.util 7 + import subprocess 8 + import sys 9 + from pathlib import Path 10 + from types import ModuleType 11 + 12 + SCRIPT = ( 13 + Path(__file__).resolve().parents[1] / "scripts" / "check_journal_config_owner.py" 14 + ) 15 + 16 + 17 + def _load_gate() -> ModuleType: 18 + spec = importlib.util.spec_from_file_location("check_journal_config_owner", SCRIPT) 19 + assert spec is not None 20 + assert spec.loader is not None 21 + module = importlib.util.module_from_spec(spec) 22 + spec.loader.exec_module(module) 23 + return module 24 + 25 + 26 + def _kinds(findings: list[tuple[int, str, str]]) -> set[str]: 27 + return {kind for _lineno, kind, _detail in findings} 28 + 29 + 30 + def _write_bad_module(tmp_path: Path, source: str) -> None: 31 + package = tmp_path / "solstone" 32 + package.mkdir() 33 + (package / "bad.py").write_text(source, encoding="utf-8") 34 + 35 + 36 + def _run_gate(tmp_path: Path) -> subprocess.CompletedProcess[str]: 37 + return subprocess.run( 38 + [sys.executable, str(SCRIPT), "--root", str(tmp_path)], 39 + check=False, 40 + capture_output=True, 41 + text=True, 42 + ) 43 + 44 + 45 + def test_scanner_flags_private_serializer_import_and_call() -> None: 46 + gate = _load_gate() 47 + 48 + findings = gate.scan_source( 49 + "from solstone.think.journal_config import _write_journal_config as raw\n" 50 + "raw({})\n" 51 + ) 52 + 53 + assert "private_serializer" in _kinds(findings) 54 + 55 + 56 + def test_scanner_flags_config_specific_atomic_replace() -> None: 57 + gate = _load_gate() 58 + 59 + findings = gate.scan_source( 60 + "from solstone.think.journal_config import get_journal_config_path\n" 61 + "from solstone.think.journal_io.atomic import atomic_replace\n" 62 + "atomic_replace(get_journal_config_path(), '{}')\n" 63 + ) 64 + 65 + assert "journal_config_replace" in _kinds(findings) 66 + 67 + 68 + def test_scanner_flags_private_serializer_wrapper() -> None: 69 + gate = _load_gate() 70 + 71 + findings = gate.scan_source( 72 + "import solstone.think.journal_config as jc\n" 73 + "write_journal_config = jc._write_journal_config\n" 74 + ) 75 + 76 + assert "private_serializer_wrapper" in _kinds(findings) 77 + 78 + 79 + def test_scanner_flags_second_config_lock() -> None: 80 + gate = _load_gate() 81 + 82 + findings = gate.scan_source( 83 + "from solstone.think.journal_config import get_journal_config_path\n" 84 + "from solstone.think.journal_io.locking import hold_lock\n" 85 + "with hold_lock(get_journal_config_path()):\n" 86 + " pass\n" 87 + ) 88 + 89 + assert "second_config_lock" in _kinds(findings) 90 + 91 + 92 + def test_scanner_flags_hand_rolled_sidecar_flock() -> None: 93 + gate = _load_gate() 94 + 95 + findings = gate.scan_source( 96 + "import fcntl\n" 97 + "from pathlib import Path\n" 98 + "lock_path = Path('journal') / 'config' / ('.' + 'journal.json' + '.lock')\n" 99 + "lock_file = open(lock_path, 'w')\n" 100 + "fcntl.flock(lock_file, fcntl.LOCK_EX)\n" 101 + ) 102 + 103 + assert "second_config_lock" in _kinds(findings) 104 + 105 + 106 + def test_scanner_allows_atomic_replace_on_non_config_domain() -> None: 107 + gate = _load_gate() 108 + 109 + findings = gate.scan_source( 110 + "from pathlib import Path\n" 111 + "from solstone.think.journal_io.atomic import atomic_replace\n" 112 + "atomic_replace(Path('config') / 'chat.json', '{}')\n" 113 + ) 114 + 115 + assert findings == [] 116 + 117 + 118 + def test_e2e_flags_private_serializer(tmp_path: Path) -> None: 119 + _write_bad_module( 120 + tmp_path, 121 + "from solstone.think.journal_config import _write_journal_config as raw\n" 122 + "raw({})\n", 123 + ) 124 + 125 + result = _run_gate(tmp_path) 126 + 127 + assert result.returncode == 1 128 + assert "journal-config-owner: NEW violations:" in result.stderr 129 + assert "private_serializer" in result.stderr 130 + 131 + 132 + def test_e2e_flags_config_specific_atomic_replace(tmp_path: Path) -> None: 133 + _write_bad_module( 134 + tmp_path, 135 + "from solstone.think.journal_config import get_journal_config_path\n" 136 + "from solstone.think.journal_io.atomic import atomic_replace\n" 137 + "atomic_replace(get_journal_config_path(), '{}')\n", 138 + ) 139 + 140 + result = _run_gate(tmp_path) 141 + 142 + assert result.returncode == 1 143 + assert "journal-config-owner: NEW violations:" in result.stderr 144 + assert "journal_config_replace" in result.stderr 145 + 146 + 147 + def test_e2e_flags_private_serializer_wrapper(tmp_path: Path) -> None: 148 + _write_bad_module( 149 + tmp_path, 150 + "import solstone.think.journal_config as jc\n" 151 + "write_journal_config = jc._write_journal_config\n", 152 + ) 153 + 154 + result = _run_gate(tmp_path) 155 + 156 + assert result.returncode == 1 157 + assert "journal-config-owner: NEW violations:" in result.stderr 158 + assert "private_serializer_wrapper" in result.stderr 159 + 160 + 161 + def test_e2e_flags_second_config_lock(tmp_path: Path) -> None: 162 + _write_bad_module( 163 + tmp_path, 164 + "from solstone.think.journal_config import get_journal_config_path\n" 165 + "from solstone.think.journal_io.locking import hold_lock\n" 166 + "with hold_lock(get_journal_config_path()):\n" 167 + " pass\n", 168 + ) 169 + 170 + result = _run_gate(tmp_path) 171 + 172 + assert result.returncode == 1 173 + assert "journal-config-owner: NEW violations:" in result.stderr 174 + assert "second_config_lock" in result.stderr 175 + 176 + 177 + def test_e2e_allows_atomic_replace_on_non_config_domain(tmp_path: Path) -> None: 178 + _write_bad_module( 179 + tmp_path, 180 + "from pathlib import Path\n" 181 + "from solstone.think.journal_io.atomic import atomic_replace\n" 182 + "atomic_replace(Path('config') / 'chat.json', '{}')\n", 183 + ) 184 + 185 + result = _run_gate(tmp_path) 186 + 187 + assert result.returncode == 0 188 + assert "journal-config-owner: pass" in result.stdout 189 + assert result.stderr == ""