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 hashlib
7import json
8import os
9import subprocess
10import sys
11import time
12from datetime import datetime, timedelta, timezone
13from pathlib import Path
14
15from solstone.think.models import LOCAL_MODEL
16from solstone.think.providers.brain_state import (
17 DEFAULT_READY_EVIDENCE_TTL,
18 begin_brain_refresh,
19 finish_brain_refresh,
20 inspect_brain_state,
21)
22
23NOW = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
24
25
26def _env(journal: Path) -> dict[str, str]:
27 env = os.environ.copy()
28 env["SOLSTONE_JOURNAL"] = str(journal)
29 return env
30
31
32def _write_config(journal: Path) -> None:
33 path = journal / "config" / "journal.json"
34 path.parent.mkdir(parents=True, exist_ok=True)
35 path.write_text(
36 json.dumps(
37 {
38 "providers": {"active": {"provider": "openai", "model": "gpt-5"}},
39 "env": {"OPENAI_API_KEY": "secret"},
40 }
41 ),
42 encoding="utf-8",
43 )
44
45
46def _write_spp_config(journal: Path) -> None:
47 credential = "endpoint-secret"
48 path = journal / "config" / "journal.json"
49 path.parent.mkdir(parents=True, exist_ok=True)
50 path.write_text(
51 json.dumps(
52 {
53 "providers": {
54 "active": {"provider": "local", "model": LOCAL_MODEL},
55 "local": {
56 "endpoint_url": "https://brain.example.test/v1",
57 "served_model_id": "served-model",
58 "credential": credential,
59 },
60 },
61 "services": {
62 "confidential": {
63 "enabled_at": NOW.isoformat(),
64 "account_id": "acct-a",
65 "endpoint_url": "https://brain.example.test/v1",
66 "served_model_id": "served-model",
67 "credential_created_at": NOW.isoformat(),
68 "credential_fingerprint_sha256": hashlib.sha256(
69 credential.encode("utf-8")
70 ).hexdigest(),
71 "prior_active": {
72 "provider": "google",
73 "model": "gemini-flash-latest",
74 },
75 "prior_local_endpoint": None,
76 }
77 },
78 "env": {},
79 }
80 ),
81 encoding="utf-8",
82 )
83
84
85def _component(now: datetime) -> dict[str, str]:
86 return {
87 "status": "ok",
88 "observed_at": now.isoformat(),
89 "expires_at": (now + DEFAULT_READY_EVIDENCE_TTL).isoformat(),
90 }
91
92
93def _write_spp_ready_record(journal: Path) -> None:
94 _write_spp_config(journal)
95 permit = begin_brain_refresh(NOW, journal_path=journal)
96 assert permit is not None
97 finish_brain_refresh(
98 permit,
99 {
100 "configuration": _component(NOW),
101 "lane_prerequisites": _component(NOW),
102 "generate": _component(NOW),
103 "cogitate": _component(NOW),
104 },
105 NOW,
106 journal_path=journal,
107 )
108
109
110def test_refresh_permit_excludes_contender_and_crash_releases(tmp_path: Path) -> None:
111 _write_config(tmp_path)
112 ready = tmp_path / "ready"
113 holder_code = f"""
114import pathlib
115import time
116from datetime import datetime, timezone
117from solstone.think.providers.brain_state import begin_brain_refresh
118now = datetime.fromisoformat({NOW.isoformat()!r})
119permit = begin_brain_refresh(now, journal_path={str(tmp_path)!r})
120assert permit is not None
121pathlib.Path({str(ready)!r}).write_text("ready")
122while True:
123 time.sleep(0.05)
124"""
125 contender_code = f"""
126from datetime import datetime
127from solstone.think.providers.brain_state import begin_brain_refresh
128now = datetime.fromisoformat({NOW.isoformat()!r})
129permit = begin_brain_refresh(now, journal_path={str(tmp_path)!r})
130if permit is None:
131 print("busy", flush=True)
132else:
133 print("free", flush=True)
134 permit.release()
135"""
136 holder = subprocess.Popen(
137 [sys.executable, "-c", holder_code],
138 cwd=Path.cwd(),
139 env=_env(tmp_path),
140 stdout=subprocess.PIPE,
141 stderr=subprocess.PIPE,
142 text=True,
143 )
144 try:
145 deadline = time.monotonic() + 10
146 while not ready.exists() and time.monotonic() < deadline:
147 time.sleep(0.05)
148 assert ready.exists()
149
150 busy = subprocess.run(
151 [sys.executable, "-c", contender_code],
152 cwd=Path.cwd(),
153 env=_env(tmp_path),
154 capture_output=True,
155 text=True,
156 check=True,
157 )
158 assert busy.stdout.strip() == "busy"
159
160 holder.terminate()
161 stdout, stderr = holder.communicate(timeout=10)
162 assert holder.returncode is not None, (stdout, stderr)
163
164 projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"]
165 assert projection["aggregate_state"] == "unknown"
166 assert projection["reason_code"] == "brain_check_interrupted"
167
168 free = subprocess.run(
169 [sys.executable, "-c", contender_code],
170 cwd=Path.cwd(),
171 env=_env(tmp_path),
172 capture_output=True,
173 text=True,
174 check=True,
175 )
176 assert free.stdout.strip() == "free"
177 finally:
178 if holder.poll() is None:
179 holder.terminate()
180 holder.wait(timeout=5)
181
182
183def test_prerequisite_renewal_permit_excludes_contender_and_crash_releases(
184 tmp_path: Path,
185) -> None:
186 _write_spp_ready_record(tmp_path)
187 ready = tmp_path / "ready"
188 holder_code = f"""
189import pathlib
190import time
191from datetime import datetime
192from solstone.think.providers.brain_state import begin_brain_prerequisite_renewal
193now = datetime.fromisoformat({(NOW + timedelta(seconds=1)).isoformat()!r})
194result = begin_brain_prerequisite_renewal(now, journal_path={str(tmp_path)!r})
195assert result["status"] == "started", result
196pathlib.Path({str(ready)!r}).write_text("ready")
197while True:
198 time.sleep(0.05)
199"""
200 contender_code = f"""
201from datetime import datetime
202from solstone.think.providers.brain_state import begin_brain_prerequisite_renewal
203now = datetime.fromisoformat({(NOW + timedelta(seconds=2)).isoformat()!r})
204result = begin_brain_prerequisite_renewal(now, journal_path={str(tmp_path)!r})
205print(result["status"], flush=True)
206if result["status"] == "started":
207 result["permit"].release()
208"""
209 holder = subprocess.Popen(
210 [sys.executable, "-c", holder_code],
211 cwd=Path.cwd(),
212 env=_env(tmp_path),
213 stdout=subprocess.PIPE,
214 stderr=subprocess.PIPE,
215 text=True,
216 )
217 try:
218 deadline = time.monotonic() + 10
219 while not ready.exists() and time.monotonic() < deadline:
220 time.sleep(0.05)
221 assert ready.exists()
222
223 busy = subprocess.run(
224 [sys.executable, "-c", contender_code],
225 cwd=Path.cwd(),
226 env=_env(tmp_path),
227 capture_output=True,
228 text=True,
229 check=True,
230 )
231 assert busy.stdout.strip() == "busy"
232
233 holder.terminate()
234 stdout, stderr = holder.communicate(timeout=10)
235 assert holder.returncode is not None, (stdout, stderr)
236
237 free = subprocess.run(
238 [sys.executable, "-c", contender_code],
239 cwd=Path.cwd(),
240 env=_env(tmp_path),
241 capture_output=True,
242 text=True,
243 check=True,
244 )
245 assert free.stdout.strip() == "started"
246 finally:
247 if holder.poll() is None:
248 holder.terminate()
249 holder.wait(timeout=5)