personal memory agent
Add activity_state segment generator for tracking facet activities
Implements a multi-facet segment generator that detects configured
activities and tracks their state across segments:
- Pre-hook loads facet's activity list and previous segment state
- Handles 1-hour timeout for activity continuity (fresh start after gaps)
- Outputs JSON with active/ended arrays tracking activity transitions
- Supports concurrent activities and same-type transitions (e.g., one
meeting ends, another starts)
Instructions optimized for focused detection:
- sources: audio+screen only (no agents dependency at priority 95)
- facets: none (pre-hook provides activity list directly)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+841
+125
muse/activity_state.md
+125
muse/activity_state.md
···
1
+
{
2
+
3
+
"title": "Activity State",
4
+
"description": "Detects configured activities present in segment and tracks state across segments",
5
+
"color": "#00897b",
6
+
"schedule": "segment",
7
+
"priority": 95,
8
+
"multi_facet": true,
9
+
"output": "json",
10
+
"hook": {"pre": "activity_state"},
11
+
"tier": 3,
12
+
"thinking_budget": 2048,
13
+
"max_output_tokens": 512,
14
+
"instructions": {
15
+
"sources": {"audio": true, "screen": true, "agents": false},
16
+
"facets": "none"
17
+
}
18
+
19
+
}
20
+
21
+
Detect which of the facet's configured activities are present in this segment.
22
+
23
+
## Input
24
+
25
+
You receive:
26
+
1. **Facet Activities** - The list of activities configured for this facet. Only detect these.
27
+
2. **Previous State** - What activities were active/ended in the prior segment (if available).
28
+
3. **Current Segment Content** - Audio transcript and screen activity from this recording window.
29
+
30
+
## Task
31
+
32
+
Analyze the current segment and determine:
33
+
- Which configured activities are happening at the END of this segment (active)
34
+
- Which previously active activities ended DURING this segment (ended)
35
+
36
+
Consider continuity: if an activity was active in the previous segment and you see evidence of it continuing, keep tracking it with the same `since` value and update the description if context has evolved.
37
+
38
+
## Output Format
39
+
40
+
Return a JSON object with two arrays:
41
+
42
+
```json
43
+
{
44
+
"active": [
45
+
{
46
+
"activity": "meeting",
47
+
"since": "143000_300",
48
+
"description": "Design review with UX team discussing navigation patterns",
49
+
"level": "high"
50
+
}
51
+
],
52
+
"ended": [
53
+
{
54
+
"activity": "email",
55
+
"since": "140000_300",
56
+
"description": "Replied to deployment notification from ops team"
57
+
}
58
+
]
59
+
}
60
+
```
61
+
62
+
### Field Definitions
63
+
64
+
**active** - Activities ongoing at segment end:
65
+
- `activity`: Activity ID from the configured list
66
+
- `since`: Segment key when this activity instance started (copy from previous if continuing, use current segment if new)
67
+
- `description`: Brief description of what this activity involves (update as context evolves)
68
+
- `level`: Engagement level - "high" (primary focus), "medium" (secondary), "low" (background)
69
+
70
+
**ended** - Activities that stopped during this segment:
71
+
- `activity`: Activity ID
72
+
- `since`: When this instance started (for duration tracking)
73
+
- `description`: Final summary of what the activity was
74
+
75
+
## Rules
76
+
77
+
1. **Only detect configured activities** - Ignore activity that doesn't match the facet's list
78
+
2. **One instance per type** - If a meeting ends and another starts, the first goes to `ended`, the new one to `active`
79
+
3. **Preserve `since`** - For continuing activities, keep the original start segment
80
+
4. **Update descriptions** - As activities continue, refine the description with new context
81
+
5. **Empty is valid** - `{"active": [], "ended": []}` is correct when no activities detected
82
+
83
+
## Examples
84
+
85
+
**New activity starts:**
86
+
```json
87
+
{
88
+
"active": [{"activity": "coding", "since": "143500_300", "description": "Implementing user auth flow", "level": "high"}],
89
+
"ended": []
90
+
}
91
+
```
92
+
93
+
**Activity continues from previous:**
94
+
```json
95
+
{
96
+
"active": [{"activity": "meeting", "since": "140000_300", "description": "Sprint planning - now discussing blockers", "level": "high"}],
97
+
"ended": []
98
+
}
99
+
```
100
+
101
+
**One activity ends, another starts (same type):**
102
+
```json
103
+
{
104
+
"active": [{"activity": "meeting", "since": "144500_300", "description": "1:1 with manager", "level": "high"}],
105
+
"ended": [{"activity": "meeting", "since": "140000_300", "description": "Sprint planning completed"}]
106
+
}
107
+
```
108
+
109
+
**Multiple concurrent activities:**
110
+
```json
111
+
{
112
+
"active": [
113
+
{"activity": "meeting", "since": "143000_300", "description": "Team standup", "level": "high"},
114
+
{"activity": "messaging", "since": "143000_300", "description": "Slack thread about deployment", "level": "low"}
115
+
],
116
+
"ended": []
117
+
}
118
+
```
119
+
120
+
**No activities detected:**
121
+
```json
122
+
{"active": [], "ended": []}
123
+
```
124
+
125
+
Return ONLY the JSON object, no other text.
+320
muse/activity_state.py
+320
muse/activity_state.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Pre-hook for activity_state generator.
5
+
6
+
Builds context for activity detection by:
7
+
1. Loading the facet's configured activities
8
+
2. Finding and loading previous segment's activity state
9
+
3. Formatting both as context for the prompt
10
+
"""
11
+
12
+
import json
13
+
import logging
14
+
import os
15
+
import re
16
+
from datetime import datetime, timedelta
17
+
18
+
from think.utils import day_path, segment_parse
19
+
20
+
logger = logging.getLogger(__name__)
21
+
22
+
# Default timeout for activity continuity (1 hour in seconds)
23
+
DEFAULT_TIMEOUT_SECONDS = 3600
24
+
25
+
26
+
def _extract_facet_from_output_path(output_path: str) -> str | None:
27
+
"""Extract facet name from output path.
28
+
29
+
Output paths for faceted generators follow the pattern:
30
+
{day}/{segment}/activity_state_{facet}.json
31
+
32
+
Returns None if facet cannot be extracted.
33
+
"""
34
+
if not output_path:
35
+
return None
36
+
37
+
# Match pattern like activity_state_work.json
38
+
filename = os.path.basename(output_path)
39
+
match = re.match(r"activity_state_([^.]+)\.json$", filename)
40
+
if match:
41
+
return match.group(1)
42
+
return None
43
+
44
+
45
+
def _parse_segment_time(segment: str) -> datetime | None:
46
+
"""Parse segment key to datetime (time only, date is placeholder)."""
47
+
start, _ = segment_parse(segment)
48
+
if start is None:
49
+
return None
50
+
# Convert time to datetime with placeholder date
51
+
return datetime(2000, 1, 1, start.hour, start.minute, start.second)
52
+
53
+
54
+
def _get_segment_end_time(segment: str) -> datetime | None:
55
+
"""Get the end time of a segment."""
56
+
_, end = segment_parse(segment)
57
+
if end is None:
58
+
return None
59
+
return datetime(2000, 1, 1, end.hour, end.minute, end.second)
60
+
61
+
62
+
def find_previous_segment(day: str, current_segment: str) -> str | None:
63
+
"""Find the segment immediately before the current one.
64
+
65
+
Scans the day directory for segment folders, sorts by time,
66
+
and returns the one before current_segment.
67
+
68
+
Returns None if current is the first segment or no segments found.
69
+
"""
70
+
day_dir = day_path(day)
71
+
if not day_dir.is_dir():
72
+
return None
73
+
74
+
# Collect valid segment folders
75
+
segments = []
76
+
for entry in os.listdir(day_dir):
77
+
entry_path = day_dir / entry
78
+
if not entry_path.is_dir():
79
+
continue
80
+
# Match segment pattern: HHMMSS_LEN or HHMMSS_LEN_suffix
81
+
if re.match(r"^\d{6}_\d+", entry):
82
+
segments.append(entry)
83
+
84
+
if not segments:
85
+
return None
86
+
87
+
# Sort by time (segment keys sort lexicographically by time)
88
+
segments.sort()
89
+
90
+
# Find current segment's position
91
+
try:
92
+
current_idx = segments.index(current_segment)
93
+
except ValueError:
94
+
# Current segment not in list - might be new
95
+
# Find where it would be inserted
96
+
for i, seg in enumerate(segments):
97
+
if seg > current_segment:
98
+
current_idx = i
99
+
break
100
+
else:
101
+
current_idx = len(segments)
102
+
103
+
# Return previous if exists
104
+
if current_idx > 0:
105
+
return segments[current_idx - 1]
106
+
return None
107
+
108
+
109
+
def load_previous_state(
110
+
day: str, segment: str, facet: str
111
+
) -> tuple[dict | None, str | None]:
112
+
"""Load activity state from a previous segment.
113
+
114
+
Returns tuple of (state, segment_key) where state is the parsed JSON
115
+
or None if not found/invalid.
116
+
"""
117
+
state_path = day_path(day) / segment / f"activity_state_{facet}.json"
118
+
if not state_path.exists():
119
+
return None, None
120
+
121
+
try:
122
+
content = state_path.read_text().strip()
123
+
if not content:
124
+
return None, segment
125
+
126
+
data = json.loads(content)
127
+
return data, segment
128
+
except (json.JSONDecodeError, OSError) as e:
129
+
logger.warning("Failed to load previous state from %s: %s", state_path, e)
130
+
return None, segment
131
+
132
+
133
+
def check_timeout(
134
+
current_segment: str,
135
+
previous_segment: str,
136
+
timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
137
+
) -> bool:
138
+
"""Check if gap between segments exceeds timeout threshold.
139
+
140
+
Returns True if timed out (gap too large), False if within threshold.
141
+
"""
142
+
current_start = _parse_segment_time(current_segment)
143
+
previous_end = _get_segment_end_time(previous_segment)
144
+
145
+
if current_start is None or previous_end is None:
146
+
# Can't determine times, assume no timeout
147
+
return False
148
+
149
+
gap = current_start - previous_end
150
+
# Handle day wraparound (negative gap means previous was yesterday)
151
+
if gap < timedelta(0):
152
+
return True # Cross-day, always timeout
153
+
154
+
return gap.total_seconds() > timeout_seconds
155
+
156
+
157
+
def format_activities_context(facet: str) -> str:
158
+
"""Format the facet's configured activities as context.
159
+
160
+
Returns markdown-formatted list of activities with descriptions.
161
+
"""
162
+
from think.activities import get_facet_activities
163
+
164
+
activities = get_facet_activities(facet)
165
+
if not activities:
166
+
return "No activities configured for this facet."
167
+
168
+
lines = ["## Facet Activities (only detect these)", ""]
169
+
for activity in activities:
170
+
activity_id = activity.get("id", "")
171
+
name = activity.get("name", activity_id)
172
+
description = activity.get("description", "")
173
+
priority = activity.get("priority", "normal")
174
+
175
+
if priority == "high":
176
+
priority_note = " [high priority]"
177
+
elif priority == "low":
178
+
priority_note = " [low priority]"
179
+
else:
180
+
priority_note = ""
181
+
182
+
if description:
183
+
lines.append(f"- **{activity_id}** ({name}){priority_note}: {description}")
184
+
else:
185
+
lines.append(f"- **{activity_id}** ({name}){priority_note}")
186
+
187
+
return "\n".join(lines)
188
+
189
+
190
+
def format_previous_state(
191
+
state: dict | None,
192
+
segment: str | None,
193
+
current_segment: str,
194
+
timed_out: bool,
195
+
) -> str:
196
+
"""Format previous state as context for the prompt.
197
+
198
+
Args:
199
+
state: Previous segment's activity state dict
200
+
segment: Previous segment key
201
+
current_segment: Current segment key
202
+
timed_out: Whether the gap exceeded timeout threshold
203
+
"""
204
+
if state is None or segment is None:
205
+
return "## Previous State\n\nNo previous segment state available (first segment or fresh start)."
206
+
207
+
if timed_out:
208
+
return "## Previous State\n\nPrevious segment too long ago (>1 hour gap). Starting fresh."
209
+
210
+
# Calculate time gap for context
211
+
current_start = _parse_segment_time(current_segment)
212
+
previous_end = _get_segment_end_time(segment)
213
+
214
+
if current_start and previous_end:
215
+
gap = current_start - previous_end
216
+
gap_minutes = int(gap.total_seconds() / 60)
217
+
time_note = f" ({gap_minutes} minutes ago)"
218
+
else:
219
+
time_note = ""
220
+
221
+
lines = [f"## Previous State (from {segment}){time_note}", ""]
222
+
223
+
active = state.get("active", [])
224
+
ended = state.get("ended", [])
225
+
226
+
if active:
227
+
lines.append("**Active activities (may be continuing):**")
228
+
for item in active:
229
+
activity_id = item.get("activity", "")
230
+
since = item.get("since", "")
231
+
description = item.get("description", "")
232
+
level = item.get("level", "")
233
+
234
+
parts = [f"- {activity_id}"]
235
+
if since:
236
+
parts.append(f"(since {since})")
237
+
if level:
238
+
parts.append(f"[{level}]")
239
+
if description:
240
+
parts.append(f": {description}")
241
+
lines.append(" ".join(parts))
242
+
lines.append("")
243
+
244
+
if ended:
245
+
lines.append("**Recently ended:**")
246
+
for item in ended:
247
+
activity_id = item.get("activity", "")
248
+
description = item.get("description", "")
249
+
lines.append(
250
+
f"- {activity_id}: {description}" if description else f"- {activity_id}"
251
+
)
252
+
lines.append("")
253
+
254
+
if not active and not ended:
255
+
lines.append("No activities were detected in the previous segment.")
256
+
257
+
return "\n".join(lines)
258
+
259
+
260
+
def pre_process(context: dict) -> dict | None:
261
+
"""Build enriched context for activity state detection.
262
+
263
+
Args:
264
+
context: PreHookContext with day, segment, output_path, transcript, meta
265
+
266
+
Returns:
267
+
Dict with modified transcript containing activity context,
268
+
or None if unable to process.
269
+
"""
270
+
day = context.get("day")
271
+
segment = context.get("segment")
272
+
output_path = context.get("output_path", "")
273
+
transcript = context.get("transcript", "")
274
+
meta = context.get("meta", {})
275
+
276
+
if not day or not segment:
277
+
logger.warning("activity_state pre-hook requires day and segment")
278
+
return None
279
+
280
+
# Extract facet from output path
281
+
facet = _extract_facet_from_output_path(output_path)
282
+
if not facet:
283
+
logger.warning("Could not extract facet from output_path: %s", output_path)
284
+
return None
285
+
286
+
# Get timeout from meta or use default
287
+
timeout_seconds = meta.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS)
288
+
289
+
# Build activity context
290
+
activities_context = format_activities_context(facet)
291
+
292
+
# Find and load previous segment state
293
+
previous_segment = find_previous_segment(day, segment)
294
+
previous_state = None
295
+
timed_out = False
296
+
297
+
if previous_segment:
298
+
# Check timeout
299
+
timed_out = check_timeout(segment, previous_segment, timeout_seconds)
300
+
301
+
if not timed_out:
302
+
previous_state, _ = load_previous_state(day, previous_segment, facet)
303
+
304
+
# Format previous state context
305
+
previous_context = format_previous_state(
306
+
previous_state, previous_segment, segment, timed_out
307
+
)
308
+
309
+
# Build enriched transcript
310
+
enriched_parts = [
311
+
activities_context,
312
+
"",
313
+
previous_context,
314
+
"",
315
+
"## Current Segment Content",
316
+
"",
317
+
transcript,
318
+
]
319
+
320
+
return {"transcript": "\n".join(enriched_parts)}
+396
tests/test_activity_state.py
+396
tests/test_activity_state.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Tests for the activity_state pre-hook module."""
5
+
6
+
import json
7
+
import os
8
+
import tempfile
9
+
from pathlib import Path
10
+
11
+
# Set up test environment before importing the module
12
+
os.environ["JOURNAL_PATH"] = "fixtures/journal"
13
+
14
+
15
+
class TestExtractFacetFromOutputPath:
16
+
"""Tests for _extract_facet_from_output_path."""
17
+
18
+
def test_extracts_facet_from_valid_path(self):
19
+
from muse.activity_state import _extract_facet_from_output_path
20
+
21
+
path = "/journal/20260130/143000_300/activity_state_work.json"
22
+
assert _extract_facet_from_output_path(path) == "work"
23
+
24
+
def test_extracts_facet_with_hyphen(self):
25
+
from muse.activity_state import _extract_facet_from_output_path
26
+
27
+
path = "/journal/20260130/143000_300/activity_state_my-project.json"
28
+
assert _extract_facet_from_output_path(path) == "my-project"
29
+
30
+
def test_returns_none_for_empty_path(self):
31
+
from muse.activity_state import _extract_facet_from_output_path
32
+
33
+
assert _extract_facet_from_output_path("") is None
34
+
assert _extract_facet_from_output_path(None) is None
35
+
36
+
def test_returns_none_for_non_matching_path(self):
37
+
from muse.activity_state import _extract_facet_from_output_path
38
+
39
+
# Different generator name
40
+
assert _extract_facet_from_output_path("/path/to/facets.json") is None
41
+
# No facet suffix
42
+
assert _extract_facet_from_output_path("/path/to/activity_state.json") is None
43
+
44
+
45
+
class TestFindPreviousSegment:
46
+
"""Tests for find_previous_segment."""
47
+
48
+
def test_finds_previous_segment(self):
49
+
from muse.activity_state import find_previous_segment
50
+
51
+
with tempfile.TemporaryDirectory() as tmpdir:
52
+
original_path = os.environ.get("JOURNAL_PATH")
53
+
os.environ["JOURNAL_PATH"] = tmpdir
54
+
55
+
try:
56
+
# Create day directory with segments
57
+
day_dir = Path(tmpdir) / "20260130"
58
+
day_dir.mkdir()
59
+
(day_dir / "100000_300").mkdir()
60
+
(day_dir / "110000_300").mkdir()
61
+
(day_dir / "120000_300").mkdir()
62
+
63
+
# Test finding previous
64
+
assert find_previous_segment("20260130", "120000_300") == "110000_300"
65
+
assert find_previous_segment("20260130", "110000_300") == "100000_300"
66
+
assert find_previous_segment("20260130", "100000_300") is None
67
+
68
+
finally:
69
+
if original_path:
70
+
os.environ["JOURNAL_PATH"] = original_path
71
+
72
+
def test_returns_none_for_nonexistent_day(self):
73
+
from muse.activity_state import find_previous_segment
74
+
75
+
with tempfile.TemporaryDirectory() as tmpdir:
76
+
original_path = os.environ.get("JOURNAL_PATH")
77
+
os.environ["JOURNAL_PATH"] = tmpdir
78
+
79
+
try:
80
+
assert find_previous_segment("20260130", "100000_300") is None
81
+
finally:
82
+
if original_path:
83
+
os.environ["JOURNAL_PATH"] = original_path
84
+
85
+
def test_handles_segments_with_suffix(self):
86
+
from muse.activity_state import find_previous_segment
87
+
88
+
with tempfile.TemporaryDirectory() as tmpdir:
89
+
original_path = os.environ.get("JOURNAL_PATH")
90
+
os.environ["JOURNAL_PATH"] = tmpdir
91
+
92
+
try:
93
+
day_dir = Path(tmpdir) / "20260130"
94
+
day_dir.mkdir()
95
+
(day_dir / "100000_300_audio").mkdir()
96
+
(day_dir / "110000_300").mkdir()
97
+
98
+
# Should still find previous
99
+
assert (
100
+
find_previous_segment("20260130", "110000_300")
101
+
== "100000_300_audio"
102
+
)
103
+
104
+
finally:
105
+
if original_path:
106
+
os.environ["JOURNAL_PATH"] = original_path
107
+
108
+
109
+
class TestCheckTimeout:
110
+
"""Tests for check_timeout."""
111
+
112
+
def test_no_timeout_within_threshold(self):
113
+
from muse.activity_state import check_timeout
114
+
115
+
# 5 minute gap (300 seconds)
116
+
assert check_timeout("100500_300", "100000_300", timeout_seconds=3600) is False
117
+
118
+
def test_timeout_exceeds_threshold(self):
119
+
from muse.activity_state import check_timeout
120
+
121
+
# 2 hour gap
122
+
assert check_timeout("120000_300", "100000_300", timeout_seconds=3600) is True
123
+
124
+
def test_uses_segment_end_time(self):
125
+
from muse.activity_state import check_timeout
126
+
127
+
# Previous segment: 10:00:00 - 10:05:00 (300 seconds)
128
+
# Current segment: 10:10:00
129
+
# Gap should be 5 minutes (10:10:00 - 10:05:00)
130
+
assert check_timeout("101000_300", "100000_300", timeout_seconds=600) is False
131
+
132
+
133
+
class TestLoadPreviousState:
134
+
"""Tests for load_previous_state."""
135
+
136
+
def test_loads_valid_state(self):
137
+
from muse.activity_state import load_previous_state
138
+
139
+
with tempfile.TemporaryDirectory() as tmpdir:
140
+
original_path = os.environ.get("JOURNAL_PATH")
141
+
os.environ["JOURNAL_PATH"] = tmpdir
142
+
143
+
try:
144
+
# Create state file
145
+
segment_dir = Path(tmpdir) / "20260130" / "100000_300"
146
+
segment_dir.mkdir(parents=True)
147
+
148
+
state = {
149
+
"active": [
150
+
{
151
+
"activity": "meeting",
152
+
"since": "100000_300",
153
+
"description": "Standup",
154
+
"level": "high",
155
+
}
156
+
],
157
+
"ended": [],
158
+
}
159
+
(segment_dir / "activity_state_work.json").write_text(json.dumps(state))
160
+
161
+
loaded, segment = load_previous_state("20260130", "100000_300", "work")
162
+
assert segment == "100000_300"
163
+
assert loaded["active"][0]["activity"] == "meeting"
164
+
165
+
finally:
166
+
if original_path:
167
+
os.environ["JOURNAL_PATH"] = original_path
168
+
169
+
def test_returns_none_for_missing_file(self):
170
+
from muse.activity_state import load_previous_state
171
+
172
+
with tempfile.TemporaryDirectory() as tmpdir:
173
+
original_path = os.environ.get("JOURNAL_PATH")
174
+
os.environ["JOURNAL_PATH"] = tmpdir
175
+
176
+
try:
177
+
segment_dir = Path(tmpdir) / "20260130" / "100000_300"
178
+
segment_dir.mkdir(parents=True)
179
+
180
+
loaded, segment = load_previous_state("20260130", "100000_300", "work")
181
+
assert loaded is None
182
+
assert segment is None
183
+
184
+
finally:
185
+
if original_path:
186
+
os.environ["JOURNAL_PATH"] = original_path
187
+
188
+
189
+
class TestFormatActivitiesContext:
190
+
"""Tests for format_activities_context."""
191
+
192
+
def test_formats_activities_list(self):
193
+
from muse.activity_state import format_activities_context
194
+
195
+
with tempfile.TemporaryDirectory() as tmpdir:
196
+
original_path = os.environ.get("JOURNAL_PATH")
197
+
os.environ["JOURNAL_PATH"] = tmpdir
198
+
199
+
try:
200
+
# Create facet with activities
201
+
facet_dir = Path(tmpdir) / "facets" / "work" / "activities"
202
+
facet_dir.mkdir(parents=True)
203
+
204
+
activities = [
205
+
{"id": "meeting"},
206
+
{"id": "coding", "priority": "high"},
207
+
]
208
+
(facet_dir / "activities.jsonl").write_text(
209
+
"\n".join(json.dumps(a) for a in activities)
210
+
)
211
+
212
+
result = format_activities_context("work")
213
+
assert "## Facet Activities" in result
214
+
assert "meeting" in result
215
+
assert "coding" in result
216
+
assert "[high priority]" in result
217
+
218
+
finally:
219
+
if original_path:
220
+
os.environ["JOURNAL_PATH"] = original_path
221
+
222
+
def test_handles_empty_activities(self):
223
+
from muse.activity_state import format_activities_context
224
+
225
+
with tempfile.TemporaryDirectory() as tmpdir:
226
+
original_path = os.environ.get("JOURNAL_PATH")
227
+
os.environ["JOURNAL_PATH"] = tmpdir
228
+
229
+
try:
230
+
# Create facet without activities
231
+
facet_dir = Path(tmpdir) / "facets" / "work"
232
+
facet_dir.mkdir(parents=True)
233
+
234
+
result = format_activities_context("work")
235
+
assert "No activities configured" in result
236
+
237
+
finally:
238
+
if original_path:
239
+
os.environ["JOURNAL_PATH"] = original_path
240
+
241
+
242
+
class TestFormatPreviousState:
243
+
"""Tests for format_previous_state."""
244
+
245
+
def test_formats_active_activities(self):
246
+
from muse.activity_state import format_previous_state
247
+
248
+
state = {
249
+
"active": [
250
+
{
251
+
"activity": "meeting",
252
+
"since": "100000_300",
253
+
"description": "Team standup",
254
+
"level": "high",
255
+
}
256
+
],
257
+
"ended": [],
258
+
}
259
+
260
+
result = format_previous_state(
261
+
state, "100000_300", "100500_300", timed_out=False
262
+
)
263
+
assert "Previous State" in result
264
+
assert "meeting" in result
265
+
assert "since 100000_300" in result
266
+
assert "Team standup" in result
267
+
268
+
def test_formats_ended_activities(self):
269
+
from muse.activity_state import format_previous_state
270
+
271
+
state = {
272
+
"active": [],
273
+
"ended": [
274
+
{
275
+
"activity": "email",
276
+
"since": "093000_300",
277
+
"description": "Replied to boss",
278
+
}
279
+
],
280
+
}
281
+
282
+
result = format_previous_state(
283
+
state, "100000_300", "100500_300", timed_out=False
284
+
)
285
+
assert "Recently ended" in result
286
+
assert "email" in result
287
+
288
+
def test_handles_timeout(self):
289
+
from muse.activity_state import format_previous_state
290
+
291
+
state = {"active": [{"activity": "meeting"}], "ended": []}
292
+
result = format_previous_state(
293
+
state, "100000_300", "120000_300", timed_out=True
294
+
)
295
+
assert "Starting fresh" in result
296
+
assert "meeting" not in result
297
+
298
+
def test_handles_no_previous_state(self):
299
+
from muse.activity_state import format_previous_state
300
+
301
+
result = format_previous_state(None, None, "100000_300", timed_out=False)
302
+
assert "No previous segment state" in result
303
+
304
+
305
+
class TestPreProcess:
306
+
"""Tests for the pre_process hook function."""
307
+
308
+
def test_builds_enriched_context(self):
309
+
from muse.activity_state import pre_process
310
+
311
+
with tempfile.TemporaryDirectory() as tmpdir:
312
+
original_path = os.environ.get("JOURNAL_PATH")
313
+
os.environ["JOURNAL_PATH"] = tmpdir
314
+
315
+
try:
316
+
# Create day and segments
317
+
day_dir = Path(tmpdir) / "20260130"
318
+
day_dir.mkdir()
319
+
(day_dir / "100000_300").mkdir()
320
+
segment_dir = day_dir / "110000_300"
321
+
segment_dir.mkdir()
322
+
323
+
# Create facet with activities
324
+
facet_dir = Path(tmpdir) / "facets" / "work" / "activities"
325
+
facet_dir.mkdir(parents=True)
326
+
(facet_dir / "activities.jsonl").write_text(
327
+
'{"id": "meeting"}\n{"id": "coding"}'
328
+
)
329
+
330
+
# Create previous state
331
+
prev_state = {
332
+
"active": [
333
+
{
334
+
"activity": "meeting",
335
+
"since": "100000_300",
336
+
"description": "Standup",
337
+
"level": "high",
338
+
}
339
+
],
340
+
"ended": [],
341
+
}
342
+
(day_dir / "100000_300" / "activity_state_work.json").write_text(
343
+
json.dumps(prev_state)
344
+
)
345
+
346
+
context = {
347
+
"day": "20260130",
348
+
"segment": "110000_300",
349
+
"output_path": "/journal/20260130/110000_300/activity_state_work.json",
350
+
"transcript": "User is typing code...",
351
+
"meta": {},
352
+
}
353
+
354
+
result = pre_process(context)
355
+
assert result is not None
356
+
assert "transcript" in result
357
+
358
+
transcript = result["transcript"]
359
+
assert "## Facet Activities" in transcript
360
+
assert "meeting" in transcript
361
+
assert "## Previous State" in transcript
362
+
assert "Standup" in transcript
363
+
assert "## Current Segment Content" in transcript
364
+
assert "User is typing code" in transcript
365
+
366
+
finally:
367
+
if original_path:
368
+
os.environ["JOURNAL_PATH"] = original_path
369
+
370
+
def test_returns_none_without_day(self):
371
+
from muse.activity_state import pre_process
372
+
373
+
context = {
374
+
"segment": "100000_300",
375
+
"output_path": "/path/to/activity_state_work.json",
376
+
}
377
+
assert pre_process(context) is None
378
+
379
+
def test_returns_none_without_segment(self):
380
+
from muse.activity_state import pre_process
381
+
382
+
context = {
383
+
"day": "20260130",
384
+
"output_path": "/path/to/activity_state_work.json",
385
+
}
386
+
assert pre_process(context) is None
387
+
388
+
def test_returns_none_without_facet_in_path(self):
389
+
from muse.activity_state import pre_process
390
+
391
+
context = {
392
+
"day": "20260130",
393
+
"segment": "100000_300",
394
+
"output_path": "/path/to/something_else.json",
395
+
}
396
+
assert pre_process(context) is None