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 stat
7from collections.abc import Iterator
8from contextlib import contextmanager
9from pathlib import Path
10from typing import Any
11
12import pytest
13
14from solstone.think.journal_io.errors import LockTimeout
15from solstone.think.providers import runtime_health
16from solstone.think.providers.runtime_health import (
17 RuntimeHealthConflictError,
18 RuntimeHealthMalformedError,
19 RuntimeHealthRecord,
20 RuntimeHealthUnavailableError,
21 consume_retry_token,
22 inspect_retry_token,
23 inspect_runtime_health,
24 make_synthetic_runtime_health,
25 observe_runtime_repair,
26 read_retry_token,
27 read_runtime_health,
28 repair_corrupt_record,
29 request_retry_token,
30 request_runtime_retry,
31 runtime_directory,
32 runtime_health_path,
33 runtime_operation_lock_path,
34 runtime_operation_path,
35 runtime_retry_token_path,
36 write_runtime_health,
37)
38
39
40@pytest.fixture
41def provider_cache_reset() -> Iterator[None]:
42 from solstone.think.providers import local_server, local_vulkan
43
44 local_vulkan.reset_detect_cache()
45 local_server.reset_parallel_slots_cache()
46 try:
47 yield
48 finally:
49 local_vulkan.reset_detect_cache()
50 local_server.reset_parallel_slots_cache()
51
52
53def _mode(path: Path) -> int:
54 return stat.S_IMODE(path.stat().st_mode)
55
56
57def _health(
58 provider: str = "local",
59 *,
60 phase: str = "observing",
61 fingerprint: str | None = "fp-1",
62) -> RuntimeHealthRecord:
63 record = make_synthetic_runtime_health(provider)
64 record["phase"] = phase
65 record["reason_code"] = "truth-observation-started"
66 record["desired_fingerprint_sha256"] = fingerprint
67 record["updated_at"] = "2026-07-19T00:00:00+00:00"
68 return record
69
70
71def _write_corrupt(path: Path, payload: bytes = b"{not-json") -> None:
72 path.parent.mkdir(parents=True, exist_ok=True)
73 path.write_bytes(payload)
74
75
76def _contains_value(value: Any, needle: str) -> bool:
77 if value == needle:
78 return True
79 if isinstance(value, dict):
80 return any(
81 _contains_value(key, needle) or _contains_value(item, needle)
82 for key, item in value.items()
83 )
84 if isinstance(value, (list, tuple, set, frozenset)):
85 return any(_contains_value(item, needle) for item in value)
86 if isinstance(value, Path):
87 return needle in str(value)
88 if isinstance(value, str):
89 return needle in value
90 return False
91
92
93def test_phase_and_reason_code_vocabularies_are_disjoint() -> None:
94 assert len(runtime_health.RUNTIME_PHASES) == 17
95 assert runtime_health.RUNTIME_PHASES.isdisjoint(runtime_health.REASON_CODES)
96 assert set().union(*runtime_health.REASON_CODE_GROUPS.values()) == set(
97 runtime_health.REASON_CODES
98 )
99
100
101def test_invalid_provider_rejected_on_entry_points(tmp_path: Path) -> None:
102 for call in (
103 lambda: runtime_health_path("mlx", journal_path=tmp_path),
104 lambda: runtime_retry_token_path("mlx", journal_path=tmp_path),
105 lambda: runtime_operation_path("mlx", journal_path=tmp_path),
106 lambda: read_runtime_health("mlx", journal_path=tmp_path),
107 lambda: read_retry_token("mlx", journal_path=tmp_path),
108 lambda: request_retry_token(
109 "mlx",
110 desired_fingerprint_sha256="fp",
111 journal_path=tmp_path,
112 ),
113 ):
114 with pytest.raises(ValueError):
115 call()
116
117
118def test_paths_modes_and_atomic_replace_for_both_records(
119 tmp_path: Path,
120 monkeypatch: pytest.MonkeyPatch,
121) -> None:
122 calls: list[tuple[Path, int | None]] = []
123 real_atomic_replace = runtime_health.atomic_replace
124
125 def spy_atomic_replace(path: Path, data: str | bytes, *, mode: int | None = None):
126 calls.append((path, mode))
127 return real_atomic_replace(path, data, mode=mode)
128
129 monkeypatch.setattr(runtime_health, "atomic_replace", spy_atomic_replace)
130
131 stored_health = write_runtime_health(_health(), journal_path=tmp_path)
132 stored_retry = request_retry_token(
133 "local",
134 desired_fingerprint_sha256="fp-1",
135 owner={"actor": "test"},
136 journal_path=tmp_path,
137 )
138
139 assert stored_health["revision"] == 1
140 assert stored_retry["revision"] == 1
141 assert runtime_health_path("local", journal_path=tmp_path) == (
142 tmp_path / "health" / "providers" / "runtime" / "local.json"
143 )
144 assert runtime_retry_token_path("local", journal_path=tmp_path) == (
145 tmp_path / "health" / "providers" / "runtime" / "local.retry-token.json"
146 )
147 assert runtime_operation_lock_path("local", journal_path=tmp_path) == (
148 tmp_path / "health" / "providers" / "runtime" / "local.operation.lock"
149 )
150 assert _mode(runtime_health_path("local", journal_path=tmp_path)) == 0o600
151 assert _mode(runtime_retry_token_path("local", journal_path=tmp_path)) == 0o600
152 assert _mode(runtime_operation_lock_path("local", journal_path=tmp_path)) == 0o600
153 assert calls == [
154 (runtime_health_path("local", journal_path=tmp_path), 0o600),
155 (runtime_retry_token_path("local", journal_path=tmp_path), 0o600),
156 ]
157 assert not list(runtime_directory(journal_path=tmp_path).glob(".tmp_*.tmp"))
158
159
160def test_records_are_independent_but_share_operation_lock(
161 tmp_path: Path,
162 monkeypatch: pytest.MonkeyPatch,
163) -> None:
164 stored_health = write_runtime_health(_health(), journal_path=tmp_path)
165 stored_retry = request_retry_token(
166 "local",
167 desired_fingerprint_sha256="fp-1",
168 journal_path=tmp_path,
169 )
170
171 updated_health = {**stored_health, "phase": "starting"}
172 updated_health["reason_code"] = "launch-requested"
173 written_health = write_runtime_health(updated_health, journal_path=tmp_path)
174 assert read_retry_token("local", journal_path=tmp_path) == stored_retry
175
176 written_retry = request_retry_token(
177 "local",
178 desired_fingerprint_sha256="fp-1",
179 owner={"actor": "second"},
180 journal_path=tmp_path,
181 )
182 assert read_runtime_health("local", journal_path=tmp_path) == written_health
183 assert written_retry["token_id"] == stored_retry["token_id"]
184
185 blocked = {runtime_operation_path("local", journal_path=tmp_path)}
186
187 @contextmanager
188 def fake_hold_lock(
189 path: Path,
190 *,
191 timeout: float = 10.0,
192 poll_interval: float = 0.05,
193 mode: int | None = None,
194 ) -> Iterator[None]:
195 del poll_interval, mode
196 if path in blocked:
197 raise LockTimeout(path=path, timeout=timeout)
198 yield
199
200 monkeypatch.setattr(runtime_health, "hold_lock", fake_hold_lock)
201
202 with pytest.raises(RuntimeHealthUnavailableError):
203 write_runtime_health(written_health, journal_path=tmp_path)
204 with pytest.raises(RuntimeHealthUnavailableError):
205 request_retry_token(
206 "local",
207 desired_fingerprint_sha256="fp-1",
208 journal_path=tmp_path,
209 )
210
211
212def test_absent_records_are_synthetic_and_read_verbs_do_not_create_directory(
213 tmp_path: Path,
214) -> None:
215 assert not runtime_directory(journal_path=tmp_path).exists()
216
217 health = read_runtime_health("local", journal_path=tmp_path)
218 retry = read_retry_token("local", journal_path=tmp_path)
219 health_inspection = inspect_runtime_health("local", journal_path=tmp_path)
220 retry_inspection = inspect_retry_token("local", journal_path=tmp_path)
221 repair_observation = observe_runtime_repair(
222 "local",
223 record_kind="health",
224 journal_path=tmp_path,
225 )
226
227 assert health["phase"] == "stopped"
228 assert health["revision"] == 0
229 assert retry["token_id"] is None
230 assert retry["revision"] == 0
231 assert health_inspection["status"] == "ok"
232 assert retry_inspection["status"] == "ok"
233 assert repair_observation["status"] == "ok"
234 assert not runtime_directory(journal_path=tmp_path).exists()
235
236
237@pytest.mark.parametrize(
238 ("record_kind", "path_func", "read_func", "inspect_func"),
239 [
240 ("health", runtime_health_path, read_runtime_health, inspect_runtime_health),
241 (
242 "retry-token",
243 runtime_retry_token_path,
244 read_retry_token,
245 inspect_retry_token,
246 ),
247 ],
248)
249def test_malformed_record_raises_without_quarantine_or_reinitialize(
250 tmp_path: Path,
251 record_kind: str,
252 path_func,
253 read_func,
254 inspect_func,
255) -> None:
256 path = path_func("local", journal_path=tmp_path)
257 _write_corrupt(path)
258 before = path.read_bytes()
259
260 with pytest.raises(RuntimeHealthMalformedError):
261 read_func("local", journal_path=tmp_path)
262
263 inspection = inspect_func("local", journal_path=tmp_path)
264 assert inspection["status"] == "corrupt"
265 assert inspection["record_kind"] == record_kind
266 assert "repair_handle" not in inspection
267 assert path.read_bytes() == before
268 assert not (runtime_directory(journal_path=tmp_path) / "corrupt").exists()
269
270
271@pytest.mark.parametrize(
272 ("record_kind", "path_func", "read_func", "inspect_func"),
273 [
274 ("health", runtime_health_path, read_runtime_health, inspect_runtime_health),
275 (
276 "retry-token",
277 runtime_retry_token_path,
278 read_retry_token,
279 inspect_retry_token,
280 ),
281 ],
282)
283@pytest.mark.parametrize(
284 "exc", [OSError("disk read failed"), PermissionError("denied")]
285)
286def test_read_oserror_and_permission_denial_are_unavailable_without_reinitialize(
287 tmp_path: Path,
288 monkeypatch: pytest.MonkeyPatch,
289 record_kind: str,
290 path_func,
291 read_func,
292 inspect_func,
293 exc: OSError,
294) -> None:
295 path = path_func("local", journal_path=tmp_path)
296 if record_kind == "health":
297 write_runtime_health(_health(), journal_path=tmp_path)
298 else:
299 request_retry_token(
300 "local",
301 desired_fingerprint_sha256="fp-1",
302 journal_path=tmp_path,
303 )
304 before = path.read_bytes()
305 original_read = runtime_health._read_record_bytes
306
307 def fail_target(target: Path) -> bytes:
308 if target == path:
309 raise exc
310 return original_read(target)
311
312 monkeypatch.setattr(runtime_health, "_read_record_bytes", fail_target)
313
314 with pytest.raises(RuntimeHealthUnavailableError):
315 read_func("local", journal_path=tmp_path)
316
317 inspection = inspect_func("local", journal_path=tmp_path)
318 assert inspection["status"] == "unavailable"
319 assert inspection["reason_code"] == "record-unavailable"
320 assert inspection["record_kind"] == record_kind
321 assert path.read_bytes() == before
322 assert not (runtime_directory(journal_path=tmp_path) / "corrupt").exists()
323
324
325def test_stale_revision_and_fingerprint_rejected(tmp_path: Path) -> None:
326 stored = write_runtime_health(_health(fingerprint="fp-1"), journal_path=tmp_path)
327 stale_revision = {**stored, "revision": 0}
328 stale_revision["phase"] = "starting"
329
330 with pytest.raises(RuntimeHealthConflictError):
331 write_runtime_health(stale_revision, journal_path=tmp_path)
332
333 changed_fingerprint = {**stored, "desired_fingerprint_sha256": "fp-2"}
334 with pytest.raises(RuntimeHealthConflictError):
335 write_runtime_health(
336 changed_fingerprint,
337 expected_desired_fingerprint_sha256="other-fp",
338 journal_path=tmp_path,
339 )
340
341 first = request_retry_token(
342 "local",
343 desired_fingerprint_sha256="fp-1",
344 journal_path=tmp_path,
345 )
346 current = request_retry_token(
347 "local",
348 desired_fingerprint_sha256="fp-1",
349 journal_path=tmp_path,
350 )
351 with pytest.raises(RuntimeHealthConflictError):
352 consume_retry_token(
353 "local",
354 token_id=str(first["token_id"]),
355 revision=first["revision"],
356 desired_fingerprint_sha256="fp-1",
357 journal_path=tmp_path,
358 )
359 with pytest.raises(RuntimeHealthConflictError):
360 consume_retry_token(
361 "local",
362 token_id=str(current["token_id"]),
363 revision=current["revision"],
364 desired_fingerprint_sha256="wrong-fp",
365 journal_path=tmp_path,
366 )
367
368
369def test_retry_token_lifecycle_coalesces_consumes_and_survives_restart(
370 tmp_path: Path,
371 monkeypatch: pytest.MonkeyPatch,
372) -> None:
373 first = request_retry_token(
374 "local",
375 desired_fingerprint_sha256="fp-1",
376 owner={"actor": "first"},
377 journal_path=tmp_path,
378 )
379 second = request_retry_token(
380 "local",
381 desired_fingerprint_sha256="fp-1",
382 owner={"actor": "second"},
383 journal_path=tmp_path,
384 )
385
386 assert second["token_id"] == first["token_id"]
387 assert second["revision"] == first["revision"] + 1
388 assert second["owner"] == {"actor": "second"}
389
390 outstanding = read_retry_token("local", journal_path=tmp_path)
391 assert outstanding == second
392
393 real_atomic_replace = runtime_health.atomic_replace
394
395 def fail_replace(path: Path, data: str | bytes, *, mode: int | None = None):
396 del path, data, mode
397 raise OSError("write failed")
398
399 monkeypatch.setattr(runtime_health, "atomic_replace", fail_replace)
400 with pytest.raises(RuntimeHealthUnavailableError):
401 consume_retry_token(
402 "local",
403 token_id=str(second["token_id"]),
404 revision=second["revision"],
405 desired_fingerprint_sha256="fp-1",
406 journal_path=tmp_path,
407 )
408 monkeypatch.setattr(runtime_health, "atomic_replace", real_atomic_replace)
409 assert read_retry_token("local", journal_path=tmp_path) == second
410
411 consumed = consume_retry_token(
412 "local",
413 token_id=str(second["token_id"]),
414 revision=second["revision"],
415 desired_fingerprint_sha256="fp-1",
416 journal_path=tmp_path,
417 )
418 assert consumed["token_id"] is None
419 assert consumed["revision"] == second["revision"] + 1
420 assert read_retry_token("local", journal_path=tmp_path) == consumed
421
422
423def test_owner_runtime_retry_requires_current_terminal_failure(tmp_path: Path) -> None:
424 failed = _health(phase="failed")
425 failed["reason_code"] = "launch-budget-exhausted"
426 stored = write_runtime_health(failed, journal_path=tmp_path)
427
428 requested = request_runtime_retry(
429 "local",
430 expected_health_revision=stored["revision"],
431 expected_retry_revision=0,
432 desired_fingerprint_sha256="fp-1",
433 owner={"source": "owner-recovery"},
434 journal_path=tmp_path,
435 )
436
437 assert requested["revision"] == 1
438 assert requested["token_id"] is not None
439 assert requested["desired_fingerprint_sha256"] == "fp-1"
440 assert requested["reason_code"] == "retry-token-requested"
441 assert requested["owner"] == {"source": "owner-recovery"}
442 assert read_runtime_health("local", journal_path=tmp_path) == stored
443
444
445@pytest.mark.parametrize("phase", ["ready", "starting", "backoff", "host-blocked"])
446def test_owner_runtime_retry_rejects_nonterminal_state(
447 tmp_path: Path,
448 phase: str,
449) -> None:
450 health = _health(phase=phase)
451 stored = write_runtime_health(health, journal_path=tmp_path)
452
453 with pytest.raises(
454 RuntimeHealthConflictError,
455 match="terminal failure",
456 ):
457 request_runtime_retry(
458 "local",
459 expected_health_revision=stored["revision"],
460 expected_retry_revision=0,
461 desired_fingerprint_sha256="fp-1",
462 journal_path=tmp_path,
463 )
464
465 assert read_retry_token("local", journal_path=tmp_path)["token_id"] is None
466
467
468def test_owner_runtime_retry_rejects_stale_and_outstanding_requests(
469 tmp_path: Path,
470) -> None:
471 failed = _health(phase="failed")
472 failed["reason_code"] = "launch-budget-exhausted"
473 stored = write_runtime_health(failed, journal_path=tmp_path)
474
475 with pytest.raises(RuntimeHealthConflictError, match="health revision"):
476 request_runtime_retry(
477 "local",
478 expected_health_revision=stored["revision"] - 1,
479 expected_retry_revision=0,
480 desired_fingerprint_sha256="fp-1",
481 journal_path=tmp_path,
482 )
483 with pytest.raises(RuntimeHealthConflictError, match="fingerprint"):
484 request_runtime_retry(
485 "local",
486 expected_health_revision=stored["revision"],
487 expected_retry_revision=0,
488 desired_fingerprint_sha256="fp-stale",
489 journal_path=tmp_path,
490 )
491
492 requested = request_runtime_retry(
493 "local",
494 expected_health_revision=stored["revision"],
495 expected_retry_revision=0,
496 desired_fingerprint_sha256="fp-1",
497 journal_path=tmp_path,
498 )
499 with pytest.raises(RuntimeHealthConflictError, match="retry-token revision"):
500 request_runtime_retry(
501 "local",
502 expected_health_revision=stored["revision"],
503 expected_retry_revision=0,
504 desired_fingerprint_sha256="fp-1",
505 journal_path=tmp_path,
506 )
507 with pytest.raises(RuntimeHealthConflictError, match="already requested"):
508 request_runtime_retry(
509 "local",
510 expected_health_revision=stored["revision"],
511 expected_retry_revision=requested["revision"],
512 desired_fingerprint_sha256="fp-1",
513 journal_path=tmp_path,
514 )
515
516
517def test_owner_runtime_retry_replaces_an_old_target_token(tmp_path: Path) -> None:
518 stale = request_retry_token(
519 "local",
520 desired_fingerprint_sha256="fp-old",
521 journal_path=tmp_path,
522 )
523 failed = _health(phase="failed")
524 failed["reason_code"] = "launch-budget-exhausted"
525 stored = write_runtime_health(failed, journal_path=tmp_path)
526
527 requested = request_runtime_retry(
528 "local",
529 expected_health_revision=stored["revision"],
530 expected_retry_revision=stale["revision"],
531 desired_fingerprint_sha256="fp-1",
532 journal_path=tmp_path,
533 )
534
535 assert requested["token_id"] != stale["token_id"]
536 assert requested["desired_fingerprint_sha256"] == "fp-1"
537 assert requested["revision"] == stale["revision"] + 1
538
539
540def test_repair_handle_allows_repair_without_parseable_token(tmp_path: Path) -> None:
541 path = runtime_health_path("local", journal_path=tmp_path)
542 _write_corrupt(path)
543
544 observation = observe_runtime_repair(
545 "local",
546 record_kind="health",
547 journal_path=tmp_path,
548 )
549 repair_handle = observation["repair_handle"]
550 repaired = repair_corrupt_record(
551 "local",
552 record_kind="health",
553 repair_handle=repair_handle,
554 journal_path=tmp_path,
555 )
556
557 assert observation["status"] == "corrupt"
558 assert repaired == make_synthetic_runtime_health("local")
559 assert read_runtime_health("local", journal_path=tmp_path) == repaired
560 assert repair_handle not in path.read_text(encoding="utf-8")
561
562
563def test_repair_handle_rejects_stale_corrupt_bytes(tmp_path: Path) -> None:
564 path = runtime_retry_token_path("local", journal_path=tmp_path)
565 _write_corrupt(path, b"{first")
566 observation = observe_runtime_repair(
567 "local",
568 record_kind="retry-token",
569 journal_path=tmp_path,
570 )
571
572 path.write_bytes(b"{second")
573
574 with pytest.raises(RuntimeHealthConflictError):
575 repair_corrupt_record(
576 "local",
577 record_kind="retry-token",
578 repair_handle=observation["repair_handle"],
579 journal_path=tmp_path,
580 )
581 assert path.read_bytes() == b"{second"
582
583
584def test_repair_handle_never_appears_in_logs_owner_copy_or_diagnostics(
585 tmp_path: Path,
586 caplog: pytest.LogCaptureFixture,
587) -> None:
588 corrupt_path = runtime_retry_token_path("local", journal_path=tmp_path)
589 _write_corrupt(corrupt_path, b"{private")
590 repair_observation = observe_runtime_repair(
591 "local",
592 record_kind="retry-token",
593 journal_path=tmp_path,
594 )
595 repair_handle = repair_observation["repair_handle"]
596
597 public_returns: list[Any] = [
598 runtime_health_path("parakeet", journal_path=tmp_path),
599 runtime_retry_token_path("parakeet", journal_path=tmp_path),
600 runtime_operation_path("parakeet", journal_path=tmp_path),
601 runtime_operation_lock_path("parakeet", journal_path=tmp_path),
602 make_synthetic_runtime_health("parakeet"),
603 runtime_health.make_synthetic_retry_token("parakeet"),
604 read_runtime_health("parakeet", journal_path=tmp_path),
605 read_retry_token("parakeet", journal_path=tmp_path),
606 inspect_runtime_health("parakeet", journal_path=tmp_path),
607 inspect_retry_token("local", journal_path=tmp_path),
608 ]
609 health = write_runtime_health(
610 _health("parakeet", fingerprint="fp-public"),
611 journal_path=tmp_path,
612 )
613 token = request_retry_token(
614 "parakeet",
615 desired_fingerprint_sha256="fp-public",
616 owner={"actor": "test"},
617 journal_path=tmp_path,
618 )
619 consumed = consume_retry_token(
620 "parakeet",
621 token_id=str(token["token_id"]),
622 revision=token["revision"],
623 desired_fingerprint_sha256="fp-public",
624 journal_path=tmp_path,
625 )
626 repaired = repair_corrupt_record(
627 "local",
628 record_kind="retry-token",
629 repair_handle=repair_handle,
630 journal_path=tmp_path,
631 )
632 public_returns.extend([health, token, consumed, repaired])
633
634 for value in public_returns:
635 assert not _contains_value(value, repair_handle)
636 for record in caplog.records:
637 assert repair_handle not in record.getMessage()
638
639
640def test_scoped_provider_cache_reset_fixture_does_not_invoke_detection(
641 provider_cache_reset,
642 monkeypatch: pytest.MonkeyPatch,
643) -> None:
644 from solstone.think.providers import local_vulkan
645
646 def fail_detection():
647 raise AssertionError("provider cache reset must not detect GPUs")
648
649 monkeypatch.setattr(local_vulkan, "_enumerate_gpus", fail_detection)
650 local_vulkan.reset_detect_cache()