From 7b19c680c110c3e519ce0989ca4e154d7f5e7afa Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Mon, 2 Feb 2026 12:09:48 -0700 Subject: [PATCH] Extract muse utilities into dedicated think/muse.py module Consolidate all muse-related functionality from think/utils.py into a new think/muse.py module for better organization and maintainability: - Move prompt loading (load_prompt, PromptContent, PromptNotFoundError) - Move config discovery (get_muse_configs, get_agent, _load_prompt_metadata) - Move instruction composition (compose_instructions, _merge_instructions_config) - Move source helpers (source_is_enabled, source_is_required, get_agent_filter) - Move output path utilities (key_to_context, get_output_topic, get_output_path) - Move hook loading (load_pre_hook, load_post_hook, _resolve_hook_path) Update all consumers to import from think.muse instead of think.utils. Create tests/test_muse.py with dedicated tests for the new module. Update documentation references from think/utils.py to think/muse.py. No legacy re-exports or backwards compatibility - all usages updated directly. Co-Authored-By: Claude Opus 4.5 --- apps/agents/routes.py | 4 +- apps/calendar/routes.py | 2 +- apps/chat/routes.py | 2 +- apps/insights/routes.py | 5 +- apps/settings/routes.py | 8 +- apps/stats/routes.py | 2 +- docs/APPS.md | 8 +- docs/CORTEX.md | 2 +- docs/JOURNAL.md | 2 +- docs/PROMPT_TEMPLATES.md | 14 +- docs/THINK.md | 2 +- muse/anticipation.py | 2 +- muse/occurrence.py | 2 +- observe/describe.py | 3 +- observe/enrich.py | 2 +- observe/extract.py | 2 +- observe/transcribe/gemini.py | 2 +- tests/test_app_agents.py | 2 +- tests/test_dream_full.py | 2 +- tests/test_entity_agents.py | 2 +- tests/test_generators.py | 42 +- tests/test_muse.py | 324 +++++++++ tests/test_output_hooks.py | 4 +- tests/test_output_path.py | 2 +- tests/test_template_substitution.py | 2 +- tests/test_think_utils.py | 331 +--------- think/agents.py | 108 +-- think/cortex.py | 5 +- think/detect_created.py | 2 +- think/detect_transcript.py | 2 +- think/dream.py | 11 +- think/hooks.py | 3 +- think/models.py | 63 +- think/muse.py | 984 ++++++++++++++++++++++++++++ think/muse_cli.py | 6 +- think/planner.py | 3 +- think/utils.py | 824 +---------------------- 37 files changed, 1424 insertions(+), 1362 deletions(-) create mode 100644 tests/test_muse.py create mode 100644 think/muse.py diff --git a/apps/agents/routes.py b/apps/agents/routes.py index 2286bafed..2645359cb 100644 --- a/apps/agents/routes.py +++ b/apps/agents/routes.py @@ -18,7 +18,7 @@ from convey import state from convey.utils import DATE_RE, format_date from think.facets import get_facets from think.models import calc_agent_cost -from think.utils import get_muse_configs +from think.muse import get_muse_configs agents_bp = Blueprint( "app:agents", @@ -357,7 +357,7 @@ def api_preview_prompt(name: str) -> Any: } """ try: - from think.utils import get_agent + from think.muse import get_agent config = get_agent(name) diff --git a/apps/calendar/routes.py b/apps/calendar/routes.py index 83c875a92..9d58763e0 100644 --- a/apps/calendar/routes.py +++ b/apps/calendar/routes.py @@ -55,7 +55,7 @@ def calendar_day_events(day: str) -> Any: return "", 404 from think.indexer.journal import get_events - from think.utils import get_muse_configs + from think.muse import get_muse_configs generators = get_muse_configs(has_tools=False, has_output=True) diff --git a/apps/chat/routes.py b/apps/chat/routes.py index a2ab76fc0..5b14f008e 100644 --- a/apps/chat/routes.py +++ b/apps/chat/routes.py @@ -113,7 +113,7 @@ def _check_provider_api_key(provider: str) -> str | None: def generate_chat_title(message: str) -> str: """Generate a short title for a chat message using configured provider.""" - from think.utils import load_prompt + from think.muse import load_prompt prompt = load_prompt("title", base_dir=Path(__file__).parent) try: diff --git a/apps/insights/routes.py b/apps/insights/routes.py index 0056ea367..70b01834b 100644 --- a/apps/insights/routes.py +++ b/apps/insights/routes.py @@ -15,7 +15,8 @@ from flask import Blueprint, jsonify, redirect, render_template, url_for from convey.utils import DATE_RE, format_date from think.models import get_usage_cost -from think.utils import day_dirs, day_path, get_muse_configs, get_output_topic +from think.muse import get_muse_configs, get_output_topic +from think.utils import day_dirs, day_path insights_bp = Blueprint( "app:insights", @@ -85,7 +86,7 @@ def insights_day(day: str) -> str: meta = info["meta"] # Get generation cost for this generator - from think.utils import key_to_context + from think.muse import key_to_context cost_data = get_usage_cost(day, context=key_to_context(key)) cost = cost_data["cost"] if cost_data["cost"] > 0 else None diff --git a/apps/settings/routes.py b/apps/settings/routes.py index 051257356..c352961a1 100644 --- a/apps/settings/routes.py +++ b/apps/settings/routes.py @@ -260,8 +260,8 @@ def get_providers() -> Any: DEFAULT_TIER, get_context_registry, ) + from think.muse import get_muse_configs from think.providers import get_provider_list - from think.utils import get_muse_configs config = get_journal_config() providers_config = config.get("providers", {}) @@ -287,7 +287,7 @@ def get_providers() -> Any: context_defaults[pattern]["has_tools"] = ctx_config["has_tools"] # Enhance muse contexts with additional metadata from get_muse_configs - from think.utils import key_to_context + from think.muse import key_to_context muse_configs = get_muse_configs(include_disabled=True) for key, info in muse_configs.items(): @@ -543,7 +543,7 @@ def get_generators() -> Any: - daily: List of daily-schedule generators """ try: - from think.utils import get_muse_configs + from think.muse import get_muse_configs # Get all generators (has output but no tools) all_generators = get_muse_configs( @@ -599,7 +599,7 @@ def update_generators() -> Any: old_contexts = old_providers.get("contexts", {}) changed_fields = {} - from think.utils import key_to_context + from think.muse import key_to_context for key, updates in request_data.items(): if not isinstance(updates, dict): diff --git a/apps/stats/routes.py b/apps/stats/routes.py index ba65c2997..02253d942 100644 --- a/apps/stats/routes.py +++ b/apps/stats/routes.py @@ -10,7 +10,7 @@ from typing import Any from flask import Blueprint, jsonify from convey import state -from think.utils import get_muse_configs +from think.muse import get_muse_configs stats_bp = Blueprint( "app:stats", diff --git a/docs/APPS.md b/docs/APPS.md index 0a4f8948f..7398638c9 100644 --- a/docs/APPS.md +++ b/docs/APPS.md @@ -333,8 +333,8 @@ def post_process(result: str, context: dict) -> str | None: **Reference implementations:** - System generator templates: `muse/*.md` (files with `schedule` field but no `tools` field) - Extraction hooks: `muse/occurrence.py`, `muse/anticipation.py` -- Discovery logic: `think/utils.py` - `get_muse_configs(has_tools=False)`, `get_output_topic()` -- Hook loading: `think/agents.py` - `load_pre_hook()`, `load_post_hook()` +- Discovery logic: `think/muse.py` - `get_muse_configs(has_tools=False)`, `get_output_topic()` +- Hook loading: `think/muse.py` - `load_pre_hook()`, `load_post_hook()` --- @@ -356,7 +356,7 @@ Define custom agents and generator templates that integrate with solstone's Cort **Reference implementations:** - System agent examples: `muse/*.md` (files with `tools` field) -- Discovery logic: `think/utils.py` - `get_muse_configs(has_tools=True)`, `get_agent()` +- Discovery logic: `think/muse.py` - `get_muse_configs(has_tools=True)`, `get_agent()` #### Instructions Configuration @@ -380,7 +380,7 @@ Both insights and agents support an optional `instructions` key for customizing - `"required"` - load, and skip generation if no content found (useful for generators that only make sense with specific input types, e.g., `"audio": "required"` for speaker detection) - For `agents` only: a dict for selective filtering, e.g., `{"entities": true, "meetings": "required", "flow": false}`. Keys are agent names (system) or `"app:topic"` (app-namespaced). An empty dict `{}` means no agents. -**Authoritative source:** `think/utils.py` - `compose_instructions()`, `_DEFAULT_INSTRUCTIONS`, `source_is_enabled()`, `source_is_required()`, `get_agent_filter()` +**Authoritative source:** `think/muse.py` - `compose_instructions()`, `_DEFAULT_INSTRUCTIONS`, `source_is_enabled()`, `source_is_required()`, `get_agent_filter()` --- diff --git a/docs/CORTEX.md b/docs/CORTEX.md index d87b64fd3..d651d51c4 100644 --- a/docs/CORTEX.md +++ b/docs/CORTEX.md @@ -264,7 +264,7 @@ Agents use configurations stored in the `muse/` directory. Each agent is a `.md` When spawning an agent: 1. Cortex passes the raw request to `sol agents` via stdin (NDJSON format) 2. The agent process (`think/agents.py`) handles all config loading via `hydrate_config()`: - - Loads agent configuration using `get_agent()` from `think/utils.py` + - Loads agent configuration using `get_agent()` from `think/muse.py` - Merges request parameters with agent defaults - Resolves provider and model based on context - Expands tool pack names to tool lists diff --git a/docs/JOURNAL.md b/docs/JOURNAL.md index 38b3898b4..b4aee57f5 100644 --- a/docs/JOURNAL.md +++ b/docs/JOURNAL.md @@ -971,7 +971,7 @@ Post-processing generates day-level outputs in the `agents/` directory that synt - `muse/*.md` – system generator templates (files with `schedule` field but no `tools` field) - `apps/{app}/muse/*.md` – app-specific generator templates -Each template is a `.md` file with JSON frontmatter containing metadata (title, description, schedule, output format). The `schedule` field is required and must be `"segment"` or `"daily"` - generators with missing or invalid schedule are skipped. Use `get_muse_configs(has_tools=False)` from `think/utils.py` to retrieve all available generators, or `get_muse_configs(has_tools=False, schedule="daily")` to get generators filtered by schedule. +Each template is a `.md` file with JSON frontmatter containing metadata (title, description, schedule, output format). The `schedule` field is required and must be `"segment"` or `"daily"` - generators with missing or invalid schedule are skipped. Use `get_muse_configs(has_tools=False)` from `think/muse.py` to retrieve all available generators, or `get_muse_configs(has_tools=False, schedule="daily")` to get generators filtered by schedule. **Output naming:** - System outputs: `agents/{topic}.md` (e.g., `agents/flow.md`, `agents/meetings.md`) diff --git a/docs/PROMPT_TEMPLATES.md b/docs/PROMPT_TEMPLATES.md index 03f7ff11a..7cda5231b 100644 --- a/docs/PROMPT_TEMPLATES.md +++ b/docs/PROMPT_TEMPLATES.md @@ -4,7 +4,7 @@ This document describes solstone's template variable system for personalizing pr ## Overview -Prompts are stored as `.md` files with optional JSON frontmatter for metadata. The prompt content is loaded via `load_prompt()` from `think/utils.py`, which uses Python's `string.Template` with `safe_substitute`. This means: +Prompts are stored as `.md` files with optional JSON frontmatter for metadata. The prompt content is loaded via `load_prompt()` from `think/muse.py`, which uses Python's `string.Template` with `safe_substitute`. This means: - Variables use `$name` or `${name}` syntax - Undefined variables are left as-is (no errors) @@ -62,7 +62,7 @@ The flattening logic converts nested objects using underscore separators. For ex **References:** - Identity configuration: [JOURNAL.md](JOURNAL.md) (identity section) -- Flattening implementation: `think/utils.py` → `_flatten_identity_to_template_vars()` +- Flattening implementation: `think/muse.py` → `_flatten_identity_to_template_vars()` ### Template Variables @@ -144,7 +144,7 @@ The system instruction establishes the journal partnership context. The user ins **Optional model configuration:** Add `max_output_tokens` (response length limit) and `thinking_budget` (model thinking token budget) to override provider defaults. Note: OpenAI uses fixed reasoning and ignores `thinking_budget`. -**Reference:** `think/utils.py` → `get_agent()` for agent configuration loading +**Reference:** `think/muse.py` → `get_agent()` for agent configuration loading ### The load_prompt() Function @@ -159,7 +159,7 @@ load_prompt( Returns a `PromptContent` named tuple with `text` (substituted content), `path` (source file), and `metadata` (frontmatter dict). -**Reference:** `think/utils.py` → `load_prompt()` +**Reference:** `think/muse.py` → `load_prompt()` ## Adding New Variables @@ -184,9 +184,9 @@ load_prompt("myprompt", context={"custom_var": "value"}) | Category | Authoritative Source | |----------|---------------------| | Identity config schema | [JOURNAL.md](JOURNAL.md) (identity section) | -| Identity flattening | `think/utils.py` (`_flatten_identity_to_template_vars`) | -| Template loading | `think/utils.py` (`_load_templates`) | -| Core load function | `think/utils.py` (`load_prompt`) | +| Identity flattening | `think/muse.py` (`_flatten_identity_to_template_vars`) | +| Template loading | `think/muse.py` (`_load_templates`) | +| Core load function | `think/muse.py` (`load_prompt`) | | Template files | `think/templates/*.md` | | Test coverage | `tests/test_template_substitution.py` | | Generator prompts | `muse/*.md` (files with `schedule` field but no `tools`) | diff --git a/docs/THINK.md b/docs/THINK.md index f9e7fce9c..2722801a1 100644 --- a/docs/THINK.md +++ b/docs/THINK.md @@ -179,7 +179,7 @@ which automatically routes to the configured provider based on context. ## Generator map keys -`think.utils.get_muse_configs(has_tools=False)` reads the `.md` prompt files under `muse/` and +`think.muse.get_muse_configs(has_tools=False)` reads the `.md` prompt files under `muse/` and returns a dictionary keyed by generator name. Each entry contains: - `path` – the prompt file path diff --git a/muse/anticipation.py b/muse/anticipation.py index 553dc84ee..3053ff773 100644 --- a/muse/anticipation.py +++ b/muse/anticipation.py @@ -19,7 +19,7 @@ from think.hooks import ( write_events_jsonl, ) from think.models import generate -from think.utils import get_output_topic, load_prompt +from think.muse import get_output_topic, load_prompt def post_process(result: str, context: dict) -> str | None: diff --git a/muse/occurrence.py b/muse/occurrence.py index 07f00f05d..045b28767 100644 --- a/muse/occurrence.py +++ b/muse/occurrence.py @@ -19,7 +19,7 @@ from think.hooks import ( write_events_jsonl, ) from think.models import generate -from think.utils import get_output_topic, load_prompt +from think.muse import get_output_topic, load_prompt def post_process(result: str, context: dict) -> str | None: diff --git a/observe/describe.py b/observe/describe.py index e7bcd0d7a..a6e39c97b 100644 --- a/observe/describe.py +++ b/observe/describe.py @@ -35,7 +35,8 @@ from observe.aruco import detect_markers, mask_convey_region, polygon_area from observe.extract import DEFAULT_MAX_EXTRACTIONS, select_frames_for_extraction from observe.utils import get_segment_key from think.callosum import callosum_send -from think.utils import get_config, get_journal, load_prompt, setup_cli +from think.muse import load_prompt +from think.utils import get_config, get_journal, setup_cli logger = logging.getLogger(__name__) diff --git a/observe/enrich.py b/observe/enrich.py index 2563359dd..ea03f82fb 100644 --- a/observe/enrich.py +++ b/observe/enrich.py @@ -23,7 +23,7 @@ from google.genai import types from observe.utils import audio_to_flac_bytes from think.models import generate -from think.utils import load_prompt +from think.muse import load_prompt logger = logging.getLogger(__name__) diff --git a/observe/extract.py b/observe/extract.py index 734afdcaf..b9551853c 100644 --- a/observe/extract.py +++ b/observe/extract.py @@ -205,7 +205,7 @@ def _ai_select_frames( If AI selection fails (will trigger fallback in caller). """ from think.models import generate - from think.utils import load_prompt + from think.muse import load_prompt # Build extraction guidance with config overrides extraction_guidance = _build_extraction_guidance(categories, config_overrides) diff --git a/observe/transcribe/gemini.py b/observe/transcribe/gemini.py index a7e061a33..1be9299af 100644 --- a/observe/transcribe/gemini.py +++ b/observe/transcribe/gemini.py @@ -32,7 +32,7 @@ from google.genai import types from observe.utils import audio_to_flac_bytes from think.models import generate -from think.utils import load_prompt +from think.muse import load_prompt logger = logging.getLogger(__name__) diff --git a/tests/test_app_agents.py b/tests/test_app_agents.py index b95f56067..587b24858 100644 --- a/tests/test_app_agents.py +++ b/tests/test_app_agents.py @@ -8,7 +8,7 @@ import os import pytest -from think.utils import _resolve_agent_path, get_agent, get_muse_configs +from think.muse import _resolve_agent_path, get_agent, get_muse_configs @pytest.fixture diff --git a/tests/test_dream_full.py b/tests/test_dream_full.py index 9215497f8..58dee1f81 100644 --- a/tests/test_dream_full.py +++ b/tests/test_dream_full.py @@ -106,7 +106,7 @@ def test_segment_mode_skips_pre_post_phases(tmp_path, monkeypatch): def test_priority_validation_required(tmp_path, monkeypatch): """Test that get_muse_configs raises error for scheduled prompts without priority.""" - from think.utils import get_muse_configs + from think.muse import get_muse_configs # Create a test muse file without priority muse_dir = Path(__file__).parent.parent / "muse" diff --git a/tests/test_entity_agents.py b/tests/test_entity_agents.py index 9d749b894..67dc6c242 100644 --- a/tests/test_entity_agents.py +++ b/tests/test_entity_agents.py @@ -7,7 +7,7 @@ import os import pytest -from think.utils import get_agent +from think.muse import get_agent @pytest.fixture diff --git a/tests/test_generators.py b/tests/test_generators.py index 66f383569..b6536a5d2 100644 --- a/tests/test_generators.py +++ b/tests/test_generators.py @@ -7,8 +7,8 @@ import os def test_get_muse_configs_generators(): """Test that system generators are discovered with source field.""" - utils = importlib.import_module("think.utils") - generators = utils.get_muse_configs(has_tools=False, has_output=True) + muse = importlib.import_module("think.muse") + generators = muse.get_muse_configs(has_tools=False, has_output=True) assert "flow" in generators info = generators["flow"] assert os.path.basename(info["path"]) == "flow.md" @@ -22,20 +22,20 @@ def test_get_muse_configs_generators(): def test_get_output_topic(): """Test generator key to filename conversion.""" - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") # System generators: key unchanged - assert utils.get_output_topic("activity") == "activity" - assert utils.get_output_topic("flow") == "flow" + assert muse.get_output_topic("activity") == "activity" + assert muse.get_output_topic("flow") == "flow" # App generators: _app_topic format - assert utils.get_output_topic("chat:sentiment") == "_chat_sentiment" - assert utils.get_output_topic("my_app:weekly_summary") == "_my_app_weekly_summary" + assert muse.get_output_topic("chat:sentiment") == "_chat_sentiment" + assert muse.get_output_topic("my_app:weekly_summary") == "_my_app_weekly_summary" def test_get_muse_configs_app_discovery(tmp_path, monkeypatch): """Test that app generators are discovered from apps/*/muse/.""" - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") # Create a fake app with a generator app_dir = tmp_path / "apps" / "test_app" / "muse" @@ -50,7 +50,7 @@ def test_get_muse_configs_app_discovery(tmp_path, monkeypatch): (tmp_path / "apps" / "test_app" / "workspace.html").write_text("

