solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for the activities module and activities agent hooks."""
5
6import json
7import logging
8import tempfile
9from pathlib import Path
10
11import pytest
12
13
14def test_get_default_activities():
15 """Test that default activities are returned correctly."""
16 from solstone.think.activities import get_default_activities
17
18 defaults = get_default_activities()
19
20 # Should return a list
21 assert isinstance(defaults, list)
22 assert len(defaults) > 0
23
24 # Each activity should have required fields
25 for activity in defaults:
26 assert "id" in activity
27 assert "name" in activity
28 assert "description" in activity
29 assert "emoji" in activity
30 assert "icon" in activity
31
32 # Check some known activities exist
33 ids = [a["id"] for a in defaults]
34 assert "meeting" in ids
35 assert "coding" in ids
36 assert "browsing" in ids
37
38 # All defaults should have instructions
39 for activity in defaults:
40 assert "instructions" in activity, f"{activity['id']} missing instructions"
41 assert isinstance(activity["instructions"], str)
42 assert len(activity["instructions"]) > 0
43
44
45def test_default_activities_have_canonical_icon_shape():
46 """Default activities keep emoji fallback and explicit Lucide icon name."""
47 from solstone.think.activities import DEFAULT_ACTIVITIES, LUCIDE_ICON_NAME_RE
48
49 assert len(DEFAULT_ACTIVITIES) == 25
50 for activity in DEFAULT_ACTIVITIES:
51 assert activity["emoji"]
52 assert LUCIDE_ICON_NAME_RE.fullmatch(activity["icon"])
53
54
55def test_get_default_activities_returns_copy():
56 """Test that get_default_activities returns a copy, not the original."""
57 from solstone.think.activities import get_default_activities
58
59 defaults1 = get_default_activities()
60 defaults2 = get_default_activities()
61
62 # Should be equal but not the same object
63 assert defaults1 == defaults2
64 assert defaults1 is not defaults2
65
66 # Modifying one should not affect the other
67 defaults1[0]["id"] = "modified"
68 assert defaults2[0]["id"] != "modified"
69
70
71def test_fallback_activity_title_uses_folded_untitled_activity():
72 from solstone.think.activities import _fallback_activity_title
73
74 assert _fallback_activity_title({}) == "untitled activity"
75
76
77def test_always_on_activities(monkeypatch):
78 """Test that always-on activities are auto-included for all facets."""
79 from solstone.think.activities import DEFAULT_ACTIVITIES, get_facet_activities
80
81 always_on = [a for a in DEFAULT_ACTIVITIES if a.get("always_on")]
82 assert len(always_on) >= 2 # messaging and email
83
84 always_on_ids = {a["id"] for a in always_on}
85 assert "messaging" in always_on_ids
86 assert "email" in always_on_ids
87
88 with tempfile.TemporaryDirectory() as tmpdir:
89 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
90
91 facet_path = Path(tmpdir) / "facets" / "test_facet"
92 facet_path.mkdir(parents=True)
93
94 # Empty facet should still have always-on activities
95 activities = get_facet_activities("test_facet")
96 activity_ids = {a["id"] for a in activities}
97 assert always_on_ids <= activity_ids
98
99 # Explicitly attaching one should not duplicate it
100 from solstone.think.activities import add_activity_to_facet
101
102 add_activity_to_facet("test_facet", "messaging")
103 activities = get_facet_activities("test_facet")
104 messaging_count = sum(1 for a in activities if a["id"] == "messaging")
105 assert messaging_count == 1
106
107
108def test_generate_activity_id():
109 """Test activity ID generation from names."""
110 from solstone.think.activities import generate_activity_id
111
112 assert generate_activity_id("My Activity") == "my_activity"
113 assert generate_activity_id("Research & Development") == "research_development"
114 assert generate_activity_id(" Spaces ") == "spaces"
115 assert generate_activity_id("123-Numbers!") == "123_numbers"
116 assert generate_activity_id("") == "activity"
117
118
119def test_facet_activities_empty():
120 """Test loading activities from a facet with no activities file.
121
122 With no activities.jsonl, all defaults are returned as the vocabulary.
123 """
124 from solstone.think.activities import DEFAULT_ACTIVITIES, get_facet_activities
125
126 activities = get_facet_activities("personal")
127 assert isinstance(activities, list)
128
129 # Should contain all defaults (full vocabulary for unconfigured facets)
130 all_default_ids = {a["id"] for a in DEFAULT_ACTIVITIES}
131 assert {a["id"] for a in activities} == all_default_ids
132
133
134def test_meeting_is_always_on():
135 """Test that meeting is marked always_on in DEFAULT_ACTIVITIES."""
136 from solstone.think.activities import DEFAULT_ACTIVITIES
137
138 always_on_ids = {a["id"] for a in DEFAULT_ACTIVITIES if a.get("always_on")}
139 assert "meeting" in always_on_ids
140 assert "email" in always_on_ids
141 assert "messaging" in always_on_ids
142
143
144def test_unconfigured_facet_returns_all_defaults(monkeypatch):
145 """Test that a facet with no activities.jsonl gets all 16 defaults."""
146 from solstone.think.activities import DEFAULT_ACTIVITIES, get_facet_activities
147
148 with tempfile.TemporaryDirectory() as tmpdir:
149 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
150
151 facet_path = Path(tmpdir) / "facets" / "new_facet"
152 facet_path.mkdir(parents=True)
153
154 activities = get_facet_activities("new_facet")
155 assert len(activities) == len(DEFAULT_ACTIVITIES)
156
157 activity_ids = {a["id"] for a in activities}
158 default_ids = {a["id"] for a in DEFAULT_ACTIVITIES}
159 assert activity_ids == default_ids
160
161 # All should be marked as not custom
162 for activity in activities:
163 assert activity.get("custom") is False
164
165
166def test_configured_facet_includes_meeting_always_on(monkeypatch):
167 """Test that a facet with explicit activities auto-includes meeting."""
168 from solstone.think.activities import get_facet_activities, save_facet_activities
169
170 with tempfile.TemporaryDirectory() as tmpdir:
171 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
172
173 facet_path = Path(tmpdir) / "facets" / "work"
174 facet_path.mkdir(parents=True)
175
176 # Save only coding — meeting, email, messaging should auto-include
177 save_facet_activities("work", [{"id": "coding"}])
178
179 activities = get_facet_activities("work")
180 activity_ids = {a["id"] for a in activities}
181
182 assert "coding" in activity_ids
183 assert "meeting" in activity_ids
184 assert "email" in activity_ids
185 assert "messaging" in activity_ids
186
187
188def test_facet_activities_roundtrip(monkeypatch):
189 """Test saving and loading activities."""
190 from solstone.think.activities import (
191 DEFAULT_ACTIVITIES,
192 _get_activities_path,
193 get_facet_activities,
194 save_facet_activities,
195 )
196
197 # Create a temp journal
198 with tempfile.TemporaryDirectory() as tmpdir:
199 # Temporarily override SOLSTONE_JOURNAL
200 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
201
202 # Create facet directory
203 facet_path = Path(tmpdir) / "facets" / "test_facet"
204 facet_path.mkdir(parents=True)
205
206 # Save some activities
207 activities = [
208 {"id": "meeting", "priority": "high"},
209 {"id": "coding", "description": "Custom coding description"},
210 {
211 "id": "browsing",
212 "instructions": "Custom browsing instructions for this facet",
213 },
214 {
215 "id": "custom_activity",
216 "name": "Custom",
217 "description": "A custom activity",
218 "instructions": "Custom activity detection hints",
219 "custom": True,
220 "emoji": "🎯",
221 "icon": "sparkles",
222 },
223 ]
224 save_facet_activities("test_facet", activities)
225
226 # Verify file was created
227 path = _get_activities_path("test_facet")
228 assert path.exists()
229
230 # Load and verify (4 saved + always-on defaults not already saved)
231 loaded = get_facet_activities("test_facet")
232 loaded_ids = {a["id"] for a in loaded}
233 saved_ids = {a["id"] for a in activities}
234 always_on_ids = {a["id"] for a in DEFAULT_ACTIVITIES if a.get("always_on")}
235 assert loaded_ids == saved_ids | always_on_ids
236
237 # Check meeting (predefined with priority override)
238 meeting = next(a for a in loaded if a["id"] == "meeting")
239 assert meeting["priority"] == "high"
240 assert meeting["custom"] is False
241 assert "name" in meeting # Should have default name
242 # Should have default instructions (no override)
243 assert "instructions" in meeting
244 assert "Levels:" in meeting["instructions"]
245
246 # Check coding (predefined with description override)
247 coding = next(a for a in loaded if a["id"] == "coding")
248 assert coding["description"] == "Custom coding description"
249 # Should keep default instructions (only description overridden)
250 assert "instructions" in coding
251
252 # Check browsing (predefined with instructions override)
253 browsing = next(a for a in loaded if a["id"] == "browsing")
254 assert browsing["instructions"] == "Custom browsing instructions for this facet"
255
256 # Check custom activity with instructions
257 custom = next(a for a in loaded if a["id"] == "custom_activity")
258 assert custom["custom"] is True
259 assert custom["name"] == "Custom"
260 assert custom["instructions"] == "Custom activity detection hints"
261 assert custom["emoji"] == "🎯"
262 assert custom["icon"] == "sparkles"
263
264
265def test_get_facet_activities_normalizes_legacy_custom_icon_without_writing(
266 monkeypatch,
267):
268 """Legacy custom records with glyph in icon are normalized in memory only."""
269 from solstone.think.activities import get_facet_activities
270
271 with tempfile.TemporaryDirectory() as tmpdir:
272 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
273 activities_dir = Path(tmpdir) / "facets" / "work" / "activities"
274 activities_dir.mkdir(parents=True)
275 activities_file = activities_dir / "activities.jsonl"
276 activities_file.write_text(
277 json.dumps(
278 {
279 "id": "legacy_custom",
280 "custom": True,
281 "name": "Legacy custom",
282 "description": "Old stored shape",
283 "icon": "🎯",
284 }
285 )
286 + "\n",
287 encoding="utf-8",
288 )
289
290 activities = get_facet_activities("work")
291
292 legacy = next(
293 activity for activity in activities if activity["id"] == "legacy_custom"
294 )
295 assert legacy["emoji"] == "🎯"
296 assert "icon" not in legacy
297 assert '"icon":' in activities_file.read_text(encoding="utf-8")
298 assert '"emoji":' not in activities_file.read_text(encoding="utf-8")
299
300
301def test_migrate_custom_activity_icons_to_emoji_is_idempotent(monkeypatch):
302 from solstone.think.activities import migrate_custom_activity_icons_to_emoji
303
304 with tempfile.TemporaryDirectory() as tmpdir:
305 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
306 activities_dir = Path(tmpdir) / "facets" / "work" / "activities"
307 activities_dir.mkdir(parents=True)
308 activities_file = activities_dir / "activities.jsonl"
309 activities_file.write_text(
310 json.dumps(
311 {
312 "id": "legacy_custom",
313 "custom": True,
314 "name": "Legacy custom",
315 "description": "Old stored shape",
316 "icon": "🎯",
317 }
318 )
319 + "\n",
320 encoding="utf-8",
321 )
322
323 first = migrate_custom_activity_icons_to_emoji()
324 second = migrate_custom_activity_icons_to_emoji()
325
326 assert first["files_scanned"] == 1
327 assert first["files_changed"] == 1
328 assert first["records_changed"] == 1
329 assert second["files_scanned"] == 1
330 assert second["files_changed"] == 0
331 assert second["records_changed"] == 0
332
333 records = [
334 json.loads(line)
335 for line in activities_file.read_text(encoding="utf-8").splitlines()
336 ]
337 assert records == [
338 {
339 "id": "legacy_custom",
340 "custom": True,
341 "name": "Legacy custom",
342 "description": "Old stored shape",
343 "emoji": "🎯",
344 }
345 ]
346
347
348def test_migrate_custom_activity_icons_to_emoji_skips_hybrid_record(
349 monkeypatch,
350 caplog,
351):
352 from solstone.think.activities import migrate_custom_activity_icons_to_emoji
353
354 with tempfile.TemporaryDirectory() as tmpdir:
355 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
356 activities_dir = Path(tmpdir) / "facets" / "work" / "activities"
357 activities_dir.mkdir(parents=True)
358 activities_file = activities_dir / "activities.jsonl"
359 original = (
360 json.dumps(
361 {
362 "id": "hybrid_custom",
363 "custom": True,
364 "name": "Hybrid custom",
365 "description": "Ambiguous stored shape",
366 "emoji": "🎯",
367 "icon": "🎨",
368 },
369 ensure_ascii=False,
370 )
371 + "\n"
372 )
373 activities_file.write_text(original, encoding="utf-8")
374
375 with caplog.at_level(logging.INFO, logger="solstone.think.activities"):
376 first = migrate_custom_activity_icons_to_emoji()
377 second = migrate_custom_activity_icons_to_emoji()
378
379 assert first["files_scanned"] == 1
380 assert first["files_changed"] == 0
381 assert first["records_changed"] == 0
382 assert second["files_scanned"] == 1
383 assert second["files_changed"] == 0
384 assert second["records_changed"] == 0
385 assert activities_file.read_text(encoding="utf-8") == original
386 assert (
387 sum(
388 "skipping activity icon migration for facet work activity hybrid_custom"
389 in record.getMessage()
390 for record in caplog.records
391 )
392 == 2
393 )
394
395
396def test_add_activity_to_facet(monkeypatch):
397 """Test adding an activity to a facet."""
398 from solstone.think.activities import (
399 add_activity_to_facet,
400 get_facet_activities,
401 remove_activity_from_facet,
402 )
403
404 with tempfile.TemporaryDirectory() as tmpdir:
405 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
406
407 facet_path = Path(tmpdir) / "facets" / "test_facet"
408 facet_path.mkdir(parents=True)
409
410 # Add a predefined activity
411 result = add_activity_to_facet("test_facet", "meeting", priority="high")
412 assert result["id"] == "meeting"
413
414 # Verify it was added (+ always-on defaults)
415 activities = get_facet_activities("test_facet")
416 activity_ids = {a["id"] for a in activities}
417 assert "meeting" in activity_ids
418
419 # Adding same activity again should not duplicate
420 prev_count = len(activities)
421 add_activity_to_facet("test_facet", "meeting")
422 activities = get_facet_activities("test_facet")
423 assert len(activities) == prev_count
424
425 # Add a predefined activity with custom instructions
426 add_activity_to_facet(
427 "test_facet",
428 "coding",
429 instructions="Focus on Python and Rust only",
430 )
431 coding = next(
432 a for a in get_facet_activities("test_facet") if a["id"] == "coding"
433 )
434 assert coding["instructions"] == "Focus on Python and Rust only"
435
436 # Add a custom activity with instructions
437 add_activity_to_facet(
438 "test_facet",
439 "3d_modeling",
440 name="3D Modeling",
441 description="Blender and CAD work",
442 instructions="Detect via: Blender, FreeCAD, OpenSCAD windows",
443 )
444 modeling = next(
445 a for a in get_facet_activities("test_facet") if a["id"] == "3d_modeling"
446 )
447 assert (
448 modeling["instructions"] == "Detect via: Blender, FreeCAD, OpenSCAD windows"
449 )
450
451 # Remove it
452 removed = remove_activity_from_facet("test_facet", "meeting")
453 assert removed is True
454
455 # Removing non-existent should return False
456 removed = remove_activity_from_facet("test_facet", "meeting")
457 assert removed is False
458
459
460def test_update_activity_in_facet(monkeypatch):
461 """Test updating an activity in a facet."""
462 from solstone.think.activities import (
463 add_activity_to_facet,
464 get_activity_by_id,
465 update_activity_in_facet,
466 )
467
468 with tempfile.TemporaryDirectory() as tmpdir:
469 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
470
471 facet_path = Path(tmpdir) / "facets" / "test_facet"
472 facet_path.mkdir(parents=True)
473
474 # Add an activity
475 add_activity_to_facet("test_facet", "meeting")
476
477 # Update it
478 updated = update_activity_in_facet(
479 "test_facet", "meeting", priority="low", description="Updated desc"
480 )
481 assert updated is not None
482 assert updated["priority"] == "low"
483 assert updated["description"] == "Updated desc"
484
485 # Update instructions
486 updated = update_activity_in_facet(
487 "test_facet",
488 "meeting",
489 instructions="Only detect scheduled meetings, not ad-hoc calls",
490 )
491 assert updated is not None
492 assert (
493 updated["instructions"]
494 == "Only detect scheduled meetings, not ad-hoc calls"
495 )
496 # Other fields should be preserved
497 assert updated["priority"] == "low"
498
499 # Verify via lookup
500 activity = get_activity_by_id("test_facet", "meeting")
501 assert activity["priority"] == "low"
502 assert (
503 activity["instructions"]
504 == "Only detect scheduled meetings, not ad-hoc calls"
505 )
506
507 # Reset instructions to default via empty string
508 from solstone.think.activities import DEFAULT_ACTIVITIES
509
510 default_instructions = next(
511 a["instructions"] for a in DEFAULT_ACTIVITIES if a["id"] == "meeting"
512 )
513 updated = update_activity_in_facet("test_facet", "meeting", instructions="")
514 assert updated is not None
515 assert updated["instructions"] == default_instructions
516
517 # Reset description to default via empty string
518 default_desc = next(
519 a["description"] for a in DEFAULT_ACTIVITIES if a["id"] == "meeting"
520 )
521 updated = update_activity_in_facet("test_facet", "meeting", description="")
522 assert updated is not None
523 assert updated["description"] == default_desc
524
525 # Reset priority to default via "normal"
526 updated = update_activity_in_facet("test_facet", "meeting", priority="normal")
527 assert updated is not None
528 assert updated["priority"] == "normal"
529
530 # Update non-existent should return None
531 result = update_activity_in_facet("test_facet", "nonexistent", priority="high")
532 assert result is None
533
534
535def test_format_activities_context_includes_instructions(monkeypatch):
536 """Test that format_activities_context renders instructions inline."""
537 from solstone.think.activities import save_facet_activities
538
539 with tempfile.TemporaryDirectory() as tmpdir:
540 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
541
542 facet_path = Path(tmpdir) / "facets" / "test_facet"
543 facet_path.mkdir(parents=True)
544
545 save_facet_activities(
546 "test_facet",
547 [
548 {"id": "coding"},
549 {
550 "id": "custom_task",
551 "name": "Custom",
552 "description": "A custom activity",
553 "instructions": "Detect via: specific app UI",
554 "custom": True,
555 },
556 ],
557 )
558
559 from solstone.talent.activity_state import format_activities_context
560
561 output = format_activities_context("test_facet")
562
563 # Predefined coding should include its default instructions
564 assert "**coding**" in output
565 assert "IDE or editor open" in output # from default instructions
566
567 # Custom activity should include its custom instructions
568 assert "**custom_task**" in output
569 assert "Detect via: specific app UI" in output
570
571
572# ---------------------------------------------------------------------------
573# Activity Records (think/activities.py)
574# ---------------------------------------------------------------------------
575
576
577class TestLevelAvg:
578 """Tests for level_avg computation."""
579
580 def test_all_high(self):
581 from solstone.think.activities import level_avg
582
583 assert level_avg(["high", "high", "high"]) == 1.0
584
585 def test_all_medium(self):
586 from solstone.think.activities import level_avg
587
588 assert level_avg(["medium", "medium"]) == 0.5
589
590 def test_all_low(self):
591 from solstone.think.activities import level_avg
592
593 assert level_avg(["low", "low"]) == 0.25
594
595 def test_mixed(self):
596 from solstone.think.activities import level_avg
597
598 assert level_avg(["high", "medium"]) == 0.75
599
600 def test_empty_defaults_to_medium(self):
601 from solstone.think.activities import level_avg
602
603 assert level_avg([]) == 0.5
604
605 def test_unknown_defaults_to_medium(self):
606 from solstone.think.activities import level_avg
607
608 assert level_avg(["unknown", "high"]) == 0.75
609
610
611class TestActivityRecordIO:
612 """Tests for append/load/update of activity records."""
613
614 def test_append_and_load(self, monkeypatch):
615 from solstone.think.activities import (
616 append_activity_record,
617 load_activity_records,
618 )
619
620 with tempfile.TemporaryDirectory() as tmpdir:
621 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
622
623 record = {
624 "id": "coding_100000_300",
625 "activity": "coding",
626 "segments": ["100000_300", "100500_300"],
627 "level_avg": 0.75,
628 "description": "Test coding session",
629 "active_entities": ["VS Code"],
630 "created_at": 1234567890000,
631 }
632
633 assert append_activity_record("work", "20260209", record) is True
634 records = load_activity_records("work", "20260209")
635 assert len(records) == 1
636 assert records[0]["id"] == "coding_100000_300"
637 assert records[0]["segments"] == ["100000_300", "100500_300"]
638 assert records[0]["title"] == "Test coding session"
639 assert records[0]["details"] == ""
640 assert records[0]["hidden"] is False
641 assert records[0]["edits"] == []
642
643 def test_append_idempotent(self, monkeypatch):
644 from solstone.think.activities import (
645 append_activity_record,
646 load_activity_records,
647 )
648
649 with tempfile.TemporaryDirectory() as tmpdir:
650 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
651
652 record = {
653 "id": "coding_100000_300",
654 "activity": "coding",
655 "segments": ["100000_300"],
656 "created_at": 1234567890000,
657 }
658
659 assert append_activity_record("work", "20260209", record) is True
660 assert append_activity_record("work", "20260209", record) is False
661
662 records = load_activity_records("work", "20260209")
663 assert len(records) == 1
664
665 def test_load_nonexistent_returns_empty(self, monkeypatch):
666 from solstone.think.activities import load_activity_records
667
668 with tempfile.TemporaryDirectory() as tmpdir:
669 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
670 assert load_activity_records("work", "20260209") == []
671
672 def test_update_description(self, monkeypatch):
673 from solstone.think.activities import (
674 append_activity_record,
675 load_activity_records,
676 update_record_description,
677 )
678
679 with tempfile.TemporaryDirectory() as tmpdir:
680 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
681
682 record = {
683 "id": "coding_100000_300",
684 "activity": "coding",
685 "description": "Original description",
686 "segments": ["100000_300"],
687 "created_at": 1234567890000,
688 }
689
690 append_activity_record("work", "20260209", record)
691 result = update_record_description(
692 "work", "20260209", "coding_100000_300", "Updated description"
693 )
694 assert result is True
695
696 records = load_activity_records("work", "20260209")
697 assert records[0]["description"] == "Updated description"
698 assert records[0]["title"] == "Updated description"
699 assert records[0]["details"] == ""
700
701 def test_update_description_with_title_and_details(self, monkeypatch):
702 from solstone.think.activities import (
703 append_activity_record,
704 load_activity_records,
705 update_record_description,
706 )
707
708 with tempfile.TemporaryDirectory() as tmpdir:
709 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
710
711 record = {
712 "id": "coding_100000_300",
713 "activity": "coding",
714 "description": "Original description",
715 "segments": ["100000_300"],
716 "created_at": 1234567890000,
717 }
718
719 append_activity_record("work", "20260209", record)
720 result = update_record_description(
721 "work",
722 "20260209",
723 "coding_100000_300",
724 "Updated description",
725 title="Focused coding",
726 details="Pairing with Alex on tests.",
727 )
728
729 assert result is True
730 records = load_activity_records("work", "20260209")
731 assert records[0]["description"] == "Updated description"
732 assert records[0]["title"] == "Focused coding"
733 assert records[0]["details"] == "Pairing with Alex on tests."
734
735 def test_update_description_none_title_and_details_only_updates_description(
736 self, monkeypatch
737 ):
738 from solstone.think.activities import (
739 append_activity_record,
740 load_activity_records,
741 update_record_description,
742 )
743
744 with tempfile.TemporaryDirectory() as tmpdir:
745 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
746
747 append_activity_record(
748 "work",
749 "20260209",
750 {
751 "id": "coding_100000_300",
752 "activity": "coding",
753 "title": "Existing title",
754 "details": "Existing details",
755 "description": "Original description",
756 "segments": ["100000_300"],
757 "created_at": 1234567890000,
758 },
759 )
760
761 assert (
762 update_record_description(
763 "work",
764 "20260209",
765 "coding_100000_300",
766 "Updated description",
767 title=None,
768 details=None,
769 )
770 is True
771 )
772
773 records = load_activity_records("work", "20260209")
774 assert records[0]["description"] == "Updated description"
775 assert records[0]["title"] == "Existing title"
776 assert records[0]["details"] == "Existing details"
777
778 def test_update_nonexistent_returns_false(self, monkeypatch):
779 from solstone.think.activities import update_record_description
780
781 with tempfile.TemporaryDirectory() as tmpdir:
782 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
783 assert (
784 update_record_description("work", "20260209", "nonexistent", "desc")
785 is False
786 )
787
788 def test_update_preserves_other_records(self, monkeypatch):
789 from solstone.think.activities import (
790 append_activity_record,
791 load_activity_records,
792 update_record_description,
793 )
794
795 with tempfile.TemporaryDirectory() as tmpdir:
796 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
797
798 r1 = {
799 "id": "coding_100000_300",
800 "activity": "coding",
801 "description": "First",
802 "segments": ["100000_300"],
803 "created_at": 1,
804 }
805 r2 = {
806 "id": "meeting_110000_300",
807 "activity": "meeting",
808 "description": "Second",
809 "segments": ["110000_300"],
810 "created_at": 2,
811 }
812
813 append_activity_record("work", "20260209", r1)
814 append_activity_record("work", "20260209", r2)
815
816 update_record_description(
817 "work", "20260209", "coding_100000_300", "Updated first"
818 )
819
820 records = load_activity_records("work", "20260209")
821 assert len(records) == 2
822 assert records[0]["description"] == "Updated first"
823 assert records[1]["description"] == "Second"
824
825 def test_update_activity_record_appends_edit(self, monkeypatch):
826 from solstone.think.activities import (
827 append_activity_record,
828 load_activity_records,
829 update_activity_record,
830 )
831
832 with tempfile.TemporaryDirectory() as tmpdir:
833 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
834
835 append_activity_record(
836 "work",
837 "20260209",
838 {
839 "id": "coding_100000_300",
840 "activity": "coding",
841 "description": "Original description",
842 "segments": ["100000_300"],
843 "created_at": 1234567890000,
844 },
845 )
846
847 updated = update_activity_record(
848 "work",
849 "20260209",
850 "coding_100000_300",
851 {"title": "Focused coding", "details": "Updated details"},
852 actor="cli:update",
853 note="updated fields: details, title",
854 )
855
856 assert updated is not None
857 assert updated["title"] == "Focused coding"
858 assert updated["details"] == "Updated details"
859 assert updated["edits"][-1]["actor"] == "cli:update"
860 assert updated["edits"][-1]["fields"] == ["title", "details"]
861
862 records = load_activity_records("work", "20260209")
863 assert records[0]["edits"][-1]["note"] == "updated fields: details, title"
864
865 def test_update_activity_record_validates_patch(self, monkeypatch):
866 from solstone.think.activities import update_activity_record
867
868 with tempfile.TemporaryDirectory() as tmpdir:
869 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
870
871 with pytest.raises(ValueError, match="patch cannot be empty"):
872 update_activity_record(
873 "work",
874 "20260209",
875 "coding_100000_300",
876 {},
877 actor="cli:update",
878 note="no-op",
879 )
880
881 with pytest.raises(ValueError, match="disallowed fields"):
882 update_activity_record(
883 "work",
884 "20260209",
885 "coding_100000_300",
886 {"activity": "meeting"},
887 actor="cli:update",
888 note="bad field",
889 )
890
891 def test_format_activities_renders_story(self):
892 from solstone.think.activities import format_activities
893
894 chunks, _meta = format_activities(
895 [
896 {
897 "id": "meeting_090000_300",
898 "activity": "meeting",
899 "description": "Team sync",
900 "segments": ["090000_300"],
901 "created_at": 1,
902 "participation": [{"name": "Mina"}],
903 "story": {
904 "body": "Aligned on the launch plan and assigned owners.",
905 "topics": ["launch", "owners"],
906 "confidence": 0.9,
907 },
908 },
909 {
910 "id": "coding_100000_300",
911 "activity": "coding",
912 "description": "Implementation block",
913 "segments": ["100000_300"],
914 "created_at": 2,
915 },
916 ]
917 )
918
919 assert (
920 "Aligned on the launch plan and assigned owners." in chunks[0]["markdown"]
921 )
922 assert "Topics: launch, owners" in chunks[0]["markdown"]
923 assert "Topics:" not in chunks[1]["markdown"]
924 assert (
925 "Aligned on the launch plan and assigned owners."
926 not in chunks[1]["markdown"]
927 )
928
929 def test_hidden_records_filtered_by_default(self, monkeypatch):
930 from solstone.think.activities import (
931 append_activity_record,
932 load_activity_records,
933 mute_activity_record,
934 unmute_activity_record,
935 )
936
937 with tempfile.TemporaryDirectory() as tmpdir:
938 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
939
940 append_activity_record(
941 "work",
942 "20260209",
943 {
944 "id": "coding_100000_300",
945 "activity": "coding",
946 "description": "Original description",
947 "segments": ["100000_300"],
948 "created_at": 1234567890000,
949 },
950 )
951
952 muted = mute_activity_record(
953 "work",
954 "20260209",
955 "coding_100000_300",
956 actor="cli:mute",
957 reason="too noisy",
958 )
959 assert muted is not None
960 assert muted["hidden"] is True
961 assert muted["edits"][-1]["note"] == "too noisy"
962
963 assert load_activity_records("work", "20260209") == []
964 hidden_records = load_activity_records(
965 "work", "20260209", include_hidden=True
966 )
967 assert len(hidden_records) == 1
968 assert hidden_records[0]["hidden"] is True
969
970 hidden_count = len(hidden_records[0]["edits"])
971 muted_again = mute_activity_record(
972 "work",
973 "20260209",
974 "coding_100000_300",
975 actor="cli:mute",
976 reason="still noisy",
977 )
978 assert muted_again is not None
979 assert len(muted_again["edits"]) == hidden_count
980
981 unmuted = unmute_activity_record(
982 "work",
983 "20260209",
984 "coding_100000_300",
985 actor="cli:unmute",
986 reason=None,
987 )
988 assert unmuted is not None
989 assert unmuted["hidden"] is False
990 assert unmuted["edits"][-1]["note"] == "unmuted"
991 assert len(load_activity_records("work", "20260209")) == 1
992
993
994# ---------------------------------------------------------------------------
995# Activities Agent Hooks (talent/activities.py)
996# ---------------------------------------------------------------------------
997
998
999def _setup_segment(tmpdir, day, segment, facet, state):
1000 """Helper to create an activity_state.json file in a segment."""
1001 talents_dir = (
1002 Path(tmpdir) / "chronicle" / day / "default" / segment / "talents" / facet
1003 )
1004 talents_dir.mkdir(parents=True, exist_ok=True)
1005 state_file = talents_dir / "activity_state.json"
1006 state_file.write_text(json.dumps(state))
1007
1008
1009class TestMakeActivityId:
1010 def test_basic(self):
1011 from solstone.think.activities import make_activity_id
1012
1013 assert make_activity_id("coding", "095809_303") == "coding_095809_303"
1014
1015 def test_with_custom_type(self):
1016 from solstone.think.activities import make_activity_id
1017
1018 assert (
1019 make_activity_id("video_editing", "120000_300")
1020 == "video_editing_120000_300"
1021 )
1022
1023
1024class TestListFacetsWithActivityState:
1025 def test_finds_facets(self, monkeypatch):
1026 from solstone.talent.activities import _list_facets_with_activity_state
1027
1028 with tempfile.TemporaryDirectory() as tmpdir:
1029 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1030
1031 _setup_segment(tmpdir, "20260209", "100000_300", "personal", [])
1032 _setup_segment(tmpdir, "20260209", "100000_300", "work", [])
1033
1034 facets = _list_facets_with_activity_state(
1035 "20260209", "100000_300", stream="default"
1036 )
1037 assert facets == ["personal", "work"]
1038
1039 def test_returns_empty_for_nonexistent(self, monkeypatch):
1040 from solstone.talent.activities import _list_facets_with_activity_state
1041
1042 with tempfile.TemporaryDirectory() as tmpdir:
1043 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1044 assert (
1045 _list_facets_with_activity_state(
1046 "20260209", "100000_300", stream="default"
1047 )
1048 == []
1049 )
1050
1051
1052class TestDetectEndedActivities:
1053 def test_explicit_ended(self):
1054 from solstone.talent.activities import _detect_ended_activities
1055
1056 prev = [
1057 {"activity": "coding", "state": "active", "since": "100000_300"},
1058 ]
1059 curr = [
1060 {"activity": "coding", "state": "ended", "since": "100000_300"},
1061 ]
1062 ended = _detect_ended_activities(prev, curr, timed_out=False)
1063 assert len(ended) == 1
1064 assert ended[0]["activity"] == "coding"
1065
1066 def test_implicit_ended(self):
1067 from solstone.talent.activities import _detect_ended_activities
1068
1069 prev = [
1070 {"activity": "coding", "state": "active", "since": "100000_300"},
1071 {"activity": "meeting", "state": "active", "since": "100000_300"},
1072 ]
1073 curr = [
1074 {"activity": "meeting", "state": "active", "since": "100000_300"},
1075 ]
1076 ended = _detect_ended_activities(prev, curr, timed_out=False)
1077 assert len(ended) == 1
1078 assert ended[0]["activity"] == "coding"
1079
1080 def test_timeout_ends_all(self):
1081 from solstone.talent.activities import _detect_ended_activities
1082
1083 prev = [
1084 {"activity": "coding", "state": "active", "since": "100000_300"},
1085 {"activity": "meeting", "state": "active", "since": "100000_300"},
1086 ]
1087 ended = _detect_ended_activities(prev, [], timed_out=True)
1088 assert len(ended) == 2
1089
1090 def test_continuing_not_ended(self):
1091 from solstone.talent.activities import _detect_ended_activities
1092
1093 prev = [
1094 {"activity": "coding", "state": "active", "since": "100000_300"},
1095 ]
1096 curr = [
1097 {"activity": "coding", "state": "active", "since": "100000_300"},
1098 ]
1099 ended = _detect_ended_activities(prev, curr, timed_out=False)
1100 assert len(ended) == 0
1101
1102 def test_ignores_previously_ended(self):
1103 from solstone.talent.activities import _detect_ended_activities
1104
1105 prev = [
1106 {"activity": "coding", "state": "ended", "since": "090000_300"},
1107 {"activity": "meeting", "state": "active", "since": "100000_300"},
1108 ]
1109 curr = []
1110 ended = _detect_ended_activities(prev, curr, timed_out=False)
1111 assert len(ended) == 1
1112 assert ended[0]["activity"] == "meeting"
1113
1114 def test_new_activity_same_type(self):
1115 """A new activity of same type with different since is not the same."""
1116 from solstone.talent.activities import _detect_ended_activities
1117
1118 prev = [
1119 {"activity": "coding", "state": "active", "since": "100000_300"},
1120 ]
1121 curr = [
1122 {"activity": "coding", "state": "active", "since": "110000_300"},
1123 ]
1124 ended = _detect_ended_activities(prev, curr, timed_out=False)
1125 assert len(ended) == 1
1126 assert ended[0]["since"] == "100000_300"
1127
1128
1129class TestWalkActivitySegments:
1130 def test_walks_segments(self, monkeypatch):
1131 from solstone.talent.activities import _walk_activity_segments
1132
1133 with tempfile.TemporaryDirectory() as tmpdir:
1134 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1135
1136 _setup_segment(
1137 tmpdir,
1138 "20260209",
1139 "100000_300",
1140 "work",
1141 [
1142 {
1143 "activity": "coding",
1144 "state": "active",
1145 "since": "100000_300",
1146 "description": "Starting work",
1147 "level": "high",
1148 "active_entities": ["VS Code"],
1149 }
1150 ],
1151 )
1152 _setup_segment(
1153 tmpdir,
1154 "20260209",
1155 "100500_300",
1156 "work",
1157 [
1158 {
1159 "activity": "coding",
1160 "state": "active",
1161 "since": "100000_300",
1162 "description": "Continuing work",
1163 "level": "medium",
1164 "active_entities": ["VS Code", "Claude Code"],
1165 }
1166 ],
1167 )
1168
1169 result = _walk_activity_segments(
1170 "20260209", "work", "coding", "100000_300", "100500_300"
1171 )
1172
1173 assert result["segments"] == ["100000_300", "100500_300"]
1174 assert len(result["descriptions"]) == 2
1175 assert result["levels"] == ["high", "medium"]
1176 assert result["active_entities"] == ["VS Code", "Claude Code"]
1177
1178 def test_deduplicates_entities(self, monkeypatch):
1179 from solstone.talent.activities import _walk_activity_segments
1180
1181 with tempfile.TemporaryDirectory() as tmpdir:
1182 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1183
1184 _setup_segment(
1185 tmpdir,
1186 "20260209",
1187 "100000_300",
1188 "work",
1189 [
1190 {
1191 "activity": "coding",
1192 "state": "active",
1193 "since": "100000_300",
1194 "level": "high",
1195 "active_entities": ["VS Code", "Git"],
1196 }
1197 ],
1198 )
1199 _setup_segment(
1200 tmpdir,
1201 "20260209",
1202 "100500_300",
1203 "work",
1204 [
1205 {
1206 "activity": "coding",
1207 "state": "active",
1208 "since": "100000_300",
1209 "level": "high",
1210 "active_entities": ["VS Code", "Claude Code"],
1211 }
1212 ],
1213 )
1214
1215 result = _walk_activity_segments(
1216 "20260209", "work", "coding", "100000_300", "100500_300"
1217 )
1218
1219 assert result["active_entities"] == ["VS Code", "Git", "Claude Code"]
1220
1221 def test_empty_when_no_match(self, monkeypatch):
1222 from solstone.talent.activities import _walk_activity_segments
1223
1224 with tempfile.TemporaryDirectory() as tmpdir:
1225 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1226 (Path(tmpdir) / "chronicle" / "20260209").mkdir(parents=True)
1227
1228 result = _walk_activity_segments(
1229 "20260209", "work", "coding", "100000_300", "100500_300"
1230 )
1231 assert result["segments"] == []
1232
1233
1234class TestPreProcess:
1235 """Tests for the activities pre_process hook."""
1236
1237 def test_skips_when_no_previous_segment(self, monkeypatch):
1238 from solstone.talent.activities import pre_process
1239
1240 with tempfile.TemporaryDirectory() as tmpdir:
1241 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1242
1243 day_dir = Path(tmpdir) / "chronicle" / "20260209"
1244 day_dir.mkdir(parents=True)
1245 (day_dir / "default" / "100000_300").mkdir(parents=True)
1246
1247 result = pre_process(
1248 {"day": "20260209", "segment": "100000_300", "stream": "default"}
1249 )
1250 assert result is not None
1251 assert "skip_reason" in result
1252
1253 def test_skips_when_no_ended_activities(self, monkeypatch):
1254 from solstone.talent.activities import pre_process
1255
1256 with tempfile.TemporaryDirectory() as tmpdir:
1257 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1258
1259 _setup_segment(
1260 tmpdir,
1261 "20260209",
1262 "100000_300",
1263 "work",
1264 [
1265 {
1266 "activity": "coding",
1267 "state": "active",
1268 "since": "100000_300",
1269 "level": "high",
1270 }
1271 ],
1272 )
1273 _setup_segment(
1274 tmpdir,
1275 "20260209",
1276 "100500_300",
1277 "work",
1278 [
1279 {
1280 "activity": "coding",
1281 "state": "active",
1282 "since": "100000_300",
1283 "level": "high",
1284 }
1285 ],
1286 )
1287
1288 result = pre_process(
1289 {"day": "20260209", "segment": "100500_300", "stream": "default"}
1290 )
1291 assert result is not None
1292 assert result.get("skip_reason") == "no_ended_activities"
1293
1294 def test_detects_ended_and_writes_record(self, monkeypatch):
1295 from solstone.talent.activities import pre_process
1296 from solstone.think.activities import load_activity_records
1297
1298 with tempfile.TemporaryDirectory() as tmpdir:
1299 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1300
1301 _setup_segment(
1302 tmpdir,
1303 "20260209",
1304 "100000_300",
1305 "work",
1306 [
1307 {
1308 "activity": "coding",
1309 "state": "active",
1310 "since": "100000_300",
1311 "description": "Starting work",
1312 "level": "high",
1313 "active_entities": ["VS Code"],
1314 }
1315 ],
1316 )
1317 _setup_segment(tmpdir, "20260209", "100500_300", "work", [])
1318
1319 result = pre_process(
1320 {"day": "20260209", "segment": "100500_300", "stream": "default"}
1321 )
1322
1323 assert "skip_reason" not in result
1324 assert "transcript" in result
1325 assert "coding_100000_300" in result["transcript"]
1326
1327 records = load_activity_records("work", "20260209")
1328 assert len(records) == 1
1329 assert records[0]["id"] == "coding_100000_300"
1330 assert records[0]["segments"] == ["100000_300"]
1331
1332 def test_idempotent_on_rerun(self, monkeypatch):
1333 from solstone.talent.activities import pre_process
1334 from solstone.think.activities import load_activity_records
1335
1336 with tempfile.TemporaryDirectory() as tmpdir:
1337 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1338
1339 _setup_segment(
1340 tmpdir,
1341 "20260209",
1342 "100000_300",
1343 "work",
1344 [
1345 {
1346 "activity": "coding",
1347 "state": "active",
1348 "since": "100000_300",
1349 "description": "Coding",
1350 "level": "high",
1351 }
1352 ],
1353 )
1354 _setup_segment(tmpdir, "20260209", "100500_300", "work", [])
1355
1356 context = {"day": "20260209", "segment": "100500_300", "stream": "default"}
1357 pre_process(context)
1358 pre_process(context)
1359
1360 records = load_activity_records("work", "20260209")
1361 assert len(records) == 1
1362
1363 def test_multi_facet_detection(self, monkeypatch):
1364 from solstone.talent.activities import pre_process
1365 from solstone.think.activities import load_activity_records
1366
1367 with tempfile.TemporaryDirectory() as tmpdir:
1368 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1369
1370 _setup_segment(
1371 tmpdir,
1372 "20260209",
1373 "100000_300",
1374 "work",
1375 [
1376 {
1377 "activity": "coding",
1378 "state": "active",
1379 "since": "100000_300",
1380 "description": "Work coding",
1381 "level": "high",
1382 }
1383 ],
1384 )
1385 _setup_segment(
1386 tmpdir,
1387 "20260209",
1388 "100000_300",
1389 "personal",
1390 [
1391 {
1392 "activity": "meeting",
1393 "state": "active",
1394 "since": "100000_300",
1395 "description": "Team standup",
1396 "level": "medium",
1397 }
1398 ],
1399 )
1400
1401 _setup_segment(tmpdir, "20260209", "100500_300", "work", [])
1402 _setup_segment(tmpdir, "20260209", "100500_300", "personal", [])
1403
1404 result = pre_process(
1405 {"day": "20260209", "segment": "100500_300", "stream": "default"}
1406 )
1407
1408 assert "skip_reason" not in result
1409 assert "#work" in result["transcript"]
1410 assert "#personal" in result["transcript"]
1411
1412 work_records = load_activity_records("work", "20260209")
1413 personal_records = load_activity_records("personal", "20260209")
1414 assert len(work_records) == 1
1415 assert len(personal_records) == 1
1416
1417 def test_multi_segment_span(self, monkeypatch):
1418 """Activity spanning multiple segments should collect all segments."""
1419 from solstone.talent.activities import pre_process
1420 from solstone.think.activities import load_activity_records
1421
1422 with tempfile.TemporaryDirectory() as tmpdir:
1423 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1424
1425 _setup_segment(
1426 tmpdir,
1427 "20260209",
1428 "100000_300",
1429 "work",
1430 [
1431 {
1432 "activity": "coding",
1433 "state": "active",
1434 "since": "100000_300",
1435 "description": "Starting",
1436 "level": "high",
1437 "active_entities": ["VS Code"],
1438 }
1439 ],
1440 )
1441 _setup_segment(
1442 tmpdir,
1443 "20260209",
1444 "100500_300",
1445 "work",
1446 [
1447 {
1448 "activity": "coding",
1449 "state": "active",
1450 "since": "100000_300",
1451 "description": "Continuing",
1452 "level": "medium",
1453 "active_entities": ["VS Code", "Git"],
1454 }
1455 ],
1456 )
1457 _setup_segment(
1458 tmpdir,
1459 "20260209",
1460 "101000_300",
1461 "work",
1462 [
1463 {
1464 "activity": "coding",
1465 "state": "active",
1466 "since": "100000_300",
1467 "description": "Finishing",
1468 "level": "high",
1469 "active_entities": ["Claude Code"],
1470 }
1471 ],
1472 )
1473 # Coding ends
1474 _setup_segment(tmpdir, "20260209", "101500_300", "work", [])
1475
1476 pre_process(
1477 {"day": "20260209", "segment": "101500_300", "stream": "default"}
1478 )
1479
1480 records = load_activity_records("work", "20260209")
1481 assert len(records) == 1
1482 r = records[0]
1483 assert r["segments"] == ["100000_300", "100500_300", "101000_300"]
1484 assert r["active_entities"] == ["VS Code", "Git", "Claude Code"]
1485 assert r["level_avg"] == 0.83 # (1.0 + 0.5 + 1.0) / 3
1486
1487
1488class TestPostProcess:
1489 """Tests for the activities post_process hook."""
1490
1491 def test_updates_descriptions(self, monkeypatch):
1492 from solstone.talent.activities import post_process
1493 from solstone.think.activities import (
1494 append_activity_record,
1495 load_activity_records,
1496 )
1497
1498 with tempfile.TemporaryDirectory() as tmpdir:
1499 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1500
1501 record = {
1502 "id": "coding_100000_300",
1503 "activity": "coding",
1504 "description": "Preliminary description",
1505 "segments": ["100000_300"],
1506 "created_at": 1,
1507 }
1508 append_activity_record("work", "20260209", record)
1509
1510 llm_result = json.dumps(
1511 {
1512 "work": [
1513 {
1514 "id": "coding_100000_300",
1515 "title": "Coding summary",
1516 "details": "Worked through test failures and cleanup.",
1517 "description": "Synthesized full description of coding session",
1518 }
1519 ]
1520 }
1521 )
1522
1523 post_process(llm_result, {"day": "20260209"})
1524
1525 records = load_activity_records("work", "20260209")
1526 assert (
1527 records[0]["description"]
1528 == "Synthesized full description of coding session"
1529 )
1530 assert records[0]["title"] == "Coding summary"
1531 assert records[0]["details"] == "Worked through test failures and cleanup."
1532
1533 def test_updates_descriptions_without_optional_fields(self, monkeypatch):
1534 from solstone.talent.activities import post_process
1535 from solstone.think.activities import (
1536 append_activity_record,
1537 load_activity_records,
1538 )
1539
1540 with tempfile.TemporaryDirectory() as tmpdir:
1541 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1542
1543 append_activity_record(
1544 "work",
1545 "20260209",
1546 {
1547 "id": "coding_100000_300",
1548 "activity": "coding",
1549 "title": "Existing title",
1550 "details": "Existing details",
1551 "description": "Preliminary description",
1552 "segments": ["100000_300"],
1553 "created_at": 1,
1554 },
1555 )
1556
1557 llm_result = json.dumps(
1558 {
1559 "work": [
1560 {
1561 "id": "coding_100000_300",
1562 "description": "Only description changed",
1563 }
1564 ]
1565 }
1566 )
1567
1568 post_process(llm_result, {"day": "20260209"})
1569
1570 records = load_activity_records("work", "20260209")
1571 assert records[0]["description"] == "Only description changed"
1572 assert records[0]["title"] == "Existing title"
1573 assert records[0]["details"] == "Existing details"
1574
1575 def test_handles_invalid_json(self):
1576 from solstone.talent.activities import post_process
1577
1578 result = post_process("not json", {"day": "20260209"})
1579 assert result is None
1580
1581 def test_handles_non_object(self):
1582 from solstone.talent.activities import post_process
1583
1584 result = post_process("[]", {"day": "20260209"})
1585 assert result is None
1586
1587 def test_returns_none(self, monkeypatch):
1588 from solstone.talent.activities import post_process
1589
1590 with tempfile.TemporaryDirectory() as tmpdir:
1591 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1592
1593 result = post_process("{}", {"day": "20260209"})
1594 assert result is None
1595
1596
1597class TestEstimateDurationMinutes:
1598 def test_single_segment(self):
1599 from solstone.think.activities import estimate_duration_minutes
1600
1601 assert estimate_duration_minutes(["100000_300"]) == 5
1602
1603 def test_multiple_segments(self):
1604 from solstone.think.activities import estimate_duration_minutes
1605
1606 assert estimate_duration_minutes(["100000_300", "100500_300"]) == 10
1607
1608 def test_empty_returns_1(self):
1609 from solstone.think.activities import estimate_duration_minutes
1610
1611 assert estimate_duration_minutes([]) == 1
1612
1613
1614class TestPreProcessMeta:
1615 """Tests for pre-hook stashing record data in meta."""
1616
1617 def test_meta_contains_activity_records(self, monkeypatch):
1618 from solstone.talent.activities import pre_process
1619
1620 with tempfile.TemporaryDirectory() as tmpdir:
1621 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1622
1623 _setup_segment(
1624 tmpdir,
1625 "20260209",
1626 "100000_300",
1627 "work",
1628 [
1629 {
1630 "activity": "coding",
1631 "state": "active",
1632 "since": "100000_300",
1633 "level": "high",
1634 "description": "Writing code",
1635 "active_entities": ["VS Code"],
1636 }
1637 ],
1638 )
1639 _setup_segment(tmpdir, "20260209", "100500_300", "work", [])
1640
1641 result = pre_process(
1642 {
1643 "day": "20260209",
1644 "segment": "100500_300",
1645 "stream": "default",
1646 "meta": {},
1647 }
1648 )
1649
1650 assert "meta" in result
1651 records = result["meta"]["activity_records"]
1652 assert "work" in records
1653 assert "coding_100000_300" in records["work"]
1654 rec = records["work"]["coding_100000_300"]
1655 assert rec["activity"] == "coding"
1656 assert rec["segments"] == ["100000_300"]
1657 assert rec["level_avg"] == 1.0
1658 assert rec["active_entities"] == ["VS Code"]
1659 assert rec["description"] == "Writing code"
1660
1661 def test_meta_multiple_facets(self, monkeypatch):
1662 from solstone.talent.activities import pre_process
1663
1664 with tempfile.TemporaryDirectory() as tmpdir:
1665 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1666
1667 _setup_segment(
1668 tmpdir,
1669 "20260209",
1670 "100000_300",
1671 "work",
1672 [
1673 {
1674 "activity": "coding",
1675 "state": "active",
1676 "since": "100000_300",
1677 "level": "high",
1678 }
1679 ],
1680 )
1681 _setup_segment(
1682 tmpdir,
1683 "20260209",
1684 "100000_300",
1685 "personal",
1686 [
1687 {
1688 "activity": "browsing",
1689 "state": "active",
1690 "since": "100000_300",
1691 "level": "low",
1692 }
1693 ],
1694 )
1695 _setup_segment(tmpdir, "20260209", "100500_300", "work", [])
1696 _setup_segment(tmpdir, "20260209", "100500_300", "personal", [])
1697
1698 result = pre_process(
1699 {"day": "20260209", "segment": "100500_300", "stream": "default"}
1700 )
1701
1702 records = result["meta"]["activity_records"]
1703 assert "work" in records
1704 assert "personal" in records
1705
1706
1707class TestPostProcessEvents:
1708 """Tests for post-hook callosum event emission."""
1709
1710 def test_emits_events_with_llm_description(self, monkeypatch):
1711 from unittest.mock import patch
1712
1713 from solstone.talent.activities import post_process
1714 from solstone.think.activities import append_activity_record
1715
1716 with tempfile.TemporaryDirectory() as tmpdir:
1717 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1718
1719 record = {
1720 "id": "coding_100000_300",
1721 "activity": "coding",
1722 "description": "Preliminary",
1723 "segments": ["100000_300"],
1724 "created_at": 1,
1725 }
1726 append_activity_record("work", "20260209", record)
1727
1728 llm_result = json.dumps(
1729 {
1730 "work": [
1731 {
1732 "id": "coding_100000_300",
1733 "description": "Full synthesized description",
1734 }
1735 ]
1736 }
1737 )
1738
1739 meta = {
1740 "activity_records": {
1741 "work": {
1742 "coding_100000_300": {
1743 "activity": "coding",
1744 "segments": ["100000_300"],
1745 "description": "Preliminary",
1746 "level_avg": 1.0,
1747 "active_entities": ["VS Code"],
1748 }
1749 }
1750 }
1751 }
1752
1753 with patch("solstone.talent.activities.callosum_send") as mock_send:
1754 mock_send.return_value = True
1755 post_process(
1756 llm_result,
1757 {"day": "20260209", "segment": "100500_300", "meta": meta},
1758 )
1759
1760 mock_send.assert_called_once_with(
1761 "activity",
1762 "recorded",
1763 facet="work",
1764 day="20260209",
1765 segment="100500_300",
1766 id="coding_100000_300",
1767 activity="coding",
1768 segments=["100000_300"],
1769 level_avg=1.0,
1770 description="Full synthesized description",
1771 active_entities=["VS Code"],
1772 )
1773
1774 def test_falls_back_to_prehook_description_with_warning(self, monkeypatch, caplog):
1775 from unittest.mock import patch
1776
1777 from solstone.talent.activities import post_process
1778
1779 with tempfile.TemporaryDirectory() as tmpdir:
1780 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1781
1782 meta = {
1783 "activity_records": {
1784 "work": {
1785 "coding_100000_300": {
1786 "activity": "coding",
1787 "segments": ["100000_300"],
1788 "description": "Pre-hook fallback desc",
1789 "level_avg": 0.5,
1790 "active_entities": [],
1791 }
1792 }
1793 }
1794 }
1795
1796 # LLM returns empty — no descriptions to update
1797 with patch("solstone.talent.activities.callosum_send") as mock_send:
1798 mock_send.return_value = True
1799 import logging
1800
1801 with caplog.at_level(
1802 logging.WARNING, logger="solstone.talent.activities"
1803 ):
1804 post_process(
1805 "{}",
1806 {"day": "20260209", "segment": "100500_300", "meta": meta},
1807 )
1808
1809 mock_send.assert_called_once()
1810 call_kwargs = mock_send.call_args[1]
1811 assert call_kwargs["description"] == "Pre-hook fallback desc"
1812 assert "No LLM description" in caplog.text
1813
1814 def test_no_events_without_meta(self, monkeypatch):
1815 from unittest.mock import patch
1816
1817 from solstone.talent.activities import post_process
1818
1819 with tempfile.TemporaryDirectory() as tmpdir:
1820 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1821
1822 with patch("solstone.talent.activities.callosum_send") as mock_send:
1823 post_process("{}", {"day": "20260209", "segment": "100500_300"})
1824 mock_send.assert_not_called()
1825
1826 def test_event_emission_failure_does_not_raise(self, monkeypatch):
1827 from unittest.mock import patch
1828
1829 from solstone.talent.activities import post_process
1830
1831 with tempfile.TemporaryDirectory() as tmpdir:
1832 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1833
1834 meta = {
1835 "activity_records": {
1836 "work": {
1837 "coding_100000_300": {
1838 "activity": "coding",
1839 "segments": ["100000_300"],
1840 "description": "desc",
1841 "level_avg": 0.5,
1842 "active_entities": [],
1843 }
1844 }
1845 }
1846 }
1847
1848 with patch("solstone.talent.activities.callosum_send") as mock_send:
1849 mock_send.side_effect = OSError("socket error")
1850 # Should not raise
1851 result = post_process(
1852 "{}",
1853 {"day": "20260209", "segment": "100500_300", "meta": meta},
1854 )
1855 assert result is None
1856
1857
1858class TestHandleActivityRecorded:
1859 """Tests for supervisor's _handle_activity_recorded handler."""
1860
1861 def test_queues_think_task(self):
1862 from unittest.mock import MagicMock, patch
1863
1864 from solstone.think.supervisor import _handle_activity_recorded
1865
1866 mock_queue = MagicMock()
1867 with patch("solstone.think.supervisor._task_queue", mock_queue):
1868 _handle_activity_recorded(
1869 {
1870 "tract": "activity",
1871 "event": "recorded",
1872 "id": "coding_100000_300",
1873 "facet": "work",
1874 "day": "20260209",
1875 }
1876 )
1877
1878 mock_queue.submit.assert_called_once_with(
1879 [
1880 "journal",
1881 "think",
1882 "--activity",
1883 "coding_100000_300",
1884 "--facet",
1885 "work",
1886 "--day",
1887 "20260209",
1888 ],
1889 day="20260209",
1890 )
1891
1892 def test_ignores_wrong_tract(self):
1893 from unittest.mock import MagicMock, patch
1894
1895 from solstone.think.supervisor import _handle_activity_recorded
1896
1897 mock_queue = MagicMock()
1898 with patch("solstone.think.supervisor._task_queue", mock_queue):
1899 _handle_activity_recorded(
1900 {
1901 "tract": "think",
1902 "event": "recorded",
1903 "id": "x",
1904 "facet": "w",
1905 "day": "d",
1906 }
1907 )
1908 mock_queue.submit.assert_not_called()
1909
1910 def test_ignores_wrong_event(self):
1911 from unittest.mock import MagicMock, patch
1912
1913 from solstone.think.supervisor import _handle_activity_recorded
1914
1915 mock_queue = MagicMock()
1916 with patch("solstone.think.supervisor._task_queue", mock_queue):
1917 _handle_activity_recorded(
1918 {
1919 "tract": "activity",
1920 "event": "other",
1921 "id": "x",
1922 "facet": "w",
1923 "day": "d",
1924 }
1925 )
1926 mock_queue.submit.assert_not_called()
1927
1928 def test_warns_on_missing_fields(self, caplog):
1929 from unittest.mock import MagicMock, patch
1930
1931 from solstone.think.supervisor import _handle_activity_recorded
1932
1933 mock_queue = MagicMock()
1934 import logging
1935
1936 with patch("solstone.think.supervisor._task_queue", mock_queue):
1937 with caplog.at_level(logging.WARNING):
1938 _handle_activity_recorded(
1939 {"tract": "activity", "event": "recorded", "id": "x"}
1940 )
1941 mock_queue.submit.assert_not_called()
1942
1943 def test_warns_when_no_task_queue(self, caplog):
1944 import logging
1945 from unittest.mock import patch
1946
1947 from solstone.think.supervisor import _handle_activity_recorded
1948
1949 with patch("solstone.think.supervisor._task_queue", None):
1950 with caplog.at_level(logging.WARNING):
1951 _handle_activity_recorded(
1952 {
1953 "tract": "activity",
1954 "event": "recorded",
1955 "id": "coding_100000_300",
1956 "facet": "work",
1957 "day": "20260209",
1958 }
1959 )
1960 assert "No task queue" in caplog.text
1961
1962
1963# ---------------------------------------------------------------------------
1964# Flush Mode Tests
1965# ---------------------------------------------------------------------------
1966
1967
1968class TestPreProcessFlush:
1969 """Tests for the activities pre_process hook in flush mode."""
1970
1971 def test_flush_ends_all_active_activities(self, monkeypatch):
1972 from solstone.talent.activities import pre_process
1973 from solstone.think.activities import load_activity_records
1974
1975 with tempfile.TemporaryDirectory() as tmpdir:
1976 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
1977
1978 # Set up a segment with an active activity (no following segment)
1979 _setup_segment(
1980 tmpdir,
1981 "20260209",
1982 "100000_300",
1983 "work",
1984 [
1985 {
1986 "activity": "coding",
1987 "state": "active",
1988 "since": "100000_300",
1989 "description": "Working on feature",
1990 "level": "high",
1991 "active_entities": ["VS Code"],
1992 }
1993 ],
1994 )
1995
1996 result = pre_process(
1997 {
1998 "day": "20260209",
1999 "segment": "100000_300",
2000 "stream": "default",
2001 "flush": True,
2002 }
2003 )
2004
2005 # Should detect the active activity as ended
2006 assert "skip_reason" not in result
2007 assert "transcript" in result
2008 assert "coding_100000_300" in result["transcript"]
2009
2010 # Record should be written
2011 records = load_activity_records("work", "20260209")
2012 assert len(records) == 1
2013 assert records[0]["id"] == "coding_100000_300"
2014 assert records[0]["segments"] == ["100000_300"]
2015
2016 def test_flush_skips_when_no_active_activities(self, monkeypatch):
2017 from solstone.talent.activities import pre_process
2018
2019 with tempfile.TemporaryDirectory() as tmpdir:
2020 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2021
2022 # Set up a segment with only ended activities
2023 _setup_segment(
2024 tmpdir,
2025 "20260209",
2026 "100000_300",
2027 "work",
2028 [
2029 {
2030 "activity": "coding",
2031 "state": "ended",
2032 "since": "090000_300",
2033 }
2034 ],
2035 )
2036
2037 result = pre_process(
2038 {
2039 "day": "20260209",
2040 "segment": "100000_300",
2041 "stream": "default",
2042 "flush": True,
2043 }
2044 )
2045 assert result["skip_reason"] == "no_active_activities"
2046
2047 def test_flush_skips_when_no_activity_state(self, monkeypatch):
2048 from solstone.talent.activities import pre_process
2049
2050 with tempfile.TemporaryDirectory() as tmpdir:
2051 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2052
2053 # Create segment dir but no activity_state files
2054 seg_dir = Path(tmpdir) / "chronicle" / "20260209" / "default" / "100000_300"
2055 seg_dir.mkdir(parents=True)
2056
2057 result = pre_process(
2058 {
2059 "day": "20260209",
2060 "segment": "100000_300",
2061 "stream": "default",
2062 "flush": True,
2063 }
2064 )
2065 assert result["skip_reason"] == "no_activity_state"
2066
2067 def test_flush_handles_multiple_facets(self, monkeypatch):
2068 from solstone.talent.activities import pre_process
2069 from solstone.think.activities import load_activity_records
2070
2071 with tempfile.TemporaryDirectory() as tmpdir:
2072 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2073
2074 _setup_segment(
2075 tmpdir,
2076 "20260209",
2077 "100000_300",
2078 "work",
2079 [
2080 {
2081 "activity": "coding",
2082 "state": "active",
2083 "since": "100000_300",
2084 "description": "Work coding",
2085 "level": "high",
2086 }
2087 ],
2088 )
2089 _setup_segment(
2090 tmpdir,
2091 "20260209",
2092 "100000_300",
2093 "personal",
2094 [
2095 {
2096 "activity": "browsing",
2097 "state": "active",
2098 "since": "100000_300",
2099 "description": "Personal browsing",
2100 "level": "low",
2101 }
2102 ],
2103 )
2104
2105 result = pre_process(
2106 {
2107 "day": "20260209",
2108 "segment": "100000_300",
2109 "stream": "default",
2110 "flush": True,
2111 }
2112 )
2113
2114 assert "skip_reason" not in result
2115 assert "transcript" in result
2116
2117 work_records = load_activity_records("work", "20260209")
2118 personal_records = load_activity_records("personal", "20260209")
2119 assert len(work_records) == 1
2120 assert len(personal_records) == 1
2121
2122 def test_flush_is_idempotent(self, monkeypatch):
2123 from solstone.talent.activities import pre_process
2124 from solstone.think.activities import load_activity_records
2125
2126 with tempfile.TemporaryDirectory() as tmpdir:
2127 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2128
2129 _setup_segment(
2130 tmpdir,
2131 "20260209",
2132 "100000_300",
2133 "work",
2134 [
2135 {
2136 "activity": "coding",
2137 "state": "active",
2138 "since": "100000_300",
2139 "description": "Coding",
2140 "level": "high",
2141 }
2142 ],
2143 )
2144
2145 context = {
2146 "day": "20260209",
2147 "segment": "100000_300",
2148 "stream": "default",
2149 "flush": True,
2150 }
2151 pre_process(context)
2152 pre_process(context)
2153
2154 records = load_activity_records("work", "20260209")
2155 assert len(records) == 1
2156
2157 def test_flush_stashes_meta_for_post_hook(self, monkeypatch):
2158 from solstone.talent.activities import pre_process
2159
2160 with tempfile.TemporaryDirectory() as tmpdir:
2161 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2162
2163 _setup_segment(
2164 tmpdir,
2165 "20260209",
2166 "100000_300",
2167 "work",
2168 [
2169 {
2170 "activity": "coding",
2171 "state": "active",
2172 "since": "100000_300",
2173 "description": "Coding",
2174 "level": "high",
2175 }
2176 ],
2177 )
2178
2179 result = pre_process(
2180 {
2181 "day": "20260209",
2182 "segment": "100000_300",
2183 "stream": "default",
2184 "flush": True,
2185 }
2186 )
2187
2188 assert "meta" in result
2189 records = result["meta"]["activity_records"]
2190 assert "work" in records
2191 assert "coding_100000_300" in records["work"]
2192
2193
2194class TestCheckSegmentFlush:
2195 """Tests for supervisor's _check_segment_flush."""
2196
2197 def test_queues_flush_after_timeout(self):
2198 import time as time_mod
2199 from unittest.mock import MagicMock, patch
2200
2201 from solstone.think.supervisor import _check_segment_flush, _flush_state
2202
2203 # Set up state as if a segment arrived over an hour ago
2204 _flush_state["last_segment_ts"] = time_mod.time() - 4000
2205 _flush_state["day"] = "20260209"
2206 _flush_state["segment"] = "100000_300"
2207 _flush_state["flushed"] = False
2208
2209 mock_queue = MagicMock()
2210 with (
2211 patch("solstone.think.supervisor._task_queue", mock_queue),
2212 patch("solstone.think.supervisor._is_remote_mode", False),
2213 ):
2214 _check_segment_flush()
2215
2216 mock_queue.submit.assert_called_once_with(
2217 [
2218 "journal",
2219 "think",
2220 "-v",
2221 "--day",
2222 "20260209",
2223 "--segment",
2224 "100000_300",
2225 "--flush",
2226 ],
2227 day="20260209",
2228 )
2229 assert _flush_state["flushed"] is True
2230
2231 def test_does_not_flush_before_timeout(self):
2232 import time as time_mod
2233 from unittest.mock import MagicMock, patch
2234
2235 from solstone.think.supervisor import _check_segment_flush, _flush_state
2236
2237 _flush_state["last_segment_ts"] = time_mod.time() - 100 # Only 100s ago
2238 _flush_state["day"] = "20260209"
2239 _flush_state["segment"] = "100000_300"
2240 _flush_state["flushed"] = False
2241
2242 mock_queue = MagicMock()
2243 with (
2244 patch("solstone.think.supervisor._task_queue", mock_queue),
2245 patch("solstone.think.supervisor._is_remote_mode", False),
2246 ):
2247 _check_segment_flush()
2248
2249 mock_queue.submit.assert_not_called()
2250 assert _flush_state["flushed"] is False
2251
2252 def test_does_not_flush_twice(self):
2253 import time as time_mod
2254 from unittest.mock import MagicMock, patch
2255
2256 from solstone.think.supervisor import _check_segment_flush, _flush_state
2257
2258 _flush_state["last_segment_ts"] = time_mod.time() - 4000
2259 _flush_state["day"] = "20260209"
2260 _flush_state["segment"] = "100000_300"
2261 _flush_state["flushed"] = True # Already flushed
2262
2263 mock_queue = MagicMock()
2264 with (
2265 patch("solstone.think.supervisor._task_queue", mock_queue),
2266 patch("solstone.think.supervisor._is_remote_mode", False),
2267 ):
2268 _check_segment_flush()
2269
2270 mock_queue.submit.assert_not_called()
2271
2272 def test_skips_in_remote_mode(self):
2273 import time as time_mod
2274 from unittest.mock import MagicMock, patch
2275
2276 from solstone.think.supervisor import _check_segment_flush, _flush_state
2277
2278 _flush_state["last_segment_ts"] = time_mod.time() - 4000
2279 _flush_state["day"] = "20260209"
2280 _flush_state["segment"] = "100000_300"
2281 _flush_state["flushed"] = False
2282
2283 mock_queue = MagicMock()
2284 with (
2285 patch("solstone.think.supervisor._task_queue", mock_queue),
2286 patch("solstone.think.supervisor._is_remote_mode", True),
2287 ):
2288 _check_segment_flush()
2289
2290 mock_queue.submit.assert_not_called()
2291
2292 def test_force_flushes_before_timeout(self):
2293 import time as time_mod
2294 from unittest.mock import MagicMock, patch
2295
2296 from solstone.think.supervisor import _check_segment_flush, _flush_state
2297
2298 # Only 100s ago — would NOT flush normally
2299 _flush_state["last_segment_ts"] = time_mod.time() - 100
2300 _flush_state["day"] = "20260209"
2301 _flush_state["segment"] = "100000_300"
2302 _flush_state["flushed"] = False
2303
2304 mock_queue = MagicMock()
2305 with (
2306 patch("solstone.think.supervisor._task_queue", mock_queue),
2307 patch("solstone.think.supervisor._is_remote_mode", False),
2308 ):
2309 _check_segment_flush(force=True)
2310
2311 mock_queue.submit.assert_called_once()
2312 assert _flush_state["flushed"] is True
2313
2314 def test_segment_observed_resets_flush_state(self):
2315 from solstone.think.supervisor import _flush_state, _handle_segment_observed
2316
2317 _flush_state["flushed"] = True
2318 _flush_state["last_segment_ts"] = 0
2319
2320 # We need to mock the thread start to avoid side effects
2321 from unittest.mock import patch
2322
2323 with patch("solstone.think.supervisor.threading"):
2324 _handle_segment_observed(
2325 {
2326 "tract": "observe",
2327 "event": "observed",
2328 "day": "20260209",
2329 "segment": "110000_300",
2330 }
2331 )
2332
2333 assert _flush_state["flushed"] is False
2334 assert _flush_state["day"] == "20260209"
2335 assert _flush_state["segment"] == "110000_300"
2336 assert _flush_state["last_segment_ts"] > 0
2337
2338
2339def _seed_activity_records(
2340 tmpdir: str, facet: str, day: str, records: list[dict]
2341) -> None:
2342 from solstone.think.activities import append_activity_record
2343
2344 for record in records:
2345 append_activity_record(facet, day, record)
2346
2347
2348def test_make_anticipation_id_builds_stable_id():
2349 from solstone.think.activities import make_anticipation_id
2350
2351 assert make_anticipation_id("meeting", "16:30:00", "2026-04-20") == (
2352 "anticipated_meeting_163000_0420"
2353 )
2354 assert make_anticipation_id("deadline", None, "2026-05-05") == (
2355 "anticipated_deadline_000000_0505"
2356 )
2357
2358
2359@pytest.mark.parametrize(
2360 ("activity_type", "start", "target_date"),
2361 [
2362 ("meeting", "9:00", "2026-04-20"),
2363 ("meeting", "09:00:00", "2026/04/20"),
2364 ("", "09:00:00", "2026-04-20"),
2365 ],
2366)
2367def test_make_anticipation_id_rejects_malformed_inputs(
2368 activity_type,
2369 start,
2370 target_date,
2371):
2372 from solstone.think.activities import make_anticipation_id
2373
2374 with pytest.raises(ValueError):
2375 make_anticipation_id(activity_type, start, target_date)
2376
2377
2378def test_dedup_anticipation_returns_empty_for_first_record(monkeypatch):
2379 from solstone.think.activities import dedup_anticipation
2380
2381 with tempfile.TemporaryDirectory() as tmpdir:
2382 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2383
2384 should_write, superseded_ids = dedup_anticipation(
2385 "work",
2386 "20260420",
2387 {"id": "anticipated_meeting_163000_0420", "title": "Yuri intro"},
2388 )
2389
2390 assert should_write is True
2391 assert superseded_ids == []
2392
2393
2394def test_dedup_anticipation_rejects_exact_id_collision(monkeypatch):
2395 from solstone.think.activities import dedup_anticipation
2396
2397 with tempfile.TemporaryDirectory() as tmpdir:
2398 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2399 _seed_activity_records(
2400 tmpdir,
2401 "work",
2402 "20260420",
2403 [
2404 {
2405 "id": "anticipated_meeting_163000_0420",
2406 "activity": "meeting",
2407 "title": "Yuri intro",
2408 "description": "Original",
2409 "source": "anticipated",
2410 }
2411 ],
2412 )
2413
2414 should_write, superseded_ids = dedup_anticipation(
2415 "work",
2416 "20260420",
2417 {"id": "anticipated_meeting_163000_0420", "title": "Yuri intro"},
2418 )
2419
2420 assert should_write is False
2421 assert superseded_ids == []
2422
2423
2424def test_dedup_anticipation_returns_fuzzy_supersede_matches(monkeypatch):
2425 from solstone.think.activities import dedup_anticipation
2426
2427 with tempfile.TemporaryDirectory() as tmpdir:
2428 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2429 _seed_activity_records(
2430 tmpdir,
2431 "work",
2432 "20260420",
2433 [
2434 {
2435 "id": "anticipated_meeting_160000_0420",
2436 "activity": "meeting",
2437 "title": "Yuri Namikawa intro call",
2438 "description": "Original",
2439 "source": "anticipated",
2440 }
2441 ],
2442 )
2443
2444 should_write, superseded_ids = dedup_anticipation(
2445 "work",
2446 "20260420",
2447 {
2448 "id": "anticipated_meeting_163000_0420",
2449 "title": "Yuri Namikawa intro call",
2450 },
2451 )
2452
2453 assert should_write is True
2454 assert superseded_ids == ["anticipated_meeting_160000_0420"]
2455
2456
2457def test_dedup_anticipation_ignores_below_threshold_and_hidden_rows(monkeypatch):
2458 from solstone.think.activities import dedup_anticipation
2459
2460 with tempfile.TemporaryDirectory() as tmpdir:
2461 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2462 _seed_activity_records(
2463 tmpdir,
2464 "work",
2465 "20260420",
2466 [
2467 {
2468 "id": "anticipated_meeting_090000_0420",
2469 "activity": "meeting",
2470 "title": "Quarterly planning summit",
2471 "description": "Visible",
2472 "source": "anticipated",
2473 },
2474 {
2475 "id": "anticipated_meeting_100000_0420",
2476 "activity": "meeting",
2477 "title": "Yuri Namikawa intro call",
2478 "description": "Hidden",
2479 "source": "anticipated",
2480 "hidden": True,
2481 },
2482 ],
2483 )
2484
2485 should_write, superseded_ids = dedup_anticipation(
2486 "work",
2487 "20260420",
2488 {
2489 "id": "anticipated_meeting_163000_0420",
2490 "title": "Scott Ward standup",
2491 },
2492 )
2493
2494 assert should_write is True
2495 assert superseded_ids == []
2496
2497
2498def test_dedup_anticipation_returns_all_matching_supersedes(monkeypatch):
2499 from solstone.think.activities import dedup_anticipation
2500
2501 with tempfile.TemporaryDirectory() as tmpdir:
2502 monkeypatch.setenv("SOLSTONE_JOURNAL", tmpdir)
2503 _seed_activity_records(
2504 tmpdir,
2505 "work",
2506 "20260420",
2507 [
2508 {
2509 "id": "anticipated_call_090000_0420",
2510 "activity": "call",
2511 "title": "Mari Zumbro intro",
2512 "description": "Old 1",
2513 "source": "anticipated",
2514 },
2515 {
2516 "id": "anticipated_call_093000_0420",
2517 "activity": "call",
2518 "title": "Mari Zumbro intro",
2519 "description": "Old 2",
2520 "source": "anticipated",
2521 },
2522 {
2523 "id": "cogitate_call_100000_300",
2524 "activity": "call",
2525 "title": "Mari Zumbro intro",
2526 "description": "Non-anticipated",
2527 "source": "cogitate",
2528 },
2529 ],
2530 )
2531
2532 should_write, superseded_ids = dedup_anticipation(
2533 "work",
2534 "20260420",
2535 {
2536 "id": "anticipated_call_103000_0420",
2537 "title": "Mari Zumbro intro",
2538 },
2539 )
2540
2541 assert should_write is True
2542 assert superseded_ids == [
2543 "anticipated_call_090000_0420",
2544 "anticipated_call_093000_0420",
2545 ]