uhhh add support for WHAT Where

AGENTS.md

@@ -16,4 +16,11 @@ uv run basedpyright --level error # error to cut spam
 
 or so i think.
 
-<!-- if this file comes with every request, tell me, and you have perms to cut the files down. goo goo ga ga. use caveman skill (only) for repeated artifacts. -->
+<!-- if this file comes with every request, tell me, and you have perms to cut the files down. goo goo ga ga. use caveman skill (only) for repeated artifacts. -->
+
+# partial impls
+
+if a capability is partial until something else frees, or whatever else of the same kind, and you note it in docs,
+you also put a short comment/docstrings at the site to state the problem, and may also refer to the docs there.
+
+<!-- should this be global? putting docstrings/comment is global -->

docs/f/mre.py

@@ -4,14 +4,16 @@ import mcp.types as mcp_types
 from fastmcp import Client
 from fastmcp.client.transports import FastMCPTransport
 from fastmcp.server import create_proxy
-from mcp.server.fastmcp import Context, FastMCP
+from fastmcp import Context, FastMCP
+from fastmcp.dependencies import CurrentContext
 from pydantic import AnyUrl
 
 server = FastMCP("notification-fixture")
 
 
 @server.tool()
-async def emit_notifications_tool(ctx: Context) -> dict[str, bool]:
+async def emit_notifications_tool(ctx: Context = CurrentContext()) -> dict[str, bool]:
+    assert ctx.request_context is not None
     await ctx.request_context.session.send_tool_list_changed()
     await ctx.request_context.session.send_resource_list_changed()
     await ctx.request_context.session.send_prompt_list_changed()

docs/superpowers/plans/2026-05-17-call-wrapper-artifacts.md

