personal memory agent
Unify dream priority system: require explicit priority for all scheduled prompts
Replace the two-phase execution model (generators then agents) with a single
priority-ordered execution pipeline. All scheduled prompts now require an
explicit priority field - no defaults.
Key changes:
- Add priority validation in get_muse_configs() that rejects scheduled prompts
without priority
- Rewrite run_prompts_by_priority() to execute all prompts in priority order,
running same-priority prompts in parallel
- Run incremental indexer --rescan-file after each generator completes
- Remove --skip-generators and --skip-agents CLI flags
- Add --run NAME flag to run single prompts
Priority bands: 10-30 generators, 40-60 analysis agents, 90+ late-stage, 99 fun
Updated all muse/*.md and apps/*/muse/*.md files with explicit priorities.
Updated docs (THINK.md, CORTEX.md, APPS.md) and tests.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+608
-729
+1
-1
apps/entities/muse/entities.md
+1
-1
apps/entities/muse/entities.md
+1
-1
apps/entities/muse/entities_review.md
+1
-1
apps/entities/muse/entities_review.md
+1
-1
apps/entities/muse/entity_observer.md
+1
-1
apps/entities/muse/entity_observer.md
+1
-1
apps/todos/muse/review.md
+1
-1
apps/todos/muse/review.md
+1
-1
apps/todos/muse/todo.md
+1
-1
apps/todos/muse/todo.md
···
4
4
"description": "Maintains the daily todos checklist by mining the journal, prioritising tasks, and applying updates via the todo_* tools.",
5
5
"color": "#ef6c00",
6
6
"schedule": "daily",
7
-
"priority": 20,
7
+
"priority": 50,
8
8
"tools": "journal, todo",
9
9
"multi_facet": true,
10
10
"group": "Todos"
+8
-2
docs/APPS.md
+8
-2
docs/APPS.md
···
271
271
- Keys are namespaced as `{app}:{topic}` (e.g., `my_app:weekly_summary`)
272
272
- Outputs go to `JOURNAL/YYYYMMDD/agents/_<app>_<topic>.md` (or `.json` if `output: "json"`)
273
273
274
-
**Metadata format:** Same schema as system generators in `muse/*.md` - JSON frontmatter includes `title`, `description`, `color`, `schedule` (required), `hook`, `output`, `max_output_tokens`, and `thinking_budget` fields. The `schedule` field must be `"segment"` or `"daily"` - generators with missing or invalid schedule are skipped with a warning. Set `output: "json"` for structured JSON output instead of markdown. Optional `max_output_tokens` sets the maximum response length; `thinking_budget` sets the model's thinking token budget (provider-specific defaults apply if omitted).
274
+
**Metadata format:** Same schema as system generators in `muse/*.md` - JSON frontmatter includes `title`, `description`, `color`, `schedule` (required), `priority` (required for scheduled prompts), `hook`, `output`, `max_output_tokens`, and `thinking_budget` fields. The `schedule` field must be `"segment"` or `"daily"`. The `priority` field is required for all scheduled prompts - prompts without explicit priority will fail validation. Set `output: "json"` for structured JSON output instead of markdown. Optional `max_output_tokens` sets the maximum response length; `thinking_budget` sets the model's thinking token budget (provider-specific defaults apply if omitted).
275
+
276
+
**Priority bands:** Prompts run in priority order (lowest first). Recommended bands:
277
+
- 10-30: Generators (content-producing prompts)
278
+
- 40-60: Analysis agents
279
+
- 90+: Late-stage agents
280
+
- 99: Fun/optional prompts
275
281
276
282
**Event extraction via hooks:** To extract structured events from generator output, use the `hook` field:
277
283
···
344
350
- Keys are namespaced as `{app}:{name}` (e.g., `my_app:helper`)
345
351
- Agents inherit all system agent capabilities (tools, scheduling, handoffs, multi-facet)
346
352
347
-
**Metadata format:** Same schema as system agents in `muse/*.md` - JSON frontmatter includes `title`, `provider`, `model`, `tools`, `schedule`, `priority`, `multi_facet`, `max_output_tokens`, and `thinking_budget` fields. Optional `max_output_tokens` sets the maximum response length; `thinking_budget` sets the model's thinking token budget (provider-specific defaults apply if omitted; OpenAI uses fixed reasoning and ignores this field). See [CORTEX.md](CORTEX.md) for agent configuration details.
353
+
**Metadata format:** Same schema as system agents in `muse/*.md` - JSON frontmatter includes `title`, `provider`, `model`, `tools`, `schedule`, `priority`, `multi_facet`, `max_output_tokens`, and `thinking_budget` fields. The `priority` field is **required** for all scheduled prompts - prompts without explicit priority will fail validation. See the priority bands documentation in [THINK.md](THINK.md#unified-priority-execution). Optional `max_output_tokens` sets the maximum response length; `thinking_budget` sets the model's thinking token budget (provider-specific defaults apply if omitted; OpenAI uses fixed reasoning and ignores this field). See [CORTEX.md](CORTEX.md) for agent configuration details.
348
354
349
355
**Template variables:** Agent prompts can use template variables like `$name`, `$preferred`, and pronoun variables. See [PROMPT_TEMPLATES.md](PROMPT_TEMPLATES.md) for the complete template system documentation.
350
356
+11
-10
docs/CORTEX.md
+11
-10
docs/CORTEX.md
···
287
287
- If omitted, defaults to "default" pack (alias for "journal")
288
288
- `schedule`: Scheduling configuration for automated execution
289
289
- `"daily"`: Run automatically at midnight each day
290
-
- `priority`: Execution order for scheduled agents (integer, default: 50)
291
-
- Lower numbers run first (e.g., priority 10 runs before priority 50)
292
-
- Used to control the order when multiple agents have the same schedule
290
+
- `priority`: Execution order for scheduled prompts (integer, **required** for scheduled prompts)
291
+
- Lower numbers run first (e.g., priority 10 runs before priority 40)
292
+
- See [THINK.md](THINK.md#unified-priority-execution) for priority bands
293
293
- `multi_facet`: Boolean flag for facet-aware agents (default: false)
294
294
- When true, the agent is spawned once for each **active** facet (see Multi-Facet Agents section)
295
295
- Each instance receives a facet-specific prompt with the facet name
···
355
355
356
356
### Execution Order
357
357
Scheduled items run in priority order (lower numbers first):
358
-
1. Items are sorted by their `priority` field (default: 50)
359
-
2. Items with the same priority run in alphabetical order by filename
360
-
3. Each item completes before the next begins
358
+
1. Items are sorted by their `priority` field (required for all scheduled prompts)
359
+
2. Items with the same priority run in parallel, then dream waits for completion
360
+
3. After each generator completes, incremental indexing runs for its output
361
361
362
-
**Priority ranges (recommended):**
363
-
- **1-20**: Foundation tasks (early processing)
364
-
- **50**: Default (most generators and agents)
365
-
- **80-99**: Synthesis tasks that consume other outputs
362
+
**Priority bands (recommended):**
363
+
- **10-30**: Generators (content-producing prompts)
364
+
- **40-60**: Analysis agents
365
+
- **90+**: Late-stage agents
366
+
- **99**: Fun/optional prompts
366
367
367
368
### Multi-Facet Agents
368
369
When an agent has `"multi_facet": true`:
+21
-2
docs/THINK.md
+21
-2
docs/THINK.md
···
25
25
26
26
```bash
27
27
sol cluster YYYYMMDD [--start HHMMSS --length MINUTES]
28
-
sol dream [--day YYYYMMDD] [--segment HHMMSS_LEN] [--force] [--skip-generators] [--skip-agents]
28
+
sol dream [--day YYYYMMDD] [--segment HHMMSS_LEN] [--force] [--run NAME]
29
29
sol supervisor [--no-observers]
30
30
sol mcp [--transport http] [--port PORT] [--path PATH]
31
31
sol cortex [--host HOST] [--port PORT] [--path PATH]
···
78
78
```
79
79
80
80
## Agent System
81
+
82
+
### Unified Priority Execution
83
+
84
+
All scheduled prompts (both generators and tool-using agents) share a unified priority system. The `sol dream` command executes prompts ordered by priority, from lowest (runs first) to highest (runs last).
85
+
86
+
**Priority is required for all scheduled prompts.** Prompts without a `priority` field will fail validation. Suggested priority bands:
87
+
88
+
| Band | Range | Use Case |
89
+
|------|-------|----------|
90
+
| Generators | 10-30 | Content-producing prompts that create `.md` files |
91
+
| Analysis Agents | 40-60 | Agents that analyze generated content |
92
+
| Late-stage | 90+ | Agents that run after most others complete |
93
+
| Fun/Optional | 99 | Low-priority or experimental prompts |
94
+
95
+
After each generator completes and creates output, the indexer runs `--rescan-file` for incremental indexing. A full `--rescan` runs in the post phase.
81
96
82
97
### Cortex: Central Agent Manager
83
98
···
244
259
245
260
System prompts in `muse/*.md` (markdown with JSON frontmatter). Apps can add custom agents in `apps/{app}/muse/`.
246
261
247
-
JSON metadata supports `title`, `provider`, `model`, `tools`, `schedule`, `priority`, `multi_facet`, and `instructions` keys. See [APPS.md](APPS.md#instructions-configuration) for the `instructions` schema that controls system prompts, facet context, and source filtering.
262
+
JSON metadata supports `title`, `provider`, `model`, `tools`, `schedule`, `priority`, `multi_facet`, and `instructions` keys.
263
+
264
+
**Important:** The `priority` field is **required** for all prompts with a `schedule`. Prompts without explicit priority will fail validation. See the [Unified Priority Execution](#unified-priority-execution) section for priority bands.
265
+
266
+
See [APPS.md](APPS.md#instructions-configuration) for the `instructions` schema that controls system prompts, facet context, and source filtering.
248
267
249
268
## Documentation
250
269
+1
muse/activity.md
+1
muse/activity.md
···
4
4
"description": "Synthesizes segment activity from screenshots and audio, focusing on observable changes and searchability.",
5
5
"color": "#00bcd4",
6
6
"schedule": "segment",
7
+
"priority": 10,
7
8
"output": "md",
8
9
"instructions": {
9
10
"sources": {"audio": true, "screen": true, "agents": false},
+1
-1
muse/daily_news.md
+1
-1
muse/daily_news.md
+1
muse/daily_schedule.md
+1
muse/daily_schedule.md
+1
-1
muse/decisionalizer.md
+1
-1
muse/decisionalizer.md
+1
muse/decisions.md
+1
muse/decisions.md
+1
muse/documentation.md
+1
muse/documentation.md
+1
muse/entities.md
+1
muse/entities.md
+1
muse/files.md
+1
muse/files.md
+1
muse/flow.md
+1
muse/flow.md
+1
muse/followups.md
+1
muse/followups.md
+1
muse/knowledge_graph.md
+1
muse/knowledge_graph.md
+1
muse/media.md
+1
muse/media.md
+1
muse/meetings.md
+1
muse/meetings.md
+1
muse/messages.md
+1
muse/messages.md
+1
muse/opportunities.md
+1
muse/opportunities.md
+1
muse/research.md
+1
muse/research.md
+1
muse/schedule.md
+1
muse/schedule.md
+1
muse/screen.md
+1
muse/screen.md
···
4
4
"description": "Creates a detailed documentary record of screen activity. Focuses on the 'what' - chronological account with preserved details, excerpts, and entities.",
5
5
"color": "#9c27b0",
6
6
"schedule": "segment",
7
+
"priority": 10,
7
8
"output": "md",
8
9
"instructions": {
9
10
"sources": {"audio": true, "screen": true, "agents": false},
+1
muse/speakers.md
+1
muse/speakers.md
+1
muse/timeline.md
+1
muse/timeline.md
+1
muse/tools.md
+1
muse/tools.md
+109
-22
tests/test_dream_full.py
+109
-22
tests/test_dream_full.py
···
1
1
# SPDX-License-Identifier: AGPL-3.0-only
2
2
# Copyright (c) 2026 sol pbc
3
3
4
+
"""Tests for the dream module unified priority system."""
5
+
4
6
import importlib
5
7
import shutil
6
8
from pathlib import Path
9
+
10
+
import pytest
7
11
8
12
FIXTURES = Path("fixtures")
9
13
···
15
19
return dest
16
20
17
21
18
-
def test_main_runs(tmp_path, monkeypatch):
22
+
def test_main_runs_with_mocked_prompts(tmp_path, monkeypatch):
23
+
"""Test that main() runs pre/post phases and prompts by priority."""
19
24
mod = importlib.import_module("think.dream")
20
25
journal = copy_journal(tmp_path)
21
26
monkeypatch.setenv("JOURNAL_PATH", str(journal))
22
-
called = []
23
-
generators_called = False
27
+
28
+
commands_run = []
29
+
prompts_run = False
24
30
25
31
def mock_run_command(cmd, day):
26
-
called.append(cmd)
27
-
return True # Return success
32
+
commands_run.append(cmd)
33
+
return True
28
34
29
35
def mock_run_queued_command(cmd, day, timeout=600):
30
-
called.append(cmd)
31
-
return True # Return success
36
+
commands_run.append(cmd)
37
+
return True
32
38
33
-
def mock_run_generators_via_cortex(day, force, segment=None):
34
-
nonlocal generators_called
35
-
generators_called = True
36
-
return (1, 0) # 1 success, 0 failures
39
+
def mock_run_prompts_by_priority(day, segment, force, verbose):
40
+
nonlocal prompts_run
41
+
prompts_run = True
42
+
return (5, 0) # 5 success, 0 failures
37
43
38
44
monkeypatch.setattr(mod, "run_command", mock_run_command)
39
45
monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command)
46
+
monkeypatch.setattr(mod, "run_prompts_by_priority", mock_run_prompts_by_priority)
47
+
monkeypatch.setattr("think.utils.load_dotenv", lambda: True)
40
48
monkeypatch.setattr(
41
-
mod, "run_generators_via_cortex", mock_run_generators_via_cortex
49
+
"sys.argv",
50
+
["sol dream", "--day", "20240101", "--force", "--verbose"],
42
51
)
43
-
# Also mock run_daily_agents to avoid agent execution
44
-
monkeypatch.setattr(mod, "run_daily_agents", lambda day: (0, 0))
52
+
53
+
mod.main()
54
+
55
+
# Verify pre-phase: sense ran
56
+
assert any(c[0] == "sol" and c[1] == "sense" for c in commands_run)
57
+
58
+
# Verify main phase: prompts ran
59
+
assert prompts_run, "run_prompts_by_priority should have been called"
60
+
61
+
# Verify post-phase: indexer rescan ran
62
+
indexer_cmds = [c for c in commands_run if c[0] == "sol" and c[1] == "indexer"]
63
+
assert len(indexer_cmds) >= 1
64
+
assert any("--rescan" in cmd for cmd in indexer_cmds)
65
+
66
+
67
+
def test_segment_mode_skips_pre_post_phases(tmp_path, monkeypatch):
68
+
"""Test that segment mode skips sense and journal-stats."""
69
+
mod = importlib.import_module("think.dream")
70
+
journal = copy_journal(tmp_path)
71
+
72
+
# Create segment directory
73
+
segment_dir = journal / "20240101" / "120000_300"
74
+
segment_dir.mkdir(parents=True, exist_ok=True)
75
+
76
+
monkeypatch.setenv("JOURNAL_PATH", str(journal))
77
+
78
+
commands_run = []
79
+
80
+
def mock_run_command(cmd, day):
81
+
commands_run.append(cmd)
82
+
return True
83
+
84
+
def mock_run_queued_command(cmd, day, timeout=600):
85
+
commands_run.append(cmd)
86
+
return True
87
+
88
+
def mock_run_prompts_by_priority(day, segment, force, verbose):
89
+
return (1, 0)
90
+
91
+
monkeypatch.setattr(mod, "run_command", mock_run_command)
92
+
monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command)
93
+
monkeypatch.setattr(mod, "run_prompts_by_priority", mock_run_prompts_by_priority)
45
94
monkeypatch.setattr("think.utils.load_dotenv", lambda: True)
46
95
monkeypatch.setattr(
47
96
"sys.argv",
48
-
["sol dream", "--day", "20240101", "--force", "--verbose", "--skip-agents"],
97
+
["sol dream", "--day", "20240101", "--segment", "120000_300"],
49
98
)
99
+
50
100
mod.main()
51
-
assert any(c[0] == "sol" and c[1] == "sense" for c in called)
52
-
# Generators now run via cortex, not as direct subprocess
53
-
assert generators_called, "run_generators_via_cortex should have been called"
54
-
# Verify indexer is called with --rescan (light mode) via queued command
55
-
indexer_cmds = [c for c in called if c[0] == "sol" and c[1] == "indexer"]
56
-
assert len(indexer_cmds) == 1
57
-
assert "--rescan" in indexer_cmds[0]
101
+
102
+
# Segment mode should NOT run sense or journal-stats
103
+
assert not any(c[1] == "sense" for c in commands_run if len(c) > 1)
104
+
assert not any(c[1] == "journal-stats" for c in commands_run if len(c) > 1)
105
+
106
+
107
+
def test_priority_validation_required(tmp_path, monkeypatch):
108
+
"""Test that get_muse_configs raises error for scheduled prompts without priority."""
109
+
from think.utils import get_muse_configs
110
+
111
+
# Create a test muse file without priority
112
+
muse_dir = Path(__file__).parent.parent / "muse"
113
+
114
+
# This test verifies the validation exists - actual validation tested in test_utils.py
115
+
# Here we just confirm all existing scheduled prompts have priority
116
+
configs = get_muse_configs(schedule="daily")
117
+
for name, config in configs.items():
118
+
assert "priority" in config, f"Scheduled prompt '{name}' missing priority"
119
+
120
+
121
+
def test_run_single_prompt_validates_schedule(tmp_path, monkeypatch):
122
+
"""Test that --run validates schedule compatibility."""
123
+
mod = importlib.import_module("think.dream")
124
+
journal = copy_journal(tmp_path)
125
+
monkeypatch.setenv("JOURNAL_PATH", str(journal))
126
+
127
+
# Mock to avoid actual execution
128
+
def mock_cortex_request(*args, **kwargs):
129
+
return "mock-id"
130
+
131
+
def mock_wait_for_agents(*args, **kwargs):
132
+
return (["mock-id"], [])
133
+
134
+
def mock_get_agent_end_state(*args, **kwargs):
135
+
return "finish"
136
+
137
+
monkeypatch.setattr(mod, "cortex_request", mock_cortex_request)
138
+
monkeypatch.setattr(mod, "wait_for_agents", mock_wait_for_agents)
139
+
monkeypatch.setattr(mod, "get_agent_end_state", mock_get_agent_end_state)
140
+
141
+
# Running a daily prompt with --segment should fail
142
+
# Note: This requires a real daily prompt in the fixtures
143
+
# For now, just verify the function exists and is callable
144
+
assert callable(mod.run_single_prompt)
+139
-123
tests/test_dream_segment.py
+139
-123
tests/test_dream_segment.py
···
92
92
assert result == ["work", "personal"]
93
93
94
94
95
-
class TestRunSegmentAgentsMultiFacet:
96
-
"""Tests for multi-facet segment agent spawning."""
95
+
class TestRunPromptsByPriority:
96
+
"""Tests for unified priority execution."""
97
97
98
-
def test_multi_facet_agent_spawns_per_facet(self, segment_dir, monkeypatch):
99
-
"""Multi-facet segment agent spawns once per detected facet."""
98
+
def test_prompts_grouped_by_priority(self, segment_dir, monkeypatch):
99
+
"""Prompts are grouped and executed by priority order."""
100
100
from think import dream
101
101
102
-
# Set up facets.json
103
-
facets_data = [
104
-
{"facet": "work", "activity": "Coding", "level": "high"},
105
-
{"facet": "personal", "activity": "Email", "level": "low"},
106
-
]
107
-
(segment_dir / "facets.json").write_text(json.dumps(facets_data))
108
-
109
-
# Track cortex_request calls
110
102
spawned = []
103
+
wait_calls = []
111
104
112
105
def mock_cortex_request(prompt, name, config=None):
113
-
spawned.append({"prompt": prompt, "name": name, "config": config})
114
-
return "mock-agent-id"
106
+
spawned.append({"name": name, "config": config})
107
+
return f"agent-{name}"
115
108
116
-
# Mock get_muse_configs to return a multi-facet segment agent
117
-
def mock_get_muse_configs(has_tools=None, **kwargs):
109
+
def mock_wait_for_agents(agent_ids, timeout=600):
110
+
wait_calls.append(agent_ids)
111
+
return (agent_ids, []) # All complete, none timed out
112
+
113
+
def mock_get_agent_end_state(agent_id):
114
+
return "finish"
115
+
116
+
def mock_get_muse_configs(schedule=None, **kwargs):
118
117
return {
119
-
"test_agent": {
120
-
"schedule": "segment",
121
-
"multi_facet": True,
122
-
"tools": "journal",
123
-
}
118
+
"low_priority": {"priority": 10, "output": "md", "schedule": "segment"},
119
+
"high_priority": {"priority": 90, "output": "md", "schedule": "segment"},
120
+
"mid_priority": {"priority": 50, "output": "md", "schedule": "segment"},
124
121
}
125
122
126
-
# Mock get_enabled_facets to return both facets as enabled
127
123
def mock_get_enabled_facets():
128
-
return {"work": {"title": "Work"}, "personal": {"title": "Personal"}}
124
+
return {"work": {"title": "Work"}}
125
+
126
+
def mock_get_active_facets(day):
127
+
return set()
128
+
129
+
def mock_run_queued_command(cmd, day, timeout=60):
130
+
return True
129
131
130
132
monkeypatch.setattr(dream, "cortex_request", mock_cortex_request)
133
+
monkeypatch.setattr(dream, "wait_for_agents", mock_wait_for_agents)
134
+
monkeypatch.setattr(dream, "get_agent_end_state", mock_get_agent_end_state)
131
135
monkeypatch.setattr(dream, "get_muse_configs", mock_get_muse_configs)
132
136
monkeypatch.setattr(dream, "get_enabled_facets", mock_get_enabled_facets)
137
+
monkeypatch.setattr(dream, "get_active_facets", mock_get_active_facets)
138
+
monkeypatch.setattr(dream, "run_queued_command", mock_run_queued_command)
133
139
134
-
count = dream.run_segment_agents("20240115", "120000_300")
140
+
success, failed = dream.run_prompts_by_priority(
141
+
"20240115", "120000_300", force=False, verbose=False
142
+
)
135
143
136
-
assert count == 2 # One per facet
137
-
assert len(spawned) == 2
144
+
assert success == 3
145
+
assert failed == 0
138
146
139
-
# Verify facet-specific config
140
-
facets_spawned = [s["config"]["facet"] for s in spawned]
141
-
assert "work" in facets_spawned
142
-
assert "personal" in facets_spawned
147
+
# Verify wait was called 3 times (once per priority group)
148
+
assert len(wait_calls) == 3
143
149
144
-
# Verify segment is included
145
-
for s in spawned:
146
-
assert s["config"]["segment"] == "120000_300"
147
-
assert s["config"]["env"]["SEGMENT_KEY"] == "120000_300"
150
+
# Verify order: priority 10 first, then 50, then 90
151
+
spawn_order = [s["name"] for s in spawned]
152
+
assert spawn_order.index("low_priority") < spawn_order.index("mid_priority")
153
+
assert spawn_order.index("mid_priority") < spawn_order.index("high_priority")
148
154
149
-
def test_non_multi_facet_agent_spawns_once(self, segment_dir, monkeypatch):
150
-
"""Regular segment agent spawns once regardless of facets."""
155
+
def test_multi_facet_prompt_spawns_per_facet(self, segment_dir, monkeypatch):
156
+
"""Multi-facet prompts spawn once per active facet."""
151
157
from think import dream
152
158
153
-
# Set up facets.json with multiple facets
159
+
# Set up facets.json
154
160
facets_data = [
155
161
{"facet": "work", "activity": "Coding", "level": "high"},
156
162
{"facet": "personal", "activity": "Email", "level": "low"},
···
160
166
spawned = []
161
167
162
168
def mock_cortex_request(prompt, name, config=None):
163
-
spawned.append({"prompt": prompt, "name": name, "config": config})
164
-
return "mock-agent-id"
169
+
spawned.append({"name": name, "config": config})
170
+
return f"agent-{len(spawned)}"
171
+
172
+
def mock_wait_for_agents(agent_ids, timeout=600):
173
+
return (agent_ids, [])
165
174
166
-
# Regular agent (no multi_facet)
167
-
def mock_get_muse_configs(has_tools=None, **kwargs):
175
+
def mock_get_agent_end_state(agent_id):
176
+
return "finish"
177
+
178
+
def mock_get_muse_configs(schedule=None, **kwargs):
168
179
return {
169
-
"regular_agent": {
180
+
"multi_facet_prompt": {
181
+
"priority": 10,
182
+
"tools": "journal",
183
+
"multi_facet": True,
170
184
"schedule": "segment",
171
-
"tools": "journal",
172
-
}
185
+
},
173
186
}
174
187
175
-
monkeypatch.setattr(dream, "cortex_request", mock_cortex_request)
176
-
monkeypatch.setattr(dream, "get_muse_configs", mock_get_muse_configs)
177
-
178
-
count = dream.run_segment_agents("20240115", "120000_300")
179
-
180
-
assert count == 1
181
-
assert len(spawned) == 1
182
-
assert "facet" not in spawned[0]["config"]
183
-
184
-
def test_multi_facet_no_facets_detected(self, segment_dir, monkeypatch):
185
-
"""Multi-facet agent with no facets detected spawns nothing."""
186
-
from think import dream
187
-
188
-
# Empty facets.json
189
-
(segment_dir / "facets.json").write_text("[]")
190
-
191
-
spawned = []
192
-
193
-
def mock_cortex_request(prompt, name, config=None):
194
-
spawned.append({"prompt": prompt, "name": name, "config": config})
195
-
return "mock-agent-id"
196
-
197
-
def mock_get_muse_configs(has_tools=None, **kwargs):
188
+
def mock_get_enabled_facets():
198
189
return {
199
-
"test_agent": {
200
-
"schedule": "segment",
201
-
"multi_facet": True,
202
-
"tools": "journal",
203
-
}
190
+
"work": {"title": "Work"},
191
+
"personal": {"title": "Personal"},
204
192
}
205
193
206
194
monkeypatch.setattr(dream, "cortex_request", mock_cortex_request)
195
+
monkeypatch.setattr(dream, "wait_for_agents", mock_wait_for_agents)
196
+
monkeypatch.setattr(dream, "get_agent_end_state", mock_get_agent_end_state)
207
197
monkeypatch.setattr(dream, "get_muse_configs", mock_get_muse_configs)
198
+
monkeypatch.setattr(dream, "get_enabled_facets", mock_get_enabled_facets)
208
199
209
-
count = dream.run_segment_agents("20240115", "120000_300")
200
+
success, failed = dream.run_prompts_by_priority(
201
+
"20240115", "120000_300", force=False, verbose=False
202
+
)
203
+
204
+
assert success == 2 # One per facet
205
+
assert len(spawned) == 2
210
206
211
-
assert count == 0
212
-
assert len(spawned) == 0
207
+
facets_spawned = [s["config"]["facet"] for s in spawned]
208
+
assert "work" in facets_spawned
209
+
assert "personal" in facets_spawned
213
210
214
-
def test_mixed_agents_spawn_correctly(self, segment_dir, monkeypatch):
215
-
"""Mix of multi-facet and regular agents spawn correctly."""
211
+
def test_muted_facets_filtered(self, segment_dir, monkeypatch):
212
+
"""Muted facets are filtered out from multi-facet prompts."""
216
213
from think import dream
217
214
218
-
facets_data = [{"facet": "work", "activity": "Coding", "level": "high"}]
215
+
# facets.json contains both work and personal
216
+
facets_data = [
217
+
{"facet": "work", "activity": "Coding", "level": "high"},
218
+
{"facet": "personal", "activity": "Email", "level": "low"},
219
+
]
219
220
(segment_dir / "facets.json").write_text(json.dumps(facets_data))
220
221
221
222
spawned = []
222
223
223
224
def mock_cortex_request(prompt, name, config=None):
224
-
spawned.append({"prompt": prompt, "name": name, "config": config})
225
-
return "mock-agent-id"
225
+
spawned.append({"name": name, "config": config})
226
+
return f"agent-{len(spawned)}"
226
227
227
-
def mock_get_muse_configs(has_tools=None, **kwargs):
228
+
def mock_wait_for_agents(agent_ids, timeout=600):
229
+
return (agent_ids, [])
230
+
231
+
def mock_get_agent_end_state(agent_id):
232
+
return "finish"
233
+
234
+
def mock_get_muse_configs(schedule=None, **kwargs):
228
235
return {
229
-
"faceted_agent": {
230
-
"schedule": "segment",
236
+
"test_prompt": {
237
+
"priority": 10,
238
+
"tools": "journal",
231
239
"multi_facet": True,
232
-
"tools": "journal",
233
-
},
234
-
"regular_agent": {
235
240
"schedule": "segment",
236
-
"tools": "journal",
237
241
},
238
242
}
239
243
240
-
# Mock get_enabled_facets to return work as enabled
244
+
# Only work is enabled (personal is muted)
241
245
def mock_get_enabled_facets():
242
246
return {"work": {"title": "Work"}}
243
247
244
248
monkeypatch.setattr(dream, "cortex_request", mock_cortex_request)
249
+
monkeypatch.setattr(dream, "wait_for_agents", mock_wait_for_agents)
250
+
monkeypatch.setattr(dream, "get_agent_end_state", mock_get_agent_end_state)
245
251
monkeypatch.setattr(dream, "get_muse_configs", mock_get_muse_configs)
246
252
monkeypatch.setattr(dream, "get_enabled_facets", mock_get_enabled_facets)
247
253
248
-
count = dream.run_segment_agents("20240115", "120000_300")
254
+
success, failed = dream.run_prompts_by_priority(
255
+
"20240115", "120000_300", force=False, verbose=False
256
+
)
249
257
250
-
assert count == 2 # 1 faceted (1 facet) + 1 regular
251
-
assert len(spawned) == 2
258
+
# Only work facet should be spawned, personal is muted
259
+
assert success == 1
260
+
assert len(spawned) == 1
261
+
assert spawned[0]["config"]["facet"] == "work"
252
262
253
-
faceted = [s for s in spawned if s["name"] == "faceted_agent"]
254
-
regular = [s for s in spawned if s["name"] == "regular_agent"]
263
+
def test_generator_triggers_incremental_indexing(self, segment_dir, monkeypatch):
264
+
"""Generators trigger incremental indexing after completion."""
265
+
from think import dream
255
266
256
-
assert len(faceted) == 1
257
-
assert faceted[0]["config"]["facet"] == "work"
267
+
indexer_calls = []
258
268
259
-
assert len(regular) == 1
260
-
assert "facet" not in regular[0]["config"]
269
+
def mock_cortex_request(prompt, name, config=None):
270
+
return f"agent-{name}"
261
271
262
-
def test_muted_facets_filtered(self, segment_dir, monkeypatch):
263
-
"""Muted facets are filtered out from segment multi-facet agents."""
264
-
from think import dream
272
+
def mock_wait_for_agents(agent_ids, timeout=600):
273
+
return (agent_ids, [])
265
274
266
-
# facets.json contains both work and personal
267
-
facets_data = [
268
-
{"facet": "work", "activity": "Coding", "level": "high"},
269
-
{"facet": "personal", "activity": "Email", "level": "low"},
270
-
]
271
-
(segment_dir / "facets.json").write_text(json.dumps(facets_data))
275
+
def mock_get_agent_end_state(agent_id):
276
+
return "finish"
272
277
273
-
spawned = []
274
-
275
-
def mock_cortex_request(prompt, name, config=None):
276
-
spawned.append({"prompt": prompt, "name": name, "config": config})
277
-
return "mock-agent-id"
278
-
279
-
def mock_get_muse_configs(has_tools=None, **kwargs):
278
+
def mock_get_muse_configs(schedule=None, **kwargs):
280
279
return {
281
-
"test_agent": {
280
+
"test_generator": {
281
+
"priority": 10,
282
+
"output": "md",
282
283
"schedule": "segment",
283
-
"multi_facet": True,
284
-
"tools": "journal",
285
-
}
284
+
},
286
285
}
287
286
288
-
# Mock get_enabled_facets to only return "work" (personal is muted)
289
287
def mock_get_enabled_facets():
290
288
return {"work": {"title": "Work"}}
291
289
290
+
def mock_get_active_facets(day):
291
+
return set()
292
+
293
+
def mock_run_queued_command(cmd, day, timeout=60):
294
+
indexer_calls.append(cmd)
295
+
return True
296
+
297
+
# Create the output file so indexer is triggered
298
+
output_file = segment_dir / "test_generator.md"
299
+
output_file.write_text("test output")
300
+
292
301
monkeypatch.setattr(dream, "cortex_request", mock_cortex_request)
302
+
monkeypatch.setattr(dream, "wait_for_agents", mock_wait_for_agents)
303
+
monkeypatch.setattr(dream, "get_agent_end_state", mock_get_agent_end_state)
293
304
monkeypatch.setattr(dream, "get_muse_configs", mock_get_muse_configs)
294
305
monkeypatch.setattr(dream, "get_enabled_facets", mock_get_enabled_facets)
306
+
monkeypatch.setattr(dream, "get_active_facets", mock_get_active_facets)
307
+
monkeypatch.setattr(dream, "run_queued_command", mock_run_queued_command)
295
308
296
-
count = dream.run_segment_agents("20240115", "120000_300")
309
+
dream.run_prompts_by_priority(
310
+
"20240115", "120000_300", force=False, verbose=False
311
+
)
297
312
298
-
# Only work facet should be spawned, personal is muted
299
-
assert count == 1
300
-
assert len(spawned) == 1
301
-
assert spawned[0]["config"]["facet"] == "work"
313
+
# Verify indexer was called with --rescan-file
314
+
assert len(indexer_calls) == 1
315
+
assert indexer_calls[0][0] == "sol"
316
+
assert indexer_calls[0][1] == "indexer"
317
+
assert "--rescan-file" in indexer_calls[0]
+6
-6
tests/test_entity_agents.py
+6
-6
tests/test_entity_agents.py
···
33
33
# Verify JSON metadata fields from entities.json
34
34
assert config.get("title") == "Entity Detector"
35
35
assert config.get("schedule") == "daily"
36
-
assert config.get("priority") == 25
36
+
assert config.get("priority") == 55
37
37
assert config.get("tools") == "journal, entities"
38
38
assert config.get("multi_facet") is True
39
39
···
53
53
# Verify JSON metadata fields from entities_review.json
54
54
assert config.get("title") == "Entity Reviewer"
55
55
assert config.get("schedule") == "daily"
56
-
assert config.get("priority") == 26
56
+
assert config.get("priority") == 56
57
57
assert config.get("tools") == "journal, entities"
58
58
assert config.get("multi_facet") is True
59
59
···
131
131
detection_config = get_agent("entities:entities")
132
132
review_config = get_agent("entities:entities_review")
133
133
134
-
detection_priority = detection_config.get("priority", 50)
135
-
review_priority = review_config.get("priority", 50)
134
+
detection_priority = detection_config["priority"]
135
+
review_priority = review_config["priority"]
136
136
137
137
# Review should run after detection
138
138
assert review_priority > detection_priority
139
-
assert detection_priority == 25
140
-
assert review_priority == 26
139
+
assert detection_priority == 55
140
+
assert review_priority == 56
+4
-4
tests/test_generate_full.py
+4
-4
tests/test_generate_full.py
···
77
77
muse_dir = Path(mod.__file__).resolve().parent.parent / "muse"
78
78
test_generator = muse_dir / "test_gen.md"
79
79
test_generator.write_text(
80
-
'{\n "schedule": "daily",\n "output": "md"\n}\n\nTest prompt'
80
+
'{\n "schedule": "daily",\n "priority": 10,\n "output": "md"\n}\n\nTest prompt'
81
81
)
82
82
83
83
try:
···
147
147
# Create generator with hook (new format)
148
148
test_generator = muse_dir / "hooked_gen.md"
149
149
test_generator.write_text(
150
-
'{\n "title": "Hooked",\n "schedule": "daily",\n "output": "md",\n "hook": {"post": "test_hook"}\n}\n\nTest prompt'
150
+
'{\n "title": "Hooked",\n "schedule": "daily",\n "priority": 10,\n "output": "md",\n "hook": {"post": "test_hook"}\n}\n\nTest prompt'
151
151
)
152
152
153
153
try:
···
203
203
muse_dir = Path(mod.__file__).resolve().parent.parent / "muse"
204
204
test_generator = muse_dir / "nohook_gen.md"
205
205
test_generator.write_text(
206
-
'{\n "schedule": "daily",\n "output": "md"\n}\n\nNo hook prompt'
206
+
'{\n "schedule": "daily",\n "priority": 10,\n "output": "md"\n}\n\nNo hook prompt'
207
207
)
208
208
209
209
try:
···
272
272
muse_dir = Path(mod.__file__).resolve().parent.parent / "muse"
273
273
test_generator = muse_dir / "empty_gen.md"
274
274
test_generator.write_text(
275
-
'{\n "schedule": "daily",\n "output": "md"\n}\n\nTest prompt'
275
+
'{\n "schedule": "daily",\n "priority": 10,\n "output": "md"\n}\n\nTest prompt'
276
276
)
277
277
278
278
try:
+5
-5
tests/test_output_hooks.py
+5
-5
tests/test_output_hooks.py
···
179
179
180
180
prompt_file = muse_dir / "hooked_test.md"
181
181
prompt_file.write_text(
182
-
'{\n "title": "Hooked",\n "schedule": "daily",\n "output": "md",\n "hook": {"post": "hooked_test"}\n}\n\nTest prompt'
182
+
'{\n "title": "Hooked",\n "schedule": "daily",\n "priority": 10,\n "output": "md",\n "hook": {"post": "hooked_test"}\n}\n\nTest prompt'
183
183
)
184
184
185
185
hook_file = muse_dir / "hooked_test.py"
···
237
237
238
238
prompt_file = muse_dir / "noop_test.md"
239
239
prompt_file.write_text(
240
-
'{\n "title": "Noop",\n "schedule": "daily",\n "output": "md",\n "hook": {"post": "noop_test"}\n}\n\nTest prompt'
240
+
'{\n "title": "Noop",\n "schedule": "daily",\n "priority": 10,\n "output": "md",\n "hook": {"post": "noop_test"}\n}\n\nTest prompt'
241
241
)
242
242
243
243
hook_file = muse_dir / "noop_test.py"
···
287
287
288
288
prompt_file = muse_dir / "broken_test.md"
289
289
prompt_file.write_text(
290
-
'{\n "title": "Broken",\n "schedule": "daily",\n "output": "md",\n "hook": {"post": "broken_test"}\n}\n\nTest prompt'
290
+
'{\n "title": "Broken",\n "schedule": "daily",\n "priority": 10,\n "output": "md",\n "hook": {"post": "broken_test"}\n}\n\nTest prompt'
291
291
)
292
292
293
293
hook_file = muse_dir / "broken_test.py"
···
574
574
575
575
prompt_file = muse_dir / "prehooked_test.md"
576
576
prompt_file.write_text(
577
-
'{\n "title": "Prehooked",\n "schedule": "daily",\n "output": "md",\n "hook": {"pre": "prehooked_test"}\n}\n\nOriginal prompt'
577
+
'{\n "title": "Prehooked",\n "schedule": "daily",\n "priority": 10,\n "output": "md",\n "hook": {"pre": "prehooked_test"}\n}\n\nOriginal prompt'
578
578
)
579
579
580
580
hook_file = muse_dir / "prehooked_test.py"
···
634
634
635
635
prompt_file = muse_dir / "both_hooks_test.md"
636
636
prompt_file.write_text(
637
-
'{\n "title": "Both Hooks",\n "schedule": "daily",\n "output": "md",\n "hook": {"pre": "both_hooks_test", "post": "both_hooks_test"}\n}\n\nOriginal prompt'
637
+
'{\n "title": "Both Hooks",\n "schedule": "daily",\n "priority": 10,\n "output": "md",\n "hook": {"pre": "both_hooks_test", "post": "both_hooks_test"}\n}\n\nOriginal prompt'
638
638
)
639
639
640
640
hook_file = muse_dir / "both_hooks_test.py"
+270
-547
think/dream.py
+270
-547
think/dream.py
···
1
1
# SPDX-License-Identifier: AGPL-3.0-only
2
2
# Copyright (c) 2026 sol pbc
3
3
4
+
"""Unified prompt execution pipeline for solstone.
5
+
6
+
Runs all scheduled prompts (generators and agents) in priority order.
7
+
Lower priority numbers run first. All prompts at the same priority
8
+
run in parallel, then dream waits for completion before the next group.
9
+
"""
10
+
4
11
import argparse
5
12
import json
6
13
import logging
···
10
17
from pathlib import Path
11
18
12
19
from think.callosum import CallosumConnection
13
-
from think.cortex_client import cortex_request, wait_for_agents
20
+
from think.cortex_client import cortex_request, get_agent_end_state, wait_for_agents
14
21
from think.facets import get_active_facets, get_enabled_facets, get_facets
15
22
from think.runner import run_task
16
23
from think.utils import (
···
19
26
day_path,
20
27
get_journal,
21
28
get_muse_configs,
29
+
get_output_path,
22
30
iso_date,
23
31
setup_cli,
24
32
)
···
28
36
29
37
30
38
def run_command(cmd: list[str], day: str) -> bool:
39
+
"""Run a shell command synchronously."""
31
40
logging.info("==> %s", " ".join(cmd))
32
-
# Extract command name for logging (e.g., "sol generate" -> "generate")
33
41
cmd_name = cmd[1] if cmd[0] == "sol" else cmd[0]
34
42
cmd_name = cmd_name.replace("-", "_")
35
43
36
-
# Use unified runner with automatic logging
37
44
try:
38
45
success, exit_code = run_task(cmd)
39
46
if not success:
···
50
57
51
58
52
59
def run_queued_command(cmd: list[str], day: str, timeout: int = 600) -> bool:
53
-
"""Run a command through supervisor's task queue and wait for completion.
54
-
55
-
This ensures the command is serialized with other requests for the same
56
-
command type (e.g., multiple indexer requests are queued, not concurrent).
57
-
58
-
Args:
59
-
cmd: Command to run (e.g., ["sol", "indexer", "--rescan"])
60
-
day: Day for logging
61
-
timeout: Maximum wait time in seconds (default 600 = 10 minutes)
62
-
63
-
Returns:
64
-
True if command succeeded, False otherwise
65
-
"""
60
+
"""Run a command through supervisor's task queue and wait for completion."""
66
61
import threading
67
62
import uuid
68
63
69
64
cmd_name = cmd[1] if cmd[0] == "sol" else cmd[0]
70
65
cmd_name_log = cmd_name.replace("-", "_")
71
-
72
-
# Generate unique ref to track this specific request
73
66
ref = f"dream-{uuid.uuid4().hex[:8]}"
74
67
75
68
logging.info("==> %s (queued, ref=%s)", " ".join(cmd), ref)
···
79
72
day_log(day, f"{cmd_name_log} error no_callosum")
80
73
return False
81
74
82
-
# Track completion via supervisor.stopped event matching our ref
83
75
result = {"completed": False, "exit_code": None}
84
76
result_event = threading.Event()
85
77
···
88
80
return
89
81
if msg.get("event") != "stopped":
90
82
return
91
-
# Match by ref to ensure we're waiting for OUR request, not another
92
83
if msg.get("ref") != ref:
93
84
return
94
-
95
85
result["completed"] = True
96
86
result["exit_code"] = msg.get("exit_code", -1)
97
87
result_event.set()
98
88
99
-
# Create a separate connection to listen for completion
100
-
# (can't reuse _callosum as it may be busy with other events)
101
89
listener = CallosumConnection()
102
90
listener.start(callback=on_message)
103
91
104
92
try:
105
-
# Emit request to supervisor with our ref for tracking
106
93
_callosum.emit("supervisor", "request", cmd=cmd, ref=ref)
107
94
108
-
# Wait for completion
109
95
if not result_event.wait(timeout=timeout):
110
96
logging.error(f"Timeout waiting for {cmd_name} to complete (ref={ref})")
111
97
day_log(day, f"{cmd_name_log} error timeout")
···
121
107
return False
122
108
123
109
return True
124
-
125
110
finally:
126
111
listener.stop()
127
112
128
113
129
-
def build_pre_generator_commands(
130
-
day: str, verbose: bool = False, segment: str | None = None
131
-
) -> list[list[str]]:
132
-
"""Build pre-generator commands (sense repair for daily mode).
114
+
def emit(event: str, **fields) -> None:
115
+
"""Emit a dream tract event if callosum is connected."""
116
+
if _callosum:
117
+
_callosum.emit("dream", event, **fields)
133
118
134
-
Args:
135
-
day: YYYYMMDD format
136
-
segment: Optional HHMMSS_LEN format (if set, skip sense)
137
-
verbose: Verbose logging
138
-
"""
139
-
commands: list[list[str]] = []
140
119
141
-
if not segment:
142
-
# Daily-only: repair routines run before generators
143
-
cmd = ["sol", "sense", "--day", day]
144
-
if verbose:
145
-
cmd.append("-v")
146
-
commands.append(cmd)
120
+
def check_callosum_available() -> bool:
121
+
"""Check if Callosum socket exists (supervisor running)."""
122
+
socket_path = Path(get_journal()) / "health" / "callosum.sock"
123
+
return socket_path.exists()
147
124
148
-
return commands
149
125
126
+
def load_segment_facets(day: str, segment: str) -> list[str]:
127
+
"""Load facet IDs from a segment's facets.json output."""
128
+
facets_file = day_path(day) / segment / "facets.json"
150
129
151
-
def build_post_generator_commands(
152
-
day: str, verbose: bool = False, segment: str | None = None
153
-
) -> list[list[str]]:
154
-
"""Build post-generator commands (indexer, journal-stats).
130
+
if not facets_file.exists():
131
+
logging.debug(f"No facets.json found for segment {segment}")
132
+
return []
155
133
156
-
Args:
157
-
day: YYYYMMDD format
158
-
segment: Optional HHMMSS_LEN format
159
-
verbose: Verbose logging
160
-
"""
161
-
commands: list[list[str]] = []
134
+
try:
135
+
content = facets_file.read_text().strip()
136
+
if not content:
137
+
return []
162
138
163
-
# Re-index (light mode: excludes historical days, mtime-cached)
164
-
indexer_cmd = ["sol", "indexer", "--rescan"]
165
-
if verbose:
166
-
indexer_cmd.append("--verbose")
167
-
commands.append(indexer_cmd)
139
+
data = json.loads(content)
140
+
if not isinstance(data, list):
141
+
logging.warning(f"facets.json is not an array: {facets_file}")
142
+
return []
168
143
169
-
# Daily-only: journal stats
170
-
if not segment:
171
-
stats_cmd = ["sol", "journal-stats"]
172
-
if verbose:
173
-
stats_cmd.append("--verbose")
174
-
commands.append(stats_cmd)
144
+
facet_ids = [item.get("facet") for item in data if item.get("facet")]
145
+
return facet_ids
175
146
176
-
return commands
147
+
except json.JSONDecodeError as e:
148
+
logging.error(f"Failed to parse facets.json for {segment}: {e}")
149
+
return []
150
+
except Exception as e:
151
+
logging.error(f"Error reading facets.json for {segment}: {e}")
152
+
return []
177
153
178
154
179
-
def run_generators_via_cortex(
180
-
day: str, force: bool, segment: str | None = None
155
+
def run_prompts_by_priority(
156
+
day: str,
157
+
segment: str | None,
158
+
force: bool,
159
+
verbose: bool,
181
160
) -> tuple[int, int]:
182
-
"""Run generators via cortex requests sequentially in priority order.
161
+
"""Run all scheduled prompts in priority order.
183
162
184
-
Generators are sorted by their `priority` field (default: 50), with lower
185
-
numbers running first. This allows generators that depend on other outputs
186
-
to run after those outputs are created.
163
+
Loads all prompts for the target schedule, groups by priority, and executes
164
+
each group in parallel. Waits for completion before proceeding to the next
165
+
priority group. For generators (prompts with output), runs incremental
166
+
indexing after each completes.
187
167
188
168
Args:
189
-
day: YYYYMMDD format
190
-
segment: Optional HHMMSS_LEN format
169
+
day: Day in YYYYMMDD format
170
+
segment: Optional segment key in HHMMSS_LEN format
171
+
force: Whether to regenerate existing outputs
172
+
verbose: Verbose logging
191
173
192
174
Returns:
193
175
Tuple of (success_count, fail_count)
194
176
"""
195
-
from think.cortex_client import get_agent_end_state
196
-
197
177
target_schedule = "segment" if segment else "daily"
198
-
generators = get_muse_configs(
199
-
has_tools=False, has_output=True, schedule=target_schedule
200
-
)
201
178
202
-
if not generators:
203
-
logging.info("No generators found for schedule: %s", target_schedule)
204
-
return (0, 0)
205
-
206
-
# Sort generators by priority (lower numbers first, default 50)
207
-
sorted_generators = sorted(
208
-
generators.items(),
209
-
key=lambda x: (x[1].get("priority", 50), x[0]),
210
-
)
211
-
212
-
logging.info(
213
-
"Running %d generators for %s via cortex: %s",
214
-
len(generators),
215
-
day,
216
-
[name for name, _ in sorted_generators],
217
-
)
218
-
219
-
success_count = 0
220
-
fail_count = 0
221
-
222
-
# Run generators sequentially in priority order
223
-
for generator_name, generator_data in sorted_generators:
224
-
logging.info("Starting generator: %s", generator_name)
225
-
226
-
# Build config for cortex request
227
-
config = {
228
-
"day": day,
229
-
"output": generator_data.get("output", "md"),
230
-
}
231
-
if segment:
232
-
config["segment"] = segment
233
-
if force:
234
-
config["force"] = True
179
+
# Load ALL scheduled prompts (both generators and agents)
180
+
all_prompts = get_muse_configs(schedule=target_schedule)
235
181
236
-
try:
237
-
# Spawn via cortex
238
-
agent_id = cortex_request(
239
-
prompt="", # Generators don't use prompt
240
-
name=generator_name,
241
-
config=config,
242
-
)
243
-
logging.info("Spawned generator %s (ID: %s)", generator_name, agent_id)
182
+
if not all_prompts:
183
+
logging.info(f"No prompts found for schedule: {target_schedule}")
184
+
return (0, 0)
244
185
245
-
# Wait for completion
246
-
completed, timed_out = wait_for_agents([agent_id], timeout=600)
247
-
248
-
if timed_out:
249
-
logging.error(
250
-
"Generator %s timed out (ID: %s)", generator_name, agent_id
251
-
)
252
-
fail_count += 1
253
-
elif completed:
254
-
# Check if it finished successfully or with error
255
-
end_state = get_agent_end_state(agent_id)
256
-
if end_state == "finish":
257
-
logging.info("Generator %s completed successfully", generator_name)
258
-
success_count += 1
259
-
else:
260
-
logging.error(
261
-
"Generator %s ended with state: %s", generator_name, end_state
262
-
)
263
-
fail_count += 1
264
-
else:
265
-
logging.error("Generator %s did not complete", generator_name)
266
-
fail_count += 1
267
-
268
-
except Exception as e:
269
-
logging.error("Failed to run generator %s: %s", generator_name, e)
270
-
fail_count += 1
271
-
272
-
return (success_count, fail_count)
273
-
274
-
275
-
def parse_args() -> argparse.ArgumentParser:
276
-
parser = argparse.ArgumentParser(
277
-
description="Run processing tasks on a journal day or segment"
278
-
)
279
-
parser.add_argument(
280
-
"--day",
281
-
help="Day folder in YYYYMMDD format (defaults to yesterday)",
282
-
)
283
-
parser.add_argument(
284
-
"--segment",
285
-
help="Segment key in HHMMSS_LEN format (processes segment topics only)",
286
-
)
287
-
parser.add_argument("--force", action="store_true", help="Overwrite existing files")
288
-
parser.add_argument(
289
-
"--skip-generators",
290
-
action="store_true",
291
-
help="Skip generator processing, run agents only",
292
-
)
293
-
parser.add_argument(
294
-
"--skip-agents",
295
-
action="store_true",
296
-
help="Skip agent processing, run generators only",
297
-
)
298
-
parser.add_argument(
299
-
"--run",
300
-
metavar="NAME",
301
-
help="Run a single prompt by name (e.g., 'activity', 'timeline')",
302
-
)
303
-
parser.add_argument(
304
-
"--facet",
305
-
metavar="NAME",
306
-
help="Target a specific facet (only used with --run for multi-facet agents)",
307
-
)
308
-
return parser
309
-
310
-
311
-
def check_callosum_available() -> bool:
312
-
"""Check if Callosum socket exists (supervisor running).
313
-
314
-
Returns True if socket exists at JOURNAL_PATH/health/callosum.sock.
315
-
"""
316
-
socket_path = Path(get_journal()) / "health" / "callosum.sock"
317
-
return socket_path.exists()
318
-
319
-
320
-
def run_daily_agents(day: str) -> tuple[int, int]:
321
-
"""Run scheduled daily agents grouped by priority.
322
-
323
-
Loads agents with schedule="daily", groups by priority field (default 50),
324
-
expands multi_facet agents to one per active non-muted facet, spawns each
325
-
group and waits for completion before proceeding to the next.
326
-
327
-
Args:
328
-
day: Day in YYYYMMDD format
329
-
330
-
Returns:
331
-
Tuple of (success_count, fail_count)
332
-
"""
333
-
# Check callosum availability (warning only - cortex_request will fail if not)
334
-
if not check_callosum_available():
335
-
logging.warning("Callosum socket not found - agents may fail to spawn")
336
-
337
-
agents = get_muse_configs(has_tools=True)
338
-
339
-
# Group agents by priority
186
+
# Group prompts by priority
340
187
priority_groups: dict[int, list[tuple[str, dict]]] = {}
341
-
for agent_name, config in agents.items():
342
-
if config.get("schedule") == "daily":
343
-
priority = config.get("priority", 50)
344
-
priority_groups.setdefault(priority, []).append((agent_name, config))
188
+
for name, config in all_prompts.items():
189
+
priority = config["priority"] # Required field, validated by get_muse_configs
190
+
priority_groups.setdefault(priority, []).append((name, config))
345
191
346
-
if not priority_groups:
347
-
logging.info("No scheduled daily agents found")
348
-
return (0, 0)
349
-
350
-
# Pre-compute shared data for multi-facet agents
192
+
# Pre-compute shared data for multi-facet prompts
351
193
day_formatted = iso_date(day)
352
194
input_summary = day_input_summary(day)
353
195
enabled_facets = get_enabled_facets()
354
-
active_facets = get_active_facets(day)
196
+
197
+
if segment:
198
+
# Segment mode: use facets from facets.json (if available)
199
+
raw_facets = load_segment_facets(day, segment)
200
+
active_facets = set(f for f in raw_facets if f in enabled_facets)
201
+
else:
202
+
# Daily mode: use facets with activity on this day
203
+
active_facets = get_active_facets(day)
355
204
356
-
total_agents = sum(len(agents_list) for agents_list in priority_groups.values())
205
+
total_prompts = sum(len(prompts) for prompts in priority_groups.values())
357
206
num_groups = len(priority_groups)
358
207
359
208
logging.info(
360
-
f"Running {total_agents} scheduled agents for {day} in {num_groups} priority groups"
209
+
f"Running {total_prompts} prompts for {day} in {num_groups} priority groups"
361
210
)
362
211
363
-
# Emit agents_started event
364
212
emit(
365
-
"agents_started",
366
-
mode="daily",
213
+
"started",
214
+
mode=target_schedule,
367
215
day=day,
368
-
count=total_agents,
216
+
segment=segment,
217
+
count=total_prompts,
369
218
groups=num_groups,
370
219
)
371
220
372
-
agents_start_time = time.time()
373
-
total_completed = 0
221
+
start_time = time.time()
222
+
total_success = 0
374
223
total_failed = 0
375
224
376
225
# Process each priority group in order
377
226
for priority in sorted(priority_groups.keys()):
378
-
agents_list = priority_groups[priority]
379
-
logging.info(f"Starting priority {priority} agents ({len(agents_list)} agents)")
227
+
prompts_list = priority_groups[priority]
228
+
logging.info(f"Starting priority {priority} ({len(prompts_list)} prompts)")
380
229
381
-
# Emit group_started event
382
230
emit(
383
231
"group_started",
384
-
mode="daily",
232
+
mode=target_schedule,
385
233
day=day,
234
+
segment=segment,
386
235
priority=priority,
387
-
count=len(agents_list),
236
+
count=len(prompts_list),
388
237
)
389
238
390
-
spawned_ids = []
239
+
spawned: list[tuple[str, str, dict]] = [] # (agent_id, name, config)
240
+
241
+
for prompt_name, config in prompts_list:
242
+
has_tools = bool(config.get("tools"))
243
+
is_generator = not has_tools
391
244
392
-
for agent_name, config in agents_list:
393
245
try:
394
-
# Check if this is a multi-facet agent
395
246
if config.get("multi_facet"):
396
247
always_run = config.get("always", False)
397
248
398
249
for facet_name in enabled_facets.keys():
399
-
# Skip inactive facets unless agent has always=true
400
250
if not always_run and facet_name not in active_facets:
401
251
logging.info(
402
-
f"Skipping {agent_name} for {facet_name}: "
252
+
f"Skipping {prompt_name} for {facet_name}: "
403
253
f"no activity on {day_formatted}"
404
254
)
405
255
continue
406
256
407
-
logging.info(f"Spawning {agent_name} for facet: {facet_name}")
257
+
logging.info(f"Spawning {prompt_name} for facet: {facet_name}")
258
+
259
+
request_config: dict = {"facet": facet_name}
260
+
if is_generator:
261
+
request_config["day"] = day
262
+
request_config["output"] = config.get("output", "md")
263
+
if force:
264
+
request_config["force"] = True
265
+
if segment:
266
+
request_config["segment"] = segment
267
+
request_config["env"] = {"SEGMENT_KEY": segment}
268
+
269
+
prompt = (
270
+
""
271
+
if is_generator
272
+
else f"Processing facet '{facet_name}' for {day_formatted}: {input_summary}. Use get_facet('{facet_name}') to load context."
273
+
)
274
+
408
275
agent_id = cortex_request(
409
-
prompt=f"Processing facet '{facet_name}' for {day_formatted}: {input_summary}. Use get_facet('{facet_name}') to load context.",
410
-
name=agent_name,
411
-
config={"facet": facet_name},
276
+
prompt=prompt,
277
+
name=prompt_name,
278
+
config=request_config,
412
279
)
413
-
spawned_ids.append(agent_id)
280
+
spawned.append((agent_id, prompt_name, config))
414
281
logging.info(
415
-
f"Started {agent_name} for {facet_name} (ID: {agent_id})"
282
+
f"Started {prompt_name} for {facet_name} (ID: {agent_id})"
416
283
)
417
284
else:
418
-
# Regular single-instance agent
285
+
# Regular single-instance prompt
286
+
logging.info(f"Spawning {prompt_name}")
287
+
288
+
request_config = {}
289
+
if is_generator:
290
+
request_config["day"] = day
291
+
request_config["output"] = config.get("output", "md")
292
+
if force:
293
+
request_config["force"] = True
294
+
if segment:
295
+
request_config["segment"] = segment
296
+
request_config["env"] = {"SEGMENT_KEY": segment}
297
+
298
+
prompt = (
299
+
""
300
+
if is_generator
301
+
else f"Running scheduled task for {day_formatted}: {input_summary}."
302
+
)
303
+
419
304
agent_id = cortex_request(
420
-
prompt=f"Running daily scheduled task for {day_formatted}: {input_summary}.",
421
-
name=agent_name,
305
+
prompt=prompt,
306
+
name=prompt_name,
307
+
config=request_config if request_config else None,
422
308
)
423
-
spawned_ids.append(agent_id)
424
-
logging.info(f"Started {agent_name} agent (ID: {agent_id})")
309
+
spawned.append((agent_id, prompt_name, config))
310
+
logging.info(f"Started {prompt_name} (ID: {agent_id})")
311
+
425
312
except Exception as e:
426
-
logging.error(f"Failed to spawn {agent_name}: {e}")
313
+
logging.error(f"Failed to spawn {prompt_name}: {e}")
427
314
total_failed += 1
428
315
429
316
# Wait for this priority group to complete
430
-
group_completed = 0
431
-
group_timed_out = 0
432
-
if spawned_ids:
433
-
logging.info(
434
-
f"Waiting for {len(spawned_ids)} agents in priority {priority}..."
435
-
)
436
-
completed, timed_out = wait_for_agents(spawned_ids, timeout=600)
437
-
group_completed = len(completed)
438
-
group_timed_out = len(timed_out)
439
-
total_completed += group_completed
440
-
total_failed += group_timed_out
317
+
group_success = 0
318
+
group_failed = 0
319
+
320
+
if spawned:
321
+
agent_ids = [agent_id for agent_id, _, _ in spawned]
322
+
logging.info(f"Waiting for {len(agent_ids)} prompts in priority {priority}...")
323
+
324
+
completed, timed_out = wait_for_agents(agent_ids, timeout=600)
441
325
442
326
if timed_out:
443
327
logging.warning(
444
-
f"Priority {priority}: {len(timed_out)} agents timed out: {timed_out}"
328
+
f"Priority {priority}: {len(timed_out)} prompts timed out: {timed_out}"
445
329
)
330
+
group_failed += len(timed_out)
331
+
332
+
# Check end states and run incremental indexing for generators
333
+
for agent_id, prompt_name, config in spawned:
334
+
if agent_id in timed_out:
335
+
continue
446
336
447
-
# Emit group_completed event
337
+
end_state = get_agent_end_state(agent_id)
338
+
if end_state == "finish":
339
+
logging.info(f"{prompt_name} completed successfully")
340
+
group_success += 1
341
+
342
+
# Incremental indexing for generators
343
+
is_generator = not bool(config.get("tools"))
344
+
if is_generator:
345
+
output_format = config.get("output", "md")
346
+
output_path = get_output_path(
347
+
day_path(day),
348
+
prompt_name,
349
+
segment=segment,
350
+
output_format=output_format,
351
+
)
352
+
353
+
if output_path.exists():
354
+
logging.debug(f"Indexing {output_path}")
355
+
run_queued_command(
356
+
["sol", "indexer", "--rescan-file", str(output_path)],
357
+
day,
358
+
timeout=60,
359
+
)
360
+
else:
361
+
logging.error(f"{prompt_name} ended with state: {end_state}")
362
+
group_failed += 1
363
+
364
+
total_success += group_success
365
+
total_failed += group_failed
366
+
448
367
emit(
449
368
"group_completed",
450
-
mode="daily",
369
+
mode=target_schedule,
451
370
day=day,
371
+
segment=segment,
452
372
priority=priority,
453
-
completed=group_completed,
454
-
timed_out=group_timed_out,
373
+
success=group_success,
374
+
failed=group_failed,
455
375
)
456
376
457
-
# Emit agents_completed event
458
-
agents_duration_ms = int((time.time() - agents_start_time) * 1000)
377
+
duration_ms = int((time.time() - start_time) * 1000)
459
378
emit(
460
-
"agents_completed",
461
-
mode="daily",
379
+
"completed",
380
+
mode=target_schedule,
462
381
day=day,
463
-
success=total_completed,
382
+
segment=segment,
383
+
success=total_success,
464
384
failed=total_failed,
465
-
duration_ms=agents_duration_ms,
466
-
)
467
-
468
-
logging.info(
469
-
f"Daily agents completed: {total_completed} succeeded, {total_failed} failed"
385
+
duration_ms=duration_ms,
470
386
)
471
-
return (total_completed, total_failed)
472
387
473
-
474
-
def load_segment_facets(day: str, segment: str) -> list[str]:
475
-
"""Load facet IDs from a segment's facets.json output.
476
-
477
-
Reads the facets.json file written by the facets generator and extracts
478
-
the list of facet IDs that were active in this segment.
479
-
480
-
Args:
481
-
day: Day in YYYYMMDD format
482
-
segment: Segment key in HHMMSS_LEN format
483
-
484
-
Returns:
485
-
List of facet IDs (e.g., ["work", "personal"]). Returns empty list if
486
-
file is missing, empty, or malformed.
487
-
"""
488
-
facets_file = day_path(day) / segment / "facets.json"
489
-
490
-
if not facets_file.exists():
491
-
logging.debug(f"No facets.json found for segment {segment}")
492
-
return []
493
-
494
-
try:
495
-
content = facets_file.read_text().strip()
496
-
if not content:
497
-
return []
498
-
499
-
data = json.loads(content)
500
-
if not isinstance(data, list):
501
-
logging.warning(f"facets.json is not an array: {facets_file}")
502
-
return []
503
-
504
-
# Extract facet IDs from the array of objects
505
-
facet_ids = [item.get("facet") for item in data if item.get("facet")]
506
-
return facet_ids
507
-
508
-
except json.JSONDecodeError as e:
509
-
logging.error(f"Failed to parse facets.json for {segment}: {e}")
510
-
return []
511
-
except Exception as e:
512
-
logging.error(f"Error reading facets.json for {segment}: {e}")
513
-
return []
514
-
515
-
516
-
def run_segment_agents(day: str, segment: str) -> int:
517
-
"""Spawn segment agents (fire-and-forget).
518
-
519
-
Loads agents with schedule="segment" and spawns each with SEGMENT_KEY env var.
520
-
Multi-facet agents are spawned once per facet detected in the segment's
521
-
facets.json output. Does NOT wait for completion.
522
-
523
-
Args:
524
-
day: Day in YYYYMMDD format
525
-
segment: Segment key in HHMMSS_LEN format
526
-
527
-
Returns:
528
-
Number of agents spawned
529
-
"""
530
-
agents = get_muse_configs(has_tools=True)
531
-
spawned = 0
532
-
533
-
# Lazy-load segment facets and enabled facets for multi-facet agents
534
-
segment_facets: list[str] | None = None
535
-
536
-
for agent_name, config in agents.items():
537
-
if config.get("schedule") != "segment":
538
-
continue
539
-
540
-
try:
541
-
if config.get("multi_facet"):
542
-
# Lazy-load and filter segment facets on first multi-facet agent
543
-
if segment_facets is None:
544
-
enabled_facet_names = set(get_enabled_facets().keys())
545
-
raw_facets = load_segment_facets(day, segment)
546
-
# Filter out muted facets (consistent with daily behavior)
547
-
segment_facets = [f for f in raw_facets if f in enabled_facet_names]
548
-
if segment_facets:
549
-
logging.info(
550
-
f"Segment {segment} facets: {', '.join(segment_facets)}"
551
-
)
552
-
else:
553
-
logging.debug(f"No enabled facets for segment {segment}")
554
-
555
-
# Spawn once per enabled facet
556
-
for facet_name in segment_facets:
557
-
cortex_request(
558
-
prompt=f"Processing facet '{facet_name}' in segment {segment} from {day}. Use get_facet('{facet_name}') to load context.",
559
-
name=agent_name,
560
-
config={
561
-
"segment": segment,
562
-
"facet": facet_name,
563
-
"env": {"SEGMENT_KEY": segment},
564
-
},
565
-
)
566
-
spawned += 1
567
-
logging.info(f"Spawned {agent_name} for facet {facet_name}")
568
-
else:
569
-
# Regular segment agent - spawn once
570
-
cortex_request(
571
-
prompt=f"Processing segment {segment} from {day}. Use available tools to analyze this specific recording window.",
572
-
name=agent_name,
573
-
config={"segment": segment, "env": {"SEGMENT_KEY": segment}},
574
-
)
575
-
spawned += 1
576
-
logging.info(f"Spawned segment agent: {agent_name}")
577
-
except Exception as e:
578
-
logging.error(f"Failed to spawn {agent_name}: {e}")
579
-
580
-
return spawned
388
+
logging.info(f"Prompts completed: {total_success} succeeded, {total_failed} failed")
389
+
return (total_success, total_failed)
581
390
582
391
583
392
def run_single_prompt(
···
599
408
Returns:
600
409
True if successful, False if failed
601
410
"""
602
-
from think.cortex_client import get_agent_end_state
603
-
604
411
# Load all configs to find the prompt
605
412
all_configs = get_muse_configs(include_disabled=True)
606
413
···
611
418
612
419
config = all_configs[name]
613
420
614
-
# Check if disabled
615
421
if config.get("disabled"):
616
422
logging.warning(f"Prompt '{name}' is disabled")
617
423
return False
618
424
619
-
# Determine if this is a generator (no tools) or tool agent (has tools)
620
425
has_tools = bool(config.get("tools"))
621
426
is_generator = not has_tools
622
427
···
635
440
)
636
441
return False
637
442
638
-
# Validate facet usage
639
443
if facet and not config.get("multi_facet"):
640
444
logging.warning(f"'{name}' is not a multi-facet agent, --facet will be ignored")
641
445
facet = None
···
643
447
day_formatted = iso_date(day)
644
448
645
449
if is_generator:
646
-
# Run as generator
647
450
logging.info(f"Running generator: {name}")
648
451
649
452
request_config = {
···
657
460
658
461
try:
659
462
agent_id = cortex_request(
660
-
prompt="", # Generators don't use prompt
463
+
prompt="",
661
464
name=name,
662
465
config=request_config,
663
466
)
664
467
logging.info(f"Spawned generator {name} (ID: {agent_id})")
665
468
666
-
# Wait for completion
667
469
completed, timed_out = wait_for_agents([agent_id], timeout=600)
668
470
669
471
if timed_out:
···
684
486
return False
685
487
686
488
else:
687
-
# Run as agent
688
489
logging.info(f"Running agent: {name}")
689
490
690
491
input_summary = day_input_summary(day)
691
492
spawned_ids = []
692
493
693
494
if config.get("multi_facet"):
694
-
# Multi-facet agent - run for specific facet or all active facets
695
495
facets_data = get_facets()
696
496
enabled_facets = {
697
497
k: v for k, v in facets_data.items() if not v.get("muted", False)
···
700
500
always_run = config.get("always", False)
701
501
702
502
if facet:
703
-
# Run for specific facet
704
503
if facet not in enabled_facets:
705
504
logging.error(f"Facet '{facet}' not found or is muted")
706
505
return False
707
506
target_facets = [facet]
708
507
else:
709
-
# Run for all active facets (or all if always=true)
710
508
target_facets = [
711
509
f for f in enabled_facets.keys() if always_run or f in active_facets
712
510
]
713
511
714
512
if not target_facets:
715
513
logging.warning(f"No active facets for {name} on {day_formatted}")
716
-
return True # Not a failure, just nothing to do
514
+
return True
717
515
718
516
for facet_name in target_facets:
719
517
try:
···
729
527
logging.error(f"Failed to spawn {name} for {facet_name}: {e}")
730
528
731
529
else:
732
-
# Regular single-instance agent
733
530
try:
734
531
request_config = {}
735
532
if segment:
···
750
547
if not spawned_ids:
751
548
return False
752
549
753
-
# Wait for all spawned agents
754
550
logging.info(f"Waiting for {len(spawned_ids)} agent(s)...")
755
551
completed, timed_out = wait_for_agents(spawned_ids, timeout=600)
756
552
757
553
if timed_out:
758
554
logging.warning(f"{len(timed_out)} agent(s) timed out: {timed_out}")
759
555
760
-
# Check end states for completed agents
761
556
error_count = 0
762
557
for agent_id in completed:
763
558
end_state = get_agent_end_state(agent_id)
···
773
568
return success
774
569
775
570
776
-
def emit(event: str, **fields) -> None:
777
-
"""Emit a dream tract event if callosum is connected."""
778
-
if _callosum:
779
-
_callosum.emit("dream", event, **fields)
571
+
def parse_args() -> argparse.ArgumentParser:
572
+
parser = argparse.ArgumentParser(
573
+
description="Run processing tasks on a journal day or segment"
574
+
)
575
+
parser.add_argument(
576
+
"--day",
577
+
help="Day folder in YYYYMMDD format (defaults to yesterday)",
578
+
)
579
+
parser.add_argument(
580
+
"--segment",
581
+
help="Segment key in HHMMSS_LEN format (processes segment topics only)",
582
+
)
583
+
parser.add_argument("--force", action="store_true", help="Overwrite existing files")
584
+
parser.add_argument(
585
+
"--run",
586
+
metavar="NAME",
587
+
help="Run a single prompt by name (e.g., 'activity', 'timeline')",
588
+
)
589
+
parser.add_argument(
590
+
"--facet",
591
+
metavar="NAME",
592
+
help="Target a specific facet (only used with --run for multi-facet agents)",
593
+
)
594
+
return parser
780
595
781
596
782
597
def main() -> None:
···
793
608
if not day_dir.is_dir():
794
609
parser.error(f"Day folder not found: {day_dir}")
795
610
796
-
# Validate --run is mutually exclusive with --skip-generators/--skip-agents
797
-
if args.run and (args.skip_generators or args.skip_agents):
798
-
parser.error("--run cannot be used with --skip-generators or --skip-agents")
799
-
800
-
# Validate --facet requires --run
801
611
if args.facet and not args.run:
802
612
parser.error("--facet requires --run")
803
613
804
-
# Handle single prompt execution mode
805
-
if args.run:
806
-
# Start callosum for cortex communication
807
-
_callosum = CallosumConnection()
808
-
_callosum.start()
809
-
try:
614
+
# Start callosum connection
615
+
_callosum = CallosumConnection()
616
+
_callosum.start()
617
+
618
+
try:
619
+
# Handle single prompt execution mode
620
+
if args.run:
810
621
success = run_single_prompt(
811
622
day=day,
812
623
name=args.run,
···
815
626
facet=args.facet,
816
627
)
817
628
sys.exit(0 if success else 1)
818
-
finally:
819
-
_callosum.stop()
820
629
821
-
# Start callosum connection for event emission
822
-
_callosum = CallosumConnection()
823
-
_callosum.start()
630
+
# Check callosum availability
631
+
if not check_callosum_available():
632
+
logging.warning("Callosum socket not found - prompts may fail to spawn")
824
633
825
-
try:
826
634
start_time = time.time()
827
-
generator_fail_count = 0
828
-
agent_fail_count = 0
829
-
830
-
# Determine mode based on segment presence
831
635
mode = "segment" if args.segment else "daily"
832
636
833
-
# Build base event fields (mode always, segment only for segment mode)
834
-
def event_fields(**extra):
835
-
fields = {"mode": mode, "day": day}
836
-
if args.segment:
837
-
fields["segment"] = args.segment
838
-
fields.update(extra)
839
-
return fields
840
-
841
-
# Emit started event
842
-
emit("started", **event_fields())
843
-
844
-
# Phase 1: Generators (pre-commands, generators via cortex, post-commands)
845
-
if not args.skip_generators:
846
-
# Run pre-generator commands (e.g., sense repair)
847
-
pre_commands = build_pre_generator_commands(
848
-
day, verbose=args.verbose, segment=args.segment
849
-
)
850
-
for cmd in pre_commands:
851
-
day_log(day, f"starting: {' '.join(cmd)}")
852
-
if not run_command(cmd, day):
853
-
generator_fail_count += 1
854
-
855
-
# Run generators via cortex
856
-
gen_success, gen_fail = run_generators_via_cortex(
857
-
day, args.force, segment=args.segment
858
-
)
859
-
generator_fail_count += gen_fail
860
-
861
-
# Run post-generator commands (indexer, journal-stats)
862
-
post_commands = build_post_generator_commands(
863
-
day, verbose=args.verbose, segment=args.segment
864
-
)
865
-
for index, cmd in enumerate(post_commands):
866
-
day_log(day, f"starting: {' '.join(cmd)}")
867
-
868
-
# Emit command event
869
-
emit(
870
-
"command",
871
-
**event_fields(
872
-
command=cmd[1], index=index, total=len(post_commands)
873
-
),
874
-
)
875
-
876
-
# Route indexer commands through supervisor queue for serialization
877
-
is_indexer = cmd[0] == "sol" and len(cmd) > 1 and cmd[1] == "indexer"
878
-
if is_indexer:
879
-
success = run_queued_command(cmd, day)
880
-
else:
881
-
success = run_command(cmd, day)
882
-
883
-
if not success:
884
-
generator_fail_count += 1
885
-
886
-
# Emit generators_completed event
887
-
emit(
888
-
"generators_completed",
889
-
**event_fields(
890
-
success=gen_success,
891
-
failed=generator_fail_count,
892
-
duration_ms=int((time.time() - start_time) * 1000),
893
-
),
894
-
)
637
+
# PRE-PHASE: Run sense repair (daily only)
638
+
if not args.segment:
639
+
logging.info("Running pre-phase: sense repair")
640
+
cmd = ["sol", "sense", "--day", day]
641
+
if args.verbose:
642
+
cmd.append("-v")
643
+
day_log(day, f"starting: {' '.join(cmd)}")
644
+
if not run_command(cmd, day):
645
+
logging.warning("Sense repair failed, continuing anyway")
895
646
896
-
logging.info(
897
-
f"Generators completed: {gen_success} succeeded, {generator_fail_count} failed"
898
-
)
899
-
900
-
# Exit early if generators failed and agents are requested
901
-
if generator_fail_count > 0 and not args.skip_agents:
902
-
logging.error("Generators failed, skipping agents")
903
-
emit(
904
-
"completed",
905
-
**event_fields(
906
-
generator_failed=generator_fail_count,
907
-
agent_failed=0,
908
-
duration_ms=int((time.time() - start_time) * 1000),
909
-
),
910
-
)
911
-
day_log(day, f"dream generators failed {generator_fail_count}")
912
-
sys.exit(1)
913
-
914
-
# Phase 2: Agents
915
-
if not args.skip_agents:
916
-
if args.segment:
917
-
# Segment mode: fire-and-forget
918
-
spawned = run_segment_agents(day, args.segment)
919
-
logging.info(f"Spawned {spawned} segment agents")
920
-
else:
921
-
# Daily mode: priority groups with waiting
922
-
agent_success, agent_fail_count = run_daily_agents(day)
647
+
# MAIN PHASE: Run all prompts by priority
648
+
success_count, fail_count = run_prompts_by_priority(
649
+
day=day,
650
+
segment=args.segment,
651
+
force=args.force,
652
+
verbose=args.verbose,
653
+
)
923
654
924
-
# Full rescan after agents (via supervisor queue for serialization)
925
-
if agent_success > 0 or agent_fail_count > 0:
926
-
logging.info("Running full index rescan after agents...")
927
-
full_rescan_cmd = ["sol", "indexer", "--rescan-full"]
928
-
if args.verbose:
929
-
full_rescan_cmd.append("--verbose")
930
-
run_queued_command(full_rescan_cmd, day, timeout=3600)
655
+
# POST-PHASE: Final indexing and stats (daily only)
656
+
if not args.segment:
657
+
logging.info("Running post-phase: indexer rescan")
658
+
rescan_cmd = ["sol", "indexer", "--rescan"]
659
+
if args.verbose:
660
+
rescan_cmd.append("--verbose")
661
+
run_queued_command(rescan_cmd, day, timeout=3600)
931
662
932
-
# Emit completed event (all processing done)
933
-
emit(
934
-
"completed",
935
-
**event_fields(
936
-
generator_failed=generator_fail_count,
937
-
agent_failed=agent_fail_count,
938
-
duration_ms=int((time.time() - start_time) * 1000),
939
-
),
940
-
)
663
+
logging.info("Running post-phase: journal stats")
664
+
stats_cmd = ["sol", "journal-stats"]
665
+
if args.verbose:
666
+
stats_cmd.append("--verbose")
667
+
run_command(stats_cmd, day)
941
668
942
669
# Build log message
943
670
msg = "dream"
944
-
if args.skip_generators:
945
-
msg += " --skip-generators"
946
-
if args.skip_agents:
947
-
msg += " --skip-agents"
948
671
if args.force:
949
672
msg += " --force"
950
-
if generator_fail_count:
951
-
msg += f" generators_failed={generator_fail_count}"
952
-
if agent_fail_count:
953
-
msg += f" agents_failed={agent_fail_count}"
673
+
if fail_count:
674
+
msg += f" failed={fail_count}"
954
675
day_log(day, msg)
955
676
956
-
# Exit with error if any failures
957
-
if generator_fail_count > 0 or agent_fail_count > 0:
958
-
total_failures = generator_fail_count + agent_fail_count
959
-
logging.error(f"{total_failures} task(s) failed, exiting with error")
677
+
duration_ms = int((time.time() - start_time) * 1000)
678
+
logging.info(f"Dream completed in {duration_ms}ms: {success_count} succeeded, {fail_count} failed")
679
+
680
+
if fail_count > 0:
681
+
logging.error(f"{fail_count} prompt(s) failed, exiting with error")
960
682
sys.exit(1)
683
+
961
684
finally:
962
685
_callosum.stop()
963
686
+8
think/utils.py
+8
think/utils.py
···
1011
1011
if "provider" in override:
1012
1012
info["provider"] = override["provider"]
1013
1013
1014
+
# Validate: scheduled prompts must have explicit priority
1015
+
for key, info in configs.items():
1016
+
if info.get("schedule") and "priority" not in info:
1017
+
raise ValueError(
1018
+
f"Scheduled prompt '{key}' is missing required 'priority' field. "
1019
+
f"All prompts with 'schedule' must declare an explicit priority."
1020
+
)
1021
+
1014
1022
return configs
1015
1023
1016
1024