Test

") # For now, just verify system generators have correct source - generators = utils.get_muse_configs(has_tools=False, has_output=True) + generators = muse.get_muse_configs(has_tools=False, has_output=True) for key, info in generators.items(): if ":" not in key: assert info.get("source") == "system", f"{key} should have source=system" @@ -58,16 +58,16 @@ def test_get_muse_configs_app_discovery(tmp_path, monkeypatch): def test_get_muse_configs_by_schedule(): """Test filtering generators by schedule.""" - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") # Get daily generators - daily = utils.get_muse_configs(has_tools=False, has_output=True, schedule="daily") + daily = muse.get_muse_configs(has_tools=False, has_output=True, schedule="daily") assert len(daily) > 0 for key, meta in daily.items(): assert meta.get("schedule") == "daily", f"{key} should have schedule=daily" # Get segment generators - segment = utils.get_muse_configs( + segment = muse.get_muse_configs( has_tools=False, has_output=True, schedule="segment" ) assert len(segment) > 0 @@ -81,23 +81,23 @@ def test_get_muse_configs_by_schedule(): # Unknown schedule returns empty dict assert ( - utils.get_muse_configs(has_tools=False, has_output=True, schedule="hourly") + muse.get_muse_configs(has_tools=False, has_output=True, schedule="hourly") == {} ) - assert utils.get_muse_configs(has_tools=False, has_output=True, schedule="") == {} + assert muse.get_muse_configs(has_tools=False, has_output=True, schedule="") == {} def test_get_muse_configs_include_disabled(monkeypatch): """Test include_disabled parameter.""" - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") # Get generators without disabled (default) - without_disabled = utils.get_muse_configs( + without_disabled = muse.get_muse_configs( has_tools=False, has_output=True, schedule="daily" ) # Get generators with disabled included - with_disabled = utils.get_muse_configs( + with_disabled = muse.get_muse_configs( has_tools=False, has_output=True, schedule="daily", include_disabled=True ) @@ -113,9 +113,9 @@ def test_scheduled_generators_have_valid_schedule(): Some generators (like importer) have output but no schedule - they're used for ad-hoc processing, not scheduled runs. """ - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") - generators = utils.get_muse_configs(has_tools=False, has_output=True) + generators = muse.get_muse_configs(has_tools=False, has_output=True) valid_schedules = ("segment", "daily") for key, meta in generators.items(): @@ -128,9 +128,9 @@ def test_scheduled_generators_have_valid_schedule(): def test_speakers_has_required_audio(): """Test that speakers generator has audio as required source.""" - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") - generators = utils.get_muse_configs( + generators = muse.get_muse_configs( has_tools=False, has_output=True, schedule="segment" ) assert "speakers" in generators diff --git a/tests/test_muse.py b/tests/test_muse.py new file mode 100644 index 000000000..cead75a85 --- /dev/null +++ b/tests/test_muse.py @@ -0,0 +1,324 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Tests for think.muse module. + +Tests for muse prompt loading, configuration, and instruction composition. +""" + +import pytest + +from think.muse import ( + _merge_instructions_config, + compose_instructions, + get_agent_filter, + source_is_enabled, + source_is_required, +) + + +# ============================================================================= +# _merge_instructions_config tests +# ============================================================================= + + +def test_merge_instructions_config_empty_overrides(): + """Test that empty overrides returns defaults copy.""" + defaults = {"system": "journal", "facets": True, "sources": {"audio": False}} + result = _merge_instructions_config(defaults, None) + assert result == defaults + assert result is not defaults # Should be a copy + + +def test_merge_instructions_config_with_overrides(): + """Test that overrides are merged correctly.""" + defaults = {"system": "journal", "facets": True, "sources": {"audio": False}} + overrides = {"system": "custom", "facets": False} + result = _merge_instructions_config(defaults, overrides) + assert result["system"] == "custom" + assert result["facets"] is False + assert result["sources"] == {"audio": False} # Preserved + + +def test_merge_instructions_config_sources_merge(): + """Test that sources dict is merged, not replaced.""" + defaults = {"system": None, "sources": {"audio": False, "screen": False}} + overrides = {"sources": {"audio": True}} + result = _merge_instructions_config(defaults, overrides) + assert result["sources"]["audio"] is True # Overridden + assert result["sources"]["screen"] is False # Preserved from defaults + + +def test_merge_instructions_config_ignores_unknown_keys(): + """Test that unknown keys in overrides are ignored.""" + defaults = {"system": "journal", "facets": True} + overrides = {"unknown_key": "value", "another": 123} + result = _merge_instructions_config(defaults, overrides) + assert "unknown_key" not in result + assert "another" not in result + + +def test_merge_instructions_config_facets_override(): + """Test that facets key can be overridden with different values.""" + defaults = {"system": "journal", "facets": True} + overrides = {"facets": "full"} + result = _merge_instructions_config(defaults, overrides) + assert result["system"] == "journal" + assert result["facets"] == "full" + + +# ============================================================================= +# compose_instructions tests +# ============================================================================= + + +class TestComposeInstructions: + """Tests for compose_instructions function.""" + + def test_default_system_instruction_is_none(self, monkeypatch, tmp_path): + """Test that default system instruction is empty (agents must opt-in).""" + think_dir = tmp_path / "think" + think_dir.mkdir() + + import think.muse + + original_file = think.muse.__file__ + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions() + + # Restore + monkeypatch.setattr(think.muse, "__file__", original_file) + + assert "system_instruction" in result + assert result["system_instruction"] == "" + assert result["system_prompt_name"] == "" + + def test_custom_system_instruction(self, monkeypatch, tmp_path): + """Test that custom system prompt can be loaded.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + custom_txt = think_dir / "custom.md" + custom_txt.write_text("Custom system instruction") + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions( + config_overrides={"system": "custom"}, + ) + + assert result["system_prompt_name"] == "custom" + assert "Custom system instruction" in result["system_instruction"] + + def test_user_instruction_loaded_when_provided(self, monkeypatch, tmp_path): + """Test that user instruction is loaded when user_prompt is provided.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + journal_txt = think_dir / "journal.md" + journal_txt.write_text("System instruction") + user_txt = think_dir / "default.md" + user_txt.write_text("User instruction content") + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions(user_prompt="default") + + assert result["user_instruction"] == "User instruction content" + + def test_user_instruction_none_when_not_provided(self, monkeypatch, tmp_path): + """Test that user instruction is None when user_prompt is not provided.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + journal_txt = think_dir / "journal.md" + journal_txt.write_text("System instruction") + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions() + + assert result["user_instruction"] is None + + def test_facets_none_excludes_facets_from_context(self, monkeypatch, tmp_path): + """Test that facets='none' excludes facet info from extra_context.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + journal_txt = think_dir / "journal.md" + journal_txt.write_text("System instruction") + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions( + include_datetime=False, + config_overrides={"facets": False}, + ) + + # With no datetime and no facets, extra_context should be empty/None + assert result["extra_context"] is None or result["extra_context"] == "" + + def test_include_datetime_false_excludes_time(self, monkeypatch, tmp_path): + """Test that include_datetime=False excludes time from context.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + journal_txt = think_dir / "journal.md" + journal_txt.write_text("System instruction") + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions( + include_datetime=False, + config_overrides={"facets": False}, + ) + + extra = result.get("extra_context") or "" + assert "Current Date and Time" not in extra + + def test_include_datetime_true_includes_time(self, monkeypatch, tmp_path): + """Test that include_datetime=True includes time in context.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + journal_txt = think_dir / "journal.md" + journal_txt.write_text("System instruction") + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions( + include_datetime=True, + config_overrides={"facets": False}, + ) + + assert "Current Date and Time" in result["extra_context"] + + def test_sources_returned_from_defaults(self, monkeypatch, tmp_path): + """Test that sources config is returned with defaults (all false).""" + think_dir = tmp_path / "think" + think_dir.mkdir() + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions() + + assert "sources" in result + assert result["sources"]["audio"] is False + assert result["sources"]["screen"] is False + assert result["sources"]["agents"] is False + + def test_sources_can_be_overridden(self, monkeypatch, tmp_path): + """Test that sources config can be overridden.""" + think_dir = tmp_path / "think" + think_dir.mkdir() + + import think.muse + + monkeypatch.setattr(think.muse, "__file__", str(think_dir / "muse.py")) + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) + + result = compose_instructions( + config_overrides={ + "sources": {"audio": True, "agents": True}, + }, + ) + + assert result["sources"]["audio"] is True # Overridden + assert result["sources"]["screen"] is False # Default preserved + assert result["sources"]["agents"] is True # Overridden + + +# ============================================================================= +# source_is_enabled / source_is_required / get_agent_filter tests +# ============================================================================= + + +def test_source_is_enabled_bool(): + """Test source_is_enabled with bool values.""" + assert source_is_enabled(True) is True + assert source_is_enabled(False) is False + + +def test_source_is_enabled_required_string(): + """Test source_is_enabled with 'required' string.""" + assert source_is_enabled("required") is True + + +def test_source_is_enabled_dict(): + """Test source_is_enabled with dict values for agents source.""" + # Dict with at least one True value -> enabled + assert source_is_enabled({"entities": True, "meetings": False}) is True + + # Dict with at least one "required" value -> enabled + assert source_is_enabled({"entities": "required", "meetings": False}) is True + + # Dict with all False values -> disabled + assert source_is_enabled({"entities": False, "meetings": False}) is False + + # Empty dict -> disabled + assert source_is_enabled({}) is False + + +def test_source_is_required_bool(): + """Test source_is_required with bool values.""" + assert source_is_required(True) is False + assert source_is_required(False) is False + + +def test_source_is_required_string(): + """Test source_is_required with 'required' string.""" + assert source_is_required("required") is True + + +def test_source_is_required_dict(): + """Test source_is_required with dict values.""" + # Dict with at least one "required" value -> required + assert source_is_required({"entities": "required", "meetings": False}) is True + + # Dict with no "required" values -> not required + assert source_is_required({"entities": True, "meetings": False}) is False + + # Empty dict -> not required + assert source_is_required({}) is False + + +def test_get_agent_filter_bool(): + """Test get_agent_filter with bool values.""" + # True -> None (all agents) + assert get_agent_filter(True) is None + + # False -> empty dict (no agents) + assert get_agent_filter(False) == {} + + +def test_get_agent_filter_required_string(): + """Test get_agent_filter with 'required' string.""" + # "required" -> None (all agents, required) + assert get_agent_filter("required") is None + + +def test_get_agent_filter_dict(): + """Test get_agent_filter with dict values.""" + # Dict -> returned as-is for filtering + filter_dict = {"entities": True, "meetings": "required", "flow": False} + assert get_agent_filter(filter_dict) == filter_dict + + # Empty dict -> empty dict (no agents) + assert get_agent_filter({}) == {} diff --git a/tests/test_output_hooks.py b/tests/test_output_hooks.py index 4b38013ab..6f5082ae2 100644 --- a/tests/test_output_hooks.py +++ b/tests/test_output_hooks.py @@ -150,7 +150,7 @@ def test_load_post_hook_file_not_found(tmp_path): def test_prompt_metadata_no_hook_path(tmp_path): """Test that _load_prompt_metadata no longer sets hook_path.""" - utils = importlib.import_module("think.utils") + muse = importlib.import_module("think.muse") md_file = tmp_path / "test_generator.md" md_file.write_text( @@ -161,7 +161,7 @@ def test_prompt_metadata_no_hook_path(tmp_path): hook_file = tmp_path / "test_generator.py" hook_file.write_text("def post_process(r, c): return r") - meta = utils._load_prompt_metadata(md_file) + meta = muse._load_prompt_metadata(md_file) # hook_path should no longer be set (hooks are loaded via load_post_hook) assert "hook_path" not in meta diff --git a/tests/test_output_path.py b/tests/test_output_path.py index 04228a776..d68eb6701 100644 --- a/tests/test_output_path.py +++ b/tests/test_output_path.py @@ -5,7 +5,7 @@ from pathlib import Path -from think.utils import get_output_path, get_output_topic +from think.muse import get_output_path, get_output_topic class TestGetOutputTopic: diff --git a/tests/test_template_substitution.py b/tests/test_template_substitution.py index ff1cd5389..97038a02c 100644 --- a/tests/test_template_substitution.py +++ b/tests/test_template_substitution.py @@ -8,7 +8,7 @@ import os import pytest -from think.utils import _flatten_identity_to_template_vars, load_prompt +from think.muse import _flatten_identity_to_template_vars, load_prompt @pytest.fixture diff --git a/tests/test_think_utils.py b/tests/test_think_utils.py index 3f0b137a4..804e966c3 100644 --- a/tests/test_think_utils.py +++ b/tests/test_think_utils.py @@ -13,12 +13,7 @@ from pathlib import Path import pytest from think.entities import load_entity_names -from think.utils import ( - _merge_instructions_config, - compose_instructions, - segment_key, - setup_cli, -) +from think.utils import segment_key, setup_cli def setup_entities_new_structure( @@ -658,233 +653,6 @@ class TestSetupCliConfigEnv: assert os.environ.get("BOOL_VAR") == "True" -class TestMergeInstructionsConfig: - """Tests for _merge_instructions_config helper.""" - - def test_returns_defaults_when_no_overrides(self): - """Test that defaults are returned when overrides is None.""" - defaults = {"system": "journal", "facets": True} - result = _merge_instructions_config(defaults, None) - assert result == defaults - # Should be a copy, not the same object - assert result is not defaults - - def test_returns_defaults_when_empty_overrides(self): - """Test that defaults are returned when overrides is empty dict.""" - defaults = {"system": "journal", "facets": True} - result = _merge_instructions_config(defaults, {}) - assert result == defaults - - def test_overrides_system_key(self): - """Test that system key can be overridden.""" - defaults = {"system": "journal", "facets": True} - overrides = {"system": "custom_prompt"} - result = _merge_instructions_config(defaults, overrides) - assert result["system"] == "custom_prompt" - assert result["facets"] is True - - def test_overrides_facets_key(self): - """Test that facets key can be overridden.""" - defaults = {"system": "journal", "facets": True} - overrides = {"facets": "full"} - result = _merge_instructions_config(defaults, overrides) - assert result["system"] == "journal" - assert result["facets"] == "full" - - def test_merges_sources_dict(self): - """Test that sources dict is merged, not replaced.""" - defaults = { - "system": "journal", - "sources": {"audio": True, "screen": True, "agents": False}, - } - overrides = {"sources": {"screen": False}} - result = _merge_instructions_config(defaults, overrides) - assert result["sources"]["audio"] is True # Preserved from defaults - assert result["sources"]["screen"] is False # Overridden - assert result["sources"]["agents"] is False # Preserved from defaults - - def test_ignores_unknown_keys(self): - """Test that unknown keys in overrides are ignored.""" - defaults = {"system": "journal", "facets": True} - overrides = {"unknown_key": "value", "another": 123} - result = _merge_instructions_config(defaults, overrides) - assert "unknown_key" not in result - assert "another" not in result - - -class TestComposeInstructions: - """Tests for compose_instructions function.""" - - def test_default_system_instruction_is_none(self, monkeypatch, tmp_path): - """Test that default system instruction is empty (agents must opt-in).""" - think_dir = tmp_path / "think" - think_dir.mkdir() - - import think.utils - - original_file = think.utils.__file__ - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions() - - # Restore - monkeypatch.setattr(think.utils, "__file__", original_file) - - assert "system_instruction" in result - assert result["system_instruction"] == "" - assert result["system_prompt_name"] == "" - - def test_custom_system_instruction(self, monkeypatch, tmp_path): - """Test that custom system prompt can be loaded.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - custom_txt = think_dir / "custom.md" - custom_txt.write_text("Custom system instruction") - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions( - config_overrides={"system": "custom"}, - ) - - assert result["system_prompt_name"] == "custom" - assert "Custom system instruction" in result["system_instruction"] - - def test_user_instruction_loaded_when_provided(self, monkeypatch, tmp_path): - """Test that user instruction is loaded when user_prompt is provided.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - journal_txt = think_dir / "journal.md" - journal_txt.write_text("System instruction") - user_txt = think_dir / "default.md" - user_txt.write_text("User instruction content") - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions(user_prompt="default") - - assert result["user_instruction"] == "User instruction content" - - def test_user_instruction_none_when_not_provided(self, monkeypatch, tmp_path): - """Test that user instruction is None when user_prompt is not provided.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - journal_txt = think_dir / "journal.md" - journal_txt.write_text("System instruction") - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions() - - assert result["user_instruction"] is None - - def test_facets_none_excludes_facets_from_context(self, monkeypatch, tmp_path): - """Test that facets='none' excludes facet info from extra_context.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - journal_txt = think_dir / "journal.md" - journal_txt.write_text("System instruction") - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions( - include_datetime=False, - config_overrides={"facets": False}, - ) - - # With no datetime and no facets, extra_context should be empty/None - assert result["extra_context"] is None or result["extra_context"] == "" - - def test_include_datetime_false_excludes_time(self, monkeypatch, tmp_path): - """Test that include_datetime=False excludes time from context.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - journal_txt = think_dir / "journal.md" - journal_txt.write_text("System instruction") - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions( - include_datetime=False, - config_overrides={"facets": False}, - ) - - extra = result.get("extra_context") or "" - assert "Current Date and Time" not in extra - - def test_include_datetime_true_includes_time(self, monkeypatch, tmp_path): - """Test that include_datetime=True includes time in context.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - journal_txt = think_dir / "journal.md" - journal_txt.write_text("System instruction") - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions( - include_datetime=True, - config_overrides={"facets": False}, - ) - - assert "Current Date and Time" in result["extra_context"] - - def test_sources_returned_from_defaults(self, monkeypatch, tmp_path): - """Test that sources config is returned with defaults (all false).""" - think_dir = tmp_path / "think" - think_dir.mkdir() - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions() - - assert "sources" in result - assert result["sources"]["audio"] is False - assert result["sources"]["screen"] is False - assert result["sources"]["agents"] is False - - def test_sources_can_be_overridden(self, monkeypatch, tmp_path): - """Test that sources config can be overridden.""" - think_dir = tmp_path / "think" - think_dir.mkdir() - - import think.utils - - monkeypatch.setattr(think.utils, "__file__", str(think_dir / "utils.py")) - monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) - - result = compose_instructions( - config_overrides={ - "sources": {"audio": True, "agents": True}, - }, - ) - - assert result["sources"]["audio"] is True # Overridden - assert result["sources"]["screen"] is False # Default preserved - assert result["sources"]["agents"] is True # Overridden - - class TestPortDiscovery: """Tests for service port discovery utilities.""" @@ -963,100 +731,3 @@ class TestPortDiscovery: # Now it should exist assert health_dir.exists() assert (health_dir / "new_service.port").read_text() == "9999" - - -# ============================================================================= -# source_is_enabled / source_is_required / get_agent_filter tests -# ============================================================================= - - -def test_source_is_enabled_bool(): - """Test source_is_enabled with bool values.""" - from think.utils import source_is_enabled - - assert source_is_enabled(True) is True - assert source_is_enabled(False) is False - - -def test_source_is_enabled_required_string(): - """Test source_is_enabled with 'required' string.""" - from think.utils import source_is_enabled - - assert source_is_enabled("required") is True - - -def test_source_is_enabled_dict(): - """Test source_is_enabled with dict values for agents source.""" - from think.utils import source_is_enabled - - # Dict with at least one True value -> enabled - assert source_is_enabled({"entities": True, "meetings": False}) is True - - # Dict with at least one "required" value -> enabled - assert source_is_enabled({"entities": "required", "meetings": False}) is True - - # Dict with all False values -> disabled - assert source_is_enabled({"entities": False, "meetings": False}) is False - - # Empty dict -> disabled - assert source_is_enabled({}) is False - - -def test_source_is_required_bool(): - """Test source_is_required with bool values.""" - from think.utils import source_is_required - - assert source_is_required(True) is False - assert source_is_required(False) is False - - -def test_source_is_required_string(): - """Test source_is_required with 'required' string.""" - from think.utils import source_is_required - - assert source_is_required("required") is True - - -def test_source_is_required_dict(): - """Test source_is_required with dict values.""" - from think.utils import source_is_required - - # Dict with at least one "required" value -> required - assert source_is_required({"entities": "required", "meetings": False}) is True - - # Dict with no "required" values -> not required - assert source_is_required({"entities": True, "meetings": False}) is False - - # Empty dict -> not required - assert source_is_required({}) is False - - -def test_get_agent_filter_bool(): - """Test get_agent_filter with bool values.""" - from think.utils import get_agent_filter - - # True -> None (all agents) - assert get_agent_filter(True) is None - - # False -> empty dict (no agents) - assert get_agent_filter(False) == {} - - -def test_get_agent_filter_required_string(): - """Test get_agent_filter with 'required' string.""" - from think.utils import get_agent_filter - - # "required" -> None (all agents, required) - assert get_agent_filter("required") is None - - -def test_get_agent_filter_dict(): - """Test get_agent_filter with dict values.""" - from think.utils import get_agent_filter - - # Dict -> returned as-is for filtering - filter_dict = {"entities": True, "meetings": "required", "flow": False} - assert get_agent_filter(filter_dict) == filter_dict - - # Empty dict -> empty dict (no agents) - assert get_agent_filter({}) == {} diff --git a/think/agents.py b/think/agents.py index eb4604876..fb45ab471 100644 --- a/think/agents.py +++ b/think/agents.py @@ -29,21 +29,25 @@ from google.genai import types from think.cluster import cluster, cluster_period, cluster_span from think.providers.shared import Event, GenerateResult -from think.utils import ( +from think.muse import ( compose_instructions, - day_log, - day_path, - format_day, - format_segment_times, get_agent_filter, get_muse_configs, get_output_path, + load_post_hook, + load_pre_hook, load_prompt, + source_is_enabled, + source_is_required, +) +from think.utils import ( + day_log, + day_path, + format_day, + format_segment_times, now_ms, segment_parse, setup_cli, - source_is_enabled, - source_is_required, ) LOG = logging.getLogger("think.agents") @@ -248,92 +252,6 @@ class PreHookContext(TypedDict, total=False): meta: dict # Full frontmatter/config -# MUSE_DIR for hook resolution -_MUSE_DIR = Path(__file__).parent.parent / "muse" - - -def _resolve_hook_path(hook_name: str) -> Path: - """Resolve hook name to file path. - - Resolution: - - Named: "name" -> muse/{name}.py - - App-qualified: "app:name" -> apps/{app}/muse/{name}.py - - Explicit path: "path/to/hook.py" -> direct path - """ - if "/" in hook_name or hook_name.endswith(".py"): - return Path(hook_name) - elif ":" in hook_name: - app, name = hook_name.split(":", 1) - return Path(__file__).parent.parent / "apps" / app / "muse" / f"{name}.py" - else: - return _MUSE_DIR / f"{hook_name}.py" - - -def _load_hook_function(config: dict, key: str, func_name: str) -> Callable | None: - """Load a hook function from config. - - Args: - config: Agent/generator config dict - key: Hook key in config ("pre" or "post") - func_name: Function name to load ("pre_process" or "post_process") - - Returns: - The hook function, or None if no hook configured. - - Raises: - ValueError: If hook file doesn't define the required function. - ImportError: If hook file cannot be loaded. - """ - import importlib.util - - hook_config = config.get("hook") - if not hook_config or not isinstance(hook_config, dict): - return None - - hook_name = hook_config.get(key) - if not hook_name: - return None - - hook_path = _resolve_hook_path(hook_name) - - if not hook_path.exists(): - raise ImportError(f"Hook file not found: {hook_path}") - - spec = importlib.util.spec_from_file_location( - f"{key}_hook_{hook_path.stem}", hook_path - ) - if spec is None or spec.loader is None: - raise ImportError(f"Cannot load hook from {hook_path}") - - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - - if not hasattr(module, func_name): - raise ValueError(f"Hook {hook_path} must define a '{func_name}' function") - - process_func = getattr(module, func_name) - if not callable(process_func): - raise ValueError(f"Hook {hook_path} '{func_name}' must be callable") - - return process_func - - -def load_post_hook(config: dict) -> Callable[[str, HookContext], str | None] | None: - """Load post-processing hook from config if defined. - - Hook config format: {"hook": {"post": "name"}} - """ - return _load_hook_function(config, "post", "post_process") - - -def load_pre_hook(config: dict) -> Callable[[PreHookContext], dict | None] | None: - """Load pre-processing hook from config if defined. - - Hook config format: {"hook": {"pre": "name"}} - """ - return _load_hook_function(config, "pre", "pre_process") - - def _build_base_context(config: dict) -> dict: """Build common context fields shared by pre and post hooks.""" context = { @@ -474,7 +392,7 @@ def hydrate_config(request: dict) -> dict: Fully hydrated config dict ready for routing """ from think.models import resolve_model_for_provider, resolve_provider - from think.utils import get_agent, key_to_context + from think.muse import get_agent, key_to_context name = request.get("name", "default") facet = request.get("facet") @@ -1235,7 +1153,7 @@ def generate_agent_output( max_output_tokens = 8192 * 6 # Build context for provider routing and token logging - from think.utils import key_to_context + from think.muse import key_to_context context = key_to_context(name) if name else "muse.system.unknown" diff --git a/think/cortex.py b/think/cortex.py index a58473eac..a1e87f27b 100644 --- a/think/cortex.py +++ b/think/cortex.py @@ -483,7 +483,7 @@ class CortexService: if usage_data and original_request: try: from think.models import log_token_usage - from think.utils import key_to_context + from think.muse import key_to_context model = original_request.get("model", "unknown") name = original_request.get("name", "unknown") @@ -659,7 +659,8 @@ class CortexService: - Multi-facet: {name}_{facet}.{ext} instead of {name}.{ext} """ try: - from think.utils import day_path, get_output_path + from think.muse import get_output_path + from think.utils import day_path # Check for explicit output_path override first if config.get("output_path"): diff --git a/think/detect_created.py b/think/detect_created.py index ebac91379..a1f271527 100644 --- a/think/detect_created.py +++ b/think/detect_created.py @@ -13,7 +13,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Optional -from .utils import load_prompt +from .muse import load_prompt def _load_system_prompt() -> str: diff --git a/think/detect_transcript.py b/think/detect_transcript.py index 9ea18f75e..c4d74d70e 100644 --- a/think/detect_transcript.py +++ b/think/detect_transcript.py @@ -10,7 +10,7 @@ import logging from pathlib import Path from typing import List, Optional -from .utils import load_prompt +from .muse import load_prompt def _load_json_prompt() -> str: diff --git a/think/dream.py b/think/dream.py index dbbde15d8..b688e9730 100644 --- a/think/dream.py +++ b/think/dream.py @@ -20,13 +20,12 @@ from think.callosum import CallosumConnection from think.cortex_client import cortex_request, get_agent_end_state, wait_for_agents from think.facets import get_active_facets, get_enabled_facets, get_facets from think.runner import run_task +from think.muse import get_muse_configs, get_output_path from think.utils import ( day_input_summary, day_log, day_path, get_journal, - get_muse_configs, - get_output_path, iso_date, setup_cli, ) @@ -319,7 +318,9 @@ def run_prompts_by_priority( if spawned: agent_ids = [agent_id for agent_id, _, _ in spawned] - logging.info(f"Waiting for {len(agent_ids)} prompts in priority {priority}...") + logging.info( + f"Waiting for {len(agent_ids)} prompts in priority {priority}..." + ) completed, timed_out = wait_for_agents(agent_ids, timeout=600) @@ -675,7 +676,9 @@ def main() -> None: day_log(day, msg) duration_ms = int((time.time() - start_time) * 1000) - logging.info(f"Dream completed in {duration_ms}ms: {success_count} succeeded, {fail_count} failed") + logging.info( + f"Dream completed in {duration_ms}ms: {success_count} succeeded, {fail_count} failed" + ) if fail_count > 0: logging.error(f"{fail_count} prompt(s) failed, exiting with error") diff --git a/think/hooks.py b/think/hooks.py index 4fffb95e1..121185587 100644 --- a/think/hooks.py +++ b/think/hooks.py @@ -131,7 +131,8 @@ def compute_output_source(context: dict) -> str: Returns: Relative path like "20240101/agents/meetings.md". """ - from think.utils import get_journal, get_output_topic + from think.muse import get_output_topic + from think.utils import get_journal day = context.get("day", "") output_path = context.get("output_path", "") diff --git a/think/models.py b/think/models.py index a8a84c22b..bfb31cd33 100644 --- a/think/models.py +++ b/think/models.py @@ -188,8 +188,8 @@ def _discover_prompt_contexts() -> Dict[str, Dict[str, Any]]: def _discover_muse_contexts() -> Dict[str, Dict[str, Any]]: """Discover muse context defaults from muse/*.md config files. - Scans system muse configs (muse/*.md) and app muse configs (apps/*/muse/*.md) - for tier/label/group metadata. Includes both tool-using agents and generators. + Uses get_muse_configs() from think.muse to load all muse configurations + and converts them to context patterns with tier/label/group metadata. Returns ------- @@ -197,56 +197,21 @@ def _discover_muse_contexts() -> Dict[str, Dict[str, Any]]: Mapping of context patterns to {tier, label, group, has_tools} dicts. Context patterns are: muse.system.{name} or muse.{app}.{name} """ + from think.muse import get_muse_configs, key_to_context + contexts = {} - # System muse configs from muse/ - muse_dir = Path(__file__).parent.parent / "muse" - if muse_dir.exists(): - for md_path in muse_dir.glob("*.md"): - config_name = md_path.stem - try: - post = frontmatter.load( - md_path, - ) - config = post.metadata if post.metadata else {} - - context = f"muse.system.{config_name}" - contexts[context] = { - "tier": config.get("tier", TIER_FLASH), - "label": config.get("label", config.get("title", config_name)), - "group": config.get("group", "Think"), - "has_tools": "tools" in config, - } - except Exception: - pass # Skip configs that can't be loaded + # Load all muse configs (including disabled for completeness) + all_configs = get_muse_configs(include_disabled=True) - # App muse configs from apps/*/muse/ - apps_dir = Path(__file__).parent.parent / "apps" - if apps_dir.is_dir(): - for app_path in apps_dir.iterdir(): - if not app_path.is_dir() or app_path.name.startswith("_"): - continue - muse_subdir = app_path / "muse" - if not muse_subdir.is_dir(): - continue - app_name = app_path.name - for md_path in muse_subdir.glob("*.md"): - config_name = md_path.stem - try: - post = frontmatter.load( - md_path, - ) - config = post.metadata if post.metadata else {} - - context = f"muse.{app_name}.{config_name}" - contexts[context] = { - "tier": config.get("tier", TIER_FLASH), - "label": config.get("label", config.get("title", config_name)), - "group": config.get("group", "Think"), - "has_tools": "tools" in config, - } - except Exception: - pass # Skip configs that can't be loaded + for key, config in all_configs.items(): + context = key_to_context(key) + contexts[context] = { + "tier": config.get("tier", TIER_FLASH), + "label": config.get("label", config.get("title", key)), + "group": config.get("group", "Think"), + "has_tools": "tools" in config, + } return contexts diff --git a/think/muse.py b/think/muse.py new file mode 100644 index 000000000..d1fdfddbc --- /dev/null +++ b/think/muse.py @@ -0,0 +1,984 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Muse prompt loading and configuration utilities. + +This module provides all functionality for loading, parsing, and configuring +muse prompts (agents and generators) from muse/*.md and apps/*/muse/*.md. + +Key functions: +- load_prompt(): Load and parse .md prompt files with template substitution +- get_muse_configs(): Discover all muse configs with filtering +- get_agent(): Load complete agent configuration by name +- compose_instructions(): Build system/user prompts from instruction config +- Hook loading: load_pre_hook(), load_post_hook() +""" + +from __future__ import annotations + +import importlib.util +import logging +import os +from pathlib import Path +from string import Template +from typing import Any, Callable, NamedTuple + +import frontmatter + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +MUSE_DIR = Path(__file__).parent.parent / "muse" +TEMPLATES_DIR = Path(__file__).parent / "templates" + +# Cached raw template content loaded from think/templates/*.md +_templates_cache: dict[str, str] | None = None + + +# --------------------------------------------------------------------------- +# Template Loading +# --------------------------------------------------------------------------- + + +def _load_raw_templates() -> dict[str, str]: + """Load raw template files from think/templates/ directory. + + Templates are cached on first load. Each .md file becomes a template + variable named after its stem (e.g., daily_preamble.md -> $daily_preamble). + + Returns + ------- + dict[str, str] + Mapping of template variable names to their raw content (no substitution). + """ + global _templates_cache + if _templates_cache is not None: + return _templates_cache + + _templates_cache = {} + if TEMPLATES_DIR.is_dir(): + for md_path in TEMPLATES_DIR.glob("*.md"): + var_name = md_path.stem + try: + post = frontmatter.load( + md_path, + ) + _templates_cache[var_name] = post.content.strip() + except Exception as exc: + logging.debug("Failed to load template %s: %s", md_path, exc) + + return _templates_cache + + +def _load_templates(template_vars: dict[str, str] | None = None) -> dict[str, str]: + """Load and substitute template files from think/templates/ directory. + + Raw templates are cached, but substitution is performed on each call + to support context-dependent variables like $date and $segment_start. + + Parameters + ---------- + template_vars: + Optional variables to substitute into templates. Templates can use + identity vars ($name, $preferred), context vars ($day, $date, + $segment_start, $segment_end), and other template vars. + + Returns + ------- + dict[str, str] + Mapping of template variable names to their substituted content. + """ + raw_templates = _load_raw_templates() + + if not template_vars: + return dict(raw_templates) + + # Substitute variables into each template + substituted = {} + for var_name, content in raw_templates.items(): + try: + template = Template(content) + substituted[var_name] = template.safe_substitute(template_vars) + except Exception as exc: + logging.debug("Template substitution failed for %s: %s", var_name, exc) + substituted[var_name] = content + + return substituted + + +# --------------------------------------------------------------------------- +# Prompt Loading +# --------------------------------------------------------------------------- + + +class PromptContent(NamedTuple): + """Container for prompt text, metadata, and its resolved path.""" + + text: str + path: Path + metadata: dict[str, Any] = {} + + +class PromptNotFoundError(FileNotFoundError): + """Raised when a prompt file cannot be located.""" + + def __init__(self, path: Path) -> None: + self.path = path + super().__init__(f"Prompt file not found: {path}") + + +def _flatten_identity_to_template_vars(identity: dict[str, Any]) -> dict[str, str]: + """Flatten identity config into template variables with uppercase-first versions. + + Parameters + ---------- + identity: + Identity configuration dictionary from get_config()['identity']. + + Returns + ------- + dict[str, str] + Template variables including flattened nested objects and uppercase-first versions. + For example: + - 'name' → identity['name'] + - 'pronouns_possessive' → identity['pronouns']['possessive'] + - 'Pronouns_possessive' → identity['pronouns']['possessive'].capitalize() + - 'bio' → identity['bio'] + """ + template_vars: dict[str, str] = {} + + # Flatten top-level and nested values + for key, value in identity.items(): + if isinstance(value, dict): + # Flatten nested dictionaries with underscore separator + for subkey, subvalue in value.items(): + var_name = f"{key}_{subkey}" + template_vars[var_name] = str(subvalue) + # Create uppercase-first version + template_vars[var_name.capitalize()] = str(subvalue).capitalize() + elif isinstance(value, (str, int, float, bool)): + # Top-level scalar values + template_vars[key] = str(value) + # Create uppercase-first version + template_vars[key.capitalize()] = str(value).capitalize() + + return template_vars + + +def load_prompt( + name: str, + base_dir: str | Path | None = None, + *, + include_journal: bool = False, + context: dict[str, Any] | None = None, +) -> PromptContent: + """Return the text contents, metadata, and path for a ``.md`` prompt file. + + Prompt files use JSON frontmatter for metadata. Supports Python + string.Template variable substitution using: + - Identity config from get_config()['identity']: + - Top-level fields: $name, $preferred, $bio, $timezone + - Nested fields with underscores: $pronouns_possessive, $pronouns_subject + - Uppercase-first versions: $Pronouns_possessive, $Name, $Bio + - Templates from think/templates/*.md: + - Each file becomes a variable named after its stem + - Example: daily_preamble.md -> $daily_preamble + - Templates are pre-processed with identity and context vars, so templates + can use $date, $preferred, etc. before being substituted into prompts + + Callers can provide additional context variables via the ``context`` parameter. + Context variables override identity and template variables if there's a collision. + Uppercase-first versions are automatically created for context variables. + + Parameters + ---------- + name: + Base filename of the prompt without the ``.md`` suffix. If the suffix is + included, it will not be duplicated. + base_dir: + Optional directory containing the prompt file. Defaults to the directory + of this module when not provided. + include_journal: + If True, prepends the content of ``think/journal.md`` to the requested + prompt. Defaults to False. Context variables are passed through to the + journal template as well. + context: + Optional dictionary of additional template variables. Values are converted + to strings. For each key, an uppercase-first version is also created + (e.g., ``{"day": "20250110"}`` adds both ``$day`` and ``$Day``). + + Returns + ------- + PromptContent + The prompt text (with surrounding whitespace removed and template variables + substituted), the resolved path to the ``.md`` file, and metadata from + the JSON frontmatter. + """ + from think.utils import get_config + + if not name: + raise ValueError("Prompt name must be provided") + + if name.endswith(".md"): + filename = name + else: + filename = f"{name}.md" + + prompt_dir = Path(base_dir) if base_dir is not None else Path(__file__).parent + prompt_path = prompt_dir / filename + try: + post = frontmatter.load( + prompt_path, + ) + text = post.content.strip() + metadata = dict(post.metadata) + except FileNotFoundError as exc: # pragma: no cover - caller handles missing prompt + raise PromptNotFoundError(prompt_path) from exc + + # Perform template substitution + try: + config = get_config() + identity = config.get("identity", {}) + template_vars = _flatten_identity_to_template_vars(identity) + + # Merge caller-provided context (overrides identity vars if collision) + if context: + for key, value in context.items(): + str_value = str(value) + template_vars[key] = str_value + # Add uppercase-first version + template_vars[key.capitalize()] = str_value.capitalize() + + # Load templates with identity and context vars so templates can use them + templates = _load_templates(template_vars) + template_vars.update(templates) + + # Use safe_substitute to avoid errors for undefined variables + template = Template(text) + text = template.safe_substitute(template_vars) + except Exception as exc: + # Log but don't fail - return original text if substitution fails + logging.debug("Template substitution failed for %s: %s", prompt_path, exc) + + # Prepend journal content if requested + if include_journal and name != "journal": + journal_content = load_prompt("journal", context=context) + text = f"{journal_content.text}\n\n{text}" + + return PromptContent(text=text, path=prompt_path, metadata=metadata) + + +# --------------------------------------------------------------------------- +# Prompt Metadata Loading +# --------------------------------------------------------------------------- + + +def _load_prompt_metadata(md_path: Path) -> dict[str, object]: + """Load prompt metadata from .md file with JSON frontmatter. + + Parameters + ---------- + md_path: + Path to the .md prompt file with JSON frontmatter. + + Returns + ------- + dict + Metadata dict with path, mtime, color, and frontmatter fields. + """ + mtime = int(md_path.stat().st_mtime) + info: dict[str, object] = { + "path": str(md_path), + "mtime": mtime, + } + + try: + post = frontmatter.load( + md_path, + ) + if post.metadata: + info.update(post.metadata) + except Exception as exc: # pragma: no cover - metadata optional + logging.debug("Error reading frontmatter from %s: %s", md_path, exc) + + # Apply default color if not specified + if "color" not in info: + info["color"] = "#6c757d" + + return info + + +# --------------------------------------------------------------------------- +# Muse Config Discovery +# --------------------------------------------------------------------------- + + +def key_to_context(key: str) -> str: + """Convert muse config key to context pattern. + + Parameters + ---------- + key: + Muse config key in format "name" (system) or "app:name" (app). + + Returns + ------- + str + Context pattern: "muse.system.{name}" or "muse.{app}.{name}". + + Examples + -------- + >>> key_to_context("meetings") + 'muse.system.meetings' + >>> key_to_context("entities:observer") + 'muse.entities.observer' + """ + if ":" in key: + app, name = key.split(":", 1) + return f"muse.{app}.{name}" + return f"muse.system.{key}" + + +def get_output_topic(key: str) -> str: + """Convert agent/generator key to filesystem-safe basename (no extension). + + Parameters + ---------- + key: + Generator key in format "topic" (system) or "app:topic" (app). + + Returns + ------- + str + Filesystem-safe name: "topic" or "_app_topic". + + Examples + -------- + >>> get_output_topic("activity") + 'activity' + >>> get_output_topic("chat:sentiment") + '_chat_sentiment' + """ + if ":" in key: + app, topic = key.split(":", 1) + return f"_{app}_{topic}" + return key + + +def get_output_path( + day_dir: "os.PathLike[str]", + key: str, + segment: str | None = None, + output_format: str | None = None, + facet: str | None = None, +) -> Path: + """Return output path for generator agent output. + + Shared utility for determining where to write generator results. + Used by think/agents.py and think/cortex.py. + + Parameters + ---------- + day_dir: + Day directory path (YYYYMMDD). + key: + Generator key or agent name (e.g., "activity", "chat:sentiment", + "decisionalizer", "entities:observer"). + segment: + Optional segment key (HHMMSS_LEN) for segment-level output. + output_format: + Output format - "json" for JSON, anything else for markdown. + facet: + Optional facet name for multi-facet agents. When provided, the facet + is appended to the filename (e.g., "newsletter_work.md"). + + Returns + ------- + Path + Output file path: + - With segment: YYYYMMDD/{segment}/{topic}.{ext} + - Without segment: YYYYMMDD/agents/{topic}.{ext} + - With facet: {topic}_{facet}.{ext} instead of {topic}.{ext} + Where topic is derived from key and ext is "json" or "md". + """ + day = Path(day_dir) + topic = get_output_topic(key) + ext = "json" if output_format == "json" else "md" + + # Append facet suffix for multi-facet agent outputs + if facet: + filename = f"{topic}_{facet}.{ext}" + else: + filename = f"{topic}.{ext}" + + if segment: + # Segment output goes directly in segment directory + return day / segment / filename + else: + # Daily output goes in agents/ subdirectory + return day / "agents" / filename + + +def get_muse_configs( + *, + has_tools: bool | None = None, + has_output: bool | None = None, + schedule: str | None = None, + include_disabled: bool = False, +) -> dict[str, dict[str, Any]]: + """Load muse configs from system and app directories. + + Unified function for loading both tool-using agents and generators from + muse/*.md files. Filters based on presence of tools/output fields. + + Args: + has_tools: If True, only configs with "tools" field (agents). + If False, only configs without "tools" field. + If None, no filtering on tools presence. + has_output: If True, only configs with "output" field (generators). + If False, only configs without "output" field. + If None, no filtering on output presence. + schedule: If provided, only configs where schedule matches this value + (e.g., "segment", "daily"). + include_disabled: If True, include configs with disabled=True. + Default False (for processing pipelines). + + Returns: + Dictionary mapping config keys to their metadata including: + - path: Path to the .md file + - source: "system" or "app" + - app: App name (only for app configs) + - All fields from frontmatter + """ + from think.utils import get_config + + configs: dict[str, dict[str, Any]] = {} + + def matches_filter(info: dict) -> bool: + """Check if config matches the filter criteria.""" + # Check has_tools filter + if has_tools is True and "tools" not in info: + return False + if has_tools is False and "tools" in info: + return False + + # Check has_output filter + if has_output is True and "output" not in info: + return False + if has_output is False and "output" in info: + return False + + # Check specific schedule value + if schedule is not None and info.get("schedule") != schedule: + return False + + # Check disabled status + if not include_disabled and info.get("disabled", False): + return False + + return True + + # System configs from muse/ + if MUSE_DIR.is_dir(): + for md_path in sorted(MUSE_DIR.glob("*.md")): + name = md_path.stem + info = _load_prompt_metadata(md_path) + + if not matches_filter(info): + continue + + info["source"] = "system" + configs[name] = info + + # App configs from apps/*/muse/ + apps_dir = Path(__file__).parent.parent / "apps" + if apps_dir.is_dir(): + for app_path in sorted(apps_dir.iterdir()): + if not app_path.is_dir() or app_path.name.startswith("_"): + continue + app_muse_dir = app_path / "muse" + if not app_muse_dir.is_dir(): + continue + app_name = app_path.name + for md_path in sorted(app_muse_dir.glob("*.md")): + item_name = md_path.stem + info = _load_prompt_metadata(md_path) + + if not matches_filter(info): + continue + + key = f"{app_name}:{item_name}" + info["source"] = "app" + info["app"] = app_name + configs[key] = info + + # Merge journal config overrides from providers.contexts + providers_config = get_config().get("providers", {}) + contexts = providers_config.get("contexts", {}) + + for key, info in configs.items(): + context_key = key_to_context(key) + + # Check for exact match in contexts + override = contexts.get(context_key) + if override and isinstance(override, dict): + # Merge supported override fields + if "disabled" in override: + info["disabled"] = override["disabled"] + if "extract" in override: + info["extract"] = override["extract"] + if "tier" in override: + info["tier"] = override["tier"] + if "provider" in override: + info["provider"] = override["provider"] + + # Validate: scheduled prompts must have explicit priority + for key, info in configs.items(): + if info.get("schedule") and "priority" not in info: + raise ValueError( + f"Scheduled prompt '{key}' is missing required 'priority' field. " + f"All prompts with 'schedule' must declare an explicit priority." + ) + + return configs + + +# --------------------------------------------------------------------------- +# Agent Resolution +# --------------------------------------------------------------------------- + + +def _resolve_agent_path(name: str) -> tuple[Path, str]: + """Resolve agent name to directory path and agent filename. + + Parameters + ---------- + name: + Agent name - either system agent (e.g., "default") or + app-namespaced agent (e.g., "chat:helper"). + + Returns + ------- + tuple[Path, str] + (agent_directory, agent_name) tuple. + """ + if ":" in name: + # App agent: "chat:helper" -> apps/chat/muse/helper + app, agent_name = name.split(":", 1) + agent_dir = Path(__file__).parent.parent / "apps" / app / "muse" + else: + # System agent: "default" -> muse/default + agent_dir = MUSE_DIR + agent_name = name + return agent_dir, agent_name + + +# --------------------------------------------------------------------------- +# Instructions Composition +# --------------------------------------------------------------------------- + +# Default instruction configuration - all false, agents must explicitly opt-in +_DEFAULT_INSTRUCTIONS = { + "system": None, + "facets": False, + "sources": { + "audio": False, + "screen": False, + "agents": False, + }, +} + + +def _merge_instructions_config(defaults: dict, overrides: dict | None) -> dict: + """Merge instruction config overrides into defaults. + + Handles nested "sources" dict specially. + + Parameters + ---------- + defaults: + Default instruction configuration. + overrides: + Optional overrides from .json "instructions" key. + + Returns + ------- + dict + Merged configuration. + """ + if not overrides: + return defaults.copy() + + result = defaults.copy() + + # Merge top-level keys + for key in ("system", "facets"): + if key in overrides: + result[key] = overrides[key] + + # Merge sources dict if present + if "sources" in overrides and isinstance(overrides["sources"], dict): + result["sources"] = {**defaults.get("sources", {}), **overrides["sources"]} + + return result + + +def compose_instructions( + *, + user_prompt: str | None = None, + user_prompt_dir: Path | None = None, + facet: str | None = None, + include_datetime: bool = True, + config_overrides: dict | None = None, +) -> dict: + """Compose instruction components for agents or generators. + + This is the shared function for building system_instruction, user_instruction, + extra_context, and sources configuration. Both agents and generators use this + to ensure consistent prompt composition. + + Parameters + ---------- + user_prompt: + Name of the user instruction prompt to load (e.g., "default" for agents). + If None, no user_instruction is included (typical for generators). + user_prompt_dir: + Directory to load user_prompt from. If None, uses think/ directory. + facet: + Optional facet name to focus on. When provided, extra_context includes + only this facet's info (detail level controlled by "facets" setting). + include_datetime: + Whether to include current date/time in extra_context. Default True + for agents (real-time chat), typically False for generators (past analysis). + config_overrides: + Optional dict from .json "instructions" key. Supported keys: + - "system": prompt name for system instruction (default: "journal") + - "facets": false | true | "full" (default: true) + false = skip facet context + true = include facet context with names only + "full" = include facet context with full descriptions + For faceted generators, shows focused facet; for unfaceted, shows all facets. + - "sources": {"audio": bool, "screen": bool, "agents": bool|dict} + The "agents" source can be: + - bool: True (all agents), False (no agents) + - "required": all agents, fail if none found + - dict: selective filtering, e.g., {"entities": true, "meetings": "required"} + + Returns + ------- + dict + Composed instruction configuration: + - system_instruction: str - loaded from "system" prompt + - system_prompt_name: str - name of system prompt (for cache keys) + - user_instruction: str | None - loaded from user_prompt if provided + - extra_context: str | None - facets + datetime + - sources: dict - {"audio": bool, "screen": bool, "agents": bool|dict} + """ + from datetime import datetime + + # Merge defaults with overrides + cfg = _merge_instructions_config(_DEFAULT_INSTRUCTIONS, config_overrides) + + result: dict = {} + + # Load system instruction (None means no system prompt) + system_name = cfg.get("system") + if system_name: + system_prompt = load_prompt(system_name) + result["system_instruction"] = system_prompt.text + result["system_prompt_name"] = system_name + else: + result["system_instruction"] = "" + result["system_prompt_name"] = "" + + # Load user instruction if specified + if user_prompt: + base_dir = user_prompt_dir if user_prompt_dir else Path(__file__).parent + user_prompt_obj = load_prompt(user_prompt, base_dir=base_dir) + result["user_instruction"] = user_prompt_obj.text + else: + result["user_instruction"] = None + + # Build extra_context based on facets setting + # Values: false (skip), true (names only), "full" (with descriptions) + extra_parts = [] + facets_setting = cfg.get("facets", False) + facets_full = facets_setting == "full" + + if facets_setting: + if facet: + # Focused facet mode: include only this facet's context + try: + from think.facets import facet_summary + + summary = facet_summary(facet, detailed=facets_full) + extra_parts.append(f"## Facet Focus\n{summary}") + except Exception: + pass # Ignore if facet can't be loaded + else: + # General mode: all facets + try: + from think.facets import facet_summaries + + summary = facet_summaries(detailed=facets_full) + if summary and summary != "No facets found.": + extra_parts.append(summary) + except Exception: + pass # Ignore if facets can't be loaded + + # Add current date/time if requested + if include_datetime: + now = datetime.now() + try: + import tzlocal + + local_tz = tzlocal.get_localzone() + now_local = now.astimezone(local_tz) + time_str = now_local.strftime("%A, %B %d, %Y at %I:%M %p %Z") + except Exception: + time_str = now.strftime("%A, %B %d, %Y at %I:%M %p") + extra_parts.append(f"## Current Date and Time\nToday is {time_str}") + + result["extra_context"] = "\n\n".join(extra_parts).strip() if extra_parts else None + + # Include sources config + result["sources"] = cfg.get("sources", _DEFAULT_INSTRUCTIONS["sources"]) + + return result + + +# --------------------------------------------------------------------------- +# Source Configuration Helpers +# --------------------------------------------------------------------------- + + +def source_is_enabled(value: bool | str | dict) -> bool: + """Check if a source should be loaded based on its config value. + + Sources can be configured as: + - False: don't load + - True: load if available + - "required": load (and generation will fail if none found) + - dict: for agents source, selective loading (e.g., {"entities": true}) + + Both True and "required" mean the source should be loaded. + A non-empty dict means the source should be loaded (with filtering). + + Args: + value: The source config value (bool, "required" string, or dict for agents) + + Returns: + True if the source should be loaded, False otherwise. + """ + if isinstance(value, dict): + # Dict means selective loading - enabled if any agent is enabled + return any(v is True or v == "required" for v in value.values()) + return value is True or value == "required" + + +def source_is_required(value: bool | str | dict) -> bool: + """Check if a source must have content for generation to proceed. + + Args: + value: The source config value (bool, "required" string, or dict for agents) + + Returns: + True if the source is required (generation should skip if no content). + For dict values, returns True if any agent is marked "required". + """ + if isinstance(value, dict): + return any(v == "required" for v in value.values()) + return value == "required" + + +def get_agent_filter(value: bool | str | dict) -> dict[str, bool | str] | None: + """Extract agent filter from sources config. + + When agents source is a dict, returns it as filter mapping agent names + to their enabled/required status. When agents source is bool or "required", + returns None to indicate all agents should be loaded. + + Args: + value: The agents source config value + + Returns: + Dict mapping agent names to bool/"required", or None for all agents. + Returns empty dict if value is False (no agents). + + Examples: + >>> get_agent_filter(True) + None # All agents + >>> get_agent_filter(False) + {} # No agents + >>> get_agent_filter({"entities": True, "meetings": "required"}) + {"entities": True, "meetings": "required"} + """ + if isinstance(value, dict): + return value + if value is False: + return {} # No agents + return None # All agents (True or "required") + + +# --------------------------------------------------------------------------- +# Agent Loading +# --------------------------------------------------------------------------- + + +def get_agent(name: str = "default", facet: str | None = None) -> dict: + """Return complete agent configuration by name. + + Loads configuration from .md file with JSON frontmatter and instruction text, + merges with runtime context. + + Parameters + ---------- + name: + Agent name to load. Can be a system agent (e.g., "default") + or an app-namespaced agent (e.g., "chat:helper" for apps/chat/muse/helper). + facet: + Optional facet name to focus on. When provided, includes detailed + information for just this facet (with full entity details) instead + of summaries of all facets. + + Returns + ------- + dict + Complete agent configuration including system_instruction, user_instruction, + extra_context, model, backend, etc. + """ + # Resolve agent path based on namespace + agent_dir, agent_name = _resolve_agent_path(name) + + # Verify agent prompt file exists + md_path = agent_dir / f"{agent_name}.md" + if not md_path.exists(): + raise FileNotFoundError(f"Agent not found: {name}") + + # Load config from frontmatter + post = frontmatter.load( + md_path, + ) + config = dict(post.metadata) if post.metadata else {} + + # Extract instructions config if present + instructions_config = config.pop("instructions", None) + + # Use compose_instructions for consistent prompt composition + instructions = compose_instructions( + user_prompt=agent_name, + user_prompt_dir=agent_dir, + facet=facet, + include_datetime=True, + config_overrides=instructions_config, + ) + + # Merge instruction results into config + config["system_instruction"] = instructions["system_instruction"] + config["user_instruction"] = instructions["user_instruction"] + if instructions["extra_context"]: + config["extra_context"] = instructions["extra_context"] + + # Set agent name + config["name"] = name + + return config + + +# --------------------------------------------------------------------------- +# Hook Loading +# --------------------------------------------------------------------------- + + +def _resolve_hook_path(hook_name: str) -> Path: + """Resolve hook name to file path. + + Resolution: + - Named: "name" -> muse/{name}.py + - App-qualified: "app:name" -> apps/{app}/muse/{name}.py + - Explicit path: "path/to/hook.py" -> direct path + """ + if "/" in hook_name or hook_name.endswith(".py"): + return Path(hook_name) + elif ":" in hook_name: + app, name = hook_name.split(":", 1) + return Path(__file__).parent.parent / "apps" / app / "muse" / f"{name}.py" + else: + return MUSE_DIR / f"{hook_name}.py" + + +def _load_hook_function(config: dict, key: str, func_name: str) -> Callable | None: + """Load a hook function from config. + + Args: + config: Agent/generator config dict + key: Hook key in config ("pre" or "post") + func_name: Function name to load ("pre_process" or "post_process") + + Returns: + The hook function, or None if no hook configured. + + Raises: + ValueError: If hook file doesn't define the required function. + ImportError: If hook file cannot be loaded. + """ + hook_config = config.get("hook") + if not hook_config or not isinstance(hook_config, dict): + return None + + hook_name = hook_config.get(key) + if not hook_name: + return None + + hook_path = _resolve_hook_path(hook_name) + + if not hook_path.exists(): + raise ImportError(f"Hook file not found: {hook_path}") + + spec = importlib.util.spec_from_file_location( + f"{key}_hook_{hook_path.stem}", hook_path + ) + if spec is None or spec.loader is None: + raise ImportError(f"Cannot load hook from {hook_path}") + + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + if not hasattr(module, func_name): + raise ValueError(f"Hook {hook_path} must define a '{func_name}' function") + + process_func = getattr(module, func_name) + if not callable(process_func): + raise ValueError(f"Hook {hook_path} '{func_name}' must be callable") + + return process_func + + +def load_post_hook(config: dict) -> Callable[[str, "HookContext"], str | None] | None: + """Load post-processing hook from config if defined. + + Hook config format: {"hook": {"post": "name"}} + + Returns: + Post-processing function or None if no hook configured. + Function signature: (result: str, context: HookContext) -> str | None + """ + return _load_hook_function(config, "post", "post_process") + + +def load_pre_hook(config: dict) -> Callable[["PreHookContext"], dict | None] | None: + """Load pre-processing hook from config if defined. + + Hook config format: {"hook": {"pre": "name"}} + + Returns: + Pre-processing function or None if no hook configured. + Function signature: (context: PreHookContext) -> dict | None + """ + return _load_hook_function(config, "pre", "pre_process") + + +# Type aliases for hook context (actual TypedDicts defined in agents.py) +HookContext = dict +PreHookContext = dict diff --git a/think/muse_cli.py b/think/muse_cli.py index ad0fe40db..5f73b90c3 100644 --- a/think/muse_cli.py +++ b/think/muse_cli.py @@ -28,13 +28,13 @@ from typing import Any import frontmatter -from think.utils import ( +from think.muse import ( MUSE_DIR, _load_prompt_metadata, get_muse_configs, get_output_topic, - setup_cli, ) +from think.utils import setup_cli # Project root for computing relative paths _PROJECT_ROOT = Path(__file__).parent.parent @@ -460,7 +460,7 @@ def show_prompt_context( config["facet"] = facet else: # Tool agent - use get_agent() to build full config with instructions - from think.utils import get_agent + from think.muse import get_agent try: agent_config = get_agent(name, facet=facet) diff --git a/think/planner.py b/think/planner.py index 484744a80..523327b5e 100644 --- a/think/planner.py +++ b/think/planner.py @@ -8,7 +8,8 @@ import os import sys from pathlib import Path -from .utils import load_prompt, setup_cli +from .muse import load_prompt +from .utils import setup_cli async def _get_mcp_tools() -> str: diff --git a/think/utils.py b/think/utils.py index bdeaa045e..fb2121f96 100644 --- a/think/utils.py +++ b/think/utils.py @@ -1,6 +1,13 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc +"""General utilities for solstone. + +This module provides core utilities for journal access, date/segment handling, +configuration loading, and CLI setup. Muse-related utilities (prompt loading, +agent configs, etc.) have been moved to think/muse.py. +""" + from __future__ import annotations import argparse @@ -11,10 +18,8 @@ import re import time from datetime import datetime from pathlib import Path -from string import Template -from typing import Any, NamedTuple, Optional +from typing import Any, Optional -import frontmatter import platformdirs from dotenv import load_dotenv from timefhuman import timefhuman @@ -28,235 +33,6 @@ def now_ms() -> int: return int(time.time() * 1000) -MUSE_DIR = Path(__file__).parent.parent / "muse" -TEMPLATES_DIR = Path(__file__).parent / "templates" - -# Cached raw template content loaded from think/templates/*.md -_templates_cache: dict[str, str] | None = None - - -def _load_raw_templates() -> dict[str, str]: - """Load raw template files from think/templates/ directory. - - Templates are cached on first load. Each .md file becomes a template - variable named after its stem (e.g., daily_preamble.md -> $daily_preamble). - - Returns - ------- - dict[str, str] - Mapping of template variable names to their raw content (no substitution). - """ - global _templates_cache - if _templates_cache is not None: - return _templates_cache - - _templates_cache = {} - if TEMPLATES_DIR.is_dir(): - for md_path in TEMPLATES_DIR.glob("*.md"): - var_name = md_path.stem - try: - post = frontmatter.load( - md_path, - ) - _templates_cache[var_name] = post.content.strip() - except Exception as exc: - logging.debug("Failed to load template %s: %s", md_path, exc) - - return _templates_cache - - -def _load_templates(template_vars: dict[str, str] | None = None) -> dict[str, str]: - """Load and substitute template files from think/templates/ directory. - - Raw templates are cached, but substitution is performed on each call - to support context-dependent variables like $date and $segment_start. - - Parameters - ---------- - template_vars: - Optional variables to substitute into templates. Templates can use - identity vars ($name, $preferred), context vars ($day, $date, - $segment_start, $segment_end), and other template vars. - - Returns - ------- - dict[str, str] - Mapping of template variable names to their substituted content. - """ - raw_templates = _load_raw_templates() - - if not template_vars: - return dict(raw_templates) - - # Substitute variables into each template - substituted = {} - for var_name, content in raw_templates.items(): - try: - template = Template(content) - substituted[var_name] = template.safe_substitute(template_vars) - except Exception as exc: - logging.debug("Template substitution failed for %s: %s", var_name, exc) - substituted[var_name] = content - - return substituted - - -class PromptContent(NamedTuple): - """Container for prompt text, metadata, and its resolved path.""" - - text: str - path: Path - metadata: dict[str, Any] = {} - - -class PromptNotFoundError(FileNotFoundError): - """Raised when a prompt file cannot be located.""" - - def __init__(self, path: Path) -> None: - self.path = path - super().__init__(f"Prompt file not found: {path}") - - -def _flatten_identity_to_template_vars(identity: dict[str, Any]) -> dict[str, str]: - """Flatten identity config into template variables with uppercase-first versions. - - Parameters - ---------- - identity: - Identity configuration dictionary from get_config()['identity']. - - Returns - ------- - dict[str, str] - Template variables including flattened nested objects and uppercase-first versions. - For example: - - 'name' → identity['name'] - - 'pronouns_possessive' → identity['pronouns']['possessive'] - - 'Pronouns_possessive' → identity['pronouns']['possessive'].capitalize() - - 'bio' → identity['bio'] - """ - template_vars: dict[str, str] = {} - - # Flatten top-level and nested values - for key, value in identity.items(): - if isinstance(value, dict): - # Flatten nested dictionaries with underscore separator - for subkey, subvalue in value.items(): - var_name = f"{key}_{subkey}" - template_vars[var_name] = str(subvalue) - # Create uppercase-first version - template_vars[var_name.capitalize()] = str(subvalue).capitalize() - elif isinstance(value, (str, int, float, bool)): - # Top-level scalar values - template_vars[key] = str(value) - # Create uppercase-first version - template_vars[key.capitalize()] = str(value).capitalize() - - return template_vars - - -def load_prompt( - name: str, - base_dir: str | Path | None = None, - *, - include_journal: bool = False, - context: dict[str, Any] | None = None, -) -> PromptContent: - """Return the text contents, metadata, and path for a ``.md`` prompt file. - - Prompt files use JSON frontmatter for metadata. Supports Python - string.Template variable substitution using: - - Identity config from get_config()['identity']: - - Top-level fields: $name, $preferred, $bio, $timezone - - Nested fields with underscores: $pronouns_possessive, $pronouns_subject - - Uppercase-first versions: $Pronouns_possessive, $Name, $Bio - - Templates from think/templates/*.md: - - Each file becomes a variable named after its stem - - Example: daily_preamble.md -> $daily_preamble - - Templates are pre-processed with identity and context vars, so templates - can use $date, $preferred, etc. before being substituted into prompts - - Callers can provide additional context variables via the ``context`` parameter. - Context variables override identity and template variables if there's a collision. - Uppercase-first versions are automatically created for context variables. - - Parameters - ---------- - name: - Base filename of the prompt without the ``.md`` suffix. If the suffix is - included, it will not be duplicated. - base_dir: - Optional directory containing the prompt file. Defaults to the directory - of this module when not provided. - include_journal: - If True, prepends the content of ``think/journal.md`` to the requested - prompt. Defaults to False. Context variables are passed through to the - journal template as well. - context: - Optional dictionary of additional template variables. Values are converted - to strings. For each key, an uppercase-first version is also created - (e.g., ``{"day": "20250110"}`` adds both ``$day`` and ``$Day``). - - Returns - ------- - PromptContent - The prompt text (with surrounding whitespace removed and template variables - substituted), the resolved path to the ``.md`` file, and metadata from - the JSON frontmatter. - """ - - if not name: - raise ValueError("Prompt name must be provided") - - if name.endswith(".md"): - filename = name - else: - filename = f"{name}.md" - - prompt_dir = Path(base_dir) if base_dir is not None else Path(__file__).parent - prompt_path = prompt_dir / filename - try: - post = frontmatter.load( - prompt_path, - ) - text = post.content.strip() - metadata = dict(post.metadata) - except FileNotFoundError as exc: # pragma: no cover - caller handles missing prompt - raise PromptNotFoundError(prompt_path) from exc - - # Perform template substitution - try: - config = get_config() - identity = config.get("identity", {}) - template_vars = _flatten_identity_to_template_vars(identity) - - # Merge caller-provided context (overrides identity vars if collision) - if context: - for key, value in context.items(): - str_value = str(value) - template_vars[key] = str_value - # Add uppercase-first version - template_vars[key.capitalize()] = str_value.capitalize() - - # Load templates with identity and context vars so templates can use them - templates = _load_templates(template_vars) - template_vars.update(templates) - - # Use safe_substitute to avoid errors for undefined variables - template = Template(text) - text = template.safe_substitute(template_vars) - except Exception as exc: - # Log but don't fail - return original text if substitution fails - logging.debug("Template substitution failed for %s: %s", prompt_path, exc) - - # Prepend journal content if requested - if include_journal and name != "journal": - journal_content = load_prompt("journal", context=context) - text = f"{journal_content.text}\n\n{text}" - - return PromptContent(text=text, path=prompt_path, metadata=metadata) - - def get_journal_info() -> tuple[str, str]: """Return the journal path and its source. @@ -759,590 +535,6 @@ def setup_cli(parser: argparse.ArgumentParser, *, parse_known: bool = False): return (args, extra) if parse_known else args -def get_output_topic(key: str) -> str: - """Convert agent/generator key to filesystem-safe basename (no extension). - - Parameters - ---------- - key: - Generator key in format "topic" (system) or "app:topic" (app). - - Returns - ------- - str - Filesystem-safe name: "topic" or "_app_topic". - - Examples - -------- - >>> get_output_topic("activity") - 'activity' - >>> get_output_topic("chat:sentiment") - '_chat_sentiment' - """ - if ":" in key: - app, topic = key.split(":", 1) - return f"_{app}_{topic}" - return key - - -def key_to_context(key: str) -> str: - """Convert muse config key to context pattern. - - Parameters - ---------- - key: - Muse config key in format "name" (system) or "app:name" (app). - - Returns - ------- - str - Context pattern: "muse.system.{name}" or "muse.{app}.{name}". - - Examples - -------- - >>> key_to_context("meetings") - 'muse.system.meetings' - >>> key_to_context("entities:observer") - 'muse.entities.observer' - """ - if ":" in key: - app, name = key.split(":", 1) - return f"muse.{app}.{name}" - return f"muse.system.{key}" - - -def get_output_path( - day_dir: os.PathLike[str], - key: str, - segment: str | None = None, - output_format: str | None = None, - facet: str | None = None, -) -> Path: - """Return output path for generator agent output. - - Shared utility for determining where to write generator results. - Used by think/agents.py and think/cortex.py. - - Parameters - ---------- - day_dir: - Day directory path (YYYYMMDD). - key: - Generator key or agent name (e.g., "activity", "chat:sentiment", - "decisionalizer", "entities:observer"). - segment: - Optional segment key (HHMMSS_LEN) for segment-level output. - output_format: - Output format - "json" for JSON, anything else for markdown. - facet: - Optional facet name for multi-facet agents. When provided, the facet - is appended to the filename (e.g., "newsletter_work.md"). - - Returns - ------- - Path - Output file path: - - With segment: YYYYMMDD/{segment}/{topic}.{ext} - - Without segment: YYYYMMDD/agents/{topic}.{ext} - - With facet: {topic}_{facet}.{ext} instead of {topic}.{ext} - Where topic is derived from key and ext is "json" or "md". - """ - day = Path(day_dir) - topic = get_output_topic(key) - ext = "json" if output_format == "json" else "md" - - # Append facet suffix for multi-facet agent outputs - if facet: - filename = f"{topic}_{facet}.{ext}" - else: - filename = f"{topic}.{ext}" - - if segment: - # Segment output goes directly in segment directory - return day / segment / filename - else: - # Daily output goes in agents/ subdirectory - return day / "agents" / filename - - -def _load_prompt_metadata(md_path: Path) -> dict[str, object]: - """Load prompt metadata from .md file with JSON frontmatter. - - Parameters - ---------- - md_path: - Path to the .md prompt file with JSON frontmatter. - - Returns - ------- - dict - Metadata dict with path, mtime, color, and frontmatter fields. - """ - mtime = int(md_path.stat().st_mtime) - info: dict[str, object] = { - "path": str(md_path), - "mtime": mtime, - } - - try: - post = frontmatter.load( - md_path, - ) - if post.metadata: - info.update(post.metadata) - except Exception as exc: # pragma: no cover - metadata optional - logging.debug("Error reading frontmatter from %s: %s", md_path, exc) - - # Apply default color if not specified - if "color" not in info: - info["color"] = "#6c757d" - - return info - - -def get_muse_configs( - *, - has_tools: bool | None = None, - has_output: bool | None = None, - schedule: str | None = None, - include_disabled: bool = False, -) -> dict[str, dict[str, Any]]: - """Load muse configs from system and app directories. - - Unified function for loading both tool-using agents and generators from - muse/*.md files. Filters based on presence of tools/output fields. - - Args: - has_tools: If True, only configs with "tools" field (agents). - If False, only configs without "tools" field. - If None, no filtering on tools presence. - has_output: If True, only configs with "output" field (generators). - If False, only configs without "output" field. - If None, no filtering on output presence. - schedule: If provided, only configs where schedule matches this value - (e.g., "segment", "daily"). - include_disabled: If True, include configs with disabled=True. - Default False (for processing pipelines). - - Returns: - Dictionary mapping config keys to their metadata including: - - path: Path to the .md file - - source: "system" or "app" - - app: App name (only for app configs) - - All fields from frontmatter - """ - configs: dict[str, dict[str, Any]] = {} - - def matches_filter(info: dict) -> bool: - """Check if config matches the filter criteria.""" - # Check has_tools filter - if has_tools is True and "tools" not in info: - return False - if has_tools is False and "tools" in info: - return False - - # Check has_output filter - if has_output is True and "output" not in info: - return False - if has_output is False and "output" in info: - return False - - # Check specific schedule value - if schedule is not None and info.get("schedule") != schedule: - return False - - # Check disabled status - if not include_disabled and info.get("disabled", False): - return False - - return True - - # System configs from muse/ - if MUSE_DIR.is_dir(): - for md_path in sorted(MUSE_DIR.glob("*.md")): - name = md_path.stem - info = _load_prompt_metadata(md_path) - - if not matches_filter(info): - continue - - info["source"] = "system" - configs[name] = info - - # App configs from apps/*/muse/ - apps_dir = Path(__file__).parent.parent / "apps" - if apps_dir.is_dir(): - for app_path in sorted(apps_dir.iterdir()): - if not app_path.is_dir() or app_path.name.startswith("_"): - continue - app_muse_dir = app_path / "muse" - if not app_muse_dir.is_dir(): - continue - app_name = app_path.name - for md_path in sorted(app_muse_dir.glob("*.md")): - item_name = md_path.stem - info = _load_prompt_metadata(md_path) - - if not matches_filter(info): - continue - - key = f"{app_name}:{item_name}" - info["source"] = "app" - info["app"] = app_name - configs[key] = info - - # Merge journal config overrides from providers.contexts - providers_config = get_config().get("providers", {}) - contexts = providers_config.get("contexts", {}) - - for key, info in configs.items(): - context_key = key_to_context(key) - - # Check for exact match in contexts - override = contexts.get(context_key) - if override and isinstance(override, dict): - # Merge supported override fields - if "disabled" in override: - info["disabled"] = override["disabled"] - if "extract" in override: - info["extract"] = override["extract"] - if "tier" in override: - info["tier"] = override["tier"] - if "provider" in override: - info["provider"] = override["provider"] - - # Validate: scheduled prompts must have explicit priority - for key, info in configs.items(): - if info.get("schedule") and "priority" not in info: - raise ValueError( - f"Scheduled prompt '{key}' is missing required 'priority' field. " - f"All prompts with 'schedule' must declare an explicit priority." - ) - - return configs - - -def _resolve_agent_path(name: str) -> tuple[Path, str]: - """Resolve agent name to directory path and agent filename. - - Parameters - ---------- - name: - Agent name - either system agent (e.g., "default") or - app-namespaced agent (e.g., "chat:helper"). - - Returns - ------- - tuple[Path, str] - (agent_directory, agent_name) tuple. - """ - if ":" in name: - # App agent: "chat:helper" -> apps/chat/muse/helper - app, agent_name = name.split(":", 1) - agent_dir = Path(__file__).parent.parent / "apps" / app / "muse" - else: - # System agent: "default" -> muse/default - agent_dir = MUSE_DIR - agent_name = name - return agent_dir, agent_name - - -# Default instruction configuration - all false, agents must explicitly opt-in -_DEFAULT_INSTRUCTIONS = { - "system": None, - "facets": False, - "sources": { - "audio": False, - "screen": False, - "agents": False, - }, -} - - -def _merge_instructions_config(defaults: dict, overrides: dict | None) -> dict: - """Merge instruction config overrides into defaults. - - Handles nested "sources" dict specially. - - Parameters - ---------- - defaults: - Default instruction configuration. - overrides: - Optional overrides from .json "instructions" key. - - Returns - ------- - dict - Merged configuration. - """ - if not overrides: - return defaults.copy() - - result = defaults.copy() - - # Merge top-level keys - for key in ("system", "facets"): - if key in overrides: - result[key] = overrides[key] - - # Merge sources dict if present - if "sources" in overrides and isinstance(overrides["sources"], dict): - result["sources"] = {**defaults.get("sources", {}), **overrides["sources"]} - - return result - - -def compose_instructions( - *, - user_prompt: str | None = None, - user_prompt_dir: Path | None = None, - facet: str | None = None, - include_datetime: bool = True, - config_overrides: dict | None = None, -) -> dict: - """Compose instruction components for agents or generators. - - This is the shared function for building system_instruction, user_instruction, - extra_context, and sources configuration. Both agents and generators use this - to ensure consistent prompt composition. - - Parameters - ---------- - user_prompt: - Name of the user instruction prompt to load (e.g., "default" for agents). - If None, no user_instruction is included (typical for generators). - user_prompt_dir: - Directory to load user_prompt from. If None, uses think/ directory. - facet: - Optional facet name to focus on. When provided, extra_context includes - only this facet's info (detail level controlled by "facets" setting). - include_datetime: - Whether to include current date/time in extra_context. Default True - for agents (real-time chat), typically False for generators (past analysis). - config_overrides: - Optional dict from .json "instructions" key. Supported keys: - - "system": prompt name for system instruction (default: "journal") - - "facets": false | true | "full" (default: true) - false = skip facet context - true = include facet context with names only - "full" = include facet context with full descriptions - For faceted generators, shows focused facet; for unfaceted, shows all facets. - - "sources": {"audio": bool, "screen": bool, "agents": bool|dict} - The "agents" source can be: - - bool: True (all agents), False (no agents) - - "required": all agents, fail if none found - - dict: selective filtering, e.g., {"entities": true, "meetings": "required"} - - Returns - ------- - dict - Composed instruction configuration: - - system_instruction: str - loaded from "system" prompt - - system_prompt_name: str - name of system prompt (for cache keys) - - user_instruction: str | None - loaded from user_prompt if provided - - extra_context: str | None - facets + datetime - - sources: dict - {"audio": bool, "screen": bool, "agents": bool|dict} - """ - # Merge defaults with overrides - cfg = _merge_instructions_config(_DEFAULT_INSTRUCTIONS, config_overrides) - - result: dict = {} - - # Load system instruction (None means no system prompt) - system_name = cfg.get("system") - if system_name: - system_prompt = load_prompt(system_name) - result["system_instruction"] = system_prompt.text - result["system_prompt_name"] = system_name - else: - result["system_instruction"] = "" - result["system_prompt_name"] = "" - - # Load user instruction if specified - if user_prompt: - base_dir = user_prompt_dir if user_prompt_dir else Path(__file__).parent - user_prompt_obj = load_prompt(user_prompt, base_dir=base_dir) - result["user_instruction"] = user_prompt_obj.text - else: - result["user_instruction"] = None - - # Build extra_context based on facets setting - # Values: false (skip), true (names only), "full" (with descriptions) - extra_parts = [] - facets_setting = cfg.get("facets", False) - facets_full = facets_setting == "full" - - if facets_setting: - if facet: - # Focused facet mode: include only this facet's context - try: - from think.facets import facet_summary - - summary = facet_summary(facet, detailed=facets_full) - extra_parts.append(f"## Facet Focus\n{summary}") - except Exception: - pass # Ignore if facet can't be loaded - else: - # General mode: all facets - try: - from think.facets import facet_summaries - - summary = facet_summaries(detailed=facets_full) - if summary and summary != "No facets found.": - extra_parts.append(summary) - except Exception: - pass # Ignore if facets can't be loaded - - # Add current date/time if requested - if include_datetime: - now = datetime.now() - try: - import tzlocal - - local_tz = tzlocal.get_localzone() - now_local = now.astimezone(local_tz) - time_str = now_local.strftime("%A, %B %d, %Y at %I:%M %p %Z") - except Exception: - time_str = now.strftime("%A, %B %d, %Y at %I:%M %p") - extra_parts.append(f"## Current Date and Time\nToday is {time_str}") - - result["extra_context"] = "\n\n".join(extra_parts).strip() if extra_parts else None - - # Include sources config - result["sources"] = cfg.get("sources", _DEFAULT_INSTRUCTIONS["sources"]) - - return result - - -def source_is_enabled(value: bool | str | dict) -> bool: - """Check if a source should be loaded based on its config value. - - Sources can be configured as: - - False: don't load - - True: load if available - - "required": load (and generation will fail if none found) - - dict: for agents source, selective loading (e.g., {"entities": true}) - - Both True and "required" mean the source should be loaded. - A non-empty dict means the source should be loaded (with filtering). - - Args: - value: The source config value (bool, "required" string, or dict for agents) - - Returns: - True if the source should be loaded, False otherwise. - """ - if isinstance(value, dict): - # Dict means selective loading - enabled if any agent is enabled - return any(v is True or v == "required" for v in value.values()) - return value is True or value == "required" - - -def source_is_required(value: bool | str | dict) -> bool: - """Check if a source must have content for generation to proceed. - - Args: - value: The source config value (bool, "required" string, or dict for agents) - - Returns: - True if the source is required (generation should skip if no content). - For dict values, returns True if any agent is marked "required". - """ - if isinstance(value, dict): - return any(v == "required" for v in value.values()) - return value == "required" - - -def get_agent_filter(value: bool | str | dict) -> dict[str, bool | str] | None: - """Extract agent filter from sources config. - - When agents source is a dict, returns it as filter mapping agent names - to their enabled/required status. When agents source is bool or "required", - returns None to indicate all agents should be loaded. - - Args: - value: The agents source config value - - Returns: - Dict mapping agent names to bool/"required", or None for all agents. - Returns empty dict if value is False (no agents). - - Examples: - >>> get_agent_filter(True) - None # All agents - >>> get_agent_filter(False) - {} # No agents - >>> get_agent_filter({"entities": True, "meetings": "required"}) - {"entities": True, "meetings": "required"} - """ - if isinstance(value, dict): - return value - if value is False: - return {} # No agents - return None # All agents (True or "required") - - -def get_agent(name: str = "default", facet: str | None = None) -> dict: - """Return complete agent configuration by name. - - Loads configuration from .md file with JSON frontmatter and instruction text, - merges with runtime context. - - Parameters - ---------- - name: - Agent name to load. Can be a system agent (e.g., "default") - or an app-namespaced agent (e.g., "chat:helper" for apps/chat/muse/helper). - facet: - Optional facet name to focus on. When provided, includes detailed - information for just this facet (with full entity details) instead - of summaries of all facets. - - Returns - ------- - dict - Complete agent configuration including system_instruction, user_instruction, - extra_context, model, backend, etc. - """ - # Resolve agent path based on namespace - agent_dir, agent_name = _resolve_agent_path(name) - - # Verify agent prompt file exists - md_path = agent_dir / f"{agent_name}.md" - if not md_path.exists(): - raise FileNotFoundError(f"Agent not found: {name}") - - # Load config from frontmatter - post = frontmatter.load( - md_path, - ) - config = dict(post.metadata) if post.metadata else {} - - # Extract instructions config if present - instructions_config = config.pop("instructions", None) - - # Use compose_instructions for consistent prompt composition - instructions = compose_instructions( - user_prompt=agent_name, - user_prompt_dir=agent_dir, - facet=facet, - include_datetime=True, - config_overrides=instructions_config, - ) - - # Merge instruction results into config - config["system_instruction"] = instructions["system_instruction"] - config["user_instruction"] = instructions["user_instruction"] - if instructions["extra_context"]: - config["extra_context"] = instructions["extra_context"] - - # Set agent name - config["name"] = name - - return config - - def create_mcp_client(http_uri: str) -> Any: """Return a FastMCP HTTP client for solstone tools.""" -- 2.51.2