feat: retrieve config value using template replacement in skill content (#209)

frostming · web-flow · commit f0c32127450f · 2026-05-13T10:09:19.000+08:00
Signed-off-by: Frost Ming &lt;me@frostming.com&gt;
diff --git 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
@@ -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
@@ -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 <CHAT_ID> --message -
+cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_send.py --chat-id <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 <CHAT_ID> --reply-to <MESSAGE_ID> --message -
+cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_send.py --chat-id <CHAT_ID> --token "$BUB_TELEGRAM_TOKEN" --reply-to <MESSAGE_ID> --message -
 Reply content here.
 EOF
 
 # Edit an existing message
-cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_edit.py --chat-id <CHAT_ID> --message-id <MESSAGE_ID> --text -
+cat << 'EOF' | uv run ${SKILL_DIR}/scripts/telegram_edit.py --chat-id <CHAT_ID> --token "$BUB_TELEGRAM_TOKEN" --message-id <MESSAGE_ID> --text -
 Updated content here.
 EOF
 ```
diff --git 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
@@ -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()