@@ -0,0 +1,64 @@
+# Call Wrapper Artifacts Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Let `wf.workflow.call_capability` execute saved `WorkflowArtifact(kind="wrapper")` artifacts by their stable artifact node name without turning arbitrary saved workflows into node capabilities.
+
+**Architecture:** Keep live `NodeSpec` execution unchanged. Extend the workflow surface resolution path so `workflow.<artifact_id>.v<version>` can resolve to a saved wrapper artifact, validate that it is wrapper-kind and interrupt-free, execute its stored plan through the existing workflow runner, and normalize the final workflow result into the same `qualified_name` / `outcome` / `output` payload shape as live capabilities.
+
+**Tech Stack:** Python, Pydantic, `wf_artifacts`, `wf_core`, pytest.
+
+---
+
+### Task 1: Pin Wrapper-Artifact Call Semantics
+
+**Files:**
+- Modify: `tests/wf_mcp/test_service.py`
+
+- [ ] **Step 1: Write the failing test**
+
+Add a test that saves a wrapper artifact with a simple one-node plan, calls `WorkflowSurfaceHandlers.call_capability()` with `workflow.<id>.v<version>`, and asserts the returned `qualified_name`, `outcome`, and `output`.
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run --with pytest pytest tests/wf_mcp/test_service.py -q`
+
+Expected: FAIL because `call_capability()` only resolves live specs today.
+
+### Task 2: Resolve Wrapper Artifacts in the Workflow Surface
+
+**Files:**
+- Modify: `src/wf_mcp/workflow_surface/handlers.py`
+
+- [ ] **Step 1: Implement minimal wrapper-artifact resolution**
+
+Add a small helper that:
+- recognizes stable artifact node names
+- loads the artifact from the store
+- rejects non-wrapper artifacts
+- rejects unsupported interrupt plans
+- executes the artifact plan with the existing workflow runner
+- converts the workflow run into the `call_capability` response payload
+
+- [ ] **Step 2: Run focused tests**
+
+Run: `uv run --with pytest pytest tests/wf_mcp/test_service.py -q`
+
+Expected: PASS.
+
+### Task 3: Verify the Whole Project
+
+**Files:**
+- No additional files.
+
+- [ ] **Step 1: Run focused workflow-surface tests**
+
+Run: `uv run --with pytest pytest tests/wf_mcp -q`
+
+Expected: PASS.
+
+- [ ] **Step 2: Run the full suite**
+
+Run: `uv run --with pytest pytest -q`
+
+Expected: PASS.

src/wf_artifacts/catalog.py

@@ -9,6 +9,9 @@ class WorkflowArtifactCatalogEntry(BaseModel):
     """NodeSpec-shaped projection of a saved workflow artifact."""
 
     name: str
+    artifact_id: str
+    version: int
+    kind: str
     display_name: str
     description: str | None = None
     outcomes: tuple[str, ...]
@@ -37,6 +40,9 @@ def artifact_catalog_entry(
     )
     return WorkflowArtifactCatalogEntry(
         name=artifact_node_name(artifact),
+        artifact_id=artifact.id,
+        version=artifact.version,
+        kind=artifact.kind,
         display_name=artifact.title,
         description=artifact.description,
         outcomes=artifact.outcomes,

src/wf_mcp/broker/artifact_tools.py

@@ -5,6 +5,7 @@ from typing import Any
 from mcp.server.fastmcp import FastMCP
 
 from ..workflow_surface import WorkflowSurfaceHandlers
+from ..models import RawWorkflowPlan
 from .service import WfMcpService
 
 
@@ -25,7 +26,7 @@ def register_artifact_tools(server: FastMCP, service: WfMcpService) -> None:
         artifact_id: str,
         version: int,
         title: str,
-        plan: dict[str, Any],
+        plan: RawWorkflowPlan,
         outcomes: list[str],
         description: str | None = None,
         required_capabilities: dict[str, dict[str, Any]] | None = None,

src/wf_mcp/broker/service/core.py

@@ -523,9 +523,9 @@ class WfMcpService:
     def compile_plan(self, plan: RawWorkflowPlan) -> Workflow:
         node_defs: dict[str, Any] = {}
         for step in plan.nodes:
-            if step.get("type") != "node":
+            if not isinstance(step, NodeUse):
                 continue
-            qualified_name = step["node"]
+            qualified_name = step.node
             spec = self._get_qualified_spec(qualified_name)
             node_defs[qualified_name] = spec.to_node_def()
 
@@ -536,8 +536,8 @@ class WfMcpService:
             "output_schema": plan.output_schema,
             "start": plan.start,
             "node_defs": [node.model_dump() for node in node_defs.values()],
-            "nodes": plan.nodes,
-            "edges": plan.edges,
+            "nodes": [node.model_dump(by_alias=True) for node in plan.nodes],
+            "edges": [edge.model_dump(by_alias=True) for edge in plan.edges],
         }
         return Workflow.model_validate(payload)
 

src/wf_mcp/models.py

@@ -4,6 +4,10 @@ from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import Any
 
+from pydantic import BaseModel
+from wf_core import Edge
+from wf_core.models.steps import Step
+
 from .capabilities import CatalogNodeEntry, CatalogPromptEntry, CatalogResourceEntry
 
 
@@ -38,15 +42,16 @@ class CatalogSnapshot:
         return age_ms > self.max_age_seconds * 1000
 
 
-@dataclass(slots=True)
-class RawWorkflowPlan:
+class RawWorkflowPlan(BaseModel):
+    """Raw authoring plan using the same graph step and edge models as core."""
+
     name: str
     input_schema: dict[str, Any]
     state_schema: dict[str, Any]
     output_schema: dict[str, Any]
     start: str
-    nodes: list[dict[str, Any]]
-    edges: list[dict[str, Any]]
+    nodes: list[Step]
+    edges: list[Edge]
 
 
 @dataclass(slots=True)

src/wf_mcp/transparent_proxy/runtime.py

@@ -35,6 +35,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     # Stable workflow control surface. Keep future workflow-capability test
     # tools pinned here too; they are distinct from raw MCP tool execution.
     "wf.workflow.list_artifacts",
+    "wf.workflow.create_artifact_from_plan",
     "wf.workflow.call_capability",
     "wf.workflow.inspect_artifact",
     "wf.workflow.list_deployments",
@@ -186,7 +187,9 @@ class ProxyRuntime:
     @staticmethod
     def _enabled_connection_ids(config: BrokerConfig) -> set[str]:
         """Return connection ids that currently contribute mounted proxies."""
-        return {connection.id for connection in config.connections if connection.enabled}
+        return {
+            connection.id for connection in config.connections if connection.enabled
+        }
 
 
 TransparentProxyRuntime = ProxyRuntime

src/wf_mcp/workflow_surface/handlers.py

@@ -4,6 +4,7 @@ from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
 from wf_artifacts import (
+    ArtifactKind,
     AvailableCapability,
     AvailableSource,
     DependencyDiagnostic,
@@ -48,18 +49,57 @@ class WorkflowSurfaceHandlers:
         payload: dict[str, Any],
     ) -> dict[str, Any]:
         """Execute one planner-visible workflow capability for authoring tests."""
+        wrapper_artifact = self._wrapper_artifact_for_capability_name(qualified_name)
+        if wrapper_artifact is not None:
+            return await self._call_wrapper_artifact(wrapper_artifact, payload)
+
         spec = self.service._get_qualified_spec(qualified_name)
         handler = build_async_registry(spec)[spec.name]
-        result = await handler(
-            payload,
-            RuntimeContext(current_node_id=spec.name),
-        )
+        result = await handler(payload, RuntimeContext(current_node_id=spec.name))
         return {
             "qualified_name": spec.name,
             "outcome": result["outcome"],
             "output": result["output"],
         }
 
+    def _wrapper_artifact_for_capability_name(
+        self,
+        qualified_name: str,
+    ) -> WorkflowArtifact | None:
+        """Resolve a saved node-like wrapper artifact from its stable capability name."""
+        parsed = _parse_artifact_capability_id(qualified_name)
+        if parsed is None or self.service.artifact_store is None:
+            return None
+        artifact_id, version = parsed
+        try:
+            artifact = self.service.artifact_store.get_artifact(artifact_id, version)
+        except KeyError:
+            return None
+        if artifact.kind != "wrapper":
+            return None
+        return artifact
+
+    async def _call_wrapper_artifact(
+        self,
+        artifact: WorkflowArtifact,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Execute a saved wrapper artifact through the workflow runner."""
+        unsupported = _unsupported_interrupt_diagnostic(artifact)
+        if unsupported is not None:
+            raise ValueError(unsupported.message)
+
+        # For now only wrapper artifacts are honest node capabilities here.
+        # Full saved workflows stay on `run_deployment` until core supports
+        # graph-as-node semantics instead of us faking subgraphs at this layer.
+        plan = _raw_plan_from_artifact(artifact)
+        run = await self.service.run_workflow_from_plan(plan, payload)
+        return {
+            "qualified_name": _artifact_capability_id(artifact),
+            "outcome": run.status.value,
+            "output": run.output,
+        }
+
     async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
@@ -87,20 +127,27 @@ class WorkflowSurfaceHandlers:
         artifact_id: str,
         version: int,
         title: str,
-        plan: dict[str, Any],
+        plan: RawWorkflowPlan | dict[str, Any],
         outcomes: Sequence[str],
+        kind: ArtifactKind = "workflow",
         description: str | None = None,
         required_capabilities: dict[str, dict[str, Any]] | None = None,
         created_from_catalog_version: str | None = None,
     ) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
+        typed_plan = (
+            plan
+            if isinstance(plan, RawWorkflowPlan)
+            else RawWorkflowPlan.model_validate(plan)
+        )
         workflow_artifact = build_workflow_artifact_from_plan(
             artifact_id=artifact_id,
             version=version,
             title=title,
+            kind=kind,
             description=description,
-            plan=plan,
+            plan=typed_plan.model_dump(mode="json", by_alias=True),
             outcomes=tuple(outcomes),
             required_capabilities={
                 name: RequiredCapability.model_validate(capability)
@@ -260,16 +307,32 @@ def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
     return f"workflow.{artifact.id}.v{artifact.version}"
 
 
+# this feels like a hack
+def _parse_artifact_capability_id(qualified_name: str) -> tuple[str, int] | None:
+    """Parse the stable `workflow.<artifact_id>.v<version>` capability name."""
+    prefix = "workflow."
+    if not qualified_name.startswith(prefix):
+        return None
+    artifact_part, separator, version_part = qualified_name[len(prefix) :].rpartition(
+        ".v"
+    )
+    if not separator or not artifact_part or not version_part.isdecimal():
+        return None
+    return artifact_part, int(version_part)
+
+
 def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
     """Validate the stored plan shape expected by the broker workflow runner."""
-    return RawWorkflowPlan(
-        name=_plan_field(artifact, "name"),
-        input_schema=_plan_field(artifact, "input_schema"),
-        state_schema=_plan_field(artifact, "state_schema"),
-        output_schema=_plan_field(artifact, "output_schema"),
-        start=_plan_field(artifact, "start"),
-        nodes=_plan_field(artifact, "nodes"),
-        edges=_plan_field(artifact, "edges"),
+    return RawWorkflowPlan.model_validate(
+        {
+            "name": _plan_field(artifact, "name"),
+            "input_schema": _plan_field(artifact, "input_schema"),
+            "state_schema": _plan_field(artifact, "state_schema"),
+            "output_schema": _plan_field(artifact, "output_schema"),
+            "start": _plan_field(artifact, "start"),
+            "nodes": _plan_field(artifact, "nodes"),
+            "edges": _plan_field(artifact, "edges"),
+        }
     )
 
 

src/wf_mcp/workflow_surface/tools.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Mapping
 
 from fastmcp import FastMCP
 
+from wf_artifacts import ArtifactKind
+from wf_artifacts.models import RequiredCapability
 from wf_mcp.broker.service import WfMcpService
 
+from ..models import RawWorkflowPlan
 from .handlers import WorkflowSurfaceHandlers
 
 
@@ -55,20 +58,32 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         artifact_id: str,
         version: int,
         title: str,
-        plan: dict[str, Any],
+        plan: RawWorkflowPlan,
         outcomes: list[str],
+        kind: ArtifactKind = "workflow",
         description: str | None = None,
-        required_capabilities: dict[str, dict[str, Any]] | None = None,
+        required_capabilities: (
+            Mapping[str, RequiredCapability | dict[str, Any]] | None
+        ) = None,
         created_from_catalog_version: str | None = None,
     ) -> dict[str, Any]:
         return await handlers.create_artifact_from_plan(
             artifact_id=artifact_id,
             version=version,
             title=title,
+            kind=kind,
             description=description,
             plan=plan,
             outcomes=outcomes,
-            required_capabilities=required_capabilities,
+            required_capabilities={
+                name: (
+                    capability.model_dump()
+                    if isinstance(capability, RequiredCapability)
+                    else capability
+                )
+                for name, capability in (required_capabilities or {}).items()
+            }
+            or None,
             created_from_catalog_version=created_from_catalog_version,
         )
 

tests/wf_mcp/test_workflow_surface.py

@@ -11,6 +11,7 @@ from wf_artifacts import (
 )
 from wf_mcp.broker import WfMcpService
 from wf_mcp.models import ConnectionConfig
+from wf_mcp.models import RawWorkflowPlan
 from wf_mcp.storage import FileStore
 from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
 
@@ -27,6 +28,9 @@ def test_workflow_surface_lists_artifact_catalog_entries() -> None:
     nodes = payload["nodes"]
     assert len(nodes) == 1
     assert nodes[0]["name"] == "workflow.summarize_docs.v1"
+    assert nodes[0]["artifact_id"] == "summarize_docs"
+    assert nodes[0]["version"] == 1
+    assert nodes[0]["kind"] == "workflow"
     assert nodes[0]["required_sources"] == ["context7"]
     assert "plan" not in nodes[0]
 
@@ -81,6 +85,35 @@ def test_workflow_surface_records_artifact_and_deployment_save_events() -> None:
     assert events[1].capability_id == "deployment.echo.personal"
 
 
+def test_workflow_surface_creates_wrapper_artifact_from_plan() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_wrapper_plan"
+    )
+    handlers = _handlers(artifact_store)
+
+    payload = asyncio.run(
+        handlers.create_artifact_from_plan(
+            artifact_id="echo_wrapper",
+            version=1,
+            title="Echo Wrapper",
+            kind="wrapper",
+            plan=_echo_artifact().plan,
+            outcomes=("completed",),
+        )
+    )
+    artifact = artifact_store.get_artifact("echo_wrapper", 1)
+
+    assert payload["saved"] is True
+    assert artifact.kind == "wrapper"
+
+
+def test_raw_workflow_plan_uses_core_step_and_edge_models() -> None:
+    plan = RawWorkflowPlan.model_validate(_echo_artifact().plan)
+
+    assert plan.nodes[0].type == "node"
+    assert plan.edges[0].outcome == "ok"
+
+
 def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_run")
     artifact_store.save_artifact(_echo_artifact())
@@ -114,6 +147,36 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     assert payload["diagnostics"] == []
 
 
+def test_workflow_surface_calls_saved_wrapper_artifact() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_wrapper_call"
+    )
+    wrapper = _echo_artifact().model_copy(
+        update={"id": "echo_wrapper", "kind": "wrapper"}
+    )
+    artifact_store.save_artifact(wrapper)
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_wrapper_call_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)
+
+    payload = asyncio.run(
+        handlers.call_capability(
+            qualified_name="workflow.echo_wrapper.v1",
+            payload={"text": "hello"},
+        )
+    )
+
+    assert payload["qualified_name"] == "workflow.echo_wrapper.v1"
+    assert payload["outcome"] == "completed"
+    assert payload["output"]["echoed"] == "hello"
+
+
 def _handlers(artifact_store: FileWorkflowArtifactStore) -> WorkflowSurfaceHandlers:
     service = WfMcpService(
         store=FileStore(local_temp_root() / "surface_mcp"),