solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import importlib.util
7import subprocess
8import sys
9from pathlib import Path
10from types import ModuleType
11
12import pytest
13
14SCRIPT = (
15 Path(__file__).resolve().parents[1] / "scripts" / "check_provider_start_commands.py"
16)
17
18
19def _load_checker() -> ModuleType:
20 spec = importlib.util.spec_from_file_location(
21 "check_provider_start_commands", SCRIPT
22 )
23 assert spec is not None
24 assert spec.loader is not None
25 module = importlib.util.module_from_spec(spec)
26 spec.loader.exec_module(module)
27 return module
28
29
30def _write_file(root: Path, rel: str, source: str) -> Path:
31 path = root / rel
32 path.parent.mkdir(parents=True, exist_ok=True)
33 path.write_text(source, encoding="utf-8")
34 return path
35
36
37def _run(root: Path) -> subprocess.CompletedProcess[str]:
38 return subprocess.run(
39 [sys.executable, str(SCRIPT), "--root", str(root)],
40 check=False,
41 capture_output=True,
42 text=True,
43 )
44
45
46def _kinds(findings: list[tuple[int, str, str]]) -> set[str]:
47 return {kind for _lineno, kind, _detail in findings}
48
49
50checker = _load_checker()
51
52
53def test_scan_flags_provider_start_commands() -> None:
54 findings = checker.scan_source(
55 "from solstone.think.callosum import callosum_send as send\n"
56 "def f(client):\n"
57 " send('supervisor', 'start_local')\n"
58 " client.emit(tract='supervisor', event='start_parakeet')\n"
59 " msg = {'tract': 'supervisor', 'event': 'start_local'}\n"
60 " return msg\n"
61 )
62
63 kinds = _kinds(findings)
64 assert "provider_start_command" in kinds
65 assert "provider_start_message" in kinds
66
67
68@pytest.mark.parametrize(
69 "source",
70 [
71 "from solstone.think import callosum\n"
72 "callosum.callosum_send('supervisor', 'start_local')\n",
73 "from solstone.think import callosum as bus\n"
74 "bus.callosum_send('supervisor', 'start_parakeet')\n",
75 "import solstone.think.callosum as callosum\n"
76 "callosum.callosum_send('supervisor', 'start_local')\n",
77 ],
78)
79def test_scan_flags_reasonable_callosum_import_spellings(source: str) -> None:
80 findings = checker.scan_source(source)
81
82 assert _kinds(findings) == {"provider_start_command"}
83
84
85def test_scan_flags_restored_handler_names() -> None:
86 findings = checker.scan_source(
87 "def _handle_supervisor_start_local(message):\n"
88 " return None\n"
89 "def _request_parakeet_server_start():\n"
90 " return None\n"
91 )
92
93 assert _kinds(findings) == {"provider_start_handler"}
94
95
96def test_scan_allows_unrelated_callosum_traffic_and_raw_strings() -> None:
97 findings = checker.scan_source(
98 "def f(client):\n"
99 " client.emit('thinking', 'start_local')\n"
100 " client.emit('supervisor', 'restart')\n"
101 " msg = {'tract': 'thinking', 'event': 'start_parakeet'}\n"
102 " raw = 'start_local start_parakeet'\n"
103 " return msg, raw\n"
104 )
105
106 assert findings == []
107
108
109def test_e2e_flags_violation(tmp_path: Path) -> None:
110 _write_file(
111 tmp_path,
112 "solstone/bad.py",
113 "from solstone.think.callosum import callosum_send\n"
114 "callosum_send('supervisor', 'start_parakeet')\n",
115 )
116
117 result = _run(tmp_path)
118
119 assert result.returncode == 1
120 assert "provider-start-commands: violations:" in result.stderr
121 assert "provider_start_command" in result.stderr
122
123
124def test_e2e_clean_source_passes(tmp_path: Path) -> None:
125 _write_file(
126 tmp_path,
127 "solstone/good.py",
128 "def f(client):\n client.emit('supervisor', 'restart')\n",
129 )
130
131 result = _run(tmp_path)
132
133 assert result.returncode == 0
134 assert "provider-start-commands: pass" in result.stdout
135 assert result.stderr == ""
136
137
138def test_allowlist_is_empty_and_ratcheted(tmp_path: Path) -> None:
139 assert checker.ALLOWLIST == {}
140 _write_file(
141 tmp_path,
142 "solstone/bad.py",
143 "def _handle_supervisor_start_parakeet(message):\n return None\n",
144 )
145 counts = checker.count_violations(tmp_path)
146
147 over, stale, tracked = checker.evaluate(tmp_path, counts)
148 assert over == []
149 assert stale == []
150 assert tracked
151
152 ratcheted = {next(iter(counts)): 0}
153 over, stale, _tracked = checker.evaluate(tmp_path, ratcheted)
154 assert over
155 assert stale == []
156
157 stale_allowlist = {("solstone/missing.py", "provider_start_handler"): 1}
158 over, stale, _tracked = checker.evaluate(tmp_path, stale_allowlist)
159 assert over
160 assert stale
161
162
163def test_landed_tree_is_clean() -> None:
164 over, stale, _tracked = checker.evaluate(checker.ROOT, checker.ALLOWLIST)
165
166 assert over == []
167 assert stale == []