feat: add source registry mutations

docs/current_roadmap.md

@@ -127,19 +127,24 @@ implementation state.
       and config entries shadow same-id registry entries with an event.
    - Next source registry slices are planned in
       [2026-06-03 source registry next slices](./superpowers/plans/2026-06-03-source-registry-next-slices.md):
-      desired-registry admin reads and safe mutation commands remain later
+      desired-registry admin reads and safe mutation commands are complete;
-      slices.
+      concrete MCP-backed `WorkflowServer` construction remains future work.
    - Completed: desired-registry admin read plumbing is available through
       `WorkflowSourceRegistryApi`, JSON-RPC methods
       (`workflow.admin.source_registry.list` / `.inspect`), and CLI commands
       (`wf admin registry list` / `wf admin registry inspect`). Local/static
       servers report unavailable instead of empty; a concrete MCP-backed
-      `WorkflowServer` construction path remains future work. No mutations added.
+      `WorkflowServer` construction path remains future work.
-   - Next executable source registry slice:
+   - Completed: desired-registry mutation operations are available through
-      [2026-06-04 source registry mutations](./superpowers/plans/2026-06-04-source-registry-mutations.md).
+      `WorkflowSourceRegistryApi` (add/update/enable/disable/remove),
-      It adds add/update/enable/disable/remove operations against desired
+      JSON-RPC methods (`workflow.admin.source_registry.add` / `.update` /
-      registry state only; config, auth records, and catalog snapshots are not
+      `.enable` / `.disable` / `.remove`), and CLI commands
-      mutated in v1.
+      (`wf admin registry add` / `update` / `enable` / `disable` / `remove`)
+      for targets that expose the registry-admin surface. Mutations target
+      persisted desired registry state only; config files, auth records, and
+      catalog snapshots are not mutated. Config-shadowed add is rejected in v1.
+      Remove requires `--confirm` in CLI; local/static servers report
+      unavailable.
    - Longer term: make the MCP frontend an adapter over these neutral workflow,
       source-admin, and config-admin surfaces so the old `wf_mcp` server entry
       point can shrink or retire.

docs/superpowers/plans/2026-06-03-source-registry-next-slices.md

@@ -68,10 +68,13 @@ The next executable slice is startup merge:
    - Concrete MCP-backed `WorkflowServer` construction remains future work.
 
 5. **Slice 5: Mutation Commands**
-   - **Status: planned.**
+   - **Status: complete.**
    - Add add/update/enable/disable/remove operations.
    - Use registry store, validation, and atomic writes.
    - Keep auth/catalog cleanup deferred.
+   - JSON-RPC/CLI calls work for targets that expose the registry-admin surface;
+     local/static servers report unavailable and concrete MCP-backed
+     `WorkflowServer` construction remains future work.
 
 ---
 
@@ -368,8 +371,8 @@ Pick one naming shape in the implementation plan.
 
 Add safe registry mutation.
 
-Implementation plan:
+Status: complete. Implementation:
-`docs/superpowers/plans/2026-06-04-source-registry-mutations.md`.
+[2026-06-04 source registry mutations](../plans/2026-06-04-source-registry-mutations.md).
 
 ### Operations
 
@@ -392,7 +395,8 @@ Implementation plan:
 
 - RPC methods exist.
 - CLI commands exist.
-- Mutations persist across process restart.
+- Mutations persist across process restart for targets backed by a registry
+  store; local/static servers report unavailable.
 - Validation errors are actionable.
 
 ---

docs/superpowers/specs/2026-06-03-store-backed-source-registry-design.md

@@ -280,14 +280,13 @@ MCP-backed `WorkflowServer` construction remains future work.
 
 ### Slice 5: Mutating RPC/CLI
 
-Implementation plan:
+Status: complete. Add/update/enable/disable/remove operations are available
-[2026-06-04 source registry mutations](../plans/2026-06-04-source-registry-mutations.md).
+through `WorkflowSourceRegistryApi`, JSON-RPC methods, and CLI commands.
-
+Mutations target persisted desired registry state only; config files, auth
-- Add add/update/enable/disable/remove operations.
+records, and catalog snapshots are not mutated. Config-shadowed add is rejected
-- Add JSON-RPC methods.
+in v1. Remove requires `--confirm` in CLI. Local/static servers report
-- Add CLI commands.
+unavailable for mutation commands. Concrete MCP-backed `WorkflowServer`
-- Validate before commit.
+construction remains future work.
-- Do not delete auth/catalog on remove in v1.
 
 ## Open Questions
 

src/wf_api/__init__.py

@@ -26,6 +26,7 @@ from .service import WorkflowApi
 from .source_admin import WorkflowSourceAdminApi
 from .source_registry_admin import (
     WorkflowSourceRegistryApi,
+    WorkflowSourceRegistryMutationProvider,
     WorkflowSourceRegistryProvider,
 )
 from .surface import (
@@ -107,6 +108,7 @@ __all__ = [
     "WorkflowSourceAdminApi",
     "WorkflowSourceAdminSurface",
     "WorkflowSourceRegistryApi",
+    "WorkflowSourceRegistryMutationProvider",
     "WorkflowSourceRegistryProvider",
     "WorkflowSourceRegistrySurface",
     "WorkflowSpecProvider",

src/wf_api/source_registry_admin.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Mapping, Sequence, Set
 from dataclasses import asdict, is_dataclass
-from typing import Any, Protocol
+from typing import Any, Protocol, runtime_checkable
 
 from wf_platform import page_items
 
@@ -15,8 +15,22 @@ class WorkflowSourceRegistryProvider(Protocol):
     def config_source_ids(self) -> Set[str]: ...
 
 
+@runtime_checkable
+class WorkflowSourceRegistryMutationProvider(Protocol):
+    """Write capabilities for source registry mutation operations."""
+
+    def add_registry_entry(self, entry: Mapping[str, Any]) -> Mapping[str, Any] | object: ...
+    def update_registry_entry(
+        self, source_id: str, patch: Mapping[str, Any]
+    ) -> Mapping[str, Any] | object: ...
+    def set_registry_entry_enabled(
+        self, source_id: str, enabled: bool
+    ) -> Mapping[str, Any] | object: ...
+    def remove_registry_entry(self, source_id: str) -> Mapping[str, Any] | object: ...
+
+
 class WorkflowSourceRegistryApi:
-    """Protocol-neutral read-only desired source registry operations.
+    """Protocol-neutral desired source registry operations.
 
     This surface is intentionally separate from WorkflowSourceAdminApi.
     WorkflowSourceAdminApi exposes observed/hydrated runtime source inventory.
@@ -28,8 +42,13 @@ class WorkflowSourceRegistryApi:
         self,
         *,
         provider: WorkflowSourceRegistryProvider,
+        mutation_provider: WorkflowSourceRegistryMutationProvider | None = None,
     ) -> None:
         self._provider = provider
+        self._mutation_provider = mutation_provider
+
+    def _is_shadowed(self, source_id: str) -> bool:
+        return source_id in self._provider.config_source_ids()
 
     async def list_registry_entries(
         self,
@@ -37,10 +56,9 @@ class WorkflowSourceRegistryApi:
         cursor: str | None = None,
         limit: int = 50,
     ) -> dict[str, Any]:
-        shadowed_ids = set(self._provider.config_source_ids())
         entries = sorted(
             (
-                _entry_summary(_payload(item), shadowed_ids)
+                _entry_summary(_payload(item), self._provider.config_source_ids())
                 for item in self._provider.list_registry_entries()
             ),
             key=lambda item: str(item.get("id", "")),
@@ -57,16 +75,86 @@ class WorkflowSourceRegistryApi:
         *,
         source_id: str,
     ) -> dict[str, Any]:
-        shadowed_ids = set(self._provider.config_source_ids())
         for item in self._provider.list_registry_entries():
             entry = _payload(item)
             if entry.get("id") == source_id:
                 return {
                     "entry": entry,
-                    "shadowed_by_config": source_id in shadowed_ids,
+                    "shadowed_by_config": self._is_shadowed(source_id),
                 }
         raise KeyError(f"unknown registry source {source_id!r}")
 
+    async def add_registry_entry(
+        self,
+        *,
+        entry: dict[str, Any],
+    ) -> dict[str, Any]:
+        if self._mutation_provider is None:
+            raise TypeError("add_registry_entry requires a mutation provider")
+        raw = self._mutation_provider.add_registry_entry(entry)
+        result = _payload(raw)
+        return {
+            "entry": result,
+            "shadowed_by_config": self._is_shadowed(result["id"]),
+        }
+
+    async def update_registry_entry(
+        self,
+        *,
+        source_id: str,
+        patch: dict[str, Any],
+    ) -> dict[str, Any]:
+        if self._mutation_provider is None:
+            raise TypeError("update_registry_entry requires a mutation provider")
+        raw = self._mutation_provider.update_registry_entry(source_id, patch)
+        result = _payload(raw)
+        return {
+            "entry": result,
+            "shadowed_by_config": self._is_shadowed(result["id"]),
+        }
+
+    async def enable_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]:
+        if self._mutation_provider is None:
+            raise TypeError("enable_registry_entry requires a mutation provider")
+        raw = self._mutation_provider.set_registry_entry_enabled(source_id, True)
+        result = _payload(raw)
+        return {
+            "entry": result,
+            "shadowed_by_config": self._is_shadowed(result["id"]),
+        }
+
+    async def disable_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]:
+        if self._mutation_provider is None:
+            raise TypeError("disable_registry_entry requires a mutation provider")
+        raw = self._mutation_provider.set_registry_entry_enabled(source_id, False)
+        result = _payload(raw)
+        return {
+            "entry": result,
+            "shadowed_by_config": self._is_shadowed(result["id"]),
+        }
+
+    async def remove_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]:
+        if self._mutation_provider is None:
+            raise TypeError("remove_registry_entry requires a mutation provider")
+        raw = self._mutation_provider.remove_registry_entry(source_id)
+        result = _payload(raw)
+        return {
+            "removed": bool(result.get("removed")),
+            "source_id": str(result.get("source_id", source_id)),
+        }
+
 
 def _payload(value: Mapping[str, Any] | object) -> dict[str, Any]:
     """Normalize provider objects without depending on MCP registry types."""
