personal memory agent
Fix audit issues: docs, duplicate test, type annotations
- Update docs/APPS.md with dict syntax for selective agent filtering
- Remove duplicate test_source_helpers from test_generators.py
- Standardize type annotations in think/cluster.py (Dict→dict, List→list, etc.)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+34
-48
+2
-1
docs/APPS.md
+2
-1
docs/APPS.md
···
372
372
- `false` - don't load this source type
373
373
- `true` - load if available
374
374
- `"required"` - load, and skip generation if no content found (useful for generators that only make sense with specific input types, e.g., `"audio": "required"` for speaker detection)
375
+
- For `agents` only: a dict for selective filtering, e.g., `{"entities": true, "meetings": "required", "flow": false}`. Keys are agent names (system) or `"app:topic"` (app-namespaced). An empty dict `{}` means no agents.
375
376
376
-
**Authoritative source:** `think/utils.py` - `compose_instructions()`, `_DEFAULT_INSTRUCTIONS`, `source_is_enabled()`, `source_is_required()`
377
+
**Authoritative source:** `think/utils.py` - `compose_instructions()`, `_DEFAULT_INSTRUCTIONS`, `source_is_enabled()`, `source_is_required()`, `get_agent_filter()`
377
378
378
379
---
379
380
-15
tests/test_generators.py
-15
tests/test_generators.py
···
126
126
), f"Generator '{key}' has invalid schedule '{sched}'"
127
127
128
128
129
-
def test_source_helpers():
130
-
"""Test source_is_enabled and source_is_required helper functions."""
131
-
utils = importlib.import_module("think.utils")
132
-
133
-
# source_is_enabled: True for True and "required"
134
-
assert utils.source_is_enabled(True) is True
135
-
assert utils.source_is_enabled("required") is True
136
-
assert utils.source_is_enabled(False) is False
137
-
138
-
# source_is_required: True only for "required"
139
-
assert utils.source_is_required("required") is True
140
-
assert utils.source_is_required(True) is False
141
-
assert utils.source_is_required(False) is False
142
-
143
-
144
129
def test_speakers_has_required_audio():
145
130
"""Test that speakers generator has audio as required source."""
146
131
utils = importlib.import_module("think.utils")
+32
-32
think/cluster.py
+32
-32
think/cluster.py
···
8
8
from collections import Counter, defaultdict
9
9
from datetime import datetime, timedelta
10
10
from pathlib import Path
11
-
from typing import Any, Dict, List, Set, Tuple
11
+
from typing import Any
12
12
13
13
from observe.screen import format_screen_text
14
14
···
42
42
43
43
44
44
def _agent_matches_filter(
45
-
filename: str, agent_filter: Dict[str, bool | str] | None
45
+
filename: str, agent_filter: dict[str, bool | str] | None
46
46
) -> bool:
47
47
"""Check if an agent output file matches the filter.
48
48
···
76
76
date_str: str,
77
77
audio: bool,
78
78
screen: bool,
79
-
agents: bool | Dict[str, bool | str],
80
-
) -> List[Dict[str, Any]]:
79
+
agents: bool | dict[str, bool | str],
80
+
) -> list[dict[str, Any]]:
81
81
"""Process a single segment directory and return entries.
82
82
83
83
Args:
···
94
94
"""
95
95
from think.utils import segment_parse
96
96
97
-
entries: List[Dict[str, Any]] = []
97
+
entries: list[dict[str, Any]] = []
98
98
99
99
start_time, end_time = segment_parse(segment_path.name)
100
100
if not start_time or not end_time:
···
198
198
199
199
200
200
def _load_entries(
201
-
day_dir: str, audio: bool, screen: bool, agents: bool | Dict[str, bool | str]
202
-
) -> List[Dict[str, Any]]:
201
+
day_dir: str, audio: bool, screen: bool, agents: bool | dict[str, bool | str]
202
+
) -> list[dict[str, Any]]:
203
203
"""Load all transcript entries from a day directory."""
204
204
from think.utils import segment_parse
205
205
206
206
date_str = _date_str(day_dir)
207
-
entries: List[Dict[str, Any]] = []
207
+
entries: list[dict[str, Any]] = []
208
208
day_path_obj = Path(day_dir)
209
209
210
210
for item in day_path_obj.iterdir():
···
218
218
219
219
220
220
def _group_entries(
221
-
entries: List[Dict[str, Any]],
222
-
) -> Dict[str, List[Dict[str, Any]]]:
221
+
entries: list[dict[str, Any]],
222
+
) -> dict[str, list[dict[str, Any]]]:
223
223
"""Group entries by segment key.
224
224
225
225
Returns dict mapping segment_key to list of entries for that segment.
226
226
"""
227
-
grouped: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
227
+
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
228
228
for e in entries:
229
229
grouped[e["segment_key"]].append(e)
230
230
return grouped
231
231
232
232
233
-
def _count_by_source(entries: List[Dict[str, Any]]) -> Dict[str, int]:
233
+
def _count_by_source(entries: list[dict[str, Any]]) -> dict[str, int]:
234
234
"""Count entries by source type (prefix).
235
235
236
236
Maps the internal prefix names to source config names:
···
258
258
}
259
259
260
260
261
-
def _groups_to_markdown(groups: Dict[str, List[Dict[str, Any]]]) -> str:
261
+
def _groups_to_markdown(groups: dict[str, list[dict[str, Any]]]) -> str:
262
262
"""Render grouped entries as markdown with segment-based headers."""
263
-
lines: List[str] = []
263
+
lines: list[str] = []
264
264
265
265
# Sort by segment start time (entries within each group have same segment_start)
266
266
def sort_key(segment_key: str) -> datetime:
···
298
298
return "\n".join(lines)
299
299
300
300
301
-
def _slots_to_ranges(slots: List[datetime]) -> List[Tuple[str, str]]:
301
+
def _slots_to_ranges(slots: list[datetime]) -> list[tuple[str, str]]:
302
302
"""Collapse 15-minute slots into start/end pairs.
303
303
304
304
Args:
···
309
309
contiguous 15-minute ranges.
310
310
"""
311
311
312
-
ranges: List[Tuple[str, str]] = []
312
+
ranges: list[tuple[str, str]] = []
313
313
if not slots:
314
314
return ranges
315
315
···
330
330
return ranges
331
331
332
332
333
-
def cluster_scan(day: str) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
333
+
def cluster_scan(day: str) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]:
334
334
"""Return 15-minute ranges with audio and screen transcripts for ``day``.
335
335
336
336
Args:
···
347
347
return [], []
348
348
349
349
date_str = _date_str(day_dir)
350
-
audio_slots: Set[datetime] = set()
351
-
screen_slots: Set[datetime] = set()
350
+
audio_slots: set[datetime] = set()
351
+
screen_slots: set[datetime] = set()
352
352
day_path_obj = Path(day_dir)
353
353
354
354
# Check timestamp subdirectories for transcript files
···
377
377
return audio_ranges, screen_ranges
378
378
379
379
380
-
def cluster_segments(day: str) -> List[Dict[str, Any]]:
380
+
def cluster_segments(day: str) -> list[dict[str, Any]]:
381
381
"""Return individual recording segments for a day with their content types.
382
382
383
383
Unlike ``cluster_scan()`` which collapses segments into 15-minute ranges,
···
400
400
return []
401
401
402
402
day_path_obj = Path(day_dir)
403
-
segments: List[Dict[str, Any]] = []
403
+
segments: list[dict[str, Any]] = []
404
404
405
405
for item in day_path_obj.iterdir():
406
406
start_time, end_time = segment_parse(item.name)
···
438
438
439
439
def cluster(
440
440
day: str,
441
-
sources: Dict[str, bool | str | Dict],
442
-
) -> Tuple[str, Dict[str, int]]:
441
+
sources: dict[str, bool | str | dict],
442
+
) -> tuple[str, dict[str, int]]:
443
443
"""Return Markdown summary for one day's JSON files and counts by source.
444
444
445
445
Args:
···
480
480
def cluster_period(
481
481
day: str,
482
482
segment: str,
483
-
sources: Dict[str, bool | str | Dict],
484
-
) -> Tuple[str, Dict[str, int]]:
483
+
sources: dict[str, bool | str | dict],
484
+
) -> tuple[str, dict[str, int]]:
485
485
"""Return Markdown summary for one segment's JSON files and counts by source.
486
486
487
487
Args:
···
517
517
518
518
519
519
def _load_entries_from_segment(
520
-
segment_dir: str, audio: bool, screen: bool, agents: bool | Dict[str, bool | str]
521
-
) -> List[Dict[str, Any]]:
520
+
segment_dir: str, audio: bool, screen: bool, agents: bool | dict[str, bool | str]
521
+
) -> list[dict[str, Any]]:
522
522
"""Load entries from a single segment directory.
523
523
524
524
Args:
···
539
539
540
540
def cluster_span(
541
541
day: str,
542
-
span: List[str],
543
-
sources: Dict[str, bool | str | Dict],
544
-
) -> Tuple[str, Dict[str, int]]:
542
+
span: list[str],
543
+
sources: dict[str, bool | str | dict],
544
+
) -> tuple[str, dict[str, int]]:
545
545
"""Return Markdown summary for a span of segments and counts by source.
546
546
547
547
A span is a list of sequential segment keys (e.g., from an import that created
···
576
576
raise ValueError(f"Segment directories not found: {', '.join(missing)}")
577
577
578
578
# Load entries from all segments in span
579
-
entries: List[Dict[str, Any]] = []
579
+
entries: list[dict[str, Any]] = []
580
580
for segment_key in span:
581
581
segment_dir = Path(day_dir) / segment_key
582
582
segment_entries = _load_entries_from_segment(
···
614
614
day: str,
615
615
start: str,
616
616
end: str,
617
-
sources: Dict[str, bool | str | Dict],
617
+
sources: dict[str, bool | str | dict],
618
618
) -> str:
619
619
"""Return markdown for ``day`` limited to ``start``-``end`` (HHMMSS).
620
620