refactor(skill): unify runtime skill snapshot source

server/utils/lifespan.py

@@ -4,7 +4,6 @@ from fastapi import FastAPI
 
 from src.services.task_service import tasker
 from src.services.mcp_service import init_mcp_servers
-from src.services.run_queue_service import close_queue_clients, get_redis_client
 from src.storage.postgres.manager import pg_manager
 from src.knowledge import knowledge_base
 from src.utils import logger

src/agents/common/backends/skills_backend.py

@@ -132,13 +132,12 @@ def create_agent_composite_backend(runtime) -> CompositeBackend:
 
 
 def _get_visible_skills_from_runtime(runtime) -> list[str]:
-    state = getattr(runtime, "state", None)
+    context = getattr(runtime, "context", None)
-    if isinstance(state, dict):
+    snapshot = getattr(context, "skill_session_snapshot", None)
-        snapshot = state.get("skill_session_snapshot")
+    if isinstance(snapshot, dict):
-        if isinstance(snapshot, dict):
+        visible = snapshot.get("visible_skills")
-            visible = snapshot.get("visible_skills")
+        if isinstance(visible, list):
-            if isinstance(visible, list):
+            return [slug for slug in visible if isinstance(slug, str) and is_valid_skill_slug(slug)]
-                return [slug for slug in visible if isinstance(slug, str) and is_valid_skill_slug(slug)]
 
-    selected = getattr(runtime.context, "skills", None) or []
+    selected = getattr(context, "skills", None) or []
     return normalize_selected_skills(selected)

src/agents/common/middlewares/runtime_config_middleware.py

@@ -18,7 +18,6 @@ from src.services.skill_resolver import (
     SkillSessionSnapshot,
     build_dependency_bundle,
     collect_prompt_metadata,
-    is_snapshot_match_selected_skills,
     normalize_selected_skills,
     resolve_session_snapshot,
 )
@@ -123,11 +122,44 @@ class RuntimeConfigMiddleware(AgentMiddleware):
             f"skills={skills_context_name}"
         )
 
+    async def abefore_agent(self, state: RuntimeConfigState, runtime) -> dict[str, Any] | None:
+        runtime_context = runtime.context
+        configured_skills = getattr(runtime_context, self.skills_context_name, None) or []
+        selected_skills = normalize_selected_skills(configured_skills)
+
+        try:
+            snapshot = await resolve_session_snapshot(selected_skills)
+        except Exception as e:
+            logger.warning(f"RuntimeConfigMiddleware: failed to resolve skill snapshot in abefore_agent: {e}")
+            snapshot = {
+                "selected_skills": selected_skills,
+                "visible_skills": [],
+                "prompt_metadata": {},
+                "dependency_map": {},
+            }
+
+        setattr(runtime_context, "skill_session_snapshot", snapshot)
+
+        if not self.enable_system_prompt_override or not self.enable_skills_prompt_override:
+            return None
+        if getattr(runtime_context, "_skills_prompt_injected", False):
+            return None
+        if not snapshot.get("visible_skills"):
+            return None
+
+        skills_meta = collect_prompt_metadata(snapshot, snapshot.get("visible_skills") or [])
+        skills_section = self._build_skills_section(skills_meta)
+        base_prompt = getattr(runtime_context, self.system_prompt_context_name, "") or ""
+        merged_prompt = f"{base_prompt}\n\n{skills_section}" if base_prompt else skills_section
+        setattr(runtime_context, self.system_prompt_context_name, merged_prompt)
+        setattr(runtime_context, "_skills_prompt_injected", True)
+        return None
+
     async def awrap_model_call(
         self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
     ) -> ModelResponse:
         runtime_context = request.runtime.context
-        snapshot, request = await self._ensure_skill_snapshot(request)
+        snapshot = self._get_skill_snapshot_from_context(runtime_context)
         overrides: dict[str, Any] = {}
 
         # 1. 模型覆盖（可选）