@@ -84,7 +172,7 @@ def _payload(value: Mapping[str, Any] | object) -> dict[str, Any]:
     )
 
 
-def _entry_summary(entry: dict[str, Any], shadowed_ids: set[str]) -> dict[str, Any]:
+def _entry_summary(entry: dict[str, Any], shadowed_ids: Set[str]) -> dict[str, Any]:
     transport = entry.get("transport")
     transport_kind = transport.get("kind") if isinstance(transport, Mapping) else None
     return {

src/wf_api/surface.py

@@ -227,7 +227,7 @@ class WorkflowAdminSurface(Protocol):
 
 
 class WorkflowSourceRegistrySurface(Protocol):
-    """Read-only desired source registry methods exposed by platform frontends."""
+    """Desired source registry methods exposed by platform frontends."""
 
     async def list_registry_entries(
         self,
@@ -242,6 +242,37 @@ class WorkflowSourceRegistrySurface(Protocol):
         source_id: str,
     ) -> dict[str, Any]: ...
 
+    async def add_registry_entry(
+        self,
+        *,
+        entry: dict[str, Any],
+    ) -> dict[str, Any]: ...
+
+    async def update_registry_entry(
+        self,
+        *,
+        source_id: str,
+        patch: dict[str, Any],
+    ) -> dict[str, Any]: ...
+
+    async def enable_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]: ...
+
+    async def disable_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]: ...
+
+    async def remove_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]: ...
+
 
 __all__ = [
     "WorkflowAdminSurface",

src/wf_cli/commands/source_registry.py

@@ -1,21 +1,32 @@
 from __future__ import annotations
 
 import asyncio
-from typing import Annotated
+import json
+from pathlib import Path
+from typing import Annotated, Any
 
 import typer
 
-from wf_cli.context import load_cli_context_from_typer
+from wf_cli.context import CliContext, load_cli_context_from_typer
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import emit_json
 
 app = typer.Typer(
     name="registry",
-    help="List and inspect desired persisted source registry entries.",
+    help="List, inspect, and mutate desired persisted source registry entries.",
     no_args_is_help=True,
 )
 
 
+def _require_registry_admin(context: CliContext) -> Any:
+    """Return the source registry admin surface or raise if unavailable."""
+    if context.source_registry_admin is None:
+        raise typer.BadParameter(
+            "source registry admin operations are not available for this server"
+        )
+    return context.source_registry_admin
+
+
 @app.command("list")
 def list_registry_entries(
     ctx: typer.Context,
@@ -31,13 +42,8 @@ def list_registry_entries(
 ) -> None:
     """List desired persisted source registry entries."""
     context = load_cli_context_from_typer(ctx)
-    if context.source_registry_admin is None:
+    admin = _require_registry_admin(context)
-        raise typer.BadParameter(
+    payload = asyncio.run(admin.list_registry_entries(cursor=cursor, limit=limit))
-            "source registry admin reads are not available for this server"
-        )
-    payload = asyncio.run(
-        context.source_registry_admin.list_registry_entries(cursor=cursor, limit=limit)
-    )
     emit_list_payload(
         payload,
         collection_key="entries",
@@ -54,11 +60,117 @@ def inspect_registry_entry(
 ) -> None:
     """Inspect one desired persisted source registry entry."""
     context = load_cli_context_from_typer(ctx)
-    if context.source_registry_admin is None:
+    admin = _require_registry_admin(context)
-        raise typer.BadParameter(
+    payload = asyncio.run(admin.inspect_registry_entry(source_id=source_id))
-            "source registry admin reads are not available for this server"
-        )
-    payload = asyncio.run(
-        context.source_registry_admin.inspect_registry_entry(source_id=source_id)
-    )
     emit_json(payload)
+
+
+@app.command("add")
+def add_registry_entry(
+    ctx: typer.Context,
+    input_json: Annotated[
+        str | None, typer.Option("--input", help="JSON payload for the new entry.")
+    ] = None,
+    input_file: Annotated[
+        str | None,
+        typer.Option("--input-file", help="Path to JSON file for the new entry."),
+    ] = None,
+) -> None:
+    """Add a new desired source registry entry."""
+    context = load_cli_context_from_typer(ctx)
+    admin = _require_registry_admin(context)
+    entry = _read_json_arg(input_json, input_file, "--input/--input-file")
+    payload = asyncio.run(admin.add_registry_entry(entry=entry))
+    emit_json(payload)
+
+
+@app.command("update")
+def update_registry_entry(
+    ctx: typer.Context,
+    source_id: Annotated[str, typer.Argument(help="Source ID to update.")],
+    patch_json: Annotated[
+        str | None, typer.Option("--patch", help="JSON patch to apply.")
+    ] = None,
+    patch_file: Annotated[
+        str | None, typer.Option("--patch-file", help="Path to JSON patch file.")
+    ] = None,
+) -> None:
+    """Update an existing desired source registry entry."""
+    context = load_cli_context_from_typer(ctx)
+    admin = _require_registry_admin(context)
+    patch = _read_json_arg(patch_json, patch_file, "--patch/--patch-file")
+    payload = asyncio.run(admin.update_registry_entry(source_id=source_id, patch=patch))
+    emit_json(payload)
+
+
+@app.command("enable")
+def enable_registry_entry(
+    ctx: typer.Context,
+    source_id: Annotated[str, typer.Argument(help="Source ID to enable.")],
+) -> None:
+    """Enable a desired source registry entry."""
+    context = load_cli_context_from_typer(ctx)
+    admin = _require_registry_admin(context)
+    payload = asyncio.run(admin.enable_registry_entry(source_id=source_id))
+    emit_json(payload)
+
+
+@app.command("disable")
+def disable_registry_entry(
+    ctx: typer.Context,
+    source_id: Annotated[str, typer.Argument(help="Source ID to disable.")],
+) -> None:
+    """Disable a desired source registry entry."""
+    context = load_cli_context_from_typer(ctx)
+    admin = _require_registry_admin(context)
+    payload = asyncio.run(admin.disable_registry_entry(source_id=source_id))
+    emit_json(payload)
+
+
+@app.command("remove")
+def remove_registry_entry(
+    ctx: typer.Context,
+    source_id: Annotated[str, typer.Argument(help="Source ID to remove.")],
+    confirm: Annotated[
+        bool, typer.Option("--confirm", help="Confirm removal.")
+    ] = False,
+) -> None:
+    """Remove a desired source registry entry (requires --confirm)."""
+    context = load_cli_context_from_typer(ctx)
+    admin = _require_registry_admin(context)
+    if not confirm:
+        raise typer.BadParameter("removal requires --confirm flag")
+    payload = asyncio.run(admin.remove_registry_entry(source_id=source_id))
+    emit_json(payload)
+
+
+def _read_json_arg(
+    inline: str | None,
+    file_path: str | None,
+    flag_names: str,
+) -> dict[str, Any]:
+    """Resolve JSON from either inline text or a file path."""
+    if inline and file_path:
+        raise typer.BadParameter(f"provide exactly one of {flag_names}")
+    if inline:
+        try:
+            value = json.loads(inline)
+        except json.JSONDecodeError as exc:
+            raise typer.BadParameter(f"invalid JSON: {exc}") from exc
+        return _require_json_object(value, flag_names)
+    if file_path:
+        try:
+            value = json.loads(Path(file_path).read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            raise typer.BadParameter(f"file not found: {file_path}")
+        except json.JSONDecodeError as exc:
+            raise typer.BadParameter(f"invalid JSON in file: {exc}") from exc
+        return _require_json_object(value, flag_names)
+    raise typer.BadParameter(f"{flag_names} is required")
+
+
+def _require_json_object(value: Any, flag_names: str) -> dict[str, Any]:
+    """Registry add/update payloads must be JSON objects, not arrays/scalars."""
+    if not isinstance(value, dict):
+        raise typer.BadParameter(f"{flag_names} must be a JSON object")
+    return value

src/wf_mcp/broker/service/source_registry_admin.py

@@ -1,21 +1,96 @@
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
+from typing import Any
+
+from wf_api.source_registry_admin import WorkflowSourceRegistryMutationProvider
 
 from ...models import ConnectionConfig
-from ...source_registry import SourceRegistryStore
+from ...source_registry import McpSourceRegistryEntry, SourceRegistryFile, SourceRegistryStore
 
 
 @dataclass(slots=True)
-class SourceRegistryAdminProvider:
+class SourceRegistryAdminProvider(WorkflowSourceRegistryMutationProvider):
-    """Read desired MCP source registry state without mutating it."""
+    """Read/write desired MCP source registry state.
+
+    Implements ``WorkflowSourceRegistryMutationProvider`` so the API layer
+    can delegate mutation operations here.
+    """
 
     source_registry_store: SourceRegistryStore
     config_connections: Sequence[ConnectionConfig] = field(default_factory=tuple)
 
-    def list_registry_entries(self) -> list[object]:
+    # -- read helpers -------------------------------------------------------
+
+    def list_registry_entries(self) -> list[McpSourceRegistryEntry]:
         return list(self.source_registry_store.load_registry().sources)
 
     def config_source_ids(self) -> set[str]:
         return {connection.id for connection in self.config_connections}
+
+    # -- private helpers ----------------------------------------------------
+
+    def _load(self) -> SourceRegistryFile:
+        return self.source_registry_store.load_registry()
+
+    def _save(self, sources: list[McpSourceRegistryEntry]) -> None:
+        registry = SourceRegistryFile(sources=sources)
+        self.source_registry_store.save_registry(registry)
+
+    def _entry_map(self, registry: SourceRegistryFile) -> dict[str, McpSourceRegistryEntry]:
+        return registry.source_map()
+
+    def _require_entry(self, source_id: str) -> McpSourceRegistryEntry:
+        registry = self._load()
+        entry = self._entry_map(registry).get(source_id)
+        if entry is None:
+            raise KeyError(f"unknown registry source {source_id!r}")
+        return entry
+
+    # -- mutation methods ---------------------------------------------------
+
+    def add_registry_entry(self, entry: Mapping[str, Any]) -> McpSourceRegistryEntry:
+        source_id = str(entry["id"])
+        if source_id in self.config_source_ids():
+            raise ValueError(
+                f"cannot add {source_id!r}: id is shadowed by a config connection"
+            )
+        validated = McpSourceRegistryEntry.model_validate(dict(entry))
+        registry = self._load()
+        if validated.id in self._entry_map(registry):
+            raise ValueError(f"duplicate registry source id {validated.id!r}")
+        self._save([*registry.sources, validated])
+        return validated
+
+    def update_registry_entry(
+        self, source_id: str, patch: Mapping[str, Any]
+    ) -> McpSourceRegistryEntry:
+        existing = self._require_entry(source_id)
+        merged = existing.model_dump(mode="json")
+        merged.update(dict(patch))
+        # v1: forbid renaming unless the new id matches source_id
+        if merged["id"] != source_id:
+            raise ValueError(
+                f"cannot change source id from {source_id!r} to {merged['id']!r}"
+            )
+        updated = McpSourceRegistryEntry.model_validate(merged)
+        registry = self._load()
+        sources = [updated if s.id == source_id else s for s in registry.sources]
+        self._save(sources)
+        return updated
+
+    def set_registry_entry_enabled(self, source_id: str, enabled: bool) -> McpSourceRegistryEntry:
+        existing = self._require_entry(source_id)
+        updated = existing.model_copy(update={"enabled": enabled})
+        registry = self._load()
+        sources = [updated if s.id == source_id else s for s in registry.sources]
+        self._save(sources)
+        return updated
+
+    def remove_registry_entry(self, source_id: str) -> dict[str, Any]:
+        self._require_entry(source_id)
+        registry = self._load()
+        sources = [s for s in registry.sources if s.id != source_id]
+        self._save(sources)
+        return {"removed": True, "source_id": source_id}

src/wf_transport_rpc_http/client_source_registry.py

@@ -4,7 +4,7 @@ from typing import Any
 
 
 class RpcSourceRegistryClientMixin:
-    """JSON-RPC implementation of read-only desired source registry surface methods."""
+    """JSON-RPC implementation of source registry surface methods."""
 
     async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
 
@@ -28,3 +28,54 @@ class RpcSourceRegistryClientMixin:
             "workflow.admin.source_registry.inspect",
             {"source_id": source_id},
         )
+
+    async def add_registry_entry(
+        self,
+        *,
+        entry: dict[str, Any],
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.admin.source_registry.add",
+            {"entry": entry},
+        )
+
+    async def update_registry_entry(
+        self,
+        *,
+        source_id: str,
+        patch: dict[str, Any],
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.admin.source_registry.update",
+            {"source_id": source_id, "patch": patch},
+        )
+
+    async def enable_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.admin.source_registry.enable",
+            {"source_id": source_id},
+        )
+
+    async def disable_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.admin.source_registry.disable",
+            {"source_id": source_id},
+        )
+
+    async def remove_registry_entry(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.admin.source_registry.remove",
+            {"source_id": source_id},
+        )

src/wf_transport_rpc_http/methods_source_registry.py

@@ -9,14 +9,20 @@ from fastapi_jsonrpc import Params
 from wf_server import WorkflowServer
 
 from .errors import WorkflowRpcError, raise_workflow_rpc_error
-from .models import InspectRegistryEntryParams, ListRegistryEntriesParams
+from .models import (
+    AddRegistryEntryParams,
+    InspectRegistryEntryParams,
+    ListRegistryEntriesParams,
+    RegistryEntryIdParams,
+    UpdateRegistryEntryParams,
+)
 
 
 def register_methods(
     entrypoint: jsonrpc.Entrypoint,
     server: WorkflowServer,
 ) -> None:
-    """Register read-only desired source registry JSON-RPC methods."""
+    """Register source registry JSON-RPC methods."""
 
     @entrypoint.method(
         name="workflow.admin.source_registry.list",
@@ -62,3 +68,109 @@ def register_methods(
             )
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.admin.source_registry.add",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_admin_source_registry_add(
+        params: AddRegistryEntryParams = Params(...),  # type: ignore[reportArgumentType]
+    ) -> dict[str, Any]:
+        if server.source_registry_admin is None:
+            raise WorkflowRpcError(
+                data={
+                    "code": "source_registry_unavailable",
+                    "message": "source registry admin mutations are not available for this server",
+                }
+            )
+        try:
+            return await server.source_registry_admin.add_registry_entry(
+                entry=params.entry,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.admin.source_registry.update",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_admin_source_registry_update(
+        params: UpdateRegistryEntryParams = Params(...),  # type: ignore[reportArgumentType]
+    ) -> dict[str, Any]:
+        if server.source_registry_admin is None:
+            raise WorkflowRpcError(
+                data={
+                    "code": "source_registry_unavailable",
+                    "message": "source registry admin mutations are not available for this server",
+                }
+            )
+        try:
+            return await server.source_registry_admin.update_registry_entry(
+                source_id=params.source_id,
+                patch=params.patch,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.admin.source_registry.enable",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_admin_source_registry_enable(
+        params: RegistryEntryIdParams = Params(...),  # type: ignore[reportArgumentType]
+    ) -> dict[str, Any]:
+        if server.source_registry_admin is None:
+            raise WorkflowRpcError(
+                data={
+                    "code": "source_registry_unavailable",
+                    "message": "source registry admin mutations are not available for this server",
+                }
+            )
+        try:
+            return await server.source_registry_admin.enable_registry_entry(
+                source_id=params.source_id,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.admin.source_registry.disable",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_admin_source_registry_disable(
+        params: RegistryEntryIdParams = Params(...),  # type: ignore[reportArgumentType]
+    ) -> dict[str, Any]:
+        if server.source_registry_admin is None:
+            raise WorkflowRpcError(
+                data={
+                    "code": "source_registry_unavailable",
+                    "message": "source registry admin mutations are not available for this server",
+                }
+            )
+        try:
+            return await server.source_registry_admin.disable_registry_entry(
+                source_id=params.source_id,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.admin.source_registry.remove",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_admin_source_registry_remove(
+        params: RegistryEntryIdParams = Params(...),  # type: ignore[reportArgumentType]
+    ) -> dict[str, Any]:
+        if server.source_registry_admin is None:
+            raise WorkflowRpcError(
+                data={
+                    "code": "source_registry_unavailable",
+                    "message": "source registry admin mutations are not available for this server",
+                }
+            )
+        try:
+            return await server.source_registry_admin.remove_registry_entry(
+                source_id=params.source_id,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)

src/wf_transport_rpc_http/models.py

@@ -188,3 +188,16 @@ class ListRegistryEntriesParams(RpcParamsModel):
 
 class InspectRegistryEntryParams(RpcParamsModel):
     source_id: str = Field(min_length=1)
+
+
+class AddRegistryEntryParams(RpcParamsModel):
+    entry: dict[str, Any]
+
+
+class UpdateRegistryEntryParams(RpcParamsModel):
+    source_id: str = Field(min_length=1)
+    patch: dict[str, Any]
+
+
+class RegistryEntryIdParams(RpcParamsModel):
+    source_id: str = Field(min_length=1)

tests/wf_api/test_source_registry_admin_api.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import asyncio
-from dataclasses import dataclass, field
+from collections.abc import Mapping
+from dataclasses import asdict, dataclass, field
 from typing import Any
 
 import pytest
@@ -145,3 +146,173 @@ def test_api_satisfies_surface_protocol() -> None:
     api: WorkflowSourceRegistrySurface = _api(FakeRegistryEntry(id="x"))
 
     assert api is not None
+
+
+class FakeMutationProvider:
+    """Mutable fake that tracks mutation calls for assertion."""
+
+    def __init__(
+        self,
+        entries: list[FakeRegistryEntry] | None = None,
+    ) -> None:
+        self._entries = list(entries) if entries else []
+
+    def add_registry_entry(self, entry: Mapping[str, Any]) -> dict[str, Any]:
+        fe = FakeRegistryEntry(**entry)
+        self._entries.append(fe)
+        return asdict(fe)
+
+    def update_registry_entry(self, source_id: str, patch: Mapping[str, Any]) -> dict[str, Any]:
+        for i, e in enumerate(self._entries):
+            if e.id == source_id:
+                merged = asdict(e)
+                merged.update(patch)
+                self._entries[i] = FakeRegistryEntry(**merged)
+                return merged
+        raise KeyError(source_id)
+
+    def set_registry_entry_enabled(self, source_id: str, enabled: bool) -> dict[str, Any]:
+        for i, e in enumerate(self._entries):
+            if e.id == source_id:
+                merged = asdict(e)
+                merged["enabled"] = enabled
+                self._entries[i] = FakeRegistryEntry(**merged)
+                return merged
+        raise KeyError(source_id)
+
+    def remove_registry_entry(self, source_id: str) -> dict[str, Any]:
+        if not any(e.id == source_id for e in self._entries):
+            raise KeyError(source_id)
+        self._entries = [e for e in self._entries if e.id != source_id]
+        return {"removed": True, "source_id": source_id}
+
+
+def _mutation_api(
+    entries: list[FakeRegistryEntry] | None = None,
+    config_ids: set[str] | None = None,
+) -> tuple[WorkflowSourceRegistryApi, FakeMutationProvider]:
+    provider = FakeRegistryProvider(list(entries) if entries else [], config_ids)
+    mutation = FakeMutationProvider(list(entries) if entries else [])
+    return WorkflowSourceRegistryApi(provider=provider, mutation_provider=mutation), mutation
+
+
+def test_add_registry_entry() -> None:
+    api, _ = _mutation_api()
+    new_entry = {"id": "new.source", "kind": "mcp", "enabled": True, "provider": "new", "account": "default", "profile": None, "transport": {"kind": "stdio"}, "auth_ref": None}
+    payload = asyncio.run(api.add_registry_entry(entry=new_entry))
+
+    assert payload["entry"]["id"] == "new.source"
+    assert payload["entry"]["provider"] == "new"
+    assert payload["shadowed_by_config"] is False
+
+
+def test_add_registry_entry_shadowed() -> None:
+    api, _ = _mutation_api(config_ids={"new.source"})
+    new_entry = {"id": "new.source", "kind": "mcp", "enabled": True, "provider": "new", "account": "default", "profile": None, "transport": {"kind": "stdio"}, "auth_ref": None}
+    payload = asyncio.run(api.add_registry_entry(entry=new_entry))
+
+    assert payload["entry"]["id"] == "new.source"
+    assert payload["shadowed_by_config"] is True
+
+
+def test_update_registry_entry() -> None:
+    api, _ = _mutation_api(
+        entries=[FakeRegistryEntry(id="upd.source", provider="old")],
+    )
+    payload = asyncio.run(api.update_registry_entry(source_id="upd.source", patch={"provider": "new"}))
+
+    assert payload["entry"]["id"] == "upd.source"
+    assert payload["entry"]["provider"] == "new"
+    assert payload["shadowed_by_config"] is False
+
+
+def test_enable_registry_entry() -> None:
+    api, _ = _mutation_api(
+        entries=[FakeRegistryEntry(id="toggle.source", enabled=False)],
+    )
+    payload = asyncio.run(api.enable_registry_entry(source_id="toggle.source"))
+
+    assert payload["entry"]["id"] == "toggle.source"
+    assert payload["entry"]["enabled"] is True
+    assert payload["shadowed_by_config"] is False
+
+
+def test_disable_registry_entry() -> None:
+    api, _ = _mutation_api(
+        entries=[FakeRegistryEntry(id="toggle.source", enabled=True)],
+    )
+    payload = asyncio.run(api.disable_registry_entry(source_id="toggle.source"))
+
+    assert payload["entry"]["id"] == "toggle.source"
+    assert payload["entry"]["enabled"] is False
+    assert payload["shadowed_by_config"] is False
+
+
+def test_remove_registry_entry() -> None:
+    api, _ = _mutation_api(
+        entries=[FakeRegistryEntry(id="rem.source")],
+    )
+    payload = asyncio.run(api.remove_registry_entry(source_id="rem.source"))
+
+    assert payload == {"removed": True, "source_id": "rem.source"}
+
+
+def test_update_nonexistent_raises_key_error() -> None:
+    api, _ = _mutation_api()
+    with pytest.raises(KeyError):
+        asyncio.run(api.update_registry_entry(source_id="missing", patch={}))
+
+
+def test_enable_nonexistent_raises_key_error() -> None:
+    api, _ = _mutation_api()
+    with pytest.raises(KeyError):
+        asyncio.run(api.enable_registry_entry(source_id="missing"))
+
+
+def test_disable_nonexistent_raises_key_error() -> None:
+    api, _ = _mutation_api()
+    with pytest.raises(KeyError):
+        asyncio.run(api.disable_registry_entry(source_id="missing"))
+
+
+def test_remove_nonexistent_raises_key_error() -> None:
+    api, _ = _mutation_api()
+    with pytest.raises(KeyError):
+        asyncio.run(api.remove_registry_entry(source_id="missing"))
+
+
+def test_add_raises_without_mutation_provider() -> None:
+    api = _api()
+    new_entry = {"id": "x", "kind": "mcp", "enabled": True}
+    with pytest.raises(TypeError, match="requires a mutation provider"):
+        asyncio.run(api.add_registry_entry(entry=new_entry))
+
+
+def test_update_raises_without_mutation_provider() -> None:
+    api = _api()
+    with pytest.raises(TypeError, match="requires a mutation provider"):
+        asyncio.run(api.update_registry_entry(source_id="x", patch={}))
+
+
+def test_enable_raises_without_mutation_provider() -> None:
+    api = _api()
+    with pytest.raises(TypeError, match="requires a mutation provider"):
+        asyncio.run(api.enable_registry_entry(source_id="x"))
+
+
+def test_disable_raises_without_mutation_provider() -> None:
+    api = _api()
+    with pytest.raises(TypeError, match="requires a mutation provider"):
+        asyncio.run(api.disable_registry_entry(source_id="x"))
+
+
+def test_remove_raises_without_mutation_provider() -> None:
+    api = _api()
+    with pytest.raises(TypeError, match="requires a mutation provider"):
+        asyncio.run(api.remove_registry_entry(source_id="x"))
+
+
+def test_api_with_mutation_satisfies_surface_protocol() -> None:
+    api, _ = _mutation_api(entries=[FakeRegistryEntry(id="x")])
+    surface: WorkflowSourceRegistrySurface = api
+    assert surface is not None

tests/wf_cli/test_source_registry.py

@@ -2,20 +2,70 @@ from __future__ import annotations
 
 import json
 from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
 
+import pytest
 from typer.testing import CliRunner
 
 from wf_cli.app import app
+from wf_cli.commands.source_registry import _read_json_arg
+from wf_cli.context import CliContext
 
 runner = CliRunner()
 
 
+def _make_config(tmp_path: Path) -> dict[str, Any]:
+    return {
+        "version": 1,
+        "client": {"target": {"kind": "local"}},
+        "server": {"store": {"kind": "filesystem", "root": str(tmp_path / "store")}},
+    }
+
+
+def _fake_context_with_admin(
+    mock_surface: MagicMock | None = None,
+) -> CliContext:
+    return CliContext(
+        config_path=Path("dummy"),
+        service=None,
+        handlers=MagicMock(),
+        source_admin=MagicMock(),
+        admin=MagicMock(),
+        source_registry_admin=mock_surface or MagicMock(),
+    )
+
+
+def _patch_load_cli_context(
+    monkeypatch: pytest.MonkeyPatch, fake_ctx: CliContext
+) -> None:
+    monkeypatch.setattr(
+        "wf_cli.commands.source_registry.load_cli_context_from_typer",
+        lambda _ctx: fake_ctx,
+    )
+
+
+def _patch_asyncio_run(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "wf_cli.commands.source_registry.asyncio.run",
+        lambda coro: coro,
+    )
+
+
+# --- help tests ---
+
+
 def test_wf_admin_registry_help_exists() -> None:
     result = runner.invoke(app, ["admin", "registry", "--help"])
 
     assert result.exit_code == 0
     assert "list" in result.output
     assert "inspect" in result.output
+    assert "add" in result.output
+    assert "update" in result.output
+    assert "enable" in result.output
+    assert "disable" in result.output
+    assert "remove" in result.output
 
 
 def test_wf_admin_registry_list_help_exists() -> None:
@@ -33,18 +83,53 @@ def test_wf_admin_registry_inspect_help_exists() -> None:
     assert "SOURCE_ID" in result.output
 
 
-def test_wf_admin_registry_list_local_static_returns_unavailable(tmp_path: Path) -> None:
+def test_wf_admin_registry_add_help_exists() -> None:
+    result = runner.invoke(app, ["admin", "registry", "add", "--help"])
+
+    assert result.exit_code == 0
+    assert "--input" in result.output
+    assert "--input-file" in result.output
+
+
+def test_wf_admin_registry_update_help_exists() -> None:
+    result = runner.invoke(app, ["admin", "registry", "update", "--help"])
+
+    assert result.exit_code == 0
+    assert "SOURCE_ID" in result.output
+    assert "--patch" in result.output
+    assert "--patch-file" in result.output
+
+
+def test_wf_admin_registry_enable_help_exists() -> None:
+    result = runner.invoke(app, ["admin", "registry", "enable", "--help"])
+
+    assert result.exit_code == 0
+    assert "SOURCE_ID" in result.output
+
+
+def test_wf_admin_registry_disable_help_exists() -> None:
+    result = runner.invoke(app, ["admin", "registry", "disable", "--help"])
+
+    assert result.exit_code == 0
+    assert "SOURCE_ID" in result.output
+
+
+def test_wf_admin_registry_remove_help_exists() -> None:
+    result = runner.invoke(app, ["admin", "registry", "remove", "--help"])
+
+    assert result.exit_code == 0
+    assert "SOURCE_ID" in result.output
+    assert "--confirm" in result.output
+
+
+# --- unavailability tests ---
+
+
+def test_wf_admin_registry_list_local_static_returns_unavailable(
+    tmp_path: Path,
+) -> None:
     config_path = tmp_path / "wf_mcp.config.json"
-    config_path.write_text(
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
-        json.dumps(
-            {
-                "version": 1,
-                "client": {"target": {"kind": "local"}},
-                "server": {"store": {"kind": "filesystem", "root": str(tmp_path / "store")}},
-            }
-        ),
-        encoding="utf-8",
-    )
 
     result = runner.invoke(
         app,
@@ -55,18 +140,11 @@ def test_wf_admin_registry_list_local_static_returns_unavailable(tmp_path: Path)
     assert "not available" in result.output
 
 
-def test_wf_admin_registry_inspect_local_static_returns_unavailable(tmp_path: Path) -> None:
+def test_wf_admin_registry_inspect_local_static_returns_unavailable(
+    tmp_path: Path,
+) -> None:
     config_path = tmp_path / "wf_mcp.config.json"
-    config_path.write_text(
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
-        json.dumps(
-            {
-                "version": 1,
-                "client": {"target": {"kind": "local"}},
-                "server": {"store": {"kind": "filesystem", "root": str(tmp_path / "store")}},
-            }
-        ),
-        encoding="utf-8",
-    )
 
     result = runner.invoke(
         app,
@@ -75,3 +153,354 @@ def test_wf_admin_registry_inspect_local_static_returns_unavailable(tmp_path: Pa
 
     assert result.exit_code != 0
     assert "not available" in result.output
+
+
+def test_wf_admin_registry_add_local_static_returns_unavailable(tmp_path: Path) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
+
+    result = runner.invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "admin",
+            "registry",
+            "add",
+            "--input",
+            '{"id": "test.source"}',
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "not available" in result.output
+
+
+def test_wf_admin_registry_update_local_static_returns_unavailable(
+    tmp_path: Path,
+) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
+
+    result = runner.invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "admin",
+            "registry",
+            "update",
+            "github.work",
+            "--patch",
+            '{"enabled": false}',
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "not available" in result.output
+
+
+def test_wf_admin_registry_enable_local_static_returns_unavailable(
+    tmp_path: Path,
+) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
+
+    result = runner.invoke(
+        app,
+        ["--config", str(config_path), "admin", "registry", "enable", "github.work"],
+    )
+
+    assert result.exit_code != 0
+    assert "not available" in result.output
+
+
+def test_wf_admin_registry_disable_local_static_returns_unavailable(
+    tmp_path: Path,
+) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
+
+    result = runner.invoke(
+        app,
+        ["--config", str(config_path), "admin", "registry", "disable", "github.work"],
+    )
+
+    assert result.exit_code != 0
+    assert "not available" in result.output
+
+
+def test_wf_admin_registry_remove_local_static_returns_unavailable(
+    tmp_path: Path,
+) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(json.dumps(_make_config(tmp_path)), encoding="utf-8")
+
+    result = runner.invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "admin",
+            "registry",
+            "remove",
+            "github.work",
+            "--confirm",
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "not available" in result.output
+
+
+# --- validation tests ---
+
+
+def test_wf_admin_registry_remove_without_confirm_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mock_admin = MagicMock()
+    mock_admin.remove_registry_entry.return_value = {}
+    fake_ctx = _fake_context_with_admin(mock_admin)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(app, ["admin", "registry", "remove", "github.work"])
+
+    assert result.exit_code != 0
+    assert "--confirm" in result.output
+    mock_admin.remove_registry_entry.assert_not_called()
+
+
+def test_wf_admin_registry_add_both_input_flags_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_ctx = _fake_context_with_admin()
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(
+        app,
+        [
+            "admin",
+            "registry",
+            "add",
+            "--input",
+            '{"id": "x"}',
+            "--input-file",
+            "dummy.json",
+        ],
+    )
+
+    assert result.exit_code != 0
+
+
+def test_wf_admin_registry_add_no_input_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_ctx = _fake_context_with_admin()
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(app, ["admin", "registry", "add"])
+
+    assert result.exit_code != 0
+
+
+def test_wf_admin_registry_add_invalid_json_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_ctx = _fake_context_with_admin()
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(
+        app,
+        ["admin", "registry", "add", "--input", "not json"],
+    )
+
+    assert result.exit_code != 0
+    assert "invalid JSON" in result.output
+
+
+def test_wf_admin_registry_update_both_patch_flags_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_ctx = _fake_context_with_admin()
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(
+        app,
+        [
+            "admin",
+            "registry",
+            "update",
+            "x",
+            "--patch",
+            '{"a": 1}',
+            "--patch-file",
+            "dummy.json",
+        ],
+    )
+
+    assert result.exit_code != 0
+
+
+def test_wf_admin_registry_update_no_patch_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_ctx = _fake_context_with_admin()
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(app, ["admin", "registry", "update", "x"])
+
+    assert result.exit_code != 0
+
+
+def test_wf_admin_registry_update_invalid_json_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_ctx = _fake_context_with_admin()
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+
+    result = runner.invoke(
+        app,
+        ["admin", "registry", "update", "x", "--patch", "not json"],
+    )
+
+    assert result.exit_code != 0
+    assert "invalid JSON" in result.output
+
+
+# --- delegation tests ---
+
+
+def test_wf_admin_registry_add_delegates_to_surface(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mock_surface = MagicMock()
+    mock_surface.add_registry_entry.return_value = {"id": "new.source"}
+    fake_ctx = _fake_context_with_admin(mock_surface)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+    _patch_asyncio_run(monkeypatch)
+
+    result = runner.invoke(
+        app,
+        ["admin", "registry", "add", "--input", '{"id": "new.source"}'],
+    )
+
+    assert result.exit_code == 0
+    mock_surface.add_registry_entry.assert_called_once_with(entry={"id": "new.source"})
+
+
+def test_wf_admin_registry_update_delegates_to_surface(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mock_surface = MagicMock()
+    mock_surface.update_registry_entry.return_value = {"id": "x", "enabled": False}
+    fake_ctx = _fake_context_with_admin(mock_surface)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+    _patch_asyncio_run(monkeypatch)
+
+    result = runner.invoke(
+        app,
+        ["admin", "registry", "update", "x", "--patch", '{"enabled": false}'],
+    )
+
+    assert result.exit_code == 0
+    mock_surface.update_registry_entry.assert_called_once_with(
+        source_id="x", patch={"enabled": False}
+    )
+
+
+def test_wf_admin_registry_enable_delegates_to_surface(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mock_surface = MagicMock()
+    mock_surface.enable_registry_entry.return_value = {"id": "x", "enabled": True}
+    fake_ctx = _fake_context_with_admin(mock_surface)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+    _patch_asyncio_run(monkeypatch)
+
+    result = runner.invoke(app, ["admin", "registry", "enable", "x"])
+
+    assert result.exit_code == 0
+    mock_surface.enable_registry_entry.assert_called_once_with(source_id="x")
+
+
+def test_wf_admin_registry_disable_delegates_to_surface(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mock_surface = MagicMock()
+    mock_surface.disable_registry_entry.return_value = {"id": "x", "enabled": False}
+    fake_ctx = _fake_context_with_admin(mock_surface)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+    _patch_asyncio_run(monkeypatch)
+
+    result = runner.invoke(app, ["admin", "registry", "disable", "x"])
+
+    assert result.exit_code == 0
+    mock_surface.disable_registry_entry.assert_called_once_with(source_id="x")
+
+
+def test_wf_admin_registry_remove_delegates_to_surface(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mock_surface = MagicMock()
+    mock_surface.remove_registry_entry.return_value = {"id": "x", "removed": True}
+    fake_ctx = _fake_context_with_admin(mock_surface)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+    _patch_asyncio_run(monkeypatch)
+
+    result = runner.invoke(app, ["admin", "registry", "remove", "x", "--confirm"])
+
+    assert result.exit_code == 0
+    mock_surface.remove_registry_entry.assert_called_once_with(source_id="x")
+
+
+# --- _read_json_arg unit tests ---
+
+
+def test_read_json_arg_inline() -> None:
+    assert _read_json_arg('{"a": 1}', None, "--input/--input-file") == {"a": 1}
+
+
+def test_read_json_arg_file(tmp_path: Path) -> None:
+    f = tmp_path / "data.json"
+    f.write_text('{"b": 2}', encoding="utf-8")
+    assert _read_json_arg(None, str(f), "--input/--input-file") == {"b": 2}
+
+
+def test_read_json_arg_both_raises() -> None:
+    with pytest.raises(Exception, match="provide exactly one"):
+        _read_json_arg('{"a": 1}', "file.json", "--input/--input-file")
+
+
+def test_read_json_arg_neither_raises() -> None:
+    with pytest.raises(Exception, match="is required"):
+        _read_json_arg(None, None, "--input/--input-file")
+
+
+def test_read_json_arg_invalid_inline() -> None:
+    with pytest.raises(Exception, match="invalid JSON"):
+        _read_json_arg("not json", None, "--input/--input-file")
+
+
+def test_read_json_arg_rejects_non_object_inline() -> None:
+    with pytest.raises(Exception, match="must be a JSON object"):
+        _read_json_arg("[1, 2]", None, "--input/--input-file")
+
+
+def test_read_json_arg_rejects_non_object_file(tmp_path: Path) -> None:
+    f = tmp_path / "array.json"
+    f.write_text("[1, 2]", encoding="utf-8")
+    with pytest.raises(Exception, match="must be a JSON object"):
+        _read_json_arg(None, str(f), "--input/--input-file")
+
+
+def test_read_json_arg_invalid_file(tmp_path: Path) -> None:
+    f = tmp_path / "bad.json"
+    f.write_text("not json", encoding="utf-8")
+    with pytest.raises(Exception, match="invalid JSON in file"):
+        _read_json_arg(None, str(f), "--input/--input-file")
+
+
+def test_read_json_arg_missing_file() -> None:
+    with pytest.raises(Exception, match="file not found"):
+        _read_json_arg(None, "/nonexistent/file.json", "--input/--input-file")

tests/wf_mcp/service/test_source_registry_admin.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from pathlib import Path
 
+import pytest
+
 from wf_mcp.broker.service.source_registry_admin import SourceRegistryAdminProvider
 from wf_mcp.models import ConnectionConfig
 from wf_mcp.source_registry import (
@@ -27,6 +29,28 @@ def _entry(source_id: str, *, provider: str = "github", account: str = "work") -
     )
 
 
+def _entry_dict(source_id: str, *, provider: str = "github", account: str = "work") -> dict:
+    return {
+        "id": source_id,
+        "provider": provider,
+        "account": account,
+        "transport": {"kind": "stdio", "command": "npx", "args": (), "env": {}},
+    }
+
+
+def _provider(
+    tmp_path: Path,
+    entries: list[McpSourceRegistryEntry] | None = None,
+    config_ids: frozenset[str] | None = None,
+) -> SourceRegistryAdminProvider:
+    store = _store_with_entries(tmp_path / "reg", *(entries or []))
+    connections = [ConnectionConfig(id=cid, server="s", account="a") for cid in (config_ids or frozenset())]
+    return SourceRegistryAdminProvider(source_registry_store=store, config_connections=connections)
+
+
+# -- read tests ------------------------------------------------------------
+
+
 def test_provider_lists_entries_from_store(tmp_path: Path) -> None:
     store = _store_with_entries(
         tmp_path / "reg",
@@ -38,7 +62,7 @@ def test_provider_lists_entries_from_store(tmp_path: Path) -> None:
     entries = provider.list_registry_entries()
 
     assert len(entries) == 2
-    ids = {getattr(e, "id", getattr(e, "get", lambda k: None)("id")) for e in entries}
+    ids = {e.id for e in entries}
     assert ids == {"alpha.work", "zeta.personal"}
 
 
@@ -67,3 +91,124 @@ def test_provider_empty_store(tmp_path: Path) -> None:
 
     assert entries == []
     assert shadowed == set()
+
+
+# -- add tests -------------------------------------------------------------
+
+
+def test_add_persists_and_round_trips(tmp_path: Path) -> None:
+    provider = _provider(tmp_path)
+    result = provider.add_registry_entry(_entry_dict("new.server"))
+
+    assert result.id == "new.server"
+    reloaded = provider.list_registry_entries()
+    assert len(reloaded) == 1
+    assert reloaded[0].id == "new.server"
+
+
+def test_add_rejects_config_shadowed_id(tmp_path: Path) -> None:
+    provider = _provider(tmp_path, config_ids=frozenset({"config.server"}))
+
+    with pytest.raises(ValueError, match="shadowed by a config connection"):
+        provider.add_registry_entry(_entry_dict("config.server"))
+
+    assert provider.list_registry_entries() == []
+
+
+def test_add_rejects_duplicate_registry_id(tmp_path: Path) -> None:
+    provider = _provider(tmp_path, entries=[_entry("existing.server")])
+
+    with pytest.raises(ValueError, match="duplicate"):
+        provider.add_registry_entry(_entry_dict("existing.server"))
+
+    assert len(provider.list_registry_entries()) == 1
+
+
+def test_add_malformed_payload_raises_validation_error(tmp_path: Path) -> None:
+    provider = _provider(tmp_path)
+
+    with pytest.raises(Exception, match="validation"):
+        provider.add_registry_entry({"id": "x"})
+
+
+# -- update tests ----------------------------------------------------------
+
+
+def test_update_persists_provider_account_transport_changes(tmp_path: Path) -> None:
+    provider = _provider(tmp_path, entries=[_entry("src.server")])
+
+    result = provider.update_registry_entry(
+        "src.server",
+        {"provider": "new_provider", "account": "new_account"},
+    )
+
+    assert result.id == "src.server"
+    assert result.provider == "new_provider"
+    assert result.account == "new_account"
+    reloaded = provider.list_registry_entries()
+    reloaded_entry = reloaded[0]
+    assert reloaded_entry.provider == "new_provider"
+    assert reloaded_entry.account == "new_account"
+
+
+def test_update_rejects_id_change(tmp_path: Path) -> None:
+    provider = _provider(tmp_path, entries=[_entry("old.name")])
+
+    with pytest.raises(ValueError, match="cannot change source id"):
+        provider.update_registry_entry("old.name", {"id": "new.name"})
+
+    # original unchanged
+    reloaded = provider.list_registry_entries()
+    assert reloaded[0].id == "old.name"
+
+
+def test_update_missing_source_raises_key_error(tmp_path: Path) -> None:
+    provider = _provider(tmp_path)
+
+    with pytest.raises(KeyError, match="unknown registry source"):
+        provider.update_registry_entry("no.such.id", {})
+
+
+# -- enable/disable tests --------------------------------------------------
+
+
+def test_enable_disable_persist(tmp_path: Path) -> None:
+    provider = _provider(tmp_path, entries=[_entry("toggle.server")])
+
+    disabled = provider.set_registry_entry_enabled("toggle.server", False)
+    assert disabled.enabled is False
+    reloaded = provider.list_registry_entries()
+    assert reloaded[0].enabled is False
+
+    enabled = provider.set_registry_entry_enabled("toggle.server", True)
+    assert enabled.enabled is True
+    reloaded = provider.list_registry_entries()
+    assert reloaded[0].enabled is True
+
+
+def test_enable_disable_missing_source_raises_key_error(tmp_path: Path) -> None:
+    provider = _provider(tmp_path)
+
+    with pytest.raises(KeyError, match="unknown registry source"):
+        provider.set_registry_entry_enabled("no.such.id", True)
+
+
+# -- remove tests ----------------------------------------------------------
+
+
+def test_remove_persists_absence_and_does_not_touch_unrelated(tmp_path: Path) -> None:
+    provider = _provider(tmp_path, entries=[_entry("keep.server"), _entry("drop.server")])
+
+    result = provider.remove_registry_entry("drop.server")
+
+    assert result == {"removed": True, "source_id": "drop.server"}
+    reloaded = provider.list_registry_entries()
+    assert len(reloaded) == 1
+    assert reloaded[0].id == "keep.server"
+
+
+def test_remove_missing_source_raises_key_error(tmp_path: Path) -> None:
+    provider = _provider(tmp_path)
+
+    with pytest.raises(KeyError, match="unknown registry source"):
+        provider.remove_registry_entry("no.such.id")

tests/wf_transport_rpc_http/test_source_registry_rpc.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 from dataclasses import dataclass, replace
+from typing import Any
 
 import httpx
 
@@ -37,9 +38,44 @@ class FakeRegistryProvider:
         return {"github.work"}
 
 
-async def _rpc(
+class FakeMutationProvider:
-    client: httpx.AsyncClient, method: str, params: dict
+    def __init__(self) -> None:
-) -> dict:
+        self.entries: dict[str, dict[str, Any]] = {
+            "github.work": {
+                "id": "github.work",
+                "kind": "mcp",
+                "enabled": True,
+                "transport": {"kind": "stdio", "command": "npx"},
+            }
+        }
+
+    def add_registry_entry(self, entry: Any) -> dict[str, Any]:
+        source_id = entry["id"]
+        self.entries[source_id] = dict(entry)
+        return self.entries[source_id]
+
+    def update_registry_entry(self, source_id: str, patch: Any) -> dict[str, Any]:
+        if source_id not in self.entries:
+            raise KeyError(f"unknown registry source {source_id!r}")
+        self.entries[source_id].update(patch)
+        return self.entries[source_id]
+
+    def set_registry_entry_enabled(
+        self, source_id: str, enabled: bool
+    ) -> dict[str, Any]:
+        if source_id not in self.entries:
+            raise KeyError(f"unknown registry source {source_id!r}")
+        self.entries[source_id]["enabled"] = enabled
+        return self.entries[source_id]
+
+    def remove_registry_entry(self, source_id: str) -> dict[str, Any]:
+        if source_id not in self.entries:
+            raise KeyError(f"unknown registry source {source_id!r}")
+        self.entries.pop(source_id)
+        return {"removed": True, "source_id": source_id}
+
+
+async def _rpc(client: httpx.AsyncClient, method: str, params: dict) -> dict:
     response = await client.post(
         "/rpc",
         json={"jsonrpc": "2.0", "id": "test", "method": method, "params": params},
@@ -48,6 +84,19 @@ async def _rpc(
     return response.json()
 
 
+def _server_with_mutation_provider(tmp_path: Any) -> Any:
+    return replace(
+        build_local_static_workflow_server(tmp_path / "store"),
+        source_registry_admin=WorkflowSourceRegistryApi(
+            provider=FakeRegistryProvider(),
+            mutation_provider=FakeMutationProvider(),
+        ),
+    )
+
+
+# --- read-only tests (unchanged) ---
+
+
 def test_rpc_source_registry_list_unavailable_on_local_static(tmp_path) -> None:
     async def scenario() -> None:
         server = build_local_static_workflow_server(tmp_path / "store")
@@ -116,6 +165,286 @@ def test_rpc_source_registry_methods_return_registry_payloads(tmp_path) -> None:
     asyncio.run(scenario())
 
 
+# --- mutation unavailable tests ---
+
+
+def test_rpc_source_registry_add_unavailable_on_local_static(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.add",
+                {"entry": {"id": "new.source", "kind": "mcp"}},
+            )
+
+        assert "error" in payload
+        assert payload["error"]["data"]["code"] == "source_registry_unavailable"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_update_unavailable_on_local_static(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.update",
+                {"source_id": "github.work", "patch": {"enabled": False}},
+            )
+
+        assert "error" in payload
+        assert payload["error"]["data"]["code"] == "source_registry_unavailable"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_enable_unavailable_on_local_static(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.enable",
+                {"source_id": "github.work"},
+            )
+
+        assert "error" in payload
+        assert payload["error"]["data"]["code"] == "source_registry_unavailable"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_disable_unavailable_on_local_static(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.disable",
+                {"source_id": "github.work"},
+            )
+
+        assert "error" in payload
+        assert payload["error"]["data"]["code"] == "source_registry_unavailable"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_remove_unavailable_on_local_static(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.remove",
+                {"source_id": "github.work"},
+            )
+
+        assert "error" in payload
+        assert payload["error"]["data"]["code"] == "source_registry_unavailable"
+
+    asyncio.run(scenario())
+
+
+# --- mutation success tests ---
+
+
+def test_rpc_source_registry_add_returns_entry(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.add",
+                {"entry": {"id": "new.mcp", "kind": "mcp", "enabled": True}},
+            )
+
+        assert "result" in payload
+        assert payload["result"]["entry"]["id"] == "new.mcp"
+        assert payload["result"]["entry"]["kind"] == "mcp"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_update_returns_entry(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.update",
+                {"source_id": "github.work", "patch": {"enabled": False}},
+            )
+
+        assert "result" in payload
+        assert payload["result"]["entry"]["id"] == "github.work"
+        assert payload["result"]["entry"]["enabled"] is False
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_enable_returns_entry(tmp_path) -> None:
+    async def scenario() -> None:
+        mutation = FakeMutationProvider()
+        mutation.entries["github.work"]["enabled"] = False
+        server = replace(
+            build_local_static_workflow_server(tmp_path / "store"),
+            source_registry_admin=WorkflowSourceRegistryApi(
+                provider=FakeRegistryProvider(),
+                mutation_provider=mutation,
+            ),
+        )
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.enable",
+                {"source_id": "github.work"},
+            )
+
+        assert "result" in payload
+        assert payload["result"]["entry"]["enabled"] is True
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_disable_returns_entry(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.disable",
+                {"source_id": "github.work"},
+            )
+
+        assert "result" in payload
+        assert payload["result"]["entry"]["enabled"] is False
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_remove_returns_removed(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.remove",
+                {"source_id": "github.work"},
+            )
+
+        assert "result" in payload
+        assert payload["result"]["removed"] is True
+        assert payload["result"]["source_id"] == "github.work"
+
+    asyncio.run(scenario())
+
+
+# --- mutation error tests ---
+
+
+def test_rpc_source_registry_add_missing_entry_raises_error(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.add",
+                {"entry": {}},
+            )
+
+        assert "error" in payload
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_update_missing_source_raises_error(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.update",
+                {"source_id": "nonexistent", "patch": {"enabled": False}},
+            )
+
+        assert "error" in payload
+
+    asyncio.run(scenario())
+
+
+def test_rpc_source_registry_remove_missing_source_raises_error(tmp_path) -> None:
+    async def scenario() -> None:
+        server = _server_with_mutation_provider(tmp_path)
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://test"
+        ) as client:
+            payload = await _rpc(
+                client,
+                "workflow.admin.source_registry.remove",
+                {"source_id": "nonexistent"},
+            )
+
+        assert "error" in payload
+
+    asyncio.run(scenario())
+
+
+# --- client method tests ---
+
+
 def test_rpc_client_source_registry_calls_correct_methods(tmp_path) -> None:
     async def scenario() -> None:
         server = build_local_static_workflow_server(tmp_path / "store")
@@ -149,3 +478,46 @@ def test_rpc_client_source_registry_calls_correct_methods(tmp_path) -> None:
         assert "source registry admin reads are not available" in inspect_error
 
     asyncio.run(scenario())
+
+
+def test_rpc_client_source_registry_mutation_methods_exist() -> None:
+    from wf_transport_rpc_http.client import RpcWorkflowApiClient
+
+    client = RpcWorkflowApiClient.__new__(RpcWorkflowApiClient)
+    calls: list[tuple[str, dict[str, Any]]] = []
+
+    async def fake_call(method: str, params: dict[str, Any]) -> dict[str, Any]:
+        calls.append((method, params))
+        return {}
+
+    client._call = fake_call  # type: ignore[assignment]
+
+    asyncio.run(client.add_registry_entry(entry={"id": "x", "kind": "mcp"}))
+    assert calls[-1] == (
+        "workflow.admin.source_registry.add",
+        {"entry": {"id": "x", "kind": "mcp"}},
+    )
+
+    asyncio.run(client.update_registry_entry(source_id="s", patch={"enabled": False}))
+    assert calls[-1] == (
+        "workflow.admin.source_registry.update",
+        {"source_id": "s", "patch": {"enabled": False}},
+    )
+
+    asyncio.run(client.enable_registry_entry(source_id="s"))
+    assert calls[-1] == (
+        "workflow.admin.source_registry.enable",
+        {"source_id": "s"},
+    )
+
+    asyncio.run(client.disable_registry_entry(source_id="s"))
+    assert calls[-1] == (
+        "workflow.admin.source_registry.disable",
+        {"source_id": "s"},
+    )
+
+    asyncio.run(client.remove_registry_entry(source_id="s"))
+    assert calls[-1] == (
+        "workflow.admin.source_registry.remove",
+        {"source_id": "s"},
+    )