solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Durability tests for ActivityStateMachine snapshots."""
5
6import json
7from pathlib import Path
8
9from solstone.think.activities import (
10 append_activity_record,
11 load_activity_records,
12 make_activity_id,
13)
14from solstone.think.activity_state_machine import ActivityStateMachine
15from solstone.think.journal_io import atomic_replace
16
17
18def _sense(content_type: str = "coding", density: str = "active", facet: str = "test"):
19 return {
20 "density": density,
21 "content_type": content_type,
22 "activity_summary": f"{content_type} work",
23 "entities": [],
24 "facets": [{"facet": facet, "activity": content_type, "level": "high"}],
25 "meeting_detected": content_type == "meeting",
26 "speakers": [],
27 "recommend": {},
28 }
29
30
31def _persist_snapshot(journal_root: Path, state_machine: ActivityStateMachine) -> None:
32 snapshot = {
33 "last_segment_key": state_machine.last_segment_key,
34 "last_segment_day": state_machine.last_segment_day,
35 "active": {
36 facet: {k: v for k, v in entry.items() if k != "_change"}
37 for facet, entry in state_machine.state.items()
38 },
39 }
40 atomic_replace(
41 journal_root / "awareness" / "activity_state.json", json.dumps(snapshot)
42 )
43
44
45def _append_ended_records(
46 state_machine: ActivityStateMachine, changes: list[dict], day: str
47) -> None:
48 completed_lookup = {}
49 for record in state_machine.get_completed_activities():
50 completed_lookup.setdefault(record["id"], record)
51 for change in changes:
52 if change.get("state") != "ended":
53 continue
54 record = completed_lookup.get(change["id"])
55 if record:
56 append_activity_record(change["facet"], day, record)
57
58
59def test_state_survives_subprocess_boundary(tmp_path: Path, monkeypatch):
60 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
61
62 sm1 = ActivityStateMachine(journal_root=tmp_path)
63 sm1.update(_sense(), "090000_300", "20260427")
64 assert set(sm1.state) == {"test"}
65 _persist_snapshot(tmp_path, sm1)
66
67 sm2 = ActivityStateMachine(journal_root=tmp_path)
68 assert sm2.last_segment_key == "090000_300"
69 assert sm2.last_segment_day == "20260427"
70 assert sm2.state["test"]["segments"] == ["090000_300"]
71
72 changes = sm2.update(_sense(density="idle"), "090500_300", "20260427")
73 ended = [change for change in changes if change.get("state") == "ended"]
74 assert len(ended) == 1
75 _append_ended_records(sm2, changes, "20260427")
76 _persist_snapshot(tmp_path, sm2)
77
78 records = load_activity_records("test", "20260427")
79 assert len(records) == 1
80 assert records[0]["id"] == make_activity_id("coding", "090000_300")
81 assert records[0]["segments"] == ["090000_300"]
82
83 sm3 = ActivityStateMachine(journal_root=tmp_path)
84 assert sm3.state == {}
85 assert sm3.last_segment_key == "090500_300"
86 assert sm3.last_segment_day == "20260427"
87
88
89def test_day_boundary_routes_ended_record_to_prior_day(tmp_path: Path, monkeypatch):
90 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
91
92 sm1 = ActivityStateMachine(journal_root=tmp_path)
93 sm1.update(_sense(), "233000_300", "20260304")
94 _persist_snapshot(tmp_path, sm1)
95
96 sm2 = ActivityStateMachine(journal_root=tmp_path)
97 routing_day = sm2.last_segment_day or "20260305"
98 changes = sm2.update(_sense(), "001500_300", "20260305")
99
100 ended = [change for change in changes if change.get("state") == "ended"]
101 active = [change for change in changes if change.get("state") == "active"]
102 assert len(ended) == 1
103 assert len(active) == 1
104 _append_ended_records(sm2, changes, routing_day)
105 _persist_snapshot(tmp_path, sm2)
106
107 prior_day_records = load_activity_records("test", "20260304")
108 current_day_records = load_activity_records("test", "20260305")
109 assert len(prior_day_records) == 1
110 assert current_day_records == []
111 assert prior_day_records[0]["segments"] == ["233000_300"]
112 assert sm2.state["test"]["since"] == "001500_300"
113 assert sm2.last_segment_day == "20260305"
114
115
116def test_three_active_segments_then_idle_writes_one_record(tmp_path: Path, monkeypatch):
117 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
118 day = "20260427"
119
120 for segment in ("090000_300", "090505_300", "091010_300"):
121 state_machine = ActivityStateMachine(journal_root=tmp_path)
122 state_machine.update(_sense(facet="work"), segment, day)
123 _persist_snapshot(tmp_path, state_machine)
124
125 state_machine = ActivityStateMachine(journal_root=tmp_path)
126 routing_day = state_machine.last_segment_day or day
127 changes = state_machine.update(
128 _sense(density="idle", facet="work"), "091515_300", day
129 )
130 ended = [change for change in changes if change.get("state") == "ended"]
131 assert len(ended) == 1
132 _append_ended_records(state_machine, changes, routing_day)
133 _persist_snapshot(tmp_path, state_machine)
134
135 records = load_activity_records("work", day)
136 assert len(records) == 1
137 assert records[0]["segments"] == ["090000_300", "090505_300", "091010_300"]
138 assert len(records[0]["segments"]) == 3
139 assert state_machine.state == {}
140
141
142def test_crash_between_append_and_snapshot_is_idempotent(tmp_path: Path, monkeypatch):
143 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
144 day = "20260427"
145
146 sm1 = ActivityStateMachine(journal_root=tmp_path)
147 sm1.update(_sense(facet="work"), "090000_300", day)
148 _persist_snapshot(tmp_path, sm1)
149
150 sm2 = ActivityStateMachine(journal_root=tmp_path)
151 routing_day = sm2.last_segment_day or day
152 changes = sm2.update(_sense(density="idle", facet="work"), "090505_300", day)
153 _append_ended_records(sm2, changes, routing_day)
154 assert len(load_activity_records("work", day)) == 1
155
156 sm3 = ActivityStateMachine(journal_root=tmp_path)
157 routing_day = sm3.last_segment_day or day
158 retry_changes = sm3.update(_sense(density="idle", facet="work"), "090505_300", day)
159 _append_ended_records(sm3, retry_changes, routing_day)
160 records = load_activity_records("work", day)
161 assert len(records) == 1
162 assert records[0]["id"] == make_activity_id("coding", "090000_300")
163
164 _persist_snapshot(tmp_path, sm3)
165 sm4 = ActivityStateMachine(journal_root=tmp_path)
166 assert sm4.state == {}
167 assert sm4.last_segment_key == "090505_300"
168 assert sm4.last_segment_day == day
169
170
171def test_batch_construction_has_no_journal_root_and_skips_snapshot(
172 tmp_path: Path, monkeypatch
173):
174 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
175 marker = {
176 "last_segment_key": "marker",
177 "last_segment_day": "20990101",
178 "active": {},
179 }
180 state_path = tmp_path / "awareness" / "activity_state.json"
181 atomic_replace(state_path, json.dumps(marker))
182 mtime_before = state_path.stat().st_mtime_ns
183
184 state_machine = ActivityStateMachine()
185 assert state_machine.journal_root is None
186 state_machine.update(_sense(facet="work"), "090000_300", "20260427")
187 state_machine.update(_sense(facet="work"), "090505_300", "20260427")
188 state_machine.update(_sense(density="idle", facet="work"), "091010_300", "20260427")
189
190 if state_machine.journal_root is not None:
191 _persist_snapshot(state_machine.journal_root, state_machine)
192
193 assert state_path.stat().st_mtime_ns == mtime_before
194 assert json.loads(state_path.read_text(encoding="utf-8")) == marker
195
196
197def test_run_segment_sense_emits_activity_events(tmp_path: Path, monkeypatch):
198 from solstone.think import thinking as think
199
200 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
201 day = "20260427"
202 stream = "default"
203 segment = "090505_300"
204
205 sm1 = ActivityStateMachine(journal_root=tmp_path)
206 sm1.update(_sense(facet="work"), "090000_300", day)
207 _persist_snapshot(tmp_path, sm1)
208
209 talents_dir = tmp_path / "chronicle" / day / stream / segment / "talents"
210 talents_dir.mkdir(parents=True)
211 (talents_dir / "sense.json").write_text(
212 json.dumps(_sense(density="idle", facet="work")),
213 encoding="utf-8",
214 )
215
216 events = []
217 monkeypatch.setattr(
218 think,
219 "get_talent_configs",
220 lambda schedule=None, **kwargs: {
221 "sense": {"priority": 10, "type": "generate", "output": "json"}
222 },
223 )
224 monkeypatch.setattr(think, "_dispatch_cortex_request", lambda **kwargs: "sense-1")
225 monkeypatch.setattr(
226 think, "_drain_priority_batch", lambda *args, **kwargs: (1, 0, [])
227 )
228 monkeypatch.setattr(
229 think, "_jsonl_log", lambda event, **fields: events.append((event, fields))
230 )
231 monkeypatch.setattr(think, "run_activity_prompts", lambda **kwargs: True)
232 monkeypatch.setattr(think, "_callosum", None)
233
234 success, failed, failed_names = think.run_segment_sense(
235 day,
236 segment,
237 refresh=False,
238 verbose=False,
239 stream=stream,
240 state_machine=ActivityStateMachine(journal_root=tmp_path),
241 )
242
243 assert (success, failed, failed_names) == (1, 0, [])
244 event_names = [event for event, _fields in events]
245 assert "activity.detected" in event_names
246 assert "activity.persisted" in event_names
247 detected = [fields for event, fields in events if event == "activity.detected"]
248 persisted = [fields for event, fields in events if event == "activity.persisted"]
249 assert detected[0]["change"] == "ended_idle"
250 assert persisted[0]["change"] == "ended_idle"