solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for the deterministic activity state machine."""
5
6
7def _sense(
8 content_type="coding",
9 density="active",
10 facets=None,
11 summary="Working on code.",
12 entities=None,
13 meeting=False,
14 speakers=None,
15):
16 """Build a Sense output payload for testing."""
17 if facets is None:
18 facets = [{"facet": "work", "activity": content_type, "level": "high"}]
19 return {
20 "density": density,
21 "content_type": content_type,
22 "activity_summary": summary,
23 "entities": entities or [],
24 "facets": facets,
25 "meeting_detected": meeting,
26 "speakers": speakers or [],
27 "recommend": {},
28 }
29
30
31class TestNewActivity:
32 def test_first_segment_starts_new(self):
33 from think.activity_state_machine import ActivityStateMachine
34
35 sm = ActivityStateMachine()
36 changes = sm.update(_sense(), "090000_300", "20260304")
37
38 assert len(changes) == 1
39 assert changes[0]["_change"] == "new"
40 assert changes[0]["activity"] == "coding"
41 assert changes[0]["state"] == "active"
42 assert changes[0]["since"] == "090000_300"
43 assert changes[0]["_facet"] == "work"
44
45
46class TestContinuation:
47 def test_same_type_continues(self):
48 from think.activity_state_machine import ActivityStateMachine
49
50 sm = ActivityStateMachine()
51 sm.update(_sense(), "090000_300", "20260304")
52 changes = sm.update(_sense(summary="Still coding."), "090500_300", "20260304")
53
54 assert len(changes) == 1
55 assert changes[0]["_change"] == "continuing"
56 assert changes[0]["since"] == "090000_300"
57 assert changes[0]["description"] == "Still coding."
58
59
60class TestContentTypeChange:
61 def test_type_change_ends_old_starts_new(self):
62 from think.activity_state_machine import ActivityStateMachine
63
64 sm = ActivityStateMachine()
65 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
66 changes = sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
67
68 assert len(changes) == 2
69 ended = [c for c in changes if c["state"] == "ended"]
70 started = [c for c in changes if c["state"] == "active"]
71 assert len(ended) == 1
72 assert ended[0]["_change"] == "ended_type_change"
73 assert ended[0]["activity"] == "coding"
74 assert len(started) == 1
75 assert started[0]["_change"] == "new"
76 assert started[0]["activity"] == "meeting"
77
78
79class TestIdleTransition:
80 def test_idle_ends_all(self):
81 from think.activity_state_machine import ActivityStateMachine
82
83 sm = ActivityStateMachine()
84 sm.update(_sense(), "090000_300", "20260304")
85 changes = sm.update(_sense(density="idle"), "090500_300", "20260304")
86
87 ended = [c for c in changes if c["state"] == "ended"]
88 assert len(ended) == 1
89 assert ended[0]["_change"] == "ended_idle"
90 assert sm.get_current_state() == []
91
92
93class TestTimeGap:
94 def test_gap_over_600s_resets(self):
95 from think.activity_state_machine import ActivityStateMachine
96
97 sm = ActivityStateMachine()
98 sm.update(_sense(), "090000_300", "20260304")
99 changes = sm.update(_sense(), "100600_300", "20260304")
100
101 ended_gap = [c for c in changes if c["_change"] == "ended_gap"]
102 assert len(ended_gap) == 1
103 new = [c for c in changes if c["_change"] == "new"]
104 assert len(new) == 1
105
106 def test_gap_equal_600s_no_reset(self):
107 from think.activity_state_machine import ActivityStateMachine
108
109 sm = ActivityStateMachine()
110 sm.update(_sense(), "090000_300", "20260304")
111 changes = sm.update(_sense(), "091500_300", "20260304")
112
113 assert all(c["_change"] != "ended_gap" for c in changes)
114
115
116class TestDayBoundary:
117 def test_day_change_resets(self):
118 from think.activity_state_machine import ActivityStateMachine
119
120 sm = ActivityStateMachine()
121 sm.update(_sense(), "230000_300", "20260304")
122 changes = sm.update(_sense(), "000000_300", "20260305")
123
124 ended_gap = [c for c in changes if c["_change"] == "ended_gap"]
125 assert len(ended_gap) == 1
126
127
128class TestMultiFacet:
129 def test_independent_facet_tracking(self):
130 from think.activity_state_machine import ActivityStateMachine
131
132 sm = ActivityStateMachine()
133 facets = [
134 {"facet": "work", "activity": "coding", "level": "high"},
135 {"facet": "personal", "activity": "browsing", "level": "low"},
136 ]
137 changes = sm.update(_sense(facets=facets), "090000_300", "20260304")
138
139 assert len(changes) == 2
140 facet_names = {c["_facet"] for c in changes}
141 assert facet_names == {"work", "personal"}
142
143
144class TestFacetDisappearing:
145 def test_facet_gone_emits_ended(self):
146 from think.activity_state_machine import ActivityStateMachine
147
148 sm = ActivityStateMachine()
149 two_facets = [
150 {"facet": "work", "activity": "coding", "level": "high"},
151 {"facet": "personal", "activity": "browsing", "level": "low"},
152 ]
153 sm.update(_sense(facets=two_facets), "090000_300", "20260304")
154 one_facet = [{"facet": "work", "activity": "coding", "level": "high"}]
155 changes = sm.update(_sense(facets=one_facet), "090500_300", "20260304")
156
157 ended = [c for c in changes if c["_change"] == "ended_facet_gone"]
158 assert len(ended) == 1
159 assert ended[0]["_facet"] == "personal"
160
161
162class TestGetCurrentState:
163 def test_returns_clean_entries(self):
164 from think.activity_state_machine import ActivityStateMachine
165
166 sm = ActivityStateMachine()
167 sm.update(_sense(), "090000_300", "20260304")
168 state = sm.get_current_state()
169
170 assert len(state) == 1
171 entry = state[0]
172 assert "id" in entry
173 assert "activity" in entry
174 assert "state" in entry and entry["state"] == "active"
175 assert "since" in entry
176 assert "level" in entry
177 assert "active_entities" in entry
178 assert "_change" not in entry
179 assert "_facet" not in entry
180 assert "_segment" not in entry
181
182
183class TestGetCompletedActivities:
184 def test_completed_format(self):
185 from think.activity_state_machine import ActivityStateMachine
186
187 sm = ActivityStateMachine()
188 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
189 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
190 completed = sm.get_completed_activities()
191
192 assert len(completed) == 1
193 rec = completed[0]
194 assert "id" in rec
195 assert "activity" in rec and rec["activity"] == "coding"
196 assert "segments" in rec and isinstance(rec["segments"], list)
197 assert "level_avg" in rec and isinstance(rec["level_avg"], float)
198 assert "description" in rec
199 assert "active_entities" in rec
200 assert "created_at" in rec
201
202
203class TestPseudoFacet:
204 def test_no_facets_uses_underscore(self):
205 from think.activity_state_machine import ActivityStateMachine
206
207 sm = ActivityStateMachine()
208 changes = sm.update(_sense(facets=[]), "090000_300", "20260304")
209
210 assert len(changes) == 1
211 assert changes[0]["_facet"] == "__"
212
213
214class TestEntityTracking:
215 def test_extracts_names(self):
216 from think.activity_state_machine import ActivityStateMachine
217
218 sm = ActivityStateMachine()
219 entities = [
220 {"type": "Person", "name": "Alice", "context": "colleague"},
221 {"type": "Tool", "name": "VSCode", "context": "editor"},
222 ]
223 changes = sm.update(_sense(entities=entities), "090000_300", "20260304")
224
225 assert changes[0]["active_entities"] == ["Alice", "VSCode"]
226
227 def test_skips_blank_names(self):
228 from think.activity_state_machine import ActivityStateMachine
229
230 sm = ActivityStateMachine()
231 entities = [
232 {"type": "Person", "name": "", "context": "unknown"},
233 {"type": "Tool", "context": "no name key"},
234 ]
235 changes = sm.update(_sense(entities=entities), "090000_300", "20260304")
236
237 assert changes[0]["active_entities"] == []