From 87f04d38905bcca201c9ab690494dbb916dcbe7f Mon Sep 17 00:00:00 2001 From: Frost Ming Date: Wed, 13 May 2026 10:02:11 +0800 Subject: [PATCH] feat: retrieve config value using template replacement in skill content Signed-off-by: Frost Ming --- src/bub/configure.py | 47 ++++++++++++++++++++++++++++++ src/bub/skills.py | 30 +++++++++++++++++-- src/skills/telegram/SKILL.md | 10 ++++--- tests/test_configure.py | 56 ++++++++++++++++++++++++++++++++++++ tests/test_skills.py | 53 ++++++++++++++++++++++++++++++++++ 5 files changed, 189 insertions(+), 7 deletions(-) diff --git a/src/bub/configure.py b/src/bub/configure.py index 0d0d6b17..eb6c6329 100644 --- a/src/bub/configure.py +++ b/src/bub/configure.py @@ -6,6 +6,7 @@ CONFIG_MAP: dict[str, list[type[BaseSettings]]] = {} ROOT = "" +MISSING = object() _global_config: dict[str, list[BaseSettings]] = {} _config_data: dict[str, Any] = {} @@ -95,6 +96,22 @@ def ensure_config[C: BaseSettings](config_cls: type[C]) -> C: return instance +def get_value(path: str, default: Any = MISSING) -> Any: + """Get a loaded config value by dotted path, preserving registered settings behavior.""" + + parts = [part for part in path.split(".") if part] + if not parts: + raise ValueError("config path must not be empty") + + value = _lookup_registered_config(parts) + if value is not MISSING: + return value + + if default is not MISSING: + return default + raise KeyError(path) + + def _copy_dict(data: dict[str, Any]) -> dict[str, Any]: copied: dict[str, Any] = {} for key, value in data.items(): @@ -105,6 +122,36 @@ def _copy_dict(data: dict[str, Any]) -> dict[str, Any]: return copied +def _lookup_registered_config(parts: list[str]) -> Any: + section, *subpath = parts + if section in CONFIG_MAP and section != ROOT: + for config_cls in CONFIG_MAP[section]: + value = _lookup_path(ensure_config(config_cls), subpath) + if value is not MISSING: + return value + + for config_cls in CONFIG_MAP.get(ROOT, []): + value = _lookup_path(ensure_config(config_cls), parts) + if value is not MISSING: + return value + + return MISSING + + +def _lookup_path(value: Any, parts: list[str]) -> Any: + current = value + for part in parts: + if isinstance(current, dict): + if part not in current: + return MISSING + current = current[part] + continue + if not hasattr(current, part): + return MISSING + current = getattr(current, part) + return current + + def _merge_into(target: dict[str, Any], incoming: dict[str, Any], path: tuple[str, ...]) -> None: for key, value in incoming.items(): existing = target.get(key) diff --git a/src/bub/skills.py b/src/bub/skills.py index 683a98a8..2bd53377 100644 --- a/src/bub/skills.py +++ b/src/bub/skills.py @@ -13,11 +13,14 @@ import yaml +import bub.configure as configure + PROJECT_SKILLS_DIR = ".agents/skills" LEGACY_SKILLS_DIR = ".agent/skills" SKILL_FILE_NAME = "SKILL.md" SKILL_SOURCES = ("project", "global", "builtin") SKILL_NAME_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +CONFIG_TEMPLATE_PATTERN = re.compile(r"\$\{\s*config\.([a-zA-Z0-9_.-]+)\s*\}") @dataclass(frozen=True) @@ -33,11 +36,15 @@ class SkillMetadata: def body(self) -> str: front_matter_pattern = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL) try: - template = string.Template(self.location.read_text(encoding="utf-8").strip()) + template_content = self.location.read_text(encoding="utf-8").strip() except OSError: return "" - content = template.safe_substitute({"SKILL_DIR": str(self.location.parent), "PYTHON": sys.executable}) - return front_matter_pattern.sub("", content, count=1).strip() + raw_content = front_matter_pattern.sub("", template_content, count=1).strip() + content = _render_config_templates(raw_content) + return string.Template(content).safe_substitute({ + "SKILL_DIR": str(self.location.parent), + "PYTHON": sys.executable, + }) def discover_skills(workspace_path: Path) -> list[SkillMetadata]: @@ -60,6 +67,23 @@ def discover_skills(workspace_path: Path) -> list[SkillMetadata]: return sorted(skills_by_name.values(), key=lambda item: item.name.casefold()) +def _render_config_templates(content: str) -> str: + def replace(match: re.Match[str]) -> str: + try: + value = configure.get_value(match.group(1), default="") + except KeyError: + return match.group(0) + if isinstance(value, str): + return value + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, int | float): + return str(value) + return yaml.safe_dump(value, sort_keys=False).strip() + + return CONFIG_TEMPLATE_PATTERN.sub(replace, content) + + def _read_skill(skill_dir: Path, *, source: str) -> SkillMetadata | None: skill_file = skill_dir / SKILL_FILE_NAME if not skill_file.is_file(): diff --git a/src/skills/telegram/SKILL.md b/src/skills/telegram/SKILL.md index 20334a78..dcb0f8c5 100644 --- a/src/skills/telegram/SKILL.md +++ b/src/skills/telegram/SKILL.md @@ -13,7 +13,9 @@ metadata: Agent-facing execution guide for Telegram outbound communication. -Assumption: `BUB_TELEGRAM_TOKEN` is already available. +Env vars: + +- `BUB_TELEGRAM_TOKEN=${config.telegram.token}` ## Required Inputs Collect these before execution: @@ -76,18 +78,18 @@ Paths are relative to this skill directory. ```bash # Send message (ALWAYS use heredoc stdin, never inline text in arguments) -cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_send.py --chat-id --message - +cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_send.py --chat-id --token "$BUB_TELEGRAM_TOKEN" --message - Your message content here. Special characters are safe: $100, "quotes", 'apostrophes', !exclamation EOF # Reply to a specific message -cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_send.py --chat-id --reply-to --message - +cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_send.py --chat-id --token "$BUB_TELEGRAM_TOKEN" --reply-to --message - Reply content here. EOF # Edit an existing message -cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_edit.py --chat-id --message-id --text - +cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_edit.py --chat-id --token "$BUB_TELEGRAM_TOKEN" --message-id --text - Updated content here. EOF ``` diff --git a/tests/test_configure.py b/tests/test_configure.py index 8f854a6a..30674206 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -75,3 +75,59 @@ def test_save_writes_yaml_and_refreshes_loaded_config(tmp_path: Path) -> None: assert configure.ensure_config(TelegramSettings).token == expected_token finally: os.chdir(previous_cwd) + + +def test_get_value_reads_registered_section_from_yaml(load_config) -> None: + with patch.dict(os.environ, {}, clear=True): + load_config( + """ +telegram: + token: yaml-token +""".strip(), + ) + + assert configure.get_value("telegram.token") == "yaml-token" + + +def test_get_value_prefers_registered_env_over_yaml(load_config) -> None: + load_config( + """ +telegram: + token: yaml-token +""".strip(), + ) + + with patch.dict(os.environ, {"BUB_TELEGRAM_TOKEN": "env-token"}, clear=True): + configure._global_config.clear() + + assert configure.get_value("telegram.token") == "env-token" + + +def test_get_value_descends_into_registered_dict_field(load_config) -> None: + with patch.dict(os.environ, {}, clear=True): + load_config( + """ +api_key: + openai: sk-yaml +""".strip(), + ) + + assert configure.get_value("api_key") == {"openai": "sk-yaml"} + assert configure.get_value("api_key.openai") == "sk-yaml" + + +def test_get_value_ignores_raw_unregistered_path(load_config) -> None: + load_config( + """ +custom: + nested: + value: raw-value +""".strip(), + ) + + with pytest.raises(KeyError): + configure.get_value("custom.nested.value") + + +def test_get_value_returns_default_for_missing_path() -> None: + assert configure.get_value("missing.value", default="fallback") == "fallback" diff --git a/tests/test_skills.py b/tests/test_skills.py index 12bcb563..617445f3 100644 --- a/tests/test_skills.py +++ b/tests/test_skills.py @@ -1,5 +1,8 @@ from pathlib import Path +from unittest.mock import patch +import bub.configure as configure +from bub.channels.telegram import TelegramSettings from bub.skills import ( SKILL_FILE_NAME, SkillMetadata, @@ -46,6 +49,56 @@ def test_skill_metadata_body_strips_frontmatter(tmp_path: Path) -> None: assert metadata.body() == "Line 1\nLine 2" +def test_skill_metadata_body_renders_config_templates(tmp_path: Path, load_config) -> None: + assert TelegramSettings.__name__ == "TelegramSettings" + skill_file = _write_skill( + tmp_path, + "demo-skill", + body='Token: "${config.telegram.token}"\nSkill dir: $SKILL_DIR', + ) + metadata = SkillMetadata( + name="demo-skill", + description="Demo", + location=skill_file, + source="project", + ) + + with patch.dict("os.environ", {}, clear=True): + load_config( + """ +telegram: + token: yaml-token +""".strip(), + ) + + body = metadata.body() + + assert 'Token: "yaml-token"' in body + assert f"Skill dir: {tmp_path / 'demo-skill'}" in body + + +def test_skill_metadata_body_renders_env_over_config(tmp_path: Path, load_config) -> None: + assert TelegramSettings.__name__ == "TelegramSettings" + skill_file = _write_skill(tmp_path, "demo-skill", body='Token: "${config.telegram.token}"') + metadata = SkillMetadata( + name="demo-skill", + description="Demo", + location=skill_file, + source="project", + ) + load_config( + """ +telegram: + token: yaml-token +""".strip(), + ) + + with patch.dict("os.environ", {"BUB_TELEGRAM_TOKEN": "env-token"}, clear=True): + configure._global_config.clear() + + assert metadata.body() == 'Token: "env-token"' + + def test_read_skill_rejects_invalid_metadata_field_type(tmp_path: Path) -> None: skill_dir = tmp_path / "bad-skill" skill_dir.mkdir()