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 json
7import multiprocessing
8import os
9import traceback
10from pathlib import Path
11from queue import Empty
12from typing import Any
13
14import pytest
15
16from solstone.think.facet_review_candidates import (
17 facet_review_candidates_path,
18 load_candidates,
19 save_candidates,
20)
21
22
23def _record_candidate_worker(
24 journal_path: str,
25 barrier: Any,
26 errors: Any,
27 index: int,
28) -> None:
29 os.environ["SOLSTONE_JOURNAL"] = journal_path
30 try:
31 barrier.wait(timeout=5)
32
33 from solstone.think.facet_review_candidates import record_facet_candidate
34
35 record_facet_candidate(
36 name=f"Candidate {index}",
37 name_key=f"candidate {index}",
38 count=index + 1,
39 window_days=14,
40 samples=[
41 {
42 "day": "20260602",
43 "stream": "archon",
44 "segment": f"09000{index}_300",
45 }
46 ],
47 day="20260602",
48 )
49 except BaseException:
50 errors.put(traceback.format_exc())
51 raise
52
53
54def _drain_errors(errors: Any) -> list[str]:
55 found = []
56 while True:
57 try:
58 found.append(errors.get_nowait())
59 except Empty:
60 return found
61
62
63def _join_processes(processes: list[Any], errors: Any) -> None:
64 for process in processes:
65 process.start()
66 for process in processes:
67 process.join(timeout=10)
68 for process in processes:
69 if process.is_alive():
70 process.terminate()
71 process.join(timeout=2)
72
73 error_text = "\n".join(_drain_errors(errors))
74 assert all(not process.is_alive() for process in processes), error_text
75 assert all(process.exitcode == 0 for process in processes), error_text
76
77
78def test_facet_review_candidate_atomic_failure_preserves_existing_bytes(
79 tmp_path: Path,
80 monkeypatch: pytest.MonkeyPatch,
81) -> None:
82 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
83 path = facet_review_candidates_path()
84 seed = {"name": "Café", "name_key": "café", "status": "open"}
85 original = json.dumps(seed, ensure_ascii=False) + "\n"
86 path.write_text(original, encoding="utf-8")
87
88 def fail_replace(_src: str, _dst: str) -> None:
89 raise OSError("simulated replace failure")
90
91 monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace)
92
93 with pytest.raises(OSError):
94 save_candidates([{"name": "Changed", "name_key": "changed"}])
95
96 assert path.read_text(encoding="utf-8") == original
97 assert list(path.parent.glob(".tmp_*")) == []
98
99
100def test_facet_review_candidate_locked_modify_survives_multiprocess_writers(
101 tmp_path: Path,
102 monkeypatch: pytest.MonkeyPatch,
103) -> None:
104 ctx = multiprocessing.get_context("spawn")
105 journal = tmp_path / "journal"
106 barrier = ctx.Barrier(4)
107 errors = ctx.Queue()
108 processes = [
109 ctx.Process(
110 target=_record_candidate_worker,
111 args=(str(journal), barrier, errors, index),
112 )
113 for index in range(4)
114 ]
115
116 _join_processes(processes, errors)
117
118 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
119 rows = load_candidates()
120 assert sorted(row["name_key"] for row in rows) == [
121 "candidate 0",
122 "candidate 1",
123 "candidate 2",
124 "candidate 3",
125 ]