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 asyncio
7import errno
8import json
9import os
10import subprocess
11import sys
12import threading
13import time
14
15import pytest
16
17from solstone.think.providers.local_admission import (
18 LocalAdmissionTimeout,
19 LocalSlotLease,
20 acquire_local_slot,
21 acquire_local_slot_async,
22 record_local_inference,
23)
24
25
26def _isolated_journal(monkeypatch, tmp_path):
27 import solstone.think.utils as think_utils
28
29 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
30 think_utils._journal_path_cache = None
31
32
33def _admission_root(tmp_path):
34 return tmp_path / "health" / "local-inference-admission"
35
36
37def _wait_for(predicate, timeout_s: float = 1.0) -> None:
38 deadline = time.monotonic() + timeout_s
39 while not predicate():
40 if time.monotonic() >= deadline:
41 raise AssertionError("condition was not met before timeout")
42 time.sleep(0.005)
43
44
45def _wait_for_ticket_count(tmp_path, count: int) -> None:
46 root = _admission_root(tmp_path)
47 _wait_for(lambda: len(list(root.glob("wait-*.ticket"))) >= count)
48
49
50def test_cross_thread_admission_never_exceeds_capacity(monkeypatch, tmp_path):
51 _isolated_journal(monkeypatch, tmp_path)
52 active = 0
53 peak = 0
54 lock = threading.Lock()
55
56 def work() -> None:
57 nonlocal active, peak
58 with acquire_local_slot(2, 2):
59 with lock:
60 active += 1
61 peak = max(peak, active)
62 time.sleep(0.04)
63 with lock:
64 active -= 1
65
66 threads = [threading.Thread(target=work) for _ in range(8)]
67 for thread in threads:
68 thread.start()
69 for thread in threads:
70 thread.join()
71
72 assert peak == 2
73
74
75def test_failure_releases_permit(monkeypatch, tmp_path):
76 _isolated_journal(monkeypatch, tmp_path)
77
78 with pytest.raises(RuntimeError, match="boom"):
79 with acquire_local_slot(1, 0.5):
80 raise RuntimeError("boom")
81
82 with acquire_local_slot(1, 0.1) as permit:
83 assert permit.slot_index == 0
84
85
86def test_sync_queue_timeout(monkeypatch, tmp_path):
87 _isolated_journal(monkeypatch, tmp_path)
88 first = acquire_local_slot(1, 0.1)
89 try:
90 with pytest.raises(LocalAdmissionTimeout):
91 acquire_local_slot(1, 0.03)
92 finally:
93 first.release()
94
95
96def test_exclusive_waits_for_single_slot_holder(monkeypatch, tmp_path):
97 _isolated_journal(monkeypatch, tmp_path)
98 holder = acquire_local_slot(2, 0.1)
99 acquired: list[int] = []
100
101 def wait_exclusive() -> None:
102 with acquire_local_slot(2, 1.0, exclusive=True) as permit:
103 acquired.append(permit.slot_index)
104
105 thread = threading.Thread(target=wait_exclusive)
106 thread.start()
107 time.sleep(0.05)
108 assert acquired == []
109
110 holder.release()
111 thread.join(timeout=1.0)
112
113 assert not thread.is_alive()
114 assert acquired == [0]
115
116
117def test_exclusive_capacity_one_degenerates_to_single_slot(monkeypatch, tmp_path):
118 _isolated_journal(monkeypatch, tmp_path)
119
120 with acquire_local_slot(1, 0.1, exclusive=True) as permit:
121 assert permit.slot_index == 0
122 assert permit.capacity == 1
123
124
125def test_exclusive_timeout_does_not_strand_partial_locks(monkeypatch, tmp_path):
126 _isolated_journal(monkeypatch, tmp_path)
127 slot_zero = acquire_local_slot(2, 0.1)
128 slot_one = acquire_local_slot(2, 0.1)
129 slot_zero.release()
130 try:
131 with pytest.raises(LocalAdmissionTimeout):
132 acquire_local_slot(2, 0.03, exclusive=True)
133
134 with acquire_local_slot(2, 0.1) as permit:
135 assert permit.slot_index == 0
136 finally:
137 slot_one.release()
138
139
140def test_exclusive_flock_error_releases_partial_slot_set(monkeypatch, tmp_path):
141 _isolated_journal(monkeypatch, tmp_path)
142
143 from solstone.think.providers import local_admission
144
145 real_flock = local_admission.fcntl.flock
146
147 def fail_second_slot(lock_file, operation):
148 if (
149 str(lock_file.name).endswith("slot-1.lock")
150 and operation & local_admission.fcntl.LOCK_EX
151 and operation & local_admission.fcntl.LOCK_NB
152 ):
153 raise OSError(errno.EIO, "simulated flock failure")
154 return real_flock(lock_file, operation)
155
156 monkeypatch.setattr(local_admission.fcntl, "flock", fail_second_slot)
157
158 with pytest.raises(OSError):
159 acquire_local_slot(2, 0.1, exclusive=True)
160
161 with acquire_local_slot(1, 0.1) as permit:
162 assert permit.slot_index == 0
163
164
165def test_exclusive_failure_releases_all_slots(monkeypatch, tmp_path):
166 _isolated_journal(monkeypatch, tmp_path)
167
168 with pytest.raises(RuntimeError, match="boom"):
169 with acquire_local_slot(2, 0.5, exclusive=True):
170 raise RuntimeError("boom")
171
172 first = acquire_local_slot(2, 0.1)
173 try:
174 second = acquire_local_slot(2, 0.1)
175 try:
176 assert {first.slot_index, second.slot_index} == {0, 1}
177 finally:
178 second.release()
179 finally:
180 first.release()
181
182
183def test_async_queued_cancellation_does_not_leak(monkeypatch, tmp_path):
184 _isolated_journal(monkeypatch, tmp_path)
185
186 async def exercise() -> None:
187 first = await acquire_local_slot_async(1, 0.5)
188 queued = asyncio.create_task(acquire_local_slot_async(1, 1.0))
189 await asyncio.sleep(0.05)
190 queued.cancel()
191 with pytest.raises(asyncio.CancelledError):
192 await queued
193 assert not list(
194 (tmp_path / "health" / "local-inference-admission").glob("wait-*.ticket")
195 )
196 first.release()
197 second = await acquire_local_slot_async(1, 0.1)
198 second.release()
199
200 asyncio.run(exercise())
201
202
203def test_sync_cancel_event_drops_ticket_and_releases_queue(monkeypatch, tmp_path):
204 _isolated_journal(monkeypatch, tmp_path)
205
206 from solstone.think.providers import local_admission
207
208 holder = acquire_local_slot(1, 0.1)
209 cancel_event = threading.Event()
210 errors: list[BaseException] = []
211
212 def wait_for_slot() -> None:
213 try:
214 acquire_local_slot(1, 2.0, cancel_event=cancel_event)
215 except BaseException as exc:
216 errors.append(exc)
217
218 thread = threading.Thread(target=wait_for_slot)
219 thread.start()
220 _wait_for_ticket_count(tmp_path, 1)
221
222 cancel_event.set()
223 thread.join(timeout=1.0)
224 holder.release()
225
226 assert not thread.is_alive()
227 assert len(errors) == 1
228 assert isinstance(errors[0], local_admission.LocalAdmissionCancelled)
229 assert not list(_admission_root(tmp_path).glob("wait-*.ticket"))
230 with acquire_local_slot(1, 0.1) as permit:
231 assert permit.slot_index == 0
232
233
234def test_sync_cancel_event_after_acquire_releases_permit(monkeypatch, tmp_path):
235 _isolated_journal(monkeypatch, tmp_path)
236
237 from solstone.think.providers import local_admission
238
239 cancel_event = threading.Event()
240 real_try_acquire = local_admission._try_acquire
241
242 def cancel_after_acquire(capacity, started, root):
243 permit = real_try_acquire(capacity, started, root)
244 if permit is not None:
245 cancel_event.set()
246 return permit
247
248 monkeypatch.setattr(local_admission, "_try_acquire", cancel_after_acquire)
249
250 with pytest.raises(local_admission.LocalAdmissionCancelled):
251 acquire_local_slot(1, 0.1, cancel_event=cancel_event)
252
253 monkeypatch.setattr(local_admission, "_try_acquire", real_try_acquire)
254 with acquire_local_slot(1, 0.1) as permit:
255 assert permit.slot_index == 0
256
257
258def test_lease_close_cancels_pending_reacquire_without_leaking(
259 monkeypatch,
260 tmp_path,
261):
262 _isolated_journal(monkeypatch, tmp_path)
263
264 from solstone.think.providers import local_admission
265
266 initial = acquire_local_slot(1, 0.1)
267 lease = LocalSlotLease(
268 capacity=1,
269 deadline=time.monotonic() + 2.0,
270 permit=initial,
271 )
272 lease.yield_slot()
273 holder = acquire_local_slot(1, 0.1)
274 errors: list[BaseException] = []
275
276 def reacquire() -> None:
277 try:
278 lease.reacquire()
279 except BaseException as exc:
280 errors.append(exc)
281
282 thread = threading.Thread(target=reacquire)
283 thread.start()
284 _wait_for_ticket_count(tmp_path, 1)
285
286 lease.close()
287 thread.join(timeout=1.0)
288 holder.release()
289
290 assert not thread.is_alive()
291 assert len(errors) == 1
292 assert isinstance(errors[0], local_admission.LocalAdmissionCancelled)
293 assert not list(_admission_root(tmp_path).glob("wait-*.ticket"))
294 with acquire_local_slot(1, 0.1) as permit:
295 assert permit.slot_index == 0
296
297
298def test_waiters_are_admitted_in_ticket_order(monkeypatch, tmp_path):
299 _isolated_journal(monkeypatch, tmp_path)
300 root = _admission_root(tmp_path)
301 first = acquire_local_slot(1, 1)
302 order: list[int] = []
303
304 def wait(index: int) -> None:
305 with acquire_local_slot(1, 2):
306 order.append(index)
307 time.sleep(0.01)
308
309 threads = []
310 for index in range(5):
311 thread = threading.Thread(target=wait, args=(index,))
312 thread.start()
313 threads.append(thread)
314 deadline = time.monotonic() + 1
315 while len(list(root.glob("wait-*.ticket"))) < index + 1:
316 assert time.monotonic() < deadline
317 time.sleep(0.005)
318
319 first.release()
320 for thread in threads:
321 thread.join()
322
323 assert order == list(range(5))
324
325
326def test_yielded_parent_reacquire_respects_existing_fifo_waiter(
327 monkeypatch,
328 tmp_path,
329):
330 _isolated_journal(monkeypatch, tmp_path)
331
332 initial = acquire_local_slot(1, 0.1)
333 lease = LocalSlotLease(
334 capacity=1,
335 deadline=time.monotonic() + 2.0,
336 permit=initial,
337 )
338 order: list[str] = []
339 waiter_entered = threading.Event()
340 release_waiter = threading.Event()
341 parent_reacquired = threading.Event()
342
343 def waiter() -> None:
344 with acquire_local_slot(1, 2.0):
345 order.append("waiter")
346 waiter_entered.set()
347 assert release_waiter.wait(1.0)
348
349 waiter_thread = threading.Thread(target=waiter)
350 waiter_thread.start()
351 _wait_for_ticket_count(tmp_path, 1)
352
353 lease.yield_slot()
354 assert waiter_entered.wait(1.0)
355
356 def parent() -> None:
357 lease.reacquire()
358 order.append("parent")
359 parent_reacquired.set()
360
361 parent_thread = threading.Thread(target=parent)
362 parent_thread.start()
363 _wait_for_ticket_count(tmp_path, 1)
364
365 assert order == ["waiter"]
366 release_waiter.set()
367 assert parent_reacquired.wait(1.0)
368 parent_thread.join(timeout=1.0)
369 waiter_thread.join(timeout=1.0)
370 lease.close()
371
372 assert not parent_thread.is_alive()
373 assert not waiter_thread.is_alive()
374 assert order == ["waiter", "parent"]
375
376
377def test_cross_process_nested_acquire_succeeds_while_parent_yielded(
378 monkeypatch,
379 tmp_path,
380):
381 _isolated_journal(monkeypatch, tmp_path)
382
383 initial = acquire_local_slot(1, 0.1)
384 lease = LocalSlotLease(
385 capacity=1,
386 deadline=time.monotonic() + 2.0,
387 permit=initial,
388 )
389 env = {**os.environ, "SOLSTONE_JOURNAL": str(tmp_path)}
390 script = """
391from solstone.think.providers.local_admission import acquire_local_slot
392with acquire_local_slot(1, 1.0) as permit:
393 assert permit.slot_index == 0
394"""
395
396 try:
397 lease.yield_slot()
398 completed = subprocess.run(
399 [sys.executable, "-c", script],
400 env=env,
401 text=True,
402 capture_output=True,
403 timeout=2.0,
404 check=False,
405 )
406 assert completed.returncode == 0, completed.stderr
407 lease.reacquire()
408 finally:
409 lease.close()
410
411 with acquire_local_slot(1, 0.1) as permit:
412 assert permit.slot_index == 0
413
414
415def test_capacity_two_two_parents_run_cross_process_nested_acquires(
416 monkeypatch,
417 tmp_path,
418):
419 _isolated_journal(monkeypatch, tmp_path)
420
421 env = {**os.environ, "SOLSTONE_JOURNAL": str(tmp_path)}
422 script = """
423from solstone.think.providers.local_admission import acquire_local_slot
424with acquire_local_slot(2, 1.0):
425 pass
426"""
427 barrier = threading.Barrier(2)
428 results: list[int] = []
429 errors: list[BaseException] = []
430 lock = threading.Lock()
431
432 def parent() -> None:
433 lease: LocalSlotLease | None = None
434 try:
435 permit = acquire_local_slot(2, 1.0)
436 lease = LocalSlotLease(
437 capacity=2,
438 deadline=time.monotonic() + 3.0,
439 permit=permit,
440 )
441 barrier.wait(timeout=1.0)
442 lease.yield_slot()
443 completed = subprocess.run(
444 [sys.executable, "-c", script],
445 env=env,
446 text=True,
447 capture_output=True,
448 timeout=2.0,
449 check=False,
450 )
451 with lock:
452 results.append(completed.returncode)
453 if completed.returncode != 0:
454 raise AssertionError(completed.stderr)
455 lease.reacquire()
456 except BaseException as exc:
457 with lock:
458 errors.append(exc)
459 finally:
460 if lease is not None:
461 lease.close()
462
463 threads = [threading.Thread(target=parent) for _ in range(2)]
464 for thread in threads:
465 thread.start()
466 for thread in threads:
467 thread.join(timeout=4.0)
468
469 assert all(not thread.is_alive() for thread in threads)
470 assert errors == []
471 assert results == [0, 0]
472 first = acquire_local_slot(2, 0.1)
473 try:
474 second = acquire_local_slot(2, 0.1)
475 try:
476 assert {first.slot_index, second.slot_index} == {0, 1}
477 finally:
478 second.release()
479 finally:
480 first.release()
481
482
483def test_stale_ticket_from_exited_owner_is_pruned(monkeypatch, tmp_path):
484 _isolated_journal(monkeypatch, tmp_path)
485 root = tmp_path / "health" / "local-inference-admission"
486 root.mkdir(parents=True)
487 stale = root / "wait-00000000000000000000-1-stale.ticket"
488 stale.write_text("", encoding="utf-8")
489
490 with acquire_local_slot(1, 0.2):
491 assert not stale.exists()
492
493
494def test_telemetry_is_durable_and_content_free(monkeypatch, tmp_path):
495 _isolated_journal(monkeypatch, tmp_path)
496 record_local_inference(
497 {
498 "timestamp": 1.0,
499 "request_id": "abc",
500 "provider": "local",
501 "model": "local/qwen3.5-4b",
502 "queue_wait_ms": 12.5,
503 "outcome": "success",
504 }
505 )
506
507 path = tmp_path / "health" / "local-inference" / time.strftime("%Y%m%d.jsonl")
508 row = json.loads(path.read_text(encoding="utf-8"))
509 assert row["request_id"] == "abc"
510 assert "prompt" not in row
511 assert "output" not in row