@@ -165,20 +197,6 @@ class RuntimeConfigMiddleware(AgentMiddleware):
             system_prompt = getattr(runtime_context, self.system_prompt_context_name, "") or ""
             merged_system_prompt = f"{cur_datetime}\n\n{system_prompt}"
 
-            configured_skills = getattr(runtime_context, self.skills_context_name, None) or []
-            if self.enable_skills_prompt_override and configured_skills:
-                if self._supports_skill_prompt(request):
-                    skills_for_prompt = configured_skills
-                    if snapshot and isinstance(snapshot.get("visible_skills"), list):
-                        skills_for_prompt = snapshot.get("visible_skills") or []
-                    skills_meta = collect_prompt_metadata(snapshot, skills_for_prompt)
-                    skills_section = self._build_skills_section(skills_meta)
-                    merged_system_prompt = f"{merged_system_prompt}\n\n{skills_section}"
-                else:
-                    logger.warning(
-                        "RuntimeConfigMiddleware: skills configured but read_file unavailable, skip skills prompt"
-                    )
-
             content_blocks = list(request.system_message.content_blocks) if request.system_message else []
             new_content = content_blocks + [{"type": "text", "text": merged_system_prompt}]
             new_system_message = SystemMessage(content=new_content)
@@ -291,30 +309,6 @@ class RuntimeConfigMiddleware(AgentMiddleware):
         logger.debug(f"RuntimeConfigMiddleware: activated skill by read_file: {slug}")
         return self._merge_activated_skill_update(result, slug)
 
-    async def _ensure_skill_snapshot(self, request: ModelRequest) -> tuple[SkillSessionSnapshot | None, ModelRequest]:
-        runtime_context = request.runtime.context
-        configured_skills = getattr(runtime_context, self.skills_context_name, None) or []
-        normalized_skills = normalize_selected_skills(configured_skills)
-        state = request.state if isinstance(request.state, dict) else {}
-
-        snapshot = self._get_skill_snapshot_from_state(state)
-        if is_snapshot_match_selected_skills(snapshot, normalized_skills):
-            return snapshot, request
-
-        try:
-            snapshot = await resolve_session_snapshot(normalized_skills)
-        except Exception as e:
-            logger.warning(f"RuntimeConfigMiddleware: failed to resolve skill snapshot, fallback empty: {e}")
-            snapshot = None
-
-        if isinstance(request.state, dict):
-            if snapshot:
-                request.state["skill_session_snapshot"] = snapshot
-            else:
-                request.state.pop("skill_session_snapshot", None)
-
-        return snapshot, request
-
     def _extract_skill_slug_from_skill_md_path(self, file_path: Any) -> str | None:
         if not isinstance(file_path, str):
             return None
@@ -345,7 +339,7 @@ class RuntimeConfigMiddleware(AgentMiddleware):
         return result
 
     def _is_visible_skill_slug(self, request: ToolCallRequest, slug: str) -> bool:
-        snapshot = self._get_skill_snapshot_from_state(request.state)
+        snapshot = self._get_skill_snapshot_from_context(request.runtime.context)
         if snapshot:
             visible_skills = snapshot.get("visible_skills")
             if isinstance(visible_skills, list):
@@ -355,10 +349,8 @@ class RuntimeConfigMiddleware(AgentMiddleware):
         normalized = normalize_selected_skills(configured_skills)
         return slug in normalized
 
-    def _get_skill_snapshot_from_state(self, state: Any) -> SkillSessionSnapshot | None:
+    def _get_skill_snapshot_from_context(self, context: Any) -> SkillSessionSnapshot | None:
-        if not isinstance(state, dict):
+        snapshot = getattr(context, "skill_session_snapshot", None)
-            return None
-        snapshot = state.get("skill_session_snapshot")
         if not isinstance(snapshot, dict):
             return None
         visible_skills = snapshot.get("visible_skills")
