draft -> wrapper

docs/wf_mcp_end_to_end_runbook.md

@@ -513,6 +513,8 @@ Concrete MCP sequence:
 7. `wf.workflow.validate_draft_workspace` if capabilities changed or you want
    to refresh diagnostics without editing the draft.
 8. `wf.workflow.create_artifact_from_workspace` after validation is clean.
+   Use `wf.workflow.create_wrapper_from_workspace` instead when the workspace
+   is a reusable wrapper around a raw capability.
 9. `wf.workflow.save_deployment`, then `validate_deployment`, then
    `run_deployment`.
 10. `wf.workflow.delete_draft_workspace` when the mutable authoring session is no
@@ -535,3 +537,7 @@ Concrete MCP sequence:
   }
 }
 ```
+
+`create_wrapper_from_workspace` accepts the same request shape except there is
+no `kind` field. It always saves `kind="wrapper"` and the result is discoverable
+as a workflow capability named `workflow.<artifact_id>.v<version>`.

docs/wf_mcp_operator_manual.md

@@ -355,6 +355,7 @@ resending the full draft each turn.
 | Refresh validation without changing revision | `wf.workflow.validate_draft_workspace` |
 | Change common draft fields without JSON Patch | `wf.workflow.set_draft_name`, `wf.workflow.set_draft_route`, `wf.workflow.set_step_input_map`, `wf.workflow.set_step_output_map` |
 | Save final workspace as artifact | `wf.workflow.create_artifact_from_workspace` |
+| Save final workspace as callable wrapper | `wf.workflow.create_wrapper_from_workspace` |
 | Clean up a draft workspace | `wf.workflow.delete_draft_workspace` |
 
 Workspace patches are optimistic-concurrency guarded. Pass the current
@@ -365,6 +366,11 @@ Workspace mutation tools use a single `request` object in MCP Inspector. That
 keeps the form grouped and lets the schema describe fields like
 `input_schema`, `output_map`, and `error_message_source`.
 
+Use `create_wrapper_from_workspace` when the draft is meant to normalize a raw
+capability into a reusable workflow-facing wrapper. It is the same validation
+path as `create_artifact_from_workspace`, but the saved artifact kind is fixed
+to `wrapper`.
+
 Minimal example:
 
 ```json

docs/workflow_capabilities.md

@@ -262,6 +262,7 @@ authoring loop:
 - `wf.workflow.list_capabilities`
   - lists compact paged enabled planner-visible workflow-ready node spec
     summaries, with optional query/source filtering
+  - also includes saved wrapper artifacts under source id `workflow`
   - includes the owning `source_id`, outcomes, and top-level input/output field
     names, but not full schemas
 - `wf.workflow.inspect_capability`
