personal memory agent
1#!/usr/bin/env python3
2# SPDX-License-Identifier: AGPL-3.0-only
3# Copyright (c) 2026 sol pbc
4
5"""Static no-Python-spawn lint for the migrated native sol surface."""
6
7from __future__ import annotations
8
9import re
10from dataclasses import dataclass
11from pathlib import Path
12
13try:
14 from scripts.build_native_sol_inventory import REPO_ROOT, discover
15except ModuleNotFoundError: # pragma: no cover - direct script execution path.
16 from build_native_sol_inventory import REPO_ROOT, discover # type: ignore[no-redef]
17
18ALLOWLIST: dict[tuple[str, str], str] = {}
19
20CLIENT_CRATES = (
21 REPO_ROOT / "core/crates/solstone-core-sol-client",
22 REPO_ROOT / "core/crates/solstone-core-sol-client-cli",
23)
24SEAM_FILE = REPO_ROOT / "core/crates/solstone-core-sol-client/src/seam.rs"
25CLI_LIB_FILE = REPO_ROOT / "core/crates/solstone-core-sol-client-cli/src/lib.rs"
26PARITY_TEST_FILE = (
27 REPO_ROOT / "core/crates/solstone-core-sol-client-cli/tests/parity.rs"
28)
29
30FORBIDDEN_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
31 ("direct-std-process-command", re.compile(r"\bstd::process::Command\b")),
32 ("direct-tokio-process", re.compile(r"\btokio::process\b")),
33 ("direct-process-command", re.compile(r"\bprocess::Command\b")),
34 ("direct-command-new", re.compile(r"\bCommand::new\s*\(")),
35 ("direct-spawn-call", re.compile(r"\.spawn\s*\(")),
36 ("direct-output-call", re.compile(r"\.output\s*\(")),
37 ("direct-exec-call", re.compile(r"\bexec(?:[lv][pe]?|ve)?\s*\(")),
38 ("pyo3-reference", re.compile(r"\b(?:pyo3|PyO3)\b")),
39 ("cpython-reference", re.compile(r"\b(?:cpython|CPython)\b")),
40 ("python-fallback-symbol", re.compile(r"\bpython_(?:fallback|dispatch)\b")),
41 ("compat-dispatch-symbol", re.compile(r"\bcompat(?:ibility)?_dispatch\b")),
42 ("fallback-to-python-symbol", re.compile(r"\bfallback_to_python\b")),
43 (
44 "python-fallback-string",
45 re.compile(r"\b(?:fallback|dispatch)[^\n\"]*python3?\b"),
46 ),
47)
48
49
50@dataclass(frozen=True)
51class Violation:
52 file: str
53 kind: str
54 detail: str
55
56 @property
57 def key(self) -> tuple[str, str]:
58 return (self.file, self.kind)
59
60
61def rel(path: Path) -> str:
62 return path.relative_to(REPO_ROOT).as_posix()
63
64
65def rust_files_to_scan() -> list[Path]:
66 files: set[Path] = set()
67 files.update(authority_sources())
68 for crate in CLIENT_CRATES:
69 files.update(crate.rglob("*.rs"))
70 return sorted(path for path in files if path.is_file())
71
72
73def authority_sources() -> set[Path]:
74 return {entry.source for entry in discover(REPO_ROOT)}
75
76
77def collect_violations() -> list[Violation]:
78 violations: list[Violation] = []
79 for path in rust_files_to_scan():
80 text = path.read_text(encoding="utf-8")
81 violations.extend(scan_forbidden_patterns(path, text))
82 violations.extend(scan_process_spawner_import(path, text))
83 violations.extend(check_required_spawn_guard_tests())
84 return violations
85
86
87def scan_forbidden_patterns(path: Path, text: str) -> list[Violation]:
88 violations: list[Violation] = []
89 for kind, pattern in FORBIDDEN_PATTERNS:
90 if pattern.search(text):
91 violations.append(
92 Violation(
93 rel(path),
94 kind,
95 "migrated native sol code must not spawn or dispatch to Python",
96 )
97 )
98 return violations
99
100
101def scan_process_spawner_import(path: Path, text: str) -> list[Violation]:
102 if path == SEAM_FILE:
103 return []
104 if re.search(r"\bProcessSpawner\b", text):
105 return [
106 Violation(
107 rel(path),
108 "process-spawner-outside-seam",
109 "ProcessSpawner is only allowed in the shared seam definition; "
110 "host implementations belong in the process shell",
111 )
112 ]
113 return []
114
115
116def check_required_spawn_guard_tests() -> list[Violation]:
117 checks = {
118 SEAM_FILE: (
119 "missing-failing-spawner-test",
120 "failing_spawner_always_errors",
121 ),
122 CLI_LIB_FILE: (
123 "missing-unsupported-without-spawn-test",
124 "classifies_unported_call_as_unsupported_without_spawn_path",
125 ),
126 PARITY_TEST_FILE: (
127 "missing-native-parity-test",
128 "native_matches_sol_call_parity_vectors",
129 ),
130 }
131 violations: list[Violation] = []
132 for path, (kind, needle) in checks.items():
133 text = path.read_text(encoding="utf-8")
134 if needle in text:
135 continue
136 violations.append(
137 Violation(
138 rel(path),
139 kind,
140 f"required Rust proof {needle} is missing",
141 )
142 )
143 return violations
144
145
146def main() -> int:
147 violations = collect_violations()
148 actual = {violation.key: violation for violation in violations}
149 allowed = set(ALLOWLIST)
150 unexpected = sorted(set(actual) - allowed)
151 stale = sorted(allowed - set(actual))
152
153 if unexpected or stale:
154 if unexpected:
155 print("native sol no-python-spawn violations:")
156 for key in unexpected:
157 violation = actual[key]
158 print(f"- {violation.file}: {violation.kind}: {violation.detail}")
159 if stale:
160 print("stale native sol no-python-spawn allowlist entries:")
161 for file, kind in stale:
162 print(f"- {file}: {kind}: {ALLOWLIST[(file, kind)]}")
163 return 1
164
165 print("native sol no-python-spawn ok")
166 return 0
167
168
169if __name__ == "__main__":
170 raise SystemExit(main())