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 facet_activity="Reviewing the open diff.",
13 entities=None,
14 meeting=False,
15 speakers=None,
16):
17 """Build a Sense output payload for testing."""
18 if facets is None:
19 facets = [{"facet": "work", "activity": facet_activity, "level": "high"}]
20 return {
21 "density": density,
22 "content_type": content_type,
23 "activity_summary": summary,
24 "entities": entities or [],
25 "facets": facets,
26 "meeting_detected": meeting,
27 "speakers": speakers or [],
28 "recommend": {},
29 }
30
31
32class TestNewActivity:
33 def test_first_segment_starts_new(self):
34 from solstone.think.activity_state_machine import ActivityStateMachine
35
36 sm = ActivityStateMachine()
37 changes = sm.update(_sense(), "090000_300", "20260304")
38
39 assert len(changes) == 1
40 assert changes[0]["_change"] == "new"
41 assert changes[0]["activity"] == "coding"
42 assert changes[0]["state"] == "active"
43 assert changes[0]["since"] == "090000_300"
44 assert changes[0]["facet"] == "work"
45
46
47class TestContinuation:
48 def test_same_type_continues(self):
49 from solstone.think.activity_state_machine import ActivityStateMachine
50
51 sm = ActivityStateMachine()
52 sm.update(_sense(), "090000_300", "20260304")
53 changes = sm.update(
54 _sense(facet_activity="Still reviewing the diff."),
55 "090500_300",
56 "20260304",
57 )
58
59 assert len(changes) == 1
60 assert changes[0]["_change"] == "continuing"
61 assert changes[0]["since"] == "090000_300"
62 assert changes[0]["description"] == "Still reviewing the diff."
63
64
65class TestContentTypeChange:
66 def test_type_change_ends_old_starts_new(self):
67 from solstone.think.activity_state_machine import ActivityStateMachine
68
69 sm = ActivityStateMachine()
70 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
71 pending = sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
72 changes = sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
73
74 assert all(c["state"] != "ended" for c in pending)
75 assert pending[0]["_change"] == "type_change_pending"
76 assert len(changes) == 2
77 ended = [c for c in changes if c["state"] == "ended"]
78 started = [c for c in changes if c["state"] == "active"]
79 assert len(ended) == 1
80 assert ended[0]["_change"] == "ended_type_change"
81 assert ended[0]["activity"] == "coding"
82 assert len(started) == 1
83 assert started[0]["_change"] == "new"
84 assert started[0]["activity"] == "meeting"
85
86
87class TestIdleTransition:
88 def test_idle_ends_all(self):
89 from solstone.think.activity_state_machine import ActivityStateMachine
90
91 sm = ActivityStateMachine()
92 sm.update(_sense(), "090000_300", "20260304")
93 changes = sm.update(_sense(density="idle"), "090500_300", "20260304")
94
95 ended = [c for c in changes if c["state"] == "ended"]
96 assert len(ended) == 1
97 assert ended[0]["_change"] == "ended_idle"
98 assert sm.state == {}
99
100
101class TestTimeGap:
102 def test_gap_over_600s_resets(self):
103 from solstone.think.activity_state_machine import ActivityStateMachine
104
105 sm = ActivityStateMachine()
106 sm.update(_sense(), "090000_300", "20260304")
107 changes = sm.update(_sense(), "100600_300", "20260304")
108
109 ended_gap = [c for c in changes if c["_change"] == "ended_gap"]
110 assert len(ended_gap) == 1
111 new = [c for c in changes if c["_change"] == "new"]
112 assert len(new) == 1
113
114 def test_gap_equal_600s_no_reset(self):
115 from solstone.think.activity_state_machine import ActivityStateMachine
116
117 sm = ActivityStateMachine()
118 sm.update(_sense(), "090000_300", "20260304")
119 changes = sm.update(_sense(), "091500_300", "20260304")
120
121 assert all(c["_change"] != "ended_gap" for c in changes)
122
123
124class TestDayBoundary:
125 def test_day_change_resets(self):
126 from solstone.think.activity_state_machine import ActivityStateMachine
127
128 sm = ActivityStateMachine()
129 sm.update(_sense(), "230000_300", "20260304")
130 changes = sm.update(_sense(), "000000_300", "20260305")
131
132 ended_gap = [c for c in changes if c["_change"] == "ended_gap"]
133 assert len(ended_gap) == 1
134
135
136class TestMultiFacet:
137 def test_independent_facet_tracking(self):
138 from solstone.think.activity_state_machine import ActivityStateMachine
139
140 sm = ActivityStateMachine()
141 facets = [
142 {"facet": "work", "activity": "coding", "level": "high"},
143 {"facet": "personal", "activity": "browsing", "level": "medium"},
144 ]
145 changes = sm.update(_sense(facets=facets), "090000_300", "20260304")
146
147 assert len(changes) == 2
148 facet_names = {c["facet"] for c in changes}
149 assert facet_names == {"work", "personal"}
150
151
152class TestLevelGate:
153 def test_explicit_low_facets_are_excluded_from_active_set(self):
154 from solstone.think.activity_state_machine import ActivityStateMachine
155
156 sm = ActivityStateMachine()
157 facets = [
158 {"facet": "solpbc", "activity": "Planning launch work.", "level": "high"},
159 {
160 "facet": "personal",
161 "activity": "Coordinating a personal errand.",
162 "level": "medium",
163 },
164 {
165 "facet": "bluesky",
166 "activity": "Checking a social thread.",
167 "level": "low",
168 },
169 {
170 "facet": "kognova",
171 "activity": "Skimming background notes.",
172 "level": "low",
173 },
174 ]
175
176 changes = sm.update(
177 _sense(facets=facets, summary="Mixed activity summary."),
178 "090000_300",
179 "20260304",
180 )
181
182 assert {change["facet"] for change in changes} == {"solpbc", "personal"}
183 assert set(sm.state) == {"solpbc", "personal"}
184 assert not any(
185 change.get("facet") in {"bluesky", "kognova"} for change in changes
186 )
187
188 def test_description_uses_each_engaged_facets_activity(self):
189 from solstone.think.activity_state_machine import ActivityStateMachine
190
191 sm = ActivityStateMachine()
192 summary = "Segment-level summary for multiple domains."
193 facets = [
194 {
195 "facet": "solpbc",
196 "activity": "Reviewing release blockers.",
197 "level": "high",
198 },
199 {
200 "facet": "personal",
201 "activity": "Sorting out afternoon plans.",
202 "level": "medium",
203 },
204 ]
205
206 changes = sm.update(
207 _sense(facets=facets, summary=summary),
208 "090000_300",
209 "20260304",
210 )
211
212 by_facet = {change["facet"]: change for change in changes}
213 assert by_facet["solpbc"]["description"] == "Reviewing release blockers."
214 assert by_facet["personal"]["description"] == "Sorting out afternoon plans."
215 assert all(change["description"] != summary for change in changes)
216
217 def test_description_falls_back_to_summary_when_facet_activity_missing(self):
218 from solstone.think.activity_state_machine import ActivityStateMachine
219
220 sm = ActivityStateMachine()
221 summary = "Segment summary fills missing facet activity."
222 facets = [
223 {"facet": "work", "activity": "", "level": "high"},
224 {"facet": "personal", "level": "medium"},
225 ]
226
227 changes = sm.update(
228 _sense(facets=facets, summary=summary),
229 "090000_300",
230 "20260304",
231 )
232
233 assert {change["facet"] for change in changes} == {"work", "personal"}
234 assert all(change["description"] == summary for change in changes)
235 assert all(change["description"] for change in changes)
236
237 def test_one_low_segment_does_not_end_prior_high_facet(self):
238 from solstone.think.activity_state_machine import ActivityStateMachine
239
240 sm = ActivityStateMachine()
241 high = [{"facet": "work", "activity": "Writing the parser.", "level": "high"}]
242 low = [{"facet": "work", "activity": "Barely checking parser.", "level": "low"}]
243
244 sm.update(_sense(facets=high), "090000_300", "20260304")
245 pending = sm.update(_sense(facets=low), "090500_300", "20260304")
246
247 assert len(pending) == 1
248 assert pending[0]["_change"] == "facet_gone_pending"
249 assert pending[0]["facet"] == "work"
250 assert sm.state["work"]["_pending_facet_misses"] == 1
251 assert not any(
252 change.get("_change") == "ended_facet_gone" for change in pending
253 )
254
255 returned = sm.update(_sense(facets=high), "091000_300", "20260304")
256
257 assert len(returned) == 1
258 assert returned[0]["_change"] == "continuing"
259 assert returned[0]["since"] == "090000_300"
260 assert sm.state["work"]["_pending_facet_misses"] == 0
261 assert sm.get_completed_activities() == []
262
263 def test_two_low_segments_end_prior_high_facet(self):
264 from solstone.think.activity_state_machine import ActivityStateMachine
265
266 sm = ActivityStateMachine()
267 high = [{"facet": "work", "activity": "Writing the parser.", "level": "high"}]
268 low = [{"facet": "work", "activity": "Barely checking parser.", "level": "low"}]
269
270 sm.update(_sense(facets=high), "090000_300", "20260304")
271 sm.update(_sense(facets=low), "090500_300", "20260304")
272 changes = sm.update(_sense(facets=low), "091000_300", "20260304")
273
274 ended = [
275 change for change in changes if change.get("_change") == "ended_facet_gone"
276 ]
277 assert len(ended) == 1
278 assert ended[0]["state"] == "ended"
279 assert ended[0]["facet"] == "work"
280 assert sm.state == {}
281 assert len(sm.get_completed_activities()) == 1
282
283 def test_all_low_fresh_segment_creates_no_activity(self):
284 from solstone.think.activity_state_machine import ActivityStateMachine
285
286 sm = ActivityStateMachine()
287 facets = [
288 {"facet": "work", "activity": "Light inbox scan.", "level": "low"},
289 {"facet": "personal", "activity": "Background reading.", "level": "low"},
290 ]
291
292 changes = sm.update(
293 _sense(facets=facets, summary="Low-attention activity."),
294 "090000_300",
295 "20260304",
296 )
297
298 assert changes == []
299 assert sm.state == {}
300 assert sm.get_completed_activities() == []
301
302 def test_all_low_segments_wind_down_prior_active_facet(self):
303 from solstone.think.activity_state_machine import ActivityStateMachine
304
305 sm = ActivityStateMachine()
306 high = [
307 {"facet": "work", "activity": "Implementing the parser.", "level": "high"}
308 ]
309 all_low = [
310 {"facet": "work", "activity": "Glancing at parser logs.", "level": "low"},
311 {
312 "facet": "personal",
313 "activity": "Skimming personal email.",
314 "level": "low",
315 },
316 ]
317
318 sm.update(_sense(facets=high), "090000_300", "20260304")
319 pending = sm.update(_sense(facets=all_low), "090500_300", "20260304")
320 ended_changes = sm.update(_sense(facets=all_low), "091000_300", "20260304")
321
322 assert len(pending) == 1
323 assert pending[0]["_change"] == "facet_gone_pending"
324 assert pending[0]["_pending_facet_misses"] == 1
325 ended = [
326 change
327 for change in ended_changes
328 if change.get("_change") == "ended_facet_gone"
329 ]
330 assert len(ended) == 1
331 assert ended[0]["facet"] == "work"
332 assert sm.state == {}
333
334
335class TestFacetDisappearing:
336 def test_facet_gone_emits_ended(self):
337 from solstone.think.activity_state_machine import ActivityStateMachine
338
339 sm = ActivityStateMachine()
340 two_facets = [
341 {"facet": "work", "activity": "coding", "level": "high"},
342 {"facet": "personal", "activity": "browsing", "level": "medium"},
343 ]
344 sm.update(_sense(facets=two_facets), "090000_300", "20260304")
345 one_facet = [{"facet": "work", "activity": "coding", "level": "high"}]
346 pending = sm.update(_sense(facets=one_facet), "090500_300", "20260304")
347 changes = sm.update(_sense(facets=one_facet), "091000_300", "20260304")
348
349 assert pending[0]["_change"] == "facet_gone_pending"
350 ended = [c for c in changes if c["_change"] == "ended_facet_gone"]
351 assert len(ended) == 1
352 assert ended[0]["facet"] == "personal"
353
354
355class TestStateShape:
356 def test_active_entries_use_persisted_field_names(self):
357 from solstone.think.activity_state_machine import ActivityStateMachine
358
359 sm = ActivityStateMachine()
360 sm.update(_sense(), "090000_300", "20260304")
361
362 assert set(sm.state) == {"work"}
363 entry = sm.state["work"]
364 assert "id" in entry
365 assert "activity" in entry
366 assert "state" in entry and entry["state"] == "active"
367 assert "since" in entry
368 assert "level" in entry
369 assert "active_entities" in entry
370 assert entry["facet"] == "work"
371 assert entry["segment"] == "090000_300"
372 assert entry["segments"] == ["090000_300"]
373 assert entry["_change"] == "new"
374 assert "_facet" not in entry
375 assert "_segment" not in entry
376 assert "_segments" not in entry
377
378
379class TestGetCompletedActivities:
380 def test_completed_format(self):
381 from solstone.think.activity_state_machine import ActivityStateMachine
382
383 sm = ActivityStateMachine()
384 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
385 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
386 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
387 completed = sm.get_completed_activities()
388
389 assert len(completed) == 1
390 rec = completed[0]
391 assert "id" in rec
392 assert "activity" in rec and rec["activity"] == "coding"
393 assert "segments" in rec and isinstance(rec["segments"], list)
394 assert rec["segments"] == ["090000_300", "090500_300"]
395 assert "level_avg" in rec and isinstance(rec["level_avg"], float)
396 assert "description" in rec
397 assert "active_entities" in rec
398 assert "created_at" in rec
399 assert isinstance(rec["created_at"], int)
400
401
402class TestSegmentAccumulation:
403 def test_continuing_accumulates_segments(self):
404 from solstone.think.activity_state_machine import ActivityStateMachine
405
406 sm = ActivityStateMachine()
407 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
408 sm.update(_sense(content_type="coding"), "090500_300", "20260304")
409 sm.update(_sense(content_type="coding"), "091000_300", "20260304")
410 # End by type change
411 sm.update(_sense(content_type="meeting"), "091500_300", "20260304")
412 sm.update(_sense(content_type="meeting"), "092000_300", "20260304")
413
414 completed = sm.get_completed_activities()
415 assert len(completed) == 1
416 rec = completed[0]
417 assert rec["segments"] == [
418 "090000_300",
419 "090500_300",
420 "091000_300",
421 "091500_300",
422 ]
423
424 def test_ten_segments_produces_ten_keys(self):
425 from solstone.think.activity_state_machine import ActivityStateMachine
426
427 sm = ActivityStateMachine()
428 for i in range(10):
429 minutes = i * 5
430 seg = f"09{minutes:02d}00_300"
431 sm.update(_sense(content_type="coding"), seg, "20260304")
432 # End with idle
433 sm.update(_sense(density="idle"), "095000_300", "20260304")
434
435 completed = sm.get_completed_activities()
436 assert len(completed) == 1
437 assert len(completed[0]["segments"]) == 10
438
439 def test_segments_accumulate_in_state(self):
440 from solstone.think.activity_state_machine import ActivityStateMachine
441
442 sm = ActivityStateMachine()
443 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
444 sm.update(_sense(content_type="coding"), "090500_300", "20260304")
445
446 assert sm.state["work"]["segments"] == ["090000_300", "090500_300"]
447
448
449class TestEntityTracking:
450 def test_extracts_names(self):
451 from solstone.think.activity_state_machine import ActivityStateMachine
452
453 sm = ActivityStateMachine()
454 entities = [
455 {"type": "Person", "name": "Alice", "context": "colleague"},
456 {"type": "Tool", "name": "VSCode", "context": "editor"},
457 ]
458 changes = sm.update(_sense(entities=entities), "090000_300", "20260304")
459
460 assert changes[0]["active_entities"] == ["Alice", "VSCode"]
461
462 def test_skips_blank_names(self):
463 from solstone.think.activity_state_machine import ActivityStateMachine
464
465 sm = ActivityStateMachine()
466 entities = [
467 {"type": "Person", "name": "", "context": "unknown"},
468 {"type": "Tool", "context": "no name key"},
469 ]
470 changes = sm.update(_sense(entities=entities), "090000_300", "20260304")
471
472 assert changes[0]["active_entities"] == []
473
474
475class TestSegmentAccumulationEdgeCases:
476 """Regression tests for segment accumulation across all ending paths."""
477
478 def test_idle_ending_preserves_accumulated_segments(self):
479 """Idle transition after multi-segment activity must carry all segments."""
480 from solstone.think.activity_state_machine import ActivityStateMachine
481
482 sm = ActivityStateMachine()
483 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
484 sm.update(_sense(content_type="coding"), "090500_300", "20260304")
485 sm.update(_sense(content_type="coding"), "091000_300", "20260304")
486 # End via idle
487 sm.update(_sense(density="idle"), "091500_300", "20260304")
488
489 completed = sm.get_completed_activities()
490 assert len(completed) == 1
491 assert completed[0]["segments"] == [
492 "090000_300",
493 "090500_300",
494 "091000_300",
495 ]
496
497 def test_facet_gone_preserves_accumulated_segments(self):
498 """Facet disappearing after multi-segment activity must carry all segments.
499
500 Note: the state machine uses the top-level content_type for all facets,
501 not the per-facet activity field. So both facets have activity=="coding".
502 """
503 from solstone.think.activity_state_machine import ActivityStateMachine
504
505 sm = ActivityStateMachine()
506 two = [
507 {"facet": "work", "activity": "coding", "level": "high"},
508 {"facet": "personal", "activity": "browsing", "level": "medium"},
509 ]
510 one = [{"facet": "work", "activity": "coding", "level": "high"}]
511 sm.update(_sense(facets=two), "090000_300", "20260304")
512 sm.update(_sense(facets=two), "090500_300", "20260304")
513 sm.update(_sense(facets=two), "091000_300", "20260304")
514 # personal disappears
515 sm.update(_sense(facets=one), "091500_300", "20260304")
516 sm.update(_sense(facets=one), "092000_300", "20260304")
517
518 completed = sm.get_completed_activities()
519 assert len(completed) == 1
520 # Activity type comes from top-level content_type, not per-facet
521 assert completed[0]["activity"] == "coding"
522 assert completed[0]["segments"] == [
523 "090000_300",
524 "090500_300",
525 "091000_300",
526 "091500_300",
527 ]
528
529 def test_gap_ending_preserves_accumulated_segments(self):
530 """Time gap after multi-segment activity must carry all segments."""
531 from solstone.think.activity_state_machine import ActivityStateMachine
532
533 sm = ActivityStateMachine()
534 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
535 sm.update(_sense(content_type="coding"), "090500_300", "20260304")
536 # 20-minute gap triggers reset (> 600s threshold)
537 sm.update(_sense(content_type="coding"), "093000_300", "20260304")
538
539 completed = sm.get_completed_activities()
540 assert len(completed) == 1
541 assert completed[0]["segments"] == ["090000_300", "090500_300"]
542
543 def test_type_change_preserves_accumulated_segments(self):
544 """Type change after multi-segment activity must carry all segments."""
545 from solstone.think.activity_state_machine import ActivityStateMachine
546
547 sm = ActivityStateMachine()
548 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
549 sm.update(_sense(content_type="coding"), "090500_300", "20260304")
550 sm.update(_sense(content_type="coding"), "091000_300", "20260304")
551 sm.update(_sense(content_type="coding"), "091500_300", "20260304")
552 sm.update(_sense(content_type="meeting"), "092000_300", "20260304")
553 sm.update(_sense(content_type="meeting"), "092500_300", "20260304")
554
555 completed = sm.get_completed_activities()
556 assert len(completed) == 1
557 assert completed[0]["segments"] == [
558 "090000_300",
559 "090500_300",
560 "091000_300",
561 "091500_300",
562 "092000_300",
563 ]
564
565 def test_single_segment_activity_has_one_segment(self):
566 """Activity lasting exactly one segment has segments: [that_segment]."""
567 from solstone.think.activity_state_machine import ActivityStateMachine
568
569 sm = ActivityStateMachine()
570 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
571 sm.update(_sense(density="idle"), "090500_300", "20260304")
572
573 completed = sm.get_completed_activities()
574 assert len(completed) == 1
575 assert completed[0]["segments"] == ["090000_300"]
576
577 def test_duplicate_segment_key_not_appended(self):
578 """Same segment key processed twice doesn't duplicate in _segments."""
579 from solstone.think.activity_state_machine import ActivityStateMachine
580
581 sm = ActivityStateMachine()
582 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
583 # Same segment key again (shouldn't happen in practice, but defensive)
584 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
585 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
586 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
587
588 completed = sm.get_completed_activities()
589 assert len(completed) == 1
590 assert completed[0]["segments"] == ["090000_300", "090500_300"]
591
592 def test_multi_facet_simultaneous_ending_all_have_segments(self):
593 """Multiple facets ending simultaneously each have their own segments."""
594 from solstone.think.activity_state_machine import ActivityStateMachine
595
596 sm = ActivityStateMachine()
597 two = [
598 {"facet": "work", "activity": "coding", "level": "high"},
599 {"facet": "personal", "activity": "browsing", "level": "medium"},
600 ]
601 sm.update(_sense(facets=two), "090000_300", "20260304")
602 sm.update(_sense(facets=two), "090500_300", "20260304")
603 sm.update(_sense(facets=two), "091000_300", "20260304")
604 # Both end via idle
605 sm.update(_sense(density="idle"), "091500_300", "20260304")
606
607 completed = sm.get_completed_activities()
608 assert len(completed) == 2
609 for rec in completed:
610 assert rec["segments"] == [
611 "090000_300",
612 "090500_300",
613 "091000_300",
614 ]
615
616
617class TestCompletedRecordFields:
618 """Verify all fields in completed records have correct types and values."""
619
620 def test_created_at_is_millisecond_int(self):
621 """created_at must be an integer in milliseconds (not ISO string)."""
622 import time
623
624 from solstone.think.activity_state_machine import ActivityStateMachine
625
626 before = int(time.time() * 1000)
627 sm = ActivityStateMachine()
628 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
629 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
630 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
631 after = int(time.time() * 1000)
632
633 rec = sm.get_completed_activities()[0]
634 assert isinstance(rec["created_at"], int)
635 assert before <= rec["created_at"] <= after
636
637 def test_created_at_comparable_to_float_cutoff(self):
638 """created_at must be comparable with float (routes.py cutoff_ts)."""
639 from datetime import datetime, timedelta
640
641 from solstone.think.activity_state_machine import ActivityStateMachine
642
643 sm = ActivityStateMachine()
644 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
645 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
646 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
647
648 rec = sm.get_completed_activities()[0]
649 # Simulate routes.py cutoff calculation
650 cutoff_ts = (datetime.now() - timedelta(hours=4)).timestamp() * 1000
651 # This must not raise TypeError
652 assert isinstance(rec["created_at"] < cutoff_ts, bool)
653 # Also must support division (routes.py line 283)
654 dt = datetime.fromtimestamp(rec["created_at"] / 1000)
655 assert isinstance(dt, datetime)
656
657 def test_level_avg_is_float(self):
658 from solstone.think.activity_state_machine import ActivityStateMachine
659
660 sm = ActivityStateMachine()
661 sm.update(
662 _sense(
663 content_type="coding",
664 facets=[{"facet": "work", "activity": "coding", "level": "high"}],
665 ),
666 "090000_300",
667 "20260304",
668 )
669 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
670 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
671
672 rec = sm.get_completed_activities()[0]
673 assert isinstance(rec["level_avg"], float)
674 assert rec["level_avg"] == 1.0 # "high" maps to 1.0
675
676 def test_all_required_fields_present(self):
677 """Completed record must have every field run_activity_prompts expects."""
678 from solstone.think.activity_state_machine import ActivityStateMachine
679
680 sm = ActivityStateMachine()
681 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
682 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
683 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
684
685 rec = sm.get_completed_activities()[0]
686 required = {
687 "id",
688 "facet",
689 "activity",
690 "segments",
691 "level_avg",
692 "description",
693 "active_entities",
694 "created_at",
695 }
696 assert required.issubset(rec.keys())
697 # No internal _fields should leak
698 assert not any(k.startswith("_") for k in rec.keys())
699
700 def test_active_entities_preserved_in_completed(self):
701 from solstone.think.activity_state_machine import ActivityStateMachine
702
703 sm = ActivityStateMachine()
704 entities = [
705 {"type": "Person", "name": "Alice", "context": "dev"},
706 {"type": "Tool", "name": "VSCode", "context": "editor"},
707 ]
708 sm.update(
709 _sense(content_type="coding", entities=entities), "090000_300", "20260304"
710 )
711 sm.update(_sense(density="idle"), "090500_300", "20260304")
712
713 rec = sm.get_completed_activities()[0]
714 assert rec["active_entities"] == ["Alice", "VSCode"]
715
716
717class TestCumulativeCompletedList:
718 """Verify get_completed_activities() is cumulative and correct."""
719
720 def test_multiple_endings_accumulate(self):
721 from solstone.think.activity_state_machine import ActivityStateMachine
722
723 sm = ActivityStateMachine()
724 # Activity 1: coding
725 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
726 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
727 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
728 # Activity 2: meeting
729 sm.update(_sense(content_type="coding"), "091500_300", "20260304")
730 sm.update(_sense(content_type="coding"), "092000_300", "20260304")
731
732 completed = sm.get_completed_activities()
733 assert len(completed) == 2
734 assert completed[0]["activity"] == "coding"
735 assert completed[1]["activity"] == "meeting"
736
737 def test_completed_list_is_copy(self):
738 """get_completed_activities returns a copy, not internal state."""
739 from solstone.think.activity_state_machine import ActivityStateMachine
740
741 sm = ActivityStateMachine()
742 sm.update(_sense(content_type="coding"), "090000_300", "20260304")
743 sm.update(_sense(content_type="meeting"), "090500_300", "20260304")
744 sm.update(_sense(content_type="meeting"), "091000_300", "20260304")
745
746 list1 = sm.get_completed_activities()
747 list2 = sm.get_completed_activities()
748 assert list1 is not list2
749 assert list1 == list2