@@ -298,9 +299,10 @@ prompts
 resources
 ```
 
-Possible later additions may include saved wrappers or workflow artifacts as
+Possible later additions may include full workflow artifacts as first-class
-first-class projected capability kinds, but they should not erase the raw versus
+projected capability kinds, but they should not erase the raw versus
-workflow-facing distinction.
+workflow-facing distinction. Saved wrapper artifacts are already projected as
+workflow capabilities because they have a node-like callable boundary today.
 
 Examples:
 
@@ -325,6 +327,8 @@ Today:
 - saved artifacts can be tagged with `kind="workflow"` or `kind="wrapper"`
 - `wf.workflow.call_capability` can execute one planner-visible workflow
   capability directly and return normalized `outcome` / `output`
+- `wf.workflow.list_capabilities` and `wf.workflow.inspect_capability` project
+  saved wrapper artifacts as workflow capabilities under source id `workflow`
 
 Not yet implemented:
 

docs/workflow_drafts.md

@@ -299,7 +299,8 @@ The workspace flow is:
 2. `wf.workflow.get_draft_workspace`
 3. `wf.workflow.patch_draft_workspace`
 4. repeat get/patch until valid
-5. `wf.workflow.create_artifact_from_workspace`
+5. `wf.workflow.create_artifact_from_workspace` for a full workflow, or
+   `wf.workflow.create_wrapper_from_workspace` for a reusable callable wrapper
 
 Workspaces are mutable and revisioned. Artifacts are immutable and versioned.
 Patch calls must include the current `revision`; stale revisions return
@@ -314,6 +315,10 @@ In MCP Inspector, workspace mutation tools accept a single `request` object.
 This is deliberate: the request object carries descriptions and validation for
 the authoring envelope while raw JSON Schema fields remain plain JSON objects.
 
+`create_wrapper_from_workspace` is intentionally just the wrapper-specific save
+path. It validates and compiles the same draft workspace, but fixes the saved
+artifact kind to `wrapper` so clients do not need to pass `kind` manually.
+
 ## Patching Drafts
 
 `patch_draft` accepts JSON Patch operations.

src/wf_mcp/transparent_proxy/runtime.py

@@ -53,6 +53,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.set_step_output_map",
     "wf.workflow.create_minimal_draft_workspace",
     "wf.workflow.create_artifact_from_workspace",
+    "wf.workflow.create_wrapper_from_workspace",
     "wf.workflow.call_capability",
     "wf.workflow.inspect_artifact",
     "wf.workflow.list_deployments",

src/wf_mcp/workflow_surface/handlers.py

@@ -97,6 +97,10 @@ class WorkflowSurfaceHandlers:
                 and query.casefold() in detail.description.casefold()
             )
         ]
+        capabilities.extend(
+            self._wrapper_capability_summaries(query=query, source_id=source_id)
+        )
+        capabilities.sort(key=lambda capability: capability["name"])
         page = page_items(capabilities, cursor=cursor, limit=limit)
         return {
             "capabilities": list(page.items),
@@ -112,6 +116,9 @@ class WorkflowSurfaceHandlers:
             for detail in source.as_inventory().capabilities.node_spec_details:
                 if detail.name == qualified_name:
                     return detail.model_dump(mode="json")
+        wrapper_detail = self._wrapper_capability_detail(qualified_name)
+        if wrapper_detail is not None:
+            return wrapper_detail
         raise KeyError(f"unknown workflow capability {qualified_name!r}")
 
     async def call_capability(
@@ -163,6 +170,71 @@ class WorkflowSurfaceHandlers:
             return None
         return artifact
 
+    def _wrapper_capability_summaries(
+        self,
+        *,
+        query: str | None,
+        source_id: str | None,
+    ) -> list[dict[str, Any]]:
+        """Project saved wrappers into workflow capability discovery rows.
+
+        Wrapper artifacts are not live source NodeSpecs, but authors need to
+        discover and test them through the same workflow-facing REPL surface.
+        Full saved workflows stay out of this projection until graph-as-node is
+        real in core.
+        """
+        if source_id not in {None, "workflow"} or self.service.artifact_store is None:
+            return []
+        rows: list[dict[str, Any]] = []
+        for artifact in self.service.artifact_store.list_artifacts():
+            if artifact.kind != "wrapper":
+                continue
+            name = _artifact_capability_id(artifact)
+            if not _matches_capability_query(
+                name,
+                artifact.description,
+                query=query,
+            ):
+                continue
+            rows.append(
+                {
+                    "name": name,
+                    "source_id": "workflow",
+                    "description": artifact.description,
+                    "outcomes": list(artifact.outcomes),
+                    "is_async": True,
+                    "input_fields": _schema_field_names(artifact.input_schema),
+                    "output_fields": _schema_field_names(artifact.output_schema),
+                }
+            )
+        return rows
+
+    def _wrapper_capability_detail(
+        self,
+        qualified_name: str,
+    ) -> dict[str, Any] | None:
+        """Return a NodeSpec-like contract for one saved wrapper artifact."""
+        artifact = self._wrapper_artifact_for_capability_name(qualified_name)
+        if artifact is None:
+            return None
+        return {
+            "name": _artifact_capability_id(artifact),
+            "source_id": "workflow",
+            "kind": "wrapper_artifact",
+            "artifact_id": artifact.id,
+            "version": artifact.version,
+            "title": artifact.title,
+            "description": artifact.description,
+            "outcomes": list(artifact.outcomes),
+            "is_async": True,
+            "input_schema": artifact.input_schema,
+            "output_schema": artifact.output_schema,
+            "required_capabilities": {
+                name: capability.model_dump(mode="json")
+                for name, capability in sorted(artifact.required_capabilities.items())
+            },
+        }
+
     async def _call_wrapper_artifact(
         self,
         artifact: WorkflowArtifact,
@@ -562,9 +634,7 @@ class WorkflowSurfaceHandlers:
                 "in": {error_source: "message"},
                 "out": {},
             }
-            routes[DEFAULT_CALL_STEP_ID][DEFAULT_ERROR_OUTCOME] = (
+            routes[DEFAULT_CALL_STEP_ID][DEFAULT_ERROR_OUTCOME] = DEFAULT_ERROR_STEP_ID
-                DEFAULT_ERROR_STEP_ID
-            )
             routes[DEFAULT_ERROR_STEP_ID] = {DEFAULT_OK_OUTCOME: "__end__"}
         draft = {
             "name": name,
@@ -618,6 +688,33 @@ class WorkflowSurfaceHandlers:
             created_from_catalog_version=created_from_catalog_version,
         )
 
+    async def create_wrapper_from_workspace(
+        self,
+        *,
+        workspace_id: str,
+        artifact_id: str,
+        version: int,
+        title: str,
+        outcomes: Sequence[str],
+        description: str | None = None,
+        required_capabilities: dict[str, dict[str, Any]] | None = None,
+        source_bindings: dict[str, str] | None = None,
+        created_from_catalog_version: str | None = None,
+    ) -> dict[str, Any]:
+        """Save the current draft workspace as a callable wrapper artifact."""
+        return await self.create_artifact_from_workspace(
+            workspace_id=workspace_id,
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            outcomes=outcomes,
+            kind="wrapper",
+            description=description,
+            required_capabilities=required_capabilities,
+            source_bindings=source_bindings,
+            created_from_catalog_version=created_from_catalog_version,
+        )
+
     def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
         try:
             return self.service._get_qualified_spec(qualified_name).outcomes
@@ -842,6 +939,21 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
+def _matches_capability_query(
+    name: str,
+    description: str | None,
+    *,
+    query: str | None,
+) -> bool:
+    """Apply the same compact capability search semantics to every row kind."""
+    if query is None:
+        return True
+    lowered = query.casefold()
+    return lowered in name.casefold() or (
+        description is not None and lowered in description.casefold()
+    )
+
+
 def _first_state_path(output_map: dict[str, str]) -> str | None:
     """Return the first mapped state path for minimal error-route bootstraps."""
     for target in output_map.values():

src/wf_mcp/workflow_surface/models.py

@@ -216,3 +216,26 @@ class CreateArtifactFromWorkspaceRequest(BaseModel):
         default=None,
         description="Optional catalog version used while authoring.",
     )
+
+
+class CreateWrapperFromWorkspaceRequest(BaseModel):
+    """Typed MCP request for saving a draft workspace as a wrapper artifact."""
+
+    workspace_id: WorkspaceId
+    artifact_id: str = Field(description="Immutable wrapper artifact id to write.")
+    version: int = Field(ge=1, description="Wrapper artifact version to write.")
+    title: str = Field(description="Human-readable wrapper title.")
+    outcomes: list[str] = Field(description="Wrapper-level outcomes.")
+    description: str | None = Field(default=None, description="Optional description.")
+    required_capabilities: dict[str, dict[str, Any]] | None = Field(
+        default=None,
+        description="Optional explicit dependency contract override.",
+    )
+    source_bindings: SourceBindings | None = Field(
+        default=None,
+        description="Optional logical-to-concrete source bindings.",
+    )
+    created_from_catalog_version: str | None = Field(
+        default=None,
+        description="Optional catalog version used while authoring.",
+    )

src/wf_mcp/workflow_surface/tools.py

@@ -14,6 +14,7 @@ from .models import (
     CreateArtifactFromWorkspaceRequest,
     CreateDraftWorkspaceRequest,
     CreateMinimalDraftWorkspaceRequest,
+    CreateWrapperFromWorkspaceRequest,
     DeleteDraftWorkspaceRequest,
     DeleteDraftWorkspaceResult,
     DraftWorkspaceListResult,
@@ -405,9 +406,38 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                     if isinstance(capability, RequiredCapability)
                     else capability
                 )
-                for name, capability in (
+                for name, capability in (request.required_capabilities or {}).items()
-                    request.required_capabilities or {}
+            }
-                ).items()
+            or None,
+            source_bindings=dict(request.source_bindings or {}),
+            created_from_catalog_version=request.created_from_catalog_version,
+        )
+
+    @server.tool(
+        name="wf.workflow.create_wrapper_from_workspace",
+        title="Create Wrapper From Workspace",
+        description=(
+            "Validate the current draft workspace and save it as a callable "
+            "wrapper artifact."
+        ),
+    )
+    async def create_wrapper_from_workspace(
+        request: CreateWrapperFromWorkspaceRequest,
+    ) -> dict[str, Any]:
+        return await handlers.create_wrapper_from_workspace(
+            workspace_id=request.workspace_id,
+            artifact_id=request.artifact_id,
+            version=request.version,
+            title=request.title,
+            description=request.description,
+            outcomes=request.outcomes,
+            required_capabilities={
+                name: (
+                    capability.model_dump()
+                    if isinstance(capability, RequiredCapability)
+                    else capability
+                )
+                for name, capability in (request.required_capabilities or {}).items()
             }
             or None,
             source_bindings=dict(request.source_bindings or {}),

tests/wf_mcp/test_server.py

@@ -76,6 +76,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_step_output_map" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
             assert "wf.workflow.create_artifact_from_workspace" in names
+            assert "wf.workflow.create_wrapper_from_workspace" in names
             assert "wf.workflow.run_deployment" in names
             call_capability_schema = tools_by_name[
                 "wf.workflow.call_capability"
@@ -101,6 +102,12 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
                 == "JSON Schema object. Keep this as ordinary JSON; "
                 "nested schema fields are passed through unchanged."
             )
+            wrapper_workspace_input = tools_by_name[
+                "wf.workflow.create_wrapper_from_workspace"
+            ].inputSchema
+            wrapper_request = wrapper_workspace_input["properties"]["request"]
+            assert "kind" not in wrapper_request["properties"]
+            assert "artifact_id" in wrapper_request["properties"]
 
             echo_result = await client.call_tool(
                 "fixture.personal.echo_tool",
@@ -206,6 +213,7 @@ def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
             assert "wf.workflow.set_step_output_map" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
             assert "wf.workflow.create_artifact_from_workspace" in names
+            assert "wf.workflow.create_wrapper_from_workspace" in names
             assert "wf.workflow.call_capability" in names
             assert "wf.workflow.inspect_artifact" in names
             assert "wf.workflow.list_deployments" in names

tests/wf_mcp/test_workflow_surface.py

@@ -111,6 +111,34 @@ def test_workflow_surface_filters_capabilities_by_source() -> None:
     assert payload["capabilities"][0]["source_id"] == "wf.mcp"
 
 
+def test_workflow_surface_lists_saved_wrapper_capabilities() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_wrapper_caps"
+    )
+    artifact_store.save_artifact(
+        _echo_artifact().model_copy(
+            update={
+                "id": "echo_wrapper",
+                "kind": "wrapper",
+                "description": "Reusable echo wrapper.",
+            }
+        )
+    )
+    artifact_store.save_artifact(_echo_artifact())
+    handlers = _handlers(artifact_store)
+
+    payload = asyncio.run(
+        handlers.list_capabilities(source_id="workflow", query="echo")
+    )
+
+    names = [capability["name"] for capability in payload["capabilities"]]
+    assert names == ["workflow.echo_wrapper.v1"]
+    assert payload["capabilities"][0]["source_id"] == "workflow"
+    assert payload["capabilities"][0]["outcomes"] == ["completed"]
+    assert payload["capabilities"][0]["input_fields"] == ["text"]
+    assert payload["capabilities"][0]["output_fields"] == ["echoed"]
+
+
 def test_workflow_surface_inspects_one_capability() -> None:
     handlers = _handlers(
         FileWorkflowArtifactStore(local_temp_root() / "surface_inspect_cap")
@@ -125,6 +153,27 @@ def test_workflow_surface_inspects_one_capability() -> None:
     assert "input_schema" in payload
 
 
+def test_workflow_surface_inspects_saved_wrapper_capability() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_inspect_wrapper_cap"
+    )
+    artifact_store.save_artifact(
+        _echo_artifact().model_copy(update={"id": "echo_wrapper", "kind": "wrapper"})
+    )
+    handlers = _handlers(artifact_store)
+
+    payload = asyncio.run(
+        handlers.inspect_capability(qualified_name="workflow.echo_wrapper.v1")
+    )
+
+    assert payload["name"] == "workflow.echo_wrapper.v1"
+    assert payload["source_id"] == "workflow"
+    assert payload["kind"] == "wrapper_artifact"
+    assert payload["artifact_id"] == "echo_wrapper"
+    assert payload["outcomes"] == ["completed"]
+    assert "input_schema" in payload
+
+
 def test_workflow_surface_validates_deployment_dependencies() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_validate")
     artifact_store.save_artifact(_artifact())
@@ -419,9 +468,7 @@ def test_workflow_surface_deletes_draft_workspace() -> None:
         )
     )
 
-    deleted = asyncio.run(
+    deleted = asyncio.run(handlers.delete_draft_workspace(workspace_id="echo_draft"))
-        handlers.delete_draft_workspace(workspace_id="echo_draft")
-    )
     deleted_again = asyncio.run(
         handlers.delete_draft_workspace(workspace_id="echo_draft")
     )
@@ -514,9 +561,7 @@ def test_workflow_surface_validates_draft_workspace_with_live_outcomes() -> None
         )
     )
 
-    payload = asyncio.run(
+    payload = asyncio.run(handlers.validate_draft_workspace(workspace_id="echo_draft"))
-        handlers.validate_draft_workspace(workspace_id="echo_draft")
-    )
     fetched = asyncio.run(handlers.get_draft_workspace(workspace_id="echo_draft"))
 
     assert payload["revision"] == 1
@@ -629,6 +674,43 @@ def test_workflow_surface_creates_artifact_from_workspace() -> None:
     assert artifact.id == "workspace_echo"
 
 
+def test_workflow_surface_creates_wrapper_from_workspace() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_workspace_wrapper"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_workspace_wrapper_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+    asyncio.run(
+        handlers.create_draft_workspace(
+            workspace_id="echo_draft",
+            draft=_echo_draft(),
+        )
+    )
+
+    result = asyncio.run(
+        handlers.create_wrapper_from_workspace(
+            workspace_id="echo_draft",
+            artifact_id="workspace_echo_wrapper",
+            version=1,
+            title="Workspace Echo Wrapper",
+            outcomes=("completed",),
+            source_bindings={"demo": "demo.personal"},
+        )
+    )
+    artifact = artifact_store.get_artifact("workspace_echo_wrapper", 1)
+
+    assert result["saved"] is True
+    assert artifact.kind == "wrapper"
+    assert artifact.plan["nodes"][0]["node"] == "demo.echo_tool"
+
+
 def test_raw_workflow_plan_uses_core_step_and_edge_models() -> None:
     plan = RawWorkflowPlan.model_validate(_echo_artifact().plan)