solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from pathlib import Path
5
6import pytest
7
8from solstone.think.day_accumulator import append_record, read_latest, read_records
9
10
11@pytest.fixture
12def journal(tmp_path, monkeypatch):
13 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
14 return tmp_path
15
16
17def _accumulator_path(journal: Path, day: str, name: str) -> Path:
18 return journal / "chronicle" / day / "talents" / f"{name}.jsonl"
19
20
21def test_first_write_creates_file_and_stamps_ts(journal):
22 day = "20260611"
23
24 append_record(day, "pulse", {"summary": "a"})
25
26 path = _accumulator_path(journal, day, "pulse")
27 assert path.is_file()
28 assert len(path.read_text(encoding="utf-8").splitlines()) == 1
29
30 records = read_records(day, "pulse")
31 assert len(records) == 1
32 assert records[0]["summary"] == "a"
33 assert "ts" in records[0]
34
35
36def test_append_record_does_not_overwrite_existing_records(journal):
37 day = "20260611"
38
39 append_record(day, "pulse", {"summary": "a", "ts": 100})
40 append_record(day, "pulse", {"summary": "b", "ts": 200})
41
42 records = read_records(day, "pulse")
43 assert [record["summary"] for record in records] == ["a", "b"]
44 assert [record["ts"] for record in records] == [100, 200]
45
46
47def test_read_latest_uses_highest_ts_and_last_file_position_for_ties(journal):
48 day = "20260611"
49
50 append_record(day, "pulse", {"summary": "a", "ts": 100})
51 append_record(day, "pulse", {"summary": "b", "ts": 300})
52 append_record(day, "pulse", {"summary": "c", "ts": 200})
53
54 assert read_latest(day, "pulse")["summary"] == "b"
55
56 append_record(day, "tie", {"summary": "first", "ts": 100})
57 append_record(day, "tie", {"summary": "second", "ts": 100})
58
59 assert read_latest(day, "tie")["summary"] == "second"
60
61
62def test_read_latest_lookback_window(journal):
63 base_day = "20260611"
64
65 append_record("20260608", "pulse", {"summary": "lookback", "ts": 100})
66
67 assert read_latest(base_day, "pulse", lookback_days=7)["summary"] == "lookback"
68 assert read_latest(base_day, "pulse", lookback_days=2) is None
69
70 append_record("20260601", "far", {"summary": "too far", "ts": 100})
71
72 assert read_latest(base_day, "far", lookback_days=7) is None
73
74
75def test_read_records_returns_ascending_stable_ts_order(journal):
76 day = "20260611"
77
78 append_record(day, "pulse", {"summary": "three", "ts": 300})
79 append_record(day, "pulse", {"summary": "one", "ts": 100})
80 append_record(day, "pulse", {"summary": "two", "ts": 200})
81
82 records = read_records(day, "pulse")
83 assert [record["summary"] for record in records] == ["one", "two", "three"]
84 assert [record["ts"] for record in records] == [100, 200, 300]
85
86
87def test_malformed_lines_are_skipped_and_dropped_on_subsequent_append(journal):
88 day = "20260611"
89
90 append_record(day, "pulse", {"summary": "valid", "ts": 100})
91 path = _accumulator_path(journal, day, "pulse")
92 with path.open("a", encoding="utf-8") as handle:
93 handle.write("not json\n")
94
95 records = read_records(day, "pulse")
96 assert len(records) == 1
97 assert records[0]["summary"] == "valid"
98 assert read_latest(day, "pulse")["summary"] == "valid"
99
100 append_record(day, "pulse", {"summary": "second", "ts": 200})
101
102 records = read_records(day, "pulse")
103 assert len(records) == 2
104 assert "not json" not in path.read_text(encoding="utf-8")
105
106
107def test_missing_day_reads_do_not_create_directories_or_files(journal):
108 day = "20200101"
109 path = _accumulator_path(journal, day, "pulse")
110
111 assert read_records(day, "pulse") == []
112 assert read_latest(day, "pulse") is None
113 assert not path.exists()
114 assert not path.parent.exists()
115
116
117def test_append_record_indexes_under_accumulator_name(journal):
118 day = "20260611"
119
120 append_record(
121 day,
122 "pulse",
123 {
124 "title": "Focus block",
125 "one_sentence": "The owner ran a deep focus block.",
126 "full_details": "The morning centered on a deep focus block.",
127 "needs_you": [],
128 "model": "test-model",
129 "generated_at": "2026-06-11T12:00:00Z",
130 "ts": 100,
131 },
132 )
133
134 from solstone.think.indexer.journal import search_journal
135
136 total, results = search_journal("focus", agent="pulse")
137 assert total >= 1
138 assert any("focus" in hit["text"].lower() for hit in results)
139
140 control_total, control_results = search_journal("focus", agent="steward")
141 assert control_total == 0
142 assert control_results == []