personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import importlib
5import os
6import uuid
7from pathlib import Path
8
9import pytest
10
11
12def test_get_talent_configs_generators():
13 """Test that system generators are discovered with source field."""
14 talent = importlib.import_module("think.talent")
15 generators = talent.get_talent_configs(type="generate")
16 assert "flow" in generators
17 info = generators["flow"]
18 assert os.path.basename(info["path"]) == "flow.md"
19 assert isinstance(info["color"], str)
20 assert isinstance(info["mtime"], int)
21 assert "title" in info
22 assert "occurrences" in info
23 # New: check source field
24 assert info.get("source") == "system"
25
26
27def test_get_output_name():
28 """Test generator key to filename conversion."""
29 talent = importlib.import_module("think.talent")
30
31 # System generators: key unchanged
32 assert talent.get_output_name("activity") == "activity"
33 assert talent.get_output_name("flow") == "flow"
34
35 # App generators: _app_name format
36 assert talent.get_output_name("chat:sentiment") == "_chat_sentiment"
37 assert talent.get_output_name("my_app:weekly_summary") == "_my_app_weekly_summary"
38
39
40def test_get_talent_configs_app_discovery(tmp_path, monkeypatch):
41 """Test that app generators are discovered from apps/*/talent/."""
42 talent = importlib.import_module("think.talent")
43
44 # Create a fake app with a generator
45 app_dir = tmp_path / "apps" / "test_app" / "talent"
46 app_dir.mkdir(parents=True)
47
48 # Create generator files with frontmatter
49 (app_dir / "custom_generator.md").write_text(
50 '{\n "title": "Custom Generator",\n "color": "#ff0000"\n}\n\nTest prompt'
51 )
52
53 # Also create workspace.html to make it a valid app (not strictly required for generators)
54 (tmp_path / "apps" / "test_app" / "workspace.html").write_text("<h1>Test</h1>")
55
56 # For now, just verify system generators have correct source
57 generators = talent.get_talent_configs(type="generate")
58 for key, info in generators.items():
59 if ":" not in key:
60 assert info.get("source") == "system", f"{key} should have source=system"
61
62
63def test_get_talent_configs_by_schedule():
64 """Test filtering generators by schedule."""
65 talent = importlib.import_module("think.talent")
66
67 # Get daily generators
68 daily = talent.get_talent_configs(type="generate", schedule="daily")
69 assert len(daily) > 0
70 for key, meta in daily.items():
71 assert meta.get("schedule") == "daily", f"{key} should have schedule=daily"
72
73 # Get segment generators
74 segment = talent.get_talent_configs(type="generate", schedule="segment")
75 assert len(segment) > 0
76 for key, meta in segment.items():
77 assert meta.get("schedule") == "segment", f"{key} should have schedule=segment"
78
79 # Verify no overlap
80 assert not set(daily.keys()) & set(segment.keys()), (
81 "daily and segment should not overlap"
82 )
83
84 # Unknown schedule returns empty dict
85 assert talent.get_talent_configs(type="generate", schedule="hourly") == {}
86 assert talent.get_talent_configs(type="generate", schedule="") == {}
87
88
89def test_get_talent_configs_include_disabled(monkeypatch):
90 """Test include_disabled parameter."""
91 talent = importlib.import_module("think.talent")
92
93 # Get generators without disabled (default)
94 without_disabled = talent.get_talent_configs(type="generate", schedule="daily")
95
96 # Get generators with disabled included
97 with_disabled = talent.get_talent_configs(
98 type="generate", schedule="daily", include_disabled=True
99 )
100
101 # Should have at least as many with disabled included
102 assert len(with_disabled) >= len(without_disabled)
103
104
105def test_scheduled_generators_have_valid_schedule():
106 """Test that scheduled generators have valid schedule field.
107
108 Generators with a schedule field must have valid values
109 ('segment', 'daily', or 'activity'). Some generators (like importer) have
110 output but no schedule - they're used for ad-hoc processing, not scheduled runs.
111 """
112 talent = importlib.import_module("think.talent")
113
114 generators = talent.get_talent_configs(type="generate")
115 valid_schedules = ("segment", "daily", "activity", "weekly")
116
117 for key, meta in generators.items():
118 sched = meta.get("schedule")
119 if sched is not None:
120 assert sched in valid_schedules, (
121 f"Generator '{key}' has invalid schedule '{sched}'"
122 )
123
124
125def test_sense_in_segment_schedule():
126 """Test that sense generator exists in segment schedule at priority 5."""
127 talent = importlib.import_module("think.talent")
128
129 generators = talent.get_talent_configs(type="generate", schedule="segment")
130 assert "sense" in generators
131
132 sense = generators["sense"]
133 assert sense.get("priority") == 5, "sense should be at priority 5"
134
135 sources = sense.get("load", {})
136
137 assert sources.get("transcripts") is True, "sense should include transcripts"
138 assert sources.get("percepts") is True, "sense should include percepts"
139
140
141def _write_temp_talent_prompt(stem: str, frontmatter: str) -> Path:
142 talent_dir = Path(__file__).resolve().parent.parent / "talent"
143 prompt_path = talent_dir / f"{stem}.md"
144 prompt_path.write_text(
145 f"{frontmatter}\n\nTemporary test prompt\n", encoding="utf-8"
146 )
147 return prompt_path
148
149
150def test_get_talent_configs_raises_on_missing_type_with_output():
151 talent = importlib.import_module("think.talent")
152 stem = f"test_missing_type_output_{uuid.uuid4().hex}"
153 prompt_path = _write_temp_talent_prompt(
154 stem,
155 '{\n "schedule": "daily",\n "priority": 10,\n "output": "md"\n}',
156 )
157 try:
158 with pytest.raises(
159 ValueError, match=rf"Prompt '{stem}'.*missing required 'type'"
160 ):
161 talent.get_talent_configs(include_disabled=True)
162 finally:
163 prompt_path.unlink(missing_ok=True)
164
165
166def test_get_talent_configs_allows_missing_type_with_tools():
167 talent = importlib.import_module("think.talent")
168 stem = f"test_missing_type_tools_{uuid.uuid4().hex}"
169 prompt_path = _write_temp_talent_prompt(
170 stem,
171 '{\n "schedule": "daily",\n "priority": 10,\n "tools": "journal"\n}',
172 )
173 try:
174 configs = talent.get_talent_configs(include_disabled=True)
175 assert stem in configs
176 assert configs[stem].get("type") is None
177 finally:
178 prompt_path.unlink(missing_ok=True)
179
180
181def test_get_talent_configs_raises_when_generate_missing_output():
182 talent = importlib.import_module("think.talent")
183 stem = f"test_generate_missing_output_{uuid.uuid4().hex}"
184 prompt_path = _write_temp_talent_prompt(
185 stem,
186 '{\n "type": "generate",\n "schedule": "daily",\n "priority": 10\n}',
187 )
188 try:
189 with pytest.raises(
190 ValueError,
191 match=rf"Prompt '{stem}'.*type='generate'.*missing required 'output'",
192 ):
193 talent.get_talent_configs(include_disabled=True)
194 finally:
195 prompt_path.unlink(missing_ok=True)
196
197
198def test_get_talent_configs_allows_cogitate_without_tools():
199 talent = importlib.import_module("think.talent")
200 stem = f"test_cogitate_missing_tools_{uuid.uuid4().hex}"
201 prompt_path = _write_temp_talent_prompt(
202 stem,
203 '{\n "type": "cogitate",\n "schedule": "daily",\n "priority": 10\n}',
204 )
205 try:
206 configs = talent.get_talent_configs(include_disabled=True)
207 assert stem in configs
208 assert configs[stem]["type"] == "cogitate"
209 finally:
210 prompt_path.unlink(missing_ok=True)
211
212
213def test_get_talent_configs_type_generate_returns_only_generate():
214 talent = importlib.import_module("think.talent")
215 generators = talent.get_talent_configs(type="generate")
216 assert generators, "Expected at least one generate prompt"
217 assert all(meta.get("type") == "generate" for meta in generators.values())
218
219
220def test_get_talent_configs_type_cogitate_returns_only_cogitate():
221 talent = importlib.import_module("think.talent")
222 cogitate_prompts = talent.get_talent_configs(type="cogitate")
223 assert cogitate_prompts, "Expected at least one cogitate prompt"
224 assert all(meta.get("type") == "cogitate" for meta in cogitate_prompts.values())