@@ -367,13 +359,6 @@ class RuntimeConfigMiddleware(AgentMiddleware):
             return None
         return snapshot
 
-    def _supports_skill_prompt(self, request: ModelRequest) -> bool:
-        """仅当请求工具中包含 read_file 时，才注入 skills 指引。"""
-        for tool in request.tools or []:
-            if getattr(tool, "name", None) == "read_file":
-                return True
-        return False
-
     def _format_skills_locations(self, sources: list[str]) -> str:
         locations = []
         for i, source_path in enumerate(sources):

test/test_runtime_config_middleware_skills.py

@@ -44,8 +44,23 @@ async def _echo_handler(request):
     return request
 
 
-def _build_request(*, skills: list[str], tools: list[str], system_prompt: str = "你是助手", state=None) -> _FakeRequest:
+def _build_request(
-    context = SimpleNamespace(system_prompt=system_prompt, skills=skills, tools=[], knowledges=[], mcps=[])
+    *,
+    skills: list[str],
+    tools: list[str],
+    system_prompt: str = "你是助手",
+    state: dict[str, Any] | None = None,
+    skill_snapshot: dict[str, Any] | None = None,
+) -> _FakeRequest:
+    context = SimpleNamespace(
+        system_prompt=system_prompt,
+        skills=skills,
+        tools=[],
+        knowledges=[],
+        mcps=[],
+    )
+    if skill_snapshot is not None:
+        context.skill_session_snapshot = skill_snapshot
     runtime = SimpleNamespace(context=context)
     return _FakeRequest(
         runtime=runtime,
@@ -58,15 +73,13 @@ def _build_request(*, skills: list[str], tools: list[str], system_prompt: str =
 def _build_tool_request(*, skills: list[str], visible_skills: list[str], file_path: str) -> _FakeToolCallRequest:
     return _FakeToolCallRequest(
         tool_call={"name": "read_file", "args": {"file_path": file_path}},
-        runtime=SimpleNamespace(context=SimpleNamespace(skills=skills)),
+        runtime=SimpleNamespace(
-        state={
+            context=SimpleNamespace(
-            "skill_session_snapshot": {
+                skills=skills,
-                "selected_skills": skills,
+                skill_session_snapshot=_build_snapshot(selected=skills, visible=visible_skills),
-                "visible_skills": visible_skills,
+            )
-                "prompt_metadata": {},
+        ),
-                "dependency_map": {},
+        state={},
-            }
-        },
     )
 
 
@@ -83,123 +96,124 @@ def _build_middleware() -> RuntimeConfigMiddleware:
     )
 
 
-def _build_snapshot(selected: list[str], metadata: dict[str, dict[str, str]] | None = None) -> dict[str, Any]:
+def _build_snapshot(
+    selected: list[str],
+    visible: list[str] | None = None,
+    metadata: dict[str, dict[str, str]] | None = None,
+    dependency_map: dict[str, dict[str, list[str]]] | None = None,
+) -> dict[str, Any]:
     return {
         "selected_skills": selected,
-        "visible_skills": selected,
+        "visible_skills": visible if visible is not None else selected,
         "prompt_metadata": metadata or {},
-        "dependency_map": {},
+        "dependency_map": dependency_map or {},
     }
 
 
 @pytest.mark.asyncio
-async def test_injects_skills_section_when_skills_configured_and_read_file_available(monkeypatch: pytest.MonkeyPatch):
+async def test_abefore_agent_resolves_visible_skills_and_preinjects_prompt(monkeypatch: pytest.MonkeyPatch):
     async def fake_resolve(selected):
-        assert selected == ["research-report"]
+        assert selected == ["deliver-prd"]
         return _build_snapshot(
-            ["research-report"],
+            selected=["deliver-prd"],
-            {
+            visible=["deliver-prd", "brainstorming"],
-                "research-report": {
+            metadata={
-                    "name": "research-report",
+                "deliver-prd": {
-                    "description": "Write structured research reports",
+                    "name": "deliver-prd",
-                    "path": "/skills/research-report/SKILL.md",
+                    "description": "deliver prd",
-                }
+                    "path": "/skills/deliver-prd/SKILL.md",
+                },
+                "brainstorming": {
+                    "name": "brainstorming",
+                    "description": "brainstorming desc",
+                    "path": "/skills/brainstorming/SKILL.md",
+                },
+            },
+            dependency_map={
+                "deliver-prd": {"tools": [], "mcps": [], "skills": ["brainstorming"]},
+                "brainstorming": {"tools": [], "mcps": [], "skills": []},
             },
         )
 
     monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
     middleware = _build_middleware()
-    request = _build_request(skills=["research-report"], tools=["read_file"])
+    request = _build_request(skills=["deliver-prd"], tools=["read_file"], system_prompt="你是助手")
 
+    await middleware.abefore_agent(request.state, request.runtime)
+
+    snapshot = request.runtime.context.skill_session_snapshot
+    assert snapshot["selected_skills"] == ["deliver-prd"]
+    assert snapshot["visible_skills"] == ["deliver-prd", "brainstorming"]
+    assert snapshot["dependency_map"]["deliver-prd"]["skills"] == ["brainstorming"]
+    assert request.runtime.context._skills_prompt_injected is True
+    assert "## Skills System" in request.runtime.context.system_prompt
+    assert "- **deliver-prd**: deliver prd" in request.runtime.context.system_prompt
+    assert "- **brainstorming**: brainstorming desc" in request.runtime.context.system_prompt
+
+
+@pytest.mark.asyncio
+async def test_abefore_agent_injection_is_idempotent(monkeypatch: pytest.MonkeyPatch):
+    async def fake_resolve(_selected):
+        return _build_snapshot(
+            selected=["alpha"],
+            metadata={"alpha": {"name": "alpha", "description": "alpha desc", "path": "/skills/alpha/SKILL.md"}},
+        )
+
+    monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
+    middleware = _build_middleware()
+    request = _build_request(skills=["alpha"], tools=["read_file"], system_prompt="base prompt")
+
+    await middleware.abefore_agent(request.state, request.runtime)
+    await middleware.abefore_agent(request.state, request.runtime)
+
+    assert request.runtime.context.system_prompt.count("## Skills System") == 1
+
+
+@pytest.mark.asyncio
+async def test_awrap_model_call_keeps_preinjected_skills_prompt(monkeypatch: pytest.MonkeyPatch):
+    middleware = _build_middleware()
+    request = _build_request(
+        skills=["alpha"],
+        tools=["read_file"],
+        system_prompt="base prompt\n\n## Skills System\n- **alpha**: alpha desc",
+    )
+
+    def raise_if_called(_skills_meta):
+        raise AssertionError("_build_skills_section should not be called in awrap_model_call")
+
+    monkeypatch.setattr(middleware, "_build_skills_section", raise_if_called)
     result = await middleware.awrap_model_call(request, _echo_handler)
     prompt = _extract_appended_prompt(result)
 
-    assert "## Skills System" in prompt
-    assert "**Skills Skills**: `/skills/` (higher priority)" in prompt
-    assert "- **research-report**: Write structured research reports" in prompt
-    assert "Read `/skills/research-report/SKILL.md` for full instructions" in prompt
-    assert "Recognize when a skill applies" in prompt
     assert "当前时间：" in prompt
-    assert "skill_session_snapshot" in result.state
+    assert "## Skills System" in prompt
+    assert "- **alpha**: alpha desc" in prompt
 
 
 @pytest.mark.asyncio
-async def test_skips_skills_section_when_context_skills_empty():
+async def test_abefore_agent_degrades_when_resolver_fails(monkeypatch: pytest.MonkeyPatch):
-    middleware = _build_middleware()
-    request = _build_request(skills=[], tools=["read_file"])
-
-    result = await middleware.awrap_model_call(request, _echo_handler)
-    prompt = _extract_appended_prompt(result)
-
-    assert "## Skills System" not in prompt
-
-
-@pytest.mark.asyncio
-async def test_skips_skills_section_without_read_file_and_logs_warning(monkeypatch: pytest.MonkeyPatch):
-    warnings: list[str] = []
-    fake_logger = SimpleNamespace(
-        debug=lambda *_args, **_kwargs: None,
-        warning=lambda message: warnings.append(message),
-    )
-    monkeypatch.setattr(runtime_middleware, "logger", fake_logger)
-
     async def fake_resolve(_selected):
-        return _build_snapshot(
+        raise RuntimeError("boom")
-            ["research-report"],
-            {
-                "research-report": {
-                    "name": "research-report",
-                    "description": "Write structured research reports",
-                    "path": "/skills/research-report/SKILL.md",
-                }
-            },
-        )
 
     monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
     middleware = _build_middleware()
-    request = _build_request(skills=["research-report"], tools=["write_file"])
+    request = _build_request(skills=["alpha"], tools=["read_file"], system_prompt="base prompt")
 
-    result = await middleware.awrap_model_call(request, _echo_handler)
+    await middleware.abefore_agent(request.state, request.runtime)
-    prompt = _extract_appended_prompt(result)
 
-    assert "## Skills System" not in prompt
+    snapshot = request.runtime.context.skill_session_snapshot
-    assert any("read_file unavailable" in msg for msg in warnings)
+    assert snapshot["selected_skills"] == ["alpha"]
+    assert snapshot["visible_skills"] == []
+    assert "## Skills System" not in request.runtime.context.system_prompt
 
 
 @pytest.mark.asyncio
-async def test_injects_skills_in_input_order_with_dedup_and_invalid_slug_skipped(monkeypatch: pytest.MonkeyPatch):
+async def test_awrap_tool_call_activates_skill_when_visible_in_context():
-    async def fake_resolve(_selected):
-        return _build_snapshot(
-            ["beta", "missing", "alpha", "beta"],
-            {
-                "beta": {"name": "beta", "description": "beta skill", "path": "/skills/beta/SKILL.md"},
-                "alpha": {"name": "alpha", "description": "alpha skill", "path": "/skills/alpha/SKILL.md"},
-            },
-        )
-
-    monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
-    middleware = _build_middleware()
-    request = _build_request(skills=["beta", "missing", "alpha", "beta"], tools=["read_file"])
-
-    result = await middleware.awrap_model_call(request, _echo_handler)
-    prompt = _extract_appended_prompt(result)
-
-    beta_line = "- **beta**: beta skill"
-    alpha_line = "- **alpha**: alpha skill"
-    assert beta_line in prompt
-    assert alpha_line in prompt
-    assert prompt.find(beta_line) < prompt.find(alpha_line)
-    assert prompt.count(beta_line) == 1
-    assert "missing" not in prompt
-
-
-@pytest.mark.asyncio
-async def test_awrap_tool_call_activates_skill_when_read_skill_md():
     middleware = _build_middleware()
     request = _build_tool_request(
-        skills=["research-report"],
+        skills=["deliver-prd"],
-        visible_skills=["research-report"],
+        visible_skills=["deliver-prd", "brainstorming"],
-        file_path="/skills/research-report/SKILL.md",
+        file_path="/skills/brainstorming/SKILL.md",
     )
 
     async def _handler(_request):
@@ -207,50 +221,16 @@ async def test_awrap_tool_call_activates_skill_when_read_skill_md():
 
     result = await middleware.awrap_tool_call(request, _handler)
     assert isinstance(result, Command)
-    assert result.update["activated_skills"] == ["research-report"]
+    assert result.update["activated_skills"] == ["brainstorming"]
-    assert len(result.update["messages"]) == 1
-
-
-@pytest.mark.asyncio
-async def test_awrap_tool_call_skips_invalid_skill_slug_path():
-    middleware = _build_middleware()
-    request = _build_tool_request(
-        skills=["research-report"],
-        visible_skills=["research-report"],
-        file_path="/skills/../SKILL.md",
-    )
-
-    async def _handler(_request):
-        return ToolMessage(content="ok", tool_call_id="tc-1")
-
-    result = await middleware.awrap_tool_call(request, _handler)
-    assert isinstance(result, ToolMessage)
-
-
-@pytest.mark.asyncio
-async def test_awrap_tool_call_merges_with_existing_command_update():
-    middleware = _build_middleware()
-    request = _build_tool_request(
-        skills=["research-report"],
-        visible_skills=["research-report"],
-        file_path="/skills/research-report/SKILL.md",
-    )
-
-    async def _handler(_request):
-        return Command(update={"messages": [ToolMessage(content="ok", tool_call_id="tc-1")], "activated_skills": ["a"]})
-
-    result = await middleware.awrap_tool_call(request, _handler)
-    assert isinstance(result, Command)
-    assert result.update["activated_skills"] == ["a", "research-report"]
 
 
 @pytest.mark.asyncio
 async def test_awrap_tool_call_denies_invisible_skill():
     middleware = _build_middleware()
     request = _build_tool_request(
-        skills=["research-report"],
+        skills=["deliver-prd"],
-        visible_skills=["alpha"],
+        visible_skills=["deliver-prd"],
-        file_path="/skills/research-report/SKILL.md",
+        file_path="/skills/brainstorming/SKILL.md",
     )
 
     async def _handler(_request):
@@ -268,12 +248,20 @@ async def test_model_call_injects_dependency_tools_and_mcps_after_activation(mon
         lambda: [_FakeTool(name="calculator"), _FakeTool(name="dep-tool")],
     )
     monkeypatch.setattr(runtime_middleware, "get_kb_based_tools", lambda db_names=None: [])
-    monkeypatch.setattr(
+
-        runtime_middleware,
+    snapshot = _build_snapshot(
-        "build_dependency_bundle",
+        selected=["alpha"],
-        lambda _snapshot, activated: {"tools": ["dep-tool"], "mcps": ["mcp-a"], "skills": activated},
+        visible=["alpha", "beta"],
+        dependency_map={"alpha": {"tools": ["dep-tool"], "mcps": ["mcp-a"], "skills": ["beta"]}},
     )
 
+    def fake_build_dependency_bundle(received_snapshot, activated):
+        assert received_snapshot == snapshot
+        assert activated == ["alpha"]
+        return {"tools": ["dep-tool"], "mcps": ["mcp-a"], "skills": ["alpha", "beta"]}
+
+    monkeypatch.setattr(runtime_middleware, "build_dependency_bundle", fake_build_dependency_bundle)
+
     async def fake_get_enabled_mcp_tools(server_name: str):
         if server_name == "mcp-a":
             return [_FakeTool(name="mcp_tool")]
@@ -290,9 +278,10 @@ async def test_model_call_injects_dependency_tools_and_mcps_after_activation(mon
     )
 
     request = _build_request(
-        skills=[],
+        skills=["alpha"],
         tools=["calculator", "dep-tool", "mcp_tool", "read_file"],
-        state={"activated_skills": ["alpha"], "skill_session_snapshot": _build_snapshot([])},
+        state={"activated_skills": ["alpha"]},
+        skill_snapshot=snapshot,
     )
 
     result = await middleware.awrap_model_call(request, _echo_handler)
@@ -300,53 +289,3 @@ async def test_model_call_injects_dependency_tools_and_mcps_after_activation(mon
     assert "dep-tool" in tool_names
     assert "mcp_tool" in tool_names
     assert "calculator" not in tool_names
-
-
-@pytest.mark.asyncio
-async def test_model_call_reuses_snapshot_until_skills_changed(monkeypatch: pytest.MonkeyPatch):
-    called = {"count": 0}
-
-    async def fake_resolve(selected):
-        called["count"] += 1
-        return _build_snapshot(selected)
-
-    monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
-
-    middleware = _build_middleware()
-    req1 = _build_request(skills=["alpha"], tools=["read_file"], state={})
-    res1 = await middleware.awrap_model_call(req1, _echo_handler)
-    assert called["count"] == 1
-
-    req2 = _build_request(skills=["alpha"], tools=["read_file"], state=res1.state)
-    await middleware.awrap_model_call(req2, _echo_handler)
-    assert called["count"] == 1
-
-    req3 = _build_request(skills=["beta"], tools=["read_file"], state=res1.state)
-    await middleware.awrap_model_call(req3, _echo_handler)
-    assert called["count"] == 2
-
-
-@pytest.mark.asyncio
-async def test_injects_dependency_skills_into_prompt(monkeypatch: pytest.MonkeyPatch):
-    async def fake_resolve(_selected):
-        return {
-            "selected_skills": ["alpha"],
-            "visible_skills": ["alpha", "beta"],
-            "prompt_metadata": {
-                "alpha": {"name": "alpha", "description": "alpha desc", "path": "/skills/alpha/SKILL.md"},
-                "beta": {"name": "beta", "description": "beta desc", "path": "/skills/beta/SKILL.md"},
-            },
-            "dependency_map": {
-                "alpha": {"tools": [], "mcps": [], "skills": ["beta"]},
-                "beta": {"tools": [], "mcps": [], "skills": []},
-            },
-        }
-
-    monkeypatch.setattr(runtime_middleware, "resolve_session_snapshot", fake_resolve)
-    middleware = _build_middleware()
-    request = _build_request(skills=["alpha"], tools=["read_file"])
-
-    result = await middleware.awrap_model_call(request, _echo_handler)
-    prompt = _extract_appended_prompt(result)
-    assert "- **alpha**: alpha desc" in prompt
-    assert "- **beta**: beta desc" in prompt

test/test_skills_backend.py

@@ -51,15 +51,16 @@ def test_composite_backend_mounts_skills_under_prefix(tmp_path, monkeypatch):
     monkeypatch.setattr(skills_backend, "get_skills_root_dir", lambda: tmp_path)
 
     runtime = SimpleNamespace(
-        context=SimpleNamespace(skills=["alpha"]),
+        context=SimpleNamespace(
-        state={
+            skills=["alpha"],
-            "skill_session_snapshot": {
+            skill_session_snapshot={
                 "selected_skills": ["alpha"],
                 "visible_skills": ["alpha", "beta"],
                 "prompt_metadata": {},
                 "dependency_map": {},
-            }
+            },
-        },
+        ),
+        state={},
     )
     composite = skills_backend.create_agent_composite_backend(runtime)
 
@@ -87,3 +88,25 @@ def test_composite_backend_fallbacks_to_context_skills_when_snapshot_missing(tmp
     skills_root = composite.ls_info("/skills/")
     skill_paths = sorted(entry.get("path") for entry in skills_root)
     assert skill_paths == ["/skills/alpha/"]
+
+
+def test_composite_backend_reads_visible_skills_from_runtime_context(tmp_path, monkeypatch):
+    _prepare_skills_dir(tmp_path)
+    monkeypatch.setattr(skills_backend, "get_skills_root_dir", lambda: tmp_path)
+
+    runtime = SimpleNamespace(
+        context=SimpleNamespace(
+            skills=["alpha"],
+            skill_session_snapshot={
+                "selected_skills": ["alpha"],
+                "visible_skills": ["alpha", "beta"],
+                "prompt_metadata": {},
+                "dependency_map": {},
+            },
+        ),
+        state=None,
+    )
+    composite = skills_backend.create_agent_composite_backend(runtime)
+    skills_root = composite.ls_info("/skills/")
+    skill_paths = sorted(entry.get("path") for entry in skills_root)
+    assert skill_paths == ["/skills/alpha/", "/skills/beta/"]