Add sol muse CLI for inspecting prompt configurations

+2

sol.py

··· 62 62 "agents": "think.agents", 63 63 "cortex": "think.cortex", 64 64 "mcp": "think.mcp", 65 + "muse": "think.muse_cli", 65 66 # convey package - web UI 66 67 "convey": "convey.cli", 67 68 "restart": "convey.restart", ··· 108 109 "agents", 109 110 "cortex", 110 111 "mcp", 112 + "muse", 111 113 ], 112 114 "Convey (web UI)": [ 113 115 "convey",

+203

tests/test_muse_cli.py

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Tests for the sol muse CLI.""" 5 + 6 + import json 7 + 8 + import pytest 9 + 10 + from think.muse_cli import ( 11 + _collect_configs, 12 + _property_tags, 13 + _scan_variables, 14 + json_output, 15 + list_prompts, 16 + show_prompt, 17 + ) 18 + 19 + 20 + def test_collect_configs_returns_prompts(): 21 + """All configs include known system prompts.""" 22 + configs = _collect_configs(include_disabled=True) 23 + assert "flow" in configs 24 + assert "activity" in configs 25 + assert "default" in configs 26 + 27 + 28 + def test_collect_configs_excludes_disabled_by_default(): 29 + """Disabled prompts are excluded unless include_disabled is set.""" 30 + without = _collect_configs(include_disabled=False) 31 + with_disabled = _collect_configs(include_disabled=True) 32 + assert len(with_disabled) >= len(without) 33 + 34 + # files.md is disabled by default 35 + disabled_keys = set(with_disabled.keys()) - set(without.keys()) 36 + assert len(disabled_keys) > 0 37 + 38 + 39 + def test_collect_configs_filter_schedule(): 40 + """Schedule filter returns only matching prompts.""" 41 + daily = _collect_configs(schedule="daily", include_disabled=True) 42 + for key, info in daily.items(): 43 + assert info.get("schedule") == "daily", f"{key} should be daily" 44 + 45 + segment = _collect_configs(schedule="segment", include_disabled=True) 46 + for key, info in segment.items(): 47 + assert info.get("schedule") == "segment", f"{key} should be segment" 48 + 49 + # No overlap 50 + assert not set(daily.keys()) & set(segment.keys()) 51 + 52 + 53 + def test_collect_configs_filter_source(): 54 + """Source filter returns only matching prompts.""" 55 + system = _collect_configs(source="system", include_disabled=True) 56 + for key, info in system.items(): 57 + assert info.get("source") == "system", f"{key} should be system" 58 + 59 + app = _collect_configs(source="app", include_disabled=True) 60 + for key, info in app.items(): 61 + assert info.get("source") == "app", f"{key} should be app" 62 + 63 + 64 + def test_property_tags_output(): 65 + """Property tags show output, tools, hook.""" 66 + assert _property_tags({"output": "md"}) == "output:md" 67 + assert _property_tags({"tools": "journal, todo"}) == "tools:journal, todo" 68 + assert _property_tags({"hook": "occurrence"}) == "hook:occurrence" 69 + 70 + tags = _property_tags({"output": "md", "hook": "occurrence"}) 71 + assert "output:md" in tags 72 + assert "hook:occurrence" in tags 73 + 74 + assert _property_tags({}) == "" 75 + assert "disabled" in _property_tags({"disabled": True}) 76 + 77 + 78 + def test_property_tags_tools_list(): 79 + """Property tags handle tools as a list.""" 80 + tags = _property_tags({"tools": ["journal", "todo"]}) 81 + assert tags == "tools:journal,todo" 82 + 83 + 84 + def test_scan_variables(): 85 + """Variable scanning finds template variables in prompt body.""" 86 + assert "name" in _scan_variables("Hello $name, welcome") 87 + assert "daily_preamble" in _scan_variables("$daily_preamble\n\n# Title") 88 + assert _scan_variables("No variables here") == [] 89 + # Deduplicates 90 + result = _scan_variables("$foo and $bar and $foo again") 91 + assert result == ["foo", "bar"] 92 + 93 + 94 + def test_list_prompts_output(capsys): 95 + """List view outputs expected groups and prompts.""" 96 + list_prompts() 97 + output = capsys.readouterr().out 98 + 99 + assert "segment:" in output 100 + assert "daily:" in output 101 + assert "activity" in output 102 + assert "flow" in output 103 + 104 + 105 + def test_list_prompts_schedule_filter(capsys): 106 + """Schedule filter shows only matching group.""" 107 + list_prompts(schedule="segment") 108 + output = capsys.readouterr().out 109 + 110 + assert "activity" in output 111 + # Should not show daily-only prompts 112 + # (but don't assert group headers since they're suppressed with filter) 113 + 114 + 115 + def test_list_prompts_disabled_shown(capsys): 116 + """--disabled includes disabled prompts.""" 117 + list_prompts(include_disabled=True) 118 + output = capsys.readouterr().out 119 + 120 + # files.md is disabled, should appear 121 + assert "files" in output 122 + 123 + 124 + def test_show_prompt_known(capsys): 125 + """Detail view shows expected fields for a known prompt.""" 126 + show_prompt("flow") 127 + output = capsys.readouterr().out 128 + 129 + assert "muse/flow.md" in output 130 + assert "title:" in output 131 + assert "schedule:" in output 132 + assert "daily" in output 133 + assert "hook:" in output 134 + assert "occurrence" in output 135 + assert "variables:" in output 136 + assert "$daily_preamble" in output 137 + assert "body:" in output 138 + assert "lines" in output 139 + 140 + 141 + def test_show_prompt_not_found(capsys): 142 + """Detail view exits with error for unknown prompt.""" 143 + with pytest.raises(SystemExit): 144 + show_prompt("nonexistent_prompt_xyz") 145 + 146 + output = capsys.readouterr().err 147 + assert "not found" in output.lower() 148 + 149 + 150 + def test_json_output_format(capsys): 151 + """JSON output produces valid JSONL with file field.""" 152 + json_output() 153 + output = capsys.readouterr().out 154 + 155 + lines = [x for x in output.strip().splitlines() if x.strip()] 156 + assert len(lines) > 0 157 + 158 + for line in lines: 159 + record = json.loads(line) 160 + assert "file" in record, f"Missing 'file' key in: {line}" 161 + assert record["file"].endswith(".md") 162 + 163 + 164 + def test_json_output_contains_known_prompts(capsys): 165 + """JSON output includes known prompts with expected fields.""" 166 + json_output(include_disabled=True) 167 + output = capsys.readouterr().out 168 + 169 + records = [json.loads(x) for x in output.strip().splitlines() if x.strip()] 170 + files = {r["file"] for r in records} 171 + assert any("flow.md" in f for f in files) 172 + assert any("activity.md" in f for f in files) 173 + 174 + # Check a specific record has expected fields 175 + flow = next(r for r in records if "flow.md" in r["file"]) 176 + assert "title" in flow 177 + assert "schedule" in flow 178 + 179 + 180 + def test_json_output_schedule_filter(capsys): 181 + """JSON output respects schedule filter.""" 182 + json_output(schedule="segment") 183 + output = capsys.readouterr().out 184 + 185 + records = [json.loads(x) for x in output.strip().splitlines() if x.strip()] 186 + for r in records: 187 + assert r.get("schedule") == "segment", f"Expected segment: {r}" 188 + 189 + 190 + def test_show_prompt_as_json(capsys): 191 + """Detail view with --json outputs single JSONL record.""" 192 + show_prompt("flow", as_json=True) 193 + output = capsys.readouterr().out 194 + 195 + lines = [x for x in output.strip().splitlines() if x.strip()] 196 + assert len(lines) == 1 197 + 198 + record = json.loads(lines[0]) 199 + assert record["file"].endswith("flow.md") 200 + assert "title" in record 201 + assert "schedule" in record 202 + # Should not contain expanded instruction text 203 + assert "system_instruction" not in record

+346

think/muse_cli.py

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """CLI for inspecting muse prompt configurations. 5 + 6 + Lists all system and app prompts with their frontmatter metadata, 7 + supports filtering by schedule and source, and provides detail views. 8 + 9 + Usage: 10 + sol muse List all prompts grouped by schedule 11 + sol muse <name> Show details for a specific prompt 12 + sol muse <name> --json Output a single prompt as JSONL 13 + sol muse --json Output all configs as JSONL 14 + sol muse --schedule daily Filter by schedule type 15 + """ 16 + 17 + from __future__ import annotations 18 + 19 + import argparse 20 + import json 21 + import re 22 + import sys 23 + from pathlib import Path 24 + from typing import Any 25 + 26 + import frontmatter 27 + 28 + from think.utils import ( 29 + MUSE_DIR, 30 + _load_prompt_metadata, 31 + get_muse_configs, 32 + setup_cli, 33 + ) 34 + 35 + # Project root for computing relative paths 36 + _PROJECT_ROOT = Path(__file__).parent.parent 37 + 38 + # Keys injected by get_agent() or internal bookkeeping — not frontmatter 39 + _SKIP_KEYS = frozenset( 40 + { 41 + "path", 42 + "mtime", 43 + "hook_path", 44 + "system_instruction", 45 + "user_instruction", 46 + "extra_context", 47 + "system_prompt_name", 48 + "sources", 49 + } 50 + ) 51 + 52 + 53 + def _relative_path(abs_path: str) -> str: 54 + """Convert absolute path to project-relative path.""" 55 + try: 56 + return str(Path(abs_path).relative_to(_PROJECT_ROOT)) 57 + except ValueError: 58 + return abs_path 59 + 60 + 61 + def _resolve_md_path(name: str) -> Path: 62 + """Resolve a prompt name to its .md file path.""" 63 + if ":" in name: 64 + app, agent_name = name.split(":", 1) 65 + return _PROJECT_ROOT / "apps" / app / "muse" / f"{agent_name}.md" 66 + return MUSE_DIR / f"{name}.md" 67 + 68 + 69 + def _resolve_file_path(key: str, info: dict[str, Any]) -> str: 70 + """Resolve the relative file path for a config entry.""" 71 + if info.get("path"): 72 + return _relative_path(str(info["path"])) 73 + # Configs loaded via get_agent() lose their path — reconstruct from key 74 + return _relative_path(str(_resolve_md_path(key))) 75 + 76 + 77 + def _scan_variables(body: str) -> list[str]: 78 + """Scan prompt body text for $template variables.""" 79 + # Match $word or ${word} but not $$ (escaped dollar signs) 80 + matches = re.findall(r"(?<!\$)\$\{?([a-zA-Z_]\w*)\}?", body) 81 + # Deduplicate preserving order 82 + seen: set[str] = set() 83 + result: list[str] = [] 84 + for m in matches: 85 + if m not in seen: 86 + seen.add(m) 87 + result.append(m) 88 + return result 89 + 90 + 91 + def _property_tags(info: dict[str, Any]) -> str: 92 + """Build compact property tags string for list view.""" 93 + tags: list[str] = [] 94 + 95 + if info.get("output"): 96 + tags.append(f"output:{info['output']}") 97 + 98 + if info.get("tools"): 99 + tools = info["tools"] 100 + if isinstance(tools, list): 101 + tools = ",".join(tools) 102 + tags.append(f"tools:{tools}") 103 + 104 + if info.get("hook"): 105 + tags.append(f"hook:{info['hook']}") 106 + 107 + if info.get("disabled"): 108 + tags.append("disabled") 109 + 110 + return " ".join(tags) 111 + 112 + 113 + def _collect_configs( 114 + *, 115 + schedule: str | None = None, 116 + source: str | None = None, 117 + include_disabled: bool = False, 118 + ) -> dict[str, dict[str, Any]]: 119 + """Collect all muse configs with optional filters applied.""" 120 + configs = get_muse_configs(schedule=schedule, include_disabled=True) 121 + 122 + filtered: dict[str, dict[str, Any]] = {} 123 + for key, info in configs.items(): 124 + if not include_disabled and info.get("disabled", False): 125 + continue 126 + if source and info.get("source") != source: 127 + continue 128 + filtered[key] = info 129 + 130 + return filtered 131 + 132 + 133 + def _to_jsonl_record(key: str, info: dict[str, Any]) -> dict[str, Any]: 134 + """Build a clean JSONL record from a config entry.""" 135 + record: dict[str, Any] = {"file": _resolve_file_path(key, info)} 136 + for k, v in info.items(): 137 + if k not in _SKIP_KEYS: 138 + record[k] = v 139 + return record 140 + 141 + 142 + def list_prompts( 143 + *, 144 + schedule: str | None = None, 145 + source: str | None = None, 146 + include_disabled: bool = False, 147 + ) -> None: 148 + """Print prompts grouped by schedule.""" 149 + all_configs = get_muse_configs(include_disabled=True) 150 + configs = _collect_configs( 151 + schedule=schedule, source=source, include_disabled=include_disabled 152 + ) 153 + 154 + if not configs: 155 + print("No prompts found matching filters.") 156 + return 157 + 158 + # Group by schedule 159 + groups: dict[str, list[tuple[str, dict[str, Any]]]] = { 160 + "segment": [], 161 + "daily": [], 162 + "unscheduled": [], 163 + } 164 + 165 + for key, info in sorted(configs.items()): 166 + sched = info.get("schedule") 167 + if sched in ("segment", "daily"): 168 + groups[sched].append((key, info)) 169 + else: 170 + groups["unscheduled"].append((key, info)) 171 + 172 + # Compute column width from longest name 173 + all_names = list(configs.keys()) 174 + name_width = max(len(n) for n in all_names) if all_names else 20 175 + name_width = max(name_width, 10) 176 + 177 + # Print each non-empty group 178 + for group_name in ("segment", "daily", "unscheduled"): 179 + items = groups[group_name] 180 + if not items: 181 + continue 182 + 183 + # Skip group header if filtering to a single schedule 184 + if not schedule: 185 + print(f"{group_name}:") 186 + 187 + for key, info in items: 188 + title = info.get("title", "") 189 + tags = _property_tags(info) 190 + src = "" 191 + if info.get("source") == "app": 192 + src = f" [{info.get('app', 'app')}]" 193 + 194 + line = f" {key:<{name_width}} {title:<32} {tags}{src}" 195 + print(line.rstrip()) 196 + 197 + if not schedule: 198 + print() 199 + 200 + # Show disabled count hint 201 + if not include_disabled: 202 + disabled_count = sum( 203 + 1 for info in all_configs.values() if info.get("disabled", False) 204 + ) 205 + if disabled_count: 206 + total = len(configs) 207 + print(f"{total} prompts ({disabled_count} disabled hidden, use --disabled)") 208 + 209 + 210 + def show_prompt(name: str, *, as_json: bool = False) -> None: 211 + """Print detailed info for a single prompt.""" 212 + md_path = _resolve_md_path(name) 213 + 214 + if not md_path.exists(): 215 + print(f"Prompt not found: {name}", file=sys.stderr) 216 + print(f" looked at: {_relative_path(str(md_path))}", file=sys.stderr) 217 + sys.exit(1) 218 + 219 + info = _load_prompt_metadata(md_path) 220 + rel_path = _relative_path(str(md_path)) 221 + 222 + # Load body once for variables and line count 223 + try: 224 + post = frontmatter.load(md_path) 225 + body = post.content.strip() 226 + except Exception: 227 + body = None 228 + 229 + if as_json: 230 + record = _to_jsonl_record(name, info) 231 + print(json.dumps(record, default=str)) 232 + return 233 + 234 + print(f"\n{rel_path}\n") 235 + 236 + # Display frontmatter fields 237 + # Order: title, description, then alphabetical for the rest 238 + priority_keys = [ 239 + "title", 240 + "description", 241 + "schedule", 242 + "output", 243 + "tools", 244 + "hook", 245 + "color", 246 + ] 247 + skip_keys = {"path", "mtime", "hook_path"} 248 + 249 + label_width = 14 250 + 251 + def print_field(key: str, value: Any) -> None: 252 + if key in skip_keys: 253 + return 254 + val_str = str(value) 255 + # Truncate long descriptions for readability 256 + if key == "description" and len(val_str) > 72: 257 + val_str = val_str[:72] + "..." 258 + # Show hook path inline 259 + if key == "hook" and info.get("hook_path"): 260 + val_str += f" \u2192 {_relative_path(str(info['hook_path']))}" 261 + print(f" {key + ':':<{label_width}} {val_str}") 262 + 263 + printed: set[str] = set() 264 + for key in priority_keys: 265 + if key in info and key not in skip_keys: 266 + print_field(key, info[key]) 267 + printed.add(key) 268 + 269 + # Remaining fields alphabetically 270 + for key in sorted(info.keys()): 271 + if key not in printed and key not in skip_keys: 272 + print_field(key, info[key]) 273 + 274 + # Template variables and body line count from single parse 275 + if body is not None: 276 + variables = _scan_variables(body) 277 + if variables: 278 + vars_str = ", ".join(f"${v}" for v in variables) 279 + print(f" {'variables:':<{label_width}} {vars_str}") 280 + 281 + line_count = len(body.splitlines()) 282 + print(f" {'body:':<{label_width}} {line_count} lines") 283 + 284 + print() 285 + 286 + 287 + def json_output( 288 + *, 289 + schedule: str | None = None, 290 + source: str | None = None, 291 + include_disabled: bool = False, 292 + ) -> None: 293 + """Print JSONL output with one config per line, including filename.""" 294 + configs = _collect_configs( 295 + schedule=schedule, source=source, include_disabled=include_disabled 296 + ) 297 + 298 + for key, info in sorted(configs.items()): 299 + print(json.dumps(_to_jsonl_record(key, info), default=str)) 300 + 301 + 302 + def main() -> None: 303 + """Entry point for sol muse.""" 304 + parser = argparse.ArgumentParser(description="Inspect muse prompt configurations") 305 + parser.add_argument("name", nargs="?", help="Show details for a specific prompt") 306 + parser.add_argument( 307 + "--schedule", 308 + choices=["daily", "segment"], 309 + help="Filter by schedule type", 310 + ) 311 + parser.add_argument( 312 + "--source", 313 + choices=["system", "app"], 314 + help="Filter by origin", 315 + ) 316 + parser.add_argument( 317 + "--disabled", 318 + action="store_true", 319 + help="Include disabled prompts", 320 + ) 321 + parser.add_argument( 322 + "--json", 323 + action="store_true", 324 + help="Output as JSONL (one JSON object per line)", 325 + ) 326 + 327 + args = setup_cli(parser) 328 + 329 + if args.name: 330 + show_prompt(args.name, as_json=args.json) 331 + elif args.json: 332 + json_output( 333 + schedule=args.schedule, 334 + source=args.source, 335 + include_disabled=args.disabled, 336 + ) 337 + else: 338 + list_prompts( 339 + schedule=args.schedule, 340 + source=args.source, 341 + include_disabled=args.disabled, 342 + ) 343 + 344 + 345 + if __name__ == "__main__": 346 + main()

Configure Feed

Configure Feed