personal memory agent
Add daily_schedule generator to find optimal maintenance windows
Introduces a new generator that analyzes journal activity patterns over
the past 7 days to identify the best time for scheduled maintenance tasks.
Uses a pre-hook to generate span summaries from segment data, replacing
the normal transcript clustering with structured activity window data.
- muse/daily_schedule.py: Pre-hook that scans segments and builds spans
- muse/daily_schedule.md: Prompt for JSON output with primary/fallback times
- think/models.py: Add TIER_LITE context for agent.daily_schedule.*
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+228
+50
muse/daily_schedule.md
+50
muse/daily_schedule.md
···
1
+
{
2
+
3
+
"title": "Maintenance Window",
4
+
"description": "Analyzes activity patterns to identify optimal times for scheduled maintenance tasks.",
5
+
"schedule": "daily",
6
+
"output": "json",
7
+
"hook": {"pre": "daily_schedule"},
8
+
"color": "#455a64",
9
+
"thinking_budget": 4096,
10
+
"max_output_tokens": 512
11
+
12
+
}
13
+
14
+
# Maintenance Window Analysis
15
+
16
+
You are given a summary of when the user was active over the past week. Each day lists time windows when activity was recorded. Times not listed represent periods of inactivity.
17
+
18
+
## Task
19
+
20
+
Find the best time to schedule daily maintenance tasks (backups, syncs, cleanups). The ideal window is:
21
+
22
+
1. **Consistent** - Inactive at that time across most or all observed days
23
+
2. **Long** - A larger gap is better than a smaller one
24
+
3. **Reliable** - Prefer times that are consistently inactive over times that vary
25
+
26
+
## Analysis Steps
27
+
28
+
1. Map out the 24-hour day and mark which hours show activity on each day
29
+
2. Identify hours that are consistently inactive across all days
30
+
3. Find the largest contiguous block of inactive hours
31
+
4. Select the midpoint of that block as the primary time
32
+
5. Select an alternate time in a different inactive block if available
33
+
34
+
## Output
35
+
36
+
Return ONLY a JSON object with this exact structure:
37
+
38
+
```json
39
+
{
40
+
"primary": "HH:MM",
41
+
"fallback": "HH:MM"
42
+
}
43
+
```
44
+
45
+
- Use 24-hour format (00:00 to 23:59)
46
+
- Primary should be in the largest consistent inactive window
47
+
- Fallback should be in a different inactive period, or 1 hour offset from primary if only one window exists
48
+
- If activity covers the entire day on all days, use "03:00" as primary and "04:00" as fallback
49
+
50
+
Return ONLY the JSON object, no other text.
+173
muse/daily_schedule.py
+173
muse/daily_schedule.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Pre-hook for daily_schedule generator.
5
+
6
+
Generates activity span data from journal segments to identify optimal
7
+
maintenance windows when the user is consistently inactive.
8
+
"""
9
+
10
+
import os
11
+
import re
12
+
from datetime import datetime, timedelta
13
+
14
+
from think.utils import get_journal
15
+
16
+
17
+
def _parse_segment(folder_name: str) -> tuple[datetime, int] | None:
18
+
"""Parse segment folder name into start time and duration."""
19
+
match = re.match(r"^(\d{6})_(\d+)(?:_\w+)?$", folder_name)
20
+
if not match:
21
+
return None
22
+
23
+
time_str, duration_str = match.groups()
24
+
try:
25
+
hour = int(time_str[0:2])
26
+
minute = int(time_str[2:4])
27
+
second = int(time_str[4:6])
28
+
duration = int(duration_str)
29
+
start = datetime(2000, 1, 1, hour, minute, second)
30
+
return start, duration
31
+
except ValueError:
32
+
return None
33
+
34
+
35
+
def _get_segments(day_path: str) -> list[tuple[datetime, datetime]]:
36
+
"""Get sorted list of segment time ranges for a day."""
37
+
segments = []
38
+
39
+
if not os.path.isdir(day_path):
40
+
return segments
41
+
42
+
for entry in os.listdir(day_path):
43
+
entry_path = os.path.join(day_path, entry)
44
+
if not os.path.isdir(entry_path):
45
+
continue
46
+
47
+
parsed = _parse_segment(entry)
48
+
if parsed is None:
49
+
continue
50
+
51
+
start, duration = parsed
52
+
end = start + timedelta(seconds=duration)
53
+
segments.append((start, end))
54
+
55
+
segments.sort(key=lambda x: x[0])
56
+
return segments
57
+
58
+
59
+
def _build_spans(
60
+
segments: list[tuple[datetime, datetime]],
61
+
gap_seconds: int = 300,
62
+
min_minutes: int = 10,
63
+
) -> list[tuple[datetime, datetime]]:
64
+
"""Group segments into spans based on gap threshold."""
65
+
if not segments:
66
+
return []
67
+
68
+
spans = []
69
+
span_start, span_end = segments[0]
70
+
71
+
for seg_start, seg_end in segments[1:]:
72
+
gap = (seg_start - span_end).total_seconds()
73
+
74
+
if gap > gap_seconds:
75
+
duration_minutes = (span_end - span_start).total_seconds() / 60
76
+
if duration_minutes >= min_minutes:
77
+
spans.append((span_start, span_end))
78
+
span_start = seg_start
79
+
80
+
span_end = seg_end
81
+
82
+
duration_minutes = (span_end - span_start).total_seconds() / 60
83
+
if duration_minutes >= min_minutes:
84
+
spans.append((span_start, span_end))
85
+
86
+
return spans
87
+
88
+
89
+
def _format_time(dt: datetime) -> str:
90
+
"""Format datetime as HH:MM."""
91
+
return dt.strftime("%H:%M")
92
+
93
+
94
+
def _format_duration(start: datetime, end: datetime) -> str:
95
+
"""Format duration as Xh Ym."""
96
+
total_minutes = int((end - start).total_seconds() / 60)
97
+
hours, minutes = divmod(total_minutes, 60)
98
+
if hours > 0 and minutes > 0:
99
+
return f"{hours}h {minutes}m"
100
+
if hours > 0:
101
+
return f"{hours}h"
102
+
return f"{minutes}m"
103
+
104
+
105
+
def _get_weekday(date_str: str) -> str:
106
+
"""Get weekday name from YYYYMMDD string."""
107
+
dt = datetime.strptime(date_str, "%Y%m%d")
108
+
return dt.strftime("%A")
109
+
110
+
111
+
def generate_span_summary(days: int = 7) -> str:
112
+
"""Generate activity span summary for the past N days.
113
+
114
+
Args:
115
+
days: Number of days of history to analyze.
116
+
117
+
Returns:
118
+
Formatted text summarizing activity windows per day.
119
+
"""
120
+
journal = get_journal()
121
+
lines = []
122
+
123
+
end_date = datetime.now()
124
+
start_date = end_date - timedelta(days=days - 1)
125
+
126
+
current = start_date
127
+
days_with_data = 0
128
+
129
+
while current <= end_date:
130
+
day_str = current.strftime("%Y%m%d")
131
+
day_path = os.path.join(journal, day_str)
132
+
133
+
segments = _get_segments(day_path)
134
+
spans = _build_spans(segments)
135
+
136
+
if spans:
137
+
days_with_data += 1
138
+
weekday = _get_weekday(day_str)
139
+
lines.append(f"{day_str} ({weekday}):")
140
+
141
+
for span_start, span_end in spans:
142
+
duration = _format_duration(span_start, span_end)
143
+
lines.append(
144
+
f" {_format_time(span_start)} - {_format_time(span_end)} ({duration})"
145
+
)
146
+
147
+
lines.append("")
148
+
149
+
current += timedelta(days=1)
150
+
151
+
if days_with_data == 0:
152
+
return "No activity data found for the past week."
153
+
154
+
header = f"Activity windows for the past {days} days ({days_with_data} days with data):\n\n"
155
+
return header + "\n".join(lines)
156
+
157
+
158
+
def pre_process(context: dict) -> dict | None:
159
+
"""Generate span data to replace transcript content.
160
+
161
+
Args:
162
+
context: PreHookContext with day, meta, etc.
163
+
164
+
Returns:
165
+
Dict with transcript replacement, or None if insufficient data.
166
+
"""
167
+
# Get lookback window from meta, default 7 days
168
+
meta = context.get("meta", {})
169
+
days = meta.get("lookback_days", 7)
170
+
171
+
span_summary = generate_span_summary(days=days)
172
+
173
+
return {"transcript": span_summary}
+5
think/models.py
+5
think/models.py