feat(skills): unified read-merge + authoring/override MCP tools

Adds the merged read surface (api/skill_catalog_view.py) over synced +
local stores with origin discrimination and override precedence, and
extends the skill MCP tools with create_skill/update_skill/delete_skill
that dispatch by origin (edit local skills; create/update/remove local
overrides for cloud skills). Wired into api.mcp.create_mcp_server.
Full non-integration suite green (564 passed).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

api/mcp.py

@@ -97,6 +97,7 @@ def create_mcp_server(
     manager: DeviceManager | None = None,
     skill_catalog_store: Any | None = None,
     skill_active_subscriptions: set[str] | None = None,
+    skill_local_store: Any | None = None,
 ) -> Any:
     try:
         from mcp.server.fastmcp import FastMCP
@@ -166,10 +167,13 @@ def create_mcp_server(
 
     if skill_catalog_store is not None:
         from api.skill_catalog_mcp import register_skill_catalog_tools
+        from storage.local_skills import LocalSkillStore
 
+        local_store = skill_local_store or LocalSkillStore()
         register_skill_catalog_tools(
             server,
             store=skill_catalog_store,
+            local_store=local_store,
             get_active_subscriptions=lambda: set(skill_active_subscriptions or set()),
             get_registered_tools=lambda: set(handlers.keys()),
         )

api/skill_catalog_mcp.py

@@ -1,11 +1,16 @@
 """MCP tool surface for the Skill Catalog.
 
 API layer per CONSTITUTION.md: MCP dependencies (FastMCP) live here.
-Reads from :mod:`storage.skill_catalog`; flow-template parameter resolution
-delegates to the existing :func:`workflow.skill_exec.resolve_skill_steps`.
+Reads go through the unified merge surface in :mod:`api.skill_catalog_view`
+(synced + local, with override precedence); flow-template parameter
+resolution delegates to the existing :func:`workflow.skill_exec.resolve_skill_steps`.
 No server-side execution primitive is exposed — flow templates are resolved
 here but executed step-by-step by the LLM via the existing device-capability
 tools (design D5).
+
+Authoring tools (``create_skill``/``update_skill``/``delete_skill``) dispatch by
+origin (design D10): they edit/delete local skills and create/update/remove
+local overrides for cloud skills, never writing to the synced store.
 """
 from __future__ import annotations
 
@@ -13,10 +18,13 @@ from collections.abc import Callable
 from typing import Any
 
 from skills_learning.models import (
+    FlowStep,
     FlowTemplateSkill,
     KnowledgeSkill,
     SkillMetadata,
 )
+from api.skill_catalog_view import SkillCatalogView, SkillSummary, SkillView
+from storage.local_skills import LocalSkillStore
 from storage.skill_catalog import SkillCatalogStore
 from workflow.skill_exec import SkillExecutionError, resolve_skill_steps
 
@@ -27,6 +35,9 @@ SKILL_TOOL_NAMES = (
     "search_skills",
     "get_skill",
     "resolve_flow_template",
+    "create_skill",
+    "update_skill",
+    "delete_skill",
 )
 
 
@@ -51,9 +62,15 @@ class MissingParameterError(SkillCatalogError):
     """Raised when required flow-template parameters are missing/invalid."""
 
 
+class SkillAuthoringError(SkillCatalogError):
+    """Raised when an authoring operation cannot be applied (e.g., deleting a
+    cloud skill that has no local override)."""
+
+
 def skill_tool_handlers(
     *,
     store: SkillCatalogStore,
+    local_store: LocalSkillStore,
     get_active_subscriptions: Callable[[], set[str]],
     get_registered_tools: Callable[[], set[str]] | None = None,
 ) -> dict[str, Callable[..., dict[str, Any]]]:
@@ -62,44 +79,41 @@ def skill_tool_handlers(
     Decoupled from FastMCP so handlers can be tested directly without
     standing up a server (mirrors :func:`api.mcp.tool_handlers`).
     """
+    view = SkillCatalogView(store, local_store)
+    local = local_store
     tools_getter = get_registered_tools or (lambda: set())
 
     def _list_skills() -> dict[str, Any]:
-        metas = store.list_skills(get_active_subscriptions())
-        return {
-            "ok": True,
-            "skills": [_metadata_to_summary(m) for m in metas],
-        }
+        summaries = view.list_skills(get_active_subscriptions())
+        return {"ok": True, "skills": [_summary_to_dict(s) for s in summaries]}
 
     def _search_skills(query: str) -> dict[str, Any]:
-        metas = store.search_skills(query, get_active_subscriptions())
-        return {
-            "ok": True,
-            "skills": [_metadata_to_summary(m) for m in metas],
-        }
+        summaries = view.search_skills(query, get_active_subscriptions())
+        return {"ok": True, "skills": [_summary_to_dict(s) for s in summaries]}
 
     def _get_skill(skill_id: str) -> dict[str, Any]:
-        skill = store.get_skill(
+        result = view.get_skill(
             skill_id,
             get_active_subscriptions(),
             registered_tools=tools_getter(),
         )
-        if skill is None:
+        if result is None:
             return _error_response(SkillNotFoundError(skill_id))
-        return {"ok": True, "skill": _skill_to_full_dict(skill)}
+        return {"ok": True, "skill": _view_to_dict(result)}
 
     def _resolve_flow_template(
         skill_id: str,
         params: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         params = params or {}
-        skill = store.get_skill(
+        result = view.get_skill(
             skill_id,
             get_active_subscriptions(),
             registered_tools=tools_getter(),
         )
-        if skill is None:
+        if result is None:
             return _error_response(SkillNotFoundError(skill_id))
+        skill = result.skill
         if not isinstance(skill, FlowTemplateSkill):
             return _error_response(
                 InvalidFlowTemplateError(
@@ -112,11 +126,62 @@ def skill_tool_handlers(
             return _error_response(MissingParameterError(str(exc)))
         return {"ok": True, "steps": steps}
 
+    def _create_skill(payload: dict[str, Any]) -> dict[str, Any]:
+        try:
+            skill = _build_skill(payload)
+        except ValueError as exc:
+            return _error_response(SkillAuthoringError(str(exc)))
+        stored = local.create_local(skill)
+        return {"ok": True, "skill": _skill_to_full_dict(stored), "origin": "local"}
+
+    def _update_skill(skill_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        try:
+            skill = _build_skill(payload)
+        except ValueError as exc:
+            return _error_response(SkillAuthoringError(str(exc)))
+        if view.is_local_skill(skill_id):
+            stored = local.update_local(
+                _with_id(skill, skill_id)
+            )
+            return {
+                "ok": True,
+                "skill": _skill_to_full_dict(stored),
+                "origin": "local",
+            }
+        # Cloud skill id (or anticipated one): create/update a local override.
+        stored = local.upsert_override(skill_id, skill)
+        return {
+            "ok": True,
+            "skill": _skill_to_full_dict(stored),
+            "origin": "cloud",
+            "locally_overridden": True,
+        }
+
+    def _delete_skill(skill_id: str) -> dict[str, Any]:
+        if view.is_local_skill(skill_id):
+            local.delete_local(skill_id)
+            return {"ok": True, "deleted": skill_id, "origin": "local"}
+        if view.has_override(skill_id):
+            local.remove_override(skill_id)
+            return {
+                "ok": True,
+                "deleted_override": skill_id,
+                "origin": "cloud",
+            }
+        return _error_response(
+            SkillAuthoringError(
+                f"skill {skill_id} is a cloud skill with no local override to remove"
+            )
+        )
+
     return {
         "list_skills": _list_skills,
         "search_skills": _search_skills,
         "get_skill": _get_skill,
         "resolve_flow_template": _resolve_flow_template,
+        "create_skill": _create_skill,
+        "update_skill": _update_skill,
+        "delete_skill": _delete_skill,
     }
 
 
@@ -124,18 +189,20 @@ def register_skill_catalog_tools(
     server: Any,
     *,
     store: SkillCatalogStore,
+    local_store: LocalSkillStore,
     get_active_subscriptions: Callable[[], set[str]],
     get_registered_tools: Callable[[], set[str]] | None = None,
 ) -> Any:
-    """Register ``list_skills``/``search_skills``/``get_skill``/
-    ``resolve_flow_template`` as MCP tools on ``server``.
+    """Register the skill MCP tools (read + authoring) on ``server``.
 
     Returns the server so the caller can chain. No batch-execute tool is
     registered (design D5): the LLM issues each resulting device-capability
-    tool call itself, preserving the Observe-Think-Act loop.
+    tool call itself, preserving the Observe-Think-Act loop. Authoring tools
+    are always registered (design D6): there is no enable/disable gate.
     """
     handlers = skill_tool_handlers(
         store=store,
+        local_store=local_store,
         get_active_subscriptions=get_active_subscriptions,
         get_registered_tools=get_registered_tools,
     )
@@ -159,25 +226,58 @@ def register_skill_catalog_tools(
     ) -> dict[str, Any]:
         return handlers["resolve_flow_template"](skill_id=skill_id, params=params)
 
+    @server.tool(name="create_skill")
+    def _create_skill(payload: dict[str, Any]) -> dict[str, Any]:
+        return handlers["create_skill"](payload=payload)
+
+    @server.tool(name="update_skill")
+    def _update_skill(skill_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        return handlers["update_skill"](skill_id=skill_id, payload=payload)
+
+    @server.tool(name="delete_skill")
+    def _delete_skill(skill_id: str) -> dict[str, Any]:
+        return handlers["delete_skill"](skill_id=skill_id)
+
     return server
 
 
-def _metadata_to_summary(meta: SkillMetadata) -> dict[str, Any]:
-    """Compact metadata for list/search — no Subscription-Platform-specific fields."""
+# ----------------------------------------------------------------------
+# Serialization helpers
+# ----------------------------------------------------------------------
+
+
+def _summary_to_dict(summary: SkillSummary) -> dict[str, Any]:
+    meta = summary.metadata
     return {
         "id": meta.id,
         "name": meta.name,
         "description": meta.description,
         "kind": meta.kind,
         "tags": list(meta.tags),
+        "origin": summary.origin,
+        "locally_overridden": summary.locally_overridden,
     }
 
 
+def _view_to_dict(view: SkillView) -> dict[str, Any]:
+    base = _skill_to_full_dict(view.skill)
+    base["origin"] = view.origin
+    base["locally_overridden"] = view.locally_overridden
+    return base
+
+
 def _skill_to_full_dict(skill: Any) -> dict[str, Any]:
-    """Full skill payload for ``get_skill``."""
-    base = _metadata_to_summary(skill.metadata)
-    base["version"] = skill.metadata.version
-    base["updated_at"] = skill.metadata.updated_at.isoformat()
+    """Full skill payload for ``get_skill`` / authoring responses."""
+    meta: SkillMetadata = skill.metadata
+    base = {
+        "id": meta.id,
+        "name": meta.name,
+        "description": meta.description,
+        "kind": meta.kind,
+        "tags": list(meta.tags),
+        "version": meta.version,
+        "updated_at": meta.updated_at.isoformat(),
+    }
     if isinstance(skill, KnowledgeSkill):
         base["content"] = skill.content
     elif isinstance(skill, FlowTemplateSkill):
@@ -188,6 +288,45 @@ def _skill_to_full_dict(skill: Any) -> dict[str, Any]:
     return base
 
 
+def _build_skill(payload: dict[str, Any]) -> KnowledgeSkill | FlowTemplateSkill:
+    """Construct a Skill from an authoring payload."""
+    kind = str(payload.get("kind") or "").strip()
+    name = str(payload.get("name") or "").strip()
+    if not name:
+        raise ValueError("skill name must not be empty")
+    if kind not in ("knowledge", "flow_template"):
+        raise ValueError(f"unsupported skill kind: {kind!r}")
+    tags = [str(tag) for tag in payload.get("tags") or []]
+    meta = SkillMetadata(name=name, kind=kind, tags=tags)  # type: ignore[arg-type]
+    if kind == "knowledge":
+        return KnowledgeSkill(metadata=meta, content=str(payload.get("content") or ""))
+    steps = [
+        FlowStep(
+            tool_name=str(step.get("tool_name") or step.get("action") or ""),
+            args=dict(step.get("args") or {}),
+        )
+        for step in (payload.get("steps") or [])
+    ]
+    parameters = {
+        str(name): dict(schema) for name, schema in (payload.get("parameters") or {}).items()
+    }
+    return FlowTemplateSkill(metadata=meta, steps=steps, parameters=parameters)
+
+
+def _with_id(skill: KnowledgeSkill | FlowTemplateSkill, skill_id: str):
+    meta = skill.metadata
+    from dataclasses import replace
+
+    new_meta = replace(meta, id=skill_id)
+    if isinstance(skill, KnowledgeSkill):
+        return KnowledgeSkill(metadata=new_meta, content=skill.content)
+    return FlowTemplateSkill(
+        metadata=new_meta,
+        steps=list(skill.steps),
+        parameters={n: dict(s) for n, s in skill.parameters.items()},
+    )
+
+
 def _error_response(exc: Exception) -> dict[str, Any]:
     return {"ok": False, "error": _semantic_skill_error(exc)}
 
@@ -200,4 +339,7 @@ def _semantic_skill_error(exc: Exception) -> str:
     if isinstance(exc, MissingParameterError):
         message = str(exc)
         return f"missing parameter: {message}" if message else "invalid parameter"
+    if isinstance(exc, SkillAuthoringError):
+        message = str(exc)
+        return f"authoring error: {message}" if message else "authoring error"
     return "operation failed"

api/skill_catalog_view.py

@@ -0,0 +1,166 @@
+"""Unified read-merge surface over the synced catalog and the local skill store.
+
+API layer per CONSTITUTION.md: this module reads from both
+:mod:`storage.skill_catalog` (cloud-synced, read-only-except-sync) and
+:mod:`storage.local_skills` (agent-authored + overrides) and presents a single
+origin-discriminated catalog. It is the only module that reads both stores.
+
+Override semantics (design D8/D9/D11): a local override shadows the cloud
+skill at read time, wins against sync updates, and is reported with
+``origin = "cloud"`` and ``locally_overridden = true`` while it shadows.
+"""
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Literal
+
+from skills_learning.models import (
+    FlowTemplateSkill,
+    KnowledgeSkill,
+    Skill,
+    SkillMetadata,
+)
+from storage.local_skills import LocalSkillStore
+from storage.skill_catalog import SkillCatalogStore
+
+Origin = Literal["cloud", "local"]
+
+
+@dataclass(frozen=True)
+class SkillSummary:
+    """Metadata + origin for list/search results."""
+
+    metadata: SkillMetadata
+    origin: Origin
+    locally_overridden: bool
+
+
+@dataclass(frozen=True)
+class SkillView:
+    """Full skill + origin for get_skill results."""
+
+    skill: KnowledgeSkill | FlowTemplateSkill
+    origin: Origin
+    locally_overridden: bool
+
+
+class SkillCatalogView:
+    """Merged read surface over synced + local skills, with override precedence."""
+
+    def __init__(self, synced: SkillCatalogStore, local: LocalSkillStore) -> None:
+        self._synced = synced
+        self._local = local
+
+    # ------------------------------------------------------------------
+    # Merged reads
+    # ------------------------------------------------------------------
+
+    def list_skills(
+        self,
+        active_subscriptions: set[str],
+    ) -> list[SkillSummary]:
+        override_ids = self._local.list_override_cloud_ids()
+        summaries: list[SkillSummary] = []
+
+        # Local authored skills.
+        for meta in self._local.list_local():
+            summaries.append(SkillSummary(meta, "local", False))
+
+        # Cloud-synced skills, applying overrides where present.
+        for meta in self._synced.list_skills(active_subscriptions):
+            if meta.id in override_ids:
+                override = self._local.get_override(meta.id)
+                if override is not None:
+                    summaries.append(
+                        SkillSummary(override.metadata, "cloud", True)
+                    )
+                    continue
+            summaries.append(SkillSummary(meta, "cloud", False))
+
+        summaries.sort(key=lambda s: s.metadata.name)
+        return summaries
+
+    def search_skills(
+        self,
+        query: str,
+        active_subscriptions: set[str],
+    ) -> list[SkillSummary]:
+        override_ids = self._local.list_override_cloud_ids()
+        summaries: list[SkillSummary] = []
+
+        for meta in self._local.search_local(query):
+            summaries.append(SkillSummary(meta, "local", False))
+
+        for meta in self._synced.search_skills(query, active_subscriptions):
+            if meta.id in override_ids:
+                override = self._local.get_override(meta.id)
+                if override is not None:
+                    summaries.append(
+                        SkillSummary(override.metadata, "cloud", True)
+                    )
+                    continue
+            summaries.append(SkillSummary(meta, "cloud", False))
+
+        return summaries
+
+    def get_skill(
+        self,
+        skill_id: str,
+        active_subscriptions: set[str],
+        *,
+        registered_tools: set[str] | None = None,
+    ) -> SkillView | None:
+        # 1. Local authored skill.
+        local_skill = self._local.get_local(skill_id)
+        if local_skill is not None:
+            return SkillView(local_skill, "local", False)
+
+        # 2. Override shadowing a cloud skill id.
+        override = self._local.get_override(skill_id)
+        if override is not None:
+            if (
+                isinstance(override, FlowTemplateSkill)
+                and registered_tools is not None
+                and not _tools_valid(override, registered_tools)
+            ):
+                return None
+            return SkillView(override, "cloud", True)
+
+        # 3. Cloud-synced skill (None for unknown AND not-visible — no leak).
+        cloud_skill = self._synced.get_skill(
+            skill_id,
+            active_subscriptions,
+            registered_tools=registered_tools,
+        )
+        if cloud_skill is not None:
+            return SkillView(cloud_skill, "cloud", False)
+        return None
+
+    # ------------------------------------------------------------------
+    # Origin classification for authoring dispatch (D10)
+    # ------------------------------------------------------------------
+
+    def is_local_skill(self, skill_id: str) -> bool:
+        return self._local.get_local(skill_id) is not None
+
+    def has_override(self, cloud_skill_id: str) -> bool:
+        return self._local.has_override(cloud_skill_id)
+
+
+def _tools_valid(
+    skill: FlowTemplateSkill, registered_tools: set[str]
+) -> bool:
+    return all(step.tool_name in registered_tools for step in skill.steps)
+
+
+def make_view_from_stores(
+    synced: SkillCatalogStore,
+    local: LocalSkillStore,
+) -> SkillCatalogView:
+    return SkillCatalogView(synced, local)
+
+
+def default_local_store() -> LocalSkillStore:
+    """Lazy default local store (created on first use)."""
+    return LocalSkillStore()